Initial MVP for QA asset backend

hy_qa_asset_backend/backend/app/services/ai_cleaner.py

@@ -0,0 +1,49 @@
+import json
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from app.config import get_settings
+from app.services.mock_data_service import mock_ai_cleaning_result
+from app.services.qa_parser import normalize_ai_item
+from app.utils.json_validator import validate_ai_qa_items
+
+
+PROMPT_PATH = Path(__file__).resolve().parents[1] / "prompts" / "qa_cleaning_prompt.md"
+
+
+class AICleaner:
+    def __init__(self) -> None:
+        self.settings = get_settings()
+
+    def _prompt(self) -> str:
+        return PROMPT_PATH.read_text(encoding="utf-8")
+
+    async def clean_transcript(self, session_id: int, transcript: str) -> list[dict[str, Any]]:
+        if not self.settings.openai_configured:
+            return [normalize_ai_item(item) for item in mock_ai_cleaning_result(session_id)]
+
+        payload = {
+            "model": self.settings.model_name,
+            "messages": [
+                {"role": "system", "content": self._prompt()},
+                {
+                    "role": "user",
+                    "content": f"场次 ID: {session_id}\n\n以下是转写稿：\n{transcript}",
+                },
+            ],
+            "temperature": 0.1,
+        }
+        base_url = (self.settings.openai_base_url or "https://api.openai.com/v1").rstrip("/")
+        headers = {"Authorization": f"Bearer {self.settings.openai_api_key}"}
+        async with httpx.AsyncClient(timeout=60) as client:
+            response = await client.post(f"{base_url}/chat/completions", headers=headers, json=payload)
+            response.raise_for_status()
+        content = response.json()["choices"][0]["message"]["content"]
+        try:
+            parsed = json.loads(content)
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"AI 返回非法 JSON: {exc}") from exc
+        return [normalize_ai_item(item) for item in validate_ai_qa_items(parsed)]
+

hy_qa_asset_backend/backend/app/services/feishu_client.py

@@ -0,0 +1,600 @@
+from __future__ import annotations
+
+import io
+import re
+import zipfile
+from datetime import date, datetime
+from typing import Any
+from xml.etree import ElementTree as ET
+
+import httpx
+from sqlalchemy.orm import Session
+
+from app import models
+from app.config import get_settings
+from app.services.mock_data_service import read_sample_transcript
+
+
+FEISHU_OPEN_API = "https://open.feishu.cn/open-apis"
+
+STATUS_TO_FEISHU = {
+    models.ProcessStatus.unprocessed.value: "unprocessed",
+    models.ProcessStatus.processing.value: "processing",
+    models.ProcessStatus.parsed.value: "parsed",
+    models.ProcessStatus.pending_review.value: "pending_review",
+    models.ProcessStatus.completed.value: "completed",
+    models.ProcessStatus.failed.value: "failed",
+}
+
+FEISHU_TO_STATUS = {
+    "unprocessed": models.ProcessStatus.unprocessed,
+    "未处理": models.ProcessStatus.unprocessed,
+    "processing": models.ProcessStatus.processing,
+    "处理中": models.ProcessStatus.processing,
+    "parsed": models.ProcessStatus.parsed,
+    "已拆解": models.ProcessStatus.parsed,
+    "pending_review": models.ProcessStatus.pending_review,
+    "待审核": models.ProcessStatus.pending_review,
+    "completed": models.ProcessStatus.completed,
+    "已完成": models.ProcessStatus.completed,
+    "failed": models.ProcessStatus.failed,
+    "失败": models.ProcessStatus.failed,
+}
+
+SESSION_FIELD_ALIASES = {
+    "session_code": ["session_code", "场次编号", "编号"],
+    "title": ["title", "标题", "答疑标题", "场次标题"],
+    "date": ["date", "日期", "答疑日期"],
+    "teachers": ["teachers", "答疑老师", "老师"],
+    "feishu_doc_url": ["feishu_doc_url", "飞书资料链接", "文档链接", "资料链接"],
+    "source_file_name": ["source_file_name", "文件名", "来源文件名"],
+    "source_text": ["source_text", "转写稿", "文字稿", "原始资料", "答疑文字稿"],
+    "process_status": ["process_status", "处理状态", "状态"],
+    "failed_reason": ["failed_reason", "失败原因"],
+}
+
+SESSION_WRITE_FIELDS = {
+    "process_status": ["处理状态", "process_status"],
+    "qa_count": ["问答总数", "qa_count"],
+    "pending_review_count": ["待审核数", "pending_review_count"],
+    "approved_count": ["已审核数", "approved_count"],
+    "standard_qa_count": ["已入库数", "standard_qa_count"],
+    "failed_reason": ["失败原因", "failed_reason"],
+}
+
+
+def _first_field(fields: dict[str, Any], names: list[str], default: Any = None) -> Any:
+    for name in names:
+        if name in fields and fields[name] not in (None, ""):
+            return fields[name]
+    return default
+
+
+def _stringify(value: Any) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, str):
+        return value
+    if isinstance(value, list):
+        parts: list[str] = []
+        for item in value:
+            if isinstance(item, dict):
+                parts.append(str(item.get("text") or item.get("name") or item.get("url") or item))
+            else:
+                parts.append(str(item))
+        return " ".join(parts)
+    if isinstance(value, dict):
+        return str(value.get("text") or value.get("name") or value.get("url") or value)
+    return str(value)
+
+
+def _parse_date(value: Any) -> date | None:
+    if value in (None, ""):
+        return None
+    if isinstance(value, (int, float)):
+        # Bitable date fields are often millisecond timestamps.
+        return datetime.fromtimestamp(value / 1000).date()
+    text = _stringify(value)
+    if not text:
+        return None
+    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d"):
+        try:
+            return datetime.strptime(text[:10], fmt).date()
+        except ValueError:
+            continue
+    return None
+
+
+def _extract_url(value: Any) -> str | None:
+    if not value:
+        return None
+    if isinstance(value, list):
+        for item in value:
+            url = _extract_url(item)
+            if url:
+                return url
+        return None
+    if isinstance(value, dict):
+        for key in ("link", "url", "text"):
+            url = _extract_url(value.get(key))
+            if url:
+                return url
+        return None
+    text = str(value)
+    match = re.search(r"https?://\S+", text)
+    return match.group(0) if match else text if text.startswith("http") else None
+
+
+def _extract_doc_id(url: str | None) -> str | None:
+    if not url:
+        return None
+    patterns = [
+        r"/docx/([A-Za-z0-9]+)",
+        r"/docs/([A-Za-z0-9]+)",
+        r"/wiki/([A-Za-z0-9]+)",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, url)
+        if match:
+            return match.group(1)
+    tail = url.rstrip("/").split("/")[-1]
+    return tail if re.match(r"^[A-Za-z0-9]{8,}$", tail) else None
+
+
+def _extract_drive_file_token(url: str | None) -> str | None:
+    if not url:
+        return None
+    match = re.search(r"/file/([A-Za-z0-9]+)", url)
+    return match.group(1) if match else None
+
+
+class FeishuAPIError(RuntimeError):
+    pass
+
+
+class FeishuClient:
+    """Feishu adapter used by the black-light workflow.
+
+    Product boundary: Feishu is the material entrance. AI only cleans and
+    pre-screens; every reusable QA still must pass human review in this system.
+    """
+
+    def __init__(self) -> None:
+        self.settings = get_settings()
+        self._tenant_access_token: str | None = None
+        self._bitable_app_token: str | None = None
+
+    @property
+    def is_mock_mode(self) -> bool:
+        return self.settings.mock_mode or not self.settings.feishu_configured
+
+    def connection_status(self) -> dict[str, Any]:
+        configured = self.settings.feishu_configured
+        status = {
+            "configured": configured,
+            "mock_mode": self.is_mock_mode,
+            "app_token_configured": bool(self.settings.feishu_app_token),
+            "wiki_node_token_configured": bool(self.settings.feishu_wiki_node_token),
+            "table_ids": {
+                "session": bool(self.settings.feishu_table_id_session),
+                "raw_qa": bool(self.settings.feishu_table_id_raw_qa),
+                "standard_qa": bool(self.settings.feishu_table_id_standard_qa),
+            },
+            "token_ok": False,
+            "message": "未配置飞书密钥，当前使用 mock mode。",
+        }
+        if not configured:
+            return status
+        try:
+            self._get_tenant_access_token()
+            self._get_bitable_app_token()
+            status["token_ok"] = True
+            status["message"] = "飞书 tenant_access_token 获取成功。"
+        except Exception as exc:  # noqa: BLE001
+            status["message"] = f"飞书连接失败：{exc}"
+        return status
+
+    def scan_unprocessed_sessions(self, db: Session) -> list[models.FeishuSession]:
+        if self.is_mock_mode:
+            return (
+                db.query(models.FeishuSession)
+                .filter(models.FeishuSession.process_status == models.ProcessStatus.unprocessed)
+                .order_by(models.FeishuSession.created_at.asc())
+                .all()
+            )
+
+        records = self._list_records(self.settings.feishu_table_id_session)
+        sessions: list[models.FeishuSession] = []
+        for record in records:
+            fields = record.get("fields", {})
+            record_id = record.get("record_id")
+            status_text = _stringify(_first_field(fields, SESSION_FIELD_ALIASES["process_status"], "unprocessed"))
+            process_status = FEISHU_TO_STATUS.get(status_text or "", models.ProcessStatus.unprocessed)
+            if process_status != models.ProcessStatus.unprocessed:
+                continue
+
+            session_code = _stringify(_first_field(fields, SESSION_FIELD_ALIASES["session_code"])) or f"FS-{record_id}"
+            session = (
+                db.query(models.FeishuSession)
+                .filter(models.FeishuSession.feishu_record_id == record_id)
+                .one_or_none()
+            )
+            if not session:
+                session = models.FeishuSession(session_code=session_code, title=session_code)
+                db.add(session)
+            session.session_code = session_code
+            session.title = _stringify(_first_field(fields, SESSION_FIELD_ALIASES["title"], session_code)) or session_code
+            session.date = _parse_date(_first_field(fields, SESSION_FIELD_ALIASES["date"]))
+            session.teachers = _stringify(_first_field(fields, SESSION_FIELD_ALIASES["teachers"]))
+            session.feishu_doc_url = _extract_url(_first_field(fields, SESSION_FIELD_ALIASES["feishu_doc_url"]))
+            session.feishu_record_id = record_id
+            session.source_file_name = _stringify(_first_field(fields, SESSION_FIELD_ALIASES["source_file_name"]))
+            session.source_text = _stringify(_first_field(fields, SESSION_FIELD_ALIASES["source_text"]))
+            session.process_status = process_status
+            session.failed_reason = _stringify(_first_field(fields, SESSION_FIELD_ALIASES["failed_reason"]))
+            sessions.append(session)
+        db.commit()
+        for session in sessions:
+            db.refresh(session)
+        return sessions
+
+    def fetch_transcript(self, session: models.FeishuSession) -> str:
+        if self.is_mock_mode:
+            return session.source_text or read_sample_transcript()
+        if session.source_text:
+            return session.source_text
+        file_token = _extract_drive_file_token(session.feishu_doc_url)
+        if file_token:
+            return self._get_drive_docx_text(file_token)
+        doc_id = _extract_doc_id(session.feishu_doc_url)
+        if doc_id:
+            return self._get_docx_raw_content(doc_id)
+        raise FeishuAPIError("场次没有转写稿文本，也没有可识别的飞书 docx 文档链接")
+
+    def update_session_status(self, session: models.FeishuSession, status: str) -> None:
+        if self.is_mock_mode or not session.feishu_record_id:
+            return
+        fields = self._build_session_update_fields(session, status)
+        self._update_record(self.settings.feishu_table_id_session, session.feishu_record_id, fields)
+
+    def write_raw_qa_items(self, session: models.FeishuSession, items: list[models.RawQAItem]) -> None:
+        if self.is_mock_mode or not self.settings.feishu_table_id_raw_qa or not items:
+            return
+        self._write_raw_qa_items_compatible(session, items)
+        return
+        records = []
+        for item in items:
+            records.append(
+                {
+                    "fields": {
+                        "问答编号": item.qa_code,
+                        "场次编号": session.session_code,
+                        "原始问题": item.raw_question,
+                        "问题整理版": item.normalized_question,
+                        "原始回答": item.raw_answer,
+                        "回答整理版": item.normalized_answer,
+                        "AI建议标准问题": item.suggested_standard_question,
+                        "AI建议标准回答": item.suggested_standard_answer,
+                        "回答人": item.answer_person,
+                        "一级主题": item.primary_topic.value,
+                        "问题标签": "、".join(item.problem_tags or []),
+                        "课程阶段": item.course_stage.value,
+                        "适用人群": "、".join(item.audience_tags or []),
+                        "情绪强度": item.emotion_intensity,
+                        "风险等级": item.risk_level.value,
+                        "风险类型": "、".join(item.risk_types or []),
+                        "风险说明": item.risk_notes,
+                        "是否需脱敏": "是" if item.need_desensitization else "否",
+                        "脱敏状态": item.desensitization_status.value,
+                        "审核状态": item.review_status.value,
+                        "建议入库": "是" if item.suggested_to_standard_qa else "否",
+                        "来源时间戳": item.source_timestamp,
+                        "审核备注": item.review_notes,
+                    }
+                }
+            )
+        self._batch_create_records(self.settings.feishu_table_id_raw_qa, records)
+
+    def write_standard_qa_items(self, items: list[models.StandardQAItem]) -> None:
+        if self.is_mock_mode or not self.settings.feishu_table_id_standard_qa or not items:
+            return
+        self._write_standard_qa_items_compatible(items)
+        return
+        records = []
+        for item in items:
+            records.append(
+                {
+                    "fields": {
+                        "标准编号": item.standard_code,
+                        "来源原始问答ID": str(item.source_raw_qa_id),
+                        "来源场次ID": str(item.session_id),
+                        "标准问题": item.standard_question,
+                        "标准回答": item.standard_answer,
+                        "相似问法": "、".join(item.similar_questions or []),
+                        "一级主题": item.primary_topic.value,
+                        "问题标签": "、".join(item.problem_tags or []),
+                        "课程阶段": item.course_stage.value,
+                        "适用人群": "、".join(item.audience_tags or []),
+                        "回答边界": item.answer_boundary,
+                        "禁止表达": "、".join(item.forbidden_expressions or []),
+                        "风险等级": item.risk_level.value,
+                        "审核状态": item.audit_status.value,
+                        "调用状态": item.call_status.value,
+                        "禁用原因": item.disabled_reason,
+                    }
+                }
+            )
+        self._batch_create_records(self.settings.feishu_table_id_standard_qa, records)
+
+    def _get_tenant_access_token(self) -> str:
+        if self._tenant_access_token:
+            return self._tenant_access_token
+        payload = {"app_id": self.settings.feishu_app_id, "app_secret": self.settings.feishu_app_secret}
+        with httpx.Client(timeout=20) as client:
+            response = client.post(f"{FEISHU_OPEN_API}/auth/v3/tenant_access_token/internal", json=payload)
+        data = response.json()
+        if response.status_code >= 400 or data.get("code") != 0:
+            raise FeishuAPIError(data.get("msg") or response.text)
+        self._tenant_access_token = data["tenant_access_token"]
+        return self._tenant_access_token
+
+    def _headers(self) -> dict[str, str]:
+        return {"Authorization": f"Bearer {self._get_tenant_access_token()}", "Content-Type": "application/json"}
+
+    def _get_bitable_app_token(self) -> str:
+        if self._bitable_app_token:
+            return self._bitable_app_token
+        if self.settings.feishu_app_token:
+            self._bitable_app_token = self.settings.feishu_app_token
+            return self._bitable_app_token
+        if not self.settings.feishu_wiki_node_token:
+            raise FeishuAPIError("缺少 FEISHU_APP_TOKEN 或 FEISHU_WIKI_NODE_TOKEN")
+        data = self._request(
+            "GET",
+            "/wiki/v2/spaces/get_node",
+            params={"token": self.settings.feishu_wiki_node_token},
+        )
+        node = data.get("node", {})
+        if node.get("obj_type") != "bitable":
+            raise FeishuAPIError(f"FEISHU_WIKI_NODE_TOKEN 指向的对象不是 bitable：{node.get('obj_type')}")
+        obj_token = node.get("obj_token")
+        if not obj_token:
+            raise FeishuAPIError("wiki 节点信息没有返回 obj_token")
+        self._bitable_app_token = obj_token
+        return self._bitable_app_token
+
+    def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
+        with httpx.Client(timeout=30) as client:
+            response = client.request(method, f"{FEISHU_OPEN_API}{path}", headers=self._headers(), **kwargs)
+        data = response.json()
+        if response.status_code >= 400 or data.get("code") != 0:
+            raise FeishuAPIError(data.get("msg") or response.text)
+        return data.get("data", {})
+
+    def _table_field_names(self, table_id: str | None) -> set[str]:
+        if not table_id:
+            return set()
+        cache = getattr(self, "_table_field_name_cache", None)
+        if cache is None:
+            cache = {}
+            self._table_field_name_cache = cache
+        if table_id in cache:
+            return cache[table_id]
+        names: set[str] = set()
+        page_token: str | None = None
+        while True:
+            params = {"page_size": 100}
+            if page_token:
+                params["page_token"] = page_token
+            data = self._request(
+                "GET",
+                f"/bitable/v1/apps/{self._get_bitable_app_token()}/tables/{table_id}/fields",
+                params=params,
+            )
+            names.update(item.get("field_name") for item in data.get("items", []) if item.get("field_name"))
+            if not data.get("has_more"):
+                break
+            page_token = data.get("page_token")
+        cache[table_id] = names
+        return names
+
+    def _filter_existing_fields(self, table_id: str | None, fields: dict[str, Any]) -> dict[str, Any]:
+        names = self._table_field_names(table_id)
+        return {key: value for key, value in fields.items() if key in names and value is not None}
+
+    @staticmethod
+    def _join_values(values: list[str] | None) -> str | None:
+        return "、".join(values or []) or None
+
+    @staticmethod
+    def _enum_value(value: Any) -> str:
+        return getattr(value, "value", value)
+
+    def _risk_label(self, value: Any) -> str:
+        return {"low": "低风险", "medium": "中风险", "high": "高风险"}.get(self._enum_value(value), str(value))
+
+    def _review_label(self, value: Any) -> str:
+        return {
+            "pending": "待审核",
+            "approved": "已通过",
+            "rejected": "已拒绝",
+            "needs_revision": "需修改",
+        }.get(self._enum_value(value), str(value))
+
+    def _build_session_update_fields_compatible(self, session: models.FeishuSession, status: str) -> dict[str, Any]:
+        status_label = {
+            models.ProcessStatus.unprocessed.value: "待处理",
+            models.ProcessStatus.processing.value: "处理中",
+            models.ProcessStatus.parsed.value: "处理中",
+            models.ProcessStatus.pending_review.value: "待审核",
+            models.ProcessStatus.completed.value: "已完成",
+            models.ProcessStatus.failed.value: "失败",
+        }.get(status, status)
+        high_risk_count = sum(1 for item in session.raw_items if item.risk_level == models.RiskLevel.high)
+        fields = {
+            "处理状态": status_label,
+            "问答总数": session.qa_count,
+            "原始问答数": session.qa_count,
+            "待审核数": session.pending_review_count,
+            "已审核数": session.approved_count,
+            "已入库数": session.standard_qa_count,
+            "标准问答数": session.standard_qa_count,
+            "高风险数": high_risk_count,
+            "失败原因": session.failed_reason,
+            "最近同步时间": int(datetime.now().timestamp() * 1000),
+        }
+        return self._filter_existing_fields(self.settings.feishu_table_id_session, fields)
+
+    def _write_raw_qa_items_compatible(self, session: models.FeishuSession, items: list[models.RawQAItem]) -> None:
+        table_id = self.settings.feishu_table_id_raw_qa
+        records = []
+        for item in items:
+            tags = (item.problem_tags or []) + [self._enum_value(item.primary_topic)]
+            fields = {
+                "问答编号": item.qa_code,
+                "场次编号": session.session_code,
+                "原始问题": item.raw_question,
+                "问题": item.normalized_question or item.raw_question,
+                "问题整理版": item.normalized_question,
+                "原始回答": item.raw_answer,
+                "回答": item.normalized_answer or item.raw_answer,
+                "回答整理版": item.normalized_answer,
+                "AI建议标准问题": item.suggested_standard_question,
+                "AI建议标准答案": item.suggested_standard_answer,
+                "答疑老师": item.answer_person or session.teachers,
+                "标签": self._join_values(tags),
+                "风险等级": self._risk_label(item.risk_level),
+                "证据片段": item.source_timestamp or (item.raw_answer or "")[:240],
+                "审核状态": self._review_label(item.review_status),
+                "入库状态": "建议入库" if item.suggested_to_standard_qa else "未入库",
+            }
+            filtered = self._filter_existing_fields(table_id, fields)
+            if filtered:
+                records.append({"fields": filtered})
+        self._batch_create_records(table_id, records)
+
+    def _write_standard_qa_items_compatible(self, items: list[models.StandardQAItem]) -> None:
+        table_id = self.settings.feishu_table_id_standard_qa
+        records = []
+        for item in items:
+            fields = {
+                "标准编号": item.standard_code,
+                "来源原始问答ID": str(item.source_raw_qa_id),
+                "来源场次ID": str(item.session_id),
+                "标准问题": item.standard_question,
+                "标准答案": item.standard_answer,
+                "相似问法": self._join_values(item.similar_questions),
+                "一级主题": self._enum_value(item.primary_topic),
+                "问题标签": self._join_values(item.problem_tags),
+                "适用人群": self._join_values(item.audience_tags),
+                "标签": self._join_values(item.problem_tags),
+                "回答边界": item.answer_boundary,
+                "禁止表达": self._join_values(item.forbidden_expressions),
+                "风险等级": self._risk_label(item.risk_level),
+                "审核状态": self._enum_value(item.audit_status),
+                "调用状态": self._enum_value(item.call_status),
+                "状态": self._enum_value(item.audit_status),
+                "是否可调用": "是" if item.call_status == models.CallStatus.callable else "否",
+                "来源场次": str(item.session_id),
+                "来源问题": str(item.source_raw_qa_id),
+                "禁用原因": item.disabled_reason,
+            }
+            filtered = self._filter_existing_fields(table_id, fields)
+            if filtered:
+                records.append({"fields": filtered})
+        self._batch_create_records(table_id, records)
+
+    def _list_records(self, table_id: str | None) -> list[dict[str, Any]]:
+        if not table_id:
+            raise FeishuAPIError("缺少飞书场次表 table_id")
+        records: list[dict[str, Any]] = []
+        page_token: str | None = None
+        while True:
+            params = {"page_size": 500}
+            if page_token:
+                params["page_token"] = page_token
+            data = self._request(
+                "GET",
+                f"/bitable/v1/apps/{self._get_bitable_app_token()}/tables/{table_id}/records",
+                params=params,
+            )
+            records.extend(data.get("items", []))
+            if not data.get("has_more"):
+                break
+            page_token = data.get("page_token")
+        return records
+
+    def _update_record(self, table_id: str | None, record_id: str, fields: dict[str, Any]) -> None:
+        if not table_id:
+            return
+        self._request(
+            "PUT",
+            f"/bitable/v1/apps/{self._get_bitable_app_token()}/tables/{table_id}/records/{record_id}",
+            json={"fields": fields},
+        )
+
+    def _batch_create_records(self, table_id: str | None, records: list[dict[str, Any]]) -> None:
+        if not table_id:
+            return
+        for start in range(0, len(records), 500):
+            chunk = records[start : start + 500]
+            self._request(
+                "POST",
+                f"/bitable/v1/apps/{self._get_bitable_app_token()}/tables/{table_id}/records/batch_create",
+                json={"records": chunk},
+            )
+
+    def _get_docx_raw_content(self, document_id: str) -> str:
+        data = self._request("GET", f"/docx/v1/documents/{document_id}/raw_content")
+        content = data.get("content") or data.get("raw_content") or data.get("text")
+        if not content:
+            raise FeishuAPIError("飞书文档纯文本接口没有返回 content")
+        return content
+
+    def _get_drive_docx_text(self, file_token: str) -> str:
+        with httpx.Client(timeout=60, follow_redirects=True) as client:
+            response = client.get(
+                f"{FEISHU_OPEN_API}/drive/v1/files/{file_token}/download",
+                headers=self._headers(),
+            )
+        if response.status_code >= 400:
+            try:
+                data = response.json()
+                message = data.get("msg") or response.text
+            except Exception:  # noqa: BLE001
+                message = response.text
+            raise FeishuAPIError(message)
+        return self._extract_docx_text(response.content)
+
+    @staticmethod
+    def _extract_docx_text(content: bytes) -> str:
+        try:
+            with zipfile.ZipFile(io.BytesIO(content)) as docx:
+                document_xml = docx.read("word/document.xml")
+        except Exception as exc:  # noqa: BLE001
+            raise FeishuAPIError(f"飞书上传文件不是可解析的 docx：{exc}") from exc
+
+        root = ET.fromstring(document_xml)
+        namespace = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
+        paragraphs: list[str] = []
+        for paragraph in root.findall(".//w:p", namespace):
+            text = "".join(node.text or "" for node in paragraph.findall(".//w:t", namespace)).strip()
+            if text:
+                paragraphs.append(text)
+        extracted = "\n".join(paragraphs).strip()
+        if not extracted:
+            raise FeishuAPIError("飞书上传 docx 没有提取到正文")
+        return extracted
+
+    def _build_session_update_fields(self, session: models.FeishuSession, status: str) -> dict[str, Any]:
+        return self._build_session_update_fields_compatible(session, status)
+        fields: dict[str, Any] = {
+            SESSION_WRITE_FIELDS["process_status"][0]: STATUS_TO_FEISHU.get(status, status),
+            SESSION_WRITE_FIELDS["qa_count"][0]: session.qa_count,
+            SESSION_WRITE_FIELDS["pending_review_count"][0]: session.pending_review_count,
+            SESSION_WRITE_FIELDS["approved_count"][0]: session.approved_count,
+            SESSION_WRITE_FIELDS["standard_qa_count"][0]: session.standard_qa_count,
+        }
+        if session.failed_reason:
+            fields[SESSION_WRITE_FIELDS["failed_reason"][0]] = session.failed_reason
+        return fields

hy_qa_asset_backend/backend/app/services/mock_data_service.py

@@ -0,0 +1,115 @@
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from app import models
+
+
+SEED_DIR = Path(__file__).resolve().parents[1] / "seed"
+SAMPLE_TRANSCRIPT_PATH = SEED_DIR / "sample_transcript_001.txt"
+
+
+def read_sample_transcript() -> str:
+    return SAMPLE_TRANSCRIPT_PATH.read_text(encoding="utf-8")
+
+
+def create_mock_session(db: Session) -> models.FeishuSession:
+    index = db.query(models.FeishuSession).count() + 1
+    session = models.FeishuSession(
+        session_code=f"MOCK-{index:04d}",
+        title="大本营周三答疑 mock 场次",
+        date=date.today(),
+        teachers="院长, 辅导老师",
+        feishu_doc_url="mock://feishu/sample_transcript_001",
+        feishu_record_id=f"mock-record-{index:04d}",
+        source_file_name="sample_transcript_001.txt",
+        source_text=read_sample_transcript(),
+        process_status=models.ProcessStatus.unprocessed,
+    )
+    db.add(session)
+    db.commit()
+    db.refresh(session)
+    return session
+
+
+def mock_ai_cleaning_result(session_id: int) -> list[dict[str, Any]]:
+    return [
+        {
+            "session_id": str(session_id),
+            "qa_id": "mock-qa-001",
+            "raw_question": "我和孩子沟通时总是忍不住发火，孩子现在一听我说话就躲开，我该怎么办？",
+            "normalized_question": "和孩子沟通时容易发火，孩子开始回避，应该如何调整？",
+            "raw_answer": "先不要急着讲道理，先看见自己发火背后的着急和无力。可以把要求先放低一点，用一句具体的话表达当下感受，比如我现在有点急，我想先停一下。",
+            "normalized_answer": "先觉察自己发火背后的着急和无力，暂停说教，把要求放低，用具体语言表达当下感受。",
+            "suggested_standard_question": "亲子沟通中总忍不住发火，如何先做自我调整？",
+            "suggested_standard_answer": "可以先暂停说教，觉察自己发火背后的着急和无力，再用具体、低压力的语言表达当下感受。这个回答不能替代个体咨询，也不承诺亲子关系会立即改善。",
+            "answer_person": "辅导老师",
+            "primary_topic": "亲子关系",
+            "problem_tags": ["亲子沟通", "情绪觉察", "发火"],
+            "course_stage": "觉知",
+            "audience_tags": ["家长", "亲子关系困扰者"],
+            "emotion_intensity": "medium",
+            "risk_level": "medium",
+            "risk_types": ["未成年人信息", "家庭矛盾"],
+            "risk_notes": "涉及孩子和家庭互动，需要人工确认是否包含可识别隐私。",
+            "need_desensitization": True,
+            "desensitization_status": "needed",
+            "suggested_review_status": "pending",
+            "suggested_to_standard_qa": True,
+            "source_timestamp": "00:03:12",
+            "review_notes": "建议审核时确认是否需要进一步脱敏孩子细节。",
+        },
+        {
+            "session_id": str(session_id),
+            "qa_id": "mock-qa-002",
+            "raw_question": "我学习大本营后知道自己总是拖延，怎么开始一个小行动？",
+            "normalized_question": "意识到自己经常拖延后，如何启动一个小行动？",
+            "raw_answer": "不要一开始就设很大的目标。先选一个今天能完成的动作，比如写三句话、整理十分钟。关键是让身体先体验到我可以开始。",
+            "normalized_answer": "先设置当天能完成的小动作，让身体体验到可以开始，而不是直接设很大目标。",
+            "suggested_standard_question": "面对拖延时，如何启动一个足够小的行动？",
+            "suggested_standard_answer": "可以先选择一个今天就能完成的小动作，例如写三句话或整理十分钟。重点不是追求完美，而是让自己先体验到可以开始。",
+            "answer_person": "院长",
+            "primary_topic": "行动力与拖延",
+            "problem_tags": ["拖延", "小行动", "自我启动"],
+            "course_stage": "大本营综合",
+            "audience_tags": ["大本营学员"],
+            "emotion_intensity": "low",
+            "risk_level": "low",
+            "risk_types": [],
+            "risk_notes": "未发现明显隐私或边界风险。",
+            "need_desensitization": False,
+            "desensitization_status": "not_needed",
+            "suggested_review_status": "pending",
+            "suggested_to_standard_qa": True,
+            "source_timestamp": "00:16:44",
+            "review_notes": "可作为标准问答候选，但仍需人工审核。",
+        },
+        {
+            "session_id": str(session_id),
+            "qa_id": "mock-qa-003",
+            "raw_question": "我朋友说自己有抑郁症，我能不能用课程里的方法帮她判断严重程度？",
+            "normalized_question": "朋友提到抑郁症时，能否用课程方法判断严重程度？",
+            "raw_answer": "这类情况不要做诊断，也不要替对方判断严重程度。你可以表达关心，建议对方寻求专业医生或心理专业人员帮助。",
+            "normalized_answer": "不要做诊断或严重程度判断，可以表达关心，并建议对方寻求专业医生或心理专业人员帮助。",
+            "suggested_standard_question": "遇到疑似心理健康问题时，课程学习者应该如何把握边界？",
+            "suggested_standard_answer": "不做心理诊断，不判断严重程度；可以表达关心，并建议对方寻求专业医生或心理专业人员帮助。",
+            "answer_person": "辅导老师",
+            "primary_topic": "其他",
+            "problem_tags": ["心理边界", "专业转介"],
+            "course_stage": "不确定",
+            "audience_tags": ["大本营学员"],
+            "emotion_intensity": "high",
+            "risk_level": "high",
+            "risk_types": ["心理 / 医疗边界"],
+            "risk_notes": "涉及心理诊断边界，必须人工复审，默认不可自动入库。",
+            "need_desensitization": False,
+            "desensitization_status": "not_needed",
+            "suggested_review_status": "pending",
+            "suggested_to_standard_qa": False,
+            "source_timestamp": "00:27:05",
+            "review_notes": "高风险边界题，不建议自动进入标准问答库。",
+        },
+    ]
+

hy_qa_asset_backend/backend/app/services/qa_parser.py

@@ -0,0 +1,41 @@
+from typing import Any
+
+from app.services.risk_detector import reinforce_risk
+
+
+PRIMARY_TOPICS = {
+    "亲子关系",
+    "婚姻关系",
+    "原生家庭",
+    "情绪模式",
+    "课程学习",
+    "个人成长",
+    "家族系统",
+    "关系内耗",
+    "行动力与拖延",
+    "其他",
+}
+COURSE_STAGES = {"觉知", "原生", "合一", "生机", "心光", "大本营综合", "不确定"}
+DESENSITIZATION_STATUSES = {"not_needed", "needed", "done", "unknown"}
+REVIEW_STATUSES = {"pending", "reviewing", "approved", "needs_revision", "rejected", "forbidden"}
+
+
+def normalize_ai_item(item: dict[str, Any]) -> dict[str, Any]:
+    item = reinforce_risk(dict(item))
+    item["primary_topic"] = item.get("primary_topic") if item.get("primary_topic") in PRIMARY_TOPICS else "其他"
+    item["course_stage"] = item.get("course_stage") if item.get("course_stage") in COURSE_STAGES else "不确定"
+    item["desensitization_status"] = (
+        item.get("desensitization_status")
+        if item.get("desensitization_status") in DESENSITIZATION_STATUSES
+        else "unknown"
+    )
+    item["suggested_review_status"] = (
+        item.get("suggested_review_status") if item.get("suggested_review_status") in REVIEW_STATUSES else "pending"
+    )
+    for key in ("problem_tags", "audience_tags", "risk_types"):
+        if not isinstance(item.get(key), list):
+            item[key] = []
+    item["suggested_to_standard_qa"] = bool(item.get("suggested_to_standard_qa"))
+    item["need_desensitization"] = bool(item.get("need_desensitization"))
+    return item
+

hy_qa_asset_backend/backend/app/services/risk_detector.py

@@ -0,0 +1,35 @@
+from typing import Any
+
+
+HIGH_RISK_KEYWORDS = ["抑郁", "自杀", "诊断", "医生", "法律", "诉讼", "离婚协议"]
+PRIVACY_KEYWORDS = ["姓名", "电话", "微信", "学校", "单位", "城市", "住址", "孩子"]
+
+
+def reinforce_risk(item: dict[str, Any]) -> dict[str, Any]:
+    text = " ".join(
+        [
+            str(item.get("raw_question", "")),
+            str(item.get("raw_answer", "")),
+            str(item.get("normalized_question", "")),
+            str(item.get("normalized_answer", "")),
+        ]
+    )
+    risk_types = list(item.get("risk_types") or [])
+    if any(keyword in text for keyword in HIGH_RISK_KEYWORDS):
+        item["risk_level"] = "high" if item.get("risk_level") != "forbidden" else "forbidden"
+        if "心理 / 医疗边界" not in risk_types:
+            risk_types.append("心理 / 医疗边界")
+    if any(keyword in text for keyword in PRIVACY_KEYWORDS):
+        item["need_desensitization"] = True
+        if item.get("desensitization_status") == "not_needed":
+            item["desensitization_status"] = "needed"
+        if "学员隐私" not in risk_types:
+            risk_types.append("学员隐私")
+        if item.get("risk_level") == "low":
+            item["risk_level"] = "medium"
+    if item.get("risk_level") not in {"low", "medium", "high", "forbidden"}:
+        item["risk_level"] = "medium"
+        risk_types.append("不确定风险")
+    item["risk_types"] = risk_types
+    return item
+

hy_qa_asset_backend/backend/app/services/scheduler.py

@@ -0,0 +1,35 @@
+from apscheduler.schedulers.background import BackgroundScheduler
+
+from app import models
+from app.config import get_settings
+from app.database import SessionLocal
+from app.services.task_runner import TaskRunner
+from app.utils.time_utils import split_cron
+
+
+scheduler = BackgroundScheduler()
+
+
+def start_scheduler() -> None:
+    settings = get_settings()
+    if scheduler.running:
+        return
+    cron_kwargs = split_cron(settings.schedule_cron)
+    scheduler.configure(timezone=settings.schedule_timezone)
+    scheduler.add_job(
+        scheduled_scan,
+        "cron",
+        id="scheduled_feishu_scan",
+        replace_existing=True,
+        **cron_kwargs,
+    )
+    scheduler.start()
+
+
+def scheduled_scan() -> None:
+    import asyncio
+
+    with SessionLocal() as db:
+        runner = TaskRunner()
+        asyncio.run(runner.run(db, task_type=models.TaskType.scheduled_scan))
+

hy_qa_asset_backend/backend/app/services/standard_qa_service.py

@@ -0,0 +1,237 @@
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app import models
+from app.services.feishu_client import FeishuClient
+
+
+def _default_reviewer_id(db: Session, reviewer_id: int | None) -> int | None:
+    if reviewer_id:
+        return reviewer_id
+    user = db.query(models.User).filter(models.User.role == models.UserRole.reviewer).first()
+    return user.id if user else None
+
+
+def add_audit_log(
+    db: Session,
+    *,
+    user_id: int | None,
+    target_type: models.TargetType,
+    target_id: int,
+    action: models.AuditAction,
+    before_status: str | None,
+    after_status: str | None,
+    comment: str | None = None,
+) -> models.AuditLog:
+    log = models.AuditLog(
+        user_id=user_id,
+        target_type=target_type,
+        target_id=target_id,
+        action=action,
+        before_status=before_status,
+        after_status=after_status,
+        comment=comment,
+    )
+    db.add(log)
+    return log
+
+
+def refresh_session_counts(db: Session, session_id: int) -> None:
+    session = db.get(models.FeishuSession, session_id)
+    if not session:
+        return
+    session.qa_count = db.query(models.RawQAItem).filter(models.RawQAItem.session_id == session_id).count()
+    session.pending_review_count = (
+        db.query(models.RawQAItem)
+        .filter(
+            models.RawQAItem.session_id == session_id,
+            models.RawQAItem.review_status.in_([models.ReviewStatus.pending, models.ReviewStatus.reviewing]),
+        )
+        .count()
+    )
+    session.approved_count = (
+        db.query(models.RawQAItem)
+        .filter(models.RawQAItem.session_id == session_id, models.RawQAItem.review_status == models.ReviewStatus.approved)
+        .count()
+    )
+    session.standard_qa_count = (
+        db.query(models.StandardQAItem).filter(models.StandardQAItem.session_id == session_id).count()
+    )
+    if session.pending_review_count == 0 and session.qa_count > 0:
+        session.process_status = models.ProcessStatus.completed
+    elif session.qa_count > 0:
+        session.process_status = models.ProcessStatus.pending_review
+
+
+def create_standard_from_raw(
+    db: Session, raw: models.RawQAItem, reviewer_id: int | None
+) -> models.StandardQAItem | None:
+    if raw.standard_item:
+        return raw.standard_item
+    if not raw.suggested_to_standard_qa:
+        return None
+    # 高风险与禁止内容即使被标记状态，也不能自动进入标准问答库。
+    if raw.risk_level in {models.RiskLevel.high, models.RiskLevel.forbidden}:
+        return None
+    if raw.desensitization_status not in {
+        models.DesensitizationStatus.not_needed,
+        models.DesensitizationStatus.done,
+    }:
+        return None
+
+    next_id = (db.query(func.count(models.StandardQAItem.id)).scalar() or 0) + 1
+    standard = models.StandardQAItem(
+        standard_code=f"STD-{next_id:05d}",
+        source_raw_qa_id=raw.id,
+        session_id=raw.session_id,
+        standard_question=raw.suggested_standard_question or raw.normalized_question or raw.raw_question,
+        standard_answer=raw.suggested_standard_answer or raw.normalized_answer or raw.raw_answer,
+        similar_questions=[raw.normalized_question] if raw.normalized_question else [],
+        primary_topic=raw.primary_topic,
+        problem_tags=raw.problem_tags,
+        course_stage=raw.course_stage,
+        audience_tags=raw.audience_tags,
+        answer_boundary="只基于已审核答疑内容复用；不做心理诊断、医疗建议、法律判断或课程效果承诺。",
+        forbidden_expressions=["保证有效", "一定改变", "诊断为", "法律上必然"],
+        risk_level=raw.risk_level,
+        audit_status=models.AuditStatus.approved,
+        call_status=models.CallStatus.not_callable,
+        last_reviewer_id=reviewer_id,
+    )
+    db.add(standard)
+    return standard
+
+
+def approve_raw_qa(
+    db: Session, raw: models.RawQAItem, reviewer_id: int | None = None, comment: str | None = None
+) -> models.RawQAItem:
+    reviewer_id = _default_reviewer_id(db, reviewer_id)
+    before = raw.review_status.value
+    raw.review_status = models.ReviewStatus.approved
+    raw.reviewer_id = reviewer_id
+    if comment:
+        raw.review_notes = comment
+    add_audit_log(
+        db,
+        user_id=reviewer_id,
+        target_type=models.TargetType.raw_qa,
+        target_id=raw.id,
+        action=models.AuditAction.approve,
+        before_status=before,
+        after_status=raw.review_status.value,
+        comment=comment,
+    )
+    standard = create_standard_from_raw(db, raw, reviewer_id)
+    if standard:
+        db.flush()
+        add_audit_log(
+            db,
+            user_id=reviewer_id,
+            target_type=models.TargetType.standard_qa,
+            target_id=standard.id,
+            action=models.AuditAction.approve,
+            before_status="none",
+            after_status=models.AuditStatus.approved.value,
+            comment="由原始问答审核通过后生成，默认 not_callable。",
+        )
+    refresh_session_counts(db, raw.session_id)
+    db.commit()
+    feishu = FeishuClient()
+    if standard:
+        try:
+            feishu.write_standard_qa_items([standard])
+        except Exception as exc:  # noqa: BLE001
+            add_audit_log(
+                db,
+                user_id=reviewer_id,
+                target_type=models.TargetType.standard_qa,
+                target_id=standard.id,
+                action=models.AuditAction.revise,
+                before_status="feishu_write",
+                after_status="failed",
+                comment=f"飞书标准问答回写失败：{exc}",
+            )
+            db.commit()
+    session = db.get(models.FeishuSession, raw.session_id)
+    if session:
+        try:
+            feishu.update_session_status(session, session.process_status.value)
+        except Exception as exc:  # noqa: BLE001
+            add_audit_log(
+                db,
+                user_id=reviewer_id,
+                target_type=models.TargetType.session,
+                target_id=session.id,
+                action=models.AuditAction.revise,
+                before_status="feishu_session_sync",
+                after_status="failed",
+                comment=f"飞书场次统计回写失败：{exc}",
+            )
+            db.commit()
+    db.refresh(raw)
+    return raw
+
+
+def set_raw_status(
+    db: Session,
+    raw: models.RawQAItem,
+    status: models.ReviewStatus,
+    action: models.AuditAction,
+    reviewer_id: int | None = None,
+    comment: str | None = None,
+) -> models.RawQAItem:
+    reviewer_id = _default_reviewer_id(db, reviewer_id)
+    before = raw.review_status.value
+    raw.review_status = status
+    raw.reviewer_id = reviewer_id
+    if comment:
+        raw.review_notes = comment
+    add_audit_log(
+        db,
+        user_id=reviewer_id,
+        target_type=models.TargetType.raw_qa,
+        target_id=raw.id,
+        action=action,
+        before_status=before,
+        after_status=status.value,
+        comment=comment,
+    )
+    refresh_session_counts(db, raw.session_id)
+    db.commit()
+    db.refresh(raw)
+    return raw
+
+
+def mark_standard_call_status(
+    db: Session,
+    standard: models.StandardQAItem,
+    status: models.CallStatus,
+    reviewer_id: int | None = None,
+    comment: str | None = None,
+) -> models.StandardQAItem:
+    reviewer_id = _default_reviewer_id(db, reviewer_id)
+    before = standard.call_status.value
+    if status == models.CallStatus.callable:
+        raw = standard.source_raw_qa
+        if (
+            standard.audit_status != models.AuditStatus.approved
+            or standard.risk_level not in {models.RiskLevel.low, models.RiskLevel.medium}
+            or raw.desensitization_status
+            not in {models.DesensitizationStatus.not_needed, models.DesensitizationStatus.done}
+        ):
+            raise ValueError("只有已审核、低/中风险、且已完成脱敏或无需脱敏的标准问答可以标记为可调用")
+    standard.call_status = status
+    standard.last_reviewer_id = reviewer_id
+    add_audit_log(
+        db,
+        user_id=reviewer_id,
+        target_type=models.TargetType.standard_qa,
+        target_id=standard.id,
+        action=models.AuditAction.mark_callable if status == models.CallStatus.callable else models.AuditAction.disable,
+        before_status=before,
+        after_status=status.value,
+        comment=comment,
+    )
+    db.commit()
+    db.refresh(standard)
+    return standard

hy_qa_asset_backend/backend/app/services/task_runner.py

@@ -0,0 +1,124 @@
+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)