diff --git a/frontend/app/ops/telegram-audit/page.tsx b/frontend/app/ops/telegram-audit/page.tsx new file mode 100644 index 00000000..5d94b0e2 --- /dev/null +++ b/frontend/app/ops/telegram-audit/page.tsx @@ -0,0 +1,5 @@ +import { TelegramAuditPageClient } from "@/components/ops/telegram-audit/TelegramAuditPageClient"; + +export default function TelegramAuditPage() { + return ; +} diff --git a/frontend/components/ops/layout/AdminSidebar.tsx b/frontend/components/ops/layout/AdminSidebar.tsx index fe890b60..7b53b621 100644 --- a/frontend/components/ops/layout/AdminSidebar.tsx +++ b/frontend/components/ops/layout/AdminSidebar.tsx @@ -14,6 +14,7 @@ import { FileText, ScrollText, Activity, + ShieldAlert, } from "lucide-react"; const navGroups = [ @@ -32,6 +33,7 @@ const navGroups = [ items: [ { href: "/ops/payments", icon: CreditCard, label: "支付管理" }, { href: "/ops/memberships", icon: UserCheck, label: "会员订阅" }, + { href: "/ops/telegram-audit", icon: ShieldAlert, label: "电报清理" }, { href: "/ops/users", icon: Users, label: "用户积分" }, ], }, diff --git a/frontend/components/ops/telegram-audit/TelegramAuditPageClient.tsx b/frontend/components/ops/telegram-audit/TelegramAuditPageClient.tsx new file mode 100644 index 00000000..136ae740 --- /dev/null +++ b/frontend/components/ops/telegram-audit/TelegramAuditPageClient.tsx @@ -0,0 +1,351 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { opsApi } from "@/lib/ops-api"; +import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { + RefreshCcw, + ShieldAlert, + Copy, + Check, + Search, + UserX, + Mail, + Calendar, + AlertTriangle, +} from "lucide-react"; + +interface TelegramAnomaly { + telegram_id: number; + username: string; + chat_id: string; + status: string; + anomaly_type: "unbound" | "expired" | "trial_only"; + reason: string; + email: string | null; + expires_at: string | null; +} + +export function TelegramAuditPageClient() { + const [loading, setLoading] = useState(true); + const [data, setData] = useState<{ + anomalies: TelegramAnomaly[]; + valid_count: number; + anomaly_count: number; + error?: string; + } | null>(null); + const [filterType, setFilterType] = useState("all"); + const [searchQuery, setSearchQuery] = useState(""); + const [copiedId, setCopiedId] = useState(null); + + const loadData = async () => { + setLoading(true); + try { + const res = await opsApi.telegramAudit(); + setData(res); + } catch (err: any) { + setData({ + anomalies: [], + valid_count: 0, + anomaly_count: 0, + error: err.message || "获取审核数据失败", + }); + } + setLoading(false); + }; + + useEffect(() => { + void loadData(); + }, []); + + const handleCopy = (id: number) => { + void navigator.clipboard.writeText(String(id)); + setCopiedId(id); + setTimeout(() => setCopiedId(null), 1500); + }; + + if (loading) { + return ( +
+ +

正在与电报服务器同步并审计群成员，可能需要几秒钟...

+
+ ); + } + + const anomalies = data?.anomalies ?? []; + const error = data?.error; + + const filtered = anomalies.filter((item) => { + const matchesFilter = filterType === "all" || item.anomaly_type === filterType; + const matchesSearch = + searchQuery.trim() === "" || + String(item.telegram_id).includes(searchQuery) || + (item.username && item.username.toLowerCase().includes(searchQuery.toLowerCase())) || + (item.email && item.email.toLowerCase().includes(searchQuery.toLowerCase())) || + (item.reason && item.reason.toLowerCase().includes(searchQuery.toLowerCase())); + return matchesFilter && matchesSearch; + }); + + const countUnbound = anomalies.filter((x) => x.anomaly_type === "unbound").length; + const countExpired = anomalies.filter((x) => x.anomaly_type === "expired").length; + const countTrial = anomalies.filter((x) => x.anomaly_type === "trial_only").length; + + return ( +
+
+
+

电报群清理与异常检测

+

+ 本页通过核对本地数据库中与 Bot 交互过（或绑定过）的用户，实时查询 Telegram 接口审计群成员资格。 +

+
+ +
+ + {error && ( + + + +
+

检测出发生错误

+

{error}

+
+ + + )} + + {/* Overview Cards */} +
+ + + 群内付费用户 (正常) + + +
{data?.valid_count ?? 0} 人
+

已绑定且在订阅期内的正常付费用户

+ + + + + + 未绑定网页账号 + + +
{countUnbound} 人
+

未能在网页端找到绑定对应 TG 的记录

+ + + + + + 订阅已过期 + + +
{countExpired} 人
+

已绑定，但当前没有任何有效订阅

+ + + + + + 仅拥有试用期 + + +
{countTrial} 人
+

仅注册了免费试用订阅（不具备入群资格）

+ + +
+ + {/* Filter and Search Bar */} + + +
+ {/* Filter Buttons */} +
+ + + + +
+ + {/* Search Input */} +
+ + setSearchQuery(e.target.value)} + className="pl-9 h-9 bg-slate-950 border-white/10 text-xs rounded-lg text-slate-100 placeholder:text-slate-500" + /> +
+
+ + + + {/* Main Table */} + + + 需清理成员列表 + + +
+ + + + + + + + + + + + + {filtered.length > 0 ? ( + filtered.map((row) => ( + + + + + + + + + )) + ) : ( + + + + )} + +
电报用户名 & ID异常类型原因绑定邮箱订阅到期日操作
+
+ {row.username.startsWith("@") ? ( + + {row.username} + + ) : ( + {row.username} + )} +
+
+ {row.telegram_id} + +
+ 		+ {row.anomaly_type === "unbound" ? ( + + 未绑定 + + ) : row.anomaly_type === "expired" ? ( + + 已到期 + + ) : ( + + 试用会员 + + )} + + {row.reason} + + {row.email ? ( +
+ + + {row.email} + +
+ ) : ( + — + )} + 		+ {row.expires_at ? ( +
+ + {row.expires_at.split("T")[0]} +
+ ) : ( + — + )} + 		+ +
+ {searchQuery ? "未找到符合搜索条件的异常成员" : "当前群内没有检测出任何已过期的异常成员 🎉"} +
+
+ + + + + +

💡 怎么手动清理？

+
    +
  1. 在上述列表中点击任意用户的 复制 ID 按钮。
    2. +
  2. 打开你的 Telegram 客户端，进入对应的群组设置面板。
    3. +
  3. 点击群成员列表中的“添加成员”或直接搜索该用户的电报 ID 或电报用户名。
    4. +
  4. 在用户资料卡中选择 Remove from Group (踢出群组) 即可。
    5. +
+ + +
+ ); +} diff --git a/frontend/lib/ops-api.ts b/frontend/lib/ops-api.ts index 1ba3fdd7..61fb03a2 100644 --- a/frontend/lib/ops-api.ts +++ b/frontend/lib/ops-api.ts @@ -126,4 +126,21 @@ export const opsApi = { }>; }>("/api/ops/training/accuracy"); }, + telegramAudit() { + return opsFetch<{ + anomalies: Array<{ + telegram_id: number; + username: string; + chat_id: string; + status: string; + anomaly_type: "unbound" | "expired" | "trial_only"; + reason: string; + email: string | null; + expires_at: string | null; + }>; + valid_count: number; + anomaly_count: number; + error?: string; + }>("/api/ops/telegram/members-audit"); + }, }; diff --git a/web/routers/ops.py b/web/routers/ops.py index a1196088..506db86e 100644 --- a/web/routers/ops.py +++ b/web/routers/ops.py @@ -22,6 +22,7 @@ from web.services.ops_api import ( search_ops_users, update_ops_config, get_ops_training_accuracy, + get_ops_telegram_audit, ) router = APIRouter(tags=["ops"]) @@ -167,3 +168,9 @@ async def ops_health_check(request: Request): async def ops_training_accuracy(request: Request): return get_ops_training_accuracy(request) + +@router.get("/api/ops/telegram/members-audit") +async def ops_telegram_audit(request: Request): + return get_ops_telegram_audit(request) + + diff --git a/web/services/ops_api.py b/web/services/ops_api.py index c42e739d..4f5d7f09 100644 --- a/web/services/ops_api.py +++ b/web/services/ops_api.py @@ -836,3 +836,175 @@ def get_ops_training_accuracy(request: Request) -> Dict[str, Any]: return {"accuracy": accuracy_data} + +def get_ops_telegram_audit(request: Request) -> Dict[str, Any]: + _require_ops(request) + import concurrent.futures + import os + import sqlite3 + import requests + from src.database.db_manager import DBManager + from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env + + db = DBManager() + + # 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 + + all_tasks = [] + 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 = [] + + from web.services.ops_api import legacy_routes + 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), + } + +