"""Klient HTTP do warstwy logicznej. Jedyny punkt styku prezentacji w dół. Przekazuje dane z formularza i odbiera opracowane wyniki. Prezentacja nie sięga bezpośrednio do bazy ani do silnika. """ 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 międzywarstwowy (LOG-32) — pusty, gdy ochrona wyłączona.""" token = os.getenv("INTERNAL_TOKEN", "") return {"X-Astrololo-Token": token} if token else {} def _link() -> link_crypto.Link | None: """Klucz łącza prezentacja↔logika. Czytany przy każdym wywołaniu, bo konfiguracja może się zmienić bez restartu procesu (testy, podmiana sekretu).""" 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 _post(self, path: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]: """Jedyna droga w dół. Celowo JEDNA: dopóki każda metoda składała żądanie sama, dołożenie nowej znaczyło, że łatwo zapomnieć o tokenie albo kluczu łącza — i tak się już raz stało (401 wyszedł dopiero na produkcji).""" with httpx.Client(timeout=timeout) as client: return link_crypto.call_json(client, "POST", f"{self.base_url}{path}", payload=payload, headers=_auth_headers(), link=_link()) def query(self, query: str, field: str, exact: bool, limit: int) -> dict[str, Any]: payload = {"query": query, "field": field, "exact": exact, "limit": limit} return self._post("/api/query", payload, settings.http_timeout) def positions( self, when_utc_iso: str, lat: float, lon: float, objects: list[str] | None = None, house_system: str = "whole_sign", stations: bool = False, zodiac: str = "tropical", tables: bool = False, house_systems: list[str] | None = None, aspect_orb: float = 8.0, aspect_luminary_bonus: float = 2.0, aspect_minor: bool = False, ) -> dict[str, Any]: """Pełny horoskop dla danego momentu — woła logic /chart/positions.""" payload = { "when_utc": when_utc_iso, "lat": lat, "lon": lon, "objects": objects, "house_system": house_system, "house_systems": house_systems, "aspect_orb": aspect_orb, "aspect_luminary_bonus": aspect_luminary_bonus, "aspect_minor": aspect_minor, "stations": stations, "zodiac": zodiac, "tables": tables, } # stacje ORAZ tabele wymagają root-findów / szukania numerycznego — dłuższy timeout timeout = max(settings.http_timeout, 60.0) if (stations or tables) else settings.http_timeout return self._post("/chart/positions", payload, timeout) def synastry( self, person_a: dict, person_b: dict, zodiac: str = "tropical", aspect_orb: float = 8.0, aspect_luminary_bonus: float = 2.0, aspect_minor: bool = False, ) -> dict[str, Any]: """Synastria dwóch osób (PRE-04) — aspekty między horoskopami.""" payload = { "person_a": person_a, "person_b": person_b, "zodiac": zodiac, "aspect_orb": aspect_orb, "aspect_luminary_bonus": aspect_luminary_bonus, "aspect_minor": aspect_minor, } return self._post("/chart/synastry", payload, settings.http_timeout) def report( self, when_utc_iso: str, lat: float, lon: float, limit: int = 5000, group: bool = False ) -> dict[str, Any]: """Sygnifikatory z obliczeń szukane w bazie — woła logic /chart/report.""" payload = {"when_utc": when_utc_iso, "lat": lat, "lon": lon, "limit": limit, "group": group} return self._post("/chart/report", payload, max(settings.http_timeout, 30.0)) def prompt( self, profile: str, when_utc_iso: str, lat: float, lon: float, budget: str = "medium", from_date: str | None = None, to_date: str | None = None, provider: str | None = None, model: str | None = None, ) -> dict[str, Any]: """Gotowy prompt do LLM z wyliczeń (LOG-29/30) — woła logic /chart/prompt. `provider` jest potrzebny dla budżetu „maksymalny kontekst modelu": limit znaków zależy wtedy od okna kontekstu konkretnego modelu.""" payload: dict[str, Any] = { "profile": profile, "when_utc": when_utc_iso, "lat": lat, "lon": lon, "budget": budget, "provider": provider, "model": model, } if from_date and to_date: payload["from_date"], payload["to_date"] = from_date, to_date return self._post("/chart/prompt", payload, max(settings.http_timeout, 60.0)) def horoscope( self, profile: str, when_utc_iso: str, lat: float, lon: float, budget: str = "medium", provider: str | None = None, model: str | None = None, from_date: str | None = None, to_date: str | None = None, ) -> dict[str, Any]: """Napisany horoskop (LOG-31) — woła logic /chart/horoscope. Generowanie tekstu trwa (zwłaszcza na modelu lokalnym), stąd długi timeout. """ payload: dict[str, Any] = { "profile": profile, "when_utc": when_utc_iso, "lat": lat, "lon": lon, "budget": budget, } if provider: payload["provider"] = provider if model: payload["model"] = model if from_date and to_date: payload["from_date"], payload["to_date"] = from_date, to_date return self._post("/chart/horoscope", payload, max(settings.http_timeout, 300.0)) def horoscope_stream(self, payload: dict[str, Any]): """Strumień postępu pisania horoskopu (NDJSON) — przekazywany do przeglądarki. Timeout jest długi, bo generowanie trwa; strumień i tak niesie heartbeat, więc cisza na łączu nie zostanie wzięta za zerwanie. Idzie przez szyfrowane łącze jak reszta ruchu w dół (PRE-16). Bez tego przy włączonym `LINK_ENCRYPTION_REQUIRED` serwer odrzuciłby nieszyfrowane żądanie i okno postępu przestałoby działać. `stream_lines` pieczętuje żądanie i odszyfrowuje odpowiedź ramka po ramce, zachowując dostarczanie na żywo. """ with httpx.Client(timeout=httpx.Timeout(None, connect=15.0)) as client: yield from link_crypto.stream_lines( client, f"{self.base_url}/chart/horoscope/stream", payload=payload, headers=_auth_headers(), link=_link()) def bases(self) -> dict[str, Any]: """Bazy dostępne na udziale + metaopis i stan włączenia (DAN-15/PRE-09).""" with httpx.Client(timeout=settings.http_timeout) as client: return link_crypto.call_json(client, "GET", f"{self.base_url}/bases", headers=_auth_headers(), link=_link()) # ── zarządzanie plikami baz (DAN-27) ──────────────────────────────── # Jedna metoda na trasę, bez sprytnego generyka: te wywołania różnią się # skutkiem (odczyt / zapis / skasowanie), a ujednolicenie ich w jedno # `call(path, payload)` zaciera tę różnicę dokładnie tam, gdzie jest ważna. def files_list(self, for_admin: bool = False) -> dict[str, Any]: return self._files_post("/files", {"for_admin": for_admin}) def files_status(self, path: str, status: str, by: str = "") -> dict[str, Any]: return self._files_post("/files/status", {"path": path, "status": status, "by": by}) def files_upload(self, filename: str, content_b64: str, by: str = "") -> dict[str, Any]: return self._files_post("/files/upload", {"filename": filename, "content_b64": content_b64, "by": by}) def files_delete(self, path: str) -> dict[str, Any]: return self._files_post("/files/delete", {"path": path}) def files_rules(self, rules: dict) -> dict[str, Any]: return self._files_post("/files/rules", {"rules": rules}) def _files_post(self, path: str, payload: dict) -> dict[str, Any]: with httpx.Client(timeout=settings.http_timeout) as client: return link_crypto.call_json(client, "POST", f"{self.base_url}{path}", payload=payload, headers=_auth_headers(), link=_link()) def llm_models(self) -> dict[str, Any]: """Katalog modeli per dostawca (podpowiedzi do pola wyboru w UI).""" with httpx.Client(timeout=settings.http_timeout) as client: return link_crypto.call_json(client, "GET", f"{self.base_url}/llm/models", headers=_auth_headers(), link=_link()) def timeline( self, when_utc_iso: str, lat: float, lon: float, from_date: str, to_date: str, interpret: bool = True, ) -> dict[str, Any]: """Oś czasu z technik (+interpretacje z bazy) — woła logic /chart/timeline.""" payload = { "when_utc": when_utc_iso, "lat": lat, "lon": lon, "from_date": from_date, "to_date": to_date, "interpret": interpret, } return self._post("/chart/timeline", payload, max(settings.http_timeout, 60.0))