cantarire-etichetare/cantarire/db.py

"""Conexiunea SQLite, migrarea de la modelul vechi si setarile aplicatiei."""
from __future__ import annotations

import sqlite3
from pathlib import Path
from typing import Any, Optional

import click
from flask import current_app, g

SCHEMA_PATH = Path(__file__).with_name("schema.sql")
SCHEMA_VERSION = 5

DEFAULT_SETTINGS: dict[str, str] = {
    "company_name": "Firma SRL",
    "company_address": "",
    "company_phone": "",
    "active_scale_profile": "",
    "default_template_id": "",
    "auto_print": "1",
    "default_operator": "",
}

DEFAULT_TEMPLATE_NAME = "Eticheta produs finit"


def _default_design_json(width: float = 100.0, height: float = 60.0) -> str:
    """Designul implicit (import lazy: `designer` depinde de `repositories`)."""
    from . import designer

    return designer.dump_design(designer.default_design(width, height))


def get_db() -> sqlite3.Connection:
    if "db" not in g:
        path = Path(current_app.config["DATABASE"])
        path.parent.mkdir(parents=True, exist_ok=True)
        db = sqlite3.connect(path, detect_types=sqlite3.PARSE_DECLTYPES)
        db.row_factory = sqlite3.Row
        db.execute("PRAGMA foreign_keys = ON")
        db.execute("PRAGMA journal_mode = WAL")
        g.db = db
    return g.db


def close_db(_exc: Optional[BaseException] = None) -> None:
    db = g.pop("db", None)
    if db is not None:
        db.close()


def _table_names(db: sqlite3.Connection) -> set[str]:
    return {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}


def _migrate_legacy(db: sqlite3.Connection) -> None:
    """Modelul vechi (`products` + `weighings`) devine `components` + istoric arhivat."""
    tables = _table_names(db)
    if "components" not in tables and "products" in tables:
        db.execute("ALTER TABLE products RENAME TO components")
    if "packages" not in tables and "weighings" in tables:
        db.execute("ALTER TABLE weighings RENAME TO legacy_weighings")


def _schema_version(db: sqlite3.Connection) -> int:
    """Versiunea schemei salvata in setari (0 daca lipseste)."""
    if "settings" not in _table_names(db):
        return 0
    row = db.execute("SELECT value FROM settings WHERE key = 'schema_version'").fetchone()
    try:
        return int(row["value"]) if row is not None else 0
    except (TypeError, ValueError):
        return 0


def _seed_settings(db: sqlite3.Connection) -> None:
    for key, value in DEFAULT_SETTINGS.items():
        db.execute("INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)", (key, value))


def _seed_template(db: sqlite3.Connection) -> None:
    row = db.execute(
        "SELECT id FROM label_templates ORDER BY is_default DESC, id LIMIT 1"
    ).fetchone()
    if row is None:
        cur = db.execute(
            "INSERT INTO label_templates(name, width_mm, height_mm, body_html, css,"
            " design_json, is_default) VALUES (?, ?, ?, ' ', '', ?, 1)",
            (DEFAULT_TEMPLATE_NAME, 100.0, 60.0, _default_design_json(100.0, 60.0)),
        )
        db.execute(
            "INSERT OR REPLACE INTO settings(key, value) VALUES ('default_template_id', ?)",
            (str(cur.lastrowid),),
        )
        return

    # Sabloanele vechi (doar HTML/CSS) primesc designul implicit, ca eticheta sa iasa mereu.
    legacy = db.execute(
        "SELECT id, width_mm, height_mm FROM label_templates"
        " WHERE design_json IS NULL OR TRIM(design_json) = ''"
    ).fetchall()
    for item in legacy:
        width = float(item["width_mm"] or 100.0)
        height = float(item["height_mm"] or 60.0)
        db.execute(
            "UPDATE label_templates SET design_json = ?, updated_at = datetime('now')"
            " WHERE id = ?",
            (_default_design_json(width, height), item["id"]),
        )


def _add_missing_columns(db: sqlite3.Connection) -> None:
    """Coloane adaugate dupa prima versiune a schemei."""
    tables = _table_names(db)
    if "label_templates" in tables:
        columns = {row[1] for row in db.execute("PRAGMA table_info(label_templates)")}
        if "design_json" not in columns:
            db.execute(
                "ALTER TABLE label_templates ADD COLUMN design_json TEXT NOT NULL DEFAULT ''"
            )
    if "recipes" in tables:
        columns = {row[1] for row in db.execute("PRAGMA table_info(recipes)")}
        # Sablonul de eticheta al fiecarui produs finit (v5).
        if "template_id" not in columns:
            db.execute(
                "ALTER TABLE recipes ADD COLUMN template_id INTEGER"
                " REFERENCES label_templates(id) ON DELETE SET NULL"
            )


def init_db() -> None:
    """Creeaza/migreaza tabelele, setarile si sablonul implicit."""
    db = get_db()
    _migrate_legacy(db)
    db.executescript(SCHEMA_PATH.read_text(encoding="utf-8"))
    _add_missing_columns(db)
    _seed_settings(db)
    _seed_template(db)
    db.execute(
        "INSERT OR REPLACE INTO settings(key, value) VALUES ('schema_version', ?)",
        (str(SCHEMA_VERSION),),
    )
    db.commit()


def get_setting(key: str, default: str = "") -> str:
    row = get_db().execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
    return row["value"] if row is not None else default


def set_setting(key: str, value: Any) -> None:
    db = get_db()
    db.execute(
        "INSERT INTO settings(key, value) VALUES (?, ?)"
        " ON CONFLICT(key) DO UPDATE SET value = excluded.value",
        (key, str(value)),
    )
    db.commit()


def all_settings() -> dict[str, str]:
    return {row["key"]: row["value"] for row in get_db().execute("SELECT key, value FROM settings")}


def init_db_command() -> None:
    init_db()
    click.echo("Baza de date a fost initializata.")


def register(app) -> None:
    app.teardown_appcontext(close_db)
    app.cli.add_command(click.command("init-db")(init_db_command))