#!/usr/bin/env python3 """ Step 6 — Fix chunk Applica correzioni dirette su step-6//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//chunks.json + step-6//report.json Output: step-6//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)