from __future__ import annotations import os from datetime import datetime from typing import Any, Dict, Optional, Tuple from loguru import logger from src.bot.settings import ( CITY_DAILY_FREE_LIMIT, DEB_DAILY_FREE_LIMIT, MESSAGE_COOLDOWN_SEC, MESSAGE_DAILY_CAP, MESSAGE_MIN_LENGTH, MESSAGE_POINTS, ) from src.database.db_manager import DBManager class BotIOLayer: """Telegram IO + points/account side effects.""" def __init__(self, bot: Any, db: DBManager): self.bot = bot self.db = db self.query_topic_map = self._parse_topic_map( os.getenv("TELEGRAM_QUERY_TOPIC_MAP") ) self.message_cooldown_map = self._parse_int_map( os.getenv("POLYWEATHER_BOT_MESSAGE_COOLDOWN_BY_CHAT") ) self.query_topic_chat_id = str( os.getenv("TELEGRAM_QUERY_TOPIC_CHAT_ID") or "" ).strip() self.query_topic_id = self._safe_int( os.getenv("TELEGRAM_QUERY_TOPIC_ID"), default=0, ) @staticmethod def display_name(user: Any) -> str: return user.username or user.first_name or f"User_{user.id}" @staticmethod def _safe_int(raw: Any, default: int = 0) -> int: try: return int(raw) except Exception: return default @staticmethod def _parse_topic_map(raw: Optional[str]) -> Dict[str, int]: """ Parse TELEGRAM_QUERY_TOPIC_MAP: - "-1003586303099:25513,-1003539418691:25514" - Supports comma/semicolon/newline separators. """ out: Dict[str, int] = {} if not raw: return out normalized = str(raw).replace("\r", ",").replace("\n", ",").replace(";", ",") for part in normalized.split(","): row = part.strip() if not row or ":" not in row: continue chat_id, topic_raw = row.split(":", 1) chat_id = str(chat_id or "").strip() topic_id = BotIOLayer._safe_int(topic_raw, default=0) if chat_id and topic_id > 0: out[chat_id] = topic_id return out @staticmethod def _parse_int_map(raw: Optional[str]) -> Dict[str, int]: """ Parse env maps like: - "-1003586303099:10,-1003539418691:20" - Supports comma/semicolon/newline separators. """ out: Dict[str, int] = {} if not raw: return out normalized = str(raw).replace("\r", ",").replace("\n", ",").replace(";", ",") for part in normalized.split(","): row = part.strip() if not row or ":" not in row: continue key, value_raw = row.split(":", 1) key = str(key or "").strip() value = BotIOLayer._safe_int(value_raw, default=-1) if key and value >= 0: out[key] = value return out def _resolve_message_cooldown(self, chat_id: Any) -> int: chat_key = str(chat_id).strip() if chat_id is not None else "" if chat_key and chat_key in self.message_cooldown_map: return self.message_cooldown_map[chat_key] return MESSAGE_COOLDOWN_SEC def _resolve_query_target( self, source_chat_id: Any, ) -> Tuple[Optional[str], int]: src = str(source_chat_id).strip() if source_chat_id is not None else "" if src and src in self.query_topic_map: return src, self.query_topic_map[src] if self.query_topic_chat_id: return self.query_topic_chat_id, self.query_topic_id # No mapping/fixed topic configured: reply directly to source chat without topic. return src or None, 0 def send_query_message( self, message: Any, text: str, *, parse_mode: str | None = None, ) -> None: chat = getattr(message, "chat", None) fallback_chat_id = getattr(chat, "id", None) target_chat_id, target_topic_id = self._resolve_query_target(fallback_chat_id) if target_chat_id is None: self.bot.send_message(message.chat.id, text, parse_mode=parse_mode) return kwargs = {} if parse_mode: kwargs["parse_mode"] = parse_mode if target_topic_id > 0: kwargs["message_thread_id"] = target_topic_id try: self.bot.send_message(target_chat_id, text, **kwargs) return except Exception as exc: logger.warning( "query topic send failed chat_id={} topic_id={} source_chat_id={} error={}", target_chat_id, target_topic_id, fallback_chat_id, exc, ) # Fallback: drop topic and send to source chat. if fallback_chat_id is not None: fallback_kwargs = dict(kwargs) fallback_kwargs.pop("message_thread_id", None) self.bot.send_message(fallback_chat_id, text, **fallback_kwargs) def ensure_query_points(self, message: Any, cost: int, label: str) -> bool: user = message.from_user self.db.upsert_user(user.id, self.display_name(user)) result = self.db.spend_points(user.id, cost) if result.get("ok"): return True balance = int(result.get("balance") or 0) required = int(result.get("required") or cost) missing = max(0, required - balance) self.send_query_message( message, ( f"❌ 积分不足,无法执行 {label}\n" f"当前积分: {balance}\n" f"需要积分: {required}\n" f"还差积分: {missing}\n\n" f"积分规则:有效发言满 {MESSAGE_MIN_LENGTH} 字获得 {MESSAGE_POINTS} 积分," f"每日上限 {MESSAGE_DAILY_CAP} 分。" ), parse_mode="HTML", ) return False def build_welcome_text(self) -> str: return ( "🚀 PolyWeather 天气查询机器人\n\n" "可用指令:\n" f"/city [城市名] 或 /pwcity [城市名] - 查询城市天气预测与实测 (免费, 每日 {CITY_DAILY_FREE_LIMIT} 次)\n" f"/deb [城市名] 或 /pwdeb [城市名] - 查看 DEB 融合预测准确率 (免费, 每日 {DEB_DAILY_FREE_LIMIT} 次)\n" "/markets - 私聊机器人查看当前市场监控摘要\n" "/top - 查看积分排行榜\n" "/id - 获取当前聊天的 Chat ID\n\n" "/diag - 查看 Bot 启动诊断\n\n" "/bind - 绑定 Supabase 账号(可选)\n" "/unbind - 解除当前 Telegram 与网页账号绑定\n\n" "🔗 机器人: @WeatherQuant_bot\n" "👥 社群: 加入 Telegram 群组\n\n" "📌 私有频道用于接收自动推送;手动查看市场概览请私聊机器人发送 /markets\n\n" "🔐 /city 与 /deb 仅限官方群成员使用。\n\n" "示例: /city 伦敦/pwcity 伦敦\n" f"💡 提示: 群内有效发言(满 {MESSAGE_MIN_LENGTH} 字)获得 {MESSAGE_POINTS} 积分," f"每日上限 {MESSAGE_DAILY_CAP} 分。" f"首次发言额外奖励 20 积分,每日首条消息 +2 积分。" ) def build_points_rank_text(self, user: Any) -> str: self.db.upsert_user(user.id, self.display_name(user)) user_info = self.db.get_user(user.id) now = datetime.now() today_str = now.strftime("%Y-%m-%d") weekly_profile = self.db.get_weekly_profile(user.id) week_key = str(weekly_profile.get("week_key") or "") leaderboard = self.db.get_weekly_leaderboard(limit=5) rank_text = f"🏆 PolyWeather 周活跃度排行榜 ({week_key})\n" rank_text += "────────────────────\n" for i, entry in enumerate(leaderboard): medal = ["🥇", "🥈", "🥉", " ", " "][i] if i < 5 else " " username = (entry.get("username") or "unknown")[:12] weekly_points = int(entry.get("weekly_points") or 0) rank_text += f"{medal} {username}: {weekly_points} 点\n" if user_info: daily_points = int(user_info.get("daily_points") or 0) daily_points_date = str(user_info.get("daily_points_date") or "") if daily_points_date != today_str: daily_points = 0 if daily_points > MESSAGE_DAILY_CAP: daily_points = MESSAGE_DAILY_CAP weekly_points = int(weekly_profile.get("weekly_points") or 0) weekly_rank = weekly_profile.get("weekly_rank") ranked_count = int(weekly_profile.get("total_ranked") or 0) weekly_rank_text = ( f"{weekly_rank}/{ranked_count}" if weekly_rank and ranked_count > 0 else "未上榜" ) daily_queries_date = str(user_info.get("daily_queries_date") or "") city_used = int(user_info.get("daily_city_queries") or 0) if daily_queries_date == today_str else 0 deb_used = int(user_info.get("daily_deb_queries") or 0) if daily_queries_date == today_str else 0 rank_text += "────────────────────\n" rank_text += ( "👤 我的状态:\n" f"┣ 累计积分: {user_info['points']}\n" f"┣ 累计发言: {user_info['message_count']} 次\n" f"┣ 本周排名: {weekly_rank_text}\n" f"┣ 本周发言积分: {weekly_points}\n" f"┣ 今日发言积分: {daily_points}/{MESSAGE_DAILY_CAP}\n" f"┗ /city 免费 ({city_used}/{CITY_DAILY_FREE_LIMIT}) | /deb 免费 ({deb_used}/{DEB_DAILY_FREE_LIMIT})" ) return rank_text def track_group_text_activity(self, message: Any) -> None: text = str(getattr(message, "text", "") or "") if text.startswith("/"): return chat = getattr(message, "chat", None) if not chat or chat.type not in ("group", "supergroup"): return user = message.from_user username = self.display_name(user) cooldown_sec = self._resolve_message_cooldown(getattr(chat, "id", None)) preview = text.strip().replace("\n", " ") if len(preview) > 80: preview = preview[:77] + "..." logger.info( "group text received chat_id={} thread_id={} user_id={} text_len={} cooldown_sec={} preview={!r}", getattr(chat, "id", None), getattr(message, "message_thread_id", None), getattr(user, "id", None), len(text.strip()), cooldown_sec, preview, ) self.db.upsert_user(user.id, username) result = self.db.add_message_activity( user.id, text=text, points_to_add=MESSAGE_POINTS, cooldown_sec=cooldown_sec, daily_cap=MESSAGE_DAILY_CAP, min_text_length=MESSAGE_MIN_LENGTH, ) if result.get("awarded"): awarded = int(result.get("points_added") or MESSAGE_POINTS) logger.info( f"message points awarded user={user.id} points=+{awarded} " f"daily_points={result.get('daily_points')}/{MESSAGE_DAILY_CAP}" ) return logger.info( "message points skipped chat_id={} thread_id={} user_id={} reason={} daily_points={} weekly_points={}", getattr(chat, "id", None), getattr(message, "message_thread_id", None), getattr(user, "id", None), result.get("reason") or "unknown", result.get("daily_points"), result.get("weekly_points"), )