2569718930@qq.com
765 lines
25 KiB
Python
765 lines
25 KiB
Python
"""Ops config / subscriptions / logs / telegram service functions."""
|
||
from __future__ import annotations
|
||
|
||
import concurrent.futures
|
||
import os
|
||
import sqlite3
|
||
import subprocess
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import requests as _requests
|
||
from fastapi import HTTPException, Request
|
||
|
||
from src.utils.runtime_secrets import get_runtime_secret_status
|
||
|
||
|
||
def _get_db():
|
||
from src.database.db_manager import DBManager as _DBManager
|
||
return _DBManager()
|
||
|
||
|
||
# ── Config key definitions ──────────────────────────────────────────
|
||
|
||
_EDITABLE_CONFIG_KEYS: dict[str, str] = {
|
||
"POLYWEATHER_AUTH_REQUIRED": "是否强制要求 Supabase 登录访问 API",
|
||
"POLYWEATHER_PAYMENT_ENABLED": "是否启用支付功能",
|
||
"POLYWEATHER_PAYMENT_POINTS_ENABLED": "是否启用积分抵扣",
|
||
"POLYWEATHER_TELEGRAM_ALERT_PUSH_ENABLED": "是否启用 Telegram 告警推送",
|
||
"POLYWEATHER_GROUP_MEMBER_PRICE_USDC": "群成员月费 (USDC)",
|
||
"POLYWEATHER_PUBLIC_PRICE_USDC": "公开月费 (USDC)",
|
||
"POLYWEATHER_PAYMENT_POINTS_PER_USDC": "积分兑换汇率 (积分/USDC)",
|
||
"POLYWEATHER_PAYMENT_POINTS_MAX_DISCOUNT_USDC": "积分最高抵扣金额 (USDC)",
|
||
"POLYWEATHER_PAYMENT_DIRECT_RECEIVER_ADDRESS": "手动转账收款钱包地址",
|
||
}
|
||
|
||
_SENSITIVE_CONFIG_KEYS: dict[str, dict[str, str]] = {
|
||
"POLYWEATHER_AMSC_SESSION_ID": {
|
||
"label": "AMSC AWOS sessionId",
|
||
"description": "中国跑道观测接口 sessionId，用于上海/北京/广州等 AMSC AWOS 数据源。",
|
||
},
|
||
}
|
||
|
||
|
||
# ── 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]]:
|
||
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
|
||
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
|
||
if not supabase_url or not service_role_key:
|
||
raise HTTPException(status_code=503, detail="Supabase not configured")
|
||
|
||
headers = {
|
||
"apikey": service_role_key,
|
||
"Authorization": f"Bearer {service_role_key}",
|
||
}
|
||
resp = _requests.get(
|
||
f"{supabase_url}/rest/v1/{table}",
|
||
headers=headers,
|
||
params=params,
|
||
timeout=timeout,
|
||
)
|
||
if not resp.ok:
|
||
raise HTTPException(
|
||
status_code=502,
|
||
detail=f"Supabase query failed for {table}: {resp.status_code}",
|
||
)
|
||
rows = resp.json() if resp.content else []
|
||
if not isinstance(rows, list):
|
||
return []
|
||
return [row for row in rows if isinstance(row, dict)]
|
||
|
||
|
||
def _supabase_service_headers(
|
||
service_role_key: str,
|
||
*,
|
||
prefer: str | None = None,
|
||
) -> dict[str, str]:
|
||
headers = {
|
||
"apikey": service_role_key,
|
||
"Authorization": f"Bearer {service_role_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
if prefer:
|
||
headers["Prefer"] = prefer
|
||
return headers
|
||
|
||
|
||
def _lookup_supabase_user_id_by_email(
|
||
supabase_url: str,
|
||
service_role_key: str,
|
||
email: str,
|
||
) -> str:
|
||
normalized_email = str(email or "").strip().lower()
|
||
if not normalized_email:
|
||
return ""
|
||
base = str(supabase_url or "").strip().rstrip("/")
|
||
headers = _supabase_service_headers(service_role_key)
|
||
|
||
profile_resp = _requests.get(
|
||
f"{base}/rest/v1/profiles",
|
||
headers=headers,
|
||
params={
|
||
"select": "id",
|
||
"email": f"eq.{normalized_email}",
|
||
"limit": "1",
|
||
},
|
||
timeout=10,
|
||
)
|
||
if profile_resp.ok:
|
||
profiles = profile_resp.json() if profile_resp.content else []
|
||
if isinstance(profiles, list) and profiles:
|
||
user_id = str((profiles[0] or {}).get("id") or "").strip()
|
||
if user_id:
|
||
return user_id
|
||
|
||
user_resp = _requests.get(
|
||
f"{base}/auth/v1/admin/users",
|
||
headers=headers,
|
||
params={"filter": f"email.eq.{normalized_email}"},
|
||
timeout=10,
|
||
)
|
||
users = user_resp.json().get("users", []) if user_resp.ok else []
|
||
return str(users[0].get("id") or "").strip() if users else ""
|
||
|
||
|
||
# ── Config ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def get_ops_config(request: Request) -> dict[str, Any]:
|
||
_require_ops(request)
|
||
|
||
configs: list[dict[str, Any]] = []
|
||
for key, desc in _EDITABLE_CONFIG_KEYS.items():
|
||
configs.append(
|
||
{
|
||
"key": key,
|
||
"value": os.getenv(key) or "",
|
||
"description": desc,
|
||
}
|
||
)
|
||
return {"configs": configs}
|
||
|
||
|
||
def update_ops_config(request: Request, key: str, value: str) -> dict[str, Any]:
|
||
_require_ops(request)
|
||
|
||
normalized_key = str(key or "").strip()
|
||
if normalized_key not in _EDITABLE_CONFIG_KEYS:
|
||
raise HTTPException(
|
||
status_code=400, detail=f"config key '{normalized_key}' is not editable"
|
||
)
|
||
os.environ[normalized_key] = str(value)
|
||
return {
|
||
"key": normalized_key,
|
||
"value": value,
|
||
"ok": True,
|
||
}
|
||
|
||
|
||
def _sensitive_config_payload(key: str) -> dict[str, Any]:
|
||
definition = _SENSITIVE_CONFIG_KEYS.get(key) or {}
|
||
metadata = get_runtime_secret_status(key)
|
||
return {
|
||
"key": key,
|
||
"label": definition.get("label") or key,
|
||
"description": definition.get("description") or "",
|
||
"configured": bool(metadata.get("configured")),
|
||
"masked": str(metadata.get("masked") or ""),
|
||
"length": int(metadata.get("length") or 0),
|
||
"updated_at": str(metadata.get("updated_at") or ""),
|
||
"updated_by": str(metadata.get("updated_by") or ""),
|
||
"source": str(metadata.get("source") or "runtime_store"),
|
||
}
|
||
|
||
|
||
def get_ops_sensitive_config(request: Request) -> dict[str, Any]:
|
||
_require_ops(request)
|
||
return {
|
||
"configs": [
|
||
_sensitive_config_payload(key)
|
||
for key in _SENSITIVE_CONFIG_KEYS
|
||
]
|
||
}
|
||
|
||
|
||
def update_ops_sensitive_config(
|
||
request: Request,
|
||
key: str,
|
||
value: str,
|
||
) -> dict[str, Any]:
|
||
admin = _require_ops(request) or {}
|
||
normalized_key = str(key or "").strip()
|
||
if normalized_key not in _SENSITIVE_CONFIG_KEYS:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"sensitive config key '{normalized_key}' is not editable",
|
||
)
|
||
secret_value = str(value or "").strip()
|
||
if not 12 <= len(secret_value) <= 256 or any(ch.isspace() for ch in secret_value):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="sessionId must be 12-256 non-whitespace characters",
|
||
)
|
||
|
||
db = _get_db()
|
||
try:
|
||
config = db.set_runtime_secret(
|
||
normalized_key,
|
||
secret_value,
|
||
updated_by=str(admin.get("email") or ""),
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
|
||
os.environ[normalized_key] = secret_value
|
||
response_config = _sensitive_config_payload(normalized_key)
|
||
response_config.update(
|
||
{
|
||
"configured": bool(config.get("configured")),
|
||
"masked": str(config.get("masked") or ""),
|
||
"length": int(config.get("length") or 0),
|
||
"updated_at": str(config.get("updated_at") or ""),
|
||
"updated_by": str(config.get("updated_by") or ""),
|
||
"source": str(config.get("source") or "runtime_store"),
|
||
}
|
||
)
|
||
# Lazy import to avoid circular dependency with ops_api
|
||
from web.services.ops_api import _check_amsc_awos_health
|
||
health = (
|
||
_check_amsc_awos_health(timeout=8)
|
||
if normalized_key == "POLYWEATHER_AMSC_SESSION_ID"
|
||
else None
|
||
)
|
||
return {"ok": True, "config": response_config, "health": health}
|
||
|
||
|
||
# ── Subscriptions ───────────────────────────────────────────────────
|
||
|
||
|
||
def grant_ops_subscription(
|
||
request: Request,
|
||
email: str,
|
||
plan_code: str = "pro_monthly",
|
||
days: int = 30,
|
||
deduct_points: int = 0,
|
||
) -> dict[str, Any]:
|
||
admin = _require_ops(request) or {}
|
||
actor_email = str(admin.get("email") or "").strip().lower()
|
||
from datetime import datetime
|
||
|
||
import web.routes as legacy_routes # lazy – avoid circular import
|
||
|
||
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
|
||
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
|
||
if not supabase_url or not service_role_key:
|
||
raise HTTPException(status_code=503, detail="Supabase not configured")
|
||
|
||
allowed_plans = {"pro_monthly"}
|
||
if plan_code not in allowed_plans:
|
||
raise HTTPException(
|
||
status_code=400, detail=f"invalid plan_code, allowed: {allowed_plans}"
|
||
)
|
||
|
||
safe_days = max(1, min(365, int(days or 30)))
|
||
safe_deduct = max(0, int(deduct_points or 0))
|
||
normalized_email = str(email or "").strip().lower()
|
||
if not normalized_email:
|
||
raise HTTPException(status_code=400, detail="email is required")
|
||
|
||
user_id = _lookup_supabase_user_id_by_email(
|
||
supabase_url,
|
||
service_role_key,
|
||
normalized_email,
|
||
)
|
||
if not user_id:
|
||
raise HTTPException(
|
||
status_code=404, detail=f"user not found: {normalized_email}"
|
||
)
|
||
|
||
now = datetime.utcnow()
|
||
starts_at = now.isoformat() + "Z"
|
||
expires_at = (now + timedelta(days=safe_days)).isoformat() + "Z"
|
||
|
||
payload = {
|
||
"user_id": user_id,
|
||
"email": normalized_email,
|
||
"plan_code": plan_code,
|
||
"starts_at": starts_at,
|
||
"expires_at": expires_at,
|
||
"source": "ops_manual_grant",
|
||
"created_at": now.isoformat() + "Z",
|
||
}
|
||
|
||
resp = _requests.post(
|
||
f"{supabase_url}/rest/v1/subscriptions",
|
||
headers=_supabase_service_headers(service_role_key, prefer="return=minimal"),
|
||
json=payload,
|
||
timeout=10,
|
||
)
|
||
if not resp.ok:
|
||
raise HTTPException(
|
||
status_code=500, detail=f"Supabase insert failed: {resp.text[:200]}"
|
||
)
|
||
legacy_routes.SUPABASE_ENTITLEMENT.invalidate_subscription_cache(user_id)
|
||
|
||
result: dict[str, Any] = {
|
||
"ok": True,
|
||
"user_id": user_id,
|
||
"plan_code": plan_code,
|
||
"days": safe_days,
|
||
"expires_at": expires_at,
|
||
}
|
||
|
||
# Optionally deduct points from the user (manual Pro grant with points payment)
|
||
if safe_deduct > 0:
|
||
db = _get_db()
|
||
deduct_result = db.deduct_points_by_supabase_email(
|
||
normalized_email,
|
||
safe_deduct,
|
||
source="ops_subscription_deduction",
|
||
actor_email=actor_email,
|
||
reference_type="subscription",
|
||
reference_id=user_id,
|
||
metadata={"plan_code": plan_code, "days": safe_days},
|
||
)
|
||
result["points_deducted"] = safe_deduct
|
||
result["points_result"] = deduct_result
|
||
|
||
try:
|
||
_get_db().append_ops_audit_event(
|
||
action="subscription_manual_grant",
|
||
actor_email=actor_email,
|
||
target_user_id=user_id,
|
||
target_email=normalized_email,
|
||
target_type="subscription",
|
||
payload={
|
||
"plan_code": plan_code,
|
||
"days": safe_days,
|
||
"expires_at": expires_at,
|
||
"deduct_points": safe_deduct,
|
||
},
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
return result
|
||
|
||
|
||
def extend_ops_subscription(
|
||
request: Request,
|
||
email: str,
|
||
additional_days: int = 30,
|
||
) -> dict[str, Any]:
|
||
admin = _require_ops(request) or {}
|
||
actor_email = str(admin.get("email") or "").strip().lower()
|
||
from datetime import datetime
|
||
|
||
import web.routes as legacy_routes # lazy – avoid circular import
|
||
|
||
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
|
||
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
|
||
if not supabase_url or not service_role_key:
|
||
raise HTTPException(status_code=503, detail="Supabase not configured")
|
||
|
||
safe_days = max(1, min(365, int(additional_days or 30)))
|
||
normalized_email = str(email or "").strip().lower()
|
||
if not normalized_email:
|
||
raise HTTPException(status_code=400, detail="email is required")
|
||
|
||
headers = _supabase_service_headers(service_role_key)
|
||
|
||
user_id = _lookup_supabase_user_id_by_email(
|
||
supabase_url,
|
||
service_role_key,
|
||
normalized_email,
|
||
)
|
||
if not user_id:
|
||
raise HTTPException(
|
||
status_code=404, detail=f"user not found: {normalized_email}"
|
||
)
|
||
|
||
# Find latest active subscription
|
||
subs_resp = _requests.get(
|
||
f"{supabase_url}/rest/v1/subscriptions",
|
||
headers=headers,
|
||
params={
|
||
"select": "id,expires_at",
|
||
"user_id": f"eq.{user_id}",
|
||
"status": "eq.active",
|
||
"order": "expires_at.desc",
|
||
"limit": "1",
|
||
},
|
||
timeout=10,
|
||
)
|
||
subs = subs_resp.json() if subs_resp.ok else []
|
||
if not subs:
|
||
raise HTTPException(
|
||
status_code=404, detail=f"no subscription found for {normalized_email}"
|
||
)
|
||
|
||
sub = subs[0]
|
||
current_expiry = sub.get("expires_at", "")
|
||
try:
|
||
dt = datetime.fromisoformat(current_expiry.replace("Z", "+00:00"))
|
||
new_expiry = (dt + timedelta(days=safe_days)).isoformat()
|
||
except Exception:
|
||
new_expiry = (datetime.utcnow() + timedelta(days=safe_days)).isoformat() + "Z"
|
||
|
||
patch_resp = _requests.patch(
|
||
f"{supabase_url}/rest/v1/subscriptions?id=eq.{sub['id']}",
|
||
headers=_supabase_service_headers(service_role_key, prefer="return=minimal"),
|
||
json={"expires_at": new_expiry},
|
||
timeout=10,
|
||
)
|
||
if patch_resp.ok:
|
||
legacy_routes.SUPABASE_ENTITLEMENT.invalidate_subscription_cache(user_id)
|
||
try:
|
||
_get_db().append_ops_audit_event(
|
||
action="subscription_manual_extend",
|
||
actor_email=actor_email,
|
||
target_user_id=user_id,
|
||
target_email=normalized_email,
|
||
target_type="subscription",
|
||
target_id=str(sub.get("id") or ""),
|
||
payload={
|
||
"additional_days": safe_days,
|
||
"previous_expires_at": current_expiry,
|
||
"new_expires_at": new_expiry,
|
||
},
|
||
)
|
||
except Exception:
|
||
pass
|
||
return {
|
||
"ok": True,
|
||
"email": normalized_email,
|
||
"additional_days": safe_days,
|
||
"new_expires_at": new_expiry,
|
||
}
|
||
raise HTTPException(
|
||
status_code=500, detail=f"Supabase update failed: {patch_resp.text[:200]}"
|
||
)
|
||
|
||
|
||
def get_ops_user_subscriptions(
|
||
request: Request,
|
||
email: str,
|
||
) -> dict[str, Any]:
|
||
"""Return ALL subscription rows for a user (by email), regardless of status."""
|
||
_require_ops(request)
|
||
|
||
supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
|
||
service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
|
||
if not supabase_url or not service_role_key:
|
||
raise HTTPException(status_code=503, detail="Supabase not configured")
|
||
|
||
normalized_email = str(email or "").strip().lower()
|
||
if not normalized_email:
|
||
raise HTTPException(status_code=400, detail="email is required")
|
||
|
||
headers = _supabase_service_headers(service_role_key)
|
||
|
||
user_id = _lookup_supabase_user_id_by_email(
|
||
supabase_url,
|
||
service_role_key,
|
||
normalized_email,
|
||
)
|
||
if not user_id:
|
||
raise HTTPException(
|
||
status_code=404, detail=f"user not found: {normalized_email}"
|
||
)
|
||
|
||
# Fetch all subscription rows for this user (no status filter)
|
||
subs_resp = _requests.get(
|
||
f"{supabase_url}/rest/v1/subscriptions",
|
||
headers=headers,
|
||
params={
|
||
"select": "id,user_id,status,plan_code,source,starts_at,expires_at,created_at,updated_at",
|
||
"user_id": f"eq.{user_id}",
|
||
"order": "created_at.desc",
|
||
"limit": "50",
|
||
},
|
||
timeout=10,
|
||
)
|
||
rows = subs_resp.json() if subs_resp.ok and subs_resp.content else []
|
||
if not isinstance(rows, list):
|
||
rows = []
|
||
|
||
return {
|
||
"email": normalized_email,
|
||
"user_id": user_id,
|
||
"subscriptions": rows,
|
||
"count": len(rows),
|
||
}
|
||
|
||
|
||
# ── Logs ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def get_ops_logs(
|
||
request: Request,
|
||
level: str = "",
|
||
lines: int = 100,
|
||
) -> dict[str, Any]:
|
||
_require_ops(request)
|
||
|
||
safe_lines = max(10, min(1000, int(lines or 100)))
|
||
log_text = ""
|
||
try:
|
||
# Read from Docker logs
|
||
result = subprocess.run(
|
||
["docker", "logs", "--tail", str(safe_lines), "polyweather_bot"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=10,
|
||
)
|
||
log_text = result.stdout or result.stderr or ""
|
||
except Exception:
|
||
pass
|
||
|
||
# Fallback to local log file if docker logs returns empty
|
||
if not log_text.strip():
|
||
log_file = "data/logs/polyweather.log"
|
||
if os.path.exists(log_file):
|
||
try:
|
||
with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
|
||
all_lines = f.readlines()
|
||
log_text = "".join(all_lines[-safe_lines:])
|
||
except Exception:
|
||
pass
|
||
|
||
log_lines = log_text.strip().split("\n") if log_text.strip() else []
|
||
|
||
if level:
|
||
level_upper = level.upper()
|
||
log_lines = [line for line in log_lines if level_upper in line.upper()]
|
||
|
||
return {
|
||
"lines": log_lines[-safe_lines:],
|
||
"total": len(log_lines),
|
||
}
|
||
|
||
|
||
# ── Telegram audit ──────────────────────────────────────────────────
|
||
|
||
|
||
def get_ops_telegram_audit(request: Request) -> Dict[str, Any]:
|
||
_require_ops(request)
|
||
|
||
|
||
from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env
|
||
# Lazy imports to avoid circular dependency with ops_api
|
||
import web.routes as legacy_routes
|
||
from web.services.ops.payments import _list_active_subscriptions_with_windows
|
||
|
||
db = _get_db()
|
||
|
||
# 1. Fetch all distinct telegram users from database
|
||
with db._get_connection() as conn:
|
||
conn.row_factory = sqlite3.Row
|
||
users_rows = conn.execute("SELECT telegram_id, username FROM users").fetchall()
|
||
bindings_rows = conn.execute(
|
||
"SELECT telegram_id, supabase_user_id, supabase_email FROM supabase_bindings"
|
||
).fetchall()
|
||
|
||
user_info = {}
|
||
for r in users_rows:
|
||
tid = int(r["telegram_id"])
|
||
user_info[tid] = {
|
||
"telegram_id": tid,
|
||
"username": r["username"] or f"ID: {tid}",
|
||
"supabase_user_id": None,
|
||
"supabase_email": None,
|
||
"is_bound": False,
|
||
}
|
||
|
||
for r in bindings_rows:
|
||
tid = int(r["telegram_id"])
|
||
if tid not in user_info:
|
||
user_info[tid] = {
|
||
"telegram_id": tid,
|
||
"username": f"ID: {tid}",
|
||
"supabase_user_id": r["supabase_user_id"],
|
||
"supabase_email": r["supabase_email"],
|
||
"is_bound": True,
|
||
}
|
||
else:
|
||
user_info[tid]["supabase_user_id"] = r["supabase_user_id"]
|
||
user_info[tid]["supabase_email"] = r["supabase_email"]
|
||
user_info[tid]["is_bound"] = True
|
||
|
||
# 2. Get Telegram Bot settings
|
||
bot_token = str(os.getenv("TELEGRAM_BOT_TOKEN") or "").strip()
|
||
chat_ids = get_telegram_chat_ids_from_env()
|
||
|
||
# Add other group IDs if configured
|
||
for env_name in [
|
||
"POLYWEATHER_TELEGRAM_GROUP_ID",
|
||
"POLYWEATHER_TELEGRAM_TOPICS_GROUP_ID",
|
||
]:
|
||
val = str(os.getenv(env_name) or "").strip()
|
||
if val and val not in chat_ids:
|
||
chat_ids.append(val)
|
||
|
||
if not bot_token or not chat_ids:
|
||
return {
|
||
"error": "Telegram Bot Token or Chat IDs not configured",
|
||
"anomalies": [],
|
||
}
|
||
|
||
# 3. Check membership status for all users in parallel
|
||
results = []
|
||
|
||
def check_user_chat(tg_id, chat_id):
|
||
try:
|
||
resp = _requests.get(
|
||
f"https://api.telegram.org/bot{bot_token}/getChatMember",
|
||
params={"chat_id": chat_id, "user_id": tg_id},
|
||
timeout=5,
|
||
)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
if data.get("ok"):
|
||
res_status = data["result"].get("status")
|
||
return chat_id, res_status
|
||
return chat_id, None
|
||
except Exception:
|
||
return chat_id, None
|
||
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
|
||
futures = {}
|
||
for tg_id in user_info.keys():
|
||
for c_id in chat_ids:
|
||
f = executor.submit(check_user_chat, tg_id, c_id)
|
||
futures[f] = (tg_id, c_id)
|
||
|
||
for f in concurrent.futures.as_completed(futures):
|
||
tg_id, c_id = futures[f]
|
||
try:
|
||
_, status = f.result()
|
||
if status in {"creator", "administrator", "member"}:
|
||
results.append((tg_id, c_id, status))
|
||
except Exception:
|
||
pass
|
||
|
||
# 4. Filter and categorize members
|
||
anomalies = []
|
||
valid_members = []
|
||
|
||
active_subs, _, used_active_window_query = _list_active_subscriptions_with_windows(
|
||
limit=5000
|
||
)
|
||
if not used_active_window_query:
|
||
active_subs = legacy_routes.SUPABASE_ENTITLEMENT.list_active_subscriptions(
|
||
limit=5000
|
||
)
|
||
active_subs_map = {}
|
||
for sub in active_subs:
|
||
uid = str(sub.get("user_id") or "").strip().lower()
|
||
if uid:
|
||
active_subs_map[uid] = sub
|
||
|
||
for tg_id, chat_id, status in results:
|
||
info = user_info[tg_id]
|
||
|
||
if not info["is_bound"]:
|
||
anomalies.append(
|
||
{
|
||
"telegram_id": tg_id,
|
||
"username": info["username"],
|
||
"chat_id": chat_id,
|
||
"status": status,
|
||
"anomaly_type": "unbound",
|
||
"reason": "未绑定网页账号",
|
||
"email": None,
|
||
"expires_at": None,
|
||
}
|
||
)
|
||
else:
|
||
uid = str(info["supabase_user_id"]).strip().lower()
|
||
sub = active_subs_map.get(uid)
|
||
is_paid = False
|
||
plan_code = ""
|
||
expires_at = None
|
||
|
||
if sub:
|
||
plan_code = str(sub.get("plan_code") or "").strip().lower()
|
||
source = str(sub.get("source") or "").strip().lower()
|
||
is_paid = "trial" not in plan_code and "trial" not in source
|
||
expires_at = sub.get("expires_at")
|
||
|
||
if not sub:
|
||
anomalies.append(
|
||
{
|
||
"telegram_id": tg_id,
|
||
"username": info["username"],
|
||
"chat_id": chat_id,
|
||
"status": status,
|
||
"anomaly_type": "expired",
|
||
"reason": "没有有效的会员订阅",
|
||
"email": info["supabase_email"],
|
||
"expires_at": None,
|
||
}
|
||
)
|
||
elif not is_paid:
|
||
anomalies.append(
|
||
{
|
||
"telegram_id": tg_id,
|
||
"username": info["username"],
|
||
"chat_id": chat_id,
|
||
"status": status,
|
||
"anomaly_type": "trial_only",
|
||
"reason": f"仅拥有试用会员 ({plan_code})",
|
||
"email": info["supabase_email"],
|
||
"expires_at": expires_at,
|
||
}
|
||
)
|
||
else:
|
||
valid_members.append(
|
||
{
|
||
"telegram_id": tg_id,
|
||
"username": info["username"],
|
||
"chat_id": chat_id,
|
||
"status": status,
|
||
"email": info["supabase_email"],
|
||
"plan_code": plan_code,
|
||
"expires_at": expires_at,
|
||
}
|
||
)
|
||
|
||
return {
|
||
"anomalies": anomalies,
|
||
"valid_count": len(valid_members),
|
||
"anomaly_count": len(anomalies),
|
||
}
|