ai-agent/scripts/backfill-artefacts.py

#!/usr/bin/env python3
"""One-off backfill: copy past conversations' generated artifacts into the
per-project artefact folders the viewer now serves.

Scans every archived transcript (services/ai-agent/data/transcripts/**), finds
mentions of generated media — docs/screenshots/… captures and images/videos
under the repo's data/ tree — and copies each file that still exists into

    <projectFolder>/.ai/artefacts/<conv-start-date>/<sessionId>/<file>

where <projectFolder> is the canonical dir of the conversation's first
project/service tag (manual conv-meta tags first, then slugs auto-mined from
the transcript the same way the backend does). Conversations with no
resolvable project folder, and mentioned files that no longer exist on disk,
are skipped. Idempotent: existing targets are never overwritten.

Run on the host from the homelab main checkout:

    python3 services/ai-agent/scripts/backfill-artefacts.py [--dry-run]
"""
from __future__ import annotations

import json
import re
import shutil
import sys
from collections import Counter
from pathlib import Path

HOME = Path.home()
HOMELAB = HOME / "homelab"
PROJECTS = HOME / "projects"
TRANSCRIPTS = HOMELAB / "services/ai-agent/data/transcripts"
META = HOMELAB / "services/ai-agent/data/conversations-meta.json"

MEDIA_EXT = r"(?:jpe?g|png|webp|gif|bmp|avif|mp4|webm|mov|m4v|ogv)"
# Generated-artifact locations only: screenshot-skill captures and the repo's
# data/ tree (the old CLAUDE.md artifact convention). Plain source-asset reads
# (public/images/…) are intentionally not treated as artefacts.
SHOT_RE = re.compile(rf"docs/screenshots/([A-Za-z0-9._/-]+\.{MEDIA_EXT})", re.I)
DATA_RE = re.compile(rf"(?:^|[/\s\"'`])data/([A-Za-z0-9._/-]+\.{MEDIA_EXT})", re.I)
# Same slug mining as backend/conversations.py (_PROJECT_RE/_SERVICE_RE).
PROJECT_RE = re.compile(r"(?:^|[/\s])projects/([A-Za-z0-9._][\w.-]*)")
SERVICE_RE = re.compile(r"(?:^|[/\s])services/([A-Za-z0-9._][\w.-]*)")
TS_RE = re.compile(r'"timestamp"\s*:\s*"(\d{4}-\d{2}-\d{2})')


def project_dir(slug: str) -> Path | None:
    for base in (PROJECTS, HOMELAB / "services"):
        d = base / Path(slug).name
        if d.is_dir() and not slug.startswith("."):
            return d
    return None


def dest_dir(meta: dict, text: str) -> Path | None:
    """The conversation's canonical project folder: manual tags win, then the
    most-mentioned auto-mined slug that maps to an existing directory."""
    for slug in list(meta.get("projects") or []) + list(meta.get("services") or []):
        d = project_dir(slug)
        if d:
            return d
    counts: Counter[str] = Counter()
    for rx in (PROJECT_RE, SERVICE_RE):
        counts.update(m.group(1) for m in rx.finditer(text))
    for slug, _n in counts.most_common():
        d = project_dir(slug)
        if d:
            return d
    return None


def transcripts() -> list[tuple[str, Path]]:
    """(sessionId, transcript path) pairs: main transcripts plus subagent
    sidechains (attributed to their parent conversation id)."""
    out = []
    for p in sorted(TRANSCRIPTS.glob("*/*.jsonl")):
        out.append((p.stem, p))
    for p in sorted(TRANSCRIPTS.glob("*/*/subagents/agent-*.jsonl")):
        out.append((p.parent.parent.name, p))
    return out


def main() -> int:
    dry = "--dry-run" in sys.argv
    try:
        meta_all = json.loads(META.read_text())
    except Exception:
        meta_all = {}

    copied = skipped_missing = existing = 0
    convs_with = 0
    per_dest: Counter[str] = Counter()
    no_dest: list[str] = []
    dates: dict[str, str] = {}  # parent sid -> conv start date

    for sid, path in transcripts():
        try:
            text = path.read_text(errors="replace")
        except OSError:
            continue
        rels = [(HOMELAB / "docs/screenshots" / m.group(1), m.group(1))
                for m in SHOT_RE.finditer(text)]
        rels += [(HOMELAB / "data" / m.group(1), m.group(1))
                 for m in DATA_RE.finditer(text)]
        if not rels:
            continue
        dest = dest_dir(meta_all.get(sid) or {}, text)
        if dest is None:
            no_dest.append(sid[:8])
            continue
        if sid not in dates:
            m = TS_RE.search(text)
            dates[sid] = m.group(1) if m else "unknown-date"
        date = dates[sid]

        seen: set[Path] = set()
        conv_hit = False
        for src, _rel in rels:
            if src in seen:
                continue
            seen.add(src)
            if not src.is_file():
                skipped_missing += 1
                continue
            target = dest / ".ai" / "artefacts" / date / sid / src.name
            if target.exists():
                existing += 1
                continue
            conv_hit = True
            copied += 1
            per_dest[str(dest)] += 1
            if not dry:
                target.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(src, target)
                # Keep standalone project repos clean (homelab's .gitignore
                # already covers services/*/.ai/artefacts/).
                gitdir = dest / ".git"
                if gitdir.is_dir():
                    excl = gitdir / "info" / "exclude"
                    try:
                        cur = excl.read_text() if excl.is_file() else ""
                        if ".ai/" not in cur.splitlines():
                            excl.parent.mkdir(parents=True, exist_ok=True)
                            excl.write_text(cur.rstrip("\n") + "\n.ai/\n"
                                            if cur else ".ai/\n")
                    except OSError:
                        pass
        if conv_hit:
            convs_with += 1

    print(f"{'DRY RUN — ' if dry else ''}copied {copied} artefacts "
          f"across {convs_with} conversations "
          f"({existing} already present, {skipped_missing} mentioned files gone)")
    for d, n in per_dest.most_common():
        print(f"  {n:4}  {d}")
    if no_dest:
        print(f"  (no project folder resolvable for {len(no_dest)} conversations "
              f"with media mentions: {', '.join(no_dest[:12])}"
              f"{'…' if len(no_dest) > 12 else ''})")
    return 0


if __name__ == "__main__":
    sys.exit(main())