| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| 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 |
|
|
|
|
| |
| |
| |
| 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 |
|
|
|
|
| |
| |
| |
| 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 = """ |
| * { 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'<meta http-equiv="refresh" content="{int(refresh)}">' if refresh else "" |
| footer = ( |
| '<footer>' |
| f'<span>Server time: {_ts(time.time())}</span>' |
| '<span>Admin web panel · GET /admin</span>' |
| '</footer>' |
| ) |
| return f"""<!doctype html> |
| <html lang="en"><head> |
| <meta charset="utf-8"> |
| <meta name="viewport" content="width=device-width, initial-scale=1"> |
| <title>{_esc(title)}</title> |
| {meta} |
| <style>{CSS}</style> |
| </head><body> |
| <div class="container"> |
| {body} |
| {footer} |
| </div> |
| </body></html>""" |
|
|
|
|
| def _nav(token, active="dash"): |
| def a(key, label): |
| cls = "active" if key == active else "" |
| return f'<a href="?token={token}&view={key}" class="{cls}">{label}</a>' |
| return ( |
| "<nav>" |
| + a("dash", "🏠 Dashboard") |
| + a("log", "📜 Activity") |
| + a("users", "👥 Users") |
| + a("banned", "🚫 Banned") |
| + a("top", "🔍 Top Phones") |
| + a("phone", "☎️ Phone Search") |
| + "</nav>" |
| ) |
|
|
|
|
| def _flash(request): |
| out = "" |
| if request.query_params.get("flash"): |
| out += f'<div class="alert success">{_esc(request.query_params["flash"])}</div>' |
| if request.query_params.get("err"): |
| out += f'<div class="alert error">{_esc(request.query_params["err"])}</div>' |
| return out |
|
|
|
|
| |
| |
| |
|
|
| 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 = ( |
| '<div class="cards">' |
| f'<div class="card"><div class="v">{s["unique_users"]}</div><div class="l">👥 Unique Users</div></div>' |
| f'<div class="card"><div class="v">{s["live_users"]}</div><div class="l">🔴 Live (5m)</div></div>' |
| f'<div class="card"><div class="v">{s["today_users"]}</div><div class="l">📅 Active Today</div></div>' |
| f'<div class="card"><div class="v">{s["today_searches"]}</div><div class="l">🔍 Searches Today</div></div>' |
| f'<div class="card"><div class="v">{s["total_searches"]}</div><div class="l">📊 Total Searches</div></div>' |
| f'<div class="card"><div class="v">{s["peak_concurrent"]}</div><div class="l">📈 Peak Concurrent</div></div>' |
| f'<div class="card"><div class="v">{s["banned_count"]}</div><div class="l">🚫 Banned</div></div>' |
| f'<div class="card"><div class="v">{"ON" if s["maintenance"] else "OFF"}</div><div class="l">🛠️ Maintenance</div></div>' |
| f'<div class="card"><div class="v">{len(app.SEARCH_CACHE)}</div><div class="l">💾 Cached Queries</div></div>' |
| f'<div class="card"><div class="v">{app.STATS["errors"]}</div><div class="l">❌ Errors</div></div>' |
| '</div>' |
| ) |
|
|
| live_rows = "" |
| for uid, ts in live[:15]: |
| info = app.UNIQUE_USERS.get(uid, {}) |
| nm = _name_of(app, uid, info) |
| live_rows += ( |
| f'<tr>' |
| f'<td><a href="?token={token}&view=user&uid={uid}">{_esc(nm)}</a></td>' |
| f'<td><code>{uid}</code></td>' |
| f'<td class="muted">{_ago(ts)}</td>' |
| f'</tr>' |
| ) |
| if not live_rows: |
| live_rows = '<tr><td colspan="3" class="empty">No active users</td></tr>' |
|
|
| 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'<tr>' |
| f'<td class="muted" style="white-space:nowrap">{_ts(e["ts"])}</td>' |
| f'<td><a href="?token={token}&view=user&uid={e["user_id"]}">{_esc(nm)}</a></td>' |
| f'<td>{_esc(e["action"])}</td>' |
| f'<td><code>{_esc(e.get("input", "") or "—")}</code></td>' |
| f'<td>{_esc(e.get("summary") or e.get("reply_type") or "—")}</td>' |
| f'</tr>' |
| ) |
| if not recent_rows: |
| recent_rows = '<tr><td colspan="5" class="empty">No activity yet</td></tr>' |
|
|
| refresh_btn = ( |
| f'<a href="?token={token}&view=dash" class="btn">⏸️ Disable auto-refresh</a>' |
| if refresh else |
| f'<a href="?token={token}&view=dash&refresh=10" class="btn">🔁 Auto-refresh (10s)</a>' |
| ) |
|
|
| maint_on, _m_reason = app._get_maintenance() |
| if maint_on: |
| maint_btn = ( |
| f'<form method="get" action="/admin" style="display:inline">' |
| f'<input type="hidden" name="token" value="{token}">' |
| f'<input type="hidden" name="action" value="maint">' |
| f'<input type="hidden" name="on" value="0">' |
| f'<button class="btn success" type="submit">🟢 Turn OFF maintenance</button>' |
| f'</form>' |
| ) |
| else: |
| maint_btn = f'<a href="?token={token}&view=maint" class="btn danger">🔴 Turn ON maintenance</a>' |
|
|
| body = f""" |
| <h1>📞 Phone Lookup Bot — Admin</h1> |
| <p class="muted">v{_esc(app.VERSION)} · Uptime: {uptime_min:.1f} min · |
| DuckDB: {"✅" if app._con_ready.is_set() else "⏳"} · |
| Warmed: {"✅" if app._warmed.is_set() else "⏳"}</p> |
| {_nav(token, "dash")} |
| {_flash(request)} |
| {cards} |
| <div class="row"> |
| {refresh_btn} |
| {maint_btn} |
| <a href="?token={token}&view=export&fmt=csv" class="btn">📥 CSV</a> |
| <a href="?token={token}&view=export&fmt=json" class="btn">📥 JSON</a> |
| </div> |
| <h2>🔴 Live Users</h2> |
| <table> |
| <thead><tr><th>User</th><th>ID</th><th>Last Active</th></tr></thead> |
| <tbody>{live_rows}</tbody> |
| </table> |
| <h2>📜 Recent Activity</h2> |
| <table> |
| <thead><tr><th>Time</th><th>User</th><th>Action</th><th>Input</th><th>Result</th></tr></thead> |
| <tbody>{recent_rows}</tbody> |
| </table> |
| <p><a href="?token={token}&view=log" class="btn">View full activity log →</a></p> |
| """ |
| 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'<tr>' |
| f'<td class="muted" style="white-space:nowrap">{_ts(e["ts"])}</td>' |
| f'<td><a href="?token={token}&view=user&uid={uid}">{_esc(nm)}</a>' |
| f'<br><span class="muted"><code>{uid}</code></span></td>' |
| f'<td>{_esc(e["action"])}</td>' |
| f'<td><code>{_esc(e.get("input", "") or "—")}</code></td>' |
| f'<td>{_esc(e.get("summary") or e.get("reply_type") or "—")}</td>' |
| f'<td class="muted">{f"{elapsed}ms" if elapsed else "—"}</td>' |
| f'<td class="muted">{_esc(e.get("source") or "—")}</td>' |
| f'</tr>' |
| ) |
| if not rows: |
| rows = '<tr><td colspan="7" class="empty">No matching entries</td></tr>' |
|
|
| 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 = '<div class="pagination">' |
| if page > 0: |
| pag += f'<a href="{page_url(page - 1)}">◀ Prev</a>' |
| pag += f'<span class="current">{page + 1} / {total_pages}</span>' |
| if page < total_pages - 1: |
| pag += f'<a href="{page_url(page + 1)}">Next ▶</a>' |
| pag += f'<span class="muted" style="margin-left:auto">{total} entries</span></div>' |
|
|
| chips = "" |
| if filter_uid: |
| chips += (f'<span class="chip active">User {filter_uid} ' |
| f'<a href="?token={token}&view=log{("&a=" + _esc(filter_action)) if filter_action else ""}">×</a></span>') |
| if filter_action: |
| chips += (f'<span class="chip active">Action: {_esc(filter_action)} ' |
| f'<a href="?token={token}&view=log{("&uid=" + _esc(filter_uid)) if filter_uid else ""}">×</a></span>') |
| if not chips: |
| chips = '<span class="muted">No filters</span>' |
|
|
| body = f""" |
| <h1>📜 Activity Log</h1> |
| {_nav(token, "log")} |
| {_flash(request)} |
| <div class="row" style="margin-bottom:12px">{chips}</div> |
| <table> |
| <thead><tr> |
| <th>Time</th><th>User</th><th>Action</th> |
| <th>Input</th><th>Result</th><th>Latency</th><th>Source</th> |
| </tr></thead> |
| <tbody>{rows}</tbody> |
| </table> |
| {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 = ' <span class="badge err">BANNED</span>' if banned else '' |
| rows += ( |
| f'<tr>' |
| f'<td><a href="?token={token}&view=user&uid={uid}">{_esc(nm)}</a>{badge}</td>' |
| f'<td><code>{uid}</code></td>' |
| f'<td>{info.get("count", 0)}</td>' |
| f'<td class="muted">{_ago(info.get("last_seen", 0))}</td>' |
| f'<td class="muted">{_ts(info.get("first_seen", 0))}</td>' |
| f'</tr>' |
| ) |
| if not rows: |
| rows = '<tr><td colspan="5" class="empty">No users yet</td></tr>' |
|
|
| pag = '<div class="pagination">' |
| if page > 0: |
| pag += f'<a href="?token={token}&view=users&page={page - 1}">◀ Prev</a>' |
| pag += f'<span class="current">{page + 1} / {total_pages}</span>' |
| if page < total_pages - 1: |
| pag += f'<a href="?token={token}&view=users&page={page + 1}">Next ▶</a>' |
| pag += f'<span class="muted" style="margin-left:auto">{total} users</span></div>' |
|
|
| body = f""" |
| <h1>👥 All Users</h1> |
| {_nav(token, "users")} |
| {_flash(request)} |
| <table> |
| <thead><tr> |
| <th>User</th><th>ID</th><th>Requests</th><th>Last Seen</th><th>First Seen</th> |
| </tr></thead> |
| <tbody>{rows}</tbody> |
| </table> |
| {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"<h1>👤 User {uid}</h1>" |
| f"{_nav(token, 'users')}" |
| f"<div class='alert error'>User not found in registry.</div>" |
| f"<a href='?token={token}&view=users' class='btn'>← Back to users</a>" |
| ) |
| 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'<tr>' |
| f'<td><a href="?token={token}&view=phone&num={phone}"><code>{phone}</code></a></td>' |
| f'<td>{h.get("count", 0)}</td>' |
| f'<td>{h.get("elapsed", 0):.1f}s</td>' |
| f'<td class="muted">{_esc(h.get("source", ""))}</td>' |
| f'<td class="muted">{_ago(h.get("ts", 0))}</td>' |
| f'</tr>' |
| ) |
| if not hist_rows: |
| hist_rows = '<tr><td colspan="5" class="empty">No searches</td></tr>' |
|
|
| 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'<tr>' |
| f'<td class="muted" style="white-space:nowrap">{_ts(e["ts"])}</td>' |
| f'<td>{_esc(e["action"])}</td>' |
| f'<td><code>{_esc(e.get("input", "") or "—")}</code></td>' |
| f'<td>{_esc(e.get("summary") or e.get("reply_type") or "—")}</td>' |
| f'</tr>' |
| ) |
| if not log_rows: |
| log_rows = '<tr><td colspan="4" class="empty">No activity</td></tr>' |
|
|
| if banned: |
| action_html = ( |
| f'<div class="alert error">🚫 Banned — reason: {_esc(reason or "no reason")}</div>' |
| f'<form method="get" action="/admin" style="display:inline">' |
| f'<input type="hidden" name="token" value="{token}">' |
| f'<input type="hidden" name="action" value="unban">' |
| f'<input type="hidden" name="uid" value="{uid}">' |
| f'<button type="submit" class="btn success">✅ Unban user</button></form>' |
| ) |
| else: |
| action_html = ( |
| f'<form method="get" action="/admin">' |
| f'<input type="hidden" name="token" value="{token}">' |
| f'<input type="hidden" name="action" value="ban">' |
| f'<input type="hidden" name="uid" value="{uid}">' |
| f'<div class="row">' |
| f'<input type="text" name="reason" placeholder="Ban reason (optional)">' |
| f'<button type="submit" class="btn danger">🚫 Ban user</button>' |
| f'</div></form>' |
| ) |
|
|
| body = f""" |
| <h1>👤 User Detail</h1> |
| {_nav(token, "users")} |
| {_flash(request)} |
| <div class="cards"> |
| <div class="card"><div class="v">{_esc(nm)}</div><div class="l">Name</div></div> |
| <div class="card"><div class="v">{uid}</div><div class="l">Telegram ID</div></div> |
| <div class="card"><div class="v">{info.get("count", 0)}</div><div class="l">Requests</div></div> |
| <div class="card"><div class="v">{_ago(info.get("last_seen", 0))}</div><div class="l">Last Seen</div></div> |
| <div class="card"><div class="v">{_ts(info.get("first_seen", 0))}</div><div class="l">First Seen</div></div> |
| </div> |
| {action_html} |
| <h2>🕘 Recent Searches</h2> |
| <table> |
| <thead><tr><th>Phone</th><th>Results</th><th>Latency</th><th>Source</th><th>When</th></tr></thead> |
| <tbody>{hist_rows}</tbody> |
| </table> |
| <h2>📜 Activity (last 20)</h2> |
| <table> |
| <thead><tr><th>Time</th><th>Action</th><th>Input</th><th>Result</th></tr></thead> |
| <tbody>{log_rows}</tbody> |
| </table> |
| <p> |
| <a href="?token={token}&view=log&uid={uid}" class="btn">View all logs for this user →</a> |
| <a href="?token={token}&view=users" class="btn">← Back to users</a> |
| </p> |
| """ |
| 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'<tr>' |
| f'<td><a href="?token={token}&view=user&uid={uid}">{_esc(nm)}</a></td>' |
| f'<td><code>{uid}</code></td>' |
| f'<td>{_esc(reason)}</td>' |
| f'<td>' |
| f'<form method="get" action="/admin" style="display:inline">' |
| f'<input type="hidden" name="token" value="{token}">' |
| f'<input type="hidden" name="action" value="unban">' |
| f'<input type="hidden" name="uid" value="{uid}">' |
| f'<button type="submit" class="btn success">Unban</button>' |
| f'</form>' |
| f'</td>' |
| f'</tr>' |
| ) |
| if not rows: |
| rows = '<tr><td colspan="4" class="empty">No banned users</td></tr>' |
|
|
| body = f""" |
| <h1>🚫 Banned Users ({len(banned)})</h1> |
| {_nav(token, "banned")} |
| {_flash(request)} |
| <table> |
| <thead><tr><th>User</th><th>ID</th><th>Reason</th><th>Action</th></tr></thead> |
| <tbody>{rows}</tbody> |
| </table> |
| """ |
| 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'<tr>' |
| f'<td>{i}</td>' |
| f'<td><a href="?token={token}&view=phone&num={_esc(num)}"><code>{_esc(num)}</code></a></td>' |
| f'<td>{cnt}</td>' |
| f'</tr>' |
| ) |
| if not rows: |
| rows = '<tr><td colspan="3" class="empty">No searches yet</td></tr>' |
|
|
| body = f""" |
| <h1>🔍 Top Searched Numbers</h1> |
| {_nav(token, "top")} |
| <table> |
| <thead><tr><th>#</th><th>Phone</th><th>Searches</th></tr></thead> |
| <tbody>{rows}</tbody> |
| </table> |
| """ |
| 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""" |
| <form method="get" action="/admin"> |
| <input type="hidden" name="token" value="{token}"> |
| <input type="hidden" name="view" value="phone"> |
| <div class="row"> |
| <input type="text" name="num" value="{_esc(num)}" |
| placeholder="10-digit number" autofocus> |
| <button type="submit" class="btn primary">Search</button> |
| </div> |
| </form> |
| """ |
|
|
| if not num: |
| body = f""" |
| <h1>☎️ Phone Lookup</h1> |
| {_nav(token, "phone")} |
| {search_form} |
| <p class="muted">Enter a phone number to see who searched it and recent activity.</p> |
| """ |
| 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'<tr>' |
| f'<td><a href="?token={token}&view=user&uid={uid}">{_esc(nm)}</a></td>' |
| f'<td><code>{uid}</code></td>' |
| f'<td>{c}</td>' |
| f'</tr>' |
| ) |
| if not searcher_rows: |
| searcher_rows = '<tr><td colspan="3" class="empty">No searchers</td></tr>' |
|
|
| 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'<tr>' |
| f'<td class="muted" style="white-space:nowrap">{_ts(e["ts"])}</td>' |
| f'<td><a href="?token={token}&view=user&uid={e["user_id"]}">{_esc(nm)}</a></td>' |
| f'<td>{_esc(e.get("summary") or e.get("reply_type") or "—")}</td>' |
| f'</tr>' |
| ) |
| if not log_rows: |
| log_rows = '<tr><td colspan="3" class="empty">No searches for this number</td></tr>' |
|
|
| body = f""" |
| <h1>☎️ Phone: <code>{_esc(num)}</code></h1> |
| {_nav(token, "phone")} |
| {_flash(request)} |
| <div class="cards"> |
| <div class="card"><div class="v">{count}</div><div class="l">Total Searches</div></div> |
| <div class="card"><div class="v">{len(searchers)}</div><div class="l">Unique Searchers</div></div> |
| </div> |
| {search_form} |
| <h2>Who searched this number</h2> |
| <table> |
| <thead><tr><th>User</th><th>ID</th><th>Searches</th></tr></thead> |
| <tbody>{searcher_rows}</tbody> |
| </table> |
| <h2>Recent searches (last 30)</h2> |
| <table> |
| <thead><tr><th>Time</th><th>User</th><th>Result</th></tr></thead> |
| <tbody>{log_rows}</tbody> |
| </table> |
| """ |
| return HTMLResponse(_page(f"Admin · Phone {num}", body)) |
|
|
|
|
| async def _view_maint_confirm(app, token, request): |
| body = f""" |
| <h1>🔴 Enable Maintenance Mode</h1> |
| {_nav(token, "dash")} |
| <div class="alert warn"> |
| This will block all non-admin users from searching. Basic commands |
| (/start, /status, /info, /myid) will still work. |
| </div> |
| <form method="get" action="/admin"> |
| <input type="hidden" name="token" value="{token}"> |
| <input type="hidden" name="action" value="maint"> |
| <input type="hidden" name="on" value="1"> |
| <div class="row"> |
| <input type="text" name="reason" placeholder="Reason (optional)"> |
| <button type="submit" class="btn danger">Enable maintenance</button> |
| <a href="?token={token}&view=dash" class="btn">Cancel</a> |
| </div> |
| </form> |
| """ |
| 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"'}, |
| ) |
|
|
|
|
| |
| |
| |
|
|
| 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, |
| ) |
|
|
|
|
| |
| |
| |
| async def admin_dashboard(request: Request): |
| app = _app() |
| if app is None: |
| return HTMLResponse( |
| "<h1>500 — Bot module not found</h1>" |
| "<p>admin_panel.py could not locate the running bot module. " |
| "Make sure app.py is imported (or running as __main__).</p>", |
| status_code=500, |
| ) |
|
|
| token = request.query_params.get("token", "") |
| if not app.WEBHOOK_SECRET or token != app.WEBHOOK_SECRET: |
| return HTMLResponse( |
| "<h1>403 — Invalid or missing token</h1>" |
| "<p>Append <code>?token=<WEBHOOK_SECRET></code> to the URL.</p>", |
| 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 = ( |
| "<h1>500 — Admin panel error</h1>" |
| f"<pre style='color:#ff9f9f;background:#0f1419;padding:16px;" |
| f"border-radius:8px;overflow:auto;border:1px solid #5c2424;'>" |
| f"{_esc(str(e))}\n\n{_esc(traceback.format_exc())}</pre>" |
| f"<a href='?token={token}' class='btn'>← Back to dashboard</a>" |
| ) |
| return HTMLResponse(_page("Error", err_html), status_code=500) |
