astrololo/services/astrodemo/app/main.py

"""astrodemo — 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. Ukrycie a nieobecność to
dwie różne rzeczy. Tutaj niczego poza tymi dwiema funkcjami NIE MA W OBRAZIE: nie
ma tras, nie ma szablonów, nie ma nawet metod w kliencie warstwy logicznej. Obraz
tej usługi można komuś oddać i nie oddaje się przy tym niczego ponad to, co widać
na ekranie — łącznie z komentarzami w tym pliku, bo one też jadą razem z obrazem.

PULE SĄ ROZDZIELONE (PRE-29). Każde konto ma własny, izolowany zbiór plików,
niewidoczny dla pozostałych kont; pula bierze się WYŁĄCZNIE z loginu zalogowanej
osoby, nigdy z pola formularza. Konta są osobne (ASTRODEMO_USERS) — żeby dało się
odciąć tę usługę jedną zmienną, bez
ruszania kont głównej aplikacji.
"""
from __future__ import annotations

import base64
import logging
import re
from functools import lru_cache
from pathlib import Path
from datetime import datetime, timedelta, timezone

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

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

# Automatyczna dokumentacja WYŁĄCZONA. /docs wypisałoby komplet tras, a ta usługa
# ma nie zdradzać nawet własnej powierzchni.
app = FastAPI(title="astrodemo", docs_url=None, redoc_url=None, openapi_url=None)
security.install(app)

# ── zasoby (PRE-27) ────────────────────────────────────────────────────────
# Katalog NIE jest montowany. Montowanie serwuje wszystko, co w nim leży, każdemu
# i bez śladu w dzienniku — a komentarz w arkuszu stylów mówił wprost, że istnieje
# więcej, niż widać na ekranie. Tu jest jawna lista: arkusz logowania jest
# publiczny, bo
# potrzebuje go strona logowania, a cokolwiek innego jest nieznane i nie istnieje.
ZASOBY_PUBLICZNE = frozenset({"styles.css"})
_KATALOG_ZASOBOW = Path("app/static")


def _bez_komentarzy(tresc: str) -> str:
    """Komentarz w pliku statycznym jedzie do przeglądarki bajt w bajt.
    W repozytorium zostaje — do odpowiedzi nie ma po co trafiać."""
    tresc = re.sub(r"/\*.*?\*/", "", tresc, flags=re.S)
    return re.sub(r"\n{3,}", "\n\n", tresc).strip() + "\n"


@lru_cache(maxsize=None)
def _zasob(nazwa: str) -> bytes | None:
    """Treść zasobu tak, jak pójdzie na drut — albo None, gdy go nie ma."""
    if nazwa not in ZASOBY_PUBLICZNE:
        return None
    sciezka = (_KATALOG_ZASOBOW / nazwa).resolve()
    # Wyjście poza katalog: `..` w ścieżce albo dowiązanie na zewnątrz.
    if _KATALOG_ZASOBOW.resolve() not in sciezka.parents or not sciezka.is_file():
        return None
    if sciezka.suffix in (".css", ".js"):
        return _bez_komentarzy(sciezka.read_text(encoding="utf-8")).encode("utf-8")
    return sciezka.read_bytes()


@app.get("/static/{nazwa:path}")
def zasob(nazwa: str):
    tresc = _zasob(nazwa)
    if tresc is None:
        raise HTTPException(status_code=404)
    typ = "text/css; charset=utf-8" if nazwa.endswith(".css") else "application/octet-stream"
    return Response(tresc, media_type=typ,
                    headers={"Cache-Control": "public, max-age=3600"})

templates = Jinja2Templates(directory="app/templates")
# Szczegóły awarii idą TUTAJ, nie na ekran (patrz _blad).
log = logging.getLogger("astrodemo")
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 astrodemo 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")


# Jedno zdanie dla wszystkich awarii i CELOWO żadnego szczegółu. Treść wyjątku
# niosła nazwę usługi, jej port i nazwy jej tras — czyli mapę tego, co stoi za tym
# ekranem, podaną komuś, kto ma widzieć wyłącznie dwa formularze.
_AWARIA = "Nie udało się teraz wykonać tej operacji. Spróbuj ponownie za chwilę."


def _blad(e: Exception) -> str:
    """Komunikat dla użytkownika. Diagnostyka idzie do dziennika, nie na ekran."""
    log.warning("operacja nieudana: %s: %s", type(e).__name__, e)
    return _AWARIA


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.

    Jedna czynność, nie dwie: „dodać plik" ma tu znaczyć, że od razu bierze udział
    w wyszukiwaniu — inaczej po wgraniu nic by się nie zmieniło i wyglądałoby to
    na zepsute.

    Plik o złym układzie nie wchodzi do użytku, ale NIE JEST tracony — zostaje
    zapisany i czeka."""
    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) ----------------
# Bez własnego wolumenu nie ma gdzie trzymać licznika pokolenia sesji, więc
# „wyrzuć ze wszystkich urządzeń" robi się zmianą hasła w ASTRODEMO_USERS.
# Wylogowanie 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:
    # Sama „ok". Nazwa warstwy mówiła, że warstw jest więcej i jak się nazywa;
    # kubernetes i compose czytają wyłącznie kod odpowiedzi.
    return {"status": "ok"}