Text rule trends

"""Shared pieces: reading file histories, file kinds, periods, centroids."""
import base64
import gzip
import glob
import json
import os
import re
from datetime import date, timedelta


def history_files(root):
    """SAMPLE/vers/*.tsv.gz and *.tsv, sorted by key."""
    files = glob.glob(os.path.join(root, "vers", "*.tsv.gz")) + glob.glob(os.path.join(root, "vers", "*.tsv"))
    return sorted(files, key=lambda p: os.path.basename(p))


def key_of(path):
    b = os.path.basename(path)
    return b[: -len(".tsv.gz")] if b.endswith(".tsv.gz") else b[: -len(".tsv")]


def open_text(path):
    return gzip.open(path, "rt") if path.endswith(".gz") else open(path, encoding="utf-8")


def load_versions(path):
    """[(author_date, claude_coauthor, body)], oldest first. An empty body is a deletion."""
    rows = []
    with open_text(path) as f:
        for line in f:
            parts = line.rstrip("\n").split("\t")
            if len(parts) < 3:
                continue
            body = base64.b64decode(parts[3]).decode("utf-8", "replace") if len(parts) > 3 and parts[3] else ""
            rows.append((parts[1], int(parts[2]), body))
    rows.sort(key=lambda r: r[0])  # oldest first, by author date string
    return rows


def load_manifest(root):
    """{key: path in repo} from SAMPLE/manifest.tsv, or {} when there is none."""
    p = os.path.join(root, "manifest.tsv")
    out = {}
    if os.path.exists(p):
        for line in open(p, encoding="utf-8"):
            f = line.rstrip("\n").split("\t")
            if len(f) >= 3:
                out[f[0]] = f[2]
    return out


class Kinds:
    """A file's kind. By default its file name (CLAUDE.md, README.md, package.json).

    kind_map is {regex: label}, tried in order against the file's path in the repo
    (or, with no manifest, the part of the key after "__"); the first match wins.
    """

    def __init__(self, root, kind_map=None):
        self.manifest = load_manifest(root)
        self.rules = [(re.compile(k), v) for k, v in (kind_map or {}).items()]

    def path(self, key):
        return self.manifest.get(key) or key.split("__", 1)[-1]

    def __call__(self, key):
        p = self.path(key)
        for rx, label in self.rules:
            if rx.search(p):
                return label
        return p.rsplit("/", 1)[-1]


def load_kind_map(s):
    if not s:
        return None
    if isinstance(s, dict):
        return s
    return json.load(open(s)) if os.path.exists(s) else json.loads(s)


# ---- periods: "2026Q1" (quarter), "2026-03" (month), "2026" (year)

def period_of(d, unit):
    if unit == "quarter":
        return f"{d[:4]}Q{(int(d[5:7]) - 1) // 3 + 1}"
    if unit == "month":
        return d[:7]
    if unit == "year":
        return d[:4]
    raise SystemExit(f"unknown period {unit!r}: use quarter, month or year")


def period_start(p):
    if "Q" in p:
        return date(int(p[:4]), 3 * int(p[5]) - 2, 1)
    if len(p) == 7:
        return date(int(p[:4]), int(p[5:7]), 1)
    return date(int(p[:4]), 1, 1)


def period_next(p):
    """The first day after the period."""
    s = period_start(p)
    if "Q" in p:
        return date(s.year + 1, 1, 1) if s.month == 10 else date(s.year, s.month + 3, 1)
    if len(p) == 7:
        return date(s.year + 1, 1, 1) if s.month == 12 else date(s.year, s.month + 1, 1)
    return date(s.year + 1, 1, 1)


def period_range(p0, p1, unit):
    out, p = [], p0
    while p <= p1:
        out.append(p)
        p = period_of(period_next(p).isoformat(), unit)
    return out


def period_end(p):
    return period_next(p) - timedelta(days=1)


# ---- centroids: JSON (text) or .npy

def load_centroids(path):
    import numpy as np
    if path.endswith(".npy"):
        return np.load(path).astype(np.float32)
    d = json.load(open(path))
    C = d["centroids"] if isinstance(d, dict) else d
    return np.array(C, dtype=np.float32)


def centroids_id(C):
    """A short fingerprint of a set of centroids (sha256 of the float32 values), so a labels file
    and a themes file can be checked against each other."""
    import hashlib
    import numpy as np
    return "sha256:" + hashlib.sha256(np.ascontiguousarray(np.asarray(C, dtype=np.float32)).tobytes()).hexdigest()[:16]


def save_centroids(path, C, **meta):
    """Each value is written as the shortest decimal that reads back as the same float32."""
    import numpy as np
    C = np.asarray(C, dtype=np.float32)
    rows = ",\n  ".join("[" + ",".join(np.format_float_positional(x, unique=True, trim="-") for x in row) + "]" for row in C)
    head = json.dumps({**meta, "id": centroids_id(C), "k": int(C.shape[0]), "dim": int(C.shape[1])})[1:-1]
    with open(path, "w") as f:
        f.write("{" + head + ',\n "centroids": [\n  ' + rows + "\n ]\n}\n")