code/canon_build.py

"""Canonical skills table (steps 1+2): one row per distinct SKILL.md content across every source, pointing at the most
complete bundle, with quality / language / safety / license flags.

Stages (resumable; each reads what the previous wrote under work/canon/):
  scan    stream every processed skills shard (downloaded in batches by short-lived child processes, deleted after use):
          meta/   one row per source record (no text): sha, source, bundle uid, repo, completeness, stars, license, ...
          text/   SKILL.md text, written only the first time a content sha is seen (identical sha = identical text)
  select  DuckDB over meta: per sha pick the canonical bundle + aggregates (copies, repos, sources, stars, safety)
  final   join canonical rows with text; re-parse frontmatter uniformly (lib_skill); add flags + tiers
  upload  processed/canonical/skills/*.parquet + stats/canonical_summary.json

sha = git blob SHA of the SKILL.md bytes (GitSkills file_sha / crawl blob_sha); ClawHub text is hashed the same way.
Usage: python code/canon_build.py [scan|select|final|upload|all]
"""
import hashlib, json, os, re, subprocess, sys, time
from pathlib import Path
import pyarrow as pa, pyarrow.parquet as pq

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "code"))
os.environ.setdefault("HOME", "/root")
from dotenv import load_dotenv
load_dotenv(ROOT / ".env")
from lib_skill import parse_skill_md, cjk_ratio

BUCKET = "Mercity/SkillsStorage"
W = ROOT / ("work/canon_test" if os.environ.get("CANON_TEST") else "work/canon")
for d in ("meta", "text", "dl", "canon", "final"): (W / d).mkdir(parents=True, exist_ok=True)
CRAWL = ["backfill_gitskills", "pass1_sourcegraph", "pass2_new"]

META = pa.schema([("sha", pa.string()), ("source", pa.string()), ("bundle_uid", pa.string()), ("repo", pa.string()),
    ("skill_dir", pa.string()), ("ref", pa.string()), ("snapshot", pa.string()), ("n_files", pa.int64()),
    ("n_files_with_content", pa.int64()), ("bundle_bytes", pa.int64()), ("listing_truncated", pa.bool_()),
    ("stars", pa.int64()), ("is_fork", pa.bool_()), ("license", pa.string()), ("language", pa.string()),
    ("last_commit_at", pa.string()), ("sec_verdict", pa.string()), ("sec_flagged", pa.bool_())])
TEXT = pa.schema([("sha", pa.string()), ("skill_md", pa.string())])


def blob_sha(text):
    b = (text or "").encode("utf-8")
    return hashlib.sha1(b"blob %d\0" % len(b) + b).hexdigest()


def list_remote(prefix):
    from huggingface_hub import HfApi
    return sorted(f.path for f in HfApi().list_bucket_tree(BUCKET, prefix=prefix, recursive=True)
                  if getattr(f, "size", None) is not None and f.path.endswith(".parquet"))


def download(paths):
    """Child process does the download so the xet client's memory is released afterwards."""
    spec = json.dumps([[p, str(W / "dl" / p.replace("/", "__"))] for p in paths])
    code = ("import json,sys; from huggingface_hub import HfApi; from dotenv import load_dotenv; load_dotenv('/root/skills-db/.env');"
            "HfApi().download_bucket_files('Mercity/SkillsStorage', [tuple(x) for x in json.loads(sys.stdin.read())])")
    for i in range(5):
        r = subprocess.run([sys.executable, "-c", code], input=spec, text=True, capture_output=True)
        if r.returncode == 0: return [W / "dl" / p.replace("/", "__") for p in paths]
        time.sleep(20 * (i + 1))
    raise RuntimeError(f"download failed: {r.stderr[-300:]}")


def repo_meta():
    """GitSkills repos (stars/license/fork/language) + topic-search stars/licenses for repos created since July."""
    t = pq.read_table(ROOT / "work/gitskills/data/repos/part-00000.parquet",
                      columns=["full_name", "stars", "is_fork", "license", "language"]).to_pylist()
    m = {r["full_name"]: (r["stars"], bool(r["is_fork"]) if r["is_fork"] is not None else None, r["license"] or None, r["language"]) for r in t}
    p = ROOT / "data/seeds/github_topics.jsonl"
    if p.exists():
        for l in open(p):
            r = json.loads(l)
            if r.get("full_name") and r["full_name"] not in m:
                m[r["full_name"]] = (r.get("stars"), r.get("fork"), (r.get("license") or None), None)
    return m


def scan():
    done_f = W / "scan_done.txt"; done = set(done_f.read_text().split()) if done_f.exists() else set()
    seen = set()
    for p in list((W / "text").glob("*.tmp")) + list((W / "meta").glob("*.tmp")): p.unlink()
    for p in sorted((W / "text").glob("*.parquet")):  # resume: rebuild the seen-set from text already written
        try:
            seen.update(pq.read_table(p, columns=["sha"]).column("sha").to_pylist())
        except Exception:   # partially written by a killed run: its shard was never marked done, so it is re-scanned
            print("  dropping unreadable", p.name, flush=True); p.unlink(); (W / "meta" / p.name).unlink(missing_ok=True)
    repos = repo_meta()
    jobs = [(r, p) for r in CRAWL for p in list_remote(f"processed/github_delta/{r}/skills/")]
    jobs += [("gitskills", p) for p in list_remote("processed/gitskills/skills/")]
    jobs += [("clawhub", "processed/clawhub/skills.parquet"), ("clawhub_api", "processed/clawhub/api_delta_skills.parquet")]
    jobs = [j for j in jobs if j[1] not in done]
    if os.environ.get("CANON_TEST"):  # 2 crawl shards, 1 GitSkills part, ClawHub API delta
        jobs = [j for j in jobs if j[0] == "pass1_sourcegraph"][:2] + [j for j in jobs if j[0] == "gitskills"][-1:] + [j for j in jobs if j[0] == "clawhub_api"]
    print(f"scan: {len(jobs)} shards to go, {len(seen):,} distinct texts so far", flush=True)
    batches, cur = [], []
    for j in jobs:  # GitSkills parts are ~170 MB each: 2 per batch; crawl shards are small: 40 per batch
        cur.append(j)
        if len(cur) >= (2 if j[0] == "gitskills" else 40): batches.append(cur); cur = []
    if cur: batches.append(cur)
    done_n = 0
    for batch in batches:
        b = done_n; done_n += len(batch)
        for (src, remote), local in zip(batch, download([p for _, p in batch])):
            meta, text = [], []
            name = remote.replace("/", "__"); writers = {}

            def flush_rows():
                for sub, rows_, sch in (("meta", meta, META), ("text", text, TEXT)):
                    if not rows_: continue
                    if sub not in writers: writers[sub] = pq.ParquetWriter(W / sub / (name + ".tmp"), sch, compression="zstd")
                    writers[sub].write_table(pa.Table.from_pylist(rows_, sch)); rows_.clear()
            if src == "gitskills":
                cols = ["skill_uid", "repo", "skill_dir", "is_skill_md", "file_sha", "skill_md", "sibling_count", "sibling_bytes",
                        "composition_truncated", "last_commit_at", "repo_stars", "repo_is_fork", "repo_license", "repo_language"]
            elif src.startswith("clawhub"):
                cols = None
            else:
                cols = ["skill_uid", "repo", "commit", "skill_dir", "file_sha", "skill_md", "n_bundle_files",
                        "n_bundle_files_with_content", "bundle_bytes", "fetched_at"]
            pf = pq.ParquetFile(local, buffer_size=1 << 20, pre_buffer=False)
            for bt in pf.iter_batches(batch_size=2000, columns=cols, use_threads=False):
                for r in bt.to_pylist():
                    if src == "gitskills":
                        if not r["is_skill_md"]: continue
                        sha = r["file_sha"]; nf = r["sibling_count"] or 0
                        meta.append({"sha": sha, "source": src, "bundle_uid": r["skill_uid"], "repo": r["repo"], "skill_dir": r["skill_dir"],
                                     "ref": None, "snapshot": "2026-07", "n_files": nf, "n_files_with_content": None,
                                     "bundle_bytes": r["sibling_bytes"], "listing_truncated": r["composition_truncated"],
                                     "stars": r["repo_stars"], "is_fork": r["repo_is_fork"], "license": r["repo_license"] or None,
                                     "language": r["repo_language"], "last_commit_at": r["last_commit_at"], "sec_verdict": None, "sec_flagged": None})
                    elif src.startswith("clawhub"):
                        sha = blob_sha(r["skill_md"]); flagged = None
                        if src == "clawhub":
                            # SkillSpector marks ~87% of ClawHub "suspicious" (useless as a filter): use the registry's own
                            # ClawScan verdict, static "malicious", and VirusTotal detections
                            flagged = (r.get("clawscan_verdict") in ("malicious", "suspicious") or r.get("static_status") == "malicious"
                                       or (r.get("virustotal_malicious_count") or 0) > 0)
                        meta.append({"sha": sha, "source": src, "bundle_uid": r["skill_uid"], "repo": f"clawhub:{r['owner']}/{r['slug']}",
                                     "skill_dir": ".", "ref": r.get("version"), "snapshot": r.get("snapshot") or "2026-10",
                                     "n_files": r.get("n_bundle_files"), "n_files_with_content": r.get("n_bundle_files"),
                                     "bundle_bytes": r.get("bundle_bytes"), "listing_truncated": False, "stars": None, "is_fork": False,
                                     "license": "MIT-0", "language": None, "last_commit_at": None,
                                     "sec_verdict": r.get("clawscan_verdict"), "sec_flagged": flagged})
                    else:
                        sha = r["file_sha"]; rm = repos.get(r["repo"]) or (None, None, None, None)
                        meta.append({"sha": sha, "source": src, "bundle_uid": r["skill_uid"], "repo": r["repo"], "skill_dir": r["skill_dir"],
                                     "ref": r["commit"], "snapshot": (r["fetched_at"] or "")[:10], "n_files": r["n_bundle_files"],
                                     "n_files_with_content": r["n_bundle_files_with_content"], "bundle_bytes": r["bundle_bytes"],
                                     "listing_truncated": False, "stars": rm[0], "is_fork": rm[1], "license": rm[2], "language": rm[3],
                                     "last_commit_at": None, "sec_verdict": None, "sec_flagged": None})
                    if sha and sha not in seen and r.get("skill_md") is not None:
                        seen.add(sha); text.append({"sha": sha, "skill_md": r["skill_md"]})
                flush_rows()
            flush_rows()
            for sub, wr in writers.items():   # atomic: tmp + rename once the shard is complete
                wr.close(); os.replace(W / sub / (name + ".tmp"), W / sub / name)
            local.unlink()
            with open(done_f, "a") as f: f.write(remote + "\n")
        print(f"  {b + len(batch)}/{len(jobs)} shards, {len(seen):,} distinct texts", flush=True)
    print("scan done", len(seen), flush=True)


def duck(mem="900MB"):
    import duckdb
    c = duckdb.connect(); c.execute(f"SET memory_limit='{mem}'; SET threads=2; SET preserve_insertion_order=false; SET temp_directory='{W}/duck_tmp'")
    return c


def select():
    """Per sha: canonical bundle = most complete, then most-starred, non-fork, freshest source."""
    c = duck()
    occ = W / "occ_counts.parquet"
    if not occ.exists():  # copies of each SKILL.md across GitHub in July (incl. verbatim copies GitSkills de-duplicated)
        from bucket_query import urls, connect
        cc = connect("400MB"); cc.execute(f"""copy (select file_sha sha, count(*) n_copies_jul, count(distinct repo) n_repos_jul
            from read_parquet({urls('processed/gitskills/occurrences/')}) where filename = 'SKILL.md' group by 1) to '{occ}' (format parquet)""")
        cc.close()
    for pre in "0123456789abcdef":   # 16 sha-prefix partitions: bounded DuckDB memory for the window over 4.4M rows
      outp = W / "canon" / f"canonical_meta-{pre}.parquet"
      if outp.exists(): continue
      c.execute(f"""copy (
      with m as (select *,
          case when source in ('backfill_gitskills','pass1_sourcegraph','pass2_new','clawhub','clawhub_api')
                    then coalesce(n_files_with_content::double / nullif(n_files, 0), 1.0)
               when coalesce(n_files, 0) = 0 then 1.0 else 0.5 end as completeness,
          case source when 'backfill_gitskills' then 3 when 'pass2_new' then 3 when 'pass1_sourcegraph' then 3
                      when 'clawhub_api' then 2 when 'clawhub' then 2 else 1 end as freshness
        from read_parquet('{W}/meta/*.parquet') where sha like '{pre}%'),
      r as (select *, row_number() over (partition by sha order by completeness desc, coalesce(stars, -1) desc,
                coalesce(is_fork, false), freshness desc, bundle_uid) rn from m),
      agg as (select sha, count(*) n_records, count(distinct repo) n_repos, list(distinct source order by source) sources,
                max(stars) max_stars, bool_or(coalesce(sec_flagged, false)) sec_flagged_any,
                bool_or(source like 'clawhub%') on_clawhub, bool_or(source not like 'clawhub%') on_github from m group by sha)
      select r.sha, r.source canon_source, r.bundle_uid canon_bundle_uid, r.repo canon_repo, r.skill_dir canon_skill_dir, r.ref canon_ref,
             r.snapshot canon_snapshot, r.n_files, r.n_files_with_content, r.completeness, r.bundle_bytes, r.listing_truncated,
             r.stars canon_stars, r.is_fork canon_is_fork, r.license canon_license, r.language canon_language, r.last_commit_at,
             r.sec_verdict canon_sec_verdict, agg.* exclude (sha), coalesce(o.n_copies_jul, 0) n_copies_jul, coalesce(o.n_repos_jul, 0) n_repos_jul
      from r join agg using (sha) left join '{occ}' o using (sha) where r.rn = 1
    ) to '{outp}.tmp' (format parquet, compression zstd)""")
      os.replace(f"{outp}.tmp", outp); print(f"  select partition {pre} done", flush=True)
    print("select done", c.execute(f"select count(*) from '{W}/canon/canonical_meta-*.parquet'").fetchone()[0], flush=True)


PERMISSIVE = {"mit", "mit-0", "apache-2.0", "bsd-2-clause", "bsd-3-clause", "isc", "0bsd", "unlicense", "cc0-1.0", "zlib", "bsl-1.0", "wtfpl"}
ATTRIB = {"cc-by-4.0", "cc-by-3.0"}
COPYLEFT = re.compile(r"^(a?gpl|lgpl|mpl|epl|eupl|cc-by-sa|cc-by-nc|osl|cpal)", re.I)
STUB = re.compile(r"\b(todo|lorem ipsum|replace this|your skill description|placeholder|tbd)\b", re.I)
SOURCE_AVAILABLE = {("anthropics/skills", d) for d in ("skills/docx", "skills/pdf", "skills/pptx", "skills/xlsx")}


def license_tier(lic, repo):
    if repo and repo.lower().startswith("trailofbits/"): return "copyleft"
    if not lic: return "none"
    l = lic.lower()
    if l in PERMISSIVE: return "permissive"
    if l in ATTRIB: return "attribution"
    if COPYLEFT.match(l): return "copyleft"
    return "other"   # NOASSERTION / custom


def _hash64(x):
    return int.from_bytes(hashlib.blake2b((x or "").strip().lower().encode(), digest_size=8).digest(), "little", signed=True)


def _part_stats(c, path, out):
    cols = ["flag_stub", "flag_cjk", "flag_template_desc", "flag_security", "train_ok", "release_ok", "frontmatter_valid"]
    summ = dict(zip(["rows"] + cols, c.execute("select count(*), " + ", ".join(f"count(*) filter (where {k})" for k in cols)
                                               + f" from '{path}'").fetchone()))
    for key, q in [("license_tier", "license_tier"), ("canon_source", "canon_source")]:
        summ[key] = dict(c.execute(f"select {q}, count(*) from '{path}' group by 1").fetchall())
    summ["train_ok_by_license_tier"] = dict(c.execute(f"select license_tier, count(*) from '{path}' where train_ok group by 1").fetchall())
    Path(out).write_text(json.dumps(summ))


def repartition_text():
    """One pass: text/*.parquet -> textp/<hex>.parquet (16 files) so each final partition reads only its own text."""
    out = W / "textp"; out.mkdir(exist_ok=True)
    if (out / "done.ok").exists(): return
    writers = {h: pq.ParquetWriter(out / f"{h}.parquet.tmp", TEXT, compression="zstd") for h in "0123456789abcdef"}
    import pyarrow.compute as pc
    for i, p in enumerate(sorted((W / "text").glob("*.parquet"))):
        t = pq.read_table(p)
        first = pc.utf8_slice_codeunits(t["sha"], 0, 1)
        for h in "0123456789abcdef":
            part = t.filter(pc.equal(first, h))
            if part.num_rows: writers[h].write_table(part)
        if i % 300 == 0: print(f"  repartition: {i} files", flush=True)
    for h, w in writers.items(): w.close(); os.replace(out / f"{h}.parquet.tmp", out / f"{h}.parquet")
    (out / "done.ok").write_text("ok\n"); print("repartition done", flush=True)


def final():
    """Pass 1: per distinct text, (desc hash, name hash) -> template descriptions = one description shared by >=50 distinct
    contents carrying >=20 different names (boilerplate spam; plain copies of one skill share the name too and are left
    to near-dup clustering). Pass 2: join canonical rows with text in 16 sha-prefix partitions (bounded RAM), re-parse
    frontmatter uniformly, add flags and tiers."""
    c = duck()
    dn = W / "canon/desc_name.parquet"
    if not dn.exists():
        rows = []; w = None
        for p in sorted((W / "text").glob("*.parquet")):
            for bt in pq.ParquetFile(p).iter_batches(batch_size=5000):
                for r in bt.to_pylist():
                    ok, name, desc, fm, body = parse_skill_md(r["skill_md"] or "")
                    if desc: rows.append({"sha": r["sha"], "dh": _hash64(desc), "nh": _hash64(name)})
            if len(rows) > 200000:
                t = pa.Table.from_pylist(rows); w = w or pq.ParquetWriter(dn, t.schema); w.write_table(t); rows = []
        if rows:
            t = pa.Table.from_pylist(rows); w = w or pq.ParquetWriter(dn, t.schema); w.write_table(t)
        w.close()
    template = set(x[0] for x in c.execute(f"""select dh from '{dn}' group by dh
                                               having count(*) >= 50 and count(distinct nh) >= 20""").fetchall())
    desc_n = dict(c.execute(f"select dh, count(*) from '{dn}' group by dh having count(*) >= 20").fetchall())
    print(f"final: {len(template):,} template descriptions", flush=True)
    tally = {}; n_rows = 0
    meta_schema = pq.read_schema(next((W / "canon").glob("canonical_meta-*.parquet")))
    OUT = pa.schema(list(meta_schema) + [pa.field(n, t) for n, t in [
        ("skill_md", pa.string()), ("name", pa.string()), ("description", pa.string()), ("frontmatter_valid", pa.bool_()),
        ("body_chars", pa.int64()), ("desc_chars", pa.int64()), ("cjk_ratio", pa.float64()), ("license_tier", pa.string()),
        ("flag_stub", pa.bool_()), ("flag_cjk", pa.bool_()), ("flag_template_desc", pa.bool_()), ("flag_security", pa.bool_()),
        ("desc_shared_count", pa.int64()), ("train_ok", pa.bool_()), ("release_ok", pa.bool_())]])   # explicit: no null-type inference
    import pyarrow.compute as pc
    for group in ["01", "23", "45", "67", "89", "ab", "cd", "ef"]:   # 8 passes over text/, 2 prefixes each: no extra disk
        todo = [pre for pre in group if not (W / "final" / f"part-{pre}.done").exists()]
        if not todo: continue
        # compact Arrow table + per-file Arrow hash join (a dict of 320k row-dicts cost ~900 MB)
        meta_tbl = pa.concat_tables([pq.read_table(W / "canon" / f"canonical_meta-{pre}.parquet") for pre in todo])
        import pyarrow.compute as pc
        si = meta_tbl.schema.get_field_index("sources")   # Arrow joins can't carry list columns: ship as "a,b" text
        meta_tbl = meta_tbl.set_column(si, "sources", pc.binary_join(meta_tbl["sources"], ","))
        writers, outs = {}, {pre: W / "final" / f"part-{pre}.parquet" for pre in todo}
        for p in sorted((W / "text").glob("*.parquet")):
            t = pq.read_table(p)
            t = t.filter(pc.is_in(pc.utf8_slice_codeunits(t["sha"], 0, 1), value_set=pa.array(todo)))
            if not t.num_rows: continue
            out = {pre: [] for pre in todo}
            for r in t.join(meta_tbl, keys="sha", join_type="inner").to_pylist():
                md = r.pop("skill_md") or ""; tr = {"sha": r["sha"]}
                r["sources"] = r["sources"].split(",") if r["sources"] else []
                ok, name, desc, fm, body = parse_skill_md(md)
                cjk = cjk_ratio(md); dl = (desc or "").strip().lower(); dh = _hash64(desc) if desc else None
                tier_lic = "permissive" if r["canon_source"].startswith("clawhub") else license_tier(r["canon_license"], r["canon_repo"])
                if (r["canon_repo"], r["canon_skill_dir"]) in SOURCE_AVAILABLE: tier_lic = "source_available"
                f_stub = len(body.strip()) < 100 or len(dl) < 20 or bool(STUB.search(dl))
                f_cjk, f_template, f_sec = cjk > 0.10, dh in template, bool(r["sec_flagged_any"])
                train_ok = bool(name and desc) and not (f_stub or f_cjk or f_sec or f_template)
                out[tr["sha"][0]].append({**r, "skill_md": md, "name": name, "description": desc, "frontmatter_valid": ok,
                            "body_chars": len(body), "desc_chars": len(desc or ""), "cjk_ratio": cjk, "license_tier": tier_lic,
                            "flag_stub": f_stub, "flag_cjk": f_cjk, "flag_template_desc": f_template, "flag_security": f_sec,
                            "desc_shared_count": desc_n.get(dh, 1) if dh is not None else 0, "train_ok": train_ok,
                            "release_ok": train_ok and tier_lic in ("permissive", "attribution")})
            pa.default_memory_pool().release_unused()   # Arrow's allocator kept join memory across files (2.3 GB by pass 8)
            for pre, rows_ in out.items():
                if not rows_: continue
                if pre not in writers: writers[pre] = pq.ParquetWriter(str(outs[pre]) + ".tmp", OUT, compression="zstd")
                writers[pre].write_table(pa.Table.from_pylist(rows_, OUT)); n_rows += len(rows_)
        del meta_tbl; pa.default_memory_pool().release_unused()
        for pre in todo:
            outp = outs[pre]
            if pre in writers:
                writers[pre].close(); os.replace(str(outp) + ".tmp", outp)
                _part_stats(c, outp, W / "final" / f"stats-{pre}.json")
                if not os.environ.get("CANON_TEST"):
                    from huggingface_hub import HfApi
                    HfApi().batch_bucket_files(BUCKET, add=[(str(outp), f"processed/canonical/skills/skills-{pre}.parquet")])
                    outp.unlink()
            (W / "final" / f"part-{pre}.done").write_text("ok\n")
            print(f"  final partition {pre}: done ({n_rows:,} rows this run)", flush=True)
    tally = {}
    for f in sorted((W / "final").glob("stats-*.json")):
        for k, v in json.loads(f.read_text()).items():
            if isinstance(v, dict):
                d = tally.setdefault(k, {})
                for k2, v2 in v.items(): d[k2] = d.get(k2, 0) + v2
            else: tally[k] = tally.get(k, 0) + v
    (W / "canonical_summary.json").write_text(json.dumps(tally, indent=1))
    print("final done", json.dumps(tally), flush=True)


def upload():
    """Partitions were shipped by final(); ship the summary and the canonical metadata table."""
    from huggingface_hub import HfApi
    HfApi().batch_bucket_files(BUCKET, add=[(str(W / "canonical_summary.json"), "stats/canonical_summary.json"),
                                            *[(str(p), f"processed/canonical/meta/{p.name}") for p in sorted((W / "canon").glob("canonical_meta-*.parquet"))]])
    print("upload done", flush=True)


if __name__ == "__main__":
    stage = sys.argv[1] if len(sys.argv) > 1 else "all"
    for s in (["scan", "select", "final", "upload"] if stage == "all" else [stage]):
        if (W / f"{s}.ok").exists(): print(f"{s}: already done", flush=True); continue
        globals()[s](); (W / f"{s}.ok").write_text("ok\n")