#!/usr/bin/env python3 """ 首发媒体综合分析主脚本 整合：首发媒体识别 + 媒体评估 """ import json import sys from pathlib import Path # 添加父目录到路径，以便导入其他脚本 sys.path.insert(0, str(Path(__file__).parent)) from identify_first_publisher import identify_first_publisher from assess_media import generate_media_assessment def generate_conclusion(first_publisher: dict, media_assessment: dict, timeline: list) -> str: """ 生成结论性描述 Args: first_publisher: 首发媒体信息 media_assessment: 媒体评估结果 timeline: 传播时间线 Returns: str: 结论性文字描述 """ media_name = first_publisher.get("media_name", "未知媒体") article_url = first_publisher.get("article_url", "") publish_time = first_publisher.get("publish_time", "未知时间") confidence = first_publisher.get("confidence", "unknown") # 从媒体评估中提取信息 basic_info = media_assessment.get("basic_info", {}) political = media_assessment.get("political_leaning", {}) credibility = media_assessment.get("credibility", {}) # 构建结论 conclusion = f"## 首发媒体分析结论\n\n" conclusion += f"**首发媒体：** {media_name}\n\n" conclusion += f"**发布时间：** {publish_time}\n\n" if article_url: conclusion += f"**原始链接：** {article_url}\n\n" # 基本信息 conclusion += "**媒体概况：**\n" if basic_info.get("established_year"): conclusion += f"- 成立时间：{basic_info['established_year']}年\n" if basic_info.get("headquarters"): conclusion += f"- 总部：{basic_info['headquarters']}\n" if basic_info.get("owner"): conclusion += f"- 机构归属：{basic_info['owner']}\n" if basic_info.get("audience_scale"): conclusion += f"- 影响范围：{basic_info['audience_scale']}\n" conclusion += "\n" # 政治倾向 conclusion += "**政治倾向：**\n" conclusion += f"- 倾向性：{political.get('leaning', '未知')}\n" if political.get('evidence'): conclusion += f"- 评估依据：{political['evidence']}\n" conclusion += "\n" # 可信度 conclusion += "**可信度评估：**\n" cred_score = credibility.get("score", 0) cred_level = credibility.get("level", "unknown") conclusion += f"- 评分：{cred_score}/100\n" conclusion += f"- 等级：{cred_level}\n" conclusion += "\n" # 判定依据 conclusion += "**判定依据：**\n" conclusion += f"- 基于时间排序，{media_name} 是所有关联新闻中最早发布的媒体\n" if confidence == "high": conclusion += f"- 时间戳解析成功，判定置信度高\n" else: conclusion += f"- 时间戳解析存在异常，判定置信度较低\n" conclusion += "\n" # 传播情况 if timeline and len(timeline) > 1: conclusion += "**传播情况：**\n" conclusion += f"- 检测到 {len(timeline)} 条相关新闻\n" if len(timeline) >= 2: delay = timeline[1].get("delay_from_previous_minutes", 0) conclusion += f"- 第二跟进媒体在 {delay} 分钟后发布\n" conclusion += "\n" conclusion += "**快速跟进媒体（前5名）：**\n" for i, item in enumerate(timeline[:5], 1): delay_text = f"（间隔{item['delay_from_previous_minutes']}分钟）" if item['delay_from_previous_minutes'] > 0 else "（首发）" conclusion += f"{i}. {item['media_name']} - {item['publish_time']} {delay_text}\n" conclusion += "\n" return conclusion def generate_media_card(media_assessment: dict) -> dict: """ 生成媒体页签卡（结构化数据） Returns: dict: 媒体画像卡片 """ basic_info = media_assessment.get("basic_info", {}) political = media_assessment.get("political_leaning", {}) credibility = media_assessment.get("credibility", {}) card = { "media_name": basic_info.get("media_name", ""), "media_url": media_assessment.get("media_url", ""), "basic_information": { "established_year": basic_info.get("established_year"), "headquarters": basic_info.get("headquarters"), "owner": basic_info.get("owner"), "audience_scale": basic_info.get("audience_scale"), "media_type": basic_info.get("media_type"), "description": basic_info.get("description"), }, "political_leaning": { "leaning": political.get("leaning"), "score": political.get("score"), "evidence": political.get("evidence"), "confidence": political.get("confidence"), }, "credibility": { "score": credibility.get("score"), "level": credibility.get("level"), "factors": credibility.get("factors"), }, "summary": media_assessment.get("summary", ""), } return card def analyze_first_publisher_comprehensive(related_news: list) -> dict: """ 首发媒体综合分析（主入口） Args: related_news: 关联新闻列表 Returns: dict: 综合分析结果 - conclusion: 结论性文字描述 - media_card: 媒体画像卡片（结构化数据） - detailed_results: 详细分析结果 """ # Step 1: 识别首发媒体 identification_result = identify_first_publisher(related_news) if "error" in identification_result: return { "error": identification_result["error"], "conclusion": f"错误：{identification_result['error']}", "media_card": None, } first_publisher = identification_result["first_publisher"] timeline = identification_result.get("propagation_timeline", []) # Step 2: 评估首发媒体 media_url = first_publisher.get("media_url", "") article_url = first_publisher.get("article_url", "") media_name = first_publisher.get("media_name", "") # 如果没有媒体URL，尝试从文章URL提取域名 if not media_url and article_url: from assess_media import extract_domain domain = extract_domain(article_url) media_url = f"https://www.{domain}" media_assessment = generate_media_assessment(media_url, article_url, media_name) # Step 3: 生成结论和媒体卡片 conclusion = generate_conclusion(first_publisher, media_assessment, timeline) media_card = generate_media_card(media_assessment) # Step 4: 整合所有结果 comprehensive_result = { "conclusion": conclusion, "media_card": media_card, "detailed_results": { "first_publisher": first_publisher, "propagation_timeline": timeline, "media_assessment": media_assessment, "analysis_notes": identification_result.get("analysis_notes", []), "total_news_count": identification_result.get("total_news_count", 0), "valid_time_count": identification_result.get("valid_time_count", 0), }, "generated_at": "2024-04-02T00:00:00Z" # 实际应用中应使用当前时间 } return comprehensive_result def main(): """命令行入口""" if len(sys.argv) > 1: # 从文件读取输入数据 input_file = sys.argv[1] with open(input_file, 'r', encoding='utf-8') as f: related_news = json.load(f) else: # 使用示例数据 related_news = [ { "media_name": "路透社", "media_url": "https://www.reuters.com", "article_url": "https://www.reuters.com/world/example-article", "title": "Example News Title", "publish_time": "2024-03-15T10:30:00Z", }, { "media_name": "BBC", "media_url": "https://www.bbc.com", "article_url": "https://www.bbc.com/news/example", "title": "Example Title", "publish_time": "2024-03-15T11:00:00Z", }, { "media_name": "CNN", "media_url": "https://www.cnn.com", "article_url": "https://www.cnn.com/world/example", "title": "Another Title", "publish_time": "2024-03-15T10:45:00Z", }, ] result = analyze_first_publisher_comprehensive(related_news) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()