import hashlib import mimetypes import re from pathlib import Path, PurePosixPath from zipfile import ZipFile from xml.etree import ElementTree from .config import get_settings CATEGORIES = ("stage1", "stage2", "stage3", "final_report", "presentation") SUPPORTED = {".pdf", ".docx", ".pptx"} NAMES = {"stage1": "阶段报告1", "stage2": "阶段报告2", "stage3": "阶段报告3", "final_report": "最终报告", "presentation": "汇报PPT"} def secure_path(relative: str, *, root: Path | None = None, must_exist=True) -> Path: root = (root or get_settings().inbox_path).resolve() if not relative or "\\" in relative or "\x00" in relative: raise ValueError("必须使用挂载目录内的相对路径") path = PurePosixPath(relative) if path.is_absolute() or any(part in ("..", "") for part in path.parts): raise ValueError("路径超出挂载目录") target = root.joinpath(*path.parts) if any(p.is_symlink() for p in (target, *target.parents) if p != root and root in p.parents): raise ValueError("不允许符号链接") resolved = target.resolve(strict=must_exist) if not resolved.is_relative_to(root): raise ValueError("路径超出挂载目录") return resolved def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def parse_student_folder(name: str): parts = name.split("_", 2) if len(parts) != 3 or not re.fullmatch(r"\d{5,20}", parts[0]) or not all(parts[1:]): return {"student_no": "", "name": "", "topic": "", "status": "needs_confirmation"} return {"student_no": parts[0], "name": parts[1].strip(), "topic": parts[2].strip(), "status": "recognized"} def extract_identity(text: str): patterns = { "student_no": r"学号\s*[:：]?\s*(\d{5,20})", "name": r"姓名\s*[:：]?\s*([^\s:：]{2,10})", "topic": r"(?:课题名称|项目名称|题目)\s*[:：]?\s*([^\n\r]{3,100})", } return {key: (match.group(1).strip() if (match := re.search(pattern, text)) else "") for key, pattern in patterns.items()} def first_text(path: Path) -> str: try: if path.suffix.lower() == ".pdf": import pymupdf as fitz with fitz.open(path) as doc: return doc[0].get_text()[:6000] if len(doc) else "" if path.suffix.lower() in {".docx", ".pptx"}: part = "word/document.xml" if path.suffix.lower() == ".docx" else "ppt/slides/slide1.xml" with ZipFile(path) as archive: root = ElementTree.fromstring(archive.read(part)) return " ".join(element.text or "" for element in root.iter() if element.tag.endswith("}t"))[:6000] except Exception as exc: raise ValueError(f"无法读取文件首页: {exc}") from exc return "" def classify(path: Path, text: str = ""): suffix = path.suffix.lower() if suffix not in SUPPORTED: return "unknown", "unsupported" def identify(value): value = value.lower().replace(" ", "") if any(term in value for term in ("汇报ppt", "汇报幻灯", "答辩ppt", "演示文稿", "presentation")): return "presentation" for digit, chinese, category in (("1", "一", "stage1"), ("2", "二", "stage2"), ("3", "三", "stage3")): if any(term in value for term in (f"阶段报告{digit}", f"阶段报告{chinese}", f"第{chinese}阶段", f"阶段{digit}报告")): return category if any(term in value for term in ("最终报告", "期末报告", "综合实践报告", "结题报告")): return "final_report" return None if suffix == ".pptx": return "presentation", "recognized" category = identify(path.stem) or identify(text[:600]) if category: return category, "needs_confirmation" if category == "presentation" and suffix == ".pdf" else "recognized" return "unknown", "needs_confirmation" def directories(relative: str = ".", *, root: Path | None = None): path = secure_path(relative, root=root) if not path.is_dir(): raise ValueError("请选择目录") base = (root or get_settings().inbox_path).resolve() return [{"name": item.name, "relative_path": item.relative_to(base).as_posix()} for item in sorted(path.iterdir()) if item.is_dir() and not item.is_symlink()] def scan(relative: str, *, root: Path | None = None): base = (root or get_settings().inbox_path).resolve() path = secure_path(relative, root=base) if not path.is_dir(): raise ValueError("请选择学生总目录") projects = [] for folder in sorted(path.iterdir()): if not folder.is_dir() or folder.is_symlink(): continue info = parse_student_folder(folder.name) files = [] extracted = {} warnings = [] for candidate in sorted(folder.iterdir()): if not candidate.is_file() or candidate.is_symlink(): continue if candidate.suffix.lower() not in SUPPORTED: continue error = "" try: text = first_text(candidate) category, status = classify(candidate, text) identity = extract_identity(text) for key, value in identity.items(): if value and key not in extracted: extracted[key] = value except ValueError as exc: category, status, error = "unknown", "unparseable", str(exc) stat = candidate.stat() files.append({"relative_path": candidate.relative_to(base).as_posix(), "category": category, "status": status, "size": stat.st_size, "mtime": stat.st_mtime, "sha256": sha256(candidate), "mime_type": mimetypes.guess_type(candidate.name)[0] or "application/octet-stream", "error": error, "first_text": text[:500] if not error else ""}) for category in CATEGORIES: matches = [file for file in files if file["category"] == category] if len(matches) > 1: for file in matches: file["status"] = "duplicate" for key, value in extracted.items(): if info.get(key) and value != info[key]: warnings.append(f"封面{key}“{value}”与目录名“{info[key]}”不一致") if warnings: info["status"] = "needs_confirmation" info.update({"relative_dir": folder.relative_to(base).as_posix(), "files": files, "detected_identity": extracted, "warnings": warnings, "missing": [category for category in CATEGORIES if not any(file["category"] == category for file in files)]}) projects.append(info) return {"relative_dir": relative, "projects": projects}