199 lines
8.8 KiB
Python
199 lines
8.8 KiB
Python
"""
|
||
配置模块:系统常量、RSS源、关键词库、评分维度、分类定义、AI配置加载
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
from pathlib import Path
|
||
|
||
# =============================================================================
|
||
# 军事科技 RSS 订阅源
|
||
# 优先读取 config/rss_feeds.txt 文件,文件不存在则使用默认12个源
|
||
# =============================================================================
|
||
def load_rss_feeds() -> list:
|
||
"""从配置文件读取RSS源,格式每行:名称|URL,#开头为注释"""
|
||
default_feeds = [
|
||
{"name": "Defense One", "url": "https://www.defenseone.com/rss/all/"},
|
||
{"name": "NEWUAS", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3246130840.rss"},
|
||
{"name": "Seapower", "url": "https://seapowermagazine.org/feed/"},
|
||
{"name": "The War Zone", "url": "https://www.twz.com/feed"},
|
||
{"name": "国防科技要闻", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3921202880.rss"},
|
||
{"name": "战略前沿技术", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3271903347.rss"},
|
||
{"name": "无人机邦", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3957361285.rss"},
|
||
{"name": "浮空飞行器", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3291643707.rss"},
|
||
{"name": "海鹰资讯", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3091381580.rss"},
|
||
{"name": "渊亭防务", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3865628826.rss"},
|
||
{"name": "电波之矛", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3004998095.rss"},
|
||
{"name": "龙牙的一座山", "url": f"{RSS_PROXY_BASE}/feed/MP_WXS_3517933009.rss"},
|
||
]
|
||
|
||
# RSS配置文件路径:项目根目录/config/rss_feeds.txt
|
||
config_file = Path(__file__).resolve().parent.parent.parent / "config" / "rss_feeds.txt"
|
||
|
||
if not config_file.exists():
|
||
return default_feeds
|
||
|
||
feeds = []
|
||
with open(config_file, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
# 跳过空行和注释行
|
||
if not line or line.startswith('#'):
|
||
continue
|
||
# 按|分割,确保格式正确
|
||
parts = line.split('|', 1)
|
||
if len(parts) == 2:
|
||
name = parts[0].strip()
|
||
url = parts[1].strip()
|
||
if name and url:
|
||
feeds.append({"name": name, "url": url})
|
||
|
||
# 如果读取到的源为空,返回默认源
|
||
return feeds if feeds else default_feeds
|
||
|
||
# =============================================================================
|
||
# RSS 代理基础 URL(read_env 尚未定义,直接读文件/file和ennv)
|
||
# =============================================================================
|
||
def _load_rss_proxy_base():
|
||
default = "https://werss.yynnice.top"
|
||
env_val = os.environ.get("RSS_PROXY_BASE")
|
||
if env_val:
|
||
return env_val
|
||
env_file = Path(__file__).resolve().parent.parent.parent / "config" / ".env"
|
||
if env_file.exists():
|
||
with open(env_file, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and not line.startswith('#') and line.startswith('RSS_PROXY_BASE='):
|
||
_, _, val = line.partition('=')
|
||
return val.strip().strip('"').strip("'")
|
||
return default
|
||
|
||
RSS_PROXY_BASE = _load_rss_proxy_base()
|
||
|
||
RSS_FEEDS = load_rss_feeds()
|
||
|
||
WECHAT_SOURCES = {'NEWUAS', '国防科技要闻', '战略前沿技术', '无人机邦', '浮空飞行器', '海鹰资讯', '渊亭防务', '电波之矛', '龙牙的一座山'}
|
||
|
||
# =============================================================================
|
||
# 抓取配置
|
||
# =============================================================================
|
||
FETCH_TIMEOUT = 30
|
||
MAX_THREADS = 15 # 抓取并发数,100+源建议15-20
|
||
AI_CONCURRENCY = 8 # AI调用并发数,根据API限流调整
|
||
TIME_WINDOW_HOURS = 24
|
||
|
||
# =============================================================================
|
||
# 关键词库(44个)
|
||
# =============================================================================
|
||
KEYWORDS = [
|
||
"首飞", "曝光", "美军", "解放军", "新型", "研发", "部署", "军演", "导弹", "战机",
|
||
"舰艇", "雷达", "芯片", "突破", "测试", "服役", "列装", "武器", "装备", "军事",
|
||
"武器系统", "冲突", "封锁", "威慑", "无人机", "AI", "人工智能", "战争", "以色列",
|
||
"伊朗", "俄罗斯", "乌克兰", "航母", "潜艇", "坦克", "防空", "反导", "激光", "电磁",
|
||
"卫星", "航天", "蜂群", "拦截", "打击", "攻击", "作战", "战略", "战术", "军工"
|
||
]
|
||
|
||
# =============================================================================
|
||
# AI评分维度(4维度加权)
|
||
# =============================================================================
|
||
SCORE_DIMENSIONS = [
|
||
{"name": "技术突破性", "weight": 0.35, "description": "是否有新技术、新武器、新架构曝光,1-10分"},
|
||
{"name": "战略价值", "weight": 0.30, "description": "对地缘政治、军事格局的影响程度,1-10分"},
|
||
{"name": "信息可信度", "weight": 0.20, "description": "来源权威性、数据可验证性,1-10分"},
|
||
{"name": "时效性", "weight": 0.15, "description": "事件的新鲜程度,1-10分"},
|
||
]
|
||
|
||
# =============================================================================
|
||
# 分类定义
|
||
# =============================================================================
|
||
CATEGORIES = {
|
||
"today_must_read": {"name": "🔥 今日必看", "desc": "最重磅、最紧急的头条要闻"},
|
||
"equipment": {"name": "⚙️ 装备动态", "desc": "新型武器、技术突破、军备交付"},
|
||
"conflict": {"name": "💥 地区冲突", "desc": "热点战事、边境对峙、反恐行动"},
|
||
"strategy": {"name": "🎯 战略政策", "desc": "国防战略、军演部署、外交安全"},
|
||
}
|
||
|
||
|
||
def _read_env_file(file_path):
|
||
"""从.env文件读取配置,返回 {key: value} 字典"""
|
||
result = {}
|
||
if file_path.exists():
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and not line.startswith('#') and '=' in line:
|
||
key, _, value = line.partition('=')
|
||
value = value.strip().strip('"').strip("'")
|
||
result[key] = value
|
||
return result
|
||
|
||
|
||
def load_ai_config():
|
||
"""加载AI配置(环境变量 > 项目config/.env > 硬编码默认值)"""
|
||
api_key = None
|
||
base_url = "https://api.openai.com/v1"
|
||
model = "gpt-4o-mini"
|
||
webzine_model = "deepseek-v3.2"
|
||
|
||
local_env = Path(__file__).resolve().parent.parent.parent / "config" / ".env"
|
||
values = _read_env_file(local_env)
|
||
if values.get('OPENAI_API_KEY'):
|
||
api_key = values['OPENAI_API_KEY']
|
||
if values.get('OPENAI_API_BASE'):
|
||
base_url = values['OPENAI_API_BASE']
|
||
if values.get('OPENAI_MODEL'):
|
||
model = values['OPENAI_MODEL']
|
||
if values.get('WEBZINE_MODEL'):
|
||
webzine_model = values['WEBZINE_MODEL']
|
||
|
||
if os.environ.get('OPENAI_API_KEY'):
|
||
api_key = os.environ['OPENAI_API_KEY']
|
||
if os.environ.get('OPENAI_API_BASE'):
|
||
base_url = os.environ['OPENAI_API_BASE']
|
||
if os.environ.get('OPENAI_MODEL'):
|
||
model = os.environ['OPENAI_MODEL']
|
||
if os.environ.get('WEBZINE_MODEL'):
|
||
webzine_model = os.environ['WEBZINE_MODEL']
|
||
|
||
return api_key, base_url, model, webzine_model
|
||
|
||
|
||
def load_push_config():
|
||
"""加载推送配置(环境变量 > config.json > 默认值)"""
|
||
config = {
|
||
"enable_wechat_push": False,
|
||
"enable_feishu_push": False,
|
||
"wechat_key": "",
|
||
"wechat_target": "",
|
||
"feishu_webhook": "",
|
||
}
|
||
|
||
config_file = Path(__file__).resolve().parent.parent.parent / "config" / "config.json"
|
||
values = _read_env_file(Path(__file__).resolve().parent.parent.parent / "config" / ".env")
|
||
|
||
if config_file.exists():
|
||
try:
|
||
with open(config_file, 'r', encoding='utf-8') as f:
|
||
json_config = json.load(f)
|
||
config["enable_wechat_push"] = json_config.get("enable_wechat_push", False)
|
||
config["enable_feishu_push"] = json_config.get("enable_feishu_push", False)
|
||
except Exception:
|
||
pass
|
||
|
||
if values.get("WECHAT_ACCOUNT"):
|
||
config["wechat_key"] = values["WECHAT_ACCOUNT"]
|
||
if values.get("WECHAT_TARGET"):
|
||
config["wechat_target"] = values["WECHAT_TARGET"]
|
||
if values.get("FEISHU_WEBHOOK"):
|
||
config["feishu_webhook"] = values["FEISHU_WEBHOOK"]
|
||
|
||
if os.environ.get("WECHAT_ACCOUNT"):
|
||
config["wechat_key"] = os.environ["WECHAT_ACCOUNT"]
|
||
if os.environ.get("WECHAT_TARGET"):
|
||
config["wechat_target"] = os.environ["WECHAT_TARGET"]
|
||
if os.environ.get("FEISHU_WEBHOOK"):
|
||
config["feishu_webhook"] = os.environ["FEISHU_WEBHOOK"]
|
||
|
||
return config
|