Spaces:
Sleeping
Sleeping
| """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( | |
| '<?xml version="1.0" encoding="UTF-8"?><oor:items ' | |
| 'xmlns:oor="http://openoffice.org/2001/registry">' | |
| '<item oor:path="/org.openoffice.Office.Common/Security/Scripting">' | |
| '<prop oor:name="MacroSecurityLevel" oor:op="fuse"><value>3</value></prop>' | |
| '</item></oor:items>', 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 | |