diff --git a/bot_listener.py b/bot_listener.py index 70bcd7ea..cbfb5e9d 100644 --- a/bot_listener.py +++ b/bot_listener.py @@ -16,6 +16,13 @@ from src.data_collection.city_risk_profiles import get_city_risk_profile # type from src.analysis.deb_algorithm import calculate_dynamic_weights, update_daily_record # noqa: E402 from src.database.db_manager import DBManager +MESSAGE_POINTS = 1 +MESSAGE_DAILY_CAP = 20 +MESSAGE_MIN_LENGTH = 4 +MESSAGE_COOLDOWN_SEC = 30 +CITY_QUERY_COST = 2 +DEB_QUERY_COST = 3 + def analyze_weather_trend(weather_data, temp_symbol, city_name=None): """Thin wrapper — delegates to shared trend_engine module.""" @@ -36,17 +43,45 @@ def start_bot(): weather = WeatherDataCollector(config) start_trade_alert_push_loop(bot, config) + def _display_name(user) -> str: + return user.username or user.first_name or f"User_{user.id}" + + def _ensure_query_points(message, cost: int, label: str) -> bool: + user = message.from_user + db.upsert_user(user.id, _display_name(user)) + result = 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) + bot.reply_to( + message, + ( + f"❌ 积分不足，无法执行 {label}\n" + f"当前积分: {balance}\n" + f"需要积分: {required}\n" + f"还差积分: {missing}\n\n" + f"积分规则：群内有效发言满 {MESSAGE_MIN_LENGTH} 字，" + f"每次 +{MESSAGE_POINTS} 分，每日上限 {MESSAGE_DAILY_CAP} 分。" + ), + parse_mode="HTML", + ) + return False + @bot.message_handler(commands=["start", "help"]) def send_welcome(message): welcome_text = ( "🌡️ PolyWeather 天气查询机器人\n\n" "可用指令:\n" - "/city [城市名] - 查询城市天气预测与实测\n" - "/deb [城市名] - 查看 DEB 融合预测准确率\n" + f"/city [城市名] - 查询城市天气预测与实测 (消耗 {CITY_QUERY_COST} 积分)\n" + f"/deb [城市名] - 查看 DEB 融合预测准确率 (消耗 {DEB_QUERY_COST} 积分)\n" "/points - 查看你的积分与排行榜\n" "/id - 获取当前聊天的 Chat ID\n\n" "示例: /city 伦敦\n" - "💡 提示: 在群内发言可获得积分，积分可用于后续解锁高级版。" + f"💡 提示: 群内有效发言满 {MESSAGE_MIN_LENGTH} 字，每次 +{MESSAGE_POINTS} 分，" + f"每日上限 {MESSAGE_DAILY_CAP} 分。" ) bot.reply_to(message, welcome_text, parse_mode="HTML") @@ -62,6 +97,7 @@ def start_bot(): def show_points(message): """显示当前用户的积分及排行榜""" user = message.from_user + db.upsert_user(user.id, _display_name(user)) user_info = db.get_user(user.id) leaderboard = db.get_leaderboard(limit=5) @@ -76,7 +112,9 @@ def start_bot(): rank_text += ( f"👤 我的状态：\n" f"└ 积分: {user_info['points']}\n" - f"└ 发言: {user_info['message_count']} 次" + f"└ 发言: {user_info['message_count']} 次\n" + f"└ 今日发言积分: {user_info.get('daily_points') or 0}/{MESSAGE_DAILY_CAP}\n" + f"└ /city 消耗: {CITY_QUERY_COST} | /deb 消耗: {DEB_QUERY_COST}" ) bot.send_message(message.chat.id, rank_text, parse_mode="HTML") @@ -110,6 +148,9 @@ def start_bot(): ) return + if not _ensure_query_points(message, DEB_QUERY_COST, "/deb"): + return + city_data = data[city_name] from datetime import datetime as _dt @@ -259,6 +300,7 @@ def start_bot(): else: lines.append("\n⏳ 尚无完整的 DEB 预测记录，明天起开始统计。") + lines.append(f"\n💳 本次消耗 {DEB_QUERY_COST} 积分。") bot.reply_to(message, "\n".join(lines), parse_mode="HTML") except Exception as e: bot.reply_to(message, f"❌ 查询失败: {e}") @@ -312,6 +354,9 @@ def start_bot(): ) return + if not _ensure_query_points(message, CITY_QUERY_COST, "/city"): + return + bot.send_message( message.chat.id, f"🔍 正在查询 {city_name.title()} 的天气数据..." ) @@ -724,6 +769,7 @@ def start_bot(): except Exception as e: logger.error(f"调用 Groq AI 分析失败: {e}") + msg_lines.append(f"\n💳 本次消耗 {CITY_QUERY_COST} 积分。") bot.send_message(message.chat.id, "\n".join(msg_lines), parse_mode="HTML") except Exception as e: @@ -737,13 +783,26 @@ def start_bot(): """全量监听消息，用于记录群内发言积分(非指令消息)""" if message.text.startswith('/'): return + if message.chat.type not in ("group", "supergroup"): + return user = message.from_user - username = user.username or user.first_name or f"User_{user.id}" + username = _display_name(user) db.upsert_user(user.id, username) - - # 增加积分 (5秒冷却防刷屏，每个自然发言给 1 分) - db.add_message_activity(user.id, points_to_add=1) + + result = db.add_message_activity( + user.id, + text=message.text, + points_to_add=MESSAGE_POINTS, + cooldown_sec=MESSAGE_COOLDOWN_SEC, + daily_cap=MESSAGE_DAILY_CAP, + min_text_length=MESSAGE_MIN_LENGTH, + ) + if result.get("awarded"): + logger.info( + f"message points awarded user={user.id} points=+{MESSAGE_POINTS} " + f"daily_points={result.get('daily_points')}/{MESSAGE_DAILY_CAP}" + ) logger.info("🤖 Bot 启动中...") bot.infinity_polling() diff --git a/src/database/db_manager.py b/src/database/db_manager.py index 6445c258..a36d8029 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -36,14 +36,26 @@ class DBManager: is_group_premium BOOLEAN DEFAULT 0, group_expiry TIMESTAMP, points INTEGER DEFAULT 0, + daily_points INTEGER DEFAULT 0, + daily_points_date TEXT, message_count INTEGER DEFAULT 0, last_message_at TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) + self._ensure_column(conn, "users", "daily_points", "INTEGER DEFAULT 0") + self._ensure_column(conn, "users", "daily_points_date", "TEXT") conn.commit() logger.info(f"Database initialized successfully path={self.db_path}") + def _ensure_column(self, conn: sqlite3.Connection, table: str, column: str, ddl: str) -> None: + existing = { + row[1] + for row in conn.execute(f"PRAGMA table_info({table})").fetchall() + } + if column not in existing: + conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}") + def get_user(self, telegram_id: int) -> Optional[Dict[str, Any]]: with self._get_connection() as conn: conn.row_factory = sqlite3.Row @@ -73,26 +85,99 @@ class DBManager: """, (telegram_id, username)) conn.commit() - def add_message_activity(self, telegram_id: int, points_to_add: int = 1): - """Update message count and add points with simple anti-spam logic.""" + def add_message_activity( + self, + telegram_id: int, + text: str, + points_to_add: int = 1, + cooldown_sec: int = 30, + daily_cap: int = 20, + min_text_length: int = 4, + ) -> Dict[str, Any]: + """Award points for valid group activity with cooldown and daily cap.""" now = datetime.now() + normalized = "".join((text or "").split()) + if len(normalized) < min_text_length: + return {"awarded": False, "reason": "too_short"} + + today_str = now.strftime("%Y-%m-%d") with self._get_connection() as conn: - cursor = conn.execute("SELECT last_message_at FROM users WHERE telegram_id = ?", (telegram_id,)) + conn.row_factory = sqlite3.Row + cursor = conn.execute( + """ + SELECT points, daily_points, daily_points_date, last_message_at + FROM users WHERE telegram_id = ? + """, + (telegram_id,), + ) row = cursor.fetchone() - if row and row[0]: - last_at = datetime.fromisoformat(row[0]) - if (now - last_at).total_seconds() < 5: # 5 second cooldown - return False - + if not row: + return {"awarded": False, "reason": "user_missing"} + + last_message_at = row["last_message_at"] + if last_message_at: + last_at = datetime.fromisoformat(last_message_at) + if (now - last_at).total_seconds() < cooldown_sec: + return {"awarded": False, "reason": "cooldown"} + + daily_points = int(row["daily_points"] or 0) + daily_points_date = row["daily_points_date"] or "" + if daily_points_date != today_str: + daily_points = 0 + + if daily_points >= daily_cap: + conn.execute( + """ + UPDATE users + SET last_message_at = ?, daily_points = ?, daily_points_date = ? + WHERE telegram_id = ? + """, + (now.isoformat(), daily_points, today_str, telegram_id), + ) + conn.commit() + return {"awarded": False, "reason": "daily_cap", "daily_points": daily_points} + conn.execute(""" UPDATE users SET message_count = message_count + 1, points = points + ?, + daily_points = ?, + daily_points_date = ?, last_message_at = ? WHERE telegram_id = ? - """, (points_to_add, now.isoformat(), telegram_id)) + """, (points_to_add, daily_points + points_to_add, today_str, now.isoformat(), telegram_id)) conn.commit() - return True + return { + "awarded": True, + "reason": "ok", + "daily_points": daily_points + points_to_add, + } + + def spend_points(self, telegram_id: int, amount: int) -> Dict[str, Any]: + if amount <= 0: + user = self.get_user(telegram_id) + return {"ok": True, "balance": int((user or {}).get("points") or 0)} + + with self._get_connection() as conn: + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT points FROM users WHERE telegram_id = ?", + (telegram_id,), + ).fetchone() + if not row: + return {"ok": False, "reason": "user_missing", "balance": 0, "required": amount} + + balance = int(row["points"] or 0) + if balance < amount: + return {"ok": False, "reason": "insufficient_points", "balance": balance, "required": amount} + + new_balance = balance - amount + conn.execute( + "UPDATE users SET points = ? WHERE telegram_id = ?", + (new_balance, telegram_id), + ) + conn.commit() + return {"ok": True, "balance": new_balance, "spent": amount} def set_premium(self, telegram_id: int, plan: str, months: int = 1): expiry = datetime.now() + timedelta(days=30 * months)