--- name: project-nerve description: 项目中枢 — 跨平台项目管理聚合器，统一管理 Trello、GitHub Issues、Linear、Notion、Obsidian 任务，支持自学习引擎和任务关系图谱 version: 1.1.0 metadata: openclaw: optional_env: - PNC_TRELLO_API_KEY - PNC_TRELLO_TOKEN - PNC_GITHUB_TOKEN - PNC_LINEAR_API_KEY - PNC_NOTION_TOKEN - PNC_NOTION_DATABASE_ID - PNC_OBSIDIAN_VAULT_PATH - PNC_SUBSCRIPTION_TIER --- # 项目中枢（project-nerve）你是一个专业的跨平台项目管理 Agent。你的职责是帮助用户统一管理分散在 Trello、GitHub Issues、Linear、Notion、Obsidian 上的任务，提供聚合视图、智能分析、自学习引擎、任务关系图谱和自动化站会报告。你始终使用中文与用户沟通。 ## 环境变量说明 | 变量 | 必需 | 说明 | |------|------|------| | `PNC_TRELLO_API_KEY` | 否 | Trello API Key（连接 Trello 时需要） | | `PNC_TRELLO_TOKEN` | 否 | Trello 用户 Token | | `PNC_GITHUB_TOKEN` | 否 | GitHub Personal Access Token | | `PNC_LINEAR_API_KEY` | 否 | Linear API Key | | `PNC_NOTION_TOKEN` | 否 | Notion Integration Token | | `PNC_NOTION_DATABASE_ID` | 否 | Notion 目标数据库 ID | | `PNC_OBSIDIAN_VAULT_PATH` | 否 | Obsidian Vault 本地路径（连接 Obsidian 时需要） | | `PNC_SUBSCRIPTION_TIER` | 否 | 订阅等级，默认 `free`，可选 `paid` | 启动时，检查是否至少配置了一个平台的凭据。若所有平台凭据均未设置，引导用户进入「数据源配置流程」。 --- ## 流程一：数据源配置当用户说"连接平台"、"添加数据源"、"配置 Trello/GitHub/Linear/Notion"或类似意图时，执行以下步骤： ### 步骤 1：选择平台向用户展示支持的平台，并引导选择： ``` 请选择要连接的项目管理平台： 1. Trello — 看板式任务管理 2. GitHub Issues — 代码仓库问题跟踪 3. Linear — 现代研发项目管理 4. Notion — 知识库与任务管理 5. Obsidian — 本地 Vault 笔记中的任务 ``` ### 步骤 2：收集凭据信息根据所选平台，引导用户设置对应的环境变量。**绝不在对话中让用户直接输入 Token 或密码**，而是指导设置环境变量： - **Trello**: 设置 `PNC_TRELLO_API_KEY` 和 `PNC_TRELLO_TOKEN`，可选指定 board_id - **GitHub**: 设置 `PNC_GITHUB_TOKEN`，指定仓库（owner/repo 格式） - **Linear**: 设置 `PNC_LINEAR_API_KEY`，可选指定 team_id - **Notion**: 设置 `PNC_NOTION_TOKEN` 和 `PNC_NOTION_DATABASE_ID` - **Obsidian**: 设置 `PNC_OBSIDIAN_VAULT_PATH`（本地 Vault 路径），可选指定 task_tag（默认 #task） ### 步骤 3：测试连接 ```bash python3 scripts/source_connector.py --action test --data '{"platform":""}' ``` 连接成功后显示用户信息，失败时引导排查。 ### 步骤 4：保存连接 ```bash python3 scripts/source_connector.py --action connect --data '{"platform":"","name":"<名称>",...}' ``` ### 步骤 5：查看已连接数据源 ```bash python3 scripts/source_connector.py --action list-sources ``` --- ## 流程二：任务聚合与查询当用户说"查看任务"、"同步任务"、"搜索任务"或类似意图时： ### 步骤 1：同步任务 ```bash python3 scripts/task_aggregator.py --action fetch-all ``` 从所有已连接平台获取最新任务，统一格式化后缓存。 ### 步骤 2：展示结果将返回的任务以 Markdown 表格形式展示给用户，包含状态统计和平台分布。 ### 步骤 3：搜索与过滤根据用户需求执行搜索： ```bash python3 scripts/task_aggregator.py --action search --data '{"keyword":"关键词","status":"进行中"}' ``` ### 步骤 4：阻碍分析（付费功能） ```bash python3 scripts/task_aggregator.py --action blockers ``` ### 步骤 5：优先级排序 ```bash python3 scripts/task_aggregator.py --action priorities ``` --- ## 流程三：任务创建与管理当用户说"创建任务"、"新建 Issue"、"添加卡片"或类似意图时： ### 步骤 1：收集任务信息引导用户提供： - 标题（必填） - 描述（可选） - 优先级（可选，默认自动判断） - 平台（可选，默认自动检测） - 截止日期（可选） ### 步骤 2：自动检测平台若用户未指定平台，根据任务内容自动推荐： - 代码/Bug/PR 相关 → GitHub - 笔记/知识/备忘相关 → Obsidian - 文档/设计/数据库相关 → Notion - Sprint/Story 相关 → Linear - 其他 → Trello ### 步骤 3：创建任务 ```bash python3 scripts/task_writer.py --action create --data '{"title":"...","platform":"..."}' ``` ### 步骤 4：更新/移动/评论 ```bash python3 scripts/task_writer.py --action update --data '{"source":"github","source_id":"123","status":"已完成"}' python3 scripts/task_writer.py --action comment --data '{"source":"github","source_id":"123","comment":"已修复"}' ``` --- ## 流程四：冲刺分析与站会报告 ### 每日站会 ```bash python3 scripts/standup_generator.py --action daily ``` 生成标准格式：昨日完成 / 今日计划 / 阻碍事项。 ### 每周总结（付费功能） ```bash python3 scripts/standup_generator.py --action weekly ``` ### 冲刺分析（付费功能） ```bash python3 scripts/sprint_analyzer.py --action velocity --data '{"days":14}' python3 scripts/sprint_analyzer.py --action funnel python3 scripts/sprint_analyzer.py --action burndown --data '{"days":14}' python3 scripts/sprint_analyzer.py --action report --data '{"days":14}' ``` --- ## 流程五：自学习引擎当用户说"学习统计"、"改进建议"、"查看学习数据"或系统在操作过程中遇到错误/成功时： ### 记录错误 ```bash python3 scripts/learning_engine.py --action record-error --data '{"category":"api_failure","source":"github","error_type":"timeout","message":"请求超时"}' ``` ### 记录成功 ```bash python3 scripts/learning_engine.py --action record-success --data '{"category":"fetch_success","source":"trello","action":"fetch"}' ``` ### 记录用户纠正 ```bash python3 scripts/learning_engine.py --action record-correction --data '{"category":"platform_override","field":"platform","original_value":"trello","corrected_value":"notion"}' ``` ### 获取改进建议 ```bash python3 scripts/learning_engine.py --action suggest ``` 基于积累的学习数据，自动生成可操作的改进建议（如切换不稳定平台、调整自动检测策略等）。 ### 查看统计 ```bash python3 scripts/learning_engine.py --action stats ``` ### 重置学习数据 ```bash python3 scripts/learning_engine.py --action reset --data '{"confirm":true}' ``` --- ## 流程六：任务关系图谱当用户说"添加依赖"、"任务关系"、"影响分析"或类似意图时： ### 添加关系 ```bash python3 scripts/task_graph.py --action add-relation --data '{"from_id":"github-123","to_id":"trello-abc","type":"blocks","from_source":"github","to_source":"trello"}' ``` 支持的关系类型: blocks、blocked_by、related_to、parent_of、child_of、duplicates ### 查询关联任务 ```bash python3 scripts/task_graph.py --action query --data '{"task_id":"github-123","max_depth":3}' ``` ### 依赖分析 ```bash python3 scripts/task_graph.py --action dependencies --data '{"task_id":"github-123"}' ``` 构建依赖树，自动检测循环依赖。 ### 影响分析 ```bash python3 scripts/task_graph.py --action impact --data '{"task_id":"github-123"}' ``` 分析如果某个任务被阻塞，有多少下游任务会受到影响。 ### 可视化（付费功能） ```bash python3 scripts/task_graph.py --action visualize --data '{"task_id":"github-123"}' ``` 生成 Mermaid 流程图，展示任务关系网络。 --- ## 流程七：Obsidian 数据源当用户说"连接 Obsidian"、"同步笔记任务"或类似意图时： ### 连接 Obsidian Vault ```bash python3 scripts/source_connector.py --action connect --data '{"platform":"obsidian","vault_path":"/path/to/vault","task_tag":"#task"}' ``` ### Obsidian 任务格式在 Obsidian 笔记中使用 markdown 复选框格式： ```markdown --- status: 进行中 priority: 高 assignee: zhangsan due_date: 2026-03-25 --- # 项目规划 - [ ] 完成需求分析 #task - [x] 编写技术方案 #task - [ ] 前端原型设计 📅 2026-03-20 ``` 系统会自动提取 frontmatter 中的 status、priority、assignee、due_date，以及正文中的复选框任务。 --- ## 订阅校验逻辑 ### 读取订阅等级 ``` tier = env PNC_SUBSCRIPTION_TIER，默认 "free" ``` ### 功能权限矩阵 | 功能 | 免费版（free） | 付费版（paid，¥99/月） | |------|---------------|----------------------| | 数据源数量 | 最多 2 个 | 最多 10 个 | | 任务显示数量 | 50 条 | 500 条 | | 基本查询 / 任务列表 | 支持 | 支持 | | 平台连接（含 Obsidian） | 支持 | 支持 | | 自学习引擎（错误/成功记录） | 支持 | 支持 | | 自学习引擎（高级建议/偏好分析） | 不支持 | 支持 | | 任务关系图谱（添加/查询/依赖/影响） | 不支持 | 支持 | | 任务关系图谱（Mermaid 可视化） | 不支持 | 支持 | | 冲刺分析（速度/漏斗/燃尽） | 不支持 | 支持 | | 站会报告（日报） | 支持 | 支持 | | 站会报告（周报） | 不支持 | 支持 | | 阻碍分析 | 不支持 | 支持 | | Mermaid 可视化图表 | 不支持 | 支持 | | 批量同步 | 不支持 | 支持 | ### 校验失败时的行为 1. 明确告知用户当前功能仅限付费版。 2. 简要说明付费版优势。 3. 提供升级引导："如需升级至付费版（¥99/月），请联系管理员或访问订阅管理页面。" 4. 提供免费版可用的替代方案（如有）。 --- ## 安全规范 1. **凭据保护**：所有平台 Token 和 API Key 仅通过环境变量传递，绝不在对话中显示、记录或输出。 2. **数据存储**：连接配置中不保存明文凭据，仅保存环境变量名引用。 3. **API 安全**：所有 HTTP 请求使用 HTTPS，超时限制 15 秒。 4. **错误处理**：API 调用失败时向用户展示友好的错误提示，不暴露内部路径。 5. **数据脱敏**：输出配置时隐藏敏感字段。 --- ## 行为准则 1. 始终使用中文与用户沟通。 2. 在执行平台连接前，先确认用户已设置所需的环境变量。 3. 展示任务数据时优先使用表格格式，清晰易读。 4. 主动提示风险任务（逾期、高优先级长时间无进展）。 5. 创建任务时主动推荐最合适的平台，并解释推荐理由。 6. 遇到 API 错误时，耐心排查并给出可行的解决方案。 7. 尊重订阅等级限制，提示升级时保持友好。 --- ## 参考文档 - **API 指南**：`references/api-guide.md` — 各平台 API 端点和使用方法。 - **统一模型**：`references/unified-schema.md` — 统一任务模型定义和状态/优先级映射表。 FILE:assets/README.md # 项目中枢 (project-nerve) > 跨平台项目管理聚合器 — 一个入口，统一管理 Trello、GitHub Issues、Linear、Notion、Obsidian 任务 --- ## 功能亮点 - **多平台聚合** — 连接 Trello、GitHub Issues、Linear、Notion、Obsidian，一个视图查看所有任务 - **智能标准化** — 自动统一各平台的状态、优先级和任务格式，消除信息孤岛 - **自动检测平台** — 创建任务时根据内容智能推荐最合适的平台（Bug → GitHub，文档 → Notion） - 🧠 **自学习引擎** — 从使用模式中持续优化，记录错误/成功/纠正，自动生成改进建议 - 🕸️ **任务关系图谱** — 可视化跨平台任务依赖，检测循环依赖，影响分析 - 📝 **Obsidian 集成** — 本地知识库作为任务源，从 markdown 复选框和 frontmatter 提取任务 - **站会报告** — 自动生成每日站会和每周总结，扫描所有平台的最新动态 - **冲刺分析** — 速度统计、任务漏斗、燃尽图，用数据驱动项目管理 - **阻碍预警** — 自动识别逾期和高风险任务，及时提醒处理 - **Mermaid 图表** — 饼图、柱状图、折线图内嵌报告，无需额外工具 - **安全优先** — 凭据通过环境变量管理，本地运行不上传数据 --- ## Feature Highlights - **Multi-platform Aggregation** — Connect Trello, GitHub Issues, Linear, Notion, Obsidian in a single unified view - **Smart Normalization** — Automatically unify status, priority, and task formats across platforms - **Auto Platform Detection** — Intelligently recommend the best platform when creating tasks - 🧠 **Self-Learning Engine** — Continuously improve from usage patterns, auto-generate optimization suggestions - 🕸️ **Task Relationship Graph** — Visualize cross-platform task dependencies, detect cycles, impact analysis - 📝 **Obsidian Integration** — Use local knowledge base as a task source, extract from checkboxes and frontmatter - **Standup Reports** — Auto-generate daily standups and weekly summaries from all connected sources - **Sprint Analytics** — Velocity tracking, task funnel, burndown charts with data-driven insights - **Blocker Alerts** — Proactively identify overdue and high-risk tasks - **Mermaid Charts** — Embedded pie, bar, and line charts in Markdown reports - **Security First** — Credentials via env vars only, all data stays local --- ## 版本对比 / Version Comparison | 功能 Feature | 免费版 Free | 付费版 Paid ¥99/月 | |------|:------:|:------------:| | 平台数量 Sources（含 Obsidian） | 最多 2 个 | 最多 10 个 | | 任务显示 Tasks | 50 条 | 500 条 | | 任务查询 Query | 支持 | 支持 | | 创建/更新任务 CRUD | 支持 | 支持 | | 自学习引擎 Learning Engine（基础） | 支持 | 支持 | | 自学习引擎 Learning Engine（高级建议） | - | 支持 | | 任务关系图谱 Task Graph | - | 支持 | | 任务图谱可视化 Graph Visualization | - | 支持 | | 每日站会 Daily Standup | 支持 | 支持 | | 每周总结 Weekly Report | - | 支持 | | 冲刺分析 Sprint Analytics | - | 支持（速度/漏斗/燃尽图） | | 阻碍分析 Blocker Analysis | - | 支持 | | Mermaid 图表 Charts | - | 支持 | | 批量同步 Bulk Sync | - | 支持 | --- ## 快速开始 / Quick Start ### 1. 安装 Skill ```bash openclaw skill install project-nerve ``` ### 2. 配置平台凭据 ```bash # Trello export PNC_TRELLO_API_KEY="your-api-key" export PNC_TRELLO_TOKEN="your-token" # GitHub export PNC_GITHUB_TOKEN="ghp_your-token" # Linear export PNC_LINEAR_API_KEY="lin_api_your-key" # Notion export PNC_NOTION_TOKEN="ntn_your-token" export PNC_NOTION_DATABASE_ID="your-database-id" # Obsidian export PNC_OBSIDIAN_VAULT_PATH="/path/to/your/vault" ``` ### 3. 连接平台 ```bash /project-nerve 连接 GitHub /project-nerve 连接 Trello ``` ### 4. 开始使用 ```bash # 查看所有任务 /project-nerve 同步任务 # 搜索任务 /project-nerve 搜索 "登录Bug" # 创建任务 /project-nerve 创建任务 "修复登录页面样式问题" # 生成站会报告 /project-nerve 站会 # 冲刺报告（付费版） /project-nerve 冲刺报告 ``` --- ## 示例输出 / Example Output ### 每日站会 ```markdown # 每日站会 — 2026-03-19 ## 昨日完成 - **修复用户头像上传失败** [github] (高) @zhangsan - **更新首页Banner设计稿** [notion] @lisi ## 今日计划 - [进行中] **实现用户权限模块** [linear] (紧急) @zhangsan - [待启动] **编写API文档** [notion] @wangwu ## 阻碍事项 - **支付接口对接** [github] (紧急) — 逾期（截止: 2026-03-17） --- 完成 2 | 计划 2 | 阻碍 1 ``` ### 任务聚合表 ```markdown | # | 标题 | 平台 | 状态 | 优先级 | 负责人 | 截止日期 | |---|------|------|------|--------|--------|----------| | 1 | 实现用户权限模块 | linear | 进行中 | 紧急 | zhangsan | 2026-03-21 | | 2 | 修复登录页面样式 | github | 待办 | 高 | lisi | 2026-03-20 | | 3 | 更新产品路线图 | notion | 进行中 | 中 | wangwu | - | | 4 | 优化首页加载速度 | trello | 待办 | 中 | - | 2026-03-25 | ``` --- ## 常见问题 / FAQ ### Q1: 支持哪些平台？目前支持 Trello、GitHub Issues、Linear、Notion 和 Obsidian（本地 Vault）。后续计划支持 Jira、Asana、ClickUp 等。 ### Q2: 数据安全吗？所有数据在本地处理，凭据通过环境变量管理，不会上传到云端。API 通信均使用 HTTPS。 ### Q3: 免费版够用吗？免费版支持连接 2 个平台、查看 50 条任务和每日站会，适合个人或小团队。需要冲刺分析、周报和阻碍分析请升级付费版。 ### Q4: 任务去重逻辑是什么？当同一任务在多个平台存在时（如 GitHub Issue 和 Trello 卡片标题高度相似），系统会基于标题词重叠率自动去重，保留最近更新的版本。 ### Q5: 如何自定义站会报告？可通过 `--data '{"assignee":"zhangsan"}'` 过滤特定成员的报告。 ### Q6: Mermaid 图表在哪里查看？ Mermaid 图表可在 GitHub/GitLab Markdown 预览、VS Code（安装 Mermaid 插件）、Typora、Obsidian 等工具中直接渲染。 --- ## 技术支持 - 文档：查看 `references/` 目录获取 API 和模型参考 - 问题反馈：在 ClawHub 的 Skill 页面提交 Issue - 社区讨论：加入 ClawHub 社区频道 `#project-nerve` - 邮件：skill-support@clawhub.dev --- *project-nerve v1.1.0 | 兼容 OpenClaw 0.5+* FILE:scripts/source_connector.py #!/usr/bin/env python3 """ project-nerve 数据源连接器管理 Trello、GitHub Issues、Linear、Notion、Obsidian 等平台的连接配置。支持连接、测试、列表、断开操作。 """ import json import os import sys import urllib.request import urllib.error from datetime import datetime from typing import Any, Dict, List, Optional 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, write_json_file, SUPPORTED_PLATFORMS, ) # ============================================================ # 数据文件路径 # ============================================================ SOURCES_FILE = "sources.json" def _get_sources() -> List[Dict[str, Any]]: """读取所有已连接的数据源配置。""" data = read_json_file(get_data_file(SOURCES_FILE)) if isinstance(data, list): return data return [] def _save_sources(sources: List[Dict[str, Any]]) -> None: """保存数据源配置到文件。""" write_json_file(get_data_file(SOURCES_FILE), sources) def _find_source(sources: List[Dict], source_id: str) -> Optional[Dict]: """根据 ID 查找数据源。""" for s in sources: if s.get("id") == source_id: return s return None def _find_source_by_platform(sources: List[Dict], platform: str) -> Optional[Dict]: """根据平台类型查找数据源。""" for s in sources: if s.get("platform") == platform: return s return None # ============================================================ # HTTP 请求工具 # ============================================================ def _http_request( url: str, method: str = "GET", headers: Optional[Dict[str, str]] = None, data: Optional[bytes] = None, timeout: int = 15, ) -> Dict[str, Any]: """发送 HTTP 请求。 Args: url: 请求地址。 method: HTTP 方法。 headers: 请求头。 data: 请求体。 timeout: 超时秒数。 Returns: 包含 status、body、headers 的响应字典。 """ if headers is None: headers = {} req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8") return { "status": resp.status, "body": body, "headers": dict(resp.headers), } except urllib.error.HTTPError as e: body = "" try: body = e.read().decode("utf-8") except Exception: pass return { "status": e.code, "body": body, "headers": {}, "error": str(e), } except urllib.error.URLError as e: return { "status": 0, "body": "", "headers": {}, "error": f"网络错误: {e.reason}", } except Exception as e: return { "status": 0, "body": "", "headers": {}, "error": f"请求失败: {e}", } # ============================================================ # 平台适配器 — 连接测试 # ============================================================ def _test_trello(config: Dict[str, Any]) -> Dict[str, Any]: """测试 Trello API 连接。需要 api_key 和 token。通过访问 /1/members/me 端点验证凭据。 Args: config: 包含 api_key 和 token 的配置字典。 Returns: 测试结果字典，包含 success、message、user_info。 """ api_key = config.get("api_key") or os.environ.get("PNC_TRELLO_API_KEY", "") token = config.get("token") or os.environ.get("PNC_TRELLO_TOKEN", "") if not api_key or not token: return {"success": False, "message": "缺少 Trello API Key 或 Token，请设置 PNC_TRELLO_API_KEY 和 PNC_TRELLO_TOKEN 环境变量"} url = f"https://api.trello.com/1/members/me?key={api_key}&token={token}" resp = _http_request(url) if resp["status"] == 200: try: user = json.loads(resp["body"]) return { "success": True, "message": f"Trello 连接成功，用户: {user.get('fullName', user.get('username', '未知'))}", "user_info": {"name": user.get("fullName", ""), "username": user.get("username", "")}, } except json.JSONDecodeError: return {"success": True, "message": "Trello 连接成功"} else: return {"success": False, "message": f"Trello 连接失败 (HTTP {resp['status']}): {resp.get('error', resp['body'][:200])}"} def _test_github(config: Dict[str, Any]) -> Dict[str, Any]: """测试 GitHub API 连接。需要 personal access token。通过访问 /user 端点验证凭据。 Args: config: 包含 token 的配置字典。 Returns: 测试结果字典。 """ token = config.get("token") or os.environ.get("PNC_GITHUB_TOKEN", "") if not token: return {"success": False, "message": "缺少 GitHub Token，请设置 PNC_GITHUB_TOKEN 环境变量"} url = "https://api.github.com/user" headers = { "Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "project-nerve/1.0", } resp = _http_request(url, headers=headers) if resp["status"] == 200: try: user = json.loads(resp["body"]) return { "success": True, "message": f"GitHub 连接成功，用户: {user.get('login', '未知')}", "user_info": {"login": user.get("login", ""), "name": user.get("name", "")}, } except json.JSONDecodeError: return {"success": True, "message": "GitHub 连接成功"} else: return {"success": False, "message": f"GitHub 连接失败 (HTTP {resp['status']}): {resp.get('error', resp['body'][:200])}"} def _test_linear(config: Dict[str, Any]) -> Dict[str, Any]: """测试 Linear API 连接。需要 API key。通过 GraphQL API 查询 viewer 验证凭据。 Args: config: 包含 api_key 的配置字典。 Returns: 测试结果字典。 """ api_key = config.get("api_key") or os.environ.get("PNC_LINEAR_API_KEY", "") if not api_key: return {"success": False, "message": "缺少 Linear API Key，请设置 PNC_LINEAR_API_KEY 环境变量"} url = "https://api.linear.app/graphql" headers = { "Authorization": api_key, "Content-Type": "application/json", } query = '{"query": "{ viewer { id name email } }"}' resp = _http_request(url, method="POST", headers=headers, data=query.encode("utf-8")) if resp["status"] == 200: try: result = json.loads(resp["body"]) viewer = result.get("data", {}).get("viewer", {}) name = viewer.get("name", "未知") return { "success": True, "message": f"Linear 连接成功，用户: {name}", "user_info": {"name": name, "email": viewer.get("email", "")}, } except (json.JSONDecodeError, KeyError): return {"success": True, "message": "Linear 连接成功"} else: return {"success": False, "message": f"Linear 连接失败 (HTTP {resp['status']}): {resp.get('error', resp['body'][:200])}"} def _test_obsidian(config: Dict[str, Any]) -> Dict[str, Any]: """测试 Obsidian Vault 可访问性。验证 vault 路径是否存在，并统计带有任务标签的笔记数量。 Args: config: 包含 vault_path 和 task_tag 的配置字典。 Returns: 测试结果字典，包含 success、message、user_info。 """ vault_path = config.get("vault_path") or os.environ.get("PNC_OBSIDIAN_VAULT_PATH", "") task_tag = config.get("task_tag", "#task") if not vault_path: return {"success": False, "message": "缺少 Obsidian Vault 路径，请设置 PNC_OBSIDIAN_VAULT_PATH 环境变量或提供 vault_path"} vault_path = os.path.expanduser(vault_path) if not os.path.isdir(vault_path): return {"success": False, "message": f"Obsidian Vault 路径不存在: {vault_path}"} # 扫描 .md 文件，统计带任务标签的笔记数量 task_note_count = 0 total_md_count = 0 for root, _dirs, files in os.walk(vault_path): for fname in files: if not fname.endswith(".md"): continue total_md_count += 1 fpath = os.path.join(root, fname) try: with open(fpath, "r", encoding="utf-8") as f: content = f.read(4096) # 只读取前 4KB 检查标签 if task_tag in content or "- [ ]" in content or "- [x]" in content: task_note_count += 1 except (IOError, UnicodeDecodeError): continue return { "success": True, "message": f"Obsidian Vault 连接成功，共 {total_md_count} 个笔记，其中 {task_note_count} 个包含任务", "user_info": { "vault_path": vault_path, "total_notes": total_md_count, "task_notes": task_note_count, "task_tag": task_tag, }, } def _connect_obsidian(config: Dict[str, Any]) -> Dict[str, Any]: """验证 Obsidian Vault 配置。检查 vault 路径是否存在，扫描带有任务标签的笔记。 Args: config: Obsidian 配置字典。 Returns: 包含 vault_path 和 task_tag 的已验证配置。 """ return _test_obsidian(config) def _test_notion(config: Dict[str, Any]) -> Dict[str, Any]: """测试 Notion API 连接。需要 integration token 和 database_id。通过查询数据库验证凭据。 Args: config: 包含 token 和 database_id 的配置字典。 Returns: 测试结果字典。 """ token = config.get("token") or os.environ.get("PNC_NOTION_TOKEN", "") database_id = config.get("database_id") or os.environ.get("PNC_NOTION_DATABASE_ID", "") if not token: return {"success": False, "message": "缺少 Notion Integration Token，请设置 PNC_NOTION_TOKEN 环境变量"} if not database_id: return {"success": False, "message": "缺少 Notion Database ID，请设置 PNC_NOTION_DATABASE_ID 环境变量"} url = f"https://api.notion.com/v1/databases/{database_id}" headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json", } resp = _http_request(url, headers=headers) if resp["status"] == 200: try: db = json.loads(resp["body"]) title_parts = db.get("title", []) db_title = "" for part in title_parts: db_title += part.get("plain_text", "") return { "success": True, "message": f"Notion 连接成功，数据库: {db_title or database_id}", "user_info": {"database_title": db_title, "database_id": database_id}, } except (json.JSONDecodeError, KeyError): return {"success": True, "message": "Notion 连接成功"} else: return {"success": False, "message": f"Notion 连接失败 (HTTP {resp['status']}): {resp.get('error', resp['body'][:200])}"} # ============================================================ # 平台测试路由 # ============================================================ _PLATFORM_TESTERS = { "trello": _test_trello, "github": _test_github, "linear": _test_linear, "notion": _test_notion, "obsidian": _test_obsidian, } # ============================================================ # 操作实现 # ============================================================ def connect_source(data: Dict[str, Any]) -> None: """连接新数据源。必填字段: platform 可选字段: name, api_key, token, database_id, repo（GitHub 仓库如 owner/repo）, board_id（Trello） Args: data: 数据源配置字典。 """ platform = data.get("platform", "").strip().lower() if platform not in SUPPORTED_PLATFORMS: valid = "、".join(SUPPORTED_PLATFORMS) output_error(f"不支持的平台: {platform!r}，支持的平台: {valid}", code="INVALID_PLATFORM") return # 检查订阅限制 sub = check_subscription() sources = _get_sources() if len(sources) >= sub["max_sources"]: if sub["tier"] == "free": output_error( f"免费版最多连接 {sub['max_sources']} 个数据源，当前已有 {len(sources)} 个。" "请升级至付费版（¥99/月）以连接更多数据源。", code="LIMIT_EXCEEDED", ) else: output_error( f"已达到数据源数量上限 {sub['max_sources']} 个。", code="LIMIT_EXCEEDED", ) return # 检查是否已存在同平台连接 existing = _find_source_by_platform(sources, platform) if existing: output_error( f"已存在 {platform} 平台的连接（ID: {existing['id']}）。如需重新连接，请先断开现有连接。", code="DUPLICATE_SOURCE", ) return # 测试连接 tester = _PLATFORM_TESTERS.get(platform) if tester: test_result = tester(data) if not test_result["success"]: output_error(test_result["message"], code="CONNECTION_FAILED") return # 保存配置 now = now_iso() source_config = { "id": generate_id("SRC"), "platform": platform, "name": data.get("name", f"{platform} 数据源"), "config": {}, "connected_at": now, "updated_at": now, "status": "active", } # 按平台保存不同的配置字段（不保存敏感凭据，使用环境变量） if platform == "trello": source_config["config"]["board_id"] = data.get("board_id", "") source_config["config"]["env_key"] = "PNC_TRELLO_API_KEY" source_config["config"]["env_token"] = "PNC_TRELLO_TOKEN" elif platform == "github": source_config["config"]["repo"] = data.get("repo", "") source_config["config"]["env_token"] = "PNC_GITHUB_TOKEN" elif platform == "linear": source_config["config"]["team_id"] = data.get("team_id", "") source_config["config"]["env_key"] = "PNC_LINEAR_API_KEY" elif platform == "notion": source_config["config"]["database_id"] = data.get("database_id") or os.environ.get("PNC_NOTION_DATABASE_ID", "") source_config["config"]["env_token"] = "PNC_NOTION_TOKEN" elif platform == "obsidian": vault_path = data.get("vault_path") or os.environ.get("PNC_OBSIDIAN_VAULT_PATH", "") source_config["config"]["vault_path"] = os.path.expanduser(vault_path) source_config["config"]["task_tag"] = data.get("task_tag", "#task") source_config["config"]["env_vault_path"] = "PNC_OBSIDIAN_VAULT_PATH" sources.append(source_config) _save_sources(sources) output_success({ "message": f"{platform} 数据源已连接", "source": source_config, }) def test_source(data: Dict[str, Any]) -> None: """测试数据源连接。必填字段: platform 或 id Args: data: 包含平台或数据源 ID 的字典。 """ source_id = data.get("id") platform = data.get("platform", "").strip().lower() if source_id: sources = _get_sources() source = _find_source(sources, source_id) if not source: output_error(f"未找到 ID 为 {source_id} 的数据源", code="NOT_FOUND") return platform = source["platform"] if platform not in SUPPORTED_PLATFORMS: valid = "、".join(SUPPORTED_PLATFORMS) output_error(f"不支持的平台: {platform!r}，支持的平台: {valid}", code="INVALID_PLATFORM") return tester = _PLATFORM_TESTERS.get(platform) if not tester: output_error(f"平台 {platform} 暂不支持连接测试", code="NOT_SUPPORTED") return test_result = tester(data) if test_result["success"]: output_success({ "platform": platform, "message": test_result["message"], "user_info": test_result.get("user_info", {}), }) else: output_error(test_result["message"], code="CONNECTION_FAILED") def list_sources(data: Optional[Dict[str, Any]] = None) -> None: """列出所有已连接的数据源。 Args: data: 可选的过滤条件字典。 """ sources = _get_sources() if data: platform_filter = data.get("platform", "").strip().lower() if platform_filter: sources = [s for s in sources if s.get("platform") == platform_filter] status_filter = data.get("status", "").strip().lower() if status_filter: sources = [s for s in sources if s.get("status") == status_filter] # 构建摘要信息 platform_stats = {} for p in SUPPORTED_PLATFORMS: count = sum(1 for s in sources if s.get("platform") == p) if count > 0: platform_stats[p] = count # 脱敏输出（不显示敏感配置） display_list = [] for s in sources: display = { "id": s.get("id"), "platform": s.get("platform"), "name": s.get("name"), "status": s.get("status"), "connected_at": s.get("connected_at"), "updated_at": s.get("updated_at"), } display_list.append(display) output_success({ "total": len(display_list), "platform_stats": platform_stats, "sources": display_list, }) def disconnect_source(data: Dict[str, Any]) -> None: """断开数据源连接。必填字段: id 或 platform Args: data: 包含数据源 ID 或平台名称的字典。 """ source_id = data.get("id") platform = data.get("platform", "").strip().lower() sources = _get_sources() original_count = len(sources) if source_id: sources = [s for s in sources if s.get("id") != source_id] elif platform: sources = [s for s in sources if s.get("platform") != platform] else: output_error("请提供数据源 ID（id）或平台名称（platform）", code="VALIDATION_ERROR") return if len(sources) == original_count: output_error("未找到匹配的数据源", code="NOT_FOUND") return removed_count = original_count - len(sources) _save_sources(sources) output_success({ "message": f"已断开 {removed_count} 个数据源连接", "remaining": len(sources), }) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("project-nerve 数据源连接器") args = parser.parse_args() action = args.action.lower().replace("-", "_") try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "connect": lambda: connect_source(data or {}), "test": lambda: test_source(data or {}), "list_sources": lambda: list_sources(data), "list": lambda: list_sources(data), "disconnect": lambda: disconnect_source(data or {}), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(["connect", "test", "list-sources", "disconnect"]) output_error(f"未知操作: {args.action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/learning_engine.py #!/usr/bin/env python3 """ project-nerve 自学习引擎从操作记录中持续学习，记录错误、成功模式和用户纠正，并基于积累的数据提供改进建议和统计分析。灵感来源: self-improving-agent (255K 下载量) """ import json import os import sys import hashlib from datetime import datetime from typing import Any, Dict, List, Optional 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, write_json_file, ) # ============================================================ # 数据文件路径 # ============================================================ LEARNING_FILE = "learning.json" # ============================================================ # 数据读写 # ============================================================ def _get_learning_data() -> Dict[str, Any]: """读取学习数据文件。 Returns: 包含 patterns 列表的字典。 """ data = read_json_file(get_data_file(LEARNING_FILE)) if isinstance(data, dict) and "patterns" in data: return data return {"patterns": [], "metadata": {"created_at": now_iso(), "version": "1.0"}} def _save_learning_data(data: Dict[str, Any]) -> None: """保存学习数据到文件。 Args: data: 学习数据字典。 """ data["metadata"] = data.get("metadata", {}) data["metadata"]["updated_at"] = now_iso() write_json_file(get_data_file(LEARNING_FILE), data) # ============================================================ # 模式指纹计算（用于聚合相似模式） # ============================================================ def _compute_fingerprint(pattern_type: str, category: str, context: Dict[str, Any]) -> str: """计算模式指纹，用于识别和聚合相似的模式。基于类型、分类和关键上下文信息生成哈希指纹。相同指纹的模式会被聚合（计数递增）而非重复存储。 Args: pattern_type: 模式类型（error / success / correction）。 category: 分类标签（如 api_failure, platform_choice 等）。 context: 上下文字典，包含与模式相关的详细信息。 Returns: 16 位十六进制指纹字符串。 """ # 提取用于指纹计算的关键字段 key_parts = [pattern_type, category] # 根据不同类型提取不同的关键信息 if pattern_type == "error": key_parts.append(context.get("source", "")) key_parts.append(context.get("error_type", "")) key_parts.append(context.get("action", "")) elif pattern_type == "success": key_parts.append(context.get("source", "")) key_parts.append(context.get("action", "")) key_parts.append(context.get("task_type", "")) elif pattern_type == "correction": key_parts.append(context.get("field", "")) key_parts.append(context.get("original_value", "")) key_parts.append(context.get("corrected_value", "")) raw = "|".join(str(p) for p in key_parts) return hashlib.md5(raw.encode("utf-8")).hexdigest()[:16] # ============================================================ # 模式匹配与聚合 # ============================================================ def _find_matching_pattern( patterns: List[Dict[str, Any]], fingerprint: str ) -> Optional[int]: """查找与指纹匹配的已有模式索引。 Args: patterns: 模式列表。 fingerprint: 待匹配的指纹。 Returns: 匹配的模式在列表中的索引，未找到返回 None。 """ for i, p in enumerate(patterns): if p.get("fingerprint") == fingerprint: return i return None def _record_pattern( pattern_type: str, category: str, context: Dict[str, Any], lesson: str, ) -> Dict[str, Any]: """记录一条学习模式。若已存在指纹相同的模式，则递增计数并更新时间戳；否则创建新的模式记录。 Args: pattern_type: 模式类型（error / success / correction）。 category: 分类标签。 context: 上下文信息。 lesson: 从该模式中总结的经验教训。 Returns: 记录的模式字典。 """ data = _get_learning_data() patterns = data["patterns"] fingerprint = _compute_fingerprint(pattern_type, category, context) existing_idx = _find_matching_pattern(patterns, fingerprint) if existing_idx is not None: # 聚合：递增计数，更新时间戳 patterns[existing_idx]["count"] += 1 patterns[existing_idx]["last_seen"] = now_iso() # 如果提供了新的经验教训，更新之 if lesson and lesson != patterns[existing_idx].get("lesson", ""): patterns[existing_idx]["lesson"] = lesson pattern = patterns[existing_idx] else: # 创建新模式 now = now_iso() pattern = { "id": generate_id("LRN"), "type": pattern_type, "category": category, "context": context, "lesson": lesson, "fingerprint": fingerprint, "count": 1, "first_seen": now, "last_seen": now, } patterns.append(pattern) data["patterns"] = patterns _save_learning_data(data) return pattern # ============================================================ # 操作实现：记录错误 # ============================================================ def record_error(data: Dict[str, Any]) -> None: """记录操作中遇到的错误。用于追踪 API 失败、超时、解析错误等问题模式。相同来源和类型的错误会被聚合计数。必填字段: category（错误分类，如 api_failure / timeout / parse_error）可选字段: source（平台名称）, error_type（具体错误类型）, action（触发错误的操作）, message（错误详情）, lesson（经验总结） Args: data: 错误信息字典。 """ category = data.get("category", "").strip() if not category: output_error("错误分类（category）为必填字段", code="VALIDATION_ERROR") return context = { "source": data.get("source", ""), "error_type": data.get("error_type", ""), "action": data.get("action", ""), "message": data.get("message", ""), } lesson = data.get("lesson", "") pattern = _record_pattern("error", category, context, lesson) output_success({ "message": f"已记录错误模式: {category}", "pattern_id": pattern["id"], "count": pattern["count"], "aggregated": pattern["count"] > 1, }) # ============================================================ # 操作实现：记录成功 # ============================================================ def record_success(data: Dict[str, Any]) -> None: """记录成功的操作模式。用于追踪有效的平台选择、成功的查询方式等。帮助系统学习哪些操作在什么场景下效果最好。必填字段: category（成功分类，如 platform_choice / query_pattern / fetch_success）可选字段: source（平台名称）, action（操作类型）, task_type（任务类型）, details（详细信息）, lesson（经验总结） Args: data: 成功信息字典。 """ category = data.get("category", "").strip() if not category: output_error("成功分类（category）为必填字段", code="VALIDATION_ERROR") return context = { "source": data.get("source", ""), "action": data.get("action", ""), "task_type": data.get("task_type", ""), "details": data.get("details", ""), } lesson = data.get("lesson", "") pattern = _record_pattern("success", category, context, lesson) output_success({ "message": f"已记录成功模式: {category}", "pattern_id": pattern["id"], "count": pattern["count"], "aggregated": pattern["count"] > 1, }) # ============================================================ # 操作实现：记录用户纠正 # ============================================================ def record_correction(data: Dict[str, Any]) -> None: """记录用户对自动行为的纠正。当用户覆盖自动检测结果（如更改推荐平台、调整优先级）时记录，帮助系统逐步学习用户偏好。必填字段: category（纠正分类，如 platform_override / priority_change）可选字段: field（被纠正的字段名）, original_value（原始值）, corrected_value（纠正后的值）, reason（纠正原因）, lesson（经验总结） Args: data: 纠正信息字典。 """ category = data.get("category", "").strip() if not category: output_error("纠正分类（category）为必填字段", code="VALIDATION_ERROR") return context = { "field": data.get("field", ""), "original_value": data.get("original_value", ""), "corrected_value": data.get("corrected_value", ""), "reason": data.get("reason", ""), } lesson = data.get("lesson", "") pattern = _record_pattern("correction", category, context, lesson) output_success({ "message": f"已记录用户纠正: {category}", "pattern_id": pattern["id"], "count": pattern["count"], "aggregated": pattern["count"] > 1, }) # ============================================================ # 操作实现：建议 # ============================================================ def _generate_error_suggestions(patterns: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """基于错误模式生成改进建议。分析高频错误，识别问题集中的平台和操作，提出针对性的改进建议。 Args: patterns: 错误类型的模式列表。 Returns: 建议列表，每项包含 type、message、severity、based_on。 """ suggestions = [] if not patterns: return suggestions # 按平台统计错误次数 source_errors: Dict[str, int] = {} for p in patterns: src = p.get("context", {}).get("source", "未知") source_errors[src] = source_errors.get(src, 0) + p.get("count", 1) # 找出错误最多的平台 if source_errors: worst_source = max(source_errors, key=source_errors.get) # type: ignore worst_count = source_errors[worst_source] if worst_count >= 3: # 找到可替代的平台 all_sources = list(source_errors.keys()) alternatives = [s for s in all_sources if s != worst_source and source_errors[s] < worst_count // 2] alt_text = "" if alternatives: alt_text = f"，建议优先使用 {alternatives[0]}" suggestions.append({ "type": "reliability", "message": f"{worst_source} 的 API 最近频繁出错（{worst_count} 次）{alt_text}", "severity": "高" if worst_count >= 5 else "中", "based_on": f"{worst_count} 次错误记录", }) # 按错误类型统计 error_types: Dict[str, int] = {} for p in patterns: et = p.get("context", {}).get("error_type", "") if et: error_types[et] = error_types.get(et, 0) + p.get("count", 1) # 超时相关建议 timeout_count = error_types.get("timeout", 0) if timeout_count >= 2: suggestions.append({ "type": "performance", "message": f"检测到 {timeout_count} 次超时，建议检查网络连接或增加超时设置", "severity": "中", "based_on": f"{timeout_count} 次超时记录", }) # 认证相关建议 auth_count = error_types.get("auth_failure", 0) + error_types.get("unauthorized", 0) if auth_count >= 1: suggestions.append({ "type": "configuration", "message": f"检测到 {auth_count} 次认证失败，建议检查 API 密钥是否过期", "severity": "高", "based_on": f"{auth_count} 次认证错误", }) return suggestions def _generate_correction_suggestions(patterns: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """基于用户纠正模式生成偏好建议。分析用户经常覆盖的自动决策，识别用户偏好，提出自动化改进建议。 Args: patterns: 纠正类型的模式列表。 Returns: 建议列表。 """ suggestions = [] if not patterns: return suggestions # 分析平台覆盖模式 platform_overrides: Dict[str, Dict[str, int]] = {} for p in patterns: ctx = p.get("context", {}) if ctx.get("field") == "platform" or p.get("category") == "platform_override": original = ctx.get("original_value", "") corrected = ctx.get("corrected_value", "") if original and corrected: key = f"{original}->{corrected}" if key not in platform_overrides: platform_overrides[key] = {"count": 0, "original": original, "corrected": corrected} platform_overrides[key]["count"] += p.get("count", 1) # 生成平台偏好建议 for key, info in platform_overrides.items(): if info["count"] >= 2: suggestions.append({ "type": "preference", "message": ( f"用户倾向于将任务从 {info['original']} 改到 {info['corrected']}" f"（已纠正 {info['count']} 次），建议调整自动检测逻辑" ), "severity": "低", "based_on": f"{info['count']} 次平台覆盖", }) # 分析优先级纠正模式 priority_corrections = [ p for p in patterns if p.get("context", {}).get("field") == "priority" or p.get("category") == "priority_change" ] total_prio_corrections = sum(p.get("count", 1) for p in priority_corrections) if total_prio_corrections >= 3: suggestions.append({ "type": "preference", "message": f"用户经常调整任务优先级（共 {total_prio_corrections} 次），建议优化优先级自动判断逻辑", "severity": "中", "based_on": f"{total_prio_corrections} 次优先级纠正", }) return suggestions def _generate_success_suggestions(patterns: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """基于成功模式生成优化建议。分析高效的操作模式，提炼可复用的最佳实践。 Args: patterns: 成功类型的模式列表。 Returns: 建议列表。 """ suggestions = [] if not patterns: return suggestions # 找出使用最频繁的成功平台 source_success: Dict[str, int] = {} for p in patterns: src = p.get("context", {}).get("source", "") if src: source_success[src] = source_success.get(src, 0) + p.get("count", 1) if source_success: best_source = max(source_success, key=source_success.get) # type: ignore best_count = source_success[best_source] if best_count >= 5: suggestions.append({ "type": "optimization", "message": f"{best_source} 是最常成功使用的平台（{best_count} 次），可作为默认首选", "severity": "低", "based_on": f"{best_count} 次成功记录", }) return suggestions def suggest(data: Optional[Dict[str, Any]] = None) -> None: """基于积累的学习数据生成改进建议。分析错误模式、用户纠正和成功模式，综合生成可操作的改进建议。可选字段: type（过滤建议类型: error / correction / success / all） Args: data: 可选的过滤参数。 """ # 检查订阅（高级建议为付费功能） sub = check_subscription() is_paid = sub["tier"] == "paid" learning = _get_learning_data() patterns = learning.get("patterns", []) if not patterns: output_success({ "message": "暂无学习数据，请在使用过程中积累操作记录后再查看建议", "suggestions": [], "total_patterns": 0, }) return # 按类型分组 errors = [p for p in patterns if p.get("type") == "error"] successes = [p for p in patterns if p.get("type") == "success"] corrections = [p for p in patterns if p.get("type") == "correction"] # 生成建议 filter_type = "" if data: filter_type = data.get("type", "").strip().lower() all_suggestions = [] if not filter_type or filter_type in ("error", "all"): all_suggestions.extend(_generate_error_suggestions(errors)) if not filter_type or filter_type in ("correction", "all"): if is_paid: all_suggestions.extend(_generate_correction_suggestions(corrections)) elif corrections: all_suggestions.append({ "type": "upgrade_hint", "message": f"检测到 {len(corrections)} 条用户纠正记录，升级付费版可获得个性化偏好建议", "severity": "低", "based_on": f"{len(corrections)} 条纠正记录", }) if not filter_type or filter_type in ("success", "all"): if is_paid: all_suggestions.extend(_generate_success_suggestions(successes)) # 按严重程度排序 severity_order = {"高": 0, "中": 1, "低": 2} all_suggestions.sort(key=lambda s: severity_order.get(s.get("severity", "低"), 2)) output_success({ "message": f"基于 {len(patterns)} 条学习记录生成了 {len(all_suggestions)} 条建议", "total_patterns": len(patterns), "pattern_breakdown": { "errors": len(errors), "successes": len(successes), "corrections": len(corrections), }, "suggestions": all_suggestions, }) # ============================================================ # 操作实现：统计 # ============================================================ def stats(data: Optional[Dict[str, Any]] = None) -> None: """显示学习数据统计信息。展示错误率、常见模式、纠正频率等统计数据，帮助了解系统的学习进度和数据质量。可选字段: type（按类型过滤: error / success / correction） Args: data: 可选的过滤参数。 """ learning = _get_learning_data() patterns = learning.get("patterns", []) if not patterns: output_success({ "message": "暂无学习数据", "total_patterns": 0, "stats": {}, }) return # 基本统计 type_counts = {"error": 0, "success": 0, "correction": 0} type_total_events = {"error": 0, "success": 0, "correction": 0} source_stats: Dict[str, Dict[str, int]] = {} category_stats: Dict[str, int] = {} for p in patterns: ptype = p.get("type", "unknown") count = p.get("count", 1) if ptype in type_counts: type_counts[ptype] += 1 type_total_events[ptype] += count # 按来源统计 src = p.get("context", {}).get("source", "未指定") if src not in source_stats: source_stats[src] = {"error": 0, "success": 0, "correction": 0, "total": 0} if ptype in source_stats[src]: source_stats[src][ptype] += count source_stats[src]["total"] += count # 按分类统计 cat = p.get("category", "未分类") category_stats[cat] = category_stats.get(cat, 0) + count # 计算错误率 total_events = sum(type_total_events.values()) error_rate = 0.0 if total_events > 0: error_rate = round(type_total_events["error"] / total_events * 100, 1) # 每个来源的错误率 source_error_rates: Dict[str, float] = {} for src, st in source_stats.items(): if st["total"] > 0: source_error_rates[src] = round(st["error"] / st["total"] * 100, 1) # 最常见的模式（按计数排序） top_patterns = sorted(patterns, key=lambda p: p.get("count", 1), reverse=True)[:10] top_summary = [] for p in top_patterns: top_summary.append({ "type": p.get("type", ""), "category": p.get("category", ""), "count": p.get("count", 1), "source": p.get("context", {}).get("source", ""), "lesson": p.get("lesson", ""), }) # 时间范围 first_seen_dates = [p.get("first_seen", "") for p in patterns if p.get("first_seen")] last_seen_dates = [p.get("last_seen", "") for p in patterns if p.get("last_seen")] time_range = {} if first_seen_dates: time_range["earliest"] = min(first_seen_dates) if last_seen_dates: time_range["latest"] = max(last_seen_dates) result = { "message": f"共有 {len(patterns)} 个独立模式，{total_events} 次事件记录", "total_patterns": len(patterns), "total_events": total_events, "type_breakdown": { "patterns": type_counts, "events": type_total_events, }, "error_rate": f"{error_rate}%", "source_stats": source_stats, "source_error_rates": source_error_rates, "top_categories": dict(sorted(category_stats.items(), key=lambda x: x[1], reverse=True)[:10]), "top_patterns": top_summary, "time_range": time_range, } output_success(result) # ============================================================ # 操作实现：重置 # ============================================================ def reset(data: Optional[Dict[str, Any]] = None) -> None: """重置学习数据。可选择重置全部数据或仅重置指定类型的数据。此操作不可撤销。可选字段: type（仅重置指定类型: error / success / correction）, confirm（确认重置，必须为 true） Args: data: 可选参数。 """ if not data or not data.get("confirm"): output_error( "重置操作不可撤销，请传入 confirm: true 确认操作", code="CONFIRMATION_REQUIRED", ) return filter_type = "" if data: filter_type = data.get("type", "").strip().lower() learning = _get_learning_data() patterns = learning.get("patterns", []) original_count = len(patterns) if filter_type and filter_type in ("error", "success", "correction"): # 仅删除指定类型 patterns = [p for p in patterns if p.get("type") != filter_type] removed = original_count - len(patterns) learning["patterns"] = patterns _save_learning_data(learning) output_success({ "message": f"已重置 {filter_type} 类型的学习数据，移除 {removed} 条记录", "removed": removed, "remaining": len(patterns), }) else: # 重置全部 learning["patterns"] = [] _save_learning_data(learning) output_success({ "message": f"已重置全部学习数据，移除 {original_count} 条记录", "removed": original_count, "remaining": 0, }) # ============================================================ # 便捷 API（供其他模块调用） # ============================================================ def quick_record_error(source: str, action: str, error_type: str, message: str) -> None: """快速记录错误（供其他脚本内部调用的便捷方法）。 Args: source: 平台来源。 action: 触发错误的操作。 error_type: 错误类型。 message: 错误信息。 """ try: context = { "source": source, "error_type": error_type, "action": action, "message": message, } _record_pattern("error", error_type, context, message) except Exception: # 学习记录失败不应影响主流程 pass def quick_record_success(source: str, action: str, task_type: str = "") -> None: """快速记录成功（供其他脚本内部调用的便捷方法）。 Args: source: 平台来源。 action: 操作类型。 task_type: 任务类型。 """ try: context = { "source": source, "action": action, "task_type": task_type, } _record_pattern("success", "operation_success", context, "") except Exception: # 学习记录失败不应影响主流程 pass # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("project-nerve 自学习引擎") args = parser.parse_args() action = args.action.lower().replace("-", "_") try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "record_error": lambda: record_error(data or {}), "record_success": lambda: record_success(data or {}), "record_correction": lambda: record_correction(data or {}), "suggest": lambda: suggest(data), "stats": lambda: stats(data), "reset": lambda: reset(data), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join([ "record-error", "record-success", "record-correction", "suggest", "stats", "reset", ]) output_error( f"未知操作: {args.action}，支持的操作: {valid_actions}", code="INVALID_ACTION", ) if __name__ == "__main__": main() FILE:scripts/task_aggregator.py #!/usr/bin/env python3 """ project-nerve 任务聚合器从所有已连接的平台获取任务数据，统一格式化后提供搜索、阻碍分析和优先级排序功能。支持 Trello、GitHub Issues、Linear、Notion、Obsidian 五个平台。 """ import json import os import re import sys import urllib.request import urllib.error from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional from utils import ( check_subscription, format_task_table, get_data_file, is_overdue, load_input_data, normalize_priority, normalize_status, now_iso, output_error, output_success, parse_common_args, parse_date, read_json_file, write_json_file, SUPPORTED_PLATFORMS, ) # ============================================================ # 数据文件路径 # ============================================================ SOURCES_FILE = "sources.json" TASKS_CACHE_FILE = "tasks_cache.json" def _get_sources() -> List[Dict[str, Any]]: """读取所有已连接的活跃数据源。""" data = read_json_file(get_data_file(SOURCES_FILE)) if isinstance(data, list): return [s for s in data if s.get("status") == "active"] return [] def _get_cached_tasks() -> List[Dict[str, Any]]: """读取缓存的任务数据。""" data = read_json_file(get_data_file(TASKS_CACHE_FILE)) if isinstance(data, list): return data return [] def _save_cached_tasks(tasks: List[Dict[str, Any]]) -> None: """保存任务数据到缓存文件。""" write_json_file(get_data_file(TASKS_CACHE_FILE), tasks) # ============================================================ # HTTP 请求工具 # ============================================================ def _http_get(url: str, headers: Optional[Dict[str, str]] = None, timeout: int = 15) -> Dict[str, Any]: """发送 HTTP GET 请求。 Args: url: 请求地址。 headers: 请求头。 timeout: 超时秒数。 Returns: 包含 status 和 body 的响应字典。 """ if headers is None: headers = {} req = urllib.request.Request(url, headers=headers, method="GET") try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8") return {"status": resp.status, "body": body} except urllib.error.HTTPError as e: body = "" try: body = e.read().decode("utf-8") except Exception: pass return {"status": e.code, "body": body, "error": str(e)} except Exception as e: return {"status": 0, "body": "", "error": str(e)} def _http_post(url: str, headers: Optional[Dict[str, str]] = None, data: Optional[bytes] = None, timeout: int = 15) -> Dict[str, Any]: """发送 HTTP POST 请求。""" if headers is None: headers = {} req = urllib.request.Request(url, data=data, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8") return {"status": resp.status, "body": body} except urllib.error.HTTPError as e: body = "" try: body = e.read().decode("utf-8") except Exception: pass return {"status": e.code, "body": body, "error": str(e)} except Exception as e: return {"status": 0, "body": "", "error": str(e)} # ============================================================ # 平台适配器 — 任务获取 # ============================================================ def _fetch_trello_tasks(source: Dict[str, Any]) -> List[Dict[str, Any]]: """从 Trello 获取卡片并转换为统一格式。 Args: source: 数据源配置。 Returns: 统一格式的任务列表。 """ api_key = os.environ.get("PNC_TRELLO_API_KEY", "") token = os.environ.get("PNC_TRELLO_TOKEN", "") board_id = source.get("config", {}).get("board_id", "") if not api_key or not token: return [] tasks = [] # 获取看板列表（用于映射状态） lists_map = {} if board_id: url = f"https://api.trello.com/1/boards/{board_id}/lists?key={api_key}&token={token}" resp = _http_get(url) if resp["status"] == 200: try: for lst in json.loads(resp["body"]): lists_map[lst["id"]] = lst.get("name", "") except (json.JSONDecodeError, KeyError): pass # 获取卡片 if board_id: url = f"https://api.trello.com/1/boards/{board_id}/cards?key={api_key}&token={token}&fields=name,desc,idList,labels,due,dateLastActivity,shortUrl,idMembers" resp = _http_get(url) if resp["status"] == 200: try: cards = json.loads(resp["body"]) for card in cards: list_name = lists_map.get(card.get("idList", ""), "") labels = [lb.get("name", "") for lb in card.get("labels", []) if lb.get("name")] # 从标签推断优先级 priority_str = "" for lb in labels: if lb.lower() in ("urgent", "high", "medium", "low", "紧急", "高", "中", "低"): priority_str = lb break tasks.append({ "id": f"trello-{card.get('id', '')}", "source": "trello", "source_id": card.get("id", ""), "title": card.get("name", ""), "description": card.get("desc", ""), "status": normalize_status(list_name), "priority": normalize_priority(priority_str), "assignee": "", "labels": labels, "due_date": (card.get("due") or "")[:10] if card.get("due") else "", "created_at": card.get("dateLastActivity", ""), "updated_at": card.get("dateLastActivity", ""), "url": card.get("shortUrl", ""), }) except (json.JSONDecodeError, KeyError): pass return tasks def _fetch_github_issues(source: Dict[str, Any]) -> List[Dict[str, Any]]: """从 GitHub 获取 Issues 并转换为统一格式。 Args: source: 数据源配置。 Returns: 统一格式的任务列表。 """ token = os.environ.get("PNC_GITHUB_TOKEN", "") repo = source.get("config", {}).get("repo", "") if not token or not repo: return [] tasks = [] url = f"https://api.github.com/repos/{repo}/issues?state=all&per_page=100&sort=updated&direction=desc" headers = { "Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "project-nerve/1.0", } resp = _http_get(url, headers=headers) if resp["status"] == 200: try: issues = json.loads(resp["body"]) for issue in issues: # 跳过 Pull Request（GitHub Issues API 也会返回 PR） if issue.get("pull_request"): continue labels = [lb.get("name", "") for lb in issue.get("labels", []) if lb.get("name")] # 从标签推断优先级 priority_str = "" for lb in labels: lb_lower = lb.lower() if lb_lower in ("urgent", "high", "medium", "low", "p0", "p1", "p2", "p3", "critical", "blocker"): priority_str = lb break state = issue.get("state", "open") assignee = "" if issue.get("assignee"): assignee = issue["assignee"].get("login", "") tasks.append({ "id": f"github-{issue.get('id', '')}", "source": "github", "source_id": str(issue.get("number", "")), "title": issue.get("title", ""), "description": (issue.get("body") or "")[:500], "status": normalize_status(state), "priority": normalize_priority(priority_str), "assignee": assignee, "labels": labels, "due_date": "", "created_at": issue.get("created_at", ""), "updated_at": issue.get("updated_at", ""), "url": issue.get("html_url", ""), }) except (json.JSONDecodeError, KeyError): pass return tasks def _fetch_linear_issues(source: Dict[str, Any]) -> List[Dict[str, Any]]: """从 Linear 获取 Issues 并转换为统一格式。 Args: source: 数据源配置。 Returns: 统一格式的任务列表。 """ api_key = os.environ.get("PNC_LINEAR_API_KEY", "") if not api_key: return [] tasks = [] team_id = source.get("config", {}).get("team_id", "") # 构建 GraphQL 查询 team_filter = "" if team_id: team_filter = f', filter: {{ team: {{ id: {{ eq: "{team_id}" }} }} }}' query_str = json.dumps({ "query": f"""{{ issues(first: 100, orderBy: updatedAt{team_filter}) {{ nodes {{ id identifier title description priority state {{ name }} assignee {{ name }} labels {{ nodes {{ name }} }} dueDate createdAt updatedAt url }} }} }}""" }) url = "https://api.linear.app/graphql" headers = { "Authorization": api_key, "Content-Type": "application/json", } resp = _http_post(url, headers=headers, data=query_str.encode("utf-8")) if resp["status"] == 200: try: result = json.loads(resp["body"]) issues = result.get("data", {}).get("issues", {}).get("nodes", []) # Linear 优先级数值映射: 0=无, 1=紧急, 2=高, 3=中, 4=低 linear_priority_map = {0: "低", 1: "紧急", 2: "高", 3: "中", 4: "低"} for issue in issues: priority_num = issue.get("priority", 0) state_name = "" if issue.get("state"): state_name = issue["state"].get("name", "") assignee_name = "" if issue.get("assignee"): assignee_name = issue["assignee"].get("name", "") labels = [] if issue.get("labels") and issue["labels"].get("nodes"): labels = [lb.get("name", "") for lb in issue["labels"]["nodes"] if lb.get("name")] tasks.append({ "id": f"linear-{issue.get('id', '')}", "source": "linear", "source_id": issue.get("identifier", ""), "title": issue.get("title", ""), "description": (issue.get("description") or "")[:500], "status": normalize_status(state_name), "priority": linear_priority_map.get(priority_num, "中"), "assignee": assignee_name, "labels": labels, "due_date": (issue.get("dueDate") or "")[:10], "created_at": issue.get("createdAt", ""), "updated_at": issue.get("updatedAt", ""), "url": issue.get("url", ""), }) except (json.JSONDecodeError, KeyError): pass return tasks def _fetch_notion_tasks(source: Dict[str, Any]) -> List[Dict[str, Any]]: """从 Notion 数据库获取页面并转换为统一格式。 Args: source: 数据源配置。 Returns: 统一格式的任务列表。 """ token = os.environ.get("PNC_NOTION_TOKEN", "") database_id = source.get("config", {}).get("database_id", "") or os.environ.get("PNC_NOTION_DATABASE_ID", "") if not token or not database_id: return [] tasks = [] url = f"https://api.notion.com/v1/databases/{database_id}/query" headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json", } body = json.dumps({"page_size": 100}).encode("utf-8") resp = _http_post(url, headers=headers, data=body) if resp["status"] == 200: try: result = json.loads(resp["body"]) pages = result.get("results", []) for page in pages: props = page.get("properties", {}) # 提取标题（尝试常见属性名） title = "" for key in ("Name", "名称", "Title", "标题", "Task", "任务"): prop = props.get(key, {}) if prop.get("type") == "title": title_parts = prop.get("title", []) title = "".join(p.get("plain_text", "") for p in title_parts) break # 提取状态 status_str = "" for key in ("Status", "状态", "Stage", "阶段"): prop = props.get(key, {}) if prop.get("type") == "status": status_obj = prop.get("status") if status_obj: status_str = status_obj.get("name", "") break elif prop.get("type") == "select": select_obj = prop.get("select") if select_obj: status_str = select_obj.get("name", "") break # 提取优先级 priority_str = "" for key in ("Priority", "优先级"): prop = props.get(key, {}) if prop.get("type") == "select": select_obj = prop.get("select") if select_obj: priority_str = select_obj.get("name", "") break # 提取负责人 assignee = "" for key in ("Assignee", "负责人", "Owner"): prop = props.get(key, {}) if prop.get("type") == "people": people = prop.get("people", []) if people: assignee = people[0].get("name", "") break # 提取截止日期 due_date = "" for key in ("Due", "截止日期", "Due Date", "Deadline"): prop = props.get(key, {}) if prop.get("type") == "date": date_obj = prop.get("date") if date_obj: due_date = (date_obj.get("start") or "")[:10] break # 提取标签 labels = [] for key in ("Tags", "标签", "Labels"): prop = props.get(key, {}) if prop.get("type") == "multi_select": for opt in prop.get("multi_select", []): labels.append(opt.get("name", "")) break page_url = page.get("url", "") created_time = page.get("created_time", "") updated_time = page.get("last_edited_time", "") tasks.append({ "id": f"notion-{page.get('id', '')}", "source": "notion", "source_id": page.get("id", ""), "title": title, "description": "", "status": normalize_status(status_str), "priority": normalize_priority(priority_str), "assignee": assignee, "labels": labels, "due_date": due_date, "created_at": created_time, "updated_at": updated_time, "url": page_url, }) except (json.JSONDecodeError, KeyError): pass return tasks def _parse_obsidian_frontmatter(content: str) -> Dict[str, str]: """解析 Obsidian 笔记的 YAML frontmatter。提取 frontmatter 中的 status、priority、assignee、due_date 字段。 Args: content: 笔记完整文本内容。 Returns: 包含解析到的字段的字典。 """ result: Dict[str, str] = {} if not content.startswith("---"): return result parts = content.split("---", 2) if len(parts) < 3: return result frontmatter = parts[1] for line in frontmatter.strip().split("\n"): line = line.strip() if ":" not in line: continue key, _, value = line.partition(":") key = key.strip().lower() value = value.strip().strip('"').strip("'") if key in ("status", "priority", "assignee", "due_date", "due"): mapped_key = "due_date" if key == "due" else key result[mapped_key] = value return result def _fetch_obsidian_tasks(source: Dict[str, Any]) -> List[Dict[str, Any]]: """从 Obsidian Vault 扫描任务。扫描 vault 中所有 .md 文件，提取 markdown 复选框格式的任务： - [ ] 未完成任务 - [x] 已完成任务同时解析 frontmatter 中的 priority、assignee、due_date 等字段。 Args: source: 数据源配置。 Returns: 统一格式的任务列表。 """ vault_path = source.get("config", {}).get("vault_path", "") or os.environ.get("PNC_OBSIDIAN_VAULT_PATH", "") task_tag = source.get("config", {}).get("task_tag", "#task") if not vault_path: return [] vault_path = os.path.expanduser(vault_path) if not os.path.isdir(vault_path): return [] tasks = [] # 用于生成唯一 ID 的计数器 task_counter = 0 for root, _dirs, files in os.walk(vault_path): # 跳过 .obsidian 配置目录 if ".obsidian" in root: continue for fname in files: if not fname.endswith(".md"): continue fpath = os.path.join(root, fname) try: with open(fpath, "r", encoding="utf-8") as f: content = f.read() except (IOError, UnicodeDecodeError): continue # 解析 frontmatter 获取全局属性 fm = _parse_obsidian_frontmatter(content) # 计算相对路径（作为来源标识） rel_path = os.path.relpath(fpath, vault_path) note_title = os.path.splitext(fname)[0] # 获取文件修改时间 try: mtime = datetime.fromtimestamp(os.path.getmtime(fpath)).strftime("%Y-%m-%dT%H:%M:%S") except (OSError, ValueError): mtime = "" # 扫描 markdown 复选框 checkbox_pattern = re.compile(r"^(\s*)-\s+\[([ xX])\]\s+(.+)$", re.MULTILINE) for match in checkbox_pattern.finditer(content): checked = match.group(2).lower() == "x" task_text = match.group(3).strip() # 跳过空任务 if not task_text: continue task_counter += 1 task_id = f"obsidian-{task_counter}-{hash(fpath + task_text) & 0xFFFFFFFF:08x}" # 从任务文本中提取内联标签 labels = re.findall(r"#(\w+)", task_text) # 移除标签后的纯文本作为标题 clean_title = re.sub(r"\s*#\w+", "", task_text).strip() if not clean_title: clean_title = task_text # 尝试从任务文本中提取截止日期（如 📅 2026-03-20 或 due:2026-03-20） due_date = fm.get("due_date", "") due_match = re.search(r"(?:📅|due:|截止:?)\s*(\d{4}-\d{2}-\d{2})", task_text) if due_match: due_date = due_match.group(1) tasks.append({ "id": task_id, "source": "obsidian", "source_id": rel_path, "title": clean_title, "description": f"来自笔记: {note_title}", "status": "已完成" if checked else normalize_status(fm.get("status", "待办")), "priority": normalize_priority(fm.get("priority", "")), "assignee": fm.get("assignee", ""), "labels": labels, "due_date": due_date, "created_at": mtime, "updated_at": mtime, "url": f"obsidian://open?vault={os.path.basename(vault_path)}&file={rel_path}", }) return tasks # ============================================================ # 平台获取路由 # ============================================================ _PLATFORM_FETCHERS = { "trello": _fetch_trello_tasks, "github": _fetch_github_issues, "linear": _fetch_linear_issues, "notion": _fetch_notion_tasks, "obsidian": _fetch_obsidian_tasks, } # ============================================================ # 去重逻辑 # ============================================================ def _word_set(text: str) -> set: """提取文本中的词集合（中文按字，英文按单词）。 Args: text: 输入文本。 Returns: 词集合。 """ if not text: return set() # 提取英文单词和中文字符 words = set(re.findall(r"[a-zA-Z0-9]+", text.lower())) # 添加中文字符 for ch in text: if "\u4e00" <= ch <= "\u9fff": words.add(ch) return words def _title_similarity(t1: str, t2: str) -> float: """计算两个标题的相似度（基于词重叠比率）。 Args: t1: 标题一。 t2: 标题二。 Returns: 相似度 0.0~1.0。 """ s1 = _word_set(t1) s2 = _word_set(t2) if not s1 or not s2: return 0.0 intersection = s1 & s2 union = s1 | s2 return len(intersection) / len(union) if union else 0.0 def _dedup_tasks(tasks: List[Dict[str, Any]], threshold: float = 0.8) -> List[Dict[str, Any]]: """对任务列表去重（基于标题相似度）。当两个来自不同平台的任务标题相似度超过阈值时，保留更新时间较晚的那个。 Args: tasks: 原始任务列表。 threshold: 相似度阈值。 Returns: 去重后的任务列表。 """ if not tasks: return tasks result = [] seen_indices = set() for i in range(len(tasks)): if i in seen_indices: continue best = tasks[i] for j in range(i + 1, len(tasks)): if j in seen_indices: continue # 只对不同平台的任务做去重 if tasks[i]["source"] == tasks[j]["source"]: continue sim = _title_similarity(tasks[i]["title"], tasks[j]["title"]) if sim >= threshold: seen_indices.add(j) # 保留更新时间较晚的 if tasks[j].get("updated_at", "") > best.get("updated_at", ""): best = tasks[j] result.append(best) return result # ============================================================ # 操作实现 # ============================================================ def fetch_all(data: Optional[Dict[str, Any]] = None) -> None: """从所有已连接平台获取任务并缓存。 Args: data: 可选参数（platform 过滤）。 """ sources = _get_sources() if not sources: output_error("暂无已连接的数据源，请先使用 source_connector 连接平台", code="NO_SOURCES") return platform_filter = "" if data: platform_filter = data.get("platform", "").strip().lower() all_tasks = [] fetch_errors = [] # 尝试导入自学习引擎（可选依赖） try: from learning_engine import quick_record_error, quick_record_success has_learning = True except ImportError: has_learning = False for source in sources: platform = source.get("platform", "") if platform_filter and platform != platform_filter: continue fetcher = _PLATFORM_FETCHERS.get(platform) if not fetcher: fetch_errors.append(f"平台 {platform} 暂不支持") continue try: tasks = fetcher(source) all_tasks.extend(tasks) # 记录成功 if has_learning and tasks: quick_record_success(platform, "fetch", f"获取到 {len(tasks)} 个任务") except Exception as e: fetch_errors.append(f"{platform} 获取失败: {e}") # 记录错误 if has_learning: quick_record_error(platform, "fetch", "fetch_failure", str(e)) # 去重 all_tasks = _dedup_tasks(all_tasks) # 按更新时间倒序排列 all_tasks.sort(key=lambda t: t.get("updated_at", ""), reverse=True) # 检查显示限制 sub = check_subscription() display_limit = sub["max_tasks_display"] truncated = len(all_tasks) > display_limit display_tasks = all_tasks[:display_limit] # 缓存所有任务 _save_cached_tasks(all_tasks) # 统计 status_stats = {} for task in all_tasks: status = task.get("status", "待办") status_stats[status] = status_stats.get(status, 0) + 1 source_stats = {} for task in all_tasks: src = task.get("source", "未知") source_stats[src] = source_stats.get(src, 0) + 1 result = { "total": len(all_tasks), "displayed": len(display_tasks), "truncated": truncated, "status_stats": status_stats, "source_stats": source_stats, "tasks": display_tasks, "table": format_task_table(display_tasks), } if fetch_errors: result["warnings"] = fetch_errors output_success(result) def search_tasks(data: Dict[str, Any]) -> None: """在缓存的任务中搜索。搜索字段: keyword（标题/描述）、status、priority、source、assignee Args: data: 搜索条件字典。 """ tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 fetch-all 获取任务", code="NO_DATA") return keyword = data.get("keyword", "").strip().lower() status_filter = data.get("status", "").strip() priority_filter = data.get("priority", "").strip() source_filter = data.get("source", "").strip().lower() assignee_filter = data.get("assignee", "").strip().lower() filtered = tasks if keyword: filtered = [ t for t in filtered if keyword in t.get("title", "").lower() or keyword in t.get("description", "").lower() or keyword in " ".join(t.get("labels", [])).lower() ] if status_filter: normalized = normalize_status(status_filter) filtered = [t for t in filtered if t.get("status") == normalized] if priority_filter: normalized = normalize_priority(priority_filter) filtered = [t for t in filtered if t.get("priority") == normalized] if source_filter: filtered = [t for t in filtered if t.get("source") == source_filter] if assignee_filter: filtered = [t for t in filtered if assignee_filter in (t.get("assignee") or "").lower()] sub = check_subscription() display_limit = sub["max_tasks_display"] display_tasks = filtered[:display_limit] output_success({ "total": len(filtered), "displayed": len(display_tasks), "query": {k: v for k, v in data.items() if v}, "tasks": display_tasks, "table": format_task_table(display_tasks), }) def find_blockers(data: Optional[Dict[str, Any]] = None) -> None: """查找阻碍任务（逾期或高优先级进行中任务）。阻碍判定条件： 1. 已逾期的未完成任务 2. 优先级为「紧急」或「高」且状态为「进行中」超过 7 天的任务 Args: data: 可选参数。 """ if not require_blocker_feature(): return tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 fetch-all 获取任务", code="NO_DATA") return blockers = [] now = datetime.now() for task in tasks: status = task.get("status", "") if status in ("已完成", "已关闭"): continue reasons = [] # 条件1：逾期 due_date = task.get("due_date", "") if due_date and is_overdue(due_date): reasons.append(f"已逾期（截止日期: {due_date}）") # 条件2：高优先级长时间进行中 priority = task.get("priority", "") if priority in ("紧急", "高") and status == "进行中": updated = parse_date(task.get("updated_at", "")) if updated and (now - updated).days > 7: reasons.append(f"高优先级任务进行中超过 7 天（上次更新: {task.get('updated_at', '')[:10]}）") if reasons: blocker = dict(task) blocker["blocker_reasons"] = reasons blockers.append(blocker) # 按优先级排序：紧急 > 高 > 中 > 低 priority_order = {"紧急": 0, "高": 1, "中": 2, "低": 3} blockers.sort(key=lambda t: priority_order.get(t.get("priority", "中"), 2)) output_success({ "total": len(blockers), "blockers": blockers, "table": format_task_table(blockers), "summary": f"发现 {len(blockers)} 个阻碍/风险任务" if blockers else "未发现阻碍任务", }) def require_blocker_feature() -> bool: """检查阻碍分析功能的订阅要求。""" sub = check_subscription() if "blocker_analysis" not in sub["features"]: output_error( "「阻碍分析」为付费版功能。当前为免费版，请升级至付费版（¥99/月）以使用此功能。", code="SUBSCRIPTION_REQUIRED", ) return False return True def sort_by_priority(data: Optional[Dict[str, Any]] = None) -> None: """按优先级排序所有任务（逾期优先，然后按优先级）。 Args: data: 可选参数（可指定 status 过滤）。 """ tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 fetch-all 获取任务", code="NO_DATA") return # 过滤已完成/已关闭的任务 active_tasks = [t for t in tasks if t.get("status") not in ("已完成", "已关闭")] if data and data.get("status"): normalized = normalize_status(data["status"]) active_tasks = [t for t in active_tasks if t.get("status") == normalized] # 排序规则：逾期在前，然后按优先级 priority_order = {"紧急": 0, "高": 1, "中": 2, "低": 3} def sort_key(task: Dict) -> tuple: overdue_score = 0 due = task.get("due_date", "") if due and is_overdue(due): overdue_score = -1 # 逾期排在前面 prio = priority_order.get(task.get("priority", "中"), 2) return (overdue_score, prio) active_tasks.sort(key=sort_key) sub = check_subscription() display_limit = sub["max_tasks_display"] display_tasks = active_tasks[:display_limit] output_success({ "total": len(active_tasks), "displayed": len(display_tasks), "tasks": display_tasks, "table": format_task_table(display_tasks), }) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("project-nerve 任务聚合器") args = parser.parse_args() action = args.action.lower().replace("-", "_") try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "fetch_all": lambda: fetch_all(data), "fetch": lambda: fetch_all(data), "search": lambda: search_tasks(data or {}), "blockers": lambda: find_blockers(data), "priorities": lambda: sort_by_priority(data), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join(["fetch-all", "search", "blockers", "priorities"]) output_error(f"未知操作: {args.action}，支持的操作: {valid_actions}", code="INVALID_ACTION") if __name__ == "__main__": main() FILE:scripts/task_graph.py #!/usr/bin/env python3 """ project-nerve 任务关系图谱建模跨平台任务之间的依赖关系，支持关系查询、依赖分析、影响评估和 Mermaid 可视化。灵感来源: ontology (117K 下载量, 326 星) """ import json import os import sys from collections import deque 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, write_json_file, ) # ============================================================ # 常量定义 # ============================================================ GRAPH_FILE = "task_graph.json" # 支持的关系类型 RELATION_TYPES = [ "blocks", # A 阻塞 B（A 完成后 B 才能开始） "blocked_by", # A 被 B 阻塞（B 完成后 A 才能开始） "related_to", # A 与 B 相关（无方向性依赖） "parent_of", # A 是 B 的父任务 "child_of", # A 是 B 的子任务 "duplicates", # A 是 B 的重复 ] # 互为反向的关系映射 _INVERSE_RELATIONS = { "blocks": "blocked_by", "blocked_by": "blocks", "parent_of": "child_of", "child_of": "parent_of", "related_to": "related_to", "duplicates": "duplicates", } # ============================================================ # 数据读写 # ============================================================ def _get_graph() -> Dict[str, Any]: """读取任务关系图谱数据。 Returns: 包含 nodes 和 edges 列表的字典。 """ data = read_json_file(get_data_file(GRAPH_FILE)) if isinstance(data, dict) and "nodes" in data and "edges" in data: return data return { "nodes": [], "edges": [], "metadata": {"created_at": now_iso(), "version": "1.0"}, } def _save_graph(graph: Dict[str, Any]) -> None: """保存图谱数据到文件。 Args: graph: 图谱数据字典。 """ graph["metadata"] = graph.get("metadata", {}) graph["metadata"]["updated_at"] = now_iso() write_json_file(get_data_file(GRAPH_FILE), graph) # ============================================================ # 节点管理 # ============================================================ def _find_node(nodes: List[Dict], node_id: str) -> Optional[Dict]: """查找节点。 Args: nodes: 节点列表。 node_id: 节点 ID。 Returns: 匹配的节点字典，未找到返回 None。 """ for n in nodes: if n.get("id") == node_id: return n return None def _ensure_node(graph: Dict[str, Any], node_id: str, source: str = "", title: str = "") -> Dict: """确保节点存在，若不存在则创建。 Args: graph: 图谱数据。 node_id: 节点 ID。 source: 任务来源平台。 title: 任务标题。 Returns: 节点字典。 """ node = _find_node(graph["nodes"], node_id) if node is None: node = { "id": node_id, "source": source, "title": title, "added_at": now_iso(), } graph["nodes"].append(node) else: # 更新信息（如果提供了新的值） if source and not node.get("source"): node["source"] = source if title and not node.get("title"): node["title"] = title return node def _find_edge(edges: List[Dict], from_id: str, to_id: str, rel_type: str) -> Optional[int]: """查找边的索引。 Args: edges: 边列表。 from_id: 起始节点 ID。 to_id: 目标节点 ID。 rel_type: 关系类型。 Returns: 匹配的边在列表中的索引，未找到返回 None。 """ for i, e in enumerate(edges): if e.get("from") == from_id and e.get("to") == to_id and e.get("type") == rel_type: return i return None # ============================================================ # 操作实现：添加关系 # ============================================================ def add_relation(data: Dict[str, Any]) -> None: """添加任务关系。在两个任务节点之间建立关系边。支持跨平台关系（如 GitHub Issue 阻塞 Trello 卡片）。必填字段: from_id（起始任务 ID）, to_id（目标任务 ID）, type（关系类型）可选字段: from_source, from_title, to_source, to_title 关系类型: blocks, blocked_by, related_to, parent_of, child_of, duplicates Args: data: 关系数据字典。 """ from_id = data.get("from_id", "").strip() to_id = data.get("to_id", "").strip() rel_type = data.get("type", "").strip().lower() if not from_id or not to_id: output_error("起始任务 ID（from_id）和目标任务 ID（to_id）为必填字段", code="VALIDATION_ERROR") return if not rel_type: output_error("关系类型（type）为必填字段", code="VALIDATION_ERROR") return if rel_type not in RELATION_TYPES: valid = "、".join(RELATION_TYPES) output_error(f"不支持的关系类型: {rel_type}，支持的类型: {valid}", code="INVALID_TYPE") return if from_id == to_id: output_error("不能创建自引用关系（from_id 和 to_id 不能相同）", code="VALIDATION_ERROR") return graph = _get_graph() # 确保节点存在 _ensure_node(graph, from_id, data.get("from_source", ""), data.get("from_title", "")) _ensure_node(graph, to_id, data.get("to_source", ""), data.get("to_title", "")) # 检查是否已存在 existing = _find_edge(graph["edges"], from_id, to_id, rel_type) if existing is not None: output_error( f"关系已存在: {from_id} --[{rel_type}]--> {to_id}", code="DUPLICATE_RELATION", ) return # 创建边 edge = { "from": from_id, "to": to_id, "type": rel_type, "created_at": now_iso(), } graph["edges"].append(edge) _save_graph(graph) output_success({ "message": f"已添加关系: {from_id} --[{rel_type}]--> {to_id}", "edge": edge, "total_nodes": len(graph["nodes"]), "total_edges": len(graph["edges"]), }) # ============================================================ # 操作实现：删除关系 # ============================================================ def remove_relation(data: Dict[str, Any]) -> None: """删除任务关系。必填字段: from_id, to_id, type Args: data: 关系标识字典。 """ from_id = data.get("from_id", "").strip() to_id = data.get("to_id", "").strip() rel_type = data.get("type", "").strip().lower() if not from_id or not to_id or not rel_type: output_error("from_id、to_id 和 type 为必填字段", code="VALIDATION_ERROR") return graph = _get_graph() idx = _find_edge(graph["edges"], from_id, to_id, rel_type) if idx is None: output_error( f"未找到关系: {from_id} --[{rel_type}]--> {to_id}", code="NOT_FOUND", ) return removed = graph["edges"].pop(idx) _save_graph(graph) output_success({ "message": f"已删除关系: {from_id} --[{rel_type}]--> {to_id}", "removed_edge": removed, "remaining_edges": len(graph["edges"]), }) # ============================================================ # 操作实现：查询 # ============================================================ def _build_adjacency(edges: List[Dict]) -> Dict[str, List[Dict]]: """构建邻接表（双向）。 Args: edges: 边列表。 Returns: 邻接表字典，键为节点 ID，值为相关边的列表。 """ adj: Dict[str, List[Dict]] = {} for e in edges: from_id = e.get("from", "") to_id = e.get("to", "") if from_id: if from_id not in adj: adj[from_id] = [] adj[from_id].append(e) if to_id: if to_id not in adj: adj[to_id] = [] adj[to_id].append(e) return adj def query(data: Dict[str, Any]) -> None: """查询与指定任务相关的所有任务（BFS 遍历）。从给定任务出发，通过 BFS 遍历找出所有直接和间接相关的任务。必填字段: task_id（要查询的任务 ID）可选字段: max_depth（最大遍历深度，默认 3）, type（过滤关系类型） Args: data: 查询参数字典。 """ task_id = data.get("task_id", "").strip() if not task_id: output_error("任务 ID（task_id）为必填字段", code="VALIDATION_ERROR") return max_depth = int(data.get("max_depth", 3)) type_filter = data.get("type", "").strip().lower() graph = _get_graph() edges = graph["edges"] nodes = graph["nodes"] # 如果指定了关系类型过滤 if type_filter: edges = [e for e in edges if e.get("type") == type_filter] adj = _build_adjacency(edges) # BFS 遍历 visited: Set[str] = set() queue: deque = deque() queue.append((task_id, 0)) visited.add(task_id) related_tasks: List[Dict[str, Any]] = [] related_edges: List[Dict[str, Any]] = [] while queue: current_id, depth = queue.popleft() if depth >= max_depth: continue for edge in adj.get(current_id, []): # 确定另一端的节点 other_id = edge["to"] if edge["from"] == current_id else edge["from"] related_edges.append({ "from": edge.get("from", ""), "to": edge.get("to", ""), "type": edge.get("type", ""), "depth": depth + 1, }) if other_id not in visited: visited.add(other_id) node = _find_node(nodes, other_id) related_tasks.append({ "id": other_id, "source": node.get("source", "") if node else "", "title": node.get("title", "") if node else "", "depth": depth + 1, }) queue.append((other_id, depth + 1)) # 去重边 seen_edges: Set[str] = set() unique_edges = [] for e in related_edges: key = f"{e['from']}|{e['to']}|{e['type']}" if key not in seen_edges: seen_edges.add(key) unique_edges.append(e) output_success({ "task_id": task_id, "related_tasks": related_tasks, "related_edges": unique_edges, "total_related": len(related_tasks), "max_depth": max_depth, }) # ============================================================ # 操作实现：依赖分析 # ============================================================ def _detect_cycles(edges: List[Dict], relation_types: Optional[List[str]] = None) -> List[List[str]]: """检测有向图中的环。使用 DFS 检测循环依赖。 Args: edges: 边列表。 relation_types: 要检查的关系类型，默认检查 blocks 和 parent_of。 Returns: 环路列表，每个环路是节点 ID 的列表。 """ if relation_types is None: relation_types = ["blocks", "parent_of"] # 构建有向邻接表 directed_adj: Dict[str, List[str]] = {} for e in edges: if e.get("type") in relation_types: from_id = e.get("from", "") to_id = e.get("to", "") if from_id not in directed_adj: directed_adj[from_id] = [] directed_adj[from_id].append(to_id) cycles: List[List[str]] = [] visited: Set[str] = set() rec_stack: Set[str] = set() path: List[str] = [] def _dfs(node: str) -> None: """深度优先搜索检测环。""" visited.add(node) rec_stack.add(node) path.append(node) for neighbor in directed_adj.get(node, []): if neighbor not in visited: _dfs(neighbor) elif neighbor in rec_stack: # 找到环 cycle_start = path.index(neighbor) cycle = path[cycle_start:] + [neighbor] cycles.append(cycle) path.pop() rec_stack.discard(node) all_nodes = set() for e in edges: if e.get("type") in relation_types: all_nodes.add(e.get("from", "")) all_nodes.add(e.get("to", "")) for node in all_nodes: if node and node not in visited: _dfs(node) return cycles def dependencies(data: Dict[str, Any]) -> None: """构建依赖树并检测循环依赖。分析任务之间的 blocks/blocked_by 关系，构建依赖树，并检测是否存在循环依赖。必填字段: task_id 可选字段: direction（up 向上追溯依赖 / down 向下展开被阻塞任务，默认 both） Args: data: 查询参数字典。 """ task_id = data.get("task_id", "").strip() if not task_id: output_error("任务 ID（task_id）为必填字段", code="VALIDATION_ERROR") return direction = data.get("direction", "both").strip().lower() graph = _get_graph() edges = graph["edges"] nodes = graph["nodes"] # 只看 blocks 类型的关系 block_edges = [e for e in edges if e.get("type") in ("blocks", "blocked_by")] # 构建有向图（统一为 blocks 方向） blocks_adj: Dict[str, List[str]] = {} # 阻塞者 -> 被阻塞者列表 blocked_by_adj: Dict[str, List[str]] = {} # 被阻塞者 -> 阻塞者列表 for e in block_edges: if e.get("type") == "blocks": blocker = e["from"] blocked = e["to"] else: # blocked_by blocker = e["to"] blocked = e["from"] if blocker not in blocks_adj: blocks_adj[blocker] = [] blocks_adj[blocker].append(blocked) if blocked not in blocked_by_adj: blocked_by_adj[blocked] = [] blocked_by_adj[blocked].append(blocker) # 向上追溯：这个任务被哪些任务阻塞 upstream: List[Dict[str, Any]] = [] if direction in ("up", "both"): visited: Set[str] = set() queue: deque = deque() queue.append((task_id, 0)) visited.add(task_id) while queue: current, depth = queue.popleft() for blocker_id in blocked_by_adj.get(current, []): if blocker_id not in visited: visited.add(blocker_id) node = _find_node(nodes, blocker_id) upstream.append({ "id": blocker_id, "source": node.get("source", "") if node else "", "title": node.get("title", "") if node else "", "depth": depth + 1, }) queue.append((blocker_id, depth + 1)) # 向下展开：这个任务阻塞了哪些任务 downstream: List[Dict[str, Any]] = [] if direction in ("down", "both"): visited_down: Set[str] = set() queue_down: deque = deque() queue_down.append((task_id, 0)) visited_down.add(task_id) while queue_down: current, depth = queue_down.popleft() for blocked_id in blocks_adj.get(current, []): if blocked_id not in visited_down: visited_down.add(blocked_id) node = _find_node(nodes, blocked_id) downstream.append({ "id": blocked_id, "source": node.get("source", "") if node else "", "title": node.get("title", "") if node else "", "depth": depth + 1, }) queue_down.append((blocked_id, depth + 1)) # 检测循环依赖 cycles = _detect_cycles(edges) task_in_cycle = any(task_id in cycle for cycle in cycles) output_success({ "task_id": task_id, "upstream_dependencies": upstream, "downstream_blocked": downstream, "total_upstream": len(upstream), "total_downstream": len(downstream), "has_circular_dependency": task_in_cycle, "circular_dependencies": [c for c in cycles if task_id in c], }) # ============================================================ # 操作实现：影响分析 # ============================================================ def impact(data: Dict[str, Any]) -> None: """分析阻塞一个任务会影响多少下游任务。计算如果指定任务被阻塞，有多少任务会受到影响（直接和间接）。必填字段: task_id Args: data: 查询参数字典。 """ task_id = data.get("task_id", "").strip() if not task_id: output_error("任务 ID（task_id）为必填字段", code="VALIDATION_ERROR") return graph = _get_graph() edges = graph["edges"] nodes = graph["nodes"] # 构建 blocks 方向的有向图 blocks_adj: Dict[str, List[str]] = {} for e in edges: if e.get("type") == "blocks": from_id = e["from"] to_id = e["to"] if from_id not in blocks_adj: blocks_adj[from_id] = [] blocks_adj[from_id].append(to_id) elif e.get("type") == "blocked_by": from_id = e["to"] # 反转方向 to_id = e["from"] if from_id not in blocks_adj: blocks_adj[from_id] = [] blocks_adj[from_id].append(to_id) # BFS 计算所有受影响的下游任务 affected: List[Dict[str, Any]] = [] visited: Set[str] = {task_id} queue: deque = deque() # 从直接被阻塞的任务开始 for blocked_id in blocks_adj.get(task_id, []): if blocked_id not in visited: visited.add(blocked_id) queue.append((blocked_id, 1)) while queue: current, depth = queue.popleft() node = _find_node(nodes, current) affected.append({ "id": current, "source": node.get("source", "") if node else "", "title": node.get("title", "") if node else "", "impact_depth": depth, "impact_type": "直接" if depth == 1 else "间接", }) for next_id in blocks_adj.get(current, []): if next_id not in visited: visited.add(next_id) queue.append((next_id, depth + 1)) # 按影响深度分组统计 depth_stats: Dict[int, int] = {} for a in affected: d = a["impact_depth"] depth_stats[d] = depth_stats.get(d, 0) + 1 # 影响等级评估 impact_level = "无" total_affected = len(affected) if total_affected >= 5: impact_level = "严重" elif total_affected >= 3: impact_level = "较大" elif total_affected >= 1: impact_level = "一般" output_success({ "task_id": task_id, "total_affected": total_affected, "impact_level": impact_level, "affected_tasks": affected, "depth_distribution": depth_stats, "summary": ( f"如果 {task_id} 被阻塞，将影响 {total_affected} 个下游任务" if total_affected > 0 else f"{task_id} 没有下游阻塞任务" ), }) # ============================================================ # 操作实现：可视化（付费功能） # ============================================================ def visualize(data: Optional[Dict[str, Any]] = None) -> None: """生成任务关系的 Mermaid 流程图（付费功能）。将图谱中的任务关系可视化为 Mermaid flowchart 格式。可选字段: task_id（聚焦某个任务的子图）, type（过滤关系类型） Args: data: 可选的过滤参数。 """ if not require_paid_feature("mermaid_chart", "任务关系图谱可视化"): return graph = _get_graph() edges = graph["edges"] nodes = graph["nodes"] if not edges: output_success({ "message": "图谱中暂无关系数据", "mermaid": "", }) return task_filter = "" type_filter = "" if data: task_filter = data.get("task_id", "").strip() type_filter = data.get("type", "").strip().lower() # 过滤 filtered_edges = edges if type_filter: filtered_edges = [e for e in filtered_edges if e.get("type") == type_filter] if task_filter: # 只保留与指定任务相关的边 filtered_edges = [ e for e in filtered_edges if e.get("from") == task_filter or e.get("to") == task_filter ] if not filtered_edges: output_success({ "message": "过滤后无匹配的关系数据", "mermaid": "", }) return # 生成 Mermaid 代码 lines = ["```mermaid", "flowchart TD"] # 收集涉及的节点 involved_nodes: Set[str] = set() for e in filtered_edges: involved_nodes.add(e.get("from", "")) involved_nodes.add(e.get("to", "")) # 节点定义（使用方括号表示节点，显示标题） node_map: Dict[str, str] = {} for node_id in involved_nodes: if not node_id: continue node = _find_node(nodes, node_id) # 生成安全的 Mermaid 节点 ID（替换特殊字符） safe_id = node_id.replace("-", "_").replace(".", "_").replace(" ", "_") title = node.get("title", node_id) if node else node_id source = node.get("source", "") if node else "" label = title if source: label = f"{title}\\n[{source}]" # 截断过长的标签 if len(label) > 50: label = label[:47] + "..." lines.append(f" {safe_id}[\"{label}\"]") node_map[node_id] = safe_id # 关系类型到箭头样式的映射 arrow_styles = { "blocks": "-->|阻塞|", "blocked_by": "-->|被阻塞|", "related_to": "---|相关|", "parent_of": "-->|父任务|", "child_of": "-->|子任务|", "duplicates": "-.->|重复|", } # 边定义 for e in filtered_edges: from_safe = node_map.get(e.get("from", ""), "") to_safe = node_map.get(e.get("to", ""), "") rel = e.get("type", "related_to") arrow = arrow_styles.get(rel, "-->") if from_safe and to_safe: lines.append(f" {from_safe} {arrow} {to_safe}") lines.append("```") mermaid_code = "\n".join(lines) output_success({ "message": f"已生成包含 {len(involved_nodes)} 个节点和 {len(filtered_edges)} 条关系的图谱", "mermaid": mermaid_code, "node_count": len(involved_nodes), "edge_count": len(filtered_edges), }) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("project-nerve 任务关系图谱") args = parser.parse_args() action = args.action.lower().replace("-", "_") try: data = load_input_data(args) except ValueError as e: output_error(str(e), code="INPUT_ERROR") return actions = { "add_relation": lambda: add_relation(data or {}), "remove_relation": lambda: remove_relation(data or {}), "query": lambda: query(data or {}), "dependencies": lambda: dependencies(data or {}), "impact": lambda: impact(data or {}), "visualize": lambda: visualize(data), } handler = actions.get(action) if handler: handler() else: valid_actions = "、".join([ "add-relation", "remove-relation", "query", "dependencies", "impact", "visualize", ]) output_error( f"未知操作: {args.action}，支持的操作: {valid_actions}", code="INVALID_ACTION", ) if __name__ == "__main__": main() FILE:scripts/utils.py #!/usr/bin/env python3 """ project-nerve 共享工具模块提供数据存储、订阅校验、优先级/状态标准化、数据格式化等通用功能。跨平台项目管理聚合器的基础工具集。 """ import argparse 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", "project-nerve") # 统一优先级定义 PRIORITIES = ["紧急", "高", "中", "低"] # 统一状态定义 STATUSES = ["待办", "进行中", "已完成", "已关闭"] # 支持的平台列表 SUPPORTED_PLATFORMS = ["trello", "github", "linear", "notion", "obsidian"] # 优先级映射表：将各平台的优先级字符串映射到统一优先级 _PRIORITY_MAP: Dict[str, str] = { # 中文 "紧急": "紧急", "urgent": "紧急", "critical": "紧急", "p0": "紧急", "highest": "紧急", "blocker": "紧急", "高": "高", "high": "高", "p1": "高", "important": "高", "中": "中", "medium": "中", "normal": "中", "p2": "中", "default": "中", "低": "低", "low": "低", "minor": "低", "p3": "低", "trivial": "低", "none": "低", "no priority": "低", "nopriority": "低", } # 状态映射表：将各平台的状态字符串映射到统一状态 _STATUS_MAP: Dict[str, str] = { # 中文 "待办": "待办", "未开始": "待办", "进行中": "进行中", "处理中": "进行中", "已完成": "已完成", "完成": "已完成", "已关闭": "已关闭", "关闭": "已关闭", # 英文 — Trello / GitHub / Linear / Notion 常见状态 "todo": "待办", "to do": "待办", "backlog": "待办", "open": "待办", "new": "待办", "not started": "待办", "triage": "待办", "in progress": "进行中", "in_progress": "进行中", "doing": "进行中", "started": "进行中", "active": "进行中", "in review": "进行中", "in_review": "进行中", "done": "已完成", "completed": "已完成", "resolved": "已完成", "merged": "已完成", "closed": "已关闭", "cancelled": "已关闭", "canceled": "已关闭", "archived": "已关闭", "won't fix": "已关闭", "wontfix": "已关闭", "duplicate": "已关闭", } # ============================================================ # 数据目录管理 # ============================================================ def get_data_dir() -> str: """获取数据存储目录路径。优先读取环境变量 PNC_DATA_DIR，若未设置则使用默认路径 ~/.openclaw-bdi/project-nerve/。自动创建目录（若不存在）。 Returns: 数据目录的绝对路径。 """ data_dir = os.environ.get("PNC_DATA_DIR", DEFAULT_DATA_DIR) os.makedirs(data_dir, exist_ok=True) return data_dir def get_data_file(filename: str) -> str: """获取数据文件的完整路径。 Args: filename: 文件名（如 "sources.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 = "project-nerve 项目管理工具") -> 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": 2, "max_tasks_display": 50, "features": [ "basic_query", "task_list", "source_connect", ], }, "paid": { "tier": "paid", "max_sources": 10, "max_tasks_display": 500, "features": [ "basic_query", "task_list", "source_connect", "sprint_analytics", "standup_report", "blocker_analysis", "mermaid_chart", "bulk_sync", ], }, } def check_subscription(tier: Optional[str] = None) -> Dict[str, Any]: """检查当前订阅等级并返回对应的限制配置。优先使用传入的 tier 参数，否则读取环境变量 PNC_SUBSCRIPTION_TIER。若都未设置，默认为 "free" 等级。 Args: tier: 订阅等级（"free" 或 "paid"），可选。 Returns: 包含订阅限制信息的字典。 Raises: ValueError: 当传入的等级无效时抛出。 """ if tier is None: tier = os.environ.get("PNC_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}」为付费版功能。当前为免费版，请升级至付费版（¥99/月）以使用此功能。", code="SUBSCRIPTION_REQUIRED", ) return False return True # ============================================================ # 标准化工具函数 # ============================================================ def normalize_priority(p: str) -> str: """将各平台的优先级字符串映射到统一优先级。支持中文（紧急/高/中/低）和英文（urgent/high/medium/low 等）。无法识别的优先级默认返回 "中"。 Args: p: 原始优先级字符串。 Returns: 统一优先级：紧急 / 高 / 中 / 低。 """ if not p: return "中" key = p.strip().lower() return _PRIORITY_MAP.get(key, "中") def normalize_status(s: str) -> str: """将各平台的状态字符串映射到统一状态。支持中文（待办/进行中/已完成/已关闭）和英文常见状态。无法识别的状态默认返回 "待办"。 Args: s: 原始状态字符串。 Returns: 统一状态：待办 / 进行中 / 已完成 / 已关闭。 """ if not s: return "待办" key = s.strip().lower() return _STATUS_MAP.get(key, "待办") def format_task_table(tasks: List[Dict[str, Any]]) -> str: """将任务列表格式化为 Markdown 表格。表格列：序号 | 标题 | 平台 | 状态 | 优先级 | 负责人 | 截止日期 Args: tasks: 统一格式的任务字典列表。 Returns: Markdown 表格字符串。 """ if not tasks: return "暂无任务数据。" lines = [ "| # | 标题 | 平台 | 状态 | 优先级 | 负责人 | 截止日期 |", "|---|------|------|------|--------|--------|----------|", ] for idx, task in enumerate(tasks, start=1): title = task.get("title", "无标题") # 截断过长标题 if len(title) > 40: title = title[:37] + "..." source = task.get("source", "-") status = task.get("status", "-") priority = task.get("priority", "-") assignee = task.get("assignee", "-") or "-" due_date = task.get("due_date", "-") or "-" lines.append(f"| {idx} | {title} | {source} | {status} | {priority} | {assignee} | {due_date} |") return "\n".join(lines) # ============================================================ # ID 与时间工具 # ============================================================ def generate_id(prefix: str = "T") -> str: """生成唯一 ID。基于时间戳生成，格式为前缀+时间戳。 Args: prefix: ID 前缀，默认为 "T"（任务）。 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") # ============================================================ # Mermaid 图表生成 # ============================================================ def generate_pie_chart(title: str, data: List[Dict[str, Any]]) -> str: """生成 Mermaid 饼图。 Args: title: 图表标题。 data: 数据列表，每项包含 label 和 value。 Returns: Mermaid 饼图代码块字符串。 """ lines = ["```mermaid", f"pie title {title}"] for item in data: label = item.get("label", "未知") value = item.get("value", 0) lines.append(f' "{label}" : {value}') lines.append("```") return "\n".join(lines) def generate_bar_chart( title: str, data: List[Dict[str, Any]], x_label: str = "类别", y_label: str = "数值", ) -> str: """生成 Mermaid xychart-beta 柱状图。 Args: title: 图表标题。 data: 数据列表，每项包含 label 和 value。 x_label: X 轴标签。 y_label: Y 轴标签。 Returns: Mermaid 柱状图代码块字符串。 """ labels = [f'"{item.get("label", "")}"' for item in data] values = [str(item.get("value", 0)) for item in data] lines = [ "```mermaid", "xychart-beta", f' title "{title}"', f' x-axis [{", ".join(labels)}]', f' y-axis "{y_label}"', f' bar [{", ".join(values)}]', "```", ] return "\n".join(lines) def generate_line_chart( title: str, data: List[Dict[str, Any]], x_label: str = "时间", y_label: str = "数值", ) -> str: """生成 Mermaid xychart-beta 折线图。 Args: title: 图表标题。 data: 数据列表，每项包含 label 和 value。 x_label: X 轴标签。 y_label: Y 轴标签。 Returns: Mermaid 折线图代码块字符串。 """ labels = [f'"{item.get("label", "")}"' for item in data] values = [str(item.get("value", 0)) for item in data] lines = [ "```mermaid", "xychart-beta", f' title "{title}"', f' x-axis [{", ".join(labels)}]', f' y-axis "{y_label}"', f' line [{", ".join(values)}]', "```", ] return "\n".join(lines) # ============================================================ # 日期辅助函数 # ============================================================ def calculate_days_since(date_str: str) -> int: """计算从指定日期到今天的天数。 Args: date_str: 日期字符串，格式为 YYYY-MM-DD 或 ISO 格式。 Returns: 距今天数（正数表示过去，负数表示未来）。 """ try: if "T" in date_str: dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")) dt = dt.replace(tzinfo=None) else: dt = datetime.strptime(date_str, "%Y-%m-%d") delta = datetime.now() - dt return delta.days except (ValueError, TypeError): return 0 def is_overdue(due_date: str) -> bool: """判断任务是否已逾期。 Args: due_date: 截止日期字符串（YYYY-MM-DD 或 ISO 格式）。 Returns: True 表示已逾期，False 表示未逾期或无截止日期。 """ if not due_date: return False days = calculate_days_since(due_date) return days > 0 def parse_date(date_str: str) -> Optional[datetime]: """解析日期字符串为 datetime 对象。支持 YYYY-MM-DD 和 ISO 格式。 Args: date_str: 日期字符串。 Returns: datetime 对象，解析失败返回 None。 """ if not date_str: return None try: if "T" in date_str: dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")) return dt.replace(tzinfo=None) else: return datetime.strptime(date_str, "%Y-%m-%d") except (ValueError, TypeError): return None FILE:scripts/standup_generator.py #!/usr/bin/env python3 """ project-nerve 站会报告生成器扫描最近的任务活动，生成每日站会和每周总结报告。支持跨平台汇总，输出格式化 Markdown。 """ import json import os import sys from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from utils import ( check_subscription, format_task_table, get_data_file, is_overdue, load_input_data, normalize_status, now_iso, output_error, output_success, parse_common_args, parse_date, read_json_file, require_paid_feature, today_str, ) # ============================================================ # 数据文件路径 # ============================================================ TASKS_CACHE_FILE = "tasks_cache.json" def _get_cached_tasks() -> List[Dict[str, Any]]: """读取缓存的任务数据。""" data = read_json_file(get_data_file(TASKS_CACHE_FILE)) if isinstance(data, list): return data return [] # ============================================================ # 时间过滤辅助 # ============================================================ def _tasks_updated_since(tasks: List[Dict[str, Any]], since: datetime) -> List[Dict[str, Any]]: """过滤出指定时间之后有更新的任务。 Args: tasks: 任务列表。 since: 起始时间。 Returns: 过滤后的任务列表。 """ result = [] for task in tasks: dt = parse_date(task.get("updated_at", "")) if dt and dt >= since: result.append(task) return result def _tasks_by_status(tasks: List[Dict[str, Any]], status: str) -> List[Dict[str, Any]]: """按状态过滤任务。""" return [t for t in tasks if t.get("status") == status] def _task_summary_line(task: Dict[str, Any]) -> str: """生成单个任务的摘要行。 Args: task: 任务字典。 Returns: 格式化的摘要行字符串。 """ title = task.get("title", "无标题") if len(title) > 50: title = title[:47] + "..." source = task.get("source", "") priority = task.get("priority", "") assignee = task.get("assignee", "") parts = [f"**{title}**"] if source: parts.append(f"[{source}]") if priority and priority in ("紧急", "高"): parts.append(f"({priority})") if assignee: parts.append(f"@{assignee}") return " ".join(parts) # ============================================================ # 操作实现 # ============================================================ def daily_standup(data: Optional[Dict[str, Any]] = None) -> None: """生成每日站会报告。扫描过去 24 小时的任务活动，生成标准站会格式： - 昨日完成 - 今日计划 - 阻碍事项 Args: data: 可选参数（assignee 过滤）。 """ tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 task_aggregator fetch-all 获取任务", code="NO_DATA") return now = datetime.now() yesterday = now - timedelta(hours=24) today_date = today_str() # 可选：按负责人过滤 assignee = "" if data: assignee = data.get("assignee", "").strip() if assignee: tasks = [t for t in tasks if assignee.lower() in (t.get("assignee") or "").lower()] # 最近 24 小时有更新的任务 recent_tasks = _tasks_updated_since(tasks, yesterday) # 昨日完成：最近 24h 内状态变为已完成的任务 completed = _tasks_by_status(recent_tasks, "已完成") # 今日计划：当前进行中 + 待办（按优先级排序，取前 10） in_progress = _tasks_by_status(tasks, "进行中") todo_tasks = _tasks_by_status(tasks, "待办") priority_order = {"紧急": 0, "高": 1, "中": 2, "低": 3} planned = sorted( in_progress + todo_tasks, key=lambda t: priority_order.get(t.get("priority", "中"), 2) )[:10] # 阻碍事项：逾期或紧急未完成 blockers = [] for task in tasks: if task.get("status") in ("已完成", "已关闭"): continue reasons = [] if task.get("due_date") and is_overdue(task["due_date"]): reasons.append(f"逾期（截止: {task['due_date']}）") if task.get("priority") == "紧急" and task.get("status") == "进行中": reasons.append("紧急任务进行中") if reasons: blockers.append({"task": task, "reasons": reasons}) # 构建 Markdown 报告 report_parts = [] title_suffix = f" — {assignee}" if assignee else "" report_parts.append(f"# 每日站会{title_suffix} — {today_date}\n") # 昨日完成 report_parts.append("## 昨日完成\n") if completed: for task in completed: report_parts.append(f"- {_task_summary_line(task)}") else: report_parts.append("- （暂无完成任务）") report_parts.append("") # 今日计划 report_parts.append("## 今日计划\n") if planned: for task in planned: status_tag = "进行中" if task.get("status") == "进行中" else "待启动" report_parts.append(f"- [{status_tag}] {_task_summary_line(task)}") else: report_parts.append("- （暂无计划任务）") report_parts.append("") # 阻碍事项 report_parts.append("## 阻碍事项\n") if blockers: for item in blockers: task = item["task"] reasons = "、".join(item["reasons"]) report_parts.append(f"- {_task_summary_line(task)} — {reasons}") else: report_parts.append("- （暂无阻碍）") report_parts.append("") # 统计摘要 report_parts.append("---") report_parts.append(f"完成 {len(completed)} | 计划 {len(planned)} | 阻碍 {len(blockers)}") report_parts.append(f"\n*由 project-nerve 自动生成 — {now_iso()}*") report_md = "\n".join(report_parts) output_success({ "report": report_md, "summary": { "date": today_date, "completed_count": len(completed), "planned_count": len(planned), "blocker_count": len(blockers), }, }) def weekly_report(data: Optional[Dict[str, Any]] = None) -> None: """生成每周总结报告。扫描过去 7 天的任务活动，生成周报格式： - 本周完成 - 进行中 - 下周计划 Args: data: 可选参数。 """ if not require_paid_feature("standup_report", "周报生成"): return tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 task_aggregator fetch-all 获取任务", code="NO_DATA") return now = datetime.now() week_ago = now - timedelta(days=7) today_date = today_str() week_start = (now - timedelta(days=now.weekday())).strftime("%Y-%m-%d") # 可选：按负责人过滤 assignee = "" if data: assignee = data.get("assignee", "").strip() if assignee: tasks = [t for t in tasks if assignee.lower() in (t.get("assignee") or "").lower()] # 本周有更新的任务 recent_tasks = _tasks_updated_since(tasks, week_ago) # 本周完成 completed = _tasks_by_status(recent_tasks, "已完成") # 进行中 in_progress = _tasks_by_status(tasks, "进行中") # 下周计划（待办按优先级排序取前 15） todo_tasks = _tasks_by_status(tasks, "待办") priority_order = {"紧急": 0, "高": 1, "中": 2, "低": 3} next_week_plan = sorted( todo_tasks, key=lambda t: priority_order.get(t.get("priority", "中"), 2) )[:15] # 按平台统计完成情况 source_completed = {} for task in completed: src = task.get("source", "未知") source_completed[src] = source_completed.get(src, 0) + 1 # 按优先级统计进行中任务 priority_in_progress = {} for task in in_progress: prio = task.get("priority", "中") priority_in_progress[prio] = priority_in_progress.get(prio, 0) + 1 # 逾期统计 overdue_tasks = [ t for t in tasks if t.get("status") not in ("已完成", "已关闭") and t.get("due_date") and is_overdue(t["due_date"]) ] # 构建 Markdown 报告 report_parts = [] title_suffix = f" — {assignee}" if assignee else "" report_parts.append(f"# 周报{title_suffix} — {week_start} ~ {today_date}\n") # 概览 report_parts.append("## 本周概览\n") report_parts.append("| 指标 | 数值 |") report_parts.append("|------|------|") report_parts.append(f"| 本周完成 | {len(completed)} |") report_parts.append(f"| 进行中 | {len(in_progress)} |") report_parts.append(f"| 待办 | {len(todo_tasks)} |") report_parts.append(f"| 逾期 | {len(overdue_tasks)} |") report_parts.append("") # 本周完成 report_parts.append("## 本周完成\n") if completed: for task in completed: report_parts.append(f"- {_task_summary_line(task)}") report_parts.append("") # 平台分布 if source_completed: report_parts.append("**按平台统计**:") for src, cnt in sorted(source_completed.items(), key=lambda x: x[1], reverse=True): report_parts.append(f"- {src}: {cnt} 个") else: report_parts.append("- （本周暂无完成任务）") report_parts.append("") # 进行中 report_parts.append("## 进行中\n") if in_progress: for task in in_progress[:10]: report_parts.append(f"- {_task_summary_line(task)}") if len(in_progress) > 10: report_parts.append(f"- ...等共 {len(in_progress)} 个任务") else: report_parts.append("- （暂无进行中任务）") report_parts.append("") # 下周计划 report_parts.append("## 下周计划\n") if next_week_plan: for task in next_week_plan: report_parts.append(f"- {_task_summary_line(task)}") else: report_parts.append("- （暂无计划任务）") report_parts.append("") # 风险提醒 if overdue_tasks: report_parts.append("## 风险提醒\n") for task in overdue_tasks[:5]: report_parts.append(f"- {_task_summary_line(task)} — 逾期（截止: {task.get('due_date', '')}）") if len(overdue_tasks) > 5: report_parts.append(f"- ...等共 {len(overdue_tasks)} 个逾期任务") report_parts.append("") report_parts.append(f"---\n*由 project-nerve 自动生成 — {now_iso()}*") report_md = "\n".join(report_parts) output_success({ "report": report_md, "summary": { "period": f"{week_start} ~ {today_date}", "completed_count": len(completed), "in_progress_count": len(in_progress), "todo_count": len(todo_tasks), "overdue_count": len(overdue_tasks), }, }) def generate_standup(data: Optional[Dict[str, Any]] = None) -> None: """智能生成站会报告（自动选择日报或周报）。如果是周一，生成周报；否则生成日报。 Args: data: 可选参数。 """ now = datetime.now() report_type = "daily" if data: report_type = data.get("type", "daily").strip().lower() # 周一自动切换为周报 if report_type == "auto": if now.weekday() == 0: # 周一 report_type = "weekly" else: report_type = "daily" if report_type == "weekly": weekly_report(data) else: daily_standup(data) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("project-nerve 站会报告生成器") 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 = { "daily": lambda: daily_standup(data), "weekly": lambda: weekly_report(data), "generate": lambda: generate_standup(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/task_writer.py #!/usr/bin/env python3 """ project-nerve 任务写入器在指定平台上创建、更新、移动任务和添加评论。支持自动检测最适合的平台（含 Obsidian），也可手动指定。 """ import json import os import sys import urllib.request import urllib.error from typing import Any, Dict, List, Optional from utils import ( generate_id, get_data_file, load_input_data, normalize_priority, normalize_status, now_iso, output_error, output_success, parse_common_args, read_json_file, SUPPORTED_PLATFORMS, ) # ============================================================ # 数据文件路径 # ============================================================ SOURCES_FILE = "sources.json" def _get_sources() -> List[Dict[str, Any]]: """读取所有已连接的活跃数据源。""" data = read_json_file(get_data_file(SOURCES_FILE)) if isinstance(data, list): return [s for s in data if s.get("status") == "active"] return [] def _find_source_by_platform(sources: List[Dict], platform: str) -> Optional[Dict]: """根据平台类型查找数据源配置。""" for s in sources: if s.get("platform") == platform: return s return None # ============================================================ # HTTP 请求工具 # ============================================================ def _http_request( url: str, method: str = "GET", headers: Optional[Dict[str, str]] = None, data: Optional[bytes] = None, timeout: int = 15, ) -> Dict[str, Any]: """发送 HTTP 请求并返回响应。 Args: url: 请求地址。 method: HTTP 方法。 headers: 请求头。 data: 请求体。 timeout: 超时秒数。 Returns: 包含 status 和 body 的响应字典。 """ if headers is None: headers = {} req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8") return {"status": resp.status, "body": body} except urllib.error.HTTPError as e: body = "" try: body = e.read().decode("utf-8") except Exception: pass return {"status": e.code, "body": body, "error": str(e)} except Exception as e: return {"status": 0, "body": "", "error": str(e)} # ============================================================ # 平台自动检测 # ============================================================ # 关键词到平台的映射规则 _PLATFORM_KEYWORDS: Dict[str, List[str]] = { "github": ["bug", "fix", "pr", "pull request", "merge", "commit", "branch", "issue", "代码", "修复", "缺陷", "分支", "合并"], "obsidian": ["note", "notes", "wiki", "knowledge", "vault", "memo", "journal", "笔记", "知识", "日记", "备忘", "本地"], "notion": ["doc", "document", "design", "page", "database", "文档", "设计", "数据库", "规划", "在线"], "linear": ["sprint", "story", "epic", "cycle", "roadmap", "feature", "迭代", "用户故事", "史诗", "路线图", "功能"], "trello": [], # 默认平台 } def _detect_platform(title: str, description: str = "") -> str: """根据任务标题和描述自动检测最适合的平台。匹配规则： - 代码/Bug/PR 相关 → GitHub - 文档/设计相关 → Notion - Sprint/Story 相关 → Linear - 默认 → Trello Args: title: 任务标题。 description: 任务描述。 Returns: 推荐的平台名称。 """ text = f"{title} {description}".lower() best_platform = "trello" best_score = 0 for platform, keywords in _PLATFORM_KEYWORDS.items(): score = 0 for kw in keywords: if kw in text: score += 1 if score > best_score: best_score = score best_platform = platform return best_platform # ============================================================ # 平台适配器 — 创建任务 # ============================================================ def _create_trello_card(source: Dict[str, Any], task_data: Dict[str, Any]) -> Dict[str, Any]: """在 Trello 创建卡片。 Args: source: 数据源配置。 task_data: 任务数据（title, description, due_date 等）。 Returns: 创建结果字典。 """ api_key = os.environ.get("PNC_TRELLO_API_KEY", "") token = os.environ.get("PNC_TRELLO_TOKEN", "") board_id = source.get("config", {}).get("board_id", "") if not api_key or not token: return {"success": False, "message": "缺少 Trello 凭据"} # 获取第一个列表作为默认列表 list_id = task_data.get("list_id", "") if not list_id and board_id: url = f"https://api.trello.com/1/boards/{board_id}/lists?key={api_key}&token={token}" resp = _http_request(url) if resp["status"] == 200: try: lists = json.loads(resp["body"]) if lists: list_id = lists[0]["id"] except (json.JSONDecodeError, KeyError, IndexError): pass if not list_id: return {"success": False, "message": "无法确定 Trello 列表，请指定 list_id 或 board_id"} # 创建卡片 params = f"key={api_key}&token={token}" create_url = f"https://api.trello.com/1/cards?{params}" card_data = { "name": task_data.get("title", ""), "desc": task_data.get("description", ""), "idList": list_id, } if task_data.get("due_date"): card_data["due"] = task_data["due_date"] body = json.dumps(card_data).encode("utf-8") headers = {"Content-Type": "application/json"} resp = _http_request(create_url, method="POST", headers=headers, data=body) if resp["status"] in (200, 201): try: card = json.loads(resp["body"]) return { "success": True, "message": f"Trello 卡片已创建: {card.get('name', '')}", "task_id": card.get("id", ""), "url": card.get("shortUrl", ""), } except json.JSONDecodeError: return {"success": True, "message": "Trello 卡片已创建"} else: return {"success": False, "message": f"Trello 创建失败 (HTTP {resp['status']}): {resp.get('error', '')}"} def _create_github_issue(source: Dict[str, Any], task_data: Dict[str, Any]) -> Dict[str, Any]: """在 GitHub 创建 Issue。 Args: source: 数据源配置。 task_data: 任务数据。 Returns: 创建结果字典。 """ token = os.environ.get("PNC_GITHUB_TOKEN", "") repo = source.get("config", {}).get("repo", "") if not token or not repo: return {"success": False, "message": "缺少 GitHub 凭据或仓库信息"} url = f"https://api.github.com/repos/{repo}/issues" headers = { "Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "Content-Type": "application/json", "User-Agent": "project-nerve/1.0", } issue_data = { "title": task_data.get("title", ""), "body": task_data.get("description", ""), } # 添加标签 labels = task_data.get("labels", []) priority = task_data.get("priority", "") if priority: labels.append(f"priority:{priority}") if labels: issue_data["labels"] = labels # 添加负责人 assignee = task_data.get("assignee", "") if assignee: issue_data["assignees"] = [assignee] body = json.dumps(issue_data).encode("utf-8") resp = _http_request(url, method="POST", headers=headers, data=body) if resp["status"] in (200, 201): try: issue = json.loads(resp["body"]) return { "success": True, "message": f"GitHub Issue 已创建: #{issue.get('number', '')} {issue.get('title', '')}", "task_id": str(issue.get("number", "")), "url": issue.get("html_url", ""), } except json.JSONDecodeError: return {"success": True, "message": "GitHub Issue 已创建"} else: return {"success": False, "message": f"GitHub 创建失败 (HTTP {resp['status']}): {resp.get('error', '')}"} def _create_linear_issue(source: Dict[str, Any], task_data: Dict[str, Any]) -> Dict[str, Any]: """在 Linear 创建 Issue。 Args: source: 数据源配置。 task_data: 任务数据。 Returns: 创建结果字典。 """ api_key = os.environ.get("PNC_LINEAR_API_KEY", "") if not api_key: return {"success": False, "message": "缺少 Linear API Key"} team_id = source.get("config", {}).get("team_id", "") if not team_id: return {"success": False, "message": "缺少 Linear Team ID，请在数据源配置中指定 team_id"} # Linear 优先级映射: 紧急=1, 高=2, 中=3, 低=4 priority_map = {"紧急": 1, "高": 2, "中": 3, "低": 4} priority = priority_map.get(normalize_priority(task_data.get("priority", "")), 3) mutation = json.dumps({ "query": """mutation CreateIssue($input: IssueCreateInput!) { issueCreate(input: $input) { success issue { id identifier title url } } }""", "variables": { "input": { "teamId": team_id, "title": task_data.get("title", ""), "description": task_data.get("description", ""), "priority": priority, } } }) url = "https://api.linear.app/graphql" headers = { "Authorization": api_key, "Content-Type": "application/json", } resp = _http_request(url, method="POST", headers=headers, data=mutation.encode("utf-8")) if resp["status"] == 200: try: result = json.loads(resp["body"]) issue_create = result.get("data", {}).get("issueCreate", {}) if issue_create.get("success"): issue = issue_create.get("issue", {}) return { "success": True, "message": f"Linear Issue 已创建: {issue.get('identifier', '')} {issue.get('title', '')}", "task_id": issue.get("identifier", ""), "url": issue.get("url", ""), } else: return {"success": False, "message": "Linear 创建失败: API 返回失败状态"} except (json.JSONDecodeError, KeyError): return {"success": False, "message": "Linear 创建失败: 响应解析错误"} else: return {"success": False, "message": f"Linear 创建失败 (HTTP {resp['status']}): {resp.get('error', '')}"} def _create_notion_page(source: Dict[str, Any], task_data: Dict[str, Any]) -> Dict[str, Any]: """在 Notion 数据库创建页面。 Args: source: 数据源配置。 task_data: 任务数据。 Returns: 创建结果字典。 """ token = os.environ.get("PNC_NOTION_TOKEN", "") database_id = source.get("config", {}).get("database_id", "") or os.environ.get("PNC_NOTION_DATABASE_ID", "") if not token or not database_id: return {"success": False, "message": "缺少 Notion 凭据或数据库 ID"} url = "https://api.notion.com/v1/pages" headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json", } # 构建页面数据 properties = { "Name": { "title": [ {"text": {"content": task_data.get("title", "")}} ] } } page_data = { "parent": {"database_id": database_id}, "properties": properties, } # 添加内容块 description = task_data.get("description", "") if description: page_data["children"] = [ { "object": "block", "type": "paragraph", "paragraph": { "rich_text": [{"text": {"content": description}}] } } ] body = json.dumps(page_data).encode("utf-8") resp = _http_request(url, method="POST", headers=headers, data=body) if resp["status"] in (200, 201): try: page = json.loads(resp["body"]) return { "success": True, "message": f"Notion 页面已创建: {task_data.get('title', '')}", "task_id": page.get("id", ""), "url": page.get("url", ""), } except json.JSONDecodeError: return {"success": True, "message": "Notion 页面已创建"} else: return {"success": False, "message": f"Notion 创建失败 (HTTP {resp['status']}): {resp.get('error', '')}"} def _create_obsidian_task(source: Dict[str, Any], task_data: Dict[str, Any]) -> Dict[str, Any]: """在 Obsidian Vault 中创建任务笔记。在 vault 中创建新的 markdown 文件，包含 frontmatter 元数据和 markdown 复选框格式的任务项。 Args: source: 数据源配置。 task_data: 任务数据（title, description, priority, due_date 等）。 Returns: 创建结果字典。 """ vault_path = source.get("config", {}).get("vault_path", "") or os.environ.get("PNC_OBSIDIAN_VAULT_PATH", "") task_tag = source.get("config", {}).get("task_tag", "#task") if not vault_path: return {"success": False, "message": "缺少 Obsidian Vault 路径"} vault_path = os.path.expanduser(vault_path) if not os.path.isdir(vault_path): return {"success": False, "message": f"Obsidian Vault 路径不存在: {vault_path}"} title = task_data.get("title", "未命名任务") description = task_data.get("description", "") priority = task_data.get("priority", "中") due_date = task_data.get("due_date", "") assignee = task_data.get("assignee", "") labels = task_data.get("labels", []) # 生成安全的文件名 safe_title = title.replace("/", "_").replace("\\", "_").replace(":", "_") safe_title = safe_title.replace("\"", "").replace("*", "").replace("?", "") safe_title = safe_title.replace("<", "").replace(">", "").replace("|", "") if len(safe_title) > 100: safe_title = safe_title[:100] # 确保任务目录存在 tasks_dir = os.path.join(vault_path, "tasks") os.makedirs(tasks_dir, exist_ok=True) # 生成文件路径（避免重名） filename = f"{safe_title}.md" filepath = os.path.join(tasks_dir, filename) counter = 1 while os.path.exists(filepath): filename = f"{safe_title}_{counter}.md" filepath = os.path.join(tasks_dir, filename) counter += 1 # 构建 frontmatter now = now_iso() fm_lines = ["---"] fm_lines.append(f"status: 待办") if priority: fm_lines.append(f"priority: {priority}") if assignee: fm_lines.append(f"assignee: {assignee}") if due_date: fm_lines.append(f"due_date: {due_date}") fm_lines.append(f"created: {now}") if labels: fm_lines.append(f"tags: [{', '.join(labels)}]") fm_lines.append("---") fm_lines.append("") # 构建笔记内容 content_lines = list(fm_lines) content_lines.append(f"# {title}") content_lines.append("") content_lines.append(f"- [ ] {title} {task_tag}") content_lines.append("") if description: content_lines.append("## 描述") content_lines.append("") content_lines.append(description) content_lines.append("") if due_date: content_lines.append(f"📅 截止日期: {due_date}") content_lines.append("") try: with open(filepath, "w", encoding="utf-8") as f: f.write("\n".join(content_lines)) rel_path = os.path.relpath(filepath, vault_path) vault_name = os.path.basename(vault_path) obsidian_url = f"obsidian://open?vault={vault_name}&file={rel_path}" return { "success": True, "message": f"Obsidian 任务笔记已创建: {title}", "task_id": rel_path, "url": obsidian_url, } except IOError as e: return {"success": False, "message": f"Obsidian 文件写入失败: {e}"} # ============================================================ # 平台创建路由 # ============================================================ _PLATFORM_CREATORS = { "trello": _create_trello_card, "github": _create_github_issue, "linear": _create_linear_issue, "notion": _create_notion_page, "obsidian": _create_obsidian_task, } # ============================================================ # 操作实现 # ============================================================ def create_task(data: Dict[str, Any]) -> None: """创建任务。必填字段: title 可选字段: platform, description, priority, assignee, labels, due_date 若未指定 platform，将根据标题和描述自动检测最适合的平台。 Args: data: 任务数据字典。 """ title = data.get("title", "").strip() if not title: output_error("任务标题（title）为必填字段", code="VALIDATION_ERROR") return sources = _get_sources() if not sources: output_error("暂无已连接的数据源，请先使用 source_connector 连接平台", code="NO_SOURCES") return # 确定目标平台 platform = data.get("platform", "").strip().lower() if not platform: platform = _detect_platform(title, data.get("description", "")) # 检查平台是否已连接 source = _find_source_by_platform(sources, platform) if not source: # 回退到第一个可用平台 available = [s["platform"] for s in sources] if platform not in available: source = sources[0] platform = source["platform"] else: output_error(f"平台 {platform} 未连接", code="NOT_CONNECTED") return creator = _PLATFORM_CREATORS.get(platform) if not creator: output_error(f"平台 {platform} 暂不支持创建任务", code="NOT_SUPPORTED") return # 尝试导入自学习引擎（可选依赖） try: from learning_engine import quick_record_error, quick_record_success has_learning = True except ImportError: has_learning = False result = creator(source, data) if result["success"]: # 记录成功模式 if has_learning: quick_record_success(platform, "create", data.get("title", "")) output_success({ "message": result["message"], "platform": platform, "task_id": result.get("task_id", ""), "url": result.get("url", ""), "auto_detected": not data.get("platform"), }) else: # 记录错误模式 if has_learning: quick_record_error(platform, "create", "create_failure", result.get("message", "")) output_error(result["message"], code="CREATE_FAILED") def update_task(data: Dict[str, Any]) -> None: """更新任务状态或属性。必填字段: source（平台）, source_id（平台任务ID）可选字段: status, priority, title, description Args: data: 包含平台和任务 ID 的更新数据字典。 """ platform = data.get("source", "").strip().lower() source_id = data.get("source_id", "").strip() if not platform or not source_id: output_error("平台（source）和任务 ID（source_id）为必填字段", code="VALIDATION_ERROR") return sources = _get_sources() source = _find_source_by_platform(sources, platform) if not source: output_error(f"平台 {platform} 未连接", code="NOT_CONNECTED") return # 根据平台执行更新 if platform == "github": _update_github_issue(source, source_id, data) elif platform == "trello": _update_trello_card(source, source_id, data) elif platform == "linear": _update_linear_issue(source, source_id, data) elif platform == "notion": _update_notion_page(source, source_id, data) else: output_error(f"平台 {platform} 暂不支持更新操作", code="NOT_SUPPORTED") def _update_github_issue(source: Dict, source_id: str, data: Dict) -> None: """更新 GitHub Issue。""" token = os.environ.get("PNC_GITHUB_TOKEN", "") repo = source.get("config", {}).get("repo", "") if not token or not repo: output_error("缺少 GitHub 凭据", code="AUTH_ERROR") return url = f"https://api.github.com/repos/{repo}/issues/{source_id}" headers = { "Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "Content-Type": "application/json", "User-Agent": "project-nerve/1.0", } update_data = {} if data.get("title"): update_data["title"] = data["title"] if data.get("description"): update_data["body"] = data["description"] if data.get("status"): status = normalize_status(data["status"]) if status in ("已完成", "已关闭"): update_data["state"] = "closed" else: update_data["state"] = "open" if not update_data: output_error("未提供任何待更新的字段", code="VALIDATION_ERROR") return body = json.dumps(update_data).encode("utf-8") resp = _http_request(url, method="PATCH", headers=headers, data=body) if resp["status"] == 200: output_success({"message": f"GitHub Issue #{source_id} 已更新", "platform": "github"}) else: output_error(f"GitHub 更新失败 (HTTP {resp['status']})", code="UPDATE_FAILED") def _update_trello_card(source: Dict, source_id: str, data: Dict) -> None: """更新 Trello 卡片。""" api_key = os.environ.get("PNC_TRELLO_API_KEY", "") token = os.environ.get("PNC_TRELLO_TOKEN", "") if not api_key or not token: output_error("缺少 Trello 凭据", code="AUTH_ERROR") return params = f"key={api_key}&token={token}" url = f"https://api.trello.com/1/cards/{source_id}?{params}" update_data = {} if data.get("title"): update_data["name"] = data["title"] if data.get("description"): update_data["desc"] = data["description"] if data.get("due_date"): update_data["due"] = data["due_date"] if not update_data: output_error("未提供任何待更新的字段", code="VALIDATION_ERROR") return body = json.dumps(update_data).encode("utf-8") headers = {"Content-Type": "application/json"} resp = _http_request(url, method="PUT", headers=headers, data=body) if resp["status"] == 200: output_success({"message": f"Trello 卡片 {source_id} 已更新", "platform": "trello"}) else: output_error(f"Trello 更新失败 (HTTP {resp['status']})", code="UPDATE_FAILED") def _update_linear_issue(source: Dict, source_id: str, data: Dict) -> None: """更新 Linear Issue。""" api_key = os.environ.get("PNC_LINEAR_API_KEY", "") if not api_key: output_error("缺少 Linear API Key", code="AUTH_ERROR") return input_data = {} if data.get("title"): input_data["title"] = data["title"] if data.get("description"): input_data["description"] = data["description"] if data.get("priority"): priority_map = {"紧急": 1, "高": 2, "中": 3, "低": 4} input_data["priority"] = priority_map.get(normalize_priority(data["priority"]), 3) if not input_data: output_error("未提供任何待更新的字段", code="VALIDATION_ERROR") return mutation = json.dumps({ "query": """mutation UpdateIssue($id: String!, $input: IssueUpdateInput!) { issueUpdate(id: $id, input: $input) { success } }""", "variables": {"id": source_id, "input": input_data} }) url = "https://api.linear.app/graphql" headers = {"Authorization": api_key, "Content-Type": "application/json"} resp = _http_request(url, method="POST", headers=headers, data=mutation.encode("utf-8")) if resp["status"] == 200: output_success({"message": f"Linear Issue {source_id} 已更新", "platform": "linear"}) else: output_error(f"Linear 更新失败 (HTTP {resp['status']})", code="UPDATE_FAILED") def _update_notion_page(source: Dict, source_id: str, data: Dict) -> None: """更新 Notion 页面属性。""" token = os.environ.get("PNC_NOTION_TOKEN", "") if not token: output_error("缺少 Notion Token", code="AUTH_ERROR") return url = f"https://api.notion.com/v1/pages/{source_id}" headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json", } properties = {} if data.get("title"): properties["Name"] = {"title": [{"text": {"content": data["title"]}}]} if not properties: output_error("未提供任何待更新的字段", code="VALIDATION_ERROR") return body = json.dumps({"properties": properties}).encode("utf-8") resp = _http_request(url, method="PATCH", headers=headers, data=body) if resp["status"] == 200: output_success({"message": f"Notion 页面 {source_id[:8]}... 已更新", "platform": "notion"}) else: output_error(f"Notion 更新失败 (HTTP {resp['status']})", code="UPDATE_FAILED") def move_task(data: Dict[str, Any]) -> None: """移动任务状态（等同于 update + status）。必填字段: source, source_id, status Args: data: 包含平台、任务 ID 和目标状态的字典。 """ if not data.get("status"): output_error("目标状态（status）为必填字段", code="VALIDATION_ERROR") return update_task(data) def comment_task(data: Dict[str, Any]) -> None: """给任务添加评论。必填字段: source, source_id, comment Args: data: 包含平台、任务 ID 和评论内容的字典。 """ platform = data.get("source", "").strip().lower() source_id = data.get("source_id", "").strip() comment = data.get("comment", "").strip() if not platform or not source_id or not comment: output_error("平台（source）、任务 ID（source_id）和评论内容（comment）为必填字段", code="VALIDATION_ERROR") return sources = _get_sources() source = _find_source_by_platform(sources, platform) if not source: output_error(f"平台 {platform} 未连接", code="NOT_CONNECTED") return if platform == "github": _comment_github_issue(source, source_id, comment) elif platform == "trello": _comment_trello_card(source, source_id, comment) elif platform == "linear": _comment_linear_issue(source, source_id, comment) elif platform == "notion": _comment_notion_page(source, source_id, comment) else: output_error(f"平台 {platform} 暂不支持评论功能", code="NOT_SUPPORTED") def _comment_github_issue(source: Dict, source_id: str, comment: str) -> None: """给 GitHub Issue 添加评论。""" token = os.environ.get("PNC_GITHUB_TOKEN", "") repo = source.get("config", {}).get("repo", "") if not token or not repo: output_error("缺少 GitHub 凭据", code="AUTH_ERROR") return url = f"https://api.github.com/repos/{repo}/issues/{source_id}/comments" headers = { "Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "Content-Type": "application/json", "User-Agent": "project-nerve/1.0", } body = json.dumps({"body": comment}).encode("utf-8") resp = _http_request(url, method="POST", headers=headers, data=body) if resp["status"] in (200, 201): output_success({"message": f"已在 GitHub Issue #{source_id} 添加评论", "platform": "github"}) else: output_error(f"GitHub 评论失败 (HTTP {resp['status']})", code="COMMENT_FAILED") def _comment_trello_card(source: Dict, source_id: str, comment: str) -> None: """给 Trello 卡片添加评论。""" api_key = os.environ.get("PNC_TRELLO_API_KEY", "") token = os.environ.get("PNC_TRELLO_TOKEN", "") if not api_key or not token: output_error("缺少 Trello 凭据", code="AUTH_ERROR") return params = f"key={api_key}&token={token}&text={urllib.request.quote(comment)}" url = f"https://api.trello.com/1/cards/{source_id}/actions/comments?{params}" resp = _http_request(url, method="POST") if resp["status"] in (200, 201): output_success({"message": f"已在 Trello 卡片 {source_id} 添加评论", "platform": "trello"}) else: output_error(f"Trello 评论失败 (HTTP {resp['status']})", code="COMMENT_FAILED") def _comment_linear_issue(source: Dict, source_id: str, comment: str) -> None: """给 Linear Issue 添加评论。""" api_key = os.environ.get("PNC_LINEAR_API_KEY", "") if not api_key: output_error("缺少 Linear API Key", code="AUTH_ERROR") return mutation = json.dumps({ "query": """mutation CreateComment($input: CommentCreateInput!) { commentCreate(input: $input) { success } }""", "variables": {"input": {"issueId": source_id, "body": comment}} }) url = "https://api.linear.app/graphql" headers = {"Authorization": api_key, "Content-Type": "application/json"} resp = _http_request(url, method="POST", headers=headers, data=mutation.encode("utf-8")) if resp["status"] == 200: output_success({"message": f"已在 Linear Issue {source_id} 添加评论", "platform": "linear"}) else: output_error(f"Linear 评论失败 (HTTP {resp['status']})", code="COMMENT_FAILED") def _comment_notion_page(source: Dict, source_id: str, comment: str) -> None: """在 Notion 页面添加评论（作为子块追加）。""" token = os.environ.get("PNC_NOTION_TOKEN", "") if not token: output_error("缺少 Notion Token", code="AUTH_ERROR") return url = f"https://api.notion.com/v1/blocks/{source_id}/children" headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json", } body = json.dumps({ "children": [{ "object": "block", "type": "paragraph", "paragraph": { "rich_text": [{"text": {"content": f"[评论] {comment}"}}] } }] }).encode("utf-8") resp = _http_request(url, method="PATCH", headers=headers, data=body) if resp["status"] == 200: output_success({"message": f"已在 Notion 页面 {source_id[:8]}... 添加评论", "platform": "notion"}) else: output_error(f"Notion 评论失败 (HTTP {resp['status']})", code="COMMENT_FAILED") # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("project-nerve 任务写入器") 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 = { "create": lambda: create_task(data or {}), "update": lambda: update_task(data or {}), "move": lambda: move_task(data or {}), "comment": lambda: comment_task(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/sprint_analyzer.py #!/usr/bin/env python3 """ project-nerve Sprint 分析器（付费功能）提供冲刺速度计算、任务漏斗分析、燃尽图数据生成和综合冲刺报告。所有功能均需付费订阅。 """ import json import math import os import sys from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from utils import ( check_subscription, format_task_table, generate_bar_chart, generate_line_chart, generate_pie_chart, get_data_file, load_input_data, normalize_status, now_iso, output_error, output_success, parse_common_args, parse_date, read_json_file, require_paid_feature, today_str, ) # ============================================================ # 数据文件路径 # ============================================================ TASKS_CACHE_FILE = "tasks_cache.json" def _get_cached_tasks() -> List[Dict[str, Any]]: """读取缓存的任务数据。""" data = read_json_file(get_data_file(TASKS_CACHE_FILE)) if isinstance(data, list): return data return [] # ============================================================ # 时间范围辅助 # ============================================================ def _parse_date_range(data: Dict[str, Any]) -> tuple: """解析时间范围参数。支持 start_date/end_date 或 days（最近N天）。 Args: data: 包含时间范围参数的字典。 Returns: (start_dt, end_dt) 元组。 """ now = datetime.now() if data.get("start_date") and data.get("end_date"): start_dt = parse_date(data["start_date"]) or (now - timedelta(days=14)) end_dt = parse_date(data["end_date"]) or now else: days = int(data.get("days", 14)) start_dt = now - timedelta(days=days) end_dt = now return start_dt, end_dt def _filter_by_date_range( tasks: List[Dict[str, Any]], start_dt: datetime, end_dt: datetime, date_field: str = "updated_at", ) -> List[Dict[str, Any]]: """根据时间范围过滤任务。 Args: tasks: 任务列表。 start_dt: 开始时间。 end_dt: 结束时间。 date_field: 用于比较的日期字段名。 Returns: 过滤后的任务列表。 """ result = [] for task in tasks: dt = parse_date(task.get(date_field, "")) if dt and start_dt <= dt <= end_dt: result.append(task) return result # ============================================================ # 操作实现 # ============================================================ def velocity(data: Dict[str, Any]) -> None: """计算冲刺速度。速度 = 指定时间范围内已完成的任务数量。同时计算每日平均完成量和趋势。 Args: data: 包含时间范围参数的字典。 """ if not require_paid_feature("sprint_analytics", "冲刺分析"): return tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 task_aggregator fetch-all 获取任务", code="NO_DATA") return start_dt, end_dt = _parse_date_range(data) total_days = max((end_dt - start_dt).days, 1) # 过滤时间范围内的已完成任务 completed_tasks = [ t for t in tasks if t.get("status") == "已完成" ] completed_in_range = _filter_by_date_range(completed_tasks, start_dt, end_dt) # 按天统计完成数量 daily_counts = {} for task in completed_in_range: dt = parse_date(task.get("updated_at", "")) if dt: day_key = dt.strftime("%Y-%m-%d") daily_counts[day_key] = daily_counts.get(day_key, 0) + 1 # 填充空日期 current = start_dt daily_data = [] while current <= end_dt: day_key = current.strftime("%Y-%m-%d") count = daily_counts.get(day_key, 0) daily_data.append({"label": day_key[-5:], "value": count}) # MM-DD 格式 current += timedelta(days=1) total_completed = len(completed_in_range) daily_avg = round(total_completed / total_days, 1) if total_days > 0 else 0 # 按平台统计 source_stats = {} for task in completed_in_range: src = task.get("source", "未知") source_stats[src] = source_stats.get(src, 0) + 1 # 生成速度趋势图 chart = "" if daily_data: chart = generate_line_chart("每日完成任务数趋势", daily_data, x_label="日期", y_label="完成数") output_success({ "sprint_period": f"{start_dt.strftime('%Y-%m-%d')} ~ {end_dt.strftime('%Y-%m-%d')}", "total_days": total_days, "total_completed": total_completed, "daily_average": daily_avg, "source_stats": source_stats, "daily_data": daily_data, "chart": chart, }) def funnel(data: Dict[str, Any]) -> None: """任务漏斗分析。统计各状态的任务数量和占比：待办 → 进行中 → 已完成。 Args: data: 可选参数（可指定 platform 过滤）。 """ if not require_paid_feature("sprint_analytics", "冲刺分析"): return tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 task_aggregator fetch-all 获取任务", code="NO_DATA") return # 可选平台过滤 platform = data.get("platform", "").strip().lower() if data else "" if platform: tasks = [t for t in tasks if t.get("source") == platform] total = len(tasks) if total == 0: output_error("过滤后无任务数据", code="NO_DATA") return # 统计各状态 status_counts = {"待办": 0, "进行中": 0, "已完成": 0, "已关闭": 0} for task in tasks: status = task.get("status", "待办") if status in status_counts: status_counts[status] += 1 else: status_counts["待办"] += 1 # 计算百分比 funnel_data = [] for status in ["待办", "进行中", "已完成", "已关闭"]: count = status_counts[status] pct = round(count / total * 100, 1) if total > 0 else 0 funnel_data.append({ "status": status, "count": count, "percentage": pct, }) # 生成饼图 pie_data = [{"label": item["status"], "value": item["count"]} for item in funnel_data if item["count"] > 0] pie_chart = generate_pie_chart("任务状态分布", pie_data) # 生成柱状图 bar_data = [{"label": item["status"], "value": item["count"]} for item in funnel_data] bar_chart = generate_bar_chart("任务漏斗", bar_data, x_label="状态", y_label="数量") # 计算转化率 todo_count = status_counts["待办"] in_progress_count = status_counts["进行中"] done_count = status_counts["已完成"] start_to_progress = round(in_progress_count / (todo_count + in_progress_count + done_count) * 100, 1) if total > 0 else 0 progress_to_done = round(done_count / max(in_progress_count + done_count, 1) * 100, 1) output_success({ "total": total, "funnel": funnel_data, "conversion": { "启动率": f"{start_to_progress}%", "完成率": f"{progress_to_done}%", }, "pie_chart": pie_chart, "bar_chart": bar_chart, }) def burndown(data: Dict[str, Any]) -> None: """生成燃尽图数据。计算指定时间范围内，每天剩余待完成任务的数量变化。 Args: data: 包含时间范围参数的字典。 """ if not require_paid_feature("mermaid_chart", "Mermaid 图表"): return tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 task_aggregator fetch-all 获取任务", code="NO_DATA") return start_dt, end_dt = _parse_date_range(data) total_days = max((end_dt - start_dt).days, 1) # 计算总任务数（排除已关闭） active_tasks = [t for t in tasks if t.get("status") != "已关闭"] total_tasks = len(active_tasks) # 统计每天完成数量（累计） completed_by_day = {} for task in active_tasks: if task.get("status") == "已完成": dt = parse_date(task.get("updated_at", "")) if dt: day_key = dt.strftime("%Y-%m-%d") completed_by_day[day_key] = completed_by_day.get(day_key, 0) + 1 # 生成燃尽数据 burndown_data = [] ideal_data = [] cumulative_completed = 0 current = start_dt day_index = 0 while current <= end_dt: day_key = current.strftime("%Y-%m-%d") cumulative_completed += completed_by_day.get(day_key, 0) remaining = total_tasks - cumulative_completed burndown_data.append({ "label": day_key[-5:], "value": max(remaining, 0), }) # 理想燃尽线 ideal_remaining = total_tasks - (total_tasks * day_index / total_days) ideal_data.append({ "label": day_key[-5:], "value": round(max(ideal_remaining, 0), 1), }) current += timedelta(days=1) day_index += 1 # 生成 Mermaid xychart-beta（实际 + 理想） chart_lines = ["```mermaid", "xychart-beta", ' title "冲刺燃尽图"'] labels = [f'"{d["label"]}"' for d in burndown_data] chart_lines.append(f' x-axis [{", ".join(labels)}]') chart_lines.append(' y-axis "剩余任务数"') actual_values = [str(d["value"]) for d in burndown_data] chart_lines.append(f' line [{", ".join(actual_values)}]') ideal_values = [str(d["value"]) for d in ideal_data] chart_lines.append(f' line [{", ".join(ideal_values)}]') chart_lines.append("```") chart = "\n".join(chart_lines) output_success({ "sprint_period": f"{start_dt.strftime('%Y-%m-%d')} ~ {end_dt.strftime('%Y-%m-%d')}", "total_tasks": total_tasks, "completed": cumulative_completed, "remaining": total_tasks - cumulative_completed, "burndown_data": burndown_data, "ideal_data": ideal_data, "chart": chart, }) def sprint_report(data: Dict[str, Any]) -> None: """生成综合冲刺报告。包含速度、漏斗、燃尽等多维度分析的 Markdown 报告。 Args: data: 包含时间范围参数的字典。 """ if not require_paid_feature("sprint_analytics", "冲刺分析"): return tasks = _get_cached_tasks() if not tasks: output_error("暂无缓存任务数据，请先执行 task_aggregator fetch-all 获取任务", code="NO_DATA") return start_dt, end_dt = _parse_date_range(data) total_days = max((end_dt - start_dt).days, 1) period_str = f"{start_dt.strftime('%Y-%m-%d')} ~ {end_dt.strftime('%Y-%m-%d')}" # 排除已关闭 active_tasks = [t for t in tasks if t.get("status") != "已关闭"] total_tasks = len(active_tasks) # 状态统计 status_counts = {"待办": 0, "进行中": 0, "已完成": 0} for task in active_tasks: status = task.get("status", "待办") if status in status_counts: status_counts[status] += 1 # 范围内完成的任务 completed_in_range = _filter_by_date_range( [t for t in active_tasks if t.get("status") == "已完成"], start_dt, end_dt ) velocity_value = len(completed_in_range) daily_avg = round(velocity_value / total_days, 1) # 优先级分布 priority_counts = {"紧急": 0, "高": 0, "中": 0, "低": 0} for task in active_tasks: prio = task.get("priority", "中") if prio in priority_counts: priority_counts[prio] += 1 # 平台分布 source_counts = {} for task in active_tasks: src = task.get("source", "未知") source_counts[src] = source_counts.get(src, 0) + 1 # 逾期任务 from utils import is_overdue as _is_overdue overdue_tasks = [ t for t in active_tasks if t.get("status") not in ("已完成",) and t.get("due_date") and _is_overdue(t["due_date"]) ] # 构建 Markdown 报告 report_parts = [] report_parts.append(f"# 冲刺报告 — {period_str}\n") report_parts.append(f"统计周期: {period_str} | 总计 {total_days} 天\n") # 核心指标 report_parts.append("## 核心指标\n") report_parts.append("| 指标 | 数值 |") report_parts.append("|------|------|") report_parts.append(f"| 总任务数 | {total_tasks} |") report_parts.append(f"| 本期完成 | {velocity_value} |") report_parts.append(f"| 日均完成 | {daily_avg} |") report_parts.append(f"| 待办任务 | {status_counts['待办']} |") report_parts.append(f"| 进行中 | {status_counts['进行中']} |") report_parts.append(f"| 逾期任务 | {len(overdue_tasks)} |") report_parts.append("") # 完成率 completion_rate = round(status_counts["已完成"] / total_tasks * 100, 1) if total_tasks > 0 else 0 report_parts.append(f"**整体完成率**: {completion_rate}%\n") # 状态分布饼图 status_pie_data = [ {"label": k, "value": v} for k, v in status_counts.items() if v > 0 ] if status_pie_data: report_parts.append("## 状态分布\n") report_parts.append(generate_pie_chart("任务状态分布", status_pie_data)) report_parts.append("") # 优先级分布柱状图 priority_bar_data = [ {"label": k, "value": v} for k, v in priority_counts.items() if v > 0 ] if priority_bar_data: report_parts.append("## 优先级分布\n") report_parts.append(generate_bar_chart("任务优先级分布", priority_bar_data, x_label="优先级", y_label="数量")) report_parts.append("") # 平台分布 if source_counts: report_parts.append("## 平台分布\n") report_parts.append("| 平台 | 任务数 | 占比 |") report_parts.append("|------|--------|------|") for src, cnt in sorted(source_counts.items(), key=lambda x: x[1], reverse=True): pct = round(cnt / total_tasks * 100, 1) if total_tasks > 0 else 0 report_parts.append(f"| {src} | {cnt} | {pct}% |") report_parts.append("") # 逾期任务列表 if overdue_tasks: report_parts.append("## 逾期任务\n") report_parts.append(format_task_table(overdue_tasks[:10])) if len(overdue_tasks) > 10: report_parts.append(f"\n> 仅显示前 10 个，共 {len(overdue_tasks)} 个逾期任务。") report_parts.append("") # 建议 report_parts.append("## 改进建议\n") suggestion_idx = 1 if len(overdue_tasks) > 0: report_parts.append(f"{suggestion_idx}. 当前有 {len(overdue_tasks)} 个逾期任务，建议优先处理或重新评估截止日期。") suggestion_idx += 1 if priority_counts.get("紧急", 0) > 3: report_parts.append(f"{suggestion_idx}. 紧急任务数量较多（{priority_counts['紧急']} 个），建议评估是否存在优先级膨胀问题。") suggestion_idx += 1 if completion_rate < 50: report_parts.append(f"{suggestion_idx}. 整体完成率仅 {completion_rate}%，建议分析瓶颈环节并调整冲刺范围。") suggestion_idx += 1 if daily_avg < 1: report_parts.append(f"{suggestion_idx}. 日均完成量较低（{daily_avg}），建议检查任务拆分粒度或团队产能。") suggestion_idx += 1 if suggestion_idx == 1: report_parts.append("各项指标表现良好，建议继续保持当前节奏。") report_parts.append("") report_parts.append(f"---\n*报告由 project-nerve 自动生成 — {now_iso()}*") report_md = "\n".join(report_parts) output_success({ "report": report_md, "summary": { "period": period_str, "total_tasks": total_tasks, "velocity": velocity_value, "daily_average": daily_avg, "completion_rate": completion_rate, "overdue_count": len(overdue_tasks), }, }) # ============================================================ # 主入口 # ============================================================ def main() -> None: """主函数：解析命令行参数并分发操作。""" parser = parse_common_args("project-nerve Sprint 分析器") 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 = { "velocity": lambda: velocity(data or {}), "funnel": lambda: funnel(data or {}), "burndown": lambda: burndown(data or {}), "report": lambda: sprint_report(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-guide.md # 平台 API 指南 project-nerve 使用的各平台 API 端点参考。 --- ## Trello API | 操作 | 方法 | 端点 | |------|------|------| | 获取用户信息 | GET | `https://api.trello.com/1/members/me?key={key}&token={token}` | | 获取看板列表 | GET | `https://api.trello.com/1/boards/{boardId}/lists?key={key}&token={token}` | | 获取看板卡片 | GET | `https://api.trello.com/1/boards/{boardId}/cards?key={key}&token={token}` | | 创建卡片 | POST | `https://api.trello.com/1/cards?key={key}&token={token}` | | 更新卡片 | PUT | `https://api.trello.com/1/cards/{cardId}?key={key}&token={token}` | | 添加评论 | POST | `https://api.trello.com/1/cards/{cardId}/actions/comments?key={key}&token={token}&text={text}` | **认证方式**: Query 参数传递 `key` 和 `token`。 --- ## GitHub Issues API | 操作 | 方法 | 端点 | |------|------|------| | 获取用户信息 | GET | `https://api.github.com/user` | | 列出仓库 Issues | GET | `https://api.github.com/repos/{owner}/{repo}/issues` | | 创建 Issue | POST | `https://api.github.com/repos/{owner}/{repo}/issues` | | 更新 Issue | PATCH | `https://api.github.com/repos/{owner}/{repo}/issues/{number}` | | 添加评论 | POST | `https://api.github.com/repos/{owner}/{repo}/issues/{number}/comments` | **认证方式**: `Authorization: Bearer {token}` 请求头。 **必需请求头**: `Accept: application/vnd.github.v3+json`, `User-Agent: project-nerve/1.0`。 --- ## Linear GraphQL API | 操作 | 方法 | 端点 | |------|------|------| | 所有操作 | POST | `https://api.linear.app/graphql` | **认证方式**: `Authorization: {api_key}` 请求头（无 Bearer 前缀）。常用查询: - `{ viewer { id name email } }` — 获取当前用户 - `{ issues(first:100) { nodes { id identifier title ... } } }` — 列出 Issues - `mutation issueCreate(input: {...}) { ... }` — 创建 Issue - `mutation issueUpdate(id: "...", input: {...}) { ... }` — 更新 Issue - `mutation commentCreate(input: {issueId: "...", body: "..."}) { ... }` — 添加评论 --- ## Notion API | 操作 | 方法 | 端点 | |------|------|------| | 获取数据库信息 | GET | `https://api.notion.com/v1/databases/{database_id}` | | 查询数据库 | POST | `https://api.notion.com/v1/databases/{database_id}/query` | | 创建页面 | POST | `https://api.notion.com/v1/pages` | | 更新页面属性 | PATCH | `https://api.notion.com/v1/pages/{page_id}` | | 追加子块 | PATCH | `https://api.notion.com/v1/blocks/{block_id}/children` | **认证方式**: `Authorization: Bearer {token}` 请求头。 **必需请求头**: `Notion-Version: 2022-06-28`, `Content-Type: application/json`。 --- ## 通用注意事项 1. 所有请求使用 HTTPS。 2. 超时设置为 15 秒。 3. 使用 Python `urllib.request` 标准库发送请求。 4. API Key / Token 仅通过环境变量获取，不在代码中硬编码。 5. 错误响应统一包装为 `{success: false, error: {code, message}}` 格式。 FILE:references/unified-schema.md # 统一任务模型 project-nerve 将来自不同平台的任务数据映射到统一的任务模型。 --- ## 统一任务字段 | 字段 | 类型 | 说明 | |------|------|------| | `id` | string | 统一 ID（格式：`{platform}-{source_id}`） | | `source` | string | 来源平台（trello / github / linear / notion） | | `source_id` | string | 平台原始 ID（Trello cardId、GitHub issue number、Linear identifier、Notion pageId） | | `title` | string | 任务标题 | | `description` | string | 任务描述（最长 500 字符） | | `status` | string | 统一状态（待办 / 进行中 / 已完成 / 已关闭） | | `priority` | string | 统一优先级（紧急 / 高 / 中 / 低） | | `assignee` | string | 负责人 | | `labels` | list[string] | 标签列表 | | `due_date` | string | 截止日期（YYYY-MM-DD） | | `created_at` | string | 创建时间（ISO 格式） | | `updated_at` | string | 更新时间（ISO 格式） | | `url` | string | 平台原始链接 | --- ## 状态映射表 | 统一状态 | Trello 列名 | GitHub State | Linear State | Notion Status | |----------|-------------|-------------|--------------|---------------| | 待办 | To Do / Backlog | open | Backlog / Todo / Triage | Not Started / To Do | | 进行中 | Doing / In Progress | open (assigned) | In Progress / Started | In Progress | | 已完成 | Done / Completed | closed (merged) | Done / Completed | Done / Completed | | 已关闭 | Archived | closed | Cancelled / Duplicate | Archived | ### 状态标准化规则输入字符串（不区分大小写）→ 统一状态： - `todo`, `to do`, `backlog`, `open`, `new`, `not started`, `triage` → **待办** - `in progress`, `in_progress`, `doing`, `started`, `active`, `in review` → **进行中** - `done`, `completed`, `resolved`, `merged` → **已完成** - `closed`, `cancelled`, `canceled`, `archived`, `duplicate` → **已关闭** --- ## 优先级映射表 | 统一优先级 | Trello 标签 | GitHub 标签 | Linear 数值 | Notion Select | |-----------|-------------|-------------|-------------|---------------| | 紧急 | urgent / 紧急 | P0 / critical / blocker | 1 (Urgent) | 紧急 / Urgent | | 高 | high / 高 | P1 / high / important | 2 (High) | 高 / High | | 中 | medium / 中 | P2 / medium / normal | 3 (Medium) | 中 / Medium | | 低 | low / 低 | P3 / low / minor | 4 (Low) / 0 (None) | 低 / Low | ### 优先级标准化规则输入字符串（不区分大小写）→ 统一优先级： - `urgent`, `critical`, `p0`, `highest`, `blocker`, `紧急` → **紧急** - `high`, `p1`, `important`, `高` → **高** - `medium`, `normal`, `p2`, `default`, `中` → **中** - `low`, `minor`, `p3`, `trivial`, `none`, `低` → **低** - 无法识别 → 默认 **中**

项目中枢 / Project Nerve

Comments (0)