"""飞书服务台(Helpdesk)知识库导出 → `support_kb_articles` 导入。 与 :mod:`app.domains.support.feishu_docs` 的 wiki 镜像**性质相反**,别混: * wiki 镜像是**机器所有**的 —— 飞书是唯一的编辑面,人在 Helix 里改会被下一轮 同步覆盖,所以 `kb_service._guard_mirrored` 直接 422 拒绝。 * 服务台 FAQ 是**一次性交接** —— 飞书服务台正在被 Helix 求助域收编,这批 问答导进来之后 Helix 就是编辑面,管理员照常增删改查。导入是幂等的 (按问题标题在本来源内 upsert),所以重新导出、重新上传只会更新内容, 不会造出第二份。 导出模板的形状(2026-08-30 用真实导出的 124 条核过): 第 1 行 填写说明(一整段中文,只占 A 列) 第 2 行 表头:`1 级分类 | 问题 | 答案 | 相似问题 | 语言` 第 3 行起 数据 「问题」列左边的所有列都是分类层级(模板允许在 3 级分类后继续加列),表头行 的位置和分类列的数量都不写死 —— 按表头文字定位。 「答案」有两种形态:纯文本,或飞书富文本导出的 JSON 数组 (`[{"type":"text","content":…}, {"type":"hyperlink","text":…,"url":…}, {"type":"img","src":…,"key":…}]`)。渲染成 Markdown 落库,检索与展示都用它。 """ from __future__ import annotations import io import json import logging import re from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from openpyxl import load_workbook from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.core.xlsx_safety import bounded_row, preflight_xlsx from app.domains.support.models import ( KB_ARCHIVED, KB_DRAFT, KB_PUBLISHED, KB_SOURCE_HELPDESK_FAQ, SupportKbArticle, ) logger = logging.getLogger(__name__) MAX_UPLOAD_BYTES = 10 * 1024 * 1024 MAX_FAQ_ROWS = 2_000 _TITLE_MAX = 255 _BODY_MAX = 16000 # keep in sync with kb_service._BODY_MAX _ALIASES_MAX = 4000 _SOURCE_CATEGORY_MAX = 64 #: 表头识别用的候选词(全部小写比较)。飞书导出模板有中英两版。 _QUESTION_HEADERS = ("问题", "question", "questions") _ANSWER_HEADERS = ("答案", "answer", "answers") _SIMILAR_HEADERS = ("相似问题", "similar question", "similar questions") _LANGUAGE_HEADERS = ("语言", "language", "locale") #: 「生效类型 / 生效时间 / 失效时间」等模板可选列 —— 认得出来就跳过,不当分类。 _IGNORED_HEADERS = ( "生效类型", "生效时间", "失效时间", "effective type", "effective time", "expiry time", ) #: 原始一级分类 → Helix 的 6 个固定 category。按**关键词**匹配而不是照抄一张 #: 标签对照表:服务台的分类名会改(「网络 & 飞连VPN」上一版叫「网络」), #: 关键词规则对没见过的标签也能给出合理归类。顺序即优先级。 _CATEGORY_RULES: tuple[tuple[str, tuple[str, ...]], ...] = ( ("network", ("vpn", "网络", "飞连", "wifi", "wi-fi", "无线", "network", "internet")), ( "account_access", ( "账号", "帐号", "权限", "登录", "登陆", "密码", "认证", "account", "access", "password", "login", "sso", "permission", ), ), ("facility", ("会议", "投屏", "工位", "门禁", "meeting", "room", "office", "facility")), ( "hardware", ( "打印", "设备", "电脑", "笔记本", "显示器", "硬件", "外设", "printer", "print", "equipment", "hardware", "laptop", "device", "monitor", ), ), ( "software", ( "软件", "邮箱", "邮件", "飞书", "google", "workspace", "lark", "software", "app", "license", "mail", "email", "zoom", "slack", # "spam" 与「垃圾邮件」是同一组条目的中英两版:少了它, # "Spam & Security" 落 other 而「垃圾邮件」落 software, # 同一件事在两种语言下被分到两处。 "spam", ), ), ) _LANGUAGE_ALIASES = { "zh": "zh", "zh-cn": "zh", "zh_cn": "zh", "cn": "zh", "中文": "zh", "简体中文": "zh", "en": "en", "en-us": "en", "en_us": "en", "english": "en", "英文": "en", "ja": "ja", "ja-jp": "ja", "jp": "ja", "日本語": "ja", "日文": "ja", } def _invalid(message: str) -> AppException: return AppException(code="INVALID_FAQ_WORKBOOK", message=message, status_code=400) # ── 富文本渲染 ────────────────────────────────────────────────────── def render_answer(raw: str) -> tuple[str, int]: """飞书富文本答案 → Markdown。返回 ``(markdown, 图片数)``。 纯文本答案原样返回(图片数 0)。图片渲染成**链接**而不是 `![]()` 内嵌: 导出里的 `src` 指向飞书图床(`…/saipan/v2/api/image/…`),实测匿名访问 返回 **401**,内嵌只会在员工页面上留一个碎图,而模型会把这个 URL 当成 一份「可以打开的资料」。做成链接:登录了飞书的人点得开,模型看到的是 「[图片]」这四个字,不会拿它编造内容。 """ text = (raw or "").strip() if not text.startswith("["): return text, 0 try: parts = json.loads(text) except (ValueError, TypeError): return text, 0 if not isinstance(parts, list): return text, 0 out: list[str] = [] images = 0 for part in parts: if not isinstance(part, dict): continue kind = part.get("type") if kind == "text": out.append(str(part.get("content") or "")) elif kind == "hyperlink": url = str(part.get("url") or "").strip() label = str(part.get("text") or "").strip() if not url: out.append(label) elif not label or label == url: out.append(url) else: out.append(f"[{label}]({url})") elif kind == "img": images += 1 src = str(part.get("src") or "").strip() out.append(f"[图片]({src})" if src else "[图片]") elif "content" in part: # 未知类型但带正文 —— 宁可多留文字,也不要静默丢内容。 out.append(str(part.get("content") or "")) rendered = "".join(out).strip() # 富文本里每个换行是一个独立的 text 片段,连续空行会攒成一大片。 rendered = re.sub(r"\n{3,}", "\n\n", rendered) return rendered, images def map_category(source_category: str) -> str: """原始分类标签 → Helix 的固定 category(认不出来 → ``other``)。""" label = (source_category or "").strip().lower() if not label: return "other" for canonical, keywords in _CATEGORY_RULES: if any(kw in label for kw in keywords): return canonical return "other" def normalize_language(value: str) -> str | None: key = (value or "").strip().lower() return _LANGUAGE_ALIASES.get(key) def normalize_aliases(value: str) -> str: """「相似问题」→ 逐行去空、去重(保序)、截断。""" seen: set[str] = set() lines: list[str] = [] for raw_line in (value or "").replace("\r\n", "\n").replace("\r", "\n").split("\n"): line = raw_line.strip() if not line or line in seen: continue seen.add(line) lines.append(line) return "\n".join(lines)[:_ALIASES_MAX] def _dedupe_key(title: str) -> str: """同一来源内判「同一条 FAQ」的键。 导出没有 id 列,问题文本就是它的身份。归一化到「去空白 + 小写」,这样 改个空格或大小写不会在库里造出第二条。 """ return re.sub(r"\s+", "", (title or "")).lower() # ── 解析 ──────────────────────────────────────────────────────────── @dataclass class ParsedFaq: title: str body: str source_category: str category: str aliases: str language: str | None images: int row_number: int @dataclass class ParseResult: rows: list[ParsedFaq] = field(default_factory=list) #: 被跳过的行:``(行号, 原因)``。永远回报,不静默丢。 skipped: list[tuple[int, str]] = field(default_factory=list) images: int = 0 def _header_index(cells: list[str], candidates: tuple[str, ...]) -> int | None: for idx, cell in enumerate(cells): low = cell.strip().lower() if any(low == c or low.startswith(c) for c in candidates): return idx return None def _cell_text(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value return str(value) def parse_faq_workbook(payload: bytes) -> ParseResult: """解析一份飞书服务台知识库导出。表头位置与分类列数量都由表头文字定位。""" if not payload: raise _invalid("文件为空") if len(payload) > MAX_UPLOAD_BYTES: raise AppException( code="FAQ_WORKBOOK_TOO_LARGE", message=f"文件超过 {MAX_UPLOAD_BYTES // (1024 * 1024)} MB 上限", status_code=413, ) preflight_xlsx(payload) try: workbook = load_workbook(io.BytesIO(payload), data_only=True, read_only=True) except Exception as exc: # noqa: BLE001 - openpyxl raises a zoo of types raise _invalid("无法读取该 xlsx 文件") from exc try: sheet = workbook.worksheets[0] raw_rows: list[list[str]] = [] for row_number, row in enumerate(sheet.iter_rows(values_only=True), start=1): if row_number > MAX_FAQ_ROWS + 10: raise AppException( code="FAQ_WORKBOOK_TOO_MANY_ROWS", message=f"最多导入 {MAX_FAQ_ROWS} 行", status_code=413, ) raw_rows.append([_cell_text(v) for v in bounded_row(row)]) finally: workbook.close() header_row = None q_idx = a_idx = None for idx, cells in enumerate(raw_rows[:10]): qi = _header_index(cells, _QUESTION_HEADERS) ai = _header_index(cells, _ANSWER_HEADERS) if qi is not None and ai is not None: header_row, q_idx, a_idx = idx, qi, ai break if header_row is None or q_idx is None or a_idx is None: raise _invalid("未找到表头行(需要包含「问题」与「答案」两列)") headers = raw_rows[header_row] sim_idx = _header_index(headers, _SIMILAR_HEADERS) lang_idx = _header_index(headers, _LANGUAGE_HEADERS) # 「问题」列左边的列都是分类层级;认得出来的模板可选列排除掉。 category_idx = [ i for i in range(q_idx) if not any(headers[i].strip().lower().startswith(x) for x in _IGNORED_HEADERS) ] result = ParseResult() for offset, cells in enumerate(raw_rows[header_row + 1 :], start=header_row + 2): if len(result.rows) >= MAX_FAQ_ROWS: raise AppException( code="FAQ_WORKBOOK_TOO_MANY_ROWS", message=f"最多导入 {MAX_FAQ_ROWS} 行", status_code=413, ) def cell(index: int | None) -> str: if index is None or index >= len(cells): return "" return cells[index].strip() if isinstance(cells[index], str) else "" title = cell(q_idx) answer_raw = cell(a_idx) if not title and not answer_raw: continue # 纯空行 —— 不算跳过 if not title: result.skipped.append((offset, "缺少问题")) continue body, images = render_answer(answer_raw) if not body: result.skipped.append((offset, "缺少答案")) continue source_category = " / ".join( part for part in (cell(i) for i in category_idx) if part )[:_SOURCE_CATEGORY_MAX] result.images += images result.rows.append( ParsedFaq( title=title[:_TITLE_MAX], body=body[:_BODY_MAX], source_category=source_category, category=map_category(source_category), aliases=normalize_aliases(cell(sim_idx)), language=normalize_language(cell(lang_idx)), images=images, row_number=offset, ) ) if not result.rows: raise _invalid("表格里没有可导入的问答行") return result # ── 导入 ──────────────────────────────────────────────────────────── async def import_faq_rows( db: AsyncSession, parsed: ParseResult, *, actor_id: Any = None, publish: bool = True, dry_run: bool = False, trace_id: str = "", ) -> dict[str, Any]: """把解析结果 upsert 进 `support_kb_articles`(`source='helpdesk_faq'`)。 幂等:按归一化后的问题文本在**本来源内**匹配。内容一字不差的行既不写库 也不清 embedding —— 重复上传同一份导出的代价必须是零,否则每次重传都要 把整个语料重新向量化一遍。 ``publish=False`` 只影响**新建**的行:已经上线的条目不会被一次导入 悄悄下线,已经归档的也不会被一次导入悄悄复活。 """ # Collapse duplicates WITHIN the file first (last one wins) and report # them. Without this, a hand-edited export with the same question twice # counts as two creates on a dry run and one on the real run — the preview # would disagree with what the import then did. deduped: dict[str, ParsedFaq] = {} for row in parsed.rows: key = _dedupe_key(row.title) if key in deduped: parsed.skipped.append((deduped[key].row_number, "问题重复,以最后一行为准")) deduped[key] = row rows_to_apply = list(deduped.values()) existing_rows = ( ( await db.execute( select(SupportKbArticle).where( SupportKbArticle.source == KB_SOURCE_HELPDESK_FAQ ) ) ) .scalars() .all() ) by_key: dict[str, SupportKbArticle] = {} for existing in existing_rows: by_key.setdefault(_dedupe_key(existing.title), existing) created = updated = unchanged = 0 reembed: list[SupportKbArticle] = [] for row in rows_to_apply: art = by_key.get(_dedupe_key(row.title)) if art is None: created += 1 if dry_run: continue new_art = SupportKbArticle( title=row.title, body=row.body, category=row.category, status=KB_PUBLISHED if publish else KB_DRAFT, source=KB_SOURCE_HELPDESK_FAQ, author_id=actor_id, published_at=datetime.now(UTC) if publish else None, search_aliases=row.aliases or None, language=row.language, source_category=row.source_category or None, ) db.add(new_art) by_key[_dedupe_key(row.title)] = new_art if publish: reembed.append(new_art) continue text_changed = art.body != row.body or (art.search_aliases or "") != row.aliases meta_changed = ( art.title != row.title or art.category != row.category or (art.language or None) != row.language or (art.source_category or None) != (row.source_category or None) ) if not text_changed and not meta_changed: unchanged += 1 continue updated += 1 if dry_run: continue art.title = row.title art.body = row.body art.category = row.category art.search_aliases = row.aliases or None art.language = row.language art.source_category = row.source_category or None if text_changed: # 可检索文本变了,旧向量就是错的 —— 与 kb_service.update 同一约定。 art.embedding = None art.embedding_model = None art.embedded_at = None if art.status == KB_PUBLISHED: reembed.append(art) if not dry_run: await db.flush() await AuditService(db).log( trace_id=trace_id or "", actor_type="user", action="support.kb.faq_imported", status="success", actor_user_id=actor_id, resource_type="support_kb_article", detail={ "created": created, "updated": updated, "unchanged": unchanged, "skipped": len(parsed.skipped), "images": parsed.images, "publish": publish, }, ) return { "dry_run": dry_run, "parsed": len(rows_to_apply), "created": created, "updated": updated, "unchanged": unchanged, # Entries an admin had taken down. An import never revives them, so # say the number out loud — otherwise "I re-imported and three are # still missing" reads as a bug. "kept_archived": sum(1 for a in existing_rows if a.status == KB_ARCHIVED), "skipped": [{"row": n, "reason": r} for n, r in parsed.skipped[:50]], "skipped_total": len(parsed.skipped), # 图片留在飞书图床里(匿名 401),导入的是链接不是图片本身 —— 明说, # 否则管理员会以为这批条目已经图文完整。 "images_linked": parsed.images, "reembed_ids": [a.id for a in reembed], } __all__ = [ "MAX_FAQ_ROWS", "MAX_UPLOAD_BYTES", "ParseResult", "ParsedFaq", "import_faq_rows", "map_category", "normalize_aliases", "normalize_language", "parse_faq_workbook", "render_answer", ]