#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 帖子溯源引擎 (Post Tracer Engine) 主入口：处理溯源请求，协调各模块工作 """ import json import asyncio from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict from pathlib import Path from reverse_search import ReverseSearchEngine from content_comparator import ContentComparator from path_analyzer import PathAnalyzer from account_evaluator import AccountEvaluator @dataclass class TracingRequest: """溯源请求""" post_id: str platform: str content: str images: List[str] # 图片URL或本地路径 videos: List[str] # 视频URL或本地路径 links: List[str] # 链接列表 metadata: Dict[str, Any] # 元数据：发布时间、账号等 @dataclass class TracingResult: """溯源结果""" request: TracingRequest timestamp: str original_source: Optional[Dict[str, Any]] alternative_sources: List[Dict[str, Any]] propagation_path: Dict[str, Any] evidence: Dict[str, Any] confidence: Dict[str, Any] class PostTracer: """帖子溯源器主类""" def __init__(self, cache_dir: str = "./cache", snapshot_dir: str = "./snapshots"): """ 初始化溯源器 Args: cache_dir: 缓存目录 snapshot_dir: 快照目录 """ self.cache_dir = Path(cache_dir) self.snapshot_dir = Path(snapshot_dir) # 创建目录 self.cache_dir.mkdir(parents=True, exist_ok=True) self.snapshot_dir.mkdir(parents=True, exist_ok=True) # 初始化模块 self.reverse_search = ReverseSearchEngine(cache_dir=str(self.cache_dir)) self.content_comparator = ContentComparator() self.path_analyzer = PathAnalyzer() self.account_evaluator = AccountEvaluator() async def trace_post(self, request: TracingRequest) -> TracingResult: """ 执行完整的溯源流程 Args: request: 溯源请求 Returns: TracingResult: 溯源结果 """ print(f"开始溯源帖子: {request.post_id}@{request.platform}") # Step 1: 反向搜索 print("Step 1: 执行反向搜索...") search_results = await self._perform_reverse_search(request) # Step 2: 内容比对 print("Step 2: 内容比对...") comparison_results = await self._compare_contents(request, search_results) # Step 3: 传播路径分析 print("Step 3: 分析传播路径...") propagation_path = await self._analyze_propagation(search_results) # Step 4: 确定原始来源 print("Step 4: 确定原始来源...") original_source, alternative_sources = await self._determine_original_source( search_results, comparison_results ) # Step 5: 账号可信度评估 print("Step 5: 评估账号可信度...") credibility_assessment = await self._evaluate_credibility( original_source, alternative_sources ) # Step 6: 置信度评估 print("Step 6: 评估置信度...") confidence = await self._assess_confidence( search_results, comparison_results, credibility_assessment ) # 构建结果 result = TracingResult( request=request, timestamp=datetime.now().isoformat(), original_source=original_source, alternative_sources=alternative_sources, propagation_path=propagation_path, evidence={ "reverse_search_results": search_results, "content_comparison": comparison_results, "source_account_assessment": credibility_assessment }, confidence=confidence ) return result async def _perform_reverse_search(self, request: TracingRequest) -> List[Dict[str, Any]]: """执行反向搜索""" results = [] # 搜索图片 for image_url in request.images: image_results = await self.reverse_search.search_by_image(image_url) results.extend(image_results) # 搜索视频关键帧 for video_url in request.videos: video_results = await self.reverse_search.search_by_video(video_url) results.extend(video_results) # 搜索链接 for link in request.links: link_results = await self.reverse_search.search_by_url(link) results.extend(link_results) # 按时间排序 results = sorted(results, key=lambda x: x.get('timestamp', ''), reverse=False) return results async def _compare_contents( self, request: TracingRequest, search_results: List[Dict[str, Any]] ) -> Dict[str, Any]: """内容比对""" comparison = {} # 文本比对 if request.content: text_similarity = self.content_comparator.compare_text( request.content, [r.get('content', '') for r in search_results] ) comparison['text_similarity'] = text_similarity # 图片比对 if request.images: image_similarity = self.content_comparator.compare_images( request.images, [r.get('image', '') for r in search_results] ) comparison['image_similarity'] = image_similarity return comparison async def _analyze_propagation( self, search_results: List[Dict[str, Any]] ) -> Dict[str, Any]: """分析传播路径""" return await self.path_analyzer.analyze(search_results) async def _determine_original_source( self, search_results: List[Dict[str, Any]], comparison_results: Dict[str, Any] ) -> tuple: """确定原始来源""" # 按发布时间排序，最早的作为候选 sorted_results = sorted( search_results, key=lambda x: x.get('timestamp', ''), reverse=False ) if not sorted_results: return None, [] # 最可能的原始来源（最早且相似度最高） original_source = sorted_results[0] # 其他可能的来源 alternative_sources = sorted_results[1:5] # 取前5个 return original_source, alternative_sources async def _evaluate_credibility( self, original_source: Optional[Dict[str, Any]], alternative_sources: List[Dict[str, Any]] ) -> Dict[str, Any]: """评估账号可信度""" assessments = {} if original_source: assessments['original'] = await self.account_evaluator.evaluate( original_source.get('account', {}) ) for idx, source in enumerate(alternative_sources): assessments[f'alternative_{idx}'] = await self.account_evaluator.evaluate( source.get('account', {}) ) return assessments async def _assess_confidence( self, search_results: List[Dict[str, Any]], comparison_results: Dict[str, Any], credibility_assessment: Dict[str, Any] ) -> Dict[str, Any]: """评估置信度""" # 基于证据完整性和一致性计算置信度 evidence_count = len(search_results) text_similarity = comparison_results.get('text_similarity', {}).get('max', 0) image_similarity = comparison_results.get('image_similarity', {}).get('max', 0) # 计算置信度分数（0-1） confidence_score = 0.0 # 证据数量权重 if evidence_count > 0: confidence_score += min(evidence_count * 0.1, 0.3) # 相似度权重 confidence_score += text_similarity * 0.35 confidence_score += image_similarity * 0.35 # 置信度截断在0-1之间 confidence_score = min(max(confidence_score, 0), 1) # 确定置信度等级 if confidence_score >= 0.8: confidence_level = "高" elif confidence_score >= 0.5: confidence_level = "中" else: confidence_level = "低" # 识别局限性 limitations = [] if evidence_count == 0: limitations.append("未找到任何反向搜索结果") if evidence_count < 3: limitations.append("找到的反向搜索结果较少") if text_similarity < 0.5: limitations.append("文本相似度较低，可能内容已被大幅修改") if image_similarity < 0.5: limitations.append("图像相似度较低，可能图片已被编辑") # 添加常见局限性 limitations.extend([ "原始帖子可能已被删除", "无法访问某些平台（如微信）进行搜索", "图片经多次编辑，最早版本难以追溯" ]) return { "confidence_level": confidence_level, "confidence_score": round(confidence_score, 2), "limitations": limitations, "data_sources": ["Google Images", "Yandex", "TinEye", "Bing"], "evidence_count": evidence_count } def save_result(self, result: TracingResult, output_path: str = "result.json"): """保存溯源结果""" with open(output_path, 'w', encoding='utf-8') as f: json.dump(asdict(result), f, ensure_ascii=False, indent=2) print(f"溯源结果已保存至: {output_path}") async def main(): """示例用法""" # 创建溯源器 tracer = PostTracer() # 构造测试请求 request = TracingRequest( post_id="test_post_001", platform="Twitter", content="这是一条测试推文", images=["https://example.com/test_image.jpg"], videos=[], links=["https://example.com/article"], metadata={ "publish_time": "2026-04-02T12:00:00Z", "author": "test_user", "account_id": "12345678" } ) # 执行溯源 result = await tracer.trace_post(request) # 保存结果 tracer.save_result(result) # 打印摘要 print("\n=== 溯源摘要 ===") print(f"置信度: {result.confidence['confidence_level']} ({result.confidence['confidence_score']})") if result.original_source: print(f"最可能的原始来源: {result.original_source.get('account_name')}@{result.original_source.get('platform')}") print(f"发布时间: {result.original_source.get('publish_time')}") print(f"帖子链接: {result.original_source.get('post_link')}") if __name__ == "__main__": asyncio.run(main())