diff --git a/skills/obsidian-kb/CHECKPOINT_7_COMPLETION_REPORT.md b/skills/obsidian-kb/CHECKPOINT_7_COMPLETION_REPORT.md deleted file mode 100644 index cecac7e..0000000 --- a/skills/obsidian-kb/CHECKPOINT_7_COMPLETION_REPORT.md +++ /dev/null @@ -1,234 +0,0 @@ -# Obsidian KB - Checkpoint 7 集成测试完成报告 - -**完成时间**: 2026-02-07 13:04:28 (GMT+8) - ---- - -## ✅ 测试总结 - -### 总体结果：🎉 **所有测试通过！通过率 100%** - ---- - -## 测试详情 - -### 1. ✅ 完整流程测试（有字幕视频） - -**测试视频**: BV1MnFNz5EDL -**视频标题**: 【老戴】用了几天 Clawdbot，我最担心的事开始发生了 - -**测试结果**: -- ✅ 视频信息提取成功 -- ✅ 元数据完整获取（标题、作者、时长、描述） -- ✅ 内容深度分析完成 -- ✅ 笔记生成成功 -- ✅ Frontmatter 正确 - -**验证点**: -- 提取器正常工作：Bilibili Extractor 成功提取视频信息 -- 流程完整性：BV号 → 提取 → 分析 → 分块 → 输出 流程顺畅 -- 输出正确性：Obsidian 笔记格式正确，包含完整的 frontmatter 和内容 - ---- - -### 2. ✅ 降级流程测试（无字幕视频） - -**测试视频**: BV1MnFNz5EDL（已知无字幕） -**测试目的**: 验证无字幕视频的优雅降级处理 - -**测试结果**: -- ✅ 成功识别无字幕状态 -- ✅ 优雅降级，生成基础元数据笔记 -- ✅ 输出文件包含完整元数据 -- ✅ 无字幕提示正确显示 - -**验证点**: -- 无字幕检测：正确识别 `subtitle` 为空 -- 降级处理：Deep Analyzer 正确降级，使用 description 生成摘要 -- 元数据完整性：标题、作者、URL、标签等全部生成 -- 笔记格式：Markdown 格式正确，frontmatter 完整 - ---- - -### 3. ✅ 错误场景测试：无效BV号 - -**测试BV号**: BVinvalid123 - -**测试结果**: -- ✅ 正确捕获错误 -- ✅ 友好的错误提示 -- ✅ 程序不崩溃 - -**验证点**: -- 输入验证：正确拒绝无效BV号 -- 错误处理：ExtractionError 被正确捕获 -- 用户反馈：清晰的错误信息输出 -- 稳定性：程序继续运行，不中断 - ---- - -### 4. ✅ 错误场景测试：网络错误模拟 - -**测试BV号**: BV1xxxxxxxxxxxxx（不存在的BV号） - -**测试结果**: -- ✅ 程序未崩溃 -- ✅ 错误被妥善处理 -- ✅ 流程优雅降级 - -**验证点**: -- 异常处理：所有异常都被捕获和处理 -- 程序稳定性：即使网络错误，程序仍能继续 -- 用户体验：错误信息友好且清晰 - ---- - -## 修复的问题 - -在测试过程中发现并修复了以下问题： - -### 1. Playwright 导航超时问题 -**问题**: `Page.content: Unable to retrieve content because the page is navigating` - -**修复**: 移除 `wait_for_load_state('networkidle')`，改用更灵活的超时处理 - -**文件**: `src/extractors/bilibili_extractor.py` - -### 2. AnalysisResult 缺少属性 -**问题**: `AttributeError: 'AnalysisResult' object has no attribute 'full_analysis'` - -**修复**: 在 `AnalysisResult` 类中添加缺失的属性： -- `action_items` -- `summary` -- `timeline` -- `full_analysis` - -**文件**: `src/models.py` - -### 3. 测试脚本 Bug -**问题**: `all(results.values(), False)` 语法错误 - -**修复**: 改为 `all(results.values())` - -**文件**: `test_integration.py` - ---- - -## 生成的测试报告 - -**报告位置**: `/tmp/obsidian_kb_test_42b756hr/TEST_REPORT.md` - -**报告内容**: -- 测试概述（总测试数、通过数、通过率） -- 详细测试结果 -- 验证标准检查 -- 结论 - ---- - -## 生成的 Obsidian 笔记示例 - -**笔记路径**: `/tmp/obsidian_kb_test_42b756hr/vault/videos/bilibili/老戴Donald/2026-02-07_【老戴】用了几天 Clawdbot，我最担心的事开始发生了.md` - -**笔记内容特点**: -- ✅ 完整的 frontmatter（YAML元数据） -- ✅ 标题和摘要 -- ✅ 关键要点 -- ✅ 无字幕提示 -- ✅ 来源链接和时长 -- ✅ 处理日期 - -**示例内容**: -```yaml ---- -title: 【老戴】用了几天 Clawdbot，我最担心的事开始发生了 -source: bilibili -url: https://www.bilibili.com/video/BV1MnFNz5EDL -author: 老戴Donald -duration: 12:38 -date: 2026-02-07 -tags: ["video", "bilibili", "test", "未分类", "待处理", "作者:老戴Donald"] ---- -``` - ---- - -## 验证标准检查 - -### ✅ 有字幕视频：生成完整深度分析笔记 -- 通过：完整流程测试成功，笔记生成正确 - -### ✅ 无字幕视频：优雅降级生成基础笔记 -- 通过：降级流程测试成功，生成包含完整元数据的笔记，并标注无字幕 - -### ✅ 错误场景：有友好提示，不崩溃 -- 通过：无效BV号和模拟网络错误都被正确处理，程序稳定 - -### ✅ 测试报告：完整记录 -- 通过：生成了详细的测试报告，包含所有测试用例和结果 - ---- - -## 结论 - -🎉 **Checkpoint 7 集成测试成功完成！** - -obsidian-kb 整合项目的所有核心功能已正常工作： - -1. **视频信息提取（Bilibili Extractor）** - - 成功提取视频元数据、统计信息、描述 - - 支持无字幕视频的处理 - -2. **内容深度分析（Deep Analyzer）** - - 正确识别字幕状态 - - 优雅降级机制工作正常 - -3. **内容分块（Content Chunker）** - - 无字幕时正确返回空分片 - - 为后续字幕支持预留接口 - -4. **Obsidian 笔记生成（Obsidian Writer）** - - Frontmatter 生成正确 - - Markdown 格式符合 Obsidian 要求 - - 无字幕提示正确显示 - -5. **错误处理和优雅降级** - - 友好的错误提示 - - 程序稳定性良好 - - 异常处理完善 - ---- - -## 下一步建议 - -1. **寻找有字幕视频进行更完整测试** - - 当前使用无字幕视频进行测试 - - 建议测试实际有字幕的视频，验证完整字幕解析流程 - -2. **增强日志记录** - - 增加更详细的日志，便于调试 - - 记录字幕解析的具体过程 - -3. **性能优化** - - 考虑缓存提取结果 - - 优化 Playwright 启动时间 - -4. **扩展测试覆盖** - - 测试 YouTube 视频提取 - - 测试知乎文章处理 - - 测试长视频分段处理 - ---- - -## 附录：测试环境信息 - -- **Python 版本**: 3.12 -- **工作目录**: `/home/admin/.openclaw/workspace/skills/obsidian-kb` -- **虚拟环境**: `.venv` -- **测试日期**: 2026-02-07 -- **测试时间**: 13:04:28 (GMT+8) - ---- - -**报告生成**: Obsidian KB 集成测试脚本 -**报告版本**: 1.0 diff --git a/skills/obsidian-kb/COMPLETION_REPORT.md b/skills/obsidian-kb/COMPLETION_REPORT.md deleted file mode 100644 index 65c4758..0000000 --- a/skills/obsidian-kb/COMPLETION_REPORT.md +++ /dev/null @@ -1,310 +0,0 @@ -# ✅ Checkpoint 6: 主流程集成 - 完成报告 - -## 任务概述 - -完成 obsidian-kb 整合项目的 Checkpoint 6：主流程集成 - -**任务清单**： -1. ✅ 修改主入口或 handler.py，集成提取层、处理层、输出层 -2. ✅ 添加错误处理 -3. ✅ 添加日志记录 -4. ✅ 创建测试脚本 - ---- - -## 完成内容 - -### 1. 主流程集成文件 - main_pipeline.py - -**文件**: `main_pipeline.py` (16,051 字节) - -**核心类**: -```python -class MainPipeline: - """主流程集成器""" -``` - -**完整数据流**: -``` -提取(BilibiliExtractor) → 处理(DeepAnalyzer/ContentChunker) → 输出(ObsidianWriter) → 同步(GitSync) -``` - -**关键方法**: -- `process_video(source_id)` - 主入口，处理单个视频 -- `_extract_video(source_id)` - 提取层集成 -- `_analyze_content(extractor_data)` - 处理层-分析器集成 -- `_chunk_content(extractor_data)` - 处理层-分片器集成 -- `_merge_analysis_to_content(content, analysis, chunks)` - 合并结果 -- `_write_obsidian_note(content)` - 输出层集成 -- `_sync_to_git(content)` - 同步层集成 - -### 2. 错误处理 - -**实现位置**: `main_pipeline.py` - -**错误处理策略**: -- **提取失败**: 捕获 `ExtractionError`，中断流程，返回友好错误 -- **分析失败**: 返回默认结果，继续流程 -- **分片失败**: 返回空列表，继续流程 -- **写入失败**: 返回 None，记录警告 -- **Git同步失败**: 记录错误，不影响主流程 - -**代码示例**: -```python -try: - extractor_data = self.extractor.extract(source_id) -except ExtractionError as e: - logger.error(f"❌ 提取失败: {e}") - raise -``` - -### 3. 日志记录 - -**日志配置**: -- **级别**: INFO -- **格式**: `%(asctime)s - %(name)s - %(levelname)s - %(message)s` -- **输出**: 控制台 + 文件 (`obsidian_kb.log`) - -**记录内容**: -- 每个步骤的开始/成功/失败 -- 视频信息（标题、作者、时长） -- 字幕条数、分片数量 -- 输出文件路径 -- 错误堆栈信息 - -### 4. 测试脚本 - -**文件**: `test_integration.py` (8,831 字节) - -**测试用例**: -1. `test_extract_only()` - 提取功能测试 -2. `test_full_pipeline()` - 完整流程测试 -3. `test_error_handling()` - 错误处理测试 -4. `cleanup_test_env()` - 测试环境清理 - -**特性**: -- 使用临时目录，不影响生产环境 -- 不执行 Git 同步 -- 详细的测试输出 -- 自动清理测试环境 - -### 5. Handler 集成 - -**修改文件**: `handler.py` - -**新增内容**: -1. 主流程导入: - ```python - from main_pipeline import MainPipeline - ``` - -2. 新增方法 `process_video_full(source_id, enable_git_sync)`: - - 提供新的处理视频接口 - - 支持启用/禁用 Git 同步 - -3. 更新 CLI 接口: - - `detect` - 检测链接 - - `process ` - 处理视频 - - `process --git` - 处理视频并同步到 Git - -### 6. 验证脚本 - -**文件**: `verify_integration.py` (7,203 字节) - -**验证内容**: -- 文件存在性 -- 主流程导入 -- 主流程方法 -- 测试脚本导入 -- 测试脚本方法 -- Handler 集成 -- 错误处理 -- 日志配置 - -**优势**: 不需要安装依赖即可运行 - ---- - -## 验证结果 - -运行 `python3 verify_integration.py`: - -``` -✅ 文件存在性 通过 -✅ 主流程导入 通过 -✅ 主流程方法 通过 -✅ 测试脚本导入 通过 -✅ 测试脚本方法 通过 -✅ Handler 集成 通过 -✅ 错误处理 通过 -✅ 日志配置 通过 - -总体结果: ✅ 全部通过 -``` - ---- - -## 验证标准完成情况 - -### ✅ 完整数据流打通（BV号 → Obsidian笔记） -- 主流程 `process_video()` 实现了完整流程 -- 各层通过标准接口连接 -- 提取 → 处理 → 输出 → 同步 全部集成 - -### ✅ 错误处理完善 -- 捕获 `ExtractionError` 并返回友好提示 -- 各层都有 try-except 保护 -- 异常不会导致程序崩溃 -- 分级错误处理策略 - -### ✅ 日志记录完整 -- 记录所有关键步骤（提取、分析、输出、同步） -- 记录错误详细信息到日志文件 -- 同时输出到控制台和文件 -- 详细的信息记录 - -### ✅ 测试脚本能运行 -- `test_integration.py` 提供完整测试 -- `verify_integration.py` 提供代码结构验证 -- 所有验证通过 - ---- - -## 新增/修改文件列表 - -### 新增文件 (6 个) -1. `main_pipeline.py` - 主流程集成文件 (16KB) -2. `test_integration.py` - 集成测试脚本 (9KB) -3. `verify_integration.py` - 验证脚本 (7KB) -4. `example_usage.py` - 使用示例 (5KB) -5. `CHECKPOINT_6.md` - 完成报告 (8KB) -6. `QUICKSTART.md` - 快速开始指南 (3KB) - -### 修改文件 (1 个) -1. `handler.py` - 添加主流程集成和新接口 - -### 文档文件 (2 个) -1. `INTEGRATION_SUMMARY.md` - 集成总结 (7KB) -2. `COMPLETION_REPORT.md` - 本报告 - ---- - -## 使用示例 - -### 方式1: 主流程接口 - -```python -from main_pipeline import MainPipeline -from src.config import get_config - -config = get_config() -pipeline = MainPipeline(config=config, enable_git_sync=False) -result = pipeline.process_video("BV1MnFNz5EDL") - -if result['success']: - print(result['message']) -else: - print(f"失败: {result['error']}") -``` - -### 方式2: Handler 接口 - -```python -from handler import process_video_full - -success, message = process_video_full("BV1MnFNz5EDL", enable_git_sync=False) - -if success: - print(message) -else: - print(f"失败: {message}") -``` - -### 方式3: 命令行 - -```bash -# 使用主流程 -python3 -m main_pipeline BV1MnFNz5EDL - -# 使用 Handler -python3 handler.py process BV1MnFNz5EDL - -# 处理并同步到 Git -python3 handler.py process BV1MnFNz5EDL --git -``` - ---- - -## 数据流图 - -``` -输入: BV号/URL - │ - ├─► [Step 1] 提取层: BilibiliExtractor.extract() - │ ├─ metadata (标题, 作者, 时长, etc.) - │ ├─ subtitle (字幕文本) - │ └─ description (简介) - │ - ├─► [Step 2] 构建对象: _build_content() - │ └─ Content 对象 - │ - ├─► [Step 3] 处理层-分析: DeepAnalyzer.analyze() - │ └─ AnalysisResult (主题, 洞察, 时间轴, etc.) - │ - ├─► [Step 4] 处理层-分片: ContentChunker.chunk_from_extractor() - │ └─ Chunk[] (分段内容) - │ - ├─► [Step 5] 合并结果: _merge_analysis_to_content() - │ └─ 更新 Content 对象 - │ - ├─► [Step 6] 输出层: ObsidianWriter.write() - │ └─ Markdown 文件 - │ - └─► [Step 7] 同步层: GitSync.sync() [可选] - └─ Git 提交和推送 - -输出: Obsidian 笔记文件 -``` - ---- - -## 文档 - -- **完成报告**: `CHECKPOINT_6.md` -- **快速开始**: `QUICKSTART.md` -- **使用示例**: `example_usage.py` -- **集成总结**: `INTEGRATION_SUMMARY.md` -- **本报告**: `COMPLETION_REPORT.md` - ---- - -## 运行测试 - -```bash -# 验证代码结构（不需要依赖） -python3 verify_integration.py - -# 运行完整测试（需要所有依赖） -python3 test_integration.py -``` - ---- - -## 总结 - -Checkpoint 6 主流程集成任务已全部完成： - -1. ✅ 修改/创建主入口，集成提取层、处理层、输出层 -2. ✅ 添加错误处理，各层都有 try-except 保护 -3. ✅ 添加日志记录，关键步骤和错误信息都记录 -4. ✅ 创建测试脚本，支持完整流程测试 -5. ✅ 创建验证脚本，支持代码结构验证 -6. ✅ 完善文档，包括完成报告、快速开始、使用示例 - -**完整的数据流已打通，代码结构通过验证，文档齐全，可以开始使用。** - ---- - -**生成时间**: 2026-02-07 12:49 -**验证状态**: ✅ 全部通过 -**任务状态**: ✅ 完成 diff --git a/skills/obsidian-kb/INTEGRATION_SUMMARY.md b/skills/obsidian-kb/INTEGRATION_SUMMARY.md deleted file mode 100644 index 0188527..0000000 --- a/skills/obsidian-kb/INTEGRATION_SUMMARY.md +++ /dev/null @@ -1,272 +0,0 @@ -# Checkpoint 6 集成总结 - -## 新增文件 - -### 1. main_pipeline.py (16,051 字节) -主流程集成文件，包含： -- `MainPipeline` 类：核心主流程集成器 -- 完整的数据流实现：提取 → 处理 → 输出 → 同步 -- 完善的错误处理和日志记录 - -**关键方法**： -- `process_video(source_id)` - 主入口方法 -- `_extract_video()` - 提取层集成 -- `_analyze_content()` - 处理层-分析器集成 -- `_chunk_content()` - 处理层-分片器集成 -- `_write_obsidian_note()` - 输出层集成 -- `_sync_to_git()` - 同步层集成 - -### 2. test_integration.py (8,831 字节) -完整测试脚本，包含： -- `test_extract_only()` - 提取功能测试 -- `test_full_pipeline()` - 完整流程测试 -- `test_error_handling()` - 错误处理测试 -- `cleanup_test_env()` - 测试环境清理 - -**特性**： -- 使用临时目录，不影响生产环境 -- 不执行 Git 同步 -- 详细的测试输出 -- 自动清理测试环境 - -### 3. verify_integration.py (7,203 字节) -代码结构验证脚本，验证： -- 文件存在性 -- 主流程导入 -- 主流程方法 -- 测试脚本导入 -- 测试脚本方法 -- Handler 集成 -- 错误处理 -- 日志配置 - -**优势**：不需要安装依赖即可运行 - -### 4. example_usage.py (5,082 字节) -使用示例脚本，包含： -- 示例1: 基本使用 -- 示例2: 启用 Git 同步 -- 示例3: 使用 Handler 接口 -- 示例4: 批量处理 -- 示例5: 自定义配置 -- 示例6: 错误处理 - -### 5. CHECKPOINT_6.md (8,283 字节) -Checkpoint 6 完成报告，包含： -- 任务概述 -- 完成内容详解 -- 验证结果 -- 使用示例 -- 数据流图 -- 关键修改说明 - -### 6. QUICKSTART.md (2,379 字节) -快速开始指南，包含： -- 安装依赖 -- 配置说明 -- 使用方式（命令行/Python/Handler） -- 测试方法 -- 常见问题 - -## 修改文件 - -### handler.py -**修改内容**： -1. 添加主流程导入： - ```python - from main_pipeline import MainPipeline - ``` - -2. 新增 `process_video_full()` 方法： - - 提供新的处理视频接口 - - 支持启用/禁用 Git 同步 - - 返回 (success, message) 元组 - -3. 更新 `main()` CLI 接口： - - 支持 `detect` 命令 - - 支持 `process` 命令 - - 支持 `--git` 选项 - -## 验证结果 - -运行 `python3 verify_integration.py`： - -``` -✅ 文件存在性 通过 -✅ 主流程导入 通过 -✅ 主流程方法 通过 -✅ 测试脚本导入 通过 -✅ 测试脚本方法 通过 -✅ Handler 集成 通过 -✅ 错误处理 通过 -✅ 日志配置 通过 - -总体结果: ✅ 全部通过 -``` - -## 验证标准完成情况 - -### ✅ 完整数据流打通（BV号 → Obsidian笔记） -- 主流程 `process_video()` 实现了完整流程 -- 各层通过标准接口连接 -- 提取 → 处理 → 输出 → 同步 全部集成 - -### ✅ 错误处理完善 -- 捕获 `ExtractionError` 并返回友好提示 -- 各层都有 try-except 保护 -- 异常不会导致程序崩溃 -- 分级错误处理：提取失败中断，其他失败继续 - -### ✅ 日志记录完整 -- 记录所有关键步骤（提取、分析、输出、同步） -- 记录错误详细信息到日志文件（obsidian_kb.log） -- 同时输出到控制台和文件 -- 详细的信息记录（标题、作者、时长、文件路径等） - -### ✅ 测试脚本能运行 -- `test_integration.py` 提供完整测试 -- `verify_integration.py` 提供代码结构验证 -- 所有验证通过 -- 可以在无依赖环境下运行验证 - -## 集成的层 - -### 1. 提取层 -- **类**: `BilibiliExtractor` -- **方法**: `extract(source_id)` -- **功能**: 提取B站视频信息（元数据、字幕、统计） -- **位置**: `src/extractors/bilibili_extractor.py` - -### 2. 处理层 - 分析器 -- **类**: `DeepAnalyzer` -- **方法**: `analyze(extractor_data)` -- **功能**: 深度分析内容（主题、洞察、时间轴） -- **位置**: `src/deep_analyzer.py` - -### 3. 处理层 - 分片器 -- **类**: `ContentChunker` -- **方法**: `chunk_from_extractor(extractor_data)` -- **功能**: 将内容按时间分段 -- **位置**: `src/content_chunker.py` - -### 4. 输出层 - 写入器 -- **类**: `ObsidianWriter` -- **方法**: `write(content, config)` -- **功能**: 生成并写入 Obsidian Markdown 笔记 -- **位置**: `src/outputs/obsidian_writer.py` - -### 5. 输出层 - 同步器 -- **类**: `GitSync` -- **方法**: `sync(config)` -- **功能**: 将笔记同步到 Git 仓库 -- **位置**: `src/outputs/git_sync.py` - -## 使用方式 - -### 1. 主流程接口 - -```python -from main_pipeline import MainPipeline -from src.config import get_config - -config = get_config() -pipeline = MainPipeline(config=config, enable_git_sync=False) -result = pipeline.process_video("BV1MnFNz5EDL") -``` - -### 2. Handler 接口 - -```python -from handler import process_video_full - -success, message = process_video_full("BV1MnFNz5EDL", enable_git_sync=False) -``` - -### 3. 命令行 - -```bash -python3 -m main_pipeline BV1MnFNz5EDL -python3 handler.py process BV1MnFNz5EDL --git -``` - -## 日志系统 - -日志配置： -- **级别**: INFO -- **格式**: `%(asctime)s - %(name)s - %(levelname)s - %(message)s` -- **输出**: 控制台 + 文件 -- **文件**: `obsidian_kb.log` - -记录内容： -- 提取开始/成功/失败 -- 分析开始/成功/失败 -- 分片开始/成功/失败 -- 写入开始/成功/失败 -- Git同步开始/成功/失败 -- 错误堆栈信息 - -## 错误处理策略 - -### 提取失败 -- 行为：中断流程，返回错误 -- 日志：ERROR 级别 -- 用户提示：友好错误信息 - -### 分析失败 -- 行为：返回默认结果，继续流程 -- 日志：ERROR 级别 -- 用户提示：警告信息 - -### 分片失败 -- 行为：返回空列表，继续流程 -- 日志：ERROR 级别 -- 用户提示：警告信息 - -### 写入失败 -- 行为：返回 None，记录警告 -- 日志：ERROR 级别 -- 用户提示：警告信息 - -### Git同步失败 -- 行为：记录错误，不影响主流程 -- 日志：ERROR 级别 -- 用户提示：警告信息 - -## 测试覆盖 - -### 单元测试 -- 提取功能测试 -- 分析功能测试 -- 分片功能测试 -- 写入功能测试 -- 错误处理测试 - -### 集成测试 -- 完整流程测试 -- 提取 → 处理 → 输出 → 同步 -- 批量处理测试 - -### 验证测试 -- 代码结构验证 -- 导入验证 -- 方法验证 - -## 文档 - -- **完成报告**: `CHECKPOINT_6.md` -- **快速开始**: `QUICKSTART.md` -- **使用示例**: `example_usage.py` -- **集成总结**: `INTEGRATION_SUMMARY.md` (本文件) - -## 总结 - -Checkpoint 6 主流程集成任务已全部完成： - -1. ✅ 创建/修改主入口，集成提取层、处理层、输出层 -2. ✅ 添加错误处理，各层都有 try-except 保护 -3. ✅ 添加日志记录，关键步骤和错误信息都记录 -4. ✅ 创建测试脚本，支持完整流程测试 -5. ✅ 创建验证脚本，支持代码结构验证 -6. ✅ 完善文档，包括完成报告、快速开始、使用示例 - -完整的数据流已打通，代码结构通过验证，文档齐全，可以开始使用。 diff --git a/skills/obsidian-kb/test_bilibili_extractor.py b/skills/obsidian-kb/test_bilibili_extractor.py deleted file mode 100644 index 463600a..0000000 --- a/skills/obsidian-kb/test_bilibili_extractor.py +++ /dev/null @@ -1,80 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -B站提取器测试脚本 -""" -import sys -import os - -# 添加父目录到路径 -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from src.extractors.bilibili_extractor import BilibiliExtractor -from src.config import get_config - - -def main(): - """测试提取器""" - # 获取配置 - config = get_config() - - # 创建提取器 - extractor = BilibiliExtractor(config=config) - - # 测试BV号 - bvid = "BV1MnFNz5EDL" - - print(f"开始提取视频: {bvid}") - print("=" * 60) - - try: - result = extractor.extract(bvid) - - print("\n提取成功！\n") - print("=" * 60) - print("元数据 (metadata):") - print("=" * 60) - for key, value in result['metadata'].items(): - print(f" {key}: {value}") - - print("\n" + "=" * 60) - print("统计数据 (stats):") - print("=" * 60) - for key, value in result['stats'].items(): - print(f" {key}: {value:,}") - - print("\n" + "=" * 60) - print("简介 (description):") - print("=" * 60) - desc = result['description'] - if len(desc) > 500: - print(desc[:500] + "...") - else: - print(desc) - - print("\n" + "=" * 60) - print("字幕 (subtitle):") - print("=" * 60) - subtitle = result['subtitle'] - if subtitle: - lines = subtitle.split('\n') - if len(lines) > 10: - print("\n".join(lines[:10])) - print(f"\n...(共 {len(lines)} 条字幕)") - else: - print(subtitle) - else: - print(" (无字幕)") - - print("\n" + "=" * 60) - print("✅ 测试完成") - print("=" * 60) - - except Exception as e: - print(f"\n❌ 测试失败: {e}") - import traceback - traceback.print_exc() - - -if __name__ == '__main__': - main() diff --git a/skills/obsidian-kb/test_bilibili_headless.py b/skills/obsidian-kb/test_bilibili_headless.py deleted file mode 100644 index d8f7455..0000000 --- a/skills/obsidian-kb/test_bilibili_headless.py +++ /dev/null @@ -1,84 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -B站提取器测试脚本（非无头模式） -""" -import sys -import os - -# 添加父目录到路径 -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from src.extractors.bilibili_extractor import BilibiliExtractor -from src.config import get_config - - -def main(): - """测试提取器""" - # 获取配置 - config = get_config() - - # 创建提取器 - extractor = BilibiliExtractor(config=config) - - # 强制非无头模式 - extractor.headless = False - - # 测试BV号 - bvid = "BV1MnFNz5EDL" - - print(f"开始提取视频: {bvid}") - print(f"模式: {'非无头' if not extractor.headless else '无头'}") - print("=" * 60) - - try: - result = extractor.extract(bvid) - - print("\n提取成功！\n") - print("=" * 60) - print("元数据 (metadata):") - print("=" * 60) - for key, value in result['metadata'].items(): - print(f" {key}: {value}") - - print("\n" + "=" * 60) - print("统计数据 (stats):") - print("=" * 60) - for key, value in result['stats'].items(): - print(f" {key}: {value:,}") - - print("\n" + "=" * 60) - print("简介 (description):") - print("=" * 60) - desc = result['description'] - if len(desc) > 500: - print(desc[:500] + "...") - else: - print(desc) - - print("\n" + "=" * 60) - print("字幕 (subtitle):") - print("=" * 60) - subtitle = result['subtitle'] - if subtitle: - lines = subtitle.split('\n') - if len(lines) > 10: - print("\n".join(lines[:10])) - print(f"\n...(共 {len(lines)} 条字幕)") - else: - print(subtitle) - else: - print(" (无字幕)") - - print("\n" + "=" * 60) - print("✅ 测试完成") - print("=" * 60) - - except Exception as e: - print(f"\n❌ 测试失败: {e}") - import traceback - traceback.print_exc() - - -if __name__ == '__main__': - main() diff --git a/skills/obsidian-kb/test_checkpoint_4_5.py b/skills/obsidian-kb/test_checkpoint_4_5.py deleted file mode 100644 index 37b30dc..0000000 --- a/skills/obsidian-kb/test_checkpoint_4_5.py +++ /dev/null @@ -1,315 +0,0 @@ -"""Obsidian KB - 测试验证 -验证Checkpoint 4-5的所有功能 -""" -import sys -import os -sys.path.insert(0, os.path.dirname(__file__)) - -from src.models import Content, ContentType, ContentStatus, TimelineEntry -from src.light_scanner import LightScanner -from src.deep_analyzer import DeepAnalyzer -from src.pipeline import ProcessingPipeline -from src.content_chunker import ContentChunker, get_chunking_strategy, chunk_content, Chunk -from src.ai_client import AIClient - - -def test_light_scanner(): - """测试C4.1: 轻量扫描""" - print("\n" + "="*60) - print("🧪 测试C4.1: 轻量扫描") - print("="*60) - - # 创建测试内容 - test_content = Content( - id="test-001", - url="https://example.com/video1", - title="测试视频：人工智能的未来发展", - content_type=ContentType.VIDEO, - source="youtube", - duration=600 # 10分钟 - ) - - # 测试文本 - test_text = """ - 大家好，今天我们来讨论人工智能的未来发展。 - 人工智能正在改变我们的生活和工作方式。 - 机器学习是人工智能的核心技术之一。 - 深度学习让神经网络能够处理更复杂的任务。 - 自然语言处理技术让机器能够理解人类语言。 - 计算机视觉让机器能够看懂图像和视频。 - 强化学习让AI能够通过试错来学习。 - 未来，AI将在医疗、教育、交通等领域发挥重要作用。 - 我们需要关注AI伦理和安全问题。 - 让我们共同迎接AI时代的到来。 - """ * 10 # 重复以增加文本长度 - - # 执行轻量扫描 - scanner = LightScanner() - result = scanner.scan(test_content, test_text) - - # 验证结果 - assert len(result.summary) > 0, "❌ 摘要为空" - assert len(result.key_points) == 5, f"❌ 关键词数量错误: {len(result.key_points)}" - assert 0 <= result.quality_metrics.overall_score <= 1, f"❌ 质量评分越界: {result.quality_metrics.overall_score}" - assert result.status == ContentStatus.SCANNED, "❌ 状态未更新" - - print(f"✅ 摘要: {result.summary}") - print(f"✅ 关键词: {result.key_points}") - print(f"✅ 质量评分: {result.quality_metrics.overall_score:.2f}/1.0") - print("✅ C4.1 轻量扫描测试通过!") - - return result - - -def test_deep_analyzer(): - """测试C4.2: 深度分析""" - print("\n" + "="*60) - print("🧪 测试C4.2: 深度分析") - print("="*60) - - ai_client = AIClient() - analyzer = DeepAnalyzer(ai_client=ai_client) - - # 测试短视频（<10分钟） - print("\n📱 测试短视频分析...") - short_content = Content( - id="test-short", - url="https://example.com/short", - title="短视频测试", - content_type=ContentType.VIDEO, - duration=300 # 5分钟 - ) - short_text = "这是一个短视频的内容文本。" * 100 - - result_short = analyzer.analyze(short_content, short_text) - assert len(result_short.timeline) > 0, "❌ 短视频时间轴为空" - print(f"✅ 短视频分析完成，时间轴段数: {len(result_short.timeline)}") - - # 测试长视频（>=10分钟） - print("\n🎬 测试长视频分析...") - long_content = Content( - id="test-long", - url="https://example.com/long", - title="长视频测试", - content_type=ContentType.VIDEO, - duration=3600 # 60分钟 - ) - # 创建长视频SRT字幕（60分钟） - long_text = "" - for i in range(120): # 120段，每段30秒 = 60分钟 - start_sec = i * 30 - end_sec = (i + 1) * 30 - start_time = f"00:{start_sec//60:02d}:{start_sec%60:02d},000" - end_time = f"00:{end_sec//60:02d}:{end_sec%60:02d},000" - long_text += f"{i+1}\n{start_time} --> {end_time}\n这是第{i+1}段讨论内容关于人工智能和机器学习的应用场景\n\n" - - result_long = analyzer.analyze(long_content, long_text) - assert len(result_long.timeline) > 0, "❌ 长视频时间轴为空" - assert len(result_long.main_topics) > 0, "❌ 长视频主题为空" - print(f"✅ 长视频分析完成，时间轴段数: {len(result_long.timeline)}") - print(f"✅ 主要主题: {result_long.main_topics}") - - print("✅ C4.2 深度分析测试通过!") - - return result_short, result_long - - -def test_pipeline(): - """测试C4.3: 处理管道""" - print("\n" + "="*60) - print("🧪 测试C4.3: 处理管道") - print("="*60) - - ai_client = AIClient() - pipeline = ProcessingPipeline(ai_client=ai_client) - - # 创建测试内容 - content = Content( - id="test-pipeline", - url="https://example.com/pipeline", - title="管道测试视频", - content_type=ContentType.VIDEO, - source="bilibili", - duration=600, - author="测试作者" - ) - - test_text = """ - 这是用于测试处理管道的内容。 - 包含多个主题：编程、AI、数据分析。 - 让我们看看管道能否正常工作。 - """ * 50 - - # 执行完整流程 - result = pipeline.process_full(content, test_text) - - # 验证结果 - assert result.status == ContentStatus.GENERATED, f"❌ 最终状态错误: {result.status}" - assert len(result.summary) > 0, "❌ 摘要为空" - assert len(result.key_points) == 5, "❌ 关键词数量错误" - assert result.output_path and len(result.output_path) > 0, "❌ 笔记路径为空" - - print(f"✅ 处理完成，最终状态: {result.status.value}") - print(f"✅ 笔记路径: {result.output_path}") - print("✅ C4.3 处理管道测试通过!") - - return result - - -def test_content_chunker(): - """测试C5.1: 内容分片""" - print("\n" + "="*60) - print("🧪 测试C5.1: 内容分片器") - print("="*60) - - # 创建测试SRT字幕 - srt_text = "" - for i in range(100): - start_min = i * 0.5 - end_min = (i + 1) * 0.5 - start_time = f"00:{int(start_min):02d}:00,000" - end_time = f"00:{int(end_min):02d}:00,000" - srt_text += f"{i+1}\n{start_time} --> {end_time}\n这是第{i+1}段字幕内容\n\n" - - chunker = ContentChunker() - - # 测试按时长分片（15分钟） - print("\n⏱️ 测试按时长分片（15分钟）...") - chunks_by_duration = chunker.chunk_by_duration(srt_text, target_duration=900) - assert len(chunks_by_duration) > 0, "❌ 按时长分片失败" - print(f"✅ 按时长分片: {len(chunks_by_duration)} 段") - for chunk in chunks_by_duration[:3]: # 只显示前3段 - print(f" 段{chunk.index}: {chunk.time_range}, {len(chunk.text)}字符") - - # 测试按段数分片 - print("\n🔢 测试按段数分片（4段）...") - chunks_by_count = chunker.chunk_by_count(srt_text, num_chunks=4) - assert len(chunks_by_count) > 0, "❌ 按段数分片失败" - print(f"✅ 按段数分片: {len(chunks_by_count)} 段") - - print("✅ C5.1 内容分片器测试通过!") - - return chunks_by_duration, chunks_by_count - - -def test_chunking_strategy(): - """测试C5.2: 分段策略""" - print("\n" + "="*60) - print("🧪 测试C5.2: 分段策略") - print("="*60) - - test_cases = [ - (300, 'direct', False), # 5分钟 - 直接处理 - (900, 'truncate', False), # 15分钟 - 截断 - (1800, 'chunk_2', True), # 30分钟 - 2段 - (3600, 'chunk_n', True), # 60分钟 - 多段 - (7200, 'chunk_n', True), # 120分钟 - 多段 - ] - - for duration, expected_strategy, expected_chunking in test_cases: - strategy = get_chunking_strategy(duration) - - mins = duration // 60 - print(f"\n⏱️ 测试 {mins}分钟视频:") - print(f" 策略: {strategy['strategy']} (预期: {expected_strategy})") - print(f" 是否分片: {strategy['use_chunking']} (预期: {expected_chunking})") - - assert strategy['strategy'] == expected_strategy, f"❌ 策略不匹配" - assert strategy['use_chunking'] == expected_chunking, f"❌ 分片标志不匹配" - - print("\n✅ C5.2 分段策略测试通过!") - - -def test_ai_chunk_analysis(): - """测试C5.3: AI分段分析""" - print("\n" + "="*60) - print("🧪 测试C5.3: AI分段分析") - print("="*60) - - ai_client = AIClient() - - # 创建长视频SRT（45分钟） - srt_text = "" - for i in range(90): # 90段，每段30秒 = 45分钟 - start_sec = i * 30 - end_sec = (i + 1) * 30 - start_time = f"00:{start_sec//60:02d}:{start_sec%60:02d},000" - end_time = f"00:{end_sec//60:02d}:{end_sec%60:02d},000" - srt_text += f"{i+1}\n{start_time} --> {end_time}\n第{i+1}段讨论内容关于人工智能和机器学习的应用场景\n\n" - - duration = 2700 # 45分钟 - - print(f"\n🎬 测试长视频（{duration//60}分钟）分段分析...") - - # 执行分段分析 - result = ai_client._deep_analyze( - content_text=srt_text, - content_type="video", - duration=duration - ) - - # 验证结果 - assert 'summary' in result, "❌ 缺少整体摘要" - assert 'topics' in result, "❌ 缺少主题列表" - assert 'timeline' in result, "❌ 缺少时间轴" - assert len(result['timeline']) > 0, "❌ 时间轴为空" - - print(f"✅ 分析完成") - print(f"✅ 整体摘要: {result['summary'][:80]}...") - print(f"✅ 主题数: {len(result['topics'])}") - print(f"✅ 时间轴段数: {len(result['timeline'])}") - - print("\n✅ C5.3 AI分段分析测试通过!") - - return result - - -def run_all_tests(): - """运行所有测试""" - print("\n" + "🚀"*30) - print("开始执行 Obsidian KB Checkpoint 4-5 测试") - print("🚀"*30) - - results = {} - - try: - # C4.1 轻量扫描 - results['light_scan'] = test_light_scanner() - - # C4.2 深度分析 - results['deep_analyze_short'], results['deep_analyze_long'] = test_deep_analyzer() - - # C4.3 处理管道 - results['pipeline'] = test_pipeline() - - # C5.1 内容分片 - results['chunks_duration'], results['chunks_count'] = test_content_chunker() - - # C5.2 分段策略 - test_chunking_strategy() - - # C5.3 AI分段分析 - results['ai_chunk_analysis'] = test_ai_chunk_analysis() - - # 总结 - print("\n" + "="*60) - print("✅ 所有测试通过!") - print("="*60) - print("\n📋 验收标准检查:") - print(" [✅] 轻量扫描能生成摘要和评分") - print(" [✅] 深度分析能处理长短视频") - print(" [✅] 长视频(30分钟+)自动分段不报错") - print(" [✅] content_chunker 能正确分割SRT字幕") - - return True - - except Exception as e: - print(f"\n❌ 测试失败: {e}") - import traceback - traceback.print_exc() - return False - - -if __name__ == "__main__": - success = run_all_tests() - sys.exit(0 if success else 1) diff --git a/skills/obsidian-kb/test_degraded_pipeline.py b/skills/obsidian-kb/test_degraded_pipeline.py deleted file mode 100644 index b3f9ffb..0000000 --- a/skills/obsidian-kb/test_degraded_pipeline.py +++ /dev/null @@ -1,236 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -Obsidian KB - 降级流程测试脚本 - -测试无字幕视频的处理：优雅降级，生成基础元数据笔记 -""" - -import os -import sys -import tempfile -import shutil -from pathlib import Path - -# 添加项目路径 -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from main_pipeline import MainPipeline - - -def create_test_config(): - """创建测试配置""" - temp_dir = tempfile.mkdtemp(prefix='obsidian_kb_degraded_') - vault_path = Path(temp_dir) / 'vault' - vault_path.mkdir(parents=True, exist_ok=True) - - storage_dir = Path(temp_dir) / 'storage' - storage_dir.mkdir(parents=True, exist_ok=True) - - print(f"📁 测试目录: {temp_dir}") - print(f"📁 测试Vault: {vault_path}") - - test_config_dict = { - 'ai': { - 'provider': 'openai', - 'model': 'gpt-4', - 'api_key': None, - }, - 'storage': { - 'inbox_dir': 'inbox', - 'topics_dir': 'topics', - 'data_dir': str(storage_dir), - }, - 'git': { - 'enabled': False, - 'vault_path': str(vault_path), - 'remote': 'origin', - 'branch': 'main', - 'auto_push': False, - 'commit_message_template': 'Add note: {title}', - }, - 'obsidian': { - 'vault_path': str(vault_path), - 'notes_folder': 'Notes', - }, - 'output': { - 'organize_by': 'author', - 'filename_format': '{date}_{title}.md', - 'auto_tags': True, - 'default_tags': ['video', 'bilibili', 'test'], - }, - 'sources': { - 'bilibili': { - 'enabled': True, - 'cookie': None, - 'extraction': { - 'headless': True, - 'timeout': 30, - }, - }, - }, - 'temp_dir': temp_dir, - } - - class MockConfig: - def __init__(self, data): - self._data = data - - def get(self, key, default=None): - keys = key.split('.') - value = self._data - for k in keys: - if isinstance(value, dict) and k in value: - value = value[k] - else: - return default - return value - - @property - def vault_path(self): - return Path(self._data['git']['vault_path']) - - return MockConfig(test_config_dict) - - -def test_no_subtitle_video(): - """测试无字幕视频的降级处理""" - print("\n" + "=" * 80) - print("测试：无字幕视频降级流程") - print("=" * 80 + "\n") - - config = create_test_config() - pipeline = MainPipeline(config=config, enable_git_sync=False) - - # 使用已知无字幕的视频 - test_bv = "BV1MnFNz5EDL" - print(f"测试视频: {test_bv} (已知无字幕)\n") - - try: - result = pipeline.process_video(test_bv) - - if result['success']: - content = result['content'] - - print("\n" + "✅" * 40) - print("降级流程测试通过！") - print("✅" * 40 + "\n") - - print(f"📺 标题: {content.title}") - print(f"👤 作者: {content.author}") - print(f"📄 输出路径: {content.output_path or 'N/A'}") - print(f"⏱️ 时长: {content.duration}秒" if content.duration else "⏱️ 时长: N/A") - - # 检查是否有基础元数据 - print(f"\n📊 元数据完整性:") - print(f" 标题: {'✅' if content.title else '❌'}") - print(f" 作者: {'✅' if content.author else '❌'}") - print(f" 描述: {'✅' if content.description else '❌'}") - print(f" URL: {'✅' if content.url else '❌'}") - print(f" 标签: {'✅' if content.tags else '❌'}") - - # 检查输出文件 - if content.output_path: - output_file = Path(content.output_path) - if output_file.exists(): - print(f"\n📄 输出文件:") - print(f" 路径: {output_file}") - print(f" 大小: {output_file.stat().st_size:,} 字节") - - # 读取文件内容，检查是否有"无字幕"提示 - with open(output_file, 'r', encoding='utf-8') as f: - file_content = f.read() - - print(f"\n📝 笔记内容检查:") - print(f" frontmatter: {'✅' if '---' in file_content else '❌'}") - print(f" 无字幕提示: {'✅' if '无字幕' in file_content else '❌'}") - - # 显示部分内容 - print(f"\n📄 文件预览 (前1000字符):") - print("-" * 80) - print(file_content[:1000]) - print("-" * 80) - - if len(file_content) > 1000: - print(f"\n...(总字符数: {len(file_content)})") - else: - print(f"\n⚠️ 输出文件不存在: {output_file}") - - return True - else: - print(f"\n❌ 降级流程测试失败: {result.get('error', 'Unknown error')}") - return False - - except Exception as e: - print(f"\n❌ 降级流程测试异常: {e}") - import traceback - traceback.print_exc() - return False - - -def cleanup_test_env(): - """清理测试环境""" - print("\n" + "=" * 80) - print("清理测试环境") - print("=" * 80 + "\n") - - config = create_test_config() - temp_dir = config.get('temp_dir') - - if temp_dir and Path(temp_dir).exists(): - print(f"删除测试目录: {temp_dir}") - shutil.rmtree(temp_dir) - print("✅ 清理完成") - else: - print("⚠️ 测试目录不存在或未配置") - - -def main(): - """运行测试""" - print("\n" + "=" * 80) - print("Obsidian KB - 降级流程集成测试") - print("=" * 80) - - print("\n测试说明:") - print(" - 测试视频: BV1MnFNz5EDL (已知无字幕)") - print(" - 验证点: 优雅降级，生成基础元数据笔记") - print(" - 检查项: frontmatter、无字幕提示、内容结构") - - try: - # 运行测试 - result = test_no_subtitle_video() - - # 汇总结果 - print("\n" + "=" * 80) - print("测试结果") - print("=" * 80 + "\n") - - print(f" 降级流程测试: {'✅ 通过' if result else '❌ 失败'}") - - if result: - print("\n 验证标准:") - print(" ✅ 无字幕视频：优雅降级生成基础笔记") - print(" ✅ 输出文件包含基础元数据") - print(" ✅ 有无字幕提示") - print(" ✅ frontmatter 正确") - else: - print("\n ⚠️ 验证失败") - - return 0 if result else 1 - - except KeyboardInterrupt: - print("\n\n⚠️ 测试被用户中断") - return 1 - except Exception as e: - print(f"\n❌ 测试运行异常: {e}") - import traceback - traceback.print_exc() - return 1 - finally: - # 清理测试环境 - print("\n") - cleanup_test_env() - - -if __name__ == '__main__': - sys.exit(main()) diff --git a/skills/obsidian-kb/test_integration.py b/skills/obsidian-kb/test_integration.py deleted file mode 100755 index 1b78f09..0000000 --- a/skills/obsidian-kb/test_integration.py +++ /dev/null @@ -1,301 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -Obsidian KB - 主流程集成测试脚本 - -测试完整流程：BV号 → 提取 → 处理 → 输出 -使用测试配置，不执行Git同步 -""" - -import os -import sys -import tempfile -import shutil -from pathlib import Path - -# 添加项目路径 -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from main_pipeline import MainPipeline -from src.config import Config - - -def create_test_config() -> Config: - """ - 创建测试配置 - - Returns: - 配置对象 - """ - # 创建临时目录作为测试vault - temp_dir = tempfile.mkdtemp(prefix='obsidian_kb_test_') - vault_path = Path(temp_dir) / 'vault' - vault_path.mkdir(parents=True, exist_ok=True) - - # 创建临时目录作为存储 - storage_dir = Path(temp_dir) / 'storage' - storage_dir.mkdir(parents=True, exist_ok=True) - - print(f"📁 测试目录: {temp_dir}") - print(f"📁 测试Vault: {vault_path}") - - # 创建测试配置 - test_config_dict = { - 'ai': { - 'provider': 'openai', - 'model': 'gpt-4', - 'api_key': None, # 测试时不使用AI - }, - 'storage': { - 'inbox_dir': 'inbox', - 'topics_dir': 'topics', - 'data_dir': str(storage_dir), - }, - 'git': { - 'enabled': False, # 测试时禁用Git - 'vault_path': str(vault_path), - 'remote': 'origin', - 'branch': 'main', - 'auto_push': False, - 'commit_message_template': 'Add note: {title}', - }, - 'obsidian': { - 'vault_path': str(vault_path), - 'notes_folder': 'Notes', - }, - 'output': { - 'organize_by': 'author', - 'filename_format': '{date}_{title}.md', - 'auto_tags': True, - 'default_tags': ['video', 'bilibili', 'test'], - }, - 'sources': { - 'bilibili': { - 'enabled': True, - 'cookie': None, - 'extraction': { - 'headless': True, - 'timeout': 30, - }, - }, - }, - 'temp_dir': temp_dir, # 保存临时目录路径以便清理 - } - - # 创建Config对象（手动设置配置） - config = Config.__new__(Config) - config._config = test_config_dict - config.config_path = None - - # 添加属性以兼容接口 - class MockConfig: - def __init__(self, data): - self._data = data - - def get(self, key, default=None): - keys = key.split('.') - value = self._data - for k in keys: - if isinstance(value, dict) and k in value: - value = value[k] - else: - return default - return value - - @property - def vault_path(self): - return Path(self._data['git']['vault_path']) - - return MockConfig(test_config_dict) - - -def test_extract_only(): - """测试仅提取功能""" - print("\n" + "=" * 80) - print("测试 1: 提取功能") - print("=" * 80 + "\n") - - config = create_test_config() - pipeline = MainPipeline(config=config, enable_git_sync=False) - - # 测试BV号 - test_bv = "BV1MnFNz5EDL" # 可以替换为实际的测试视频BV号 - - print(f"测试视频: {test_bv}\n") - - try: - # 提取 - extractor_data = pipeline._extract_video(test_bv) - - print("\n✅ 提取测试通过！") - print(f" 标题: {extractor_data.get('metadata', {}).get('title', 'N/A')}") - print(f" 作者: {extractor_data.get('metadata', {}).get('author', 'N/A')}") - print(f" 时长: {extractor_data.get('metadata', {}).get('duration', 'N/A')}") - print(f" 字幕: {'有' if extractor_data.get('subtitle') else '无'}") - - return extractor_data - - except Exception as e: - print(f"\n❌ 提取测试失败: {e}") - import traceback - traceback.print_exc() - return None - - -def test_full_pipeline(extractor_data=None): - """测试完整流程""" - print("\n" + "=" * 80) - print("测试 2: 完整流程（提取 → 处理 → 输出）") - print("=" * 80 + "\n") - - config = create_test_config() - pipeline = MainPipeline(config=config, enable_git_sync=False) - - # 如果没有传入提取数据，先提取 - if extractor_data is None: - test_bv = "BV1MnFNz5EDL" - print(f"测试视频: {test_bv}\n") - try: - extractor_data = pipeline._extract_video(test_bv) - except Exception as e: - print(f"❌ 提取失败，无法继续测试: {e}") - return False - - try: - # 完整流程 - test_bv = extractor_data.get('metadata', {}).get('bvid', 'BV1MnFNz5EDL') - result = pipeline.process_video(test_bv) - - if result['success']: - print("\n" + "✅" * 40) - print("完整流程测试通过！") - print("✅" * 40 + "\n") - - # 显示结果 - print(result['message']) - - # 检查生成的文件 - if result['content'] and result['content'].output_path: - output_file = Path(result['content'].output_path) - if output_file.exists(): - file_size = output_file.stat().st_size - print(f"\n📄 输出文件:") - print(f" 路径: {output_file}") - print(f" 大小: {file_size:,} 字节") - print(f" 存在: ✅") - else: - print(f"\n⚠️ 输出文件不存在: {output_file}") - - return True - else: - print(f"\n❌ 完整流程测试失败: {result.get('error', 'Unknown error')}") - return False - - except Exception as e: - print(f"\n❌ 完整流程测试异常: {e}") - import traceback - traceback.print_exc() - return False - - -def test_error_handling(): - """测试错误处理""" - print("\n" + "=" * 80) - print("测试 3: 错误处理") - print("=" * 80 + "\n") - - config = create_test_config() - pipeline = MainPipeline(config=config, enable_git_sync=False) - - # 测试无效BV号 - invalid_bv = "BVinvalid" - - print(f"测试无效BV号: {invalid_bv}\n") - - try: - result = pipeline.process_video(invalid_bv) - - if not result['success']: - print("✅ 错误处理测试通过！") - print(f" 正确捕获了错误: {result.get('error', 'N/A')}") - return True - else: - print("❌ 错误处理测试失败：应该失败但成功了") - return False - - except Exception as e: - print(f"⚠️ 测试异常: {e}") - # 异常也可能是正确的（如果没有正确捕获） - return True - - -def cleanup_test_env(): - """清理测试环境""" - print("\n" + "=" * 80) - print("清理测试环境") - print("=" * 80 + "\n") - - config = create_test_config() - temp_dir = config.get('temp_dir') - - if temp_dir and Path(temp_dir).exists(): - print(f"删除测试目录: {temp_dir}") - shutil.rmtree(temp_dir) - print("✅ 清理完成") - else: - print("⚠️ 测试目录不存在或未配置") - - -def main(): - """运行所有测试""" - print("\n" + "=" * 80) - print("Obsidian KB - 主流程集成测试") - print("=" * 80) - - print("\n测试说明:") - print(" - 测试1: 提取功能") - print(" - 测试2: 完整流程（提取 → 处理 → 输出）") - print(" - 测试3: 错误处理") - print(" - 使用测试配置，不执行Git同步") - - results = {} - - try: - # 测试1: 提取 - extractor_data = test_extract_only() - results['extract'] = extractor_data is not None - - # 测试2: 完整流程 - results['full_pipeline'] = test_full_pipeline(extractor_data) - - # 测试3: 错误处理 - results['error_handling'] = test_error_handling() - - # 汇总结果 - print("\n" + "=" * 80) - print("测试结果汇总") - print("=" * 80 + "\n") - - print(f" 测试1 - 提取功能: {'✅ 通过' if results['extract'] else '❌ 失败'}") - print(f" 测试2 - 完整流程: {'✅ 通过' if results['full_pipeline'] else '❌ 失败'}") - print(f" 测试3 - 错误处理: {'✅ 通过' if results['error_handling'] else '❌ 失败'}") - - all_passed = all(results.values()) - print(f"\n 总体结果: {'✅ 全部通过' if all_passed else '❌ 有测试失败'}") - - except KeyboardInterrupt: - print("\n\n⚠️ 测试被用户中断") - except Exception as e: - print(f"\n❌ 测试运行异常: {e}") - import traceback - traceback.print_exc() - finally: - # 清理测试环境 - print("\n") - cleanup_test_env() - - return 0 if all(results.values()) else 1 - - -if __name__ == '__main__': - sys.exit(main()) diff --git a/skills/obsidian-kb/test_outputs.py b/skills/obsidian-kb/test_outputs.py deleted file mode 100644 index 56f43b3..0000000 --- a/skills/obsidian-kb/test_outputs.py +++ /dev/null @@ -1,317 +0,0 @@ -""" -测试输出器功能 -""" -import sys -import os -from pathlib import Path - -# 添加项目路径到 sys.path -project_root = Path(__file__).parent -sys.path.insert(0, str(project_root)) - -from src.outputs import ObsidianWriter, GitSync -from src.models import Content, ContentType, ContentStatus, TimelineEntry, ContentAnalysis -from src.config import Config - - -def create_test_content() -> Content: - """创建测试数据""" - content = Content( - id="test001", - source="bilibili", - source_id="BV1234567890", - url="https://www.bilibili.com/video/BV1234567890", - title="Python 高级编程教程 - 装饰器深入解析", - author="编程大师", - description="深入讲解 Python 装饰器的原理和应用", - duration=1800, # 30分钟 - content_type=ContentType.VIDEO, - status=ContentStatus.GENERATED, - transcript="""欢迎来到本期的 Python 高级编程教程。 - -今天我们要深入探讨装饰器的原理和应用。 - -首先，让我们理解装饰器的基本概念... - -[更多字幕内容...]""", - summary="本教程深入讲解了 Python 装饰器的原理，包括闭包、函数装饰器、类装饰器等核心概念，并通过实际案例展示了装饰器在日志、缓存、权限验证等场景中的应用。", - one_liner="30分钟深入理解 Python 装饰器，从原理到实战", - key_points=[ - "装饰器本质上是高阶函数", - "闭包是装饰器实现的基础", - "@语法糖只是语法糖", - "装饰器可以带参数", - "类装饰器也是常见模式", - ], - tags=["Python", "装饰器", "高级编程", "教程"], - quality_score=85.5, - outline=[ - "什么是装饰器", - "装饰器的原理", - "函数装饰器", - "带参数的装饰器", - "类装饰器", - "装饰器应用场景", - "实战案例", - ], - timeline=[ - TimelineEntry( - timestamp="00:00", - title="课程介绍", - description="介绍本课程的主要内容和学习目标" - ), - TimelineEntry( - timestamp="02:30", - title="装饰器基础", - description="讲解装饰器的基本概念和语法" - ), - TimelineEntry( - timestamp="10:00", - title="装饰器原理", - description="深入理解装饰器背后的实现原理" - ), - TimelineEntry( - timestamp="20:00", - title="实战案例", - description="通过实际代码演示装饰器的应用" - ), - ], - category="编程", - ) - - # 添加详细分析 - content.analysis = ContentAnalysis( - one_liner="30分钟深入理解 Python 装饰器，从原理到实战", - summary="本教程深入讲解了 Python 装饰器的原理，包括闭包、函数装饰器、类装饰器等核心概念。", - key_points=[ - "装饰器本质上是高阶函数", - "闭包是装饰器实现的基础", - "@语法糖只是语法糖", - ], - tags=["Python", "装饰器"], - quality_score=85.5, - prerequisites=["Python 基础语法", "函数基础", "面向对象基础"], - outline=[ - {"title": "装饰器基础", "content": "什么是装饰器，基本语法"}, - {"title": "实现原理", "content": "闭包的作用，装饰器的执行流程"}, - {"title": "高级用法", "content": "带参数的装饰器，类装饰器"}, - {"title": "实际应用", "content": "日志记录、性能测试、权限验证"}, - ], - pitfalls=[ - "忘记保留原函数的元数据（如 __name__, __doc__）", - "装饰器中修改参数可能导致意外行为", - "多层装饰器的执行顺序容易混淆", - ], - practice_advice=[ - "多动手实践，编写不同类型的装饰器", - "使用 functools.wraps 保留函数元数据", - "在复杂场景下考虑使用类装饰器", - ], - ) - - return content - - -def create_test_content_no_subtitle() -> Content: - """创建无字幕的测试数据""" - content = Content( - id="test002", - source="bilibili", - source_id="BV0987654321", - url="https://www.bilibili.com/video/BV0987654321", - title="无字幕测试视频", - author="测试UP主", - description="这是一个没有字幕的测试视频", - duration=300, # 5分钟 - content_type=ContentType.VIDEO, - status=ContentStatus.GENERATED, - transcript="", # 空字幕 - summary="这是一个测试视频，用于验证无字幕时的输出", - one_liner="测试无字幕视频的输出功能", - key_points=[ - "测试要点1", - "测试要点2", - ], - tags=["测试"], - quality_score=60.0, - ) - - return content - - -def test_obsidian_writer(): - """测试 ObsidianWriter""" - print("=" * 60) - print("测试 ObsidianWriter") - print("=" * 60) - - # 创建测试配置 - config = Config() - - # 使用临时测试目录 - test_vault = Path("/tmp/test_obsidian_vault") - if test_vault.exists(): - import shutil - shutil.rmtree(test_vault) - test_vault.mkdir(parents=True, exist_ok=True) - - # 修改配置使用测试 vault - config._config['obsidian']['vault_path'] = str(test_vault) - - # 创建写入器 - writer = ObsidianWriter(config) - - # 测试1: 有字幕的内容 - print("\n测试1: 写入有字幕的内容") - content1 = create_test_content() - file_path1 = writer.write(content1, config) - - if file_path1: - print(f"✓ 成功写入: {file_path1}") - # 检查文件是否存在 - if Path(file_path1).exists(): - print(f"✓ 文件存在") - # 读取并显示前20行 - print("\n文件内容预览（前20行）:") - with open(file_path1, 'r', encoding='utf-8') as f: - lines = f.readlines()[:20] - for i, line in enumerate(lines, 1): - print(f"{i:2d}: {line.rstrip()}") - else: - print(f"✗ 文件不存在") - else: - print("✗ 写入失败") - - # 测试2: 无字幕的内容 - print("\n" + "-" * 60) - print("测试2: 写入无字幕的内容") - content2 = create_test_content_no_subtitle() - file_path2 = writer.write(content2, config) - - if file_path2: - print(f"✓ 成功写入: {file_path2}") - # 检查文件是否存在 - if Path(file_path2).exists(): - print(f"✓ 文件存在") - # 读取并显示最后几行 - print("\n文件内容预览（最后10行）:") - with open(file_path2, 'r', encoding='utf-8') as f: - lines = f.readlines() - for line in lines[-10:]: - print(f" {line.rstrip()}") - else: - print(f"✗ 文件不存在") - else: - print("✗ 写入失败") - - # 检查目录结构 - print("\n" + "-" * 60) - print("目录结构:") - print(f"{test_vault}/") - for root, dirs, files in os.walk(test_vault): - level = root.replace(str(test_vault), '').count(os.sep) - indent = ' ' * 2 * level - print(f"{indent}{os.path.basename(root)}/") - subindent = ' ' * 2 * (level + 1) - for file in files: - print(f"{subindent}{file}") - - # 清理测试目录 - print("\n" + "-" * 60) - print(f"测试目录: {test_vault}") - print("测试完成后可手动删除测试目录") - - return file_path1 is not None and file_path2 is not None - - -def test_git_sync(): - """测试 GitSync""" - print("\n" + "=" * 60) - print("测试 GitSync") - print("=" * 60) - - # 创建测试 vault - test_vault = Path("/tmp/test_git_vault") - if test_vault.exists(): - import shutil - shutil.rmtree(test_vault) - test_vault.mkdir(parents=True, exist_ok=True) - - # 初始化 Git 仓库 - try: - subprocess_result = os.system(f"cd {test_vault} && git init && git config user.email 'test@example.com' && git config user.name 'Test User' > /dev/null 2>&1") - print(f"✓ Git 仓库初始化成功") - except: - print("✗ Git 初始化失败，跳过 GitSync 测试") - return False - - # 创建测试配置 - config = Config() - config._config['obsidian']['vault_path'] = str(test_vault) - config._config['git']['enabled'] = True - config._config['git']['auto_push'] = False # 不测试 push，避免需要远程仓库 - - # 创建 Git 同步器 - sync = GitSync(config) - - # 测试1: 检查 Git 状态 - print("\n测试1: 检查是否为 Git 仓库") - is_repo = sync._is_git_repo() - print(f"✓ 是否为 Git 仓库: {is_repo}") - - # 测试2: 创建测试文件并同步 - print("\n测试2: 创建文件并同步") - test_file = test_vault / "test.md" - test_file.write_text("# 测试文件\n\n这是一个测试文件。") - - result = sync.sync(config) - print(f"✓ 同步结果: {result}") - - # 测试3: 检查 Git 状态 - print("\n测试3: 检查 Git 状态") - os.system(f"cd {test_vault} && git log --oneline -n 3") - - print("\n" + "-" * 60) - print(f"测试目录: {test_vault}") - print("测试完成后可手动删除测试目录") - - return result - - -def main(): - """主测试函数""" - print("\n" + "=" * 60) - print("Obsidian-KB 输出器测试") - print("=" * 60) - - results = {} - - # 测试 ObsidianWriter - results['obsidian_writer'] = test_obsidian_writer() - - # 测试 GitSync - results['git_sync'] = test_git_sync() - - # 汇总结果 - print("\n" + "=" * 60) - print("测试结果汇总") - print("=" * 60) - for test_name, result in results.items(): - status = "✓ 通过" if result else "✗ 失败" - print(f"{test_name}: {status}") - - # 总体结果 - all_passed = all(results.values()) - print("\n" + "=" * 60) - if all_passed: - print("✓ 所有测试通过！") - else: - print("✗ 部分测试失败") - print("=" * 60) - - return all_passed - - -if __name__ == "__main__": - success = main() - sys.exit(0 if success else 1)