"""公共 AI 页面路由(/k/**)。 规则: - 零 JS、零 Cookie、零登录、SSR 输出、标准 HTML - - - 限流:内存 TokenBucket - 统一 404 防存在性探测 """ import json from fastapi import APIRouter, Depends, Query, Request, Response from fastapi.responses import HTMLResponse, PlainTextResponse from sqlalchemy.orm import Session from app.api.deps import get_db from app.core.errors import NotFoundError, RateLimitedError from app.core.rate_limit import check_rate_limit from app.core.security import decrypt_token from app.services.kb_public_service import KbPublicService router = APIRouter(prefix="/k", tags=["public"]) 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 '' # --- 知识库入口(后缀路由必须先于无后缀路由注册)--- @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) docs, _ = svc.list_documents(kb) lines = [f"# {kb.name}", ""] if kb.description: lines.append(kb.description) lines.append("") lines.append("## 文档列表") lines.append("") for doc in docs: title = doc.title or doc.original_filename lines.append(f"### {title}") lines.append(f"- 类型:{doc.file_ext}") 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("") 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) docs, _ = svc.list_documents(kb) lines = [kb.name, "=" * len(kb.name), ""] if kb.description: lines.append(kb.description) lines.append("") lines.append("文档列表:") lines.append("") 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("") return PlainTextResponse(content="\n".join(lines), media_type="text/plain") @router.get("/{token}.json") def kb_index_json( token: str, request: Request, db: Session = Depends(get_db), ) -> Response: """知识库首页(JSON)。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) docs, _ = svc.list_documents(kb) doc_list = [] for doc in docs: 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, }) data = { "name": kb.name, "description": kb.description, "document_count": len(doc_list), "documents": doc_list, } return Response( content=json.dumps(data, ensure_ascii=False, indent=2), media_type="application/json", ) @router.get("/{token}") def kb_index_html( token: str, request: Request, page: int = Query(1, ge=1), db: Session = Depends(get_db), ) -> HTMLResponse: """知识库首页(HTML)。""" _rate_limit(request, token) svc = KbPublicService(db) kb = svc.get_kb_by_token(token) docs, total = svc.list_documents(kb, page=page, page_size=50) doc_rows = "" for doc in docs: doc_url = f"/k/{token}/doc/{decrypt_token(doc.doc_token_encrypted) if doc.doc_token_encrypted else ''}" keywords_html = f'关键词:{doc.keywords}' if doc.keywords else "" summary_html = f'
{doc.content_summary or ""}
' if doc.content_summary else "" status_badge = "" if doc.status != "READY": status_badge = f' [{doc.status}]' doc_rows += f"""第 {page} / {total_pages} 页
' html = f"""" + kb.description + "
" if kb.description else ""}暂无文档。
"} {pagination} """ return HTMLResponse(content=html) # --- 单文档访问 --- @router.get("/{token}/doc/{doc_token}") def doc_page_html( token: str, doc_token: str, request: Request, 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) markdown_content = svc.get_document_markdown(doc) # Markdown → HTML(简单转换) html_content = _markdown_to_html(markdown_content) keywords_html = f"关键词:{doc.keywords}
" if doc.keywords else "" html = f"""" + (item.get('description') or '') + "
"}共找到 {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 转换(安全处理)。""" import re # 转义 HTML 特殊字符 html = markdown.replace("&", "&").replace("<", "<").replace(">", ">") # 标题 html = re.sub(r"^#### (.+)$", r"{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