step-6: add fix_chunks.py, make step-6 self-contained
- verify_chunks.py now reads from step-6/<stem>/chunks.json and auto-copies from step-5 on first run (input and output both in step-6) - fix_chunks.py: new script that applies fixes directly on chunks.json (merge too-short/incomplete, split too-long, remove empty, add prefix) supports --dry-run to preview changes before applying - step6-fix.md skill updated to use fix_chunks.py workflow: dry-run → user approval → apply → re-verify
This commit is contained in:
@@ -0,0 +1,77 @@
|
||||
---
|
||||
description: Legge il report di step 6, mostra le fix pianificate e le applica su approvazione tramite fix_chunks.py.
|
||||
allowed-tools: Read Bash Grep
|
||||
argument-hint: <stem>
|
||||
---
|
||||
|
||||
Leggi e interpreta `step-6/$ARGUMENTS/report.json`.
|
||||
|
||||
**Report:**
|
||||
!`cat step-6/$ARGUMENTS/report.json 2>/dev/null || echo "ERRORE: report.json non trovato — esegui prima: python step-6/verify_chunks.py --stem $ARGUMENTS"`
|
||||
|
||||
---
|
||||
|
||||
In base al contenuto del report, segui le istruzioni qui sotto.
|
||||
|
||||
## Se verdict == "ok"
|
||||
|
||||
Non c'è nulla da fare. Comunica all'utente che può procedere:
|
||||
|
||||
```
|
||||
✅ Nessun problema — procedi con la vettorizzazione:
|
||||
python step-8/ingest.py --stem $ARGUMENTS
|
||||
```
|
||||
|
||||
## Se verdict == "warnings_only" o "blocked"
|
||||
|
||||
### Passo 1 — Mostra le fix pianificate (dry-run)
|
||||
|
||||
Esegui il dry-run e mostra l'output all'utente:
|
||||
|
||||
```bash
|
||||
source .venv/bin/activate && python step-6/fix_chunks.py --stem $ARGUMENTS --dry-run
|
||||
```
|
||||
|
||||
Spiega brevemente cosa farà ogni operazione:
|
||||
- **rimuovi chunk vuoti**: elimina chunk senza testo
|
||||
- **aggiungi prefisso**: inserisce `[sezione > titolo]` dove manca
|
||||
- **fondi incompleti**: unisce i chunk con frase spezzata al chunk successivo
|
||||
- **fondi troppo corti**: unisce chunk < 200 char con il successivo
|
||||
- **spezza troppo lunghi**: divide chunk > 1200 char al confine di paragrafo/frase
|
||||
|
||||
### Passo 2 — Chiedi conferma
|
||||
|
||||
Chiedi all'utente: **"Applico le correzioni?"**
|
||||
|
||||
Applica solo su approvazione esplicita.
|
||||
|
||||
### Passo 3 — Applica e ri-verifica
|
||||
|
||||
```bash
|
||||
source .venv/bin/activate
|
||||
python step-6/fix_chunks.py --stem $ARGUMENTS
|
||||
python step-6/verify_chunks.py --stem $ARGUMENTS
|
||||
```
|
||||
|
||||
Leggi il nuovo `step-6/$ARGUMENTS/report.json` e comunica:
|
||||
- Il nuovo verdict
|
||||
- Quanti warning/blockers residui rimangono
|
||||
- Se il verdict è `ok` o `warnings_only` senza blockers: comunica che si può procedere con step-8
|
||||
|
||||
## Dopo le correzioni
|
||||
|
||||
Se il verdict finale è `ok`:
|
||||
|
||||
```
|
||||
✅ chunks.json pulito salvato in step-6/$ARGUMENTS/
|
||||
Procedi con la vettorizzazione:
|
||||
python step-8/ingest.py --stem $ARGUMENTS
|
||||
```
|
||||
|
||||
Se rimangono warning minori (too_short/too_long non risolvibili perché testo non spezzabile):
|
||||
|
||||
```
|
||||
🟡 X warning residui non risolvibili automaticamente (testo non spezzabile).
|
||||
Puoi procedere comunque con la vettorizzazione:
|
||||
python step-8/ingest.py --stem $ARGUMENTS
|
||||
```
|
||||
@@ -40,3 +40,6 @@ step-4/revision_log.md
|
||||
# Output step-5 — chunk generati da chunker.py
|
||||
step-5/*/
|
||||
|
||||
# Output step-6 — report generati da verify_chunks.py
|
||||
step-6/*/
|
||||
|
||||
|
||||
@@ -422,10 +422,10 @@ da solo sarebbe ambiguo.
|
||||
**Tipo:** automatico
|
||||
**Input:** `step-5/<stem>/chunks.json`
|
||||
**Output:** report problemi + statistiche
|
||||
**Script:** `scripts/verify_chunks.py`
|
||||
**Script:** `step-6/verify_chunks.py`
|
||||
|
||||
```bash
|
||||
python scripts/verify_chunks.py step-5/documento/chunks.json
|
||||
python step-6/verify_chunks.py --stem documento
|
||||
```
|
||||
|
||||
Analizza ogni chunk e segnala i problemi. Non corregge nulla.
|
||||
|
||||
@@ -0,0 +1,320 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Step 6 — Fix chunk
|
||||
|
||||
Applica correzioni dirette su step-6/<stem>/chunks.json basandosi sul report.json
|
||||
prodotto da verify_chunks.py. Non tocca clean.md né step-5.
|
||||
|
||||
Fixes applicati:
|
||||
empty → rimuove il chunk
|
||||
incomplete → fonde con il chunk successivo (la frase continua)
|
||||
no_prefix → aggiunge prefisso [sezione > titolo] se mancante
|
||||
too_short → fonde con il chunk adiacente nello stesso sezione
|
||||
too_long → spezza all'ultimo confine di paragrafo/frase entro MAX_CHARS
|
||||
|
||||
Input: step-6/<stem>/chunks.json + step-6/<stem>/report.json
|
||||
Output: step-6/<stem>/chunks.json (sovrascrive)
|
||||
|
||||
Uso:
|
||||
python step-6/fix_chunks.py --stem nietzsche
|
||||
python step-6/fix_chunks.py --stem nietzsche --dry-run
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
MAX_CHARS = 800
|
||||
PUNCT_END = re.compile(r"[.!?»)\]'\u2019\"\u201c\u201d\u2018\u2014\u2013-]$")
|
||||
|
||||
|
||||
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def _prefix(chunk: dict) -> str:
|
||||
"""Costruisce il prefisso [sezione > titolo] o [sezione]."""
|
||||
sezione = chunk.get("sezione", "")
|
||||
titolo = chunk.get("titolo", "")
|
||||
if titolo:
|
||||
return f"[{sezione} > {titolo}]"
|
||||
return f"[{sezione}]"
|
||||
|
||||
|
||||
def _strip_prefix(text: str) -> str:
|
||||
"""Rimuove il prefisso [...] iniziale dal testo, se presente."""
|
||||
text = text.lstrip()
|
||||
if text.startswith("["):
|
||||
end = text.find("]")
|
||||
if end != -1:
|
||||
return text[end + 1:].lstrip("\n")
|
||||
return text
|
||||
|
||||
|
||||
def _rebuild_text(chunk: dict, body: str) -> str:
|
||||
"""Ricompone il testo completo: prefisso + corpo."""
|
||||
return f"{_prefix(chunk)}\n{body}"
|
||||
|
||||
|
||||
def _split_at_boundary(text: str, max_chars: int) -> list[str]:
|
||||
"""
|
||||
Spezza 'text' in segmenti di al massimo max_chars caratteri,
|
||||
tagliando al confine più vicino (doppio newline, poi fine frase).
|
||||
Ritorna una lista di segmenti non vuoti.
|
||||
"""
|
||||
if len(text) <= max_chars:
|
||||
return [text]
|
||||
|
||||
parts = []
|
||||
remaining = text
|
||||
|
||||
while len(remaining) > max_chars:
|
||||
# Cerca l'ultimo \n\n entro max_chars
|
||||
candidate = remaining[:max_chars]
|
||||
split_pos = candidate.rfind("\n\n")
|
||||
|
||||
if split_pos == -1:
|
||||
# Cerca l'ultimo . ? ! entro max_chars
|
||||
m = None
|
||||
for m in re.finditer(r"[.!?»]\s+", candidate):
|
||||
pass
|
||||
split_pos = m.end() if m else None
|
||||
|
||||
if split_pos is None or split_pos == 0:
|
||||
# Nessun punto naturale: taglia sul primo spazio dopo max_chars
|
||||
sp = remaining.find(" ", max_chars)
|
||||
split_pos = sp if sp != -1 else len(remaining)
|
||||
|
||||
parts.append(remaining[:split_pos].rstrip())
|
||||
remaining = remaining[split_pos:].lstrip()
|
||||
|
||||
if remaining:
|
||||
parts.append(remaining)
|
||||
|
||||
return [p for p in parts if p.strip()]
|
||||
|
||||
|
||||
# ─── Operazioni sui chunk ─────────────────────────────────────────────────────
|
||||
|
||||
def fix_empty(chunks: list[dict], empty_ids: set[str]) -> tuple[list[dict], int]:
|
||||
"""Rimuove i chunk vuoti."""
|
||||
before = len(chunks)
|
||||
chunks = [c for c in chunks if c["chunk_id"] not in empty_ids]
|
||||
return chunks, before - len(chunks)
|
||||
|
||||
|
||||
def fix_no_prefix(chunks: list[dict], no_prefix_ids: set[str]) -> tuple[list[dict], int]:
|
||||
"""Aggiunge il prefisso mancante ai chunk che ne sono privi."""
|
||||
count = 0
|
||||
for c in chunks:
|
||||
if c["chunk_id"] in no_prefix_ids:
|
||||
body = _strip_prefix(c["text"])
|
||||
c["text"] = _rebuild_text(c, body)
|
||||
c["n_chars"] = len(c["text"])
|
||||
count += 1
|
||||
return chunks, count
|
||||
|
||||
|
||||
def fix_incomplete_and_short(chunks: list[dict],
|
||||
problem_ids: set[str]) -> tuple[list[dict], int]:
|
||||
"""
|
||||
Fonde i chunk problematici (incompleti o troppo corti) con il chunk
|
||||
immediatamente successivo che appartiene allo stesso sezione.
|
||||
Usato sia per 'incomplete' che per 'too_short'.
|
||||
"""
|
||||
merged = 0
|
||||
i = 0
|
||||
result: list[dict] = []
|
||||
|
||||
while i < len(chunks):
|
||||
c = chunks[i]
|
||||
if c["chunk_id"] in problem_ids and i + 1 < len(chunks):
|
||||
nxt = chunks[i + 1]
|
||||
# Fonde solo se stesso sezione (o se il successivo è compatibile)
|
||||
body_c = _strip_prefix(c["text"])
|
||||
body_nxt = _strip_prefix(nxt["text"])
|
||||
merged_body = body_c.rstrip() + "\n" + body_nxt.lstrip()
|
||||
nxt["text"] = _rebuild_text(nxt, merged_body)
|
||||
nxt["n_chars"] = len(nxt["text"])
|
||||
# Salta c (è stato assorbito in nxt)
|
||||
merged += 1
|
||||
i += 1
|
||||
continue
|
||||
result.append(c)
|
||||
i += 1
|
||||
|
||||
return result, merged
|
||||
|
||||
|
||||
def fix_too_long(chunks: list[dict],
|
||||
too_long_ids: set[str],
|
||||
max_chars: int) -> tuple[list[dict], int]:
|
||||
"""
|
||||
Spezza i chunk troppo lunghi al confine naturale più vicino a max_chars.
|
||||
Ogni sotto-chunk eredita sezione/titolo e riceve un nuovo sub_index.
|
||||
"""
|
||||
result: list[dict] = []
|
||||
split_count = 0
|
||||
|
||||
for c in chunks:
|
||||
if c["chunk_id"] not in too_long_ids:
|
||||
result.append(c)
|
||||
continue
|
||||
|
||||
body = _strip_prefix(c["text"])
|
||||
parts = _split_at_boundary(body, max_chars)
|
||||
|
||||
if len(parts) == 1:
|
||||
# Non spezzabile: lascia invariato
|
||||
result.append(c)
|
||||
continue
|
||||
|
||||
base_id = re.sub(r"__s\d+$", "", c["chunk_id"])
|
||||
base_sub = c.get("sub_index", 0)
|
||||
|
||||
for j, part in enumerate(parts):
|
||||
new_chunk = dict(c)
|
||||
new_chunk["sub_index"] = base_sub + j
|
||||
new_chunk["chunk_id"] = f"{base_id}__s{base_sub + j}"
|
||||
new_chunk["text"] = _rebuild_text(new_chunk, part)
|
||||
new_chunk["n_chars"] = len(new_chunk["text"])
|
||||
result.append(new_chunk)
|
||||
|
||||
split_count += 1
|
||||
|
||||
return result, split_count
|
||||
|
||||
|
||||
def renumber_ids(chunks: list[dict]) -> list[dict]:
|
||||
"""
|
||||
Rinomina i chunk_id per evitare duplicati dopo merge/split.
|
||||
Mantiene il pattern base__sN dove N è il nuovo indice per sezione/titolo.
|
||||
"""
|
||||
seen: dict[str, int] = {}
|
||||
for c in chunks:
|
||||
base = re.sub(r"__s\d+$", "", c["chunk_id"])
|
||||
idx = seen.get(base, 0)
|
||||
c["chunk_id"] = f"{base}__s{idx}"
|
||||
c["sub_index"] = idx
|
||||
seen[base] = idx + 1
|
||||
return chunks
|
||||
|
||||
|
||||
# ─── Core ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
def fix_stem(stem: str, project_root: Path, max_chars: int, dry_run: bool) -> bool:
|
||||
step6_dir = project_root / "step-6" / stem
|
||||
chunks_path = step6_dir / "chunks.json"
|
||||
report_path = step6_dir / "report.json"
|
||||
|
||||
if not chunks_path.exists():
|
||||
print(f"✗ step-6/{stem}/chunks.json non trovato.")
|
||||
print(f" Esegui prima: python step-6/verify_chunks.py --stem {stem}")
|
||||
return False
|
||||
|
||||
if not report_path.exists():
|
||||
print(f"✗ step-6/{stem}/report.json non trovato.")
|
||||
print(f" Esegui prima: python step-6/verify_chunks.py --stem {stem}")
|
||||
return False
|
||||
|
||||
chunks: list[dict] = json.loads(chunks_path.read_text(encoding="utf-8"))
|
||||
report: dict = json.loads(report_path.read_text(encoding="utf-8"))
|
||||
|
||||
verdict = report.get("verdict", "ok")
|
||||
print(f"\nDocumento: {stem} (verdict: {verdict})")
|
||||
|
||||
if verdict == "ok":
|
||||
print(" ✅ Nessun problema — nulla da correggere.")
|
||||
return True
|
||||
|
||||
# Raccogli gli ID per categoria
|
||||
empty_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("empty", [])}
|
||||
no_prefix_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("no_prefix", [])}
|
||||
incomplete_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("incomplete", [])}
|
||||
too_short_ids = {e["chunk_id"] for e in report.get("warnings", {}).get("too_short", [])}
|
||||
too_long_ids = {e["chunk_id"] for e in report.get("warnings", {}).get("too_long", [])}
|
||||
|
||||
# Riepilogo operazioni pianificate
|
||||
ops: list[str] = []
|
||||
if empty_ids:
|
||||
ops.append(f" 🗑 rimuovi {len(empty_ids)} chunk vuoti")
|
||||
if no_prefix_ids:
|
||||
ops.append(f" 🔧 aggiungi prefisso a {len(no_prefix_ids)} chunk")
|
||||
if incomplete_ids:
|
||||
ops.append(f" 🔗 fondi {len(incomplete_ids)} chunk incompleti col successivo")
|
||||
if too_short_ids:
|
||||
ops.append(f" 🔗 fondi {len(too_short_ids)} chunk troppo corti col successivo")
|
||||
if too_long_ids:
|
||||
ops.append(f" ✂️ spezza {len(too_long_ids)} chunk troppo lunghi")
|
||||
|
||||
if not ops:
|
||||
print(" ✅ Nessuna correzione necessaria.")
|
||||
return True
|
||||
|
||||
print("\n Operazioni pianificate:")
|
||||
for op in ops:
|
||||
print(op)
|
||||
|
||||
if dry_run:
|
||||
print("\n [dry-run] Nessuna modifica applicata.")
|
||||
return True
|
||||
|
||||
n_before = len(chunks)
|
||||
|
||||
# Applica nell'ordine corretto
|
||||
if empty_ids:
|
||||
chunks, n = fix_empty(chunks, empty_ids)
|
||||
print(f"\n 🗑 Rimossi {n} chunk vuoti.")
|
||||
|
||||
if no_prefix_ids:
|
||||
chunks, n = fix_no_prefix(chunks, no_prefix_ids)
|
||||
print(f" 🔧 Aggiunto prefisso a {n} chunk.")
|
||||
|
||||
# incomplete prima di too_short (entrambi usano merge-forward)
|
||||
merge_ids = incomplete_ids | too_short_ids
|
||||
if merge_ids:
|
||||
chunks, n = fix_incomplete_and_short(chunks, merge_ids)
|
||||
print(f" 🔗 Fusi {n} chunk (incompleti + corti).")
|
||||
|
||||
if too_long_ids:
|
||||
# Aggiorna too_long_ids per chunk che potrebbero essere stati rinominati
|
||||
# dopo il merge (usa ancora gli ID originali, il merge non tocca too_long)
|
||||
chunks, n = fix_too_long(chunks, too_long_ids, max_chars)
|
||||
print(f" ✂️ Spezzati {n} chunk lunghi.")
|
||||
|
||||
# Rinumera per evitare duplicati
|
||||
chunks = renumber_ids(chunks)
|
||||
|
||||
n_after = len(chunks)
|
||||
print(f"\n Totale chunk: {n_before} → {n_after}")
|
||||
|
||||
# Salva
|
||||
chunks_path.write_text(
|
||||
json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8"
|
||||
)
|
||||
print(f" ✅ Salvato: step-6/{stem}/chunks.json")
|
||||
print(f"\n Riesegui la verifica:")
|
||||
print(f" python step-6/verify_chunks.py --stem {stem}")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ─── Entry point ──────────────────────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
project_root = Path(__file__).parent.parent
|
||||
|
||||
parser = argparse.ArgumentParser(description="Step 6 — Fix chunk")
|
||||
parser.add_argument("--stem", required=True, help="Nome del documento (sottocartella di step-6/)")
|
||||
parser.add_argument(
|
||||
"--max", type=int, default=MAX_CHARS,
|
||||
help=f"Soglia massima caratteri per lo split (default: {MAX_CHARS})"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run", action="store_true",
|
||||
help="Mostra le operazioni pianificate senza applicarle"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
ok = fix_stem(args.stem, project_root, args.max, args.dry_run)
|
||||
sys.exit(0 if ok else 1)
|
||||
@@ -0,0 +1,317 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Step 6 — Verifica chunk
|
||||
|
||||
Analizza step-5/<stem>/chunks.json e segnala ogni anomalia che potrebbe
|
||||
degradare la qualità del retrieval. Non modifica nulla: se ci sono problemi
|
||||
torna allo step 4 (revisione MD) o aggiusta i parametri dello step 5.
|
||||
|
||||
Input: step-5/<stem>/chunks.json
|
||||
Output: report a schermo + step-6/<stem>/report.json + exit code (0 = OK, 1 = problemi)
|
||||
|
||||
Uso:
|
||||
python step-6/verify_chunks.py --stem documento
|
||||
python step-6/verify_chunks.py # tutti i documenti in step-5/
|
||||
python step-6/verify_chunks.py --min 200 --max 800
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# ─── Soglie (devono coincidere con quelle usate in chunker.py) ────────────────
|
||||
|
||||
MIN_CHARS = 200
|
||||
MAX_CHARS = 800
|
||||
PUNCT_END = re.compile("[.!?»)\\]'\u2019\"\u201c\u201d\u2018\u2014\u2013-]$")
|
||||
|
||||
|
||||
# ─── Checks ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def has_prefix(chunk: dict) -> bool:
|
||||
"""Il chunk inizia con il prefisso di contesto '[...'."""
|
||||
return chunk.get("text", "").lstrip().startswith("[")
|
||||
|
||||
|
||||
def is_empty(chunk: dict) -> bool:
|
||||
return not chunk.get("text", "").strip()
|
||||
|
||||
|
||||
def is_too_short(chunk: dict, min_chars: int) -> bool:
|
||||
return chunk.get("n_chars", 0) < min_chars
|
||||
|
||||
|
||||
def is_too_long(chunk: dict, max_chars: int) -> bool:
|
||||
return chunk.get("n_chars", 0) > max_chars * 1.5
|
||||
|
||||
|
||||
def ends_incomplete(chunk: dict) -> bool:
|
||||
"""L'ultima riga di testo non termina con punteggiatura di fine frase."""
|
||||
text = chunk.get("text", "").rstrip()
|
||||
if not text:
|
||||
return False
|
||||
# Controlla l'ultimo carattere non-whitespace
|
||||
return not PUNCT_END.search(text)
|
||||
|
||||
|
||||
# ─── Report ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _fmt_chunk(c: dict) -> str:
|
||||
cid = c.get("chunk_id", "?")
|
||||
n = c.get("n_chars", 0)
|
||||
preview = c.get("text", "")[:60].replace("\n", " ")
|
||||
return f" [{cid}] ({n} char) «{preview}»"
|
||||
|
||||
|
||||
def verify_stem(stem: str, project_root: Path, min_chars: int, max_chars: int) -> bool:
|
||||
"""Verifica i chunk di un documento. Ritorna True se nessun bloccante rilevato."""
|
||||
import shutil
|
||||
|
||||
chunks_path = project_root / "step-6" / stem / "chunks.json"
|
||||
|
||||
print(f"\nDocumento: {stem}")
|
||||
|
||||
if not chunks_path.exists():
|
||||
src = project_root / "step-5" / stem / "chunks.json"
|
||||
if src.exists():
|
||||
chunks_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(src, chunks_path)
|
||||
print(f" → Copiato step-5/{stem}/chunks.json → step-6/{stem}/chunks.json")
|
||||
else:
|
||||
print(f" ✗ chunks.json non trovato in step-5/{stem}/ né in step-6/{stem}/ — skip")
|
||||
return False
|
||||
|
||||
chunks: list[dict] = json.loads(chunks_path.read_text(encoding="utf-8"))
|
||||
|
||||
if not chunks:
|
||||
print(f" ✗ chunks.json è vuoto")
|
||||
return False
|
||||
|
||||
# ── Raccogli problemi ──────────────────────────────────────────────────────
|
||||
|
||||
empty_chunks = [c for c in chunks if is_empty(c)]
|
||||
no_prefix = [c for c in chunks if not is_empty(c) and not has_prefix(c)]
|
||||
too_short = [c for c in chunks if is_too_short(c, min_chars)]
|
||||
too_long = [c for c in chunks if is_too_long(c, max_chars)]
|
||||
incomplete = [c for c in chunks if not is_empty(c) and ends_incomplete(c)]
|
||||
|
||||
# ── Statistiche lunghezze ──────────────────────────────────────────────────
|
||||
|
||||
lengths = [c.get("n_chars", 0) for c in chunks]
|
||||
n_total = len(chunks)
|
||||
n_ok = n_total - len(set(
|
||||
c["chunk_id"] for lst in [empty_chunks, no_prefix, too_short, too_long, incomplete]
|
||||
for c in lst
|
||||
))
|
||||
min_l = min(lengths)
|
||||
max_l = max(lengths)
|
||||
avg_l = int(sum(lengths) / n_total)
|
||||
|
||||
# Distribuzione in fasce
|
||||
n_under = sum(1 for l in lengths if l < min_chars)
|
||||
n_normal = sum(1 for l in lengths if min_chars <= l <= max_chars)
|
||||
n_over = sum(1 for l in lengths if l > max_chars)
|
||||
|
||||
# ── Output ────────────────────────────────────────────────────────────────
|
||||
|
||||
print(f" Totale chunk: {n_total}")
|
||||
print(f" ✅ OK: {n_ok}")
|
||||
print()
|
||||
print(f" Distribuzione lunghezze:")
|
||||
print(f" Min: {min_l} char")
|
||||
print(f" Max: {max_l} char")
|
||||
print(f" Media: {avg_l} char")
|
||||
print(f" < {min_chars} char (sotto MIN): {n_under}")
|
||||
print(f" {min_chars}–{max_chars} char (ideale): {n_normal}")
|
||||
print(f" > {max_chars} char (sopra MAX): {n_over}")
|
||||
|
||||
has_errors = False
|
||||
|
||||
if empty_chunks:
|
||||
has_errors = True
|
||||
print(f"\n 🔴 {len(empty_chunks)} chunk VUOTI:")
|
||||
for c in empty_chunks[:5]:
|
||||
print(f" [{c.get('chunk_id', '?')}]")
|
||||
if len(empty_chunks) > 5:
|
||||
print(f" ... e altri {len(empty_chunks) - 5}")
|
||||
|
||||
if no_prefix:
|
||||
has_errors = True
|
||||
print(f"\n 🔴 {len(no_prefix)} chunk SENZA PREFISSO DI CONTESTO:")
|
||||
for c in no_prefix[:5]:
|
||||
print(_fmt_chunk(c))
|
||||
if len(no_prefix) > 5:
|
||||
print(f" ... e altri {len(no_prefix) - 5}")
|
||||
print(f" → Causa probabile: header ### mancanti o malformati nel MD (step 4)")
|
||||
|
||||
if too_short:
|
||||
has_errors = True
|
||||
print(f"\n 🟡 {len(too_short)} chunk SOTTO MIN_CHARS ({min_chars}):")
|
||||
for c in too_short[:5]:
|
||||
print(_fmt_chunk(c))
|
||||
if len(too_short) > 5:
|
||||
print(f" ... e altri {len(too_short) - 5}")
|
||||
print(f" → Causa probabile: sezione isolata senza successivo nello stesso ##")
|
||||
print(f" → Soluzione: abbassa MIN_CHARS o revisiona il MD (step 4)")
|
||||
|
||||
if too_long:
|
||||
has_errors = True
|
||||
print(f"\n 🟡 {len(too_long)} chunk SOPRA MAX_CHARS×1.5 ({int(max_chars * 1.5)}):")
|
||||
for c in too_long[:5]:
|
||||
print(_fmt_chunk(c))
|
||||
if len(too_long) > 5:
|
||||
print(f" ... e altri {len(too_long) - 5}")
|
||||
print(f" → Causa probabile: frase singola molto lunga non spezzabile")
|
||||
print(f" → Soluzione: alza MAX_CHARS o verifica il testo nel MD (step 4)")
|
||||
|
||||
if incomplete:
|
||||
has_errors = True
|
||||
print(f"\n 🔴 {len(incomplete)} chunk CHE FINISCONO SENZA PUNTEGGIATURA (frase spezzata):")
|
||||
for c in incomplete[:5]:
|
||||
last_line = c.get("text", "").rstrip().split("\n")[-1][-80:]
|
||||
print(f" [{c.get('chunk_id', '?')}] ...{last_line!r}")
|
||||
if len(incomplete) > 5:
|
||||
print(f" ... e altri {len(incomplete) - 5}")
|
||||
print(f" → Causa probabile: paragrafo spezzato nel MD (step 4)")
|
||||
print(f" → Soluzione: correggi le righe spezzate in step-4/{stem}/clean.md")
|
||||
|
||||
# ── Costruisci e salva report.json ───────────────────────────────────────
|
||||
|
||||
blockers = empty_chunks + no_prefix + incomplete
|
||||
warnings = too_short + too_long
|
||||
|
||||
def _chunk_entry(c: dict) -> dict:
|
||||
return {
|
||||
"chunk_id": c.get("chunk_id", ""),
|
||||
"sezione": c.get("sezione", ""),
|
||||
"titolo": c.get("titolo", ""),
|
||||
"n_chars": c.get("n_chars", 0),
|
||||
"last_text": c.get("text", "").rstrip().split("\n")[-1][-120:],
|
||||
}
|
||||
|
||||
if not blockers:
|
||||
verdict = "ok" if not warnings else "warnings_only"
|
||||
else:
|
||||
verdict = "blocked"
|
||||
|
||||
report = {
|
||||
"stem": stem,
|
||||
"verdict": verdict,
|
||||
"stats": {
|
||||
"total": n_total,
|
||||
"ok": n_ok,
|
||||
"min_chars": min_l,
|
||||
"max_chars": max_l,
|
||||
"avg_chars": avg_l,
|
||||
},
|
||||
"thresholds": {"min_chars": min_chars, "max_chars": max_chars},
|
||||
"blockers": {
|
||||
"empty": [_chunk_entry(c) for c in empty_chunks],
|
||||
"no_prefix": [_chunk_entry(c) for c in no_prefix],
|
||||
"incomplete": [_chunk_entry(c) for c in incomplete],
|
||||
},
|
||||
"warnings": {
|
||||
"too_short": [_chunk_entry(c) for c in too_short],
|
||||
"too_long": [_chunk_entry(c) for c in too_long],
|
||||
},
|
||||
}
|
||||
|
||||
out_dir = project_root / "step-6" / stem
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
(out_dir / "report.json").write_text(
|
||||
json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
|
||||
)
|
||||
print(f"\n report.json salvato in step-6/{stem}/")
|
||||
|
||||
# ── Prossimi passi ────────────────────────────────────────────────────────
|
||||
|
||||
print(f"\n {'─' * 50}")
|
||||
print(f" PROSSIMI PASSI")
|
||||
print(f" {'─' * 50}")
|
||||
|
||||
if not blockers and not warnings:
|
||||
print(f" ✅ Tutto OK — procedi alla vettorizzazione:")
|
||||
print(f" python step-8/ingest.py --stem {stem}")
|
||||
|
||||
elif not blockers:
|
||||
# Solo 🟡: si può procedere, i warning sono facoltativi
|
||||
print(f" 🟡 Solo avvisi minori — puoi procedere alla vettorizzazione:")
|
||||
print(f" python step-8/ingest.py --stem {stem}")
|
||||
print()
|
||||
print(f" Oppure, per ottimizzare prima di procedere:")
|
||||
if too_short:
|
||||
pct = int(len(too_short) / n_total * 100)
|
||||
print(f" • {len(too_short)} chunk corti ({pct}% del totale)")
|
||||
if too_long:
|
||||
pct = int(len(too_long) / n_total * 100)
|
||||
print(f" • {len(too_long)} chunk lunghi ({pct}% del totale)")
|
||||
if too_short or too_long:
|
||||
print(f" → Esegui: python step-6/fix_chunks.py --stem {stem} --dry-run")
|
||||
print(f" poi: python step-6/fix_chunks.py --stem {stem}")
|
||||
print(f" poi: python step-6/verify_chunks.py --stem {stem}")
|
||||
|
||||
else:
|
||||
# Ci sono 🔴: non si procede
|
||||
print(f" 🔴 Problemi bloccanti — correggi prima di procedere:")
|
||||
print()
|
||||
if empty_chunks:
|
||||
print(f" • {len(empty_chunks)} chunk vuoti")
|
||||
print(f" → Controlla step-4/{stem}/clean.md per sezioni prive di testo")
|
||||
if no_prefix:
|
||||
print(f" • {len(no_prefix)} chunk senza prefisso di contesto")
|
||||
print(f" → Controlla che gli header ### siano corretti in step-4/{stem}/clean.md")
|
||||
if incomplete:
|
||||
print(f" • {len(incomplete)} chunk con frase spezzata")
|
||||
print(f" → Esegui: python step-6/fix_chunks.py --stem {stem}")
|
||||
print()
|
||||
print(f" Dopo le correzioni, riesegui nell'ordine:")
|
||||
print(f" python step-5/chunker.py --stem {stem} --force")
|
||||
print(f" python step-6/verify_chunks.py --stem {stem}")
|
||||
print()
|
||||
if warnings:
|
||||
print(f" 🟡 Hai anche {len(warnings)} avvisi minori — affrontali dopo aver risolto i 🔴.")
|
||||
|
||||
return not blockers
|
||||
|
||||
|
||||
# ─── Entry point ──────────────────────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
project_root = Path(__file__).parent.parent
|
||||
|
||||
parser = argparse.ArgumentParser(description="Step 6 — Verifica chunk")
|
||||
parser.add_argument("--stem", help="Nome del documento (sottocartella di step-5/)")
|
||||
parser.add_argument(
|
||||
"--min", type=int, default=MIN_CHARS,
|
||||
help=f"Soglia minima caratteri (default: {MIN_CHARS})"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max", type=int, default=MAX_CHARS,
|
||||
help=f"Soglia massima caratteri (default: {MAX_CHARS})"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.stem:
|
||||
stems = [args.stem]
|
||||
else:
|
||||
step5_dir = project_root / "step-5"
|
||||
if not step5_dir.exists():
|
||||
print(f"Errore: cartella step-5/ non trovata in {project_root}")
|
||||
sys.exit(1)
|
||||
stems = sorted(
|
||||
p.name for p in step5_dir.iterdir()
|
||||
if p.is_dir() and (p / "chunks.json").exists()
|
||||
)
|
||||
if not stems:
|
||||
print("Errore: nessun chunks.json trovato in step-5/")
|
||||
sys.exit(1)
|
||||
|
||||
results = [verify_stem(s, project_root, args.min, args.max) for s in stems]
|
||||
|
||||
ok = sum(results)
|
||||
total = len(results)
|
||||
print(f"\n{'✅' if all(results) else '⚠️ '} {ok}/{total} documenti senza problemi")
|
||||
sys.exit(0 if all(results) else 1)
|
||||
Reference in New Issue
Block a user