601 lines
26 KiB
Python
601 lines
26 KiB
Python
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
|