SoMaCoSF Corpus 2026-04-28 — 394 rows: 205 gists + 189 repos, Ghost Catalog fields, ES.exe-ready CSV

build_corpus.py

#!/usr/bin/env python3
"""
SoMaCo Corpus Builder — build_corpus.py
Slurps all SoMaCoSF gists + repos via gh CLI into Ghost Catalog–formatted CSV.

Ghost Catalog fields (SOM-DOC convention):
  file_id, type, entity_id, description, category, visibility,
  language, created_at, updated_at, url, files_or_stars,
  is_fork, stack, status, tags

Output: D:/somacosf/index/somacosf_corpus.csv
"""

import subprocess
import json
import csv
import re
import sys
from datetime import datetime, timezone
from pathlib import Path

OUT_PATH = Path(r"D:\somacosf\index\somacosf_corpus.csv")
NOW = datetime.now(timezone.utc)

# ── Category inference ─────────────────────────────────────────────────────────

CATEGORY_RULES = [
    (r"platform|mission.control|dashboard|pulse|says.network|hub",  "APP"),
    (r"uuid|gyst|protocol|spoctalk|locus|ghost.catalog",             "LIB"),
    (r"api|endpoint|ingest|harvester|signal",                        "API"),
    (r"doc|spec|prd|architecture|guide|readme|brief",                "DOC"),
    (r"config|\.env|settings|scaffold",                              "CFG"),
    (r"script|tool|cli|ps1|bat|automation",                         "SCR"),
    (r"component|card|panel|widget",                                 "CMP"),
    (r"page|route|/pulse|/says",                                     "PAG"),
]

STACK_RULES = [
    (r"next\.?js|nextjs|react|tailwind|vercel",  "next.js"),
    (r"electron",                                  "electron"),
    (r"python|py\b|flask|fastapi|uv\b",           "python"),
    (r"typescript|ts\b",                           "typescript"),
    (r"javascript|js\b",                           "javascript"),
    (r"rust\b",                                    "rust"),
    (r"go\b|golang",                               "go"),
    (r"powershell|ps1",                            "powershell"),
    (r"solidity|evm|base\b|usdc",                 "web3"),
    (r"sqlite|postgres|db\b",                      "db"),
]

TAG_KEYWORDS = [
    "uuid", "gyst", "harvester", "polymarket", "kalshi", "spoctalk",
    "prediction", "signal", "locus", "says", "popsoc", "vertex",
    "mcp", "agent", "cursor", "claude", "llm", "ollama",
    "three.js", "webgl", "webgpu", "manim", "svelte", "electron",
    "oligarch", "intelligence", "election", "dnc",
]

def infer_category(text: str) -> str:
    t = text.lower()
    for pattern, cat in CATEGORY_RULES:
        if re.search(pattern, t):
            return cat
    return "DOC"

def infer_stack(text: str, lang: str) -> str:
    combined = (text + " " + (lang or "")).lower()
    stacks = []
    for pattern, stack in STACK_RULES:
        if re.search(pattern, combined):
            stacks.append(stack)
    return ",".join(stacks[:3]) if stacks else (lang or "").lower() or "—"

def infer_tags(text: str) -> str:
    t = text.lower()
    return ",".join(k for k in TAG_KEYWORDS if k in t)[:120]

def infer_status(updated_at: str) -> str:
    try:
        dt = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
        age_days = (NOW - dt).days
        if age_days < 30:  return "active"
        if age_days < 90:  return "warm"
        return "stale"
    except Exception:
        return "unknown"

# ── Gist parsing (from gh gist list plain-text) ────────────────────────────────

def fetch_gists() -> list[dict]:
    """Run gh gist list and parse tab-separated output."""
    result = subprocess.run(
        ["gh", "gist", "list", "--limit", "300"],
        capture_output=True, text=True, encoding="utf-8", errors="replace"
    )
    rows = []
    for i, line in enumerate(result.stdout.strip().splitlines()):
        parts = line.split("\t")
        if len(parts) < 5:
            continue
        gist_id    = parts[0].strip()
        desc       = parts[1].strip()
        files_str  = parts[2].strip()   # "3 files" or "1 file"
        visibility = parts[3].strip()   # public / secret
        created_at = parts[4].strip()   # ISO 8601

        file_count = int(re.search(r"\d+", files_str).group()) if re.search(r"\d+", files_str) else 1
        seq = str(i + 1).zfill(4)

        rows.append({
            "file_id":        f"SOM-GST-{seq}",
            "type":           "GIST",
            "entity_id":      gist_id,
            "description":    desc[:200],
            "category":       infer_category(desc),
            "visibility":     visibility,
            "language":       "—",
            "created_at":     created_at,
            "updated_at":     created_at,
            "url":            f"https://gist.github.com/SoMaCoSF/{gist_id}",
            "files_or_stars": str(file_count),
            "is_fork":        "false",
            "stack":          infer_stack(desc, ""),
            "status":         infer_status(created_at),
            "tags":           infer_tags(desc),
        })
    return rows

# ── Repo parsing (from gh repo list JSON) ─────────────────────────────────────

def fetch_repos() -> list[dict]:
    result = subprocess.run(
        ["gh", "repo", "list", "SoMaCoSF", "--limit", "300",
         "--json", "name,description,url,createdAt,updatedAt,primaryLanguage,isPrivate,isFork"],
        capture_output=True, text=True, encoding="utf-8", errors="replace"
    )
    try:
        data = json.loads(result.stdout)
    except json.JSONDecodeError:
        print(f"[WARN] Could not parse repos JSON: {result.stdout[:200]}", file=sys.stderr)
        return []

    rows = []
    for i, repo in enumerate(data):
        name       = repo.get("name", "")
        desc       = repo.get("description") or ""
        url        = repo.get("url", "")
        created_at = repo.get("createdAt", "")
        updated_at = repo.get("updatedAt", "")
        lang_obj   = repo.get("primaryLanguage") or {}
        lang       = lang_obj.get("name", "") if lang_obj else ""
        is_private = repo.get("isPrivate", False)
        is_fork    = repo.get("isFork", False)
        seq        = str(i + 1).zfill(4)

        combined = name + " " + desc
        visibility = "private" if is_private else "public"

        rows.append({
            "file_id":        f"SOM-REP-{seq}",
            "type":           "REPO",
            "entity_id":      name,
            "description":    desc[:200],
            "category":       infer_category(combined),
            "visibility":     visibility,
            "language":       lang,
            "created_at":     created_at,
            "updated_at":     updated_at,
            "url":            url,
            "files_or_stars": "—",
            "is_fork":        str(is_fork).lower(),
            "stack":          infer_stack(combined, lang),
            "status":         infer_status(updated_at),
            "tags":           infer_tags(combined),
        })
    return rows

# ── Main ───────────────────────────────────────────────────────────────────────

FIELDS = [
    "file_id", "type", "entity_id", "description", "category",
    "visibility", "language", "created_at", "updated_at", "url",
    "files_or_stars", "is_fork", "stack", "status", "tags",
]

def main():
    print("Fetching gists...")
    gists = fetch_gists()
    print(f"  {len(gists)} gists")

    print("Fetching repos...")
    repos = fetch_repos()
    print(f"  {len(repos)} repos")

    all_rows = gists + repos

    # Sort: active first, then by updated_at desc
    status_order = {"active": 0, "warm": 1, "stale": 2, "unknown": 3}
    all_rows.sort(key=lambda r: (
        status_order.get(r["status"], 9),
        r["updated_at"]
    ), reverse=False)
    # secondary: updated_at desc within same status bucket
    all_rows.sort(key=lambda r: r["updated_at"], reverse=True)
    all_rows.sort(key=lambda r: status_order.get(r["status"], 9))

    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    with open(OUT_PATH, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(all_rows)

    print(f"\nCorpus written: {OUT_PATH}")
    print(f"Total rows: {len(all_rows)}  ({len(gists)} gists + {len(repos)} repos)")

    # Summary breakdown
    cats  = {}
    stks  = {}
    stats = {}
    for r in all_rows:
        cats[r["category"]] = cats.get(r["category"], 0) + 1
        stats[r["status"]]  = stats.get(r["status"], 0) + 1
        for s in r["stack"].split(","):
            s = s.strip()
            if s and s != "—":
                stks[s] = stks.get(s, 0) + 1

    print("\nBy category:")
    for k, v in sorted(cats.items(), key=lambda x: -x[1]):
        print(f"  {k:6}  {v}")
    print("\nBy status:")
    for k, v in sorted(stats.items(), key=lambda x: -x[1]):
        print(f"  {k:8}  {v}")
    print("\nTop stacks:")
    for k, v in sorted(stks.items(), key=lambda x: -x[1])[:10]:
        print(f"  {k:20}  {v}")

if __name__ == "__main__":
    main()

somacosf_corpus.csv