| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673 |
- # coding=utf-8
- """
- TrendRadar 主程序
- 热点新闻聚合与分析工具
- 支持: python -m trendradar
- """
- import os
- import re
- import webbrowser
- from pathlib import Path
- from typing import Dict, List, Tuple, Optional
- import requests
- from trendradar.context import AppContext
- from trendradar import __version__
- from trendradar.core import load_config
- from trendradar.core.analyzer import convert_keyword_stats_to_platform_stats
- from trendradar.crawler import DataFetcher
- from trendradar.storage import convert_crawl_results_to_news_data
- from trendradar.utils.time import is_within_days
- from trendradar.ai import AIAnalyzer, AIAnalysisResult
- def _parse_version(version_str: str) -> Tuple[int, int, int]:
- """解析版本号字符串为元组"""
- try:
- parts = version_str.strip().split(".")
- if len(parts) >= 3:
- return int(parts[0]), int(parts[1]), int(parts[2])
- return 0, 0, 0
- except:
- return 0, 0, 0
- def _compare_version(local: str, remote: str) -> str:
- """比较版本号,返回状态文字"""
- local_tuple = _parse_version(local)
- remote_tuple = _parse_version(remote)
- if local_tuple < remote_tuple:
- return "⚠️ 需要更新"
- elif local_tuple > remote_tuple:
- return "🔮 超前版本"
- else:
- return "✅ 已是最新"
- def _fetch_remote_version(version_url: str, proxy_url: Optional[str] = None) -> Optional[str]:
- """获取远程版本号"""
- try:
- proxies = None
- if proxy_url:
- proxies = {"http": proxy_url, "https": proxy_url}
- headers = {
- "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
- "Accept": "text/plain, */*",
- "Cache-Control": "no-cache",
- }
- response = requests.get(version_url, proxies=proxies, headers=headers, timeout=10)
- response.raise_for_status()
- return response.text.strip()
- except Exception as e:
- print(f"[版本检查] 获取远程版本失败: {e}")
- return None
- def _parse_config_versions(content: str) -> Dict[str, str]:
- """解析配置文件版本内容为字典"""
- versions = {}
- try:
- if not content:
- return versions
- for line in content.splitlines():
- line = line.strip()
- if not line or "=" not in line:
- continue
- name, version = line.split("=", 1)
- versions[name.strip()] = version.strip()
- except Exception as e:
- print(f"[版本检查] 解析配置版本失败: {e}")
- return versions
- def check_all_versions(
- version_url: str,
- configs_version_url: Optional[str] = None,
- proxy_url: Optional[str] = None
- ) -> Tuple[bool, Optional[str]]:
- """
- 统一版本检查:程序版本 + 配置文件版本
- Args:
- version_url: 远程程序版本检查 URL
- configs_version_url: 远程配置文件版本检查 URL (返回格式: filename=version)
- proxy_url: 代理 URL
- Returns:
- (need_update, remote_version): 程序是否需要更新及远程版本号
- """
- # 获取远程版本
- remote_version = _fetch_remote_version(version_url, proxy_url)
- # 获取远程配置版本(如果有提供 URL)
- remote_config_versions = {}
- if configs_version_url:
- content = _fetch_remote_version(configs_version_url, proxy_url)
- if content:
- remote_config_versions = _parse_config_versions(content)
- print("=" * 60)
- print("版本检查")
- print("=" * 60)
- if remote_version:
- print(f"远程程序版本: {remote_version}")
- else:
- print("远程程序版本: 获取失败")
- if configs_version_url:
- if remote_config_versions:
- print(f"远程配置清单: 获取成功 ({len(remote_config_versions)} 个文件)")
- else:
- print("远程配置清单: 获取失败或为空")
- print("-" * 60)
- program_status = _compare_version(__version__, remote_version) if remote_version else "(无法比较)"
- print(f" 主程序版本: {__version__} {program_status}")
- config_files = [
- Path("config/config.yaml"),
- Path("config/frequency_words.txt"),
- Path("config/ai_analysis_prompt.txt"),
- Path("config/ai_translation_prompt.txt"),
- ]
- version_pattern = re.compile(r"Version:\s*(\d+\.\d+\.\d+)", re.IGNORECASE)
- for config_file in config_files:
- if not config_file.exists():
- print(f" {config_file.name}: 文件不存在")
- continue
- try:
- with open(config_file, "r", encoding="utf-8") as f:
- local_version = None
- for i, line in enumerate(f):
- if i >= 20:
- break
- match = version_pattern.search(line)
- if match:
- local_version = match.group(1)
- break
- # 获取该文件的远程版本
- target_remote_version = remote_config_versions.get(config_file.name)
- if local_version:
- if target_remote_version:
- status = _compare_version(local_version, target_remote_version)
- print(f" {config_file.name}: {local_version} {status}")
- else:
- print(f" {config_file.name}: {local_version} (未找到远程版本)")
- else:
- print(f" {config_file.name}: 未找到本地版本号")
- except Exception as e:
- print(f" {config_file.name}: 读取失败 - {e}")
- print("=" * 60)
- # 返回程序版本的更新状态
- if remote_version:
- need_update = _parse_version(__version__) < _parse_version(remote_version)
- return need_update, remote_version if need_update else None
- return False, None
- # === 主分析器 ===
- class NewsAnalyzer:
- """新闻分析器"""
- # 模式策略定义
- MODE_STRATEGIES = {
- "incremental": {
- "mode_name": "增量模式",
- "description": "增量模式(只关注新增新闻,无新增时不推送)",
- "report_type": "增量分析",
- "should_send_notification": True,
- },
- "current": {
- "mode_name": "当前榜单模式",
- "description": "当前榜单模式(当前榜单匹配新闻 + 新增新闻区域 + 按时推送)",
- "report_type": "当前榜单",
- "should_send_notification": True,
- },
- "daily": {
- "mode_name": "全天汇总模式",
- "description": "全天汇总模式(所有匹配新闻 + 新增新闻区域 + 按时推送)",
- "report_type": "全天汇总",
- "should_send_notification": True,
- },
- }
- def __init__(self, config: Optional[Dict] = None):
- # 使用传入的配置或加载新配置
- if config is None:
- print("正在加载配置...")
- config = load_config()
- print(f"TrendRadar v{__version__} 配置加载完成")
- print(f"监控平台数量: {len(config['PLATFORMS'])}")
- print(f"时区: {config.get('TIMEZONE', 'Asia/Shanghai')}")
- # 创建应用上下文
- self.ctx = AppContext(config)
- self.request_interval = self.ctx.config["REQUEST_INTERVAL"]
- self.report_mode = self.ctx.config["REPORT_MODE"]
- self.rank_threshold = self.ctx.rank_threshold
- self.is_github_actions = os.environ.get("GITHUB_ACTIONS") == "true"
- self.is_docker_container = self._detect_docker_environment()
- self.update_info = None
- self.proxy_url = None
- self._setup_proxy()
- self.data_fetcher = DataFetcher(self.proxy_url)
- # 初始化存储管理器(使用 AppContext)
- self._init_storage_manager()
- # 注意:update_info 由 main() 函数设置,避免重复请求远程版本
- def _init_storage_manager(self) -> None:
- """初始化存储管理器(使用 AppContext)"""
- # 获取数据保留天数(支持环境变量覆盖)
- env_retention = os.environ.get("STORAGE_RETENTION_DAYS", "").strip()
- if env_retention:
- # 环境变量覆盖配置
- self.ctx.config["STORAGE"]["RETENTION_DAYS"] = int(env_retention)
- self.storage_manager = self.ctx.get_storage_manager()
- print(f"存储后端: {self.storage_manager.backend_name}")
- retention_days = self.ctx.config.get("STORAGE", {}).get("RETENTION_DAYS", 0)
- if retention_days > 0:
- print(f"数据保留天数: {retention_days} 天")
- def _detect_docker_environment(self) -> bool:
- """检测是否运行在 Docker 容器中"""
- try:
- if os.environ.get("DOCKER_CONTAINER") == "true":
- return True
- if os.path.exists("/.dockerenv"):
- return True
- return False
- except Exception:
- return False
- def _should_open_browser(self) -> bool:
- """判断是否应该打开浏览器"""
- return not self.is_github_actions and not self.is_docker_container
- def _setup_proxy(self) -> None:
- """设置代理配置"""
- if not self.is_github_actions and self.ctx.config["USE_PROXY"]:
- self.proxy_url = self.ctx.config["DEFAULT_PROXY"]
- print("本地环境,使用代理")
- elif not self.is_github_actions and not self.ctx.config["USE_PROXY"]:
- print("本地环境,未启用代理")
- else:
- print("GitHub Actions环境,不使用代理")
- def _set_update_info_from_config(self) -> None:
- """从已缓存的远程版本设置更新信息(不再重复请求)"""
- try:
- version_url = self.ctx.config.get("VERSION_CHECK_URL", "")
- if not version_url:
- return
- remote_version = _fetch_remote_version(version_url, self.proxy_url)
- if remote_version:
- need_update = _parse_version(__version__) < _parse_version(remote_version)
- if need_update:
- self.update_info = {
- "current_version": __version__,
- "remote_version": remote_version,
- }
- except Exception as e:
- print(f"版本检查出错: {e}")
- def _get_mode_strategy(self) -> Dict:
- """获取当前模式的策略配置"""
- return self.MODE_STRATEGIES.get(self.report_mode, self.MODE_STRATEGIES["daily"])
- def _has_notification_configured(self) -> bool:
- """检查是否配置了任何通知渠道"""
- cfg = self.ctx.config
- return any(
- [
- cfg["FEISHU_WEBHOOK_URL"],
- cfg["DINGTALK_WEBHOOK_URL"],
- cfg["WEWORK_WEBHOOK_URL"],
- (cfg["TELEGRAM_BOT_TOKEN"] and cfg["TELEGRAM_CHAT_ID"]),
- (
- cfg["EMAIL_FROM"]
- and cfg["EMAIL_PASSWORD"]
- and cfg["EMAIL_TO"]
- ),
- (cfg["NTFY_SERVER_URL"] and cfg["NTFY_TOPIC"]),
- cfg["BARK_URL"],
- cfg["SLACK_WEBHOOK_URL"],
- cfg["GENERIC_WEBHOOK_URL"],
- ]
- )
- def _has_valid_content(
- self, stats: List[Dict], new_titles: Optional[Dict] = None
- ) -> bool:
- """检查是否有有效的新闻内容"""
- if self.report_mode == "incremental":
- # 增量模式:只要有匹配的新闻就推送
- # count_word_frequency 已经确保只处理新增的新闻(包括当天第一次爬取的情况)
- has_matched_news = any(stat["count"] > 0 for stat in stats)
- return has_matched_news
- elif self.report_mode == "current":
- # current模式:只要stats有内容就说明有匹配的新闻
- return any(stat["count"] > 0 for stat in stats)
- else:
- # 当日汇总模式下,检查是否有匹配的频率词新闻或新增新闻
- has_matched_news = any(stat["count"] > 0 for stat in stats)
- has_new_news = bool(
- new_titles and any(len(titles) > 0 for titles in new_titles.values())
- )
- return has_matched_news or has_new_news
- def _prepare_ai_analysis_data(
- self,
- ai_mode: str,
- current_results: Optional[Dict] = None,
- current_id_to_name: Optional[Dict] = None,
- ) -> Tuple[List[Dict], Optional[Dict]]:
- """
- 为 AI 分析准备指定模式的数据
- Args:
- ai_mode: AI 分析模式 (daily/current/incremental)
- current_results: 当前抓取的结果(用于 incremental 模式)
- current_id_to_name: 当前的平台映射(用于 incremental 模式)
- Returns:
- Tuple[stats, id_to_name]: 统计数据和平台映射
- """
- try:
- word_groups, filter_words, global_filters = self.ctx.load_frequency_words()
- if ai_mode == "incremental":
- # incremental 模式:使用当前抓取的数据
- if not current_results or not current_id_to_name:
- print("[AI] incremental 模式需要当前抓取数据,但未提供")
- return [], None
- # 准备当前时间信息
- time_info = self.ctx.format_time()
- title_info = self._prepare_current_title_info(current_results, time_info)
- # 检测新增标题
- new_titles = self.ctx.detect_new_titles(list(current_results.keys()))
- # 统计计算
- stats, _ = self.ctx.count_frequency(
- current_results,
- word_groups,
- filter_words,
- current_id_to_name,
- title_info,
- new_titles,
- mode="incremental",
- global_filters=global_filters,
- quiet=True,
- )
- # 如果是 platform 模式,转换数据结构
- if self.ctx.display_mode == "platform" and stats:
- stats = convert_keyword_stats_to_platform_stats(
- stats,
- self.ctx.weight_config,
- self.ctx.rank_threshold,
- )
- return stats, current_id_to_name
- elif ai_mode in ["daily", "current"]:
- # 加载历史数据
- analysis_data = self._load_analysis_data(quiet=True)
- if not analysis_data:
- print(f"[AI] 无法加载历史数据用于 {ai_mode} 模式分析")
- return [], None
- (
- all_results,
- id_to_name,
- title_info,
- new_titles,
- _,
- _,
- _,
- ) = analysis_data
- # 统计计算
- stats, _ = self.ctx.count_frequency(
- all_results,
- word_groups,
- filter_words,
- id_to_name,
- title_info,
- new_titles,
- mode=ai_mode,
- global_filters=global_filters,
- quiet=True,
- )
- # 如果是 platform 模式,转换数据结构
- if self.ctx.display_mode == "platform" and stats:
- stats = convert_keyword_stats_to_platform_stats(
- stats,
- self.ctx.weight_config,
- self.ctx.rank_threshold,
- )
- return stats, id_to_name
- else:
- print(f"[AI] 未知的 AI 模式: {ai_mode}")
- return [], None
- except Exception as e:
- print(f"[AI] 准备 {ai_mode} 模式数据时出错: {e}")
- if self.ctx.config.get("DEBUG", False):
- import traceback
- traceback.print_exc()
- return [], None
- def _run_ai_analysis(
- self,
- stats: List[Dict],
- rss_items: Optional[List[Dict]],
- mode: str,
- report_type: str,
- id_to_name: Optional[Dict],
- current_results: Optional[Dict] = None,
- ) -> Optional[AIAnalysisResult]:
- """执行 AI 分析"""
- analysis_config = self.ctx.config.get("AI_ANALYSIS", {})
- if not analysis_config.get("ENABLED", False):
- return None
- # AI 分析时间窗口控制
- analysis_window = analysis_config.get("ANALYSIS_WINDOW", {})
- if analysis_window.get("ENABLED", False):
- push_manager = self.ctx.create_push_manager()
- time_range_start = analysis_window["TIME_RANGE"]["START"]
- time_range_end = analysis_window["TIME_RANGE"]["END"]
- if not push_manager.is_in_time_range(time_range_start, time_range_end):
- now = self.ctx.get_time()
- print(
- f"[AI] 分析窗口控制:当前时间 {now.strftime('%H:%M')} 不在分析时间窗口 {time_range_start}-{time_range_end} 内,跳过 AI 分析"
- )
- return None
- if analysis_window.get("ONCE_PER_DAY", False):
- # 检查今天是否已经进行过 AI 分析
- if push_manager.storage_backend.has_ai_analyzed_today():
- print(f"[AI] 分析窗口控制:今天已分析过,跳过本次 AI 分析")
- return None
- else:
- print(f"[AI] 分析窗口控制:今天首次分析")
- print("[AI] 正在进行 AI 分析...")
- try:
- ai_config = self.ctx.config.get("AI", {})
- debug_mode = self.ctx.config.get("DEBUG", False)
- analyzer = AIAnalyzer(ai_config, analysis_config, self.ctx.get_time, debug=debug_mode)
- # 确定 AI 分析使用的模式
- ai_mode_config = analysis_config.get("MODE", "follow_report")
- if ai_mode_config == "follow_report":
- # 跟随推送报告模式
- ai_mode = mode
- ai_stats = stats
- ai_id_to_name = id_to_name
- elif ai_mode_config in ["daily", "current", "incremental"]:
- # 使用独立配置的模式,需要重新准备数据
- ai_mode = ai_mode_config
- if ai_mode != mode:
- print(f"[AI] 使用独立分析模式: {ai_mode} (推送模式: {mode})")
- print(f"[AI] 正在准备 {ai_mode} 模式的数据...")
- # 根据 AI 模式重新准备数据
- ai_stats, ai_id_to_name = self._prepare_ai_analysis_data(
- ai_mode, current_results, id_to_name
- )
- if not ai_stats:
- print(f"[AI] 警告: 无法准备 {ai_mode} 模式的数据,回退到推送模式数据")
- ai_stats = stats
- ai_id_to_name = id_to_name
- ai_mode = mode
- else:
- ai_stats = stats
- ai_id_to_name = id_to_name
- else:
- # 配置错误,回退到跟随模式
- print(f"[AI] 警告: 无效的 ai_analysis.mode 配置 '{ai_mode_config}',使用推送模式 '{mode}'")
- ai_mode = mode
- ai_stats = stats
- ai_id_to_name = id_to_name
- # 提取平台列表
- platforms = list(ai_id_to_name.values()) if ai_id_to_name else []
- # 提取关键词列表
- keywords = [s.get("word", "") for s in ai_stats if s.get("word")] if ai_stats else []
- # 确定报告类型
- if ai_mode != mode:
- # 根据 AI 模式确定报告类型
- ai_report_type = {
- "daily": "当日汇总",
- "current": "当前榜单",
- "incremental": "增量更新"
- }.get(ai_mode, report_type)
- else:
- ai_report_type = report_type
- result = analyzer.analyze(
- stats=ai_stats,
- rss_stats=rss_items,
- report_mode=ai_mode,
- report_type=ai_report_type,
- platforms=platforms,
- keywords=keywords,
- )
- # 设置 AI 分析使用的模式
- if result.success:
- result.ai_mode = ai_mode
- if result.error:
- # 成功但有警告(如 JSON 解析问题但使用了原始文本)
- print(f"[AI] 分析完成(有警告: {result.error})")
- else:
- print("[AI] 分析完成")
- # 记录 AI 分析(如果启用了 once_per_day)
- if analysis_window.get("ENABLED", False) and analysis_window.get("ONCE_PER_DAY", False):
- push_manager = self.ctx.create_push_manager()
- push_manager.storage_backend.record_ai_analysis(ai_mode)
- else:
- print(f"[AI] 分析失败: {result.error}")
- return result
- except Exception as e:
- import traceback
- error_type = type(e).__name__
- error_msg = str(e)
- # 截断过长的错误消息
- if len(error_msg) > 200:
- error_msg = error_msg[:200] + "..."
- print(f"[AI] 分析出错 ({error_type}): {error_msg}")
- # 详细错误日志到 stderr
- import sys
- print(f"[AI] 详细错误堆栈:", file=sys.stderr)
- traceback.print_exc(file=sys.stderr)
- return AIAnalysisResult(success=False, error=f"{error_type}: {error_msg}")
- def _load_analysis_data(
- self,
- quiet: bool = False,
- ) -> Optional[Tuple[Dict, Dict, Dict, Dict, List, List]]:
- """统一的数据加载和预处理,使用当前监控平台列表过滤历史数据"""
- try:
- # 获取当前配置的监控平台ID列表
- current_platform_ids = self.ctx.platform_ids
- if not quiet:
- print(f"当前监控平台: {current_platform_ids}")
- all_results, id_to_name, title_info = self.ctx.read_today_titles(
- current_platform_ids, quiet=quiet
- )
- if not all_results:
- print("没有找到当天的数据")
- return None
- total_titles = sum(len(titles) for titles in all_results.values())
- if not quiet:
- print(f"读取到 {total_titles} 个标题(已按当前监控平台过滤)")
- new_titles = self.ctx.detect_new_titles(current_platform_ids, quiet=quiet)
- word_groups, filter_words, global_filters = self.ctx.load_frequency_words()
- return (
- all_results,
- id_to_name,
- title_info,
- new_titles,
- word_groups,
- filter_words,
- global_filters,
- )
- except Exception as e:
- print(f"数据加载失败: {e}")
- return None
- def _prepare_current_title_info(self, results: Dict, time_info: str) -> Dict:
- """从当前抓取结果构建标题信息"""
- title_info = {}
- for source_id, titles_data in results.items():
- title_info[source_id] = {}
- for title, title_data in titles_data.items():
- ranks = title_data.get("ranks", [])
- url = title_data.get("url", "")
- mobile_url = title_data.get("mobileUrl", "")
- title_info[source_id][title] = {
- "first_time": time_info,
- "last_time": time_info,
- "count": 1,
- "ranks": ranks,
- "url": url,
- "mobileUrl": mobile_url,
- }
- return title_info
- def _prepare_standalone_data(
- self,
- results: Dict,
- id_to_name: Dict,
- title_info: Optional[Dict] = None,
- rss_items: Optional[List[Dict]] = None,
- ) -> Optional[Dict]:
- """
- 从原始数据中提取独立展示区数据
- Args:
- results: 原始爬取结果 {platform_id: {title: title_data}}
- id_to_name: 平台 ID 到名称的映射
- title_info: 标题元信息(含排名历史、时间等)
- rss_items: RSS 条目列表
- Returns:
- 独立展示数据字典,如果未启用返回 None
- """
- display_config = self.ctx.config.get("DISPLAY", {})
- regions = display_config.get("REGIONS", {})
- standalone_config = display_config.get("STANDALONE", {})
- if not regions.get("STANDALONE", False):
- return None
- platform_ids = standalone_config.get("PLATFORMS", [])
- rss_feed_ids = standalone_config.get("RSS_FEEDS", [])
- max_items = standalone_config.get("MAX_ITEMS", 20)
- if not platform_ids and not rss_feed_ids:
- return None
- standalone_data = {
- "platforms": [],
- "rss_feeds": [],
- }
- # 找出最新批次时间(类似 current 模式的过滤逻辑)
- latest_time = None
- if title_info:
- for source_titles in title_info.values():
- for title_data in source_titles.values():
- last_time = title_data.get("last_time", "")
- if last_time:
- if latest_time is None or last_time > latest_time:
- latest_time = last_time
- # 提取热榜平台数据
- for platform_id in platform_ids:
- if platform_id not in results:
- continue
- platform_name = id_to_name.get(platform_id, platform_id)
- platform_titles = results[platform_id]
- items = []
- for title, title_data in platform_titles.items():
- # 获取元信息(如果有 title_info)
- meta = {}
- if title_info and platform_id in title_info and title in title_info[platform_id]:
- meta = title_info[platform_id][title]
- # 只保留当前在榜的话题(last_time 等于最新时间)
- if latest_time and meta:
- if meta.get("last_time") != latest_time:
- continue
- # 使用当前热榜的排名数据(title_data)进行排序
- # title_data 包含的是爬虫返回的当前排名,用于保证独立展示区的顺序与热榜一致
- current_ranks = title_data.get("ranks", [])
- current_rank = current_ranks[-1] if current_ranks else 0
- # 用于显示的排名范围:合并历史排名和当前排名
- historical_ranks = meta.get("ranks", []) if meta else []
- # 合并去重,保持顺序
- all_ranks = historical_ranks.copy()
- for rank in current_ranks:
- if rank not in all_ranks:
- all_ranks.append(rank)
- display_ranks = all_ranks if all_ranks else current_ranks
- item = {
- "title": title,
- "url": title_data.get("url", ""),
- "mobileUrl": title_data.get("mobileUrl", ""),
- "rank": current_rank, # 用于排序的当前排名
- "ranks": display_ranks, # 用于显示的排名范围(历史+当前)
- "first_time": meta.get("first_time", ""),
- "last_time": meta.get("last_time", ""),
- "count": meta.get("count", 1),
- }
- items.append(item)
- # 按当前排名排序
- items.sort(key=lambda x: x["rank"] if x["rank"] > 0 else 9999)
- # 限制条数
- if max_items > 0:
- items = items[:max_items]
- if items:
- standalone_data["platforms"].append({
- "id": platform_id,
- "name": platform_name,
- "items": items,
- })
- # 提取 RSS 数据
- if rss_items and rss_feed_ids:
- # 按 feed_id 分组
- feed_items_map = {}
- for item in rss_items:
- feed_id = item.get("feed_id", "")
- if feed_id in rss_feed_ids:
- if feed_id not in feed_items_map:
- feed_items_map[feed_id] = {
- "name": item.get("feed_name", feed_id),
- "items": [],
- }
- feed_items_map[feed_id]["items"].append({
- "title": item.get("title", ""),
- "url": item.get("url", ""),
- "published_at": item.get("published_at", ""),
- "author": item.get("author", ""),
- })
- # 限制条数并添加到结果
- for feed_id in rss_feed_ids:
- if feed_id in feed_items_map:
- feed_data = feed_items_map[feed_id]
- items = feed_data["items"]
- if max_items > 0:
- items = items[:max_items]
- if items:
- standalone_data["rss_feeds"].append({
- "id": feed_id,
- "name": feed_data["name"],
- "items": items,
- })
- # 如果没有任何数据,返回 None
- if not standalone_data["platforms"] and not standalone_data["rss_feeds"]:
- return None
- return standalone_data
- def _run_analysis_pipeline(
- self,
- data_source: Dict,
- mode: str,
- title_info: Dict,
- new_titles: Dict,
- word_groups: List[Dict],
- filter_words: List[str],
- id_to_name: Dict,
- failed_ids: Optional[List] = None,
- global_filters: Optional[List[str]] = None,
- quiet: bool = False,
- rss_items: Optional[List[Dict]] = None,
- rss_new_items: Optional[List[Dict]] = None,
- standalone_data: Optional[Dict] = None,
- ) -> Tuple[List[Dict], Optional[str], Optional[AIAnalysisResult]]:
- """统一的分析流水线:数据处理 → 统计计算 → AI分析 → HTML生成"""
- # 统计计算(使用 AppContext)
- stats, total_titles = self.ctx.count_frequency(
- data_source,
- word_groups,
- filter_words,
- id_to_name,
- title_info,
- new_titles,
- mode=mode,
- global_filters=global_filters,
- quiet=quiet,
- )
- # 如果是 platform 模式,转换数据结构
- if self.ctx.display_mode == "platform" and stats:
- stats = convert_keyword_stats_to_platform_stats(
- stats,
- self.ctx.weight_config,
- self.ctx.rank_threshold,
- )
- # AI 分析(如果启用,用于 HTML 报告)
- ai_result = None
- ai_config = self.ctx.config.get("AI_ANALYSIS", {})
- if ai_config.get("ENABLED", False) and stats:
- # 获取模式策略来确定报告类型
- mode_strategy = self._get_mode_strategy()
- report_type = mode_strategy["report_type"]
- ai_result = self._run_ai_analysis(
- stats, rss_items, mode, report_type, id_to_name, current_results=data_source
- )
- # HTML生成(如果启用)
- html_file = None
- if self.ctx.config["STORAGE"]["FORMATS"]["HTML"]:
- html_file = self.ctx.generate_html(
- stats,
- total_titles,
- failed_ids=failed_ids,
- new_titles=new_titles,
- id_to_name=id_to_name,
- mode=mode,
- update_info=self.update_info if self.ctx.config["SHOW_VERSION_UPDATE"] else None,
- rss_items=rss_items,
- rss_new_items=rss_new_items,
- ai_analysis=ai_result,
- standalone_data=standalone_data,
- )
- return stats, html_file, ai_result
- def _send_notification_if_needed(
- self,
- stats: List[Dict],
- report_type: str,
- mode: str,
- failed_ids: Optional[List] = None,
- new_titles: Optional[Dict] = None,
- id_to_name: Optional[Dict] = None,
- html_file_path: Optional[str] = None,
- rss_items: Optional[List[Dict]] = None,
- rss_new_items: Optional[List[Dict]] = None,
- standalone_data: Optional[Dict] = None,
- ai_result: Optional[AIAnalysisResult] = None,
- current_results: Optional[Dict] = None,
- ) -> bool:
- """统一的通知发送逻辑,包含所有判断条件,支持热榜+RSS合并推送+AI分析+独立展示区"""
- has_notification = self._has_notification_configured()
- cfg = self.ctx.config
- # 检查是否有有效内容(热榜或RSS)
- has_news_content = self._has_valid_content(stats, new_titles)
- has_rss_content = bool(rss_items and len(rss_items) > 0)
- has_any_content = has_news_content or has_rss_content
- # 计算热榜匹配条数
- news_count = sum(len(stat.get("titles", [])) for stat in stats) if stats else 0
- # rss_items 是统计列表 [{"word": "xx", "count": 5, ...}],需累加 count
- rss_count = sum(stat.get("count", 0) for stat in rss_items) if rss_items else 0
- if (
- cfg["ENABLE_NOTIFICATION"]
- and has_notification
- and has_any_content
- ):
- # 输出推送内容统计
- content_parts = []
- if news_count > 0:
- content_parts.append(f"热榜 {news_count} 条")
- if rss_count > 0:
- content_parts.append(f"RSS {rss_count} 条")
- total_count = news_count + rss_count
- print(f"[推送] 准备发送:{' + '.join(content_parts)},合计 {total_count} 条")
- # 推送窗口控制
- if cfg["PUSH_WINDOW"]["ENABLED"]:
- push_manager = self.ctx.create_push_manager()
- time_range_start = cfg["PUSH_WINDOW"]["TIME_RANGE"]["START"]
- time_range_end = cfg["PUSH_WINDOW"]["TIME_RANGE"]["END"]
- if not push_manager.is_in_time_range(time_range_start, time_range_end):
- now = self.ctx.get_time()
- print(
- f"推送窗口控制:当前时间 {now.strftime('%H:%M')} 不在推送时间窗口 {time_range_start}-{time_range_end} 内,跳过推送"
- )
- return False
- if cfg["PUSH_WINDOW"]["ONCE_PER_DAY"]:
- if push_manager.has_pushed_today():
- print(f"推送窗口控制:今天已推送过,跳过本次推送")
- return False
- else:
- print(f"推送窗口控制:今天首次推送")
- # AI 分析:优先使用传入的结果,避免重复分析
- if ai_result is None:
- ai_config = cfg.get("AI_ANALYSIS", {})
- if ai_config.get("ENABLED", False):
- ai_result = self._run_ai_analysis(
- stats, rss_items, mode, report_type, id_to_name, current_results=current_results
- )
- # 准备报告数据
- report_data = self.ctx.prepare_report(stats, failed_ids, new_titles, id_to_name, mode)
- # 是否发送版本更新信息
- update_info_to_send = self.update_info if cfg["SHOW_VERSION_UPDATE"] else None
- # 使用 NotificationDispatcher 发送到所有渠道(合并热榜+RSS+AI分析+独立展示区)
- dispatcher = self.ctx.create_notification_dispatcher()
- results = dispatcher.dispatch_all(
- report_data=report_data,
- report_type=report_type,
- update_info=update_info_to_send,
- proxy_url=self.proxy_url,
- mode=mode,
- html_file_path=html_file_path,
- rss_items=rss_items,
- rss_new_items=rss_new_items,
- ai_analysis=ai_result,
- standalone_data=standalone_data,
- )
- if not results:
- print("未配置任何通知渠道,跳过通知发送")
- return False
- # 如果成功发送了任何通知,且启用了每天只推一次,则记录推送
- if (
- cfg["PUSH_WINDOW"]["ENABLED"]
- and cfg["PUSH_WINDOW"]["ONCE_PER_DAY"]
- and any(results.values())
- ):
- push_manager = self.ctx.create_push_manager()
- push_manager.record_push(report_type)
- return True
- elif cfg["ENABLE_NOTIFICATION"] and not has_notification:
- print("⚠️ 警告:通知功能已启用但未配置任何通知渠道,将跳过通知发送")
- elif not cfg["ENABLE_NOTIFICATION"]:
- print(f"跳过{report_type}通知:通知功能已禁用")
- elif (
- cfg["ENABLE_NOTIFICATION"]
- and has_notification
- and not has_any_content
- ):
- mode_strategy = self._get_mode_strategy()
- if self.report_mode == "incremental":
- if not has_rss_content:
- print("跳过通知:增量模式下未检测到匹配的新闻和RSS")
- else:
- print("跳过通知:增量模式下新闻未匹配到关键词")
- else:
- print(
- f"跳过通知:{mode_strategy['mode_name']}下未检测到匹配的新闻"
- )
- return False
- def _initialize_and_check_config(self) -> None:
- """通用初始化和配置检查"""
- now = self.ctx.get_time()
- print(f"当前北京时间: {now.strftime('%Y-%m-%d %H:%M:%S')}")
- if not self.ctx.config["ENABLE_CRAWLER"]:
- print("爬虫功能已禁用(ENABLE_CRAWLER=False),程序退出")
- return
- has_notification = self._has_notification_configured()
- if not self.ctx.config["ENABLE_NOTIFICATION"]:
- print("通知功能已禁用(ENABLE_NOTIFICATION=False),将只进行数据抓取")
- elif not has_notification:
- print("未配置任何通知渠道,将只进行数据抓取,不发送通知")
- else:
- print("通知功能已启用,将发送通知")
- mode_strategy = self._get_mode_strategy()
- print(f"报告模式: {self.report_mode}")
- print(f"运行模式: {mode_strategy['description']}")
- def _crawl_data(self) -> Tuple[Dict, Dict, List]:
- """执行数据爬取"""
- ids = []
- for platform in self.ctx.platforms:
- if "name" in platform:
- ids.append((platform["id"], platform["name"]))
- else:
- ids.append(platform["id"])
- print(
- f"配置的监控平台: {[p.get('name', p['id']) for p in self.ctx.platforms]}"
- )
- print(f"开始爬取数据,请求间隔 {self.request_interval} 毫秒")
- Path("output").mkdir(parents=True, exist_ok=True)
- results, id_to_name, failed_ids = self.data_fetcher.crawl_websites(
- ids, self.request_interval
- )
- # 转换为 NewsData 格式并保存到存储后端
- crawl_time = self.ctx.format_time()
- crawl_date = self.ctx.format_date()
- news_data = convert_crawl_results_to_news_data(
- results, id_to_name, failed_ids, crawl_time, crawl_date
- )
- # 保存到存储后端(SQLite)
- if self.storage_manager.save_news_data(news_data):
- print(f"数据已保存到存储后端: {self.storage_manager.backend_name}")
- # 保存 TXT 快照(如果启用)
- txt_file = self.storage_manager.save_txt_snapshot(news_data)
- if txt_file:
- print(f"TXT 快照已保存: {txt_file}")
- # 兼容:同时保存到原有 TXT 格式(确保向后兼容)
- if self.ctx.config["STORAGE"]["FORMATS"]["TXT"]:
- title_file = self.ctx.save_titles(results, id_to_name, failed_ids)
- print(f"标题已保存到: {title_file}")
- return results, id_to_name, failed_ids
- def _crawl_rss_data(self) -> Tuple[Optional[List[Dict]], Optional[List[Dict]], Optional[List[Dict]]]:
- """
- 执行 RSS 数据抓取
- Returns:
- (rss_items, rss_new_items, raw_rss_items) 元组:
- - rss_items: 统计条目列表(按模式处理,用于统计区块)
- - rss_new_items: 新增条目列表(用于新增区块)
- - raw_rss_items: 原始 RSS 条目列表(用于独立展示区)
- 如果未启用或失败返回 (None, None, None)
- """
- if not self.ctx.rss_enabled:
- return None, None, None
- rss_feeds = self.ctx.rss_feeds
- if not rss_feeds:
- print("[RSS] 未配置任何 RSS 源")
- return None, None, None
- try:
- from trendradar.crawler.rss import RSSFetcher, RSSFeedConfig
- # 构建 RSS 源配置
- feeds = []
- for feed_config in rss_feeds:
- # 读取并验证单个 feed 的 max_age_days(可选)
- max_age_days_raw = feed_config.get("max_age_days")
- max_age_days = None
- if max_age_days_raw is not None:
- try:
- max_age_days = int(max_age_days_raw)
- if max_age_days < 0:
- feed_id = feed_config.get("id", "unknown")
- print(f"[警告] RSS feed '{feed_id}' 的 max_age_days 为负数,将使用全局默认值")
- max_age_days = None
- except (ValueError, TypeError):
- feed_id = feed_config.get("id", "unknown")
- print(f"[警告] RSS feed '{feed_id}' 的 max_age_days 格式错误:{max_age_days_raw}")
- max_age_days = None
- feed = RSSFeedConfig(
- id=feed_config.get("id", ""),
- name=feed_config.get("name", ""),
- url=feed_config.get("url", ""),
- max_items=feed_config.get("max_items", 50),
- enabled=feed_config.get("enabled", True),
- max_age_days=max_age_days, # None=使用全局,0=禁用,>0=覆盖
- )
- if feed.id and feed.url and feed.enabled:
- feeds.append(feed)
- if not feeds:
- print("[RSS] 没有启用的 RSS 源")
- return None, None, None
- # 创建抓取器
- rss_config = self.ctx.rss_config
- # RSS 代理:优先使用 RSS 专属代理,否则使用爬虫默认代理
- rss_proxy_url = rss_config.get("PROXY_URL", "") or self.proxy_url or ""
- # 获取配置的时区
- timezone = self.ctx.config.get("TIMEZONE", "Asia/Shanghai")
- # 获取新鲜度过滤配置
- freshness_config = rss_config.get("FRESHNESS_FILTER", {})
- freshness_enabled = freshness_config.get("ENABLED", True)
- default_max_age_days = freshness_config.get("MAX_AGE_DAYS", 3)
- fetcher = RSSFetcher(
- feeds=feeds,
- request_interval=rss_config.get("REQUEST_INTERVAL", 2000),
- timeout=rss_config.get("TIMEOUT", 15),
- use_proxy=rss_config.get("USE_PROXY", False),
- proxy_url=rss_proxy_url,
- timezone=timezone,
- freshness_enabled=freshness_enabled,
- default_max_age_days=default_max_age_days,
- )
- # 抓取数据
- rss_data = fetcher.fetch_all()
- # 保存到存储后端
- if self.storage_manager.save_rss_data(rss_data):
- print(f"[RSS] 数据已保存到存储后端")
- # 处理 RSS 数据(按模式过滤)并返回用于合并推送
- return self._process_rss_data_by_mode(rss_data)
- else:
- print(f"[RSS] 数据保存失败")
- return None, None, None
- except ImportError as e:
- print(f"[RSS] 缺少依赖: {e}")
- print("[RSS] 请安装 feedparser: pip install feedparser")
- return None, None, None
- except Exception as e:
- print(f"[RSS] 抓取失败: {e}")
- return None, None, None
- def _process_rss_data_by_mode(self, rss_data) -> Tuple[Optional[List[Dict]], Optional[List[Dict]], Optional[List[Dict]]]:
- """
- 按报告模式处理 RSS 数据,返回与热榜相同格式的统计结构
- 三种模式:
- - daily: 当日汇总,统计=当天所有条目,新增=本次新增条目
- - current: 当前榜单,统计=当前榜单条目,新增=本次新增条目
- - incremental: 增量模式,统计=新增条目,新增=无
- Args:
- rss_data: 当前抓取的 RSSData 对象
- Returns:
- (rss_stats, rss_new_stats, raw_rss_items) 元组:
- - rss_stats: RSS 关键词统计列表(与热榜 stats 格式一致)
- - rss_new_stats: RSS 新增关键词统计列表(与热榜 stats 格式一致)
- - raw_rss_items: 原始 RSS 条目列表(用于独立展示区)
- """
- from trendradar.core.analyzer import count_rss_frequency
- # 从 display.regions.rss 统一控制 RSS 分析和展示
- rss_display_enabled = self.ctx.config.get("DISPLAY", {}).get("REGIONS", {}).get("RSS", True)
- # 加载关键词配置
- try:
- word_groups, filter_words, global_filters = self.ctx.load_frequency_words()
- except FileNotFoundError:
- word_groups, filter_words, global_filters = [], [], []
- timezone = self.ctx.timezone
- max_news_per_keyword = self.ctx.config.get("MAX_NEWS_PER_KEYWORD", 0)
- sort_by_position_first = self.ctx.config.get("SORT_BY_POSITION_FIRST", False)
- rss_stats = None
- rss_new_stats = None
- raw_rss_items = None # 原始 RSS 条目列表(用于独立展示区)
- # 1. 首先获取原始条目(用于独立展示区,不受 display.regions.rss 影响)
- # 根据模式获取原始条目
- if self.report_mode == "incremental":
- new_items_dict = self.storage_manager.detect_new_rss_items(rss_data)
- if new_items_dict:
- raw_rss_items = self._convert_rss_items_to_list(new_items_dict, rss_data.id_to_name)
- elif self.report_mode == "current":
- latest_data = self.storage_manager.get_latest_rss_data(rss_data.date)
- if latest_data:
- raw_rss_items = self._convert_rss_items_to_list(latest_data.items, latest_data.id_to_name)
- else: # daily
- all_data = self.storage_manager.get_rss_data(rss_data.date)
- if all_data:
- raw_rss_items = self._convert_rss_items_to_list(all_data.items, all_data.id_to_name)
- # 如果 RSS 展示未启用,跳过关键词分析,只返回原始条目用于独立展示区
- if not rss_display_enabled:
- return None, None, raw_rss_items
- # 2. 获取新增条目(用于统计)
- new_items_dict = self.storage_manager.detect_new_rss_items(rss_data)
- new_items_list = None
- if new_items_dict:
- new_items_list = self._convert_rss_items_to_list(new_items_dict, rss_data.id_to_name)
- if new_items_list:
- print(f"[RSS] 检测到 {len(new_items_list)} 条新增")
- # 3. 根据模式获取统计条目
- if self.report_mode == "incremental":
- # 增量模式:统计条目就是新增条目
- if not new_items_list:
- print("[RSS] 增量模式:没有新增 RSS 条目")
- return None, None, raw_rss_items
- rss_stats, total = count_rss_frequency(
- rss_items=new_items_list,
- word_groups=word_groups,
- filter_words=filter_words,
- global_filters=global_filters,
- new_items=new_items_list, # 增量模式所有都是新增
- max_news_per_keyword=max_news_per_keyword,
- sort_by_position_first=sort_by_position_first,
- timezone=timezone,
- rank_threshold=self.rank_threshold,
- quiet=False,
- )
- if not rss_stats:
- print("[RSS] 增量模式:关键词匹配后没有内容")
- # 即使关键词匹配为空,也返回原始条目用于独立展示区
- return None, None, raw_rss_items
- elif self.report_mode == "current":
- # 当前榜单模式:统计=当前榜单所有条目
- # raw_rss_items 已在前面获取
- if not raw_rss_items:
- print("[RSS] 当前榜单模式:没有 RSS 数据")
- return None, None, None
- rss_stats, total = count_rss_frequency(
- rss_items=raw_rss_items,
- word_groups=word_groups,
- filter_words=filter_words,
- global_filters=global_filters,
- new_items=new_items_list, # 标记新增
- max_news_per_keyword=max_news_per_keyword,
- sort_by_position_first=sort_by_position_first,
- timezone=timezone,
- rank_threshold=self.rank_threshold,
- quiet=False,
- )
- if not rss_stats:
- print("[RSS] 当前榜单模式:关键词匹配后没有内容")
- # 即使关键词匹配为空,也返回原始条目用于独立展示区
- return None, None, raw_rss_items
- # 生成新增统计
- if new_items_list:
- rss_new_stats, _ = count_rss_frequency(
- rss_items=new_items_list,
- word_groups=word_groups,
- filter_words=filter_words,
- global_filters=global_filters,
- new_items=new_items_list,
- max_news_per_keyword=max_news_per_keyword,
- sort_by_position_first=sort_by_position_first,
- timezone=timezone,
- rank_threshold=self.rank_threshold,
- quiet=True,
- )
- else:
- # daily 模式:统计=当天所有条目
- # raw_rss_items 已在前面获取
- if not raw_rss_items:
- print("[RSS] 当日汇总模式:没有 RSS 数据")
- return None, None, None
- rss_stats, total = count_rss_frequency(
- rss_items=raw_rss_items,
- word_groups=word_groups,
- filter_words=filter_words,
- global_filters=global_filters,
- new_items=new_items_list, # 标记新增
- max_news_per_keyword=max_news_per_keyword,
- sort_by_position_first=sort_by_position_first,
- timezone=timezone,
- rank_threshold=self.rank_threshold,
- quiet=False,
- )
- if not rss_stats:
- print("[RSS] 当日汇总模式:关键词匹配后没有内容")
- # 即使关键词匹配为空,也返回原始条目用于独立展示区
- return None, None, raw_rss_items
- # 生成新增统计
- if new_items_list:
- rss_new_stats, _ = count_rss_frequency(
- rss_items=new_items_list,
- word_groups=word_groups,
- filter_words=filter_words,
- global_filters=global_filters,
- new_items=new_items_list,
- max_news_per_keyword=max_news_per_keyword,
- sort_by_position_first=sort_by_position_first,
- timezone=timezone,
- rank_threshold=self.rank_threshold,
- quiet=True,
- )
- return rss_stats, rss_new_stats, raw_rss_items
- def _convert_rss_items_to_list(self, items_dict: Dict, id_to_name: Dict) -> List[Dict]:
- """将 RSS 条目字典转换为列表格式,并应用新鲜度过滤(用于推送)"""
- rss_items = []
- filtered_count = 0
- # 获取新鲜度过滤配置
- rss_config = self.ctx.rss_config
- freshness_config = rss_config.get("FRESHNESS_FILTER", {})
- freshness_enabled = freshness_config.get("ENABLED", True)
- default_max_age_days = freshness_config.get("MAX_AGE_DAYS", 3)
- timezone = self.ctx.config.get("TIMEZONE", "Asia/Shanghai")
- # 构建 feed_id -> max_age_days 的映射
- feed_max_age_map = {}
- for feed_cfg in self.ctx.rss_feeds:
- feed_id = feed_cfg.get("id", "")
- max_age = feed_cfg.get("max_age_days")
- if max_age is not None:
- try:
- feed_max_age_map[feed_id] = int(max_age)
- except (ValueError, TypeError):
- pass
- for feed_id, items in items_dict.items():
- # 确定此 feed 的 max_age_days
- max_days = feed_max_age_map.get(feed_id)
- if max_days is None:
- max_days = default_max_age_days
- for item in items:
- # 应用新鲜度过滤(仅在启用时)
- if freshness_enabled and max_days > 0:
- if item.published_at and not is_within_days(item.published_at, max_days, timezone):
- filtered_count += 1
- continue # 跳过超过指定天数的文章
- rss_items.append({
- "title": item.title,
- "feed_id": feed_id,
- "feed_name": id_to_name.get(feed_id, feed_id),
- "url": item.url,
- "published_at": item.published_at,
- "summary": item.summary,
- "author": item.author,
- })
- # 输出过滤统计
- if filtered_count > 0:
- print(f"[RSS] 新鲜度过滤:跳过 {filtered_count} 篇超过指定天数的旧文章(仍保留在数据库中)")
- return rss_items
- def _filter_rss_by_keywords(self, rss_items: List[Dict]) -> List[Dict]:
- """使用 frequency_words.txt 过滤 RSS 条目"""
- try:
- word_groups, filter_words, global_filters = self.ctx.load_frequency_words()
- if word_groups or filter_words or global_filters:
- from trendradar.core.frequency import matches_word_groups
- filtered_items = []
- for item in rss_items:
- title = item.get("title", "")
- if matches_word_groups(title, word_groups, filter_words, global_filters):
- filtered_items.append(item)
- original_count = len(rss_items)
- rss_items = filtered_items
- print(f"[RSS] 关键词过滤后剩余 {len(rss_items)}/{original_count} 条")
- if not rss_items:
- print("[RSS] 关键词过滤后没有匹配内容")
- return []
- except FileNotFoundError:
- # frequency_words.txt 不存在时跳过过滤
- pass
- return rss_items
- def _generate_rss_html_report(self, rss_items: list, feeds_info: dict) -> str:
- """生成 RSS HTML 报告"""
- try:
- from trendradar.report.rss_html import render_rss_html_content
- from pathlib import Path
- html_content = render_rss_html_content(
- rss_items=rss_items,
- total_count=len(rss_items),
- feeds_info=feeds_info,
- get_time_func=self.ctx.get_time,
- )
- # 保存 HTML 文件(扁平化结构:output/html/日期/)
- date_folder = self.ctx.format_date()
- time_filename = self.ctx.format_time()
- output_dir = Path("output") / "html" / date_folder
- output_dir.mkdir(parents=True, exist_ok=True)
- file_path = output_dir / f"rss_{time_filename}.html"
- with open(file_path, "w", encoding="utf-8") as f:
- f.write(html_content)
- print(f"[RSS] HTML 报告已生成: {file_path}")
- return str(file_path)
- except Exception as e:
- print(f"[RSS] 生成 HTML 报告失败: {e}")
- return None
- def _execute_mode_strategy(
- self, mode_strategy: Dict, results: Dict, id_to_name: Dict, failed_ids: List,
- rss_items: Optional[List[Dict]] = None,
- rss_new_items: Optional[List[Dict]] = None,
- raw_rss_items: Optional[List[Dict]] = None,
- ) -> Optional[str]:
- """执行模式特定逻辑,支持热榜+RSS合并推送
- 简化后的逻辑:
- - 每次运行都生成 HTML 报告(时间戳快照 + latest/{mode}.html + index.html)
- - 根据模式发送通知
- """
- # 获取当前监控平台ID列表
- current_platform_ids = self.ctx.platform_ids
- new_titles = self.ctx.detect_new_titles(current_platform_ids)
- time_info = self.ctx.format_time()
- if self.ctx.config["STORAGE"]["FORMATS"]["TXT"]:
- self.ctx.save_titles(results, id_to_name, failed_ids)
- word_groups, filter_words, global_filters = self.ctx.load_frequency_words()
- html_file = None
- stats = []
- ai_result = None
- title_info = None
- # current 模式需要使用完整的历史数据
- if self.report_mode == "current":
- analysis_data = self._load_analysis_data()
- if analysis_data:
- (
- all_results,
- historical_id_to_name,
- historical_title_info,
- historical_new_titles,
- _,
- _,
- _,
- ) = analysis_data
- print(
- f"current模式:使用过滤后的历史数据,包含平台:{list(all_results.keys())}"
- )
- # 使用历史数据准备独立展示区数据(包含完整的 title_info)
- standalone_data = self._prepare_standalone_data(
- all_results, historical_id_to_name, historical_title_info, raw_rss_items
- )
- stats, html_file, ai_result = self._run_analysis_pipeline(
- all_results,
- self.report_mode,
- historical_title_info,
- historical_new_titles,
- word_groups,
- filter_words,
- historical_id_to_name,
- failed_ids=failed_ids,
- global_filters=global_filters,
- rss_items=rss_items,
- rss_new_items=rss_new_items,
- standalone_data=standalone_data,
- )
- combined_id_to_name = {**historical_id_to_name, **id_to_name}
- new_titles = historical_new_titles
- id_to_name = combined_id_to_name
- title_info = historical_title_info
- results = all_results
- else:
- print("❌ 严重错误:无法读取刚保存的数据文件")
- raise RuntimeError("数据一致性检查失败:保存后立即读取失败")
- elif self.report_mode == "daily":
- # daily 模式:使用全天累计数据
- analysis_data = self._load_analysis_data()
- if analysis_data:
- (
- all_results,
- historical_id_to_name,
- historical_title_info,
- historical_new_titles,
- _,
- _,
- _,
- ) = analysis_data
- # 使用历史数据准备独立展示区数据(包含完整的 title_info)
- standalone_data = self._prepare_standalone_data(
- all_results, historical_id_to_name, historical_title_info, raw_rss_items
- )
- stats, html_file, ai_result = self._run_analysis_pipeline(
- all_results,
- self.report_mode,
- historical_title_info,
- historical_new_titles,
- word_groups,
- filter_words,
- historical_id_to_name,
- failed_ids=failed_ids,
- global_filters=global_filters,
- rss_items=rss_items,
- rss_new_items=rss_new_items,
- standalone_data=standalone_data,
- )
- combined_id_to_name = {**historical_id_to_name, **id_to_name}
- new_titles = historical_new_titles
- id_to_name = combined_id_to_name
- title_info = historical_title_info
- results = all_results
- else:
- # 没有历史数据时使用当前数据
- title_info = self._prepare_current_title_info(results, time_info)
- standalone_data = self._prepare_standalone_data(
- results, id_to_name, title_info, raw_rss_items
- )
- stats, html_file, ai_result = self._run_analysis_pipeline(
- results,
- self.report_mode,
- title_info,
- new_titles,
- word_groups,
- filter_words,
- id_to_name,
- failed_ids=failed_ids,
- global_filters=global_filters,
- rss_items=rss_items,
- rss_new_items=rss_new_items,
- standalone_data=standalone_data,
- )
- else:
- # incremental 模式:只使用当前抓取的数据
- title_info = self._prepare_current_title_info(results, time_info)
- standalone_data = self._prepare_standalone_data(
- results, id_to_name, title_info, raw_rss_items
- )
- stats, html_file, ai_result = self._run_analysis_pipeline(
- results,
- self.report_mode,
- title_info,
- new_titles,
- word_groups,
- filter_words,
- id_to_name,
- failed_ids=failed_ids,
- global_filters=global_filters,
- rss_items=rss_items,
- rss_new_items=rss_new_items,
- standalone_data=standalone_data,
- )
- if html_file:
- print(f"HTML报告已生成: {html_file}")
- print(f"最新报告已更新: output/html/latest/{self.report_mode}.html")
- # 发送通知
- if mode_strategy["should_send_notification"]:
- standalone_data = self._prepare_standalone_data(
- results, id_to_name, title_info, raw_rss_items
- )
- self._send_notification_if_needed(
- stats,
- mode_strategy["report_type"],
- self.report_mode,
- failed_ids=failed_ids,
- new_titles=new_titles,
- id_to_name=id_to_name,
- html_file_path=html_file,
- rss_items=rss_items,
- rss_new_items=rss_new_items,
- standalone_data=standalone_data,
- ai_result=ai_result,
- current_results=results,
- )
- # 打开浏览器(仅在非容器环境)
- if self._should_open_browser() and html_file:
- file_url = "file://" + str(Path(html_file).resolve())
- print(f"正在打开HTML报告: {file_url}")
- webbrowser.open(file_url)
- elif self.is_docker_container and html_file:
- print(f"HTML报告已生成(Docker环境): {html_file}")
- return html_file
- def run(self) -> None:
- """执行分析流程"""
- try:
- self._initialize_and_check_config()
- mode_strategy = self._get_mode_strategy()
- # 抓取热榜数据
- results, id_to_name, failed_ids = self._crawl_data()
- # 抓取 RSS 数据(如果启用),返回统计条目、新增条目和原始条目
- rss_items, rss_new_items, raw_rss_items = self._crawl_rss_data()
- # 执行模式策略,传递 RSS 数据用于合并推送
- self._execute_mode_strategy(
- mode_strategy, results, id_to_name, failed_ids,
- rss_items=rss_items, rss_new_items=rss_new_items,
- raw_rss_items=raw_rss_items
- )
- except Exception as e:
- print(f"分析流程执行出错: {e}")
- if self.ctx.config.get("DEBUG", False):
- raise
- finally:
- # 清理资源(包括过期数据清理和数据库连接关闭)
- self.ctx.cleanup()
- def main():
- """主程序入口"""
- debug_mode = False
- try:
- # 先加载配置以获取 version_check_url
- config = load_config()
- version_url = config.get("VERSION_CHECK_URL", "")
- configs_version_url = config.get("CONFIGS_VERSION_CHECK_URL", "")
- # 统一版本检查(程序版本 + 配置文件版本,只请求一次远程)
- need_update = False
- remote_version = None
- if version_url:
- need_update, remote_version = check_all_versions(version_url, configs_version_url)
- # 复用已加载的配置,避免重复加载
- analyzer = NewsAnalyzer(config=config)
- # 设置更新信息(复用已获取的远程版本,不再重复请求)
- if analyzer.is_github_actions and need_update and remote_version:
- analyzer.update_info = {
- "current_version": __version__,
- "remote_version": remote_version,
- }
- # 获取 debug 配置
- debug_mode = analyzer.ctx.config.get("DEBUG", False)
- analyzer.run()
- except FileNotFoundError as e:
- print(f"❌ 配置文件错误: {e}")
- print("\n请确保以下文件存在:")
- print(" • config/config.yaml")
- print(" • config/frequency_words.txt")
- print("\n参考项目文档进行正确配置")
- except Exception as e:
- print(f"❌ 程序运行错误: {e}")
- if debug_mode:
- raise
- if __name__ == "__main__":
- main()
|