Files
astrololo/services/astroklient/app/clients/logic_client.py
T
gitea e7b6d5013b feat(astroklient): wersja demonstracyjna o dwóch funkcjach (PRE-28)
Osobna warstwa prezentacji: dodanie pliku bazy i zapytanie o interpretację
urodzeniową. Nic więcej.

OSOBNA USŁUGA, 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. Test porównuje zbiór tras aplikacji i zbiór metod
klienta z listą dokładną, więc dopisanie czegokolwiek zapala się od razu.

WGRANIE I WŁĄCZENIE TO JEDNA CZYNNOŚĆ. W pełnej aplikacji to dwie osobne decyzje
(DAN-27), bo tam ktoś nad tym panuje. Tutaj „dodać plik do bazy" musi 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, a komunikat nie zdradza reguł, bo te
zna wyłącznie administrator. Osobny test szuka w komunikacie śladów mechanizmu.

KONTO OSOBNE (DEMO_USER/DEMO_PASSWORD), nie współdzielone z główną aplikacją.
Demo pracuje na TEJ SAMEJ warstwie danych co produkcja — świadoma decyzja
właściciela — więc kto ma do niego dostęp, czyta oryginalne bazy, a jego wgrania
trafiają do produkcyjnego zbioru. Własne poświadczenia pozwalają odciąć demo jedną
zmienną, bez ruszania kont głównej aplikacji i bez zmiany hasła komukolwiek.
Zapisane wprost w README usługi i w manifeście, nie tylko w tej wiadomości.

Rozmowa z warstwą logiczną idzie tym samym szyfrowanym łączem (PRE-16) i pod tym
samym tokenem międzywarstwowym (LOG-32) — demo nie jest furtką omijającą ochronę.
Automatyczna dokumentacja wyłączona, jak w pełnej aplikacji: /docs wypisałoby
komplet tras, a demo ma nie zdradzać nawet własnej powierzchni.

Zależności celowo krótsze niż w prezentacji: bez Excela, bez stref czasowych
z lokalizacji, bez niczego pod kosmogram. Każda zbędna zależność w obrazie demo to
kolejna rzecz do pilnowania.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-26 11:56:45 +02:00

71 lines
3.2 KiB
Python

"""Klient warstwy logicznej — CELOWO okrojony do dwóch funkcji (PRE-28).
Astroklient nie dziedziczy klienta z pełnej prezentacji i nie importuje go: ma tu
być wyłącznie to, czego naprawdę używa. Metoda, której nie ma, nie zostanie
przypadkiem zawołana ani nie podpowie, że po drugiej stronie jest coś więcej.
Rozmowa idzie tym samym szyfrowanym łączem co reszta (PRE-16) i pod tym samym
tokenem międzywarstwowym (LOG-32) — demo nie jest furtką omijającą ochronę.
"""
from __future__ import annotations
import os
from typing import Any
import httpx
from app import link_crypto
from app.config import settings
def _auth_headers() -> dict[str, str]:
token = os.getenv("INTERNAL_TOKEN", "")
return {"X-Astrololo-Token": token} if token else {}
def _link() -> link_crypto.Link | None:
"""Klucz łącza czytany przy KAŻDYM wywołaniu — pod przeżywa rotację sekretu
bez restartu, a brak klucza przy wymaganym szyfrowaniu ma zatrzymać żądanie
ZANIM cokolwiek wyjdzie jawnym tekstem."""
key = link_crypto.key_from_env(link_crypto.ENV_PRESENTATION_LOGIC)
return link_crypto.Link(key) if key else None
class LogicClient:
def __init__(self, base_url: str | None = None) -> None:
self.base_url = (base_url or settings.logic_url).rstrip("/")
def report(self, when_utc_iso: str, lat: float, lon: float) -> dict[str, Any]:
"""Interpretacja urodzeniowa: sygnifikatory z obliczeń szukane w bazach.
Jedyne pytanie, jakie astroklient zadaje o treść — bez grupowania,
bez promptów, bez modelu językowego."""
payload = {"when_utc": when_utc_iso, "lat": lat, "lon": lon,
"limit": settings.report_limit, "group": False}
with httpx.Client(timeout=settings.http_timeout) as client:
return link_crypto.call_json(client, "POST", f"{self.base_url}/chart/report",
payload=payload, headers=_auth_headers(), link=_link())
def files_upload(self, filename: str, content_b64: str, by: str = "") -> dict[str, Any]:
with httpx.Client(timeout=max(settings.http_timeout, 60.0)) as client:
return link_crypto.call_json(
client, "POST", f"{self.base_url}/files/upload",
payload={"filename": filename, "content_b64": content_b64, "by": by},
headers=_auth_headers(), link=_link())
def files_status(self, path: str, status: str, by: str = "") -> dict[str, Any]:
"""Włączenie wgranego pliku do użytku. W demo wgranie i włączenie to JEDNA
czynność: „dodać do bazy" ma znaczyć, że plik od razu bierze udział
w wyszukiwaniu — inaczej demo wyglądałoby na zepsute."""
with httpx.Client(timeout=settings.http_timeout) as client:
return link_crypto.call_json(
client, "POST", f"{self.base_url}/files/status",
payload={"path": path, "status": status, "by": by},
headers=_auth_headers(), link=_link())
def health(self) -> dict[str, Any]:
with httpx.Client(timeout=settings.http_timeout) as client:
r = client.get(f"{self.base_url}/health", headers=_auth_headers())
r.raise_for_status()
return r.json()