ss-tools/backend/src/plugins/translate/dictionary_import_export.py

# #region DictionaryImportExport [C:3] [TYPE Class] [SEMANTICS dictionary, import, export, csv, migration]
# @BRIEF Import/export entries as CSV/TSV with conflict detection, and migration of old entries.
# @RELATION DEPENDS_ON -> [DictionaryEntry]
# @RELATION DEPENDS_ON -> [TerminologyDictionary]

import csv
import io
from typing import Any

from sqlalchemy.orm import Session

from ...core.logger import belief_scope, logger
from ...models.translate import DictionaryEntry, TerminologyDictionary
from ._utils import _detect_delimiter, _normalize_term


class DictionaryImportExport:
    """Import/export dictionary entries as CSV/TSV, and migrate old-style entries."""

    # region import_entries [TYPE Function]
    # @PURPOSE: Import entries from CSV/TSV content with duplicate detection and conflict resolution.
    @staticmethod
    def import_entries(
        db: Session, dict_id: str, content: str,
        delimiter: str | None = None,
        on_conflict: str = "overwrite",
        preview_only: bool = False,
        default_source_language: str | None = None,
        default_target_language: str | None = None,
    ) -> dict[str, Any]:
        with belief_scope("DictionaryImportExport.import_entries"):
            dictionary = db.query(TerminologyDictionary).filter(TerminologyDictionary.id == dict_id).first()
            if not dictionary:
                raise ValueError(f"Dictionary not found: {dict_id}")

            if not delimiter:
                delimiter = _detect_delimiter(content)
                logger.reason("Detected delimiter", {"delimiter": repr(delimiter)})

            if delimiter not in (",", "\t"):
                raise ValueError(f"Unsupported delimiter: {delimiter!r}. Use ',' or '\\t'.")

            reader = csv.DictReader(io.StringIO(content), delimiter=delimiter)
            required_fields = {"source_term", "target_term"}
            if not reader.fieldnames or not required_fields.issubset(reader.fieldnames):
                raise ValueError(
                    f"CSV/TSV must have at least 'source_term' and 'target_term' columns. "
                    f"Got: {reader.fieldnames}"
                )

            result: dict[str, Any] = {"total": 0, "created": 0, "updated": 0, "skipped": 0, "errors": [], "preview": []}
            rows = list(reader)
            result["total"] = len(rows)

            for row_idx, row in enumerate(rows):
                try:
                    source_term = row.get("source_term", "").strip()
                    target_term = row.get("target_term", "").strip()
                    context_notes = row.get("context_notes", "").strip() or None
                    source_language = row.get("source_language", "").strip() or default_source_language or "und"
                    target_language = row.get("target_language", "").strip() or default_target_language or "und"

                    if not source_term or not target_term:
                        result["errors"].append({"line": row_idx + 2, "error": "Both source_term and target_term are required", "row": row})
                        continue

                    normalized = _normalize_term(source_term)

                    if preview_only:
                        existing = (
                            db.query(DictionaryEntry)
                            .filter(
                                DictionaryEntry.dictionary_id == dict_id,
                                DictionaryEntry.source_term_normalized == normalized,
                                DictionaryEntry.source_language == source_language,
                                DictionaryEntry.target_language == target_language,
                            )
                            .first()
                        )
                        result["preview"].append({
                            "line": row_idx + 2, "source_term": source_term, "target_term": target_term,
                            "context_notes": context_notes, "source_language": source_language,
                            "target_language": target_language, "is_conflict": existing is not None,
                            "existing_target_term": existing.target_term if existing else None,
                        })
                        continue

                    existing = (
                        db.query(DictionaryEntry)
                        .filter(
                            DictionaryEntry.dictionary_id == dict_id,
                            DictionaryEntry.source_term_normalized == normalized,
                            DictionaryEntry.source_language == source_language,
                            DictionaryEntry.target_language == target_language,
                        )
                        .first()
                    )

                    if existing:
                        if on_conflict == "overwrite":
                            existing.source_term = source_term
                            existing.target_term = target_term
                            existing.context_notes = context_notes
                            existing.source_language = source_language
                            existing.target_language = target_language
                            result["updated"] += 1
                        elif on_conflict == "keep_existing":
                            result["skipped"] += 1
                        else:
                            result["errors"].append({"line": row_idx + 2, "error": f"Conflict on '{source_term}' and on_conflict='cancel'", "row": row})
                            continue
                    else:
                        entry = DictionaryEntry(
                            dictionary_id=dict_id, source_term=source_term,
                            source_term_normalized=normalized, target_term=target_term,
                            context_notes=context_notes, source_language=source_language,
                            target_language=target_language,
                        )
                        db.add(entry)
                        result["created"] += 1
                except Exception as e:
                    result["errors"].append({"line": row_idx + 2, "error": str(e), "row": row})

            if not preview_only:
                db.commit()

            logger.reflect("Import complete", {"dict_id": dict_id, "total": result["total"], "created": result["created"]})
            return result
    # endregion import_entries

    # region export_entries [TYPE Function]
    # @PURPOSE: Export all entries as CSV string with language columns.
    @staticmethod
    def export_entries(db: Session, dict_id: str, delimiter: str = ",") -> str:
        with belief_scope("DictionaryImportExport.export_entries"):
            dictionary = db.query(TerminologyDictionary).filter(TerminologyDictionary.id == dict_id).first()
            if not dictionary:
                raise ValueError(f"Dictionary not found: {dict_id}")

            entries = (
                db.query(DictionaryEntry)
                .filter(DictionaryEntry.dictionary_id == dict_id)
                .order_by(DictionaryEntry.source_term.asc())
                .all()
            )

            output = io.StringIO()
            fieldnames = ["source_term", "target_term", "source_language", "target_language", "context_notes", "context_data", "usage_notes"]
            writer = csv.DictWriter(output, fieldnames=fieldnames, delimiter=delimiter)
            writer.writeheader()
            for entry in entries:
                writer.writerow({
                    "source_term": entry.source_term, "target_term": entry.target_term,
                    "source_language": entry.source_language, "target_language": entry.target_language,
                    "context_notes": entry.context_notes or "", "context_data": entry.context_data,
                    "usage_notes": entry.usage_notes or "",
                })

            result = output.getvalue()
            logger.reflect("Entries exported", {"dict_id": dict_id, "count": len(entries)})
            return result
    # endregion export_entries

    # region migrate_old_entries [TYPE Function]
    # @PURPOSE: Migrate existing single-language dictionary entries to populate source_language and target_language.
    @staticmethod
    def migrate_old_entries(db: Session) -> dict[str, int]:
        with belief_scope("DictionaryImportExport.migrate_old_entries"):
            logger.reason("Starting migration of old dictionary entries")
            migrated_source = 0
            migrated_target = 0
            skipped = 0

            entries = (
                db.query(DictionaryEntry)
                .filter(
                    (DictionaryEntry.source_language == "und") |
                    (DictionaryEntry.target_language == "und")
                )
                .all()
            )

            for entry in entries:
                if entry.source_language == "und":
                    if entry.origin_source_language:
                        entry.source_language = entry.origin_source_language
                        migrated_source += 1
                if entry.source_language == "und" and entry.target_language == "und":
                    skipped += 1

            db.commit()
            logger.reflect("Migration complete", {"migrated_source": migrated_source, "total_processed": len(entries)})
            return {"migrated_source": migrated_source, "migrated_target": migrated_target, "skipped": skipped, "total_processed": len(entries)}
    # endregion migrate_old_entries

# #endregion DictionaryImportExport