FabledCurator/backend/app/models/import_settings.py

"""ImportSettings — single-row table holding the importer's tunable knobs.

Enforced as a single row via a CHECK (id = 1) constraint. The application
always SELECTs id=1 and never inserts/deletes after the initial migration.
"""

from sqlalchemy import Boolean, CheckConstraint, Float, Integer, Text, select
from sqlalchemy.orm import Mapped, mapped_column

from .base import Base


class ImportSettings(Base):
    __tablename__ = "import_settings"
    # Bare constraint name — Base.metadata's naming convention applies the
    # ck_<table>_<name> prefix, producing the final ck_import_settings_singleton.
    __table_args__ = (CheckConstraint("id = 1", name="singleton"),)

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    import_scan_path: Mapped[str] = mapped_column(Text, nullable=False, default="/import")

    min_width: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    min_height: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    skip_transparent: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    transparency_threshold: Mapped[float] = mapped_column(Float, nullable=False, default=0.9)

    skip_single_color: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    single_color_threshold: Mapped[float] = mapped_column(Float, nullable=False, default=0.95)
    single_color_tolerance: Mapped[int] = mapped_column(Integer, nullable=False, default=30)

    phash_threshold: Mapped[int] = mapped_column(Integer, nullable=False, default=10)

    # FC-3c downloader knobs
    download_rate_limit_seconds: Mapped[float] = mapped_column(
        Float, nullable=False, default=3.0
    )
    download_validate_files: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True
    )

    # FC-3d scheduling knobs
    download_schedule_default_seconds: Mapped[int] = mapped_column(
        Integer, nullable=False, default=28800
    )
    download_event_retention_days: Mapped[int] = mapped_column(
        Integer, nullable=False, default=90
    )
    download_failure_warning_threshold: Mapped[int] = mapped_column(
        Integer, nullable=False, default=5
    )

    # FC-3h backup knobs.
    backup_db_nightly_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False,
    )
    backup_db_nightly_hour_utc: Mapped[int] = mapped_column(
        Integer, nullable=False, default=3,
    )
    backup_db_keep_last_n: Mapped[int] = mapped_column(
        Integer, nullable=False, default=14,
    )
    backup_images_keep_last_n: Mapped[int] = mapped_column(
        Integer, nullable=False, default=3,
    )

    # FC-6.3 series continuation matcher. enabled gates the rescan; threshold is
    # the weighted-score cut-off (0..1) above which a pending suggestion is made.
    series_suggest_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True,
    )
    series_suggest_threshold: Mapped[float] = mapped_column(
        Float, nullable=False, default=0.5,
    )

    # #830 off-platform file-host downloads — per-host enable lever (default on,
    # rule #26). Column names are extdl_<host>_enabled so the worker reads them
    # via getattr(settings, f"extdl_{host}_enabled", True).
    extdl_mega_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true",
    )
    extdl_gdrive_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true",
    )
    extdl_mediafire_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true",
    )
    extdl_dropbox_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true",
    )
    extdl_pixeldrain_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true",
    )

    # -- Post-text translation via the Interpreter LAN service (milestone 143).
    # Off by default with NO default host — it needs a reachable Interpreter
    # service (the operator's, behind a reverse proxy), which not every install
    # has; the operator sets the URL and flips it on. Empty base_url OR disabled
    # → the translate sweep no-ops.
    translation_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="false",
    )
    interpreter_base_url: Mapped[str] = mapped_column(
        Text, nullable=False, default="", server_default="",
    )
    translation_target_lang: Mapped[str] = mapped_column(
        Text, nullable=False, default="en", server_default="en",
    )
    # The latin-script acceptance floor for the translation gate: a translation
    # whose Interpreter-reported confidence is below this is kept as the original
    # (operator-tunable, milestone 155). Default 0.9 — stricter than the old
    # hardcoded 0.8, because Interpreter confidently mis-detects short ASCII
    # English (e.g. "… WIP Part 1") as a European language at ~0.86. CJK stays
    # trusted regardless (script-detected). Per-post overrides handle the misses.
    translation_min_confidence: Mapped[float] = mapped_column(
        Float, nullable=False, default=0.9, server_default="0.9",
    )

    # Title-based WIP auto-tagging (task #1458). When a freshly-imported post's
    # TITLE explicitly declares work-in-progress ("WIP" / "work in progress"),
    # the importer applies the `wip` system tag to its images — the artist's own
    # label, used to keep unfinished pieces out of the Explore/gallery browse. ON
    # by default (rule 26 — the feature works out of the box). Gates only the
    # LIVE import hook; the existing catalogue is caught by the operator-triggered
    # "Scan existing posts" backfill (which runs regardless of this flag).
    wip_title_tagging_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="true",
    )

    @classmethod
    async def load(cls, session) -> ImportSettings:
        """The singleton settings row (id=1), via an async session."""
        return (await session.execute(select(cls).where(cls.id == 1))).scalar_one()

    @classmethod
    def load_sync(cls, session) -> ImportSettings:
        """The singleton settings row (id=1), via a sync session."""
        return session.execute(select(cls).where(cls.id == 1)).scalar_one()