97 lines
4.3 KiB
Python
97 lines
4.3 KiB
Python
"""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
|