Automated Changeset Generator & History Auditor

#!/usr/bin/env python3
import argparse
import glob
import json
import os
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path

def sh(cmd, *, text=True):
    return subprocess.check_output(cmd, text=text)

def git(args):
    return sh(['git', *args])

LOCKFILE_NAMES = {
    'pnpm-lock.yaml',
    'package-lock.json',
    'yarn.lock',
    'npm-shrinkwrap.json',
}


def is_lockfile(path: str) -> bool:
    base = os.path.basename(path)
    return base in LOCKFILE_NAMES


def is_ignored_file(path: str) -> bool:
    return path == '.gitignore' or is_lockfile(path)


def slugify_package_name(name: str) -> str:
    # '@bits/prettier-config-svelte' -> 'bits-prettier-config-svelte'
    # 'content-sync' -> 'content-sync'
    s = name.strip()
    if s.startswith('@'):
        s = s[1:]
    s = s.replace('/', '-')
    s = re.sub(r'[^a-zA-Z0-9._-]+', '-', s)
    s = re.sub(r'-{2,}', '-', s).strip('-')
    return s.lower()


def parse_changed_packages_list(path: Path):
    # Lines: '- @bits/ui (packages/ui)'
    pkgs = []
    pat = re.compile(r'^-\s+([^ ]+)\s+\(([^)]+)\)\s*$')
    for line in path.read_text(encoding='utf-8').splitlines():
        line = line.strip()
        if not line:
            continue
        m = pat.match(line)
        if not m:
            raise SystemExit(f'changed-packages line did not match expected format: {line!r}')
        pkgs.append((m.group(1), m.group(2)))
    return pkgs


def read_pkg_name(scope_path: str) -> str:
    pj = Path(scope_path) / 'package.json'
    if not pj.exists():
        raise FileNotFoundError(f'missing package.json at {pj}')
    data = json.loads(pj.read_text(encoding='utf-8'))
    name = data.get('name')
    if not name:
        raise ValueError(f'package.json has no name: {pj}')
    return name


def read_pkg_version(scope_path: str) -> str:
    pj = Path(scope_path) / 'package.json'
    data = json.loads(pj.read_text(encoding='utf-8'))
    v = data.get('version')
    if not v:
        raise ValueError(f'package.json has no version: {pj}')
    return v


def bump_rank(b: str) -> int:
    return {'patch': 0, 'minor': 1, 'major': 2}[b]


def bump_from_commit_message(subject: str, body: str) -> str:
    full = subject + "\n" + body
    if 'BREAKING CHANGE' in full:
        return 'major'
    if re.search(r'\b(redesign|foundation)\b', full, flags=re.I):
        return 'major'
    if subject.startswith('feat:') or subject.startswith('feat('):
        return 'minor'
    return 'patch'


def parse_subject_prefix(subject_line: str):
    # Handle autosquash prefixes
    raw = subject_line.strip()

    fixup = False
    if raw.startswith('fixup! '):
        fixup = True
        raw = raw[len('fixup! '):].lstrip()

    # Conventional prefix: type(scope): or type:
    m = re.match(r'^([a-zA-Z]+)(\([^)]*\))?:\s*(.*)$', raw)
    if not m:
        return None

    typ = m.group(1)
    scope = m.group(2) or ''
    rest = m.group(3) or ''

    # Normalize types
    if fixup:
        typ = 'fix'
    elif typ == 'style':
        typ = 'chore'

    prefix = f"{typ}{scope}:"
    return prefix, rest


def strip_leading_prefix(text: str) -> str:
    # Remove any leading 'type(scope):' or 'type:' from subject text when using subject as raw bullet text.
    return re.sub(r'^[a-zA-Z]+(\([^)]*\))?:\s*', '', text.strip())


def commit_paths(commit: str):
    out = git(['show', '--name-only', '--format=', commit])
    paths = [p.strip() for p in out.splitlines() if p.strip()]
    return paths


def commit_message(commit: str):
    msg = git(['show', '-s', '--format=%B', commit])
    lines = msg.splitlines()
    subject = ''
    for ln in lines:
        if ln.strip():
            subject = ln.rstrip('\n')
            break
    body = msg[len(subject):].lstrip('\n') if subject else msg
    return subject, body


def scopes_by_path(paths):
    scopes = set()
    for p in paths:
        m = re.match(r'^(packages|sites|tools|apps)/[^/]+', p)
        if m:
            scopes.add(m.group(0))
    return sorted(scopes)


def ensure_no_existing_changesets(ts: int):
    existing = glob.glob(f'.changeset/*-{ts}.md')
    if existing:
        raise SystemExit(f'Refusing to overwrite existing .changeset files for ts={ts}: {existing[:3]}...')


def write_changeset_file(path: Path, package_name: str, bump: str, bullets):
    fm = f"---\n\"{package_name}\": {bump}\n---\n\n"
    content = fm + "\n".join(bullets) + "\n"
    path.write_text(content, encoding='utf-8')


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--batch', required=True)
    ap.add_argument('--timestamp', required=True, type=int)
    ap.add_argument('--changed-packages', required=True)
    ap.add_argument('--out-dir', required=True)
    ap.add_argument('--baseline', required=True)
    ap.add_argument('--write-to-changeset', action='store_true')
    args = ap.parse_args()

    batch_path = Path(args.batch)
    ts = args.timestamp
    out_dir = Path(args.out_dir)
    changed_list = Path(args.changed_packages)

    commits = [l.strip() for l in batch_path.read_text(encoding='utf-8').splitlines() if l.strip()]
    if not commits:
        raise SystemExit('empty batch list')

    changed_pkgs = parse_changed_packages_list(changed_list)
    scope_to_pkgname = {scope: read_pkg_name(scope) for _, scope in changed_pkgs}

    # Accumulators
    pkg_bullets = {}  # pkgName -> [bullet, ...]
    pkg_bump = {}     # pkgName -> bump
    commit_index = []

    for h in commits:
        short = git(['rev-parse', '--short', h]).strip()
        subject, body = commit_message(h)
        prefix_parsed = parse_subject_prefix(subject)
        if not prefix_parsed:
            raise SystemExit(f'commit {short} subject does not match conventional prefix: {subject!r}')
        cc_prefix, subject_rest = prefix_parsed

        paths = commit_paths(h)
        # Phase 1 "raw capture" is completeness-first:
        # - Do NOT skip commits based on heuristics (formatting-only, lockfile-only, etc.)
        # - We still ignore lockfile/.gitignore paths when mapping to package scopes so they land in REPO_ONLY.
        paths_non_ignored = [p for p in paths if not is_ignored_file(p)]

        scopes = scopes_by_path(paths_non_ignored)
        if not scopes:
            scopes = ['REPO_ONLY']

        body_bullets_raw = [ln.strip() for ln in body.splitlines() if ln.startswith('- ')]
        bullet_texts = []
        if body_bullets_raw:
            for ln in body_bullets_raw:
                bullet_texts.append(ln[2:].rstrip())
        else:
            bullet_texts.append(strip_leading_prefix(subject))

        commit_index.append({
            'hash': h,
            'short': short,
            'status': 'included',
            'scopes': scopes,
            'prefix': cc_prefix,
            'bullets_in_body': len(body_bullets_raw),
            'bullets_generated': len(bullet_texts),
        })

        for scope in scopes:
            if scope == 'REPO_ONLY':
                pkg_name = '@bits/repo'
                scope_path = None
            else:
                pkg_name = scope_to_pkgname.get(scope)
                if not pkg_name:
                    raise SystemExit(f'could not map scope {scope!r} to package name (missing package.json?)')
                scope_path = scope

            v = '0.0.0'
            if scope_path is not None:
                v = read_pkg_version(scope_path)
            bump = 'major' if re.match(r'^0\.0\.\d+$', v) else bump_from_commit_message(subject, body)
            prev = pkg_bump.get(pkg_name, 'patch')
            pkg_bump[pkg_name] = bump if bump_rank(bump) > bump_rank(prev) else prev

            out = pkg_bullets.setdefault(pkg_name, [])
            for bt in bullet_texts:
                out.append(f"- [{short}] {cc_prefix} {bt}")

    out_dir.mkdir(parents=True, exist_ok=False)

    written = []
    for pkg_name, bullets in sorted(pkg_bullets.items(), key=lambda kv: kv[0]):
        bump = pkg_bump.get(pkg_name, 'patch')
        slug = 'repo' if pkg_name == '@bits/repo' else slugify_package_name(pkg_name)
        fname = f"{slug}-{bump}-{ts}.md"
        p_out = out_dir / fname
        write_changeset_file(p_out, pkg_name, bump, bullets)
        written.append(str(p_out))

        if args.write_to_changeset:
            ensure_no_existing_changesets(ts)
            p_repo = Path('.changeset') / fname
            write_changeset_file(p_repo, pkg_name, bump, bullets)

    (out_dir / 'commit-index.json').write_text(json.dumps(commit_index, indent=2), encoding='utf-8')

    print(json.dumps({
        'batch': str(batch_path),
        'timestamp': ts,
        'commits_total': len(commits),
        'commits_included': sum(1 for c in commit_index if c.get('status') == 'included'),
        'packages_created': sorted(pkg_bullets.keys()),
        'changeset_files_count': len(written),
    }, indent=2))


if __name__ == '__main__':
    main()