Spaces:
Running
Running
File size: 4,681 Bytes
9abe007 456bf68 9abe007 456bf68 9abe007 456bf68 9abe007 456bf68 9abe007 456bf68 9abe007 456bf68 9abe007 456bf68 9abe007 456bf68 9abe007 aad1d52 9abe007 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """
corpus_loader.py
────────────────
Carga pares (HTR, groundtruth) desde disco.
Formatos soportados:
1. JSON → lista de objetos {"id","htr","gt","type","region","date"}
2. CSV → columnas: id, htr, gt, type, region, date
3. TXT → carpeta con ficheros *.htr.txt y *.gt.txt (mismo nombre base)
Uso:
from corpus_loader import CorpusLoader
loader = CorpusLoader("./corpus")
pairs = loader.load() # lista de dicts
"""
import json
import csv
import os
from pathlib import Path
from typing import List, Dict
class CorpusLoader:
def __init__(self, corpus_path: str):
self.corpus_path = Path(corpus_path)
def load(self) -> List[Dict]:
"""Detecta formato y carga todos los pares disponibles."""
pairs = []
if not self.corpus_path.exists():
print(f"⚠ Corpus path '{self.corpus_path}' no existe. Usando pares de ejemplo.")
return []
# ── JSON ──────────────────────────────────────────────────────────────
for f in sorted(self.corpus_path.glob("*.json")):
try:
with open(f, encoding="utf-8") as fh:
data = json.load(fh)
if isinstance(data, list):
pairs.extend(self._normalize(data, source=f.stem))
elif isinstance(data, dict): # un solo documento
pairs.append(self._normalize_one(data, source=f.stem))
print(f" JSON cargado: {f.name} ({len(data)} pares)")
except Exception as e:
print(f" Error leyendo {f.name}: {e}")
# ── CSV ───────────────────────────────────────────────────────────────
for f in sorted(self.corpus_path.glob("*.csv")):
try:
with open(f, encoding="utf-8", newline="") as fh:
reader = csv.DictReader(fh)
rows = list(reader)
pairs.extend(self._normalize(rows, source=f.stem))
print(f" CSV cargado: {f.name} ({len(rows)} pares)")
except Exception as e:
print(f" Error leyendo {f.name}: {e}")
# ── TXT pareado ───────────────────────────────────────────────────────
htr_files = sorted(self.corpus_path.glob("*.htr.txt"))
for htr_file in htr_files:
gt_file = htr_file.with_suffix("").with_suffix(".gt.txt")
if not gt_file.exists():
print(f" Sin GT para {htr_file.name}, omitido.")
continue
try:
htr_text = htr_file.read_text(encoding="utf-8").strip()
gt_text = gt_file.read_text(encoding="utf-8").strip()
pairs.append({
"id": htr_file.stem.replace(".htr", ""),
"htr": htr_text,
"gt": gt_text,
"type": "desconocido",
"region": "desconocida",
"date": "",
"source": "txt",
})
except Exception as e:
print(f" Error leyendo {htr_file.name}: {e}")
if htr_files:
print(f" TXT cargados: {len(htr_files)} pares")
print(f"\n Total pares cargados desde disco: {len(pairs)}")
return pairs
# ── helpers ───────────────────────────────────────────────────────────────
def _normalize(self, rows: List[Dict], source: str) -> List[Dict]:
return [self._normalize_one(r, source) for r in rows if r.get("htr") and r.get("gt")]
def _normalize_one(self, row: Dict, source: str) -> Dict:
return {
"id": str(row.get("id", source)),
"htr": str(row.get("htr", "")).strip(),
"gt": str(row.get("gt", "")).strip(),
"type": str(row.get("type", "desconocido")),
"region": str(row.get("region", "desconocida")),
"date": str(row.get("date", "")),
"caligrafia": str(row.get("caligrafia", "desconocida")),
"source": source,
"corrections": row.get("corrections", []),
}
|