新增积分转账功能：支持管理员手动扣除和划转用户积分

web/routers/ops.py

@@ -14,6 +14,7 @@ from web.services.ops_api import (
     get_ops_weekly_leaderboard,
     get_ops_user_subscriptions,
     grant_ops_points,
+    transfer_ops_points,
     grant_ops_subscription,
     list_ops_memberships,
     list_ops_payment_incidents,
@@ -78,6 +79,17 @@ async def ops_grant_points(request: Request, body: GrantPointsRequest):
     return grant_ops_points(request, body)
 
 
+@router.post("/api/ops/users/transfer-points")
+async def ops_transfer_points(request: Request):
+    import json as _json
+    body_bytes = await request.body()
+    body = _json.loads(body_bytes.decode("utf-8"))
+    from_email = str(body.get("from_email") or "").strip()
+    to_email = str(body.get("to_email") or "").strip()
+    amount = int(body.get("amount") or 0)
+    return transfer_ops_points(request, from_email=from_email, to_email=to_email, amount=amount)
+
+
 @router.get("/api/ops/analytics/funnel")
 async def ops_analytics_funnel(request: Request, days: int = 30):
     return get_ops_analytics_funnel(request, days=days)

web/services/ops_api.py

@@ -239,6 +239,29 @@ def grant_ops_points(request: Request, body: GrantPointsRequest) -> Dict[str, An
     return result
 
 
+def transfer_ops_points(
+    request: Request,
+    from_email: str = "",
+    to_email: str = "",
+    amount: int = 0,
+) -> Dict[str, Any]:
+    """Transfer points from one user to another."""
+    admin = _require_ops(request) or {}
+    from_email = str(from_email or "").strip()
+    to_email = str(to_email or "").strip()
+    amount = int(amount or 0)
+    if not from_email or not to_email:
+        raise HTTPException(status_code=400, detail="from_email and to_email are required")
+    if amount <= 0:
+        raise HTTPException(status_code=400, detail="amount must be positive")
+    db = DBManager()
+    result = db.transfer_points_by_email(from_email, to_email, amount)
+    result["operator_email"] = admin.get("email")
+    if not result.get("ok"):
+        raise HTTPException(status_code=400, detail=result)
+    return result
+
+
 def get_ops_analytics_funnel(request: Request, days: int = 30) -> Dict[str, Any]:
     _require_ops(request)
     db = DBManager()