| 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 |
|
|
| |
| 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(): |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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, |
| ) |
|
|
| |
| 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, |
| ) |
