"""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)