"""
app.py — Google Maps leads dashboard (Streamlit @ HF Space)

Reads scrape runs from an HF Dataset repo (the bridge). Each scrape run is a
separate file under runs/ — pick one run or merge them all. Can trigger a new
scrape on Kaggle (fire-and-forget). The table has an "explored" checkbox per
lead; progress is saved to the HF Dataset so it survives restarts.

Configuration (env vars or Streamlit secrets):
    HF_DATASET_REPO  e.g. "username/gmaps-leads"   (required)
    HF_TOKEN         HF WRITE token                (data read + tracking save + Kaggle push)
    KAGGLE_USERNAME  kaggle username               (optional: enables the trigger)
    KAGGLE_KEY       kaggle legacy API key         (optional: enables the trigger)
"""

import json
import os

import folium
import pandas as pd
import plotly.express as px
import streamlit as st
from huggingface_hub import HfApi, hf_hub_download
from streamlit_folium import st_folium

# ---------- palette (dataviz reference, light mode) ----------
SURFACE = "#fcfcfb"
INK = "#0b0b0b"
INK_MUTED = "#898781"
GRID = "#e1e0d9"
BASELINE = "#c3c2b7"
SERIES = "#2a78d6"                                # categorical slot 1 (blue)
SEQ_RAMP = ["#86b6ef", "#2a78d6", "#0d366b"]       # sequential blue 250/450/700
FONT = 'system-ui, -apple-system, "Segoe UI", sans-serif'

TRACK_FILE = "tracking/explored.json"
DELETED_FILE = "tracking/deleted.json"

st.set_page_config(page_title="GMaps Leads", page_icon="🗺️", layout="wide")


def conf(name: str, default: str = "") -> str:
    val = os.environ.get(name)
    if val:
        return val
    try:
        return st.secrets[name]
    except Exception:
        return default


REPO_ID = conf("HF_DATASET_REPO")
HF_TOKEN = conf("HF_TOKEN")


def score(row: pd.Series) -> int:
    s = 0
    if not row.get("website"):
        s += 3
    if row["reviews_count"] >= 50 and row["rating"] < 4.0:
        s += 2
    if row["reviews_count"] < 10:
        s += 1
    return s


def lead_key(row) -> str:
    return f"{row['name']}|{row.get('address', '')}"


@st.cache_data(ttl=600, show_spinner="Listing scrape runs...")
def list_runs(repo_id: str, token: str) -> list:
    """All run CSVs in the dataset, newest first. Falls back to legacy leads.csv."""
    files = HfApi(token=token or None).list_repo_files(repo_id, repo_type="dataset")
    runs = sorted((f for f in files if f.startswith("runs/") and f.endswith(".csv")),
                  reverse=True)
    if not runs and "leads.csv" in files:
        runs = ["leads.csv"]
    return runs


@st.cache_data(ttl=600, show_spinner="Fetching data from the HF Dataset...")
def load_csv(repo_id: str, token: str, filename: str) -> pd.DataFrame:
    path = hf_hub_download(repo_id=repo_id, filename=filename,
                           repo_type="dataset", token=token or None)
    df = pd.read_csv(path)

    df["reviews_count"] = (
        pd.to_numeric(df.get("reviews_count"), errors="coerce").fillna(0).astype(int)
    )
    df["rating"] = pd.to_numeric(df.get("rating"), errors="coerce").fillna(0.0)
    for col in ("website", "phone", "category", "address", "email"):
        if col in df.columns:
            df[col] = df[col].fillna("")
    if "score" not in df.columns:
        df["score"] = df.apply(score, axis=1)
    df["lat"] = pd.to_numeric(df.get("lat"), errors="coerce")
    df["lng"] = pd.to_numeric(df.get("lng"), errors="coerce")
    return df


@st.cache_data(ttl=600)
def repo_last_updated(repo_id: str, token: str):
    return HfApi(token=token or None).dataset_info(repo_id).last_modified


@st.cache_data(ttl=60)
def load_tracking(repo_id: str, token: str) -> dict:
    """Explored-checkbox state, persisted in the dataset repo."""
    try:
        path = hf_hub_download(repo_id=repo_id, filename=TRACK_FILE,
                               repo_type="dataset", token=token or None)
        with open(path, encoding="utf-8") as fh:
            return json.load(fh)
    except Exception:
        return {}


def save_tracking(repo_id: str, token: str, data: dict) -> None:
    HfApi(token=token).upload_file(
        path_or_fileobj=json.dumps(data, ensure_ascii=False).encode("utf-8"),
        path_in_repo=TRACK_FILE,
        repo_id=repo_id,
        repo_type="dataset",
        commit_message="update explored tracking",
    )


@st.cache_data(ttl=60)
def load_deleted(repo_id: str, token: str) -> list:
    """Lead keys the user removed from the dashboard (tombstones)."""
    try:
        path = hf_hub_download(repo_id=repo_id, filename=DELETED_FILE,
                               repo_type="dataset", token=token or None)
        with open(path, encoding="utf-8") as fh:
            return json.load(fh)
    except Exception:
        return []


def save_deleted(repo_id: str, token: str, keys: list) -> None:
    HfApi(token=token).upload_file(
        path_or_fileobj=json.dumps(sorted(set(keys)),
                                   ensure_ascii=False).encode("utf-8"),
        path_in_repo=DELETED_FILE,
        repo_id=repo_id,
        repo_type="dataset",
        commit_message="update deleted leads",
    )


def run_label(filename: str) -> str:
    """'runs/20260704-080501_rumah-makan-yogyakarta.csv' -> readable label."""
    if filename == "leads.csv":
        return "leads.csv (legacy)"
    stem = filename.removeprefix("runs/").removesuffix(".csv")
    ts, _, slug = stem.partition("_")
    return f"{slug or 'run'}  ·  {ts}"


def style(fig, title: str):
    fig.update_layout(
        title=dict(text=title, font=dict(size=15, color=INK)),
        plot_bgcolor=SURFACE, paper_bgcolor=SURFACE,
        font=dict(family=FONT, color=INK, size=12),
        margin=dict(l=10, r=10, t=45, b=10),
        showlegend=False,
        xaxis=dict(gridcolor=GRID, linecolor=BASELINE, zerolinecolor=BASELINE,
                   tickfont=dict(color=INK_MUTED)),
        yaxis=dict(gridcolor=GRID, linecolor=BASELINE, zerolinecolor=BASELINE,
                   tickfont=dict(color=INK_MUTED)),
    )
    return fig


# ---------- header ----------
st.title("🗺️ Leads Dashboard — Google Maps")

if not REPO_ID:
    st.error("Set `HF_DATASET_REPO` (env var / secret) first, e.g. `username/gmaps-leads`.")
    st.stop()

try:
    runs = list_runs(REPO_ID, HF_TOKEN)
    updated = repo_last_updated(REPO_ID, HF_TOKEN)
except Exception as e:
    st.error(f"Failed to read the dataset `{REPO_ID}`: {e}")
    st.stop()

top_l, top_r = st.columns([5, 1], vertical_alignment="center")
top_l.caption(f"Source: `{REPO_ID}` · {len(runs)} run(s) · dataset last updated: "
              f"**{updated:%d %b %Y %H:%M} UTC**")
if top_r.button("🔄 Reload data", use_container_width=True):
    st.cache_data.clear()
    st.rerun()

# ---------- trigger a new scrape (ALWAYS visible, even with no data) ----------
KAGGLE_USERNAME = conf("KAGGLE_USERNAME")
KAGGLE_KEY = conf("KAGGLE_KEY")

with st.expander("🚀 Trigger a new scrape (runs on Kaggle)", expanded=not runs):
    if not (KAGGLE_USERNAME and KAGGLE_KEY):
        st.info(
            "To enable this, add to the Space settings: `KAGGLE_USERNAME` (variable) "
            "and `KAGGLE_KEY` (secret, legacy API key from kaggle.com/settings). "
            "`HF_TOKEN` must be a **write** token."
        )
    else:
        import trigger

        with st.form("trigger_form"):
            t1, t2, t3 = st.columns(3)
            t_keyword = t1.text_input("Keyword", "coffee shop")
            t_city = t2.text_input("City", "Yogyakarta")
            t_max = t3.number_input("Max results", min_value=10, max_value=200,
                                    value=40, step=10)
            t4, t5, t6 = st.columns(3)
            t_min_rating = t4.number_input("Min. rating (0 = off)", min_value=0.0,
                                           max_value=5.0, value=0.0, step=0.1)
            t_min_reviews = t5.number_input("Min. total reviews (0 = off)",
                                            min_value=0, max_value=100_000,
                                            value=0, step=10)
            t_only_new = t6.checkbox(
                "Only new leads",
                help="Drop leads that already exist in previous runs, so a "
                     "rescrape adds only businesses you haven't seen yet.")
            t_details = st.checkbox("Fetch phone/website details (slower)", value=True)
            t7, t8 = st.columns(2)
            t_max_reviews = t7.number_input(
                "Review texts per lead (0 = off)", min_value=0, max_value=200,
                value=0, step=5,
                help="Needs 'Fetch phone/website details'. Pulls full review "
                     "text — makes each lead noticeably slower to scrape.")
            t_max_photos = t8.number_input(
                "Photo URLs per lead (0 = off)", min_value=0, max_value=200,
                value=0, step=5,
                help="Needs 'Fetch phone/website details'. Pulls direct links "
                     "to listing photos.")
            submitted = st.form_submit_button("▶️ Start scrape on Kaggle")

        if submitted:
            with st.spinner("Pushing the kernel to Kaggle..."):
                try:
                    out = trigger.trigger_scrape(
                        keyword=t_keyword, city=t_city, max_results=int(t_max),
                        with_details=t_details,
                        min_reviews=int(t_min_reviews),
                        min_rating=float(t_min_rating),
                        only_new=bool(t_only_new),
                        max_reviews=int(t_max_reviews),
                        max_photos=int(t_max_photos),
                        hf_repo=REPO_ID, hf_token=HF_TOKEN,
                        kaggle_username=KAGGLE_USERNAME, kaggle_key=KAGGLE_KEY,
                    )
                    st.session_state["trigger_msg"] = (
                        "✅ Kernel pushed — Kaggle is running the scraper now. "
                        "The result appears as a NEW run in the dropdown in "
                        "~15–60 min — click **Reload data** then.\n\n"
                        + (out or "(no CLI output)")
                    )
                except Exception as e:
                    st.session_state["trigger_msg"] = f"❌ Trigger failed: {e}"

        if st.session_state.get("trigger_msg"):
            st.info(st.session_state["trigger_msg"])

        if st.button("🔍 Check kernel status"):
            try:
                st.session_state["kernel_status"] = trigger.kernel_status(
                    KAGGLE_USERNAME, KAGGLE_KEY)
            except Exception as e:
                st.session_state["kernel_status"] = f"Status check failed: {e}"

        if st.session_state.get("kernel_status"):
            st.code(st.session_state["kernel_status"] or "(empty status)")

# ---------- run selector ----------
if not runs:
    st.warning("No scrape runs in the dataset yet — trigger one above.")
    st.stop()

MERGED = "🧩 All runs (merged & deduped)"
options = ([MERGED] + runs) if len(runs) > 1 else runs
sel_l, sel_r = st.columns([4, 1], vertical_alignment="bottom")
choice = sel_l.selectbox("Scrape run", options, format_func=lambda f:
                         f if f == MERGED else run_label(f))

if choice != MERGED:
    with sel_r.popover("🗑️ Delete run", use_container_width=True):
        st.warning(f"Permanently delete `{run_label(choice)}` "
                   "from the HF Dataset?")
        if st.button("Yes, delete this run", type="primary"):
            try:
                HfApi(token=HF_TOKEN).delete_file(
                    choice, REPO_ID, repo_type="dataset",
                    commit_message=f"delete {choice}")
                st.cache_data.clear()
                st.rerun()
            except Exception as e:
                st.error(f"Delete failed (HF_TOKEN must be a write token): {e}")

try:
    if choice == MERGED:
        frames = [load_csv(REPO_ID, HF_TOKEN, f) for f in runs]
        df = (pd.concat(frames, ignore_index=True)
              .drop_duplicates(subset=["name", "address"], keep="first"))
    else:
        df = load_csv(REPO_ID, HF_TOKEN, choice)
except Exception as e:
    st.error(f"Failed to load `{choice}`: {e}")
    st.stop()

# Hide leads the user removed from the dashboard (tombstones survive rescrapes)
deleted_keys = set(load_deleted(REPO_ID, HF_TOKEN))
if deleted_keys and len(df):
    df = df[~df.apply(lambda r: lead_key(r) in deleted_keys, axis=1)]

if df.empty:
    st.warning("This run contains 0 leads (probably a failed/captcha'd scrape, "
               "or every lead was deleted). "
               "Pick another run or trigger a new one above.")
    st.stop()

# ---------- sidebar filters ----------
st.sidebar.header("Filters")
min_rev = st.sidebar.slider("Min. review count", 0,
                            max(int(df["reviews_count"].max()), 1), 0)
min_rating = st.sidebar.slider("Min. rating", 0.0, 5.0, 0.0, 0.1)
no_web_only = st.sidebar.checkbox("Only leads without a website")

f = df[(df["reviews_count"] >= min_rev) & (df["rating"] >= min_rating)]
if no_web_only:
    f = f[f["website"] == ""]

# ---------- stat tiles ----------
tracking = load_tracking(REPO_ID, HF_TOKEN)
explored_count = sum(1 for _, r in f.iterrows() if tracking.get(lead_key(r), False))

c1, c2, c3, c4 = st.columns(4)
c1.metric("Total leads", len(f))
c2.metric("Without website", int((f["website"] == "").sum()))
c3.metric("Median rating", f"{f['rating'].median():.1f}" if len(f) else "—")
c4.metric("Explored", f"{explored_count}/{len(f)}")

if not len(f):
    st.warning("No leads pass the current filters.")
    st.stop()

# ---------- charts ----------
left, right = st.columns(2)

hist = px.histogram(f, x="reviews_count", nbins=30,
                    color_discrete_sequence=[SERIES],
                    labels={"reviews_count": "review count"})
left.plotly_chart(style(hist, "Review count distribution"),
                  use_container_width=True)

scat = px.scatter(f, x="reviews_count", y="rating", color="score",
                  color_continuous_scale=SEQ_RAMP, hover_name="name",
                  labels={"reviews_count": "review count", "score": "score"})
scat.update_traces(marker=dict(size=9))
scat.update_coloraxes(colorbar_title_text="score",
                      colorbar_tickfont_color=INK_MUTED)
right.plotly_chart(style(scat, "Rating vs reviews (darker = higher score)"),
                   use_container_width=True)

# ---------- map ----------
st.subheader("Lead map")
st.caption("Marker color = score: 🔵 light 0–1 · 🔵 medium 2–3 · 🔵 dark 4+")
geo = f.dropna(subset=["lat", "lng"])
if len(geo):
    m = folium.Map(location=[geo["lat"].mean(), geo["lng"].mean()],
                   zoom_start=12, tiles="cartodbpositron")
    for _, r in geo.iterrows():
        color = SEQ_RAMP[0] if r["score"] <= 1 else (
            SEQ_RAMP[1] if r["score"] <= 3 else SEQ_RAMP[2])
        folium.CircleMarker(
            location=[r["lat"], r["lng"]], radius=6,
            color=color, fill=True, fill_color=color, fill_opacity=0.85,
            popup=folium.Popup(
                f"<b>{r['name']}</b><br>⭐ {r['rating']} ({r['reviews_count']} reviews)"
                f"<br>score: {r['score']}", max_width=280),
        ).add_to(m)
    st_folium(m, use_container_width=True, height=480, returned_objects=[])
else:
    st.info("No lat/lng coordinates in the filtered data.")

# ---------- table with explored checkboxes ----------
st.subheader("Leads table (sorted by score)")
st.caption("Tick ✅ **explored** for leads you've already checked, then click "
           "**Save progress**. Tick 🗑️ **delete** and click **Delete selected** "
           "to remove leads — both are stored in the HF Dataset and survive restarts.")

show_cols = [c for c in ("score", "name", "category", "rating", "reviews_count",
                         "address", "phone", "website", "email", "maps_url",
                         "reviews", "photos")
             if c in f.columns]
table = f.sort_values(["score", "reviews_count"],
                      ascending=[False, False])[show_cols].reset_index(drop=True)
table.insert(0, "explored",
             [tracking.get(lead_key(r), False) for _, r in table.iterrows()])
table.insert(1, "delete", False)

edited = st.data_editor(
    table,
    use_container_width=True,
    hide_index=True,
    disabled=show_cols,                      # only the checkboxes are editable
    column_config={
        "explored": st.column_config.CheckboxColumn("✅ explored"),
        "delete": st.column_config.CheckboxColumn("🗑️ delete"),
        "maps_url": st.column_config.LinkColumn("maps_url", display_text="open map"),
        "website": st.column_config.LinkColumn("website"),
    },
    key="leads_editor",
)

save_l, del_l, spacer, dl_r = st.columns([1.2, 1.2, 2.6, 1.6])

if save_l.button("💾 Save progress", use_container_width=True):
    new_tracking = dict(tracking)
    for _, r in edited.iterrows():
        new_tracking[lead_key(r)] = bool(r["explored"])
    try:
        save_tracking(REPO_ID, HF_TOKEN, new_tracking)
        load_tracking.clear()
        st.success("Progress saved to the HF Dataset.")
        st.rerun()
    except Exception as e:
        st.error(f"Save failed (HF_TOKEN must be a write token): {e}")

if del_l.button("🗑️ Delete selected", use_container_width=True):
    to_delete = [lead_key(r) for _, r in edited.iterrows() if r["delete"]]
    if not to_delete:
        st.info("No leads ticked for deletion.")
    else:
        try:
            save_deleted(REPO_ID, HF_TOKEN, list(deleted_keys) + to_delete)
            load_deleted.clear()
            st.success(f"{len(to_delete)} lead(s) removed from the dashboard.")
            st.rerun()
        except Exception as e:
            st.error(f"Delete failed (HF_TOKEN must be a write token): {e}")

dl_r.download_button("⬇️ Download filtered CSV",
                     edited.drop(columns=["delete"])
                           .to_csv(index=False).encode("utf-8"),
                     "leads_filtered.csv", "text/csv",
                     use_container_width=True)

if deleted_keys:
    with st.expander(f"♻️ {len(deleted_keys)} deleted lead(s)"):
        st.write("\n".join(f"- {k.split('|')[0]}" for k in sorted(deleted_keys)))
        if st.button("Restore all deleted leads"):
            try:
                save_deleted(REPO_ID, HF_TOKEN, [])
                load_deleted.clear()
                st.rerun()
            except Exception as e:
                st.error(f"Restore failed: {e}")