Add script to grant points by Supabase email

scripts/grant_points.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+
+PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, PROJECT_ROOT)
+
+
+def main() -> int:
+    from src.database.db_manager import DBManager
+
+    parser = argparse.ArgumentParser(description="Manually grant PolyWeather points.")
+    parser.add_argument("--email", required=True, help="Bound Supabase email")
+    parser.add_argument("--points", type=int, required=True, help="Points to add")
+    args = parser.parse_args()
+
+    db = DBManager()
+    result = db.grant_points_by_supabase_email(args.email, args.points)
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return 0 if result.get("ok") else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+

src/database/db_manager.py

@@ -354,6 +354,54 @@ class DBManager:
         except Exception:
             return 0
 
+    def grant_points_by_supabase_email(
+        self,
+        supabase_email: str,
+        amount: int,
+    ) -> Dict[str, Any]:
+        email = str(supabase_email or "").strip().lower()
+        points = int(amount or 0)
+        if not email:
+            return {"ok": False, "reason": "invalid_supabase_email"}
+        if points <= 0:
+            return {"ok": False, "reason": "invalid_amount"}
+
+        with self._get_connection() as conn:
+            conn.row_factory = sqlite3.Row
+            row = conn.execute(
+                """
+                SELECT telegram_id, username, points, supabase_email
+                FROM users
+                WHERE lower(trim(COALESCE(supabase_email, ''))) = ?
+                LIMIT 1
+                """,
+                (email,),
+            ).fetchone()
+            if not row:
+                return {"ok": False, "reason": "user_not_found", "supabase_email": email}
+
+            telegram_id = int(row["telegram_id"] or 0)
+            before = int(row["points"] or 0)
+            after = before + points
+            conn.execute(
+                """
+                UPDATE users
+                SET points = ?
+                WHERE telegram_id = ?
+                """,
+                (after, telegram_id),
+            )
+            conn.commit()
+            return {
+                "ok": True,
+                "telegram_id": telegram_id,
+                "username": str(row["username"] or ""),
+                "supabase_email": str(row["supabase_email"] or email),
+                "points_before": before,
+                "points_added": points,
+                "points_after": after,
+            }
+
     def upsert_user(self, telegram_id: int, username: str):
         with self._get_connection() as conn:
             conn.execute("""

tests/test_grant_points.py

@@ -0,0 +1,22 @@
+from src.database.db_manager import DBManager
+
+
+def test_grant_points_by_supabase_email(tmp_path):
+    db = DBManager(str(tmp_path / "test.db"))
+    db.upsert_user(1001, "eraer")
+    with db._get_connection() as conn:  # noqa: SLF001
+        conn.execute(
+            """
+            UPDATE users
+            SET points = ?, supabase_email = ?
+            WHERE telegram_id = ?
+            """,
+            (50, "eraer031905@gmail.com", 1001),
+        )
+        conn.commit()
+
+    result = db.grant_points_by_supabase_email("eraer031905@gmail.com", 300)
+    assert result["ok"] is True
+    assert result["points_before"] == 50
+    assert result["points_added"] == 300
+    assert result["points_after"] == 350