#!/usr/bin/env python3 """Lleva a D1 la FRASE ENTERA cuando el ancla por contexto no agarra. De donde salen estos casos. Son correcciones ya verificadas contra la imagen del impreso y escritas en el repo, pero que no se pudieron sincronizar: el texto de D1 alrededor del cambio difiere lo suficiente como para que ningun ancla —ni de nueve palabras ni de una— aparezca exactamente una vez. Son 51. La diferencia con `empujar_pendientes_d1.py` es el tamano de lo que se mueve: alli se sustituye un tramo anclado en su contexto; aqui se sustituye la FRASE, localizada por parecido. Eso arrastra necesariamente las demas diferencias que D1 tuviera dentro de esa misma frase, y por eso: - se exige un parecido alto (>= 0.60) para no confundir de frase; - se LISTA palabra por palabra lo que cambia de mas, para que quede por escrito que se ha metido en el mismo viaje; - se rechaza si la frase de D1 no aparece exactamente una vez. El repo es la fuente de verdad —corregido contra el impreso— y D1 es la copia que sirve vedabase.cc; cuando difieren, manda el repo. Pero «manda» no puede significar «sin mirar»: de ahi el listado de colaterales. Uso: python3 empujar_frase_d1.py [--aplicar] [--min 0.60] """ import json, re, os, sys, difflib, subprocess, unicodedata, pathlib, collections HERE = pathlib.Path(__file__).resolve().parent ASTRO = pathlib.Path('/Users/jaganat/git_projects/astro_vedabase') REPO = pathlib.Path('/Users/jaganat/git_projects/vedabase-original') sys.path.insert(0, str(HERE)) from sincronizar_diff import d1, q from verificar_paridad import ref_de def sd(s): s = unicodedata.normalize('NFD', s or '') s = ''.join(c for c in s if unicodedata.category(c) != 'Mn') return ' '.join(re.sub(r'[^A-Za-z0-9]+', ' ', s.lower()).split()) def frases(texto): """Trozos del texto cortados en frontera de FRASE, con sus offsets.""" out = []; ini = 0 for m in re.finditer(r'[.!?]["»”\')\]*_]*\s+|\n\n', texto): out.append((ini, m.end())); ini = m.end() if ini < len(texto): out.append((ini, len(texto))) return out def mejor_ventana(nuevo, texto, minimo=0.60): """El tramo de `texto` mas parecido a `nuevo`, cortado en frontera de frase. La primera version deslizaba una ventana del mismo NUMERO DE PALABRAS, y eso la dejaba empezar y acabar a media frase: los «colaterales» que reportaba («cubits.» -> «», «» -> «one») eran bordes desalineados, no diferencias. Cortando por frases, lo que se sustituye son unidades completas y el diff de dentro si es real. Se prueban runs de 1 a 4 frases consecutivas, que es lo que puede abarcar una correccion.""" fr = frases(texto) if not fr: return None objetivo = sd(nuevo) mejor = None for i in range(len(fr)): for k in range(1, 5): if i + k > len(fr): break a, b = fr[i][0], fr[i + k - 1][1] cand = texto[a:b] # descartar de entrada lo que no puede parecerse por tamano if not (0.5 <= len(sd(cand)) / max(1, len(objetivo)) <= 2.0): continue r = difflib.SequenceMatcher(None, objetivo, sd(cand)).ratio() if not mejor or r > mejor[3]: mejor = (a, b, cand, r) if not mejor or mejor[3] < minimo: return None return mejor def main(): aplicar = '--aplicar' in sys.argv minimo = float(sys.argv[sys.argv.index('--min') + 1]) if '--min' in sys.argv else 0.60 casos = json.load(open(HERE / 'divergencias_reales.json', encoding='utf-8')) casos = [c for c in casos if c.get('nuevo')] print(f'divergencias con texto nuevo verificado: {len(casos)}') # LA FRASE ENTERA DEL REPO, no el fragmento. Los `nuevo` que guardamos son # tramos del diff («causes one to dance»), no oraciones. Sustituir una frase # completa de D1 por un tramo la truncaria. Aqui se busca el fragmento en el # fichero del repo y se extiende a la frase que lo contiene. from verificar_paridad import ref_de as _r inv = {} for f in subprocess.run(['git', 'ls-files'], capture_output=True, text=True, cwd=str(REPO)).stdout.splitlines(): if f.endswith('.md') and not f.startswith('translations/'): rr = _r(f) if rr: inv.setdefault(rr, f) res = collections.Counter(); colaterales = [] porref = collections.defaultdict(list) for c in casos: f = inv.get(c['ref']) frag = c['nuevo'] entera = frag if f and (REPO / f).exists(): t = (REPO / f).read_text(encoding='utf-8') i = t.find(frag) if i >= 0: for a, b in frases(t): if a <= i < b: # extender mientras el fragmento siga sin caber entero fin = b for a2, b2 in frases(t): if a2 >= b and a2 < i + len(frag): fin = b2 entera = t[a:fin].strip() break if entera != frag: res['fragmento extendido a la frase del repo'] += 1 porref[c['ref']].append(entera) for ref, nuevos in sorted(porref.items()): filas = d1(f"SELECT COALESCE(purport,'') p, COALESCE(translation,'') t, " f"COALESCE(synonyms,'') s FROM verses WHERE ref={q(ref)} AND lang='en'", js=True) if not filas: res['sin fila en D1'] += len(nuevos); continue r0 = filas[0] stmts = [] for nuevo in nuevos: if sd(nuevo) in sd(r0['p'] + ' ' + r0['t'] + ' ' + r0['s']): res['ya estaba en D1'] += 1; continue elegido = None for campo in ('purport', 'translation', 'synonyms'): val = r0[{'purport': 'p', 'translation': 't', 'synonyms': 's'}[campo]] m = mejor_ventana(nuevo, val, minimo) if not m: continue if not elegido or m[3] > elegido[3]: elegido = (campo, m[0], m[1], m[3], val, m[2]) if not elegido: res[f'sin frase parecida (>= {minimo})'] += 1; continue campo, a, b, par, val, viejo = elegido viejo = viejo.rstrip() if val.count(viejo) != 1: res['la frase de D1 aparece mas de una vez'] += 1; continue # que se lleva por delante, ademas de lo que ibamos a cambiar sm = difflib.SequenceMatcher(None, viejo.split(), nuevo.split()) dif = [(viejo.split()[i1:i2], nuevo.split()[j1:j2]) for op, i1, i2, j1, j2 in sm.get_opcodes() if op != 'equal'] colaterales.append({'ref': ref, 'parecido': round(par, 3), 'd1': viejo[:220], 'repo': nuevo[:220], 'bloques': [[' '.join(x), ' '.join(y)] for x, y in dif]}) stmts.append(f"UPDATE verses SET {campo}=REPLACE({campo},{q(viejo)},{q(nuevo)}) " f"WHERE ref={q(ref)} AND lang='en';") res['frase sustituida'] += 1 if not stmts: continue print(f' {ref}: {len(stmts)}') if aplicar: sql = HERE / '_frase.sql' sql.write_text('\n'.join(stmts) + '\n', encoding='utf-8') p = subprocess.run(['npx', 'wrangler', 'd1', 'execute', 'vedabase-search-db', '--remote', '--file', str(sql)], capture_output=True, text=True, cwd=str(ASTRO), env={**os.environ, 'CF_API_TOKEN': '', 'CLOUDFLARE_API_TOKEN': ''}) sql.unlink(missing_ok=True) if 'rows_written' not in p.stdout: res['FALLO en D1'] += 1; print(' FALLO') print() for k, v in res.most_common(): print(f' {v:5} {k}') json.dump(colaterales, open(HERE / 'colaterales_d1.json', 'w'), ensure_ascii=False, indent=1) extra = sum(max(0, len(c['bloques']) - 1) for c in colaterales) print(f'\ncambios colaterales que viajan en las frases: {extra} ' f'(detalle en colaterales_d1.json)') for c in colaterales[:6]: if len(c['bloques']) > 1: print(f" {c['ref']} (parecido {c['parecido']}):") for x, y in c['bloques'][:3]: print(f" D1 {x[:44]!r} -> repo {y[:44]!r}") if not aplicar: print('\n(ensayo: anade --aplicar)') return 0 if __name__ == '__main__': sys.exit(main())