refactor(openclaw): 抽出 Telegram API helper

CONSTITUTION.md

@@ -2,7 +2,7 @@
 
 > 本文件定義專案開發的核心準則與不可違反的規範
 > **建立日期**: 2026-01-12
-> **當前版本**: V10.22 (Legacy 5888 入口清理版)
+> **當前版本**: V10.23 (OpenClaw Bot Telegram helper 拆分版)
 > **最後更新**: 2026-04-30
 
 ---

TODO_NEXT_STEPS.txt

@@ -31,9 +31,10 @@
    - ElephantAlpha transient fallback：NVIDIA NIM timeout、connection error、429 與 5xx 會嘗試下一個 fallback model；400 等非暫時性請求錯誤不重試。
    - 模組化治理守門：新增 `docs/guides/modularization_governance.md`、`docs/memory/code_modularization_inventory_20260430.md` 與 `tests/test_modularization_governance.py`，盤點並鎖住 15 個 >800 行 Python 大檔。
    - Legacy 5888 入口清理：刪除 `tests/main_test.py` standalone Flask 死碼，測試與自動匯入文件改用 Port 80 入口。
+   - OpenClaw Bot 第一刀拆分：Telegram API send/retry/file upload helper 已移至 `services/openclaw_bot/telegram_api.py`，避免 Blueprint 持續承接 delivery glue。
 
 【下次待辦】
-   - 依 inventory 優先拆 `routes/openclaw_bot_routes.py`、`routes/sales_routes.py`、`scheduler.py`。
+   - 繼續依 inventory 拆 `routes/openclaw_bot_routes.py`：下一刀優先拆 menu keyboard 或 report formatting。
    - 觀察 Prometheus scrape 後 `momo_ai_*` baseline 與非 baseline 事件序列是否持續穩定。
    - Superset panel 設定與 Smoke 摘要成效觀察。
 

app.py

@@ -95,8 +95,8 @@ except Exception as e:
     sys_log.error(f"無法檢測磁碟空間: {e}")
 
 # 🚩 系統版本定義 (備份與顯示用)
-# 🚩 2026-04-30 V10.22: Legacy standalone test and port cleanup
-SYSTEM_VERSION = "V10.22"
+# 🚩 2026-04-30 V10.23: OpenClaw Telegram API helper extraction
+SYSTEM_VERSION = "V10.23"
 
 # ==========================================
 # 🔒 SQL Injection 防護函數

config.py

@@ -254,7 +254,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
 # ==========================================
 # 系統版本與路徑
 # ==========================================
-SYSTEM_VERSION = "V10.22"
+SYSTEM_VERSION = "V10.23"
 LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
 public_url = PUBLIC_URL  # 用於模板顯示
 

docs/memory/code_modularization_inventory_20260430.md

@@ -13,7 +13,7 @@
 
 | 行數 | 檔案 | 分類 | 拆分方向 |
 |---:|---|---|---|
-| 5543 | `routes/openclaw_bot_routes.py` | P0 巨型 Blueprint | route / bot command service / report service / scheduler hook |
+| 5437 | `routes/openclaw_bot_routes.py` | P0 巨型 Blueprint | route / bot command service / report service / scheduler hook |
 | 2653 | `routes/sales_routes.py` | P0 巨型 Blueprint | page routes / API routes / chart query service / calendar service |
 | 2644 | `scheduler.py` | P0 排程總管 | task registry / crawler jobs / report jobs / notification jobs |
 | 1662 | `routes/ai_routes.py` | P1 AI Blueprint | route glue / AI orchestration service / prompt builders |
@@ -31,7 +31,7 @@
 
 ## 工作項目
 
-1. P0：拆 `routes/openclaw_bot_routes.py`，先把非 HTTP 邏輯搬到 `services/openclaw_bot/` 子模組。
+1. P0：持續拆 `routes/openclaw_bot_routes.py`；Telegram API helper 已先搬到 `services/openclaw_bot/telegram_api.py`，下一步拆 menu keyboard 或 report formatting。
 2. P0：拆 `routes/sales_routes.py`，先把 chart/query/calendar 計算搬到 `services/sales/`。
 3. P0：拆 `scheduler.py`，建立 `jobs/` 或 `services/scheduler/` task registry。
 4. P1：把 `routes/ai_routes.py` 與 `routes/vendor_routes.py` 的資料處理移出 route。

docs/memory/history_logs.md

@@ -44,6 +44,7 @@
 - **ElephantAlpha transient fallback**: NVIDIA NIM primary model timeout、connection error、429 與 5xx 會嘗試下一個 fallback model，400 等非暫時性請求錯誤不重試。
 - **模組化治理守門**: 盤點 15 個超過 800 行 Python 大檔，新增 `docs/guides/modularization_governance.md` 與 `tests/test_modularization_governance.py`，防止未分類巨檔再長出來。
 - **Legacy 5888 入口清理**: 刪除 `tests/main_test.py` standalone Flask 死碼，測試與自動匯入文件改用 Port 80 `/auto_import` 入口。
+- **OpenClaw Bot 第一刀拆分**: Telegram API send/retry/file upload helper 移到 `services/openclaw_bot/telegram_api.py`，`routes/openclaw_bot_routes.py` 往 thin Blueprint 收斂。
 
 ### 2026-04-28~29：Phase 3e 重構大戰 + daily_sales cache 隱形 bug 根除
 - **app.py 縮減 -10.8%**: 7,386 → 6,590 行，11 commits 全綠零 502。

routes/openclaw_bot_routes.py

@@ -36,6 +36,14 @@ from services.mcp_context_service import (
     get_dcard_trends, get_youtube_trending, get_taiwan_weather,
     get_twbank_exchange_rates, get_upcoming_events,
 )
+from services.openclaw_bot.telegram_api import (
+    _tg,
+    answer_callback,
+    send_document,
+    send_message,
+    send_photo,
+    send_typing,
+)
 try:
     from services.openclaw_learning_service import (
         build_rag_context, store_conversation, store_insight,
@@ -134,119 +142,6 @@ def _check_rate_limit(user_id: int) -> bool:
 TRIGGER_KEYWORDS = []  # 空 = 全部回應（小龍蝦是專用業務群組）
 
 
-# ── Telegram API ──────────────────────────────────────────────
-def _tg(method: str, payload: dict):
-    try:
-        r = requests.post(f"{BOT_API_URL}/{method}", json=payload, timeout=10)
-        if not r.ok:
-            sys_log.warning(f"[OpenClawBot] {method} failed: {r.text[:200]}")
-        return r.json()
-    except Exception as e:
-        sys_log.error(f"[OpenClawBot] {method} error: {e}")
-        return {}
-
-
-def _strip_markdown(text: str) -> str:
-    """移除 Telegram Markdown v1 格式符號，轉為純文字（send fallback 用）"""
-    import re
-    # 移除 *bold*, _italic_, `code`, [link](url)
-    text = re.sub(r'\*([^*]+)\*', r'\1', text)
-    text = re.sub(r'_([^_]+)_', r'\1', text)
-    text = re.sub(r'`([^`]+)`', r'\1', text)
-    text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', text)
-    return text
-
-
-def send_message(chat_id, text, reply_to=None, keyboard=None, parse_mode='Markdown'):
-    """送出 Telegram 訊息。Markdown 解析失敗時自動降級為純文字重送。"""
-    def _build_payload(txt, pm):
-        p = {'chat_id': chat_id, 'text': txt, 'disable_web_page_preview': True}
-        if pm:
-            p['parse_mode'] = pm
-        if reply_to:
-            p['reply_to_message_id'] = reply_to
-        if keyboard:
-            p['reply_markup'] = {'inline_keyboard': keyboard}
-        return p
-
-    # 第一次嘗試（帶 parse_mode）
-    result = _tg('sendMessage', _build_payload(text, parse_mode))
-    if result.get('ok'):
-        return result
-
-    # Markdown 解析失敗 → 降級為純文字重送
-    if parse_mode and not result.get('ok'):
-        err = result.get('description', '')
-        if 'parse' in err.lower() or 'entity' in err.lower() or 'can\'t find' in err.lower():
-            sys_log.warning(f"[OpenClawBot] Markdown parse failed, retrying as plain text")
-            plain = _strip_markdown(text)
-            result2 = _tg('sendMessage', _build_payload(plain, None))
-            if result2.get('ok'):
-                return result2
-
-    # 最後嘗試：截斷過長文字
-    if len(text) > 4000:
-        sys_log.warning(f"[OpenClawBot] Message too long ({len(text)}), truncating")
-        result3 = _tg('sendMessage', _build_payload(_strip_markdown(text[:3900]) + '\n...(訊息過長已截斷)', None))
-        return result3
-
-    return result
-
-
-def answer_callback(cq_id, text=''):
-    _tg('answerCallbackQuery', {'callback_query_id': cq_id, 'text': text})
-
-
-def send_typing(chat_id):
-    try:
-        requests.post(f"{BOT_API_URL}/sendChatAction",
-                      json={'chat_id': chat_id, 'action': 'typing'}, timeout=5)
-    except Exception:
-        pass
-
-
-def send_photo(chat_id, file_path, caption='', reply_to=None):
-    """透過 Telegram Bot API 傳送圖片"""
-    try:
-        with open(file_path, 'rb') as f:
-            data = {'chat_id': chat_id}
-            if caption:
-                data['caption'] = caption
-            if reply_to:
-                data['reply_to_message_id'] = reply_to
-            r = requests.post(
-                f"{BOT_API_URL}/sendPhoto",
-                data=data, files={'photo': f}, timeout=30
-            )
-        if not r.ok:
-            sys_log.warning(f"[OpenClawBot] sendPhoto failed: {r.text[:200]}")
-        return r.json()
-    except Exception as e:
-        sys_log.error(f"[OpenClawBot] sendPhoto error: {e}")
-        return {}
-
-
-def send_document(chat_id, file_path, caption='', reply_to=None):
-    """透過 Telegram Bot API 傳送文件"""
-    try:
-        with open(file_path, 'rb') as f:
-            data = {'chat_id': chat_id}
-            if caption:
-                data['caption'] = caption
-            if reply_to:
-                data['reply_to_message_id'] = reply_to
-            r = requests.post(
-                f"{BOT_API_URL}/sendDocument",
-                data=data, files={'document': f}, timeout=30
-            )
-        if not r.ok:
-            sys_log.warning(f"[OpenClawBot] sendDocument failed: {r.text[:200]}")
-        return r.json()
-    except Exception as e:
-        sys_log.error(f"[OpenClawBot] sendDocument error: {e}")
-        return {}
-
-
 # ── 目標管理（記憶體，跨 session 用 DB 儲存）─────────────────────
 _GOALS: dict = {}  # {'daily','monthly','quarterly','half','yearly': float}
 _scheduler = None

services/openclaw_bot/__init__.py

@@ -0,0 +1 @@
+"""OpenClaw Bot service helpers."""

services/openclaw_bot/telegram_api.py

@@ -0,0 +1,134 @@
+"""Telegram Bot API helpers for OpenClaw Bot.
+
+Route modules should keep only request handling and command dispatch. Telegram
+delivery, Markdown fallback, and file upload glue live here so they can be
+reused and tested without importing the 5k-line Blueprint.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+
+import requests
+
+from services.logger_manager import SystemLogger
+
+
+BOT_TOKEN = os.getenv("OPENCLAW_BOT_TOKEN", "")
+BOT_API_URL = f"https://api.telegram.org/bot{BOT_TOKEN}"
+sys_log = SystemLogger("OpenClawBot").get_logger()
+
+
+def _tg(method: str, payload: dict):
+    try:
+        response = requests.post(f"{BOT_API_URL}/{method}", json=payload, timeout=10)
+        if not response.ok:
+            sys_log.warning(f"[OpenClawBot] {method} failed: {response.text[:200]}")
+        return response.json()
+    except Exception as exc:
+        sys_log.error(f"[OpenClawBot] {method} error: {exc}")
+        return {}
+
+
+def _strip_markdown(text: str) -> str:
+    """移除 Telegram Markdown v1 格式符號，轉為純文字（send fallback 用）"""
+    text = re.sub(r"\*([^*]+)\*", r"\1", text)
+    text = re.sub(r"_([^_]+)_", r"\1", text)
+    text = re.sub(r"`([^`]+)`", r"\1", text)
+    text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text)
+    return text
+
+
+def send_message(chat_id, text, reply_to=None, keyboard=None, parse_mode="Markdown"):
+    """送出 Telegram 訊息。Markdown 解析失敗時自動降級為純文字重送。"""
+
+    def _build_payload(txt, pm):
+        payload = {"chat_id": chat_id, "text": txt, "disable_web_page_preview": True}
+        if pm:
+            payload["parse_mode"] = pm
+        if reply_to:
+            payload["reply_to_message_id"] = reply_to
+        if keyboard:
+            payload["reply_markup"] = {"inline_keyboard": keyboard}
+        return payload
+
+    result = _tg("sendMessage", _build_payload(text, parse_mode))
+    if result.get("ok"):
+        return result
+
+    if parse_mode and not result.get("ok"):
+        err = result.get("description", "")
+        if "parse" in err.lower() or "entity" in err.lower() or "can't find" in err.lower():
+            sys_log.warning("[OpenClawBot] Markdown parse failed, retrying as plain text")
+            result2 = _tg("sendMessage", _build_payload(_strip_markdown(text), None))
+            if result2.get("ok"):
+                return result2
+
+    if len(text) > 4000:
+        sys_log.warning(f"[OpenClawBot] Message too long ({len(text)}), truncating")
+        truncated = _strip_markdown(text[:3900]) + "\n...(訊息過長已截斷)"
+        return _tg("sendMessage", _build_payload(truncated, None))
+
+    return result
+
+
+def answer_callback(cq_id, text=""):
+    return _tg("answerCallbackQuery", {"callback_query_id": cq_id, "text": text})
+
+
+def send_typing(chat_id):
+    try:
+        requests.post(
+            f"{BOT_API_URL}/sendChatAction",
+            json={"chat_id": chat_id, "action": "typing"},
+            timeout=5,
+        )
+    except Exception as exc:
+        sys_log.debug(f"[OpenClawBot] sendChatAction skipped: {exc}")
+
+
+def send_photo(chat_id, file_path, caption="", reply_to=None):
+    """透過 Telegram Bot API 傳送圖片"""
+    try:
+        with open(file_path, "rb") as handle:
+            data = {"chat_id": chat_id}
+            if caption:
+                data["caption"] = caption
+            if reply_to:
+                data["reply_to_message_id"] = reply_to
+            response = requests.post(
+                f"{BOT_API_URL}/sendPhoto",
+                data=data,
+                files={"photo": handle},
+                timeout=30,
+            )
+        if not response.ok:
+            sys_log.warning(f"[OpenClawBot] sendPhoto failed: {response.text[:200]}")
+        return response.json()
+    except Exception as exc:
+        sys_log.error(f"[OpenClawBot] sendPhoto error: {exc}")
+        return {}
+
+
+def send_document(chat_id, file_path, caption="", reply_to=None):
+    """透過 Telegram Bot API 傳送文件"""
+    try:
+        with open(file_path, "rb") as handle:
+            data = {"chat_id": chat_id}
+            if caption:
+                data["caption"] = caption
+            if reply_to:
+                data["reply_to_message_id"] = reply_to
+            response = requests.post(
+                f"{BOT_API_URL}/sendDocument",
+                data=data,
+                files={"document": handle},
+                timeout=30,
+            )
+        if not response.ok:
+            sys_log.warning(f"[OpenClawBot] sendDocument failed: {response.text[:200]}")
+        return response.json()
+    except Exception as exc:
+        sys_log.error(f"[OpenClawBot] sendDocument error: {exc}")
+        return {}

tests/test_openclaw_bot_telegram_api.py

@@ -0,0 +1,88 @@
+from pathlib import Path
+
+
+class FakeResponse:
+    def __init__(self, ok=True, payload=None, text=""):
+        self.ok = ok
+        self._payload = payload if payload is not None else {"ok": ok}
+        self.text = text
+
+    def json(self):
+        return self._payload
+
+
+def test_send_message_retries_plain_text_when_markdown_parse_fails(monkeypatch):
+    from services.openclaw_bot import telegram_api
+
+    calls = []
+
+    def fake_post(url, json=None, **_kwargs):
+        calls.append((url, json))
+        if len(calls) == 1:
+            return FakeResponse(False, {"ok": False, "description": "can't parse entities"})
+        return FakeResponse(True, {"ok": True, "result": {"message_id": 1}})
+
+    monkeypatch.setattr(telegram_api, "BOT_API_URL", "https://telegram.test/botTOKEN")
+    monkeypatch.setattr(telegram_api.requests, "post", fake_post)
+
+    result = telegram_api.send_message(123, "*bold* [link](https://example.com)")
+
+    assert result["ok"] is True
+    assert len(calls) == 2
+    assert calls[0][1]["parse_mode"] == "Markdown"
+    assert "parse_mode" not in calls[1][1]
+    assert calls[1][1]["text"] == "bold link"
+
+
+def test_send_message_truncates_overlong_text_after_failed_send(monkeypatch):
+    from services.openclaw_bot import telegram_api
+
+    calls = []
+
+    def fake_post(url, json=None, **_kwargs):
+        calls.append((url, json))
+        if len(calls) == 1:
+            return FakeResponse(False, {"ok": False, "description": "message is too long"})
+        return FakeResponse(True, {"ok": True})
+
+    monkeypatch.setattr(telegram_api, "BOT_API_URL", "https://telegram.test/botTOKEN")
+    monkeypatch.setattr(telegram_api.requests, "post", fake_post)
+
+    result = telegram_api.send_message(123, "*" + ("x" * 4100) + "*")
+
+    assert result["ok"] is True
+    assert len(calls) == 2
+    assert len(calls[1][1]["text"]) < 4000
+    assert calls[1][1]["text"].endswith("...(訊息過長已截斷)")
+
+
+def test_send_document_posts_file_payload(monkeypatch, tmp_path):
+    from services.openclaw_bot import telegram_api
+
+    uploaded = {}
+    document = tmp_path / "report.txt"
+    document.write_text("hello", encoding="utf-8")
+
+    def fake_post(url, data=None, files=None, **_kwargs):
+        uploaded["url"] = url
+        uploaded["data"] = data
+        uploaded["file_name"] = Path(files["document"].name).name
+        return FakeResponse(True, {"ok": True, "result": {"document": {}}})
+
+    monkeypatch.setattr(telegram_api, "BOT_API_URL", "https://telegram.test/botTOKEN")
+    monkeypatch.setattr(telegram_api.requests, "post", fake_post)
+
+    result = telegram_api.send_document(123, document, caption="caption", reply_to=9)
+
+    assert result["ok"] is True
+    assert uploaded["url"] == "https://telegram.test/botTOKEN/sendDocument"
+    assert uploaded["data"] == {"chat_id": 123, "caption": "caption", "reply_to_message_id": 9}
+    assert uploaded["file_name"] == "report.txt"
+
+
+def test_openclaw_routes_keep_tg_helper_import_for_webhook_management():
+    route_source = Path("routes/openclaw_bot_routes.py").read_text(encoding="utf-8")
+
+    assert "_tg('setMyCommands'" in route_source
+    assert "_tg('setWebhook'" in route_source
+    assert "    _tg,\n" in route_source