125 lines
5.4 KiB
Python
125 lines
5.4 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from app import models
|
|
from app.services.ai_cleaner import AICleaner
|
|
from app.services.feishu_client import FeishuClient
|
|
from app.services.mock_data_service import create_mock_session
|
|
from app.services.standard_qa_service import refresh_session_counts
|
|
from app.utils.time_utils import utc_now
|
|
|
|
|
|
class TaskRunner:
|
|
def __init__(self) -> None:
|
|
self.feishu = FeishuClient()
|
|
self.ai = AICleaner()
|
|
|
|
async def run(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
task_type: models.TaskType = models.TaskType.manual_scan,
|
|
session_id: int | None = None,
|
|
reprocess: bool = False,
|
|
) -> models.TaskRun:
|
|
task = models.TaskRun(task_type=task_type, task_status=models.TaskStatus.running, started_at=utc_now())
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
try:
|
|
sessions = self._resolve_sessions(db, session_id=session_id)
|
|
if not sessions and self.feishu.is_mock_mode and session_id is None:
|
|
sessions = [create_mock_session(db)]
|
|
task.sessions_scanned = len(sessions)
|
|
|
|
for session in sessions:
|
|
try:
|
|
created_count = await self._process_session(db, session, reprocess=reprocess)
|
|
task.sessions_processed += 1
|
|
task.qa_created += created_count
|
|
except Exception as exc: # noqa: BLE001
|
|
task.qa_failed += 1
|
|
session.process_status = models.ProcessStatus.failed
|
|
session.failed_reason = str(exc)
|
|
task.error_message = str(exc)
|
|
db.add(session)
|
|
db.commit()
|
|
self.feishu.update_session_status(session, session.process_status.value)
|
|
|
|
task.task_status = (
|
|
models.TaskStatus.success
|
|
if task.qa_failed == 0
|
|
else models.TaskStatus.partial_success
|
|
if task.sessions_processed > 0
|
|
else models.TaskStatus.failed
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
task.task_status = models.TaskStatus.failed
|
|
task.error_message = str(exc)
|
|
finally:
|
|
task.ended_at = utc_now()
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
def _resolve_sessions(self, db: Session, session_id: int | None) -> list[models.FeishuSession]:
|
|
if session_id is not None:
|
|
session = db.get(models.FeishuSession, session_id)
|
|
return [session] if session else []
|
|
return self.feishu.scan_unprocessed_sessions(db)
|
|
|
|
async def _process_session(self, db: Session, session: models.FeishuSession, *, reprocess: bool = False) -> int:
|
|
if reprocess:
|
|
db.query(models.StandardQAItem).filter(models.StandardQAItem.session_id == session.id).delete()
|
|
db.query(models.RawQAItem).filter(models.RawQAItem.session_id == session.id).delete()
|
|
db.commit()
|
|
|
|
session.process_status = models.ProcessStatus.processing
|
|
session.failed_reason = None
|
|
db.add(session)
|
|
db.commit()
|
|
self.feishu.update_session_status(session, session.process_status.value)
|
|
|
|
transcript = self.feishu.fetch_transcript(session)
|
|
session.source_text = transcript
|
|
ai_items = await self.ai.clean_transcript(session.id, transcript)
|
|
|
|
created: list[models.RawQAItem] = []
|
|
for index, item in enumerate(ai_items, start=1):
|
|
raw = models.RawQAItem(
|
|
session_id=session.id,
|
|
qa_code=f"{session.session_code}-QA-{index:03d}",
|
|
raw_question=item["raw_question"],
|
|
normalized_question=item.get("normalized_question"),
|
|
raw_answer=item["raw_answer"],
|
|
normalized_answer=item.get("normalized_answer"),
|
|
suggested_standard_question=item.get("suggested_standard_question"),
|
|
suggested_standard_answer=item.get("suggested_standard_answer"),
|
|
answer_person=item.get("answer_person"),
|
|
primary_topic=item.get("primary_topic", "其他"),
|
|
problem_tags=item.get("problem_tags", []),
|
|
course_stage=item.get("course_stage", "不确定"),
|
|
audience_tags=item.get("audience_tags", []),
|
|
emotion_intensity=item.get("emotion_intensity"),
|
|
risk_level=item.get("risk_level", "medium"),
|
|
risk_types=item.get("risk_types", []),
|
|
risk_notes=item.get("risk_notes"),
|
|
need_desensitization=item.get("need_desensitization", False),
|
|
desensitization_status=item.get("desensitization_status", "unknown"),
|
|
review_status=models.ReviewStatus.pending,
|
|
review_notes=item.get("review_notes"),
|
|
suggested_to_standard_qa=item.get("suggested_to_standard_qa", True),
|
|
source_timestamp=item.get("source_timestamp"),
|
|
)
|
|
db.add(raw)
|
|
created.append(raw)
|
|
|
|
session.process_status = models.ProcessStatus.pending_review
|
|
db.add(session)
|
|
db.commit()
|
|
refresh_session_counts(db, session.id)
|
|
db.commit()
|
|
self.feishu.write_raw_qa_items(session, created)
|
|
self.feishu.update_session_status(session, session.process_status.value)
|
|
return len(created)
|