Disambiguate shared glossary aliases without discarding full entity names

src/translation/glossary.py

@@ -101,11 +101,50 @@ class BookGlossary:
         return hashlib.sha256(json.dumps(self.terms, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
 
 
+def merge_proposal(terms, proposal, conflicts, ambiguous_aliases):
+    """Keep distinct full names; never force a shared surname onto one entity."""
+    try:
+        term = copy.deepcopy(BookGlossary([proposal]).terms[0])
+    except (ValueError, AttributeError) as exc:
+        conflicts.append({'proposal': proposal, 'reason': str(exc)})
+        return terms
+    if term['source'].casefold() in ambiguous_aliases:
+        conflicts.append({'proposal': term, 'reason': 'ambiguous source name'})
+        return terms
+    term['aliases'] = [a for a in term.get('aliases', []) if a.casefold() not in ambiguous_aliases]
+    candidate = copy.deepcopy(terms)
+    existing = next((t for t in candidate if term['source'].casefold() in
+                     {a.casefold() for a in [t['source']] + t.get('aliases', [])}), None)
+    if existing and existing['target'] != term['target']:
+        conflicts.append({'existing': copy.deepcopy(existing), 'proposal': term, 'reason': 'canonical translation conflict'})
+        return terms
+    for other in candidate:
+        if other is existing or other['target'] == term['target']:
+            continue
+        shared = ({a.casefold() for a in term['aliases']} &
+                  {a.casefold() for a in [other['source']] + other.get('aliases', [])})
+        for alias in shared:
+            conflicts.append({'existing': copy.deepcopy(other), 'proposal': copy.deepcopy(term),
+                              'reason': 'shared alias between distinct entities', 'alias': alias})
+            term['aliases'] = [a for a in term['aliases'] if a.casefold() != alias]
+            if alias != other['source'].casefold():
+                ambiguous_aliases.add(alias)
+                for stored in candidate:
+                    stored['aliases'] = [a for a in stored.get('aliases', []) if a.casefold() != alias]
+    if existing:
+        existing['aliases'] = sorted(set(existing.get('aliases', []) + term['aliases']) - {existing['source']})
+    else:
+        candidate.append(term)
+    BookGlossary(candidate)
+    return candidate
+
+
 async def prepare_glossary(entries, llm, path, override=None, auto=True, chunk_chars=24000):
     """Scan the whole book before translation; never race glossary updates."""
     path = Path(path)
     digest = source_digest(entries)
     terms, conflicts, start = [], [], 0
+    ambiguous_aliases = set()
     if override:
         return BookGlossary.load(override)
     if path.exists():
@@ -115,6 +154,7 @@ async def prepare_glossary(entries, llm, path, override=None, auto=True, chunk_c
                 return BookGlossary(data['terms'])
             if data.get('chunk_chars') == chunk_chars and data.get('model') == getattr(llm, 'model', ''):
                 terms, conflicts, start = data['terms'], data.get('conflicts', []), data.get('next_chunk', 0)
+                ambiguous_aliases = set(data.get('ambiguous_aliases', []))
     if not auto:
         return BookGlossary()
     chunks, current, size = [], [], 0
@@ -139,31 +179,12 @@ async def prepare_glossary(entries, llm, path, override=None, auto=True, chunk_c
         # Validate proposals individually: a contradictory model suggestion must
         # not discard the already frozen terms or stop the entire book scan.
         for proposal in json.loads(response)['terms']:
-            try:
-                term = BookGlossary([proposal]).terms[0]
-            except (ValueError, AttributeError) as exc:
-                conflicts.append({'proposal': proposal, 'reason': str(exc)})
-                continue
-            aliases = {a.casefold() for a in [term['source']] + term.get('aliases', [])}
-            existing = next((t for t in terms if aliases & {a.casefold() for a in [t['source']] + t.get('aliases', [])}), None)
-            if existing and existing['target'] != term['target']:
-                conflicts.append({'existing': existing, 'proposal': term})
-                continue
-            candidate = copy.deepcopy(terms)
-            if existing:
-                candidate[terms.index(existing)]['aliases'] = sorted(set(existing.get('aliases', []) + [term['source']] + term.get('aliases', [])) - {existing['source']})
-            else:
-                candidate.append(term)
-            try:
-                BookGlossary(candidate)
-            except ValueError as exc:
-                conflicts.append({'proposal': term, 'reason': str(exc)})
-                continue
-            terms = candidate
+            terms = merge_proposal(terms, proposal, conflicts, ambiguous_aliases)
         temp = path.with_suffix('.tmp')
         temp.write_text(json.dumps({'source_digest': digest, 'terms': terms, 'conflicts': conflicts,
                                    'scan_complete': index + 1 == len(chunks), 'next_chunk': index + 1,
-                                   'chunk_chars': chunk_chars, 'model': getattr(llm, 'model', '')},
+                                   'chunk_chars': chunk_chars, 'model': getattr(llm, 'model', ''),
+                                   'ambiguous_aliases': sorted(ambiguous_aliases)},
                                   ensure_ascii=False, indent=2), encoding='utf-8')
         temp.replace(path)
         logger.info('Terminology scan %s/%s: %s terms, %s conflicts recorded', index + 1, len(chunks), len(terms), len(conflicts))