"""Data migration utilities for upgrading glossary and other data formats.""" from __future__ import annotations import json from pathlib import Path from typing import Any from src.models.glossary import EnhancedGlossaryTerm, TheologicalDomain def infer_domain_from_notes(notes: str | None, english: str) -> TheologicalDomain: """Infer theological domain from notes and English translation. Args: notes: Notes field from the term. english: English translation. Returns: Best-guess theological domain. """ if not notes: notes = "" notes_lower = notes.lower() english_lower = english.lower() # Christology indicators christology_terms = [ "son", "christ", "messiah", "lord", "i am", "word", "logos", ] if any(term in notes_lower or term in english_lower for term in christology_terms): return TheologicalDomain.CHRISTOLOGY # Pneumatology indicators pneumatology_terms = ["spirit", "holy spirit", "paraclete", "helper", "advocate"] if any( term in notes_lower or term in english_lower for term in pneumatology_terms ): return TheologicalDomain.PNEUMATOLOGY # Soteriology indicators soteriology_terms = [ "grace", "salvation", "save", "redeem", "faith", "believe", "justif", ] if any(term in notes_lower or term in english_lower for term in soteriology_terms): return TheologicalDomain.SOTERIOLOGY # Hamartiology indicators hamartiology_terms = ["sin", "transgression", "iniquity", "flesh"] if any( term in notes_lower or term in english_lower for term in hamartiology_terms ): return TheologicalDomain.HAMARTIOLOGY # Theology Proper indicators theology_proper_terms = ["god", "father", "divine", "glory", "holy"] if any( term in notes_lower or term in english_lower for term in theology_proper_terms ): return TheologicalDomain.THEOLOGY_PROPER return TheologicalDomain.GENERAL def migrate_glossary_v1_to_v2(data: dict[str, Any]) -> dict[str, Any]: """Migrate glossary from v1 format to v2 format with enhanced metadata. Args: data: V1 glossary data with 'terms' list. Returns: V2 glossary data with enhanced fields and version marker. """ v1_terms = data.get("terms", []) v2_terms = [] for v1_term in v1_terms: # Create enhanced term with inferred domain domain = infer_domain_from_notes( v1_term.get("notes"), v1_term.get("english", "") ) v2_term = { # Core fields (preserved) "original": v1_term["original"], "transliteration": v1_term["transliteration"], "japanese": v1_term["japanese"], "english": v1_term["english"], "notes": v1_term.get("notes"), # Enhanced fields (defaults) "definition": None, "semantic_range": [], "domain": domain.value, "usage_examples": [], "frequency": 0, "related_terms": [], "avoid_translations": [], } v2_terms.append(v2_term) return { "version": "2.0", "terms": v2_terms, } def migrate_glossary_file( input_path: Path, output_path: Path | None = None, backup: bool = True ) -> list[EnhancedGlossaryTerm]: """Migrate a glossary JSON file from v1 to v2 format. Args: input_path: Path to the v1 glossary file. output_path: Path for the output file. If None, overwrites input. backup: Whether to create a backup of the original file. Returns: List of migrated EnhancedGlossaryTerm objects. """ if output_path is None: output_path = input_path # Load existing data with input_path.open("r", encoding="utf-8") as f: v1_data = json.load(f) # Check if already v2 if v1_data.get("version") == "2.0": terms = v1_data.get("terms", []) return [EnhancedGlossaryTerm(**term) for term in terms] # Create backup if backup and output_path == input_path: backup_path = input_path.with_suffix(".json.v1.bak") with backup_path.open("w", encoding="utf-8") as f: json.dump(v1_data, f, ensure_ascii=False, indent=2) # Migrate v2_data = migrate_glossary_v1_to_v2(v1_data) # Write output with output_path.open("w", encoding="utf-8") as f: json.dump(v2_data, f, ensure_ascii=False, indent=2) return [EnhancedGlossaryTerm(**term) for term in v2_data["terms"]] def load_glossary_any_version(path: Path) -> list[EnhancedGlossaryTerm]: """Load glossary from any version, converting to EnhancedGlossaryTerm. Args: path: Path to the glossary JSON file. Returns: List of EnhancedGlossaryTerm objects. """ if not path.exists(): return [] with path.open("r", encoding="utf-8") as f: data = json.load(f) # Check version version = data.get("version", "1.0") terms = data.get("terms", []) if version == "2.0": # Already v2, load directly return [EnhancedGlossaryTerm(**term) for term in terms] # V1 format - convert on load. Preserve any enhanced fields (e.g. # inflected_forms) that may already be present in the entries. converted_terms = [] for v1_term in terms: term_data = dict(v1_term) if not term_data.get("domain"): term_data["domain"] = infer_domain_from_notes( v1_term.get("notes"), v1_term.get("english", "") ) converted_terms.append(EnhancedGlossaryTerm(**term_data)) return converted_terms