import hashlib import io import json import zipfile from pathlib import Path from typing import Any from fastapi import HTTPException from sqlalchemy import delete, func, select from sqlalchemy.orm import Session from app.grading import answers_json_to_map, grade_item from app.models import ( ExamPaperMap, ExaminationExamExaminee, ExaminationExaminee, ExaminationExamineeExamAnswer, ExaminationExamineeExamAnswerItem, ExaminationPaperItem, ExaminationPaperSource, ImportBatch, ) from app.sql_parser import find_file_content, rows_from_sql def load_zip_files(zip_bytes: bytes) -> dict[str, bytes]: out: dict[str, bytes] = {} with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf: for info in zf.infolist(): if info.is_dir(): continue if info.filename.lower().endswith(".sql"): out[info.filename] = zf.read(info) return out def load_dir_sql(dir_path: Path) -> dict[str, bytes]: out: dict[str, bytes] = {} for p in dir_path.rglob("*.sql"): out[str(p.relative_to(dir_path))] = p.read_bytes() return out def _require_sql_kind(files: dict[str, bytes], kind: str, label: str) -> str: text = find_file_content(files, kind) if not text: raise HTTPException(400, detail=f"ZIP 缺少 {label} SQL 文件") return text def _paper_identity(row: dict[str, Any]) -> tuple[str, str]: return (str(row.get("external_paper_id", "")), str(row.get("revision", ""))) def _collect_exam_ids(exam_rows: list[dict[str, Any]]) -> list[int]: ids = sorted({int(r["exam_id"]) for r in exam_rows if r.get("exam_id") is not None}) if not ids: raise HTTPException(400, detail="考试考生 SQL 中未找到有效的 exam_id") return ids def _parse_exam_to_paper_sql_map(exam_sql: str) -> dict[int, int]: out: dict[int, int] = {} for r in rows_from_sql(exam_sql): eid = r.get("id") if eid is None: continue psid = r.get("paper_source_id") if psid is None: psid = r.get("paper_id") if psid is None: continue out[int(eid)] = int(psid) return out def _resolve_exam_paper_map( exam_ids: list[int], paper_rows: list[dict[str, Any]], files: dict[str, bytes], ) -> dict[int, int]: """exam_id -> paper_source_id（SQL 文件中的试卷 id）。""" keys = {_paper_identity(r) for r in paper_rows} sql_paper_ids = {int(r["id"]) for r in paper_rows if r.get("id") is not None} if len(keys) <= 1: if not paper_rows: raise HTTPException(400, detail="试卷 SQL 无有效数据") canonical = int(paper_rows[0]["id"]) return {eid: canonical for eid in exam_ids} exam_text = find_file_content(files, "exam") if not exam_text: raise HTTPException( 400, detail="ZIP 含多份试卷时，必须提供 examination_exam SQL，用于关联每场考试的试卷", ) meta = _parse_exam_to_paper_sql_map(exam_text) if not meta: raise HTTPException(400, detail="examination_exam SQL 无有效数据或未包含 paper_source_id") result: dict[int, int] = {} for eid in exam_ids: ps_sql = meta.get(eid) if ps_sql is None: raise HTTPException(400, detail=f"examination_exam 中缺少 exam_id={eid} 的试卷关联") if ps_sql not in sql_paper_ids: raise HTTPException( 400, detail=f"exam_id={eid} 关联的 paper_source_id={ps_sql} 不在 ZIP 试卷 SQL 中", ) result[eid] = ps_sql return result def _get_or_create_paper_source( db: Session, batch: ImportBatch, row: dict[str, Any], ) -> int: ext_id, revision = _paper_identity(row) existing = db.scalar( select(ExaminationPaperSource.id).where( ExaminationPaperSource.external_paper_id == ext_id, ExaminationPaperSource.revision == revision, ) ) if existing: return int(existing) ps = ExaminationPaperSource( id=int(row["id"]), external_paper_id=ext_id, revision=revision, sha256=str(row.get("sha256") or ""), paper_name=str(row.get("paper_name") or ""), occupation=str(row.get("occupation") or ""), level=str(row.get("level") or ""), subject=int(row["subject"]) if row.get("subject") is not None else None, struct_json=str(row.get("struct_json") or ""), item_count=int(row["item_count"]) if row.get("item_count") is not None else None, import_batch_id=batch.id, ) db.add(ps) db.flush() return int(ps.id) def _items_for_grading( db: Session, batch: ImportBatch, sql_ps_id: int, db_ps_id: int, item_rows: list[dict[str, Any]], ) -> dict[str, ExaminationPaperItem]: ps = db.get(ExaminationPaperSource, db_ps_id) if not ps: raise HTTPException(500, detail=f"试卷 {db_ps_id} 不存在") if ps.import_batch_id != batch.id: items = list( db.scalars( select(ExaminationPaperItem).where(ExaminationPaperItem.paper_source_id == db_ps_id) ) ) return {it.external_item_id: it for it in items} items_by_ext: dict[str, ExaminationPaperItem] = {} total_max = 0.0 for r in item_rows: if int(r.get("paper_source_id") or 0) != sql_ps_id: continue ms = float(r.get("max_score") or 1) total_max += ms it = ExaminationPaperItem( id=int(r["id"]), paper_source_id=db_ps_id, external_item_id=str(r["external_item_id"]), section_ref=str(r.get("section_ref") or ""), part_ref=str(r.get("part_ref") or ""), part_name=str(r.get("part_name") or ""), item_type=int(r["item_type"]) if r.get("item_type") is not None else None, item_order=int(r["item_order"]) if r.get("item_order") is not None else None, title_html=str(r.get("title_html") or ""), detail_json=str(r.get("detail_json") or ""), max_score=ms, reference_answer=str(r.get("reference_answer") or ""), answer_rules=str(r.get("answer_rules") or ""), ) db.add(it) items_by_ext[it.external_item_id] = it ps.total_score = total_max return items_by_ext def run_import( db: Session, subject_name: str, zip_filename: str, file_bytes: bytes, user_id: int | None, ) -> dict[str, Any]: files = load_zip_files(file_bytes) sha = hashlib.sha256(file_bytes).hexdigest() examinee_sql = _require_sql_kind(files, "examinee", "考生") exam_examinee_sql = _require_sql_kind(files, "exam_examinee", "考试考生") paper_sql = _require_sql_kind(files, "paper_source", "试卷") item_sql = _require_sql_kind(files, "paper_item", "试题") answer_sql = _require_sql_kind(files, "answer", "考生答案") paper_rows = rows_from_sql(paper_sql) if not paper_rows: raise HTTPException(400, detail="试卷 SQL 无有效数据") exam_rows = rows_from_sql(exam_examinee_sql) if not exam_rows: raise HTTPException(400, detail="考试考生 SQL 无有效数据") exam_ids = _collect_exam_ids(exam_rows) for eid in exam_ids: if db.scalar(select(ExamPaperMap.id).where(ExamPaperMap.exam_id == eid)): raise HTTPException(409, detail=f"exam_id {eid} 已存在，请先清除对应导入批次") exam_to_sql_paper = _resolve_exam_paper_map(exam_ids, paper_rows, files) paper_row_by_sql_id = {int(r["id"]): r for r in paper_rows if r.get("id") is not None} needed_sql_paper_ids = {exam_to_sql_paper[e] for e in exam_ids} for sql_id in needed_sql_paper_ids: if sql_id not in paper_row_by_sql_id: raise HTTPException(400, detail=f"试卷 SQL 缺少 id={sql_id}") warnings: list[dict[str, Any]] = [] report: dict[str, Any] = {"warnings": warnings, "counts": {}} primary_paper_row = paper_row_by_sql_id[exam_to_sql_paper[exam_ids[0]]] ext_id, revision = _paper_identity(primary_paper_row) try: batch = ImportBatch( subject_name=subject_name.strip(), zip_filename=zip_filename, sha256=sha, status="success", imported_by=user_id, external_paper_id=ext_id, revision=revision, exam_id=exam_ids[0], exam_ids_json=json.dumps(exam_ids, ensure_ascii=False), occupation=str(primary_paper_row.get("occupation") or ""), level=str(primary_paper_row.get("level") or ""), subject_code=int(primary_paper_row["subject"]) if primary_paper_row.get("subject") is not None else None, ) db.add(batch) db.flush() sql_to_db_paper: dict[int, int] = {} for sql_id in needed_sql_paper_ids: sql_to_db_paper[sql_id] = _get_or_create_paper_source( db, batch, paper_row_by_sql_id[sql_id] ) batch.paper_source_id = sql_to_db_paper[exam_to_sql_paper[exam_ids[0]]] item_rows = rows_from_sql(item_sql) grading_items: dict[int, dict[str, ExaminationPaperItem]] = {} for sql_id in set(exam_to_sql_paper.values()): db_ps = sql_to_db_paper[sql_id] grading_items[db_ps] = _items_for_grading(db, batch, sql_id, db_ps, item_rows) for r in rows_from_sql(examinee_sql): eid = int(r["id"]) existing = db.get(ExaminationExaminee, eid) if existing: existing.username = r.get("username") existing.nickname = r.get("nickname") existing.id_card = r.get("id_card") existing.mobile = r.get("mobile") existing.sex = int(r["sex"]) if r.get("sex") is not None else None else: db.add( ExaminationExaminee( id=eid, username=r.get("username"), nickname=r.get("nickname"), id_card=r.get("id_card"), mobile=r.get("mobile"), sex=int(r["sex"]) if r.get("sex") is not None else None, deleted=int(r.get("deleted") or 0), ) ) db.flush() answer_rows = {int(r["exam_examinee_id"]): r for r in rows_from_sql(answer_sql)} for r in exam_rows: aid = int(r["id"]) row_exam_id = int(r["exam_id"]) imported_score = float(r.get("score") or 0) sql_ps = exam_to_sql_paper[row_exam_id] db_ps = sql_to_db_paper[sql_ps] items_by_ext = grading_items[db_ps] attempt = ExaminationExamExaminee( id=aid, exam_id=row_exam_id, examinee_id=int(r["examinee_id"]), room_id=r.get("room_id"), seat_number=r.get("seat_number"), status=int(r["status"]) if r.get("status") is not None else None, score=0, imported_score=imported_score, level_str=r.get("level_str"), start_time=r.get("start_time"), end_time=r.get("end_time"), subject=int(r["subject"]) if r.get("subject") is not None else None, import_batch_id=batch.id, ) db.add(attempt) ar = answer_rows.get(aid) if not ar: raise HTTPException(400, detail=f"缺少考生答案 exam_examinee_id={aid}") answers_str = str(ar.get("answers") or "[]") db.add( ExaminationExamineeExamAnswer( id=int(ar["id"]), exam_examinee_id=aid, answers=answers_str, deleted=int(ar.get("deleted") or 0), ) ) amap = answers_json_to_map(answers_str) calc = 0.0 for ext, item in items_by_ext.items(): ua = amap.get(ext, "") ok, sc = grade_item( ua, item.reference_answer or "", item.answer_rules, item.item_type, item.max_score, ) calc += sc db.add( ExaminationExamineeExamAnswerItem( exam_examinee_id=aid, external_item_id=ext, answer_raw=ua, is_correct=ok, score=sc, ) ) attempt.score = calc if abs(calc - imported_score) > 0.01: warnings.append( { "exam_examinee_id": aid, "imported_score": imported_score, "calculated_score": calc, } ) for eid in exam_ids: db.add( ExamPaperMap( exam_id=eid, paper_source_id=sql_to_db_paper[exam_to_sql_paper[eid]], import_batch_id=batch.id, ) ) report["counts"] = { "examinees": len(rows_from_sql(examinee_sql)), "attempts": len(exam_rows), "items": len(item_rows), "exam_id": exam_ids[0], "exam_ids": exam_ids, "papers": len(set(sql_to_db_paper.values())), } batch.report_json = json.dumps(report, ensure_ascii=False) db.commit() db.refresh(batch) return {"batch_id": batch.id, "report": report} except HTTPException: db.rollback() raise except Exception as e: db.rollback() raise HTTPException(500, detail=f"导入失败: {e}") from e def _paper_owned_by_batch(db: Session, ps_id: int, batch_id: int) -> bool: ps = db.get(ExaminationPaperSource, ps_id) return ps is not None and ps.import_batch_id == batch_id def _paper_still_referenced(db: Session, ps_id: int, batch_id: int) -> bool: other_maps = db.scalar( select(func.count()) .select_from(ExamPaperMap) .where( ExamPaperMap.paper_source_id == ps_id, ExamPaperMap.import_batch_id != batch_id, ) ) return (other_maps or 0) > 0 def purge_batch(db: Session, batch_id: int) -> dict[str, int]: batch = db.get(ImportBatch, batch_id) if not batch: raise HTTPException(404, detail="导入批次不存在") counts: dict[str, int] = {} attempt_ids = list( db.scalars( select(ExaminationExamExaminee.id).where( ExaminationExamExaminee.import_batch_id == batch_id ) ) ) examinee_ids_to_check: set[int] = set() if attempt_ids: examinee_ids_to_check = set( db.scalars( select(ExaminationExamExaminee.examinee_id).where( ExaminationExamExaminee.id.in_(attempt_ids) ) ) ) counts["answer_item"] = db.execute( delete(ExaminationExamineeExamAnswerItem).where( ExaminationExamineeExamAnswerItem.exam_examinee_id.in_(attempt_ids) ) ).rowcount counts["exam_answer"] = db.execute( delete(ExaminationExamineeExamAnswer).where( ExaminationExamineeExamAnswer.exam_examinee_id.in_(attempt_ids) ) ).rowcount counts["exam_examinee"] = db.execute( delete(ExaminationExamExaminee).where(ExaminationExamExaminee.import_batch_id == batch_id) ).rowcount paper_ids_to_check = set( db.scalars( select(ExamPaperMap.paper_source_id).where(ExamPaperMap.import_batch_id == batch_id) ) ) if batch.paper_source_id: paper_ids_to_check.add(batch.paper_source_id) counts["exam_paper_map"] = db.execute( delete(ExamPaperMap).where(ExamPaperMap.import_batch_id == batch_id) ).rowcount for ps_id in paper_ids_to_check: if not _paper_owned_by_batch(db, ps_id, batch_id): continue if _paper_still_referenced(db, ps_id, batch_id): continue counts["paper_item"] = counts.get("paper_item", 0) + ( db.execute( delete(ExaminationPaperItem).where(ExaminationPaperItem.paper_source_id == ps_id) ).rowcount or 0 ) counts["paper_source"] = counts.get("paper_source", 0) + ( db.execute(delete(ExaminationPaperSource).where(ExaminationPaperSource.id == ps_id)).rowcount or 0 ) if examinee_ids_to_check: for eid in examinee_ids_to_check: ref = db.scalar( select(func.count()) .select_from(ExaminationExamExaminee) .where(ExaminationExamExaminee.examinee_id == eid) ) if ref == 0: db.execute(delete(ExaminationExaminee).where(ExaminationExaminee.id == eid)) counts["import_batch"] = db.execute(delete(ImportBatch).where(ImportBatch.id == batch_id)).rowcount db.commit() return counts def parse_batch_exam_ids(batch: ImportBatch) -> list[int]: if batch.exam_ids_json: try: raw = json.loads(batch.exam_ids_json) if isinstance(raw, list) and raw: return [int(x) for x in raw] except (json.JSONDecodeError, TypeError, ValueError): pass if batch.exam_id is not None: return [int(batch.exam_id)] return []