"""Bounded file ingestion. Office renders locally; inference runs on OpenRouter.""" import base64 import csv import io import json import shutil import subprocess import tempfile import zipfile from pathlib import Path ALLOWED = [".pdf", ".doc", ".docx", ".ppt", ".pptx", ".xls", ".xlsx", ".csv", ".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tif", ".tiff"] OFFICE = {".doc", ".docx", ".ppt", ".pptx"} IMAGES = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tif", ".tiff"} MAX_FILE_BYTES = 15 * 1024 * 1024 MAX_BATCH_BYTES = 20 * 1024 * 1024 MAX_TEXT = 250_000 MAX_CELLS = 100_000 MAX_PDF_PAGES = 100 class FileError(ValueError): pass def check_zip(path: Path) -> None: if path.suffix.lower() not in {".docx", ".pptx", ".xlsx"}: return with zipfile.ZipFile(path) as archive: entries = archive.infolist() if len(entries) > 15_000 or sum(x.file_size for x in entries) > 100 * 1024 * 1024: raise FileError("Office 文件解压后过大，请拆分后上传。") if any("vbaproject" in x.filename.lower() for x in entries): raise FileError("不接受包含宏的 Office 文件，请另存为无宏版本。") def data_url(data: bytes, mime: str) -> str: return f"data:{mime};base64," + base64.b64encode(data).decode("ascii") def pdf_part(data: bytes, name: str) -> tuple[dict, int]: from pypdf import PdfReader if len(data) > MAX_FILE_BYTES: raise FileError("转换后的 PDF 超过 15 MB，请拆分后上传。") pdf = PdfReader(io.BytesIO(data)) if pdf.is_encrypted: raise FileError("PDF 已加密，请先移除密码。") pages = len(pdf.pages) if not 0 < pages <= MAX_PDF_PAGES: raise FileError("单次附件合计最多 100 页 PDF，请拆分文档。") return {"type": "file", "file": {"filename": name, "file_data": data_url(data, "application/pdf")}}, pages def office_pdf(path: Path) -> bytes: executable = shutil.which("libreoffice") or shutil.which("soffice") if not executable: raise FileError("缺少 LibreOffice。Space 需安装 packages.txt；也可先将文件另存为 PDF。") with tempfile.TemporaryDirectory(prefix="claude-office-") as folder: root = Path(folder) source = root / ("document" + path.suffix.lower()) shutil.copyfile(path, source) profile = root / "profile" settings = profile / "user" settings.mkdir(parents=True) # Block Office macros in the isolated conversion profile. (settings / "registrymodifications.xcu").write_text( '' '' '3' '', encoding="utf-8") try: result = subprocess.run([ executable, f"-env:UserInstallation={profile.as_uri()}", "--headless", "--nologo", "--nodefault", "--norestore", "--convert-to", "pdf", "--outdir", str(root), str(source), ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=90, check=False) except subprocess.TimeoutExpired: raise FileError("Office 转换超过 90 秒，请拆分文档或先导出 PDF。") from None output = root / "document.pdf" if result.returncode != 0 or not output.exists(): raise FileError("Office 转换失败，可能是加密或损坏文件；请先导出 PDF。") if output.stat().st_size > MAX_FILE_BYTES: raise FileError("转换后的 PDF 超过 15 MB，请拆分后上传。") return output.read_bytes() def spreadsheet_text(path: Path) -> str: output = ["电子表格内容；保留工作表名、行号与单元格位置。公式为原式，不在服务器执行。"] length, cells = len(output[0]), 0 def add(text: str) -> None: nonlocal length length += len(text) + 1 if length > MAX_TEXT: raise FileError("表格文字超过 25 万字符，请按工作表或行拆分；未发送截断内容。") output.append(text) def rows(name: str, iterator) -> None: nonlocal cells add("\n工作表：" + name) for number, row in enumerate(iterator, 1): values = list(row) cells += len(values) if cells > MAX_CELLS: raise FileError("表格扫描范围超过 10 万个单元格，请缩小使用区域或拆分文件。") if any(v is not None and v != "" for v in values): add(f"行 {number}: " + json.dumps(values, ensure_ascii=False, default=str)) suffix = path.suffix.lower() if suffix == ".xlsx": from openpyxl import load_workbook workbook = load_workbook(path, read_only=True, data_only=False, keep_links=False) try: for sheet in workbook: if sheet.max_row * sheet.max_column > MAX_CELLS: raise FileError("工作表范围超过 10 万个单元格，请清理空白格式区域或拆分文件。") rows(sheet.title, sheet.iter_rows(values_only=True)) finally: workbook.close() elif suffix == ".xls": import xlrd workbook = xlrd.open_workbook(path, on_demand=True) try: for sheet in workbook.sheets(): rows(sheet.name, (sheet.row_values(i) for i in range(sheet.nrows))) add("说明：旧版 XLS 使用已保存的计算结果，日期可能为 Excel 日期序列值。") finally: workbook.release_resources() else: raw = path.read_bytes() try: text = raw.decode("utf-8-sig") except UnicodeDecodeError: text = raw.decode("gb18030") rows(path.name, csv.reader(io.StringIO(text))) return "\n".join(output) def prepare_files(paths: list[str]) -> tuple[list[dict], list[str], int]: """Return model content parts, display summaries, and PDF page count.""" if len(paths) > 8: raise FileError("每次最多上传 8 个文件。") parts, notes, pages, total_size = [], [], 0, 0 for filename in paths: path = Path(filename) suffix = path.suffix.lower() if not path.is_file() or suffix not in ALLOWED: raise FileError("文件不存在或格式不支持。") size = path.stat().st_size total_size += size if size > MAX_FILE_BYTES or total_size > MAX_BATCH_BYTES: raise FileError("单文件最多 15 MB，每次附件合计最多 20 MB。") try: check_zip(path) if suffix in OFFICE or suffix == ".pdf": data = office_pdf(path) if suffix in OFFICE else path.read_bytes() part, count = pdf_part(data, path.stem + ".pdf") pages += count if pages > MAX_PDF_PAGES: raise FileError("单次附件合计最多 100 页 PDF，请拆分后上传。") parts.append(part) notes.append(f"{path.name} · {count} 页" + (" · 已转为 PDF" if suffix in OFFICE else "")) elif suffix in IMAGES: from PIL import Image, ImageOps Image.MAX_IMAGE_PIXELS = 25_000_000 with Image.open(path) as source: if getattr(source, "n_frames", 1) > 1: raise FileError("暂不接受多帧图片，请拆分为单张图片后上传。") picture = ImageOps.exif_transpose(source).convert("RGB") resized = max(picture.size) > 2048 picture.thumbnail((2048, 2048)) buffer = io.BytesIO() picture.save(buffer, format="JPEG", quality=92) parts.extend([{"type": "text", "text": f"图片附件：{path.name}"}, {"type": "image_url", "image_url": {"url": data_url(buffer.getvalue(), "image/jpeg")}}]) notes.append(path.name + (" · 已缩放至最长边 2048px" if resized else "")) else: text = spreadsheet_text(path) parts.append({"type": "text", "text": f"附件：{path.name}\n{text}"}) notes.append(f"{path.name} · 表格文本 {len(text):,} 字符") except FileError: raise except Exception: raise FileError(f"无法解析 {path.name}，请检查文件是否损坏、加密或与扩展名不符。") from None return parts, notes, pages