"""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"""