From b53f8a3eee9de636e7d8ca46039f6cd4ea6f6c17 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 14 Jun 2026 03:49:18 +0800 Subject: [PATCH] Guard Telegram forum topic routing --- deploy.sh | 55 +++++++++++++++++++++++++ src/utils/telegram_push.py | 30 ++++++++++++-- tests/test_deployment_runtime_config.py | 10 +++++ tests/test_telegram_hashtags.py | 38 +++++++++++++++++ 4 files changed, 130 insertions(+), 3 deletions(-) diff --git a/deploy.sh b/deploy.sh index 6669ae10..9a69fd57 100644 --- a/deploy.sh +++ b/deploy.sh @@ -29,6 +29,61 @@ unset GHCR_PAT cd "$COMPOSE_DIR" git fetch origin main && git reset --hard origin/main +sync_city_thread_ids() { + local runtime_dir="${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}" + local repo_file="$COMPOSE_DIR/data/city_thread_ids.json" + local target_file="$runtime_dir/city_thread_ids.json" + + if [ ! -f "$repo_file" ]; then + echo "No repository city_thread_ids.json to sync" + return 0 + fi + + mkdir -p "$runtime_dir" + REPO_CITY_THREAD_IDS_FILE="$repo_file" TARGET_CITY_THREAD_IDS_FILE="$target_file" python - <<'PY' +import json +import os +import time + +repo_file = os.environ["REPO_CITY_THREAD_IDS_FILE"] +target_file = os.environ["TARGET_CITY_THREAD_IDS_FILE"] + +with open(repo_file, "r", encoding="utf-8") as f: + repo_data = json.load(f) +if not isinstance(repo_data, dict): + raise SystemExit("repository city_thread_ids.json must contain an object") + +target_data = {} +if os.path.isfile(target_file) and os.path.getsize(target_file) > 0: + try: + with open(target_file, "r", encoding="utf-8") as f: + loaded = json.load(f) + if isinstance(loaded, dict): + target_data = loaded + else: + raise ValueError("target file is not an object") + except Exception as exc: + backup = f"{target_file}.invalid.{int(time.time())}" + os.replace(target_file, backup) + print(f"Backed up invalid city_thread_ids.json to {backup}: {exc}") + +merged = dict(repo_data) +merged.update(target_data) + +if merged != target_data: + tmp_file = f"{target_file}.tmp" + with open(tmp_file, "w", encoding="utf-8") as f: + json.dump(merged, f, ensure_ascii=False, indent=2, sort_keys=True) + f.write("\n") + os.replace(tmp_file, target_file) + print(f"Synced city_thread_ids.json: {len(target_data)} -> {len(merged)} cities") +else: + print(f"city_thread_ids.json already up to date: {len(target_data)} cities") +PY +} + +sync_city_thread_ids + PREVIOUS_TAG="" if [ -f "$TAG_FILE" ]; then PREVIOUS_TAG=$(cat "$TAG_FILE") diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index f3c0cfb8..7a6b090a 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -18,7 +18,7 @@ from src.database.runtime_state import ( get_state_storage_mode, ) from src.data_collection.city_registry import CITY_REGISTRY -from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env, parse_telegram_chat_ids from src.utils.telegram_i18n import ( copy_text as _copy, is_bilingual as _is_bilingual, @@ -33,7 +33,7 @@ _CITY_THREAD_IDS_PATH = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "data", "city_thread_ids.json", ) -_FORUM_CHAT_ID = "-1003927451869" +_DEFAULT_FORUM_CHAT_ID = "-1003927451869" _city_thread_ids: dict = {} # Shared HTTP session for AROME and auxiliary queries (connection reuse) @@ -106,9 +106,25 @@ def _load_city_thread_ids() -> dict: return {} +def _forum_chat_ids() -> Set[str]: + return set( + parse_telegram_chat_ids( + os.getenv("TELEGRAM_FORUM_CHAT_ID"), + os.getenv("POLYWEATHER_TELEGRAM_TOPICS_GROUP_ID"), + os.getenv("POLYWEATHER_TELEGRAM_GROUP_ID"), + _DEFAULT_FORUM_CHAT_ID, + ) + ) + + +def _is_forum_chat_id(chat_id: Any) -> bool: + chat_key = str(chat_id or "").strip() + return bool(chat_key and chat_key in _forum_chat_ids()) + + def _resolve_thread_id(chat_id: str, city: str) -> int: """Return message_thread_id for a given chat and city, or 0 if not a forum topic.""" - if chat_id != _FORUM_CHAT_ID: + if not _is_forum_chat_id(chat_id): return 0 mapping = _load_city_thread_ids() city_key = (city or "").strip().lower() @@ -1739,6 +1755,14 @@ def _process_airport_city( thread_id = _resolve_thread_id(chat_id, city) if thread_id: kwargs["message_thread_id"] = thread_id + elif _is_forum_chat_id(chat_id): + logger.warning( + "airport push skipped missing forum thread mapping city={} chat_id={} mapping_cities={}", + city, + chat_id, + len(_load_city_thread_ids()), + ) + continue _rate_limited_send(bot, chat_id, message, **kwargs) sent = True except Exception as exc: diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index 01c22a2a..3700ac81 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -172,6 +172,16 @@ def test_deploy_script_retries_image_pull_for_registry_propagation(): assert "docker compose pull && pull_ok=1 && break" in script +def test_deploy_script_syncs_city_thread_ids_into_runtime_volume(): + script = (ROOT / "deploy.sh").read_text(encoding="utf-8") + + assert "sync_city_thread_ids()" in script + assert 'repo_file="$COMPOSE_DIR/data/city_thread_ids.json"' in script + assert 'target_file="$runtime_dir/city_thread_ids.json"' in script + assert "merged.update(target_data)" in script + assert script.index("sync_city_thread_ids") < script.index("Updating Redis dependency") + + def test_deploy_script_retries_startup_smoke_checks(): script = (ROOT / "deploy.sh").read_text(encoding="utf-8") diff --git a/tests/test_telegram_hashtags.py b/tests/test_telegram_hashtags.py index b79d4c64..0b3a70e2 100644 --- a/tests/test_telegram_hashtags.py +++ b/tests/test_telegram_hashtags.py @@ -445,6 +445,44 @@ def test_airport_push_retries_main_chat_when_forum_thread_is_missing(monkeypatch ] +def test_airport_push_does_not_fall_back_to_general_when_forum_mapping_is_missing(monkeypatch): + import src.utils.telegram_push as telegram_push + + calls = [] + + monkeypatch.setattr( + telegram_push, + "_load_airport_city_weather_for_push", + lambda _city: { + "local_time": "15:22", + "current": {"temp": 30.0}, + "deb": {"prediction": 31.0}, + "airport_primary": { + "temp": 30.0, + "obs_time": "2026-06-11T07:52:00+00:00", + }, + }, + ) + monkeypatch.setattr(telegram_push, "_resolve_thread_id", lambda _chat, _city: 0) + monkeypatch.setattr(telegram_push, "_is_forum_chat_id", lambda _chat: True) + monkeypatch.setattr( + telegram_push, + "_rate_limited_send", + lambda _bot, chat_id, _message, **kwargs: calls.append((chat_id, kwargs)), + ) + + result = telegram_push._process_airport_city( + "chengdu", + now_ts=1781164400, + last_city={}, + chat_ids=["-1003927451869"], + bot=object(), + ) + + assert result is None + assert calls == [] + + def test_high_freq_airport_push_workers_default_to_one_for_shared_cpu(monkeypatch): source = Path("src/utils/telegram_push.py").read_text(encoding="utf-8") assert 'TELEGRAM_AIRPORT_PUSH_MAX_WORKERS", 1' in source