feat: Implement user point system with message activity tracking and command cost deductions, introducing a new db_manager.py for database operations.

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"❌ 积分不足，无法执行 <b>{label}</b>\n"
+                f"当前积分: <code>{balance}</code>\n"
+                f"需要积分: <code>{required}</code>\n"
+                f"还差积分: <code>{missing}</code>\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 = (
             "🌡️ <b>PolyWeather 天气查询机器人</b>\n\n"
             "可用指令:\n"
-            "/city [城市名] - 查询城市天气预测与实测\n"
+            f"/city [城市名] - 查询城市天气预测与实测 (消耗 {CITY_QUERY_COST} 积分)\n"
-            "/deb [城市名] - 查看 DEB 融合预测准确率\n"
+            f"/deb [城市名] - 查看 DEB 融合预测准确率 (消耗 {DEB_QUERY_COST} 积分)\n"
             "/points - 查看你的积分与排行榜\n"
             "/id - 获取当前聊天的 Chat ID\n\n"
             "示例: <code>/city 伦敦</code>\n"
-            "💡 <i>提示: 在群内发言可获得积分，积分可用于后续解锁高级版。</i>"
+            f"💡 <i>提示: 群内有效发言满 {MESSAGE_MIN_LENGTH} 字，每次 +{MESSAGE_POINTS} 分，"
+            f"每日上限 {MESSAGE_DAILY_CAP} 分。</i>"
         )
         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"👤 <b>我的状态：</b>\n"
                 f"└ 积分: <code>{user_info['points']}</code>\n"
-                f"└ 发言: <code>{user_info['message_count']}</code> 次"
+                f"└ 发言: <code>{user_info['message_count']}</code> 次\n"
+                f"└ 今日发言积分: <code>{user_info.get('daily_points') or 0}/{MESSAGE_DAILY_CAP}</code>\n"
+                f"└ /city 消耗: <code>{CITY_QUERY_COST}</code> | /deb 消耗: <code>{DEB_QUERY_COST}</code>"
             )
         
         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💳 本次消耗 <code>{DEB_QUERY_COST}</code> 积分。")
             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💳 本次消耗 <b>{CITY_QUERY_COST}</b> 积分。")
             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 分)
+        result = db.add_message_activity(
-        db.add_message_activity(user.id, points_to_add=1)
+            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()

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):
+    def add_message_activity(
-        """Update message count and add points with simple anti-spam logic."""
+        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]:
+            if not row:
-                last_at = datetime.fromisoformat(row[0])
+                return {"awarded": False, "reason": "user_missing"}
-                if (now - last_at).total_seconds() < 5:  # 5 second cooldown
+
-                    return False
+            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)