code/dedup_count.py

"""Exact distinct-skill counts across every processed source (reads only the needed columns from the bucket).
  - content level: distinct SKILL.md git blob SHAs across GitSkills + all GitHub crawl runs (same hash space)
  - routing level: distinct (lower(name), lower(description)) pairs across all sources incl. ClawHub
Writes logs/dedup_count.json.
"""
import json, sys, time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "code"))
from bucket_query import urls, connect

t0 = time.time()
c = connect("600MB")
c.execute("SET threads=4; SET preserve_insertion_order=false")
c.execute(f"SET temp_directory='{ROOT}/work/duck_tmp'")
runs = ["pass1_sourcegraph", "pass2_new", "backfill_gitskills"]
c.execute(f"""create table gh as
    select 'gitskills' src, file_sha sha, lower(trim(name)) n, lower(trim(description)) d
      from read_parquet({urls('processed/gitskills/skills/')}) where is_skill_md
    """ + "".join(f"""
    union all select '{r}', file_sha, lower(trim(name)), lower(trim(description))
      from read_parquet({urls(f'processed/github_delta/{r}/skills/')})""" for r in runs))
print("loaded github rows", c.execute("select count(*) from gh").fetchone()[0], f"{time.time()-t0:.0f}s", flush=True)
c.execute(f"""create table ch as
    select 'clawhub' src, lower(trim(name)) n, lower(trim(description)) d from read_parquet({urls('processed/clawhub/skills.parquet', '')})
    union all select 'clawhub_api', lower(trim(name)), lower(trim(description)) from read_parquet({urls('processed/clawhub/api_delta_skills.parquet', '')})""")

R = {}
R["rows_by_source"] = dict(c.execute("select src, count(*) from gh group by 1").fetchall() + c.execute("select src, count(*) from ch group by 1").fetchall())
R["distinct_skill_md_by_source"] = dict(c.execute("select src, count(distinct sha) from gh group by 1").fetchall())
R["distinct_skill_md_all_github"] = c.execute("select count(distinct sha) from gh").fetchone()[0]
R["new_vs_gitskills_by_run"] = dict(c.execute("""select src, count(distinct sha) from gh
    where src <> 'gitskills' and sha not in (select sha from gh where src = 'gitskills') group by 1""").fetchall())
R["distinct_new_vs_gitskills_all_runs"] = c.execute("""select count(distinct sha) from gh
    where src <> 'gitskills' and sha not in (select sha from gh where src = 'gitskills')""").fetchone()[0]
c.execute("create table nd as select n, d from gh where n is not null and d is not null union all select n, d from ch where n is not null and d is not null")
R["distinct_name_desc_all_sources"] = c.execute("select count(*) from (select distinct n, d from nd)").fetchone()[0]
R["distinct_name_desc_github"] = c.execute("select count(*) from (select distinct n, d from gh where n is not null and d is not null)").fetchone()[0]
R["clawhub_name_desc_not_on_github"] = c.execute("""select count(*) from (select distinct n, d from ch where n is not null and d is not null
    except select distinct n, d from gh where n is not null and d is not null)""").fetchone()[0]
R["seconds"] = round(time.time() - t0)
(ROOT / "logs/dedup_count.json").write_text(json.dumps(R, indent=1))
print(json.dumps(R, indent=1), flush=True)