fuel_track/app/database_migrations.py

from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from sqlalchemy import inspect, text
from .extensions import db


@dataclass(frozen=True)
class Migration:
    version: str
    upgrade: Callable[[], None]


def add_station_access_mode() -> None:
    inspector = inspect(db.engine)
    columns = {column["name"] for column in inspector.get_columns("company_settings")}
    if "station_access_mode" not in columns:
        db.session.execute(text(
            "ALTER TABLE company_settings ADD COLUMN station_access_mode VARCHAR(40) "
            "NOT NULL DEFAULT 'all_prefer_favorites'"
        ))


def add_fuel_entry_vat_action_snapshot() -> None:
    inspector = inspect(db.engine)
    columns = {column["name"] for column in inspector.get_columns("fuel_entry")}
    additions = {
        "vat_action_frozen": "BOOLEAN NOT NULL DEFAULT 0",
        "vat_action_applied": "BOOLEAN NOT NULL DEFAULT 0",
        "vat_action_name": "VARCHAR(160)",
        "vat_action_rate": "NUMERIC(6, 3)",
        "vat_action_start": "DATE",
        "vat_action_end": "DATE",
    }
    for name, definition in additions.items():
        if name not in columns:
            db.session.execute(text(f"ALTER TABLE fuel_entry ADD COLUMN {name} {definition}"))

    settings = dict(db.session.execute(text(
        "SELECT key, value FROM app_setting WHERE key IN "
        "('global_vat_override_enabled','global_vat_override_name','global_vat_override_rate',"
        "'global_vat_override_start','global_vat_override_end')"
    )).all())
    if settings.get("global_vat_override_enabled") == "1" and settings.get("global_vat_override_rate"):
        conditions = ["vat_action_applied = 0"]
        params = {
            "name": settings.get("global_vat_override_name") or "Projekt rządowy",
            "rate": settings["global_vat_override_rate"],
            "start": settings.get("global_vat_override_start") or None,
            "end": settings.get("global_vat_override_end") or None,
        }
        if params["start"]:
            conditions.append("DATE(fueled_at) >= DATE(:start)")
        if params["end"]:
            conditions.append("DATE(fueled_at) <= DATE(:end)")
        db.session.execute(text(f"""
            UPDATE fuel_entry
            SET vat_action_frozen = 1,
                vat_action_applied = 1,
                vat_action_name = :name,
                vat_action_rate = :rate,
                vat_action_start = :start,
                vat_action_end = :end
            WHERE {' AND '.join(conditions)}
        """), params)
    db.session.execute(text("UPDATE fuel_entry SET vat_action_frozen = 1 WHERE vat_action_frozen = 0"))



def add_fuel_entry_full_settlement_snapshot() -> None:
    inspector = inspect(db.engine)
    columns = {column["name"] for column in inspector.get_columns("fuel_entry")}
    additions = {
        "settlement_frozen": "BOOLEAN NOT NULL DEFAULT 0",
        "snapshot_vat_rate": "NUMERIC(6, 3)", "snapshot_vat_deduction_percent": "NUMERIC(6, 3)",
        "snapshot_uses_last_price": "BOOLEAN NOT NULL DEFAULT 0", "snapshot_discount_percent": "NUMERIC(6, 3)",
        "snapshot_surcharge_per_liter": "NUMERIC(10, 4)", "snapshot_rule_source": "VARCHAR(160)",
        "snapshot_base_net_price": "NUMERIC(12, 6)", "snapshot_settlement_net_price": "NUMERIC(12, 6)",
        "result_retail_net_price": "NUMERIC(12, 6)", "result_invoice_price": "NUMERIC(12, 6)",
        "result_effective_price": "NUMERIC(12, 6)", "result_retail_gross": "NUMERIC(14, 4)",
        "result_retail_net_total": "NUMERIC(14, 4)", "result_settlement_net_total": "NUMERIC(14, 4)",
        "result_invoice_gross": "NUMERIC(14, 4)", "result_deductible_vat": "NUMERIC(14, 4)",
        "result_effective_cost": "NUMERIC(14, 4)", "result_saving_net": "NUMERIC(14, 4)",
        "result_gross_saving": "NUMERIC(14, 4)", "settlement_snapshot_source": "VARCHAR(120)",
        "settlement_migration_note": "VARCHAR(255)"
    }
    for name, definition in additions.items():
        if name not in columns:
            db.session.execute(text(f"ALTER TABLE fuel_entry ADD COLUMN {name} {definition}"))


MIGRATIONS: tuple[Migration, ...] = (
    Migration("20260713_company_station_access_mode", add_station_access_mode),
    Migration("20260719_fuel_entry_vat_action_snapshot", add_fuel_entry_vat_action_snapshot),
    Migration("20260720_fuel_entry_full_settlement_snapshot", add_fuel_entry_full_settlement_snapshot),
)

def ensure_migration_table() -> None:
    inspector = inspect(db.engine)
    if inspector.has_table("schema_migration"):
        return

    db.session.execute(text("""
        CREATE TABLE schema_migration (
            version VARCHAR(100) PRIMARY KEY,
            applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
    """))
    db.session.commit()


def migrate_database() -> list[str]:
    """Uruchamia wyłącznie niewykonane migracje."""
    ensure_migration_table()

    applied = {
        row[0]
        for row in db.session.execute(text("SELECT version FROM schema_migration"))
    }
    completed: list[str] = []

    for migration in MIGRATIONS:
        if migration.version in applied:
            continue

        try:
            migration.upgrade()
            db.session.execute(
                text("INSERT INTO schema_migration (version) VALUES (:version)"),
                {"version": migration.version},
            )
            db.session.commit()
            completed.append(migration.version)
        except Exception:
            db.session.rollback()
            raise

    return completed


def main() -> None:
    from . import create_app

    app = create_app()
    with app.app_context():
        applied = migrate_database()

    if applied:
        print("Wykonane migracje: " + ", ".join(applied))
    else:
        print("Brak nowych migracji.")


if __name__ == "__main__":
    main()