81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
from pathlib import Path
|
|
from typing import List, Dict, Optional
|
|
import json
|
|
from src.common.data_model import ManifestEntry
|
|
from src.common.utils import setup_logger
|
|
|
|
logger = setup_logger("manifest_manager")
|
|
|
|
class ManifestManager:
|
|
"""
|
|
Manages the translation manifest (Source of Truth).
|
|
Handles persistence and state updates.
|
|
"""
|
|
|
|
def __init__(self, manifest_path: Path):
|
|
self.manifest_path = manifest_path
|
|
self.entries: List[ManifestEntry] = []
|
|
self._entries_map: Dict[str, ManifestEntry] = {}
|
|
|
|
def load(self):
|
|
"""Loads manifest from disk if it exists."""
|
|
if not self.manifest_path.exists():
|
|
logger.info(f"Manifest not found at {self.manifest_path}, starting empty.")
|
|
return
|
|
|
|
try:
|
|
with open(self.manifest_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
self.entries = [ManifestEntry.model_validate(item) for item in data]
|
|
self._rebuild_map()
|
|
logger.info(f"Loaded {len(self.entries)} entries from manifest.")
|
|
except Exception as e:
|
|
logger.error(f"Failed to load manifest: {e}")
|
|
raise
|
|
|
|
def save(self):
|
|
"""Saves current state to disk."""
|
|
try:
|
|
# Pydantic v2: model_dump(mode='json') or just list dump
|
|
data = [entry.model_dump(mode='json') for entry in self.entries]
|
|
self.manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
temp = self.manifest_path.with_suffix('.tmp')
|
|
with open(temp, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
temp.replace(self.manifest_path)
|
|
logger.info(f"Saved {len(self.entries)} entries to manifest.")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save manifest: {e}")
|
|
raise
|
|
|
|
def add_entries(self, new_entries: List[ManifestEntry]):
|
|
"""
|
|
Adds new entries to the manifest.
|
|
If an entry with the same ID exists, it keeps the EXISTING one (to preserve translations).
|
|
"""
|
|
# Re-extraction is authoritative for source text/maps. Keep a translation
|
|
# only if both stayed identical; remove stale entries after force-clean.
|
|
reconciled = []
|
|
for entry in new_entries:
|
|
previous = self._entries_map.get(entry.entry_id)
|
|
if previous and previous.original_text == entry.original_text and previous.placeholders == entry.placeholders:
|
|
previous.context = entry.context
|
|
reconciled.append(previous)
|
|
else:
|
|
reconciled.append(entry)
|
|
self.entries = reconciled
|
|
self._rebuild_map()
|
|
|
|
def update_translation(self, entry_id: str, translation: str):
|
|
"""Updates translation for a specific entry."""
|
|
if entry_id in self._entries_map:
|
|
self._entries_map[entry_id].translated_text = translation
|
|
else:
|
|
logger.warning(f"Attempted to update translation for unknown ID: {entry_id}")
|
|
|
|
def get_entry(self, entry_id: str) -> Optional[ManifestEntry]:
|
|
return self._entries_map.get(entry_id)
|
|
|
|
def _rebuild_map(self):
|
|
self._entries_map = {e.entry_id: e for e in self.entries}
|