ewoooc/run_telegram_bot.py

#!/usr/bin/env python3
"""
Telegram Bot 獨立執行腳本

用法:
    python run_telegram_bot.py

環境變數:
    TELEGRAM_BOT_TOKEN: Telegram Bot Token (必填)

功能:
    - 啟動 Telegram Bot 監聽
    - 每日 09:00 推播趨勢摘要
    - 處理用戶指令：/trend, /search, /copy, /keywords, /daily, /settings
"""

import os
import sys
import asyncio
import logging
import re
from datetime import datetime, time
from dotenv import load_dotenv

# 載入環境變數
load_dotenv()

# 設定日誌
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('logs/telegram_bot.log', encoding='utf-8')
    ]
)
logger = logging.getLogger('TelegramBot')

_TELEGRAM_BOT_URL_RE = re.compile(r"(api\.telegram\.org/bot)[^/\s\"]+")
_TELEGRAM_TOKEN_RE = re.compile(r"\b\d{8,12}:[A-Za-z0-9_-]{32,64}\b")


def _redact_sensitive_log_value(value):
    if not isinstance(value, str):
        return value
    value = _TELEGRAM_BOT_URL_RE.sub(r"\1<redacted>", value)
    return _TELEGRAM_TOKEN_RE.sub("<telegram-token>", value)


class SensitiveTokenFilter(logging.Filter):
    def filter(self, record):
        record.msg = _redact_sensitive_log_value(record.msg)
        if isinstance(record.args, tuple):
            record.args = tuple(_redact_sensitive_log_value(arg) for arg in record.args)
        elif isinstance(record.args, dict):
            record.args = {
                key: _redact_sensitive_log_value(value)
                for key, value in record.args.items()
            }
        return True


for handler in logging.getLogger().handlers:
    handler.addFilter(SensitiveTokenFilter())

# python-telegram-bot uses httpx internally; INFO logs include Telegram Bot API URLs.
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)

def check_dependencies():
    """檢查必要的套件"""
    try:
        from telegram import Update
        from telegram.ext import Application
        logger.info("✅ python-telegram-bot 已安裝")
        return True
    except ImportError:
        logger.error("❌ 請安裝 python-telegram-bot: pip install python-telegram-bot")
        return False

def check_token():
    """檢查 Bot Token"""
    token = os.getenv('TELEGRAM_BOT_TOKEN')
    if not token:
        logger.error("❌ 請在 .env 設定 TELEGRAM_BOT_TOKEN")
        logger.info("   1. 在 Telegram 搜尋 @BotFather")
        logger.info("   2. 發送 /newbot 建立新 Bot")
        logger.info("   3. 複製 Token 到 .env 檔案")
        return None
    logger.info("✅ Bot Token 已設定")
    return token

async def main():
    """主程式"""
    print("=" * 60)
    print("  MOMO Pro System - Telegram Bot")
    print("=" * 60)
    print()

    # 檢查依賴
    if not check_dependencies():
        sys.exit(1)

    # 檢查 Token
    token = check_token()
    if not token:
        sys.exit(1)

    # 導入 Bot 服務
    try:
        from services.telegram_bot_service import TelegramBotService
        logger.info("✅ TelegramBotService 已載入")
    except ImportError as e:
        logger.error(f"❌ 無法載入 TelegramBotService: {e}")
        sys.exit(1)

    # 建立 Bot 服務
    bot_service = TelegramBotService(token)

    if not bot_service.should_run_polling():
        logger.warning(
            "Webhook 已設定，Polling Bot 已跳過啟動；請使用 OpenClaw webhook 路徑處理互動。"
        )
        return

    # 取得 Application
    app = bot_service.get_application()

    if not app:
        logger.error("❌ 無法建立 Bot Application")
        sys.exit(1)

    # 設定每日推播排程 (每天 09:00)
    from telegram.ext import JobQueue

    async def daily_summary_job(context):
        """每日摘要推播任務"""
        logger.info("📤 執行每日趨勢摘要推播...")
        await bot_service.send_daily_summary()

    # 啟動 Bot
    logger.info("🚀 啟動 Telegram Bot...")
    logger.info("   指令列表:")
    logger.info("   /trend [分類] - 查看熱門趨勢")
    logger.info("   /search [關鍵字] - AI 網路搜尋")
    logger.info("   /copy [商品名] - 生成行銷文案")
    logger.info("   /keywords - 熱門關鍵字")
    logger.info("   /daily - 每日趨勢摘要")
    logger.info("   /settings - 通知設定")
    print()

    # 初始化並啟動
    await app.initialize()
    await app.start()

    # 設定每日推播 (09:00 台北時間)
    # 注意：此排程（send_daily_summary）與 momo-scheduler 的 run_daily_report_task 功能不同：
    #   - 此處：從 TrendAnalysis 表推播「趨勢摘要」給訂閱 notify_daily_summary=True 的 Telegram 用戶
    #   - momo-scheduler：呼叫 OpenClaw generate_daily_report()，產出業績快報＋競品威脅＋圖表推播
    # 兩者各自負責不同數據源與受眾，均需保留。
    job_queue = app.job_queue
    if job_queue:
        # 計算下一個 09:00 的時間
        target_time = time(hour=9, minute=0, second=0)
        job_queue.run_daily(daily_summary_job, time=target_time, name='daily_summary')
        logger.info("📅 已設定每日 09:00 推播趨勢摘要")

    # 開始輪詢
    await app.updater.start_polling(drop_pending_updates=True)

    logger.info("✅ Bot 已啟動，按 Ctrl+C 停止")

    # 保持運行
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        logger.info("🛑 收到停止信號...")
    finally:
        await app.updater.stop()
        await app.stop()
        await app.shutdown()
        logger.info("👋 Bot 已停止")

if __name__ == '__main__':
    # 確保 logs 目錄存在
    os.makedirs('logs', exist_ok=True)

    # 執行
    asyncio.run(main())