astrololo/services/astroklient-demo/app/main.py

"""astroklient-demo — wersja DEMONSTRACYJNA (PRE-28).

DWIE FUNKCJE I ANI JEDNEJ WIĘCEJ:
  1. dodanie pliku bazy (wgranie + włączenie do użytku),
  2. zapytanie o interpretację urodzeniową dla podanych danych wejściowych.

DLACZEGO OSOBNA USŁUGA, A NIE KONTO Z OGRANICZENIAMI. Mechanizm uprawnień z PRE-27
umiałby to ukryć w pełnej aplikacji, ale ukrycie a nieobecność to dwie różne rzeczy.
Tutaj pozostałych funkcji NIE MA W OBRAZIE: nie ma tras, nie ma szablonów, nie ma
nawet metod w kliencie warstwy logicznej. Demo można komuś oddać, nie oddając
przy okazji kodu reszty programu.

UWAGA, ŚWIADOMA DECYZJA WŁAŚCICIELA: astroklient-demo pracuje na TEJ SAMEJ warstwie
danych co produkcja. Kto ma do niego dostęp, czyta oryginalne bazy interpretacyjne,
a pliki przez niego wgrane trafiają do produkcyjnego zbioru. Dlatego konto jest
osobne (DEMO_USER/DEMO_PASSWORD) — żeby dało się odciąć demo jedną zmienną, bez
ruszania kont głównej aplikacji.
"""
from __future__ import annotations

import base64
from datetime import datetime, timedelta, timezone

import httpx
from fastapi import FastAPI, File, Form, Request, UploadFile
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from app import link_crypto, security, session
from app.clients.logic_client import LogicClient

# Automatyczna dokumentacja WYŁĄCZONA — tak samo jak w pełnej aplikacji. /docs
# wypisałoby komplet tras, a demo ma nie zdradzać nawet własnej powierzchni.
app = FastAPI(title="astroklient-demo", docs_url=None, redoc_url=None, openapi_url=None)
security.install(app)

app.mount("/static", StaticFiles(directory="app/static"), name="static")
templates = Jinja2Templates(directory="app/templates")
templates.env.globals["current_user"] = lambda request: getattr(request.state, "user", "") or ""
templates.env.globals["auth_on"] = security.auth_enabled

def _logic(request: Request) -> LogicClient:
    """Klient związany z pulą ZALOGOWANEGO konta.

    Budowany PER ŻĄDANIE, nie raz na moduł: pula musi wynikać z tego, kto się
    zalogował, a nie z globalnego stanu procesu. Login bierzemy z request.state,
    które ustawia warstwa logowania — nigdy z pola formularza, bo wtedy
    wystarczyłoby podstawić cudzy login, żeby czytać cudze bazy."""
    return LogicClient(tenant=getattr(request.state, "user", "") or "")

# Stan pliku „w użyciu" — ta sama nazwa, co w warstwie danych (DAN-27). Wpisana
# tutaj wprost, bo astroklient-demo nie importuje tamtej warstwy; literówka objawiłaby
# się dopiero na żywo, więc pilnuje jej test.
ACTIVE = "active"


def _build_utc(date: str, time: str, tz_offset: float) -> tuple[str, str]:
    """Data + godzina lokalna + offset → chwila w UTC.

    Offset podaje się wprost, bez wyszukiwania strefy z współrzędnych — demo ma
    dwie funkcje, a strefy to trzecia."""
    try:
        local = datetime.strptime(f"{date} {time}", "%Y-%m-%d %H:%M")
    except ValueError as e:
        raise ValueError("Podaj datę jako RRRR-MM-DD i godzinę jako GG:MM.") from e
    moment = local.replace(tzinfo=timezone(timedelta(hours=tz_offset)))
    return moment.astimezone(timezone.utc).isoformat(), local.strftime("%Y-%m-%d %H:%M")


def _blad(e: Exception) -> str:
    if isinstance(e, httpx.HTTPStatusError):
        try:
            return str(e.response.json().get("detail") or e)
        except Exception:                                # noqa: BLE001
            return str(e)
    if isinstance(e, link_crypto.LinkError):
        return f"Łącze do warstwy logicznej nie działa: {e}"
    return f"Warstwa logiczna niedostępna: {e}"


def _ile_rekordow(report: dict) -> int:
    return sum(len(s.get("samples") or []) for p in (report.get("positions") or [])
               for s in (p.get("significators") or []))


@app.get("/", response_class=HTMLResponse)
def formularz(request: Request, error: str = "", done: str = ""):
    return templates.TemplateResponse(request, "index.html",
                                      {"result": None, "form": {}, "error": error, "done": done})


@app.post("/", response_class=HTMLResponse)
def interpretacja(request: Request, date: str = Form(...), time: str = Form(...),
                  tz_offset: float = Form(0.0), lat: float = Form(0.0), lon: float = Form(0.0)):
    """Interpretacja urodzeniowa dla podanych danych. Jedyne pytanie o treść."""
    form = {"date": date, "time": time, "tz_offset": tz_offset, "lat": lat, "lon": lon}
    ctx: dict = {"form": form, "result": None, "error": None, "done": "", "moment": None}
    try:
        iso_utc, label = _build_utc(date, time, tz_offset)
        ctx["moment"] = label
        ctx["result"] = _logic(request).report(when_utc_iso=iso_utc, lat=lat, lon=lon)
        # Ile treści baz wyszło jednym zapytaniem — sama liczba, nigdy treść (PRE-17).
        security.audit_records(request, _ile_rekordow(ctx["result"]))
    except ValueError as e:
        ctx["error"] = str(e)
    except (httpx.HTTPError, link_crypto.LinkError) as e:
        ctx["error"] = _blad(e)
    return templates.TemplateResponse(request, "index.html", ctx)


@app.post("/plik")
async def dodaj_plik(request: Request, upload: UploadFile = File(...)):
    """Dodanie pliku bazy: wgranie ORAZ włączenie do użytku.

    W pełnej aplikacji to dwie osobne decyzje (DAN-27), bo tam ktoś nad tym panuje.
    Tutaj „dodać do bazy" ma znaczyć, że plik od razu bierze udział w wyszukiwaniu —
    inaczej po wgraniu nic by się nie zmieniło i demo wyglądałoby na zepsute.

    Walidacja zostaje: plik o złym układzie NIE wchodzi do użytku, ale też NIE JEST
    tracony — leży dalej na udziale i decyzję o nim podejmuje administrator
    w pełnej aplikacji."""
    raw = await upload.read()
    if not raw:
        return RedirectResponse("/?error=Pusty+plik.", status_code=303)

    who = getattr(request.state, "user", "") or "demo"
    try:
        out = _logic(request).files_upload(upload.filename or "baza.xlsx",
                                 base64.b64encode(raw).decode("ascii"), by=who)
    except (httpx.HTTPError, link_crypto.LinkError) as e:
        return _wroc(error=_blad(e))

    nazwa = out.get("name") or "plik"
    if not out.get("accepted"):
        # Bez słowa o regułach — te zna wyłącznie administrator (DAN-27).
        return _wroc(done=f"Plik „{nazwa}” został przyjęty, ale nie nadaje się "
                          f"do użycia w tej postaci.")
    try:
        _logic(request).files_status(out["path"], ACTIVE, by=who)
    except (httpx.HTTPError, link_crypto.LinkError) as e:
        return _wroc(error=f"Plik „{nazwa}” wgrany, ale nie udało się go włączyć: {_blad(e)}")
    return _wroc(done=f"Dodano „{nazwa}”. Baza jest już używana przy wyszukiwaniu.")


def _wroc(error: str = "", done: str = "") -> RedirectResponse:
    """Po zapisie PRZEKIEROWANIE, nie render — odświeżenie strony nie może
    wgrać pliku drugi raz."""
    from urllib.parse import urlencode

    q = urlencode({k: v for k, v in (("error", error), ("done", done)) if v})
    return RedirectResponse(f"/{'?' + q if q else ''}", status_code=303)


# ---------------- Logowanie i wylogowanie (LOG-34) ----------------
# Ta sama konstrukcja co w pełnej aplikacji. Różnica jest jedna i wynika z tego,
# że demo nie ma własnego wolumenu: nie ma licznika pokolenia sesji, więc
# „wyrzuć zewsząd" robi się zmianą hasła w DEMO_USERS. Wylogowanie i tak działa
# natychmiast, bo polega na skasowaniu ciasteczka.

def _bezpieczne_dokad(dokad: str) -> str:
    """Adres powrotu przepuszczony przez sito — bez tego `?dokad=https://obcy.pl`
    zamieniłby ekran logowania w narzędzie do wyłudzania haseł."""
    dokad = (dokad or "").strip()
    if not dokad.startswith("/") or dokad.startswith("//") or "\\" in dokad:
        return "/"
    return dokad.split("?", 1)[0].split("#", 1)[0] or "/"


@app.get("/logowanie", response_class=HTMLResponse)
def logowanie_form(request: Request, dokad: str = "/", blad: str = ""):
    if security.auth_enabled() and security.principal(request):
        return RedirectResponse(_bezpieczne_dokad(dokad), status_code=303)
    return templates.TemplateResponse(request, "logowanie.html",
                                      {"dokad": _bezpieczne_dokad(dokad), "blad": blad})


@app.post("/logowanie")
def logowanie(request: Request, login: str = Form(""), haslo: str = Form(""),
              dokad: str = Form("/")):
    cel = _bezpieczne_dokad(dokad)
    kto = security.verify(login.strip(), haslo)
    if kto is None:
        security.audit_login(request, login.strip(), udane=False)
        return templates.TemplateResponse(
            request, "logowanie.html",
            {"dokad": cel, "blad": "Nieprawidłowy login lub hasło.", "login": login},
            status_code=401)
    security.audit_login(request, kto, udane=True)
    odpowiedz = RedirectResponse(cel, status_code=303)
    odpowiedz.set_cookie(session.COOKIE, security.issue_session(kto),
                         max_age=session.max_age(),
                         **session.cookie_params(secure=security.cookies_secure()))
    return odpowiedz


@app.post("/wyloguj")
def wyloguj(request: Request):
    """POST, nie odsyłacz: pod adresem GET wystarczyłby obrazek na obcej stronie,
    żeby wylogować kogoś bez jego wiedzy."""
    security.audit_logout(request)
    odpowiedz = RedirectResponse("/logowanie?wylogowano=1", status_code=303)
    odpowiedz.delete_cookie(session.COOKIE,
                            **session.cookie_params(secure=security.cookies_secure()))
    return odpowiedz


@app.get("/health")
def health() -> dict:
    return {"status": "ok", "layer": "astroklient-demo"}