""" cultiva_doc_cache.py

Pipeline de extracción de documentos para CULTIVA IA con caché por hash de contenido.

Problema real:

• 300 docs/semana → ~40% duplicados
• Coste extracción: ~0,03 €/pág + 2-8 s/fichero (GPT-4o + OCR)
• Ahorro estimado: ~18 €/semana + ~15 min CPU sólo evitando duplicados

Patrón aplicado: SHA-256(contenido binario) → .cache/.json • Path-independiente: renombrar / mover = HIT • Auto-invalida: cambio de contenido = MISS automático • SRP: la función pura extract_text() no sabe nada de caché • Corrupción → MISS, nunca crash

Uso: python resultado.py docs/ --cache-dir .cache python resultado.py docs/ --no-cache """

from future import annotations

import hashlib import json import logging import time from dataclasses import dataclass from pathlib import Path from typing import Any

──────────────────────────────────────────────────────────────────────────────

Logging

──────────────────────────────────────────────────────────────────────────────

logging.basicConfig( format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", level=logging.INFO, ) logger = logging.getLogger("cultiva.doc_cache")

──────────────────────────────────────────────────────────────────────────────

1. Hash de contenido (no de ruta)

──────────────────────────────────────────────────────────────────────────────

_HASH_CHUNK_SIZE = 65_536 # 64 KB por chunk — seguro para PDFs grandes

def compute_file_hash(path: Path) -> str: """SHA-256 del contenido binario del fichero, en chunks de 64 KB.

Path-independiente: el mismo contenido bajo distinto nombre/ruta
devuelve el mismo hash → cache hit garantizado.
"""
if not path.is_file():
    raise FileNotFoundError(f"Fichero no encontrado: {path}")
sha256 = hashlib.sha256()
with open(path, "rb") as fh:
    while chunk := fh.read(_HASH_CHUNK_SIZE):
        sha256.update(chunk)
return sha256.hexdigest()

──────────────────────────────────────────────────────────────────────────────

2. Dataclass inmutable para el resultado extraído

──────────────────────────────────────────────────────────────────────────────

@dataclass(frozen=True, slots=True) class ExtractedDocument: """Resultado de la extracción de un documento.""" pages: int word_count: int language: str summary: str entities: tuple[str, ...] # nombres, empresas, fechas detectadas extraction_cost_eur: float # coste real incurrido en la extracción extracted_at: str # ISO-8601

@dataclass(frozen=True, slots=True) class CacheEntry: """Entrada de caché: une el hash del fichero con su resultado.""" file_hash: str source_path: str # ruta original (informativa, no clave) document: ExtractedDocument

──────────────────────────────────────────────────────────────────────────────

3. Serialización manual (evita problemas con nested frozen dataclasses)

──────────────────────────────────────────────────────────────────────────────

def _serialize_entry(entry: CacheEntry) -> dict[str, Any]: doc = entry.document return { "file_hash": entry.file_hash, "source_path": entry.source_path, "document": { "pages": doc.pages, "word_count": doc.word_count, "language": doc.language, "summary": doc.summary, "entities": list(doc.entities), "extraction_cost_eur": doc.extraction_cost_eur, "extracted_at": doc.extracted_at, }, }

def _deserialize_entry(data: dict[str, Any]) -> CacheEntry: doc_data = data["document"] doc = ExtractedDocument( pages=int(doc_data["pages"]), word_count=int(doc_data["word_count"]), language=str(doc_data["language"]), summary=str(doc_data["summary"]), entities=tuple(doc_data.get("entities", [])), extraction_cost_eur=float(doc_data["extraction_cost_eur"]), extracted_at=str(doc_data["extracted_at"]), ) return CacheEntry( file_hash=str(data["file_hash"]), source_path=str(data["source_path"]), document=doc, )

──────────────────────────────────────────────────────────────────────────────

4. Lectura / escritura de caché (nombrado {hash}.json → O(1) lookup)

──────────────────────────────────────────────────────────────────────────────

def write_cache(cache_dir: Path, entry: CacheEntry) -> None: """Persiste una entrada en /.json.""" cache_dir.mkdir(parents=True, exist_ok=True) cache_file = cache_dir / f"{entry.file_hash}.json" payload = json.dumps(_serialize_entry(entry), ensure_ascii=False, indent=2) cache_file.write_text(payload, encoding="utf-8")

def read_cache(cache_dir: Path, file_hash: str) -> CacheEntry | None: """Devuelve la entrada cacheada o None (miss / corrupción → miss, nunca crash).""" cache_file = cache_dir / f"{file_hash}.json" if not cache_file.is_file(): return None try: raw = cache_file.read_text(encoding="utf-8") data = json.loads(raw) return _deserialize_entry(data) except (json.JSONDecodeError, ValueError, KeyError, TypeError): logger.warning("Caché corrupta para hash %s — se tratará como miss.", file_hash[:12]) return None # Graceful degradation

──────────────────────────────────────────────────────────────────────────────

5. Función PURA de extracción (no sabe nada de caché — SRP)

──────────────────────────────────────────────────────────────────────────────

def extract_text(file_path: Path) -> ExtractedDocument: """Simula extracción costosa con GPT-4o + OCR.

En producción llamaría a:
  • pdfminer / pymupdf para texto plano
  • Azure Document Intelligence para OCR
  • GPT-4o para resumen + entidades

Aquí simulamos el coste y tiempo reales.
"""
import random
from datetime import datetime, timezone

random.seed(int.from_bytes(file_path.name.encode(), "little"))

pages = random.randint(2, 40)
cost_per_page = 0.03
proc_seconds = pages * 0.25          # 250 ms / página simulados
time.sleep(min(proc_seconds, 0.05))  # reducido para demo

return ExtractedDocument(
    pages=pages,
    word_count=pages * random.randint(280, 420),
    language="es" if random.random() > 0.25 else "en",
    summary=(
        f"Documento de {pages} páginas procesado vía OCR+GPT-4o. "
        "Contiene cláusulas contractuales, anexos técnicos y firma digital."
    ),
    entities=tuple(random.sample(
        ["CULTIVA IA SLU", "Álvaro Gimeno", "Contrato Marco 2026",
         "Anexo A", "Madrid", "CIF B12345678", "enero 2026"],
        k=random.randint(3, 5),
    )),
    extraction_cost_eur=round(pages * cost_per_page, 4),
    extracted_at=datetime.now(timezone.utc).isoformat(),
)

──────────────────────────────────────────────────────────────────────────────

6. Capa de servicio: caché + extracción (SRP)

──────────────────────────────────────────────────────────────────────────────

def extract_with_cache( file_path: Path, *, cache_enabled: bool = True, cache_dir: Path = Path(".cache"), ) -> tuple[ExtractedDocument, bool]: """Service layer: hash → lookup → (hit | extracción + escritura).

Returns:
    (ExtractedDocument, cache_hit: bool)
"""
if not cache_enabled:
    logger.info("Caché desactivada — extrayendo: %s", file_path.name)
    return extract_text(file_path), False

file_hash = compute_file_hash(file_path)

cached = read_cache(cache_dir, file_hash)
if cached is not None:
    logger.info(
        "✅ HIT   %s  (hash=%s…)", file_path.name, file_hash[:12]
    )
    return cached.document, True

logger.info(
    "🔄 MISS  %s  (hash=%s…) — extrayendo…", file_path.name, file_hash[:12]
)
doc = extract_text(file_path)
entry = CacheEntry(
    file_hash=file_hash,
    source_path=str(file_path),
    document=doc,
)
write_cache(cache_dir, entry)
return doc, False

──────────────────────────────────────────────────────────────────────────────

7. CLI

──────────────────────────────────────────────────────────────────────────────

def _run_batch( docs_dir: Path, cache_dir: Path, cache_enabled: bool, ) -> None: """Procesa todos los ficheros de docs_dir e imprime estadísticas.""" files = sorted(docs_dir.glob("*")) if not files: logger.warning("No se encontraron ficheros en %s", docs_dir) return

hits = misses = 0
total_cost_saved = 0.0
total_cost_paid = 0.0
t0 = time.perf_counter()

for fp in files:
    if not fp.is_file():
        continue
    doc, was_hit = extract_with_cache(fp, cache_enabled=cache_enabled, cache_dir=cache_dir)
    if was_hit:
        hits += 1
        total_cost_saved += doc.extraction_cost_eur
    else:
        misses += 1
        total_cost_paid += doc.extraction_cost_eur

elapsed = time.perf_counter() - t0
total = hits + misses

print("\n" + "=" * 58)
print("  CULTIVA IA — Resumen de pipeline doc-cache")
print("=" * 58)
print(f"  Ficheros procesados : {total:>4}")
print(f"  Cache HITs          : {hits:>4}  ({hits/total*100:.0f}%)")
print(f"  Cache MISSes        : {misses:>4}  ({misses/total*100:.0f}%)")
print(f"  Coste ahorrado      : {total_cost_saved:>7.4f} €")
print(f"  Coste incurrido     : {total_cost_paid:>7.4f} €")
print(f"  Tiempo total        : {elapsed:>6.2f} s")
print(f"  Directorio caché    : {cache_dir}")
print("=" * 58 + "\n")

def main() -> None: import argparse

parser = argparse.ArgumentParser(
    description="CULTIVA IA — Pipeline de extracción con caché por hash",
)
parser.add_argument(
    "docs_dir",
    type=Path,
    help="Directorio con documentos a procesar",
)
parser.add_argument(
    "--cache-dir",
    type=Path,
    default=Path(".cache"),
    help="Directorio de caché (default: .cache)",
)
parser.add_argument(
    "--no-cache",
    action="store_true",
    help="Desactiva la caché (fuerza reextracción)",
)
args = parser.parse_args()
_run_batch(
    docs_dir=args.docs_dir,
    cache_dir=args.cache_dir,
    cache_enabled=not args.no_cache,
)

if name == "main": main()