"""Chunker: zerlegt ein Repo mechanisch (kein LLM) in kartengroße Scan-Aufträge. Phase-0-Befund: dateigroße LLM-Calls dauern Minuten und große Dateien stallen — deshalb Chunks von ~CHUNK_ZEILEN Zeilen, geschnitten ENTLANG von Top-Level-Symbolen (nie mitten durch eine Funktion). v1 scannt nur Python-Dateien. """ import re from pathlib import Path CHUNK_ZEILEN = 250 # Zielgröße; ein einzelnes Riesen-Symbol darf größer sein ENDUNGEN = {".py"} # v1: nur Python (Creator/Planer-Backends); Rest später AUSSCHLUSS_DIRS = {".git", ".planer", "node_modules", "__pycache__", ".pytest_cache", "dist", "build", "storage", "venv", ".venv", ".idea", ".vscode", "tests"} # v1: Test-Code erzeugt in den Sichten nur Rauschen AUSSCHLUSS_DATEIEN = re.compile(r"^\.env") # Secrets: nie lesen (Leitprinzip 8) _SYMBOL_RE = re.compile(r"^(?:async\s+)?(?:def|class)\s+(\w+)") def dateien(wurzel: Path) -> list[Path]: """Scanbare Quelldateien unter `wurzel`, Ausschlussliste angewandt, sortiert.""" out = [] for p in sorted(wurzel.rglob("*")): if not p.is_file() or p.suffix not in ENDUNGEN: continue rel = p.relative_to(wurzel) if any(teil in AUSSCHLUSS_DIRS for teil in rel.parts[:-1]): continue if AUSSCHLUSS_DATEIEN.match(p.name): continue out.append(p) return out def symbol_zeilen(zeilen: list[str]) -> list[tuple[int, str]]: """(zeilennummer_0basiert, symbolname) aller Top-Level-def/class, in Dateireihenfolge.""" return [(i, m.group(1)) for i, z in enumerate(zeilen) if (m := _SYMBOL_RE.match(z))] def top_level_symbole(text: str) -> set[str]: """Alle Top-Level-Symbole (def/class/Konstanten) — Abdeckungs-Referenz für Gates.""" out = {name for _, name in symbol_zeilen(text.splitlines())} out |= {m.group(1) for m in re.finditer(r"^(\w+)\s*[:=]", text, re.M)} return out def _block_start(zeilen: list[str], symbol_zeile: int) -> int: """Dekoratoren/Kommentare direkt über dem Symbol gehören zum Block.""" i = symbol_zeile while i > 0 and (zeilen[i - 1].startswith("@") or zeilen[i - 1].lstrip().startswith("#")): i -= 1 return i def schneide(datei_rel: str, text: str) -> list[dict]: """Datei → Chunks entlang Top-Level-Symbolen, je ~CHUNK_ZEILEN Zeilen. Chunk: {datei, start, ende (exklusiv, 0-basiert), symbole, text, groesse}. Der Kopf vor dem ersten Symbol (Docstring/Imports/Konstanten) gehört zum ersten Chunk.""" zeilen = text.splitlines() symbole = symbol_zeilen(zeilen) if not symbole: # Datei ohne def/class (Config, __init__): ein Chunk, Datei-Anker return [{"datei": datei_rel, "start": 0, "ende": len(zeilen), "symbole": [], "text": text, "groesse": len(zeilen)}] if zeilen else [] # Schnittpunkte: Blockanfänge der Symbole; Kopf hängt am ersten Block. grenzen = [( _block_start(zeilen, z), name) for z, name in symbole] chunks: list[dict] = [] start = 0 aktuelle_symbole: list[str] = [] for idx, (block_start, name) in enumerate(grenzen): naechster = grenzen[idx + 1][0] if idx + 1 < len(grenzen) else len(zeilen) if aktuelle_symbole and (naechster - start) > CHUNK_ZEILEN: # Symbol beginnt neuen Chunk, sonst würde der aktuelle zu groß chunks.append(_chunk(datei_rel, zeilen, start, block_start, aktuelle_symbole)) start = block_start aktuelle_symbole = [] aktuelle_symbole.append(name) chunks.append(_chunk(datei_rel, zeilen, start, len(zeilen), aktuelle_symbole)) return chunks def _chunk(datei: str, zeilen: list[str], start: int, ende: int, symbole: list[str]) -> dict: return {"datei": datei, "start": start, "ende": ende, "symbole": list(symbole), "text": "\n".join(zeilen[start:ende]), "groesse": ende - start} def chunks_fuer_repo(wurzel: Path, nur_dateien: set[str] | None = None) -> list[dict]: """Alle Chunks des Repos; `nur_dateien` (relative Pfade) schränkt ein (Re-Scan).""" out = [] for p in dateien(wurzel): rel = str(p.relative_to(wurzel)) if nur_dateien is not None and rel not in nur_dateien: continue out.extend(schneide(rel, p.read_text(encoding="utf-8", errors="replace"))) return out