414 lines
12 KiB
Python
414 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
import argparse,json,tempfile
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from zipfile import ZIP_DEFLATED,ZIP_STORED,ZipFile,ZipInfo
|
|
from lxml import etree
|
|
|
|
def replace_once(node,old,new):
|
|
hits=[]
|
|
for e in node.iter():
|
|
if e.text and old in e.text:hits.append((e,'text'))
|
|
if e.tail and old in e.tail:hits.append((e,'tail'))
|
|
if len(hits)!=1:raise RuntimeError(f'Attesa 1 occorrenza di {old!r}, trovate {len(hits)}')
|
|
e,field=hits[0];value=getattr(e,field)
|
|
if value.count(old)!=1:raise RuntimeError(f'{old!r} compare piu volte nello stesso nodo')
|
|
setattr(e,field,value.replace(old,new,1))
|
|
|
|
|
|
def minimal_replacement(before, after):
|
|
"""
|
|
Ricava la singola porzione modificata tra due testi.
|
|
|
|
Esempio:
|
|
before = "Puoi dido forte."
|
|
after = "Puoi dirlo forte."
|
|
|
|
Restituisce:
|
|
("dido", "dirlo")
|
|
|
|
La sostituzione può essere applicata a un nodo
|
|
testuale senza eliminare il markup XHTML.
|
|
"""
|
|
|
|
if before == after:
|
|
raise RuntimeError(
|
|
"La riscrittura non modifica il testo"
|
|
)
|
|
|
|
prefix = 0
|
|
limit = min(len(before), len(after))
|
|
|
|
while (
|
|
prefix < limit
|
|
and before[prefix] == after[prefix]
|
|
):
|
|
prefix += 1
|
|
|
|
suffix = 0
|
|
|
|
while (
|
|
suffix < len(before) - prefix
|
|
and suffix < len(after) - prefix
|
|
and before[-1 - suffix]
|
|
== after[-1 - suffix]
|
|
):
|
|
suffix += 1
|
|
|
|
if suffix:
|
|
old_part = before[prefix:-suffix]
|
|
new_part = after[prefix:-suffix]
|
|
else:
|
|
old_part = before[prefix:]
|
|
new_part = after[prefix:]
|
|
|
|
if not old_part:
|
|
raise RuntimeError(
|
|
"La riscrittura richiede un inserimento "
|
|
"puro, non gestibile in sicurezza"
|
|
)
|
|
|
|
return old_part, new_part
|
|
|
|
|
|
def iter_text_fields(node):
|
|
"""
|
|
Restituisce i campi text e tail nell'ordine
|
|
in cui appaiono nel testo XHTML.
|
|
"""
|
|
|
|
for element in node.iter():
|
|
if element.text:
|
|
yield element, "text", element.text
|
|
|
|
for child in element:
|
|
if child.tail:
|
|
yield child, "tail", child.tail
|
|
|
|
|
|
def build_normalized_mapping(node):
|
|
"""
|
|
Costruisce lo stesso testo normalizzato usato
|
|
durante l'analisi:
|
|
|
|
re.sub(r"\s+", " ", text).strip()
|
|
|
|
Per ogni carattere normalizzato conserva il
|
|
riferimento al campo XHTML e alla posizione
|
|
originale.
|
|
"""
|
|
|
|
fields = list(iter_text_fields(node))
|
|
|
|
normalized = []
|
|
mapping = []
|
|
|
|
started = False
|
|
pending_space = None
|
|
|
|
for element, field, value in fields:
|
|
for position, character in enumerate(value):
|
|
if character.isspace():
|
|
if started and pending_space is None:
|
|
pending_space = (
|
|
element,
|
|
field,
|
|
position,
|
|
)
|
|
|
|
continue
|
|
|
|
if pending_space is not None:
|
|
normalized.append(" ")
|
|
mapping.append(pending_space)
|
|
pending_space = None
|
|
|
|
normalized.append(character)
|
|
mapping.append((
|
|
element,
|
|
field,
|
|
position,
|
|
))
|
|
|
|
started = True
|
|
|
|
return "".join(normalized), mapping
|
|
|
|
|
|
def replace_at_normalized_offset(
|
|
node,
|
|
offset,
|
|
original,
|
|
replacement,
|
|
):
|
|
"""
|
|
Sostituisce l'occorrenza che in LanguageTool
|
|
inizia esattamente a offset.
|
|
|
|
La modifica è consentita solo se tutti i
|
|
caratteri dell'originale appartengono allo
|
|
stesso campo text/tail XHTML.
|
|
"""
|
|
|
|
normalized, mapping = (
|
|
build_normalized_mapping(node)
|
|
)
|
|
|
|
offset = int(offset)
|
|
end = offset + len(original)
|
|
|
|
if offset < 0 or end > len(normalized):
|
|
raise RuntimeError(
|
|
f"Offset {offset} non valido per "
|
|
f"{original!r}; lunghezza testo "
|
|
f"normalizzato: {len(normalized)}"
|
|
)
|
|
|
|
found = normalized[offset:end]
|
|
|
|
if found != original:
|
|
# LanguageTool e il parser XHTML possono
|
|
# differire di pochi caratteri nella
|
|
# normalizzazione degli spazi.
|
|
search_radius = 4
|
|
search_start = max(
|
|
0,
|
|
offset - search_radius,
|
|
)
|
|
search_end = min(
|
|
len(normalized),
|
|
offset
|
|
+ len(original)
|
|
+ search_radius,
|
|
)
|
|
|
|
nearby = normalized[
|
|
search_start:search_end
|
|
]
|
|
|
|
relative_positions = []
|
|
cursor = 0
|
|
|
|
while True:
|
|
position = nearby.find(
|
|
original,
|
|
cursor,
|
|
)
|
|
|
|
if position < 0:
|
|
break
|
|
|
|
relative_positions.append(
|
|
position
|
|
)
|
|
cursor = position + 1
|
|
|
|
candidates = [
|
|
search_start + position
|
|
for position in relative_positions
|
|
]
|
|
|
|
if len(candidates) != 1:
|
|
preview_start = max(
|
|
0,
|
|
offset - 40,
|
|
)
|
|
preview_end = min(
|
|
len(normalized),
|
|
end + 40,
|
|
)
|
|
|
|
raise RuntimeError(
|
|
f"Testo non coincidente all'offset "
|
|
f"{offset}: atteso {original!r}, "
|
|
f"trovato {found!r}; "
|
|
f"corrispondenze vicine: "
|
|
f"{candidates}. Contesto: "
|
|
f"{normalized[preview_start:preview_end]!r}"
|
|
)
|
|
|
|
corrected_offset = candidates[0]
|
|
|
|
print(
|
|
"Offset corretto:",
|
|
repr(original),
|
|
offset,
|
|
"->",
|
|
corrected_offset,
|
|
flush=True,
|
|
)
|
|
|
|
offset = corrected_offset
|
|
end = offset + len(original)
|
|
|
|
positions = mapping[offset:end]
|
|
|
|
if not positions:
|
|
raise RuntimeError(
|
|
f"Nessuna posizione XHTML per "
|
|
f"{original!r}"
|
|
)
|
|
|
|
first_element, first_field, first_position = (
|
|
positions[0]
|
|
)
|
|
|
|
for element, field, _ in positions:
|
|
if (
|
|
element is not first_element
|
|
or field != first_field
|
|
):
|
|
raise RuntimeError(
|
|
f"La modifica {original!r} -> "
|
|
f"{replacement!r} attraversa più "
|
|
"nodi XHTML; applicazione bloccata"
|
|
)
|
|
|
|
actual_positions = [
|
|
position
|
|
for _, _, position in positions
|
|
]
|
|
|
|
value = getattr(
|
|
first_element,
|
|
first_field,
|
|
)
|
|
|
|
last_position = actual_positions[-1]
|
|
|
|
if last_position < first_position:
|
|
raise RuntimeError(
|
|
f"Intervallo XHTML non valido per "
|
|
f"{original!r}"
|
|
)
|
|
|
|
# L'intervallo raw può contenere più spazi,
|
|
# tabulazioni o ritorni a capo, mentre il
|
|
# rapporto usa spazi normalizzati.
|
|
raw_actual = value[
|
|
first_position:
|
|
last_position + 1
|
|
]
|
|
|
|
normalized_actual = " ".join(
|
|
raw_actual.split()
|
|
)
|
|
|
|
normalized_original = " ".join(
|
|
original.split()
|
|
)
|
|
|
|
if normalized_actual != normalized_original:
|
|
raise RuntimeError(
|
|
f"Il nodo XHTML contiene "
|
|
f"{raw_actual!r}, normalizzato come "
|
|
f"{normalized_actual!r}, invece di "
|
|
f"{normalized_original!r}"
|
|
)
|
|
|
|
updated = (
|
|
value[:first_position]
|
|
+ replacement
|
|
+ value[last_position + 1:]
|
|
)
|
|
|
|
setattr(
|
|
first_element,
|
|
first_field,
|
|
updated,
|
|
)
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser();p.add_argument('source',type=Path);p.add_argument('--changes',type=Path,required=True);p.add_argument('--output',type=Path,required=True);p.add_argument('--report',type=Path,default=Path('reports/applied_changes.json'));a=p.parse_args()
|
|
if a.source.resolve()==a.output.resolve():raise SystemExit('Output e sorgente non possono coincidere')
|
|
manifest=json.loads(a.changes.read_text(encoding='utf-8'));changes=manifest.get('approved',[]);rewrites=manifest.get('sentence_rewrites',[])
|
|
if not changes and not rewrites:raise SystemExit('Nessuna modifica approvata o riscrittura')
|
|
grouped=defaultdict(list)
|
|
for c in changes:grouped[c['file']].append({**c,'change_type':'replacement'})
|
|
for c in rewrites:grouped[c['file']].append({**c,'change_type':'rewrite','original':c['original_context'],'replacement':c['rewritten_context']})
|
|
applied=[];parser=etree.XMLParser(recover=False,resolve_entities=False,no_network=True,remove_blank_text=False)
|
|
with tempfile.TemporaryDirectory(prefix='epub-review-') as td:
|
|
root=Path(td)
|
|
with ZipFile(a.source) as z:z.extractall(root)
|
|
for name,items in grouped.items():
|
|
path=root/name
|
|
if not path.exists():raise RuntimeError(f'File mancante: {name}')
|
|
tree=etree.parse(str(path),parser);pars=tree.xpath("//*[local-name()='p']")
|
|
for c in sorted(
|
|
items,
|
|
key=lambda item: (
|
|
int(item["paragraph"]),
|
|
int(item.get("offset", -1)),
|
|
),
|
|
reverse=True,
|
|
):
|
|
idx=int(c['paragraph']);node=pars[idx];before=' '.join(''.join(node.itertext()).split())
|
|
if c["change_type"] == "rewrite":
|
|
expected = " ".join(
|
|
c["original_context"].split()
|
|
)
|
|
rewritten = " ".join(
|
|
c["rewritten_context"].split()
|
|
)
|
|
|
|
if before != expected:
|
|
raise RuntimeError(
|
|
f"Il paragrafo {idx} in {name} "
|
|
"non coincide con il testo originale"
|
|
)
|
|
|
|
if len(list(node)) == 0:
|
|
# Paragrafo senza markup interno:
|
|
# la sostituzione completa è sicura.
|
|
node.text = rewritten
|
|
else:
|
|
# Paragrafo con corsivi, link o span:
|
|
# applica soltanto la differenza minima.
|
|
old_part, new_part = (
|
|
minimal_replacement(
|
|
before,
|
|
rewritten,
|
|
)
|
|
)
|
|
|
|
replace_once(
|
|
node,
|
|
old_part,
|
|
new_part,
|
|
)
|
|
else:
|
|
if c["original"] not in before:
|
|
raise RuntimeError(
|
|
f"{c['original']!r} non trovato "
|
|
f"in {name}, paragrafo {idx}"
|
|
)
|
|
|
|
correction_offset = c.get("offset")
|
|
|
|
if correction_offset is None:
|
|
# Compatibilità con manifesti precedenti
|
|
# che non contengono l'offset.
|
|
replace_once(
|
|
node,
|
|
c["original"],
|
|
c["replacement"],
|
|
)
|
|
else:
|
|
replace_at_normalized_offset(
|
|
node=node,
|
|
offset=int(correction_offset),
|
|
original=c["original"],
|
|
replacement=c["replacement"],
|
|
)
|
|
after=' '.join(''.join(node.itertext()).split());applied.append({**c,'before':before,'after':after})
|
|
tree.write(str(path),encoding='utf-8',xml_declaration=True,pretty_print=False)
|
|
a.output.parent.mkdir(parents=True,exist_ok=True)
|
|
with ZipFile(a.output,'w') as out:
|
|
mt=root/'mimetype'
|
|
if mt.exists():
|
|
info=ZipInfo('mimetype');info.compress_type=ZIP_STORED;out.writestr(info,mt.read_bytes())
|
|
for f in sorted(root.rglob('*')):
|
|
if f.is_file() and f.relative_to(root).as_posix()!='mimetype':out.write(f,f.relative_to(root).as_posix(),compress_type=ZIP_DEFLATED)
|
|
a.report.parent.mkdir(parents=True,exist_ok=True);a.report.write_text(json.dumps({'source':str(a.source),'output':str(a.output),'changes_applied':len(applied),'applied':applied},ensure_ascii=False,indent=2),encoding='utf-8')
|
|
print('Nuovo EPUB:',a.output);print('Modifiche applicate:',len(applied));print('Rapporto:',a.report)
|
|
if __name__=='__main__':main()
|