From 11c62dd383cc16c6e09b976ff0249cd18d3f24b5 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Tue, 10 Feb 2026 20:31:33 +0800 Subject: [PATCH] =?UTF-8?q?refactor(kb-workflow):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E4=B8=BA=E7=9C=9F=E6=AD=A3=E7=9A=84=E7=BC=96=E6=8E=92=E8=80=85?= =?UTF-8?q?=EF=BC=8C=E4=BD=BF=E7=94=A8sessions=5Fspawn=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E5=AD=90agent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - kb_workflow.py: 重构为编排者，使用sessions_spawn调用子技能 - skill_main.py: 新增技能入口，处理用户消息 - SKILL.md: 更新文档，明确架构和流程 - SkillBug.md: 新增问题记录文档 设计原则: - workflow不读文件内容，从文件名提取平台 - 使用sessions_spawn调用子agent - 子技能自己定义输入输出格式 --- skills/kb-workflow/SKILL.md | 356 ++++--- skills/kb-workflow/SkillBug.md | 99 ++ skills/kb-workflow/kb_workflow.py | 1492 +++++++++++++---------------- skills/kb-workflow/skill_main.py | 133 +++ 4 files changed, 1049 insertions(+), 1031 deletions(-) create mode 100644 skills/kb-workflow/SkillBug.md create mode 100644 skills/kb-workflow/skill_main.py diff --git a/skills/kb-workflow/SKILL.md b/skills/kb-workflow/SKILL.md index 1aa036b..962805d 100644 --- a/skills/kb-workflow/SKILL.md +++ b/skills/kb-workflow/SKILL.md @@ -13,7 +13,7 @@ KB Workflow 是一个完整的知识库管理系统，包含： **核心功能**： - 📥 **收件箱扫描** - 扫描四个子目录 (pending, metadata, content, analyzed) -- 🔄 **内容处理** - 调用技能2-4处理内容 +- 🔄 **内容处理** - 调用子agent处理各阶段 - 📦 **Git版本管理** - 归档后自动提交并推送 - 🔔 **订阅管理** - 跟踪订阅源更新 - 🧹 **自动清理** - 定期清理临时文件 @@ -27,218 +27,213 @@ KB Workflow 是一个完整的知识库管理系统，包含： | 触发词 | 行为 | 示例 | |--------|------|------| -| `保存到知识库` + 链接 | 添加到收件箱并处理 | "保存到知识库 https://www.bilibili.com/video/BVxxx" | | `添加到知识库` + 链接 | 添加到收件箱并处理 | "添加到知识库 https://www.bilibili.com/video/BVxxx" | -| `订阅到知识库` + 链接 | 添加到订阅列表 | "订阅到知识库 https://www.bilibili.com/video/BVxxx" | -| `订阅知识库` + 链接 | 添加到订阅列表 | "订阅知识库 https://www.bilibili.com/video/BVxxx" | +| `保存到知识库` + 链接 | 添加到收件箱并处理 | "保存到知识库 https://www.bilibili.com/video/BVxxx" | ### 支持的平台 -| 平台 | 保存 | 订阅 | -|------|------|------| -| Bilibili | ✅ | ✅ | -| YouTube | ✅ | ❌ | -| 知乎 | ✅ | ❌ | +| 平台 | 子技能 | +|------|--------| +| Bilibili | kb-bilibili-metadata, kb-bilibili-content | +| 知乎 | kb-zhihu-metadata | + +## 架构设计 + +``` +用户: "添加到知识库 https://www.bilibili.com/video/BV1ci4y1V7tb" + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ kb-workflow (主代理) │ +│ │ +│ 职责: │ +│ 1. 接收用户消息，识别触发词和链接 │ +│ 2. 查重（根据URL/文件名） │ +│ 3. 调用子agent处理各阶段 │ +│ 4. 文件流转（mv） │ +│ 5. Git版本管理 │ +│ 6. 归档和清理 │ +└─────────────────────────────────────────────────────────┘ + │ │ │ │ + ▼ ▼ ▼ ▼ +┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ +│ kb-link-parser│ │kb-bilibili- │ │kb-bilibili- │ │ kb-deep- │ +│ (子agent) │ │ metadata │ │ content │ │ analyzer │ +│ │ │ (子agent) │ │ (子agent) │ │ (子agent) │ +└──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘ +``` + +## 处理流程 + +### 完整流程（6个阶段） + +``` +用户发送"添加到知识库 + 链接" + │ + ▼ +kb-workflow: + 1. 查重（根据URL/文件名） + 2. 确认目录存在 + 3. 调用 kb-link-parser (sessions_spawn) + │ + ▼ +kb-link-parser (子agent): + 解析URL + 输出: pending/bilibili_BV1ci4y1V7tb.link.json + 返回: {"success": true/false, "output_file": "pending/..."} + │ + ▼ +kb-workflow: + 从文件名提取平台 "bilibili" → 调用 kb-bilibili-metadata + │ + ▼ +kb-bilibili-metadata (子agent): + 获取元数据 + 输出: metadata/bilibili_BV1ci4y1V7tb.meta.json + 返回: {"success": true/false, "output_file": "metadata/..."} + │ + ▼ +kb-workflow: + 调用 kb-bilibili-content + │ + ▼ +kb-bilibili-content (子agent): + 下载视频+字幕 + 输出: content/downloads/*.mp4, *.srt + 返回: {"success": true/false, "output_file": "...", "subtitles": "..."} + │ + ▼ +kb-workflow: + 拿到字幕文件 → 调用 kb-ocr-processor + │ + ▼ +kb-ocr-processor (子agent): + OCR处理 + 输出: content/bilibili_BV1ci4y1V7tb.txt + 返回: {"success": true/false, "output_file": "content/..."} + │ + ▼ +kb-workflow: + 拿到txt文件 → 调用 kb-deep-analyzer + │ + ▼ +kb-deep-analyzer (子agent): + 深度分析 + 输出: analyzed/bilibili_BV1ci4y1V7tb.md + 返回: {"success": true/false, "output_file": "analyzed/..."} + │ + ▼ +kb-workflow: + mv → archive/YYYY-MM-DD/ + git add → commit → push + 汇报完成 +``` ## 目录结构 ``` -knowledge-base/ -├── kb.config.yaml # 配置文件 -├── .gitignore # Git忽略规则 -├── pending/ # 待处理内容 (*.md) -├── metadata/ # 链接解析结果 (*.link.json) +知识库根目录 KB_BASE_PATH/ +├── archive/ # Git归档目录 +│ ├── YYYY-MM-DD/ # 按日期归档 +│ └── subscriptions/ # 订阅数据 +├── pending/ # 链接解析结果 (*.link.json) +├── metadata/ # 元数据 (*.meta.json) ├── content/ -│ ├── json/ # 元数据 (*.meta.json) -│ ├── downloads/ # 视频文件 (*.mp4/*.flv) - 7天后清理 -│ └── subtitles/ # 字幕文件 (*.srt/*.lrc) - 7天后清理 -├── frames/ # 视频截帧 (*.jpg) - 7天后清理 -├── analyzed/ # 分析报告 (*.md) -└── archive/ # Git归档目录 - ├── YYYY-MM-DD/ # 按日期归档 - └── subscriptions/ # 订阅数据 - ├── subscriptions.json # 订阅配置 - └── processed.json # 已处理项目 +│ ├── downloads/ # 视频文件 (*.mp4) - 7天后清理 +│ └── subtitles/ # 字幕文件 (*.srt) - 7天后清理 +├── frames/ # 视频截帧 (*.jpg) - 7天后清理 +└── analyzed/ # 分析报告 (*.md) ``` -## Git版本管理 +## 设计原则 -### 配置 +### workflow 职责 -设置环境变量： -```bash -export KB_GIT_ROOT="/path/to/archive" # Git仓库目录 -export KB_GIT_REMOTE="git@github.com:user/repo.git" # 远程仓库 -export KB_GIT_USER="Username" # 用户名 -export KB_GIT_EMAIL="user@example.com" # 邮箱 +| 操作 | 说明 | +|------|------| +| 查重 | 根据URL/文件名查重 | +| 创建目录 | 确保 base_path 下目录存在 | +| 调用子agent | 使用 sessions_spawn 调用子技能 | +| 提取文件名 | 从子agent返回的output_file提取 | +| 提取平台 | 从文件名第一个下划线之前提取（不读内容） | +| 传递信息 | 给子agent提示词 + 需要的路径 | +| 文件流转 | 最后归档 | +| Git | commit + push | + +### workflow 不管 + +- 不读文件内容 +- 不关心子技能内部格式 +- 子技能自己定义输入输出规范 + +### 子agent 调用方式 + +```python +# kb-workflow 调用子agent +result = sessions_spawn( + agentId="子技能名称", + task="提示词 + 需要的路径信息", + timeoutSeconds=120 +) ``` -### 工作流程 +### 子技能返回格式 -``` -归档项目 → Git add → Git commit → Git push → 清理临时文件 -``` - -- 每次归档生成一个commit -- 自动推送到远程仓库（重试3次） -- 汇报中体现Git状态 - -## 订阅管理 - -### 数据结构 +所有子技能应返回JSON： ```json { - "version": "1.0", - "subscriptions": [ - { - "id": "sub_001", - "platform": "bilibili", - "type": "series", - "identifier": "123456", - "name": "UE5教程", - "subscribed_at": "2026-02-10T12:00:00", - "last_checked": null, - "last_update": null, - "status": "active" - } - ], - "processed_items": [ - { - "id": "BVxxx", - "title": "视频标题", - "processed_at": "2026-02-10T14:00:00", - "archived": true - } - ] + "success": true, + "output_file": "完整文件路径", + "subtitles": "字幕文件路径（可选）" } ``` -### 命令 +## 环境变量 ```bash -# 添加订阅 -python kb_workflow.py --trigger subscriptions --add "bilibili" "series" "123456" "教程名称" - -# 检查更新 -python kb_workflow.py --trigger subscriptions --check - -# 查看状态 -python kb_workflow.py --trigger subscriptions --status -``` - -## 自动清理策略 - -### 清理规则 - -| 类型 | 保留 | 清理周期 | -|------|------|---------| -| 分析报告 | ✅ 永久 | - | -| 链接信息 | ✅ 永久 | - | -| 元数据 | ✅ 永久 | - | -| 订阅数据 | ✅ Git管理 | - | -| 视频文件 | ❌ | 7天 | -| 截帧文件 | ❌ | 7天 | - -### 命令 - -```bash -# 清理30天前的临时文件 -python kb_workflow.py --trigger cleanup --days 30 - -# 清理单个项目 -python kb_workflow.py --trigger cleanup --project "项目名称" +export KB_BASE_PATH="/path/to/knowledge-base" # 知识库根目录 +export KB_GIT_ROOT="/path/to/archive" # Git仓库目录 +export KB_GIT_REMOTE="git@github.com:user/repo.git" # 远程仓库 +export KB_GIT_USER="Username" # 用户名 +export KB_GIT_EMAIL="user@example.com" # 邮箱 ``` ## Usage -### 命令行用法 +### 作为 OpenClaw 技能 + +当用户发送消息时自动触发： +``` +用户: "添加到知识库 https://www.bilibili.com/video/BVxxx" +``` + +### 作为子代理 + +```python +# 从其他技能调用 +result = sessions_spawn( + agentId="kb-workflow", + task="添加到知识库 https://www.bilibili.com/video/BVxxx" +) +``` + +### 命令行 ```bash -# 查看状态 -python kb_workflow.py --trigger status - -# 处理待处理内容 -python kb_workflow.py --trigger manual --scope pending --limit 5 - -# 初始化Git -python kb_workflow.py --trigger git-init - -# 查看Git状态 -python kb_workflow.py --trigger git-status - -# 清理过期文件 -python kb_workflow.py --trigger cleanup --days 30 +# 处理单个URL +python kb_workflow.py --url "https://www.bilibili.com/video/BVxxx" ``` -### 参数说明 +## 子技能集成 -| 参数 | 说明 | 默认值 | -|------|------|--------| -| `--trigger` | 触发类型 | `status` | -| `--scope` | 处理范围 | `pending` | -| `--limit` | 处理数量限制 | 无限制 | -| `--days` | 清理天数 | 30 | - -## Integration - -### 技能2-4 集成 - -KB Workflow 协调以下子技能： - -1. **Skill 1: Link Parser** - 解析 URL/链接 -2. **Skill 2: Metadata** - 提取元数据 -3. **Skill 3: Content** - 获取内容 -4. **Skill 4: Deep Analyzer** - 深度分析 - -## Configuration - -### 配置文件 (kb.config.yaml) - -```yaml -git: - enabled: true - remote: "git@github.com:user/repo.git" - user: "Username" - email: "user@example.com" - -cleanup: - video_retention_days: 7 - frame_retention_days: 7 - auto_cleanup: true - -subscriptions: - check_interval_hours: 24 -``` - -## Troubleshooting - -### Git未初始化 - -```bash -# 检查配置 -python kb_workflow.py --trigger git-status - -# 初始化 -export KB_GIT_ROOT="/path/to/archive" -python kb_workflow.py --trigger git-init -``` - -### 推送失败 - -- 检查网络连接 -- 验证远程仓库权限 -- 查看错误日志 - -### 存储空间过大 - -```bash -# 清理过期文件 -python kb_workflow.py --trigger cleanup --days 7 - -# 查看存储使用情况 -python kb_workflow.py --trigger status -``` +| 阶段 | 子技能 | 输出文件 | +|------|--------|----------| +| 链接解析 | kb-link-parser | pending/{platform}_{id}.link.json | +| 元数据 | kb-bilibili-metadata | metadata/{platform}_{id}.meta.json | +| 内容 | kb-bilibili-content | content/downloads/{platform}_{id}.mp4, content/subtitles/{platform}_{id}.srt | +| OCR | kb-ocr-processor | content/{platform}_{id}.txt | +| 分析 | kb-deep-analyzer | analyzed/{platform}_{id}.md | ## Related Skills @@ -246,3 +241,4 @@ python kb_workflow.py --trigger status - **kb-bilibili-metadata** - B站元数据 - **kb-bilibili-content** - B站内容获取 - **kb-deep-analyzer** - 深度分析 +- **kb-ocr-processor** - OCR处理 diff --git a/skills/kb-workflow/SkillBug.md b/skills/kb-workflow/SkillBug.md new file mode 100644 index 0000000..7d38f50 --- /dev/null +++ b/skills/kb-workflow/SkillBug.md @@ -0,0 +1,99 @@ +# SkillBug.md + +## 问题记录 + +### 2026-02-10: 缺少自动写入pending目录的功能 + +**状态：** ✅ 已修复 + +**解决方案：** +1. 创建 `skill_main.py` 作为技能入口 +2. 创建 `kb_workflow.py` 实现完整流程 +3. 接收用户消息后自动处理 + +--- + +### 2026-02-10: kb-workflow 自己实现链接解析而非调用子技能 + +**状态：** ✅ 已修复 + +**解决方案：** +重构 `kb_workflow.py`： +1. 使用 `sessions_spawn` 调用子agent +2. 不自己实现链接解析，调用 `kb-link-parser` +3. 从文件名提取平台（不读内容） + +```python +# 调用子agent +result = sessions_spawn( + agentId="kb-link-parser", + task="解析URL，输出到pending目录" +) +``` + +--- + +### 2026-02-10: _extract_metadata 是空壳 + +**状态：** ✅ 已修复 + +**解决方案：** +实现真正的子agent调用： +```python +def call_metadata_agent(platform, link_file, metadata_dir): + agent_id = METADATA_AGENTS.get(platform) # bilibili → kb-bilibili-metadata + return call_subagent(agent_id, task, timeout=120) +``` + +--- + +## 重构总结 + +### 架构变化 + +**之前（错误）：** +``` +kb_workflow.py (自己造轮子) +├── _extract_link() - 自己解析URL +├── _extract_metadata() - 空壳 +``` + +**之后（正确）：** +``` +kb-workflow (主代理) +├── sessions_spawn() 调用子agent +│ ├── kb-link-parser → pending/*.link.json +│ ├── kb-bilibili-metadata → metadata/*.meta.json +│ ├── kb-bilibili-content → content/downloads/*.mp4 +│ ├── kb-ocr-processor → content/*.txt +│ └── kb-deep-analyzer → analyzed/*.md +└── 文件流转 + Git归档 +``` + +### 设计原则 + +| workflow 管 | workflow 不管 | +|-------------|---------------| +| 查重 | 读文件内容 | +| 调用子agent | 子技能内部格式 | +| 提取文件名（从返回结果） | 子技能逻辑 | +| 提取平台（从文件名） | | +| 传递路径信息 | | +| Git归档 | | + +### 文件变更 + +| 文件 | 操作 | 说明 | +|------|------|------| +| `kb_workflow.py` | 重构 | 使用 sessions_spawn 调用子agent | +| `skill_main.py` | 新增 | 技能入口 | +| `SKILL.md` | 更新 | 明确架构和流程 | + +--- + +## 待办 + +- [ ] 测试完整的处理流程 +- [ ] 验证 sessions_spawn 调用是否正常 +- [ ] 完善错误处理 +- [ ] 添加单元测试 diff --git a/skills/kb-workflow/kb_workflow.py b/skills/kb-workflow/kb_workflow.py index e8aa08b..b1e9ca4 100644 --- a/skills/kb-workflow/kb_workflow.py +++ b/skills/kb-workflow/kb_workflow.py @@ -1,914 +1,704 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -KB Workflow - Knowledge Base Workflow Orchestration +KB Workflow - Knowledge Base Workflow Orchestrator 知识库工作流编排技能。 -## 文件存放规则 +核心职责： +1. 接收用户消息，识别触发词和链接 +2. 查重（根据URL/文件名） +3. 调用子agent处理各阶段 +4. 文件流转（mv） +5. Git版本管理 +6. 归档和清理 -``` -knowledge-base/ -├── kb.config.yaml # 配置文件 -├── .gitignore # Git忽略规则 -├── pending/ # 待处理内容 (*.md) -├── metadata/ # 链接解析结果 (*.link.json) -├── content/ -│ ├── json/ # 元数据 (*.meta.json) -│ ├── downloads/ # 视频文件 (*.mp4/*.flv) - 7天后清理 -│ └── subtitles/ # 字幕文件 (*.srt/*.lrc) - 7天后清理 -├── frames/ # 视频截帧 (*.jpg) - 7天后清理 -├── analyzed/ # 分析报告 (*.md) -└── archive/ # Git归档目录 - ├── YYYY-MM-DD/ # 按日期归档 - └── subscriptions/ # 订阅数据 - ├── subscriptions.json # 订阅配置 - └── processed.json # 已处理项目 -``` - -## Git版本管理 - -配置环境变量： -- KB_GIT_ROOT: archive目录路径 -- KB_GIT_REMOTE: 远程仓库地址 -- KB_GIT_USER: 用户名 -- KB_GIT_EMAIL: 邮箱 - -Usage: - python kb_workflow.py --trigger manual --scope pending --limit 5 +设计原则： +- workflow不读文件内容 +- 从文件名提取平台（第一个下划线之前） +- 子技能自己定义输入输出格式 +- workflow只传递提示词 + 需要的路径信息 """ import os import sys import json import argparse -import logging import subprocess import shutil -from datetime import datetime, timedelta +import re +from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any -import yaml -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) -logger = logging.getLogger('KBWorkflow') +# ==================== 子技能映射 ==================== + +LINK_PARSER_AGENT = "kb-link-parser" +METADATA_AGENTS = { + "bilibili": "kb-bilibili-metadata", + "zhihu": "kb-zhihu-metadata", +} +CONTENT_AGENTS = { + "bilibili": "kb-bilibili-content", +} +OCR_AGENT = "kb-ocr-processor" +ANALYZER_AGENT = "kb-deep-analyzer" -class KBWorkflow: - """Knowledge Base Workflow Orchestrator""" +def extract_platform(filename: str) -> str: + """从文件名提取平台（第一个下划线之前）""" + match = re.match(r'^([a-zA-Z0-9]+)_(.+)$', filename) + if match: + return match.group(1) + return "unknown" + + +def extract_base_name(filename: str) -> str: + """从文件名提取基础名称（去掉扩展名）""" + return Path(filename).stem + + +def ensure_directories(base_path: Path) -> Dict[str, Path]: + """确保目录存在""" + dirs = { + "pending": base_path / "pending", + "metadata": base_path / "metadata", + "content": base_path / "content", + "content_downloads": base_path / "content" / "downloads", + "content_subtitles": base_path / "content" / "subtitles", + "analyzed": base_path / "analyzed", + "archive": base_path / "archive", + "archive_subscriptions": base_path / "archive" / "subscriptions", + } - STAGES = ['pending', 'metadata', 'content', 'analyzed'] + for name, dir_path in dirs.items(): + dir_path.mkdir(parents=True, exist_ok=True) - # Git配置环境变量名 - ENV_GIT_ROOT = 'KB_GIT_ROOT' - ENV_GIT_REMOTE = 'KB_GIT_REMOTE' - ENV_GIT_USER = 'KB_GIT_USER' - ENV_GIT_EMAIL = 'KB_GIT_EMAIL' + return dirs + + +def check_duplicate(base_path: Path, url: str) -> Dict[str, Any]: + """查重检查""" + metadata_dir = base_path / "metadata" + archive_dir = base_path / "archive" - def __init__(self, - base_path: str = None, - skills_path: str = None): - """ - Initialize KB Workflow - - Args: - base_path: Base path for knowledge base - skills_path: Path to skills directory - """ - self.base_path = Path(base_path) if base_path else Path(__file__).resolve().parent.parent / "knowledge-base" - self.skills_path = Path(skills_path) if skills_path else Path(__file__).resolve().parent.parent - - # Git配置 - self.git_root = os.environ.get(self.ENV_GIT_ROOT, '') - self.git_remote = os.environ.get(self.ENV_GIT_REMOTE, '') - self.git_user = os.environ.get(self.ENV_GIT_USER, '') - self.git_email = os.environ.get(self.ENV_GIT_EMAIL, '') - - # 创建目录结构 - self._ensure_directories() - - # 加载配置 - self.config = self._load_config() - - # 检查Git - self.git_status = self._check_git() - - logger.info(f"KB Workflow initialized") - logger.info(f"Base: {self.base_path}") - logger.info(f"Git: {'✓' if self.git_root else '✗'}") + # 从URL提取ID + bv_match = re.search(r'BV[a-zA-Z0-9]+', url) + content_id = bv_match.group() if bv_match else None - def _ensure_directories(self): - """创建目录结构""" - # 主目录 - for subdir in self.STAGES: - dir_path = self.base_path / subdir - if not dir_path.exists(): - dir_path.mkdir(parents=True, exist_ok=True) - - # 内容子目录 - for subdir in ['json', 'downloads', 'subtitles']: - dir_path = self.base_path / 'content' / subdir - if not dir_path.exists(): - dir_path.mkdir(parents=True, exist_ok=True) - - # 截帧目录 - frames_dir = self.base_path / 'frames' - if not frames_dir.exists(): - frames_dir.mkdir(parents=True, exist_ok=True) - - # 归档目录 - archive_dir = self.base_path / 'archive' - if not archive_dir.exists(): - archive_dir.mkdir(parents=True, exist_ok=True) - - # 订阅目录 - subs_dir = archive_dir / 'subscriptions' - if not subs_dir.exists(): - subs_dir.mkdir(parents=True, exist_ok=True) + result = {"duplicate": False, "status": None, "files": []} - def _load_config(self) -> Dict: - """加载配置文件""" - config_path = self.base_path / 'kb.config.yaml' - default_config = { - 'git': { - 'enabled': bool(self.git_root), - 'remote': self.git_remote, - 'user': self.git_user, - 'email': self.git_email - }, - 'cleanup': { - 'video_retention_days': 7, - 'frame_retention_days': 7, - 'auto_cleanup': True - }, - 'subscriptions': { - 'check_interval_hours': 24 - } - } - - if config_path.exists(): - try: - with open(config_path, 'r', encoding='utf-8') as f: - user_config = yaml.safe_load(f) or {} - # 合并配置 - for section in default_config: - if section in user_config: - default_config[section].update(user_config[section]) - except Exception as e: - logger.warning(f"Failed to load config: {e}") - - return default_config + if metadata_dir.exists(): + for f in metadata_dir.iterdir(): + if f.is_file() and content_id and content_id in f.name: + result["duplicate"] = True + result["status"] = "metadata" + result["files"].append(str(f)) - # ==================== Git管理 ==================== + if archive_dir.exists(): + for date_dir in archive_dir.iterdir(): + if date_dir.is_dir(): + for f in date_dir.iterdir(): + if f.is_file() and content_id and content_id in f.name: + result["duplicate"] = True + result["status"] = "archived" + result["files"].append(str(f)) - def _check_git(self) -> Dict[str, Any]: - """检查Git状态""" - status = { - 'initialized': False, - 'remote_configured': False, - 'remote_url': '', - 'status': 'not configured' - } - - git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' - - if not str(git_root): - status['status'] = 'not configured - set KB_GIT_ROOT' - return status - - git_dir = git_root / '.git' - if not git_dir.exists(): - status['status'] = 'not initialized' - return status - - status['initialized'] = True - - # 检查远程 - try: - result = subprocess.run( - ['git', 'remote', '-v'], - cwd=str(git_root), - capture_output=True, text=True - ) - if result.returncode == 0 and result.stdout: - status['remote_configured'] = True - status['remote_url'] = result.stdout.split()[1] if result.stdout.split() else '' - status['status'] = 'ready' - else: - status['status'] = 'no remote' - except Exception as e: - status['status'] = f'error: {e}' - - return status + return result + + +def call_subagent(agent_id: str, task: str, timeout: int = 120) -> Dict[str, Any]: + """调用子agent（通过 OpenClaw sessions_spawn 或命令行）""" + print(f"\n{'='*50}") + print(f"调用子技能: {agent_id}") + print(f"{'='*50}") - def _git_init(self) -> bool: - """初始化Git仓库""" - git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' - - if not git_root.exists(): - git_root.mkdir(parents=True, exist_ok=True) - - git_dir = git_root / '.git' - if git_dir.exists(): - logger.info("Git already initialized") - return True - - try: - # 初始化 - subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True) - - # 配置用户 - if self.git_user: - subprocess.run(['git', 'config', 'user.name', self.git_user], cwd=str(git_root)) - if self.git_email: - subprocess.run(['git', 'config', 'user.email', self.git_email], cwd=str(git_root)) - - # 创建.gitignore - ignore_content = """# 知识库Git归档 -*.tmp -*.log -""" - (git_root / '.gitignore').write_text(ignore_content) - - # 初始提交 - subprocess.run(['git', 'add', '.'], cwd=str(git_root)) - commit_msg = f"Initial commit - Knowledge base archive" - subprocess.run(['git', 'commit', '-m', commit_msg], cwd=str(git_root)) - - logger.info(f"Git initialized at {git_root}") - return True - - except Exception as e: - logger.error(f"Git init failed: {e}") - return False - - def _git_add_commit(self, files: List[str], message: str) -> bool: - """Git add and commit""" - git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' - - if not (git_root / '.git').exists(): - logger.warning("Git not initialized") - return False - - try: - # Add - for f in files: - subprocess.run(['git', 'add', f], cwd=str(git_root), capture_output=True) - - # Commit - result = subprocess.run( - ['git', 'commit', '-m', message], - cwd=str(git_root), - capture_output=True, text=True - ) - - if result.returncode == 0: - logger.info(f"Committed: {message}") - return True - else: - logger.warning(f"Commit failed: {result.stderr}") - return False - - except Exception as e: - logger.error(f"Git error: {e}") - return False - - def _git_push(self, retries: int = 3) -> Dict[str, Any]: - """Git push with retry""" - git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' - result = {'success': False, 'message': '', 'retries': 0} - - if not self.git_remote: - result['message'] = 'no remote configured' - return result - - if not (git_root / '.git').exists(): - result['message'] = 'git not initialized' - return result - - for i in range(retries): - result['retries'] = i + 1 - try: - proc = subprocess.run( - ['git', 'push', 'origin', 'master'], - cwd=str(git_root), - capture_output=True, text=True, - timeout=60 - ) - - if proc.returncode == 0: - result['success'] = True - result['message'] = 'pushed successfully' - return result - else: - result['message'] = proc.stderr - if i < retries - 1: - import time - time.sleep(2 ** i) # 指数退避 - - except subprocess.TimeoutExpired: - result['message'] = 'timeout' - except Exception as e: - result['message'] = str(e) - - logger.warning(f"Push failed after {retries} attempts") + # 尝试使用 sessions_spawn（OpenClaw 环境） + try: + from openclaw_tool import sessions_spawn + result = sessions_spawn( + agentId=agent_id, + task=task, + timeoutSeconds=timeout + ) + print(f"\n子技能返回结果:") + print(json.dumps(result, ensure_ascii=False, indent=2)) return result + except ImportError: + # 回退到命令行调用 + pass - def _git_status(self) -> Dict: - """获取Git状态""" - git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' + # 命令行回退：调用子技能的脚本 + print(f"使用命令行方式调用: {agent_id}") + + # 技能目录路径 + skills_path = Path(__file__).resolve().parent.parent + + # 从任务中提取关键信息 + import re + + # 提取URL + url_match = re.search(r'URL: (\S+)', task) + url = url_match.group(1) if url_match else "" + + # 提取输出目录 + output_match = re.search(r'输出目录: (\S+)', task) + output_dir = output_match.group(1) if output_match else "" + + # 提取输出文件名格式 + filename_match = re.search(r'输出文件名格式: (\S+)', task) + filename_format = filename_match.group(1) if filename_match else "{platform}_{id}.json" + + # 提取链接文件路径 + link_file_match = re.search(r'链接文件: (\S+)', task) + link_file = link_file_match.group(1) if link_file_match else "" + + # 提取元数据文件路径 + meta_file_match = re.search(r'元数据文件: (\S+)', task) + meta_file = meta_file_match.group(1) if meta_file_match else "" + + # kb-link-parser + if agent_id == "kb-link-parser": + cmd = ['python3', str(skills_path / "kb-link-parser" / "link-parser.py"), '--url', url] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + if proc.returncode == 0: + try: + data = json.loads(proc.stdout) + if data.get('success'): + platform = data.get('platform', 'unknown') + content_id = data.get('id', 'unknown') + ext = '.link.json' + filename = f"{platform}_{content_id}{ext}" + output_file = Path(output_dir) / filename + output_file.parent.mkdir(parents=True, exist_ok=True) + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + print(f"保存文件: {output_file}") + return {"success": True, "output_file": str(output_file)} + else: + return {"success": False, "error": data.get('error', 'unknown')} + except json.JSONDecodeError: + return {"success": False, "error": f"解析失败: {proc.stdout[:200]}"} + else: + return {"success": False, "error": proc.stderr} + except Exception as e: + return {"success": False, "error": str(e)} + + # kb-bilibili-metadata + elif agent_id == "kb-bilibili-metadata": + # 从链接文件或URL中提取BV号 + bv = None + if link_file and Path(link_file).exists(): + with open(link_file, 'r', encoding='utf-8') as f: + link_data = json.load(f) + bv = link_data.get('id') + if not bv: + bv_match = re.search(r'(BV[a-zA-Z0-9]+)', url or link_file) + bv = bv_match.group(1) if bv_match else None - if not (git_root / '.git').exists(): - return {'error': 'not initialized'} + if not bv: + return {"success": False, "error": "无法提取BV号"} + + # 生成输出文件名 + filename = f"bilibili_{bv}.meta.json" + output_file = Path(output_dir) / filename + output_file.parent.mkdir(parents=True, exist_ok=True) + + cmd = [ + 'python3', str(skills_path / "kb-bilibili-metadata" / "bilibili_metadata.py"), + '--id', bv, + '--type', 'video', + '--output', str(output_file) + ] try: - # 获取远程URL - remote = subprocess.run( - ['git', 'remote', 'get-url', 'origin'], - cwd=str(git_root), capture_output=True, text=True - ) - remote_url = remote.stdout.strip() if remote.returncode == 0 else '' - - # 获取提交数 - log = subprocess.run( - ['git', 'rev-list', '--count', 'HEAD'], - cwd=str(git_root), capture_output=True, text=True - ) - commits = log.stdout.strip() if log.returncode == 0 else '0' - - return { - 'initialized': True, - 'remote_url': remote_url, - 'commits': commits, - 'status': 'ready' if remote_url else 'no remote' + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + if proc.returncode == 0: + print(f"保存文件: {output_file}") + return {"success": True, "output_file": str(output_file)} + else: + return {"success": False, "error": proc.stderr} + except Exception as e: + return {"success": False, "error": str(e)} + + # kb-bilibili-content + elif agent_id == "kb-bilibili-content": + bv_match = re.search(r'(BV[a-zA-Z0-9]+)', url or meta_file) + bv = bv_match.group(1) if bv_match else None + + if not bv: + return {"success": False, "error": "无法提取BV号"} + + downloads_dir = Path(output_dir) / "downloads" + subtitles_dir = Path(output_dir) / "subtitles" + downloads_dir.mkdir(parents=True, exist_ok=True) + subtitles_dir.mkdir(parents=True, exist_ok=True) + + video_file = None + subtitle_file = None + + # 下载视频 + cmd_video = [ + 'python3', str(skills_path / "kb-bilibili-content" / "bilibili_content.py"), + '--id', bv, + '--type', 'video', + '--output', output_dir + ] + + try: + proc = subprocess.run(cmd_video, capture_output=True, text=True, timeout=timeout) + if proc.returncode == 0: + # 查找生成的视频文件（可能包含BV号） + for f in Path(output_dir).iterdir(): + if f.is_file() and bv in f.name and f.suffix in ['.flv', '.mp4']: + # 重命名为标准格式 + new_name = f"bilibili_{bv}{f.suffix}" + new_path = downloads_dir / new_name + f.rename(new_path) + video_file = str(new_path) + break + except Exception as e: + print(f"视频下载警告: {e}") + + # 下载字幕 + cmd_subtitle = [ + 'python3', str(skills_path / "kb-bilibili-content" / "bilibili_content.py"), + '--id', bv, + '--type', 'subtitle', + '--output', output_dir + ] + + try: + proc = subprocess.run(cmd_subtitle, capture_output=True, text=True, timeout=timeout) + if proc.returncode == 0: + # 查找生成的字幕文件 + for f in Path(output_dir).iterdir(): + if f.is_file() and bv in f.name and f.suffix in ['.srt', '.lrc', '.ass']: + # 重命名为标准格式 + new_name = f"bilibili_{bv}{f.suffix}" + new_path = subtitles_dir / new_name + f.rename(new_path) + subtitle_file = str(new_path) + break + except Exception as e: + print(f"字幕下载警告: {e}") + + if video_file or subtitle_file: + return {"success": True, "output_file": video_file, "subtitles": subtitle_file} + else: + return {"success": False, "error": "下载未生成任何文件"} + + # kb-ocr-processor + elif agent_id == "kb-ocr-processor": + # 查找字幕文件 + subtitles_dir = Path(output_dir) / "subtitles" + subtitle_file = None + if subtitles_dir.exists(): + for f in subtitles_dir.iterdir(): + if f.suffix in ['.srt', '.lrc']: + subtitle_file = str(f) + break + + if not subtitle_file: + return {"success": False, "error": "未找到字幕文件"} + + # 生成输出文件名 + text_file = Path(output_dir) / f"{Path(subtitle_file).stem}.txt" + + cmd = [ + 'python3', str(skills_path / "kb-ocr-processor" / "ocr_processor.py"), + '--input', subtitle_file, + '--output', str(text_file) + ] + + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + if proc.returncode == 0: + print(f"保存文件: {text_file}") + return {"success": True, "output_file": str(text_file)} + else: + return {"success": False, "error": proc.stderr} + except Exception as e: + return {"success": False, "error": str(e)} + + # kb-deep-analyzer + elif agent_id == "kb-deep-analyzer": + # 生成输出文件名 + meta_name = Path(meta_file).stem if meta_file else "unknown" + filename = f"{meta_name}.md" + output_file = Path(output_dir) / filename + output_file.parent.mkdir(parents=True, exist_ok=True) + + # 从任务中提取文字内容文件路径 + text_file_match = re.search(r'文字内容文件: (\S+)', task) + text_content_file = text_file_match.group(1) if text_file_match else "" + + # 如果没有文字内容文件，使用 metadata 作为 content + content_file = meta_file + if text_content_file and Path(text_content_file).exists(): + # 从文字文件创建 content JSON + content_json = { + "type": "transcript", + "source": text_content_file, + "text": Path(text_content_file).read_text(encoding='utf-8')[:10000] if Path(text_content_file).exists() else "" } + content_file = Path(output_dir) / f"{meta_name}.content.json" + with open(content_file, 'w', encoding='utf-8') as f: + json.dump(content_json, f, ensure_ascii=False, indent=2) - except Exception as e: - return {'error': str(e)} - - # ==================== 订阅管理 ==================== - - def _load_subscriptions(self) -> Dict: - """加载订阅数据""" - subs_file = self.base_path / 'archive' / 'subscriptions' / 'subscriptions.json' - - default_subs = { - 'version': '1.0', - 'updated_at': datetime.now().isoformat(), - 'subscriptions': [], - 'processed_items': [] - } - - if subs_file.exists(): - try: - with open(subs_file, 'r', encoding='utf-8') as f: - return json.load(f) - except Exception as e: - logger.warning(f"Failed to load subscriptions: {e}") - - return default_subs - - def _save_subscriptions(self, data: Dict) -> bool: - """保存订阅数据""" - subs_file = self.base_path / 'archive' / 'subscriptions' / 'subscriptions.json' + cmd = [ + 'python3', str(skills_path / "kb-deep-analyzer" / "deep_analyzer.py"), + '--metadata', meta_file or output_dir, + '--content', str(content_file), + '--output', str(output_file) + ] try: - data['updated_at'] = datetime.now().isoformat() - with open(subs_file, 'w', encoding='utf-8') as f: - json.dump(data, f, ensure_ascii=False, indent=2) - - # Git提交 - self._git_add_commit( - ['subscriptions/subscriptions.json'], - f"Update subscriptions - {datetime.now().strftime('%Y-%m-%d %H:%M')}" - ) - - return True + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + if proc.returncode == 0: + print(f"保存文件: {output_file}") + return {"success": True, "output_file": str(output_file)} + else: + return {"success": False, "error": proc.stderr} except Exception as e: - logger.error(f"Failed to save subscriptions: {e}") - return False + return {"success": False, "error": str(e)} - def add_subscription(self, platform: str, subs_type: str, identifier: str, name: str = '') -> Dict: - """添加订阅""" - subs = self._load_subscriptions() - - sub_id = f"sub_{len(subs['subscriptions']) + 1:03d}" - - new_sub = { - 'id': sub_id, - 'platform': platform, - 'type': subs_type, - 'identifier': identifier, - 'name': name, - 'subscribed_at': datetime.now().isoformat(), - 'last_checked': None, - 'last_update': None, - 'status': 'active' - } - - subs['subscriptions'].append(new_sub) - self._save_subscriptions(subs) - - logger.info(f"Added subscription: {sub_id} - {platform}/{subs_type}/{identifier}") - return new_sub + return {"success": False, "error": f"暂未实现命令行回退: {agent_id}"} + + +def call_link_parser(url: str, pending_dir: Path) -> Dict[str, Any]: + """调用 kb-link-parser""" + task = f"""解析以下URL，将结果保存到指定目录： + +URL: {url} + +输出目录: {pending_dir} +输出文件名格式: {{platform}}_{{id}}.link.json + +请解析URL中的平台和内容ID，将解析结果保存为JSON文件。 +返回结果时告诉我输出文件的完整路径。 + +只需要解析URL并保存结果，不需要下载内容。""" - def mark_item_processed(self, item_id: str, title: str = '', archived: bool = True) -> bool: - """标记项目为已处理""" - subs = self._load_subscriptions() + return call_subagent(LINK_PARSER_AGENT, task, timeout=60) + + +def call_metadata_agent(platform: str, link_file: Path, metadata_dir: Path) -> Dict[str, Any]: + """调用对应的元数据获取技能""" + agent_id = METADATA_AGENTS.get(platform) + if not agent_id: + return {"success": False, "error": f"不支持的平台: {platform}"} + + task = f"""获取链接的元数据： + +链接文件: {link_file} +输出目录: {metadata_dir} + +请读取链接文件，提取URL，然后获取该链接的元数据（标题、作者、播放量等）。 +将元数据保存为JSON文件，文件名格式: {{platform}}_{{id}}.meta.json + +返回结果时告诉我输出文件的完整路径。""" + + return call_subagent(agent_id, task, timeout=120) + + +def call_content_agent(platform: str, meta_file: Path, content_dir: Path) -> Dict[str, Any]: + """调用对应的内容获取技能""" + agent_id = CONTENT_AGENTS.get(platform) + if not agent_id: + return {"success": False, "error": f"不支持的平台: {platform}"} + + task = f"""下载视频内容： + +元数据文件: {meta_file} +下载目录: {content_dir} + +请读取元数据文件，获取视频信息，然后下载视频和字幕。 +视频保存到: content/downloads/ +字幕保存到: content/subtitles/ + +文件名格式: {{platform}}_{{id}}.mp4 和 {{platform}}_{{id}}.srt + +返回结果时告诉我下载的视频文件和字幕文件的完整路径。""" + + return call_subagent(agent_id, task, timeout=600) + + +def call_ocr_processor(subtitle_file: Path, content_dir: Path) -> Dict[str, Any]: + """调用 OCR 处理""" + task = f"""处理字幕文件，提取文字内容： + +字幕文件: {subtitle_file} +输出目录: {content_dir} + +请读取字幕文件，将内容转换为纯文本。 +输出文件名格式: {{platform}}_{{id}}.txt + +返回结果时告诉我输出文件的完整路径。""" + + return call_subagent(OCR_AGENT, task, timeout=120) + + +def call_deep_analyzer(meta_file: Path, text_file: Path, analyzed_dir: Path) -> Dict[str, Any]: + """调用深度分析""" + task = f"""深度分析内容： + +元数据文件: {meta_file} +文字内容文件: {text_file} +输出目录: {analyzed_dir} + +请读取元数据和文字内容，进行深度分析。 +生成分析报告，保存为Markdown文件。 +文件名格式: {{platform}}_{{id}}.md + +分析报告应包含： +- 一句话总结 +- 核心要点 +- 时间轴（如适用） +- 推荐标签 + +返回结果时告诉我输出文件的完整路径。""" + + return call_subagent(ANALYZER_AGENT, task, timeout=180) + + +def archive_and_git(base_path: Path, analyzed_file: Path, git_root: Path, git_user: str, git_email: str, git_remote: str) -> Dict[str, Any]: + """归档并提交Git""" + result = { + "success": False, + "archive_path": "", + "git_commit": False, + "git_push": None, + "error": None + } + + # 1. 创建归档目录 + date_str = datetime.now().strftime('%Y-%m-%d') + date_dir = base_path / "archive" / date_str + date_dir.mkdir(parents=True, exist_ok=True) + + # 2. 复制文件到归档目录 + archive_file = date_dir / analyzed_file.name + shutil.copy(analyzed_file, archive_file) + result["archive_path"] = str(archive_file) + + # 3. 复制元数据 + meta_file = base_path / "metadata" / f"{extract_base_name(analyzed_file.name)}.meta.json" + if meta_file.exists(): + shutil.copy(meta_file, date_dir / meta_file.name) + + # 4. Git操作 + git_dir = git_root / ".git" + if not git_dir.exists(): + subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True) + if git_user: + subprocess.run(['git', 'config', 'user.name', git_user], cwd=str(git_root)) + if git_email: + subprocess.run(['git', 'config', 'user.email', git_email], cwd=str(git_root)) + + if (git_root / ".git").exists(): + subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True) + commit_msg = f"Archive: {analyzed_file.name} - {datetime.now().isoformat()}" + commit_result = subprocess.run( + ['git', 'commit', '-m', commit_msg], + cwd=str(git_root), capture_output=True, text=True + ) + result["git_commit"] = commit_result.returncode == 0 - # 检查是否已存在 - existing = [p for p in subs['processed_items'] if p['id'] == item_id] - if existing: - return True # 已存在，不重复添加 + if git_remote and commit_result.returncode == 0: + push_result = subprocess.run( + ['git', 'push', 'origin', 'master'], + cwd=str(git_root), capture_output=True, text=True, timeout=60 + ) + result["git_push"] = {"success": push_result.returncode == 0} + + result["success"] = True + return result + + +def process_url(url: str, base_path: Path = None) -> Dict[str, Any]: + """处理单个URL的完整流程""" + if not base_path: + base_path = Path.home() / "work" / "knowledge-base" + + git_root = os.environ.get('KB_GIT_ROOT', str(base_path / 'archive')) + git_user = os.environ.get('KB_GIT_USER', '') + git_email = os.environ.get('KB_GIT_EMAIL', '') + git_remote = os.environ.get('KB_GIT_REMOTE', '') + + dirs = ensure_directories(base_path) + + result = { + "url": url, + "success": False, + "stages": [], + "final_file": None, + "archive": None, + "error": None + } + + try: + # 查重 + print(f"\n{'='*50}") + print(f"步骤1: 查重") + print(f"{'='*50}") - new_item = { - 'id': item_id, - 'title': title, - 'processed_at': datetime.now().isoformat(), - 'archived': archived - } + dup_result = check_duplicate(base_path, url) + if dup_result["duplicate"]: + result["error"] = f"已存在: {dup_result['status']}" + result["duplicate"] = True + result["existing_files"] = dup_result["files"] + return result - subs['processed_items'].append(new_item) + # kb-link-parser + print(f"\n{'='*50}") + print(f"步骤2: 链接解析 (kb-link-parser)") + print(f"{'='*50}") - # Git提交 - self._git_add_commit( - ['subscriptions/subscriptions.json'], - f"Mark processed: {item_id}" + link_result = call_link_parser(url, dirs["pending"]) + if not link_result.get("success"): + result["error"] = f"链接解析失败: {link_result.get('error')}" + return result + + link_file = link_result.get("output_file") + if not link_file: + result["error"] = "链接解析未返回输出文件" + return result + + link_file = Path(link_file) + result["stages"].append({"stage": "link_parsing", "file": str(link_file)}) + + # 提取平台 + filename = link_file.name + platform = extract_platform(filename) + content_id = extract_base_name(filename) + + print(f"\n平台: {platform}") + print(f"内容ID: {content_id}") + + # 元数据 + print(f"\n{'='*50}") + print(f"步骤3: 元数据提取 ({METADATA_AGENTS.get(platform, 'unknown')})") + print(f"{'='*50}") + + meta_result = call_metadata_agent(platform, link_file, dirs["metadata"]) + if not meta_result.get("success"): + result["error"] = f"元数据获取失败: {meta_result.get('error')}" + return result + + meta_file = meta_result.get("output_file") + if not meta_file: + result["error"] = "元数据获取未返回输出文件" + return result + + meta_file = Path(meta_file) + result["stages"].append({"stage": "metadata", "file": str(meta_file)}) + + # 内容下载 + print(f"\n{'='*50}") + print(f"步骤4: 内容下载 ({CONTENT_AGENTS.get(platform, 'unknown')})") + print(f"{'='*50}") + + content_result = call_content_agent(platform, meta_file, dirs["content"]) + if not content_result.get("success"): + result["error"] = f"内容下载失败: {content_result.get('error')}" + return result + + video_file = content_result.get("output_file") + subtitle_file = content_result.get("subtitle") or content_result.get("subtitles") + + result["stages"].append({"stage": "content", "video": video_file, "subtitles": subtitle_file}) + + # OCR + text_file = None + if subtitle_file: + print(f"\n{'='*50}") + print(f"步骤5: OCR处理 (kb-ocr-processor)") + print(f"{'='*50}") + + ocr_result = call_ocr_processor(Path(subtitle_file), dirs["content"]) + text_file = ocr_result.get("output_file") if ocr_result.get("success") else None + + if text_file: + result["stages"].append({"stage": "ocr", "file": text_file}) + else: + print(f"\n无字幕文件，跳过OCR处理") + + # 深度分析 + print(f"\n{'='*50}") + print(f"步骤6: 深度分析 (kb-deep-analyzer)") + print(f"{'='*50}") + + analyze_result = call_deep_analyzer( + meta_file, + Path(text_file) if text_file else meta_file, + dirs["analyzed"] + ) + if not analyze_result.get("success"): + result["error"] = f"深度分析失败: {analyze_result.get('error')}" + return result + + analyzed_file = analyze_result.get("output_file") + if not analyzed_file: + result["error"] = "深度分析未返回输出文件" + return result + + analyzed_file = Path(analyzed_file) + result["stages"].append({"stage": "analyzer", "file": str(analyzed_file)}) + result["final_file"] = str(analyzed_file) + + # 归档和Git + print(f"\n{'='*50}") + print(f"步骤7: 归档和Git") + print(f"{'='*50}") + + archive_result = archive_and_git( + base_path, analyzed_file, + Path(git_root), git_user, git_email, git_remote ) - return self._save_subscriptions(subs) + result["archive"] = archive_result + result["success"] = True + + print(f"\n{'='*50}") + print(f"处理完成！") + print(f"{'='*50}") - def is_item_processed(self, item_id: str) -> bool: - """检查项目是否已处理""" - subs = self._load_subscriptions() - return any(p['id'] == item_id for p in subs['processed_items']) + except Exception as e: + result["error"] = str(e) + import traceback + traceback.print_exc() - # ==================== 清理策略 ==================== - - def cleanup_project(self, base_name: str) -> Dict[str, Any]: - """清理单个项目的临时文件""" - result = {'cleaned': [], 'errors': []} - - # 清理视频文件 - downloads_dir = self.base_path / 'content' / 'downloads' - for f in downloads_dir.glob(f'{base_name}.*'): - try: - f.unlink() - result['cleaned'].append(f"downloads/{f.name}") - except Exception as e: - result['errors'].append(str(e)) - - # 清理截帧 - frames_dir = self.base_path / 'frames' - for f in frames_dir.glob(f'{base_name}_*'): - try: - f.unlink() - result['cleaned'].append(f"frames/{f.name}") - except Exception as e: - result['errors'].append(str(e)) - - if result['cleaned']: - logger.info(f"Cleaned project {base_name}: {len(result['cleaned'])} files") - - return result - - def cleanup_expired(self, days: int = 30) -> Dict[str, Any]: - """清理过期临时文件""" - result = {'cleaned': [], 'errors': [], 'count': 0} - - cutoff = datetime.now() - timedelta(days=days) - - # 清理下载目录 - for f in (self.base_path / 'content' / 'downloads').iterdir(): - if f.is_file(): - mtime = datetime.fromtimestamp(f.stat().st_mtime) - if mtime < cutoff: - try: - f.unlink() - result['cleaned'].append(str(f.relative_to(self.base_path))) - result['count'] += 1 - except Exception as e: - result['errors'].append(str(e)) - - # 清理截帧目录 - for f in (self.base_path / 'frames').iterdir(): - if f.is_file(): - mtime = datetime.fromtimestamp(f.stat().st_mtime) - if mtime < cutoff: - try: - f.unlink() - result['cleaned'].append(str(f.relative_to(self.base_path))) - result['count'] += 1 - except Exception as e: - result['errors'].append(str(e)) - - logger.info(f"Expired cleanup: {result['count']} files") - return result - - # ==================== 归档 ==================== - - def archive_item(self, base_name: str, report_content: str) -> Dict[str, Any]: - """归档项目""" - result = { - 'success': False, - 'archive_path': '', - 'git_commit': False, - 'git_push': None, - 'cleanup': None, - 'errors': [] - } - - # 1. 创建归档目录 - date_dir = self.base_path / 'archive' / datetime.now().strftime('%Y-%m-%d') - date_dir.mkdir(parents=True, exist_ok=True) - - # 2. 复制分析报告 - report_path = date_dir / f"{base_name}.md" - with open(report_path, 'w', encoding='utf-8') as f: - f.write(report_content) - result['archive_path'] = str(report_path) - - # 3. 复制元数据 - meta_src = self.base_path / 'content' / 'json' / f"{base_name}.meta.json" - if meta_src.exists(): - shutil.copy(meta_src, date_dir / f"{base_name}.meta.json") - - # 4. Git提交 - if self._git_add_commit( - [datetime.now().strftime('%Y-%m-%d')], - f"Archive: {base_name}" - ): - result['git_commit'] = True - - # 5. 推送到远程 - push_result = self._git_push() - result['git_push'] = push_result - - # 6. 清理临时文件 - result['cleanup'] = self.cleanup_project(base_name) - - result['success'] = True - return result - - # ==================== 链接解析 ==================== - - def _extract_link(self, file_path: Path) -> Optional[Dict[str, Any]]: - """解析链接""" - try: - content = file_path.read_text(encoding='utf-8') - - import re - url_pattern = r'https?://[^\s<>"]+' - urls = re.findall(url_pattern, content) - - if not urls: - return None - - url = urls[0] - result = { - "platform": self._detect_platform(url), - "type": self._detect_type(url), - "id": self._extract_id(url), - "url": url, - "original_content": content - } - - logger.info(f"Parsed: {result['platform']}/{result['type']} - {result['id']}") - return result - - except Exception as e: - logger.error(f"Link extraction failed: {e}") - return None - - def _detect_platform(self, url: str) -> str: - """检测平台""" - if 'bilibili.com' in url or 'b23.tv' in url: - return 'bilibili' - elif 'zhihu.com' in url or 'zhuanlan.zhihu.com' in url: - return 'zhihu' - return 'unknown' - - def _detect_type(self, url: str) -> str: - """检测类型""" - if '/video/' in url or '/zvideo/' in url: - return 'video' - elif '/read/' in url or '/p/' in url: - return 'article' - elif 'space.bilibili.com' in url: - return 'up主' - return 'unknown' - - def _extract_id(self, url: str) -> str: - """提取ID""" - import re - bv = re.search(r'BV[a-zA-Z0-9]{10}', url) - if bv: return bv.group() - cv = re.search(r'cv(\d+)', url) - if cv: return f"cv{cv.group(1)}" - num = re.search(r'/(\d+)', url) - if num: return num.group(1) - return '' - - # ==================== 主处理流程 ==================== - - def _save_json(self, data: Dict, subdir: str, base_name: str, suffix: str) -> Path: - """保存JSON""" - file_path = self.base_path / subdir / f"{base_name}{suffix}" - with open(file_path, 'w', encoding='utf-8') as f: - json.dump(data, f, ensure_ascii=False, indent=2) - logger.info(f"Saved: {subdir}/{base_name}{suffix}") - return file_path - - def _move_to_stage(self, file_path: Path, to_stage: str) -> Path: - """移动文件""" - new_path = self.base_path / to_stage / file_path.name - if file_path.exists(): - file_path.rename(new_path) - logger.info(f"Moved: {file_path.name} → {to_stage}/") - return new_path - - def process_item(self, item: Dict) -> Dict[str, Any]: - """处理单个项目""" - result = { - "item": item, - "success": False, - "stages_completed": [], - "files_created": [], - "archive": None, - "error": None - } - - file_path = Path(item['path']) - base_name = file_path.stem - - try: - logger.info(f"Processing: {file_path.name}") - - # 阶段1: 链接解析 - link_info = self._extract_link(file_path) - if not link_info: - result["error"] = "无法解析链接" - return result - - result["stages_completed"].append("link_parsing") - self._save_json(link_info, 'metadata', base_name, '.link.json') - - # 阶段2: 元数据提取 - logger.info(f"Stage 2: Extracting metadata...") - metadata = self._extract_metadata(link_info) - if 'error' not in metadata: - self._save_json(metadata, 'content/json', base_name, '.meta.json') - result["stages_completed"].append("metadata_extraction") - - # 标记为已处理 - item_id = metadata.get('bvid') or metadata.get('id') or base_name - self.mark_item_processed(item_id, metadata.get('title', '')) - - # 移动原文件 - self._move_to_stage(file_path, 'metadata') - - result["success"] = len(result["stages_completed"]) >= 2 - logger.info(f"Complete: {len(result['stages_completed'])} stages") - - except Exception as e: - result["error"] = str(e) - logger.error(f"Processing failed: {e}") - - return result - - def _extract_metadata(self, link_info: Dict) -> Dict: - """提取元数据""" - platform = link_info.get('platform', '') - content_type = link_info.get('type', '') - content_id = link_info.get('id', '') - - metadata = { - "platform": platform, - "type": content_type, - "id": content_id, - "url": link_info.get('url', ''), - "source_file": link_info.get('source_file', ''), - "extracted_at": datetime.now().isoformat() - } - - # 调用相应技能的提取逻辑... - # (简化处理) - return metadata - - def process_all(self, scope: str = "pending", limit: int = None) -> Dict[str, Any]: - """处理所有项目""" - summary = {"processed": 0, "failed": 0, "results": []} - - inbox = self.scan_inbox(scope=scope) - - for stage, items in inbox.items(): - if limit and summary['processed'] >= limit: - break - - for item in items: - if limit and summary['processed'] >= limit: - break - - result = self.process_item(item) - summary["results"].append(result) - - if result["success"]: - summary["processed"] += 1 - else: - summary["failed"] += 1 - - return summary - - # ==================== 状态和汇报 ==================== - - def scan_inbox(self, scope: str = "all") -> Dict[str, List[Dict]]: - """扫描收件箱""" - result = {} - - stages = self.STAGES if scope == "all" else ([scope] if scope in self.STAGES else self.STAGES) - - for stage in stages: - dir_path = self.base_path / stage - items = [] - - if dir_path.exists(): - for item_path in dir_path.iterdir(): - if item_path.is_file(): - items.append({ - 'name': item_path.name, - 'path': str(item_path), - 'stage': stage, - 'modified': datetime.fromtimestamp( - item_path.stat().st_mtime - ).isoformat() - }) - - result[stage] = items - - return result - - def get_status(self) -> Dict[str, Any]: - """获取状态""" - inbox = self.scan_inbox(scope="all") - - # 统计文件 - file_counts = {} - total_size = 0 - - for root, dirs, files in os.walk(self.base_path): - for f in files: - file_path = Path(root) / f - ext = file_path.suffix - file_counts[ext] = file_counts.get(ext, 0) + 1 - total_size += file_path.stat().st_size - - # Git状态 - git_info = self._git_status() - - # 订阅状态 - subs = self._load_subscriptions() - - return { - "status": "ready", - "timestamp": datetime.now().isoformat(), - "stages": inbox, - "total_items": sum(len(items) for items in inbox.values()), - "total_size_mb": round(total_size / 1024 / 1024, 2), - "file_counts": file_counts, - "git": git_info, - "subscriptions": { - "count": len(subs.get('subscriptions', [])), - "processed": len(subs.get('processed_items', [])) - }, - "directories": { - 'pending': '待处理内容', - 'metadata': '链接解析结果', - 'content/json': '元数据', - 'content/downloads': '视频文件(临时)', - 'content/subtitles': '字幕文件(临时)', - 'frames': '截帧(临时)', - 'analyzed': '分析报告', - 'archive/': 'Git归档' - } - } - - def print_status(self): - """打印状态""" - status = self.get_status() - - print("\n" + "="*50) - print("KB Workflow Status") - print("="*50) - - print(f"\n📁 阶段统计:") - for stage, items in status['stages'].items(): - print(f" {stage}: {len(items)} 项") - - print(f"\n💾 存储: {status['total_size_mb']} MB") - - print(f"\n🔗 Git:") - git = status.get('git', {}) - if git.get('initialized'): - print(f" ✓ 已初始化") - print(f" 远程: {git.get('remote_url', '未配置')}") - print(f" 提交: {git.get('commits', 0)} 个") - else: - print(f" ✗ 未初始化") - print(f" 状态: {git.get('status', 'unknown')}") - - print(f"\n📋 订阅:") - subs = status.get('subscriptions', {}) - print(f" 订阅数: {subs.get('count', 0)}") - print(f" 已处理: {subs.get('processed', 0)}") - - print(f"\n📚 目录说明:") - for dir, desc in status.get('directories', {}).items(): - print(f" {dir}: {desc}") - - # 清理建议 - if status['total_size_mb'] > 1000: - print(f"\n💡 建议: 存储空间较大 ({status['total_size_mb']} MB)") - print(f" 运行清理: python kb_workflow.py --trigger cleanup --days 30") - - print() + return result -# ==================== CLI ==================== - def main(): + """CLI入口""" parser = argparse.ArgumentParser( - description='KB Workflow - Knowledge Base Workflow Orchestration', - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -使用示例: - # 查看状态 - python kb_workflow.py --trigger status - - # 处理待处理内容 - python kb_workflow.py --trigger manual --scope pending --limit 5 - - # 初始化Git - python kb_workflow.py --trigger git-init - - # 清理过期文件 - python kb_workflow.py --trigger cleanup --days 30 - - # 查看Git状态 - python kb_workflow.py --trigger git-status - -配置环境变量: - export KB_GIT_ROOT="/path/to/archive" # Git仓库目录 - export KB_GIT_REMOTE="git@github.com:user/repo.git" # 远程仓库 - export KB_GIT_USER="Username" # 用户名 - export KB_GIT_EMAIL="user@example.com" # 邮箱 - """ + description='KB Workflow - Knowledge Base Workflow Orchestrator', + formatter_class=argparse.RawDescriptionHelpFormatter ) - parser.add_argument('--trigger', type=str, default='status', - choices=['status', 'manual', 'cleanup', 'git-init', 'git-status', 'archive'], - help='触发类型') - parser.add_argument('--scope', type=str, default='pending', - choices=['all', 'pending', 'metadata', 'content', 'analyzed'], - help='处理范围') - parser.add_argument('--limit', type=int, default=None, - help='处理数量限制') - parser.add_argument('--days', type=int, default=30, - help='清理天数') - parser.add_argument('--base-path', type=str, default=None, - help='知识库基础路径') + parser.add_argument('--url', type=str, help='要处理的URL') + parser.add_argument('--base-path', type=str, help='知识库根目录') args = parser.parse_args() - workflow = KBWorkflow(base_path=args.base_path) + if not args.url: + print("请提供URL: --url https://...") + sys.exit(1) - if args.trigger == 'status': - workflow.print_status() + base_path = Path(args.base_path) if args.base_path else None + result = process_url(args.url, base_path) - elif args.trigger == 'manual': - result = workflow.process_all(scope=args.scope, limit=args.limit) - print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) + print(f"\n{'='*50}") + print("KB Workflow Result") + print(f"{'='*50}") + print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) - elif args.trigger == 'cleanup': - result = workflow.cleanup_expired(days=args.days) - print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) - - elif args.trigger == 'git-init': - success = workflow._git_init() - print(f"Git init: {'success' if success else 'failed'}") - - elif args.trigger == 'git-status': - status = workflow._git_status() - print(json.dumps(status, ensure_ascii=False, indent=2, default=str)) - - else: - print(f"Unknown trigger: {args.trigger}") - return 1 - - return 0 + sys.exit(0 if result.get('success') else 1) if __name__ == '__main__': - sys.exit(main()) + main() diff --git a/skills/kb-workflow/skill_main.py b/skills/kb-workflow/skill_main.py new file mode 100644 index 0000000..80c3ba5 --- /dev/null +++ b/skills/kb-workflow/skill_main.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +KB Workflow Skill Main Entry + +当用户发送包含触发词的消息时自动触发： +- "添加到知识库 + 链接" +- "保存到知识库 + 链接" + +职责： +1. 接收用户消息 +2. 识别触发词和链接 +3. 调用 kb_workflow.process_url 处理 +4. 汇报结果 +""" + +import re +import sys +import json +from pathlib import Path + +# 触发词模式 +TRIGGER_PATTERNS = [ + r'添加到知识库\s+(https?://[^\s]+)', + r'保存到知识库\s+(https?://[^\s]+)', +] + + +def extract_link(message: str) -> str | None: + """从消息中提取链接""" + for pattern in TRIGGER_PATTERNS: + match = re.search(pattern, message) + if match: + return match.group(1) + return None + + +def main(message: str) -> Dict[str, Any]: + """ + 主入口函数 - 当 kb-workflow 作为子代理被调用时执行 + """ + result = { + "success": False, + "url": None, + "workflow_result": None, + "error": None + } + + try: + # 1. 提取链接 + url = extract_link(message) + if not url: + result["error"] = "未找到链接" + return result + + result["url"] = url + + # 2. 调用workflow处理 + from kb_workflow import process_url + + import os + base_path_str = os.environ.get('KB_BASE_PATH', '') + base_path = Path(base_path_str) if base_path_str else None + + # 处理 + workflow_result = process_url(url, base_path) + result["workflow_result"] = workflow_result + + if workflow_result.get("success"): + result["success"] = True + + except Exception as e: + result["error"] = str(e) + import traceback + traceback.print_exc() + + return result + + +def print_result(result: Dict[str, Any]): + """打印结果""" + print("\n" + "="*50) + print("KB Workflow Result") + print("="*50) + + print(f"\nURL: {result.get('url', 'N/A')}") + + wr = result.get('workflow_result', {}) + if wr: + print(f"\n处理状态: {'成功' if wr.get('success') else '失败'}") + + if wr.get('stages'): + print(f"\n处理阶段:") + for stage in wr['stages']: + stage_name = stage.get('stage', 'unknown') + print(f" ✓ {stage_name}") + + if wr.get('final_file'): + print(f"\n最终文件: {wr['final_file']}") + + archive = wr.get('archive', {}) + if archive: + print(f"\n归档:") + print(f" 归档路径: {archive.get('archive_path', 'N/A')}") + print(f" Git提交: {'是' if archive.get('git_commit') else '否'}") + push = archive.get('git_push', {}) + if push: + print(f" Git推送: {'成功' if push.get('success') else '失败'}") + + if wr.get('error'): + print(f"\n错误: {wr['error']}") + + if result.get('error'): + print(f"\n错误: {result['error']}") + + print() + + +if __name__ == "__main__": + # 从参数获取消息（作为子代理时） + if len(sys.argv) > 1: + message = sys.argv[1] + else: + message = sys.stdin.read() + + if not message: + print("请提供消息内容") + sys.exit(1) + + result = main(message) + print_result(result) + + sys.exit(0 if result.get('success') else 1)