astrololo/services/data/app/bases.py

"""Przegląd baz na udziale i ich globalne włączanie/wyłączanie (DAN-15 / PRE-09).

Wymaganie przedefiniowane pod model serwerowy: pliki baz leżą na stałym NFS, więc
nie wybiera się folderu — potrzeba za to WIDZIEĆ, jakie bazy są dostępne (nazwa +
metaopis) i móc globalnie zdecydować, które biorą udział w interpretacji.

Stan przełączników trzymamy DEKLARATYWNIE w zmiennej `DISABLED_BASES`, a nie w
pliku, bo warstwa danych nie ma gdzie trwale zapisywać: udział z bazami jest
montowany read-only, a katalog cache to `emptyDir` (ginie przy restarcie poda).
Zapis do pliku po cichu wracałby więc do stanu sprzed restartu — a ciche
przywrócenie wyłączonej bazy jest gorsze niż konieczność edycji konfiguracji.

Dopasowanie jest tolerancyjne: wpis pasuje po nazwie pliku ALBO po ścieżce
względnej — żeby dało się wyłączyć zarówno „stara_baza.xlsx", jak i
„archiwum/stara_baza.xlsx".
"""
from __future__ import annotations

import os
from datetime import datetime, timezone
from pathlib import Path


def disabled_entries() -> list[str]:
    return [e.strip() for e in os.getenv("DISABLED_BASES", "").split(",") if e.strip()]


def _relative(path: Path, root: Path) -> str:
    try:
        return str(path.relative_to(root))
    except ValueError:
        return path.name


def is_enabled(path: str | Path, root: str | Path, entries: list[str] | None = None) -> bool:
    """Czy baza bierze udział w wyszukiwaniu interpretacji."""
    entries = disabled_entries() if entries is None else entries
    if not entries:
        return True
    p = Path(path)
    rel = _relative(p, Path(root))
    return not any(e == p.name or e == rel for e in entries)


def list_bases(root: str | Path, paths: list[str]) -> list[dict]:
    """Bazy dostępne na udziale + metaopis. Celowo TANI opis (dane z systemu
    plików): przy setkach plików liczenie rekordów oznaczałoby wczytanie każdego."""
    root = Path(root)
    entries = disabled_entries()
    out: list[dict] = []
    for raw in paths:
        p = Path(raw)
        try:
            st = p.stat()
            size_mb, modified = round(st.st_size / (1024 * 1024), 2), st.st_mtime
        except OSError:
            size_mb, modified = None, None
        out.append({
            "name": p.name,
            "path": _relative(p, root),
            "size_mb": size_mb,
            "modified": (datetime.fromtimestamp(modified, tz=timezone.utc).strftime("%Y-%m-%d")
                         if modified else None),
            "enabled": is_enabled(p, root, entries),
        })
    return out