150 lines
6.8 KiB
Python
150 lines
6.8 KiB
Python
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}
|