import os import logging import gradio as gr from db import models as dbm from admin.auth import check_admin_password from admin.exports import export_json, export_csv, export_markdown from utils.quota import quota_snapshot log = logging.getLogger("admin.panel") ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "tom123") def _gate(supplied: str) -> bool: return check_admin_password(supplied or "", ADMIN_PASSWORD) def build_admin_tabs(admin_password: str = None): global ADMIN_PASSWORD if admin_password: ADMIN_PASSWORD = admin_password # --- Login gate --- with gr.Row(): pw = gr.Textbox(label="Admin password", type="password", scale=3) login_btn = gr.Button("Login", scale=1, variant="primary") status_md = gr.Markdown("🔒 Locked. Enter password above.") with gr.Column(visible=False) as admin_area: with gr.Tabs(): # ============ Dashboard ============ with gr.Tab("📊 Dashboard"): dash_md = gr.Markdown() refresh_dash = gr.Button("🔄 Refresh") def _dash(): s = dbm.stats() q = quota_snapshot() return ( f"### Stats\n" f"- Total users: **{s['total_users']}**\n" f"- Active users: **{s['active_users']}**\n" f"- Total messages: **{s['total_messages']}**\n" f"- Messages today: **{s['messages_today']}**\n\n" f"### Quota\n" f"- GPU seconds today: **{q['seconds_used']:.1f}s** / {q['seconds_limit']}s\n" f"- Turns today: **{q['turns_today']}**" ) refresh_dash.click(_dash, outputs=dash_md, queue=False, show_progress=False) # ============ Users ============ with gr.Tab("👥 Users"): users_table = gr.Dataframe( headers=["user_id", "username", "token", "active", "created_at", "last_seen", "turns"], interactive=False, wrap=True, ) def _users_rows(): return [[u["user_id"], u.get("username"), u.get("token"), bool(u.get("is_active")), u.get("created_at"), u.get("last_seen"), u.get("total_turns")] for u in dbm.list_users()] refresh_users = gr.Button("🔄 Refresh users") refresh_users.click(_users_rows, outputs=users_table, queue=False, show_progress=False) gr.Markdown("### Generate a token") with gr.Row(): new_uid = gr.Number(label="Assign to Telegram user_id (blank = unassigned)", precision=0, scale=3) gen_btn = gr.Button("🎟️ Generate", variant="primary", scale=1) gen_out = gr.Textbox(label="Generated token", interactive=False) def _gen(uid): uid_int = int(uid) if uid else None return dbm.generate_token(uid_int) gen_btn.click(_gen, inputs=[new_uid], outputs=gen_out, queue=False, show_progress=False) gr.Markdown("### Manage users") with gr.Row(): revoke_uid = gr.Number(label="Revoke user_id", precision=0, scale=2) revoke_btn = gr.Button("🚫 Revoke", scale=1) revoke_out = gr.Markdown() def _revoke(uid): if not uid: return "Enter a user_id." dbm.revoke_user(int(uid)) return f"Revoked {int(uid)}." revoke_btn.click(_revoke, inputs=[revoke_uid], outputs=revoke_out, queue=False, show_progress=False) with gr.Row(): del_uid = gr.Number(label="Delete user_id", precision=0, scale=2) del_btn = gr.Button("🗑️ Delete (permanent)", variant="stop", scale=1) del_out = gr.Markdown() def _del(uid): if not uid: return "Enter a user_id." dbm.delete_user(int(uid)) return f"Deleted user {int(uid)}." del_btn.click(_del, inputs=[del_uid], outputs=del_out, queue=False, show_progress=False) # ============ Conversations (live context) ============ with gr.Tab("💬 Conversations"): gr.Markdown( "_Live context only — cleared by user `/reset`._ " "See **📁 Sessions** tab for the permanent archive." ) with gr.Row(): conv_uid = gr.Number(label="Filter user_id (blank = all)", precision=0, scale=2) conv_limit = gr.Number(label="Limit", value=200, precision=0, scale=1) conv_btn = gr.Button("🔍 Load", variant="primary", scale=1) conv_table = gr.Dataframe( headers=["id", "user_id", "role", "content", "timestamp", "latency_ms", "gpu_time_ms"], interactive=False, wrap=True, ) def _load_conv(uid, lim): uid_int = int(uid) if uid else None lim_int = int(lim) if lim else 200 rows = dbm.list_conversations(uid_int, lim_int) return [[r["id"], r["user_id"], r["role"], (r["content"][:300] + "…" if len(r["content"]) > 300 else r["content"]), r["timestamp"], r.get("latency_ms"), r.get("gpu_time_ms")] for r in rows] conv_btn.click(_load_conv, inputs=[conv_uid, conv_limit], outputs=conv_table, queue=False, show_progress=False) gr.Markdown("### Export (live + full archive)") with gr.Row(): exp_json_btn = gr.Button("📦 JSON") exp_csv_btn = gr.Button("📊 CSV") exp_md_btn = gr.Button("📝 Markdown") exp_file = gr.File(label="Download export", interactive=False) def _write_json(): p = "/tmp/export.json" with open(p, "w", encoding="utf-8") as f: f.write(export_json()) return p def _write_csv(): p = "/tmp/export.csv" with open(p, "w", encoding="utf-8") as f: f.write(export_csv()) return p def _write_md(): p = "/tmp/export.md" with open(p, "w", encoding="utf-8") as f: f.write(export_markdown()) return p exp_json_btn.click(_write_json, outputs=exp_file, queue=False, show_progress=False) exp_csv_btn.click(_write_csv, outputs=exp_file, queue=False, show_progress=False) exp_md_btn.click(_write_md, outputs=exp_file, queue=False, show_progress=False) # ============ Sessions (per-session markdown archive) ============ with gr.Tab("📁 Sessions"): gr.Markdown( "**Permanent archive.** One markdown file per session, " "stored at `/data/chat_logs/{user_id}/{timestamp}.md`. " "User `/reset` closes a session and starts a new file — " "**nothing is ever deleted** except by admin action here." ) with gr.Row(): sess_uid = gr.Number(label="Filter by user_id (blank = all)", precision=0, scale=2) sess_refresh = gr.Button("🔄 Refresh sessions", variant="primary", scale=1) sess_table = gr.Dataframe( headers=["user_id", "filename", "size_bytes", "modified", "path"], interactive=False, wrap=True, ) def _sess_rows(uid): uid_int = int(uid) if uid else None rows = dbm.archive_list(uid_int) return [[r["user_id"], r["filename"], r["size"], r["modified"], r["path"]] for r in rows] sess_refresh.click(_sess_rows, inputs=[sess_uid], outputs=sess_table, queue=False, show_progress=False) gr.Markdown("### View a session") with gr.Row(): view_uid = gr.Number(label="User ID", precision=0, scale=1) view_file = gr.Textbox(label="Filename (e.g. 20261016T123045.md)", scale=3) view_btn = gr.Button("📖 Load", variant="primary", scale=1) view_code = gr.Code(label="Session transcript", language="markdown", lines=25, interactive=False) def _view(uid, fn): if not uid or not fn: return "" return dbm.archive_read(int(uid), fn.strip()) view_btn.click(_view, inputs=[view_uid, view_file], outputs=view_code, queue=False, show_progress=False) gr.Markdown("### Download full transcript for a user") with gr.Row(): dl_uid = gr.Number(label="User ID", precision=0, scale=2) dl_btn = gr.Button("📦 Build combined .md", scale=1) dl_file = gr.File(label="Combined transcript", interactive=False) def _build_combined(uid): if not uid: return None p = f"/tmp/transcript_{int(uid)}.md" with open(p, "w", encoding="utf-8") as f: f.write(dbm.archive_export_user(int(uid))) return p dl_btn.click(_build_combined, inputs=[dl_uid], outputs=dl_file, queue=False, show_progress=False) gr.Markdown("### Manage sessions") with gr.Row(): del_uid = gr.Number(label="Delete one file — user_id", precision=0, scale=1) del_file = gr.Textbox(label="Filename", scale=3) del_btn = gr.Button("🗑️ Delete file", variant="stop", scale=1) del_out = gr.Markdown() def _del_file(uid, fn): if not uid or not fn: return "Enter user_id and filename." ok = dbm.archive_delete(int(uid), fn.strip()) return ("✅ Deleted." if ok else "⚠️ File not found.") del_btn.click(_del_file, inputs=[del_uid, del_file], outputs=del_out, queue=False, show_progress=False) with gr.Row(): del_user_uid = gr.Number(label="Delete ALL files for user_id", precision=0, scale=2) del_user_btn = gr.Button("🗑️ Delete user's sessions", variant="stop", scale=1) del_user_out = gr.Markdown() def _del_user(uid): if not uid: return "Enter a user_id." n = dbm.archive_delete_user(int(uid)) return f"🗑️ Deleted {n} session file(s) for {int(uid)}." del_user_btn.click(_del_user, inputs=[del_user_uid], outputs=del_user_out, queue=False, show_progress=False) gr.Markdown("---") gr.Markdown("### ⚠️ Danger zone") clear_all_btn = gr.Button("🗑️ Delete ALL sessions for ALL users", variant="stop") clear_all_out = gr.Markdown() def _clear_all(): n = dbm.archive_delete_all() return f"🗑️ Deleted {n} session file(s) total." clear_all_btn.click(_clear_all, outputs=clear_all_out, queue=False, show_progress=False) # ============ Tokens ============ with gr.Tab("🎟️ Tokens"): tok_table = gr.Dataframe( headers=["token", "assigned_to", "active", "created_at", "turns_used", "last_used"], interactive=False, wrap=True, ) def _tok_rows(): return [[t["token"], t.get("assigned_to"), bool(t.get("is_active")), t.get("created_at"), t.get("turns_used"), t.get("last_used")] for t in dbm.list_tokens()] refresh_tok = gr.Button("🔄 Refresh tokens") refresh_tok.click(_tok_rows, outputs=tok_table, queue=False, show_progress=False) gr.Markdown("### Bulk generate") with gr.Row(): bulk_count = gr.Number(label="Count", value=5, precision=0, scale=1) bulk_btn = gr.Button("Generate batch", variant="primary", scale=1) bulk_out = gr.Textbox(label="New tokens (one per line)", lines=8, interactive=False) def _bulk(n): n = int(n) if n else 0 return "\n".join(dbm.generate_token(None) for _ in range(n)) bulk_btn.click(_bulk, inputs=[bulk_count], outputs=bulk_out, queue=False, show_progress=False) gr.Markdown("### Revoke a token") with gr.Row(): revoke_tok = gr.Textbox(label="Token", scale=3) revoke_tok_btn = gr.Button("🚫 Revoke", variant="stop", scale=1) revoke_tok_out = gr.Markdown() def _revoke_tok(t): if not t: return "Enter a token." dbm.revoke_token(t.strip()) return "✅ Revoked." revoke_tok_btn.click(_revoke_tok, inputs=[revoke_tok], outputs=revoke_tok_out, queue=False, show_progress=False) # ============ Settings ============ with gr.Tab("⚙️ Settings") as settings_tab: gr.Markdown("Changes apply immediately. Use **🔄 Reload from DB** if values look stale.") window = gr.Number(label="conversation_window (turns)", value=6, precision=0) max_new = gr.Number(label="max_new_tokens", value=1024, precision=0) temp = gr.Number(label="temperature", value=0.7) topp = gr.Number(label="top_p", value=0.9) sys_prompt = gr.Textbox(label="system_prompt", value="", lines=10) daily_limit = gr.Number(label="daily_turn_limit", value=100, precision=0) with gr.Row(): save_settings = gr.Button("💾 Save settings", variant="primary") reload_settings = gr.Button("🔄 Reload from DB") settings_out = gr.Markdown() def _load_settings(): return ( int(dbm.get_setting("conversation_window", "6")), int(dbm.get_setting("max_new_tokens", "1024")), float(dbm.get_setting("temperature", "0.7")), float(dbm.get_setting("top_p", "0.9")), dbm.get_setting("system_prompt", "You are a helpful assistant."), int(dbm.get_setting("daily_turn_limit", "100")), ) def _save(w, mn, t, tp, sp, dl): dbm.set_setting("conversation_window", int(w)) dbm.set_setting("max_new_tokens", int(mn)) dbm.set_setting("temperature", float(t)) dbm.set_setting("top_p", float(tp)) dbm.set_setting("system_prompt", sp) dbm.set_setting("daily_turn_limit", int(dl)) import datetime as _d return (f"✅ Saved at {_d.datetime.now().strftime('%H:%M:%S')}. " f"Prompt length: {len(sp)} chars.") save_settings.click( _save, inputs=[window, max_new, temp, topp, sys_prompt, daily_limit], outputs=settings_out, queue=False, show_progress=False, ) reload_settings.click( _load_settings, outputs=[window, max_new, temp, topp, sys_prompt, daily_limit], queue=False, show_progress=False, ) settings_tab.select( _load_settings, outputs=[window, max_new, temp, topp, sys_prompt, daily_limit], queue=False, show_progress=False, ) # ============ Health ============ with gr.Tab("🩺 Health"): health_md = gr.Markdown() recent_gpu = gr.Dataframe( headers=["id", "user_id", "timestamp", "duration_ms", "status"], interactive=False, wrap=True, ) refresh_health = gr.Button("🔄 Refresh") def _health(): q = quota_snapshot() return (f"GPU seconds today: **{q['seconds_used']:.1f}s** / " f"{q['seconds_limit']}s\n\nTurns today: **{q['turns_today']}**") def _recent(): return [[r["id"], r.get("user_id"), r["timestamp"], r.get("duration_ms"), r.get("status")] for r in dbm.recent_gpu_log(50)] refresh_health.click(_health, outputs=health_md, queue=False, show_progress=False) refresh_health.click(_recent, outputs=recent_gpu, queue=False, show_progress=False) def _login(p): if _gate(p): return gr.update(visible=True), "✅ Logged in as admin." return gr.update(visible=False), "❌ Wrong password." login_btn.click( _login, inputs=[pw], outputs=[admin_area, status_md], queue=False, show_progress=False, )