"""Podstawa aplikacji: wspólne obiekty i pomocnicy, z których korzystają ekrany. Tu mieszka to, co ma KAŻDY produkt zbudowany z tego kodu: obiekt aplikacji, szablony, klient warstwy logicznej, serwowanie zasobów i garść funkcji pomocniczych. Ani jednej trasy ekranu — te są w `app/ekrany/*.py`, po jednym module na ekran, a o tym, które wchodzą, decyduje lista importów w `main.py`. Dlaczego tak: dopóki wszystko siedziało w jednym pliku, „produkt bez tego ekranu" znaczyło „produkt z ekranem, tylko zablokowanym". Plik i tak jechał do obrazu, razem z nazwami funkcji, których nie było jak włączyć. """ # build-marker: 2026-07-25 wymuszenie nowego obrazu po incydencie z tagiem :latest from __future__ import annotations import hashlib import logging import re from datetime import datetime, timedelta, timezone from functools import lru_cache from pathlib import Path from fastapi import FastAPI, HTTPException, Request from fastapi.responses import Response from fastapi.templating import Jinja2Templates from app import rozszerzenia from app import chartwheel as chartwheel_mod from app import features from app import produkt from app import session from app.house_systems import HOUSE_SYSTEMS, LIMITED as HOUSE_LIMITED, label as house_label from app import geocode, security from app.clients.logic_client import LogicClient from app.config import DEFAULT_LOCATION_LABEL, default_form # Automatyczna dokumentacja WYŁĄCZONA. /docs, /redoc i /openapi.json wypisują # KOMPLET tras aplikacji — czyli spis wszystkich funkcji, łącznie z tymi, o których # konto ograniczone nie ma prawa wiedzieć (PRE-27). Ochrona zakładek nic by nie dała, # gdyby obok leżał ich katalog. Znalezione testem przechodzącym po trasach aplikacji. app = FastAPI(title="astrololo · warstwa prezentacji", docs_url=None, redoc_url=None, openapi_url=None) # Katalogi modułów opcjonalnych idą PRZED wspólnym: moduł może dołożyć szablon, # którego wspólny zestaw nie ma. Wspólne szablony wołają je przez neutralne # gniazda `{% include "_dodatki_*" ignore missing %}` — bez modułu gniazdo jest # po prostu puste, a nie „wyłączone". templates = Jinja2Templates(directory=[*rozszerzenia.katalog_szablonow(), "app/templates"]) # Szczegóły awarii idą TUTAJ, nie na ekran (patrz _logic_error). log = logging.getLogger("astrololo.prezentacja") logic = LogicClient() security.install(app) # logowanie + limit żądań (LOG-32) # ── cache-busting statyki (PRE-27) ───────────────────────────────────────── # Po deployu przeglądarka trzymała stare styles.css / *.js (ten sam URL → cache). # Doklejamy do URL-a krótki HASH TREŚCI pliku: zmieni się plik → zmieni się URL → # przeglądarka pobierze nowy; bez zmian URL zostaje ten sam (cache działa dalej). # Hash liczony raz na proces (lru_cache) — nowy pod po deployu = świeży hash. _STATIC_DIR = Path("app/static") _TYPY_ZASOBOW = { ".css": "text/css; charset=utf-8", ".js": "application/javascript; charset=utf-8", ".svg": "image/svg+xml", ".png": "image/png", ".woff2": "font/woff2", } def _bez_komentarzy(tresc: str, sufiks: str) -> str: """Usuwa komentarze z arkusza albo skryptu przed wysłaniem. Komentarz w pliku statycznym jedzie do przeglądarki bajt w bajt, a te komentarze opisują funkcje pełnym zdaniem po polsku — łącznie z tym, czego konto ograniczone NIE widzi i kto to widzi. Zostają w repozytorium, bo tam są potrzebne; do odpowiedzi nie mają po co trafiać. Kasujemy tylko komentarze blokowe i CAŁE wiersze `//`. Komentarz na końcu wiersza zostaje CELOWO: żeby go usunąć, trzeba rozpoznać, czy `//` nie jest w napisie albo w wyrażeniu regularnym — a pomyłka tutaj psuje skrypt po cichu. Test pilnuje słownika zakazanych słów, więc reszta by i tak wyszła. """ tresc = re.sub(r"/\*.*?\*/", "", tresc, flags=re.S) if sufiks == ".js": tresc = re.sub(r"(?m)^[ \t]*//.*$\n?", "", tresc) return re.sub(r"\n{3,}", "\n\n", tresc).strip() + "\n" @lru_cache(maxsize=None) def _asset_body(name: str) -> bytes | None: """Treść zasobu tak, jak pójdzie na drut — albo None, gdy go nie ma.""" for katalog in (*rozszerzenia.katalog_zasobow(), str(_STATIC_DIR)): korzen = Path(katalog).resolve() sciezka = (Path(katalog) / name).resolve() # Wyjście poza katalog zasobów: `..` w ścieżce albo dowiązanie na zewnątrz. if korzen in sciezka.parents and sciezka.is_file(): break else: return None surowe = sciezka.read_bytes() sufiks = sciezka.suffix.lower() # Kod obcy zostaje nietknięty: minifikowany plik potrafi mieć `//` w napisie # albo w wyrażeniu regularnym, a on i tak nie mówi nic o naszym programie. if sufiks in (".css", ".js") and not name.startswith("vendor/"): try: return _bez_komentarzy(surowe.decode("utf-8"), sufiks).encode("utf-8") except UnicodeDecodeError: return surowe return surowe @lru_cache(maxsize=None) def _asset_version(name: str) -> str: tresc = _asset_body(name) # Hash liczony z treści WYSYŁANEJ, nie z pliku na dysku: inaczej zmiana # samego komentarza podbijałaby adres, choć odpowiedź jest ta sama. return hashlib.md5(tresc).hexdigest()[:8] if tresc is not None else "0" def static_url(name: str) -> str: return f"/static/{name}?v={_asset_version(name)}" @app.get("/static/{nazwa:path}") def static_file(nazwa: str): """Zasoby idą przez bramkę uprawnień jak każda inna trasa. Dopóki `/static/` było w PUBLIC_PREFIXES, katalog omijał całe sprawdzanie: ukrycie przycisku w szablonie nic nie dawało, bo obok leżał skrypt tej samej funkcji, pod nazwą, którą wystarczyło zgadnąć. Uprawnienie zasobu sprawdza middleware (features.static_required); tutaj zostaje samo wydanie pliku. """ tresc = _asset_body(nazwa) if tresc is None: raise HTTPException(status_code=404) return Response( tresc, media_type=_TYPY_ZASOBOW.get(Path(nazwa).suffix.lower(), "application/octet-stream"), # Adres niesie hash treści, więc wersja pod danym adresem jest niezmienna. headers={"Cache-Control": "public, max-age=31536000, immutable"}, ) # Nazwa produktu z modułu, nie z szablonu: szablony są wspólne dla kilku # produktów, a nazwa nie. templates.env.globals["produkt"] = produkt.NAZWA templates.env.globals["static"] = static_url # Pozycje specjalne przefiltrowane uprawnieniem — szablon dostaje gotową listę # i nie musi wiedzieć, czego która wymaga. templates.env.globals["nav_extra"] = lambda request: [ (k, a, e) for k, a, e, u in features.NAWIGACJA_SPECJALNA if features.can(_perms(request), u)] # Katalog systemów domów — jedno źródło nazw dla wszystkich szablonów. templates.env.globals["HOUSE_SYSTEMS"] = HOUSE_SYSTEMS templates.env.globals["house_label"] = house_label templates.env.globals["HOUSE_LIMITED"] = HOUSE_LIMITED templates.env.globals["WHEEL_ORIENTATIONS"] = chartwheel_mod.ORIENTATIONS def _perms(request: Request) -> frozenset[str]: """Uprawnienia zalogowanego konta — pusty zbiór, gdy nie wiadomo kto to.""" who = getattr(request.state, "principal", None) return who.permissions if who is not None else frozenset() # Szablony pytają o uprawnienia PRZEZ TE DWIE FUNKCJE, nigdy o konto wprost. # Dzięki temu nie da się w szablonie napisać warunku na login — a to jest # dokładnie ten rodzaj sprytu, który po roku zostawia kogoś z dostępem, o którym # nikt już nie pamięta. templates.env.globals["nav_screens"] = lambda request: features.screens_for(_perms(request)) # Kto jest zalogowany i czy logowanie w ogóle działa — pod przycisk wyjścia. templates.env.globals["current_user"] = lambda request: getattr(request.state, "user", "") or "" templates.env.globals["auth_on"] = security.auth_enabled templates.env.globals["can"] = lambda request, key: key in _perms(request) # Czy to konto ma uprawnienie NIENADAWALNE — czyli takie, którego nie da się # przyznać z żadnego ekranu. Nazwa neutralna, bo szablon jest wspólny. templates.env.globals["widok_diagnostyczny"] = lambda request: features.can( _perms(request), features.ADMIN) def _limit_options(request: Request, opts: dict) -> dict: """Ścina opcje, których konto nie ma prawa użyć — PO STRONIE SERWERA. Ukrycie pola w formularzu chroni tylko przed przypadkiem. Kto zna nazwy pól, wyśle je wprost — a wtedy policzyłby funkcję, której mu nie przyznano, i to jeszcze zobaczyłby ją w wyniku. Granicą jest ta funkcja, nie szablon. Wartości zastępcze są DOMYŚLNE, nie puste: konto ograniczone ma dostać działający horoskop w podstawowej postaci, a nie komunikat o braku czegoś. """ perms = _perms(request) if "advanced_calc" not in perms: opts.update(stations=False, tables=False, aspect_minor=False, zodiac="tropical") if "houses_compare" not in perms: opts.update(house_system="whole_sign", house_systems=[], wheel_orientation=chartwheel_mod.ASC_LEFT) return opts def _landing(request: Request) -> str | None: """Pierwszy ekran dostępny dla konta — dokąd odesłać z korzenia.""" allowed = features.screens_for(_perms(request)) return allowed[0].href if allowed else None def _build_utc(date: str, time: str, tz_offset: float) -> tuple[str, str]: """Z lokalnej daty/godziny + przesunięcia strefy → moment UTC. Zwraca (iso_utc, etykieta_czytelna). UTC = czas lokalny − offset. """ local = datetime.fromisoformat(f"{date}T{time}") utc = (local - timedelta(hours=tz_offset)).replace(tzinfo=timezone.utc) label = utc.strftime("%Y-%m-%d %H:%M UTC") return utc.isoformat(), label # Akcja formularza → uprawnienie, którego wymaga. Akcji spoza tej mapy nie # ograniczamy; akcja Z mapy, na którą konto nie ma prawa, jest traktowana jak # NIEZNANA — czyli wraca do zachowania domyślnego ekranu. # Akcje własne tego ekranu plus te, które wnoszą moduły opcjonalne. Aplikacja # nie wypisuje ich z nazwy — gdyby wypisywała, nazwa funkcji byłaby w kodzie # także tam, gdzie modułu nie ma. _AKCJE_POD_UPRAWNIENIEM = {"export": "export", **rozszerzenia.akcje()} def _dozwolona_akcja(request: Request, action: str, domyslna: str) -> str: """Akcja, której konto nie ma prawa użyć, zachowuje się jak literówka. NIE zwracamy błędu ani 404: komunikat „brak uprawnień do generowania" sam w sobie mówiłby, że taka funkcja istnieje (PRE-27). Ukrycie przycisku chroni tylko przed przypadkiem — kto zna nazwę pola, wyśle ją wprost, i to TU musi się zatrzymać.""" wymagane = _AKCJE_POD_UPRAWNIENIEM.get(action) if wymagane and wymagane not in _perms(request): return domyslna return action def _report_records(report: dict | None) -> int: """Ile rekordów baz niesie raport — do dziennika audytowego (PRE-17). Liczymy próbki we wszystkich fasetach: to one są treścią z baz. Sama liczba, nigdy treść — dziennik ma pokazywać SKALĘ dostępu, nie powielać danych.""" total = 0 for o in (report or {}).get("objects") or []: for f in o.get("facets") or []: total += len(f.get("samples") or []) return total # Jedno zdanie dla wszystkich awarii — i CELOWO żadnego szczegółu. # Poprzednia treść wypisywała na ekran nazwę endpointu (/chart/positions), nazwę # podsystemu („silnik efemeryd"), nazwę gałęzi rozwojowej i — przez str(wyjątku # httpx) — wewnętrzny adres z portem sąsiedniej usługi. Dla konta, które ma # widzieć wyłącznie Pliki, to był komplet: co program liczy, że stoi za nim # łańcuch usług, i gdzie zapukać. _AWARIA = "Nie udało się teraz wykonać tej operacji. Spróbuj ponownie za chwilę." def _logic_error(e: Exception) -> str: """Komunikat dla użytkownika. Diagnostyka idzie do dziennika, nie na ekran. Odsiew siedzi TUTAJ, a nie w siedemnastu miejscach wywołania: gdyby zależał od tego, czy ktoś pamiętał przekazać uprawnienia, prędzej czy później jedno wywołanie zostałoby pominięte — i nie byłoby tego jak zauważyć. """ log.warning("warstwa logiczna: %s: %s", type(e).__name__, e) return _AWARIA features.zarejestruj(trasy={("GET", "/health"): None}) @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"}