2569718930@qq.com
1164 lines
43 KiB
Python
1164 lines
43 KiB
Python
"""Ops payments / billing / membership service functions."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from fastapi import HTTPException, Request
|
|
|
|
from src.database.db_manager import DBManager # type hints
|
|
import web.routes as legacy_routes
|
|
|
|
|
|
def _get_db():
|
|
from src.database.db_manager import DBManager as _DBManager
|
|
return _DBManager()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Shared helpers
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
def _require_ops(request: Request) -> Dict[str, Any] | None:
|
|
from web.services.ops_api import _require_ops as _real
|
|
return _real(request)
|
|
|
|
|
|
def _parse_iso_datetime(value: Any) -> Optional[datetime]:
|
|
text = str(value or "").strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
|
|
except Exception:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def _to_utc_iso(value: datetime) -> str:
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def _supabase_rest_rows(
|
|
table: str,
|
|
params: Dict[str, Any],
|
|
*,
|
|
timeout: int = 10,
|
|
) -> List[Dict[str, Any]]:
|
|
from web.services.ops_api import _supabase_rest_rows as _real
|
|
|
|
return _real(table, params, timeout=timeout)
|
|
|
|
|
|
def _app_analytics_actor_key(row: Dict[str, Any]) -> str:
|
|
payload = row.get("payload")
|
|
payload = payload if isinstance(payload, dict) else {}
|
|
user_id = str(row.get("user_id") or payload.get("user_id") or "").strip().lower()
|
|
client_id = str(row.get("client_id") or "").strip()
|
|
session_id = str(row.get("session_id") or "").strip()
|
|
if user_id:
|
|
return f"user:{user_id}"
|
|
if client_id:
|
|
return f"client:{client_id}"
|
|
if session_id:
|
|
return f"session:{session_id}"
|
|
return f"event:{row.get('id')}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Payment helpers
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
def _payment_explorer_url(chain: Any, tx_hash: Any) -> str:
|
|
tx = str(tx_hash or "").strip()
|
|
if not tx:
|
|
return ""
|
|
chain_text = str(chain or "").strip().lower()
|
|
base = "https://etherscan.io" if "eth" in chain_text else "https://polygonscan.com"
|
|
return f"{base}/tx/{tx}"
|
|
|
|
|
|
def _risk_issue(
|
|
*,
|
|
category: str,
|
|
severity: str,
|
|
title: str,
|
|
detail: str,
|
|
user_id: Any = "",
|
|
created_at: Any = "",
|
|
reference: Any = "",
|
|
payload: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"category": category,
|
|
"severity": severity,
|
|
"title": title,
|
|
"detail": detail,
|
|
"user_id": str(user_id or ""),
|
|
"created_at": str(created_at or ""),
|
|
"reference": str(reference or ""),
|
|
"payload": payload or {},
|
|
}
|
|
|
|
|
|
def _normalize_payment_incident(item: Dict[str, Any]) -> Dict[str, Any]:
|
|
payload = item.get("payload") if isinstance(item, dict) else {}
|
|
payload = payload if isinstance(payload, dict) else {}
|
|
confirm_failure = (
|
|
payload.get("confirm_failure")
|
|
if isinstance(payload.get("confirm_failure"), dict)
|
|
else {}
|
|
)
|
|
reason = str(
|
|
payload.get("reason")
|
|
or confirm_failure.get("reason")
|
|
or payload.get("error")
|
|
or "unknown"
|
|
).strip().lower()
|
|
detail = str(
|
|
payload.get("detail")
|
|
or confirm_failure.get("detail")
|
|
or payload.get("message")
|
|
or payload.get("error")
|
|
or ""
|
|
).strip()
|
|
resolved_at = str(payload.get("resolved_at") or "").strip()
|
|
return {
|
|
**item,
|
|
"payload": payload,
|
|
"reason": reason or "unknown",
|
|
"detail": detail,
|
|
"intent_id": str(
|
|
payload.get("intent_id")
|
|
or payload.get("payment_intent_id")
|
|
or confirm_failure.get("intent_id")
|
|
or ""
|
|
).strip(),
|
|
"user_id": str(payload.get("user_id") or "").strip(),
|
|
"tx_hash": str(
|
|
payload.get("tx_hash")
|
|
or confirm_failure.get("tx_hash")
|
|
or ""
|
|
).strip(),
|
|
"refund_case_id": payload.get("refund_case_id"),
|
|
"refund_status": str(payload.get("refund_status") or "").strip(),
|
|
"resolved": bool(resolved_at),
|
|
"resolved_at": resolved_at,
|
|
"resolved_by": str(payload.get("resolved_by") or "").strip(),
|
|
}
|
|
|
|
|
|
def _payment_incident_group_key(item: Dict[str, Any]) -> Tuple[str, str, str, str]:
|
|
reason = str(item.get("reason") or "unknown").strip().lower()
|
|
intent_id = str(item.get("intent_id") or "").strip().lower()
|
|
tx_hash = str(item.get("tx_hash") or "").strip().lower()
|
|
user_id = str(item.get("user_id") or "").strip().lower()
|
|
if intent_id or tx_hash:
|
|
return reason, user_id, intent_id, tx_hash
|
|
return reason, user_id, f"event:{item.get('id')}", ""
|
|
|
|
|
|
def _group_payment_incidents(
|
|
incidents: List[Dict[str, Any]],
|
|
*,
|
|
reason: str = "",
|
|
include_resolved: bool = False,
|
|
) -> Dict[str, Any]:
|
|
normalized_reason = str(reason or "").strip().lower()
|
|
groups: Dict[Tuple[str, str, str, str], Dict[str, Any]] = {}
|
|
raw_total = 0
|
|
|
|
for item in incidents:
|
|
normalized_item = _normalize_payment_incident(item)
|
|
item_reason = str(normalized_item.get("reason") or "").strip().lower()
|
|
resolved = bool(normalized_item.get("resolved"))
|
|
if normalized_reason and item_reason != normalized_reason:
|
|
continue
|
|
if not include_resolved and resolved:
|
|
continue
|
|
raw_total += 1
|
|
|
|
key = _payment_incident_group_key(normalized_item)
|
|
created_at = str(normalized_item.get("created_at") or "").strip()
|
|
event_id = int(normalized_item.get("id") or 0)
|
|
existing = groups.get(key)
|
|
if existing is None:
|
|
grouped = {
|
|
**normalized_item,
|
|
"occurrence_count": 1,
|
|
"event_ids": [event_id] if event_id > 0 else [],
|
|
"first_seen_at": created_at,
|
|
"last_seen_at": created_at,
|
|
}
|
|
groups[key] = grouped
|
|
continue
|
|
|
|
existing["occurrence_count"] = int(existing.get("occurrence_count") or 1) + 1
|
|
if event_id > 0:
|
|
existing.setdefault("event_ids", []).append(event_id)
|
|
first_seen = str(existing.get("first_seen_at") or "").strip()
|
|
last_seen = str(existing.get("last_seen_at") or "").strip()
|
|
if created_at and (not first_seen or created_at < first_seen):
|
|
existing["first_seen_at"] = created_at
|
|
if created_at and (not last_seen or created_at > last_seen):
|
|
existing["last_seen_at"] = created_at
|
|
|
|
grouped_items = list(groups.values())
|
|
grouped_items.sort(
|
|
key=lambda item: (
|
|
str(item.get("last_seen_at") or item.get("created_at") or ""),
|
|
int(item.get("id") or 0),
|
|
),
|
|
reverse=True,
|
|
)
|
|
return {
|
|
"incidents": grouped_items,
|
|
"raw_total": raw_total,
|
|
"total": len(grouped_items),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Payment incident API
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
def list_ops_payment_incidents(
|
|
request: Request,
|
|
limit: int = 50,
|
|
reason: str = "",
|
|
include_resolved: bool = False,
|
|
) -> Dict[str, Any]:
|
|
_require_ops(request)
|
|
db = _get_db()
|
|
safe_limit = max(1, min(int(limit or 50), 200))
|
|
incidents: List[Dict[str, Any]] = []
|
|
for event_type in ("payment_intent_failed", "payment_refund_required"):
|
|
try:
|
|
rows = db.list_payment_audit_events(
|
|
limit=max(safe_limit, 500),
|
|
event_type=event_type,
|
|
)
|
|
incidents.extend(
|
|
row for row in rows
|
|
if str(row.get("event_type") or "").strip().lower() == event_type
|
|
)
|
|
except Exception:
|
|
continue
|
|
terminal_refund_statuses = {"refunded", "rejected", "closed"}
|
|
list_refund_cases = getattr(db, "list_refund_cases", None)
|
|
if callable(list_refund_cases):
|
|
try:
|
|
refund_cases = list_refund_cases(limit=max(safe_limit, 500))
|
|
except Exception:
|
|
refund_cases = []
|
|
for case in refund_cases:
|
|
if not isinstance(case, dict):
|
|
continue
|
|
status = str(case.get("status") or "").strip().lower()
|
|
if not include_resolved and status in terminal_refund_statuses:
|
|
continue
|
|
incidents.append(
|
|
{
|
|
"id": int(case.get("id") or 0),
|
|
"event_type": "payment_refund_case",
|
|
"payload": {
|
|
"reason": str(case.get("reason") or "refund_required"),
|
|
"intent_id": case.get("intent_id"),
|
|
"user_id": case.get("user_id"),
|
|
"tx_hash": case.get("tx_hash"),
|
|
"refund_case_id": case.get("id"),
|
|
"refund_status": status,
|
|
},
|
|
"created_at": case.get("created_at"),
|
|
}
|
|
)
|
|
grouped = _group_payment_incidents(
|
|
incidents,
|
|
reason=reason,
|
|
include_resolved=include_resolved,
|
|
)
|
|
return {
|
|
**grouped,
|
|
"incidents": grouped["incidents"][:safe_limit],
|
|
}
|
|
|
|
|
|
def list_ops_refund_cases(
|
|
request: Request,
|
|
limit: int = 50,
|
|
status: str = "",
|
|
) -> Dict[str, Any]:
|
|
_require_ops(request)
|
|
db = _get_db()
|
|
safe_limit = max(1, min(int(limit or 50), 200))
|
|
return {
|
|
"refunds": db.list_refund_cases(
|
|
limit=safe_limit,
|
|
status=str(status or "").strip().lower() or None,
|
|
)
|
|
}
|
|
|
|
|
|
def create_ops_refund_case(
|
|
request: Request,
|
|
*,
|
|
reason: str,
|
|
intent_id: str = "",
|
|
tx_hash: str = "",
|
|
user_id: str = "",
|
|
amount_usdc: str = "",
|
|
note: str = "",
|
|
) -> Dict[str, Any]:
|
|
admin = _require_ops(request) or {}
|
|
actor_email = str(admin.get("email") or "").strip().lower()
|
|
db = _get_db()
|
|
created = db.create_refund_case(
|
|
reason=reason,
|
|
intent_id=intent_id,
|
|
tx_hash=tx_hash,
|
|
user_id=user_id,
|
|
amount_usdc=amount_usdc,
|
|
created_by=actor_email,
|
|
note=note,
|
|
)
|
|
if not created or created.get("ok") is False:
|
|
raise HTTPException(status_code=400, detail=created or "refund_case_failed")
|
|
db.append_ops_audit_event(
|
|
action="refund_case_create",
|
|
actor_email=actor_email,
|
|
target_user_id=str(user_id or ""),
|
|
target_type="refund_case",
|
|
target_id=str(created.get("id") or ""),
|
|
payload={
|
|
"reason": reason,
|
|
"intent_id": intent_id,
|
|
"tx_hash": tx_hash,
|
|
"amount_usdc": amount_usdc,
|
|
},
|
|
)
|
|
db.append_payment_audit_event(
|
|
"payment_refund_required",
|
|
{
|
|
"reason": str(reason or "refund_required").strip().lower(),
|
|
"intent_id": intent_id,
|
|
"user_id": user_id,
|
|
"tx_hash": tx_hash,
|
|
"refund_case_id": created.get("id"),
|
|
},
|
|
)
|
|
return {"ok": True, "refund": created}
|
|
|
|
|
|
def update_ops_refund_case(
|
|
request: Request,
|
|
*,
|
|
case_id: int,
|
|
status: str,
|
|
note: str = "",
|
|
) -> Dict[str, Any]:
|
|
admin = _require_ops(request) or {}
|
|
actor_email = str(admin.get("email") or "").strip().lower()
|
|
db = _get_db()
|
|
updated = db.update_refund_case(
|
|
case_id,
|
|
status=status,
|
|
handled_by=actor_email,
|
|
note=note,
|
|
)
|
|
if not updated:
|
|
raise HTTPException(status_code=404, detail="refund_case_not_found")
|
|
db.append_ops_audit_event(
|
|
action="refund_case_update",
|
|
actor_email=actor_email,
|
|
target_user_id=str(updated.get("user_id") or ""),
|
|
target_type="refund_case",
|
|
target_id=str(case_id),
|
|
payload={
|
|
"status": status,
|
|
"note": note,
|
|
"intent_id": updated.get("intent_id"),
|
|
"tx_hash": updated.get("tx_hash"),
|
|
},
|
|
)
|
|
return {"ok": True, "refund": updated}
|
|
|
|
|
|
def resolve_ops_payment_incident(request: Request, event_id: int) -> Dict[str, Any]:
|
|
admin = _require_ops(request) or {}
|
|
db = _get_db()
|
|
resolved_group = db.mark_related_payment_audit_events_resolved(
|
|
event_id, str(admin.get("email") or "")
|
|
)
|
|
if not resolved_group:
|
|
raise HTTPException(status_code=404, detail="payment_incident_not_found")
|
|
return {
|
|
"ok": True,
|
|
"incident": resolved_group[0],
|
|
"resolved_count": len(resolved_group),
|
|
"resolved_event_ids": [int(item.get("id") or 0) for item in resolved_group],
|
|
}
|
|
|
|
|
|
def list_ops_payments(
|
|
request: Request,
|
|
limit: int = 50,
|
|
) -> Dict[str, Any]:
|
|
"""List successful payment records from Supabase."""
|
|
_require_ops(request)
|
|
safe_limit = max(1, min(int(limit or 50), 200))
|
|
rows = _supabase_rest_rows(
|
|
"payments",
|
|
{
|
|
"select": "id,user_id,amount,currency,chain,tx_hash,status,created_at",
|
|
"order": "created_at.desc",
|
|
"limit": str(safe_limit),
|
|
},
|
|
)
|
|
return {"payments": rows, "total": len(rows)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Billing risk
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
def get_ops_billing_risk(
|
|
request: Request,
|
|
days: int = 30,
|
|
limit: int = 80,
|
|
) -> Dict[str, Any]:
|
|
"""Summarize trial, payment, referral, and points risk signals for ops."""
|
|
_require_ops(request)
|
|
db = _get_db()
|
|
now = datetime.now(timezone.utc)
|
|
safe_days = max(1, min(int(days or 30), 120))
|
|
safe_limit = max(10, min(int(limit or 80), 200))
|
|
since_dt = now - timedelta(days=safe_days)
|
|
month_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
query_errors: List[Dict[str, str]] = []
|
|
|
|
def collect(table: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
try:
|
|
return _supabase_rest_rows(table, params)
|
|
except Exception as exc:
|
|
query_errors.append({"table": table, "error": str(exc)[:180]})
|
|
return []
|
|
|
|
intents = collect(
|
|
"payment_intents",
|
|
{
|
|
"select": (
|
|
"id,user_id,plan_code,chain_id,status,expires_at,tx_hash,"
|
|
"metadata,created_at,updated_at"
|
|
),
|
|
"order": "updated_at.desc",
|
|
"limit": str(max(safe_limit * 3, 100)),
|
|
},
|
|
)
|
|
referral_attributions = collect(
|
|
"referral_attributions",
|
|
{
|
|
"select": (
|
|
"id,referrer_user_id,referred_user_id,code,status,"
|
|
"converted_payment_intent_id,converted_tx_hash,converted_at,"
|
|
"created_at,updated_at"
|
|
),
|
|
"order": "updated_at.desc",
|
|
"limit": str(safe_limit),
|
|
},
|
|
)
|
|
referral_rewards = collect(
|
|
"referral_rewards",
|
|
{
|
|
"select": (
|
|
"id,referral_attribution_id,referrer_user_id,referred_user_id,"
|
|
"payment_intent_id,tx_hash,reward_days,reward_points,created_at"
|
|
),
|
|
"order": "created_at.desc",
|
|
"limit": str(safe_limit),
|
|
},
|
|
)
|
|
trial_claims = collect(
|
|
"trial_claims",
|
|
{
|
|
"select": "id,user_id,email,telegram_user_id,claimed_at,created_at",
|
|
"order": "created_at.desc",
|
|
"limit": str(max(safe_limit * 10, 500)),
|
|
},
|
|
)
|
|
trial_subscription_rows = collect(
|
|
"subscriptions",
|
|
{
|
|
"select": (
|
|
"id,user_id,plan_code,source,status,starts_at,expires_at,"
|
|
"created_at,updated_at"
|
|
),
|
|
"or": "(source.eq.signup_trial,plan_code.eq.signup_trial_3d)",
|
|
"order": "created_at.desc",
|
|
"limit": str(max(safe_limit * 20, 1000)),
|
|
},
|
|
)
|
|
active_subscription_rows = collect(
|
|
"subscriptions",
|
|
{
|
|
"select": (
|
|
"id,user_id,plan_code,source,status,starts_at,expires_at,"
|
|
"created_at,updated_at"
|
|
),
|
|
"status": "eq.active",
|
|
"order": "created_at.desc",
|
|
"limit": str(max(safe_limit * 20, 1000)),
|
|
},
|
|
)
|
|
subscription_rows: List[Dict[str, Any]] = []
|
|
seen_subscription_keys: set[str] = set()
|
|
for row in [*trial_subscription_rows, *active_subscription_rows]:
|
|
key = str(row.get("id") or "").strip() or (
|
|
f"{row.get('user_id')}:{row.get('plan_code')}:{row.get('source')}:"
|
|
f"{row.get('starts_at')}:{row.get('expires_at')}"
|
|
)
|
|
if key in seen_subscription_keys:
|
|
continue
|
|
seen_subscription_keys.add(key)
|
|
subscription_rows.append(row)
|
|
entitlement_trial_events = collect(
|
|
"entitlement_events",
|
|
{
|
|
"select": "id,user_id,action,payload,created_at",
|
|
"action": "in.(signup_trial_claimed,signup_trial_granted)",
|
|
"order": "created_at.desc",
|
|
"limit": str(max(safe_limit * 10, 500)),
|
|
},
|
|
)
|
|
|
|
issues: List[Dict[str, Any]] = []
|
|
stuck_intents: List[Dict[str, Any]] = []
|
|
points_issues: List[Dict[str, Any]] = []
|
|
|
|
for intent in intents:
|
|
status = str(intent.get("status") or "").strip().lower()
|
|
updated_at = _parse_iso_datetime(intent.get("updated_at"))
|
|
created_at = _parse_iso_datetime(intent.get("created_at"))
|
|
expires_at = _parse_iso_datetime(intent.get("expires_at"))
|
|
age_min = (
|
|
int((now - (updated_at or created_at or now)).total_seconds() // 60)
|
|
if (updated_at or created_at)
|
|
else 0
|
|
)
|
|
intent_id = str(intent.get("id") or "")
|
|
user_id = str(intent.get("user_id") or "")
|
|
metadata = intent.get("metadata")
|
|
metadata = metadata if isinstance(metadata, dict) else {}
|
|
redemption = metadata.get("points_redemption")
|
|
redemption = redemption if isinstance(redemption, dict) else {}
|
|
|
|
if status == "submitted" and age_min >= 10:
|
|
row = {
|
|
"id": intent_id,
|
|
"user_id": user_id,
|
|
"plan_code": intent.get("plan_code"),
|
|
"status": status,
|
|
"age_min": age_min,
|
|
"tx_hash": intent.get("tx_hash"),
|
|
"updated_at": intent.get("updated_at"),
|
|
}
|
|
stuck_intents.append(row)
|
|
issues.append(
|
|
_risk_issue(
|
|
category="payment_intent",
|
|
severity="high",
|
|
title="Submitted intent 超过 10 分钟未确认",
|
|
detail=f"{intent_id} 已提交 {age_min} 分钟，可能需要补单或检查确认循环。",
|
|
user_id=user_id,
|
|
created_at=intent.get("updated_at") or intent.get("created_at"),
|
|
reference=intent_id,
|
|
payload=row,
|
|
)
|
|
)
|
|
elif status == "created" and expires_at and expires_at < now:
|
|
row = {
|
|
"id": intent_id,
|
|
"user_id": user_id,
|
|
"plan_code": intent.get("plan_code"),
|
|
"status": status,
|
|
"expires_at": intent.get("expires_at"),
|
|
"age_min": age_min,
|
|
}
|
|
stuck_intents.append(row)
|
|
issues.append(
|
|
_risk_issue(
|
|
category="payment_intent",
|
|
severity="medium",
|
|
title="Created intent 已过期但未关闭",
|
|
detail=f"{intent_id} 已过期，用户可能离开支付流程。",
|
|
user_id=user_id,
|
|
created_at=intent.get("created_at"),
|
|
reference=intent_id,
|
|
payload=row,
|
|
)
|
|
)
|
|
|
|
if bool(redemption.get("applied")):
|
|
planned = int(redemption.get("points_to_consume") or 0)
|
|
consumed = bool(redemption.get("consumed"))
|
|
consumed_points = int(redemption.get("consumed_points") or 0)
|
|
if status == "confirmed" and not consumed:
|
|
row = {
|
|
"intent_id": intent_id,
|
|
"user_id": user_id,
|
|
"status": status,
|
|
"planned_points": planned,
|
|
"consumed_points": consumed_points,
|
|
"updated_at": intent.get("updated_at"),
|
|
}
|
|
points_issues.append(row)
|
|
issues.append(
|
|
_risk_issue(
|
|
category="points_redemption",
|
|
severity="high",
|
|
title="订单已确认但积分未扣减",
|
|
detail=f"{intent_id} 标记使用积分，但 confirmed metadata 未显示 consumed。",
|
|
user_id=user_id,
|
|
created_at=intent.get("updated_at") or intent.get("created_at"),
|
|
reference=intent_id,
|
|
payload=row,
|
|
)
|
|
)
|
|
elif status == "confirmed" and planned > 0 and 0 < consumed_points < planned:
|
|
row = {
|
|
"intent_id": intent_id,
|
|
"user_id": user_id,
|
|
"status": status,
|
|
"planned_points": planned,
|
|
"consumed_points": consumed_points,
|
|
"updated_at": intent.get("updated_at"),
|
|
}
|
|
points_issues.append(row)
|
|
issues.append(
|
|
_risk_issue(
|
|
category="points_redemption",
|
|
severity="medium",
|
|
title="积分抵扣只扣了部分积分",
|
|
detail=f"{intent_id} 计划扣 {planned}，实际扣 {consumed_points}。",
|
|
user_id=user_id,
|
|
created_at=intent.get("updated_at") or intent.get("created_at"),
|
|
reference=intent_id,
|
|
payload=row,
|
|
)
|
|
)
|
|
|
|
reward_by_attribution = {
|
|
str(row.get("referral_attribution_id") or ""): row
|
|
for row in referral_rewards
|
|
if row.get("referral_attribution_id") is not None
|
|
}
|
|
monthly_cap_hits: List[Dict[str, Any]] = []
|
|
referral_settlement_issues: List[Dict[str, Any]] = []
|
|
|
|
for attribution in referral_attributions:
|
|
status = str(attribution.get("status") or "").strip().lower()
|
|
attribution_id = str(attribution.get("id") or "")
|
|
updated_at = _parse_iso_datetime(
|
|
attribution.get("updated_at") or attribution.get("converted_at") or attribution.get("created_at")
|
|
)
|
|
if status == "capped" and (not updated_at or updated_at >= month_start):
|
|
row = {
|
|
"id": attribution_id,
|
|
"code": attribution.get("code"),
|
|
"referrer_user_id": attribution.get("referrer_user_id"),
|
|
"referred_user_id": attribution.get("referred_user_id"),
|
|
"updated_at": attribution.get("updated_at"),
|
|
}
|
|
monthly_cap_hits.append(row)
|
|
issues.append(
|
|
_risk_issue(
|
|
category="referral",
|
|
severity="medium",
|
|
title="邀请奖励月度上限命中",
|
|
detail=f"邀请码 {attribution.get('code') or ''} 的推荐奖励已被月度上限拦截。",
|
|
user_id=attribution.get("referrer_user_id"),
|
|
created_at=attribution.get("updated_at") or attribution.get("created_at"),
|
|
reference=attribution_id,
|
|
payload=row,
|
|
)
|
|
)
|
|
if status == "converted" and attribution_id not in reward_by_attribution:
|
|
row = {
|
|
"id": attribution_id,
|
|
"code": attribution.get("code"),
|
|
"referrer_user_id": attribution.get("referrer_user_id"),
|
|
"referred_user_id": attribution.get("referred_user_id"),
|
|
"converted_payment_intent_id": attribution.get("converted_payment_intent_id"),
|
|
"converted_at": attribution.get("converted_at"),
|
|
}
|
|
referral_settlement_issues.append(row)
|
|
issues.append(
|
|
_risk_issue(
|
|
category="referral",
|
|
severity="high",
|
|
title="推荐已转化但没有奖励记录",
|
|
detail=f"归因 {attribution_id} 已 converted，但 referral_rewards 未找到对应记录。",
|
|
user_id=attribution.get("referrer_user_id"),
|
|
created_at=attribution.get("converted_at") or attribution.get("updated_at"),
|
|
reference=attribution_id,
|
|
payload=row,
|
|
)
|
|
)
|
|
|
|
events = db.list_app_analytics_events(limit=20000, since_iso=since_dt.isoformat())
|
|
signup_rows = [
|
|
row
|
|
for row in events
|
|
if str(row.get("event_type") or "").strip().lower()
|
|
in {"signup_success", "signup_completed"}
|
|
]
|
|
|
|
def normalize_user_key(value: Any) -> str:
|
|
return str(value or "").strip().lower()
|
|
|
|
def analytics_payload(row: Dict[str, Any]) -> Dict[str, Any]:
|
|
payload = row.get("payload")
|
|
return payload if isinstance(payload, dict) else {}
|
|
|
|
def analytics_correlation_keys(row: Dict[str, Any]) -> set[str]:
|
|
payload = analytics_payload(row)
|
|
keys: set[str] = set()
|
|
user_id = normalize_user_key(row.get("user_id") or payload.get("user_id"))
|
|
client_id = str(row.get("client_id") or "").strip()
|
|
session_id = str(row.get("session_id") or "").strip()
|
|
if user_id:
|
|
keys.add(f"user:{user_id}")
|
|
if client_id:
|
|
keys.add(f"client:{client_id}")
|
|
if session_id:
|
|
keys.add(f"session:{session_id}")
|
|
return keys
|
|
|
|
signup_intent_keys: set[str] = set()
|
|
for row in events:
|
|
if str(row.get("event_type") or "").strip().lower() != "login_start":
|
|
continue
|
|
payload = analytics_payload(row)
|
|
mode = str(payload.get("mode") or payload.get("auth_mode") or "").strip().lower()
|
|
if mode == "signup":
|
|
signup_intent_keys.update(analytics_correlation_keys(row))
|
|
|
|
def has_signup_intent(row: Dict[str, Any]) -> bool:
|
|
payload = analytics_payload(row)
|
|
mode = str(payload.get("mode") or payload.get("auth_mode") or "").strip().lower()
|
|
if mode == "signup" or payload.get("signup_intent") is True:
|
|
return True
|
|
return bool(analytics_correlation_keys(row).intersection(signup_intent_keys))
|
|
|
|
trial_actor_keys = {
|
|
_app_analytics_actor_key(row)
|
|
for row in events
|
|
if str(row.get("event_type") or "").strip().lower() == "trial_created"
|
|
}
|
|
subscription_user_keys = {
|
|
normalize_user_key(row.get("user_id"))
|
|
for row in subscription_rows
|
|
if normalize_user_key(row.get("user_id"))
|
|
}
|
|
trial_subscription_user_keys = {
|
|
normalize_user_key(row.get("user_id"))
|
|
for row in subscription_rows
|
|
if normalize_user_key(row.get("user_id"))
|
|
and (
|
|
str(row.get("plan_code") or "").strip().lower() == "signup_trial_3d"
|
|
or str(row.get("source") or "").strip().lower() == "signup_trial"
|
|
)
|
|
}
|
|
trial_claim_user_keys = {
|
|
normalize_user_key(row.get("user_id"))
|
|
for row in trial_claims
|
|
if normalize_user_key(row.get("user_id"))
|
|
}
|
|
trial_event_user_keys: set[str] = set()
|
|
for row in entitlement_trial_events:
|
|
event_user_id = normalize_user_key(row.get("user_id"))
|
|
payload = row.get("payload") if isinstance(row.get("payload"), dict) else {}
|
|
payload_user_id = normalize_user_key(payload.get("user_id"))
|
|
if event_user_id:
|
|
trial_event_user_keys.add(event_user_id)
|
|
if payload_user_id:
|
|
trial_event_user_keys.add(payload_user_id)
|
|
|
|
backend_trial_user_keys = (
|
|
trial_subscription_user_keys | trial_claim_user_keys | trial_event_user_keys
|
|
)
|
|
trial_gaps: List[Dict[str, Any]] = []
|
|
|
|
for claim in trial_claims:
|
|
claim_user_id = normalize_user_key(claim.get("user_id"))
|
|
if not claim_user_id or claim_user_id in trial_subscription_user_keys:
|
|
continue
|
|
gap = {
|
|
"claim_id": claim.get("id"),
|
|
"user_id": claim.get("user_id"),
|
|
"email": claim.get("email"),
|
|
"created_at": claim.get("created_at") or claim.get("claimed_at"),
|
|
"reason": "trial_claim_without_subscription",
|
|
}
|
|
trial_gaps.append(gap)
|
|
if len(trial_gaps) <= 20:
|
|
issues.append(
|
|
_risk_issue(
|
|
category="signup_trial",
|
|
severity="high",
|
|
title="试用 claim 已写入但订阅缺失",
|
|
detail="trial_claims 已记录该用户领取试用，但 subscriptions 中没有 signup_trial_3d 记录。",
|
|
user_id=gap.get("user_id"),
|
|
created_at=gap.get("created_at"),
|
|
reference=str(gap.get("claim_id") or ""),
|
|
payload=gap,
|
|
)
|
|
)
|
|
|
|
for row in signup_rows[:300]:
|
|
if not has_signup_intent(row):
|
|
continue
|
|
actor_key = _app_analytics_actor_key(row)
|
|
if actor_key in trial_actor_keys:
|
|
continue
|
|
payload = analytics_payload(row)
|
|
signup_user_id = normalize_user_key(row.get("user_id") or payload.get("user_id"))
|
|
if not signup_user_id:
|
|
continue
|
|
if (
|
|
signup_user_id in backend_trial_user_keys
|
|
or signup_user_id in subscription_user_keys
|
|
):
|
|
continue
|
|
gap = {
|
|
"event_id": row.get("id"),
|
|
"actor_key": actor_key,
|
|
"user_id": signup_user_id,
|
|
"created_at": row.get("created_at"),
|
|
"reason": "signup_without_backend_trial_evidence",
|
|
}
|
|
trial_gaps.append(gap)
|
|
if len(trial_gaps) <= 20:
|
|
issues.append(
|
|
_risk_issue(
|
|
category="signup_trial",
|
|
severity="high",
|
|
title="注册成功后未发现后端试用记录",
|
|
detail=(
|
|
"该用户进入 signup_success，但没有 trial_created、trial_claims、"
|
|
"signup_trial subscription 或其他有效订阅证据。"
|
|
),
|
|
user_id=gap.get("user_id"),
|
|
created_at=gap.get("created_at"),
|
|
reference=str(gap.get("event_id") or ""),
|
|
payload=gap,
|
|
)
|
|
)
|
|
|
|
payment_incidents_raw = db.list_payment_audit_events(
|
|
limit=max(safe_limit, 500),
|
|
event_type="payment_intent_failed",
|
|
)
|
|
grouped_payment_incidents = _group_payment_incidents(payment_incidents_raw)
|
|
unresolved_incidents = grouped_payment_incidents["incidents"]
|
|
|
|
issues.sort(key=lambda item: str(item.get("created_at") or ""), reverse=True)
|
|
recent_rewards = [
|
|
{
|
|
"id": row.get("id"),
|
|
"referral_attribution_id": row.get("referral_attribution_id"),
|
|
"referrer_user_id": row.get("referrer_user_id"),
|
|
"referred_user_id": row.get("referred_user_id"),
|
|
"payment_intent_id": row.get("payment_intent_id"),
|
|
"reward_points": int(row.get("reward_points") or 0),
|
|
"reward_days": int(row.get("reward_days") or 0),
|
|
"tx_hash": row.get("tx_hash"),
|
|
"explorer_url": _payment_explorer_url("polygon", row.get("tx_hash")),
|
|
"created_at": row.get("created_at"),
|
|
}
|
|
for row in referral_rewards[:safe_limit]
|
|
]
|
|
|
|
return {
|
|
"checked_at": _to_utc_iso(now),
|
|
"window_days": safe_days,
|
|
"summary": {
|
|
"issues": len(issues),
|
|
"stuck_intents": len(stuck_intents),
|
|
"trial_gaps": len(trial_gaps),
|
|
"payment_incidents": grouped_payment_incidents["total"],
|
|
"payment_incident_events": grouped_payment_incidents["raw_total"],
|
|
"points_discount_issues": len(points_issues),
|
|
"referral_settlement_issues": len(referral_settlement_issues),
|
|
"monthly_cap_hits": len(monthly_cap_hits),
|
|
"recent_referral_rewards": len(recent_rewards),
|
|
"recent_trial_claims": len(trial_claims),
|
|
},
|
|
"issues": issues[:safe_limit],
|
|
"stuck_intents": stuck_intents[:safe_limit],
|
|
"trial_gaps": trial_gaps[:safe_limit],
|
|
"payment_incidents": unresolved_incidents[:safe_limit],
|
|
"points_discount_issues": points_issues[:safe_limit],
|
|
"referral_settlement_issues": referral_settlement_issues[:safe_limit],
|
|
"monthly_cap_hits": monthly_cap_hits[:safe_limit],
|
|
"recent_referral_rewards": recent_rewards,
|
|
"recent_trial_claims": trial_claims,
|
|
"query_errors": query_errors,
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Membership helpers
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
def _list_active_subscriptions_with_windows(
|
|
limit: int,
|
|
) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]], bool]:
|
|
active_window_query = getattr(
|
|
legacy_routes.SUPABASE_ENTITLEMENT,
|
|
"list_active_subscription_windows",
|
|
None,
|
|
)
|
|
if not callable(active_window_query):
|
|
return [], {}, False
|
|
try:
|
|
active_window_payload = active_window_query(limit=limit)
|
|
except Exception:
|
|
return [], {}, False
|
|
if not isinstance(active_window_payload, dict):
|
|
return [], {}, False
|
|
|
|
active_window_subscriptions = active_window_payload.get("subscriptions")
|
|
active_window_windows = active_window_payload.get("windows")
|
|
if not isinstance(active_window_subscriptions, list):
|
|
return [], {}, False
|
|
if not active_window_subscriptions and not (
|
|
isinstance(active_window_windows, dict) and active_window_windows
|
|
):
|
|
return [], {}, False
|
|
|
|
subscriptions = [
|
|
item for item in active_window_subscriptions if isinstance(item, dict)
|
|
]
|
|
subscription_windows = (
|
|
active_window_windows if isinstance(active_window_windows, dict) else {}
|
|
)
|
|
return subscriptions, subscription_windows, True
|
|
|
|
|
|
def _build_membership_rows(
|
|
db: DBManager,
|
|
subscriptions: list[dict[str, Any]],
|
|
subscription_windows: dict[str, dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
subscription_user_ids = [str(item.get("user_id") or "") for item in subscriptions]
|
|
user_map = db.get_users_by_supabase_user_ids(subscription_user_ids)
|
|
unresolved_user_ids = [
|
|
user_id
|
|
for user_id in subscription_user_ids
|
|
if str(user_id or "").strip().lower()
|
|
and not str(
|
|
(user_map.get(str(user_id).strip().lower(), {}) or {}).get("supabase_email")
|
|
or ""
|
|
).strip()
|
|
]
|
|
auth_user_map = legacy_routes.SUPABASE_ENTITLEMENT.get_auth_users(
|
|
unresolved_user_ids
|
|
)
|
|
if not subscription_windows:
|
|
subscription_windows = legacy_routes.SUPABASE_ENTITLEMENT.list_subscription_windows(
|
|
subscription_user_ids,
|
|
bypass_cache=True,
|
|
)
|
|
deduped: dict[str, dict] = {}
|
|
for item in subscriptions:
|
|
user_id = str(item.get("user_id") or "").strip().lower()
|
|
local_user = user_map.get(user_id, {})
|
|
auth_user = auth_user_map.get(user_id, {})
|
|
subscription_window = subscription_windows.get(user_id, {})
|
|
current_expires_at = item.get("expires_at")
|
|
total_expires_at = (
|
|
subscription_window.get("total_expires_at")
|
|
if isinstance(subscription_window, dict)
|
|
else None
|
|
)
|
|
queued_days = (
|
|
int(subscription_window.get("queued_days") or 0)
|
|
if isinstance(subscription_window, dict)
|
|
else 0
|
|
)
|
|
queued_count = (
|
|
int(subscription_window.get("queued_count") or 0)
|
|
if isinstance(subscription_window, dict)
|
|
else 0
|
|
)
|
|
source = str(item.get("source") or "")
|
|
is_trial = source == "signup_trial" or str(
|
|
item.get("plan_code") or ""
|
|
).startswith("signup_trial")
|
|
row = {
|
|
"user_id": user_id,
|
|
"email": str(
|
|
auth_user.get("email") or local_user.get("supabase_email") or ""
|
|
),
|
|
"telegram_id": local_user.get("telegram_id"),
|
|
"username": local_user.get("username"),
|
|
"registered_at": local_user.get("created_at")
|
|
or auth_user.get("created_at"),
|
|
"plan_code": item.get("plan_code"),
|
|
"source": source,
|
|
"is_trial": is_trial,
|
|
"starts_at": item.get("starts_at"),
|
|
"current_expires_at": current_expires_at,
|
|
"total_expires_at": total_expires_at or current_expires_at,
|
|
"expires_at": total_expires_at or current_expires_at,
|
|
"queued_days": queued_days,
|
|
"queued_count": queued_count,
|
|
}
|
|
existing = deduped.get(user_id)
|
|
existing_expires = str(existing.get("expires_at") or "") if existing else ""
|
|
current_expires = str(row.get("expires_at") or "")
|
|
if existing is None or current_expires > existing_expires:
|
|
deduped[user_id] = row
|
|
return sorted(
|
|
deduped.values(),
|
|
key=lambda item: str(item.get("expires_at") or ""),
|
|
)
|
|
|
|
|
|
def _build_membership_growth(
|
|
subscriptions: list[dict[str, Any]],
|
|
days: int,
|
|
) -> dict[str, Any]:
|
|
safe_days = max(7, min(365, int(days or 90)))
|
|
now = datetime.utcnow()
|
|
cutoff = now - timedelta(days=safe_days)
|
|
|
|
trial_by_day: dict[str, int] = defaultdict(int)
|
|
paid_by_day: dict[str, int] = defaultdict(int)
|
|
running = 0
|
|
|
|
for item in subscriptions:
|
|
starts_raw = str(item.get("starts_at") or "").strip()
|
|
if not starts_raw:
|
|
continue
|
|
try:
|
|
dt = datetime.fromisoformat(starts_raw.replace("Z", "+00:00"))
|
|
if dt.tzinfo is not None:
|
|
dt = dt.replace(tzinfo=None)
|
|
except Exception:
|
|
continue
|
|
if dt < cutoff:
|
|
continue
|
|
day_key = dt.strftime("%Y-%m-%d")
|
|
source = str(item.get("source") or "").strip().lower()
|
|
plan = str(item.get("plan_code") or "").strip().lower()
|
|
is_trial = source == "signup_trial" or plan.startswith("signup_trial")
|
|
if is_trial:
|
|
trial_by_day[day_key] += 1
|
|
else:
|
|
paid_by_day[day_key] += 1
|
|
|
|
daily = []
|
|
cursor = cutoff.date()
|
|
while cursor <= now.date():
|
|
key = cursor.isoformat()
|
|
tc = trial_by_day.get(key, 0)
|
|
pc = paid_by_day.get(key, 0)
|
|
total = tc + pc
|
|
running += total
|
|
daily.append(
|
|
{
|
|
"date": key,
|
|
"trial": tc,
|
|
"paid": pc,
|
|
"total": total,
|
|
"cumulative": running,
|
|
}
|
|
)
|
|
cursor += timedelta(days=1)
|
|
|
|
return {"days": safe_days, "daily": daily}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Membership API
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
def list_ops_memberships(request: Request, limit: int = 200) -> Dict[str, Any]:
|
|
_require_ops(request)
|
|
db = _get_db()
|
|
reconcile_enabled = (
|
|
str(os.getenv("POLYWEATHER_OPS_MEMBERSHIPS_RECONCILE_ENABLED") or "")
|
|
.strip()
|
|
.lower()
|
|
in {"1", "true", "yes", "on"}
|
|
)
|
|
if reconcile_enabled and getattr(legacy_routes.PAYMENT_CHECKOUT, "enabled", False):
|
|
try:
|
|
legacy_routes.PAYMENT_CHECKOUT.reconcile_recent_intents(
|
|
limit=min(max(int(limit or 200), 20), 200)
|
|
)
|
|
except Exception:
|
|
pass
|
|
subscriptions, subscription_windows, used_active_window_query = (
|
|
_list_active_subscriptions_with_windows(limit)
|
|
)
|
|
if not used_active_window_query:
|
|
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(
|
|
limit=limit
|
|
)
|
|
return {
|
|
"memberships": _build_membership_rows(
|
|
db,
|
|
subscriptions,
|
|
subscription_windows,
|
|
)
|
|
}
|
|
|
|
|
|
def get_ops_memberships_growth(request: Request, days: int = 90) -> dict[str, Any]:
|
|
_require_ops(request)
|
|
|
|
subscriptions, _, used_active_window_query = _list_active_subscriptions_with_windows(
|
|
limit=5000
|
|
)
|
|
if not used_active_window_query:
|
|
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(
|
|
limit=5000
|
|
)
|
|
return _build_membership_growth(subscriptions, days)
|
|
|
|
|
|
def get_ops_memberships_overview(
|
|
request: Request,
|
|
limit: int = 200,
|
|
days: int = 90,
|
|
) -> dict[str, Any]:
|
|
_require_ops(request)
|
|
db = _get_db()
|
|
safe_limit = max(1, min(int(limit or 200), 1000))
|
|
query_limit = max(safe_limit, 5000)
|
|
subscriptions, subscription_windows, used_active_window_query = (
|
|
_list_active_subscriptions_with_windows(limit=query_limit)
|
|
)
|
|
if not used_active_window_query:
|
|
subscriptions = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(
|
|
limit=query_limit
|
|
)
|
|
membership_subscriptions = subscriptions[:safe_limit]
|
|
return {
|
|
"memberships": _build_membership_rows(
|
|
db,
|
|
membership_subscriptions,
|
|
subscription_windows,
|
|
),
|
|
**_build_membership_growth(subscriptions, days),
|
|
}
|