# ══════════════════════════════════════════════════════════════════ # admin_panel.py — Web admin dashboard for Phone Lookup Bot # # Mounts at: GET /admin?token= # # Views (all take ?token=... first): # ?view=dash – dashboard + live users + recent activity # ?view=log[&page=N][&uid=X][&a=action] – full activity log, filterable # ?view=users[&page=N] – paginated user list # ?view=user&uid=X – single-user drill-down # ?view=banned – banned users # ?view=top – top searched phone numbers # ?view=phone[&num=XXXXXXXXXX] – phone search across log # ?view=export&fmt=csv|json – download activity log # ?view=maint – maintenance enable confirmation page # # Actions (all take ?token=... first): # ?action=ban&uid=X&reason=... – ban a user # ?action=unban&uid=X – unban a user # ?action=maint&on=0|1[&reason=...] – toggle maintenance # # All actions redirect (303) back to the relevant view with a flash msg. # Token = WEBHOOK_SECRET from app.py env (same one used for Telegram). # ══════════════════════════════════════════════════════════════════ import sys import io import csv import json import html import time import traceback from collections import Counter from fastapi import Request from fastapi.responses import HTMLResponse, Response, RedirectResponse # ───────────────────────────────────────────── # Locate the running bot module (works with __main__ / app / gradio-cli) # ───────────────────────────────────────────── def _app(): for name in ("app", "__main__"): m = sys.modules.get(name) if m is not None and hasattr(m, "ACTIVITY_LOG") and hasattr(m, "UNIQUE_USERS"): return m for m in list(sys.modules.values()): if m is None: continue if (hasattr(m, "ACTIVITY_LOG") and hasattr(m, "UNIQUE_USERS") and hasattr(m, "WEBHOOK_SECRET")): return m return None # ───────────────────────────────────────────── # Small helpers # ───────────────────────────────────────────── def _esc(s): return html.escape(str(s)) if s is not None else "" def _ts(t): return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(t)) if t else "—" def _ago(t): if not t: return "—" d = time.time() - t if d < 60: return f"{int(d)}s" if d < 3600: return f"{int(d / 60)}m" if d < 86400: return f"{int(d / 3600)}h" return f"{int(d / 86400)}d" def _name_of(app, uid, user=None): info = app.UNIQUE_USERS.get(uid) or {} uname = (user or {}).get("username") or info.get("username") first = (user or {}).get("first_name") or info.get("first_name") if uname: return f"@{uname}" if first: return first return f"(anon {uid})" # ───────────────────────────────────────────── # CSS + page shell # ───────────────────────────────────────────── CSS = """ * { box-sizing: border-box; } body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; background: #0f1419; color: #e6e6e6; margin: 0; padding: 0; font-size: 14px; line-height: 1.4; } .container { max-width: 1500px; margin: 0 auto; padding: 24px; } h1 { font-size: 22px; margin: 0 0 6px 0; font-weight: 600; } h2 { font-size: 16px; margin: 28px 0 10px 0; color: #a9b4bd; font-weight: 600; } a { color: #7ea5e5; text-decoration: none; } a:hover { text-decoration: underline; } p.muted, .muted { color: #6f7b87; font-size: 12px; } nav { display: flex; gap: 4px; margin: 16px 0 20px 0; background: #1a1f26; padding: 8px; border-radius: 10px; flex-wrap: wrap; border: 1px solid #2a323c; } nav a { color: #8a9ba8; padding: 8px 14px; border-radius: 6px; font-size: 13px; font-weight: 500; } nav a:hover { background: #232a33; color: #e6e6e6; text-decoration: none; } nav a.active { background: #c96442; color: white; } .cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(160px, 1fr)); gap: 12px; margin: 16px 0 24px 0; } .card { background: #1a1f26; border-radius: 10px; padding: 14px 18px; border: 1px solid #2a323c; } .card .v { font-size: 24px; font-weight: 700; color: #c96442; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } .card .l { font-size: 10px; color: #6f7b87; text-transform: uppercase; letter-spacing: 0.6px; margin-top: 4px; font-weight: 600; } table { width: 100%; background: #1a1f26; border-collapse: collapse; border-radius: 10px; overflow: hidden; border: 1px solid #2a323c; font-size: 13px; } th, td { padding: 10px 12px; text-align: left; border-bottom: 1px solid #232a33; vertical-align: top; } th { background: #232a33; color: #8a9ba8; font-weight: 600; font-size: 10px; text-transform: uppercase; letter-spacing: 0.6px; } tr:hover td { background: #1e242b; } tr:last-child td { border-bottom: none; } td { color: #cdd5dc; } code { background: #0f1419; padding: 2px 6px; border-radius: 4px; font-size: 12px; color: #e6e6e6; font-family: "SF Mono", Consolas, "Monaco", monospace; } .btn { display: inline-block; padding: 7px 13px; border-radius: 6px; font-size: 12px; font-weight: 500; text-decoration: none; border: none; cursor: pointer; background: #2a323c; color: #cdd5dc; font-family: inherit; line-height: 1; } .btn:hover { background: #354252; text-decoration: none; } .btn.primary { background: #c96442; color: white; } .btn.primary:hover { background: #b85234; } .btn.danger { background: #7a2929; color: #ff9f9f; } .btn.danger:hover { background: #943232; } .btn.success { background: #245c3e; color: #7ee5a3; } .btn.success:hover { background: #2d7050; } .pagination { display: flex; gap: 6px; margin-top: 16px; align-items: center; flex-wrap: wrap; } .pagination a, .pagination span { padding: 6px 12px; border-radius: 6px; font-size: 13px; background: #232a33; color: #cdd5dc; } .pagination a:hover { background: #354252; text-decoration: none; } .pagination .current { background: #c96442; color: white; font-weight: 600; } .chip { display: inline-flex; align-items: center; gap: 6px; padding: 4px 10px; border-radius: 999px; font-size: 11px; background: #232a33; color: #8a9ba8; margin-right: 6px; } .chip.active { background: #c96442; color: white; } .chip a { color: white; font-weight: bold; } .alert { padding: 11px 15px; border-radius: 8px; margin: 12px 0; font-size: 13px; border: 1px solid; } .alert.success { background: #163a24; color: #7ee5a3; border-color: #245c3e; } .alert.error { background: #3a1818; color: #ff9f9f; border-color: #5c2424; } .alert.warn { background: #3a2e14; color: #e5d17e; border-color: #5c4a24; } .alert.info { background: #172a3a; color: #7ea5e5; border-color: #243c5c; } .empty { text-align: center; padding: 36px; color: #6f7b87; } input[type="text"], input[type="number"], textarea, select { background: #0f1419; border: 1px solid #2a323c; color: #e6e6e6; padding: 8px 12px; border-radius: 6px; font-size: 14px; font-family: inherit; width: 100%; max-width: 340px; } input:focus, textarea:focus, select:focus { outline: none; border-color: #c96442; } .row { display: flex; gap: 10px; align-items: center; margin-bottom: 12px; flex-wrap: wrap; } .badge { display: inline-block; padding: 2px 8px; border-radius: 4px; font-size: 10px; font-weight: 700; text-transform: uppercase; letter-spacing: 0.4px; } .badge.ok { background: #245c3e; color: #7ee5a3; } .badge.warn { background: #5c4a24; color: #e5d17e; } .badge.err { background: #5c2424; color: #ff9f9f; } footer { margin-top: 40px; padding-top: 16px; border-top: 1px solid #232a33; color: #6f7b87; font-size: 11px; display: flex; justify-content: space-between; flex-wrap: wrap; gap: 8px; } """ def _page(title, body, refresh=None): meta = f'' if refresh else "" footer = ( '' ) return f""" {_esc(title)} {meta}
{body} {footer}
""" def _nav(token, active="dash"): def a(key, label): cls = "active" if key == active else "" return f'{label}' return ( "" ) def _flash(request): out = "" if request.query_params.get("flash"): out += f'
{_esc(request.query_params["flash"])}
' if request.query_params.get("err"): out += f'
{_esc(request.query_params["err"])}
' return out # ═════════════════════════════════════════════ # VIEWS # ═════════════════════════════════════════════ async def _view_dash(app, token, request): s = app._get_stats() uptime_min = (time.time() - app.STATS["started"]) / 60 live = app._get_live_users() if hasattr(app, "_get_live_users") else [] refresh = request.query_params.get("refresh") cards = ( '
' f'
{s["unique_users"]}
👥 Unique Users
' f'
{s["live_users"]}
🔴 Live (5m)
' f'
{s["today_users"]}
📅 Active Today
' f'
{s["today_searches"]}
🔍 Searches Today
' f'
{s["total_searches"]}
📊 Total Searches
' f'
{s["peak_concurrent"]}
📈 Peak Concurrent
' f'
{s["banned_count"]}
🚫 Banned
' f'
{"ON" if s["maintenance"] else "OFF"}
🛠️ Maintenance
' f'
{len(app.SEARCH_CACHE)}
💾 Cached Queries
' f'
{app.STATS["errors"]}
❌ Errors
' '
' ) live_rows = "" for uid, ts in live[:15]: info = app.UNIQUE_USERS.get(uid, {}) nm = _name_of(app, uid, info) live_rows += ( f'' f'{_esc(nm)}' f'{uid}' f'{_ago(ts)}' f'' ) if not live_rows: live_rows = 'No active users' entries = list(app.ACTIVITY_LOG) entries.reverse() recent_rows = "" for e in entries[:15]: nm = e.get("username") or e.get("first_name") or f"(anon {e['user_id']})" if nm and not nm.startswith("(") and not nm.startswith("@"): nm = e.get("first_name") recent_rows += ( f'' f'{_ts(e["ts"])}' f'{_esc(nm)}' f'{_esc(e["action"])}' f'{_esc(e.get("input", "") or "—")}' f'{_esc(e.get("summary") or e.get("reply_type") or "—")}' f'' ) if not recent_rows: recent_rows = 'No activity yet' refresh_btn = ( f'⏸️ Disable auto-refresh' if refresh else f'🔁 Auto-refresh (10s)' ) maint_on, _m_reason = app._get_maintenance() if maint_on: maint_btn = ( f'
' f'' f'' f'' f'' f'
' ) else: maint_btn = f'🔴 Turn ON maintenance' body = f"""

📞 Phone Lookup Bot — Admin

v{_esc(app.VERSION)} · Uptime: {uptime_min:.1f} min · DuckDB: {"✅" if app._con_ready.is_set() else "⏳"} · Warmed: {"✅" if app._warmed.is_set() else "⏳"}

{_nav(token, "dash")} {_flash(request)} {cards}
{refresh_btn} {maint_btn} 📥 CSV 📥 JSON

🔴 Live Users

{live_rows}
UserIDLast Active

📜 Recent Activity

{recent_rows}
TimeUserActionInputResult

View full activity log →

""" return HTMLResponse(_page("Admin · Dashboard", body, refresh=int(refresh) if refresh else None)) async def _view_log(app, token, request): try: page = max(0, int(request.query_params.get("page", "0"))) except ValueError: page = 0 filter_uid = request.query_params.get("uid") filter_action = request.query_params.get("a") page_size = 40 entries = list(app.ACTIVITY_LOG) entries.reverse() if filter_uid: try: uid = int(filter_uid) entries = [e for e in entries if e["user_id"] == uid] except (ValueError, TypeError): filter_uid = None if filter_action: entries = [e for e in entries if e["action"] == filter_action] total = len(entries) total_pages = max(1, (total + page_size - 1) // page_size) page = min(page, total_pages - 1) chunk = entries[page * page_size : (page + 1) * page_size] rows = "" for e in chunk: nm = e.get("username") if nm: nm = f"@{nm}" else: nm = e.get("first_name") or "—" uid = e["user_id"] elapsed = e.get("elapsed_ms", 0) rows += ( f'' f'{_ts(e["ts"])}' f'{_esc(nm)}' f'
{uid}' f'{_esc(e["action"])}' f'{_esc(e.get("input", "") or "—")}' f'{_esc(e.get("summary") or e.get("reply_type") or "—")}' f'{f"{elapsed}ms" if elapsed else "—"}' f'{_esc(e.get("source") or "—")}' f'' ) if not rows: rows = 'No matching entries' def page_url(p): qs = f"?token={token}&view=log&page={p}" if filter_uid: qs += f"&uid={filter_uid}" if filter_action: qs += f"&a={filter_action}" return qs pag = '' chips = "" if filter_uid: chips += (f'User {filter_uid} ' f'×') if filter_action: chips += (f'Action: {_esc(filter_action)} ' f'×') if not chips: chips = 'No filters' body = f"""

📜 Activity Log

{_nav(token, "log")} {_flash(request)}
{chips}
{rows}
TimeUserAction InputResultLatencySource
{pag} """ return HTMLResponse(_page("Admin · Activity Log", body)) async def _view_users(app, token, request): try: page = max(0, int(request.query_params.get("page", "0"))) except ValueError: page = 0 page_size = 40 users = sorted(app.UNIQUE_USERS.items(), key=lambda kv: -kv[1].get("last_seen", 0)) total = len(users) total_pages = max(1, (total + page_size - 1) // page_size) page = min(page, total_pages - 1) chunk = users[page * page_size : (page + 1) * page_size] rows = "" for uid, info in chunk: nm = _name_of(app, uid, info) banned = uid in app.BANNED_USERS badge = ' BANNED' if banned else '' rows += ( f'' f'{_esc(nm)}{badge}' f'{uid}' f'{info.get("count", 0)}' f'{_ago(info.get("last_seen", 0))}' f'{_ts(info.get("first_seen", 0))}' f'' ) if not rows: rows = 'No users yet' pag = '' body = f"""

👥 All Users

{_nav(token, "users")} {_flash(request)} {rows}
UserIDRequestsLast SeenFirst Seen
{pag} """ return HTMLResponse(_page("Admin · Users", body)) async def _view_user(app, token, request, uid): info = app.UNIQUE_USERS.get(uid) if not info: body = ( f"

👤 User {uid}

" f"{_nav(token, 'users')}" f"
User not found in registry.
" f"← Back to users" ) return HTMLResponse(_page(f"User {uid}", body), status_code=404) nm = _name_of(app, uid, info) banned = uid in app.BANNED_USERS reason = app.BAN_REASONS.get(uid, "") if banned else "" hist = list(app.USER_HISTORY.get(uid, [])) hist_rows = "" for h in hist: phone = _esc(h.get("phone", "")) hist_rows += ( f'' f'{phone}' f'{h.get("count", 0)}' f'{h.get("elapsed", 0):.1f}s' f'{_esc(h.get("source", ""))}' f'{_ago(h.get("ts", 0))}' f'' ) if not hist_rows: hist_rows = 'No searches' entries = [e for e in app.ACTIVITY_LOG if e["user_id"] == uid] entries.reverse() log_rows = "" for e in entries[:20]: log_rows += ( f'' f'{_ts(e["ts"])}' f'{_esc(e["action"])}' f'{_esc(e.get("input", "") or "—")}' f'{_esc(e.get("summary") or e.get("reply_type") or "—")}' f'' ) if not log_rows: log_rows = 'No activity' if banned: action_html = ( f'
🚫 Banned — reason: {_esc(reason or "no reason")}
' f'
' f'' f'' f'' f'
' ) else: action_html = ( f'
' f'' f'' f'' f'
' f'' f'' f'
' ) body = f"""

👤 User Detail

{_nav(token, "users")} {_flash(request)}
{_esc(nm)}
Name
{uid}
Telegram ID
{info.get("count", 0)}
Requests
{_ago(info.get("last_seen", 0))}
Last Seen
{_ts(info.get("first_seen", 0))}
First Seen
{action_html}

🕘 Recent Searches

{hist_rows}
PhoneResultsLatencySourceWhen

📜 Activity (last 20)

{log_rows}
TimeActionInputResult

View all logs for this user → ← Back to users

""" return HTMLResponse(_page(f"User · {nm}", body)) async def _view_banned(app, token, request): banned = list(app.BANNED_USERS) rows = "" for uid in banned: info = app.UNIQUE_USERS.get(uid, {}) nm = _name_of(app, uid, info) reason = app.BAN_REASONS.get(uid, "") or "—" rows += ( f'' f'{_esc(nm)}' f'{uid}' f'{_esc(reason)}' f'' f'
' f'' f'' f'' f'' f'
' f'' f'' ) if not rows: rows = 'No banned users' body = f"""

🚫 Banned Users ({len(banned)})

{_nav(token, "banned")} {_flash(request)} {rows}
UserIDReasonAction
""" return HTMLResponse(_page("Admin · Banned", body)) async def _view_top(app, token, request): rows = "" for i, (num, cnt) in enumerate(app.PHONE_COUNTS.most_common(50), 1): rows += ( f'' f'{i}' f'{_esc(num)}' f'{cnt}' f'' ) if not rows: rows = 'No searches yet' body = f"""

🔍 Top Searched Numbers

{_nav(token, "top")} {rows}
#PhoneSearches
""" return HTMLResponse(_page("Admin · Top Phones", body)) async def _view_phone(app, token, request): num = (request.query_params.get("num") or "").strip() search_form = f"""
""" if not num: body = f"""

☎️ Phone Lookup

{_nav(token, "phone")} {search_form}

Enter a phone number to see who searched it and recent activity.

""" return HTMLResponse(_page("Admin · Phone", body)) entries = [e for e in app.ACTIVITY_LOG if e.get("input") == num and e["action"] == "search"] entries.reverse() count = app.PHONE_COUNTS.get(num, 0) searchers = Counter(e["user_id"] for e in entries) searcher_rows = "" for uid, c in searchers.most_common(25): info = app.UNIQUE_USERS.get(uid, {}) nm = _name_of(app, uid, info) searcher_rows += ( f'' f'{_esc(nm)}' f'{uid}' f'{c}' f'' ) if not searcher_rows: searcher_rows = 'No searchers' log_rows = "" for e in entries[:30]: nm = e.get("username") if nm: nm = f"@{nm}" else: nm = e.get("first_name") or "—" log_rows += ( f'' f'{_ts(e["ts"])}' f'{_esc(nm)}' f'{_esc(e.get("summary") or e.get("reply_type") or "—")}' f'' ) if not log_rows: log_rows = 'No searches for this number' body = f"""

☎️ Phone: {_esc(num)}

{_nav(token, "phone")} {_flash(request)}
{count}
Total Searches
{len(searchers)}
Unique Searchers
{search_form}

Who searched this number

{searcher_rows}
UserIDSearches

Recent searches (last 30)

{log_rows}
TimeUserResult
""" return HTMLResponse(_page(f"Admin · Phone {num}", body)) async def _view_maint_confirm(app, token, request): body = f"""

🔴 Enable Maintenance Mode

{_nav(token, "dash")}
This will block all non-admin users from searching. Basic commands (/start, /status, /info, /myid) will still work.
Cancel
""" return HTMLResponse(_page("Admin · Maintenance", body)) async def _view_export(app, token, request): fmt = (request.query_params.get("fmt") or "csv").lower() entries = list(app.ACTIVITY_LOG) if fmt == "json": data = [] for e in entries: data.append({ "timestamp": _ts(e["ts"]), "user_id": e["user_id"], "username": e.get("username"), "first_name": e.get("first_name"), "action": e["action"], "input": e.get("input"), "reply_type": e.get("reply_type"), "summary": e.get("summary"), "elapsed_ms": e.get("elapsed_ms", 0), "source": e.get("source"), }) content = json.dumps(data, indent=2, ensure_ascii=False).encode("utf-8") return Response( content=content, media_type="application/json", headers={"Content-Disposition": 'attachment; filename="activity_log.json"'}, ) buf = io.StringIO() w = csv.writer(buf) w.writerow([ "timestamp", "user_id", "username", "first_name", "action", "input", "reply_type", "summary", "elapsed_ms", "source", ]) for e in entries: w.writerow([ _ts(e["ts"]), e["user_id"], e.get("username") or "", e.get("first_name") or "", e["action"], e.get("input", ""), e.get("reply_type", ""), e.get("summary", ""), e.get("elapsed_ms", 0), e.get("source", ""), ]) content = buf.getvalue().encode("utf-8") return Response( content=content, media_type="text/csv", headers={"Content-Disposition": 'attachment; filename="activity_log.csv"'}, ) # ═════════════════════════════════════════════ # ACTIONS # ═════════════════════════════════════════════ async def _action_ban(app, token, request): try: uid = int(request.query_params.get("uid", "0")) except (ValueError, TypeError): uid = 0 if uid <= 0: return RedirectResponse( f"/admin?token={token}&view=users&err=Invalid+user+ID", status_code=303, ) reason = (request.query_params.get("reason") or "")[:200] app._ban_user(uid, reason) app._log_activity({"id": 0}, "admin", f"web-ban {uid}", "banned", reason or "no reason") return RedirectResponse( f"/admin?token={token}&view=user&uid={uid}&flash=User+{uid}+banned", status_code=303, ) async def _action_unban(app, token, request): try: uid = int(request.query_params.get("uid", "0")) except (ValueError, TypeError): uid = 0 if uid <= 0: return RedirectResponse( f"/admin?token={token}&view=banned&err=Invalid+user+ID", status_code=303, ) ok = app._unban_user(uid) app._log_activity({"id": 0}, "admin", f"web-unban {uid}", "unbanned") msg = f"User+{uid}+unbanned" if ok else f"User+{uid}+was+not+banned" return RedirectResponse( f"/admin?token={token}&view=banned&flash={msg}", status_code=303, ) async def _action_maintenance(app, token, request): on = request.query_params.get("on", "1") in ("1", "true", "on", "yes") reason = (request.query_params.get("reason") or "")[:100] app._set_maintenance(on, reason) app._log_activity({"id": 0}, "admin", f"web-maintenance-{'on' if on else 'off'}", "state", reason) msg = "Maintenance+ON" if on else "Maintenance+OFF" return RedirectResponse( f"/admin?token={token}&view=dash&flash={msg}", status_code=303, ) # ═════════════════════════════════════════════ # MAIN ENTRYPOINT — mounted at /admin by app.py # ═════════════════════════════════════════════ async def admin_dashboard(request: Request): app = _app() if app is None: return HTMLResponse( "

500 — Bot module not found

" "

admin_panel.py could not locate the running bot module. " "Make sure app.py is imported (or running as __main__).

", status_code=500, ) token = request.query_params.get("token", "") if not app.WEBHOOK_SECRET or token != app.WEBHOOK_SECRET: return HTMLResponse( "

403 — Invalid or missing token

" "

Append ?token=<WEBHOOK_SECRET> to the URL.

", status_code=403, ) try: action = request.query_params.get("action") if action == "ban": return await _action_ban(app, token, request) if action == "unban": return await _action_unban(app, token, request) if action == "maint": return await _action_maintenance(app, token, request) view = request.query_params.get("view", "dash") if view == "dash": return await _view_dash(app, token, request) if view == "log": return await _view_log(app, token, request) if view == "users": return await _view_users(app, token, request) if view == "user": try: uid = int(request.query_params.get("uid", "0")) except (ValueError, TypeError): return RedirectResponse( f"/admin?token={token}&view=users&err=Invalid+user+ID", status_code=303, ) return await _view_user(app, token, request, uid) if view == "banned": return await _view_banned(app, token, request) if view == "top": return await _view_top(app, token, request) if view == "phone": return await _view_phone(app, token, request) if view == "export": return await _view_export(app, token, request) if view == "maint": return await _view_maint_confirm(app, token, request) return await _view_dash(app, token, request) except Exception as e: err_html = ( "

500 — Admin panel error

" f"
"
            f"{_esc(str(e))}\n\n{_esc(traceback.format_exc())}
" f"← Back to dashboard" ) return HTMLResponse(_page("Error", err_html), status_code=500)