"""Shared business operations used by REST routes and the conversation agent.""" import hashlib import json import mimetypes from pathlib import Path from uuid import uuid4 from fastapi import HTTPException from redis import Redis from rq import Queue, Retry from sqlalchemy import delete, func, select from .config import get_settings from .exports import export_excel, export_project_pdf from .models import (AuditLog, CourseVersion, CriterionScore, IndexTask, KnowledgeChunk, KnowledgeDocument, Project, RubricVersion, ScoringBatch, Student, Submission, SubmissionFile, TeacherRevision) from .parsing import parse_document from .rag import INDEX_VERSION, ingest from .scoring import file_index_lock, run_batch from .settings_store import model_config from .storage import CATEGORIES, SUPPORTED, secure_path, sha256 def found(db, cls, entity_id): item = db.get(cls, entity_id) if not item: raise HTTPException(404, f"{cls.__name__}不存在") return item def audit(db, action, entity_type, entity_id, details=None): db.add(AuditLog(action=action, entity_type=entity_type, entity_id=entity_id, details=details or {})) def remove_file_index(db, file): for document in db.scalars(select(KnowledgeDocument).where(KnowledgeDocument.submission_file_id == file.id)).all(): db.execute(delete(KnowledgeChunk).where(KnowledgeChunk.document_id == document.id)) db.delete(document) file.parsed, file.parse_status = {}, "pending" def confirm_import(data, db): version = found(db, CourseVersion, data.course_version_id) if not version.published: raise HTTPException(400, "请先发布课程版本") imported = [] files_to_index = [] for item in data.projects: folder = secure_path(item.relative_dir) if not folder.is_dir(): raise ValueError("学生目录不存在") student = db.scalar(select(Student).where(Student.student_no == item.student_no)) if not student: student = Student(student_no=item.student_no, name=item.name) db.add(student) db.flush() else: student.name = item.name project = db.scalar(select(Project).where(Project.course_version_id == version.id, Project.source_dir == item.relative_dir)) if not project: project = Project(course_version_id=version.id, student_id=student.id, source_dir=item.relative_dir, topic=item.topic, direction=item.direction) db.add(project) db.flush() else: project.student_id, project.topic, project.direction = student.id, item.topic, item.direction selected = set() selected_paths = set() for file in item.files: path = secure_path(file.relative_path) if path.parent != folder or not path.is_file() or path.suffix.lower() not in SUPPORTED: raise ValueError("材料必须是学生目录内的PDF、DOCX或PPTX文件") if sha256(path) != file.sha256: raise ValueError("文件已变化，请重新扫描") if file.status == "ignored": continue if file.category not in CATEGORIES or file.category in selected or file.relative_path in selected_paths: raise ValueError("请为每类材料只确认一个文件") selected.add(file.category) selected_paths.add(file.relative_path) stat = path.stat() record = db.scalar(select(SubmissionFile).where(SubmissionFile.project_id == project.id, SubmissionFile.relative_path == file.relative_path)) if not record: record = SubmissionFile(project_id=project.id, relative_path=file.relative_path, category=file.category, mime_type=mimetypes.guess_type(path.name)[0] or "application/octet-stream", size=stat.st_size, mtime=stat.st_mtime, sha256=file.sha256, recognition_status="recognized") db.add(record) db.flush() else: if record.sha256 != file.sha256 or record.category != file.category: remove_file_index(db, record) record.category, record.sha256, record.size, record.mtime, record.recognition_status = file.category, file.sha256, stat.st_size, stat.st_mtime, "recognized" doc = db.scalar(select(KnowledgeDocument).where(KnowledgeDocument.submission_file_id == record.id, KnowledgeDocument.source_hash == record.sha256, KnowledgeDocument.index_version == INDEX_VERSION, KnowledgeDocument.status == "ready")) if record.parse_status != "ready" or not doc: files_to_index.append(record.id) for old in db.scalars(select(SubmissionFile).where(SubmissionFile.project_id == project.id)).all(): if old.relative_path not in selected_paths: old.recognition_status = "ignored" if old.parse_status == "ready": remove_file_index(db, old) for category in CATEGORIES: submission = db.scalar(select(Submission).where(Submission.project_id == project.id, Submission.category == category)) if not submission: submission = Submission(project_id=project.id, category=category, status="missing") db.add(submission) db.flush() submission.status = "submitted" if category in selected else "missing" for file in db.scalars(select(SubmissionFile).where(SubmissionFile.project_id == project.id, SubmissionFile.category == category, SubmissionFile.recognition_status == "recognized")).all(): file.submission_id = submission.id audit(db, "import", "project", project.id, {"files": len(selected)}) imported.append(project.id) tasks = [] for file_id in sorted(set(files_to_index)): current = db.scalar(select(IndexTask).where(IndexTask.submission_file_id == file_id, IndexTask.status.in_(["queued", "running"])).order_by(IndexTask.id.desc())) if not current: current = IndexTask(course_version_id=version.id, submission_file_id=file_id, kind="project") db.add(current) db.flush() tasks.append(current.id) db.commit() for task_id in tasks: enqueue_index_task(db, task_id) return {"project_ids": imported, "index_task_ids": tasks} def enqueue_index_task(db, task_id: int): from .indexing import run_index_task task = found(db, IndexTask, task_id) if task.status not in ("queued", "failed"): raise HTTPException(409, "索引任务不可重试") try: Queue("wisegrade", connection=Redis.from_url(get_settings().redis_url)).enqueue( run_index_task, task.id, job_id=f"wisegrade-index-{task.id}", job_timeout="3h") task.status, task.error = "queued", "" except Exception as exc: task.status, task.error = "failed", f"无法提交索引任务: {type(exc).__name__}" db.commit() return task def retry_index_task(db, source_id: int): source = db.scalar(select(IndexTask).where(IndexTask.id == source_id).with_for_update()) if not source: raise HTTPException(404, "索引任务不存在") if source.status != "failed": db.rollback() raise HTTPException(409, "仅失败的索引任务可重试") existing = db.scalar(select(IndexTask).where(IndexTask.id > source.id, IndexTask.course_version_id == source.course_version_id, IndexTask.submission_file_id == source.submission_file_id, IndexTask.kind == source.kind, IndexTask.status.in_(["queued", "running", "completed"])) .order_by(IndexTask.id.desc())) if existing: db.commit() return existing retry = IndexTask(course_version_id=source.course_version_id, submission_file_id=source.submission_file_id, kind=source.kind, status="queued") db.add(retry) db.commit() return enqueue_index_task(db, retry.id) def ingest_course(course_version_id: int, db): with file_index_lock(-course_version_id): return _ingest_course_locked(course_version_id, db) def _ingest_course_locked(course_version_id: int, db): version = found(db, CourseVersion, course_version_id) root = Path("/app/course-docs") processed = get_settings().processed_path / f"course-{course_version_id}" completed, errors = [], {} for filename in version.templates: if Path(filename).name != filename: errors[filename] = "非法课程资料路径" continue path = root / filename if not path.is_file(): errors[filename] = "课程资料文件不存在" continue try: category = "stage1" if "阶段报告1" in filename else "stage2" if "阶段报告2" in filename else "stage3" if "阶段报告3" in filename else "course" parsed = parse_document(path, category, processed / hashlib.sha256(filename.encode()).hexdigest()[:16]) ingest(db, course_version_id=version.id, project_id=None, submission_file_id=None, title=filename, category=category, source_hash=sha256(path), pages=parsed["pages"]) completed.append(filename) except Exception as exc: errors[filename] = str(exc)[:500] rubric = db.scalar(select(RubricVersion).where(RubricVersion.course_version_id == version.id, RubricVersion.published.is_(True)).order_by(RubricVersion.version.desc())) virtual = [ ("课程目标、毕业要求与选题方向", {"objectives": version.objectives, "indicators": version.indicators, "directions": version.directions}), ("已发布评分规则", {"criteria": rubric.criteria, "weights": rubric.weights} if rubric else {}), ] for title, payload in virtual: if not payload: continue try: content = json.dumps(payload, ensure_ascii=False, indent=2) ingest(db, course_version_id=version.id, project_id=None, submission_file_id=None, title=title, category="course", source_hash=hashlib.sha256(content.encode()).hexdigest(), pages=[{"page_number": 1, "section": title, "text": content, "tables": []}]) completed.append(title) except Exception as exc: errors[title] = str(exc)[:500] return {"completed": completed, "errors": errors} def create_batch(data, db): version = found(db, CourseVersion, data.course_version_id) rubric = found(db, RubricVersion, data.rubric_version_id) if not version.published or not rubric.published or rubric.course_version_id != version.id: raise HTTPException(400, "请选择同一课程版本下已发布的规则") settings = get_settings() batch = ScoringBatch(course_version_id=version.id, rubric_version_id=rubric.id, course_snapshot={key: getattr(version, key) for key in ("version", "title", "objectives", "indicators", "directions", "prompts", "templates")}, rubric_snapshot={"version": rubric.version, "criteria": rubric.criteria, "weights": rubric.weights}, model_config_snapshot=model_config(db), embedding_model_name=settings.embedding_model_name, embedding_model_version=settings.embedding_model_version, status="created", project_ids=db.scalars(select(Project.id).where( Project.course_version_id == version.id).order_by(Project.id)).all()) db.add(batch) db.flush() audit(db, "create", "scoring_batch", batch.id) db.commit() return {"id": batch.id, "status": batch.status} def start_batch(batch_id: int, db): batch = db.scalar(select(ScoringBatch).where(ScoringBatch.id == batch_id).with_for_update()) if not batch: raise HTTPException(404, "评分批次不存在") if batch.status not in ("created", "failed", "partial"): db.rollback() raise HTTPException(409, "批次已入队、运行中或已完成") try: Queue("wisegrade", connection=Redis.from_url(get_settings().redis_url)).enqueue(run_batch, batch_id, job_id=f"wisegrade-batch-{batch_id}-{uuid4().hex}", job_timeout="12h", retry=Retry(max=1, interval=[60])) except Exception as exc: db.rollback() batch = found(db, ScoringBatch, batch_id) batch.status, batch.error = "failed", f"无法提交后台任务: {type(exc).__name__}" db.commit() raise HTTPException(503, f"无法提交后台任务: {type(exc).__name__}") from exc batch.status, batch.error, batch.finished_at = "queued", "", None db.commit() return {"id": batch.id, "status": batch.status} def review_score(score_id: int, data, db): score = found(db, CriterionScore, score_id) batch = found(db, ScoringBatch, score.batch_id) criterion = next((item for item in batch.rubric_snapshot["criteria"] if item["criterion_id"] == score.criterion_id), None) if not criterion or data.score > criterion["max_score"]: raise HTTPException(400, "教师分数超出规则满分") db.add(TeacherRevision(score_id=score.id, old_score=score.teacher_score, new_score=data.score, old_comment=score.teacher_comment, new_comment=data.comment, reason=data.reason)) score.teacher_score, score.teacher_comment, score.status = data.score, data.comment, "reviewed" audit(db, "review", "criterion_score", score.id, {"reason": data.reason}) db.commit() return {"id": score.id, "model_score": score.model_score, "teacher_score": score.teacher_score, "final_score": score.final_score, "teacher_comment": score.teacher_comment}