{kb.name}
+ {"" + kb.description + "
" if kb.description else ""} + + {all_content} + + +diff --git a/backend/app/public/routes.py b/backend/app/public/routes.py index 27edd67..3913541 100644 --- a/backend/app/public/routes.py +++ b/backend/app/public/routes.py @@ -1,5 +1,11 @@ """公共 AI 页面路由(/k/**)。 +核心设计: +- 主页 /k/{token} = 目录索引(TOC),显示所有目录及其文档数量 +- 目录页 /k/{token}/category/{path} = 该目录下的文档列表 +- 文档页 /k/{token}/doc/{doc_token} = 单个文档内容 +- AI 先看目录索引 → 根据用户描述选择目录 → 进入该目录看文档 + 规则: - 零 JS、零 Cookie、零登录、SSR 输出、标准 HTML - @@ -9,6 +15,7 @@ """ import json +import re from fastapi import APIRouter, Depends, Query, Request, Response from fastapi.responses import HTMLResponse, PlainTextResponse @@ -27,7 +34,7 @@ router = APIRouter(prefix="/k", tags=["public"]) 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,不阻塞响应)。""" + """记录访问日志(best-effort)。""" try: ua = request.headers.get("user-agent", "") AccessLogService(db).record( @@ -38,11 +45,10 @@ def _log_access(db: Session, kb_id: str, path: str, request: Request, doc_id: st request_type=req_type, ) except Exception: - pass # 日志失败不影响响应 + 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() @@ -56,7 +62,87 @@ 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") @@ -65,33 +151,39 @@ def kb_index_markdown( request: Request, db: Session = Depends(get_db), ) -> PlainTextResponse: - """知识库首页(Markdown)。""" + """知识库目录索引(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") - docs, _ = svc.list_documents(kb) + + 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("请通过以下目录链接访问对应内容:") 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("") + 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") @@ -102,48 +194,272 @@ def kb_index_text( 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") - docs, _ = svc.list_documents(kb) + + 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("目录结构:") 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("") + 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}.json") -def kb_index_json( +@router.get("/{token}") +def kb_index_html( token: str, request: Request, db: Session = Depends(get_db), -) -> Response: - """知识库首页(JSON)。""" +) -> 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}.json", request, req_type="json") + _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}
{summary}
\n' + if keywords: + content_html += f'关键词:{keywords}
\n' + content_html += f'" + kb.description + "
" if kb.description else ""} + + {all_content} + + +关键词:{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""" -