48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
"""Persistent RQ indexing jobs; scoring uses the same parser and file lock."""
|
|
from sqlalchemy import select
|
|
|
|
from .db import SessionLocal
|
|
from .models import IndexTask, SubmissionFile, now
|
|
from .scoring import parse_submission
|
|
|
|
|
|
def run_index_task(task_id: int, *, session_factory=SessionLocal, embedder=None):
|
|
with session_factory() as db:
|
|
task = db.get(IndexTask, task_id)
|
|
if not task or task.status == "completed":
|
|
return
|
|
task.status, task.error = "running", ""
|
|
db.commit()
|
|
try:
|
|
if task.kind == "project":
|
|
file = db.get(SubmissionFile, task.submission_file_id)
|
|
if not file or file.recognition_status != "recognized":
|
|
raise ValueError("材料已取消选择，请重新预览")
|
|
kwargs = {"embedder": embedder} if embedder else {}
|
|
parsed = parse_submission(db, file, **kwargs)
|
|
task.result = {"file_id": file.id, "pages": len(parsed.get("pages", []))}
|
|
elif task.kind == "course":
|
|
from .services import ingest_course
|
|
task.result = ingest_course(task.course_version_id, db)
|
|
if task.result["errors"]:
|
|
raise ValueError("部分课程资料入库失败：" + ", ".join(task.result["errors"]))
|
|
else:
|
|
raise ValueError("未知索引任务类型")
|
|
task.status = "completed"
|
|
except Exception as exc:
|
|
task.status, task.error = "failed", str(exc)[:1000]
|
|
task.updated_at = now()
|
|
db.commit()
|
|
|
|
|
|
def create_course_index_task(db, course_version_id: int):
|
|
from .services import enqueue_index_task
|
|
existing = db.scalar(select(IndexTask).where(IndexTask.course_version_id == course_version_id,
|
|
IndexTask.kind == "course", IndexTask.status.in_(["queued", "running"])).order_by(IndexTask.id.desc()))
|
|
task = existing or IndexTask(course_version_id=course_version_id, kind="course", status="queued")
|
|
if not existing:
|
|
db.add(task)
|
|
db.commit()
|
|
enqueue_index_task(db, task.id)
|
|
return task
|