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