Add Telegram topic binding command

scripts/create_forum_topics.py

@@ -41,22 +41,23 @@ BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "").strip()
 CHAT_ID = -1003927451869
 
 CITIES: list[tuple[str, str]] = [
+    ("🇰🇷 Seoul", "seoul"),
+    ("🇰🇷 Busan", "busan"),
+    ("🇯🇵 Tokyo", "tokyo"),
     ("🇨🇳 Beijing", "beijing"),
     ("🇨🇳 Shanghai", "shanghai"),
     ("🇨🇳 Guangzhou", "guangzhou"),
+    ("🇨🇳 Shenzhen / Lau Fau Shan", "shenzhen"),
     ("🇨🇳 Qingdao", "qingdao"),
     ("🇨🇳 Chengdu", "chengdu"),
     ("🇨🇳 Chongqing", "chongqing"),
     ("🇨🇳 Wuhan", "wuhan"),
     ("🇭🇰 Hong Kong", "hong kong"),
-    ("🇭🇰 Lau Fau Shan", "lau fau shan"),
     ("🇹🇼 Taipei", "taipei"),
-    ("🇰🇷 Seoul", "seoul"),
-    ("🇰🇷 Busan", "busan"),
-    ("🇯🇵 Tokyo", "tokyo"),
     ("🇸🇬 Singapore", "singapore"),
     ("🇹🇷 Istanbul", "istanbul"),
     ("🇹🇷 Ankara", "ankara"),
+    ("🇮🇱 Tel Aviv", "tel aviv"),
     ("🇫🇮 Helsinki", "helsinki"),
     ("🇳🇱 Amsterdam", "amsterdam"),
     ("🇫🇷 Paris", "paris"),

src/bot/handlers/basic.py

@@ -9,14 +9,34 @@ from loguru import logger  # type: ignore
 
 from src.bot.command_parser import extract_command_name
 from src.bot.command_parser import looks_like_slash_command
+from src.bot.command_parser import split_command_and_args
 from src.bot.io_layer import BotIOLayer
 from src.bot.observability import CommandTrace
 from src.bot.runtime_coordinator import RuntimeStatus, render_runtime_status_html
 from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT
 from src.auth.telegram_group_pricing import TelegramGroupPricing, TELEGRAM_MEMBER_STATUSES
-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
 
-_BASIC_COMMANDS = {"start", "help", "id", "top", "diag", "bind", "unbind"}
+_BASIC_COMMANDS = {"start", "help", "id", "top", "diag", "bind", "unbind", "bindtopic"}
+
+
+def normalize_airport_push_city(raw: str) -> str:
+    from src.utils.telegram_push import normalize_airport_push_city as _normalize
+
+    return _normalize(raw)
+
+
+def record_city_thread_id(city: str, thread_id: int) -> dict:
+    from src.utils.telegram_push import record_city_thread_id as _record
+
+    return _record(city, thread_id)
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
 
 
 class BasicCommandHandler:
@@ -59,6 +79,10 @@ class BasicCommandHandler:
         def _unbind(message):
             self._dispatch(message)
 
+        @self.bot.message_handler(commands=["bindtopic"])
+        def _bindtopic(message):
+            self._dispatch(message)
+
         if hasattr(self.bot, "chat_join_request_handler"):
             @self.bot.chat_join_request_handler(func=lambda request: True)
             def _chat_join_request(request):
@@ -118,6 +142,9 @@ class BasicCommandHandler:
         if command == "unbind":
             self.handle_unbind(message)
             return
+        if command == "bindtopic":
+            self.handle_bindtopic(message)
+            return
 
     def handle_start_help(self, message: Any) -> None:
         trace = CommandTrace("/start", message)
@@ -158,6 +185,83 @@ class BasicCommandHandler:
         )
         return "replied"
 
+    def _configured_forum_chat_ids(self) -> 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"),
+                os.getenv("TELEGRAM_CHAT_IDS"),
+                os.getenv("TELEGRAM_CHAT_ID"),
+            )
+        )
+
+    def _can_bind_topic(self, message: Any) -> bool:
+        if not _env_bool("POLYWEATHER_TOPIC_BIND_ADMIN_ONLY", True):
+            return True
+        chat = getattr(message, "chat", None)
+        user = getattr(message, "from_user", None)
+        chat_id = getattr(chat, "id", None)
+        user_id = getattr(user, "id", None)
+        if chat_id is None or user_id is None or not hasattr(self.bot, "get_chat_member"):
+            return False
+        try:
+            member = self.bot.get_chat_member(chat_id, user_id)
+        except Exception as exc:
+            logger.warning("bindtopic admin check failed chat_id={} user_id={}: {}", chat_id, user_id, exc)
+            return False
+        status = str(getattr(member, "status", "") or "").strip().lower()
+        return status in {"creator", "administrator"}
+
+    def handle_bindtopic(self, message: Any) -> str:
+        trace = CommandTrace("/bindtopic", message)
+        try:
+            chat = getattr(message, "chat", None)
+            chat_id = getattr(chat, "id", None)
+            chat_type = str(getattr(chat, "type", "") or "").strip().lower()
+            if chat_type not in {"group", "supergroup"} or chat_id is None:
+                self.bot.reply_to(message, "❌ 请在目标 Telegram 群子话题里执行 /bindtopic <city>。")
+                trace.set_status("blocked", "not_group_topic")
+                return "not_group_topic"
+            configured_chat_ids = self._configured_forum_chat_ids()
+            if configured_chat_ids and str(chat_id) not in configured_chat_ids:
+                self.bot.reply_to(message, "❌ 当前群不在推送目标配置里，已拒绝绑定。")
+                trace.set_status("blocked", "chat_not_configured")
+                return "chat_not_configured"
+            thread_id = int(getattr(message, "message_thread_id", 0) or 0)
+            if thread_id <= 0:
+                self.bot.reply_to(message, "❌ 请进入具体子话题后再执行 /bindtopic <city>。")
+                trace.set_status("blocked", "missing_thread_id")
+                return "missing_thread_id"
+            if not self._can_bind_topic(message):
+                self.bot.reply_to(message, "❌ 只有群管理员可以绑定推送子话题。")
+                trace.set_status("blocked", "unauthorized")
+                return "unauthorized"
+            _command, raw_city = split_command_and_args(getattr(message, "text", "") or "")
+            city = normalize_airport_push_city(raw_city)
+            if not city:
+                self.bot.reply_to(
+                    message,
+                    "❌ 未识别城市。用法：<code>/bindtopic seoul</code> 或 <code>/bindtopic LauFauShan</code>",
+                    parse_mode="HTML",
+                )
+                trace.set_status("blocked", "unsupported_city")
+                return "unsupported_city"
+            result = record_city_thread_id(city, thread_id)
+            self.bot.reply_to(
+                message,
+                (
+                    "✅ 已绑定推送子话题\n"
+                    f"city=<code>{html.escape(city)}</code>\n"
+                    f"thread_id=<code>{int(result.get('thread_id') or thread_id)}</code>"
+                ),
+                parse_mode="HTML",
+            )
+            trace.set_status("ok", city)
+            return "bound"
+        finally:
+            trace.emit()
+
     def _prompt_bind_from_web_token(self, message: Any, token: str) -> str:
         if not token:
             self.bot.reply_to(message, "❌ 绑定链接无效，请回到网页重新点击一键绑定。")

src/bot/io_layer.py

@@ -197,7 +197,8 @@ class BotIOLayer:
             "/id - 获取当前聊天的 Chat ID\n\n"
             "/diag - 查看 Bot 启动诊断\n\n"
             "/bind - 绑定 Supabase 账号（可选）\n"
-            "/unbind - 解除当前 Telegram 与网页账号绑定\n\n"
+            "/unbind - 解除当前 Telegram 与网页账号绑定\n"
+            "/bindtopic <city> - 管理员在子话题内绑定城市推送\n\n"
             "🔗 机器人: <a href=\"https://t.me/polyyuanbot\">@polyyuanbot</a>\n"
             "👥 社群: <a href=\"https://t.me/+Io5H9oVHFmVjOTQ5\">加入 Telegram 群组</a>"
         )

src/utils/telegram_push.py

@@ -17,6 +17,7 @@ from src.database.runtime_state import (
     TelegramAlertStateRepository,
     get_state_storage_mode,
 )
+from src.data_collection.city_registry import ALIASES
 from src.data_collection.city_registry import CITY_REGISTRY
 from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env, parse_telegram_chat_ids
 from src.utils.telegram_i18n import (
@@ -35,6 +36,7 @@ _CITY_THREAD_IDS_PATH = os.path.join(
 )
 _DEFAULT_FORUM_CHAT_ID = "-1003927451869"
 _city_thread_ids: dict = {}
+_CITY_THREAD_IDS_LOCK = threading.Lock()
 
 # Shared HTTP session for AROME and auxiliary queries (connection reuse)
 _HTTP_SESSION: Optional[requests_lib.Session] = None
@@ -106,6 +108,75 @@ def _load_city_thread_ids() -> dict:
     return {}
 
 
+def normalize_airport_push_city(raw: str) -> str:
+    city = str(raw or "").strip().lower().replace("-", " ")
+    city = re.sub(r"\s+", " ", city)
+    compact = city.replace(" ", "")
+    city = ALIASES.get(city, ALIASES.get(compact, city))
+    if city not in HIGH_FREQ_AIRPORT_CITIES:
+        city = {
+            candidate.replace(" ", ""): candidate
+            for candidate in HIGH_FREQ_AIRPORT_CITIES
+        }.get(compact, city)
+    if city in HIGH_FREQ_AIRPORT_CITIES:
+        return city
+    return ""
+
+
+def _city_thread_ids_write_path() -> str:
+    env_path = str(os.getenv("POLYWEATHER_CITY_THREAD_IDS_PATH") or "").strip()
+    if env_path:
+        return env_path
+    for path in (
+        "/var/lib/polyweather/city_thread_ids.json",
+        "/app/data/city_thread_ids.json",
+        _CITY_THREAD_IDS_PATH,
+    ):
+        if os.path.isfile(path):
+            return path
+    return "/var/lib/polyweather/city_thread_ids.json"
+
+
+def record_city_thread_id(city: str, thread_id: int) -> dict:
+    """Persist a forum topic mapping and update the in-process cache."""
+
+    global _city_thread_ids
+    normalized_city = normalize_airport_push_city(city)
+    if not normalized_city:
+        raise ValueError(f"unsupported airport push city: {city}")
+    try:
+        normalized_thread_id = int(thread_id)
+    except Exception as exc:
+        raise ValueError(f"invalid message_thread_id: {thread_id}") from exc
+    if normalized_thread_id <= 0:
+        raise ValueError(f"invalid message_thread_id: {thread_id}")
+
+    with _CITY_THREAD_IDS_LOCK:
+        path = _city_thread_ids_write_path()
+        mapping = dict(_load_city_thread_ids())
+        mapping[normalized_city] = normalized_thread_id
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        tmp_path = f"{path}.tmp"
+        with open(tmp_path, "w", encoding="utf-8") as f:
+            json.dump(mapping, f, ensure_ascii=False, indent=2, sort_keys=True)
+            f.write("\n")
+        os.replace(tmp_path, path)
+        _city_thread_ids = mapping
+        logger.info(
+            "recorded forum topic mapping city={} thread_id={} path={} total={}",
+            normalized_city,
+            normalized_thread_id,
+            path,
+            len(mapping),
+        )
+        return {
+            "city": normalized_city,
+            "thread_id": normalized_thread_id,
+            "path": path,
+            "total": len(mapping),
+        }
+
+
 def _forum_chat_ids() -> Set[str]:
     return set(
         parse_telegram_chat_ids(
@@ -626,6 +697,7 @@ HIGH_FREQ_AIRPORT_CITIES = {
     "seoul", "singapore", "busan", "tokyo", "ankara", "helsinki", "amsterdam",
     "istanbul", "paris", "hong kong", "taipei",
     "beijing", "shanghai", "guangzhou", "qingdao", "chengdu", "chongqing", "wuhan",
+    "shenzhen",
     "new york", "los angeles", "chicago", "denver", "atlanta",
     "miami", "san francisco", "houston", "dallas", "austin", "seattle",
     "tel aviv",
@@ -638,7 +710,7 @@ HIGH_FREQ_AIRPORT_ICAO = {
     "ankara": "17128", "helsinki": "EFHK", "amsterdam": "EHAM", "istanbul": "17058",
     "paris": "LFPB", "hong kong": "HKO", "taipei": "466920",
     "beijing": "ZBAA", "shanghai": "ZSPD", "guangzhou": "ZGGG", "qingdao": "ZSQD",
-    "chengdu": "ZUUU", "chongqing": "ZUCK", "wuhan": "ZHHH",
+    "chengdu": "ZUUU", "chongqing": "ZUCK", "wuhan": "ZHHH", "shenzhen": "LFS",
     "new york": "KLGA", "los angeles": "KLAX", "chicago": "KORD",
     "denver": "KBKF", "atlanta": "KATL", "miami": "KMIA",
     "san francisco": "KSFO", "houston": "KHOU", "dallas": "KDAL",
@@ -1145,6 +1217,7 @@ def _build_airport_status_message(
                    "taipei": "Songshan", "beijing": "Capital", "shanghai": "Pudong",
                    "guangzhou": "Baiyun", "qingdao": "Jiaodong",
                    "chengdu": "Shuangliu", "chongqing": "Jiangbei", "wuhan": "Tianhe",
+                   "shenzhen": "Lau Fau Shan",
                    "new york": "LaGuardia", "los angeles": "LAX", "chicago": "O'Hare",
                    "denver": "Buckley", "atlanta": "Hartsfield", "miami": "Intl",
                    "san francisco": "SFO", "houston": "Hobby", "dallas": "Love Field",

tests/test_bot_basic_handler.py

@@ -11,6 +11,7 @@ class DummyBot:
         self.approved_join_requests = []
         self.declined_join_requests = []
         self.callback_handlers = []
+        self.chat_member_status = "administrator"
 
     def reply_to(self, message, text, parse_mode=None, disable_web_page_preview=None, **kwargs):
         self.replies.append(
@@ -65,6 +66,9 @@ class DummyBot:
     def decline_chat_join_request(self, chat_id, user_id):
         self.declined_join_requests.append({"chat_id": chat_id, "user_id": user_id})
 
+    def get_chat_member(self, chat_id, user_id):
+        return SimpleNamespace(status=self.chat_member_status)
+
 
 def _message(text: str):
     return SimpleNamespace(
@@ -74,6 +78,15 @@ def _message(text: str):
     )
 
 
+def _topic_message(text: str, *, thread_id: int = 13102):
+    return SimpleNamespace(
+        text=text,
+        from_user=SimpleNamespace(id=1, username="admin", first_name="Admin"),
+        chat=SimpleNamespace(id=-1003927451869, type="supergroup"),
+        message_thread_id=thread_id,
+    )
+
+
 def test_basic_handler_diag_returns_html():
     runtime = RuntimeStatus(
         started_at="2026-03-12 00:00:00 UTC",
@@ -100,6 +113,78 @@ def test_basic_handler_diag_returns_html():
     assert "Bot 启动诊断" in bot.replies[0]["text"]
 
 
+def test_bindtopic_records_current_forum_thread(monkeypatch):
+    import src.bot.handlers.basic as basic
+
+    saved = []
+    monkeypatch.setenv("TELEGRAM_CHAT_IDS", "-1003927451869")
+    monkeypatch.setenv("POLYWEATHER_TELEGRAM_GROUP_ID", "-1003927451869")
+    monkeypatch.setattr(
+        basic,
+        "record_city_thread_id",
+        lambda city, thread_id: saved.append((city, thread_id))
+        or {"city": city, "thread_id": thread_id},
+    )
+
+    bot = DummyBot()
+    handler = BasicCommandHandler(
+        bot=bot,
+        io_layer=SimpleNamespace(
+            build_welcome_text=lambda: "WELCOME",
+            build_points_rank_text=lambda _user: "TOP",
+        ),
+        runtime_status_provider=lambda: RuntimeStatus(
+            started_at="2026-03-12 00:00:00 UTC",
+            loops=[],
+            command_access_mode="public",
+            protected_commands=[],
+            required_group_chat_id="-1003927451869",
+        ),
+    )
+
+    result = handler.handle_bindtopic(_topic_message("/bindtopic seoul"))
+
+    assert result == "bound"
+    assert saved == [("seoul", 13102)]
+    assert "seoul" in bot.replies[0]["text"].lower()
+    assert "13102" in bot.replies[0]["text"]
+
+
+def test_bindtopic_rejects_non_admin(monkeypatch):
+    import src.bot.handlers.basic as basic
+
+    saved = []
+    monkeypatch.setenv("TELEGRAM_CHAT_IDS", "-1003927451869")
+    monkeypatch.setattr(
+        basic,
+        "record_city_thread_id",
+        lambda city, thread_id: saved.append((city, thread_id)),
+    )
+
+    bot = DummyBot()
+    bot.chat_member_status = "member"
+    handler = BasicCommandHandler(
+        bot=bot,
+        io_layer=SimpleNamespace(
+            build_welcome_text=lambda: "WELCOME",
+            build_points_rank_text=lambda _user: "TOP",
+        ),
+        runtime_status_provider=lambda: RuntimeStatus(
+            started_at="2026-03-12 00:00:00 UTC",
+            loops=[],
+            command_access_mode="public",
+            protected_commands=[],
+            required_group_chat_id="-1003927451869",
+        ),
+    )
+
+    result = handler.handle_bindtopic(_topic_message("/bindtopic seoul"))
+
+    assert result == "unauthorized"
+    assert saved == []
+    assert "管理员" in bot.replies[0]["text"]
+
+
 def test_start_bind_token_binds_telegram_to_web_account():
     bot = DummyBot()
     consumed = []

tests/test_telegram_hashtags.py

@@ -5,6 +5,7 @@ from src.utils.telegram_push import (
     _build_airport_status_message,
     _compute_slope_15m,
     _due_airport_cities,
+    normalize_airport_push_city,
     _parse_observation_time_epoch,
     _run_high_freq_airport_cycle,
     _telegram_push_language,
@@ -165,9 +166,14 @@ def test_singapore_is_in_telegram_push_city_lists():
     assert HIGH_FREQ_AIRPORT_ICAO["singapore"] == "WSSS"
 
 
-def test_shenzhen_is_not_in_airport_push_city_lists():
+def test_shenzhen_lau_fau_shan_topic_is_in_airport_push_city_lists():
-    assert "shenzhen" not in HIGH_FREQ_AIRPORT_CITIES
+    assert normalize_airport_push_city("LauFauShan") == "shenzhen"
-    assert "shenzhen" not in HIGH_FREQ_AIRPORT_ICAO
+    assert normalize_airport_push_city("HongKong") == "hong kong"
+    assert normalize_airport_push_city("NewYork") == "new york"
+    assert normalize_airport_push_city("LosAngeles") == "los angeles"
+    assert normalize_airport_push_city("SanFrancisco") == "san francisco"
+    assert "shenzhen" in HIGH_FREQ_AIRPORT_CITIES
+    assert HIGH_FREQ_AIRPORT_ICAO["shenzhen"] == "LFS"
 
 
 def test_china_airport_push_defaults_to_one_minute_city_interval():
@@ -182,6 +188,10 @@ def test_china_airport_push_defaults_to_one_minute_city_interval():
     assert _AIRPORT_PUSH_INTERVAL["wuhan"] == 60
 
 
+def test_shenzhen_lau_fau_shan_push_uses_hko_ten_minute_interval():
+    assert _AIRPORT_PUSH_INTERVAL["shenzhen"] == 600
+
+
 def test_airport_push_prioritizes_china_markets():
     due = _due_airport_cities(
         {"paris", "shanghai", "wuhan", "ankara", "beijing"},