Text rule trends

#!/usr/bin/env python3
"""Extract line-level items (by default "rules") from every version of every file, and follow
each item's life: when it appeared, and when it was deleted.

Reads SAMPLE/vers/*.tsv.gz or *.tsv as written by github-file-history (one row per version:
sha, author_date, claude_coauthor 0/1, base64 body; an empty body is a deletion).

Usage:
  python3 extract.py SAMPLE_DIR [--extractor rules|lines|regex|plugin] [--regex RE]
                     [--plugin FILE.py:FUNC] [--min-words N] [--max-words N] [--match 0.6]
                     [--order datetime|string] [--out FILE]

Extractors:
  rules   (default) a list item, or a line starting with a capital letter, whose text (after an
          optional bold label) starts with an imperative or normative word (always, never, use,
          run, avoid, must, should, ...). Fenced code, headings and table rows are skipped.
          Default words: 4 to 45.
  lines   every non-blank line outside fenced code, list markers removed. Default words: 1 to 45.
  regex   lines (outside fenced code) that match --regex. The item is group 1 if the pattern has
          a group, else the whole match. Example, Markdown headings: --regex '^#{1,3}\\s+(.+)'
  plugin  your own function: --plugin myextract.py:items. It gets the file body (str) and returns
          a list of item strings. Word limits and the identity rules below still apply.

Words are counted after normalising: lower case, code spans -> "code", links -> their text,
digits and markdown removed. An item keeps its identity across a rewording when most of its
words survive: token Jaccard >= --match (default 0.6) against an unmatched item of the
previous version.

Versions are replayed oldest first. --order datetime (default) sorts by the parsed author date,
so mixed UTC offsets order correctly. --order string sorts by the date text, as some older
runs did.

Output (default SAMPLE_DIR/rules.json):
  nfiles    files with at least one version
  nhistory  history files found (a file whose history came back empty counts here, not in nfiles)
  life      one record per item: {kind, born, died, text, norm, repo}
            kind  the file's kind (its file name unless --kind-map says otherwise)
            born  author date of the version where it first appeared
            died  author date of the version where it disappeared, or null
            text  the line as written; norm its normalised form
            repo  the history file's name (<key>.tsv.gz or <key>.tsv), i.e. one file in one
                  repo; the key is <owner>_<repo>__<path>. Kept as "repo" for compatibility
  files     {key: {kind, first, deleted}}: first version date, and the date of the deletion if
            the last version deleted the file
  co_month  {"YYYY-MM": [versions with a Claude co-author trailer, versions]}
  settings  the extractor settings used
"""
import argparse
import collections
import importlib.util
import json
import os
import re
import sys
from datetime import datetime

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import Kinds, history_files, key_of, load_kind_map, load_versions  # noqa: E402

WORD = re.compile(r"[a-z][a-z0-9_+#.-]*")
IMPERATIVE = re.compile(r"^(always|never|do|don'?t|does not|avoid|prefer|use|run|keep|make sure|ensure|follow|add|write|read|check|test|update|create|remove|delete|put|place|import|call|ask|verify|commit|document|must|should|only|before|after|when|if|no |not |all |every|treat|leave|stop|start|wait|search|look|consider|default|instead|don't)\b")


def norm(line):
    t = re.sub(r"`[^`]*`", " code ", line.lower())
    t = re.sub(r"\[([^\]]*)\]\([^)]*\)", r"\1", t)
    t = re.sub(r"[*_>#]", " ", t)
    t = re.sub(r"\d+", " ", t)
    return " ".join(WORD.findall(t))


def rule_lines(body):
    """The default heuristic: imperative list items and capitalised lines."""
    out, fence = [], False
    for raw in body.splitlines():
        s = raw.strip()
        if s.startswith("```"):
            fence = not fence
            continue
        if fence or not s or s.startswith("#") or s.startswith("|"):
            continue
        m = re.match(r"^([-*+]|\d+[.)])\s+(.*)", s)
        text = m.group(2) if m else s
        if not m and not text[:1].isupper():
            continue  # a wrapped continuation line, not a rule
        head = re.sub(r"^\*\*[^*]{1,60}\*\*\s*[:.-]?\s*", "", text)  # drop a bold label
        if IMPERATIVE.match(head.lower()):
            out.append(text)
    return out


def plain_lines(body, rx=None):
    out, fence = [], False
    for raw in body.splitlines():
        s = raw.strip()
        if s.startswith("```"):
            fence = not fence
            continue
        if fence or not s:
            continue
        if rx is None:
            m = re.match(r"^([-*+]|\d+[.)])\s+(.*)", s)
            out.append(m.group(2) if m else s)
        else:
            m = rx.search(s)
            if m:
                out.append(m.group(1) if m.groups() else m.group(0))
    return out


def load_plugin(spec):
    path, _, func = spec.rpartition(":")
    if not path:
        raise SystemExit("--plugin wants FILE.py:FUNCTION")
    mod_spec = importlib.util.spec_from_file_location("extract_plugin", path)
    mod = importlib.util.module_from_spec(mod_spec)
    mod_spec.loader.exec_module(mod)
    return getattr(mod, func)


def make_items(a):
    if a.extractor == "rules":
        get = rule_lines
    elif a.extractor == "lines":
        get = plain_lines
    elif a.extractor == "regex":
        if not a.regex:
            raise SystemExit("--extractor regex needs --regex")
        rx = re.compile(a.regex)
        get = lambda body: plain_lines(body, rx)
    else:
        if not a.plugin:
            raise SystemExit("--extractor plugin needs --plugin FILE.py:FUNC")
        get = load_plugin(a.plugin)
    lo = a.min_words if a.min_words is not None else (4 if a.extractor == "rules" else 1)
    hi = a.max_words

    def items(body):
        seen, uniq = set(), []
        for text in get(body):
            n = norm(text)
            if lo <= len(n.split()) <= hi and n not in seen:
                seen.add(n)
                uniq.append((n, text.strip()))
        return uniq
    return items, lo


def jacc(a, b):
    A, B = set(a.split()), set(b.split())
    return len(A & B) / max(1, len(A | B))


def parse_dt(s):
    try:
        return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return float("inf")


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("sample")
    ap.add_argument("--extractor", default="rules", choices=["rules", "lines", "regex", "plugin"])
    ap.add_argument("--regex")
    ap.add_argument("--plugin")
    ap.add_argument("--min-words", type=int)
    ap.add_argument("--max-words", type=int, default=45)
    ap.add_argument("--match", type=float, default=0.6, help="Jaccard needed to keep identity through a rewording")
    ap.add_argument("--order", default="datetime", choices=["datetime", "string"])
    ap.add_argument("--kind-map", help='JSON file or string {"regex": "label"} matched against the path')
    ap.add_argument("--out", help="default SAMPLE_DIR/rules.json")
    a = ap.parse_args()

    items_of, lo = make_items(a)
    kinds = Kinds(a.sample, load_kind_map(a.kind_map))
    files = history_files(a.sample)
    if not files:
        sys.exit(f"no {a.sample}/vers/*.tsv.gz or *.tsv")
    life, nfiles, info = [], 0, {}
    co_month = collections.defaultdict(lambda: [0, 0])
    for path in files:
        key = key_of(path)
        kind = kinds(key)
        vers = load_versions(path)
        if a.order == "datetime":
            vers.sort(key=lambda r: parse_dt(r[0]))
        if not vers:
            continue
        nfiles += 1
        info[key] = {"kind": kind, "first": vers[0][0], "deleted": vers[-1][0] if not vers[-1][2] else None}
        alive = {}  # norm -> record
        for d, co, body in vers:
            co_month[d[:7]][0] += co
            co_month[d[:7]][1] += 1
            matched, nxt = set(), {}
            for n, text in items_of(body):
                rec = alive.get(n)
                if rec is None:
                    best = max(((jacc(n, k), k) for k in alive if k not in matched), default=(0, None))
                    if best[0] >= a.match:
                        rec = alive[best[1]]
                        matched.add(best[1])
                if rec is None:
                    rec = {"kind": kind, "born": d, "died": None, "text": text, "norm": n, "repo": os.path.basename(path)}
                    life.append(rec)
                else:
                    matched.add(rec["norm"])
                rec["norm"] = n
                nxt[n] = rec
            kept = {id(r) for r in nxt.values()}
            for rec in alive.values():
                if id(rec) not in kept:
                    rec["died"] = d
            alive = nxt
    settings = {"extractor": a.extractor, "regex": a.regex, "plugin": a.plugin, "min_words": lo,
                "max_words": a.max_words, "match": a.match, "order": a.order}
    out = a.out or os.path.join(a.sample, "rules.json")
    with open(out, "w") as f:
        json.dump({"nfiles": nfiles, "nhistory": len(files), "life": life, "files": info, "co_month": co_month, "settings": settings}, f)
    with_items = len({r["repo"] for r in life})
    gone = sum(r["died"] is not None for r in life)
    print(f"files {nfiles}  with items {with_items}  items {len(life)}  later deleted {gone / max(1, len(life)):.0%}  -> {out}")


if __name__ == "__main__":
    main()