"""公共 AI 页面路由（/k/**）。核心设计： - 主页 /k/{token} = 目录索引（TOC），显示所有目录及其文档数量 - 目录页 /k/{token}/category/{path} = 该目录下的文档列表 - 文档页 /k/{token}/doc/{doc_token} = 单个文档内容 - AI 先看目录索引 → 根据用户描述选择目录 → 进入该目录看文档规则： - 零 JS、零 Cookie、零登录、SSR 输出、标准 HTML - - - 限流：内存 TokenBucket - 统一 404 防存在性探测 """ import json import re from fastapi import APIRouter, Depends, Query, Request, Response from fastapi.responses import HTMLResponse, PlainTextResponse from fastapi import Path as PathParam from sqlalchemy.orm import Session from app.api.deps import get_db from app.core.errors import LinkExpiredError, NotFoundError, RateLimitedError from app.core.rate_limit import check_rate_limit from app.core.security import decrypt_token from app.models.document_category import DocumentCategory from app.services.access_log_service import AccessLogService from app.services.kb_public_service import KbPublicService router = APIRouter(prefix="/k", tags=["public"]) async def link_expired_handler(request: Request, exc: LinkExpiredError) -> Response: """链接过期：HTML 路径返回友好失效页，JSON/MD/TXT 返回对应格式的错误信息。""" path = request.url.path message = "链接已失效，请联系分享者重新生成链接。" if path.endswith(".json"): return Response( status_code=410, content=json.dumps({"code": "LINK_EXPIRED", "message": message, "detail": None}, ensure_ascii=False), media_type="application/json", ) if path.endswith(".md") or path.endswith(".txt"): return PlainTextResponse(content=message, status_code=410, media_type="text/plain; charset=utf-8") return HTMLResponse( status_code=410, content=f""" 链接已失效

⏰

链接无效或已失效

该知识库链接已过期。
请联系分享者重新生成链接后，获取新的访问地址。

""", ) def _log_access(db: Session, kb_id: str, path: str, request: Request, doc_id: str | None = None, req_type: str | None = None) -> None: """记录访问日志（best-effort）。""" try: ua = request.headers.get("user-agent", "") AccessLogService(db).record( knowledge_base_id=kb_id, document_id=doc_id, path=path, user_agent=ua, request_type=req_type, ) except Exception: pass def _rate_limit(request: Request, token: str) -> None: ip = request.client.host if request.client else "unknown" if not check_rate_limit(token_key=token[:16], ip_key=ip): raise RateLimitedError() def _robots_meta() -> str: return '' def _referrer_meta() -> str: return '' def _get_category_by_path(db: Session, kb_id: str, path: str) -> DocumentCategory | None: """根据路径查找分类。""" from sqlalchemy import select stmt = select(DocumentCategory).where( DocumentCategory.knowledge_base_id == kb_id, DocumentCategory.path == path, ) return db.scalars(stmt).first() # ============================================================ # 目录索引（主页）- AI 从这里了解知识库结构 # ============================================================ @router.get("/{token}.json") def kb_index_json( token: str, request: Request, db: Session = Depends(get_db), ) -> Response: """知识库完整内容（JSON）- 按目录分组，包含所有文档内容。 AI 访问此接口获取知识库全部内容，按目录结构组织。 """ _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) _log_access(db, kb.id, f"/k/{token}.json", request, req_type="json") category_tree = svc.get_category_tree(kb) # 获取所有文档（按目录分组） docs, _ = svc.list_documents(kb) # 按 category_id 分组 doc_map: dict[str, list] = {} for doc in docs: cat_id = doc.category_id or "__uncategorized__" if cat_id not in doc_map: doc_map[cat_id] = [] doc_map[cat_id].append({ "title": doc.title or doc.original_filename, "file_type": doc.file_ext, "description": doc.description, "summary": doc.content_summary, "keywords": doc.keywords.split(",") if doc.keywords else [], "updated_at": doc.updated_at, }) # 递归构建目录 + 文档 def build_tree_with_docs(nodes: list) -> list: result = [] for node in nodes: cat_docs = doc_map.get(node["id"], []) node_data = { "name": node["name"], "path": node["path"], "is_folder": node["is_folder"], "documents": cat_docs, "children": build_tree_with_docs(node.get("children", [])), } result.append(node_data) return result tree_with_docs = build_tree_with_docs(category_tree) # 未分类文档 uncategorized = doc_map.get("__uncategorized__", []) data = { "name": kb.name, "description": kb.description, "categories": tree_with_docs, "uncategorized_documents": uncategorized, } return Response( content=json.dumps(data, ensure_ascii=False, indent=2), media_type="application/json", ) @router.get("/{token}.md") def kb_index_markdown( token: str, request: Request, db: Session = Depends(get_db), ) -> PlainTextResponse: """知识库目录索引（Markdown）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) _log_access(db, kb.id, f"/k/{token}.md", request, req_type="md") category_tree = svc.get_category_tree(kb) lines = [f"# {kb.name}", ""] if kb.description: lines.append(kb.description) lines.append("") lines.append("## 目录结构") lines.append("") lines.append("请通过以下目录链接访问对应内容：") lines.append("") def render_tree_md(nodes: list, level: int = 0): for node in nodes: prefix = " " * level icon = "📁" if node["is_folder"] else "📄" doc_count = f" ({node['doc_count']}篇)" if node["doc_count"] > 0 else "" url = f"/k/{token}/category{node['path']}" lines.append(f"{prefix}- {icon} [{node['name']}]({url}){doc_count}") if node.get("children"): render_tree_md(node["children"], level + 1) render_tree_md(category_tree) lines.append("") lines.append("---") lines.append(f"访问各目录链接查看具体文档。如需搜索：/k/{token}/search?q=关键词") return PlainTextResponse(content="\n".join(lines), media_type="text/markdown") @router.get("/{token}.txt") def kb_index_text( token: str, request: Request, db: Session = Depends(get_db), ) -> PlainTextResponse: """知识库目录索引（纯文本）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) _log_access(db, kb.id, f"/k/{token}.txt", request, req_type="txt") category_tree = svc.get_category_tree(kb) lines = [kb.name, "=" * len(kb.name), ""] if kb.description: lines.append(kb.description) lines.append("") lines.append("目录结构：") lines.append("") def render_tree_txt(nodes: list, level: int = 0): for node in nodes: prefix = " " * level icon = "[文件夹]" if node["is_folder"] else "[文档]" doc_count = f" ({node['doc_count']}篇)" if node["doc_count"] > 0 else "" url = f"/k/{token}/category{node['path']}" lines.append(f"{prefix}{icon} {node['name']}{doc_count}") lines.append(f"{prefix} → {url}") if node.get("children"): render_tree_txt(node["children"], level + 1) render_tree_txt(category_tree) return PlainTextResponse(content="\n".join(lines), media_type="text/plain") @router.get("/{token}") def kb_index_html( token: str, request: Request, db: Session = Depends(get_db), ) -> HTMLResponse: """知识库完整内容（HTML）- 按目录分组显示所有文档。 AI 直接读取此页面即可获取全部内容，无需点击。人类用户可通过左侧目录快速跳转到对应章节。 """ _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) _log_access(db, kb.id, f"/k/{token}", request, req_type="html") category_tree = svc.get_category_tree(kb) docs, _ = svc.list_documents(kb) # 按 category_id 分组 doc_map: dict[str, list] = {} for doc in docs: cat_id = doc.category_id or "__uncategorized__" if cat_id not in doc_map: doc_map[cat_id] = [] doc_map[cat_id].append(doc) # 递归渲染目录 + 文档（按目录顺序） section_counter = [0] def render_section(nodes: list, level: int = 0) -> tuple[str, str]: """返回 (内容HTML, 目录HTML)""" content_html = "" toc_html = "" for node in nodes: section_counter[0] += 1 section_id = f"section-{section_counter[0]}" icon = "📁" if node["is_folder"] else "📄" heading_tag = "h2" if level == 0 else "h3" if level == 1 else "h4" indent = " " * level # 目录项 doc_count = f" ({node['doc_count']}篇)" if node["doc_count"] > 0 else "" toc_html += f'{indent}

' toc_html += f'{icon} {node["name"]}{doc_count}

\n' # 内容标题 content_html += f'<{heading_tag} id="{section_id}">{icon} {node["name"]}\n' # 该目录下的文档 cat_docs = doc_map.get(node["id"], []) for doc in cat_docs: doc_token = decrypt_token(doc.doc_token_encrypted) if doc.doc_token_encrypted else "" doc_url = f"/k/{token}/doc/{doc_token}" summary = doc.content_summary or "" keywords = doc.keywords or "" status_badge = f' [{doc.status}]' if doc.status != "READY" else "" content_html += f'

\n' content_html += f'

{doc.title or doc.original_filename}{status_badge}

\n' content_html += f'

类型：{doc.file_ext} | 更新：{doc.updated_at}

\n' if summary: content_html += f'

{summary}

\n' if keywords: content_html += f'

关键词：{keywords}

\n' content_html += f'

\n' # 子目录 if node.get("children"): child_content, child_toc = render_section(node["children"], level + 1) content_html += child_content toc_html += child_toc return content_html, toc_html all_content, all_toc = render_section(category_tree) # 未分类文档 uncategorized = doc_map.get("__uncategorized__", []) if uncategorized: all_content += '

📄 未分类文档

\n' for doc in uncategorized: doc_token = decrypt_token(doc.doc_token_encrypted) if doc.doc_token_encrypted else "" doc_url = f"/k/{token}/doc/{doc_token}" all_content += f'

{doc.title or doc.original_filename}

\n' html = f""" {kb.name} {_robots_meta()} {_referrer_meta()}

{kb.name}

" + kb.description + "

" if kb.description else ""} {all_content}

""" return HTMLResponse(content=html) # ============================================================ # 目录页面 - 显示某个目录下的文档 # ============================================================ @router.get("/{token}/category/{path:path}.json") def category_json( token: str, path: str, request: Request, db: Session = Depends(get_db), ) -> Response: """目录下的文档列表（JSON）。 path 格式：01公司层/公司基本信息（不含前后斜杠） """ _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) # 规范化路径 category_path = f"/{path.strip('/')}/" cat = _get_category_by_path(db, kb.id, category_path) if cat is None: raise NotFoundError("目录不存在。") _log_access(db, kb.id, f"/k/{token}/category/{path}.json", request, req_type="category_json") # 获取该目录下的文档 docs, total = svc.list_documents(kb, category_id=cat.id) doc_list = [] for doc in docs: doc_token = decrypt_token(doc.doc_token_encrypted) if doc.doc_token_encrypted else "" doc_list.append({ "title": doc.title or doc.original_filename, "file_type": doc.file_ext, "description": doc.description, "summary": doc.content_summary, "keywords": doc.keywords.split(",") if doc.keywords else [], "updated_at": doc.updated_at, "url": f"/k/{token}/doc/{doc_token}", }) data = { "category": cat.name, "category_path": cat.path, "document_count": total, "documents": doc_list, "back_to_index": f"/k/{token}.json", } return Response( content=json.dumps(data, ensure_ascii=False, indent=2), media_type="application/json", ) @router.get("/{token}/category/{path:path}.md") def category_markdown( token: str, path: str, request: Request, db: Session = Depends(get_db), ) -> PlainTextResponse: """目录下的文档列表（Markdown）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) category_path = f"/{path.strip('/')}/" cat = _get_category_by_path(db, kb.id, category_path) if cat is None: raise NotFoundError("目录不存在。") _log_access(db, kb.id, f"/k/{token}/category/{path}.md", request, req_type="category_md") docs, _ = svc.list_documents(kb, category_id=cat.id) lines = [f"# {cat.name}", ""] lines.append(f"知识库：{kb.name}") lines.append("") if docs: lines.append("## 文档列表") lines.append("") for doc in docs: title = doc.title or doc.original_filename lines.append(f"### {title}") if doc.description: lines.append(f"- 描述：{doc.description}") if doc.keywords: lines.append(f"- 关键词：{doc.keywords}") if doc.content_summary: lines.append(f"- 摘要：{doc.content_summary}") lines.append(f"- 更新：{doc.updated_at}") lines.append("") else: lines.append("暂无文档。") return PlainTextResponse(content="\n".join(lines), media_type="text/markdown") @router.get("/{token}/category/{path:path}.txt") def category_text( token: str, path: str, request: Request, db: Session = Depends(get_db), ) -> PlainTextResponse: """目录下的文档列表（纯文本）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) category_path = f"/{path.strip('/')}/" cat = _get_category_by_path(db, kb.id, category_path) if cat is None: raise NotFoundError("目录不存在。") _log_access(db, kb.id, f"/k/{token}/category/{path}.txt", request, req_type="category_txt") docs, _ = svc.list_documents(kb, category_id=cat.id) lines = [cat.name, "=" * len(cat.name), ""] lines.append(f"知识库：{kb.name}") lines.append("") if docs: for i, doc in enumerate(docs, 1): title = doc.title or doc.original_filename lines.append(f"{i}. {title}") if doc.description: lines.append(f" 描述：{doc.description}") if doc.keywords: lines.append(f" 关键词：{doc.keywords}") lines.append("") else: lines.append("暂无文档。") return PlainTextResponse(content="\n".join(lines), media_type="text/plain") @router.get("/{token}/category/{path:path}") def category_html( token: str, path: str, request: Request, db: Session = Depends(get_db), ) -> HTMLResponse: """目录下的文档列表（HTML）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) category_path = f"/{path.strip('/')}/" cat = _get_category_by_path(db, kb.id, category_path) if cat is None: raise NotFoundError("目录不存在。") _log_access(db, kb.id, f"/k/{token}/category/{path}", request, req_type="category_html") docs, total = svc.list_documents(kb, category_id=cat.id) doc_rows = "" for doc in docs: doc_token = decrypt_token(doc.doc_token_encrypted) if doc.doc_token_encrypted else "" doc_url = f"/k/{token}/doc/{doc_token}" keywords_html = f'

关键词：{doc.keywords}

' if doc.keywords else "" summary_html = f'

{doc.content_summary or ""}

' if doc.content_summary else "" doc_rows += f"""

{doc.title or doc.original_filename}

类型：{doc.file_ext} | 更新：{doc.updated_at}

{summary_html} {keywords_html}

""" html = f""" {cat.name} - {kb.name} {_robots_meta()} {_referrer_meta()}

{cat.name}

共 {total} 篇文档

{doc_rows if doc_rows else "

该目录暂无文档。

"} """ return HTMLResponse(content=html) # ============================================================ # 单文档访问 # ============================================================ @router.get("/{token}/doc/{doc_token}.md") def doc_page_markdown( token: str, doc_token: str, request: Request, db: Session = Depends(get_db), ) -> PlainTextResponse: """文档内容（Markdown）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) doc = svc.get_document_by_token(kb, doc_token) _log_access(db, kb.id, f"/k/{token}/doc/{doc_token}.md", request, doc_id=doc.id, req_type="doc_md") return PlainTextResponse(content=svc.get_document_content(doc), media_type="text/markdown") @router.get("/{token}/doc/{doc_token}.txt") def doc_page_text( token: str, doc_token: str, request: Request, db: Session = Depends(get_db), ) -> PlainTextResponse: """文档内容（纯文本）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) doc = svc.get_document_by_token(kb, doc_token) _log_access(db, kb.id, f"/k/{token}/doc/{doc_token}.txt", request, doc_id=doc.id, req_type="doc_txt") content = svc.get_document_content(doc) text = re.sub(r"[#*_`\[\]()>]", "", content) text = re.sub(r"\n{3,}", "\n\n", text) return PlainTextResponse(content=text.strip(), media_type="text/plain") @router.get("/{token}/doc/{doc_token}") def doc_page_html( token: str, doc_token: str = PathParam(pattern=r"^[A-Za-z0-9_-]+$"), request: Request = None, db: Session = Depends(get_db), ) -> HTMLResponse: """文档内容（HTML）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) doc = svc.get_document_by_token(kb, doc_token) _log_access(db, kb.id, f"/k/{token}/doc/{doc_token}", request, doc_id=doc.id, req_type="doc_html") content = svc.get_document_content(doc) html_content = _markdown_to_html(content) # 获取文档所属目录 category_name = "" category_path = "" if doc.category_id: cat = db.get(DocumentCategory, doc.category_id) if cat: category_name = cat.name category_path = cat.path keywords_html = f"

关键词：{doc.keywords}

" if doc.keywords else "" breadcrumb = f'📚 {kb.name}' if category_name: breadcrumb += f' > {category_name}' html = f""" {doc.title or doc.original_filename} {_robots_meta()} {_referrer_meta()}

{doc.title or doc.original_filename}

类型：{doc.file_ext} | 更新：{doc.updated_at}

描述：" + doc.description + "

" if doc.description else ""} {keywords_html}

{html_content}

""" return HTMLResponse(content=html) # ============================================================ # 搜索 # ============================================================ @router.get("/{token}/search") def search_html( token: str, q: str = Query(..., min_length=1), page: int = Query(1, ge=1), request: Request = None, db: Session = Depends(get_db), ) -> HTMLResponse: """搜索文档（HTML）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) _log_access(db, kb.id, f"/k/{token}/search?q={q}", request, req_type="search") results, total = svc.search_documents(kb, q, page=page) result_items = "" for item in results: doc_url = f"/k/{token}/doc/{item['url_hint'] or ''}" category_path = item.get('category_path', '') category_badge = f' [{category_path}]' if category_path else "" result_items += f"""

{item['title']}{category_badge}

类型：{item['file_type']} | 更新：{item['updated_at']}

" + (item.get('description') or '') + "

""" html = f""" 搜索：{q} - {kb.name} {_robots_meta()} {_referrer_meta()}

搜索：{q}

共找到 {total} 个结果

{result_items if result_items else "

未找到相关文档。

← 返回目录索引

""" return HTMLResponse(content=html) @router.get("/{token}/search.json") def search_json( token: str, q: str = Query(..., min_length=1), page: int = Query(1, ge=1), request: Request = None, db: Session = Depends(get_db), ) -> Response: """搜索文档（JSON）。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) results, total = svc.search_documents(kb, q, page=page) data = { "query": q, "total": total, "results": results, } return Response( content=json.dumps(data, ensure_ascii=False, indent=2), media_type="application/json", ) # ============================================================ # 工具函数 # ============================================================ def _markdown_to_html(markdown: str) -> str: """简单 Markdown → HTML 转换（安全处理）。""" html = markdown.replace("&", "&").replace("<", "<").replace(">", ">") html = re.sub(r"^#### (.+)$", r"

\1

", html, flags=re.MULTILINE) html = re.sub(r"^### (.+)$", r"

\1

", html, flags=re.MULTILINE) html = re.sub(r"^## (.+)$", r"

\1

", html, flags=re.MULTILINE) html = re.sub(r"^# (.+)$", r"

\1

", html, flags=re.MULTILINE) html = re.sub(r"\*\*(.+?)\*\*", r"\1", html) html = re.sub(r"\*(.+?)\*", r"\1", html) html = re.sub(r"```[\s\S]*?```", lambda m: f"

{m.group(0)[3:-3]}

", html) html = re.sub(r"`([^`]+)`", r"\1", html) html = re.sub(r"\n\n+", "

", html) html = f"

{html}

" return html