"""Run history + user dashboard + admin analytics (SQLite, same DB as accounts)."""
import csv
import html
import os
import tempfile
import time
from pathlib import Path

import auth as A
import db as _dbmod

# every tool that writes to the run history (charts, the admin's tool filter and the tool table follow this list)
TOOLS = ["Smart Humanize", "AI Detector", "Humanizer", "Proofread", "Journal Formatter", "Journal Finder"]
TOOL_COLORS = {"Smart Humanize": "#47148B", "AI Detector": "#50B4EF", "Humanizer": "#E92E7F", "Grammar": "#50BF49",
               "Proofread": "#50BF49", "Journal Formatter": "#FE5828", "Journal Finder": "#F5C518"}
SEARCH_TOOLS = ("Journal Finder",)                       # runs that are a search, not a text or a document
RETENTION_DAYS = int(os.getenv("HISTORY_DAYS", "180"))


def _init(con):
    if _dbmod.IS_MYSQL:
        return con
    con.execute("""CREATE TABLE IF NOT EXISTS runs (
        id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT DEFAULT CURRENT_TIMESTAMP, who TEXT, email TEXT,
        tool TEXT, kind TEXT, name TEXT, words INTEGER, ai_before INTEGER, ai_after INTEGER,
        secs REAL, status TEXT, output TEXT, engine TEXT)""")
    con.execute("CREATE INDEX IF NOT EXISTS runs_email ON runs(email, ts)")
    con.execute("CREATE INDEX IF NOT EXISTS runs_ts ON runs(ts)")
    return con


def db():
    return _init(A._db())


# ---------------------------------------------------------------- logging
def log_run(user, request, tool, document, meta, secs, status, error=""):
    try:
        with db() as con:
            con.execute(
                "INSERT INTO runs (who, email, tool, kind, name, words, ai_before, ai_after, secs, status, "
                "output, engine) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
                (A._who(user, request), (user or {}).get("email"), tool, "doc" if document else "text",
                 (meta.get("name") or ("Pasted text" if not document else ""))[:120], meta.get("words"),
                 meta.get("before"), meta.get("after"), round(secs, 1),
                 status if not error else "error: " + error.strip("\x27\" ")[:150], meta.get("output"), meta.get("engine")))
            con.execute(f"DELETE FROM runs WHERE ts < datetime('now','-{RETENTION_DAYS} days')")
    except Exception as err:
        print(f"[dashboard] log failed: {err}")


# ---------------------------------------------------------------- helpers
def _tiles(items):
    cells = "".join(f'<div class="cs-stat"><div class="n">{v}</div><div class="l">{k}</div></div>'
                    for k, v in items)
    return f'<div class="cs-stats">{cells}</div>'


def _fmt_pct(v):
    return "—" if v is None else f"{v}%"


def _daily(con, where, params, days):
    """Rows for a stacked bar chart: one row per (day, tool) with zero-filled days."""
    import pandas as pd
    rows = con.execute(
        f"SELECT date(ts) d, tool, COUNT(*) FROM runs WHERE ts >= date('now', ?) {where} GROUP BY d, tool",
        (f"-{days - 1} days", *params)).fetchall()
    got = {(d, t): n for d, t, n in rows}
    today = time.gmtime()
    out = []
    for i in range(days - 1, -1, -1):
        d = time.strftime("%Y-%m-%d", time.gmtime(time.mktime(today) - i * 86400))
        for t in TOOLS:
            out.append({"Day": d[5:], "Tool": t, "Runs": got.get((d, t), 0)})
    return pd.DataFrame(out)


HIST_COLS = ["When (UTC)", "Tool", "Input", "Type", "Words", "AI before", "AI after", "Time (s)", "Status"]


def _clean(t):
    import privacy
    return privacy.clean(t)


def _hist_rows(rows):
    return [[(ts or "")[:16], tool, name or "—",
             "Search" if tool in SEARCH_TOOLS else "Document" if kind == "doc" else "Text",
             words if words is not None else "—", _fmt_pct(b), _fmt_pct(a), secs,
             "✅ OK" if status == "ok" else "⚠️ " + (_clean((status or "")[7:200]) or "Could not finish")[:80]]
            for ts, tool, name, kind, words, b, a, secs, status in rows]


# ---------------------------------------------------------------- pay per use (plans.charges)
def _charges(where="", params=()):
    """Rows of the pay-per-use charges table ([] when it does not exist yet or cannot be read)."""
    try:
        import plans as P
        with P._db() as con:
            return con.execute("SELECT ref, ts, email, tool, name, words, cost_usd, amount_inr, status, paid_ts "
                               f"FROM charges {where} ORDER BY id DESC", params).fetchall()
    except Exception as err:
        print(f"[dashboard] charges not read: {err}")
        return []


def _pending_html(email):
    """Results of this customer that are waiting for payment, each with its price, reference and a Pay link."""
    rows = _charges("WHERE email=? AND status='pending'", ((email or "").lower(),))[:5]
    if not rows:
        return ""
    import plans as P
    items = []
    for ref, ts, em, tool, name, words, cost, inr, status, paid in rows:
        link = html.escape(P.pay_link(dict(ref=ref, amount_inr=inr or 0, tool=tool, name=name, email=em)), quote=True)
        items.append(f'<div class="cs-nudge">🔒 <b>{html.escape(str(name or "Document"))}</b> ({html.escape(tool or "")}, '
                     f'{(ts or "")[:10]}) is waiting for payment: <b>₹{(inr or 0):,.0f}</b>, reference '
                     f'<b>{html.escape(ref)}</b>. <a class="cs-up-link" href="{link}" target="_blank" '
                     f'rel="noopener">Pay now</a> — then open the tool\'s tab and press “I have paid: unlock”.</div>')
    return "".join(items)


# ---------------------------------------------------------------- user dashboard
def user_dashboard(user):
    """(tiles_html, chart_df, history_rows, file_choices)"""
    import pandas as pd
    if not user:
        return ("", pd.DataFrame(columns=["Day", "Tool", "Runs"]), [], [])
    e = user["email"]
    with db() as con:
        tot, docs, texts, ok = con.execute(
            "SELECT COUNT(*), SUM(kind='doc'), SUM(kind='text' AND tool!='Journal Finder'), SUM(status='ok') "
            "FROM runs WHERE email=?", (e,)).fetchone()
        formatted, searches, proofs = con.execute(
            "SELECT SUM(tool='Journal Formatter' AND status='ok'), SUM(tool='Journal Finder' AND status='ok'), "
            "SUM(tool='Proofread' AND status='ok') FROM runs WHERE email=?", (e,)).fetchone()
        red = con.execute(
            "SELECT AVG(ai_before - ai_after), SUM(words) FROM runs WHERE email=? AND status='ok' "
            "AND ai_before IS NOT NULL AND ai_after IS NOT NULL", (e,)).fetchone()
        words_all = con.execute("SELECT COALESCE(SUM(words),0) FROM runs WHERE email=? AND status='ok'",
                                (e,)).fetchone()[0]
        chart = _daily(con, "AND email=?", (e,), 14)
        hist = con.execute("SELECT ts, tool, name, kind, words, ai_before, ai_after, secs, status FROM runs "
                           "WHERE email=? ORDER BY ts DESC, id DESC LIMIT 100", (e,)).fetchall()
        files = con.execute("SELECT id, ts, tool, name, output FROM runs WHERE email=? AND status='ok' "
                            "AND output IS NOT NULL ORDER BY id DESC LIMIT 40", (e,)).fetchall()
    who = A._who(user, None)
    lim = A.user_limits(e)
    used = []
    for kind, label in (("doc", "docs"), ("text", "texts")):
        n = A.usage_today(who, kind)
        used.append(f"{n}/{lim[kind] if lim[kind] > 0 else '∞'} {label}")
    tiles = _tiles([
        ("Total runs", f"{tot or 0:,}"),
        ("Documents", f"{docs or 0:,}"),
        ("Pasted texts", f"{texts or 0:,}"),
        ("Words processed", f"{words_all or 0:,}"),
        ("Avg AI reduction", "—" if red[0] is None else f"{red[0]:.0f} pts"),
        ("Proofreads", f"{proofs or 0:,}"),
        ("Papers formatted", f"{formatted or 0:,}"),
        ("Journal searches", f"{searches or 0:,}"),
        ("Used today", " · ".join(used) if not A.is_admin(user) else "Unlimited"),
    ]) + _pending_html(e)
    choices = [(f"#{i} · {ts[:16]} · {tool} · {name or 'Pasted text'}", str(i))
               for i, ts, tool, name, out in files if out and Path(out).exists()]
    return tiles, chart, _hist_rows(hist), choices


def user_file(user, run_id):
    if not user or not run_id:
        return None
    with db() as con:
        row = con.execute("SELECT output FROM runs WHERE id=? AND email=?", (int(run_id), user["email"])).fetchone()
    return row[0] if row and row[0] and Path(row[0]).exists() else None


# ---------------------------------------------------------------- admin analytics
def admin_overview(user):
    """(tiles_html, daily_chart_df, tool_df, top_users_rows)"""
    A._need_admin(user)
    import pandas as pd
    with db() as con:
        users = con.execute("SELECT COUNT(*), SUM(active=1), SUM(created >= datetime('now','-7 days')) "
                            "FROM users").fetchone()
        today = con.execute("SELECT COUNT(*), SUM(email IS NULL), SUM(status!='ok') FROM runs "
                            "WHERE date(ts)=date('now')").fetchone()
        week = con.execute("SELECT COUNT(*), COUNT(DISTINCT who), SUM(status!='ok'), "
                           "AVG(CASE WHEN status='ok' THEN secs END) FROM runs "
                           "WHERE ts >= datetime('now','-7 days')").fetchone()
        red = con.execute("SELECT AVG(ai_before - ai_after) FROM runs WHERE status='ok' AND tool!='AI Detector' "
                          "AND ai_before IS NOT NULL AND ai_after IS NOT NULL "
                          "AND ts >= datetime('now','-30 days')").fetchone()[0]
        cpu = con.execute("SELECT COUNT(*) FROM runs WHERE engine='CPU' AND ts >= datetime('now','-7 days')"
                          ).fetchone()[0]
        chart = _daily(con, "", (), 30)
        tools = con.execute("SELECT tool, kind, COUNT(*) FROM runs WHERE ts >= datetime('now','-30 days') "
                            "GROUP BY tool, kind").fetchall()
        top = con.execute(
            "SELECT r.email, u.name, COUNT(*), SUM(r.kind='doc'), COALESCE(SUM(r.words),0), MAX(r.ts) FROM runs r "
            "LEFT JOIN users u ON u.email=r.email WHERE r.email IS NOT NULL AND r.ts >= datetime('now','-30 days') "
            "GROUP BY r.email ORDER BY COUNT(*) DESC LIMIT 15").fetchall()
    by_tool = {}
    for t, k, n in tools:
        by_tool[t] = by_tool.get(t, 0) + n
    pending = _charges("WHERE status='pending'")
    paid = _charges("WHERE status='paid' AND paid_ts >= datetime('now','-30 days')")
    tiles = _tiles([
        ("Users", f"{users[0] or 0:,}"), ("Active accounts", f"{users[1] or 0:,}"),
        ("New users (7d)", f"{users[2] or 0:,}"), ("Runs today", f"{today[0] or 0:,}"),
        ("Guest runs today", f"{today[1] or 0:,}"), ("Runs (7d)", f"{week[0] or 0:,}"),
        ("Unique visitors (7d)", f"{week[1] or 0:,}"), ("Errors (7d)", f"{week[2] or 0:,}"),
        ("Avg job time (7d)", "—" if week[3] is None else f"{week[3]:.0f}s"),
        ("Avg AI reduction (30d)", "—" if red is None else f"{red:.0f} pts"),
        ("Runs on CPU (7d)", f"{cpu:,}"),
        ("Papers formatted (30d)", f"{by_tool.get('Journal Formatter', 0):,}"),
        ("Journal searches (30d)", f"{by_tool.get('Journal Finder', 0):,}"),
        ("Proofreads (30d)", f"{by_tool.get('Proofread', 0):,}"),
        ("Pay-per-use awaiting payment", f"{len(pending):,} · ₹{sum(r[7] or 0 for r in pending):,.0f}"),
        ("Pay-per-use paid (30d)", f"{len(paid):,} · ₹{sum(r[7] or 0 for r in paid):,.0f}"),
        ("Pay-per-use cost (30d)", f"${sum(r[6] or 0 for r in paid):,.2f}"),
    ])
    tdf = pd.DataFrame([{"Tool": t, "Type": "Search" if t in SEARCH_TOOLS else "Document" if k == "doc" else "Text",
                         "Runs": n} for t, k, n in tools] or [{"Tool": t, "Type": "Text", "Runs": 0} for t in TOOLS])
    top_rows = [[name or "—", e, n, d or 0, f"{w:,}", (last or "")[:16]] for e, name, n, d, w, last in top]
    return tiles, chart, tdf, top_rows


ACT_COLS = ["When (UTC)", "User", "Tool", "Input", "Type", "Words", "AI before", "AI after", "Time (s)",
            "Engine", "Status"]


def _mask(who, email):
    if email:
        return email
    ip = (who or "")[3:]
    parts = ip.split(".")
    return "guest " + (".".join(parts[:3]) + ".x" if len(parts) == 4 else ip[:12] + "…")


def admin_activity(user, query="", tool="All", status="All", limit=300):
    A._need_admin(user)
    where, params = ["1=1"], []
    q = (query or "").strip().lower()
    if q:
        where.append("(lower(COALESCE(email,'')) LIKE ? OR lower(COALESCE(name,'')) LIKE ? OR who LIKE ?)")
        params += [f"%{q}%"] * 3
    if tool and tool != "All":
        where.append("tool=?")
        params.append(tool)
    if status == "Errors only":
        where.append("status!='ok'")
    elif status == "Guests only":
        where.append("email IS NULL")
    with db() as con:
        rows = con.execute(
            f"SELECT ts, who, email, tool, name, kind, words, ai_before, ai_after, secs, engine, status FROM runs "
            f"WHERE {' AND '.join(where)} ORDER BY ts DESC, id DESC LIMIT {int(limit)}", params).fetchall()
    return [[(ts or "")[:16], _mask(who, e), t, n or "—",
             "Search" if t in SEARCH_TOOLS else "Document" if k == "doc" else "Text",
             w if w is not None else "—", _fmt_pct(b), _fmt_pct(a), s, eng or "—",
             "✅ OK" if st == "ok" else "⚠️ " + (st or "")[7:80]]
            for ts, who, e, t, n, k, w, b, a, s, eng, st in rows]


def admin_export_activity(user):
    A._need_admin(user)
    folder = Path(tempfile.mkdtemp(dir=A.OUT_DIR))
    path = folder / "activity.csv"
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(ACT_COLS)
        w.writerows(admin_activity(user, limit=100000))
    return str(path)


def admin_user_detail(user, email):
    """(summary_html, history_rows) for one user."""
    A._need_admin(user)
    email = (email or "").strip().lower()
    if not email:
        return '<div class="cs-authmsg cs-bad">Enter a user email.</div>', []
    with db() as con:
        u = con.execute("SELECT name, created, last_login, active, doc_limit, text_limit FROM users WHERE email=?",
                        (email,)).fetchone()
        if not u:
            return f'<div class="cs-authmsg cs-bad">No user found with {html.escape(email)}.</div>', []
        s = con.execute("SELECT COUNT(*), SUM(kind='doc'), SUM(kind='text'), COALESCE(SUM(words),0), "
                        "SUM(status!='ok') FROM runs WHERE email=?", (email,)).fetchone()
        hist = con.execute("SELECT ts, tool, name, kind, words, ai_before, ai_after, secs, status FROM runs "
                           "WHERE email=? ORDER BY ts DESC, id DESC LIMIT 200", (email,)).fetchall()
    lim = A.user_limits(email)
    fmt = lambda v, custom: ("∞" if v <= 0 else str(v)) + (" (custom)" if custom is not None else "")
    summary = (f'<div class="cs-auth-title" style="text-align:left">{html.escape(u[0] or "")} '
               f'<small style="color:#888;font-weight:500">{html.escape(email)}</small></div>'
               + _tiles([("Joined", (u[1] or "")[:10]), ("Last login", (u[2] or "")[:16]),
                         ("Status", "Active" if u[3] else "Disabled"), ("Runs", f"{s[0] or 0:,}"),
                         ("Docs / texts", f"{s[1] or 0} / {s[2] or 0}"), ("Words", f"{s[3]:,}"),
                         ("Errors", f"{s[4] or 0}"),
                         ("Daily limits (doc / text)", f"{fmt(lim['doc'], u[4])} / {fmt(lim['text'], u[5])}")]))
    return summary, _hist_rows(hist)