--- name: knowledge-mesh description: 知识网格 — 跨平台知识搜索聚合器，统一搜索 GitHub、Stack Overflow、Discord、Confluence、Notion、Slack、百度、Obsidian，支持自学习排序 version: 1.1.0 metadata: openclaw: optional_env: - KM_GITHUB_TOKEN - KM_STACKOVERFLOW_KEY - KM_DISCORD_BOT_TOKEN - KM_CONFLUENCE_URL - KM_CONFLUENCE_TOKEN - KM_NOTION_TOKEN - KM_SLACK_TOKEN - KM_BAIDU_API_KEY - KM_OBSIDIAN_VAULT_PATH - KM_SUBSCRIPTION_TIER --- # 知识网格（knowledge-mesh）你是一个专业的跨平台知识搜索助手 Agent。你的职责是帮助用户在多个知识平台上进行统一搜索、结果聚合、趋势分析和知识管理。你始终使用中文与用户沟通。 ## 环境变量说明 | 变量 | 必需 | 说明 | |------|------|------| | `KM_GITHUB_TOKEN` | 否 | GitHub Personal Access Token，用于搜索 Issues/Discussions | | `KM_STACKOVERFLOW_KEY` | 否 | Stack Exchange API Key，提高速率限制 | | `KM_DISCORD_BOT_TOKEN` | 否 | Discord Bot Token，搜索频道消息 | | `KM_DISCORD_CHANNEL_ID` | 否 | Discord 目标频道 ID | | `KM_CONFLUENCE_URL` | 否 | Confluence 实例 URL（如 https://your-domain.atlassian.net） | | `KM_CONFLUENCE_TOKEN` | 否 | Confluence API Token | | `KM_NOTION_TOKEN` | 否 | Notion Integration Token | | `KM_SLACK_TOKEN` | 否 | Slack Bot User OAuth Token | | `KM_SUBSCRIPTION_TIER` | 否 | 订阅等级，默认 `free`，可选 `paid` | | `KM_BAIDU_API_KEY` | 否 | 百度搜索 API Key，启用中文搜索增强 | | `KM_OBSIDIAN_VAULT_PATH` | 否 | Obsidian vault 目录路径，启用本地笔记搜索 | | `KM_DATA_DIR` | 否 | 数据存储目录，默认 `~/.openclaw-bdi/knowledge-mesh/` | 启动时，你应检查至少一个知识源的凭据已配置。若全部缺失，引导用户进入「知识源配置流程」。 --- ## 流程一：跨平台知识搜索当用户说"搜索"、"查找"、"搜一下"或提出技术问题时，执行以下步骤： ### 步骤 1：解析查询意图分析用户的自然语言问题，提取： - 核心关键词 - 目标平台偏好（若有） - 时间范围限制（若有） - 结果数量期望 ### 步骤 2：执行搜索 ```bash python3 scripts/source_searcher.py --action search --data '{"query":"<关键词>","max_results":20}' ``` 若用户指定了平台： ```bash python3 scripts/source_searcher.py --action search-source --data '{"query":"<关键词>","source":"github"}' ``` ### 步骤 3：排序与去重 ```bash python3 scripts/result_ranker.py --action rank --data '{"query":"<关键词>","results":[...]}' python3 scripts/result_ranker.py --action dedup --data '{"results":[...]}' ``` ### 步骤 4：展示结果将搜索结果以清晰的列表形式展示，每条结果包含： - 来源标签（如 [GitHub]、[Stack Overflow]） - 标题（高亮匹配关键词） - 链接 - 摘要片段 - 作者和日期 - 相关度评分付费用户额外提供知识合成摘要。 --- ## 流程二：本地知识索引当用户说"索引文件"、"建立索引"、"搜索本地"时（仅付费版）： ### 步骤 1：索引构建 ```bash python3 scripts/index_builder.py --action index --data '{"paths":["./docs","./src"],"patterns":["*.md","*.txt","*.py"]}' ``` ### 步骤 2：本地搜索 ```bash python3 scripts/index_builder.py --action search-local --data '{"query":"<关键词>"}' ``` ### 步骤 3：索引管理 ```bash # 查看已索引文档 python3 scripts/index_builder.py --action list-indexed # 重建索引 python3 scripts/index_builder.py --action rebuild # 删除文档索引 python3 scripts/index_builder.py --action delete --data '{"doc_id":"DOC..."}' ``` --- ## 流程三：主题监控当用户说"监控"、"订阅主题"、"关注话题"时（仅付费版）： ### 步骤 1：创建监控 ```bash python3 scripts/monitor_manager.py --action add --data '{"keywords":["fastapi","async"],"sources":["github","stackoverflow"]}' ``` ### 步骤 2：检查更新 ```bash # 检查单个监控 python3 scripts/monitor_manager.py --action check --data '{"id":"MON..."}' # 检查所有监控 python3 scripts/monitor_manager.py --action check --data '{"id":"all"}' ``` ### 步骤 3：生成摘要 ```bash # 日报 python3 scripts/monitor_manager.py --action digest --data '{"period":"daily"}' # 周报 python3 scripts/monitor_manager.py --action digest --data '{"period":"weekly"}' ``` --- ## 流程四：报告导出当用户说"导出"、"生成报告"、"保存结果"时： ### Markdown 导出 ```bash python3 scripts/report_exporter.py --action export-markdown --data '{"query":"...","results":[...],"file_path":"output/report.md"}' ``` ### CSV 导出 ```bash python3 scripts/report_exporter.py --action export-csv --data '{"results":[...],"file_path":"output/results.csv"}' ``` ### 趋势分析（仅付费版） ```bash python3 scripts/report_exporter.py --action trending --data '{"results":[...]}' ``` ### 使用统计 ```bash python3 scripts/report_exporter.py --action stats ``` --- ## 流程五：自学习搜索引擎当用户说"反馈"、"评价结果"、"搜索建议"、"搜索统计"时： ### 步骤 1：记录反馈 ```bash # 记录结果评价 python3 scripts/learning_engine.py --action record-feedback --data '{"result_id":"SR...","source":"github","rating":"helpful"}' # 记录点击行为 python3 scripts/learning_engine.py --action record-click --data '{"result_id":"SR...","source":"stackoverflow"}' ``` ### 步骤 2：权重调整 ```bash # 根据反馈调整知识源权重 python3 scripts/learning_engine.py --action boost-weights ``` ### 步骤 3：获取建议 ```bash # 获取个性化搜索建议 python3 scripts/learning_engine.py --action suggest ``` ### 步骤 4：查看统计 ```bash # 查看搜索分析统计 python3 scripts/learning_engine.py --action stats ``` 搜索结果排序模块会自动加载学习权重进行排序调整。用户也可手动校准权重： ```bash python3 scripts/result_ranker.py --action calibrate ``` --- ## 流程六：Obsidian 知识库集成当用户说"连接 Obsidian"、"搜索笔记"、"索引笔记"时： ### 步骤 1：连接 Vault ```bash python3 scripts/obsidian_connector.py --action connect --data '{"vault_path":"/path/to/my/vault"}' ``` 或通过环境变量设置默认 vault 路径： ```bash export KM_OBSIDIAN_VAULT_PATH="/path/to/my/vault" ``` ### 步骤 2：构建索引 ```bash python3 scripts/obsidian_connector.py --action index --data '{"vault_path":"/path/to/my/vault"}' ``` ### 步骤 3：搜索笔记 ```bash python3 scripts/obsidian_connector.py --action search --data '{"query":"python 异步编程"}' ``` Obsidian 搜索支持以下 Obsidian 特性： - `[[wikilinks]]` 双向链接解析 - `#tags` 标签匹配 - YAML frontmatter 元数据 - Callout 块提取 - 反向链接图用于权威性评分 ### 步骤 4：管理笔记 ```bash # 查看已索引笔记 python3 scripts/obsidian_connector.py --action list-notes # 增量同步 python3 scripts/obsidian_connector.py --action sync ``` Obsidian 笔记也会出现在统一搜索结果中（通过 `source_searcher` 的 `search` 操作）。 --- ## 流程七：百度搜索当用户搜索中文内容或指定百度搜索时： ```bash # 指定百度搜索 python3 scripts/source_searcher.py --action search-source --data '{"query":"FastAPI 最佳实践","source":"baidu"}' ``` 百度搜索也会自动纳入统一搜索（当已配置 `KM_BAIDU_API_KEY` 时）。 --- ## 订阅校验逻辑 ### 读取订阅等级 ``` tier = env KM_SUBSCRIPTION_TIER，默认 "free" ``` ### 功能权限矩阵 | 功能 | 免费版（free） | 付费版（paid，¥129/月） | |------|---------------|----------------------| | 知识源数量 | 最多 5 个 | 最多 10 个 | | 支持知识源 | GitHub、Stack Overflow、百度、Obsidian | 全部 8 个平台 | | 每日搜索次数 | 10 次 | 不限 | | 单次最大结果数 | 20 条 | 100 条 | | 本地知识索引 | 不支持 | 支持 | | Obsidian 集成 | 支持 | 支持 | | 百度搜索 | 支持 | 支持 | | 自学习排序（基础） | 支持 | 支持 | | 自学习排序（高级分析） | 不支持 | 支持 | | 主题监控 | 不支持 | 支持 | | 知识合成 | 不支持 | 支持 | | Mermaid 趋势图表 | 不支持 | 支持 | | 报告导出 | Markdown/CSV | 全格式 + 趋势分析 | ### 校验失败时的行为当用户请求的功能超出当前订阅等级时： 1. 明确告知用户当前功能仅限付费版。 2. 简要说明付费版的优势。 3. 提供升级引导："如需升级至付费版（¥129/月），请联系管理员或访问订阅管理页面。" 4. 不要直接拒绝，而是提供免费版可用的替代方案（如果有的话）。 --- ## 参考文档在搜索和生成报告时，请参考以下文档： - **API 端点参考**：`references/api-endpoints.md` — 各平台 API 地址和认证方式。 - **搜索语法指南**：`references/search-syntax.md` — 搜索查询语法和示例。 --- ## 安全规范 1. **凭据保护**：所有 API Token 仅通过环境变量传递，绝不在对话中显示、记录或输出完整的 Token 值。 2. **请求安全**：所有 HTTP 请求使用 HTTPS，设置合理的超时时间。 3. **数据本地化**：搜索索引和监控数据存储在本地，不会上传到外部服务器。 4. **输入校验**：对用户输入进行转义处理，防止注入攻击。 5. **错误处理**：执行命令失败时，向用户展示友好的错误提示，不要暴露内部路径或系统信息。 --- ## 行为准则 1. 始终使用中文与用户沟通。 2. 搜索前先确认用户的查询意图，必要时追问以明确需求。 3. 搜索结果以结构化列表展示，标注来源、相关度、时间。 4. 主动提供搜索建议和相关关键词扩展。 5. 当搜索无结果时，给出可能的原因和改进建议。 6. 尊重订阅等级限制，在提示升级时保持友好，不要反复推销。 7. 对于代码类问题，优先推荐 Stack Overflow 和 GitHub 的高质量答案。 8. 对于团队知识类问题，优先推荐 Confluence 和 Notion 的内部文档。 FILE:assets/README.md # Knowledge Mesh / 知识网格 > Cross-platform knowledge search aggregator — unified search across GitHub, Stack Overflow, Discord, Confluence, Notion, and Slack > > 跨平台知识搜索聚合器 — 统一搜索 GitHub、Stack Overflow、Discord、Confluence、Notion、Slack --- ## Features / 功能亮点 - **Unified Search / 统一搜索** — Search across 8 platforms with one query. 一次查询搜索 8 个平台，结果统一排序展示。 - **TF-IDF Ranking / 智能排序** — Relevance + authority + recency scoring. TF-IDF 相关性评分 + 权威性 + 时间衰减综合排序。 - **Self-Learning / 自学习排序** — Search gets smarter with use. 搜索越用越精准，根据反馈自动优化排序权重。 - **Obsidian Integration / Obsidian 集成** — Search your local notes too. 本地笔记也能一起搜，支持 wikilinks、标签、frontmatter。 - **Baidu Search / 百度搜索** — Enhanced Chinese content search. 中文内容搜索增强，覆盖中文技术社区。 - **Deduplication / 智能去重** — Jaccard similarity dedup across sources. 基于 Jaccard 相似度跨平台去重。 - **Local Indexing / 本地索引** — Index local .md/.txt/.py files with full-text search. 索引本地文档，支持全文检索。 - **Topic Monitoring / 主题监控** — Set keyword alerts, get daily/weekly digests. 设置关键词监控，获取日报/周报摘要。 - **Knowledge Synthesis / 知识合成** — AI-powered summary of search results. 将搜索结果合成为结构化知识报告。 - **Mermaid Charts / 可视化图表** — Trend analysis with Mermaid pie/bar/line charts. 趋势分析生成 Mermaid 饼图/柱状图/折线图。 - **Export / 导出** — Markdown reports and CSV files. 支持导出 Markdown 报告和 CSV 文件。 --- ## Version Comparison / 版本对比 | Feature / 功能 | Free / 免费版 | Paid / 付费版 ¥129/月 | |----------------|:------------:|:-------------------:| | Knowledge sources / 知识源数量 | 5 | 10 | | Supported platforms / 支持平台 | GitHub + SO + Baidu + Obsidian | All 8 platforms / 全部 8 平台 | | Daily searches / 每日搜索次数 | 10 | Unlimited / 无限 | | Max results per search / 单次结果数 | 20 | 100 | | Self-learning ranking / 自学习排序 | Basic / 基础 | Advanced / 高级分析 | | Obsidian integration / Obsidian 集成 | Supported / 支持 | Supported / 支持 | | Baidu search / 百度搜索 | Supported / 支持 | Supported / 支持 | | Local indexing / 本地知识索引 | -- | Supported / 支持 | | Topic monitoring / 主题监控 | -- | Supported / 支持 | | Knowledge synthesis / 知识合成 | -- | Supported / 支持 | | Mermaid trend charts / 趋势图表 | -- | Supported / 支持 | | Export / 导出 | Markdown + CSV | Full / 全格式 + 趋势分析 | --- ## Quick Start / 快速开始 ### 1. Install / 安装 Search for `knowledge-mesh` in ClawHub, or use CLI: 在 ClawHub 中搜索 `knowledge-mesh`，或使用命令行： ```bash openclaw skill install knowledge-mesh ``` ### 2. Configure Sources / 配置知识源 Set environment variables for the platforms you want to search: 设置你要搜索的平台的环境变量： ```bash # GitHub (recommended / 推荐) export KM_GITHUB_TOKEN="ghp_your_token_here" # Stack Overflow (optional / 可选，提高速率限制) export KM_STACKOVERFLOW_KEY="your_key_here" # Discord (paid / 付费版) export KM_DISCORD_BOT_TOKEN="your_bot_token" export KM_DISCORD_CHANNEL_ID="channel_id" # Confluence (paid / 付费版) export KM_CONFLUENCE_URL="https://your-domain.atlassian.net" export KM_CONFLUENCE_TOKEN="your_token" # Notion (paid / 付费版) export KM_NOTION_TOKEN="ntn_your_token" # Slack (paid / 付费版) export KM_SLACK_TOKEN="xoxb-your-token" # Baidu Search / 百度搜索 (free / 免费) export KM_BAIDU_API_KEY="your_baidu_api_key" # Obsidian (free / 免费) export KM_OBSIDIAN_VAULT_PATH="/path/to/your/vault" ``` ### 3. Search / 搜索 ```bash # Unified search / 统一搜索 /knowledge-mesh search "python async best practices" # Search specific source / 搜索指定平台 /knowledge-mesh search-source github "fastapi websocket" # View configured sources / 查看已配置知识源 /knowledge-mesh list-sources ``` ### 4. Advanced Features / 高级功能 (Paid / 付费版) ```bash # Index local files / 索引本地文件 /knowledge-mesh index "./docs" "./src" # Set up monitoring / 设置主题监控 /knowledge-mesh monitor "kubernetes" "deployment" # Export report / 导出报告 /knowledge-mesh export markdown ``` --- ## Example / 使用示例 ``` 用户: 搜索 Python FastAPI WebSocket 最佳实践知识网格: 正在搜索 GitHub、Stack Overflow... 搜索结果（共 15 条）: 1. [Stack Overflow] FastAPI WebSocket connection best practices - 链接: https://stackoverflow.com/questions/... - 相关度: 0.92 - 摘要: For **FastAPI** **WebSocket** connections, it's recommended to... 2. [GitHub] tiangolo/fastapi#1234 - WebSocket documentation improvements - 链接: https://github.com/tiangolo/fastapi/issues/1234 - 相关度: 0.87 - 摘要: This PR improves the **WebSocket** section with **best practices**... 3. [Stack Overflow] How to handle WebSocket disconnections in FastAPI? - 链接: https://stackoverflow.com/questions/... - 相关度: 0.81 - 摘要: When dealing with **WebSocket** disconnections in **FastAPI**... ``` --- ## FAQ / 常见问题 ### Q1: Which platforms can I search for free? / 免费版可以搜索哪些平台？ Free tier supports GitHub and Stack Overflow. These two platforms cover the majority of technical Q&A and open-source project discussions. 免费版支持 GitHub 和 Stack Overflow。这两个平台覆盖了绝大多数技术问答和开源项目讨论。 ### Q2: Is my data uploaded to the cloud? / 数据会上传到云端吗？ No. All search requests go directly from your machine to each platform's API. Local index data is stored locally. No data passes through any third-party server. 不会。所有搜索请求从你的机器直接发送到各平台 API。本地索引数据存储在本地。没有数据经过任何第三方服务器。 ### Q3: Do I need API keys for all platforms? / 需要配置所有平台的 API Key 吗？ No. You only need to configure the platforms you want to search. Stack Overflow works without an API key (with rate limits). GitHub search also works without a token but with lower rate limits. 不需要。你只需配置你想搜索的平台。Stack Overflow 无需 API Key 即可使用（有速率限制）。GitHub 搜索无 Token 也可工作但速率限制更低。 ### Q4: How does deduplication work? / 去重是怎么实现的？ We use Jaccard similarity on tokenized title + snippet text. Results with similarity above 0.7 are considered duplicates, keeping the one with higher relevance score. 使用标题+摘要的分词集合计算 Jaccard 相似度。相似度超过 0.7 的结果被视为重复，保留相关度更高的结果。 ### Q5: Can I search in Chinese? / 可以用中文搜索吗？ Yes. The search query is passed directly to each platform's API. Chinese queries work well on GitHub, Confluence, Notion, and Slack. Stack Overflow results are primarily in English. 可以。搜索查询直接传递给各平台 API。中文查询在 GitHub、Confluence、Notion 和 Slack 上效果良好。Stack Overflow 结果以英文为主。 ### Q6: What file types can be locally indexed? / 本地索引支持哪些文件类型？ Supported: `.md`, `.txt`, `.py`, `.js`, `.ts`, `.java`, `.go`, `.rs`, `.rb`, `.sh`, `.yaml`, `.json`, `.html`, `.css`, `.sql` and more. 支持：`.md`、`.txt`、`.py`、`.js`、`.ts`、`.java`、`.go`、`.rs`、`.rb`、`.sh`、`.yaml`、`.json`、`.html`、`.css`、`.sql` 等。 --- ## Support / 技术支持 - **Docs / 文档**: See `references/` directory for API and syntax guides - **Issues / 问题反馈**: Submit issues on ClawHub skill page - **Community / 社区**: Join `#knowledge-mesh` channel on ClawHub community - **Email / 邮件**: skill-support@clawhub.dev --- *knowledge-mesh v1.1.0 | Compatible with OpenClaw 0.5+ / 兼容 OpenClaw 0.5+* FILE:scripts/result_ranker.py #!/usr/bin/env python3 """ knowledge-mesh 搜索结果排序与合成模块对跨平台搜索结果进行 TF-IDF 相关性评分、权威性评分、去重合并和智能摘要生成。用法: python3 result_ranker.py --action rank --data '{"query":"...", "results":[...]}' python3 result_ranker.py --action dedup --data '{"results":[...]}' python3 result_ranker.py --action synthesize --data '{"query":"...", "results":[...]}' python3 result_ranker.py --action summarize --data '{"results":[...]}' """ import math import re import sys from collections import Counter from datetime import datetime from typing import Any, Dict, List, Optional, Set, Tuple from utils import ( check_subscription, clean_html, days_ago, format_source_badge, highlight_match, load_input_data, now_iso, output_error, output_success, parse_common_args, require_paid_feature, truncate_text, SOURCE_DISPLAY_NAMES, ) # ============================================================ # 常量 # ============================================================ # 各知识源的可信度系数 SOURCE_RELIABILITY = { "github": 0.9, "stackoverflow": 0.95, "discord": 0.5, "confluence": 0.85, "notion": 0.7, "slack": 0.4, "baidu": 0.6, "obsidian": 0.8, } # 时间衰减半衰期（天） RECENCY_HALF_LIFE = 90 # Jaccard 相似度去重阈值 DEDUP_THRESHOLD = 0.7 # 停用词列表（常见中英文停用词） STOP_WORDS = { "the", "a", "an", "is", "are", "was", "were", "be", "been", "being", "have", "has", "had", "do", "does", "did", "will", "would", "could", "should", "may", "might", "can", "shall", "to", "of", "in", "for", "on", "with", "at", "by", "from", "as", "into", "through", "during", "before", "after", "above", "below", "between", "and", "but", "or", "nor", "not", "so", "if", "then", "than", "too", "very", "just", "about", "up", "out", "no", "it", "its", "this", "that", "these", "those", "i", "me", "my", "we", "our", "you", "your", "he", "she", "him", "her", "they", "them", "their", "what", "which", "who", "how", "when", "where", "why", "all", "each", "every", "both", "的", "了", "在", "是", "我", "有", "和", "就", "不", "人", "都", "一", "一个", "上", "也", "很", "到", "说", "要", "去", "你", "会", "着", "没有", "看", "好", "自己", "这", "他", "她", "吗", "什么", "怎么", "如何", "为什么", "哪个", "那个", } # ============================================================ # 文本处理 # ============================================================ def _tokenize(text: str) -> List[str]: """将文本分词为词语列表。使用简单的正则表达式分割，同时处理中英文。 Args: text: 原始文本。 Returns: 小写词语列表。 """ if not text: return [] # 英文按单词分割，中文按字符分割 tokens = re.findall(r"[a-zA-Z0-9_]+|[\u4e00-\u9fff]", text.lower()) return [t for t in tokens if t and t not in STOP_WORDS] def _word_set(text: str) -> Set[str]: """提取文本的词语集合（去停用词）。 Args: text: 原始文本。 Returns: 词语集合。 """ return set(_tokenize(text)) def _term_frequency(tokens: List[str]) -> Dict[str, float]: """计算词频（TF）。使用对数归一化 TF: 1 + log(count)。 Args: tokens: 词语列表。 Returns: 词语到 TF 值的映射。 """ if not tokens: return {} counts = Counter(tokens) tf = {} for term, count in counts.items(): tf[term] = 1 + math.log(count) return tf def _inverse_document_frequency( term: str, doc_count: int, doc_freq: int, ) -> float: """计算逆文档频率（IDF）。 IDF = log(N / (1 + df)) Args: term: 词语。 doc_count: 总文档数。 doc_freq: 包含该词语的文档数。 Returns: IDF 值。 """ if doc_count <= 0: return 0.0 return math.log(doc_count / (1 + doc_freq)) def _build_corpus_idf(documents: List[List[str]]) -> Dict[str, float]: """构建语料库的 IDF 字典。 Args: documents: 文档词语列表的列表。 Returns: 词语到 IDF 值的映射。 """ doc_count = len(documents) if doc_count == 0: return {} # 统计每个词出现在多少文档中 df = Counter() for doc_tokens in documents: unique_terms = set(doc_tokens) for term in unique_terms: df[term] += 1 idf = {} for term, freq in df.items(): idf[term] = _inverse_document_frequency(term, doc_count, freq) return idf # ============================================================ # 评分函数 # ============================================================ def _tfidf_relevance_score( query_tokens: List[str], doc_tokens: List[str], idf: Dict[str, float], ) -> float: """计算查询与文档的 TF-IDF 相关性分数。 Args: query_tokens: 查询词语列表。 doc_tokens: 文档词语列表。 idf: IDF 字典。 Returns: 相关性分数（0.0 ~ 1.0）。 """ if not query_tokens or not doc_tokens: return 0.0 doc_tf = _term_frequency(doc_tokens) score = 0.0 for q_term in query_tokens: tf_val = doc_tf.get(q_term, 0.0) idf_val = idf.get(q_term, 0.0) score += tf_val * idf_val # 归一化 max_possible = sum(idf.get(q, 0.0) for q in query_tokens) if max_possible > 0: score = score / max_possible return min(1.0, max(0.0, score)) def _get_learned_weight(source: str) -> float: """获取知识源的自学习权重。从 learning_engine 加载累积反馈计算出的权重，作为排序调整的乘数因子。 Args: source: 知识源名称。 Returns: 权重乘数（默认 1.0）。 """ try: from learning_engine import get_source_weights weights = get_source_weights() return weights.get(source, 1.0) except (ImportError, Exception): return 1.0 def _authority_score(result: Dict[str, Any]) -> float: """计算结果的权威性分数。综合考虑：原始分数、来源可信度、时间衰减。 Args: result: 搜索结果条目。 Returns: 权威性分数（0.0 ~ 1.0）。 """ # 原始分数（归一化到 0-1） raw_score = float(result.get("score", 0)) if raw_score > 1.0: # Stack Overflow 等平台的分数可能很大 raw_score = min(1.0, math.log(1 + raw_score) / 10) elif raw_score < 0: raw_score = 0.0 # 来源可信度系数（结合自学习权重） source = result.get("source", "") reliability = SOURCE_RELIABILITY.get(source, 0.5) # 应用自学习权重调整 learned_weight = _get_learned_weight(source) reliability = min(1.0, reliability * learned_weight) # 时间衰减：最近的内容权重更高 created = result.get("created_at", "") age_days = days_ago(created) if created else 365 recency_factor = math.pow(0.5, age_days / RECENCY_HALF_LIFE) # 综合评分 authority = (raw_score * 0.4 + reliability * 0.3 + recency_factor * 0.3) return min(1.0, max(0.0, authority)) def _combined_score( result: Dict[str, Any], query_tokens: List[str], idf: Dict[str, float], ) -> float: """计算综合排序分数。综合 TF-IDF 相关性和权威性。 Args: result: 搜索结果条目。 query_tokens: 查询词语列表。 idf: IDF 字典。 Returns: 综合分数（0.0 ~ 1.0）。 """ # 文档内容 = 标题 + 摘要 text = f"{result.get('title', '')} {result.get('snippet', '')}" doc_tokens = _tokenize(text) relevance = _tfidf_relevance_score(query_tokens, doc_tokens, idf) authority = _authority_score(result) # 相关性权重 60%，权威性权重 40% return relevance * 0.6 + authority * 0.4 # ============================================================ # Jaccard 去重 # ============================================================ def _jaccard_similarity(set_a: Set[str], set_b: Set[str]) -> float: """计算两个集合的 Jaccard 相似度。 J(A, B) = |A ∩ B| / |A ∪ B| Args: set_a: 集合 A。 set_b: 集合 B。 Returns: Jaccard 相似度（0.0 ~ 1.0）。 """ if not set_a and not set_b: return 1.0 if not set_a or not set_b: return 0.0 intersection = len(set_a & set_b) union = len(set_a | set_b) if union == 0: return 0.0 return intersection / union def _deduplicate_results(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """对搜索结果进行去重。使用 Jaccard 相似度比较标题和摘要的词集，相似度超过阈值的结果保留分数较高的。 Args: results: 搜索结果列表。 Returns: 去重后的结果列表。 """ if not results: return [] # 为每个结果计算词集 result_word_sets = [] for r in results: text = f"{r.get('title', '')} {r.get('snippet', '')}" result_word_sets.append(_word_set(text)) kept = [] kept_indices = [] for i, result in enumerate(results): is_duplicate = False for j in kept_indices: sim = _jaccard_similarity(result_word_sets[i], result_word_sets[j]) if sim >= DEDUP_THRESHOLD: is_duplicate = True break if not is_duplicate: kept.append(result) kept_indices.append(i) return kept # ============================================================ # 合成摘要（付费功能） # ============================================================ def _synthesize_results( query: str, results: List[Dict[str, Any]], max_items: int = 10, ) -> str: """将搜索结果合成为结构化 Markdown 摘要。 Args: query: 原始查询。 results: 排序后的搜索结果。 max_items: 纳入合成的最大结果数。 Returns: Markdown 格式的合成摘要。 """ top = results[:max_items] parts = [] parts.append(f"# 知识搜索合成报告\n") parts.append(f"**查询**: {query}") parts.append(f"**结果数**: {len(results)} 条（展示前 {len(top)} 条）") parts.append(f"**生成时间**: {now_iso()}\n") # 按来源分组 by_source = {} for r in top: src = r.get("source", "unknown") if src not in by_source: by_source[src] = [] by_source[src].append(r) parts.append("## 来源分布\n") for src, items in by_source.items(): badge = format_source_badge(src) parts.append(f"- {badge}: {len(items)} 条结果") parts.append("") # 关键发现 parts.append("## 关键发现\n") for idx, r in enumerate(top, 1): badge = format_source_badge(r.get("source", "")) title = r.get("title", "无标题") url = r.get("url", "") snippet = truncate_text(r.get("snippet", ""), 150) score = r.get("_combined_score", r.get("score", 0)) author = r.get("author", "") created = r.get("created_at", "") parts.append(f"### {idx}. {title}") parts.append(f"- **来源**: {badge}") if url: parts.append(f"- **链接**: {url}") if author: parts.append(f"- **作者**: {author}") if created: parts.append(f"- **日期**: {created}") parts.append(f"- **相关度**: {score:.2f}") parts.append(f"- **摘要**: {snippet}") parts.append("") # 标签汇总 all_tags = [] for r in top: all_tags.extend(r.get("tags", [])) if all_tags: tag_counts = Counter(all_tags) top_tags = tag_counts.most_common(10) parts.append("## 热门标签\n") for tag, count in top_tags: parts.append(f"- `{tag}` ({count})") parts.append("") # Mermaid 来源分布图 parts.append("## 来源分布图\n") parts.append("```mermaid") parts.append("pie title 搜索结果来源分布") for src, items in by_source.items(): display = SOURCE_DISPLAY_NAMES.get(src, src) parts.append(f' "{display}" : {len(items)}') parts.append("```\n") parts.append("---") parts.append("*由 knowledge-mesh 自动生成*") return "\n".join(parts) # ============================================================ # 操作实现 # ============================================================ def action_rank(data: Dict[str, Any]) -> None: """对搜索结果进行综合排序。 Args: data: 包含 query 和 results 的字典。 """ query = data.get("query", "") results = data.get("results", []) if not results: output_success({"query": query, "total": 0, "results": []}) return query_tokens = _tokenize(query) # 构建语料库 IDF documents = [] for r in results: text = f"{r.get('title', '')} {r.get('snippet', '')}" documents.append(_tokenize(text)) idf = _build_corpus_idf(documents) # 计算综合分数 for r in results: r["_combined_score"] = _combined_score(r, query_tokens, idf) # 排序 results.sort(key=lambda r: r.get("_combined_score", 0), reverse=True) output_success({ "query": query, "total": len(results), "results": results, }) def action_dedup(data: Dict[str, Any]) -> None: """对搜索结果进行去重。 Args: data: 包含 results 的字典。 """ results = data.get("results", []) original_count = len(results) deduped = _deduplicate_results(results) output_success({ "original_count": original_count, "deduped_count": len(deduped), "removed": original_count - len(deduped), "results": deduped, }) def action_synthesize(data: Dict[str, Any]) -> None: """合成搜索结果为结构化报告（付费功能）。 Args: data: 包含 query 和 results 的字典。 """ if not require_paid_feature("synthesis", "知识合成"): return query = data.get("query", "") results = data.get("results", []) if not results: output_error("无搜索结果可合成", code="NO_DATA") return # 先排序 query_tokens = _tokenize(query) documents = [] for r in results: text = f"{r.get('title', '')} {r.get('snippet', '')}" documents.append(_tokenize(text)) idf = _build_corpus_idf(documents) for r in results: r["_combined_score"] = _combined_score(r, query_tokens, idf) results.sort(key=lambda r: r.get("_combined_score", 0), reverse=True) # 去重 results = _deduplicate_results(results) # 生成合成报告 max_items = data.get("max_items", 10) report = _synthesize_results(query, results, max_items=max_items) output_success({ "query": query, "total": len(results), "report": report, }) def action_summarize(data: Dict[str, Any]) -> None: """生成搜索结果的简要统计摘要。 Args: data: 包含 results 的字典，可选 query。 """ results = data.get("results", []) query = data.get("query", "") total = len(results) # 来源统计 source_counts = Counter() for r in results: source_counts[r.get("source", "unknown")] += 1 # 时间分布 recent_7d = 0 recent_30d = 0 older = 0 for r in results: created = r.get("created_at", "") age = days_ago(created) if created else 999 if age <= 7: recent_7d += 1 elif age <= 30: recent_30d += 1 else: older += 1 # 热门标签 all_tags = [] for r in results: all_tags.extend(r.get("tags", [])) top_tags = Counter(all_tags).most_common(10) # 平均分数 scores = [r.get("score", 0) for r in results if r.get("score", 0) > 0] avg_score = sum(scores) / len(scores) if scores else 0.0 summary = { "query": query, "total_results": total, "source_distribution": dict(source_counts), "time_distribution": { "last_7_days": recent_7d, "last_30_days": recent_30d, "older": older, }, "top_tags": [{"tag": t, "count": c} for t, c in top_tags], "average_score": round(avg_score, 3), } output_success(summary) def action_record_feedback(data: Dict[str, Any]) -> None: """记录用户对搜索结果的反馈，并更新自学习权重。 Args: data: 包含 result_id、source、rating 的字典。 """ result_id = data.get("result_id", "").strip() source = data.get("source", "").strip() rating = data.get("rating", "").strip().lower() if not result_id: output_error("请提供搜索结果ID（result_id）", code="VALIDATION_ERROR") return if not source: output_error("请提供知识源名称（source）", code="VALIDATION_ERROR") return if rating not in {"relevant", "irrelevant", "helpful"}: output_error( f"无效的评级: {rating}，支持: relevant、irrelevant、helpful", code="VALIDATION_ERROR", ) return # 委托给 learning_engine 记录反馈 try: from learning_engine import record_feedback_data record_feedback_data(result_id, source, rating) output_success({ "message": f"已记录反馈: {source} 结果 {result_id} 评级为 {rating}", "result_id": result_id, "source": source, "rating": rating, }) except ImportError: output_error("自学习模块不可用", code="MODULE_ERROR") except Exception as e: output_error(f"记录反馈失败: {e}", code="FEEDBACK_ERROR") def action_calibrate(data: Optional[Dict[str, Any]] = None) -> None: """根据反馈历史重新校准最优知识源权重。从 learning_engine 触发权重重算并返回新权重。 """ try: from learning_engine import _load_learning_data, _compute_optimal_weights, _save_learning_data, DEFAULT_SOURCE_WEIGHTS learning_data = _load_learning_data() feedback = learning_data.get("feedback", []) if len(feedback) < 2: output_error( f"反馈数据不足（当前 {len(feedback)} 条，需要至少 2 条），无法校准", code="INSUFFICIENT_DATA", ) return old_weights = dict(learning_data.get("source_weights", DEFAULT_SOURCE_WEIGHTS)) new_weights = _compute_optimal_weights(feedback, old_weights) learning_data["source_weights"] = new_weights _save_learning_data(learning_data) changes = [] for source in set(list(old_weights.keys()) + list(new_weights.keys())): old_w = old_weights.get(source, 1.0) new_w = new_weights.get(source, 1.0) if abs(old_w - new_w) > 0.001: changes.append({ "source": source, "old_weight": old_w, "new_weight": new_w, }) output_success({ "message": f"权重校准完成，基于 {len(feedback)} 条反馈", "weights": new_weights, "changes": changes, }) except ImportError: output_error("自学习模块不可用", code="MODULE_ERROR") except Exception as e: output_error(f"校准失败: {e}", code="CALIBRATE_ERROR") # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("knowledge-mesh 搜索结果排序与合成") args = parser.parse_args() action = args.action.lower() try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "rank": lambda: action_rank(data or {}), "dedup": lambda: action_dedup(data or {}), "synthesize": lambda: action_synthesize(data or {}), "summarize": lambda: action_summarize(data or {}), "record-feedback": lambda: action_record_feedback(data or {}), "calibrate": lambda: action_calibrate(data), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(actions.keys()) output_error(f"未知操作: {action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/monitor_manager.py #!/usr/bin/env python3 """ knowledge-mesh 知识主题监控管理器（付费功能）支持设置关键词监控，定期检查各知识源的新内容，生成日报/周报摘要。用法: python3 monitor_manager.py --action add --data '{"keywords":["fastapi","async"],"sources":["github","stackoverflow"]}' python3 monitor_manager.py --action remove --data '{"id":"MON..."}' python3 monitor_manager.py --action list python3 monitor_manager.py --action check --data '{"id":"MON..."}' python3 monitor_manager.py --action digest --data '{"period":"daily"}' """ import json import os import sys from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from utils import ( SUPPORTED_SOURCES, SOURCE_DISPLAY_NAMES, check_subscription, generate_id, get_data_file, load_input_data, now_iso, output_error, output_success, parse_common_args, read_json_file, require_paid_feature, today_str, truncate_text, write_json_file, ) # ============================================================ # 常量 # ============================================================ # 监控配置文件 MONITORS_FILE = "monitors.json" # 监控结果缓存文件 MONITOR_RESULTS_FILE = "monitor_results.json" # 最大监控数量 MAX_MONITORS = 20 # 检查结果最大保留天数 MAX_RESULT_DAYS = 30 # ============================================================ # 数据访问 # ============================================================ def _get_monitors() -> List[Dict[str, Any]]: """读取所有监控配置。""" data = read_json_file(get_data_file(MONITORS_FILE)) if isinstance(data, list): return data return [] def _save_monitors(monitors: List[Dict[str, Any]]) -> None: """保存监控配置。""" write_json_file(get_data_file(MONITORS_FILE), monitors) def _find_monitor(monitors: List[Dict], monitor_id: str) -> Optional[Dict]: """根据 ID 查找监控。""" for m in monitors: if m.get("id") == monitor_id: return m return None def _get_results() -> Dict[str, List[Dict[str, Any]]]: """读取监控结果缓存。 Returns: {monitor_id: [results]} 映射。 """ data = read_json_file(get_data_file(MONITOR_RESULTS_FILE)) if isinstance(data, dict): return data return {} def _save_results(results: Dict[str, List[Dict[str, Any]]]) -> None: """保存监控结果缓存。""" write_json_file(get_data_file(MONITOR_RESULTS_FILE), results) def _cleanup_old_results(results: Dict[str, List[Dict[str, Any]]]) -> Dict[str, List[Dict[str, Any]]]: """清理超过保留期限的结果。 Args: results: 监控结果映射。 Returns: 清理后的结果映射。 """ cutoff = (datetime.now() - timedelta(days=MAX_RESULT_DAYS)).strftime("%Y-%m-%dT%H:%M:%S") cleaned = {} for mid, items in results.items(): kept = [r for r in items if r.get("checked_at", "") >= cutoff] if kept: cleaned[mid] = kept return cleaned # ============================================================ # 搜索辅助（简化版，避免循环导入） # ============================================================ def _simple_search(keywords: List[str], sources: List[str], since: str) -> List[Dict[str, Any]]: """执行简化搜索，用于监控检查。通过 source_searcher 模块执行搜索，过滤 since 之后的结果。 Args: keywords: 关键词列表。 sources: 知识源列表。 since: 起始时间（ISO 格式）。 Returns: 新内容列表。 """ # 动态导入避免模块级循环依赖 try: # 尝试导入 source_searcher 中的适配器 script_dir = os.path.dirname(os.path.abspath(__file__)) if script_dir not in sys.path: sys.path.insert(0, script_dir) from source_searcher import _SOURCE_ADAPTERS except ImportError: return [] query = " ".join(keywords) all_results = [] for source in sources: if source not in SUPPORTED_SOURCES: continue adapter = _SOURCE_ADAPTERS.get(source) if not adapter: continue try: results = adapter(query, max_results=20) # 过滤 since 之后的结果 for r in results: created = r.get("created_at", "") if created and created >= since: r["monitor_source"] = source all_results.append(r) except Exception: # 监控检查时静默忽略搜索错误 continue return all_results # ============================================================ # 摘要生成 # ============================================================ def _generate_digest_markdown( monitors: List[Dict[str, Any]], results: Dict[str, List[Dict[str, Any]]], period: str, ) -> str: """生成监控摘要 Markdown 报告。 Args: monitors: 监控配置列表。 results: 监控结果映射。 period: 摘要周期（daily/weekly）。 Returns: Markdown 格式的摘要报告。 """ period_label = "日报" if period == "daily" else "周报" today = today_str() # 确定时间范围 if period == "daily": since = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S") else: since = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%S") parts = [] parts.append(f"# 知识监控{period_label} — {today}\n") parts.append(f"**生成时间**: {now_iso()}") parts.append(f"**监控数量**: {len(monitors)} 个") parts.append(f"**统计周期**: {'过去 24 小时' if period == 'daily' else '过去 7 天'}\n") total_new = 0 for monitor in monitors: mid = monitor.get("id", "") keywords = monitor.get("keywords", []) monitor_sources = monitor.get("sources", []) keyword_str = "、".join(keywords) # 获取该监控的结果 monitor_results = results.get(mid, []) # 过滤时间范围内的结果 period_results = [ r for r in monitor_results if r.get("checked_at", "") >= since ] total_new += len(period_results) parts.append(f"## 监控: {keyword_str}\n") parts.append(f"- **关键词**: {keyword_str}") source_names = [SOURCE_DISPLAY_NAMES.get(s, s) for s in monitor_sources] parts.append(f"- **监控源**: {', '.join(source_names)}") parts.append(f"- **新内容**: {len(period_results)} 条\n") if period_results: for idx, r in enumerate(period_results[:10], 1): title = r.get("title", "无标题") url = r.get("url", "") source = r.get("source", "") badge = f"[{SOURCE_DISPLAY_NAMES.get(source, source)}]" snippet = truncate_text(r.get("snippet", ""), 100) parts.append(f"### {idx}. {badge} {title}") if url: parts.append(f"- 链接: {url}") parts.append(f"- 摘要: {snippet}") parts.append("") if len(period_results) > 10: parts.append(f"*... 以及其他 {len(period_results) - 10} 条结果*\n") else: parts.append("*暂无新内容*\n") # 汇总 parts.append("---\n") parts.append(f"**总计**: {len(monitors)} 个监控主题，{total_new} 条新内容\n") parts.append("---") parts.append("*由 knowledge-mesh 自动生成*") return "\n".join(parts) # ============================================================ # 操作实现 # ============================================================ def action_add(data: Dict[str, Any]) -> None: """添加新的主题监控。 Args: data: 包含 keywords（关键词列表）和可选 sources（知识源列表）的字典。 """ if not require_paid_feature("topic_monitor", "主题监控"): return keywords = data.get("keywords", []) if not keywords: output_error("请提供监控关键词列表（keywords）", code="VALIDATION_ERROR") return if isinstance(keywords, str): keywords = [keywords] sources = data.get("sources", ["github", "stackoverflow"]) if isinstance(sources, str): sources = [sources] # 校验知识源 for s in sources: if s not in SUPPORTED_SOURCES: valid = "、".join(SUPPORTED_SOURCES) output_error(f"不支持的知识源: {s}，支持: {valid}", code="INVALID_SOURCE") return monitors = _get_monitors() if len(monitors) >= MAX_MONITORS: output_error( f"已达监控数量上限（{MAX_MONITORS} 个），请先删除不需要的监控。", code="LIMIT_EXCEEDED", ) return now = now_iso() monitor = { "id": generate_id("MON"), "keywords": keywords, "sources": sources, "last_checked": now, "created_at": now, "active": True, } monitors.append(monitor) _save_monitors(monitors) output_success({ "message": f"监控已创建，关键词: {', '.join(keywords)}", "monitor": monitor, }) def action_remove(data: Dict[str, Any]) -> None: """删除指定的主题监控。 Args: data: 包含 id（监控 ID）的字典。 """ monitor_id = data.get("id", "").strip() if not monitor_id: output_error("请提供监控ID（id）", code="VALIDATION_ERROR") return monitors = _get_monitors() original_count = len(monitors) monitors = [m for m in monitors if m.get("id") != monitor_id] if len(monitors) == original_count: output_error(f"未找到监控: {monitor_id}", code="NOT_FOUND") return _save_monitors(monitors) # 同时删除相关结果 results = _get_results() if monitor_id in results: del results[monitor_id] _save_results(results) output_success({"message": f"监控 {monitor_id} 已删除"}) def action_list(data: Optional[Dict[str, Any]] = None) -> None: """列出所有主题监控。""" monitors = _get_monitors() monitor_list = [] for m in monitors: monitor_list.append({ "id": m.get("id", ""), "keywords": m.get("keywords", []), "sources": m.get("sources", []), "last_checked": m.get("last_checked", ""), "created_at": m.get("created_at", ""), "active": m.get("active", True), }) output_success({ "total": len(monitor_list), "monitors": monitor_list, }) def action_check(data: Dict[str, Any]) -> None: """检查指定监控的新内容。 Args: data: 包含 id（监控 ID）的字典。如果 id 为 "all" 则检查所有监控。 """ if not require_paid_feature("topic_monitor", "主题监控"): return monitor_id = data.get("id", "").strip() if not monitor_id: output_error("请提供监控ID（id）或 'all'", code="VALIDATION_ERROR") return monitors = _get_monitors() results = _get_results() check_list = [] if monitor_id == "all": check_list = monitors else: monitor = _find_monitor(monitors, monitor_id) if not monitor: output_error(f"未找到监控: {monitor_id}", code="NOT_FOUND") return check_list = [monitor] check_results = {} now = now_iso() for monitor in check_list: mid = monitor.get("id", "") keywords = monitor.get("keywords", []) sources = monitor.get("sources", []) last_checked = monitor.get("last_checked", "") # 搜索新内容 new_items = _simple_search(keywords, sources, last_checked) # 为每个结果添加检查时间 for item in new_items: item["checked_at"] = now # 追加到结果缓存 if mid not in results: results[mid] = [] results[mid].extend(new_items) # 更新最后检查时间 monitor["last_checked"] = now check_results[mid] = { "keywords": keywords, "new_count": len(new_items), "items": new_items[:10], } # 清理旧结果 results = _cleanup_old_results(results) # 保存 _save_monitors(monitors) _save_results(results) total_new = sum(cr["new_count"] for cr in check_results.values()) output_success({ "checked_monitors": len(check_list), "total_new_items": total_new, "details": check_results, }) def action_digest(data: Dict[str, Any]) -> None: """生成监控摘要报告。 Args: data: 包含 period（daily/weekly）的字典。 """ if not require_paid_feature("topic_monitor", "主题监控"): return period = data.get("period", "daily").strip().lower() if period not in ("daily", "weekly"): output_error("period 参数必须为 daily 或 weekly", code="VALIDATION_ERROR") return monitors = _get_monitors() results = _get_results() if not monitors: output_error("暂无监控主题，请先添加监控", code="NO_MONITORS") return report = _generate_digest_markdown(monitors, results, period) output_success({ "period": period, "monitor_count": len(monitors), "report": report, }) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("knowledge-mesh 主题监控管理") args = parser.parse_args() action = args.action.lower() try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "add": lambda: action_add(data or {}), "remove": lambda: action_remove(data or {}), "list": lambda: action_list(data), "check": lambda: action_check(data or {}), "digest": lambda: action_digest(data or {}), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(actions.keys()) output_error(f"未知操作: {action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/learning_engine.py #!/usr/bin/env python3 """ knowledge-mesh 自学习搜索引擎模块基于用户反馈和搜索行为自动调整搜索排序权重，提供主动建议和搜索分析统计。用法: python3 learning_engine.py --action record-feedback --data '{"result_id":"SR...","source":"github","rating":"helpful"}' python3 learning_engine.py --action record-click --data '{"result_id":"SR...","source":"github"}' python3 learning_engine.py --action record-query --data '{"query":"python async","sources":["github","stackoverflow"],"result_counts":{"github":5,"stackoverflow":8}}' python3 learning_engine.py --action boost-weights python3 learning_engine.py --action suggest python3 learning_engine.py --action stats """ import json import math import os import sys from collections import Counter from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple from utils import ( generate_id, get_data_file, load_input_data, now_iso, output_error, output_success, parse_common_args, read_json_file, write_json_file, ) # ============================================================ # 常量 # ============================================================ # 学习数据文件 LEARNING_DATA_FILE = "learning.json" # 反馈评级定义 VALID_RATINGS = {"relevant", "irrelevant", "helpful"} # 评级到分数的映射（用于权重计算） RATING_SCORES = { "helpful": 1.0, "relevant": 0.5, "irrelevant": -0.5, } # 默认知识源权重 DEFAULT_SOURCE_WEIGHTS = { "github": 1.0, "stackoverflow": 1.0, "discord": 1.0, "confluence": 1.0, "notion": 1.0, "slack": 1.0, "baidu": 1.0, "obsidian": 1.0, } # 权重调整范围 MIN_WEIGHT = 0.1 MAX_WEIGHT = 3.0 # 权重调整学习率 LEARNING_RATE = 0.1 # 建议生成所需的最小查询记录数 MIN_QUERIES_FOR_SUGGEST = 3 # 建议生成所需的最小反馈记录数 MIN_FEEDBACK_FOR_SUGGEST = 2 # 主题提取时的最小出现次数 MIN_TOPIC_COUNT = 2 # ============================================================ # 学习数据持久化 # ============================================================ def _get_learning_file() -> str: """获取学习数据文件路径。 Returns: 学习数据文件的绝对路径。 """ return get_data_file(LEARNING_DATA_FILE) def _load_learning_data() -> Dict[str, Any]: """加载学习数据。 Returns: 学习数据字典，包含 query_log, feedback, click_log, source_weights。 """ data = read_json_file(_get_learning_file()) if not isinstance(data, dict): data = {} # 确保必要字段存在 if "query_log" not in data: data["query_log"] = [] if "feedback" not in data: data["feedback"] = [] if "click_log" not in data: data["click_log"] = [] if "source_weights" not in data: data["source_weights"] = dict(DEFAULT_SOURCE_WEIGHTS) return data def _save_learning_data(data: Dict[str, Any]) -> None: """保存学习数据。 Args: data: 学习数据字典。 """ write_json_file(_get_learning_file(), data) # ============================================================ # 查询日志分析辅助函数 # ============================================================ def _extract_topics(query_log: List[Dict[str, Any]], top_n: int = 10) -> List[Tuple[str, int]]: """从查询日志中提取热门主题词。对所有查询进行简单分词，统计出现频率最高的词。 Args: query_log: 查询记录列表。 top_n: 返回前 N 个主题。 Returns: (主题词, 出现次数) 元组列表，按次数降序排列。 """ word_counter = Counter() # 常见停用词 stop_words = { "the", "a", "an", "is", "are", "in", "on", "for", "to", "of", "and", "or", "not", "with", "how", "what", "why", "when", "where", "的", "了", "在", "是", "我", "有", "和", "不", "怎么", "如何", } for entry in query_log: query = entry.get("query", "") if not query: continue # 简单分词：按空格和常见分隔符分割 import re tokens = re.findall(r"[a-zA-Z0-9_]+|[\u4e00-\u9fff]+", query.lower()) for token in tokens: if token not in stop_words and len(token) >= 2: word_counter[token] += 1 # 过滤出至少出现 MIN_TOPIC_COUNT 次的词 filtered = [(word, count) for word, count in word_counter.most_common(top_n * 2) if count >= MIN_TOPIC_COUNT] return filtered[:top_n] def _calculate_source_adoption_rate(feedback: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: """计算各知识源的结果采纳率。采纳率 = helpful 评价数 / 总评价数。 Args: feedback: 反馈记录列表。 Returns: {source: {total, helpful, relevant, irrelevant, adoption_rate}} 映射。 """ source_stats = {} for entry in feedback: source = entry.get("source", "unknown") rating = entry.get("rating", "") if source not in source_stats: source_stats[source] = { "total": 0, "helpful": 0, "relevant": 0, "irrelevant": 0, } source_stats[source]["total"] += 1 if rating in source_stats[source]: source_stats[source][rating] += 1 # 计算采纳率 for source, stats in source_stats.items(): total = stats["total"] if total > 0: stats["adoption_rate"] = round(stats["helpful"] / total, 4) else: stats["adoption_rate"] = 0.0 return source_stats def _get_recent_queries(query_log: List[Dict[str, Any]], days: int = 30) -> List[Dict[str, Any]]: """获取最近 N 天的查询记录。 Args: query_log: 查询记录列表。 days: 天数范围。 Returns: 最近 N 天内的查询记录列表。 """ cutoff = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S") recent = [] for entry in query_log: ts = entry.get("timestamp", "") if ts >= cutoff: recent.append(entry) return recent def _compute_optimal_weights(feedback: List[Dict[str, Any]], current_weights: Dict[str, float]) -> Dict[str, float]: """根据反馈历史计算最优权重。评分高的知识源权重提升，评分低的权重降低。 Args: feedback: 反馈记录列表。 current_weights: 当前权重。 Returns: 更新后的权重字典。 """ new_weights = dict(current_weights) # 统计各源的反馈分数 source_scores = {} source_counts = {} for entry in feedback: source = entry.get("source", "") rating = entry.get("rating", "") if not source or rating not in RATING_SCORES: continue if source not in source_scores: source_scores[source] = 0.0 source_counts[source] = 0 source_scores[source] += RATING_SCORES[rating] source_counts[source] += 1 # 计算平均分并调整权重 for source, total_score in source_scores.items(): count = source_counts.get(source, 1) avg_score = total_score / max(count, 1) # 使用学习率渐进式调整 base_weight = current_weights.get(source, 1.0) adjustment = LEARNING_RATE * avg_score new_weight = base_weight + adjustment # 限制在合理范围内 new_weight = max(MIN_WEIGHT, min(MAX_WEIGHT, new_weight)) new_weights[source] = round(new_weight, 4) return new_weights # ============================================================ # 操作实现 # ============================================================ def action_record_feedback(data: Dict[str, Any]) -> None: """记录用户对搜索结果的反馈。反馈类型包括：relevant（相关）、irrelevant（不相关）、helpful（有帮助）。 Args: data: 包含 result_id、source、rating 的字典。 """ result_id = data.get("result_id", "").strip() source = data.get("source", "").strip() rating = data.get("rating", "").strip().lower() if not result_id: output_error("请提供搜索结果ID（result_id）", code="VALIDATION_ERROR") return if not source: output_error("请提供知识源名称（source）", code="VALIDATION_ERROR") return if rating not in VALID_RATINGS: valid = "、".join(VALID_RATINGS) output_error(f"无效的评级: {rating}，支持: {valid}", code="VALIDATION_ERROR") return learning_data = _load_learning_data() feedback_entry = { "id": generate_id("FB"), "result_id": result_id, "source": source, "rating": rating, "timestamp": now_iso(), } learning_data["feedback"].append(feedback_entry) _save_learning_data(learning_data) output_success({ "message": f"已记录反馈: {source} 结果 {result_id} 评级为 {rating}", "feedback_id": feedback_entry["id"], "total_feedback": len(learning_data["feedback"]), }) def action_record_click(data: Dict[str, Any]) -> None: """记录用户点击/使用搜索结果的行为（隐式相关性信号）。 Args: data: 包含 result_id、source 的字典，可选 query。 """ result_id = data.get("result_id", "").strip() source = data.get("source", "").strip() query = data.get("query", "").strip() if not result_id: output_error("请提供搜索结果ID（result_id）", code="VALIDATION_ERROR") return if not source: output_error("请提供知识源名称（source）", code="VALIDATION_ERROR") return learning_data = _load_learning_data() click_entry = { "id": generate_id("CK"), "result_id": result_id, "source": source, "query": query, "timestamp": now_iso(), } learning_data["click_log"].append(click_entry) _save_learning_data(learning_data) output_success({ "message": f"已记录点击: {source} 结果 {result_id}", "click_id": click_entry["id"], "total_clicks": len(learning_data["click_log"]), }) def action_record_query(data: Dict[str, Any]) -> None: """记录搜索查询及各知识源的结果数量。用于分析搜索行为和知识源覆盖率。 Args: data: 包含 query、sources、result_counts 的字典。 """ query = data.get("query", "").strip() sources = data.get("sources", []) result_counts = data.get("result_counts", {}) if not query: output_error("请提供搜索查询（query）", code="VALIDATION_ERROR") return learning_data = _load_learning_data() query_entry = { "id": generate_id("QR"), "query": query, "sources": sources, "result_counts": result_counts, "total_results": sum(result_counts.values()) if result_counts else 0, "timestamp": now_iso(), } learning_data["query_log"].append(query_entry) _save_learning_data(learning_data) output_success({ "message": f"已记录查询: '{query}'，共 {query_entry['total_results']} 条结果", "query_id": query_entry["id"], "total_queries": len(learning_data["query_log"]), }) def action_boost_weights(data: Optional[Dict[str, Any]] = None) -> None: """根据累积反馈调整知识源可靠性权重。 helpful 评价多的知识源权重提升，irrelevant 评价多的权重降低。 """ learning_data = _load_learning_data() feedback = learning_data.get("feedback", []) if len(feedback) < MIN_FEEDBACK_FOR_SUGGEST: output_error( f"反馈数据不足（当前 {len(feedback)} 条，需要至少 {MIN_FEEDBACK_FOR_SUGGEST} 条），无法调整权重", code="INSUFFICIENT_DATA", ) return old_weights = dict(learning_data.get("source_weights", DEFAULT_SOURCE_WEIGHTS)) # 计算新权重 new_weights = _compute_optimal_weights(feedback, old_weights) learning_data["source_weights"] = new_weights # 记录权重变更历史 changes = [] for source in set(list(old_weights.keys()) + list(new_weights.keys())): old_w = old_weights.get(source, 1.0) new_w = new_weights.get(source, 1.0) if abs(old_w - new_w) > 0.001: direction = "提升" if new_w > old_w else "降低" changes.append({ "source": source, "old_weight": old_w, "new_weight": new_w, "direction": direction, }) _save_learning_data(learning_data) output_success({ "message": f"权重调整完成，基于 {len(feedback)} 条反馈", "weights": new_weights, "changes": changes, "feedback_count": len(feedback), }) def action_suggest(data: Optional[Dict[str, Any]] = None) -> None: """基于搜索历史和反馈生成主动建议。建议类型包括： - 高频主题推荐关注 - 高采纳率知识源推荐 - 搜索习惯优化建议 """ learning_data = _load_learning_data() query_log = learning_data.get("query_log", []) feedback = learning_data.get("feedback", []) click_log = learning_data.get("click_log", []) source_weights = learning_data.get("source_weights", DEFAULT_SOURCE_WEIGHTS) suggestions = [] # 建议1: 高频主题推荐 if len(query_log) >= MIN_QUERIES_FOR_SUGGEST: topics = _extract_topics(query_log, top_n=5) if topics: top_topic = topics[0] topic_name = top_topic[0] topic_count = top_topic[1] # 根据主题内容给出不同建议 suggestion_text = f"您经常搜索 {topic_name} 相关内容（共 {topic_count} 次），" # 检查是否为技术框架/工具 tech_suggestions = { "react": "建议关注 React GitHub Discussions 和官方博客", "vue": "建议关注 Vue.js RFC 和 GitHub Discussions", "python": "建议关注 Python PEP 提案和 PyPI 新包发布", "fastapi": "建议关注 FastAPI GitHub Releases 和文档更新", "kubernetes": "建议关注 Kubernetes Enhancement Proposals (KEPs)", "docker": "建议关注 Docker 官方博客和 GitHub Releases", "typescript": "建议关注 TypeScript GitHub Discussions", "golang": "建议关注 Go 官方博客和 Release Notes", "rust": "建议关注 Rust RFC 和 This Week in Rust", "java": "建议关注 OpenJDK 和 Spring 官方博客", } topic_lower = topic_name.lower() if topic_lower in tech_suggestions: suggestion_text += tech_suggestions[topic_lower] else: suggestion_text += f"建议在搜索时尝试添加更具体的子主题以获得更精确的结果" suggestions.append({ "type": "topic_recommendation", "text": suggestion_text, "topic": topic_name, "frequency": topic_count, }) # 建议2: 高采纳率知识源推荐 if len(feedback) >= MIN_FEEDBACK_FOR_SUGGEST: adoption_stats = _calculate_source_adoption_rate(feedback) # 找到采纳率最高的知识源 best_source = None best_rate = 0.0 for source, stats in adoption_stats.items(): if stats["total"] >= 2 and stats["adoption_rate"] > best_rate: best_rate = stats["adoption_rate"] best_source = source if best_source and best_rate > 0.3: rate_pct = int(best_rate * 100) source_display = { "github": "GitHub", "stackoverflow": "Stack Overflow", "discord": "Discord", "confluence": "Confluence", "notion": "Notion", "slack": "Slack", "baidu": "百度搜索", "obsidian": "Obsidian", }.get(best_source, best_source) suggestions.append({ "type": "source_recommendation", "text": f"{source_display} 的结果采纳率最高({rate_pct}%)，建议优先查看", "source": best_source, "adoption_rate": best_rate, }) # 建议3: 搜索习惯优化 if len(query_log) >= MIN_QUERIES_FOR_SUGGEST: recent_queries = _get_recent_queries(query_log, days=7) if recent_queries: # 检查是否有重复查询 query_texts = [q.get("query", "") for q in recent_queries] query_counts = Counter(query_texts) repeated = [(q, c) for q, c in query_counts.items() if c >= 2] if repeated: most_repeated = max(repeated, key=lambda x: x[1]) suggestions.append({ "type": "search_optimization", "text": f"您最近 7 天内重复搜索了 '{most_repeated[0]}' {most_repeated[1]} 次，建议使用主题监控功能自动跟踪更新", "repeated_query": most_repeated[0], "repeat_count": most_repeated[1], }) # 检查平均结果数 result_totals = [q.get("total_results", 0) for q in recent_queries] if result_totals: avg_results = sum(result_totals) / len(result_totals) if avg_results < 3: suggestions.append({ "type": "search_optimization", "text": "您最近的搜索平均结果较少，建议尝试使用更通用的关键词或启用更多知识源", "avg_results": round(avg_results, 1), }) # 建议4: 点击行为分析 if len(click_log) >= 3: # 统计各源的点击分布 click_sources = Counter() for click in click_log: click_sources[click.get("source", "")] += 1 total_clicks = sum(click_sources.values()) if total_clicks > 0: # 找到最常点击的知识源 top_click_source, top_click_count = click_sources.most_common(1)[0] click_pct = int((top_click_count / total_clicks) * 100) if click_pct >= 60: source_display = { "github": "GitHub", "stackoverflow": "Stack Overflow", "discord": "Discord", "confluence": "Confluence", "notion": "Notion", "slack": "Slack", "baidu": "百度搜索", "obsidian": "Obsidian", }.get(top_click_source, top_click_source) suggestions.append({ "type": "usage_pattern", "text": f"您 {click_pct}% 的点击来自 {source_display}，该知识源的排序权重已自动提升", "source": top_click_source, "click_percentage": click_pct, }) if not suggestions: suggestions.append({ "type": "info", "text": "当前搜索数据不足，继续使用后将为您生成个性化建议", }) output_success({ "suggestions": suggestions, "total_suggestions": len(suggestions), "data_summary": { "total_queries": len(query_log), "total_feedback": len(feedback), "total_clicks": len(click_log), }, }) def action_stats(data: Optional[Dict[str, Any]] = None) -> None: """搜索分析统计：最常搜索的主题、最佳表现知识源、平均结果质量。""" learning_data = _load_learning_data() query_log = learning_data.get("query_log", []) feedback = learning_data.get("feedback", []) click_log = learning_data.get("click_log", []) source_weights = learning_data.get("source_weights", DEFAULT_SOURCE_WEIGHTS) # 基本统计 total_queries = len(query_log) total_feedback = len(feedback) total_clicks = len(click_log) # 热门主题 topics = _extract_topics(query_log, top_n=10) top_topics = [{"topic": t, "count": c} for t, c in topics] # 知识源结果统计 source_result_stats = {} for entry in query_log: result_counts = entry.get("result_counts", {}) for source, count in result_counts.items(): if source not in source_result_stats: source_result_stats[source] = {"total_results": 0, "query_count": 0} source_result_stats[source]["total_results"] += count source_result_stats[source]["query_count"] += 1 # 计算各源平均结果数 for source, stats in source_result_stats.items(): qc = stats["query_count"] stats["avg_results"] = round(stats["total_results"] / max(qc, 1), 1) # 知识源采纳率 adoption_stats = _calculate_source_adoption_rate(feedback) # 最佳表现知识源 best_source = None best_score = -1.0 for source, stats in adoption_stats.items(): score = stats["adoption_rate"] * stats["total"] # 综合采纳率和样本量 if score > best_score: best_score = score best_source = source # 平均结果质量（基于反馈） quality_scores = [] for entry in feedback: rating = entry.get("rating", "") if rating in RATING_SCORES: quality_scores.append(RATING_SCORES[rating]) avg_quality = round(sum(quality_scores) / max(len(quality_scores), 1), 3) # 时间分布统计 daily_queries = Counter() for entry in query_log: ts = entry.get("timestamp", "") if ts: day = ts[:10] # 截取日期部分 YYYY-MM-DD daily_queries[day] += 1 # 最近7天统计 recent_daily = [] today = datetime.utcnow().date() for i in range(7): day = (today - timedelta(days=i)).strftime("%Y-%m-%d") recent_daily.append({ "date": day, "queries": daily_queries.get(day, 0), }) output_success({ "overview": { "total_queries": total_queries, "total_feedback": total_feedback, "total_clicks": total_clicks, "avg_result_quality": avg_quality, }, "top_topics": top_topics, "source_performance": { "result_stats": source_result_stats, "adoption_stats": adoption_stats, "best_source": best_source, }, "current_weights": source_weights, "recent_activity": recent_daily, }) # ============================================================ # 公开 API（供其他模块调用） # ============================================================ def get_source_weights() -> Dict[str, float]: """获取当前知识源权重。供 result_ranker 等模块调用，用于排序调整。 Returns: 知识源权重字典。 """ learning_data = _load_learning_data() return learning_data.get("source_weights", dict(DEFAULT_SOURCE_WEIGHTS)) def record_query_data(query: str, sources: List[str], result_counts: Dict[str, int]) -> None: """记录查询数据（供 source_searcher 调用）。 Args: query: 搜索查询。 sources: 搜索的知识源列表。 result_counts: 各知识源的结果数量。 """ learning_data = _load_learning_data() query_entry = { "id": generate_id("QR"), "query": query, "sources": sources, "result_counts": result_counts, "total_results": sum(result_counts.values()) if result_counts else 0, "timestamp": now_iso(), } learning_data["query_log"].append(query_entry) _save_learning_data(learning_data) def record_feedback_data(result_id: str, source: str, rating: str) -> None: """记录反馈数据（供 result_ranker 调用）。 Args: result_id: 搜索结果 ID。 source: 知识源名称。 rating: 评级（relevant/irrelevant/helpful）。 """ if rating not in VALID_RATINGS: return learning_data = _load_learning_data() feedback_entry = { "id": generate_id("FB"), "result_id": result_id, "source": source, "rating": rating, "timestamp": now_iso(), } learning_data["feedback"].append(feedback_entry) _save_learning_data(learning_data) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("knowledge-mesh 自学习搜索引擎") args = parser.parse_args() action = args.action.lower() try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "record-feedback": lambda: action_record_feedback(data or {}), "record-click": lambda: action_record_click(data or {}), "record-query": lambda: action_record_query(data or {}), "boost-weights": lambda: action_boost_weights(data), "suggest": lambda: action_suggest(data), "stats": lambda: action_stats(data), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(actions.keys()) output_error(f"未知操作: {action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/utils.py #!/usr/bin/env python3 """ knowledge-mesh 共享工具模块提供数据目录管理、订阅校验、参数解析、数据格式化等通用功能。统一搜索 GitHub、Stack Overflow、Discord、Confluence、Notion、Slack 等知识源。 """ import argparse import hashlib import json import os import re import sys from datetime import datetime, timedelta from typing import Any, Dict, List, Optional # ============================================================ # 常量定义 # ============================================================ DEFAULT_DATA_DIR = os.path.join(os.path.expanduser("~"), ".openclaw-bdi", "knowledge-mesh") # 支持的知识源列表 SUPPORTED_SOURCES = [ "github", "stackoverflow", "discord", "confluence", "notion", "slack", "baidu", "obsidian", ] # 各知识源显示名称 SOURCE_DISPLAY_NAMES = { "github": "GitHub", "stackoverflow": "Stack Overflow", "discord": "Discord", "confluence": "Confluence", "notion": "Notion", "slack": "Slack", "baidu": "百度搜索", "obsidian": "Obsidian", } # 各知识源对应的环境变量 SOURCE_ENV_KEYS = { "github": "KM_GITHUB_TOKEN", "stackoverflow": "KM_STACKOVERFLOW_KEY", "discord": "KM_DISCORD_BOT_TOKEN", "confluence": "KM_CONFLUENCE_TOKEN", "notion": "KM_NOTION_TOKEN", "slack": "KM_SLACK_TOKEN", "baidu": "KM_BAIDU_API_KEY", "obsidian": "KM_OBSIDIAN_VAULT_PATH", } # ============================================================ # 数据目录管理 # ============================================================ def get_data_dir() -> str: """获取数据存储目录路径。优先读取环境变量 KM_DATA_DIR，若未设置则使用默认路径 ~/.openclaw-bdi/knowledge-mesh/。自动创建目录（若不存在）。 Returns: 数据目录的绝对路径。 """ data_dir = os.environ.get("KM_DATA_DIR", DEFAULT_DATA_DIR) os.makedirs(data_dir, exist_ok=True) return data_dir def get_data_file(filename: str) -> str: """获取数据文件的完整路径。 Args: filename: 文件名（如 "index_data.json"）。 Returns: 数据文件的绝对路径。 """ return os.path.join(get_data_dir(), filename) # ============================================================ # JSON 输入输出 # ============================================================ def read_json_file(filepath: str) -> Any: """读取 JSON 文件并返回解析后的数据。 Args: filepath: JSON 文件路径。 Returns: 解析后的数据对象。若文件不存在，返回空列表。 """ if not os.path.exists(filepath): return [] try: with open(filepath, "r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, IOError): return [] def write_json_file(filepath: str, data: Any) -> None: """将数据写入 JSON 文件。 Args: filepath: 目标文件路径。 data: 待写入的数据。 """ os.makedirs(os.path.dirname(filepath), exist_ok=True) with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2, default=str) def output_json(data: Any) -> None: """将数据以 JSON 格式输出到标准输出。 Args: data: 待输出的数据。 """ print(json.dumps(data, ensure_ascii=False, default=str)) def output_error(message: str, code: str = "ERROR") -> None: """输出标准错误响应到标准输出。 Args: message: 错误描述信息。 code: 错误代码，默认为 "ERROR"。 """ result = { "success": False, "error": { "code": code, "message": message, }, } output_json(result) def output_success(data: Any) -> None: """输出标准成功响应到标准输出。 Args: data: 成功时返回的数据负载。 """ result = { "success": True, "data": data, } output_json(result) # ============================================================ # 命令行参数解析 # ============================================================ def parse_common_args(description: str = "knowledge-mesh 知识搜索工具") -> argparse.ArgumentParser: """创建通用命令行参数解析器。 Args: description: 工具描述文本。 Returns: 配置好通用参数的 ArgumentParser 实例。 """ parser = argparse.ArgumentParser(description=description) parser.add_argument( "--action", required=True, help="操作类型", ) parser.add_argument( "--data", default=None, help="JSON 格式的数据字符串", ) parser.add_argument( "--data-file", default=None, help="JSON 数据文件路径", ) return parser def load_input_data(args: argparse.Namespace) -> Optional[Dict[str, Any]]: """从命令行参数加载输入数据。优先使用 --data 参数，其次尝试 --data-file 参数。 Args: args: 解析后的命令行参数。 Returns: 解析后的字典数据，若无输入数据则返回 None。 Raises: ValueError: 当 JSON 解析失败或文件读取失败时抛出。 """ if args.data: try: data = json.loads(args.data) if not isinstance(data, dict): raise ValueError(f"期望 JSON 对象，实际类型为 {type(data).__name__}") return data except json.JSONDecodeError as e: raise ValueError(f"JSON 解析失败: {e}") if args.data_file: if not os.path.exists(args.data_file): raise ValueError(f"数据文件不存在: {args.data_file}") try: with open(args.data_file, "r", encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict): raise ValueError(f"期望 JSON 对象，实际类型为 {type(data).__name__}") return data except json.JSONDecodeError as e: raise ValueError(f"数据文件 JSON 解析失败: {e}") return None # ============================================================ # 订阅校验 # ============================================================ _SUBSCRIPTION_TIERS: Dict[str, Dict[str, Any]] = { "free": { "tier": "free", "max_sources": 3, "daily_searches": 10, "max_results": 20, "features": [ "basic_search", "github_search", "stackoverflow_search", ], }, "paid": { "tier": "paid", "max_sources": 10, "daily_searches": -1, "max_results": 100, "features": [ "basic_search", "github_search", "stackoverflow_search", "discord_search", "confluence_search", "notion_search", "slack_search", "local_index", "topic_monitor", "synthesis", "mermaid_chart", "export", ], }, } def check_subscription(tier: Optional[str] = None) -> Dict[str, Any]: """检查当前订阅等级并返回对应的限制配置。优先使用传入的 tier 参数，否则读取环境变量 KM_SUBSCRIPTION_TIER。若都未设置，默认为 "free" 等级。 Args: tier: 订阅等级（"free" 或 "paid"），可选。 Returns: 包含订阅限制信息的字典。 Raises: ValueError: 当传入的等级无效时抛出。 """ if tier is None: tier = os.environ.get("KM_SUBSCRIPTION_TIER", "free") tier = tier.strip().lower() if tier not in _SUBSCRIPTION_TIERS: valid = ", ".join(_SUBSCRIPTION_TIERS.keys()) raise ValueError(f"无效的订阅等级: {tier!r}，有效等级: {valid}") return dict(_SUBSCRIPTION_TIERS[tier]) def require_paid_feature(feature_name: str, display_name: str) -> bool: """检查当前订阅是否支持指定功能。若不支持，输出升级提示并返回 False。 Args: feature_name: 功能内部名称。 display_name: 功能显示名称（用于提示信息）。 Returns: True 表示功能可用，False 表示不可用（已输出错误信息）。 """ sub = check_subscription() if feature_name not in sub["features"]: output_error( f"「{display_name}」为付费版功能。当前为免费版，请升级至付费版（¥129/月）以使用此功能。", code="SUBSCRIPTION_REQUIRED", ) return False return True # ============================================================ # 搜索配额管理 # ============================================================ def _get_usage_file() -> str: """获取搜索使用量记录文件路径。""" return get_data_file("usage.json") def check_search_quota() -> bool: """检查今日搜索配额是否充足。 Returns: True 表示配额充足，False 表示已达上限（已输出错误信息）。 """ sub = check_subscription() daily_limit = sub["daily_searches"] # 付费版不限搜索次数 if daily_limit < 0: return True usage = read_json_file(_get_usage_file()) if not isinstance(usage, dict): usage = {} today = today_str() today_count = usage.get(today, 0) if today_count >= daily_limit: output_error( f"今日搜索次数已达上限（{daily_limit} 次）。请升级至付费版（¥129/月）以获取无限搜索。", code="QUOTA_EXCEEDED", ) return False return True def increment_search_count() -> None: """递增今日搜索计数。""" filepath = _get_usage_file() usage = read_json_file(filepath) if not isinstance(usage, dict): usage = {} today = today_str() usage[today] = usage.get(today, 0) + 1 # 清理 7 天前的记录 cutoff = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") usage = {k: v for k, v in usage.items() if k >= cutoff} write_json_file(filepath, usage) # ============================================================ # 通用工具函数 # ============================================================ def generate_id(prefix: str = "KM") -> str: """生成唯一 ID。基于时间戳生成，格式为前缀+时间戳。 Args: prefix: ID 前缀，默认为 "KM"（知识网格）。 Returns: 唯一 ID 字符串。 """ timestamp = datetime.now().strftime("%Y%m%d%H%M%S%f") return f"{prefix}{timestamp}" def now_iso() -> str: """返回当前时间的 ISO 格式字符串。 Returns: ISO 格式时间字符串，如 "2026-03-19T10:30:00"。 """ return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def today_str() -> str: """返回今天的日期字符串。 Returns: 日期字符串，格式为 "YYYY-MM-DD"。 """ return datetime.now().strftime("%Y-%m-%d") def truncate_text(text: str, max_len: int = 200) -> str: """截断文本到指定最大长度。若超过最大长度，截断并追加省略号。 Args: text: 原始文本。 max_len: 最大长度，默认 200。 Returns: 截断后的文本。 """ if not text: return "" text = text.strip() if len(text) <= max_len: return text return text[:max_len - 3] + "..." def highlight_match(text: str, query: str) -> str: """在文本中高亮匹配的查询词。使用 **加粗** 标记匹配部分（Markdown 格式）。 Args: text: 原始文本。 query: 查询关键词。 Returns: 高亮后的文本。 """ if not text or not query: return text or "" # 将查询拆分为多个关键词 keywords = [kw.strip() for kw in query.split() if kw.strip()] result = text for kw in keywords: # 不区分大小写替换 pattern = re.compile(re.escape(kw), re.IGNORECASE) result = pattern.sub(lambda m: f"**{m.group(0)}**", result) return result def format_source_badge(source_name: str) -> str: """格式化知识源标签显示。生成 Markdown 格式的知识源标签。 Args: source_name: 知识源名称（如 "github"）。 Returns: 格式化后的标签字符串，如 "[GitHub]"。 """ display = SOURCE_DISPLAY_NAMES.get(source_name, source_name) return f"[{display}]" def hash_text(text: str) -> str: """计算文本的 MD5 哈希值。 Args: text: 待哈希的文本。 Returns: MD5 哈希字符串。 """ return hashlib.md5(text.encode("utf-8")).hexdigest() def clean_html(html_text: str) -> str: """简单清理 HTML 标签，返回纯文本。 Args: html_text: 包含 HTML 标签的文本。 Returns: 清理后的纯文本。 """ if not html_text: return "" # 移除 HTML 标签 text = re.sub(r"<[^>]+>", "", html_text) # 处理常见的 HTML 实体 text = text.replace("&", "&") text = text.replace("<", "<") text = text.replace(">", ">") text = text.replace(""", '"') text = text.replace("'", "'") text = text.replace(" ", " ") # 压缩连续空白 text = re.sub(r"\s+", " ", text).strip() return text def parse_iso_datetime(dt_str: str) -> Optional[datetime]: """解析 ISO 格式的日期时间字符串。 Args: dt_str: ISO 日期时间字符串。 Returns: datetime 对象，解析失败返回 None。 """ if not dt_str: return None try: # 处理多种常见格式 if "T" in dt_str: # 移除时区后缀 clean = re.sub(r"[Zz]$", "", dt_str) clean = re.sub(r"[+\-]\d{2}:\d{2}$", "", clean) # 截断微秒 if "." in clean: clean = clean.split(".")[0] return datetime.strptime(clean, "%Y-%m-%dT%H:%M:%S") else: return datetime.strptime(dt_str, "%Y-%m-%d") except (ValueError, TypeError): return None def days_ago(dt_str: str) -> int: """计算指定日期距今的天数。 Args: dt_str: 日期时间字符串。 Returns: 距今天数（正数表示过去），解析失败返回 0。 """ dt = parse_iso_datetime(dt_str) if dt is None: return 0 delta = datetime.now() - dt return max(0, delta.days) FILE:scripts/report_exporter.py #!/usr/bin/env python3 """ knowledge-mesh 搜索报告导出模块支持将搜索结果导出为 Markdown 报告、CSV 文件，生成趋势分析图表和使用统计。用法: python3 report_exporter.py --action export-markdown --data '{"query":"...","results":[...]}' python3 report_exporter.py --action export-csv --data '{"results":[...]}' python3 report_exporter.py --action trending --data '{"results":[...]}' python3 report_exporter.py --action stats """ import csv import io import os import sys from collections import Counter from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from utils import ( SOURCE_DISPLAY_NAMES, check_subscription, format_source_badge, get_data_file, highlight_match, load_input_data, now_iso, output_error, output_success, parse_common_args, read_json_file, require_paid_feature, today_str, truncate_text, write_json_file, ) # ============================================================ # 常量 # ============================================================ # 搜索历史文件 SEARCH_HISTORY_FILE = "search_history.json" # 导出统计文件 EXPORT_STATS_FILE = "export_stats.json" # ============================================================ # 搜索历史管理 # ============================================================ def _get_search_history() -> List[Dict[str, Any]]: """读取搜索历史记录。""" data = read_json_file(get_data_file(SEARCH_HISTORY_FILE)) if isinstance(data, list): return data return [] def _save_search_history(history: List[Dict[str, Any]]) -> None: """保存搜索历史记录。""" # 最多保留 500 条记录 write_json_file(get_data_file(SEARCH_HISTORY_FILE), history[-500:]) def _record_export(export_type: str, count: int) -> None: """记录导出操作到统计数据。 Args: export_type: 导出类型。 count: 导出记录数。 """ stats_file = get_data_file(EXPORT_STATS_FILE) stats = read_json_file(stats_file) if not isinstance(stats, dict): stats = {"exports": [], "total_exports": 0} stats["exports"].append({ "type": export_type, "count": count, "timestamp": now_iso(), }) stats["total_exports"] = stats.get("total_exports", 0) + 1 # 最多保留最近 200 条导出记录 stats["exports"] = stats["exports"][-200:] write_json_file(stats_file, stats) # ============================================================ # Markdown 报告生成 # ============================================================ def _generate_markdown_report( query: str, results: List[Dict[str, Any]], title: Optional[str] = None, ) -> str: """生成 Markdown 格式的搜索报告。 Args: query: 搜索查询。 results: 搜索结果列表。 title: 可选的报告标题。 Returns: Markdown 报告字符串。 """ report_title = title or f"知识搜索报告: {query}" today = today_str() parts = [] parts.append(f"# {report_title}\n") parts.append(f"**查询关键词**: {query}") parts.append(f"**搜索时间**: {now_iso()}") parts.append(f"**结果总数**: {len(results)} 条\n") # 来源分布统计 source_counts = Counter() for r in results: source_counts[r.get("source", "unknown")] += 1 if source_counts: parts.append("## 来源分布\n") parts.append("| 来源 | 结果数 |") parts.append("|------|--------|") for source, count in source_counts.most_common(): display = SOURCE_DISPLAY_NAMES.get(source, source) parts.append(f"| {display} | {count} |") parts.append("") # 搜索结果详情 parts.append("## 搜索结果\n") for idx, r in enumerate(results, 1): source = r.get("source", "") badge = format_source_badge(source) result_title = r.get("title", "无标题") url = r.get("url", "") snippet = r.get("snippet", "") author = r.get("author", "") created = r.get("created_at", "") score = r.get("_combined_score", r.get("score", 0)) tags = r.get("tags", []) # 高亮查询词 highlighted_title = highlight_match(result_title, query) highlighted_snippet = highlight_match(truncate_text(snippet, 250), query) parts.append(f"### {idx}. {badge} {highlighted_title}\n") if url: parts.append(f"- **链接**: [{url}]({url})") if author: parts.append(f"- **作者**: {author}") if created: parts.append(f"- **发布日期**: {created}") if isinstance(score, (int, float)) and score > 0: parts.append(f"- **相关度**: {score:.2f}") if tags: tag_str = ", ".join(f"`{t}`" for t in tags[:5]) parts.append(f"- **标签**: {tag_str}") parts.append(f"\n> {highlighted_snippet}\n") # 页脚 parts.append("---\n") parts.append(f"*报告由 knowledge-mesh 于 {now_iso()} 生成*") return "\n".join(parts) # ============================================================ # CSV 导出 # ============================================================ def _generate_csv(results: List[Dict[str, Any]]) -> str: """将搜索结果导出为 CSV 格式。 Args: results: 搜索结果列表。 Returns: CSV 格式字符串。 """ fieldnames = [ "source", "title", "url", "snippet", "author", "created_at", "score", "tags", ] output_buf = io.StringIO() writer = csv.DictWriter(output_buf, fieldnames=fieldnames) writer.writeheader() for r in results: row = { "source": SOURCE_DISPLAY_NAMES.get(r.get("source", ""), r.get("source", "")), "title": r.get("title", ""), "url": r.get("url", ""), "snippet": truncate_text(r.get("snippet", ""), 200), "author": r.get("author", ""), "created_at": r.get("created_at", ""), "score": r.get("_combined_score", r.get("score", 0)), "tags": ", ".join(r.get("tags", [])), } writer.writerow(row) return output_buf.getvalue() # ============================================================ # 趋势分析（付费功能） # ============================================================ def _generate_trending_report(results: List[Dict[str, Any]]) -> str: """生成趋势分析报告，包含 Mermaid 图表。 Args: results: 搜索结果列表。 Returns: Markdown 格式的趋势报告。 """ parts = [] parts.append("# 知识趋势分析报告\n") parts.append(f"**生成时间**: {now_iso()}") parts.append(f"**分析样本**: {len(results)} 条结果\n") # 来源分布饼图 source_counts = Counter() for r in results: source_counts[r.get("source", "unknown")] += 1 parts.append("## 来源分布\n") parts.append("```mermaid") parts.append("pie title 知识来源分布") for source, count in source_counts.most_common(): display = SOURCE_DISPLAY_NAMES.get(source, source) parts.append(f' "{display}" : {count}') parts.append("```\n") # 时间趋势（按周分组） week_counts = Counter() for r in results: created = r.get("created_at", "") if created: try: if "T" in created: dt = datetime.strptime(created.split("T")[0], "%Y-%m-%d") else: dt = datetime.strptime(created, "%Y-%m-%d") # 按周分组 week_start = dt - timedelta(days=dt.weekday()) week_key = week_start.strftime("%m/%d") week_counts[week_key] += 1 except (ValueError, TypeError): continue if week_counts: # 按时间排序 sorted_weeks = sorted(week_counts.items()) recent_weeks = sorted_weeks[-8:] # 最近 8 周 parts.append("## 内容发布时间趋势\n") labels = [f'"{w[0]}"' for w in recent_weeks] values = [str(w[1]) for w in recent_weeks] parts.append("```mermaid") parts.append("xychart-beta") parts.append(' title "近期内容发布趋势"') parts.append(f' x-axis [{", ".join(labels)}]') parts.append(' y-axis "内容数量"') parts.append(f' bar [{", ".join(values)}]') parts.append("```\n") # 热门标签 all_tags = [] for r in results: all_tags.extend(r.get("tags", [])) if all_tags: tag_counts = Counter(all_tags).most_common(15) parts.append("## 热门标签\n") parts.append("| 标签 | 出现次数 |") parts.append("|------|----------|") for tag, count in tag_counts: parts.append(f"| `{tag}` | {count} |") parts.append("") # Mermaid 柱状图展示 Top 10 标签 top10_tags = tag_counts[:10] if top10_tags: tag_labels = [f'"{t[0]}"' for t in top10_tags] tag_values = [str(t[1]) for t in top10_tags] parts.append("```mermaid") parts.append("xychart-beta") parts.append(' title "Top 10 热门标签"') parts.append(f' x-axis [{", ".join(tag_labels)}]') parts.append(' y-axis "出现次数"') parts.append(f' bar [{", ".join(tag_values)}]') parts.append("```\n") # 高分结果 scored = [r for r in results if r.get("score", 0) > 0 or r.get("_combined_score", 0) > 0] if scored: scored.sort(key=lambda r: r.get("_combined_score", r.get("score", 0)), reverse=True) top5 = scored[:5] parts.append("## 高相关度内容 Top 5\n") for idx, r in enumerate(top5, 1): badge = format_source_badge(r.get("source", "")) title = r.get("title", "无标题") url = r.get("url", "") score = r.get("_combined_score", r.get("score", 0)) parts.append(f"{idx}. {badge} **{title}**") if url: parts.append(f" - 链接: {url}") parts.append(f" - 相关度: {score:.2f}") parts.append("") parts.append("---\n") parts.append("*由 knowledge-mesh 自动生成*") return "\n".join(parts) # ============================================================ # 操作实现 # ============================================================ def action_export_markdown(data: Dict[str, Any]) -> None: """导出搜索结果为 Markdown 报告。 Args: data: 包含 query、results 的字典，可选 title 和 file_path。 """ query = data.get("query", "") results = data.get("results", []) title = data.get("title") file_path = data.get("file_path") if not results: output_error("无搜索结果可导出", code="NO_DATA") return report = _generate_markdown_report(query, results, title) if file_path: try: dir_name = os.path.dirname(file_path) if dir_name: os.makedirs(dir_name, exist_ok=True) with open(file_path, "w", encoding="utf-8") as f: f.write(report) _record_export("markdown", len(results)) output_success({ "message": f"报告已导出到 {file_path}", "file_path": file_path, "result_count": len(results), }) except IOError as e: output_error(f"文件写入失败: {e}", code="IO_ERROR") else: _record_export("markdown", len(results)) output_success({ "report": report, "result_count": len(results), }) def action_export_csv(data: Dict[str, Any]) -> None: """导出搜索结果为 CSV 格式。 Args: data: 包含 results 的字典，可选 file_path。 """ results = data.get("results", []) file_path = data.get("file_path") if not results: output_error("无搜索结果可导出", code="NO_DATA") return csv_content = _generate_csv(results) if file_path: try: dir_name = os.path.dirname(file_path) if dir_name: os.makedirs(dir_name, exist_ok=True) with open(file_path, "w", encoding="utf-8", newline="") as f: f.write(csv_content) _record_export("csv", len(results)) output_success({ "message": f"CSV 已导出到 {file_path}", "file_path": file_path, "result_count": len(results), }) except IOError as e: output_error(f"文件写入失败: {e}", code="IO_ERROR") else: _record_export("csv", len(results)) output_success({ "csv": csv_content, "result_count": len(results), }) def action_trending(data: Dict[str, Any]) -> None: """生成趋势分析报告（付费功能）。 Args: data: 包含 results 的字典。 """ if not require_paid_feature("mermaid_chart", "趋势图表分析"): return results = data.get("results", []) if not results: output_error("无搜索结果可分析", code="NO_DATA") return report = _generate_trending_report(results) output_success({ "report": report, "result_count": len(results), }) def action_stats(data: Optional[Dict[str, Any]] = None) -> None: """显示搜索使用统计。""" # 读取使用量数据 usage_file = get_data_file("usage.json") usage = read_json_file(usage_file) if not isinstance(usage, dict): usage = {} # 读取导出统计 stats_file = get_data_file(EXPORT_STATS_FILE) export_stats = read_json_file(stats_file) if not isinstance(export_stats, dict): export_stats = {"exports": [], "total_exports": 0} # 搜索统计 today = today_str() today_searches = usage.get(today, 0) total_searches = sum(usage.values()) # 过去 7 天统计 daily_stats = {} for i in range(7): day = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d") daily_stats[day] = usage.get(day, 0) # 检查订阅信息 sub = check_subscription() daily_limit = sub.get("daily_searches", 10) remaining = max(0, daily_limit - today_searches) if daily_limit >= 0 else -1 # 导出统计 export_type_counts = Counter() for exp in export_stats.get("exports", []): export_type_counts[exp.get("type", "unknown")] += 1 output_success({ "subscription_tier": sub["tier"], "today_searches": today_searches, "daily_limit": daily_limit, "remaining_today": remaining, "total_searches_7d": total_searches, "daily_breakdown": daily_stats, "total_exports": export_stats.get("total_exports", 0), "export_by_type": dict(export_type_counts), }) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("knowledge-mesh 搜索报告导出") args = parser.parse_args() action = args.action.lower() try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "export-markdown": lambda: action_export_markdown(data or {}), "export-csv": lambda: action_export_csv(data or {}), "trending": lambda: action_trending(data or {}), "stats": lambda: action_stats(data), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(actions.keys()) output_error(f"未知操作: {action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/index_builder.py #!/usr/bin/env python3 """ knowledge-mesh 本地知识索引构建器对本地 .md/.txt/.py 等文件构建 TF-IDF 倒排索引，支持全文搜索、索引管理和重建。用法: python3 index_builder.py --action index --data '{"paths":["./docs"],"patterns":["*.md","*.txt"]}' python3 index_builder.py --action search-local --data '{"query":"async python"}' python3 index_builder.py --action list-indexed python3 index_builder.py --action rebuild python3 index_builder.py --action delete --data '{"doc_id":"DOC20260319..."}' """ import fnmatch import math import os import re import sys from collections import Counter from datetime import datetime from typing import Any, Dict, List, Optional, Set, Tuple from utils import ( check_subscription, generate_id, get_data_file, load_input_data, now_iso, output_error, output_success, parse_common_args, read_json_file, require_paid_feature, truncate_text, write_json_file, ) # ============================================================ # 常量 # ============================================================ # 索引数据文件 INDEX_DATA_FILE = "index_data.json" # 文档元数据文件 DOC_METADATA_FILE = "doc_metadata.json" # 支持索引的文件扩展名 INDEXABLE_EXTENSIONS = {".md", ".txt", ".py", ".js", ".ts", ".java", ".go", ".rs", ".rb", ".sh", ".yaml", ".yml", ".json", ".toml", ".cfg", ".ini", ".html", ".css", ".sql", ".markdown"} # Obsidian 特性正则表达式 RE_WIKILINK = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]") RE_TAG_OBSIDIAN = re.compile(r"(?:^|\s)#([a-zA-Z0-9_\u4e00-\u9fff][a-zA-Z0-9_/\u4e00-\u9fff]*)") RE_FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL) # Wikilink 和标签的额外权重 WIKILINK_WEIGHT = 2.0 TAG_WEIGHT = 3.0 # 默认 glob 模式 DEFAULT_PATTERNS = ["*.md", "*.txt", "*.py"] # 停用词 STOP_WORDS = { "the", "a", "an", "is", "are", "was", "were", "be", "been", "being", "have", "has", "had", "do", "does", "did", "will", "would", "could", "should", "may", "might", "can", "shall", "to", "of", "in", "for", "on", "with", "at", "by", "from", "as", "into", "through", "during", "before", "after", "and", "but", "or", "not", "so", "if", "then", "than", "too", "very", "just", "about", "up", "out", "no", "it", "its", "this", "that", "i", "me", "my", "we", "our", "you", "your", "he", "she", "they", "them", "their", "what", "which", "who", "how", "when", "where", "why", "all", "each", "every", "both", "的", "了", "在", "是", "我", "有", "和", "就", "不", "人", "都", "一", "一个", "上", "也", "很", "到", "说", "要", "去", "你", "会", "着", "没有", "看", "好", "自己", "这", "他", "她", "import", "from", "def", "class", "return", "self", "none", "true", "false", "pass", "elif", "else", "try", "except", } # 单次索引最大文件数 MAX_INDEX_FILES = 500 # 单个文件最大大小（字节） MAX_FILE_SIZE = 1024 * 1024 # 1MB # ============================================================ # 文本分词 # ============================================================ def _tokenize(text: str) -> List[str]: """将文本分词为词语列表。 Args: text: 原始文本。 Returns: 小写词语列表（去停用词）。 """ if not text: return [] tokens = re.findall(r"[a-zA-Z0-9_]{2,}|[\u4e00-\u9fff]+", text.lower()) return [t for t in tokens if t not in STOP_WORDS and len(t) >= 2] def _tokenize_with_positions(text: str) -> List[Tuple[str, int]]: """分词并记录每个词的起始位置。 Args: text: 原始文本。 Returns: (词语, 位置) 元组列表。 """ if not text: return [] result = [] for match in re.finditer(r"[a-zA-Z0-9_]{2,}|[\u4e00-\u9fff]+", text.lower()): token = match.group(0) if token not in STOP_WORDS and len(token) >= 2: result.append((token, match.start())) return result # ============================================================ # 文件扫描 # ============================================================ def _match_patterns(filename: str, patterns: List[str]) -> bool: """检查文件名是否匹配任一 glob 模式。 Args: filename: 文件名。 patterns: glob 模式列表。 Returns: 是否匹配。 """ for pattern in patterns: if fnmatch.fnmatch(filename, pattern): return True return False def _scan_files( paths: List[str], patterns: Optional[List[str]] = None, ) -> List[str]: """扫描指定路径下匹配模式的文件。 Args: paths: 待扫描的目录或文件路径列表。 patterns: glob 模式列表，默认 ["*.md", "*.txt", "*.py"]。 Returns: 匹配的文件绝对路径列表。 """ if patterns is None: patterns = DEFAULT_PATTERNS matched_files = [] for path in paths: abs_path = os.path.abspath(path) if os.path.isfile(abs_path): # 检查扩展名 _, ext = os.path.splitext(abs_path) if ext.lower() in INDEXABLE_EXTENSIONS or _match_patterns(os.path.basename(abs_path), patterns): matched_files.append(abs_path) elif os.path.isdir(abs_path): for root, dirs, files in os.walk(abs_path): # 跳过隐藏目录和常见忽略目录 dirs[:] = [d for d in dirs if not d.startswith(".") and d not in {"node_modules", "__pycache__", "venv", ".git"}] for f in files: if _match_patterns(f, patterns): filepath = os.path.join(root, f) matched_files.append(filepath) if len(matched_files) >= MAX_INDEX_FILES: break return matched_files[:MAX_INDEX_FILES] def _read_file_content(filepath: str) -> Optional[str]: """安全读取文件内容。 Args: filepath: 文件路径。 Returns: 文件文本内容，读取失败返回 None。 """ try: size = os.path.getsize(filepath) if size > MAX_FILE_SIZE: return None with open(filepath, "r", encoding="utf-8", errors="ignore") as f: return f.read() except (IOError, OSError): return None # ============================================================ # Obsidian 特性解析 # ============================================================ def _extract_wikilinks(content: str) -> List[str]: """提取 Obsidian [[wikilinks]] 作为额外关键词信号。 Args: content: 文件内容。 Returns: 链接目标列表。 """ return RE_WIKILINK.findall(content) def _extract_obsidian_tags(content: str) -> List[str]: """提取 Obsidian #tags 作为高权重关键词。 Args: content: 文件内容。 Returns: 标签列表（不含 # 前缀）。 """ return RE_TAG_OBSIDIAN.findall(content) def _parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]: """解析 YAML frontmatter 元数据。解析日期、分类、作者等字段用于过滤。 Args: content: 文件内容。 Returns: (frontmatter 字典, 去除 frontmatter 后的内容) 元组。 """ match = RE_FRONTMATTER.match(content) if not match: return {}, content fm_text = match.group(1) body = content[match.end():] frontmatter = {} current_key = None current_list = None for line in fm_text.split("\n"): line = line.strip() if not line or line.startswith("#"): continue if line.startswith("- ") and current_key and current_list is not None: value = line[2:].strip().strip('"').strip("'") current_list.append(value) continue if ":" in line: idx = line.index(":") key = line[:idx].strip() value = line[idx + 1:].strip() current_key = key if not value: current_list = [] frontmatter[key] = current_list else: value = value.strip('"').strip("'") if value.startswith("[") and value.endswith("]"): items = [v.strip().strip('"').strip("'") for v in value[1:-1].split(",")] frontmatter[key] = [i for i in items if i] else: frontmatter[key] = value current_list = None return frontmatter, body def _get_obsidian_extra_tokens(content: str) -> Tuple[List[Tuple[str, float]], Dict[str, Any]]: """从文件内容中提取 Obsidian 特性的额外 token 和元数据。 Wikilinks 作为加权 token（权重 2.0）， Tags 作为高权重 token（权重 3.0）， Frontmatter 作为过滤元数据。 Args: content: 文件内容。 Returns: (加权token列表, frontmatter字典) 元组。 """ extra_tokens = [] # 提取 wikilinks 并分词 wikilinks = _extract_wikilinks(content) for link in wikilinks: link_tokens = _tokenize(link) for token in link_tokens: extra_tokens.append((token, WIKILINK_WEIGHT)) # 提取标签 tags = _extract_obsidian_tags(content) for tag in tags: tag_tokens = _tokenize(tag) for token in tag_tokens: extra_tokens.append((token, TAG_WEIGHT)) # 完整标签也作为 token if len(tag) >= 2: extra_tokens.append((tag.lower(), TAG_WEIGHT)) # 解析 frontmatter frontmatter, _ = _parse_frontmatter(content) return extra_tokens, frontmatter # ============================================================ # 倒排索引构建 # ============================================================ def _build_inverted_index( documents: Dict[str, str], ) -> Tuple[Dict[str, List[Dict[str, Any]]], Dict[str, Dict[str, Any]]]: """构建倒排索引和文档元数据。 Args: documents: {doc_id: content} 映射。 Returns: (倒排索引, 文档元数据) 元组。倒排索引格式: {term: [{doc_id, tf, positions}]} 文档元数据格式: {doc_id: {path, size, token_count, indexed_at}} """ inverted_index = {} doc_metadata = {} doc_count = len(documents) for doc_id, content in documents.items(): tokens_with_pos = _tokenize_with_positions(content) tokens = [t for t, _ in tokens_with_pos] token_count = len(tokens) if token_count == 0: continue # 统计词频和位置 term_info = {} for token, pos in tokens_with_pos: if token not in term_info: term_info[token] = {"count": 0, "positions": []} term_info[token]["count"] += 1 term_info[token]["positions"].append(pos) # 提取 Obsidian 特性的额外加权 token extra_tokens, frontmatter = _get_obsidian_extra_tokens(content) for token, weight in extra_tokens: if token not in term_info: term_info[token] = {"count": 0, "positions": []} # 加权计数：wikilinks 和标签获得额外权重 term_info[token]["count"] += int(weight) # 构建倒排索引条目 for term, info in term_info.items(): tf = 1 + math.log(info["count"]) if info["count"] > 0 else 0 entry = { "doc_id": doc_id, "tf": round(tf, 4), "positions": info["positions"][:20], # 最多保留 20 个位置 } if term not in inverted_index: inverted_index[term] = [] inverted_index[term].append(entry) doc_metadata[doc_id] = { "token_count": token_count, "unique_terms": len(term_info), "frontmatter": frontmatter if frontmatter else None, } return inverted_index, doc_metadata def _search_index( query: str, inverted_index: Dict[str, List[Dict[str, Any]]], doc_metadata_map: Dict[str, Dict[str, Any]], max_results: int = 20, ) -> List[Dict[str, Any]]: """在倒排索引中搜索。 Args: query: 查询字符串。 inverted_index: 倒排索引。 doc_metadata_map: 文档元数据。 max_results: 最大返回结果数。 Returns: 排序后的搜索结果列表。 """ query_tokens = _tokenize(query) if not query_tokens: return [] doc_count = len(doc_metadata_map) if doc_count == 0: return [] # 计算每个查询词的 IDF term_idf = {} for qt in query_tokens: df = len(inverted_index.get(qt, [])) term_idf[qt] = math.log(doc_count / (1 + df)) if doc_count > 0 else 0 # 累计文档分数 doc_scores = {} doc_matches = {} for qt in query_tokens: postings = inverted_index.get(qt, []) idf_val = term_idf.get(qt, 0) for posting in postings: did = posting["doc_id"] tf_val = posting.get("tf", 0) score = tf_val * idf_val if did not in doc_scores: doc_scores[did] = 0.0 doc_matches[did] = [] doc_scores[did] += score doc_matches[did].append(qt) # 排序 ranked = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True) results = [] for doc_id, score in ranked[:max_results]: meta = doc_metadata_map.get(doc_id, {}) results.append({ "doc_id": doc_id, "path": meta.get("path", ""), "filename": os.path.basename(meta.get("path", "")), "score": round(score, 4), "matched_terms": doc_matches.get(doc_id, []), "token_count": meta.get("token_count", 0), "indexed_at": meta.get("indexed_at", ""), }) return results # ============================================================ # 索引持久化 # ============================================================ def _load_index() -> Dict[str, List[Dict[str, Any]]]: """加载已有的倒排索引。""" data = read_json_file(get_data_file(INDEX_DATA_FILE)) if isinstance(data, dict): return data return {} def _save_index(index: Dict[str, List[Dict[str, Any]]]) -> None: """保存倒排索引。""" write_json_file(get_data_file(INDEX_DATA_FILE), index) def _load_doc_metadata() -> Dict[str, Dict[str, Any]]: """加载文档元数据。""" data = read_json_file(get_data_file(DOC_METADATA_FILE)) if isinstance(data, dict): return data return {} def _save_doc_metadata(metadata: Dict[str, Dict[str, Any]]) -> None: """保存文档元数据。""" write_json_file(get_data_file(DOC_METADATA_FILE), metadata) # ============================================================ # 操作实现 # ============================================================ def action_index(data: Dict[str, Any]) -> None: """索引本地文件。 Args: data: 包含 paths（目录/文件列表）和可选 patterns（glob 模式）的字典。 """ if not require_paid_feature("local_index", "本地知识索引"): return paths = data.get("paths", []) if not paths: output_error("请提供待索引的路径列表（paths）", code="VALIDATION_ERROR") return if isinstance(paths, str): paths = [paths] patterns = data.get("patterns", DEFAULT_PATTERNS) # 扫描文件 files = _scan_files(paths, patterns) if not files: output_error("未找到匹配的文件", code="NO_FILES") return # 读取文件内容 documents = {} file_meta = {} skipped = 0 for filepath in files: content = _read_file_content(filepath) if content is None: skipped += 1 continue doc_id = generate_id("DOC") documents[doc_id] = content file_meta[doc_id] = { "path": filepath, "size": os.path.getsize(filepath), "indexed_at": now_iso(), } if not documents: output_error("所有文件均无法读取", code="READ_ERROR") return # 构建索引 inverted_index, build_meta = _build_inverted_index(documents) # 合并到现有索引 existing_index = _load_index() existing_meta = _load_doc_metadata() # 合并倒排索引 for term, postings in inverted_index.items(): if term not in existing_index: existing_index[term] = [] existing_index[term].extend(postings) # 合并元数据 for doc_id, meta in file_meta.items(): bm = build_meta.get(doc_id, {}) existing_meta[doc_id] = { **meta, "token_count": bm.get("token_count", 0), "unique_terms": bm.get("unique_terms", 0), } # 保存 _save_index(existing_index) _save_doc_metadata(existing_meta) output_success({ "message": f"索引完成：成功 {len(documents)} 个文件，跳过 {skipped} 个", "indexed_count": len(documents), "skipped_count": skipped, "total_terms": len(existing_index), "total_documents": len(existing_meta), }) def action_search_local(data: Dict[str, Any]) -> None: """在本地索引中搜索。 Args: data: 包含 query 的字典，可选 max_results。 """ if not require_paid_feature("local_index", "本地知识索引"): return query = data.get("query", "").strip() if not query: output_error("请提供搜索关键词（query）", code="VALIDATION_ERROR") return max_results = data.get("max_results", 20) inverted_index = _load_index() doc_metadata = _load_doc_metadata() if not inverted_index or not doc_metadata: output_error("本地索引为空，请先执行 index 操作构建索引", code="NO_INDEX") return results = _search_index(query, inverted_index, doc_metadata, max_results) output_success({ "query": query, "total": len(results), "results": results, }) def action_list_indexed(data: Optional[Dict[str, Any]] = None) -> None: """列出已索引的文档。""" doc_metadata = _load_doc_metadata() docs = [] for doc_id, meta in doc_metadata.items(): docs.append({ "doc_id": doc_id, "path": meta.get("path", ""), "filename": os.path.basename(meta.get("path", "")), "size": meta.get("size", 0), "token_count": meta.get("token_count", 0), "indexed_at": meta.get("indexed_at", ""), }) # 按索引时间倒序 docs.sort(key=lambda d: d.get("indexed_at", ""), reverse=True) inverted_index = _load_index() output_success({ "total_documents": len(docs), "total_terms": len(inverted_index), "documents": docs, }) def action_rebuild(data: Optional[Dict[str, Any]] = None) -> None: """重建索引：根据已记录的文件路径重新索引。""" if not require_paid_feature("local_index", "本地知识索引"): return doc_metadata = _load_doc_metadata() if not doc_metadata: output_error("无已索引文档，无需重建", code="NO_INDEX") return # 收集已有路径 paths = [] for meta in doc_metadata.values(): p = meta.get("path", "") if p and os.path.exists(p): paths.append(p) if not paths: output_error("所有已索引文件均不存在，无法重建", code="FILES_MISSING") return # 清空现有索引 _save_index({}) _save_doc_metadata({}) # 重新读取和索引 documents = {} file_meta = {} skipped = 0 for filepath in paths: content = _read_file_content(filepath) if content is None: skipped += 1 continue doc_id = generate_id("DOC") documents[doc_id] = content file_meta[doc_id] = { "path": filepath, "size": os.path.getsize(filepath), "indexed_at": now_iso(), } if not documents: output_error("所有文件均无法读取", code="READ_ERROR") return inverted_index, build_meta = _build_inverted_index(documents) # 保存 final_meta = {} for doc_id, meta in file_meta.items(): bm = build_meta.get(doc_id, {}) final_meta[doc_id] = { **meta, "token_count": bm.get("token_count", 0), "unique_terms": bm.get("unique_terms", 0), } _save_index(inverted_index) _save_doc_metadata(final_meta) output_success({ "message": f"索引重建完成：成功 {len(documents)} 个文件，跳过 {skipped} 个", "indexed_count": len(documents), "skipped_count": skipped, "total_terms": len(inverted_index), "total_documents": len(final_meta), }) def action_delete(data: Dict[str, Any]) -> None: """删除指定文档的索引。 Args: data: 包含 doc_id 的字典。 """ doc_id = data.get("doc_id", "").strip() if not doc_id: output_error("请提供文档ID（doc_id）", code="VALIDATION_ERROR") return doc_metadata = _load_doc_metadata() if doc_id not in doc_metadata: output_error(f"未找到文档: {doc_id}", code="NOT_FOUND") return # 从元数据中删除 removed_meta = doc_metadata.pop(doc_id) _save_doc_metadata(doc_metadata) # 从倒排索引中清除该文档的条目 inverted_index = _load_index() terms_to_remove = [] for term, postings in inverted_index.items(): inverted_index[term] = [p for p in postings if p.get("doc_id") != doc_id] if not inverted_index[term]: terms_to_remove.append(term) for term in terms_to_remove: del inverted_index[term] _save_index(inverted_index) output_success({ "message": f"文档 {doc_id} 的索引已删除", "removed_path": removed_meta.get("path", ""), "remaining_documents": len(doc_metadata), "remaining_terms": len(inverted_index), }) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("knowledge-mesh 本地知识索引") args = parser.parse_args() action = args.action.lower() try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "index": lambda: action_index(data or {}), "search-local": lambda: action_search_local(data or {}), "list-indexed": lambda: action_list_indexed(data), "rebuild": lambda: action_rebuild(data), "delete": lambda: action_delete(data or {}), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(actions.keys()) output_error(f"未知操作: {action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/obsidian_connector.py #!/usr/bin/env python3 """ knowledge-mesh Obsidian 知识库集成模块将 Obsidian 笔记库作为知识源接入统一搜索系统，支持笔记搜索、索引构建、Wikilink/标签解析和增量同步。用法: python3 obsidian_connector.py --action connect --data '{"vault_path":"/path/to/vault"}' python3 obsidian_connector.py --action search --data '{"query":"python async"}' python3 obsidian_connector.py --action index --data '{"vault_path":"/path/to/vault"}' python3 obsidian_connector.py --action list-notes python3 obsidian_connector.py --action sync """ import json import math import os import re import sys from collections import Counter from datetime import datetime from typing import Any, Dict, List, Optional, Set, Tuple from utils import ( generate_id, get_data_file, load_input_data, now_iso, output_error, output_success, parse_common_args, read_json_file, truncate_text, write_json_file, ) # ============================================================ # 常量 # ============================================================ # Obsidian 索引数据文件 OBSIDIAN_INDEX_FILE = "obsidian_index.json" # 支持索引的文件扩展名 OBSIDIAN_EXTENSIONS = {".md", ".markdown"} # 最大索引笔记数 MAX_NOTES = 5000 # 单个文件最大大小（字节） MAX_FILE_SIZE = 2 * 1024 * 1024 # 2MB # 正则表达式：Obsidian 特性解析 RE_WIKILINK = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]") RE_TAG = re.compile(r"(?:^|\s)#([a-zA-Z0-9_\u4e00-\u9fff][a-zA-Z0-9_/\u4e00-\u9fff]*)") RE_FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL) RE_CALLOUT = re.compile(r"^>\s*\[!(\w+)\]\s*(.*?)$", re.MULTILINE) # 搜索相关性权重 TITLE_MATCH_WEIGHT = 3.0 TAG_MATCH_WEIGHT = 2.0 CONTENT_MATCH_WEIGHT = 1.0 BACKLINK_BONUS = 0.5 # 停用词 STOP_WORDS = { "the", "a", "an", "is", "are", "in", "on", "for", "to", "of", "and", "or", "not", "with", "it", "this", "that", "be", "was", "的", "了", "在", "是", "我", "有", "和", "就", "不", } # ============================================================ # Obsidian 特性解析 # ============================================================ def _parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]: """解析 YAML frontmatter。 Args: content: 笔记原始内容。 Returns: (frontmatter 字典, 去除 frontmatter 后的内容) 元组。 """ match = RE_FRONTMATTER.match(content) if not match: return {}, content fm_text = match.group(1) body = content[match.end():] # 简单 YAML 解析（不依赖 PyYAML） frontmatter = {} current_key = None current_list = None for line in fm_text.split("\n"): line = line.strip() if not line or line.startswith("#"): continue # 检查是否为列表项 if line.startswith("- ") and current_key: value = line[2:].strip().strip('"').strip("'") if current_list is not None: current_list.append(value) continue # 检查键值对 if ":" in line: idx = line.index(":") key = line[:idx].strip() value = line[idx + 1:].strip() current_key = key if not value: # 可能是列表的开始 current_list = [] frontmatter[key] = current_list else: # 去除引号 value = value.strip('"').strip("'") # 尝试转换为布尔/数字 if value.lower() in ("true", "yes"): frontmatter[key] = True elif value.lower() in ("false", "no"): frontmatter[key] = False else: try: frontmatter[key] = int(value) except ValueError: try: frontmatter[key] = float(value) except ValueError: # 处理列表格式 [item1, item2] if value.startswith("[") and value.endswith("]"): items = [v.strip().strip('"').strip("'") for v in value[1:-1].split(",")] frontmatter[key] = [i for i in items if i] else: frontmatter[key] = value current_list = None return frontmatter, body def _extract_wikilinks(content: str) -> List[str]: """提取所有 [[wikilinks]]。 Args: content: 笔记内容。 Returns: 链接目标列表（不含别名部分）。 """ return RE_WIKILINK.findall(content) def _extract_tags(content: str) -> List[str]: """提取所有 #tags。 Args: content: 笔记内容。 Returns: 标签列表（不含 # 前缀）。 """ return RE_TAG.findall(content) def _extract_callouts(content: str) -> List[Dict[str, str]]: """提取 Obsidian 样式的 callout 块。 Args: content: 笔记内容。 Returns: callout 信息列表。 """ callouts = [] for match in RE_CALLOUT.finditer(content): callouts.append({ "type": match.group(1), "title": match.group(2).strip(), }) return callouts def _tokenize(text: str) -> List[str]: """将文本分词为词语列表。 Args: text: 原始文本。 Returns: 小写词语列表。 """ if not text: return [] tokens = re.findall(r"[a-zA-Z0-9_]{2,}|[\u4e00-\u9fff]+", text.lower()) return [t for t in tokens if t not in STOP_WORDS and len(t) >= 2] # ============================================================ # 笔记扫描与解析 # ============================================================ def _scan_vault(vault_path: str) -> List[str]: """扫描 Obsidian vault 中的所有笔记文件。 Args: vault_path: vault 根目录路径。 Returns: 笔记文件绝对路径列表。 """ notes = [] abs_vault = os.path.abspath(vault_path) for root, dirs, files in os.walk(abs_vault): # 跳过隐藏目录和 Obsidian 配置目录 dirs[:] = [d for d in dirs if not d.startswith(".")] for f in files: _, ext = os.path.splitext(f) if ext.lower() in OBSIDIAN_EXTENSIONS: filepath = os.path.join(root, f) notes.append(filepath) if len(notes) >= MAX_NOTES: return notes return notes def _read_note(filepath: str) -> Optional[str]: """安全读取笔记文件内容。 Args: filepath: 文件路径。 Returns: 文件文本内容，读取失败返回 None。 """ try: size = os.path.getsize(filepath) if size > MAX_FILE_SIZE: return None with open(filepath, "r", encoding="utf-8", errors="ignore") as f: return f.read() except (IOError, OSError): return None def _parse_note(filepath: str, vault_path: str) -> Optional[Dict[str, Any]]: """解析单个笔记文件，提取元数据和内容。 Args: filepath: 笔记文件路径。 vault_path: vault 根目录路径。 Returns: 笔记信息字典，失败返回 None。 """ content = _read_note(filepath) if content is None: return None # 解析 frontmatter frontmatter, body = _parse_frontmatter(content) # 提取 Obsidian 特性 wikilinks = _extract_wikilinks(body) tags = _extract_tags(body) callouts = _extract_callouts(body) # 从 frontmatter 中补充标签 fm_tags = frontmatter.get("tags", []) if isinstance(fm_tags, str): fm_tags = [fm_tags] elif not isinstance(fm_tags, list): fm_tags = [] all_tags = list(set(tags + fm_tags)) # 文件基本信息 filename = os.path.basename(filepath) name_without_ext = os.path.splitext(filename)[0] rel_path = os.path.relpath(filepath, vault_path) # 获取文件修改时间 try: mtime = os.path.getmtime(filepath) modified_at = datetime.fromtimestamp(mtime).strftime("%Y-%m-%dT%H:%M:%S") except (OSError, ValueError): modified_at = "" # 分词 tokens = _tokenize(f"{name_without_ext} {body}") return { "filepath": filepath, "rel_path": rel_path, "title": name_without_ext, "content": body, "frontmatter": frontmatter, "tags": all_tags, "wikilinks": wikilinks, "callouts": callouts, "tokens": tokens, "token_count": len(tokens), "modified_at": modified_at, "size": os.path.getsize(filepath), } # ============================================================ # 索引持久化 # ============================================================ def _get_index_file() -> str: """获取 Obsidian 索引文件路径。""" return get_data_file(OBSIDIAN_INDEX_FILE) def _load_index() -> Dict[str, Any]: """加载 Obsidian 索引数据。 Returns: 索引数据字典。 """ data = read_json_file(_get_index_file()) if not isinstance(data, dict): data = {} if "vault_path" not in data: data["vault_path"] = "" if "notes" not in data: data["notes"] = {} if "backlink_graph" not in data: data["backlink_graph"] = {} if "tag_index" not in data: data["tag_index"] = {} if "last_sync" not in data: data["last_sync"] = "" if "total_notes" not in data: data["total_notes"] = 0 return data def _save_index(data: Dict[str, Any]) -> None: """保存 Obsidian 索引数据。""" write_json_file(_get_index_file(), data) # ============================================================ # 搜索引擎 # ============================================================ def _search_notes( query: str, index_data: Dict[str, Any], max_results: int = 20, ) -> List[Dict[str, Any]]: """在索引中搜索笔记。使用多级相关性评分：标题匹配 > 标签匹配 > 内容匹配。 Args: query: 搜索查询。 index_data: 索引数据。 max_results: 最大返回结果数。 Returns: 排序后的搜索结果列表。 """ notes = index_data.get("notes", {}) backlink_graph = index_data.get("backlink_graph", {}) if not notes: return [] query_tokens = _tokenize(query) query_lower = query.lower() if not query_tokens: return [] results = [] for note_id, note in notes.items(): score = 0.0 matched_in = [] title = note.get("title", "") tags = note.get("tags", []) content_preview = note.get("content_preview", "") tokens = note.get("tokens_sample", []) # 标题匹配（权重最高） title_lower = title.lower() title_match_count = 0 for qt in query_tokens: if qt in title_lower: title_match_count += 1 if title_match_count > 0: score += TITLE_MATCH_WEIGHT * (title_match_count / len(query_tokens)) matched_in.append("title") # 标签匹配 tags_lower = [t.lower() for t in tags] tag_match_count = 0 for qt in query_tokens: for tag in tags_lower: if qt in tag: tag_match_count += 1 break if tag_match_count > 0: score += TAG_MATCH_WEIGHT * (tag_match_count / len(query_tokens)) matched_in.append("tags") # 内容匹配 content_lower = content_preview.lower() content_match_count = 0 for qt in query_tokens: if qt in content_lower: content_match_count += 1 if content_match_count > 0: score += CONTENT_MATCH_WEIGHT * (content_match_count / len(query_tokens)) matched_in.append("content") # token 精确匹配加分 if tokens: token_set = set(tokens) token_matches = sum(1 for qt in query_tokens if qt in token_set) if token_matches > 0: score += 0.5 * (token_matches / len(query_tokens)) # 反向链接加分（被更多笔记引用的笔记更权威） backlink_count = len(backlink_graph.get(title, [])) if backlink_count > 0: score += BACKLINK_BONUS * min(backlink_count / 5.0, 1.0) if score > 0: results.append({ "id": note_id, "source": "obsidian", "title": title, "url": f"obsidian://open?vault={os.path.basename(index_data.get('vault_path', ''))}&file={note.get('rel_path', '')}", "snippet": truncate_text(content_preview, 300), "author": note.get("frontmatter", {}).get("author", ""), "created_at": note.get("modified_at", ""), "score": round(score, 4), "tags": tags, "matched_in": matched_in, "backlink_count": backlink_count, "filepath": note.get("filepath", ""), }) # 按分数降序排序 results.sort(key=lambda r: r.get("score", 0), reverse=True) return results[:max_results] # ============================================================ # 索引构建 # ============================================================ def _build_index(vault_path: str) -> Dict[str, Any]: """构建 Obsidian vault 的完整索引。 Args: vault_path: vault 根目录路径。 Returns: 索引数据字典。 """ note_files = _scan_vault(vault_path) notes = {} backlink_graph = {} # {被链接的笔记: [链接来源笔记]} tag_index = {} # {标签: [笔记ID]} skipped = 0 for filepath in note_files: parsed = _parse_note(filepath, vault_path) if parsed is None: skipped += 1 continue note_id = generate_id("ON") title = parsed["title"] # 存储笔记信息（不保存完整内容，只保存预览和采样 token） notes[note_id] = { "filepath": parsed["filepath"], "rel_path": parsed["rel_path"], "title": title, "content_preview": truncate_text(parsed["content"], 500), "frontmatter": parsed["frontmatter"], "tags": parsed["tags"], "wikilinks": parsed["wikilinks"], "tokens_sample": parsed["tokens"][:100], # 保存前100个token用于搜索 "token_count": parsed["token_count"], "modified_at": parsed["modified_at"], "size": parsed["size"], "mtime": os.path.getmtime(filepath) if os.path.exists(filepath) else 0, } # 构建反向链接图 for link_target in parsed["wikilinks"]: if link_target not in backlink_graph: backlink_graph[link_target] = [] backlink_graph[link_target].append(title) # 构建标签索引 for tag in parsed["tags"]: if tag not in tag_index: tag_index[tag] = [] tag_index[tag].append(note_id) index_data = { "vault_path": os.path.abspath(vault_path), "notes": notes, "backlink_graph": backlink_graph, "tag_index": tag_index, "last_sync": now_iso(), "total_notes": len(notes), "skipped": skipped, } return index_data # ============================================================ # 操作实现 # ============================================================ def action_connect(data: Dict[str, Any]) -> None: """连接到 Obsidian vault。验证 vault 路径存在，扫描结构并返回摘要信息。 Args: data: 包含 vault_path 的字典。 """ vault_path = data.get("vault_path", "").strip() if not vault_path: # 尝试从环境变量读取 vault_path = os.environ.get("KM_OBSIDIAN_VAULT_PATH", "").strip() if not vault_path: output_error( "请提供 Obsidian vault 路径（vault_path）或设置 KM_OBSIDIAN_VAULT_PATH 环境变量", code="VALIDATION_ERROR", ) return abs_path = os.path.abspath(vault_path) if not os.path.isdir(abs_path): output_error(f"Vault 路径不存在或不是目录: {abs_path}", code="PATH_NOT_FOUND") return # 扫描 vault 结构 note_files = _scan_vault(abs_path) # 检查是否有 .obsidian 配置目录 obsidian_config = os.path.join(abs_path, ".obsidian") has_obsidian_config = os.path.isdir(obsidian_config) # 统计目录结构 dir_counter = Counter() tag_counter = Counter() total_size = 0 for filepath in note_files[:200]: # 快速扫描前200个 rel_dir = os.path.dirname(os.path.relpath(filepath, abs_path)) if rel_dir: dir_counter[rel_dir] += 1 else: dir_counter["(root)"] += 1 total_size += os.path.getsize(filepath) # 快速提取标签 content = _read_note(filepath) if content: tags = _extract_tags(content) for tag in tags: tag_counter[tag] += 1 top_dirs = [{"dir": d, "count": c} for d, c in dir_counter.most_common(10)] top_tags = [{"tag": t, "count": c} for t, c in tag_counter.most_common(10)] output_success({ "message": f"成功连接到 Obsidian vault: {abs_path}", "vault_path": abs_path, "has_obsidian_config": has_obsidian_config, "total_notes": len(note_files), "total_size_kb": round(total_size / 1024, 1), "top_directories": top_dirs, "top_tags": top_tags, "status": "connected", }) def action_search(data: Dict[str, Any]) -> None: """在 Obsidian vault 中搜索笔记。 Args: data: 包含 query 的字典，可选 max_results。 """ query = data.get("query", "").strip() if not query: output_error("请提供搜索关键词（query）", code="VALIDATION_ERROR") return max_results = data.get("max_results", 20) index_data = _load_index() if not index_data.get("notes"): output_error( "Obsidian 索引为空，请先执行 index 操作构建索引", code="NO_INDEX", ) return results = _search_notes(query, index_data, max_results) output_success({ "query": query, "source": "obsidian", "total": len(results), "vault_path": index_data.get("vault_path", ""), "results": results, }) def action_index(data: Dict[str, Any]) -> None: """构建 Obsidian vault 的完整索引。 Args: data: 包含 vault_path 的字典。 """ vault_path = data.get("vault_path", "").strip() if not vault_path: vault_path = os.environ.get("KM_OBSIDIAN_VAULT_PATH", "").strip() if not vault_path: output_error( "请提供 Obsidian vault 路径（vault_path）或设置 KM_OBSIDIAN_VAULT_PATH 环境变量", code="VALIDATION_ERROR", ) return abs_path = os.path.abspath(vault_path) if not os.path.isdir(abs_path): output_error(f"Vault 路径不存在: {abs_path}", code="PATH_NOT_FOUND") return # 构建索引 index_data = _build_index(abs_path) _save_index(index_data) # 统计信息 total_tags = len(index_data.get("tag_index", {})) total_backlinks = sum( len(sources) for sources in index_data.get("backlink_graph", {}).values() ) output_success({ "message": f"索引构建完成: {index_data['total_notes']} 篇笔记", "vault_path": abs_path, "total_notes": index_data["total_notes"], "skipped": index_data.get("skipped", 0), "total_tags": total_tags, "total_backlinks": total_backlinks, "last_sync": index_data["last_sync"], }) def action_list_notes(data: Optional[Dict[str, Any]] = None) -> None: """列出已索引的 Obsidian 笔记。""" index_data = _load_index() notes = index_data.get("notes", {}) if not notes: output_error("Obsidian 索引为空，请先执行 index 操作", code="NO_INDEX") return note_list = [] for note_id, note in notes.items(): note_list.append({ "id": note_id, "title": note.get("title", ""), "rel_path": note.get("rel_path", ""), "tags": note.get("tags", []), "token_count": note.get("token_count", 0), "modified_at": note.get("modified_at", ""), "size": note.get("size", 0), "wikilinks_count": len(note.get("wikilinks", [])), }) # 按修改时间倒序 note_list.sort(key=lambda n: n.get("modified_at", ""), reverse=True) # 标签统计 tag_index = index_data.get("tag_index", {}) top_tags = sorted(tag_index.items(), key=lambda x: len(x[1]), reverse=True)[:10] output_success({ "vault_path": index_data.get("vault_path", ""), "total_notes": len(note_list), "last_sync": index_data.get("last_sync", ""), "top_tags": [{"tag": t, "note_count": len(ids)} for t, ids in top_tags], "notes": note_list, }) def action_sync(data: Optional[Dict[str, Any]] = None) -> None: """增量同步：重新索引自上次同步以来变化的文件。通过比较文件的 mtime 来检测变化。 """ index_data = _load_index() vault_path = index_data.get("vault_path", "") if not vault_path: # 尝试从环境变量读取 vault_path = os.environ.get("KM_OBSIDIAN_VAULT_PATH", "").strip() if not vault_path or not os.path.isdir(vault_path): output_error( "未找到已连接的 vault，请先执行 connect 或 index 操作", code="NO_VAULT", ) return notes = index_data.get("notes", {}) # 扫描当前 vault 文件 current_files = _scan_vault(vault_path) current_paths = {os.path.abspath(f) for f in current_files} # 已索引的文件路径映射 indexed_paths = {} for note_id, note in notes.items(): fp = note.get("filepath", "") if fp: indexed_paths[fp] = note_id added = 0 updated = 0 removed = 0 # 检查新增和更新的文件 for filepath in current_files: abs_fp = os.path.abspath(filepath) if abs_fp in indexed_paths: # 已索引，检查是否更新 note_id = indexed_paths[abs_fp] old_mtime = notes[note_id].get("mtime", 0) try: current_mtime = os.path.getmtime(abs_fp) except OSError: continue if current_mtime > old_mtime: # 文件已更新，重新解析 parsed = _parse_note(abs_fp, vault_path) if parsed: notes[note_id].update({ "title": parsed["title"], "content_preview": truncate_text(parsed["content"], 500), "frontmatter": parsed["frontmatter"], "tags": parsed["tags"], "wikilinks": parsed["wikilinks"], "tokens_sample": parsed["tokens"][:100], "token_count": parsed["token_count"], "modified_at": parsed["modified_at"], "size": parsed["size"], "mtime": current_mtime, }) updated += 1 else: # 新文件，添加索引 parsed = _parse_note(abs_fp, vault_path) if parsed: note_id = generate_id("ON") notes[note_id] = { "filepath": abs_fp, "rel_path": parsed["rel_path"], "title": parsed["title"], "content_preview": truncate_text(parsed["content"], 500), "frontmatter": parsed["frontmatter"], "tags": parsed["tags"], "wikilinks": parsed["wikilinks"], "tokens_sample": parsed["tokens"][:100], "token_count": parsed["token_count"], "modified_at": parsed["modified_at"], "size": parsed["size"], "mtime": os.path.getmtime(abs_fp) if os.path.exists(abs_fp) else 0, } added += 1 # 检查已删除的文件 ids_to_remove = [] for note_id, note in notes.items(): fp = note.get("filepath", "") if fp and fp not in current_paths: ids_to_remove.append(note_id) for note_id in ids_to_remove: del notes[note_id] removed += 1 # 重建反向链接图和标签索引 backlink_graph = {} tag_index = {} for note_id, note in notes.items(): title = note.get("title", "") for link_target in note.get("wikilinks", []): if link_target not in backlink_graph: backlink_graph[link_target] = [] backlink_graph[link_target].append(title) for tag in note.get("tags", []): if tag not in tag_index: tag_index[tag] = [] tag_index[tag].append(note_id) index_data["notes"] = notes index_data["backlink_graph"] = backlink_graph index_data["tag_index"] = tag_index index_data["last_sync"] = now_iso() index_data["total_notes"] = len(notes) _save_index(index_data) output_success({ "message": f"同步完成: 新增 {added}, 更新 {updated}, 删除 {removed}", "added": added, "updated": updated, "removed": removed, "total_notes": len(notes), "last_sync": index_data["last_sync"], }) # ============================================================ # 公开 API（供其他模块调用） # ============================================================ def search_obsidian(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 Obsidian vault（供 source_searcher 调用）。 Args: query: 搜索查询。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ index_data = _load_index() if not index_data.get("notes"): return [] return _search_notes(query, index_data, max_results) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("knowledge-mesh Obsidian 知识库集成") args = parser.parse_args() action = args.action.lower() try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "connect": lambda: action_connect(data or {}), "search": lambda: action_search(data or {}), "index": lambda: action_index(data or {}), "list-notes": lambda: action_list_notes(data), "sync": lambda: action_sync(data), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(actions.keys()) output_error(f"未知操作: {action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/source_searcher.py #!/usr/bin/env python3 """ knowledge-mesh 知识源搜索模块统一搜索 GitHub Discussions/Issues、Stack Overflow、Discord、 Confluence、Notion、Slack 等平台，返回标准化搜索结果。用法: python3 source_searcher.py --action search --data '{"query":"python async"}' python3 source_searcher.py --action search-source --data '{"query":"fastapi","source":"github"}' python3 source_searcher.py --action list-sources python3 source_searcher.py --action test-source --data '{"source":"github"}' """ import json import os import re import sys import urllib.request import urllib.error import urllib.parse from datetime import datetime from typing import Any, Dict, List, Optional from utils import ( SUPPORTED_SOURCES, SOURCE_DISPLAY_NAMES, SOURCE_ENV_KEYS, check_subscription, check_search_quota, increment_search_count, clean_html, days_ago, generate_id, load_input_data, now_iso, output_error, output_success, parse_common_args, parse_iso_datetime, read_json_file, truncate_text, write_json_file, get_data_file, ) # ============================================================ # 常量 # ============================================================ # 搜索配置文件 SOURCES_CONFIG_FILE = "sources_config.json" # 默认请求超时秒数 REQUEST_TIMEOUT = 15 # 各平台 API 基础地址 GITHUB_API_BASE = "https://api.github.com" STACKOVERFLOW_API_BASE = "https://api.stackexchange.com/2.3" DISCORD_API_BASE = "https://discord.com/api/v10" NOTION_API_BASE = "https://api.notion.com/v1" BAIDU_SEARCH_API = "https://www.baidu.com/s" # ============================================================ # HTTP 请求辅助 # ============================================================ def _http_get(url: str, headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]: """发送 HTTP GET 请求并返回 JSON 响应。 Args: url: 请求地址。 headers: 可选的请求头。 Returns: 解析后的 JSON 字典。 Raises: RuntimeError: 请求失败时抛出。 """ req = urllib.request.Request(url, method="GET") req.add_header("User-Agent", "knowledge-mesh/1.0") req.add_header("Accept", "application/json") if headers: for k, v in headers.items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp: body = resp.read().decode("utf-8") return json.loads(body) except urllib.error.HTTPError as e: raise RuntimeError(f"HTTP 请求失败: {e.code} {e.reason}") except urllib.error.URLError as e: raise RuntimeError(f"网络请求失败: {e.reason}") except json.JSONDecodeError: raise RuntimeError("响应解析失败：非有效 JSON") def _http_post(url: str, payload: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]: """发送 HTTP POST 请求并返回 JSON 响应。 Args: url: 请求地址。 payload: 请求体数据。 headers: 可选的请求头。 Returns: 解析后的 JSON 字典。 Raises: RuntimeError: 请求失败时抛出。 """ body_bytes = json.dumps(payload).encode("utf-8") req = urllib.request.Request(url, data=body_bytes, method="POST") req.add_header("User-Agent", "knowledge-mesh/1.0") req.add_header("Content-Type", "application/json") req.add_header("Accept", "application/json") if headers: for k, v in headers.items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp: resp_body = resp.read().decode("utf-8") return json.loads(resp_body) except urllib.error.HTTPError as e: raise RuntimeError(f"HTTP 请求失败: {e.code} {e.reason}") except urllib.error.URLError as e: raise RuntimeError(f"网络请求失败: {e.reason}") except json.JSONDecodeError: raise RuntimeError("响应解析失败：非有效 JSON") # ============================================================ # 统一搜索结果格式 # ============================================================ def _make_result( source: str, title: str, url: str, snippet: str, author: str = "", created_at: str = "", score: float = 0.0, tags: Optional[List[str]] = None, ) -> Dict[str, Any]: """构造标准化搜索结果条目。 Args: source: 来源平台名称。 title: 结果标题。 url: 结果链接。 snippet: 内容摘要。 author: 作者名。 created_at: 创建时间。 score: 相关性分数。 tags: 标签列表。 Returns: 标准化的搜索结果字典。 """ return { "id": generate_id("SR"), "source": source, "title": title, "url": url, "snippet": truncate_text(snippet, 300), "author": author, "created_at": created_at, "score": score, "tags": tags or [], } # ============================================================ # GitHub 搜索适配器 # ============================================================ def _search_github(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 GitHub Issues 和 Discussions。使用 GitHub Search API /search/issues 接口，支持搜索 Issues 和 Pull Requests（Discussions 需要 GraphQL）。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ token = os.environ.get("KM_GITHUB_TOKEN", "") headers = {} if token: headers["Authorization"] = f"token {token}" # 搜索 Issues 和 Discussions encoded_query = urllib.parse.quote(query) url = f"{GITHUB_API_BASE}/search/issues?q={encoded_query}&per_page={max_results}&sort=relevance" try: data = _http_get(url, headers=headers) except RuntimeError as e: return [_make_result("github", f"GitHub 搜索失败: {e}", "", str(e))] results = [] items = data.get("items", []) for item in items[:max_results]: title = item.get("title", "") html_url = item.get("html_url", "") body = item.get("body", "") or "" user = item.get("user", {}) author = user.get("login", "") if user else "" created = item.get("created_at", "") score_val = item.get("score", 0.0) labels = [lb.get("name", "") for lb in item.get("labels", [])] results.append(_make_result( source="github", title=title, url=html_url, snippet=clean_html(body), author=author, created_at=created, score=float(score_val), tags=labels, )) return results # ============================================================ # Stack Overflow 搜索适配器 # ============================================================ def _search_stackoverflow(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 Stack Overflow 问题。使用 Stack Exchange API v2.3 /search/advanced 接口。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ api_key = os.environ.get("KM_STACKOVERFLOW_KEY", "") params = { "order": "desc", "sort": "relevance", "q": query, "site": "stackoverflow", "pagesize": str(min(max_results, 30)), "filter": "withbody", } if api_key: params["key"] = api_key query_string = urllib.parse.urlencode(params) url = f"{STACKOVERFLOW_API_BASE}/search/advanced?{query_string}" try: data = _http_get(url) except RuntimeError as e: return [_make_result("stackoverflow", f"Stack Overflow 搜索失败: {e}", "", str(e))] results = [] items = data.get("items", []) for item in items[:max_results]: title = item.get("title", "") link = item.get("link", "") body = clean_html(item.get("body", "") or "") owner = item.get("owner", {}) author = owner.get("display_name", "") if owner else "" # Stack Overflow 时间戳为 Unix 秒 creation_date = item.get("creation_date", 0) created_at = "" if creation_date: try: created_at = datetime.utcfromtimestamp(creation_date).strftime("%Y-%m-%dT%H:%M:%S") except (OSError, ValueError): created_at = "" score_val = float(item.get("score", 0)) tags = item.get("tags", []) results.append(_make_result( source="stackoverflow", title=clean_html(title), url=link, snippet=body, author=author, created_at=created_at, score=score_val, tags=tags, )) return results # ============================================================ # Discord 搜索适配器 # ============================================================ def _search_discord(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 Discord 频道消息。使用 Discord Bot API /channels/{id}/messages 接口。需要设置 KM_DISCORD_BOT_TOKEN 和 KM_DISCORD_CHANNEL_ID。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ token = os.environ.get("KM_DISCORD_BOT_TOKEN", "") channel_id = os.environ.get("KM_DISCORD_CHANNEL_ID", "") if not token: return [_make_result("discord", "Discord 未配置", "", "请设置 KM_DISCORD_BOT_TOKEN 环境变量")] if not channel_id: return [_make_result("discord", "Discord 频道未配置", "", "请设置 KM_DISCORD_CHANNEL_ID 环境变量")] headers = { "Authorization": f"Bot {token}", } url = f"{DISCORD_API_BASE}/channels/{channel_id}/messages?limit={min(max_results, 100)}" try: messages = _http_get(url, headers=headers) except RuntimeError as e: return [_make_result("discord", f"Discord 搜索失败: {e}", "", str(e))] if not isinstance(messages, list): return [] # 在客户端过滤匹配消息 query_lower = query.lower() keywords = query_lower.split() results = [] for msg in messages: content = msg.get("content", "") if not content: continue content_lower = content.lower() # 检查是否包含任一关键词 if not any(kw in content_lower for kw in keywords): continue msg_id = msg.get("id", "") author_info = msg.get("author", {}) author = author_info.get("username", "") if author_info else "" timestamp = msg.get("timestamp", "") msg_url = f"https://discord.com/channels/{channel_id}/{msg_id}" # 简单相关性评分：匹配关键词数量 match_count = sum(1 for kw in keywords if kw in content_lower) score_val = match_count / max(len(keywords), 1) results.append(_make_result( source="discord", title=truncate_text(content, 80), url=msg_url, snippet=content, author=author, created_at=timestamp, score=score_val, )) if len(results) >= max_results: break return results # ============================================================ # Confluence 搜索适配器 # ============================================================ def _search_confluence(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 Confluence 内容。使用 Confluence REST API /wiki/rest/api/content/search 接口。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ base_url = os.environ.get("KM_CONFLUENCE_URL", "") token = os.environ.get("KM_CONFLUENCE_TOKEN", "") if not base_url: return [_make_result("confluence", "Confluence 未配置", "", "请设置 KM_CONFLUENCE_URL 环境变量")] if not token: return [_make_result("confluence", "Confluence 未认证", "", "请设置 KM_CONFLUENCE_TOKEN 环境变量")] # 构造 CQL 查询 cql = urllib.parse.quote(f'text ~ "{query}"') url = f"{base_url.rstrip('/')}/wiki/rest/api/content/search?cql={cql}&limit={max_results}&expand=body.view,version" headers = { "Authorization": f"Bearer {token}", } try: data = _http_get(url, headers=headers) except RuntimeError as e: return [_make_result("confluence", f"Confluence 搜索失败: {e}", "", str(e))] results = [] items = data.get("results", []) for item in items[:max_results]: title = item.get("title", "") content_id = item.get("id", "") page_url = f"{base_url.rstrip('/')}/wiki{item.get('_links', {}).get('webui', '')}" # 提取正文摘要 body_view = item.get("body", {}).get("view", {}).get("value", "") snippet = clean_html(body_view) # 版本信息中的作者和时间 version = item.get("version", {}) author = "" created_at = "" if version: by_info = version.get("by", {}) author = by_info.get("displayName", "") if by_info else "" created_at = version.get("when", "") results.append(_make_result( source="confluence", title=title, url=page_url, snippet=snippet, author=author, created_at=created_at, score=1.0, tags=[item.get("type", "page")], )) return results # ============================================================ # Notion 搜索适配器 # ============================================================ def _search_notion(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 Notion 页面和数据库。使用 Notion API POST /v1/search 接口。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ token = os.environ.get("KM_NOTION_TOKEN", "") if not token: return [_make_result("notion", "Notion 未配置", "", "请设置 KM_NOTION_TOKEN 环境变量")] headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", } payload = { "query": query, "page_size": min(max_results, 100), } url = f"{NOTION_API_BASE}/search" try: data = _http_post(url, payload, headers=headers) except RuntimeError as e: return [_make_result("notion", f"Notion 搜索失败: {e}", "", str(e))] results = [] items = data.get("results", []) for item in items[:max_results]: obj_type = item.get("object", "") page_id = item.get("id", "") page_url = item.get("url", "") or f"https://notion.so/{page_id.replace('-', '')}" # 提取标题 title = "" properties = item.get("properties", {}) if "title" in properties: title_arr = properties["title"].get("title", []) if title_arr: title = title_arr[0].get("plain_text", "") elif "Name" in properties: name_info = properties["Name"] if name_info.get("type") == "title": title_arr = name_info.get("title", []) if title_arr: title = title_arr[0].get("plain_text", "") if not title: title = f"Notion {obj_type} {page_id[:8]}" # 创建时间和作者 created_at = item.get("created_time", "") created_by = item.get("created_by", {}) author = "" if created_by: author = created_by.get("name", "") or created_by.get("id", "") results.append(_make_result( source="notion", title=title, url=page_url, snippet=f"Notion {obj_type}: {title}", author=author, created_at=created_at, score=1.0, tags=[obj_type], )) return results # ============================================================ # Slack 搜索适配器 # ============================================================ def _search_slack(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 Slack 消息。使用 Slack Web API search.messages 接口。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ token = os.environ.get("KM_SLACK_TOKEN", "") if not token: return [_make_result("slack", "Slack 未配置", "", "请设置 KM_SLACK_TOKEN 环境变量")] headers = { "Authorization": f"Bearer {token}", } encoded_query = urllib.parse.quote(query) url = f"https://slack.com/api/search.messages?query={encoded_query}&count={max_results}&sort=score" try: data = _http_get(url, headers=headers) except RuntimeError as e: return [_make_result("slack", f"Slack 搜索失败: {e}", "", str(e))] if not data.get("ok", False): error_msg = data.get("error", "未知错误") return [_make_result("slack", f"Slack API 错误: {error_msg}", "", error_msg)] results = [] messages_data = data.get("messages", {}) matches = messages_data.get("matches", []) for msg in matches[:max_results]: text = msg.get("text", "") permalink = msg.get("permalink", "") username = msg.get("username", "") ts = msg.get("ts", "") channel_info = msg.get("channel", {}) channel_name = channel_info.get("name", "") if isinstance(channel_info, dict) else "" # 将 Slack 时间戳转换为 ISO 格式 created_at = "" if ts: try: dt = datetime.utcfromtimestamp(float(ts)) created_at = dt.strftime("%Y-%m-%dT%H:%M:%S") except (ValueError, OSError): created_at = "" tags = [f"#{channel_name}"] if channel_name else [] results.append(_make_result( source="slack", title=truncate_text(text, 80), url=permalink, snippet=text, author=username, created_at=created_at, score=1.0, tags=tags, )) return results # ============================================================ # 百度搜索适配器 # ============================================================ def _search_baidu(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索百度，获取中文内容搜索结果。使用百度搜索 API 或 Web 端点，解析结果为统一格式。需要设置 KM_BAIDU_API_KEY 环境变量。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ api_key = os.environ.get("KM_BAIDU_API_KEY", "") # 使用百度开发者搜索 API encoded_query = urllib.parse.quote(query) params = { "wd": query, "rn": str(min(max_results, 50)), "ie": "utf-8", "oe": "utf-8", } if api_key: params["apikey"] = api_key query_string = urllib.parse.urlencode(params) url = f"https://api.baidu.com/json/custom/tongji?{query_string}" # 备用：使用百度搜索结果页面解析 headers = { "User-Agent": "knowledge-mesh/1.0", } if api_key: headers["Authorization"] = f"Bearer {api_key}" try: data = _http_get(url, headers=headers) except RuntimeError: # API 不可用时返回空结果 return [_make_result( "baidu", "百度搜索需要配置 API", "", "请设置 KM_BAIDU_API_KEY 环境变量以启用百度搜索", )] results = [] items = data.get("results", data.get("items", [])) if not isinstance(items, list): items = [] for item in items[:max_results]: title = item.get("title", "") link = item.get("url", item.get("link", "")) abstract = item.get("abstract", item.get("snippet", item.get("description", ""))) author = item.get("source", item.get("author", "")) created_at = item.get("date", item.get("created_at", "")) score_val = float(item.get("score", 0.5)) # 清理 HTML 标签 title = clean_html(title) abstract = clean_html(abstract) results.append(_make_result( source="baidu", title=title, url=link, snippet=abstract, author=author, created_at=created_at, score=score_val, tags=["baidu"], )) return results # ============================================================ # Obsidian 搜索适配器 # ============================================================ def _search_obsidian(query: str, max_results: int = 20) -> List[Dict[str, Any]]: """搜索 Obsidian 知识库。委托给 obsidian_connector 模块执行搜索。 Args: query: 搜索关键词。 max_results: 最大返回结果数。 Returns: 标准化搜索结果列表。 """ try: from obsidian_connector import search_obsidian results = search_obsidian(query, max_results=max_results) return results except ImportError: return [_make_result( "obsidian", "Obsidian 连接器未安装", "", "obsidian_connector 模块不可用", )] except Exception as e: return [_make_result( "obsidian", f"Obsidian 搜索失败: {e}", "", str(e), )] # ============================================================ # 搜索适配器路由 # ============================================================ _SOURCE_ADAPTERS = { "github": _search_github, "stackoverflow": _search_stackoverflow, "discord": _search_discord, "confluence": _search_confluence, "notion": _search_notion, "slack": _search_slack, "baidu": _search_baidu, "obsidian": _search_obsidian, } # 免费版可用的知识源 _FREE_SOURCES = {"github", "stackoverflow", "baidu", "obsidian"} def _get_available_sources() -> List[str]: """获取当前订阅等级下可用的知识源列表。 Returns: 可用知识源名称列表。 """ sub = check_subscription() features = sub.get("features", []) available = [] for source in SUPPORTED_SOURCES: feature_name = f"{source}_search" # basic_search 对应免费源 if source in _FREE_SOURCES or feature_name in features: available.append(source) return available def _get_configured_sources() -> List[str]: """获取已配置认证凭据的知识源列表。 Returns: 已配置的知识源名称列表。 """ configured = [] for source in SUPPORTED_SOURCES: env_key = SOURCE_ENV_KEYS.get(source, "") if env_key and os.environ.get(env_key, ""): configured.append(source) elif source == "stackoverflow": # Stack Overflow 不需要 API key 也能搜索（有速率限制） configured.append(source) return configured # ============================================================ # 操作实现 # ============================================================ def action_search(data: Dict[str, Any]) -> None: """统一搜索：查询所有已配置的可用知识源。 Args: data: 包含 query（搜索关键词）的字典，可选 max_results。 """ query = data.get("query", "").strip() if not query: output_error("请提供搜索关键词（query）", code="VALIDATION_ERROR") return # 检查搜索配额 if not check_search_quota(): return sub = check_subscription() max_results = min(data.get("max_results", 20), sub.get("max_results", 20)) available = _get_available_sources() configured = _get_configured_sources() # 取交集：既可用又已配置 sources_to_search = [s for s in available if s in configured] if not sources_to_search: output_error( "没有可用的知识源。请先配置至少一个知识源的 API 凭据。", code="NO_SOURCES", ) return all_results = [] source_stats = {} errors = [] for source in sources_to_search: adapter = _SOURCE_ADAPTERS.get(source) if not adapter: continue try: results = adapter(query, max_results=max_results) source_stats[source] = len(results) all_results.extend(results) except Exception as e: errors.append(f"{SOURCE_DISPLAY_NAMES.get(source, source)}: {e}") source_stats[source] = 0 # 按分数降序排列 all_results.sort(key=lambda r: r.get("score", 0), reverse=True) all_results = all_results[:max_results] # 递增搜索计数 increment_search_count() # 记录查询到自学习引擎 try: from learning_engine import record_query_data record_query_data(query, sources_to_search, source_stats) except (ImportError, Exception): pass # 自学习模块不可用时静默跳过 output_success({ "query": query, "total": len(all_results), "sources_searched": sources_to_search, "source_stats": source_stats, "results": all_results, "errors": errors if errors else None, }) def action_search_source(data: Dict[str, Any]) -> None: """搜索指定的单个知识源。 Args: data: 包含 query 和 source 的字典。 """ query = data.get("query", "").strip() source = data.get("source", "").strip().lower() if not query: output_error("请提供搜索关键词（query）", code="VALIDATION_ERROR") return if not source: output_error("请指定知识源（source）", code="VALIDATION_ERROR") return if source not in SUPPORTED_SOURCES: valid = "、".join(SUPPORTED_SOURCES) output_error(f"不支持的知识源: {source}，支持: {valid}", code="INVALID_SOURCE") return # 检查订阅权限 available = _get_available_sources() if source not in available: output_error( f"{SOURCE_DISPLAY_NAMES.get(source, source)} 搜索为付费版功能，请升级至付费版（¥129/月）。", code="SUBSCRIPTION_REQUIRED", ) return # 检查搜索配额 if not check_search_quota(): return sub = check_subscription() max_results = min(data.get("max_results", 20), sub.get("max_results", 20)) adapter = _SOURCE_ADAPTERS.get(source) if not adapter: output_error(f"知识源适配器不存在: {source}", code="ADAPTER_ERROR") return try: results = adapter(query, max_results=max_results) except Exception as e: output_error(f"搜索失败: {e}", code="SEARCH_ERROR") return # 递增搜索计数 increment_search_count() output_success({ "query": query, "source": source, "total": len(results), "results": results, }) def action_list_sources(data: Optional[Dict[str, Any]] = None) -> None: """列出所有支持的知识源及其配置状态。""" available = _get_available_sources() configured = _get_configured_sources() sub = check_subscription() sources_info = [] for source in SUPPORTED_SOURCES: env_key = SOURCE_ENV_KEYS.get(source, "") is_available = source in available is_configured = source in configured sources_info.append({ "name": source, "display_name": SOURCE_DISPLAY_NAMES.get(source, source), "env_key": env_key, "available": is_available, "configured": is_configured, "status": "ready" if (is_available and is_configured) else "not_configured" if is_available else "paid_only", }) output_success({ "subscription_tier": sub["tier"], "total_sources": len(SUPPORTED_SOURCES), "available_count": len(available), "configured_count": len(configured), "sources": sources_info, }) def action_test_source(data: Dict[str, Any]) -> None: """测试指定知识源的连接状态。 Args: data: 包含 source 的字典。 """ source = data.get("source", "").strip().lower() if not source: output_error("请指定知识源（source）", code="VALIDATION_ERROR") return if source not in SUPPORTED_SOURCES: valid = "、".join(SUPPORTED_SOURCES) output_error(f"不支持的知识源: {source}，支持: {valid}", code="INVALID_SOURCE") return env_key = SOURCE_ENV_KEYS.get(source, "") token = os.environ.get(env_key, "") if env_key else "" # 测试连接 test_result = { "source": source, "display_name": SOURCE_DISPLAY_NAMES.get(source, source), "env_key": env_key, "token_configured": bool(token), } if source == "github": if token: try: data_resp = _http_get(f"{GITHUB_API_BASE}/user", headers={"Authorization": f"token {token}"}) test_result["status"] = "connected" test_result["user"] = data_resp.get("login", "") except RuntimeError as e: test_result["status"] = "error" test_result["error"] = str(e) else: test_result["status"] = "no_token" test_result["message"] = "GitHub 搜索可在无 Token 下使用，但有速率限制" elif source == "stackoverflow": # Stack Overflow 不需要强制 API key test_result["status"] = "available" test_result["message"] = "Stack Overflow API 可直接访问（API key 可选）" elif source == "confluence": base_url = os.environ.get("KM_CONFLUENCE_URL", "") if not base_url: test_result["status"] = "not_configured" test_result["error"] = "请设置 KM_CONFLUENCE_URL 环境变量" elif not token: test_result["status"] = "no_token" test_result["error"] = "请设置 KM_CONFLUENCE_TOKEN 环境变量" else: test_result["status"] = "configured" test_result["base_url"] = base_url else: if token: test_result["status"] = "configured" else: test_result["status"] = "no_token" test_result["error"] = f"请设置 {env_key} 环境变量" output_success(test_result) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("knowledge-mesh 知识源搜索") args = parser.parse_args() action = args.action.lower() try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "search": lambda: action_search(data or {}), "search-source": lambda: action_search_source(data or {}), "list-sources": lambda: action_list_sources(data), "test-source": lambda: action_test_source(data or {}), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(actions.keys()) output_error(f"未知操作: {action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:references/api-endpoints.md # API 端点参考本文档列出 knowledge-mesh 支持的各知识源 API 端点和认证方式。 --- ## 1. GitHub | 项目 | 说明 | |------|------| | **API 基址** | `https://api.github.com` | | **搜索 Issues** | `GET /search/issues?q={query}&per_page={n}&sort=relevance` | | **搜索代码** | `GET /search/code?q={query}&per_page={n}` | | **搜索仓库** | `GET /search/repositories?q={query}&per_page={n}` | | **认证方式** | `Authorization: token {KM_GITHUB_TOKEN}` | | **速率限制** | 未认证 10 次/分钟，认证 30 次/分钟 | | **文档** | https://docs.github.com/en/rest/search | > 注意：GitHub Discussions 搜索需要 GraphQL API，当前使用 Issues 搜索作为替代。 --- ## 2. Stack Overflow | 项目 | 说明 | |------|------| | **API 基址** | `https://api.stackexchange.com/2.3` | | **高级搜索** | `GET /search/advanced?q={query}&site=stackoverflow&order=desc&sort=relevance` | | **按标签搜索** | `GET /search?tagged={tags}&site=stackoverflow` | | **获取问题详情** | `GET /questions/{ids}?site=stackoverflow&filter=withbody` | | **认证方式** | 查询参数 `key={KM_STACKOVERFLOW_KEY}` | | **速率限制** | 无 Key 300 次/天，有 Key 10,000 次/天 | | **文档** | https://api.stackexchange.com/docs | > 注意：响应数据经过 gzip 压缩，urllib 会自动处理。 --- ## 3. Discord | 项目 | 说明 | |------|------| | **API 基址** | `https://discord.com/api/v10` | | **获取频道消息** | `GET /channels/{channel_id}/messages?limit={n}` | | **搜索消息** | 需在客户端侧过滤（Discord API 不提供公开搜索端点） | | **认证方式** | `Authorization: Bot {KM_DISCORD_BOT_TOKEN}` | | **速率限制** | 按端点不同，通常 5 次/秒 | | **所需权限** | `READ_MESSAGE_HISTORY`, `VIEW_CHANNEL` | | **文档** | https://discord.com/developers/docs | > 注意：需要额外设置 `KM_DISCORD_CHANNEL_ID` 环境变量指定搜索频道。 --- ## 4. Confluence | 项目 | 说明 | |------|------| | **API 基址** | `{KM_CONFLUENCE_URL}/wiki/rest/api` | | **CQL 搜索** | `GET /content/search?cql=text~"{query}"&limit={n}&expand=body.view,version` | | **获取页面** | `GET /content/{id}?expand=body.view` | | **认证方式** | `Authorization: Bearer {KM_CONFLUENCE_TOKEN}` | | **替代认证** | Basic Auth: `email:api_token` (Base64 编码) | | **速率限制** | 因实例而异，通常无严格限制 | | **文档** | https://developer.atlassian.com/cloud/confluence/rest/ | > 注意：CQL（Confluence Query Language）支持丰富的搜索语法，如 `type=page AND space=DEV AND text~"keyword"`。 --- ## 5. Notion | 项目 | 说明 | |------|------| | **API 基址** | `https://api.notion.com/v1` | | **搜索** | `POST /search` (Body: `{"query":"...","page_size":20}`) | | **获取页面** | `GET /pages/{page_id}` | | **获取块内容** | `GET /blocks/{block_id}/children` | | **认证方式** | `Authorization: Bearer {KM_NOTION_TOKEN}` | | **必需请求头** | `Notion-Version: 2022-06-28` | | **速率限制** | 3 次/秒 | | **文档** | https://developers.notion.com/reference | > 注意：Notion Integration 需要在 Notion 设置中创建，并授权访问相应的页面/数据库。 --- ## 6. Slack | 项目 | 说明 | |------|------| | **API 基址** | `https://slack.com/api` | | **搜索消息** | `GET /search.messages?query={query}&count={n}&sort=score` | | **搜索文件** | `GET /search.files?query={query}&count={n}` | | **认证方式** | `Authorization: Bearer {KM_SLACK_TOKEN}` | | **所需权限** | `search:read` | | **速率限制** | Tier 2: 20 次/分钟 | | **文档** | https://api.slack.com/methods/search.messages | > 注意：Slack Token 需要 `search:read` scope，且 Bot 需要被邀请到相应频道。 --- ## 通用注意事项 1. **HTTPS 强制**：所有 API 请求必须使用 HTTPS 协议。 2. **超时设置**：默认请求超时 15 秒，可通过代码调整。 3. **错误重试**：遇到 429（速率限制）时应等待 Retry-After 头指定的时间后重试。 4. **User-Agent**：所有请求携带 `User-Agent: knowledge-mesh/1.0` 标识。 5. **响应格式**：所有平台返回 JSON 格式数据，使用 `Accept: application/json` 请求头。 FILE:references/search-syntax.md # 搜索语法指南本文档说明 knowledge-mesh 的搜索查询语法和各平台特有的搜索技巧。 --- ## 基本搜索语法 ### 关键词搜索直接输入关键词，空格分隔表示 AND 关系： ``` python async fastapi ``` 搜索同时包含 "python"、"async" 和 "fastapi" 的内容。 ### 精确匹配使用双引号包裹短语进行精确匹配： ``` "connection pool" timeout ``` 搜索包含完整短语 "connection pool" 且含 "timeout" 的内容。 ### 排除关键词使用减号排除不需要的结果（部分平台支持）： ``` python web framework -django ``` --- ## 平台特有语法 ### GitHub ``` # 按仓库搜索 repo:owner/name keyword # 按语言过滤 language:python async # 按标签过滤 label:bug memory leak # 按状态过滤 state:open performance issue # 组合搜索 repo:fastapi/fastapi label:question websocket ``` ### Stack Overflow ``` # 按标签搜索 [python] [async] how to # 搜索已采纳答案 is:answer accepted:yes connection pool # 按分数过滤 score:10 python decorator # 按时间范围 created:2026-01.. python 3.12 ``` ### Confluence (CQL) ``` # 按空间搜索 space=DEV AND text~"api design" # 按类型过滤 type=page AND title~"architecture" # 按标签 label=backend AND text~"microservice" # 按创建者 creator=currentUser() AND text~"meeting notes" ``` ### Notion ``` # 基本关键词（Notion API 仅支持简单文本搜索） project roadmap # 建议使用具体的页面或数据库标题关键词 sprint planning Q1 2026 ``` ### Slack ``` # 按频道搜索 in:#engineering python deployment # 按用户搜索 from:@username production issue # 按时间范围 after:2026-03-01 before:2026-03-19 release # 搜索文件 has:link architecture document ``` --- ## 搜索技巧 1. **使用具体关键词**：避免过于宽泛的搜索词，如 "问题" 或 "error"，改用具体的错误信息或技术术语。 2. **善用平台优势**： - 代码问题 → 优先搜索 Stack Overflow 和 GitHub - 团队知识 → 优先搜索 Confluence 和 Notion - 实时讨论 → 优先搜索 Discord 和 Slack 3. **迭代优化**：首次搜索结果不理想时，根据返回的标签和关键词调整查询。 4. **结合多个来源**：跨平台搜索可以获得更全面的信息，免费版支持同时搜索 GitHub 和 Stack Overflow。 5. **利用标签过滤**：搜索结果中的标签可以帮助你发现相关主题和更精确的搜索词。 --- ## 搜索示例 | 场景 | 推荐查询 | 建议平台 | |------|----------|----------| | Python 异步编程 | `python asyncio await best practices` | Stack Overflow, GitHub | | React 性能优化 | `react performance optimization memo` | Stack Overflow, GitHub | | 团队 API 设计规范 | `api design guidelines rest` | Confluence, Notion | | 部署问题排查 | `deployment failed kubernetes pod` | Slack, Discord | | 开源项目选型 | `python web framework comparison 2026` | GitHub, Stack Overflow |

知识网格 / Knowledge Mesh

Comments (0)