init: TrendRadar v6.10.0 with custom config

This commit is contained in:
2026-08-27 19:32:53 +08:00
commit cb2b0ceef8
182 changed files with 61644 additions and 0 deletions
+49
View File
@@ -0,0 +1,49 @@
# coding=utf-8
"""
核心模块 - 配置管理和核心工具
"""
from trendradar.core.config import (
parse_multi_account_config,
validate_paired_configs,
limit_accounts,
get_account_at_index,
)
from trendradar.core.loader import load_config
from trendradar.core.frequency import load_frequency_words, matches_word_groups
from trendradar.core.scheduler import Scheduler, ResolvedSchedule
from trendradar.core.data import (
read_all_today_titles_from_storage,
read_all_today_titles,
detect_latest_new_titles_from_storage,
detect_latest_new_titles,
)
from trendradar.core.analyzer import (
calculate_news_weight,
format_time_display,
count_word_frequency,
count_rss_frequency,
)
__all__ = [
"parse_multi_account_config",
"validate_paired_configs",
"limit_accounts",
"get_account_at_index",
"load_config",
"load_frequency_words",
"matches_word_groups",
# 数据处理
"read_all_today_titles_from_storage",
"read_all_today_titles",
"detect_latest_new_titles_from_storage",
"detect_latest_new_titles",
# 统计分析
"calculate_news_weight",
"format_time_display",
"count_word_frequency",
"count_rss_frequency",
# 调度器
"Scheduler",
"ResolvedSchedule",
]
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+779
View File
@@ -0,0 +1,779 @@
# coding=utf-8
"""
统计分析模块
提供新闻统计和分析功能:
- calculate_news_weight: 计算新闻权重
- format_time_display: 格式化时间显示
- count_word_frequency: 统计词频
"""
from typing import Dict, List, Tuple, Optional, Callable
from trendradar.core.frequency import matches_word_groups, _word_matches
from trendradar.utils.time import DEFAULT_TIMEZONE
def calculate_news_weight(
title_data: Dict,
rank_threshold: int,
weight_config: Dict,
) -> float:
"""
计算新闻权重,用于排序
Args:
title_data: 标题数据,包含 ranks 和 count
rank_threshold: 排名阈值
weight_config: 权重配置 {RANK_WEIGHT, FREQUENCY_WEIGHT, HOTNESS_WEIGHT}
Returns:
float: 计算出的权重值
"""
ranks = title_data.get("ranks", [])
if not ranks:
return 0.0
count = title_data.get("count", len(ranks))
# 单次遍历计算排名分数总和与高排名次数
rank_score_sum = 0
high_rank_count = 0
for rank in ranks:
rank_score_sum += 11 - min(rank, 10)
if rank <= rank_threshold:
high_rank_count += 1
# 归一化到 0~100(与 frequency_weight、hotness_weight 量纲对齐)
rank_weight = (rank_score_sum / len(ranks)) * 10
# 频次权重:min(出现次数, 10) × 10
frequency_weight = min(count, 10) * 10
# 热度加成:高排名次数 / 总出现次数 × 100
hotness_ratio = high_rank_count / len(ranks)
hotness_weight = hotness_ratio * 100
total_weight = (
rank_weight * weight_config["RANK_WEIGHT"]
+ frequency_weight * weight_config["FREQUENCY_WEIGHT"]
+ hotness_weight * weight_config["HOTNESS_WEIGHT"]
)
return total_weight
def format_time_display(
first_time: str,
last_time: str,
convert_time_func: Callable[[str], str],
) -> str:
"""
格式化时间显示(将 HH-MM 转换为 HH:MM)
Args:
first_time: 首次出现时间
last_time: 最后出现时间
convert_time_func: 时间格式转换函数
Returns:
str: 格式化后的时间显示字符串
"""
if not first_time:
return ""
# 转换为显示格式
first_display = convert_time_func(first_time)
last_display = convert_time_func(last_time)
if first_display == last_display or not last_display:
return first_display
else:
return f"[{first_display} ~ {last_display}]"
def count_word_frequency(
results: Dict,
word_groups: List[Dict],
filter_words: List[str],
id_to_name: Dict,
title_info: Optional[Dict] = None,
rank_threshold: int = 3,
new_titles: Optional[Dict] = None,
mode: str = "daily",
global_filters: Optional[List[str]] = None,
weight_config: Optional[Dict] = None,
max_news_per_keyword: int = 0,
sort_by_position_first: bool = False,
is_first_crawl_func: Optional[Callable[[], bool]] = None,
convert_time_func: Optional[Callable[[str], str]] = None,
quiet: bool = False,
) -> Tuple[List[Dict], int]:
"""
统计词频,支持必须词、频率词、过滤词、全局过滤词,并标记新增标题
Args:
results: 抓取结果 {source_id: {title: title_data}}
word_groups: 词组配置列表
filter_words: 过滤词列表
id_to_name: ID 到名称的映射
title_info: 标题统计信息(可选)
rank_threshold: 排名阈值
new_titles: 新增标题(可选)
mode: 报告模式 (daily/incremental/current)
global_filters: 全局过滤词(可选)
weight_config: 权重配置
max_news_per_keyword: 每个关键词最大显示数量
sort_by_position_first: 是否优先按配置位置排序
is_first_crawl_func: 检测是否是当天第一次爬取的函数
convert_time_func: 时间格式转换函数
quiet: 是否静默模式(不打印日志)
Returns:
Tuple[List[Dict], int]: (统计结果列表, 总标题数)
"""
# 默认权重配置
if weight_config is None:
weight_config = {
"RANK_WEIGHT": 0.6,
"FREQUENCY_WEIGHT": 0.3,
"HOTNESS_WEIGHT": 0.1,
}
# 默认时间转换函数
if convert_time_func is None:
convert_time_func = lambda x: x
# 默认首次爬取检测函数
if is_first_crawl_func is None:
is_first_crawl_func = lambda: True
# 如果没有配置词组,创建一个包含所有新闻的虚拟词组
if not word_groups:
print("频率词配置为空,将显示所有新闻")
word_groups = [{"required": [], "normal": [], "group_key": "全部新闻"}]
filter_words = [] # 清空过滤词,显示所有新闻
is_first_today = is_first_crawl_func()
# 确定处理的数据源和新增标记逻辑
if mode == "incremental":
if is_first_today:
# 增量模式 + 当天第一次:处理所有新闻,都标记为新增
results_to_process = results
all_news_are_new = True
else:
# 增量模式 + 当天非第一次:只处理新增的新闻
results_to_process = new_titles if new_titles else {}
all_news_are_new = True
elif mode == "current":
# current 模式:只处理当前时间批次的新闻,但统计信息来自全部历史
if title_info:
latest_time = None
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
# 只处理 last_time 等于最新时间的新闻
if latest_time:
results_to_process = {}
for source_id, source_titles in results.items():
if source_id in title_info:
filtered_titles = {}
for title, title_data in source_titles.items():
if title in title_info[source_id]:
info = title_info[source_id][title]
if info.get("last_time") == latest_time:
filtered_titles[title] = title_data
if filtered_titles:
results_to_process[source_id] = filtered_titles
if not quiet:
print(
f"当前榜单模式:最新时间 {latest_time},筛选出 {sum(len(titles) for titles in results_to_process.values())} 条当前榜单新闻"
)
else:
results_to_process = results
else:
results_to_process = results
all_news_are_new = False
else:
# 当日汇总模式:处理所有新闻
results_to_process = results
all_news_are_new = False
total_input_news = sum(len(titles) for titles in results.values())
filter_status = (
"全部显示"
if len(word_groups) == 1 and word_groups[0]["group_key"] == "全部新闻"
else "频率词过滤"
)
print(f"当日汇总模式:处理 {total_input_news} 条新闻,模式:{filter_status}")
word_stats = {}
total_titles = 0
processed_titles = {}
matched_new_count = 0
if title_info is None:
title_info = {}
if new_titles is None:
new_titles = {}
for group in word_groups:
group_key = group["group_key"]
word_stats[group_key] = {"count": 0, "titles": {}}
for source_id, titles_data in results_to_process.items():
total_titles += len(titles_data)
if source_id not in processed_titles:
processed_titles[source_id] = {}
for title, title_data in titles_data.items():
if title in processed_titles.get(source_id, {}):
continue
# 使用统一的匹配逻辑
matches_frequency_words = matches_word_groups(
title, word_groups, filter_words, global_filters
)
if not matches_frequency_words:
continue
# 如果是增量模式或 current 模式第一次,统计匹配的新增新闻数量
if (mode == "incremental" and all_news_are_new) or (
mode == "current" and is_first_today
):
matched_new_count += 1
source_ranks = title_data.get("ranks", [])
source_url = title_data.get("url", "")
source_mobile_url = title_data.get("mobileUrl", "")
# 找到匹配的词组(防御性转换确保类型安全)
title_lower = str(title).lower() if not isinstance(title, str) else title.lower()
for group in word_groups:
required_words = group["required"]
normal_words = group["normal"]
# 如果是"全部新闻"模式,所有标题都匹配第一个(唯一的)词组
if len(word_groups) == 1 and word_groups[0]["group_key"] == "全部新闻":
group_key = group["group_key"]
word_stats[group_key]["count"] += 1
if source_id not in word_stats[group_key]["titles"]:
word_stats[group_key]["titles"][source_id] = []
else:
# 原有的匹配逻辑(支持正则语法)
if required_words:
all_required_present = all(
_word_matches(req_item, title_lower)
for req_item in required_words
)
if not all_required_present:
continue
if normal_words:
any_normal_present = any(
_word_matches(normal_item, title_lower)
for normal_item in normal_words
)
if not any_normal_present:
continue
group_key = group["group_key"]
word_stats[group_key]["count"] += 1
if source_id not in word_stats[group_key]["titles"]:
word_stats[group_key]["titles"][source_id] = []
first_time = ""
last_time = ""
count_info = 1
ranks = source_ranks if source_ranks else []
url = source_url
mobile_url = source_mobile_url
rank_timeline = []
# 对于 current 模式,从历史统计信息中获取完整数据
if (
mode == "current"
and title_info
and source_id in title_info
and title in title_info[source_id]
):
info = title_info[source_id][title]
first_time = info.get("first_time", "")
last_time = info.get("last_time", "")
count_info = info.get("count", 1)
if "ranks" in info and info["ranks"]:
ranks = info["ranks"]
url = info.get("url", source_url)
mobile_url = info.get("mobileUrl", source_mobile_url)
rank_timeline = info.get("rank_timeline", [])
elif (
title_info
and source_id in title_info
and title in title_info[source_id]
):
info = title_info[source_id][title]
first_time = info.get("first_time", "")
last_time = info.get("last_time", "")
count_info = info.get("count", 1)
if "ranks" in info and info["ranks"]:
ranks = info["ranks"]
url = info.get("url", source_url)
mobile_url = info.get("mobileUrl", source_mobile_url)
rank_timeline = info.get("rank_timeline", [])
if not ranks:
ranks = [99]
time_display = format_time_display(first_time, last_time, convert_time_func)
source_name = id_to_name.get(source_id, source_id)
# 判断是否为新增
is_new = False
if all_news_are_new:
# 增量模式下所有处理的新闻都是新增,或者当天第一次的所有新闻都是新增
is_new = True
elif new_titles and source_id in new_titles:
# 检查是否在新增列表中
new_titles_for_source = new_titles[source_id]
is_new = title in new_titles_for_source
word_stats[group_key]["titles"][source_id].append(
{
"title": title,
"source_name": source_name,
"first_time": first_time,
"last_time": last_time,
"time_display": time_display,
"count": count_info,
"ranks": ranks,
"rank_threshold": rank_threshold,
"url": url,
"mobileUrl": mobile_url,
"is_new": is_new,
"rank_timeline": rank_timeline,
}
)
if source_id not in processed_titles:
processed_titles[source_id] = {}
processed_titles[source_id][title] = True
break
# 最后统一打印汇总信息
if mode == "incremental":
if is_first_today:
total_input_news = sum(len(titles) for titles in results.values())
filter_status = (
"全部显示"
if len(word_groups) == 1 and word_groups[0]["group_key"] == "全部新闻"
else "频率词匹配"
)
if not quiet:
print(
f"增量模式:当天第一次爬取,{total_input_news} 条新闻中有 {matched_new_count} 条{filter_status}"
)
else:
if new_titles:
total_new_count = sum(len(titles) for titles in new_titles.values())
filter_status = (
"全部显示"
if len(word_groups) == 1
and word_groups[0]["group_key"] == "全部新闻"
else "匹配频率词"
)
if not quiet:
print(
f"增量模式:{total_new_count} 条新增新闻中,有 {matched_new_count} 条{filter_status}"
)
if matched_new_count == 0 and len(word_groups) > 1:
print("增量模式:没有新增新闻匹配频率词,将不会发送通知")
else:
if not quiet:
print("增量模式:未检测到新增新闻")
elif mode == "current":
total_input_news = sum(len(titles) for titles in results_to_process.values())
if is_first_today:
filter_status = (
"全部显示"
if len(word_groups) == 1 and word_groups[0]["group_key"] == "全部新闻"
else "频率词匹配"
)
if not quiet:
print(
f"当前榜单模式:当天第一次爬取,{total_input_news} 条当前榜单新闻中有 {matched_new_count} 条{filter_status}"
)
else:
matched_count = sum(stat["count"] for stat in word_stats.values())
filter_status = (
"全部显示"
if len(word_groups) == 1 and word_groups[0]["group_key"] == "全部新闻"
else "频率词匹配"
)
if not quiet:
print(
f"当前榜单模式:{total_input_news} 条当前榜单新闻中有 {matched_count} 条{filter_status}"
)
stats = []
# 创建 group_key 到位置、最大数量、显示名称的映射
group_key_to_position = {
group["group_key"]: idx for idx, group in enumerate(word_groups)
}
group_key_to_max_count = {
group["group_key"]: group.get("max_count", 0) for group in word_groups
}
group_key_to_display_name = {
group["group_key"]: group.get("display_name") for group in word_groups
}
for group_key, data in word_stats.items():
all_titles = []
for source_id, title_list in data["titles"].items():
all_titles.extend(title_list)
# 按权重排序
sorted_titles = sorted(
all_titles,
key=lambda x: (
-calculate_news_weight(x, rank_threshold, weight_config),
min(x["ranks"]) if x["ranks"] else 999,
-x["count"],
),
)
# 应用最大显示数量限制(优先级:单独配置 > 全局配置)
group_max_count = group_key_to_max_count.get(group_key, 0)
if group_max_count == 0:
# 使用全局配置
group_max_count = max_news_per_keyword
if group_max_count > 0:
sorted_titles = sorted_titles[:group_max_count]
# 优先使用 display_name,否则使用 group_key
display_word = group_key_to_display_name.get(group_key) or group_key
stats.append(
{
"word": display_word,
"count": data["count"],
"position": group_key_to_position.get(group_key, 999),
"titles": sorted_titles,
"percentage": (
round(data["count"] / total_titles * 100, 2)
if total_titles > 0
else 0
),
}
)
# 根据配置选择排序优先级
if sort_by_position_first:
# 先按配置位置,再按热点条数
stats.sort(key=lambda x: (x["position"], -x["count"]))
else:
# 先按热点条数,再按配置位置(原逻辑)
stats.sort(key=lambda x: (-x["count"], x["position"]))
# 打印过滤后的匹配新闻数
matched_news_count = sum(len(stat["titles"]) for stat in stats if stat["count"] > 0)
if not quiet and mode == "daily":
print(f"当日汇总模式:处理 {total_titles} 条新闻,模式:频率词过滤")
print(f"频率词过滤后:{matched_news_count} 条新闻匹配")
return stats, total_titles
def count_rss_frequency(
rss_items: List[Dict],
word_groups: List[Dict],
filter_words: List[str],
global_filters: Optional[List[str]] = None,
new_items: Optional[List[Dict]] = None,
max_news_per_keyword: int = 0,
sort_by_position_first: bool = False,
timezone: str = DEFAULT_TIMEZONE,
rank_threshold: int = 5,
quiet: bool = False,
) -> Tuple[List[Dict], int]:
"""
按关键词分组统计 RSS 条目(与热榜统计格式一致)
Args:
rss_items: RSS 条目列表,每个条目包含:
- title: 标题
- feed_id: RSS 源 ID
- feed_name: RSS 源名称
- url: 文章链接
- published_at: 发布时间(ISO 格式)
word_groups: 词组配置列表
filter_words: 过滤词列表
global_filters: 全局过滤词(可选)
new_items: 新增条目列表(可选,用于标记 is_new)
max_news_per_keyword: 每个关键词最大显示数量
sort_by_position_first: 是否优先按配置位置排序
timezone: 时区名称(用于时间格式化)
quiet: 是否静默模式
Returns:
Tuple[List[Dict], int]: (统计结果列表, 总条目数)
统计结果格式与热榜一致:
[
{
"word": "关键词",
"count": 5,
"position": 0,
"titles": [
{
"title": "标题",
"source_name": "Hacker News",
"time_display": "12-29 08:20",
"count": 1,
"ranks": [1], # RSS 用发布时间顺序作为排名
"rank_threshold": 50,
"url": "...",
"mobile_url": "",
"is_new": True/False
}
],
"percentage": 10.0
}
]
"""
from trendradar.utils.time import format_iso_time_friendly
if not rss_items:
return [], 0
# 如果没有配置词组,创建一个包含所有条目的虚拟词组
if not word_groups:
if not quiet:
print("[RSS] 频率词配置为空,将显示所有 RSS 条目")
word_groups = [{"required": [], "normal": [], "group_key": "全部 RSS"}]
filter_words = []
# 创建新增条目的 URL 集合,用于快速查找
new_urls = set()
if new_items:
for item in new_items:
if item.get("url"):
new_urls.add(item["url"])
# 初始化词组统计
word_stats = {}
for group in word_groups:
group_key = group["group_key"]
word_stats[group_key] = {"count": 0, "titles": []}
total_items = len(rss_items)
processed_urls = set() # 用于去重
# 为每个条目分配一个基于发布时间的"排名"
# 按发布时间排序,最新的排在前面
sorted_items = sorted(
rss_items,
key=lambda x: x.get("published_at", ""),
reverse=True
)
url_to_rank = {item.get("url", ""): idx + 1 for idx, item in enumerate(sorted_items)}
for item in rss_items:
title = item.get("title", "")
url = item.get("url", "")
# 去重
if url and url in processed_urls:
continue
if url:
processed_urls.add(url)
# 使用统一的匹配逻辑
if not matches_word_groups(title, word_groups, filter_words, global_filters):
continue
# 找到匹配的词组
title_lower = title.lower()
for group in word_groups:
required_words = group["required"]
normal_words = group["normal"]
group_key = group["group_key"]
# "全部 RSS" 模式:所有条目都匹配
if len(word_groups) == 1 and word_groups[0]["group_key"] == "全部 RSS":
matched = True
else:
# 检查必须词(支持正则语法)
if required_words:
all_required_present = all(
_word_matches(req_item, title_lower)
for req_item in required_words
)
if not all_required_present:
continue
# 检查普通词(支持正则语法)
if normal_words:
any_normal_present = any(
_word_matches(normal_item, title_lower)
for normal_item in normal_words
)
if not any_normal_present:
continue
matched = True
if matched:
word_stats[group_key]["count"] += 1
# 格式化时间显示
published_at = item.get("published_at", "")
time_display = format_iso_time_friendly(published_at, timezone, include_date=True) if published_at else ""
# 判断是否为新增
is_new = url in new_urls if url else False
# 获取排名(基于发布时间顺序)
rank = url_to_rank.get(url, 99) if url else 99
title_data = {
"title": title,
"source_name": item.get("feed_name", item.get("feed_id", "RSS")),
"time_display": time_display,
"count": 1, # RSS 条目通常只出现一次
"ranks": [rank],
"rank_threshold": rank_threshold,
"url": url,
"mobile_url": "",
"is_new": is_new,
}
word_stats[group_key]["titles"].append(title_data)
break # 一个条目只匹配第一个词组
# 构建统计结果
stats = []
group_key_to_position = {
group["group_key"]: idx for idx, group in enumerate(word_groups)
}
group_key_to_max_count = {
group["group_key"]: group.get("max_count", 0) for group in word_groups
}
group_key_to_display_name = {
group["group_key"]: group.get("display_name") for group in word_groups
}
for group_key, data in word_stats.items():
if data["count"] == 0:
continue
# 按发布时间排序(最新在前)
sorted_titles = sorted(
data["titles"],
key=lambda x: x["ranks"][0] if x["ranks"] else 999
)
# 应用最大显示数量限制
group_max_count = group_key_to_max_count.get(group_key, 0)
if group_max_count == 0:
group_max_count = max_news_per_keyword
if group_max_count > 0:
sorted_titles = sorted_titles[:group_max_count]
# 优先使用 display_name,否则使用 group_key
display_word = group_key_to_display_name.get(group_key) or group_key
stats.append({
"word": display_word,
"count": data["count"],
"position": group_key_to_position.get(group_key, 999),
"titles": sorted_titles,
"percentage": round(data["count"] / total_items * 100, 2) if total_items > 0 else 0,
})
# 排序
if sort_by_position_first:
stats.sort(key=lambda x: (x["position"], -x["count"]))
else:
stats.sort(key=lambda x: (-x["count"], x["position"]))
matched_count = sum(stat["count"] for stat in stats)
if not quiet:
print(f"[RSS] 关键词分组统计:{matched_count}/{total_items} 条匹配")
return stats, total_items
def convert_keyword_stats_to_platform_stats(
keyword_stats: List[Dict],
weight_config: Dict,
rank_threshold: int = 5,
) -> List[Dict]:
"""
将按关键词分组的统计数据转换为按平台分组的统计数据
Args:
keyword_stats: 原始按关键词分组的统计数据
weight_config: 权重配置
rank_threshold: 排名阈值
Returns:
按平台分组的统计数据,格式与原 stats 一致
"""
# 1. 收集所有新闻,按平台分组
platform_map: Dict[str, List[Dict]] = {}
for stat in keyword_stats:
keyword = stat["word"]
for title_data in stat["titles"]:
source_name = title_data["source_name"]
if source_name not in platform_map:
platform_map[source_name] = []
# 复制 title_data 并添加匹配的关键词
title_with_keyword = title_data.copy()
title_with_keyword["matched_keyword"] = keyword
platform_map[source_name].append(title_with_keyword)
# 2. 去重(同一平台下相同标题只保留一条,保留第一个匹配的关键词)
for source_name, titles in platform_map.items():
seen_titles: Dict[str, bool] = {}
unique_titles = []
for title_data in titles:
title_text = title_data["title"]
if title_text not in seen_titles:
seen_titles[title_text] = True
unique_titles.append(title_data)
platform_map[source_name] = unique_titles
# 3. 按权重排序每个平台内的新闻
for source_name, titles in platform_map.items():
platform_map[source_name] = sorted(
titles,
key=lambda x: (
-calculate_news_weight(x, rank_threshold, weight_config),
min(x["ranks"]) if x["ranks"] else 999,
-x["count"],
),
)
# 4. 构建平台统计结果
platform_stats = []
for source_name, titles in platform_map.items():
platform_stats.append({
"word": source_name, # 平台名作为分组标识
"count": len(titles),
"titles": titles,
"percentage": 0, # 可后续计算
})
# 5. 按新闻条数排序平台
platform_stats.sort(key=lambda x: -x["count"])
return platform_stats
+91
View File
@@ -0,0 +1,91 @@
# coding=utf-8
"""
CDN 回退模块
为版本检查等远程请求提供多源回退能力。
默认使用 GitHub 原始链接,失败后自动切换到 CDN 备用源。
同一会话中记住可用源的索引,后续请求从该源开始尝试。
"""
import re
import logging
from typing import Optional
import requests
logger = logging.getLogger(__name__)
_GITHUB_RAW_PATTERN = re.compile(
r"^https://raw\.githubusercontent\.com/sansan0/TrendRadar/(?:refs/heads/)?master/(.+)$"
)
_ALL_SOURCES = [
"https://raw.githubusercontent.com/sansan0/TrendRadar/refs/heads/master/",
"https://fastly.jsdelivr.net/gh/sansan0/TrendRadar@master/",
"https://cdn.jsdelivr.net/gh/sansan0/TrendRadar@master/",
"https://gcore.jsdelivr.net/gh/sansan0/TrendRadar@master/",
]
_SOURCE_LABELS = {
_ALL_SOURCES[0]: "GitHub",
_ALL_SOURCES[1]: "fastly.jsdelivr.net",
_ALL_SOURCES[2]: "cdn.jsdelivr.net",
_ALL_SOURCES[3]: "gcore.jsdelivr.net",
}
_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/plain, */*",
"Cache-Control": "no-cache",
}
_TIMEOUT = 5
_state = {"last_ok": 0}
def _extract_path(url: str) -> Optional[str]:
m = _GITHUB_RAW_PATTERN.match(url)
return m.group(1) if m else None
def _do_request(url: str, proxies: Optional[dict]) -> str:
resp = requests.get(url, headers=_HEADERS, proxies=proxies, timeout=_TIMEOUT)
resp.raise_for_status()
return resp.text.strip()
def fetch_with_fallback(
url: str,
proxy_url: Optional[str] = None,
) -> Optional[str]:
"""从上次成功的源开始轮转尝试,非 GitHub 链接直接请求。"""
proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
path = _extract_path(url)
if path is None:
try:
return _do_request(url, proxies)
except Exception as e:
logger.warning("[版本检查] 获取失败: %s", e)
return None
n = len(_ALL_SOURCES)
start = _state["last_ok"]
for offset in range(n):
idx = (start + offset) % n
source = _ALL_SOURCES[idx]
try:
content = _do_request(source + path, proxies)
if idx != start:
label = _SOURCE_LABELS.get(source, source)
logger.info("[版本检查] 已切换到: %s", label)
_state["last_ok"] = idx
return content
except Exception:
label = _SOURCE_LABELS.get(source, source)
logger.debug("[版本检查] %s 不可用,尝试下一个源", label)
logger.warning("[版本检查] 所有源均不可用")
return None
+152
View File
@@ -0,0 +1,152 @@
# coding=utf-8
"""
配置工具模块 - 多账号配置解析和验证
提供多账号推送配置的解析、验证和限制功能
"""
from typing import Dict, List, Optional, Tuple
def parse_multi_account_config(config_value: str, separator: str = ";") -> List[str]:
"""
解析多账号配置,返回账号列表
Args:
config_value: 配置值字符串,多个账号用分隔符分隔
separator: 分隔符,默认为 ;
Returns:
账号列表,空字符串会被保留(用于占位)
Examples:
>>> parse_multi_account_config("url1;url2;url3")
['url1', 'url2', 'url3']
>>> parse_multi_account_config(";token2") # 第一个账号无token
['', 'token2']
>>> parse_multi_account_config("")
[]
"""
if not config_value:
return []
# 保留空字符串用于占位(如 ";token2" 表示第一个账号无token)
accounts = [acc.strip() for acc in config_value.split(separator)]
# 过滤掉全部为空的情况
if all(not acc for acc in accounts):
return []
return accounts
def validate_paired_configs(
configs: Dict[str, List[str]],
channel_name: str,
required_keys: Optional[List[str]] = None
) -> Tuple[bool, int]:
"""
验证配对配置的数量是否一致
对于需要多个配置项配对的渠道(如 Telegram 的 token 和 chat_id),
验证所有配置项的账号数量是否一致。
Args:
configs: 配置字典,key 为配置名,value 为账号列表
channel_name: 渠道名称,用于日志输出
required_keys: 必须有值的配置项列表
Returns:
(是否验证通过, 账号数量)
Examples:
>>> validate_paired_configs({
... "token": ["t1", "t2"],
... "chat_id": ["c1", "c2"]
... }, "Telegram", ["token", "chat_id"])
(True, 2)
>>> validate_paired_configs({
... "token": ["t1", "t2"],
... "chat_id": ["c1"] # 数量不匹配
... }, "Telegram", ["token", "chat_id"])
(False, 0)
"""
# 过滤掉空列表
non_empty_configs = {k: v for k, v in configs.items() if v}
if not non_empty_configs:
return True, 0
# 检查必须项
if required_keys:
for key in required_keys:
if key not in non_empty_configs or not non_empty_configs[key]:
return True, 0 # 必须项为空,视为未配置
# 获取所有非空配置的长度
lengths = {k: len(v) for k, v in non_empty_configs.items()}
unique_lengths = set(lengths.values())
if len(unique_lengths) > 1:
print(f"❌ {channel_name} 配置错误:配对配置数量不一致,将跳过该渠道推送")
for key, length in lengths.items():
print(f" - {key}: {length} 个")
return False, 0
return True, list(unique_lengths)[0] if unique_lengths else 0
def limit_accounts(
accounts: List[str],
max_count: int,
channel_name: str
) -> List[str]:
"""
限制账号数量
当配置的账号数量超过最大限制时,只使用前 N 个账号,
并输出警告信息。
Args:
accounts: 账号列表
max_count: 最大账号数量
channel_name: 渠道名称,用于日志输出
Returns:
限制后的账号列表
Examples:
>>> limit_accounts(["a1", "a2", "a3"], 2, "飞书")
⚠️ 飞书 配置了 3 个账号,超过最大限制 2,只使用前 2 个
['a1', 'a2']
"""
if len(accounts) > max_count:
print(f"⚠️ {channel_name} 配置了 {len(accounts)} 个账号,超过最大限制 {max_count},只使用前 {max_count} 个")
print(f" ⚠️ 警告:如果你是 fork 用户,过多账号可能导致 GitHub Actions 运行时间过长,存在账号风险")
return accounts[:max_count]
return accounts
def get_account_at_index(accounts: List[str], index: int, default: str = "") -> str:
"""
安全获取指定索引的账号值
当索引超出范围或账号值为空时,返回默认值。
Args:
accounts: 账号列表
index: 索引
default: 默认值
Returns:
账号值或默认值
Examples:
>>> get_account_at_index(["a", "b", "c"], 1)
'b'
>>> get_account_at_index(["a", "", "c"], 1, "default")
'default'
>>> get_account_at_index(["a"], 5, "default")
'default'
"""
if index < len(accounts):
return accounts[index] if accounts[index] else default
return default
+218
View File
@@ -0,0 +1,218 @@
# coding=utf-8
"""
数据处理模块
提供数据读取和检测功能:
- read_all_today_titles: 从存储后端读取当天所有标题
- detect_latest_new_titles: 检测最新批次的新增标题
Author: TrendRadar Team
"""
from typing import Dict, List, Tuple, Optional
def read_all_today_titles_from_storage(
storage_manager,
current_platform_ids: Optional[List[str]] = None,
) -> Tuple[Dict, Dict, Dict]:
"""
从存储后端读取当天所有标题(SQLite 数据)
Args:
storage_manager: 存储管理器实例
current_platform_ids: 当前监控的平台 ID 列表(用于过滤)
Returns:
Tuple[Dict, Dict, Dict]: (all_results, id_to_name, title_info)
"""
try:
news_data = storage_manager.get_today_all_data()
if not news_data or not news_data.items:
return {}, {}, {}
all_results = {}
final_id_to_name = {}
title_info = {}
for source_id, news_list in news_data.items.items():
# 按平台过滤
if current_platform_ids is not None and source_id not in current_platform_ids:
continue
# 获取来源名称
source_name = news_data.id_to_name.get(source_id, source_id)
final_id_to_name[source_id] = source_name
if source_id not in all_results:
all_results[source_id] = {}
title_info[source_id] = {}
for item in news_list:
title = item.title
ranks = item.ranks or [item.rank]
first_time = item.first_time or item.crawl_time
last_time = item.last_time or item.crawl_time
count = item.count
rank_timeline = item.rank_timeline
all_results[source_id][title] = {
"ranks": ranks,
"url": item.url or "",
"mobileUrl": item.mobile_url or "",
}
title_info[source_id][title] = {
"first_time": first_time,
"last_time": last_time,
"count": count,
"ranks": ranks,
"url": item.url or "",
"mobileUrl": item.mobile_url or "",
"rank_timeline": rank_timeline,
}
return all_results, final_id_to_name, title_info
except Exception as e:
print(f"[存储] 从存储后端读取数据失败: {e}")
return {}, {}, {}
def read_all_today_titles(
storage_manager,
current_platform_ids: Optional[List[str]] = None,
quiet: bool = False,
) -> Tuple[Dict, Dict, Dict]:
"""
读取当天所有标题(从存储后端)
Args:
storage_manager: 存储管理器实例
current_platform_ids: 当前监控的平台 ID 列表(用于过滤)
quiet: 是否静默模式(不打印日志)
Returns:
Tuple[Dict, Dict, Dict]: (all_results, id_to_name, title_info)
"""
all_results, final_id_to_name, title_info = read_all_today_titles_from_storage(
storage_manager, current_platform_ids
)
if not quiet:
if all_results:
total_count = sum(len(titles) for titles in all_results.values())
print(f"[存储] 已从存储后端读取 {total_count} 条标题")
else:
print("[存储] 当天暂无数据")
return all_results, final_id_to_name, title_info
def detect_latest_new_titles_from_storage(
storage_manager,
current_platform_ids: Optional[List[str]] = None,
) -> Dict:
"""
从存储后端检测最新批次的新增标题
Args:
storage_manager: 存储管理器实例
current_platform_ids: 当前监控的平台 ID 列表(用于过滤)
Returns:
Dict: 新增标题 {source_id: {title: title_data}}
"""
try:
# 获取最新抓取数据
latest_data = storage_manager.get_latest_crawl_data()
if not latest_data or not latest_data.items:
return {}
# 获取所有历史数据
all_data = storage_manager.get_today_all_data()
if not all_data or not all_data.items:
# 没有历史数据(第一次抓取),不应该有"新增"标题
return {}
# 获取最新批次时间
latest_time = latest_data.crawl_time
# 步骤1:收集最新批次的标题(last_crawl_time = latest_time 的标题)
latest_titles = {}
for source_id, news_list in latest_data.items.items():
if current_platform_ids is not None and source_id not in current_platform_ids:
continue
latest_titles[source_id] = {}
for item in news_list:
latest_titles[source_id][item.title] = {
"ranks": [item.rank],
"url": item.url or "",
"mobileUrl": item.mobile_url or "",
}
# 步骤2:收集历史标题
# 关键逻辑:一个标题只要其 first_crawl_time < latest_time,就是历史标题
# 这样即使同一标题有多条记录(URL 不同),只要任何一条是历史的,该标题就算历史
historical_titles = {}
for source_id, news_list in all_data.items.items():
if current_platform_ids is not None and source_id not in current_platform_ids:
continue
historical_titles[source_id] = set()
for item in news_list:
first_time = item.first_time or item.crawl_time
# 如果该记录的首次出现时间早于最新批次,则该标题是历史标题
if first_time < latest_time:
historical_titles[source_id].add(item.title)
# 检查是否是当天第一次抓取(没有任何历史标题)
# 如果所有平台的历史标题集合都为空,说明只有一个抓取批次
# 在这种情况下,将所有最新批次的标题视为"新增"(用于增量模式的第一次推送)
has_historical_data = any(len(titles) > 0 for titles in historical_titles.values())
if not has_historical_data:
# 第一次爬取:返回所有最新标题作为"新增"
return latest_titles
# 步骤3:找出新增标题 = 最新批次标题 - 历史标题
new_titles = {}
for source_id, source_latest_titles in latest_titles.items():
historical_set = historical_titles.get(source_id, set())
source_new_titles = {}
for title, title_data in source_latest_titles.items():
if title not in historical_set:
source_new_titles[title] = title_data
if source_new_titles:
new_titles[source_id] = source_new_titles
return new_titles
except Exception as e:
print(f"[存储] 从存储后端检测新标题失败: {e}")
return {}
def detect_latest_new_titles(
storage_manager,
current_platform_ids: Optional[List[str]] = None,
quiet: bool = False,
) -> Dict:
"""
检测当日最新批次的新增标题(从存储后端)
Args:
storage_manager: 存储管理器实例
current_platform_ids: 当前监控的平台 ID 列表(用于过滤)
quiet: 是否静默模式(不打印日志)
Returns:
Dict: 新增标题 {source_id: {title: title_data}}
"""
new_titles = detect_latest_new_titles_from_storage(storage_manager, current_platform_ids)
if new_titles and not quiet:
total_new = sum(len(titles) for titles in new_titles.values())
print(f"[存储] 从存储后端检测到 {total_new} 条新增标题")
return new_titles
+309
View File
@@ -0,0 +1,309 @@
# coding=utf-8
"""
频率词配置加载模块
负责从配置文件加载频率词规则,支持:
- 普通词组
- 必须词(+前缀)
- 过滤词(!前缀)
- 全局过滤词([GLOBAL_FILTER] 区域)
- 最大显示数量(@前缀)
- 正则表达式(/pattern/ 语法)
- 显示名称(=> 别名 语法)
- 组别名([组别名] 语法,作为词组第一行)
"""
import os
import re
from pathlib import Path
from typing import Dict, List, Tuple, Optional, Union
def _parse_word(word: str) -> Dict:
"""
解析单个词,识别是否为正则表达式,支持显示名称
Args:
word: 原始配置行 (e.g. "/京东|刘强东/ => 京东")
Returns:
Dict: 包含 word, is_regex, pattern, display_name
"""
display_name = None
# 1. 优先处理显示名称 (=>)
# 先切分出 "配置内容" 和 "显示名称"
if '=>' in word:
parts = re.split(r'\s*=>\s*', word, 1)
word_config = parts[0].strip()
# 只有当 => 右边有内容时才作为 display_name
if len(parts) > 1 and parts[1].strip():
display_name = parts[1].strip()
else:
word_config = word.strip()
# 2. 解析正则表达式
# 规则:以 / 开头,以 / 结尾(可能跟 flags),中间内容贪婪提取
# [a-z]*$ 表示允许末尾有 flags (如 i, g),但在下面代码中会被忽略
regex_match = re.match(r'^/(.+)/[a-z]*$', word_config)
if regex_match:
pattern_str = regex_match.group(1)
try:
pattern = re.compile(pattern_str, re.IGNORECASE)
return {
"word": pattern_str,
"is_regex": True,
"pattern": pattern,
"display_name": display_name,
}
except re.error as e:
print(f"Warning: Invalid regex pattern '/{pattern_str}/': {e}")
pass
return {
"word": word_config,
"is_regex": False,
"pattern": None,
"display_name": display_name
}
def _word_matches(word_config: Union[str, Dict], title_lower: str) -> bool:
"""
检查词是否在标题中匹配
Args:
word_config: 词配置(字符串或字典)
title_lower: 小写的标题
Returns:
是否匹配
"""
if isinstance(word_config, str):
# 向后兼容:纯字符串
return word_config.lower() in title_lower
if word_config.get("is_regex") and word_config.get("pattern"):
# 正则匹配
return bool(word_config["pattern"].search(title_lower))
else:
# 子字符串匹配
return word_config["word"].lower() in title_lower
def load_frequency_words(
frequency_file: Optional[str] = None,
) -> Tuple[List[Dict], List[str], List[str]]:
"""
加载频率词配置
配置文件格式说明:
- 每个词组由空行分隔
- [GLOBAL_FILTER] 区域定义全局过滤词
- [WORD_GROUPS] 区域定义词组(默认)
词组语法:
- 普通词:直接写入,任意匹配即可
- +词:必须词,所有必须词都要匹配
- !词:过滤词,匹配则排除
- @数字:该词组最多显示的条数
Args:
frequency_file: 频率词配置文件路径,默认从环境变量 FREQUENCY_WORDS_PATH 获取或使用 config/frequency_words.txt,短文件名从 config/custom/keyword/ 查找
Returns:
(词组列表, 词组内过滤词, 全局过滤词)
Raises:
FileNotFoundError: 频率词文件不存在
"""
if frequency_file is None:
frequency_file = os.environ.get(
"FREQUENCY_WORDS_PATH", "config/frequency_words.txt"
)
frequency_path = Path(frequency_file)
if not frequency_path.exists():
# 尝试作为短文件名,拼接 config/custom/keyword/ 前缀
custom_path = Path("config/custom/keyword") / frequency_file
if custom_path.exists():
frequency_path = custom_path
else:
raise FileNotFoundError(f"频率词文件 {frequency_file} 不存在")
with open(frequency_path, "r", encoding="utf-8") as f:
content = f.read()
word_groups = [group.strip() for group in content.split("\n\n") if group.strip()]
processed_groups = []
filter_words = []
global_filters = []
# 默认区域(向后兼容)
current_section = "WORD_GROUPS"
for group in word_groups:
# 过滤空行和注释行(# 开头)
lines = [line.strip() for line in group.split("\n") if line.strip() and not line.strip().startswith("#")]
if not lines:
continue
# 检查是否为区域标记
if lines[0].startswith("[") and lines[0].endswith("]"):
section_name = lines[0][1:-1].upper()
if section_name in ("GLOBAL_FILTER", "WORD_GROUPS"):
current_section = section_name
lines = lines[1:] # 移除标记行
# 处理全局过滤区域
if current_section == "GLOBAL_FILTER":
# 直接添加所有非空行到全局过滤列表
for line in lines:
# 忽略特殊语法前缀,只提取纯文本
if line.startswith(("!", "+", "@")):
continue # 全局过滤区不支持特殊语法
if line:
global_filters.append(line)
continue
# 处理词组区域
words = lines
group_alias = None # 组别名([别名] 语法)
# 检查第一行是否为组别名(非区域标记)
if words and words[0].startswith("[") and words[0].endswith("]"):
potential_alias = words[0][1:-1].strip()
# 排除区域标记(GLOBAL_FILTER, WORD_GROUPS)
if potential_alias.upper() not in ("GLOBAL_FILTER", "WORD_GROUPS"):
group_alias = potential_alias
words = words[1:] # 移除组别名行
group_required_words = []
group_normal_words = []
group_max_count = 0 # 默认不限制
for word in words:
if word.startswith("@"):
# 解析最大显示数量(只接受正整数)
try:
count = int(word[1:])
if count > 0:
group_max_count = count
except (ValueError, IndexError):
pass # 忽略无效的@数字格式
elif word.startswith("!"):
# 过滤词(支持正则语法)
filter_word = word[1:]
parsed = _parse_word(filter_word)
filter_words.append(parsed)
elif word.startswith("+"):
# 必须词(支持正则语法)
req_word = word[1:]
group_required_words.append(_parse_word(req_word))
else:
# 普通词(支持正则语法)
group_normal_words.append(_parse_word(word))
if group_required_words or group_normal_words:
if group_normal_words:
group_key = " ".join(w["word"] for w in group_normal_words)
else:
group_key = " ".join(w["word"] for w in group_required_words)
# 生成显示名称
# 优先级:组别名 > 行别名拼接 > 关键词拼接
if group_alias:
# 有组别名,直接使用
display_name = group_alias
else:
# 没有组别名,拼接每行的显示名(行别名或关键词本身)
all_words = group_normal_words + group_required_words
display_parts = []
for w in all_words:
# 优先使用行别名,否则使用关键词本身
part = w.get("display_name") or w["word"]
display_parts.append(part)
# 用 " / " 拼接多个词
display_name = " / ".join(display_parts) if display_parts else None
processed_groups.append(
{
"required": group_required_words,
"normal": group_normal_words,
"group_key": group_key,
"display_name": display_name, # 可能为 None
"max_count": group_max_count,
}
)
return processed_groups, filter_words, global_filters
def matches_word_groups(
title: str,
word_groups: List[Dict],
filter_words: List,
global_filters: Optional[List[str]] = None
) -> bool:
"""
检查标题是否匹配词组规则
Args:
title: 标题文本
word_groups: 词组列表
filter_words: 过滤词列表(可以是字符串列表或字典列表)
global_filters: 全局过滤词列表
Returns:
是否匹配
"""
# 防御性类型检查:确保 title 是有效字符串
if not isinstance(title, str):
title = str(title) if title is not None else ""
if not title.strip():
return False
title_lower = title.lower()
# 全局过滤检查(优先级最高)
if global_filters:
if any(global_word.lower() in title_lower for global_word in global_filters):
return False
# 如果没有配置词组,则匹配所有标题(支持显示全部新闻)
if not word_groups:
return True
# 过滤词检查(兼容新旧格式)
for filter_item in filter_words:
if _word_matches(filter_item, title_lower):
return False
# 词组匹配检查
for group in word_groups:
required_words = group["required"]
normal_words = group["normal"]
# 必须词检查
if required_words:
all_required_present = all(
_word_matches(req_item, title_lower) for req_item in required_words
)
if not all_required_present:
continue
# 普通词检查
if normal_words:
any_normal_present = any(
_word_matches(normal_item, title_lower) for normal_item in normal_words
)
if not any_normal_present:
continue
return True
return False
+606
View File
@@ -0,0 +1,606 @@
# coding=utf-8
"""
配置加载模块
负责从 YAML 配置文件和环境变量加载配置。
"""
import os
from pathlib import Path
from typing import Dict, Any, Optional
import yaml
from .config import parse_multi_account_config, validate_paired_configs
from trendradar.utils.time import DEFAULT_TIMEZONE
def _get_env_bool(key: str) -> Optional[bool]:
"""从环境变量获取布尔值,如果未设置返回 None"""
value = os.environ.get(key, "").strip().lower()
if not value:
return None
return value in ("true", "1")
def _get_env_int_or_none(key: str) -> Optional[int]:
"""从环境变量获取整数值,未设置时返回 None"""
value = os.environ.get(key, "").strip()
if not value:
return None
try:
return int(value)
except ValueError:
return None
def _get_env_str(key: str, default: str = "") -> str:
"""从环境变量获取字符串值"""
return os.environ.get(key, "").strip() or default
def _load_app_config(config_data: Dict) -> Dict:
"""加载应用配置"""
app_config = config_data.get("app", {})
advanced = config_data.get("advanced", {})
return {
"VERSION_CHECK_URL": advanced.get("version_check_url", ""),
"CONFIGS_VERSION_CHECK_URL": advanced.get("configs_version_check_url", ""),
"SHOW_VERSION_UPDATE": app_config.get("show_version_update", True),
"TIMEZONE": _get_env_str("TIMEZONE") or app_config.get("timezone", DEFAULT_TIMEZONE),
"DEBUG": _get_env_bool("DEBUG") if _get_env_bool("DEBUG") is not None else advanced.get("debug", False),
}
def _load_crawler_config(config_data: Dict) -> Dict:
"""加载爬虫配置"""
advanced = config_data.get("advanced", {})
crawler_config = advanced.get("crawler", {})
platforms_config = config_data.get("platforms", {})
return {
"REQUEST_INTERVAL": crawler_config.get("request_interval", 100),
"USE_PROXY": crawler_config.get("use_proxy", False),
"DEFAULT_PROXY": crawler_config.get("default_proxy", ""),
"ENABLE_CRAWLER": platforms_config.get("enabled", True),
"PLATFORMS_API_URL": _get_env_str("PLATFORMS_API_URL") or platforms_config.get("api_url", ""),
}
def _load_report_config(config_data: Dict) -> Dict:
"""加载报告配置"""
report_config = config_data.get("report", {})
# 环境变量覆盖
sort_by_position_env = _get_env_bool("SORT_BY_POSITION_FIRST")
max_news_env = _get_env_int_or_none("MAX_NEWS_PER_KEYWORD")
return {
"REPORT_MODE": report_config.get("mode", "daily"),
"DISPLAY_MODE": report_config.get("display_mode", "keyword"),
"RANK_THRESHOLD": report_config.get("rank_threshold", 10),
"SORT_BY_POSITION_FIRST": sort_by_position_env if sort_by_position_env is not None else report_config.get("sort_by_position_first", False),
"MAX_NEWS_PER_KEYWORD": max_news_env if max_news_env is not None else report_config.get("max_news_per_keyword", 0),
}
def _load_notification_config(config_data: Dict) -> Dict:
"""加载通知配置"""
notification = config_data.get("notification", {})
advanced = config_data.get("advanced", {})
batch_size = advanced.get("batch_size", {})
max_accounts_env = _get_env_int_or_none("MAX_ACCOUNTS_PER_CHANNEL")
return {
"ENABLE_NOTIFICATION": notification.get("enabled", True),
"MESSAGE_BATCH_SIZE": batch_size.get("default", 4000),
"DINGTALK_BATCH_SIZE": batch_size.get("dingtalk", 20000),
"FEISHU_BATCH_SIZE": batch_size.get("feishu", 29000),
"BARK_BATCH_SIZE": batch_size.get("bark", 3600),
"SLACK_BATCH_SIZE": batch_size.get("slack", 4000),
"BATCH_SEND_INTERVAL": advanced.get("batch_send_interval", 1.0),
"FEISHU_MESSAGE_SEPARATOR": advanced.get("feishu_message_separator", "---"),
"MAX_ACCOUNTS_PER_CHANNEL": max_accounts_env if max_accounts_env is not None else advanced.get("max_accounts_per_channel", 3),
}
def _load_schedule_config(config_data: Dict) -> Dict:
"""
加载统一调度配置
从 config.yaml 的 schedule 段读取,支持环境变量覆盖。
"""
schedule = config_data.get("schedule", {})
# 环境变量覆盖
enabled_env = _get_env_bool("SCHEDULE_ENABLED")
preset_env = _get_env_str("SCHEDULE_PRESET")
enabled = enabled_env if enabled_env is not None else schedule.get("enabled", False)
preset = preset_env or schedule.get("preset", "always_on")
return {
"enabled": enabled,
"preset": preset,
}
def _load_timeline_data(config_dir: str = "config") -> Dict:
"""
加载 timeline.yaml
Args:
config_dir: 配置目录路径
Returns:
timeline.yaml 的完整数据,找不到时返回空模板
"""
timeline_path = Path(config_dir) / "timeline.yaml"
if not timeline_path.exists():
print(f"[调度] timeline.yaml 未找到: {timeline_path},使用空模板")
return {
"presets": {},
"custom": {
"default": {
"collect": True,
"analyze": False,
"push": False,
"report_mode": "current",
"ai_mode": "follow_report",
"once": {"analyze": False, "push": False},
},
"periods": {},
"day_plans": {"all_day": {"periods": []}},
"week_map": {i: "all_day" for i in range(1, 8)},
},
}
with open(timeline_path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
print(f"[调度] timeline.yaml 加载成功: {timeline_path}")
return data or {}
def _load_weight_config(config_data: Dict) -> Dict:
"""加载权重配置"""
advanced = config_data.get("advanced", {})
weight = advanced.get("weight", {})
return {
"RANK_WEIGHT": weight.get("rank", 0.6),
"FREQUENCY_WEIGHT": weight.get("frequency", 0.3),
"HOTNESS_WEIGHT": weight.get("hotness", 0.1),
}
def _load_rss_config(config_data: Dict) -> Dict:
"""加载 RSS 配置"""
rss = config_data.get("rss", {})
advanced = config_data.get("advanced", {})
advanced_rss = advanced.get("rss", {})
advanced_crawler = advanced.get("crawler", {})
# RSS 代理配置:优先使用 RSS 专属代理,否则复用 crawler 的 default_proxy
rss_proxy_url = advanced_rss.get("proxy_url", "") or advanced_crawler.get("default_proxy", "")
# 新鲜度过滤配置
freshness_filter = rss.get("freshness_filter", {})
# 验证并设置 max_age_days 默认值
raw_max_age = freshness_filter.get("max_age_days", 3)
try:
max_age_days = int(raw_max_age)
if max_age_days < 0:
print(f"[警告] RSS freshness_filter.max_age_days 为负数 ({max_age_days}),使用默认值 3")
max_age_days = 3
except (ValueError, TypeError):
print(f"[警告] RSS freshness_filter.max_age_days 格式错误 ({raw_max_age}),使用默认值 3")
max_age_days = 3
# RSS 配置直接从 config.yaml 读取,不再支持环境变量
return {
"ENABLED": rss.get("enabled", False),
"REQUEST_INTERVAL": advanced_rss.get("request_interval", 2000),
"TIMEOUT": advanced_rss.get("timeout", 15),
"USE_PROXY": advanced_rss.get("use_proxy", False),
"PROXY_URL": rss_proxy_url,
"FEEDS": rss.get("feeds", []),
"FRESHNESS_FILTER": {
"ENABLED": freshness_filter.get("enabled", True), # 默认启用
"MAX_AGE_DAYS": max_age_days,
},
}
def _load_display_config(config_data: Dict) -> Dict:
"""加载推送内容显示配置"""
display = config_data.get("display", {})
regions = display.get("regions", {})
standalone = display.get("standalone", {})
# 默认区域顺序
default_region_order = ["hotlist", "rss", "new_items", "standalone", "ai_analysis"]
region_order = display.get("region_order", default_region_order)
# 验证 region_order 中的值是否合法
valid_regions = {"hotlist", "rss", "new_items", "standalone", "ai_analysis"}
region_order = [r for r in region_order if r in valid_regions]
# 如果过滤后为空,使用默认顺序
if not region_order:
region_order = default_region_order
return {
# 区域显示顺序
"REGION_ORDER": region_order,
# 区域开关
"REGIONS": {
"HOTLIST": regions.get("hotlist", True),
"NEW_ITEMS": regions.get("new_items", True),
"RSS": regions.get("rss", True),
"STANDALONE": regions.get("standalone", False),
"AI_ANALYSIS": regions.get("ai_analysis", True),
},
# 独立展示区配置
"STANDALONE": {
"PLATFORMS": standalone.get("platforms", []),
"RSS_FEEDS": standalone.get("rss_feeds", []),
"MAX_ITEMS": standalone.get("max_items", 20),
},
}
def _load_ai_config(config_data: Dict) -> Dict:
"""加载 AI 模型配置(LiteLLM 格式)"""
ai_config = config_data.get("ai", {})
timeout_env = _get_env_int_or_none("AI_TIMEOUT")
return {
# LiteLLM 核心配置
"MODEL": _get_env_str("AI_MODEL") or ai_config.get("model", ""),
"API_KEY": _get_env_str("AI_API_KEY") or ai_config.get("api_key", ""),
"API_BASE": _get_env_str("AI_API_BASE") or ai_config.get("api_base", ""),
# 生成参数
"TIMEOUT": timeout_env if timeout_env is not None else ai_config.get("timeout", 120),
"TEMPERATURE": ai_config.get("temperature", 1.0),
"MAX_TOKENS": ai_config.get("max_tokens", 5000),
# LiteLLM 高级选项
"NUM_RETRIES": ai_config.get("num_retries", 2),
"FALLBACK_MODELS": ai_config.get("fallback_models", []),
"EXTRA_PARAMS": ai_config.get("extra_params", {}),
}
def _load_ai_analysis_config(config_data: Dict) -> Dict:
"""加载 AI 分析配置(功能配置,模型配置见 _load_ai_config)"""
ai_config = config_data.get("ai_analysis", {})
enabled_env = _get_env_bool("AI_ANALYSIS_ENABLED")
return {
"ENABLED": enabled_env if enabled_env is not None else ai_config.get("enabled", False),
"LANGUAGE": ai_config.get("language", "Chinese"),
"PROMPT_FILE": ai_config.get("prompt_file", "ai_analysis_prompt.txt"),
"MODE": ai_config.get("mode", "follow_report"),
"MAX_NEWS_FOR_ANALYSIS": ai_config.get("max_news_for_analysis", 50),
"INCLUDE_RSS": ai_config.get("include_rss", True),
"INCLUDE_RANK_TIMELINE": ai_config.get("include_rank_timeline", False),
"INCLUDE_STANDALONE": ai_config.get("include_standalone", False),
}
def _load_ai_translation_config(config_data: Dict) -> Dict:
"""加载 AI 翻译配置(功能配置,模型配置见 _load_ai_config)"""
trans_config = config_data.get("ai_translation", {})
enabled_env = _get_env_bool("AI_TRANSLATION_ENABLED")
scope = trans_config.get("scope", {})
return {
"ENABLED": enabled_env if enabled_env is not None else trans_config.get("enabled", False),
"LANGUAGE": _get_env_str("AI_TRANSLATION_LANGUAGE") or trans_config.get("language", "English"),
"PROMPT_FILE": trans_config.get("prompt_file", "ai_translation_prompt.txt"),
"SCOPE": {
"HOTLIST": scope.get("hotlist", True),
"RSS": scope.get("rss", True),
"STANDALONE": scope.get("standalone", True),
},
}
def _load_ai_filter_config(config_data: Dict) -> Dict:
"""加载 AI 智能筛选配置(由 filter.method 控制是否启用)"""
ai_filter = config_data.get("ai_filter", {})
return {
"BATCH_SIZE": ai_filter.get("batch_size", 200),
"BATCH_INTERVAL": ai_filter.get("batch_interval", 5),
"INTERESTS_FILE": ai_filter.get("interests_file"), # None = 使用默认 config/ai_interests.txt
"PROMPT_FILE": ai_filter.get("prompt_file", "prompt.txt"),
"EXTRACT_PROMPT_FILE": ai_filter.get("extract_prompt_file", "extract_prompt.txt"),
"UPDATE_TAGS_PROMPT_FILE": ai_filter.get("update_tags_prompt_file", "update_tags_prompt.txt"),
"RECLASSIFY_THRESHOLD": ai_filter.get("reclassify_threshold", 0.6),
"MIN_SCORE": float(ai_filter.get("min_score", 0)),
}
def _load_filter_config(config_data: Dict) -> Dict:
"""加载筛选策略配置"""
filter_cfg = config_data.get("filter", {})
# 环境变量兼容:AI_FILTER_ENABLED=true → method=ai
env_ai_filter = _get_env_bool("AI_FILTER_ENABLED")
method = filter_cfg.get("method", "keyword")
if env_ai_filter is True:
method = "ai"
# 兼容旧配置:如果 ai_filter.enabled=true 且未显式设置 filter.method
if method == "keyword" and not filter_cfg.get("method"):
ai_filter = config_data.get("ai_filter", {})
if ai_filter.get("enabled", False):
method = "ai"
return {
"METHOD": method, # "keyword" | "ai"
"PRIORITY_SORT_ENABLED": filter_cfg.get("priority_sort_enabled", False), # AI 模式标签优先级排序开关
}
def _load_storage_config(config_data: Dict) -> Dict:
"""加载存储配置"""
storage = config_data.get("storage", {})
formats = storage.get("formats", {})
local = storage.get("local", {})
remote = storage.get("remote", {})
pull = storage.get("pull", {})
txt_enabled_env = _get_env_bool("STORAGE_TXT_ENABLED")
html_enabled_env = _get_env_bool("STORAGE_HTML_ENABLED")
pull_enabled_env = _get_env_bool("PULL_ENABLED")
local_retention_env = _get_env_int_or_none("LOCAL_RETENTION_DAYS")
remote_retention_env = _get_env_int_or_none("REMOTE_RETENTION_DAYS")
pull_days_env = _get_env_int_or_none("PULL_DAYS")
return {
"BACKEND": _get_env_str("STORAGE_BACKEND") or storage.get("backend", "auto"),
"FORMATS": {
"SQLITE": formats.get("sqlite", True),
"TXT": txt_enabled_env if txt_enabled_env is not None else formats.get("txt", True),
"HTML": html_enabled_env if html_enabled_env is not None else formats.get("html", True),
},
"LOCAL": {
"DATA_DIR": local.get("data_dir", "output"),
"RETENTION_DAYS": local_retention_env if local_retention_env is not None else local.get("retention_days", 0),
},
"REMOTE": {
"ENDPOINT_URL": _get_env_str("S3_ENDPOINT_URL") or remote.get("endpoint_url", ""),
"BUCKET_NAME": _get_env_str("S3_BUCKET_NAME") or remote.get("bucket_name", ""),
"ACCESS_KEY_ID": _get_env_str("S3_ACCESS_KEY_ID") or remote.get("access_key_id", ""),
"SECRET_ACCESS_KEY": _get_env_str("S3_SECRET_ACCESS_KEY") or remote.get("secret_access_key", ""),
"REGION": _get_env_str("S3_REGION") or remote.get("region", ""),
"RETENTION_DAYS": remote_retention_env if remote_retention_env is not None else remote.get("retention_days", 0),
},
"PULL": {
"ENABLED": pull_enabled_env if pull_enabled_env is not None else pull.get("enabled", False),
"DAYS": pull_days_env if pull_days_env is not None else pull.get("days", 7),
},
}
def _load_webhook_config(config_data: Dict) -> Dict:
"""加载 Webhook 配置"""
notification = config_data.get("notification", {})
channels = notification.get("channels", {})
# 各渠道配置
feishu = channels.get("feishu", {})
dingtalk = channels.get("dingtalk", {})
wework = channels.get("wework", {})
telegram = channels.get("telegram", {})
email = channels.get("email", {})
ntfy = channels.get("ntfy", {})
bark = channels.get("bark", {})
slack = channels.get("slack", {})
generic = channels.get("generic_webhook", {})
return {
# 飞书
"FEISHU_WEBHOOK_URL": _get_env_str("FEISHU_WEBHOOK_URL") or feishu.get("webhook_url", ""),
# 钉钉
"DINGTALK_WEBHOOK_URL": _get_env_str("DINGTALK_WEBHOOK_URL") or dingtalk.get("webhook_url", ""),
# 企业微信
"WEWORK_WEBHOOK_URL": _get_env_str("WEWORK_WEBHOOK_URL") or wework.get("webhook_url", ""),
"WEWORK_MSG_TYPE": _get_env_str("WEWORK_MSG_TYPE") or wework.get("msg_type", "markdown"),
# Telegram
"TELEGRAM_BOT_TOKEN": _get_env_str("TELEGRAM_BOT_TOKEN") or telegram.get("bot_token", ""),
"TELEGRAM_CHAT_ID": _get_env_str("TELEGRAM_CHAT_ID") or telegram.get("chat_id", ""),
# 邮件
"EMAIL_FROM": _get_env_str("EMAIL_FROM") or email.get("from", ""),
"EMAIL_PASSWORD": _get_env_str("EMAIL_PASSWORD") or email.get("password", ""),
"EMAIL_TO": _get_env_str("EMAIL_TO") or email.get("to", ""),
"EMAIL_SMTP_SERVER": _get_env_str("EMAIL_SMTP_SERVER") or email.get("smtp_server", ""),
"EMAIL_SMTP_PORT": _get_env_str("EMAIL_SMTP_PORT") or email.get("smtp_port", ""),
# ntfy
"NTFY_SERVER_URL": _get_env_str("NTFY_SERVER_URL") or ntfy.get("server_url") or "https://ntfy.sh",
"NTFY_TOPIC": _get_env_str("NTFY_TOPIC") or ntfy.get("topic", ""),
"NTFY_TOKEN": _get_env_str("NTFY_TOKEN") or ntfy.get("token", ""),
# Bark
"BARK_URL": _get_env_str("BARK_URL") or bark.get("url", ""),
# Slack
"SLACK_WEBHOOK_URL": _get_env_str("SLACK_WEBHOOK_URL") or slack.get("webhook_url", ""),
# 通用 Webhook
"GENERIC_WEBHOOK_URL": _get_env_str("GENERIC_WEBHOOK_URL") or generic.get("webhook_url", ""),
"GENERIC_WEBHOOK_TEMPLATE": _get_env_str("GENERIC_WEBHOOK_TEMPLATE") or generic.get("payload_template", ""),
}
def _print_notification_sources(config: Dict) -> None:
"""打印通知渠道配置来源信息"""
notification_sources = []
max_accounts = config["MAX_ACCOUNTS_PER_CHANNEL"]
if config["FEISHU_WEBHOOK_URL"]:
accounts = parse_multi_account_config(config["FEISHU_WEBHOOK_URL"])
count = min(len(accounts), max_accounts)
source = "环境变量" if os.environ.get("FEISHU_WEBHOOK_URL") else "配置文件"
notification_sources.append(f"飞书({source}, {count}个账号)")
if config["DINGTALK_WEBHOOK_URL"]:
accounts = parse_multi_account_config(config["DINGTALK_WEBHOOK_URL"])
count = min(len(accounts), max_accounts)
source = "环境变量" if os.environ.get("DINGTALK_WEBHOOK_URL") else "配置文件"
notification_sources.append(f"钉钉({source}, {count}个账号)")
if config["WEWORK_WEBHOOK_URL"]:
accounts = parse_multi_account_config(config["WEWORK_WEBHOOK_URL"])
count = min(len(accounts), max_accounts)
source = "环境变量" if os.environ.get("WEWORK_WEBHOOK_URL") else "配置文件"
notification_sources.append(f"企业微信({source}, {count}个账号)")
if config["TELEGRAM_BOT_TOKEN"] and config["TELEGRAM_CHAT_ID"]:
tokens = parse_multi_account_config(config["TELEGRAM_BOT_TOKEN"])
chat_ids = parse_multi_account_config(config["TELEGRAM_CHAT_ID"])
valid, count = validate_paired_configs(
{"bot_token": tokens, "chat_id": chat_ids},
"Telegram",
required_keys=["bot_token", "chat_id"]
)
if valid and count > 0:
count = min(count, max_accounts)
token_source = "环境变量" if os.environ.get("TELEGRAM_BOT_TOKEN") else "配置文件"
notification_sources.append(f"Telegram({token_source}, {count}个账号)")
if config["EMAIL_FROM"] and config["EMAIL_PASSWORD"] and config["EMAIL_TO"]:
from_source = "环境变量" if os.environ.get("EMAIL_FROM") else "配置文件"
notification_sources.append(f"邮件({from_source})")
if config["NTFY_SERVER_URL"] and config["NTFY_TOPIC"]:
topics = parse_multi_account_config(config["NTFY_TOPIC"])
tokens = parse_multi_account_config(config["NTFY_TOKEN"])
if tokens:
valid, count = validate_paired_configs(
{"topic": topics, "token": tokens},
"ntfy"
)
if valid and count > 0:
count = min(count, max_accounts)
server_source = "环境变量" if os.environ.get("NTFY_SERVER_URL") else "配置文件"
notification_sources.append(f"ntfy({server_source}, {count}个账号)")
else:
count = min(len(topics), max_accounts)
server_source = "环境变量" if os.environ.get("NTFY_SERVER_URL") else "配置文件"
notification_sources.append(f"ntfy({server_source}, {count}个账号)")
if config["BARK_URL"]:
accounts = parse_multi_account_config(config["BARK_URL"])
count = min(len(accounts), max_accounts)
bark_source = "环境变量" if os.environ.get("BARK_URL") else "配置文件"
notification_sources.append(f"Bark({bark_source}, {count}个账号)")
if config["SLACK_WEBHOOK_URL"]:
accounts = parse_multi_account_config(config["SLACK_WEBHOOK_URL"])
count = min(len(accounts), max_accounts)
slack_source = "环境变量" if os.environ.get("SLACK_WEBHOOK_URL") else "配置文件"
notification_sources.append(f"Slack({slack_source}, {count}个账号)")
if config.get("GENERIC_WEBHOOK_URL"):
accounts = parse_multi_account_config(config["GENERIC_WEBHOOK_URL"])
count = min(len(accounts), max_accounts)
source = "环境变量" if os.environ.get("GENERIC_WEBHOOK_URL") else "配置文件"
notification_sources.append(f"通用Webhook({source}, {count}个账号)")
if notification_sources:
print(f"通知渠道配置来源: {', '.join(notification_sources)}")
print(f"每个渠道最大账号数: {max_accounts}")
else:
print("未配置任何通知渠道")
def load_config(config_path: Optional[str] = None) -> Dict[str, Any]:
"""
加载配置文件
Args:
config_path: 配置文件路径,默认从环境变量 CONFIG_PATH 获取或使用 config/config.yaml
Returns:
包含所有配置的字典
Raises:
FileNotFoundError: 配置文件不存在
"""
if config_path is None:
config_path = os.environ.get("CONFIG_PATH", "config/config.yaml")
if not Path(config_path).exists():
raise FileNotFoundError(f"配置文件 {config_path} 不存在")
with open(config_path, "r", encoding="utf-8") as f:
config_data = yaml.safe_load(f)
print(f"配置文件加载成功: {config_path}")
# 合并所有配置
config = {}
# 应用配置
config.update(_load_app_config(config_data))
# 爬虫配置
config.update(_load_crawler_config(config_data))
# 报告配置
config.update(_load_report_config(config_data))
# 通知配置
config.update(_load_notification_config(config_data))
# 统一调度配置
config["SCHEDULE"] = _load_schedule_config(config_data)
config["_TIMELINE_DATA"] = _load_timeline_data(
str(Path(config_path).parent) if config_path else "config"
)
# 权重配置
config["WEIGHT_CONFIG"] = _load_weight_config(config_data)
# 平台配置
platforms_config = config_data.get("platforms", {})
config["PLATFORMS"] = [p for p in platforms_config.get("sources", []) if p.get("enabled", True)]
# RSS 配置
config["RSS"] = _load_rss_config(config_data)
# AI 模型共享配置
config["AI"] = _load_ai_config(config_data)
# AI 分析配置
config["AI_ANALYSIS"] = _load_ai_analysis_config(config_data)
# AI 翻译配置
config["AI_TRANSLATION"] = _load_ai_translation_config(config_data)
# AI 智能筛选配置
config["AI_FILTER"] = _load_ai_filter_config(config_data)
# 筛选策略配置
config["FILTER"] = _load_filter_config(config_data)
# 推送内容显示配置
config["DISPLAY"] = _load_display_config(config_data)
# 存储配置
config["STORAGE"] = _load_storage_config(config_data)
# Webhook 配置
config.update(_load_webhook_config(config_data))
# 打印通知渠道配置来源
_print_notification_sources(config)
return config
+431
View File
@@ -0,0 +1,431 @@
# coding=utf-8
"""
时间线调度器
统一的时间线调度系统,替代分散的 push_window / analysis_window 逻辑。
基于 periods + day_plans + week_map 模型实现灵活的时间段调度。
"""
import copy
import re
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional
from datetime import datetime
@dataclass
class ResolvedSchedule:
"""当前时间解析后的调度结果"""
period_key: Optional[str] # 命中的 period key,None=默认配置
period_name: Optional[str] # 命中的展示名称
day_plan: str # 当前日计划
collect: bool
analyze: bool
push: bool
report_mode: str
ai_mode: str
once_analyze: bool
once_push: bool
frequency_file: Optional[str] = None # 频率词文件路径,None=使用默认
filter_method: Optional[str] = None # 筛选策略: "keyword"|"ai",None=使用全局配置
interests_file: Optional[str] = None # AI 筛选兴趣文件,None=使用默认
class Scheduler:
"""
时间线调度器
根据 timeline 配置(periods + day_plans + week_map)解析当前时间应执行的行为。
支持:
- 预设模板 + 自定义模式
- 跨日时间段(如 22:00-07:00)
- 每天 / 每周差异化配置
- once 执行去重(analyze / push 独立维度)
- 冲突策略(error_on_overlap / last_wins)
"""
def __init__(
self,
schedule_config: Dict[str, Any],
timeline_data: Dict[str, Any],
storage_backend: Any,
get_time_func: Callable[[], datetime],
fallback_report_mode: str = "current",
):
"""
初始化调度器
Args:
schedule_config: config.yaml 中的 schedule 段(含 preset 等)
timeline_data: timeline.yaml 的完整数据
storage_backend: 存储后端(用于 once 去重记录)
get_time_func: 获取当前时间的函数(应使用配置的时区)
fallback_report_mode: 调度未启用时回退使用的 report_mode(来自 config.yaml 的 report.mode)
"""
self.schedule_config = schedule_config
self.storage = storage_backend
self.get_time = get_time_func
self.enabled = schedule_config.get("enabled", True)
self.fallback_report_mode = fallback_report_mode
# 加载并构建最终 timeline
self.timeline = self._build_timeline(schedule_config, timeline_data)
if self.enabled:
self._validate_timeline(self.timeline)
def _build_timeline(
self,
schedule_config: Dict[str, Any],
timeline_data: Dict[str, Any],
) -> Dict[str, Any]:
"""从 preset 或 custom 构建 timeline"""
preset = schedule_config.get("preset", "always_on")
if preset == "custom":
timeline = copy.deepcopy(timeline_data.get("custom", {}))
else:
presets = timeline_data.get("presets", {})
if preset not in presets:
raise ValueError(
f"未知的预设模板: '{preset}',可选值: "
f"{', '.join(presets.keys())}, custom"
)
timeline = copy.deepcopy(presets[preset])
# 确保 periods 是 dict(可能为空 {})
if timeline.get("periods") is None:
timeline["periods"] = {}
return timeline
def resolve(self) -> ResolvedSchedule:
"""
解析当前时间对应的调度配置
Returns:
ResolvedSchedule 包含当前应执行的行为
"""
if not self.enabled:
# 调度未启用时返回默认的全功能配置,report_mode 回退使用 config.yaml 的 report.mode
return ResolvedSchedule(
period_key=None,
period_name=None,
day_plan="disabled",
collect=True,
analyze=True,
push=True,
report_mode=self.fallback_report_mode,
ai_mode="follow_report",
once_analyze=False,
once_push=False,
)
now = self.get_time()
weekday = now.isoweekday() # 1=周一 ... 7=周日
now_hhmm = now.strftime("%H:%M")
# 查找当天的日计划
day_plan_key = self.timeline["week_map"].get(weekday)
if day_plan_key is None:
raise ValueError(f"week_map 缺少星期映射: {weekday}")
day_plan = self.timeline["day_plans"].get(day_plan_key)
if day_plan is None:
raise ValueError(f"week_map[{weekday}] 引用了不存在的 day_plan: {day_plan_key}")
# 查找当前活跃的时间段
period_key = self._find_active_period(now_hhmm, day_plan)
# 合并默认配置和时间段配置
merged = self._merge_with_default(period_key)
# 打印调度日志
weekday_names = {1: "一", 2: "二", 3: "三", 4: "四", 5: "五", 6: "六", 7: "日"}
period_display = "默认配置(未命中任何时间段)"
if period_key:
period_cfg = self.timeline["periods"][period_key]
period_name = period_cfg.get("name", period_key)
start = period_cfg.get("start", "?")
end = period_cfg.get("end", "?")
period_display = f"{period_name} ({start}-{end})"
print(f"[调度] 星期{weekday_names.get(weekday, '?')},日计划: {day_plan_key}")
print(f"[调度] 当前时间段: {period_display}")
resolved = ResolvedSchedule(
period_key=period_key,
period_name=(
self.timeline["periods"][period_key].get("name")
if period_key
else None
),
day_plan=day_plan_key,
collect=merged.get("collect", True),
analyze=merged.get("analyze", False),
push=merged.get("push", False),
report_mode=merged.get("report_mode", "current"),
ai_mode=self._resolve_ai_mode(merged),
once_analyze=merged.get("once", {}).get("analyze", False),
once_push=merged.get("once", {}).get("push", False),
frequency_file=merged.get("frequency_file"),
filter_method=merged.get("filter_method"),
interests_file=merged.get("interests_file"),
)
# 打印行为摘要
actions = []
if resolved.collect:
actions.append("采集")
if resolved.analyze:
actions.append(f"分析(AI:{resolved.ai_mode})")
if resolved.push:
actions.append(f"推送(模式:{resolved.report_mode})")
print(f"[调度] 行为: {', '.join(actions) if actions else '无'}")
if resolved.frequency_file:
print(f"[调度] 频率词文件: {resolved.frequency_file}")
return resolved
def _find_active_period(
self, now_hhmm: str, day_plan: Dict[str, Any]
) -> Optional[str]:
"""
查找当前时间命中的活跃时间段
Args:
now_hhmm: 当前时间 HH:MM
day_plan: 日计划配置
Returns:
命中的 period key,或 None
"""
candidates = []
for idx, key in enumerate(day_plan.get("periods", [])):
period = self.timeline["periods"].get(key)
if period is None:
continue
if self._in_range(now_hhmm, period["start"], period["end"]):
candidates.append((idx, key))
if not candidates:
return None
# 检查冲突
if len(candidates) > 1:
policy = self.timeline.get("overlap", {}).get("policy", "error_on_overlap")
conflicting = [c[1] for c in candidates]
if policy == "error_on_overlap":
raise ValueError(
f"检测到时间段重叠冲突: {', '.join(conflicting)} 在 {now_hhmm} 重叠。"
f"请调整时间段配置,或将 overlap.policy 设为 'last_wins'"
)
# last_wins:输出重叠警告,列表中后面的优先
print(
f"[调度] 检测到时间段重叠: {', '.join(conflicting)} 在 {now_hhmm} 重叠"
)
winner = candidates[-1]
print(f"[调度] 冲突策略: last_wins,生效时间段: {winner[1]}")
return winner[1]
return candidates[0][1]
@staticmethod
def _in_range(now_hhmm: str, start: str, end: str) -> bool:
"""
检查时间是否在范围内(支持跨日)
Args:
now_hhmm: 当前时间 HH:MM
start: 开始时间 HH:MM
end: 结束时间 HH:MM
Returns:
是否在范围内
"""
if start <= end:
# 正常范围,如 08:00-09:00(半开区间 [start, end))
return start <= now_hhmm < end
else:
# 跨日范围,如 22:00-07:00(半开区间 [start, end))
return now_hhmm >= start or now_hhmm < end
def _merge_with_default(self, period_key: Optional[str]) -> Dict[str, Any]:
"""合并默认配置和时间段配置"""
base = copy.deepcopy(self.timeline.get("default", {}))
if not period_key:
return base
period = copy.deepcopy(self.timeline["periods"][period_key])
# 先合并 once 子对象
merged_once = dict(base.get("once", {}))
merged_once.update(period.get("once", {}))
# 标量字段覆盖
base.update(period)
# 恢复合并后的 once
if merged_once:
base["once"] = merged_once
return base
@staticmethod
def _resolve_ai_mode(cfg: Dict[str, Any]) -> str:
"""解析最终的 AI 模式"""
ai_mode = cfg.get("ai_mode", "follow_report")
if ai_mode == "follow_report":
return cfg.get("report_mode", "current")
return ai_mode
def already_executed(self, period_key: str, action: str, date_str: str) -> bool:
"""
检查指定时间段的某个 action 今天是否已执行
Args:
period_key: 时间段 key
action: 动作类型 (analyze / push)
date_str: 日期 YYYY-MM-DD
Returns:
是否已执行
"""
return self.storage.has_period_executed(date_str, period_key, action)
def record_execution(self, period_key: str, action: str, date_str: str) -> None:
"""
记录时间段的 action 执行
Args:
period_key: 时间段 key
action: 动作类型 (analyze / push)
date_str: 日期 YYYY-MM-DD
"""
self.storage.record_period_execution(date_str, period_key, action)
# ========================================
# 校验
# ========================================
def _validate_timeline(self, timeline: Dict[str, Any]) -> None:
"""
启动时校验 timeline 配置
Raises:
ValueError: 配置不合法时抛出
"""
required_top_keys = ["default", "periods", "day_plans", "week_map"]
for key in required_top_keys:
if key not in timeline:
raise ValueError(f"timeline 缺少必须字段: {key}")
# week_map 必须覆盖 1..7
for day in range(1, 8):
if day not in timeline["week_map"]:
raise ValueError(f"week_map 缺少星期映射: {day}")
# day_plan 引用完整性
for day, plan_key in timeline["week_map"].items():
if plan_key not in timeline["day_plans"]:
raise ValueError(
f"week_map[{day}] 引用了不存在的 day_plan: {plan_key}"
)
# period 引用完整性
for plan_key, plan in timeline["day_plans"].items():
for period_key in plan.get("periods", []):
if period_key not in timeline["periods"]:
raise ValueError(
f"day_plan[{plan_key}] 引用了不存在的 period: {period_key}"
)
# 时间格式校验
for period_key, period in timeline["periods"].items():
if "start" not in period or "end" not in period:
raise ValueError(
f"period '{period_key}' 缺少 start 或 end 字段"
)
self._validate_hhmm(period["start"], f"{period_key}.start")
self._validate_hhmm(period["end"], f"{period_key}.end")
if period["start"] == period["end"]:
raise ValueError(
f"period '{period_key}' 的 start 与 end 不能相同: {period['start']}"
)
# 检查冲突策略下的重叠
policy = timeline.get("overlap", {}).get("policy", "error_on_overlap")
if policy == "error_on_overlap":
self._check_period_overlaps(timeline)
def _check_period_overlaps(self, timeline: Dict[str, Any]) -> None:
"""
检查每个日计划中的时间段是否存在重叠
仅在 overlap.policy == "error_on_overlap" 时调用
"""
periods = timeline.get("periods", {})
for plan_key, plan in timeline["day_plans"].items():
period_keys = plan.get("periods", [])
if len(period_keys) <= 1:
continue
# 收集每个时间段的范围
ranges = []
for pk in period_keys:
p = periods.get(pk, {})
if "start" in p and "end" in p:
ranges.append((pk, p["start"], p["end"]))
# 两两检查重叠
for i in range(len(ranges)):
for j in range(i + 1, len(ranges)):
if self._ranges_overlap(
ranges[i][1], ranges[i][2],
ranges[j][1], ranges[j][2],
):
raise ValueError(
f"day_plan '{plan_key}' 中时间段 '{ranges[i][0]}' "
f"({ranges[i][1]}-{ranges[i][2]}) 与 '{ranges[j][0]}' "
f"({ranges[j][1]}-{ranges[j][2]}) 存在重叠。"
f"请调整时间段,或将 overlap.policy 设为 'last_wins'"
)
@staticmethod
def _ranges_overlap(s1: str, e1: str, s2: str, e2: str) -> bool:
"""检查两个时间范围是否重叠(支持跨日)"""
def to_minutes(t: str) -> int:
h, m = t.split(":")
return int(h) * 60 + int(m)
def expand_range(start: str, end: str) -> List[tuple]:
"""将时间范围展开为分钟段列表,跨日时拆分为两段"""
s = to_minutes(start)
e = to_minutes(end)
if s <= e:
return [(s, e)]
else:
# 跨日:拆分为 [start, 24:00) 和 [00:00, end)
return [(s, 24 * 60), (0, e)]
segs1 = expand_range(s1, e1)
segs2 = expand_range(s2, e2)
for a_start, a_end in segs1:
for b_start, b_end in segs2:
# 两个半开区间有重叠的条件
if a_start < b_end and b_start < a_end:
return True
return False
@staticmethod
def _validate_hhmm(value: str, field_name: str) -> None:
"""校验 HH:MM 格式"""
if not re.match(r"^\d{2}:\d{2}$", value):
raise ValueError(f"{field_name} 格式错误: '{value}',期望 HH:MM")
h, m = value.split(":")
if not (0 <= int(h) <= 23 and 0 <= int(m) <= 59):
raise ValueError(f"{field_name} 时间值超出范围: '{value}'")