""" File parser for Offer Catcher. Supports .txt, .md, .pdf, and .docx. Pure-local parsing: no file is saved to disk and no API call is made. """ from typing import Tuple def extract_text_from_upload(uploaded_file) -> Tuple[str, str]: """ Read a Streamlit UploadedFile and extract text. Returns: (text, message) - text: extracted resume text, or "" on failure - message: UI-friendly status message """ if uploaded_file is None: return "", "未上传文件。" filename = getattr(uploaded_file, "name", "unknown") suffix = "" if "." in filename: suffix = "." + filename.rsplit(".", 1)[-1].lower() if suffix in (".txt", ".md"): return _read_text_file(uploaded_file, filename) if suffix == ".pdf": return _read_pdf(uploaded_file, filename) if suffix == ".docx": return _read_docx(uploaded_file, filename) if suffix == ".doc": return ( "", f"不支持 .doc 格式({filename})。\n" "请先将文件另存为 .docx 或 PDF,再重新上传。", ) return "", f"不支持的文件格式:{suffix}({filename})。" def _reset_file(uploaded_file) -> None: if hasattr(uploaded_file, "seek"): try: uploaded_file.seek(0) except Exception: pass def _read_text_file(uploaded_file, filename: str) -> Tuple[str, str]: raw = uploaded_file.read() _reset_file(uploaded_file) if not raw: return "", f"文件为空({filename})。" text = None for encoding in ("utf-8", "utf-8-sig", "gbk", "gb18030", "latin-1"): try: text = raw.decode(encoding) break except UnicodeDecodeError: continue if text is None: return "", f"无法识别文件编码({filename})。请另存为 UTF-8 文本后重试。" text = text.strip() if len(text) < 30: return "", f"文件内容过少(仅 {len(text)} 字符,{filename}),请检查文件是否正确。" return text, f"已解析 {filename}({len(text)} 字符,文本格式)。" def _read_pdf(uploaded_file, filename: str) -> Tuple[str, str]: try: import pypdf except ImportError: return ( "", f"服务器未安装 pypdf,无法解析 PDF({filename})。\n" "请在 requirements.txt 中加入 pypdf>=4.0.0 并重新部署。", ) try: _reset_file(uploaded_file) reader = pypdf.PdfReader(uploaded_file) parts = [] for page in reader.pages: try: page_text = page.extract_text() except Exception: page_text = "" if page_text: parts.append(page_text) text = "\n".join(parts).strip() if len(text) < 30: return ( "", f"PDF 解析结果过少(仅 {len(text)} 字符,{filename})。\n" "可能是扫描版 PDF,请尝试复制文本后粘贴,或上传 DOCX/PDF 文字版。", ) return text, f"已解析 {filename}({len(text)} 字符,PDF {len(reader.pages)} 页)。" except Exception as exc: return "", f"PDF 解析失败({filename}):{exc}" def _read_docx(uploaded_file, filename: str) -> Tuple[str, str]: try: import docx except ImportError: return ( "", f"服务器未安装 python-docx,无法解析 DOCX({filename})。\n" "请在 requirements.txt 中加入 python-docx>=1.1.0 并重新部署。", ) try: _reset_file(uploaded_file) document = docx.Document(uploaded_file) parts = [] for paragraph in document.paragraphs: value = paragraph.text.strip() if value: parts.append(value) for table in document.tables: for row in table.rows: cells = [cell.text.strip() for cell in row.cells if cell.text.strip()] if cells: parts.append(" | ".join(cells)) text = "\n".join(parts).strip() if len(text) < 30: return "", f"DOCX 解析结果过少(仅 {len(text)} 字符,{filename}),请检查文件是否正确。" return text, f"已解析 {filename}({len(text)} 字符,DOCX)。" except Exception as exc: return "", f"DOCX 解析失败({filename}):{exc}"