"""ETERNAL HuggingFace Spaces firehose: api/spaces walk + raw file scan.

Anonymous. api 500/5min, resolvers 3000/5min. List sort=lastModified desc;
per space tree/main -> raw text files. Seen = id:lastModified (rescan on
change). mark_seen ONLY on success.
"""
import os
KF_BASE = os.path.dirname(os.path.abspath(__file__))
os.makedirs(os.path.join(KF_BASE, "data"), exist_ok=True)
import hashlib
import sqlite3
import sys
import threading
import time

import requests

DB = os.environ.get("KF_DB") or os.path.join(KF_BASE, "data/keys.db")
LOG = os.path.join(KF_BASE, "data/hfspaces_eternal.log")
SEEN_FILE = os.path.join(KF_BASE, "data/hfspaces_seen.txt")
N_THREADS = 4
SPACES_PER_CYCLE = 100
FILE_CAP = 500_000
SLEEP = 120

from key_patterns import scan_text

GARBAGE = ("xxxx", "your", "YOUR", "EXAMPLE", "example", "n0tr3al",
           "placeholder", "CHANGEME", "changeme", "dummy", "DUMMY",
           "abcdefghijklmnopqrst", "1234567890abcdef")

SCAN_EXT = (".py", ".json", ".yaml", ".yml", ".toml", ".env", ".sh", ".js",
            ".ts", ".cfg", ".ini", ".md", ".txt", ".ipynb", ".xml")

UA = {"User-Agent": "keyfarm-research/1.0"}


def log(msg):
    line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n"
    with open(LOG, "a", encoding="utf-8") as f:
        f.write(line)


conn = sqlite3.connect(DB, timeout=120, check_same_thread=False,
                       isolation_level=None)
lock = threading.Lock()
total = 0


def db_add(key, prov, repo):
    global total
    with lock:
        conn.execute(
            "INSERT INTO keys (hash, val, prov, status, found, repo) "
            "VALUES (?,?,?,?,?,?) "
            "ON CONFLICT(hash) DO UPDATE SET status='NEW', "
            "found=excluded.found, repo=excluded.repo "
            "WHERE keys.status IN ('DEAD','ERR')",
            (hashlib.sha1(key.encode()).hexdigest()[:16],
             key, prov, "NEW", "hf", repo))
        total += 1


def load_seen():
    s = {}
    try:
        with open(SEEN_FILE, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if "\t" in line:
                    k, v = line.split("\t", 1)
                    s[k] = v
    except Exception:
        pass
    return s


seen = load_seen()


def mark_seen(sid, mtime):
    seen[sid] = mtime
    with open(SEEN_FILE, "a", encoding="utf-8") as f:
        f.write(f"{sid}\t{mtime}\n")


def _get(url, timeout=20):
    for attempt in (1, 2):
        try:
            r = requests.get(url, headers=UA, timeout=timeout)
            if r.status_code == 429:
                time.sleep(30)
                continue
            return r
        except Exception:
            if attempt == 2:
                return None
            time.sleep(5)
    return None


def work(sid):
    """Scan one space: tree + raw text files. -1 on final fail."""
    n = 0
    r = _get(f"https://huggingface.co/api/spaces/{sid}/tree/main?recursive=true")
    if r is None:
        return -1
    if r.status_code == 404:
        return 0
    if r.status_code != 200:
        return -1
    try:
        tree = r.json()
    except Exception:
        return -1
    paths = []
    for ent in tree:
        if ent.get("type") != "file":
            continue
        p, sz = ent.get("path", ""), ent.get("size", 0)
        if not p or sz > FILE_CAP:
            continue
        if not any(p.lower().endswith(e) for e in SCAN_EXT):
            continue
        paths.append(p)
    for p in paths:
        r2 = _get(f"https://huggingface.co/spaces/{sid}/raw/main/{p}")
        if r2 is None or r2.status_code != 200:
            continue
        text = r2.text
        for prov, key in scan_text(text):
            if isinstance(key, tuple):
                key = key[0]
            if any(g in key for g in GARBAGE) or len(key) < 30:
                continue
            db_add(key, prov, f"hf:{sid}")
            n += 1
    return n


def cycle():
    global total
    total = 0
    done = 0
    offset = 0
    todo = []
    while done < SPACES_PER_CYCLE and offset < 1000:
        url = ("https://huggingface.co/api/spaces?limit=100&sort=lastModified"
               f"&direction=-1&skip={offset}")
        r = _get(url, timeout=25)
        if r is None or r.status_code != 200:
            log(f"list-err offset={offset} "
                f"{r.status_code if r is not None else 'net'}")
            break
        try:
            batch = r.json()
        except Exception:
            break
        if not batch:
            break
        for sp in batch:
            sid = sp.get("id")
            mtime = (sp.get("lastModified") or "")[:19]
            if not sid:
                continue
            if seen.get(sid) == mtime:
                continue
            todo.append((sid, mtime))
            done += 1
            if done >= SPACES_PER_CYCLE:
                break
        offset += 100
        time.sleep(2)

    ok = fail = 0
    from concurrent.futures import ThreadPoolExecutor, as_completed
    with ThreadPoolExecutor(max_workers=N_THREADS) as ex:
        futs = {ex.submit(work, sid): (sid, mt) for sid, mt in todo}
        for fut in as_completed(futs):
            sid, mt = futs[fut]
            try:
                res = fut.result()
            except Exception:
                res = -1
            if res == -1:
                fail += 1
            else:
                ok += 1
                mark_seen(sid, mt)

    log(f"cycle: todo={len(todo)} ok={ok} fail={fail} +{total} keys")


def main():
    log("hfspaces_eternal started")
    if "--once" in sys.argv:
        cycle()
        return
    while True:
        try:
            cycle()
        except Exception as e:
            log(f"cycle-err {type(e).__name__}: {e}")
        time.sleep(SLEEP)


if __name__ == "__main__":
    main()