diff --git a/config/config.yaml b/config/config.yaml index b743c7c76a447..bb5dddf5d07a8 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -395,6 +395,12 @@ ai: max_tokens: 5000 # 最大生成 token 数 # 注意：如果 API 不支持此参数(报 HTTP 400)，请设为 0 以禁用发送 + + # 推理强度（仅推理型模型有效，如 gpt-5 / o 系列 / grok；普通模型请留空） + # 留空 = 不发送该参数 + # 可选值: minimal / low / medium / high + reasoning_effort: "" + # 高级选项 num_retries: 1 # 失败重试次数 fallback_models: [] # 备用模型列表（可选） diff --git a/docker/docker-compose-build.yml b/docker/docker-compose-build.yml index bed6d4549e0c0..ad52f9132829f 100644 --- a/docker/docker-compose-build.yml +++ b/docker/docker-compose-build.yml @@ -12,6 +12,9 @@ services: volumes: - ../config:/app/config:ro - ../output:/app/output + - ../trendradar:/app/trendradar + - ../docker/manage.py:/app/manage.py + - ../docker/entrypoint.sh:/entrypoint.sh:ro environment: - TZ=Asia/Shanghai diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 7948f3a298077..10f43748bb82d 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -10,6 +10,9 @@ services: volumes: - ../config:/app/config:ro - ../output:/app/output + - ../trendradar:/app/trendradar + - ../docker/manage.py:/app/manage.py + - ../docker/entrypoint.sh:/entrypoint.sh:ro environment: - TZ=Asia/Shanghai diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh old mode 100644 new mode 100755 index a14de73ef6c82..fdf4b6eb0209a --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -31,16 +31,16 @@ case "${RUN_MODE:-cron}" in exit 1 fi + # 先启动 Web 服务器，避免 IMMEDIATE_RUN 抓取阻塞面板访问 + echo "🌐 启动 Web 服务器..." + python manage.py start_webserver + # 立即执行一次（如果配置了） if [ "${IMMEDIATE_RUN:-false}" = "true" ]; then echo "▶️ 立即执行一次" python -m trendradar fi - # 启动 Web 服务器 - echo "🌐 启动 Web 服务器..." - python manage.py start_webserver - echo "⏰ 启动supercronic: $CRON_EXPR" echo "🎯 supercronic 将作为 PID 1 运行" diff --git a/docker/manage.py b/docker/manage.py index d58685c4006bf..7f7b7268394af 100644 --- a/docker/manage.py +++ b/docker/manage.py @@ -454,7 +454,10 @@ def _is_expected_webserver_process(pid: int) -> bool: cmdline = _read_proc_cmdline(pid) if not cmdline: return False - return "http.server" in cmdline and str(WEBSERVER_PORT) in cmdline + return ( + str(WEBSERVER_PORT) in cmdline + and ("http.server" in cmdline or "trendradar.web_control" in cmdline) + ) def _terminate_webserver_process(pid: int, require_expected: bool = True) -> bool: @@ -531,7 +534,7 @@ def _cleanup_stale_pid(): def start_webserver(): """启动 Web 服务器托管 output 目录""" print(f"🌐 启动 Web 服务器 (端口: {WEBSERVER_PORT})...") - print(f" 🔒 安全提示：仅提供静态文件访问，限制在 {WEBSERVER_DIR} 目录") + print(f" 🔒 报告目录: {WEBSERVER_DIR}；控制面板可手动抓取 / 分析") # 检查是否已经运行 if Path(WEBSERVER_PID_FILE).exists(): @@ -564,12 +567,21 @@ def start_webserver(): # 启动 HTTP 服务器 # 使用 --bind 绑定到 0.0.0.0 使容器内部可访问 # 工作目录限制在 WEBSERVER_DIR，防止访问其他目录 + here = Path(__file__).resolve().parent + project_root = str(here if (here / "trendradar").is_dir() else here.parent) process = subprocess.Popen( - [sys.executable, '-m', 'http.server', str(WEBSERVER_PORT), '--bind', '0.0.0.0'], - cwd=WEBSERVER_DIR, + [ + sys.executable, + "-m", + "trendradar.web_control", + str(WEBSERVER_PORT), + WEBSERVER_DIR, + project_root, + ], + cwd=project_root, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - start_new_session=True + start_new_session=True, ) # 等待一下确保服务器启动 @@ -581,7 +593,8 @@ def start_webserver(): with open(WEBSERVER_PID_FILE, 'w') as f: f.write(str(process.pid)) print(f" ✅ Web 服务器已启动 (PID: {process.pid})") - print(f" 📁 服务目录: {WEBSERVER_DIR} (只读，仅静态文件)") + print(f" 📁 服务目录: {WEBSERVER_DIR}") + print(f" 🎛️ 控制面板: 报告页顶部可手动抓取 / AI 分析 / 切换预设") print(f" 🌐 访问地址: http://localhost:{WEBSERVER_PORT}") print(f" 📄 首页: http://localhost:{WEBSERVER_PORT}/index.html") print(" 💡 停止服务: python manage.py stop_webserver") diff --git a/tests/test_ai_client.py b/tests/test_ai_client.py new file mode 100644 index 0000000000000..89191383b4745 --- /dev/null +++ b/tests/test_ai_client.py @@ -0,0 +1,79 @@ +# coding=utf-8 +import unittest +from unittest.mock import patch + + +def _fake_response(): + class Message: + content = "ok" + + class Choice: + message = Message() + + class Resp: + choices = [Choice()] + + return Resp() + + +class AIClientParamTests(unittest.TestCase): + def _capture(self, config, kwargs=None): + from trendradar.ai import client as client_mod + + captured = {} + + def fake_completion(**params): + captured.update(params) + return _fake_response() + + with patch.object(client_mod, "completion", fake_completion): + client = client_mod.AIClient(config) + client.chat([{"role": "user", "content": "hi"}], **(kwargs or {})) + return captured + + def test_reasoning_effort_sent_when_set(self): + # openai/ 前缀（含自定义兼容端点）走 extra_body 透传，避免 litellm 白名单校验 + params = self._capture( + {"MODEL": "openai/custom-reasoner", "API_KEY": "sk-x", "REASONING_EFFORT": " HIGH "} + ) + self.assertEqual(params["extra_body"]["reasoning_effort"], "high") + self.assertNotIn("reasoning_effort", params) + + def test_reasoning_effort_top_level_for_other_providers(self): + # 非 openai 提供商走顶层参数，由 litellm 完成跨商映射 + params = self._capture( + {"MODEL": "anthropic/claude-sonnet-4", "API_KEY": "sk-x", "REASONING_EFFORT": "high"} + ) + self.assertEqual(params["reasoning_effort"], "high") + self.assertNotIn("extra_body", params) + + def test_reasoning_effort_omitted_when_empty(self): + params = self._capture( + {"MODEL": "openai/gpt-4o", "API_KEY": "sk-x", "REASONING_EFFORT": ""} + ) + self.assertNotIn("reasoning_effort", params) + self.assertNotIn("extra_body", params) + + def test_reasoning_effort_overridable_per_call(self): + params = self._capture( + {"MODEL": "openai/gpt-5", "API_KEY": "sk-x", "REASONING_EFFORT": "low"}, + {"reasoning_effort": "high"}, + ) + self.assertEqual(params["extra_body"]["reasoning_effort"], "high") + + def test_extra_params_merged_without_override(self): + params = self._capture( + { + "MODEL": "openai/gpt-4o", + "API_KEY": "sk-x", + "TEMPERATURE": 0.5, + "EXTRA_PARAMS": {"top_p": 0.9, "temperature": 2.0}, + } + ) + self.assertEqual(params["top_p"], 0.9) + # 显式配置的 temperature 优先于 extra_params + self.assertEqual(params["temperature"], 0.5) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_web_control.py b/tests/test_web_control.py new file mode 100644 index 0000000000000..a610e53c39295 --- /dev/null +++ b/tests/test_web_control.py @@ -0,0 +1,414 @@ +# coding=utf-8 +import json +import os +import threading +import unittest +from http.client import HTTPConnection +from http.server import ThreadingHTTPServer +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import MagicMock, patch + + +class ParseJobProgressTests(unittest.TestCase): + def test_parse_platform_and_rss_progress(self): + from trendradar.web_control import parse_job_progress + + log = """ +配置的监控平台: ['今日头条', '百度热搜', '微博'] +开始爬取数据，请求间隔 2000 毫秒 +正在获取 今日头条（1/3）... +获取 toutiao 成功（缓存数据） +正在获取 百度热搜（2/3）... +获取 baidu 成功（缓存数据） +正在获取 微博（3/3）... +获取 weibo 成功（缓存数据） +成功: ['toutiao', 'baidu', 'weibo'], 失败: [] +[RSS] 开始抓取 2 个 RSS 源... +[RSS] 正在获取 Hacker News（1/2）... +[RSS] Hacker News: 获取 20 条 +""" + progress = parse_job_progress(log, mode="crawl") + self.assertEqual(progress["phase"], "rss") + self.assertIn("Hacker News", progress["message"]) + self.assertEqual(progress["platforms_total"], 3) + self.assertEqual(progress["rss_total"], 2) + self.assertGreaterEqual(progress["percent"], 50) + self.assertLess(progress["percent"], 100) + + def test_parse_ai_phase(self): + from trendradar.web_control import parse_job_progress + + progress = parse_job_progress("[AI] 正在进行 AI 分析...\n", mode="analyze") + self.assertEqual(progress["phase"], "ai") + self.assertIn("AI", progress["message"]) + self.assertGreaterEqual(progress["percent"], 40) + + +class WebControlApiTests(unittest.TestCase): + def setUp(self): + from trendradar.web_control import ControlState, make_handler + + self.tmp = TemporaryDirectory() + self.output = Path(self.tmp.name) + (self.output / "index.html").write_text("report", encoding="utf-8") + + # 模拟项目根目录：config/frequency_words.txt + config/config.yaml + self.project_root = self.output / "project" + self.config_dir = self.project_root / "config" + self.config_dir.mkdir(parents=True) + (self.config_dir / "frequency_words.txt").write_text( + "[GLOBAL_FILTER]\n震惊\n\n[WORD_GROUPS]\n华为\n", encoding="utf-8" + ) + (self.config_dir / "config.yaml").write_text( + "rss:\n" + " enabled: true\n" + " feeds:\n" + " - id: hacker-news\n" + " name: Hacker News\n" + " url: https://hnrss.org/frontpage\n" + "ai:\n" + " model: deepseek/deepseek-v4-flash\n" + " api_key: \"\"\n" + " api_base: \"\"\n", + encoding="utf-8", + ) + + self.runner = MagicMock() + self.runner.is_running.return_value = False + self.runner.status.return_value = { + "running": False, + "mode": None, + "error": None, + "progress": {"phase": "idle", "message": "", "percent": 0}, + "log_lines": 0, + } + self.runner.read_log.return_value = { + "text": "hello log", + "lines": 1, + "tail": 1, + "path": ".webui-job.log", + "exists": True, + } + self.state = ControlState( + output_dir=self.output, runner=self.runner, project_root=self.project_root + ) + + handler = make_handler(self.state) + self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + self.port = self.server.server_address[1] + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.tmp.cleanup() + + def _request(self, method, path, body=None): + conn = HTTPConnection("127.0.0.1", self.port, timeout=5) + payload = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Content-Type": "application/json"} if payload else {} + conn.request(method, path, body=payload, headers=headers) + resp = conn.getresponse() + data = resp.read() + conn.close() + return resp.status, resp.getheader("Content-Type"), data + + def test_root_serves_index_and_injects_toolbar(self): + status, content_type, data = self._request("GET", "/") + html = data.decode("utf-8") + self.assertEqual(status, 200) + self.assertIn("text/html", content_type) + self.assertIn("tr-toolbar", html) + self.assertIn("tr-progress-wrap", html) + self.assertIn("tr-log-panel", html) + self.assertIn("report", html) + + def test_get_settings_contains_presets_and_ai_flag(self): + status, _, data = self._request("GET", "/api/settings") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertIn("ai_analysis_enabled", payload) + ids = [item["id"] for item in payload["presets"]] + self.assertIn("morning_evening", ids) + self.assertIn("off", ids) + + def test_post_settings_persists_overlay(self): + status, _, data = self._request( + "POST", + "/api/settings", + {"ai_analysis_enabled": False, "schedule_preset": "office_hours"}, + ) + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertFalse(payload["ai_analysis_enabled"]) + self.assertEqual(payload["schedule_preset"], "office_hours") + self.assertTrue(payload["schedule_enabled"]) + + _, _, again = self._request("GET", "/api/settings") + self.assertEqual(json.loads(again)["schedule_preset"], "office_hours") + + def test_post_run_crawl_starts_job(self): + status, _, data = self._request("POST", "/api/run", {"mode": "crawl"}) + payload = json.loads(data) + self.assertEqual(status, 202) + self.assertEqual(payload["mode"], "crawl") + self.runner.start.assert_called_once_with("crawl") + + def test_post_run_analyze_starts_job(self): + status, _, data = self._request("POST", "/api/run", {"mode": "analyze"}) + self.assertEqual(status, 202) + self.assertEqual(json.loads(data)["mode"], "analyze") + self.runner.start.assert_called_once_with("analyze") + + def test_post_run_rejects_when_busy(self): + self.runner.is_running.return_value = True + status, _, data = self._request("POST", "/api/run", {"mode": "crawl"}) + self.assertEqual(status, 409) + self.assertIn("running", json.loads(data)["error"]) + + def test_get_logs_returns_text_and_progress(self): + status, _, data = self._request("GET", "/api/logs?tail=100") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["text"], "hello log") + self.assertIn("progress", payload) + self.runner.read_log.assert_called() + # tail query should be forwarded + args, kwargs = self.runner.read_log.call_args + self.assertEqual(kwargs.get("tail") or (args[0] if args else None), 100) + + def test_get_topics_reads_file_when_no_overlay(self): + status, _, data = self._request("GET", "/api/topics") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "file") + self.assertIn("[WORD_GROUPS]", payload["content"]) + self.assertIn("华为", payload["content"]) + self.assertGreaterEqual(payload["group_count"], 1) + + def test_post_and_delete_topics(self): + content = "[WORD_GROUPS]\nwebui词\n" + status, _, data = self._request("POST", "/api/topics", {"content": content}) + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "overlay") + self.assertIn("webui词", payload["content"]) + self.assertEqual(payload["group_count"], 1) + + _, _, again = self._request("GET", "/api/topics") + self.assertEqual(json.loads(again)["source"], "overlay") + + status, _, data = self._request("DELETE", "/api/topics") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "file") + self.assertIn("华为", payload["content"]) + + def test_post_topics_rejects_non_string(self): + status, _, data = self._request("POST", "/api/topics", {"content": 123}) + self.assertEqual(status, 400) + self.assertIn("error", json.loads(data)) + + def test_get_feeds_reads_yaml_when_no_overlay(self): + status, _, data = self._request("GET", "/api/feeds") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "config") + self.assertTrue(payload["rss_enabled"]) + self.assertEqual(len(payload["feeds"]), 1) + self.assertEqual(payload["feeds"][0]["id"], "hacker-news") + self.assertTrue(payload["feeds"][0]["enabled"]) + + def test_post_and_delete_feeds(self): + body = { + "rss_enabled": False, + "feeds": [ + {"name": "My Feed", "url": "https://example.com/rss.xml"}, + { + "id": "second", + "name": "Second", + "url": "https://example.org/feed", + "enabled": False, + "max_age_days": 5, + }, + ], + } + status, _, data = self._request("POST", "/api/feeds", body) + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "overlay") + self.assertFalse(payload["rss_enabled"]) + self.assertEqual(payload["feeds"][0]["id"], "my-feed") + self.assertEqual(payload["feeds"][1]["max_age_days"], 5) + + # overlay 应被 apply_overlay 应用到运行配置 + from trendradar.webui_settings import apply_overlay + + config = { + "RSS": { + "ENABLED": True, + "FEEDS": [{"id": "old", "name": "旧", "url": "https://old.example/rss"}], + } + } + apply_overlay(config, output_dir=self.output) + self.assertFalse(config["RSS"]["ENABLED"]) + self.assertEqual(len(config["RSS"]["FEEDS"]), 2) + + status, _, data = self._request("DELETE", "/api/feeds") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "config") + self.assertEqual(payload["feeds"][0]["id"], "hacker-news") + + def test_post_feeds_rejects_invalid(self): + status, _, data = self._request( + "POST", "/api/feeds", {"rss_enabled": True, "feeds": [{"name": "x", "url": "not-a-url"}]} + ) + self.assertEqual(status, 400) + self.assertIn("error", json.loads(data)) + + def _clean_ai_env(self): + return patch.dict( + os.environ, {"AI_API_KEY": "", "AI_MODEL": "", "AI_API_BASE": ""}, clear=False + ) + + def test_get_ai_reads_yaml_when_no_overlay(self): + with self._clean_ai_env(): + status, _, data = self._request("GET", "/api/ai") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "config") + self.assertEqual(payload["model"], "deepseek/deepseek-v4-flash") + self.assertFalse(payload["api_key_set"]) + self.assertIsNone(payload["api_key_masked"]) + self.assertEqual(payload["reasoning_effort"], "") + # 默认不返回明文密钥字段 + self.assertNotIn("api_key", payload) + + def test_post_and_delete_ai(self): + with self._clean_ai_env(): + status, _, data = self._request( + "POST", + "/api/ai", + { + "model": "openai/gpt-4o", + "api_base": "", + "api_key": "sk-test-1234567890", + "reasoning_effort": "high", + }, + ) + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "overlay") + self.assertTrue(payload["api_key_set"]) + self.assertEqual(payload["api_key_masked"], "sk-***7890") + self.assertEqual(payload["reasoning_effort"], "high") + # 明文密钥绝不能出现在响应里 + self.assertNotIn("sk-test-1234567890", data.decode("utf-8")) + + # 密钥留空再次保存：密钥保留，仅更新模型 + status, _, data = self._request( + "POST", + "/api/ai", + {"model": "deepseek/deepseek-v4-pro", "api_base": "", "api_key": "", "reasoning_effort": ""}, + ) + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["model"], "deepseek/deepseek-v4-pro") + self.assertEqual(payload["reasoning_effort"], "") + self.assertTrue(payload["api_key_set"]) + self.assertNotIn("sk-test-1234567890", data.decode("utf-8")) + + # overlay 应覆盖运行配置中的 AI 设置 + from trendradar.webui_settings import apply_overlay + + config = {"AI": {"MODEL": "old", "API_KEY": "old", "API_BASE": "", "REASONING_EFFORT": ""}} + apply_overlay(config, output_dir=self.output) + self.assertEqual(config["AI"]["MODEL"], "deepseek/deepseek-v4-pro") + self.assertEqual(config["AI"]["API_KEY"], "sk-test-1234567890") + self.assertEqual(config["AI"]["REASONING_EFFORT"], "") + + status, _, data = self._request("DELETE", "/api/ai") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["source"], "config") + self.assertFalse(payload["api_key_set"]) + + def test_get_ai_reveal_returns_plaintext_on_request(self): + with self._clean_ai_env(): + self._request( + "POST", "/api/ai", {"model": "m", "api_base": "", "api_key": "sk-reveal-123456", "reasoning_effort": ""} + ) + # 普通GET：无明文 + _, _, data = self._request("GET", "/api/ai") + self.assertNotIn("sk-reveal-123456", data.decode("utf-8")) + # reveal=1：返回明文（面板“显示”按钮查看已保存密钥） + status, _, data = self._request("GET", "/api/ai?reveal=1") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertEqual(payload["api_key"], "sk-reveal-123456") + self._request("DELETE", "/api/ai") + + def test_post_ai_rejects_invalid(self): + with self._clean_ai_env(): + status, _, data = self._request("POST", "/api/ai", {"model": " ", "api_key": "sk-x"}) + self.assertEqual(status, 400) + self.assertIn("model", json.loads(data)["error"]) + status, _, data = self._request( + "POST", "/api/ai", {"model": "m", "api_base": "ftp://x"} + ) + self.assertEqual(status, 400) + self.assertIn("api_base", json.loads(data)["error"]) + status, _, data = self._request( + "POST", "/api/ai", {"model": "m", "reasoning_effort": "extreme"} + ) + self.assertEqual(status, 400) + self.assertIn("reasoning_effort", json.loads(data)["error"]) + + def test_toolbar_contains_config_modal(self): + status, _, data = self._request("GET", "/") + html = data.decode("utf-8") + self.assertEqual(status, 200) + self.assertIn("tr-config-overlay", html) + self.assertIn("tr-topics-text", html) + self.assertIn("tr-feeds-list", html) + self.assertIn("tr-panel-ai", html) + self.assertIn("tr-ai-key", html) + + def test_get_status_includes_progress_field(self): + self.runner.status.return_value = { + "running": True, + "mode": "crawl", + "error": None, + "progress": {"phase": "platforms", "message": "正在抓取热榜：微博（3/11）", "percent": 40}, + "log_lines": 12, + } + status, _, data = self._request("GET", "/api/status") + payload = json.loads(data) + self.assertEqual(status, 200) + self.assertTrue(payload["running"]) + self.assertEqual(payload["progress"]["percent"], 40) + self.assertIn("微博", payload["progress"]["message"]) + + +class JobRunnerLogTests(unittest.TestCase): + def test_read_log_tail(self): + from trendradar.web_control import JobRunner + + with TemporaryDirectory() as tmp: + output = Path(tmp) + runner = JobRunner(project_root=tmp, output_dir=output) + log_path = output / ".webui-job.log" + log_path.write_text("\n".join(f"line-{i}" for i in range(1, 21)), encoding="utf-8") + payload = runner.read_log(tail=5) + self.assertTrue(payload["exists"]) + self.assertEqual(payload["lines"], 20) + self.assertIn("line-20", payload["text"]) + self.assertIn("已省略前 15 行", payload["text"]) + self.assertNotIn("line-1\n", payload["text"].split("…", 1)[-1] if "…" in payload["text"] else payload["text"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_webui_settings.py b/tests/test_webui_settings.py new file mode 100644 index 0000000000000..e18c0c3ef5a38 --- /dev/null +++ b/tests/test_webui_settings.py @@ -0,0 +1,299 @@ +# coding=utf-8 +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + + +class WebuiSettingsTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.output = Path(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_missing_overlay_returns_empty_dict(self): + from trendradar.webui_settings import load_overlay + + self.assertEqual(load_overlay(self.output), {}) + + def test_save_and_load_roundtrip(self): + from trendradar.webui_settings import load_overlay, save_overlay + + saved = save_overlay( + { + "ai_analysis_enabled": False, + "schedule_enabled": True, + "schedule_preset": "morning_evening", + }, + self.output, + ) + self.assertFalse(saved["ai_analysis_enabled"]) + self.assertEqual(load_overlay(self.output)["schedule_preset"], "morning_evening") + self.assertTrue((self.output / ".webui.json").exists()) + + def test_rejects_unknown_preset(self): + from trendradar.webui_settings import save_overlay + + with self.assertRaises(ValueError): + save_overlay({"schedule_preset": "not-a-preset"}, self.output) + + def test_apply_overlay_overrides_yaml_and_env(self): + from trendradar.webui_settings import apply_overlay, save_overlay + + save_overlay( + { + "ai_analysis_enabled": False, + "schedule_enabled": True, + "schedule_preset": "office_hours", + }, + self.output, + ) + config = { + "AI_ANALYSIS": {"ENABLED": True}, + "SCHEDULE": {"enabled": False, "preset": "always_on"}, + } + apply_overlay(config, output_dir=self.output) + self.assertFalse(config["AI_ANALYSIS"]["ENABLED"]) + self.assertTrue(config["SCHEDULE"]["enabled"]) + self.assertEqual(config["SCHEDULE"]["preset"], "office_hours") + + def test_one_shot_env_beats_overlay(self): + from trendradar.webui_settings import apply_overlay, save_overlay + + save_overlay({"ai_analysis_enabled": False}, self.output) + config = {"AI_ANALYSIS": {"ENABLED": False}, "SCHEDULE": {"enabled": False, "preset": "always_on"}} + with patch.dict(os.environ, {"WEBUI_RUN_AI": "true"}, clear=False): + apply_overlay(config, output_dir=self.output) + self.assertTrue(config["AI_ANALYSIS"]["ENABLED"]) + + def test_preset_off_disables_schedule(self): + from trendradar.webui_settings import save_overlay + + saved = save_overlay({"schedule_preset": "off"}, self.output) + self.assertFalse(saved["schedule_enabled"]) + self.assertEqual(saved["schedule_preset"], "off") + + +class WebuiSettingsTopicsTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.output = Path(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_save_and_load_frequency_words(self): + from trendradar.webui_settings import load_overlay, save_overlay + + content = "[GLOBAL_FILTER]\n震惊\n\n[WORD_GROUPS]\n华为\n/抖音|TikTok/ => 字节跳动\n" + save_overlay({"frequency_words": content}, self.output) + self.assertEqual(load_overlay(self.output)["frequency_words"], content) + + def test_rejects_non_string_frequency_words(self): + from trendradar.webui_settings import save_overlay + + with self.assertRaises(ValueError): + save_overlay({"frequency_words": ["not", "a", "string"]}, self.output) + + def test_remove_overlay_keys_restores_file_behavior(self): + from trendradar.webui_settings import load_overlay, remove_overlay_keys, save_overlay + + save_overlay({"frequency_words": "华为", "ai_analysis_enabled": False}, self.output) + remove_overlay_keys(["frequency_words"], self.output) + data = load_overlay(self.output) + self.assertNotIn("frequency_words", data) + self.assertFalse(data["ai_analysis_enabled"]) + + +class WebuiSettingsFeedsTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.output = Path(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_save_valid_feeds_normalized(self): + from trendradar.webui_settings import load_overlay, save_overlay + + save_overlay( + { + "rss": { + "enabled": True, + "feeds": [ + {"name": "Hacker News", "url": "https://hnrss.org/frontpage"}, + { + "id": "ruanyifeng", + "name": "阮一峰", + "url": "http://www.ruanyifeng.com/blog/atom.xml", + "enabled": False, + "max_age_days": "3", + }, + ], + } + }, + self.output, + ) + feeds = load_overlay(self.output)["rss"]["feeds"] + self.assertEqual(feeds[0]["id"], "hacker-news") + self.assertTrue(feeds[0]["enabled"]) + self.assertNotIn("max_age_days", feeds[0]) + self.assertEqual(feeds[1]["max_age_days"], 3) + self.assertFalse(feeds[1]["enabled"]) + + def test_chinese_name_falls_back_to_url_host(self): + from trendradar.webui_settings import save_overlay, load_overlay + + save_overlay( + { + "rss": { + "feeds": [ + {"name": "阮一峰", "url": "http://www.ruanyifeng.com/blog/atom.xml"}, + {"name": "阮一峰备份", "url": "http://www.ruanyifeng.com/blog/feed.xml"}, + ], + } + }, + self.output, + ) + feeds = load_overlay(self.output)["rss"]["feeds"] + self.assertEqual(feeds[0]["id"], "ruanyifeng") + self.assertEqual(feeds[1]["id"], "ruanyifeng-2") + + def test_rejects_bad_url_and_duplicate_ids(self): + from trendradar.webui_settings import save_overlay + + with self.assertRaises(ValueError): + save_overlay( + {"rss": {"feeds": [{"id": "a", "url": "ftp://example.com/rss"}]}}, + self.output, + ) + with self.assertRaises(ValueError): + save_overlay( + { + "rss": { + "feeds": [ + {"id": "a", "url": "https://x.com/1"}, + {"id": "a", "url": "https://x.com/2"}, + ] + } + }, + self.output, + ) + + def test_rejects_unknown_rss_fields(self): + from trendradar.webui_settings import save_overlay + + with self.assertRaises(ValueError): + save_overlay({"rss": {"enabled": True, "timeout": 15}}, self.output) + + def test_apply_overlay_overrides_rss_config(self): + from trendradar.webui_settings import apply_overlay, save_overlay + + feeds = [{"id": "hn", "name": "HN", "url": "https://hnrss.org/frontpage", "enabled": True}] + save_overlay({"rss": {"enabled": False, "feeds": feeds}}, self.output) + config = { + "RSS": { + "ENABLED": True, + "FEEDS": [{"id": "old", "name": "旧", "url": "https://old.example/rss"}], + } + } + apply_overlay(config, output_dir=self.output) + self.assertFalse(config["RSS"]["ENABLED"]) + self.assertEqual(config["RSS"]["FEEDS"], feeds) + + +class WebuiSettingsAiTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.output = Path(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_save_valid_ai(self): + from trendradar.webui_settings import load_overlay, save_overlay + + save_overlay( + {"ai": {"model": "deepseek/deepseek-v4-flash", "api_key": "sk-test-123456789", "api_base": ""}}, + self.output, + ) + data = load_overlay(self.output)["ai"] + self.assertEqual(data["model"], "deepseek/deepseek-v4-flash") + self.assertEqual(data["api_key"], "sk-test-123456789") + self.assertEqual(data["api_base"], "") + + def test_rejects_empty_model_and_bad_base(self): + from trendradar.webui_settings import save_overlay + + with self.assertRaises(ValueError): + save_overlay({"ai": {"model": " "}}, self.output) + with self.assertRaises(ValueError): + save_overlay({"ai": {"model": "m", "api_base": "ftp://x"}}, self.output) + with self.assertRaises(ValueError): + save_overlay({"ai": {"model": "m", "timeout": 5}}, self.output) + + def test_reasoning_effort_validated_and_applied(self): + from trendradar.webui_settings import apply_overlay, load_overlay, save_overlay + + with self.assertRaises(ValueError): + save_overlay({"ai": {"model": "m", "reasoning_effort": "extreme"}}, self.output) + + save_overlay({"ai": {"model": "m", "reasoning_effort": "HIGH"}}, self.output) + self.assertEqual(load_overlay(self.output)["ai"]["reasoning_effort"], "high") + + config = {"AI": {"MODEL": "m", "REASONING_EFFORT": ""}} + apply_overlay(config, output_dir=self.output) + self.assertEqual(config["AI"]["REASONING_EFFORT"], "high") + + def test_mask_secret(self): + from trendradar.webui_settings import mask_secret + + self.assertIsNone(mask_secret("")) + self.assertIsNone(mask_secret(None)) + self.assertEqual(mask_secret("short"), "***") + self.assertEqual(mask_secret("sk-1234567890abcd"), "sk-***abcd") + + def test_apply_overlay_overrides_ai(self): + from trendradar.webui_settings import apply_overlay, save_overlay + + save_overlay( + {"ai": {"model": "openai/gpt-4o", "api_key": "sk-new-key", "api_base": ""}}, + self.output, + ) + config = {"AI": {"MODEL": "old-model", "API_KEY": "old-key", "API_BASE": "https://old/v1"}} + apply_overlay(config, output_dir=self.output) + self.assertEqual(config["AI"]["MODEL"], "openai/gpt-4o") + self.assertEqual(config["AI"]["API_KEY"], "sk-new-key") + self.assertEqual(config["AI"]["API_BASE"], "") + + +class FrequencyWordsOverlayTests(unittest.TestCase): + def test_context_load_prefers_overlay(self): + from trendradar.context import AppContext + from trendradar.webui_settings import save_overlay + + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) + save_overlay({"frequency_words": "[WORD_GROUPS]\noverlay词"}, output) + config = {"STORAGE": {"LOCAL": {"DATA_DIR": str(output)}}} + ctx = AppContext(config) + groups, _filters, _global = ctx.load_frequency_words() + self.assertEqual(len(groups), 1) + self.assertEqual(groups[0]["normal"][0]["word"], "overlay词") + + def test_context_load_falls_back_to_file(self): + from trendradar.context import AppContext + + with tempfile.TemporaryDirectory() as tmp: + config = {"STORAGE": {"LOCAL": {"DATA_DIR": tmp}}} + ctx = AppContext(config) + # 未设置 overlay 且指定不存在的文件时应抛 FileNotFoundError + with self.assertRaises(FileNotFoundError): + ctx.load_frequency_words("config/__no_such_file__.txt") + + +if __name__ == "__main__": + unittest.main() diff --git a/trendradar/ai/analyzer.py b/trendradar/ai/analyzer.py index 7408afd1b3d48..59cac776cd3d2 100644 --- a/trendradar/ai/analyzer.py +++ b/trendradar/ai/analyzer.py @@ -140,7 +140,9 @@ def analyze( timeout = self.ai_config.get("TIMEOUT", 120) max_tokens = self.ai_config.get("MAX_TOKENS", 5000) - print(f"[AI] 参数: timeout={timeout}, max_tokens={max_tokens}") + reasoning_effort = str(self.ai_config.get("REASONING_EFFORT") or "").strip() + effort_display = f", reasoning_effort={reasoning_effort}" if reasoning_effort else "" + print(f"[AI] 参数: timeout={timeout}, max_tokens={max_tokens}{effort_display}") if not self.client.api_key: return AIAnalysisResult( diff --git a/trendradar/ai/client.py b/trendradar/ai/client.py index a2a974e01ba55..5de873d4e672c 100644 --- a/trendradar/ai/client.py +++ b/trendradar/ai/client.py @@ -29,6 +29,8 @@ def __init__(self, config: Dict[str, Any]): - TIMEOUT: 请求超时时间（秒） - NUM_RETRIES: 重试次数（可选） - FALLBACK_MODELS: 备用模型列表（可选） + - REASONING_EFFORT: 推理强度（可选，minimal/low/medium/high，仅推理型模型支持） + - EXTRA_PARAMS: 额外请求参数（可选，优先级低于显式参数与调用 kwargs） """ self.model = config.get("MODEL", "deepseek/deepseek-chat") self.api_key = config.get("API_KEY") or os.environ.get("AI_API_KEY", "") @@ -38,6 +40,8 @@ def __init__(self, config: Dict[str, Any]): self.timeout = config.get("TIMEOUT", 120) self.num_retries = config.get("NUM_RETRIES", 2) self.fallback_models = config.get("FALLBACK_MODELS", []) + self.reasoning_effort = str(config.get("REASONING_EFFORT") or "").strip().lower() + self.extra_params = dict(config.get("EXTRA_PARAMS") or {}) def chat( self, @@ -83,6 +87,26 @@ def chat( if self.fallback_models: params["fallbacks"] = self.fallback_models + # 添加推理强度（仅推理型模型支持，未设置时不发送；可被调用参数覆盖） + reasoning_effort = str( + kwargs.get("reasoning_effort") or self.reasoning_effort or "" + ).strip() + if reasoning_effort: + if self.model.startswith("openai/"): + # openai/ 前缀多为自定义兼容端点，litellm 会按模型名做参数白名单校验， + # 自定义模型名不在白名单会抛 UnsupportedParamsError； + # 走 extra_body 原样透传（真正的 OpenAI 推理模型同样接受） + extra_body = dict(params.get("extra_body") or {}) + extra_body["reasoning_effort"] = reasoning_effort + params["extra_body"] = extra_body + else: + # 其他提供商走顶层参数，由 litellm 完成参数映射（如 anthropic 思考预算） + params["reasoning_effort"] = reasoning_effort + + # 合并 extra_params（显式参数优先） + for key, value in self.extra_params.items(): + params.setdefault(key, value) + # 合并其他额外参数 for key, value in kwargs.items(): if key not in params: diff --git a/trendradar/context.py b/trendradar/context.py index 79bc2a9819eb6..7267e2f561fbf 100644 --- a/trendradar/context.py +++ b/trendradar/context.py @@ -237,7 +237,17 @@ def is_first_crawl(self) -> bool: def load_frequency_words( self, frequency_file: Optional[str] = None ) -> Tuple[List[Dict], List[str], List[str]]: - """加载频率词配置""" + """加载频率词配置（Web UI overlay 中的主题词优先于文件）""" + from trendradar.webui_settings import load_overlay + from trendradar.core.frequency import parse_frequency_words_content + + output_dir = ( + self.config.get("STORAGE", {}).get("LOCAL", {}).get("DATA_DIR") or "output" + ) + overlay = load_overlay(output_dir) + content = overlay.get("frequency_words") + if isinstance(content, str) and content.strip(): + return parse_frequency_words_content(content) return load_frequency_words(frequency_file) def matches_word_groups( diff --git a/trendradar/core/__init__.py b/trendradar/core/__init__.py index 255439fbb8156..339d87c109e2c 100644 --- a/trendradar/core/__init__.py +++ b/trendradar/core/__init__.py @@ -10,7 +10,11 @@ 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.frequency import ( + load_frequency_words, + matches_word_groups, + parse_frequency_words_content, +) from trendradar.core.scheduler import Scheduler, ResolvedSchedule from trendradar.core.data import ( read_all_today_titles_from_storage, @@ -33,6 +37,7 @@ "load_config", "load_frequency_words", "matches_word_groups", + "parse_frequency_words_content", # 数据处理 "read_all_today_titles_from_storage", "read_all_today_titles", diff --git a/trendradar/core/frequency.py b/trendradar/core/frequency.py index 561741ac28455..fcf843469299c 100644 --- a/trendradar/core/frequency.py +++ b/trendradar/core/frequency.py @@ -136,6 +136,23 @@ def load_frequency_words( with open(frequency_path, "r", encoding="utf-8") as f: content = f.read() + return parse_frequency_words_content(content) + + +def parse_frequency_words_content( + content: str, +) -> Tuple[List[Dict], List[str], List[str]]: + """ + 解析频率词配置文本（与文件格式一致） + + 供需要直接解析文本内容的调用方使用（如 Web UI 的 overlay 覆盖配置）。 + + Args: + content: 频率词配置文本 + + Returns: + (词组列表, 词组内过滤词, 全局过滤词) + """ word_groups = [group.strip() for group in content.split("\n\n") if group.strip()] processed_groups = [] diff --git a/trendradar/core/loader.py b/trendradar/core/loader.py index 49e83918a106b..8757a88973943 100644 --- a/trendradar/core/loader.py +++ b/trendradar/core/loader.py @@ -270,6 +270,7 @@ def _load_ai_config(config_data: Dict) -> Dict: # LiteLLM 高级选项 "NUM_RETRIES": ai_config.get("num_retries", 2), "FALLBACK_MODELS": ai_config.get("fallback_models", []), + "REASONING_EFFORT": str(ai_config.get("reasoning_effort") or "").strip().lower(), "EXTRA_PARAMS": ai_config.get("extra_params", {}), } @@ -603,4 +604,11 @@ def load_config(config_path: Optional[str] = None) -> Dict[str, Any]: # 打印通知渠道配置来源 _print_notification_sources(config) + from trendradar.webui_settings import apply_overlay + + output_dir = ( + config.get("STORAGE", {}).get("LOCAL", {}).get("DATA_DIR") or "output" + ) + apply_overlay(config, output_dir=output_dir) + return config diff --git a/trendradar/crawler/fetcher.py b/trendradar/crawler/fetcher.py index 7fbc6d745fa3a..01f0700738f9f 100644 --- a/trendradar/crawler/fetcher.py +++ b/trendradar/crawler/fetcher.py @@ -172,6 +172,7 @@ def crawl_websites( failed_ids = [] domain_rules = domain_rules or {} + total = len(ids_list) for i, id_info in enumerate(ids_list): if isinstance(id_info, tuple): id_value, name = id_info @@ -180,6 +181,7 @@ def crawl_websites( name = id_value id_to_name[id_value] = name + print(f"正在获取 {name}（{i + 1}/{total}）...", flush=True) response, _, _ = self.fetch_data(id_info) if response: diff --git a/trendradar/crawler/rss/fetcher.py b/trendradar/crawler/rss/fetcher.py index 01addc8d9dd8b..ba968f947b996 100644 --- a/trendradar/crawler/rss/fetcher.py +++ b/trendradar/crawler/rss/fetcher.py @@ -167,7 +167,8 @@ def fetch_all(self) -> RSSData: crawl_time = now.strftime("%H:%M") crawl_date = now.strftime("%Y-%m-%d") - print(f"[RSS] 开始抓取 {len(self.feeds)} 个 RSS 源...") + total = len(self.feeds) + print(f"[RSS] 开始抓取 {total} 个 RSS 源...") for i, feed in enumerate(self.feeds): # 请求间隔（带随机波动） @@ -176,6 +177,7 @@ def fetch_all(self) -> RSSData: jitter = random.uniform(-0.2, 0.2) * interval time.sleep(interval + jitter) + print(f"[RSS] 正在获取 {feed.name}（{i + 1}/{total}）...", flush=True) items, error = self.fetch_feed(feed) id_to_name[feed.id] = feed.name diff --git a/trendradar/web_control.py b/trendradar/web_control.py new file mode 100644 index 0000000000000..045d63e72f100 --- /dev/null +++ b/trendradar/web_control.py @@ -0,0 +1,1442 @@ +# coding=utf-8 +"""本地 Web 控制面板：静态报告 + 手动抓取 / AI 分析 / 预设 / 进度与日志。""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +import threading +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any, Dict, List, Optional +from urllib.parse import parse_qs, urlparse + +import yaml + +from trendradar.webui_settings import ( + PRESETS, + _validate_feeds, + load_overlay, + mask_secret, + remove_overlay_keys, + save_overlay, + validate_frequency_words, +) + +TOOLBAR_MARKER = "tr-toolbar" +JOB_LOG_NAME = ".webui-job.log" +LOG_TAIL_DEFAULT = 400 +LOG_TAIL_MAX = 2000 + +# 从任务日志中识别阶段 / 当前条目 +_RE_PLATFORM_LIST = re.compile(r"配置的监控平台:\s*(\[[^\]]*\])") +_RE_FETCHING = re.compile(r"正在获取\s+(.+?)（(\d+)/(\d+)）") +_RE_FETCHED = re.compile(r"获取\s+(\S+)\s+成功") +_RE_FETCH_FAIL = re.compile(r"请求\s+(\S+)\s+失败") +_RE_RSS_START = re.compile(r"\[RSS\]\s*开始抓取\s+(\d+)\s+个") +_RE_RSS_FETCHING = re.compile(r"\[RSS\]\s*正在获取\s+(.+?)（(\d+)/(\d+)）") +_RE_RSS_DONE = re.compile(r"\[RSS\]\s*(.+?):\s*(获取\s+\d+\s*条|.+)") +_RE_AI = re.compile(r"\[AI\]\s*(.+)") +_RE_TRANSLATE = re.compile(r"\[翻译\]\s*(.+)") +_RE_PUSH = re.compile(r"\[推送\]\s*(.+)") +_RE_HTML = re.compile(r"HTML报告已生成") +_RE_CRAWL_START = re.compile(r"开始爬取数据") +_RE_SUCCESS_SUMMARY = re.compile(r"^成功:\s*") + + +def _safe_literal_list(text: str) -> List[str]: + try: + value = json.loads(text.replace("'", '"')) + if isinstance(value, list): + return [str(x) for x in value] + except (TypeError, ValueError, json.JSONDecodeError): + pass + return [] + + +def _parse_words_for_api(content: str): + """解析主题词文本用于统计预览，异常时按空配置处理。""" + try: + from trendradar.core.frequency import parse_frequency_words_content + + return parse_frequency_words_content(content) + except Exception: + return [], [], [] + + +def parse_job_progress(log_text: str, mode: Optional[str] = None) -> Dict[str, Any]: + """从任务日志解析当前阶段与进度百分比。""" + lines = [ln.strip() for ln in (log_text or "").splitlines() if ln.strip()] + phase = "starting" + message = "准备中…" + current = "" + step = 0 + total = 0 + platforms_total = 0 + rss_total = 0 + platforms_done = 0 + rss_done = 0 + weight_platform = 55 + weight_rss = 25 + weight_pipeline = 20 + + for line in lines: + m = _RE_PLATFORM_LIST.search(line) + if m: + platforms_total = max(platforms_total, len(_safe_literal_list(m.group(1)))) + continue + + if _RE_CRAWL_START.search(line): + phase = "platforms" + message = "开始抓取热榜平台" + continue + + m = _RE_FETCHING.search(line) + if m: + phase = "platforms" + current = m.group(1).strip() + step = int(m.group(2)) + total = int(m.group(3)) + platforms_total = max(platforms_total, total) + platforms_done = max(platforms_done, step - 1) + message = f"正在抓取热榜：{current}（{step}/{total}）" + continue + + m = _RE_FETCHED.search(line) + if m: + phase = "platforms" + platforms_done += 1 + if platforms_total: + platforms_done = min(platforms_done, platforms_total) + current = m.group(1) + message = f"已完成热榜：{current}" + continue + + m = _RE_FETCH_FAIL.search(line) + if m: + phase = "platforms" + current = m.group(1) + message = f"热榜失败：{current}" + continue + + if _RE_SUCCESS_SUMMARY.search(line): + phase = "platforms_done" + if platforms_total: + platforms_done = platforms_total + message = "热榜抓取完成" + continue + + m = _RE_RSS_START.search(line) + if m: + phase = "rss" + rss_total = max(rss_total, int(m.group(1))) + message = f"开始抓取 RSS（{rss_total} 个源）" + continue + + m = _RE_RSS_FETCHING.search(line) + if m: + phase = "rss" + current = m.group(1).strip() + step = int(m.group(2)) + total = int(m.group(3)) + rss_total = max(rss_total, total) + rss_done = max(rss_done, step - 1) + message = f"正在抓取 RSS：{current}（{step}/{total}）" + continue + + m = _RE_RSS_DONE.search(line) + if m and "开始抓取" not in line and "抓取完成" not in line and "正在获取" not in line: + name = m.group(1).strip() + if name and not name.startswith("["): + phase = "rss" + current = name + rss_done += 1 + if rss_total: + rss_done = min(rss_done, rss_total) + detail = m.group(2).strip() + message = f"RSS {name}：{detail}" + continue + + if "[RSS] 抓取完成" in line: + phase = "rss_done" + if rss_total: + rss_done = rss_total + message = "RSS 抓取完成" + continue + + m = _RE_AI.search(line) + if m: + phase = "ai" + current = m.group(1).strip() + message = f"AI：{current}" + continue + + m = _RE_TRANSLATE.search(line) + if m: + phase = "translate" + current = m.group(1).strip() + message = f"翻译：{current}" + continue + + m = _RE_PUSH.search(line) + if m: + phase = "push" + current = m.group(1).strip() + message = f"推送：{current}" + continue + + if _RE_HTML.search(line): + phase = "report" + message = "正在生成报告" + continue + + # 估算百分比 + percent = 0 + if mode == "analyze": + if phase in ("starting",): + percent = 5 + elif phase in ("platforms", "platforms_done", "rss", "rss_done"): + percent = 15 + elif phase == "ai": + percent = 55 + elif phase == "translate": + percent = 75 + elif phase == "push": + percent = 88 + elif phase == "report": + percent = 95 + else: + percent = 30 + else: + p_part = 0.0 + if platforms_total > 0: + # 正在抓第 step 项时，算 step-0.3 完成感 + active = platforms_done + if phase == "platforms" and step and total: + active = max(active, step - 0.3) + p_part = min(1.0, active / platforms_total) * weight_platform + elif phase in ("platforms", "platforms_done"): + p_part = weight_platform * (0.5 if phase == "platforms" else 1.0) + + r_part = 0.0 + if phase in ("rss", "rss_done") or rss_done or rss_total: + if rss_total > 0: + active = rss_done + if phase == "rss" and step and total and "RSS" in message: + active = max(active, step - 0.3) + r_part = min(1.0, active / rss_total) * weight_rss + elif phase == "rss_done": + r_part = float(weight_rss) + else: + r_part = weight_rss * 0.4 + if phase in ("platforms_done",) and not rss_total: + # 尚不知是否有 RSS + r_part = 0 + + pipe = 0.0 + if phase == "ai": + pipe = weight_pipeline * 0.45 + elif phase == "translate": + pipe = weight_pipeline * 0.65 + elif phase == "push": + pipe = weight_pipeline * 0.85 + elif phase == "report": + pipe = weight_pipeline * 0.95 + elif phase in ("rss_done",) or (platforms_total and platforms_done >= platforms_total and rss_total and rss_done >= rss_total): + pipe = weight_pipeline * 0.2 + + if phase == "starting": + percent = 2 + else: + base = 0 + if platforms_total or phase in ("platforms", "platforms_done", "rss", "rss_done", "ai", "translate", "push", "report"): + base = p_part + r_part + pipe + percent = int(max(2, min(99, round(base)))) + + return { + "phase": phase, + "message": message, + "current": current, + "step": step or None, + "total": total or None, + "platforms_done": platforms_done or None, + "platforms_total": platforms_total or None, + "rss_done": rss_done or None, + "rss_total": rss_total or None, + "percent": percent, + } + + +TOOLBAR_HTML = """ + +
+
+ + + + + + +
+
+
+ 准备中… + 0% +
+
+
+
+
+
+ 任务日志 + + + + +
+ 
暂无日志
+
+
+
+ +
+ +""" + + +class JobRunner: + def __init__(self, project_root: str | Path = ".", output_dir: str | Path = "output"): + self.project_root = Path(project_root) + self.output_dir = Path(output_dir) + self._lock = threading.Lock() + self._proc: Optional[subprocess.Popen] = None + self._mode: Optional[str] = None + self._error: Optional[str] = None + self._log_path = self.output_dir / JOB_LOG_NAME + + def is_running(self) -> bool: + with self._lock: + return self._proc is not None and self._proc.poll() is None + + def log_path(self) -> Path: + return self._log_path + + def read_log(self, tail: int = LOG_TAIL_DEFAULT) -> Dict[str, Any]: + path = self._log_path + tail = max(1, min(int(tail or LOG_TAIL_DEFAULT), LOG_TAIL_MAX)) + if not path.exists(): + return {"text": "", "lines": 0, "path": str(path.name), "exists": False} + try: + # 任务日志通常不大；按行截取尾部即可 + content = path.read_text(encoding="utf-8", errors="replace") + except OSError as exc: + return {"text": f"[读取日志失败] {exc}", "lines": 0, "path": str(path.name), "exists": True} + lines = content.splitlines() + sliced = lines[-tail:] if tail < len(lines) else lines + text = "\n".join(sliced) + if len(lines) > tail: + text = f"… 已省略前 {len(lines) - tail} 行 …\n" + text + return { + "text": text, + "lines": len(lines), + "tail": len(sliced), + "path": str(path.name), + "exists": True, + } + + def status(self) -> Dict[str, Any]: + with self._lock: + running = self._proc is not None and self._proc.poll() is None + mode = self._mode + error = self._error + log_info = self.read_log(tail=LOG_TAIL_DEFAULT) + progress = parse_job_progress(log_info.get("text") or "", mode=mode) + if not running and not mode and not (log_info.get("text") or "").strip(): + progress = { + "phase": "idle", + "message": "", + "current": "", + "percent": 0, + "step": None, + "total": None, + "platforms_done": None, + "platforms_total": None, + "rss_done": None, + "rss_total": None, + } + elif not running and mode and not error: + progress = dict(progress) + progress["phase"] = "done" + progress["message"] = progress.get("message") or "已完成" + progress["percent"] = 100 + return { + "running": running, + "mode": mode, + "error": error, + "progress": progress, + "log_lines": log_info.get("lines") or 0, + } + + def start(self, mode: str) -> None: + if mode not in ("crawl", "analyze"): + raise ValueError(f"未知模式: {mode}") + with self._lock: + if self._proc is not None and self._proc.poll() is None: + raise RuntimeError("running") + env = os.environ.copy() + if mode == "analyze": + env["WEBUI_RUN_AI"] = "true" + env["SCHEDULE_ENABLED"] = "false" + else: + env["WEBUI_RUN_AI"] = "false" + self.output_dir.mkdir(parents=True, exist_ok=True) + log_file = open(self._log_path, "w", encoding="utf-8") + # 行缓冲，便于前端实时看到进度 + try: + if hasattr(log_file, "reconfigure"): + log_file.reconfigure(line_buffering=True) + except Exception: + pass + self._error = None + self._mode = mode + self._proc = subprocess.Popen( + [sys.executable, "-u", "-m", "trendradar"], + cwd=str(self.project_root), + env=env, + stdout=log_file, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + threading.Thread(target=self._watch, args=(log_file,), daemon=True).start() + + def _watch(self, log_file) -> None: + proc = self._proc + if proc is None: + log_file.close() + return + code = proc.wait() + try: + log_file.flush() + except Exception: + pass + log_file.close() + with self._lock: + if code != 0: + self._error = f"退出码 {code}" + self._proc = None + + +class ControlState: + def __init__( + self, + output_dir: str | Path, + runner: Any = None, + project_root: str | Path = ".", + ): + self.output_dir = Path(output_dir) + self.runner = runner or JobRunner(project_root=project_root, output_dir=self.output_dir) + self.project_root = Path(project_root) + + # === 配置文件定位 === + + def _config_dir(self) -> Path: + env_path = os.environ.get("CONFIG_PATH", "") + if env_path: + return Path(env_path).expanduser().resolve().parent + return self.project_root / "config" + + def _frequency_words_path(self) -> Path: + env_path = os.environ.get("FREQUENCY_WORDS_PATH", "") + if env_path and Path(env_path).exists(): + return Path(env_path) + candidate = self._config_dir() / "frequency_words.txt" + if candidate.exists(): + return candidate + return Path(env_path) if env_path else candidate + + def _config_yaml_path(self) -> Path: + env_path = os.environ.get("CONFIG_PATH", "") + if env_path: + return Path(env_path).expanduser() + return self._config_dir() / "config.yaml" + + # === 主题词 === + + def _read_frequency_file(self) -> str: + try: + return self._frequency_words_path().read_text(encoding="utf-8") + except OSError: + return "" + + def topics_payload(self) -> Dict[str, Any]: + overlay = load_overlay(self.output_dir) + override = overlay.get("frequency_words") + file_content = self._read_frequency_file() + using_override = isinstance(override, str) and override.strip() != "" + content = override if isinstance(override, str) else file_content + groups, _filters, _global_filters = _parse_words_for_api(content) + return { + "content": content, + "source": "overlay" if using_override else "file", + "file_content": file_content, + "file_path": str(self._frequency_words_path()), + "group_count": len(groups), + } + + # === RSS 订阅 === + + def _read_yaml_feeds(self) -> Dict[str, Any]: + try: + data = yaml.safe_load(self._config_yaml_path().read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return {"enabled": None, "feeds": []} + if not isinstance(data, dict): + return {"enabled": None, "feeds": []} + rss = data.get("rss", {}) or {} + feeds = rss.get("feeds", []) or [] + return {"enabled": rss.get("enabled"), "feeds": feeds if isinstance(feeds, list) else []} + + def feeds_payload(self) -> Dict[str, Any]: + overlay = load_overlay(self.output_dir) + override = overlay.get("rss") + yaml_data = self._read_yaml_feeds() + if isinstance(override, dict) and (override.get("feeds") is not None): + feeds = override.get("feeds") or [] + enabled = override.get("enabled") + if enabled is None: + enabled = yaml_data.get("enabled") + if enabled is None: + enabled = True + source = "overlay" + else: + feeds = yaml_data.get("feeds") or [] + enabled = yaml_data.get("enabled") + if enabled is None: + enabled = True + source = "config" + normalized = [ + { + "id": str(f.get("id") or ""), + "name": str(f.get("name") or f.get("id") or ""), + "url": str(f.get("url") or ""), + "enabled": bool(f.get("enabled", True)), + "max_age_days": f.get("max_age_days"), + } + for f in feeds + if isinstance(f, dict) + ] + return { + "rss_enabled": bool(enabled), + "feeds": normalized, + "source": source, + "config_path": str(self._config_yaml_path()), + } + + # === AI 配置 === + + def _read_yaml_ai(self) -> Dict[str, Any]: + try: + data = yaml.safe_load(self._config_yaml_path().read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return {} + if not isinstance(data, dict): + return {} + ai = data.get("ai", {}) + return ai if isinstance(ai, dict) else {} + + def _effective_ai(self) -> Dict[str, Any]: + """计算生效的 AI 配置（overlay > 环境变量 > config.yaml），含明文密钥。""" + overlay_ai = load_overlay(self.output_dir).get("ai") + has_overlay = isinstance(overlay_ai, dict) + env_key = os.environ.get("AI_API_KEY", "").strip() + env_model = os.environ.get("AI_MODEL", "").strip() + env_base = os.environ.get("AI_API_BASE", "").strip() + yaml_ai = self._read_yaml_ai() + yaml_key = str(yaml_ai.get("api_key") or "").strip() + + def pick(*candidates: Any) -> str: + for candidate in candidates: + if candidate: + return str(candidate) + return "" + + effective_key = pick( + has_overlay and overlay_ai.get("api_key"), + env_key, + yaml_key, + ) + if has_overlay and str(overlay_ai.get("api_key") or "").strip(): + key_source = "overlay" + elif env_key: + key_source = "env" + elif yaml_key: + key_source = "config" + else: + key_source = None + + if has_overlay: + source = "overlay" + elif env_key or env_model or env_base: + source = "env" + else: + source = "config" + # overlay 存在时 api_base / reasoning_effort 以 overlay 为准（空串 = 用户已清空，不回落） + if has_overlay: + api_base = str(overlay_ai.get("api_base") or "") + reasoning_effort = str(overlay_ai.get("reasoning_effort") or "") + else: + api_base = pick(env_base, str(yaml_ai.get("api_base") or "")) + reasoning_effort = str(yaml_ai.get("reasoning_effort") or "").strip().lower() + return { + "model": pick( + has_overlay and overlay_ai.get("model"), + env_model, + str(yaml_ai.get("model") or ""), + ), + "api_base": api_base, + "reasoning_effort": reasoning_effort, + "api_key": effective_key, + "api_key_source": key_source, + "source": source, + } + + def ai_payload(self, reveal: bool = False) -> Dict[str, Any]: + payload = self._effective_ai() + key = payload.pop("api_key") + payload.update( + { + "api_key_set": bool(key), + "api_key_masked": mask_secret(key), + } + ) + # 明文密钥仅在显式请求 reveal 时返回（供面板“显示”按钮查看已保存密钥） + if reveal: + payload["api_key"] = key + return payload + + def settings(self) -> Dict[str, Any]: + overlay = load_overlay(self.output_dir) + env_ai = os.environ.get("AI_ANALYSIS_ENABLED", "").strip().lower() + default_ai = True if not env_ai else env_ai in ("true", "1", "yes") + preset = overlay.get("schedule_preset") or "off" + enabled = overlay.get("schedule_enabled") + if enabled is None: + enabled = preset != "off" + payload = { + "ai_analysis_enabled": overlay.get("ai_analysis_enabled", default_ai), + "schedule_enabled": bool(enabled), + "schedule_preset": preset, + "presets": [dict(item) for item in PRESETS], + "job": self.runner.status(), + } + return payload + + +def _json_bytes(payload: Dict[str, Any]) -> bytes: + return json.dumps(payload, ensure_ascii=False).encode("utf-8") + + +def make_handler(state: ControlState): + class Handler(SimpleHTTPRequestHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, directory=str(state.output_dir), **kwargs) + + def log_message(self, format, *args): + sys.stderr.write("%s - %s\n" % (self.address_string(), format % args)) + + def _send_json(self, status: int, payload: Dict[str, Any]): + body = _json_bytes(payload) + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def _read_json(self) -> Dict[str, Any]: + length = int(self.headers.get("Content-Length") or 0) + raw = self.rfile.read(length) if length else b"{}" + if not raw: + return {} + data = json.loads(raw.decode("utf-8")) + if not isinstance(data, dict): + raise ValueError("JSON 必须是对象") + return data + + def do_HEAD(self): + # manage.py 用 HEAD / 探活；API 与注入 HTML 走同一路由 + parsed = urlparse(self.path) + path = parsed.path + if path.startswith("/api/"): + self.send_response(200) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", "2") + self.send_header("Cache-Control", "no-store") + self.end_headers() + return + if path in ("/", "/index.html") or path.endswith(".html"): + target = state.output_dir / "index.html" if path in ("/", "/index.html") else Path(self.translate_path(self.path)) + if target.is_file(): + body = target.read_bytes() + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + return + self.send_error(404, "File not found") + return + super().do_HEAD() + + def do_GET(self): + parsed = urlparse(self.path) + path = parsed.path + if path == "/api/settings": + self._send_json(200, state.settings()) + return + if path == "/api/topics": + self._send_json(200, state.topics_payload()) + return + if path == "/api/feeds": + self._send_json(200, state.feeds_payload()) + return + if path == "/api/ai": + qs = parse_qs(parsed.query or "") + reveal = (qs.get("reveal") or [""])[0].strip().lower() in ("1", "true", "yes") + self._send_json(200, state.ai_payload(reveal=reveal)) + return + if path == "/api/status": + self._send_json(200, state.runner.status()) + return + if path == "/api/logs": + qs = parse_qs(parsed.query or "") + try: + tail = int((qs.get("tail") or [LOG_TAIL_DEFAULT])[0]) + except (TypeError, ValueError): + tail = LOG_TAIL_DEFAULT + log_payload = state.runner.read_log(tail=tail) + st = state.runner.status() + log_payload.update( + { + "running": st.get("running"), + "mode": st.get("mode"), + "error": st.get("error"), + "progress": st.get("progress"), + } + ) + self._send_json(200, log_payload) + return + if path in ("/", "/index.html"): + self._serve_html(state.output_dir / "index.html") + return + if path.endswith(".html"): + fs_path = Path(self.translate_path(self.path)) + if fs_path.is_file(): + self._serve_html(fs_path) + return + super().do_GET() + + def do_POST(self): + path = urlparse(self.path).path + try: + payload = self._read_json() + except (ValueError, json.JSONDecodeError) as exc: + self._send_json(400, {"error": str(exc)}) + return + if path == "/api/settings": + try: + save_overlay(payload, state.output_dir) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + self._send_json(200, state.settings()) + return + if path == "/api/topics": + try: + content = validate_frequency_words(payload.get("content")) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + try: + save_overlay({"frequency_words": content}, state.output_dir) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + self._send_json(200, state.topics_payload()) + return + if path == "/api/feeds": + try: + feeds = _validate_feeds(payload.get("feeds")) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + rss_enabled = payload.get("rss_enabled") + try: + save_overlay( + {"rss": {"enabled": bool(rss_enabled), "feeds": feeds}}, + state.output_dir, + ) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + self._send_json(200, state.feeds_payload()) + return + if path == "/api/ai": + model = str(payload.get("model") or "").strip() + if not model: + self._send_json(400, {"error": "model 不能为空"}) + return + api_base = str(payload.get("api_base") or "").strip() + if api_base and not api_base.startswith(("http://", "https://")): + self._send_json(400, {"error": "api_base 必须以 http:// 或 https:// 开头"}) + return + effort = str(payload.get("reasoning_effort") or "").strip().lower() + # 密钥留空 = 保持已保存的密钥不变（不回传明文，前端无法回填） + new_key = str(payload.get("api_key") or "").strip() + existing_ai = load_overlay(state.output_dir).get("ai") + ai_data: Dict[str, Any] = { + "model": model, + "api_base": api_base, + "reasoning_effort": effort, + } + if new_key: + ai_data["api_key"] = new_key + elif isinstance(existing_ai, dict) and existing_ai.get("api_key"): + ai_data["api_key"] = existing_ai["api_key"] + try: + save_overlay({"ai": ai_data}, state.output_dir) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + self._send_json(200, state.ai_payload()) + return + if path == "/api/run": + mode = payload.get("mode") + if mode not in ("crawl", "analyze"): + self._send_json(400, {"error": "mode 必须是 crawl 或 analyze"}) + return + if state.runner.is_running(): + self._send_json(409, {"error": "running", "job": state.runner.status()}) + return + try: + state.runner.start(mode) + except RuntimeError: + self._send_json(409, {"error": "running", "job": state.runner.status()}) + return + self._send_json(202, {"ok": True, "mode": mode, "job": state.runner.status()}) + return + self._send_json(404, {"error": "not found"}) + + def do_DELETE(self): + path = urlparse(self.path).path + if path == "/api/topics": + remove_overlay_keys(["frequency_words"], state.output_dir) + self._send_json(200, state.topics_payload()) + return + if path == "/api/feeds": + remove_overlay_keys(["rss"], state.output_dir) + self._send_json(200, state.feeds_payload()) + return + if path == "/api/ai": + remove_overlay_keys(["ai"], state.output_dir) + self._send_json(200, state.ai_payload()) + return + self._send_json(404, {"error": "not found"}) + + def _serve_html(self, path: Path): + if not path.is_file(): + self.send_error(404, "File not found") + return + html = path.read_text(encoding="utf-8") + if TOOLBAR_MARKER not in html: + injected = TOOLBAR_HTML + lower = html.lower() + idx = lower.find("= 0: + gt = html.find(">", idx) + html = html[: gt + 1] + injected + html[gt + 1 :] + else: + html = injected + html + body = html.encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + return Handler + + +def start_control_server( + port: int, + output_dir: str | Path, + project_root: str | Path = ".", + bind: str = "0.0.0.0", +) -> None: + state = ControlState(output_dir=output_dir, project_root=project_root) + handler = make_handler(state) + server = ThreadingHTTPServer((bind, port), handler) + print(f" 🎛️ 控制面板: http://127.0.0.1:{port}/") + server.serve_forever() + + +def main() -> None: + port = int(sys.argv[1]) if len(sys.argv) > 1 else int(os.environ.get("WEBSERVER_PORT", "8080")) + output_dir = sys.argv[2] if len(sys.argv) > 2 else os.environ.get("WEBSERVER_DIR", "output") + project_root = sys.argv[3] if len(sys.argv) > 3 else os.environ.get("TRENDRADAR_ROOT", ".") + start_control_server(port, output_dir, project_root=project_root) + + +if __name__ == "__main__": + main() diff --git a/trendradar/webui_settings.py b/trendradar/webui_settings.py new file mode 100644 index 0000000000000..637ba78a7f07c --- /dev/null +++ b/trendradar/webui_settings.py @@ -0,0 +1,291 @@ +# coding=utf-8 +"""Web 控制面板的持久化设置（覆盖 config.yaml / 环境变量）。 + +overlay 保存在 output/.webui.json，键说明： +- schedule_preset / schedule_enabled / ai_analysis_enabled: 调度与 AI 开关 +- frequency_words: 主题词配置全文（覆盖 config/frequency_words.txt） +- rss: {"enabled": bool, "feeds": [...]} 覆盖 config.yaml 的 rss.feeds +- ai: {"model": str, "api_key": str, "api_base": str} 覆盖 ai 模型配置（优先级高于环境变量） +""" + +from __future__ import annotations + +import json +import os +import re +from pathlib import Path +from typing import Any, Dict, List, Optional + +OVERLAY_FILENAME = ".webui.json" + +PRESETS = ( + {"id": "off", "label": "关闭调度", "description": "每次定时任务都抓取；是否分析由 AI 开关决定"}, + {"id": "always_on", "label": "全天监控", "description": "全天候采集，有新增就推送，默认定时不跑 AI"}, + {"id": "morning_evening", "label": "早晚汇总", "description": "全天推送当前热点，晚间做一次当日汇总分析"}, + {"id": "office_hours", "label": "办公时间", "description": "工作日到岗 / 午间 / 收工三段式"}, + {"id": "night_owl", "label": "夜猫子", "description": "午后速览 + 深夜全天汇总"}, +) + +KNOWN_PRESETS = {item["id"] for item in PRESETS} + +# RSS 订阅源允许的键 +_FEED_KEYS = {"id", "name", "url", "enabled", "max_age_days"} + +# 推理强度可选值（空串 = 不发送） +REASONING_EFFORTS = {"", "minimal", "low", "medium", "high"} + +_RE_FEED_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]*$") + + +def slugify_feed_id(source: Any) -> str: + """把名称/URL 转成合法的 feed id；无法提取时返回空串。""" + text = str(source or "").strip() + if not text: + return "" + # URL 来源时取主机名（去掉 www. 前缀与 TLD 后缀） + host_match = re.match(r"^https?://([^/:]+)", text, re.IGNORECASE) + if host_match: + text = host_match.group(1) + if text.lower().startswith("www."): + text = text[4:] + labels = text.split(".") + if len(labels) > 1: + text = "-".join(labels[:-1]) or text + return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") + + +def _validate_feeds(feeds: Any) -> List[Dict[str, Any]]: + """校验并归一化 RSS 订阅源列表，返回可写入 overlay 的列表。""" + if not isinstance(feeds, list): + raise ValueError("feeds 必须是列表") + normalized: List[Dict[str, Any]] = [] + seen_ids: set[str] = set() + for index, feed in enumerate(feeds, start=1): + if not isinstance(feed, dict): + raise ValueError(f"第 {index} 个订阅源必须是对象") + unknown = set(feed.keys()) - _FEED_KEYS + if unknown: + raise ValueError(f"第 {index} 个订阅源含未知字段: {', '.join(sorted(unknown))}") + + url = str(feed.get("url") or "").strip() + if not url.startswith(("http://", "https://")): + raise ValueError(f"第 {index} 个订阅源的 url 必须以 http:// 或 https:// 开头") + + feed_id = str(feed.get("id") or "").strip() + if not feed_id: + # 名称无法转 slug（如纯中文）时退回 URL 主机名；再冲突则追加序号 + base = ( + slugify_feed_id(feed.get("name")) + or slugify_feed_id(feed.get("url")) + or "feed" + ) + feed_id = base + suffix = 2 + while feed_id in seen_ids: + feed_id = f"{base}-{suffix}" + suffix += 1 + if not _RE_FEED_ID.match(feed_id): + raise ValueError(f"第 {index} 个订阅源 id 含非法字符（仅限字母数字、下划线、连字符）") + if feed_id in seen_ids: + raise ValueError(f"订阅源 id 重复: {feed_id}") + seen_ids.add(feed_id) + + name = str(feed.get("name") or "").strip() or feed_id + + raw_max_age = feed.get("max_age_days") + if raw_max_age in (None, ""): + max_age_days = None + else: + try: + max_age_days = int(raw_max_age) + except (TypeError, ValueError): + raise ValueError(f"第 {index} 个订阅源 max_age_days 必须是整数") + if max_age_days < 0: + raise ValueError(f"第 {index} 个订阅源 max_age_days 不能为负数") + + normalized.append( + { + "id": feed_id, + "name": name, + "url": url, + "enabled": bool(feed.get("enabled", True)), + **({"max_age_days": max_age_days} if max_age_days is not None else {}), + } + ) + return normalized + + +def validate_frequency_words(content: Any) -> str: + """校验主题词配置文本，返回规整后的字符串。""" + if not isinstance(content, str): + raise ValueError("主题词内容必须是文本") + if "\x00" in content: + raise ValueError("主题词内容含有非法字符") + return content + + +def mask_secret(value: Any) -> Optional[str]: + """脱敏展示密钥，仅保留首尾少量字符。""" + text = str(value or "") + if not text: + return None + if len(text) <= 8: + return "***" + return f"{text[:3]}***{text[-4:]}" + + +def _validate_ai_override(ai: Any) -> Dict[str, Any]: + """校验 AI 模型配置覆盖项（model / api_key / api_base / reasoning_effort）。""" + if not isinstance(ai, dict): + raise ValueError("ai 必须是对象") + unknown = set(ai.keys()) - {"model", "api_key", "api_base", "reasoning_effort"} + if unknown: + raise ValueError(f"ai 含未知字段: {', '.join(sorted(unknown))}") + + model = str(ai.get("model") or "").strip() + if not model: + raise ValueError("model 不能为空") + + result: Dict[str, Any] = {"model": model} + + if ai.get("api_key") is not None: + api_key = str(ai["api_key"]).strip() + if not api_key: + raise ValueError("api_key 不能为空（保持不变请直接省略该字段）") + result["api_key"] = api_key + + api_base = str(ai.get("api_base") or "").strip() + if api_base and not api_base.startswith(("http://", "https://")): + raise ValueError("api_base 必须以 http:// 或 https:// 开头") + result["api_base"] = api_base + + effort = str(ai.get("reasoning_effort") or "").strip().lower() + if effort not in REASONING_EFFORTS: + raise ValueError(f"reasoning_effort 仅支持: minimal / low / medium / high（留空为不设置）") + result["reasoning_effort"] = effort + return result + + +def overlay_path(output_dir: str | Path = "output") -> Path: + return Path(output_dir) / OVERLAY_FILENAME + + +def load_overlay(output_dir: str | Path = "output") -> Dict[str, Any]: + path = overlay_path(output_dir) + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return data if isinstance(data, dict) else {} + + +def save_overlay(data: Dict[str, Any], output_dir: str | Path = "output") -> Dict[str, Any]: + current = load_overlay(output_dir) + incoming = dict(data or {}) + + if "schedule_preset" in incoming: + preset = str(incoming["schedule_preset"]) + if preset not in KNOWN_PRESETS: + raise ValueError(f"未知预设: {preset}") + incoming["schedule_preset"] = preset + incoming["schedule_enabled"] = preset != "off" + + if "ai_analysis_enabled" in incoming: + incoming["ai_analysis_enabled"] = bool(incoming["ai_analysis_enabled"]) + if "schedule_enabled" in incoming: + incoming["schedule_enabled"] = bool(incoming["schedule_enabled"]) + + if "frequency_words" in incoming: + incoming["frequency_words"] = validate_frequency_words(incoming["frequency_words"]) + + if "rss" in incoming: + incoming["rss"] = _validate_rss_override(incoming["rss"]) + + if "ai" in incoming: + incoming["ai"] = _validate_ai_override(incoming["ai"]) + + current.update(incoming) + return _write_overlay(current, output_dir) + + +def remove_overlay_keys(keys: List[str], output_dir: str | Path = "output") -> Dict[str, Any]: + """从 overlay 中移除指定键（恢复使用配置文件），其余键保留。""" + current = load_overlay(output_dir) + for key in keys: + current.pop(key, None) + return _write_overlay(current, output_dir) + + +def _write_overlay(data: Dict[str, Any], output_dir: str | Path = "output") -> Dict[str, Any]: + path = overlay_path(output_dir) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + return data + + +def _validate_rss_override(rss: Any) -> Dict[str, Any]: + """校验 rss 覆盖项，仅保留 enabled / feeds。""" + if not isinstance(rss, dict): + raise ValueError("rss 必须是对象") + unknown = set(rss.keys()) - {"enabled", "feeds"} + if unknown: + raise ValueError(f"rss 含未知字段: {', '.join(sorted(unknown))}") + result: Dict[str, Any] = {} + if "enabled" in rss and rss["enabled"] is not None: + result["enabled"] = bool(rss["enabled"]) + if "feeds" in rss and rss["feeds"] is not None: + result["feeds"] = _validate_feeds(rss["feeds"]) + return result + + +def _env_bool(key: str) -> Optional[bool]: + value = os.environ.get(key, "").strip().lower() + if not value: + return None + return value in ("true", "1", "yes") + + +def apply_overlay( + config: Dict[str, Any], + overlay: Optional[Dict[str, Any]] = None, + output_dir: str | Path = "output", +) -> Dict[str, Any]: + data = load_overlay(output_dir) if overlay is None else overlay + + if data.get("ai_analysis_enabled") is not None: + config.setdefault("AI_ANALYSIS", {})["ENABLED"] = bool(data["ai_analysis_enabled"]) + if data.get("schedule_enabled") is not None: + config.setdefault("SCHEDULE", {})["enabled"] = bool(data["schedule_enabled"]) + preset = data.get("schedule_preset") + if preset and preset != "off": + config.setdefault("SCHEDULE", {})["preset"] = preset + elif preset == "off": + config.setdefault("SCHEDULE", {})["enabled"] = False + + rss_override = data.get("rss") + if isinstance(rss_override, dict): + rss_config = config.setdefault("RSS", {}) + if rss_override.get("enabled") is not None: + rss_config["ENABLED"] = bool(rss_override["enabled"]) + if rss_override.get("feeds") is not None: + rss_config["FEEDS"] = rss_override["feeds"] + + ai_override = data.get("ai") + if isinstance(ai_override, dict): + ai_config = config.setdefault("AI", {}) + if ai_override.get("model"): + ai_config["MODEL"] = ai_override["model"] + if ai_override.get("api_key"): + ai_config["API_KEY"] = ai_override["api_key"] + if "api_base" in ai_override: + ai_config["API_BASE"] = str(ai_override.get("api_base") or "") + if "reasoning_effort" in ai_override: + ai_config["REASONING_EFFORT"] = str(ai_override.get("reasoning_effort") or "") + + force_ai = _env_bool("WEBUI_RUN_AI") + if force_ai is not None: + config.setdefault("AI_ANALYSIS", {})["ENABLED"] = force_ai + + return config