app.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Titleist++ unified CSV-first app (interesting-only homepage)

import os, csv, sqlite3, time, threading, ipaddress
from pathlib import Path
from collections import Counter
from flask import Flask, jsonify, Response, stream_with_context

def env(name, default): return os.environ.get(name, default)

SQLITE_PATH = env("TITLEIST_DB", "ct_hits.sqlite")
HITS_CSV    = env("TITLEIST_HITS_CSV", "ct_hits.csv")
HITS_PART   = env("TITLEIST_HITS_PART", "ct_hits.csv.part")
HITS_TABLE  = env("TITLEIST_TABLE", "hit")
COL_TS      = env("COL_TS", "ts")
COL_HOST    = env("COL_HOST", "host")
COL_TLD     = env("COL_TLD", "tld")
COL_INT     = env("COL_INTERESTING", "interesting")
SQL_BATCH   = int(env("HITS_SQL_BATCH", "100000"))
HITS_CHUNK_LINES = int(env("HITS_CHUNK_LINES", "10000"))

TYPO_CSV    = env("TITLEIST_TYPO_CSV", "typosquat_bucketed.csv")
LIST_SEP    = env("TITLEIST_LIST_SEP", "|")
PORT        = int(env("PORT", "8010"))

TRUE_SET = {"true","1","t","yes","y","True","TRUE","Yes","Y"}

def _truthy(v) -> bool:
    return str(v).strip() in TRUE_SET

def _root(host: str) -> str:
    h = (host or "").strip().lower()
    if "." not in h: return h
    parts = h.split("."); return ".".join(parts[-2:])

def _cidr24(ip: str):
    try:
        ip_obj = ipaddress.ip_address(str(ip))
        if isinstance(ip_obj, ipaddress.IPv4Address):
            return ".".join(str(ip_obj).split(".")[:3]) + ".0/24"
        hextets = ip_obj.exploded.split(":")
        return ":".join(hextets[:3]) + "::/48"
    except Exception:
        return None

def _split_items(text: str):
    if not text: return []
    text = str(text)
    parts = [p.strip() for p in text.split(LIST_SEP) if p.strip()]
    if len(parts) <= 1:
        alt = "|" if LIST_SEP != "|" else ","
        p2 = [p.strip() for p in text.split(alt) if p.strip()]
        return p2 if len(p2) > len(parts) else parts
    return parts

class HitsState:
    def __init__(self):
        self.lock = threading.Lock()
        self.running = False
        self.phase = "idle"
        self.error = None
        self.rows = 0
        self.rows_interesting = 0
        self.updated = 0.0
        self.headers = []
        self.daily = Counter()
        self.tlds  = Counter()
        self.hosts = Counter()

hits = HitsState()

def _choose_hits_path() -> str:
    part = Path(HITS_PART)
    if part.exists() and part.stat().st_size > 0:
        return str(part)
    return HITS_CSV

def dump_sqlite_to_csv_part():
    path_part = Path(HITS_PART)
    path_csv  = Path(HITS_CSV)
    try:
        with hits.lock:
            hits.phase = "dumping"
            hits.error = None
        con = sqlite3.connect(SQLITE_PATH)
        cur = con.cursor()
        cols = [r[1] for r in cur.execute(f"PRAGMA table_info({HITS_TABLE})").fetchall()]
        if not cols:
            cur.execute(f"SELECT * FROM {HITS_TABLE} LIMIT 1")
            cols = [d[0] for d in cur.description]
        with open(path_part, "w", encoding="utf-8", newline="") as f:
            w = csv.writer(f)
            w.writerow(cols)
            cur.execute(f"SELECT * FROM {HITS_TABLE}")
            while True:
                rows = cur.fetchmany(SQL_BATCH)
                if not rows: break
                w.writerows(rows)
                f.flush()
        con.close()
        os.replace(path_part, path_csv)
        with hits.lock:
            hits.phase = "done"
            hits.updated = time.time()
    except Exception as e:
        with hits.lock:
            hits.phase = "error"
            hits.error = str(e)
        raise

def tail_hits_csv_live():
    last_path = None
    f = None
    rdr = None
    aliases = {
        "ts":          [COL_TS, "ts", "timestamp", "time", "datetime", "date", "created_at"],
        "host":        [COL_HOST, "host", "domain", "name", "fqdn"],
        "tld":         [COL_TLD, "tld", "suffix", "zone"],
        "interesting": [COL_INT, "interesting", "is_interesting", "flag", "label", "poi", "poi_flag"],
    }
    key = {"ts": None, "host": None, "tld": None, "interesting": None}

    with hits.lock:
        hits.phase = "processing"
        hits.rows = 0
        hits.rows_interesting = 0
        hits.daily.clear(); hits.tlds.clear(); hits.hosts.clear()
        hits.updated = time.time()

    try:
        while True:
            path = _choose_hits_path()
            if not path or not Path(path).exists():
                time.sleep(0.25); continue

            if path != last_path:
                if f:
                    try: f.close()
                    except: pass
                f = open(path, "r", encoding="utf-8-sig", newline="")
                rdr = csv.DictReader(f)
                hdrs = [h.strip() for h in (rdr.fieldnames or [])]
                with hits.lock:
                    hits.headers = hdrs

                def resolve(wants):
                    for w in wants:
                        if w in hdrs: return w
                        wl = w.lower()
                        for h in hdrs:
                            if h.lower() == wl: return h
                    return None

                key["ts"]          = resolve(aliases["ts"])
                key["host"]        = resolve(aliases["host"])
                key["tld"]         = resolve(aliases["tld"])
                key["interesting"] = resolve(aliases["interesting"])

                print(f"[hits] tailing: {path}")
                print(f"[hits] header map: ts={key['ts']}, host={key['host']}, tld={key['tld']}, interesting={key['interesting']}")
                last_path = path

            progressed = False

            loc_rows = 0
            loc_rows_int = 0
            loc_daily = Counter()
            loc_tlds  = Counter()
            loc_hosts = Counter()

            for _ in range(HITS_CHUNK_LINES):
                try:
                    row = next(rdr)
                except StopIteration:
                    break
                progressed = True
                loc_rows += 1
                iv = row.get(key["interesting"]) if key["interesting"] else None
                if not _truthy(iv):
                    continue
                loc_rows_int += 1
                ts   = row.get(key["ts"], "")   if key["ts"]   else ""
                host = row.get(key["host"], "") if key["host"] else ""
                tld  = row.get(key["tld"], "")  if key["tld"]  else ""
                if ts:   loc_daily[str(ts)[:10]] += 1
                if tld:  loc_tlds[str(tld)]     += 1
                if host: loc_hosts[str(host)]   += 1

            if loc_rows or loc_rows_int or loc_daily or loc_tlds or loc_hosts:
                with hits.lock:
                    hits.rows += loc_rows
                    hits.rows_interesting += loc_rows_int
                    hits.daily.update(loc_daily)
                    hits.tlds.update(loc_tlds)
                    hits.hosts.update(loc_hosts)
                    hits.updated = time.time()

                if (hits.rows % (HITS_CHUNK_LINES * 10)) < HITS_CHUNK_LINES:
                    print(f"[hits] processed rows={hits.rows:,} (interesting={hits.rows_interesting:,})")

            time.sleep(0.35 if not progressed else 0.05)

    except Exception as e:
        with hits.lock:
            hits.phase = "error"
            hits.error = str(e)
        print("[hits] ERROR:", e)
        raise
    finally:
        if f:
            try: f.close()
            except: pass

def start_hits_pipeline(rebuild: bool):
    with hits.lock:
        if hits.running: return False
        hits.running = True
        hits.error = None
    def runner():
        try:
            if rebuild:
                try: Path(HITS_PART).unlink()
                except: pass
                threading.Thread(target=dump_sqlite_to_csv_part, daemon=True).start()
            tail_hits_csv_live()
        finally:
            with hits.lock:
                hits.running = False
                hits.updated = time.time()
    threading.Thread(target=runner, daemon=True).start()
    return True

class TypoState:
    def __init__(self):
        self.lock = threading.Lock()
        self.running = False
        self.rows = 0
        self.updated = 0.0
        self.error = None
        self.buckets = Counter()
        self.vendor  = Counter()
        self.ips     = Counter()
        self.cidrs   = Counter()
        self.ns      = Counter()
        self.mx      = Counter()
        self.wild_true = 0
        self.wild_total = 0

typo = TypoState()

def _choose_typo_path() -> str:
    base = Path(TYPO_CSV)
    parts = sorted(base.parent.glob(base.stem.replace(".csv","") + "*.csv.part"))
    if parts: return str(parts[-1])
    return str(base)

def tail_typosquat_csv_live():
    last_path, f, rdr = None, None, None
    with typo.lock:
        typo.running = True
        typo.rows = 0
        typo.updated = time.time()
        typo.error = None
        typo.buckets.clear(); typo.vendor.clear()
        typo.ips.clear(); typo.cidrs.clear(); typo.ns.clear(); typo.mx.clear()
        typo.wild_true = 0; typo.wild_total = 0

    aliases = {
        "bucket":["bucket","class","label"], "vendor":["vendor","hoster","provider"],
        "ip_list":["ip_list","ips","iplist"], "ns_list":["ns_list","ns","nameservers"],
        "mx_list":["mx_list","mx","mailservers"], "wildcard":["wildcard","wc","has_wildcard"],
    }
    key = {}

    try:
        while True:
            path = _choose_typo_path()
            if not Path(path).exists():
                time.sleep(0.3); continue

            if path != last_path:
                if f:
                    try: f.close()
                    except: pass
                f = open(path, "r", encoding="utf-8-sig", newline="")
                rdr = csv.DictReader(f)
                hdrs = rdr.fieldnames or []
                for k,opts in aliases.items():
                    key[k] = next((h for h in opts if h in hdrs), None)
                last_path = path
                print(f"[typo] tailing: {path}", flush=True)

            progressed = False

            loc_rows = 0
            loc_b = Counter(); loc_v = Counter()
            loc_ips = Counter(); loc_cidrs = Counter()
            loc_ns = Counter(); loc_mx = Counter()
            loc_w_true = 0; loc_w_tot = 0

            try:
                for _ in range(10000):
                    row = next(rdr)
                    progressed = True
                    loc_rows += 1
                    b = (row.get(key["bucket"], "") if key["bucket"] else "") or ""
                    v = (row.get(key["vendor"], "") if key["vendor"] else "") or ""
                    ips = _split_items(row.get(key["ip_list"], "")) if key["ip_list"] else []
                    nss = _split_items(row.get(key["ns_list"], "")) if key["ns_list"] else []
                    mxs = _split_items(row.get(key["mx_list"], "")) if key["mx_list"] else []
                    wc = str(row.get(key["wildcard"], "") or "").strip() in TRUE_SET
                    if b: loc_b[b] += 1
                    if v: loc_v[v] += 1
                    loc_w_tot += 1
                    if wc: loc_w_true += 1
                    for ip in ips:
                        loc_ips[ip] += 1
                        c = _cidr24(ip)
                        if c: loc_cidrs[c] += 1
                    for ns in nss: loc_ns[_root(ns)] += 1
                    for mx in mxs: loc_mx[_root(mx)] += 1
            except StopIteration:
                pass

            if progressed:
                with typo.lock:
                    typo.rows += loc_rows
                    typo.buckets.update(loc_b); typo.vendor.update(loc_v)
                    typo.ips.update(loc_ips);   typo.cidrs.update(loc_cidrs)
                    typo.ns.update(loc_ns);     typo.mx.update(loc_mx)
                    typo.wild_true += loc_w_true
                    typo.wild_total += loc_w_tot
                    typo.updated = time.time()
                time.sleep(0.05)
            else:
                with typo.lock: typo.updated = time.time()
                time.sleep(0.4)

    except Exception as e:
        with typo.lock:
            typo.error = str(e)
        raise
    finally:
        if f:
            try: f.close()
            except: pass
        with typo.lock:
            typo.running = False
            typo.updated = time.time()

def start_typo_tailer():
    with typo.lock:
        if typo.running: return False
        typo.running = True
    threading.Thread(target=tail_typosquat_csv_live, daemon=True).start()
    return True

app = Flask(__name__)

@app.get("/")
def home():
    return """<!doctype html><html><head>
<meta charset='utf-8'><meta name='viewport' content='width=device-width,initial-scale=1'>
<title>Titleist++ — Live Hits (interesting only)</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.1/dist/chart.umd.min.js"></script>
<style>
body{font-family:system-ui,Segoe UI,Roboto,Ubuntu,Arial,sans-serif;background:#0b1020;color:#e7ebf2;margin:0}
header{padding:16px 24px;background:#0f1733;border-bottom:1px solid #1d2a57;display:flex;gap:12px;align-items:center}
button.btn,a.btn{cursor:pointer;color:#e7ebf2;text-decoration:none;border:1px solid #2b3c86;border-radius:10px;padding:6px 10px;background:#1e2a5f}
button.btn:hover,a.btn:hover{background:#283a7e}
main{padding:16px 24px;max-width:1200px;margin:0 auto}
.grid{display:grid;grid-template-columns:1fr 1fr;gap:16px}
.panel{background:#121a3a;border:1px solid #21306b;border-radius:12px;padding:16px}
small{color:#a9b3d1}
@media(max-width:900px){.grid{grid-template-columns:1fr}}
.panel canvas{width:100%;height:280px;display:block}
</style>
</head><body>
<header>
  <button id="rebuild" class="btn">Rebuild from SQLite → CSV</button>
  <small id="stat">idle</small>
</header>
<main>
  <div class='grid'>
    <div class='panel'><h3>Daily Volume (interesting)</h3><canvas id='cDaily'></canvas></div>
    <div class='panel'><h3>Top TLDs (interesting)</h3><canvas id='cTLDs'></canvas></div>
    <div class='panel'><h3>Top Hosts (interesting)</h3><canvas id='cHosts'></canvas></div>
    <div class='panel'><h3>Interesting by TLD</h3><canvas id='cIntr'></canvas></div>
  </div>
</main>
<script>
(function(){
  if (typeof Chart === 'undefined') { console.error('Chart.js not found'); return; }

  // Make robust numeric
  const toNum = v => {
    const n = Number(String(v ?? '').replace(/,/g,'').trim());
    return Number.isFinite(n) && n >= 0 ? n : 0;
  };

  // Build data points as {x, y} so parsing is unambiguous.
  function makeBarPoints(labels, values, horiz){
    const L = Array.isArray(labels) ? labels : [];
    const V = Array.isArray(values) ? values.map(toNum) : [];
    const n = Math.min(L.length, V.length);
    const pts = new Array(n);
    for (let i=0;i<n;i++){
      if (horiz) pts[i] = { x: V[i], y: String(L[i] ?? '') };
      else       pts[i] = { x: String(L[i] ?? ''), y: V[i] };
    }
    return pts;
  }

  function ensureBar(id, horiz=false, label='count'){
    const el = document.getElementById(id);
    if (!el.__chart){
      el.__chart = new Chart(el, {
        type:'bar',
        data:{ datasets:[{ label, data: [], backgroundColor:'rgba(88,144,255,0.55)', borderColor:'rgba(88,144,255,1)', borderWidth:1 }] },
        options:Object.assign({
          responsive:true, maintainAspectRatio:true,
          parsing:true,         // ← use {x,y} points
          indexAxis: horiz ? 'y' : 'x',
          scales: horiz ? {
            x:{ type:'linear', beginAtZero:true, grid:{color:'#1d2a57'}, ticks:{color:'#a9b3d1'} },
            y:{ type:'category', grid:{color:'#1d2a57'}, ticks:{color:'#a9b3d1'} }
          } : {
            x:{ type:'category', grid:{color:'#1d2a57'}, ticks:{color:'#a9b3d1'} },
            y:{ type:'linear', beginAtZero:true, grid:{color:'#1d2a57'}, ticks:{color:'#a9b3d1'} }
          }
        }, {})
      });
    }
    return el.__chart;
  }

  function ensureLine(id, label='count'){
    const el = document.getElementById(id);
    if (!el.__chart){
      el.__chart = new Chart(el, {
        type:'line',
        data:{ datasets:[{ label, data: [], fill:false, tension:0.15, borderWidth:2, pointRadius:2, borderColor:'rgba(88,144,255,1)' }] },
        options:{
          responsive:true,  maintainAspectRatio:true,
          parsing:true,     // we will feed {x: label, y: value}
          scales:{
            x:{ type:'category', grid:{color:'#1d2a57'}, ticks:{color:'#a9b3d1'} },
            y:{ type:'linear', beginAtZero:true, grid:{color:'#1d2a57'}, ticks:{color:'#a9b3d1'} }
          }
        }
      });
    }
    return el.__chart;
  }

  const ch = {
    daily: ensureLine('cDaily','count'),
    tlds:  ensureBar('cTLDs',false,'count'),
    hosts: ensureBar('cHosts',true,'count'),
    intr:  ensureBar('cIntr',true,'interesting'),
  };

  async function j(u,opts){
    const r = await fetch(u, Object.assign({cache:'no-store'}, opts||{}));
    if(!r.ok) throw new Error(`${r.status} ${r.statusText}`);
    return r.json();
  }
  const setStatus = t => { const el=document.getElementById('stat'); if(el) el.textContent=t; };

  function applyBar(chart, labels, values, horiz){
    chart.data.datasets[0].data = makeBarPoints(labels, values, horiz);
    // set a reasonable suggestedMax on the numeric axis
    const maxV = Math.max(0, ...values.map(toNum));
    const axis = horiz ? 'x' : 'y';
    chart.options.scales[axis].beginAtZero = true;
    chart.options.scales[axis].suggestedMax = maxV > 0 ? Math.ceil(maxV * 1.1) : 1;
    chart.update();
  }

  function applyLine(chart, labels, values){
    const pts = [];
    for (let i=0;i<Math.min(labels.length, values.length); i++){
      pts.push({ x:String(labels[i] ?? ''), y:toNum(values[i]) });
    }
    chart.data.datasets[0].data = pts;
    const maxV = Math.max(0, ...values.map(toNum));
    chart.options.scales.y.suggestedMax = maxV > 0 ? Math.ceil(maxV * 1.1) : 1;
    chart.update();
  }

  let inFlight = false;
  async function refresh(){
    if (inFlight) return;
    inFlight = true;
    try{
      const st = await j('/api/hits/status');
      setStatus(`${st.phase||'processing'} • rows=${(st.rows||0).toLocaleString()} • interesting=${(st.rows_interesting||0).toLocaleString()}${st.error?(' • error='+st.error):''}`);

      const [daily,tlds,hosts,intr] = await Promise.allSettled([
        j('/api/hits/data/daily'),
        j('/api/hits/data/tlds'),
        j('/api/hits/data/hosts'),
        j('/api/hits/data/interesting'),
      ]);

      if (daily.status==='fulfilled'){
        const arr = (daily.value||[]).slice().sort((a,b)=>String(a.day).localeCompare(String(b.day)));
        applyLine(ch.daily, arr.map(x=>x.day), arr.map(x=>x.count));
      }
      if (tlds.status==='fulfilled'){
        applyBar(ch.tlds,  tlds.value.map(x=>x.tld),  tlds.value.map(x=>x.count), false);
      }
      if (hosts.status==='fulfilled'){
        applyBar(ch.hosts, hosts.value.map(x=>x.host), hosts.value.map(x=>x.count), true);
      }
      if (intr.status==='fulfilled'){
        applyBar(ch.intr,  intr.value.map(x=>x.tld),  intr.value.map(x=>x.count), true);
      }
    }catch(e){
      console.warn('[ui] refresh error', e);
      setStatus('error: ' + (e.message||e));
    }finally{
      inFlight = false;
    }
  }

  document.getElementById('rebuild').addEventListener('click', async ()=>{
    try{ await j('/api/hits/restart',{method:'POST'}); }catch(e){}
  });

  try{
    const es = new EventSource('/stream/hits');
    es.addEventListener('hello', refresh);
    es.addEventListener('update', refresh);
    es.onerror = ()=>{};
  }catch(e){}

  refresh();
  setInterval(()=>{ requestAnimationFrame(refresh); }, 1000);
})();
</script>
</body></html>"""

# ---------- Typosquat page (unchanged) ----------
@app.get("/typosquat")
def typosquat_page():
    return """<!doctype html><html><head>
<meta charset='utf-8'><meta name='viewport' content='width=device-width,initial-scale=1'>
<title>Titleist++ — Typosquats</title>
<script src='https://cdn.jsdelivr.net/npm/chart.js'></script>
<style>
body{font-family:system-ui,Segoe UI,Roboto,Ubuntu,Arial,sans-serif;background:#0b1020;color:#e7ebf2;margin:0}
header{padding:16px 24px;background:#0f1733;border-bottom:1px solid #1d2a57;display:flex;gap:12px;align-items:center}
button.btn,a.btn{cursor:pointer;color:#e7ebf2;text-decoration:none;border:1px solid #2b3c86;border-radius:10px;padding:6px 10px;background:#1e2a5f}
button.btn:hover,a.btn:hover{background:#283a7e}
main{padding:16px 24px;max-width:1200px;margin:0 auto}
.grid{display:grid;grid-template-columns:1fr 1fr;gap:16px}
.panel{background:#121a3a;border:1px solid #21306b;border-radius:12px;padding:16px}
small{color:#a9b3d1}
@media(max-width:900px){.grid{grid-template-columns:1fr}}
.panel canvas{width:100%;height:280px;display:block}
</style>
</head><body>
<header>
  <a class='btn' href='/'>← Back</a>
  <button id="watch" class="btn">Watch live</button>
  <small id='sum'>idle</small>
</header>
<main>
  <div class='grid'>
    <div class='panel'><h3>Buckets</h3><canvas id='buckets'></canvas></div>
    <div class='panel'><h3>Vendors</h3><canvas id='vendor'></canvas></div>
    <div class='panel'><h3>Top IPs</h3><canvas id='ips'></canvas></div>
    <div class='panel'><h3>Top Subnets</h3><canvas id='cidrs'></canvas></div>
    <div class='panel'><h3>Top NS roots</h3><canvas id='ns'></canvas></div>
    <div class='panel'><h3>Top MX roots</h3><canvas id='mx'></canvas></div>
  </div>
</main>
<script>
async function j(u,opts){const r=await fetch(u,opts||{}); if(!r.ok) throw new Error(await r.text()); return r.json();}
function toNum(v){const n=Number(String(v??'').replace(/,/g,'')); return Number.isFinite(n)&&n>=0?n:0;}
function bar(id,pts,hor=false,lbl='count'){
  const ctx=document.getElementById(id);
  return new Chart(ctx,{type:'bar',
    data:{datasets:[{label:lbl,data:pts,backgroundColor:'rgba(88,144,255,0.55)'}]},
    options:Object.assign({
      responsive:true,maintainAspectRatio:true,parsing:true,
      indexAxis:hor?'y':'x',
      scales: hor? {x:{type:'linear',beginAtZero:true}, y:{type:'category'}} : {x:{type:'category'}, y:{type:'linear',beginAtZero:true}}
    },{})
  });
}
let chB,chV,chIPs,chCIDRs,chN,chM;
function pts(labels,values,hor){const n=Math.min(labels.length,values.length);const a=[];for(let i=0;i<n;i++){const L=String(labels[i]??'');const V=toNum(values[i]);a.push(hor?{x:V,y:L}:{x:L,y:V});}return a;}
function applyMax(ch,hor,vals){const m=Math.max(0,...vals.map(toNum));const axis=hor?'x':'y';ch.options.scales[axis].suggestedMax=m>0?Math.ceil(m*1.1):1;ch.update();}
async function renderOnce(){
  const d = await j('/api/typo/live/data');
  document.getElementById('sum').textContent = `${(d.rows||0).toLocaleString()} rows • wildcard ${(d.wildcard_rate||0).toFixed(3)}`;
  const set=(ref,id,L,V,hor=false)=>{if(ref)ref.destroy(); const ch=bar(id,pts(L,V,hor),hor); applyMax(ch,hor,V); return ch;};
  chB=set(chB,'buckets',d.buckets.map(x=>x.bucket),d.buckets.map(x=>x.count));
  chV=set(chV,'vendor', d.vendor.map(x=>x.vendor), d.vendor.map(x=>x.count),true);
  chIPs=set(chIPs,'ips', d.ips.map(x=>x.ip), d.ips.map(x=>x.count),true);
  chCIDRs=set(chCIDRs,'cidrs', d.cidrs.map(x=>x.cidr), d.cidrs.map(x=>x.count),true);
  chN=set(chN,'ns', d.ns.map(x=>x.ns), d.ns.map(x=>x.count),true);
  chM=set(chM,'mx', d.mx.map(x=>x.mx), d.mx.map(x=>x.count),true);
}
document.getElementById('watch').addEventListener('click', async ()=>{
  await j('/api/typo/live/start', {method:'POST'});
  const es = new EventSource('/stream/typo');
  es.addEventListener('hello', renderOnce);
  es.addEventListener('progress', renderOnce);
});
</script>
</body></html>"""

def _ser_hits_daily():
    with hits.lock:
        items = list(hits.daily.items())
    items.sort(key=lambda kv: kv[0])
    return [{"day": k, "count": int(v)} for k,v in items]

def _ser_hits_tlds(n=25):
    with hits.lock:
        items = list(hits.tlds.items())
    items.sort(key=lambda kv: kv[1], reverse=True)
    return [{"tld": k, "count": int(v)} for k, v in items[:n]]

def _ser_hits_hosts(n=25):
    with hits.lock:
        items = list(hits.hosts.items())
    items.sort(key=lambda kv: kv[1], reverse=True)
    return [{"host": k, "count": int(v)} for k, v in items[:n]]

def _ser_hits_intr(n=25):
    return _ser_hits_tlds(n)

@app.post("/api/hits/restart")
def api_hits_restart():
    started = start_hits_pipeline(rebuild=True)
    return jsonify({"started": bool(started)})

@app.get("/api/hits/status")
def api_hits_status():
    with hits.lock:
        return jsonify({
            "running": hits.running, "phase": hits.phase,
            "rows": hits.rows, "rows_interesting": hits.rows_interesting,
            "updated": hits.updated, "error": hits.error
        })

@app.get("/api/hits/data/<metric>")
def api_hits_data(metric):
    if metric == "daily":       return jsonify(_ser_hits_daily())
    if metric == "tlds":        return jsonify(_ser_hits_tlds())
    if metric == "hosts":       return jsonify(_ser_hits_hosts())
    if metric == "interesting": return jsonify(_ser_hits_intr())
    return jsonify({"error":"unknown metric"}), 404

@app.get("/stream/hits")
def stream_hits():
    @stream_with_context
    def gen():
        last = -1
        last_heartbeat = time.time()
        yield "event: hello\n" "data: {}\n\n"
        while True:
            with hits.lock:
                stamp = hits.updated
            if stamp != last:
                last = stamp
                yield "event: update\n" "data: {}\n\n"
            if time.time() - last_heartbeat > 15:
                last_heartbeat = time.time()
                yield ": keepalive\n\n"
            time.sleep(0.6)
    return Response(
        gen(),
        mimetype="text/event-stream",
        headers={"Cache-Control":"no-cache","X-Accel-Buffering":"no","Connection":"keep-alive"},
    )

@app.post("/api/typo/live/start")
def api_typo_start():
    started = start_typo_tailer()
    return jsonify({"started": bool(started)})

@app.get("/api/typo/live/data")
def api_typo_data():
    with typo.lock:
        return jsonify({
            "rows": typo.rows,
            "wildcard_rate": (typo.wild_true/typo.wild_total) if typo.wild_total else 0.0,
            "buckets": [{"bucket": k, "count": int(v)} for k,v in typo.buckets.most_common()],
            "vendor":  [{"vendor": k, "count": int(v)} for k,v in typo.vendor.most_common(20)],
            "ips":     [{"ip": k, "count": int(v)} for k,v in typo.ips.most_common(20)],
            "cidrs":   [{"cidr": k, "count": int(v)} for k,v in typo.cidrs.most_common(20)],
            "ns":      [{"ns": k, "count": int(v)} for k,v in typo.ns.most_common(20)],
            "mx":      [{"mx": k, "count": int(v)} for k,v in typo.mx.most_common(20)],
        })

@app.get("/stream/typo")
def stream_typo():
    @stream_with_context
    def gen():
        last = -1
        yield "event: hello\n" + "data: {}\n\n"
        while True:
            with typo.lock: rows = typo.rows
            if rows != last:
                last = rows
                yield f"event: progress\n" + f"data: {{\"rows\": {rows}}}\n\n"
            time.sleep(0.2)
    return Response(gen(), mimetype="text/event-stream")

@app.after_request
def no_cache(resp):
    resp.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
    resp.headers['Pragma'] = 'no-cache'
    resp.headers['Expires'] = '0'
    return resp

if __name__ == "__main__":
    print(f"[Titleist++] CSV-first app on :{PORT}")
    print(f"  SQLite → {SQLITE_PATH}")
    print(f"  Hits CSV → {HITS_CSV}  (writer: {HITS_PART})")
    print(f"  Typo  CSV → {TYPO_CSV}")
    if Path(HITS_CSV).exists() or Path(HITS_PART).exists():
        start_hits_pipeline(rebuild=False)

    from werkzeug.serving import run_simple
    run_simple("0.0.0.0", PORT, app, use_reloader=False, use_debugger=False, threaded=True)