app.py

"""CC Backlink Explorer - a Majestic/Ahrefs-style backlink & rank tool built on the
Common Crawl domain-level web graph (CC-MAIN-2026 Apr-May-Jun release).

Architecture (built for speed on a free/basic CPU HF Space):
  - domain_vertices2.parquet (~0.9GB, incl. precomputed outbound_count) and
    domain_ranks.parquet (~2.5GB) are downloaded locally at startup and queried
    from local disk (instant).
  - domain_edges.parquet (~19GB, 3.9B edges, sorted by to_id, classic LFS storage
    -- NOT Xet, so plain HTTP range requests hit the CDN directly with no
    per-request reconstruction overhead) stays remote and is queried via DuckDB's
    httpfs with pre-resolved signed CDN URL.
  - CRITICAL for speed: every remote query is a PLAIN filter on domain_edges only
    (`WHERE to_id = X` / `WHERE to_id IN (...)`), never a JOIN mixed into the same
    query as the remote scan. Because the file is sorted by to_id, a plain filter
    lets DuckDB prune row groups via zone maps and only fetch the handful of row
    groups that matter (seconds, even for domains with 500k+ backlinks). So we
    always do it in two phases: (1) fetch raw ids remotely with a bare filter,
    (2) enrich that (small, already-local) id set locally.
  - Phase (2) does NOT use a SQL JOIN against the 121M-row vertices table.
    Measured via EXPLAIN: DuckDB does a full ~121M-row SEQ_SCAN for that join
    regardless of how few ids are being looked up (it never uses the id index
    for JOIN probes) -- that scan, not the network, was the real multi-second
    bottleneck. Instead, since domain ids are a dense 0..N-1 range, harmonic/
    pagerank/outbound are loaded as plain numpy arrays indexed *positionally*
    by id (true O(1) lookups) and domain names as a compact Arrow string array
    (~3GB for 121M domains) that supports fast vectorized take(). This turns an
    O(121M) scan into an O(k) gather, k = number of ids being enriched. We do
    NOT also keep a DuckDB copy of these 121M rows -- holding both at once
    OOM'd the 16GB cpu-basic Space; columns are read one-at-a-time at startup
    to keep the load-time memory peak low.
  - The full result set from phase (1)/(2) is cached in a Gradio State so that
    pagination and "download full CSV" never re-hit the network.
  - Every remote query is ALSO cached process-wide (in-memory LRU keyed by the
    query text, see `_query_cache`), so re-querying the same domain / re-running
    the same gap analysis later in the same Space session is instant, with zero
    network round-trips.
  - CSV export uses `gr.DownloadButton`: one click computes the CSV (from the
    already-fetched, in-memory full result set) and the browser downloads it
    immediately -- no separate "hosted file" step, nothing is uploaded anywhere,
    it's just served straight from the Space's own temp disk for that instant.
"""
import os
import time
import tempfile
from collections import OrderedDict

import duckdb
import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.parquet as pq
import requests as _r
import gradio as gr
from huggingface_hub import hf_hub_download

DATASET_REPO = os.environ.get("CC_DATASET_REPO", "metehan777/cc-webgraph-domain-2026-jun")
HF_TOKEN = os.environ.get("HF_TOKEN", "")
RESOLVE_URL = f"https://huggingface.co/datasets/{DATASET_REPO}/resolve/main/domain_edges.parquet"
TOTAL_DOMAINS = 121_091_933

print("=== app.py booting ===", flush=True)
_boot = time.time()
print("Downloading vertices3 (all domain attrs inline) ...", flush=True)
VERTICES = hf_hub_download(DATASET_REPO, "domain_vertices3.parquet", repo_type="dataset", token=HF_TOKEN or None)
print(f"Local file ready in {time.time() - _boot:.1f}s.", flush=True)

# DuckDB is used ONLY to range-scan the remote edges parquet. We deliberately
# do NOT load the 121M-row vertices table into DuckDB -- doing that *and* the
# numpy/Arrow arrays below at the same time held ~2x the data resident in RAM
# and OOM'd the 16GB cpu-basic Space (RUNTIME_ERROR). All per-domain attribute
# lookups are served from the arrays instead.
con = duckdb.connect()
con.execute("INSTALL httpfs; LOAD httpfs;")
con.execute("SET enable_http_metadata_cache=true;")
con.execute("SET enable_object_cache=true;")

# --- id-indexed attribute arrays (the whole in-memory model) ---------------
# IMPORTANT LEARNING: DuckDB does NOT use an ART index to speed up a JOIN/IN
# against a 121M-row table -- EXPLAIN shows a full SEQ_SCAN every time. That
# scan (not the remote edge fetch) was the real multi-second bottleneck.
# Since domain ids are a dense 0..N-1 set, we store attributes in arrays indexed
# *positionally by id* (true O(1) gather). The parquet is sorted by rev_domain,
# so id is a permutation of 0..N-1; we scatter each column into id order.
# Columns are read ONE AT A TIME (never the whole table at once) to keep the
# startup memory peak well under the Space's RAM limit. rev_domain stays as a
# compact Arrow string array (~3GB) instead of ~121M exploded Python str objects.
print("Loading domain attributes (Arrow, single read) ...", flush=True)
_t = time.time()
# Read the whole 0.9GB file ONCE. (Reading it column-by-column was ~5x the IO
# and on a throttled 2-vCPU free Space that alone blew startup past the
# scheduler's patience -> stuck APP_STARTING.) A single ~4GB Arrow table plus
# the numpy arrays below peaks well under the 16GB limit.
_tbl = pq.read_table(VERTICES, columns=["id", "rev_domain", "n_hosts", "outbound_count", "harmonic_pos", "pr_pos"])
print(f"  parquet read in {time.time() - _t:.1f}s", flush=True)
_ids = _tbl.column("id").to_numpy()
ARR_N = int(len(_ids))
if int(_ids.min()) != 0 or int(_ids.max()) != ARR_N - 1:
    raise RuntimeError("domain id column is not a dense 0..N-1 range")
_inv = np.empty(ARR_N, dtype=np.int64)
_inv[_ids] = np.arange(ARR_N, dtype=np.int64)  # _inv[id] = row position holding that id


def _scatter_num(col_name, fill_zero=False):
    src = _tbl.column(col_name).to_numpy(zero_copy_only=False).astype("float64")
    out = np.full(ARR_N, np.nan, dtype="float64")
    out[_ids] = src
    if fill_zero:
        out = np.nan_to_num(out, nan=0.0)
    return out


ARR_HARMONIC = _scatter_num("harmonic_pos")
ARR_PR = _scatter_num("pr_pos")
ARR_OUTBOUND = _scatter_num("outbound_count", fill_zero=True).astype("int64")
ARR_NHOSTS = _scatter_num("n_hosts", fill_zero=True).astype("int64")
ARR_REV_ARROW = pc.take(_tbl.column("rev_domain"), pa.array(_inv))  # positionally aligned: ARR_REV_ARROW[id] == rev_domain
del _tbl, _ids, _inv
TOTAL_DOMAINS = ARR_N
print(f"Domain arrays ready in {time.time() - _t:.1f}s ({ARR_N:,} domains).", flush=True)

# The edges file is stored on classic HF LFS -> S3. Hitting resolve/main issues a
# 302 to a *method-specific* presigned S3 URL (valid ~1h). We must resolve with GET
# (not HEAD) so the presigned signature is valid for DuckDB's ranged GETs, cache the
# URL, and transparently re-resolve when it expires (403).
_edges_cache = {"url": None, "ts": 0.0}
_EDGES_TTL = 45 * 60  # refresh well before the 1h expiry


def _resolve_edges_url(force=False):
    if not force and _edges_cache["url"] and (time.time() - _edges_cache["ts"] < _EDGES_TTL):
        return _edges_cache["url"]
    headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
    # stream=True + read nothing: HF hands out a GET-valid presigned URL.
    with _r.get(RESOLVE_URL, headers=headers, allow_redirects=True, stream=True, timeout=30) as resp:
        resp.raise_for_status()
        url = resp.url
    _edges_cache["url"] = url
    _edges_cache["ts"] = time.time()
    return url


# Transparent result cache for remote queries, keyed by the literal SQL text
# (which already contains the specific to_id/domain, and never the presigned
# CDN URL itself since that's swapped in at call time). The dataset is a static
# crawl snapshot, so cached results never go stale within a Space's lifetime.
# This means re-querying the same domain (or re-running the same gap analysis)
# is instant on repeat, with zero network round-trips.
_query_cache: "OrderedDict[str, pd.DataFrame]" = OrderedDict()
_QUERY_CACHE_MAX = 300


def q_edges(sql_template: str):
    """Run a query against the remote edges file, transparently refreshing the
    presigned URL on expiry (403/HTTP error) and caching results by query text.
    sql_template must contain '{EDGES}'."""
    if sql_template in _query_cache:
        _query_cache.move_to_end(sql_template)
        return _query_cache[sql_template]

    url = _resolve_edges_url()
    edges = f"read_parquet('{url}')"
    try:
        result = con.execute(sql_template.format(EDGES=edges)).df()
    except duckdb.HTTPException:
        url = _resolve_edges_url(force=True)
        edges = f"read_parquet('{url}')"
        result = con.execute(sql_template.format(EDGES=edges)).df()

    _query_cache[sql_template] = result
    if len(_query_cache) > _QUERY_CACHE_MAX:
        _query_cache.popitem(last=False)
    return result


# Resolve the CDN URL + warm the httpfs footer cache in a BACKGROUND thread.
# This must NOT block module import: it runs before demo.launch(), and a cold
# metadata read of the 19GB remote edges file can be slow (or hang) -- if it
# blocked here the Gradio port would never bind and the Space would sit in
# "APP_STARTING" forever. Doing it in a daemon thread lets the app go live
# immediately; the first user query just pays the warm-up cost if it hasn't
# finished yet.
def _warm_edges_bg():
    try:
        _u = _resolve_edges_url()
        print(f"CDN URL ready: ...{_u[-60:]}", flush=True)
        _t = time.time()
        q_edges("SELECT from_id FROM {EDGES} WHERE to_id = 1")
        print(f"Edges footer warmed in {time.time() - _t:.1f}s", flush=True)
    except Exception as e:
        print(f"WARN could not pre-resolve/warm edges URL: {e}", flush=True)


import threading
threading.Thread(target=_warm_edges_bg, daemon=True).start()


def rev(domain: str) -> str:
    return ".".join(reversed(domain.strip().lower().lstrip("www.").split(".")))


def fwd(revdom: str) -> str:
    return ".".join(reversed(revdom.split(".")))


def _lookup(domain):
    """Resolve a domain to (id, n_hosts, outbound_count, harmonic_pos, pr_pos).
    id == the array position, since arrays are aligned positionally by id."""
    r = rev(domain)
    pos = pc.index(ARR_REV_ARROW, pa.scalar(r, type=ARR_REV_ARROW.type)).as_py()
    if pos is None or pos < 0:
        return None
    h = ARR_HARMONIC[pos]
    p = ARR_PR[pos]
    return (
        int(pos),
        int(ARR_NHOSTS[pos]),
        int(ARR_OUTBOUND[pos]),
        None if np.isnan(h) else int(h),
        None if np.isnan(p) else int(p),
    )


def _rows_from_ids(ids: np.ndarray) -> pd.DataFrame:
    """O(k) enrichment via direct positional array indexing -- no scan, no SQL,
    no hashing. Sorted by harmonic rank (NULLs last)."""
    harmonic = ARR_HARMONIC[ids]
    pagerank = ARR_PR[ids]
    outbound = ARR_OUTBOUND[ids]
    sort_key = np.where(np.isnan(harmonic), np.inf, harmonic)
    order = np.argsort(sort_key, kind="stable")
    domains = pc.take(ARR_REV_ARROW, pa.array(ids[order])).to_pylist()
    return pd.DataFrame({
        "domain": [fwd(d) for d in domains],
        "harmonic_rank": pd.array(harmonic[order], dtype="Int64"),
        "pagerank_rank": pd.array(pagerank[order], dtype="Int64"),
        "outbound_count": outbound[order],
    })


def _join_ids_locally(ids_df: pd.DataFrame, id_col: str = "from_id") -> pd.DataFrame:
    """Enrich a (small, already-local) id set with each referring domain's name
    and authority. Never touches the network; O(k) positional array gather."""
    if len(ids_df) == 0:
        return pd.DataFrame(columns=["domain", "harmonic_rank", "pagerank_rank", "outbound_count"])
    return _rows_from_ids(ids_df[id_col].to_numpy())


def _paginate(df: pd.DataFrame, page: int, page_size: int):
    if df is None or len(df) == 0:
        return df, "0 / 0"
    n = len(df)
    n_pages = max(1, (n + page_size - 1) // page_size)
    page = max(1, min(page, n_pages))
    start = (page - 1) * page_size
    return df.iloc[start:start + page_size], f"Page {page} / {n_pages}  ({n:,} total rows)"


def _to_csv_file(df: pd.DataFrame, name: str):
    if df is None or len(df) == 0:
        return None
    path = os.path.join(tempfile.gettempdir(), name)
    df.to_csv(path, index=False)
    return path


# ---------------------------------------------------------------- Backlink report

def backlink_report(domain, page_size):
    domain = (domain or "").strip()
    if not domain:
        return "Enter a domain.", None, None, "0 / 0"
    row = _lookup(domain)
    if not row:
        return f"**{domain}** was not found in the Common Crawl web graph.", None, None, "0 / 0"
    did, nhosts, ob, self_harmonic, self_pr = row
    rk = (self_harmonic, self_pr) if self_harmonic is not None else None

    t0 = time.time()
    ids_df = q_edges(f"SELECT from_id FROM {{EDGES}} WHERE to_id = {did}")
    t1 = time.time()
    full = _join_ids_locally(ids_df, "from_id")
    fetch_s = t1 - t0
    join_s = time.time() - t1

    harmonic = f"#{rk[0]:,}" if rk else "-"
    pagerank = f"#{rk[1]:,}" if rk else "-"
    pct_s = f"(top {rk[0] / TOTAL_DOMAINS * 100:.4f}%)" if rk else ""

    cards = f"""
<div style="display:flex;gap:14px;flex-wrap:wrap;margin:10px 0">
  <div style="flex:1;min-width:150px;background:#eef4ff;border-radius:10px;padding:14px;text-align:center">
    <div style="font-size:26px;font-weight:700;color:#1a3d7c">{len(full):,}</div>
    <div style="font-size:12px;color:#555">Referring domains (backlinks)</div>
  </div>
  <div style="flex:1;min-width:150px;background:#eefaf0;border-radius:10px;padding:14px;text-align:center">
    <div style="font-size:26px;font-weight:700;color:#1a7c3d">{harmonic}</div>
    <div style="font-size:12px;color:#555">Harmonic centrality rank {pct_s}</div>
  </div>
  <div style="flex:1;min-width:150px;background:#fff7ea;border-radius:10px;padding:14px;text-align:center">
    <div style="font-size:26px;font-weight:700;color:#a06c00">{pagerank}</div>
    <div style="font-size:12px;color:#555">PageRank rank</div>
  </div>
  <div style="flex:1;min-width:150px;background:#f5eeff;border-radius:10px;padding:14px;text-align:center">
    <div style="font-size:26px;font-weight:700;color:#5b1a7c">{ob:,}</div>
    <div style="font-size:12px;color:#555">Outbound linked domains</div>
  </div>
  <div style="flex:1;min-width:150px;background:#fdeef2;border-radius:10px;padding:14px;text-align:center">
    <div style="font-size:26px;font-weight:700;color:#a01a3d">{nhosts}</div>
    <div style="font-size:12px;color:#555">Crawled hosts (subdomains)</div>
  </div>
</div>
<div style="font-size:11px;color:#888">remote fetch: {fetch_s:.1f}s · enrich/sort: {join_s:.1f}s · {domain}</div>
"""
    page_df, pager = _paginate(full, 1, int(page_size))
    return cards, full, page_df, pager


def report_change_page(full_df, page, page_size, direction):
    if full_df is None:
        return None, "0 / 0", page
    n_pages = max(1, (len(full_df) + int(page_size) - 1) // int(page_size))
    page = max(1, min(page + direction, n_pages))
    page_df, pager = _paginate(full_df, page, int(page_size))
    return page_df, pager, page


def report_export(full_df, domain):
    return _to_csv_file(full_df, f"{(domain or 'domain').strip()}_backlinks.csv")


# ---------------------------------------------------------------- Link gap

def _compute_gap(target, competitors):
    """Core link-gap computation shared by the UI and the JSON API. Returns
    (note, high_full_df, niche_full_df, min_comp, missing_list)."""
    target = (target or "").strip()
    comps = [c.strip() for c in (competitors or "").replace("\n", ",").split(",") if c.strip()]
    empty_df = pd.DataFrame()
    if not target or not comps:
        return ("Enter a target domain and at least one competitor.", empty_df, empty_df, 1, [])
    tv = _lookup(target)
    if not tv:
        return (f"**{target}** not found in the web graph.", empty_df, empty_df, 1, [])
    tid = tv[0]
    comp_ids, missing = [], []
    for c in comps:
        v = _lookup(c)
        (comp_ids if v else missing).append(v[0] if v else c)
    if not comp_ids:
        return ("None of the competitors were found.", empty_df, empty_df, 1, missing)
    ids_list = ",".join(str(i) for i in comp_ids)

    t0 = time.time()
    comp_df = q_edges(f"SELECT DISTINCT from_id FROM {{EDGES}} WHERE to_id IN ({ids_list})")
    tgt_df = q_edges(f"SELECT DISTINCT from_id FROM {{EDGES}} WHERE to_id = {tid}")
    counts_df = q_edges(f"SELECT from_id, count(DISTINCT to_id) AS competitors_linked FROM {{EDGES}} WHERE to_id IN ({ids_list}) GROUP BY from_id")
    fetch_s = time.time() - t0

    con.register("_comp", comp_df)
    con.register("_tgt", tgt_df)
    gap_ids = con.execute("SELECT from_id FROM _comp WHERE from_id NOT IN (SELECT from_id FROM _tgt)").df()
    con.unregister("_comp")
    con.unregister("_tgt")

    high_full = _join_ids_locally(gap_ids, "from_id") if len(gap_ids) else pd.DataFrame()

    min_comp = 2 if len(comp_ids) >= 2 else 1
    con.register("_counts", counts_df)
    con.register("_tgt2", tgt_df)
    niche_ids = con.execute(f"""
        SELECT from_id, competitors_linked FROM _counts
        WHERE from_id NOT IN (SELECT from_id FROM _tgt2) AND competitors_linked >= {min_comp}
    """).df()
    con.unregister("_counts")
    con.unregister("_tgt2")

    if len(niche_ids):
        ids = niche_ids["from_id"].to_numpy()
        harmonic = ARR_HARMONIC[ids]
        mask = (harmonic >= 300) & (harmonic <= 300000)  # NaN comparisons are False -> ranks excluded, matching SQL BETWEEN
        ids, harmonic = ids[mask], harmonic[mask]
        pagerank = ARR_PR[ids]
        outbound = ARR_OUTBOUND[ids]
        comp_linked = niche_ids["competitors_linked"].to_numpy()[mask]
        order = np.lexsort((harmonic, -comp_linked))  # primary: competitors_linked DESC, secondary: harmonic ASC
        domains = pc.take(ARR_REV_ARROW, pa.array(ids[order])).to_pylist()
        niche_full = pd.DataFrame({
            "domain": [fwd(d) for d in domains],
            "harmonic_rank": pd.array(harmonic[order], dtype="Int64"),
            "pagerank_rank": pd.array(pagerank[order], dtype="Int64"),
            "outbound_count": outbound[order],
            "competitors_linked": comp_linked[order],
        })
    else:
        niche_full = pd.DataFrame()

    note = f"### Link-gap for {target}\nCompetitors analyzed: {', '.join(comps)} · fetch: {fetch_s:.1f}s"
    if missing:
        note += f"\n\n_(not found, skipped: {', '.join(missing)})_"
    note += f"\n\n**{len(high_full):,}** total gap domains found · **{len(niche_full):,}** are niche-relevant & attainable (link to ≥{min_comp} competitor(s), harmonic rank 300–300,000)."
    return note, high_full, niche_full, min_comp, missing


def link_gap(target, competitors, page_size):
    note, high_full, niche_full, _min_comp, _missing = _compute_gap(target, competitors)
    high_page, high_pager = _paginate(high_full, 1, int(page_size))
    niche_page, niche_pager = _paginate(niche_full, 1, int(page_size))
    return note, high_full, high_page, high_pager, niche_full, niche_page, niche_pager


def gap_change_page(full_df, page, page_size, direction):
    if full_df is None:
        return None, "0 / 0", page
    n_pages = max(1, (len(full_df) + int(page_size) - 1) // int(page_size))
    page = max(1, min(page + direction, n_pages))
    page_df, pager = _paginate(full_df, page, int(page_size))
    return page_df, pager, page


def gap_export(full_df, target, suffix):
    return _to_csv_file(full_df, f"{(target or 'domain').strip()}_{suffix}.csv")


# ---------------------------------------------------------------- JSON API
# Dedicated endpoints for external clients (e.g. the Cloudflare Worker UI).
# These return the FULL result set as clean JSON in one call -- no HTML card
# parsing and no page_size/dropdown limits like the Gradio UI functions have.

def _records(df):
    """DataFrame -> list[dict] with NaN/NA -> None and native python scalars."""
    if df is None or len(df) == 0:
        return []
    recs = df.astype(object).where(pd.notnull(df), None).to_dict("records")
    out = []
    for r in recs:
        out.append({
            k: (int(v) if isinstance(v, np.integer) else float(v) if isinstance(v, np.floating) else v)
            for k, v in r.items()
        })
    return out


def api_backlink_report(domain):
    domain = (domain or "").strip()
    if not domain:
        return {"found": False, "error": "empty domain"}
    row = _lookup(domain)
    if not row:
        return {"found": False, "domain": domain}
    did, nhosts, ob, sh, spr = row
    full = _join_ids_locally(q_edges(f"SELECT from_id FROM {{EDGES}} WHERE to_id = {did}"), "from_id")
    return {
        "found": True,
        "domain": domain,
        "referringDomains": int(len(full)),
        "harmonicRank": sh,
        "pagerankRank": spr,
        "harmonicTopPct": round(sh / TOTAL_DOMAINS * 100, 4) if sh else None,
        "outboundCount": int(ob),
        "crawledHosts": int(nhosts),
        "rows": _records(full),
    }


def api_link_gap(target, competitors):
    note, high_full, niche_full, min_comp, missing = _compute_gap(target, competitors)
    return {
        "target": (target or "").strip(),
        "note": note,
        "minCompetitors": min_comp,
        "missing": missing,
        "high": _records(high_full),
        "niche": _records(niche_full),
    }


CSS = """
.gradio-container {max-width: 1250px !important}
h1 {color:#1a3d7c}
.pager-row {align-items:center}
"""

with gr.Blocks(title="CC Backlink Explorer", css=CSS, theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🔗 CC Backlink Explorer\nMajestic/Ahrefs-style backlinks & rank booster, built on the **Common Crawl** domain web graph (2026 Apr–May–Jun · 121M domains · 3.9B links). Harmonic centrality ≈ Trust Flow, PageRank ≈ Citation Flow (no anchor-text / referring-IP data at domain-graph granularity).")

    with gr.Tab("Backlink report"):
        with gr.Row():
            d_in = gr.Textbox(label="Domain", placeholder="stripe.com", scale=3)
            d_page_size = gr.Dropdown([25, 50, 100, 250, 500], value=50, label="Rows per page", scale=1)
            d_btn = gr.Button("Analyze", variant="primary", scale=1)
        d_md = gr.Markdown()
        d_full_state = gr.State(None)
        d_page_state = gr.State(1)
        with gr.Row(elem_classes="pager-row"):
            d_prev = gr.Button("◀ Prev", size="sm")
            d_pager = gr.Markdown("0 / 0")
            d_next = gr.Button("Next ▶", size="sm")
            d_export = gr.DownloadButton("⬇ Download full CSV", size="sm")
        d_df = gr.Dataframe(label="Referring domains (sorted by their own authority)", wrap=True)

        d_btn.click(backlink_report, [d_in, d_page_size], [d_md, d_full_state, d_df, d_pager])
        d_in.submit(backlink_report, [d_in, d_page_size], [d_md, d_full_state, d_df, d_pager])
        d_prev.click(lambda f, p, s: report_change_page(f, p, s, -1), [d_full_state, d_page_state, d_page_size], [d_df, d_pager, d_page_state])
        d_next.click(lambda f, p, s: report_change_page(f, p, s, 1), [d_full_state, d_page_state, d_page_size], [d_df, d_pager, d_page_state])
        d_export.click(report_export, [d_full_state, d_in], d_export)

    with gr.Tab("Link gap / rank booster"):
        gr.Markdown("Find high-authority domains that link to your **competitors but not you** — the best targets to raise your harmonic/PageRank.")
        with gr.Row():
            g_t = gr.Textbox(label="Your domain", placeholder="metehan.ai", scale=2)
            g_c = gr.Textbox(label="Competitors (comma-separated)", placeholder="peec.ai, ahrefs.com, semrush.com", scale=3)
            g_page_size = gr.Dropdown([25, 50, 100, 250, 500], value=50, label="Rows per page", scale=1)
        g_btn = gr.Button("Find link gaps", variant="primary")
        g_md = gr.Markdown()
        g_high_state = gr.State(None)
        g_niche_state_full = gr.State(None)
        g_high_page = gr.State(1)
        g_niche_page = gr.State(1)

        gr.Markdown("**Actionable / niche-relevant targets** (link to ≥2 competitors, realistic attainable authority):")
        with gr.Row(elem_classes="pager-row"):
            gn_prev = gr.Button("◀ Prev", size="sm")
            gn_pager = gr.Markdown("0 / 0")
            gn_next = gr.Button("Next ▶", size="sm")
            gn_export = gr.DownloadButton("⬇ Download full CSV", size="sm")
        g_niche = gr.Dataframe(wrap=True)

        gr.Markdown("**All gap domains** (every domain linking to a competitor but not you, sorted by authority):")
        with gr.Row(elem_classes="pager-row"):
            gh_prev = gr.Button("◀ Prev", size="sm")
            gh_pager = gr.Markdown("0 / 0")
            gh_next = gr.Button("Next ▶", size="sm")
            gh_export = gr.DownloadButton("⬇ Download full CSV", size="sm")
        g_high = gr.Dataframe(wrap=True)

        g_btn.click(
            link_gap, [g_t, g_c, g_page_size],
            [g_md, g_high_state, g_high, gh_pager, g_niche_state_full, g_niche, gn_pager],
        )
        gh_prev.click(lambda f, p, s: gap_change_page(f, p, s, -1), [g_high_state, g_high_page, g_page_size], [g_high, gh_pager, g_high_page])
        gh_next.click(lambda f, p, s: gap_change_page(f, p, s, 1), [g_high_state, g_high_page, g_page_size], [g_high, gh_pager, g_high_page])
        gh_export.click(lambda f, t: gap_export(f, t, "link_gap_all"), [g_high_state, g_t], gh_export)
        gn_prev.click(lambda f, p, s: gap_change_page(f, p, s, -1), [g_niche_state_full, g_niche_page, g_page_size], [g_niche, gn_pager, g_niche_page])
        gn_next.click(lambda f, p, s: gap_change_page(f, p, s, 1), [g_niche_state_full, g_niche_page, g_page_size], [g_niche, gn_pager, g_niche_page])
        gn_export.click(lambda f, t: gap_export(f, t, "link_gap_niche"), [g_niche_state_full, g_t], gn_export)

    gr.Markdown("<sub>Data: Common Crawl web graph (CC-MAIN-2026 Apr-May-Jun), harmonic centrality & PageRank. URL-level backlinks are not available in the web graph (host & domain level only). Full result sets are always available via CSV export, not just the on-screen page.</sub>")

    # Hidden JSON API endpoints for external clients (Cloudflare Worker UI).
    # api_backlink(domain) and api_gap(target, competitors) each return the full
    # result set as JSON in a single call -- no dropdown/page_size constraints.
    with gr.Row(visible=False):
        _api_dom = gr.Textbox()
        _api_report_out = gr.JSON()
        _api_report_btn = gr.Button()
        _api_gap_t = gr.Textbox()
        _api_gap_c = gr.Textbox()
        _api_gap_out = gr.JSON()
        _api_gap_btn = gr.Button()
    _api_report_btn.click(api_backlink_report, _api_dom, _api_report_out, api_name="api_backlink")
    _api_gap_btn.click(api_link_gap, [_api_gap_t, _api_gap_c], _api_gap_out, api_name="api_gap")

if __name__ == "__main__":
    demo.queue(max_size=32, default_concurrency_limit=4).launch()