96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
"""简历文字处理
|
|
|
|
将上传的简历文件提取为纯文本，供后续 AI 重写/结构化使用。
|
|
支持格式：.txt / .md / .pdf / .docx
|
|
所有解析均为异步（阻塞解析下沉到线程池，避免阻塞事件循环）。
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
|
|
import fitz # PyMuPDF
|
|
from docx import Document
|
|
|
|
from app.core.asserts import Assert
|
|
from app.core.logger import log
|
|
|
|
# 支持的简历文件类型
|
|
SUPPORTED_EXTENSIONS = (".txt", ".md", ".pdf", ".docx")
|
|
|
|
# 纯文本解码尝试的编码顺序
|
|
_TEXT_ENCODINGS = ("utf-8", "gbk", "gb2312", "latin-1")
|
|
|
|
|
|
async def _parse_txt(content: bytes) -> str:
|
|
"""解析纯文本 (.txt / .md)：自动探测编码，返回全文文本"""
|
|
|
|
def _decode() -> str:
|
|
for encoding in _TEXT_ENCODINGS:
|
|
try:
|
|
return content.decode(encoding)
|
|
except (UnicodeDecodeError, LookupError):
|
|
continue
|
|
# 全部失败时以替换字符兜底，保证不抛异常
|
|
return content.decode("utf-8", errors="replace")
|
|
|
|
return await asyncio.to_thread(_decode)
|
|
|
|
|
|
async def _parse_pdf(content: bytes) -> str:
|
|
"""解析 PDF：使用 PyMuPDF 逐页 get_text() 拼接为全文文本"""
|
|
|
|
def _extract() -> str:
|
|
parts: list[str] = []
|
|
with fitz.open(stream=content, filetype="pdf") as doc:
|
|
for page in doc:
|
|
parts.append(page.get_text())
|
|
return "\n".join(parts)
|
|
|
|
return await asyncio.to_thread(_extract)
|
|
|
|
|
|
async def _parse_docx(content: bytes) -> str:
|
|
"""解析 Word (.docx)：拼接段落文本，表格行用 \t 连接，返回全文文本"""
|
|
|
|
def _extract() -> str:
|
|
try:
|
|
doc = Document(io.BytesIO(content))
|
|
except Exception:
|
|
raise ValueError("无法解析该 Word 文件，请确认为有效的 .docx 格式")
|
|
parts: list[str] = []
|
|
for para in doc.paragraphs:
|
|
text = para.text.strip()
|
|
if text:
|
|
parts.append(text)
|
|
for table in doc.tables:
|
|
for row in table.rows:
|
|
row_text = "\t".join(cell.text.strip() for cell in row.cells)
|
|
if row_text.strip():
|
|
parts.append(row_text)
|
|
return "\n".join(parts)
|
|
|
|
return await asyncio.to_thread(_extract)
|
|
|
|
|
|
# 后缀 → 解析函数。新增格式只需在此登记一行，并同步 SUPPORTED_EXTENSIONS。
|
|
_PARSERS = {
|
|
".txt": _parse_txt,
|
|
".md": _parse_txt,
|
|
".pdf": _parse_pdf,
|
|
".docx": _parse_docx,
|
|
}
|
|
|
|
|
|
async def extract_text(filename: str, content: bytes) -> str:
|
|
"""简历文字提取统一入口：先校验文件类型受支持，再按后缀路由提取全文文本"""
|
|
suffix = filename[filename.rfind("."):].lower() if "." in filename else ""
|
|
Assert.is_true(
|
|
suffix in SUPPORTED_EXTENSIONS,
|
|
f"不支持的文件类型: {suffix or '未知'}，支持: {', '.join(SUPPORTED_EXTENSIONS)}",
|
|
)
|
|
|
|
log.info(f"开始提取简历文本: {filename}，类型: {suffix}")
|
|
text = await _PARSERS[suffix](content)
|
|
log.info(f"简历文本提取完成: {filename}，字符数: {len(text)}")
|
|
return text
|