scripts/build_analysis.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import math
import os
import subprocess
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any


ROOT = Path("/home/smlflg/Projekte")
OUT = Path(__file__).resolve().parents[1] / "data" / "analysis.json"
WINDOW_END = datetime.fromisoformat("2026-03-20T23:59:59+01:00")
WINDOW_START = datetime.fromisoformat("2026-02-19T00:00:00+01:00")
SESSION_GAP_MINUTES = 120
IGNORED_PROJECTS = {
    "Sidecar",
    "Sidecar-V5",
    "Sidecar-V6",
    "Sidecar-V7-Lite",
    "Sidecar-V7-V2",
    "Sidecar-V8-Plan",
}
TOP_RECOMMENDATION_ORDER = [
    "Sidecar-ng",
    "MyAIGame",
    "Sidecar-V7",
    "ClaudeCodePanel",
    "Sidecar-Evolution",
    "AgentArena",
]
RANKING_SCORES = {
    "Sidecar-ng": {
        "leverage": 5.0,
        "reusability": 4.6,
        "strategic_value": 5.0,
        "product_readiness": 4.5,
        "synergy": 5.0,
        "workflow_fit": 5.0,
    },
    "MyAIGame": {
        "leverage": 4.4,
        "reusability": 5.0,
        "strategic_value": 4.4,
        "product_readiness": 3.4,
        "synergy": 4.8,
        "workflow_fit": 4.9,
    },
    "Sidecar-V7": {
        "leverage": 4.3,
        "reusability": 4.3,
        "strategic_value": 4.4,
        "product_readiness": 4.5,
        "synergy": 4.3,
        "workflow_fit": 4.4,
    },
    "ClaudeCodePanel": {
        "leverage": 4.0,
        "reusability": 3.4,
        "strategic_value": 4.2,
        "product_readiness": 4.8,
        "synergy": 4.1,
        "workflow_fit": 4.8,
    },
    "Sidecar-Evolution": {
        "leverage": 3.5,
        "reusability": 4.4,
        "strategic_value": 4.8,
        "product_readiness": 2.3,
        "synergy": 4.6,
        "workflow_fit": 3.8,
    },
    "AgentArena": {
        "leverage": 3.9,
        "reusability": 4.1,
        "strategic_value": 4.3,
        "product_readiness": 3.8,
        "synergy": 4.0,
        "workflow_fit": 3.7,
    },
}
PROJECT_NOTES = {
    "Sidecar-ng": "Stateless Hook statt Daemon, direkte Claude-Code-Integration, dokumentierte Evals und klares Accuracy-Signal.",
    "MyAIGame": "Breites Workflow-Toolkit mit Skills, Hooks, Session-Browser und MCP-Setups als Multiplikator fuer andere Projekte.",
    "Sidecar-V7": "Reifste volle Sidecar-Ausbaustufe mit hoher Produktnaehe, aber auch sichtbarer Komplexitaetslast.",
    "ClaudeCodePanel": "Starke Observability-Schicht fuer Sessions, Kosten, Swarms und Events mit direktem Nutzwert im Alltag.",
    "Sidecar-Evolution": "Meta-Repo fuer KEEP/PARK/KILL-Entscheidungen und Evaluationsarchaeologie, strategisch stark aber indirekt.",
    "AgentArena": "Eval-Harness mit Benchmark-Logik und Visual-Explainer-Potenzial, aber weniger direkt in den taeglichen Workflow eingebettet.",
}


@dataclass
class Session:
    start: datetime
    end: datetime
    commit_count: int

    @property
    def span_hours(self) -> float:
        return max(0.0, (self.end - self.start).total_seconds() / 3600)

    @property
    def estimate_hours(self) -> float:
        padded = self.span_hours + 0.65 + max(0, self.commit_count - 1) * 0.18
        return max(0.5, min(4.5, padded))


def run_git(repo: Path, *args: str) -> str:
    result = subprocess.run(
        ["git", "-C", str(repo), *args],
        check=False,
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()


def list_repos() -> list[Path]:
    repos = []
    for entry in sorted(ROOT.iterdir()):
        if not entry.is_dir() or entry.name.startswith("."):
            continue
        if (entry / ".git").is_dir():
            repos.append(entry)
    return repos


def parse_log(repo: Path) -> list[dict[str, Any]]:
    out = run_git(
        repo,
        "log",
        f"--since={WINDOW_START.isoformat(sep=' ')}",
        f"--until={WINDOW_END.isoformat(sep=' ')}",
        "--date=iso-strict",
        "--pretty=format:%ad%x1f%s%x1f%h",
    )
    commits = []
    for line in out.splitlines():
        if not line.strip():
            continue
        date_str, subject, short_hash = line.split("\x1f")
        ts = datetime.fromisoformat(date_str)
        commits.append(
            {
                "timestamp": ts,
                "date": ts.strftime("%Y-%m-%d"),
                "time": ts.strftime("%H:%M"),
                "subject": subject,
                "hash": short_hash,
            }
        )
    return commits


def build_sessions(commits: list[dict[str, Any]]) -> list[Session]:
    if not commits:
        return []
    ordered = sorted((c["timestamp"] for c in commits))
    sessions: list[Session] = []
    current = [ordered[0]]
    for ts in ordered[1:]:
        if (ts - current[-1]).total_seconds() / 60 > SESSION_GAP_MINUTES:
            sessions.append(Session(current[0], current[-1], len(current)))
            current = [ts]
        else:
            current.append(ts)
    sessions.append(Session(current[0], current[-1], len(current)))
    return sessions


def find_artifacts(repo: Path) -> list[Path]:
    artifacts = []
    patterns = ("README", "PLAN", "REVIEW", "BACKLOG", "FOR_SMLFLG")
    for path in repo.rglob("*"):
        if path.is_dir():
            continue
        if ".git" in path.parts:
            continue
        if len(path.relative_to(repo).parts) > 3:
            continue
        if any(path.name.upper().startswith(prefix) for prefix in patterns):
            artifacts.append(path)
    return sorted(artifacts)


def latest_activity(repo: Path) -> str | None:
    latest = 0.0
    for path in repo.rglob("*"):
        if ".git" in path.parts:
            continue
        try:
            if path.is_file():
                latest = max(latest, path.stat().st_mtime)
        except OSError:
            continue
    if not latest:
        return None
    return datetime.fromtimestamp(latest).strftime("%Y-%m-%d %H:%M")


def in_window(date_text: str | None) -> bool:
    if not date_text:
        return False
    ts = datetime.strptime(date_text, "%Y-%m-%d %H:%M")
    return WINDOW_START.replace(tzinfo=None) <= ts <= WINDOW_END.replace(tzinfo=None)


def classify_project(repo: Path, artifacts: list[Path]) -> tuple[str, str]:
    names = {p.name.lower() for p in repo.iterdir()}
    text = " ".join(path.name.lower() for path in artifacts)
    repo_name = repo.name.lower()
    if "sidecar" in repo_name:
        return "Sidecar / Meta Tooling", "Agent workflow instrumentation, hooks, evals or platforming."
    if any(token in repo_name for token in ("agent", "swarm", "arena")):
        return "Agent Infrastructure", "Agent evaluation, orchestration or execution surface."
    if {"templates", "static"}.issubset(names) or "htmx" in text:
        return "Vertical App", "Local web app pattern with templates/static/tests suggests end-user MVPs."
    if any(token in repo_name for token in ("panel", "dashboard", "monitor", "counter", "voicemode")):
        return "Personal Workflow OS", "Observability or desktop support tooling for your own coding environment."
    if any(token in repo_name for token in ("prompt", "writing", "bookmark")):
        return "Knowledge / Content", "Prompt, content or knowledge capture system."
    return "Tooling", "General-purpose local project tooling."


def summarize_commit_types(commits: list[dict[str, Any]]) -> dict[str, int]:
    counts = Counter()
    for commit in commits:
        subject = commit["subject"].lower()
        for key in ("feat", "fix", "test", "docs", "refactor", "chore"):
            if subject.startswith(f"{key}:") or subject.startswith(f"{key}(") or subject.startswith(f"{key} "):
                counts[key] += 1
                break
        else:
            counts["other"] += 1
    return dict(counts)


def has_uncommitted_work(repo: Path) -> bool:
    return bool(run_git(repo, "status", "--short"))


def repo_summary(repo: Path) -> dict[str, Any]:
    commits = parse_log(repo)
    sessions = build_sessions(commits)
    artifacts = find_artifacts(repo)
    cluster, cluster_reason = classify_project(repo, artifacts)
    branch = run_git(repo, "rev-parse", "--abbrev-ref", "HEAD") or "unknown"
    status = run_git(repo, "status", "--short")
    status_label = "clean"
    if status:
        status_label = "dirty" if any(not line.startswith("??") for line in status.splitlines()) else "untracked-only"
    commit_days = sorted({c["date"] for c in commits})
    commit_type_counts = summarize_commit_types(commits)
    plan_count = sum(1 for path in artifacts if path.name.upper().startswith("PLAN"))
    review_count = sum(1 for path in artifacts if path.name.upper().startswith("REVIEW"))
    readme_count = sum(1 for path in artifacts if path.name.upper().startswith("README"))
    backlog_count = sum(1 for path in artifacts if path.name.upper().startswith("BACKLOG"))
    has_tests = (repo / "tests").exists() or (repo / "e2e").exists()
    has_eval = (repo / "eval").exists()
    has_docs = (repo / "docs").exists()
    return {
        "name": repo.name,
        "path": str(repo),
        "branch": branch,
        "status": status_label,
        "has_uncommitted_work": has_uncommitted_work(repo),
        "latest_activity": latest_activity(repo),
        "cluster": cluster,
        "cluster_reason": cluster_reason,
        "commit_count": len(commits),
        "commit_days": commit_days,
        "commit_day_count": len(commit_days),
        "commits_sample": [
            {"date": c["date"], "time": c["time"], "subject": c["subject"], "hash": c["hash"]}
            for c in commits[:8]
        ],
        "first_commit": commits[-1]["date"] if commits else None,
        "last_commit": commits[0]["date"] if commits else None,
        "artifact_counts": {
            "readme": readme_count,
            "plan": plan_count,
            "review": review_count,
            "backlog": backlog_count,
        },
        "structure_markers": {
            "tests": has_tests,
            "eval": has_eval,
            "docs": has_docs,
            "systemd": (repo / "systemd").exists(),
            "knowledge": (repo / "knowledge").exists(),
            "adapters": (repo / "adapters").exists(),
            "core": (repo / "core").exists(),
            "samuel": (repo / "samuel").exists(),
        },
        "commit_type_counts": commit_type_counts,
        "session_count": len(sessions),
        "sessions": [
            {
                "start": s.start.isoformat(),
                "end": s.end.isoformat(),
                "commit_count": s.commit_count,
                "estimate_hours": round(s.estimate_hours, 2),
            }
            for s in sessions
        ],
        "estimated_hours": round(sum(s.estimate_hours for s in sessions), 2),
    }


def estimate_work_hours(projects: list[dict[str, Any]]) -> dict[str, Any]:
    active = [
        p
        for p in projects
        if p["commit_count"] > 0 or (p["has_uncommitted_work"] and in_window(p["latest_activity"]))
    ]
    session_hours = sum(p["estimated_hours"] for p in active)
    artifact_bonus = 0.0
    uncertainty = []
    for project in active:
        artifact_count = sum(project["artifact_counts"].values())
        if artifact_count:
            artifact_bonus += min(1.0, 0.15 * artifact_count)
        if project["has_uncommitted_work"] and project["commit_count"] == 0 and in_window(project["latest_activity"]):
            artifact_bonus += 0.75
            uncertainty.append(f"{project['name']} hatte lokale Aktivitaet ohne Commit-Spur im Fenster.")
    best = round(session_hours + artifact_bonus, 1)
    lower = round(max(1.0, session_hours * 0.72), 1)
    upper = round(best * 1.38 + 4.0, 1)
    return {
        "best_estimate_hours": best,
        "lower_bound_hours": lower,
        "upper_bound_hours": upper,
        "active_project_count": len(active),
        "method": [
            "Commit-Timestamps im Fenster 2026-02-19 bis 2026-03-20 zu Sessions gebuendelt (neue Session ab >120 Minuten Luecke).",
            "Pro Session Schaetzung aus beobachtetem Commit-Span plus Puffer fuer Vor-/Nacharbeit, gedeckelt um einzelne Squash- oder Burst-Sessions nicht zu uebergewichten.",
            "Zusatzbonus fuer PLAN/REVIEW/BACKLOG-Artefakte und uncommitted Aktivitaet, weil Cloud-Code-Arbeit oft ausserhalb des finalen Commits sichtbar wird.",
        ],
        "uncertainties": uncertainty
        + [
            "Nicht jede Arbeitsminute landet in Git; besonders lokale Experimente, Copy/Paste, Prompting und verworfene Ansaetze fehlen teilweise.",
            "Einige Commits koennen squash-/batch-artig sein und daher mehr reale Arbeit reprasentieren als die Zeitdifferenzen zeigen.",
        ],
    }


def build_timeline(projects: list[dict[str, Any]]) -> list[dict[str, Any]]:
    days = []
    cursor = WINDOW_START.date()
    end_date = WINDOW_END.date()
    while cursor <= end_date:
        day_projects = []
        commit_total = 0
        for project in projects:
            matches = [c for c in project["commits_sample"]]  # placeholder for schema consistency
            count = sum(1 for session in project["sessions"] if session["start"][:10] == cursor.isoformat())
            if count:
                day_projects.append({"name": project["name"], "session_count": count})
            commit_total += sum(1 for d in project["commit_days"] if d == cursor.isoformat())
        intensity = min(4, commit_total)
        days.append(
            {
                "date": cursor.isoformat(),
                "commit_days": commit_total,
                "intensity": intensity,
                "projects": day_projects,
            }
        )
        cursor += timedelta(days=1)
    return days


def workflow_patterns(projects: list[dict[str, Any]]) -> dict[str, Any]:
    commit_types = Counter()
    active_projects = [p for p in projects if p["commit_count"] > 0]
    day_to_projects: dict[str, set[str]] = defaultdict(set)
    for project in active_projects:
        commit_types.update(project["commit_type_counts"])
        for day in project["commit_days"]:
            day_to_projects[day].add(project["name"])
    multi_project_days = sum(1 for names in day_to_projects.values() if len(names) >= 2)
    prototype_signals = commit_types["feat"] + commit_types["other"]
    refinement_signals = commit_types["fix"] + commit_types["test"] + commit_types["refactor"] + commit_types["docs"]
    eval_projects = [
        p["name"]
        for p in projects
        if p["structure_markers"]["eval"] or p["artifact_counts"]["review"] or p["commit_type_counts"].get("test", 0)
    ]
    top_active = sorted(active_projects, key=lambda p: p["commit_count"], reverse=True)[:5]
    return {
        "startup_pattern": "Du startest Projekte meist mit schriftlicher Rahmung: READMEs, PLAN-Dateien und Kontext-Artefakte tauchen sehr frueh auf und bleiben als Steuerungsoberflaeche erhalten.",
        "iteration_pattern": "Die dominante Schleife ist nicht linearer Build, sondern schnelle Zyklen aus feat/fix/test/docs. Sidecar-Projekte zeigen besonders stark: erst Scope aufspannen, dann messen, dann radikal zusammenstreichen.",
        "parallel_vs_serial": {
            "label": "Hybrid mit Sidecar als seriellem Hauptstrang",
            "evidence": f"{multi_project_days} Tage im Fenster zeigen Aktivitaet in mindestens zwei Projekten; gleichzeitig konzentrieren sich die meisten Commits auf wenige Kernrepos mit Sidecar-ng, Sidecar-V7, ClaudeCodePanel und AgentArena.",
        },
        "prototyping_vs_refinement": {
            "prototype_signals": prototype_signals,
            "refinement_signals": refinement_signals,
            "interpretation": "Der Monat kippt nach fruehem Scope-Aufbau deutlich in Refinement: Tests, Doku, Refactors und Eval-Artefakte sind in den aktiven Kernprojekten stark sichtbar.",
        },
        "evaluation_loops": {
            "projects": eval_projects[:8],
            "interpretation": "Eval-/Review-Loops sind kein Nachgedanke. AgentArena, Sidecar-ng, Sidecar-Evolution und ClaudeCodePanel zeigen Mess-, Review- oder Validierungsartefakte direkt im Repo.",
        },
        "preferred_project_types": [
            "Agentische Infrastruktur und Eval-Harnesses",
            "Persoenliche Workflow- und Observability-Tools",
            "Lokale Python-Web-MVPs mit templates/static/tests",
        ],
        "meta_systems": [
            "Selbstinstrumentierung des eigenen Entwicklungsprozesses",
            "Core-vs-Personalization-Trennung in reiferen Systemen",
            "Planungsrepo und Implementierungsrepo bewusst auseinanderziehen",
        ],
        "featured_projects": [
            {"name": project["name"], "commit_count": project["commit_count"], "cluster": project["cluster"]}
            for project in top_active
        ],
    }


def ranking(projects: list[dict[str, Any]]) -> dict[str, Any]:
    lookup = {project["name"]: project for project in projects}
    rows = []
    for name in TOP_RECOMMENDATION_ORDER:
        if name not in lookup:
            continue
        scores = RANKING_SCORES[name]
        weighted = round(
            (
                scores["leverage"] * 0.24
                + scores["reusability"] * 0.16
                + scores["strategic_value"] * 0.18
                + scores["product_readiness"] * 0.14
                + scores["synergy"] * 0.16
                + scores["workflow_fit"] * 0.12
            ),
            2,
        )
        project = lookup[name]
        rows.append(
            {
                "name": name,
                "score": weighted,
                "scores": scores,
                "commit_count": project["commit_count"],
                "last_commit": project["last_commit"],
                "summary": PROJECT_NOTES[name],
            }
        )
    rows.sort(key=lambda row: row["score"], reverse=True)
    return {
        "top_recommendation": rows[0],
        "top_three": rows[:3],
        "shortlist": rows[:5],
        "method": "Editorial scoring auf sechs Kriterien, kombiniert mit Aktivitaetssignalen, Doku-Dichte und Synergie zum Cloud-Code-Workflow.",
    }


def build_executive_summary(projects: list[dict[str, Any]], work_estimate: dict[str, Any], rank: dict[str, Any]) -> dict[str, Any]:
    active = sorted(
        (
            p
            for p in projects
            if p["commit_count"] > 0 or (p["has_uncommitted_work"] and in_window(p["latest_activity"]))
        ),
        key=lambda p: p["commit_count"],
        reverse=True,
    )
    headline = (
        f"Im 30-Tage-Fenster liegen geschaetzt {work_estimate['best_estimate_hours']} Stunden Cloud-Code-Arbeit "
        f"ueber {work_estimate['active_project_count']} aktiv sichtbare Projekte."
    )
    return {
        "headline": headline,
        "key_findings": [
            "Der Monat war kein einzelnes Produktprojekt, sondern ein Portfolio aus Meta-Tooling, Evaluationssystemen und persoenlichen Workflow-Oberflaechen.",
            "Die staerkste Prozesssignatur ist instrumentierte Iteration: PLAN/REVIEW/EVAL-Dateien und test-/docs-lastige Commits begleiten die Umsetzung eng.",
            f"Die beste Weiterentwicklungswette ist {rank['top_recommendation']['name']}, weil dort Hebel, Reifegrad und direkter Workflow-Fit am staerksten zusammenfallen.",
        ],
        "top_activity": [
            {
                "name": project["name"],
                "commit_count": project["commit_count"],
                "session_count": project["session_count"],
                "cluster": project["cluster"],
            }
            for project in active[:6]
        ],
    }


def build_clusters(projects: list[dict[str, Any]]) -> list[dict[str, Any]]:
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for project in projects:
        grouped[project["cluster"]].append(project)
    output = []
    for cluster, items in grouped.items():
        output.append(
            {
                "name": cluster,
                "count": len(items),
                "projects": [
                    {
                        "name": item["name"],
                        "commit_count": item["commit_count"],
                        "status": item["status"],
                    }
                    for item in sorted(items, key=lambda p: p["commit_count"], reverse=True)[:8]
                ],
                "summary": items[0]["cluster_reason"],
            }
        )
    output.sort(key=lambda item: item["count"], reverse=True)
    return output


def build_evidence(projects: list[dict[str, Any]]) -> list[dict[str, Any]]:
    lookup = {p["name"]: p for p in projects}
    evidence_rows = []
    for name in ["Sidecar-ng", "Sidecar-V7", "ClaudeCodePanel", "AgentArena", "MyAIGame", "Sidecar-Evolution", "Sidecar-V8"]:
        project = lookup.get(name)
        if not project:
            continue
        evidence_rows.append(
            {
                "project": name,
                "commits": project["commit_count"],
                "last_commit": project["last_commit"],
                "artifacts": project["artifact_counts"],
                "markers": project["structure_markers"],
                "sample_messages": [entry["subject"] for entry in project["commits_sample"][:3]],
            }
        )
    return evidence_rows


def build_payload() -> dict[str, Any]:
    repos = [repo for repo in list_repos() if repo.name not in IGNORED_PROJECTS]
    projects = [repo_summary(repo) for repo in repos]
    projects.sort(key=lambda project: (project["commit_count"], project["estimated_hours"]), reverse=True)
    work_estimate = estimate_work_hours(projects)
    rank = ranking(projects)
    return {
        "meta": {
            "generated_at": datetime.now().isoformat(timespec="seconds"),
            "source_root": str(ROOT),
            "window_start": WINDOW_START.date().isoformat(),
            "window_end": WINDOW_END.date().isoformat(),
            "ignored_projects": sorted(IGNORED_PROJECTS),
            "project_count": len(projects),
        },
        "executive_summary": build_executive_summary(projects, work_estimate, rank),
        "work_estimate": work_estimate,
        "workflow_patterns": workflow_patterns(projects),
        "project_clusters": build_clusters(projects),
        "timeline": build_timeline(projects),
        "rankings": rank,
        "evidence": build_evidence(projects),
        "projects": projects,
    }


def main() -> None:
    payload = build_payload()
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    print(f"Wrote {OUT}")


if __name__ == "__main__":
    main()