Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 gitea
|
d4a37aca52 | ||
|
gitea
|
e39c31618d |
@@ -1,32 +0,0 @@
|
||||
name: build-swisseph
|
||||
|
||||
# Osobny pipeline dla silnika B (Swiss Ephemeris, AGPL) — celowo ODDZIELONY od
|
||||
# głównego build.yaml (data/logic/presentation). Buduje się tylko, gdy zmienia się
|
||||
# sam silnik, i nie miesza obrazu AGPL do pipeline'u permisywnego produktu.
|
||||
#
|
||||
# Obraz konsumuje profil deployu `astrololo-swisseph` w repo `deploy`.
|
||||
on:
|
||||
push:
|
||||
branches: [master]
|
||||
paths:
|
||||
- 'services/engine-swisseph/**'
|
||||
- '.gitea/workflows/build-swisseph.yaml'
|
||||
workflow_dispatch: {} # ręczne odpalenie (bootstrap pierwszego obrazu)
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Login
|
||||
run: echo "${{ secrets.REGISTRY_TOKEN }}" | docker login gitea.czernobog.pl -u gitea --password-stdin
|
||||
- name: Build & push engine-swisseph (AGPL, izolowany)
|
||||
run: |
|
||||
TAG=${GITHUB_SHA::8}
|
||||
IMG=gitea.czernobog.pl/gitea/astrololo-engine-swisseph
|
||||
# Obraz kompiluje pyswisseph ze źródeł (brak wheeli dla cp312) — build jest
|
||||
# zarazem realnym testem Dockerfile'a.
|
||||
docker build -t $IMG:$TAG -t $IMG:latest ./services/engine-swisseph
|
||||
docker push $IMG:$TAG
|
||||
docker push $IMG:latest
|
||||
echo "Zbudowano i wypchnięto: $IMG:$TAG (+ latest)"
|
||||
@@ -51,57 +51,6 @@ jobs:
|
||||
EPHEMERIS_DIR: ${{ github.workspace }}/services/logic/.ephemeris
|
||||
run: pytest tests -q -rs
|
||||
|
||||
presentation-tests:
|
||||
name: Testy warstwy prezentacji (dostęp do baz)
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
cache: pip
|
||||
cache-dependency-path: services/presentation/requirements-dev.txt
|
||||
- name: Instalacja zależności
|
||||
run: pip install -r services/presentation/requirements-dev.txt
|
||||
# Bramka chroniąca oryginalne bazy — nietestowany kod ochronny jest gorszy
|
||||
# niż jego brak, bo daje złudzenie zabezpieczenia.
|
||||
- name: Testy (pytest)
|
||||
working-directory: services/presentation
|
||||
env:
|
||||
PYTHONPATH: .
|
||||
run: pytest tests -q -rs
|
||||
|
||||
swisseph-image:
|
||||
name: Build obrazu silnika B (swisseph)
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
# Obraz kompiluje pyswisseph ze źródeł (brak wheeli dla cp312), więc ten
|
||||
# build jest realnym testem Dockerfile'a — nie tylko pobraniem paczek.
|
||||
- name: docker build
|
||||
run: docker build -t astrololo/engine-swisseph:ci services/engine-swisseph
|
||||
|
||||
- name: Smoke test (health + pozycje)
|
||||
run: |
|
||||
docker run -d --name swe -p 8003:8003 astrololo/engine-swisseph:ci
|
||||
for i in $(seq 1 30); do
|
||||
curl -fsS http://localhost:8003/health >/dev/null 2>&1 && break
|
||||
sleep 1
|
||||
done
|
||||
curl -fsS http://localhost:8003/health
|
||||
echo
|
||||
# Horoskop referencyjny (30.04.1984) — ten sam, na którym opieramy testy
|
||||
# silnika własnego; sprawdzamy, że silnik B faktycznie liczy.
|
||||
curl -fsS -X POST http://localhost:8003/positions \
|
||||
-H 'Content-Type: application/json' \
|
||||
-d '{"when_utc":"1984-04-30T09:20:00Z","lat":50.0647,"lon":19.9450}'
|
||||
echo
|
||||
|
||||
- name: Logi kontenera (gdy coś padło)
|
||||
if: failure()
|
||||
run: docker logs swe || true
|
||||
|
||||
compile-all:
|
||||
name: Kontrola składni wszystkich warstw
|
||||
runs-on: ubuntu-latest
|
||||
Binary file not shown.
@@ -10,7 +10,6 @@ from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from app import security
|
||||
from app.config import settings
|
||||
from app.models import HealthInfo, SearchQuery, SearchResult
|
||||
from app.providers.factory import build_provider
|
||||
@@ -25,7 +24,6 @@ async def lifespan(app: FastAPI):
|
||||
|
||||
|
||||
app = FastAPI(title="astrololo · warstwa bazodanowa", lifespan=lifespan)
|
||||
security.install(app, "danych") # token międzywarstwowy (LOG-32)
|
||||
|
||||
|
||||
@app.post("/search", response_model=SearchResult)
|
||||
|
||||
@@ -18,10 +18,7 @@ class SearchQuery(BaseModel):
|
||||
key: str = Field(..., description="Pole/kolumna kanoniczna, po której szukamy, np. 'name'.")
|
||||
value: str = Field(..., description="Szukana wartość.")
|
||||
exact: bool = Field(False, description="Dopasowanie dokładne vs. zawieranie (contains).")
|
||||
# Górny limit celowo niski: to zapytanie oddaje SUROWE wiersze baz, więc wysoki
|
||||
# pułap zamienia je w narzędzie do masowego pobrania (LOG-32). 5000 = tyle, ile
|
||||
# realnie potrzebuje build_report na jeden obiekt.
|
||||
limit: int = Field(50, ge=1, le=5000)
|
||||
limit: int = Field(50, ge=1, le=50000)
|
||||
fields: list[str] | None = Field(
|
||||
None, description="Lista pól kanonicznych do zwrócenia; None = wszystkie."
|
||||
)
|
||||
|
||||
@@ -1,50 +0,0 @@
|
||||
"""Uwierzytelnianie międzywarstwowe (LOG-32).
|
||||
|
||||
Warstwa danych oddaje SUROWE wiersze baz — to najbardziej wrażliwy punkt całego
|
||||
systemu. Bez tego kontrolera wystarczyłoby uderzyć w nią bezpośrednio, z pominięciem
|
||||
i logiki, i logowania w UI. Gdy ustawiono INTERNAL_TOKEN, każde żądanie (poza /health)
|
||||
musi go przynieść w nagłówku X-Astrololo-Token.
|
||||
|
||||
Bez INTERNAL_TOKEN kontrola jest wyłączona (dev / zgodność wstecz) — wtedy przy
|
||||
starcie leci ostrzeżenie.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import secrets
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
log = logging.getLogger("astrololo.security")
|
||||
|
||||
HEADER = "X-Astrololo-Token"
|
||||
PUBLIC_PATHS = frozenset({"/health"})
|
||||
|
||||
|
||||
def token() -> str:
|
||||
"""Czytany leniwie — konfiguracja może się zmienić bez importu modułu."""
|
||||
return os.getenv("INTERNAL_TOKEN", "")
|
||||
|
||||
|
||||
def enabled() -> bool:
|
||||
return bool(token())
|
||||
|
||||
|
||||
def install(app, layer: str) -> None:
|
||||
if not enabled():
|
||||
log.warning(
|
||||
"UWAGA: INTERNAL_TOKEN nie ustawiony — warstwa %s przyjmuje żądania od "
|
||||
"kogokolwiek, kto ma do niej dostęp sieciowy.", layer,
|
||||
)
|
||||
|
||||
@app.middleware("http")
|
||||
async def _guard(request: Request, call_next):
|
||||
if request.url.path in PUBLIC_PATHS or not enabled():
|
||||
return await call_next(request)
|
||||
got = request.headers.get(HEADER, "")
|
||||
if not secrets.compare_digest(got, token()):
|
||||
return JSONResponse({"detail": "Brak lub błędny token międzywarstwowy."},
|
||||
status_code=401)
|
||||
return await call_next(request)
|
||||
@@ -1,34 +1,10 @@
|
||||
# Build wieloetapowy — bo `pyswisseph` to rozszerzenie C bez gotowych wheeli.
|
||||
#
|
||||
# Na PyPI (2.10.3.2) wheels kończą się na cp311 i obejmują wyłącznie i686/x86_64.
|
||||
# Dla Pythona 3.12 oraz dla arm64 pip ZAWSZE kompiluje ze źródeł, a `-slim` nie ma
|
||||
# kompilatora — dlatego jednoetapowy build tu padał. Kompilujemy w etapie builder,
|
||||
# a do obrazu finalnego wchodzi już tylko gotowy wheel (bez toolchaina).
|
||||
|
||||
FROM python:3.12-slim AS builder
|
||||
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends build-essential \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
WORKDIR /build
|
||||
COPY requirements.txt .
|
||||
RUN pip wheel --no-cache-dir --wheel-dir /wheels -r requirements.txt
|
||||
|
||||
|
||||
FROM python:3.12-slim
|
||||
|
||||
WORKDIR /app
|
||||
COPY --from=builder /wheels /wheels
|
||||
COPY requirements.txt .
|
||||
RUN pip install --no-cache-dir --no-index --find-links=/wheels -r requirements.txt \
|
||||
&& rm -rf /wheels
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
COPY . .
|
||||
|
||||
# Sanity check na etapie budowania: brak działającego swissepha ma wywalić build,
|
||||
# a nie dopiero pierwszy request.
|
||||
RUN python -c "import swisseph as swe; swe.set_ephe_path(None); print('swisseph OK', swe.version)"
|
||||
|
||||
EXPOSE 8003
|
||||
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8003"]
|
||||
|
||||
@@ -15,23 +15,9 @@ permisywnej reszty systemu. **Nie wchodzi do dystrybucji zamkniętego produktu.*
|
||||
|
||||
Tryb Moshiera (`FLG_MOSEPH`) — bez plików efemeryd, zero konfiguracji.
|
||||
|
||||
## Build obrazu
|
||||
```bash
|
||||
docker compose --profile comparison build engine-swisseph
|
||||
```
|
||||
Dockerfile jest **wieloetapowy** i to nie jest ozdobnik: `pyswisseph` to rozszerzenie
|
||||
C, a na PyPI (2.10.3.2) gotowe wheels kończą się na **cp311** i obejmują wyłącznie
|
||||
**i686/x86_64**. Dla Pythona 3.12 oraz dla arm64 pip musi kompilować ze źródeł, więc
|
||||
sam `python:3.12-slim` (bez kompilatora) build wywracał. Kompilacja idzie w etapie
|
||||
`builder` (`build-essential`), a do obrazu finalnego trafia już tylko gotowy wheel —
|
||||
runtime zostaje czysty i mały. Pierwszy build trwa ~1–2 min, kolejne idą z cache warstw.
|
||||
|
||||
Build kończy się sanity-checkiem (`import swisseph`), żeby niedziałający silnik
|
||||
wykrzaczył build, a nie dopiero pierwszy request.
|
||||
|
||||
## Uruchomienie (tylko profil porównawczy / dev / CI)
|
||||
```bash
|
||||
pip install -r requirements.txt # wymaga kompilatora C (patrz wyżej)
|
||||
pip install -r requirements.txt
|
||||
uvicorn app.main:app --port 8003
|
||||
```
|
||||
Następnie w warstwie logicznej ustaw `ENGINE_SWISSEPH_URL=http://localhost:8003`,
|
||||
|
||||
@@ -6,7 +6,7 @@ i nie w bazie.
|
||||
|
||||
## API
|
||||
- `POST /api/query` → `QueryRequest` → `QueryResponse`
|
||||
- `POST /chart/positions` → `{when_utc, lat, lon, house_system?}` → pełny horoskop: pozycje (LOG-01) + osie i domy (LOG-05) + aspekty główne z applying/separating (LOG-06) + opcjonalnie stacje planet (`stations:true`, LOG-03). Zwraca też sektę i 7 Lots hermetycznych z domami (LOG-08). Obiekty: 10 planet + mean NN/SN/Lilith (LOG-02). `house_system`: `whole_sign` (dom.) / `equal` / `porphyry`.
|
||||
- `POST /chart/positions` → `{when_utc, lat, lon, house_system?}` → pełny horoskop: pozycje (LOG-01) + osie i domy (LOG-05) + aspekty główne z applying/separating (LOG-06) + opcjonalnie stacje planet (`stations:true`, LOG-03). Obiekty: 10 planet + mean NN/SN/Lilith (LOG-02). `house_system`: `whole_sign` (dom.) / `equal` / `porphyry`.
|
||||
- `POST /chart/report` → `{when_utc, lat, lon, limit?}` → wynik obliczeń wyszukany w bazie: fasety sygnifikatorów **w znaku / w domu / w aspekcie**, z rozwinięciem skrótów, odsiewaniem duplikatów (ten sam sygnifikator i opis), rankingiem siły (LOG-21) oraz opcją group (grupowanie identycznych opisów)
|
||||
- `POST /chart/profections` → `{when_utc, lat, lon, start_age?, count?}` → profekcje roczne: wiek, profektowany Asc, Władca Roku (+MC/Su/Mo) (LOG-10)
|
||||
- `POST /chart/return` → `{when_utc, lat, lon, kind, around?}` → Solar/Lunar Return: moment powrotu + pełny horoskop na ten moment (LOG-12)
|
||||
|
||||
@@ -24,8 +24,6 @@ ABBREVIATIONS: dict[str, str] = {
|
||||
"NN": "North Node", "SN": "South Node", "Lilith": "Lilith", "Chiron": "Chiron",
|
||||
# osie
|
||||
"Asc": "Ascendant", "Dsc": "Descendant", "MC": "Midheaven", "IC": "Imum Coeli",
|
||||
# Lots (punkty arabskie)
|
||||
"PF": "Part of Fortune", "Fortune": "Part of Fortune", "Spirit": "Lot of Spirit",
|
||||
# aspekty
|
||||
"conj": "conjunction", "sex": "sextile", "sq": "square", "tri": "trine",
|
||||
"opp": "opposition", "semisex": "semisextile", "semisq": "semisquare",
|
||||
|
||||
@@ -5,7 +5,6 @@ Jedyny punkt styku w dół. Gdyby warstwa bazodanowa zmieniła implementację
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
@@ -13,12 +12,6 @@ import httpx
|
||||
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 {}
|
||||
|
||||
|
||||
class DataClient:
|
||||
def __init__(self, base_url: str | None = None) -> None:
|
||||
self.base_url = (base_url or settings.data_url).rstrip("/")
|
||||
@@ -33,12 +26,12 @@ class DataClient:
|
||||
) -> dict[str, Any]:
|
||||
payload = {"key": key, "value": value, "exact": exact, "limit": limit, "fields": fields}
|
||||
with httpx.Client(timeout=max(settings.http_timeout, 30.0)) as client:
|
||||
r = client.post(f"{self.base_url}/search", json=payload, headers=_auth_headers())
|
||||
r = client.post(f"{self.base_url}/search", json=payload)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
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 = client.get(f"{self.base_url}/health")
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
@@ -27,18 +27,6 @@ LUMINARIES = {"Sun", "Moon"}
|
||||
DEFAULT_ORB = 8.0
|
||||
LUMINARY_BONUS = 2.0
|
||||
|
||||
# Pary sztywno powiązane definicyjnie — kąt między nimi wynika z samej definicji
|
||||
# punktu, nie z układu nieba (SN = NN + 180°). Aspekt taki zawsze wychodzi
|
||||
# dokładny (orb 0,00°) i nie niesie żadnej informacji astrologicznej, więc
|
||||
# wycinamy go z wyników: zaśmieca listę w UI i zjada budżet promptu do LLM.
|
||||
RIGID_PAIRS = frozenset({
|
||||
frozenset({"North Node", "South Node"}),
|
||||
})
|
||||
|
||||
|
||||
def _is_rigid(name_a: str, name_b: str) -> bool:
|
||||
return frozenset({name_a, name_b}) in RIGID_PAIRS
|
||||
|
||||
|
||||
def separation(a: float, b: float) -> float:
|
||||
"""Najmniejsza separacja kątowa [0,180]."""
|
||||
@@ -68,16 +56,12 @@ def find_aspects(
|
||||
|
||||
Zwraca listę aspektów głównych; gdy znane są prędkości, każdy aspekt ma
|
||||
applying (bool) i skrót 'as': 'A'/'S' (aplikacyjny/separacyjny).
|
||||
|
||||
Pary z RIGID_PAIRS (np. NN/SN) są pomijane — ich kąt jest definicyjny.
|
||||
"""
|
||||
out: list[dict] = []
|
||||
n = len(positions)
|
||||
for i in range(n):
|
||||
for j in range(i + 1, n):
|
||||
a, b = positions[i], positions[j]
|
||||
if _is_rigid(a["name"], b["name"]):
|
||||
continue
|
||||
la, lb = a.get("decimal"), b.get("decimal")
|
||||
if la is None or lb is None:
|
||||
continue
|
||||
|
||||
@@ -7,51 +7,27 @@ pozycje.
|
||||
from __future__ import annotations
|
||||
|
||||
from app.engine import houses as H
|
||||
from app.engine import zodiac as Z
|
||||
from app.engine.base import EphemerisEngine
|
||||
from app.engine.formats import SIGNS, absolute, decimal, in_sign, norm360, sign_index
|
||||
from app.engine.formats import SIGNS, in_sign, norm360, sign_index
|
||||
from app.engine.models import ChartMoment
|
||||
|
||||
|
||||
def _fmt(name: str, lon: float, off: float = 0.0) -> dict:
|
||||
lon = norm360(lon - off)
|
||||
def _fmt(name: str, lon: float) -> dict:
|
||||
return {
|
||||
"name": name,
|
||||
"sign": SIGNS[sign_index(lon)],
|
||||
"in_sign": in_sign(lon),
|
||||
"decimal": round(lon, 6),
|
||||
"decimal": round(norm360(lon), 6),
|
||||
}
|
||||
|
||||
|
||||
def _shift_pos(pdict: dict, off: float) -> None:
|
||||
"""Przelicza etykiety pozycji na wybrany zodiak (in-place). off=0 → bez zmian."""
|
||||
if not off:
|
||||
return
|
||||
lon = norm360(pdict["decimal"] - off)
|
||||
pdict["sign"] = SIGNS[sign_index(lon)]
|
||||
pdict["in_sign"] = in_sign(lon)
|
||||
pdict["absolute"] = absolute(lon)
|
||||
pdict["decimal"] = decimal(lon)
|
||||
|
||||
|
||||
def build_chart(engine: EphemerisEngine, moment: ChartMoment, house_system: str = H.WHOLE_SIGN,
|
||||
lots_method: str = "degree", zodiac: str = Z.TROPICAL) -> dict:
|
||||
def build_chart(engine: EphemerisEngine, moment: ChartMoment, house_system: str = H.WHOLE_SIGN) -> dict:
|
||||
from app.engine.aspects import find_aspects
|
||||
|
||||
positions = engine.positions(moment)
|
||||
result: dict = {"engine": engine.name, "positions": [p.as_dict() for p in positions]}
|
||||
# aspekty liczymy PRZED zmianą zodiaku — kąty między obiektami są niezmiennicze
|
||||
result["aspects"] = find_aspects(result["positions"]) # aspekty (LOG-06)
|
||||
|
||||
# offset zodiaku (LOG-04): syderyczny = ayanamsa, draconic = długość węzła
|
||||
node_lon = next((p.longitude for p in positions if p.name == "North Node"), None)
|
||||
off = Z.offset(zodiac, Z.julian_day(moment.when_utc), node_lon)
|
||||
result["zodiac"] = zodiac
|
||||
if zodiac in Z.SIDEREAL:
|
||||
result["ayanamsha"] = round(off, 6)
|
||||
for pdict in result["positions"]:
|
||||
_shift_pos(pdict, off)
|
||||
|
||||
if not hasattr(engine, "sidereal"):
|
||||
return result
|
||||
|
||||
@@ -59,37 +35,19 @@ def build_chart(engine: EphemerisEngine, moment: ChartMoment, house_system: str
|
||||
asc = H.compute_asc(ramc, eps, moment.lat)
|
||||
mc = H.compute_mc(ramc, eps)
|
||||
system = house_system if house_system in H.SYSTEMS else H.WHOLE_SIGN
|
||||
cusp_list = H.cusps(asc, mc, system) # tropikalne — geometria domów jest niezmiennicza
|
||||
cusp_list = H.cusps(asc, mc, system)
|
||||
|
||||
result["house_system"] = system
|
||||
result["angles"] = {
|
||||
"Asc": _fmt("Asc", asc, off),
|
||||
"MC": _fmt("MC", mc, off),
|
||||
"Dsc": _fmt("Dsc", norm360(asc + 180.0), off),
|
||||
"IC": _fmt("IC", norm360(mc + 180.0), off),
|
||||
"Asc": _fmt("Asc", asc),
|
||||
"MC": _fmt("MC", mc),
|
||||
"Dsc": _fmt("Dsc", norm360(asc + 180.0)),
|
||||
"IC": _fmt("IC", norm360(mc + 180.0)),
|
||||
}
|
||||
result["cusps"] = [
|
||||
{"house": i + 1, "sign": SIGNS[sign_index(norm360(c - off))],
|
||||
"in_sign": in_sign(norm360(c - off))}
|
||||
{"house": i + 1, "sign": SIGNS[sign_index(c)], "in_sign": in_sign(c)}
|
||||
for i, c in enumerate(cusp_list)
|
||||
]
|
||||
for pdict, obj in zip(result["positions"], positions):
|
||||
pdict["house"] = H.assign_house(obj.longitude, cusp_list) # dom po długości tropikalnej
|
||||
|
||||
# Lots (LOG-08) — wymagają Asc i sekty (dzień/noc)
|
||||
from app.engine.firdaria import is_day_birth
|
||||
from app.engine.lots import compute_lots
|
||||
|
||||
pts = {p.name: p.longitude for p in positions}
|
||||
pts["Asc"] = asc
|
||||
day = is_day_birth(pts["Sun"], asc, mc) if "Sun" in pts else True
|
||||
result["sect"] = "day" if day else "night"
|
||||
result["lots"] = [
|
||||
{**lot,
|
||||
"longitude": decimal(norm360(lot["longitude"] - off)), # w wybranym zodiaku
|
||||
"sign": SIGNS[sign_index(norm360(lot["longitude"] - off))],
|
||||
"in_sign": in_sign(norm360(lot["longitude"] - off)),
|
||||
"house": H.assign_house(lot["longitude"], cusp_list)} # dom po długości tropikalnej
|
||||
for lot in compute_lots(pts, day, lots_method)
|
||||
]
|
||||
pdict["house"] = H.assign_house(obj.longitude, cusp_list)
|
||||
return result
|
||||
|
||||
@@ -1,60 +0,0 @@
|
||||
"""Lots / punkty arabskie (LOG-08) — 7 Lots hermetycznych.
|
||||
|
||||
Formuła: Lot = C + A − B (od punktu C odmierzamy odległość między A i B).
|
||||
Większość Lots **odwraca się w horoskopach nocnych** (zamiana A↔B) — np.
|
||||
Fortuna: dzień Asc + Mo − Su, noc Asc + Su − Mo.
|
||||
|
||||
Dwa warianty liczenia (notes3):
|
||||
- `degree` (domyślny) — dokładny stopień,
|
||||
- `sign` — liczone całymi znakami (Lot wypada na 0° wyliczonego znaku).
|
||||
|
||||
Kolejność ma znaczenie: Fortuna i Duch liczone są pierwsze, bo pozostałe Lots
|
||||
odwołują się do nich.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from app.engine.formats import norm360, sign_index
|
||||
|
||||
# (nazwa, C, A, B, odwracalny w nocy)
|
||||
LOT_DEFS: list[tuple[str, str, str, str, bool]] = [
|
||||
("Fortune", "Asc", "Moon", "Sun", True),
|
||||
("Spirit", "Asc", "Sun", "Moon", True),
|
||||
("Eros", "Asc", "Venus", "Spirit", True),
|
||||
("Necessity", "Asc", "Fortune", "Mercury", True),
|
||||
("Courage", "Asc", "Fortune", "Mars", True),
|
||||
("Victory", "Asc", "Jupiter", "Spirit", True),
|
||||
("Nemesis", "Asc", "Fortune", "Saturn", True),
|
||||
]
|
||||
|
||||
METHODS = ("degree", "sign")
|
||||
|
||||
|
||||
def compute_lots(
|
||||
points: dict[str, float], is_day: bool, method: str = "degree"
|
||||
) -> list[dict]:
|
||||
"""points: nazwa → długość natalna (wymagane Asc + planety formuł).
|
||||
|
||||
Zwraca listę {name, longitude, formula} w kolejności definicji.
|
||||
"""
|
||||
if method not in METHODS:
|
||||
raise ValueError(f"nieznana metoda liczenia Lots: {method}")
|
||||
|
||||
vals = dict(points)
|
||||
out: list[dict] = []
|
||||
for name, c, a, b, reversible in LOT_DEFS:
|
||||
first, second = (a, b) if (is_day or not reversible) else (b, a)
|
||||
if any(k not in vals for k in (c, first, second)):
|
||||
continue # brak składnika — pomijamy
|
||||
if method == "sign":
|
||||
idx = (sign_index(vals[c]) + sign_index(vals[first])
|
||||
- sign_index(vals[second])) % 12
|
||||
lon = idx * 30.0
|
||||
else:
|
||||
lon = norm360(vals[c] + vals[first] - vals[second])
|
||||
vals[name] = lon # dostępny dla kolejnych Lots
|
||||
out.append({
|
||||
"name": name,
|
||||
"longitude": lon,
|
||||
"formula": f"{c} + {first} − {second}",
|
||||
})
|
||||
return out
|
||||
@@ -1,103 +0,0 @@
|
||||
"""Systemy zodiaku (LOG-04): tropikalny, syderyczny (ayanamsy), draconic + RA.
|
||||
|
||||
Wszystkie pozycje silnika są liczone **tropikalnie of-date** (kontrakt LOG-28).
|
||||
Zmiana zodiaku to — dla zodiaków ekliptycznych — jednolite przesunięcie długości:
|
||||
|
||||
długość_docelowa = (długość_tropikalna − offset) mod 360
|
||||
|
||||
gdzie offset to:
|
||||
- **syderyczny**: ayanamsa (kąt między tropikalnym a syderycznym punktem Barana),
|
||||
- **draconic**: długość wznoszącego węzła Księżyca (węzeł = 0° draconic),
|
||||
- **tropikalny**: 0.
|
||||
|
||||
Ponieważ to stałe przesunięcie obiektu ORAZ cusps, **numery domów się nie zmieniają**
|
||||
(geometria jest niezmiennicza względem obrotu) — przesuwamy tylko etykiety znaków.
|
||||
|
||||
Model ayanamsy: `ayan(jd) = ayan0 + B·x + C·x²`, gdzie `x = jd − J2000`. Prędkość
|
||||
precesji (B, C) jest **wspólna** dla wszystkich ayanams; różni je tylko stała `ayan0`
|
||||
(wybór syderycznego zera). Stałe skalibrowano do Swiss Ephemeris jako wyroczni —
|
||||
zgodność do ~0,02" w latach 1900–2100 (patrz tests/test_zodiac.py).
|
||||
|
||||
RA (right ascension): konwersja ekliptyka→równik dla przyszłego widoku równikowego.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from datetime import datetime
|
||||
|
||||
from app.engine.formats import norm360
|
||||
|
||||
TROPICAL = "tropical"
|
||||
DRACONIC = "draconic"
|
||||
|
||||
# stała ayanamsy w J2000.0 (°) — skalibrowana do swisseph (get_ayanamsa_ut)
|
||||
_AYAN0 = {
|
||||
"lahiri": 23.857092,
|
||||
"fagan_bradley": 24.740300,
|
||||
"krishnamurti": 23.760240,
|
||||
}
|
||||
_J2000 = 2451545.0
|
||||
_B = 3.824459e-5 # °/dobę — liniowy człon precesji (wspólny)
|
||||
_C = 2.304e-13 # °/dobę² — drobne przyspieszenie (wspólne)
|
||||
|
||||
# nazwy zodiaków akceptowane przez API
|
||||
SIDEREAL = tuple(f"sidereal_{k}" for k in _AYAN0) # sidereal_lahiri, ...
|
||||
SYSTEMS = (TROPICAL, *SIDEREAL, DRACONIC)
|
||||
|
||||
|
||||
def julian_day(dt: datetime) -> float:
|
||||
"""Julian Day (UT) z momentu UTC — algorytm Meeusa (kalendarz gregoriański)."""
|
||||
y, m = dt.year, dt.month
|
||||
day = dt.day + (dt.hour + dt.minute / 60.0 + dt.second / 3600.0
|
||||
+ dt.microsecond / 3.6e9) / 24.0
|
||||
if m <= 2:
|
||||
y -= 1
|
||||
m += 12
|
||||
a = y // 100
|
||||
b = 2 - a + a // 4
|
||||
return math.floor(365.25 * (y + 4716)) + math.floor(30.6001 * (m + 1)) + day + b - 1524.5
|
||||
|
||||
|
||||
def ayanamsha(name: str, jd: float) -> float:
|
||||
"""Ayanamsa [°] danej szkoły dla Julian Day (UT)."""
|
||||
key = name[len("sidereal_"):] if name.startswith("sidereal_") else name
|
||||
if key not in _AYAN0:
|
||||
raise ValueError(f"Nieznana ayanamsa: {name!r} (dostępne: {', '.join(_AYAN0)})")
|
||||
x = jd - _J2000
|
||||
return _AYAN0[key] + _B * x + _C * x * x
|
||||
|
||||
|
||||
def offset(zodiac: str, jd: float, node_lon: float | None = None) -> float:
|
||||
"""Ile odjąć od długości tropikalnej, by dostać wybrany zodiak.
|
||||
|
||||
`node_lon` (tropikalna długość węzła wznoszącego) wymagana tylko dla draconic.
|
||||
"""
|
||||
if zodiac == TROPICAL:
|
||||
return 0.0
|
||||
if zodiac == DRACONIC:
|
||||
if node_lon is None:
|
||||
raise ValueError("draconic wymaga długości węzła (node_lon)")
|
||||
return norm360(node_lon)
|
||||
if zodiac in SIDEREAL:
|
||||
return ayanamsha(zodiac, jd)
|
||||
raise ValueError(f"Nieznany zodiak: {zodiac!r} (dostępne: {', '.join(SYSTEMS)})")
|
||||
|
||||
|
||||
def apply(lon: float, off: float) -> float:
|
||||
"""Długość w docelowym zodiaku."""
|
||||
return norm360(lon - off)
|
||||
|
||||
|
||||
def to_equatorial(lon: float, lat: float, eps: float) -> tuple[float, float]:
|
||||
"""Ekliptyka (λ, β) → równik: (RA, deklinacja) w stopniach. Wszystko w °.
|
||||
|
||||
RA rośnie 0–360°; deklinacja w [−90, 90].
|
||||
"""
|
||||
lam, bet, e = math.radians(lon), math.radians(lat), math.radians(eps)
|
||||
sin_dec = math.sin(bet) * math.cos(e) + math.cos(bet) * math.sin(e) * math.sin(lam)
|
||||
dec = math.asin(max(-1.0, min(1.0, sin_dec)))
|
||||
ra = math.atan2(
|
||||
math.sin(lam) * math.cos(e) - math.tan(bet) * math.sin(e),
|
||||
math.cos(lam),
|
||||
)
|
||||
return norm360(math.degrees(ra)), math.degrees(dec)
|
||||
@@ -1 +0,0 @@
|
||||
"""Warstwa dostawców modeli językowych (LOG-31)."""
|
||||
@@ -1,42 +0,0 @@
|
||||
"""Kontrakt dostawcy modelu językowego (LOG-31).
|
||||
|
||||
Analogicznie do `EphemerisEngine` (LOG-24): prezentacja i reszta logiki nie wiedzą,
|
||||
kto pisze tekst — lokalny model na naszym sprzęcie czy dostawca w chmurze.
|
||||
|
||||
Kluczowa własność dla bezpieczeństwa baz (LOG-32): każdy dostawca deklaruje
|
||||
`leaves_lan`. Prompt niesie ORYGINALNE opisy z baz, więc interfejs musi jawnie
|
||||
mówić, czy ta treść opuszcza naszą sieć — UI ma na tej podstawie ostrzegać.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass
|
||||
class Completion:
|
||||
"""Wynik generowania — tekst + metryki do rozliczenia i podglądu."""
|
||||
|
||||
text: str
|
||||
model: str
|
||||
provider: str
|
||||
leaves_lan: bool
|
||||
usage: dict = field(default_factory=dict) # prompt_tokens / completion_tokens
|
||||
|
||||
|
||||
class LLMError(RuntimeError):
|
||||
"""Błąd wołania modelu — z komunikatem nadającym się do pokazania użytkownikowi."""
|
||||
|
||||
|
||||
class LLMProvider(ABC):
|
||||
name: str = "?"
|
||||
#: czy treść promptu (a więc opisy z baz) opuszcza naszą sieć
|
||||
leaves_lan: bool = True
|
||||
|
||||
@abstractmethod
|
||||
def generate(self, prompt: str, max_tokens: int) -> Completion:
|
||||
"""Zwraca gotowy tekst. Rzuca LLMError przy niepowodzeniu."""
|
||||
|
||||
@abstractmethod
|
||||
def health(self) -> dict:
|
||||
"""Czy dostawca jest osiągalny i skonfigurowany."""
|
||||
@@ -1,102 +0,0 @@
|
||||
"""Wybór dostawcy LLM (LOG-31) — jedyne miejsce znające konkretne implementacje.
|
||||
|
||||
Domyślny jest **model lokalny**: prompt niesie oryginalne opisy z baz, więc
|
||||
domyślnie nic nie opuszcza naszej sieci (LOG-32). Chmurę włącza się świadomie —
|
||||
przez konfigurację albo pojedyncze żądanie.
|
||||
|
||||
Konfiguracja jest **per dostawca**, bo UI pozwala przełączać go przy każdym żądaniu.
|
||||
Wspólne `LLM_*` nie wystarczy: ustawienie `LLM_BASE_URL` na lokalny model kierowałoby
|
||||
tam także żądania do OpenAI, a `LLM_MODEL=llama3.1:8b` kazałoby Anthropic użyć modelu
|
||||
llama. Dlatego każdy dostawca ma własny komplet zmiennych.
|
||||
|
||||
Zmienne środowiskowe:
|
||||
LLM_PROVIDER local (domyślnie) | openai | anthropic — dostawca domyślny
|
||||
LLM_TIMEOUT sekundy (domyślnie 120)
|
||||
LLM_MAX_TOKENS limit długości odpowiedzi (domyślnie 2000)
|
||||
|
||||
<DOSTAWCA>_MODEL / _BASE_URL / _API_KEY — konfiguracja konkretnego dostawcy:
|
||||
LOCAL_MODEL, LOCAL_BASE_URL (klucz zwykle zbędny)
|
||||
OPENAI_MODEL, OPENAI_BASE_URL, OPENAI_API_KEY
|
||||
ANTHROPIC_MODEL, ANTHROPIC_BASE_URL, ANTHROPIC_API_KEY
|
||||
|
||||
Klucze WYŁĄCZNIE z sekretu — nigdy w repo, w UI ani w logach.
|
||||
|
||||
Zgodność wstecz: wspólne `LLM_MODEL` / `LLM_BASE_URL` / `LLM_API_KEY` nadal działają,
|
||||
ale stosują się TYLKO do dostawcy domyślnego (LLM_PROVIDER) — czyli konfiguracja
|
||||
instalacji jednodostawcowej zostaje nietknięta, a pozostali dostawcy jej nie dziedziczą.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from app.llm.base import LLMError, LLMProvider
|
||||
from app.llm.providers import AnthropicProvider, ChatCompletionsProvider
|
||||
|
||||
LOCAL = "local"
|
||||
OPENAI = "openai"
|
||||
ANTHROPIC = "anthropic"
|
||||
PROVIDERS = (LOCAL, OPENAI, ANTHROPIC)
|
||||
|
||||
_DEFAULT_MODEL = {
|
||||
LOCAL: "llama3.1:8b",
|
||||
OPENAI: "gpt-4o-mini",
|
||||
ANTHROPIC: "claude-sonnet-5",
|
||||
}
|
||||
_DEFAULT_URL = {
|
||||
# Ollama i vLLM wystawiają zgodne API pod /v1
|
||||
LOCAL: "http://localhost:11434/v1",
|
||||
OPENAI: "https://api.openai.com/v1",
|
||||
ANTHROPIC: "https://api.anthropic.com",
|
||||
}
|
||||
|
||||
|
||||
def default_provider_name() -> str:
|
||||
return os.getenv("LLM_PROVIDER", LOCAL).lower()
|
||||
|
||||
|
||||
def max_tokens() -> int:
|
||||
return int(os.getenv("LLM_MAX_TOKENS", "2000"))
|
||||
|
||||
|
||||
def timeout() -> float:
|
||||
return float(os.getenv("LLM_TIMEOUT", "120"))
|
||||
|
||||
|
||||
def setting(provider: str, suffix: str, fallback: str = "") -> str:
|
||||
"""Ustawienie dostawcy: <DOSTAWCA>_<SUFIKS> → LLM_<SUFIKS> → wbudowana domyślna.
|
||||
|
||||
Wspólne `LLM_*` stosuje się WYŁĄCZNIE do dostawcy domyślnego — inaczej adres
|
||||
lokalnego modelu przejąłby żądania do chmury (i odwrotnie).
|
||||
"""
|
||||
specific = os.getenv(f"{provider.upper()}_{suffix}")
|
||||
if specific:
|
||||
return specific
|
||||
if provider == default_provider_name():
|
||||
generic = os.getenv(f"LLM_{suffix}")
|
||||
if generic:
|
||||
return generic
|
||||
return fallback
|
||||
|
||||
|
||||
def build_provider(name: str | None = None) -> LLMProvider:
|
||||
name = (name or default_provider_name()).lower()
|
||||
if name not in PROVIDERS:
|
||||
raise LLMError(f"Nieznany dostawca LLM: {name!r} (dostępne: {', '.join(PROVIDERS)})")
|
||||
|
||||
model = setting(name, "MODEL", _DEFAULT_MODEL[name])
|
||||
base_url = setting(name, "BASE_URL", _DEFAULT_URL[name])
|
||||
api_key = setting(name, "API_KEY")
|
||||
|
||||
if name in (OPENAI, ANTHROPIC) and not api_key:
|
||||
raise LLMError(
|
||||
f"Brak klucza dla dostawcy {name} — ustaw {name.upper()}_API_KEY "
|
||||
f"(z sekretu). Model lokalny klucza nie wymaga."
|
||||
)
|
||||
if name == ANTHROPIC:
|
||||
return AnthropicProvider(base_url, model, api_key, timeout())
|
||||
if name == OPENAI:
|
||||
return ChatCompletionsProvider(OPENAI, base_url, model, api_key, timeout(),
|
||||
leaves_lan=True)
|
||||
# lokalny — klucz zwykle zbędny; treść NIE opuszcza sieci
|
||||
return ChatCompletionsProvider(LOCAL, base_url, model, api_key, timeout(),
|
||||
leaves_lan=False)
|
||||
@@ -1,143 +0,0 @@
|
||||
"""Implementacje dostawców LLM (LOG-31) — na samym httpx, bez SDK.
|
||||
|
||||
Świadomie bez bibliotek `openai` / `anthropic`: lokalny serwer modelu (Ollama,
|
||||
vLLM, llama.cpp) i OpenAI mówią **tym samym** protokołem `/chat/completions`,
|
||||
więc jedna implementacja obsługuje oba — różni je tylko adres i klucz. Anthropic
|
||||
ma własny kształt `/v1/messages`, stąd druga klasa. Mniej zależności, mniej
|
||||
powierzchni ataku, pełna kontrola nad tym, co wychodzi z sieci.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import httpx
|
||||
|
||||
from app.llm.base import Completion, LLMError, LLMProvider
|
||||
|
||||
RETRY_STATUSES = {429, 500, 502, 503, 504}
|
||||
MAX_ATTEMPTS = 3
|
||||
|
||||
|
||||
def _post_with_retry(url: str, headers: dict, payload: dict, timeout: float) -> dict:
|
||||
"""POST z ponawianiem i backoffem — chroni przed chwilowym 429/5xx."""
|
||||
last: Exception | None = None
|
||||
for attempt in range(MAX_ATTEMPTS):
|
||||
try:
|
||||
with httpx.Client(timeout=timeout) as client:
|
||||
r = client.post(url, json=payload, headers=headers)
|
||||
if r.status_code in RETRY_STATUSES and attempt < MAX_ATTEMPTS - 1:
|
||||
time.sleep(2 ** attempt)
|
||||
continue
|
||||
if r.status_code >= 400:
|
||||
raise LLMError(f"Model odpowiedział błędem {r.status_code}: {r.text[:300]}")
|
||||
return r.json()
|
||||
except httpx.TimeoutException as e:
|
||||
last = e
|
||||
if attempt < MAX_ATTEMPTS - 1:
|
||||
time.sleep(2 ** attempt)
|
||||
continue
|
||||
raise LLMError(
|
||||
f"Model nie odpowiedział w czasie {timeout:.0f}s. Dłuższe horoskopy "
|
||||
f"wymagają większego LLM_TIMEOUT albo mniejszego budżetu promptu."
|
||||
) from e
|
||||
except httpx.HTTPError as e:
|
||||
last = e
|
||||
if attempt < MAX_ATTEMPTS - 1:
|
||||
time.sleep(2 ** attempt)
|
||||
continue
|
||||
raise LLMError(f"Nie udało się połączyć z modelem: {e}") from e
|
||||
raise LLMError(f"Nie udało się wywołać modelu: {last}")
|
||||
|
||||
|
||||
class ChatCompletionsProvider(LLMProvider):
|
||||
"""Protokół OpenAI `/chat/completions` — lokalny serwer modelu ORAZ OpenAI."""
|
||||
|
||||
def __init__(self, name: str, base_url: str, model: str, api_key: str = "",
|
||||
timeout: float = 120.0, leaves_lan: bool = True) -> None:
|
||||
self.name = name
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.model = model
|
||||
self.api_key = api_key
|
||||
self.timeout = timeout
|
||||
self.leaves_lan = leaves_lan
|
||||
|
||||
def _headers(self) -> dict:
|
||||
h = {"Content-Type": "application/json"}
|
||||
if self.api_key:
|
||||
h["Authorization"] = f"Bearer {self.api_key}"
|
||||
return h
|
||||
|
||||
def generate(self, prompt: str, max_tokens: int) -> Completion:
|
||||
data = _post_with_retry(
|
||||
f"{self.base_url}/chat/completions", self._headers(),
|
||||
{
|
||||
"model": self.model,
|
||||
"max_tokens": max_tokens,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
},
|
||||
self.timeout,
|
||||
)
|
||||
try:
|
||||
text = data["choices"][0]["message"]["content"]
|
||||
except (KeyError, IndexError, TypeError) as e:
|
||||
raise LLMError(f"Nieoczekiwany kształt odpowiedzi modelu: {str(data)[:300]}") from e
|
||||
return Completion(
|
||||
text=text, model=data.get("model", self.model), provider=self.name,
|
||||
leaves_lan=self.leaves_lan, usage=data.get("usage") or {},
|
||||
)
|
||||
|
||||
def health(self) -> dict:
|
||||
info = {"provider": self.name, "model": self.model, "leaves_lan": self.leaves_lan}
|
||||
try:
|
||||
with httpx.Client(timeout=min(self.timeout, 10.0)) as client:
|
||||
r = client.get(f"{self.base_url}/models", headers=self._headers())
|
||||
info["status"] = "ok" if r.status_code < 400 else f"http {r.status_code}"
|
||||
except httpx.HTTPError as e:
|
||||
info["status"] = f"down: {e}"
|
||||
return info
|
||||
|
||||
|
||||
class AnthropicProvider(LLMProvider):
|
||||
"""Protokół Anthropic `/v1/messages`."""
|
||||
|
||||
leaves_lan = True
|
||||
|
||||
def __init__(self, base_url: str, model: str, api_key: str = "",
|
||||
timeout: float = 120.0) -> None:
|
||||
self.name = "anthropic"
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.model = model
|
||||
self.api_key = api_key
|
||||
self.timeout = timeout
|
||||
|
||||
def generate(self, prompt: str, max_tokens: int) -> Completion:
|
||||
if not self.api_key:
|
||||
raise LLMError("Brak LLM_API_KEY — dostawca anthropic wymaga klucza.")
|
||||
data = _post_with_retry(
|
||||
f"{self.base_url}/v1/messages",
|
||||
{
|
||||
"Content-Type": "application/json",
|
||||
"x-api-key": self.api_key,
|
||||
"anthropic-version": "2023-06-01",
|
||||
},
|
||||
{
|
||||
"model": self.model,
|
||||
"max_tokens": max_tokens,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
},
|
||||
self.timeout,
|
||||
)
|
||||
try:
|
||||
text = "".join(b.get("text", "") for b in data["content"] if b.get("type") == "text")
|
||||
except (KeyError, TypeError) as e:
|
||||
raise LLMError(f"Nieoczekiwany kształt odpowiedzi modelu: {str(data)[:300]}") from e
|
||||
return Completion(
|
||||
text=text, model=data.get("model", self.model), provider=self.name,
|
||||
leaves_lan=True, usage=data.get("usage") or {},
|
||||
)
|
||||
|
||||
def health(self) -> dict:
|
||||
return {
|
||||
"provider": self.name, "model": self.model, "leaves_lan": True,
|
||||
"status": "ok (klucz ustawiony)" if self.api_key else "brak LLM_API_KEY",
|
||||
}
|
||||
+1
-135
@@ -12,14 +12,12 @@ import httpx
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
from app import security
|
||||
from app.clients.data_client import DataClient
|
||||
from app.models import QueryRequest, QueryResponse
|
||||
from app.service import QueryService
|
||||
|
||||
app = FastAPI(title="astrololo · warstwa logiczna")
|
||||
service = QueryService()
|
||||
security.install(app, "logiczna") # token międzywarstwowy (LOG-32)
|
||||
|
||||
# --- silnik efemeryd (LOG-24): budowany leniwie, by nie wymagać Skyfielda do startu ---
|
||||
_engine = None
|
||||
@@ -41,7 +39,6 @@ class PositionsRequest(BaseModel):
|
||||
objects: list[str] | None = None
|
||||
house_system: str = "whole_sign" # whole_sign | equal | porphyry
|
||||
stations: bool = False # licz stacje (LOG-03; wolniejsze — root-findy)
|
||||
zodiac: str = "tropical" # LOG-04: tropical | sidereal_{lahiri,fagan_bradley,krishnamurti} | draconic
|
||||
|
||||
|
||||
@app.post("/api/query", response_model=QueryResponse)
|
||||
@@ -61,10 +58,7 @@ def chart_positions(req: PositionsRequest) -> dict:
|
||||
|
||||
engine = get_engine()
|
||||
moment = ChartMoment(when_utc=req.when_utc, lat=req.lat, lon=req.lon)
|
||||
try:
|
||||
chart = build_chart(engine, moment, req.house_system, zodiac=req.zodiac)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=422, detail=str(e))
|
||||
chart = build_chart(engine, moment, req.house_system)
|
||||
if req.stations:
|
||||
from app.engine.stations import find_stations
|
||||
|
||||
@@ -123,134 +117,6 @@ def chart_report(req: ReportRequest) -> dict:
|
||||
return {"engine": engine.name, **report}
|
||||
|
||||
|
||||
class PromptRequest(BaseModel):
|
||||
"""Wejście generatora promptu (LOG-29/30)."""
|
||||
profile: str = "natal" # natal | period
|
||||
when_utc: datetime
|
||||
lat: float = 0.0
|
||||
lon: float = 0.0
|
||||
budget: str = "medium" # concise | medium | extensive
|
||||
limit: int = 5000
|
||||
# tylko dla profilu period:
|
||||
from_date: str | None = None
|
||||
to_date: str | None = None
|
||||
techniques: list[str] | None = None
|
||||
|
||||
|
||||
@app.post("/chart/prompt")
|
||||
def chart_prompt(req: PromptRequest) -> dict:
|
||||
"""Gotowy prompt do LLM z naszych wyliczeń (LOG-29) z budżetowaniem (LOG-30).
|
||||
|
||||
profile=natal → horoskop urodzeniowy (ekran Interpretacje)
|
||||
profile=period → horoskop na wybrany okres (ekran Kalendarz)
|
||||
|
||||
Nie woła żadnego modelu — zwraca sam prompt i statystyki redukcji, żeby dało się
|
||||
go obejrzeć i skopiować. Wysyłkę do modelu doda LOG-31.
|
||||
"""
|
||||
from app.engine.chart import build_chart
|
||||
from app.engine.models import ChartMoment
|
||||
from app.prompt import build_natal_prompt, build_period_prompt
|
||||
|
||||
engine = get_engine()
|
||||
moment = ChartMoment(when_utc=req.when_utc, lat=req.lat, lon=req.lon)
|
||||
chart = build_chart(engine, moment)
|
||||
label = req.when_utc.strftime("%Y-%m-%d %H:%M UTC")
|
||||
data_error = None
|
||||
|
||||
# Warstwa danych dokłada wyłącznie WSKAZANIA. Wyliczenia (horoskop, oś czasu) są
|
||||
# od niej niezależne — gdy padnie, prompt musi zachować wszystko, co policzyliśmy.
|
||||
try:
|
||||
if req.profile == "natal":
|
||||
from app.significators import build_report
|
||||
|
||||
report: dict = {"objects": []}
|
||||
try:
|
||||
report = build_report(
|
||||
chart["positions"], DataClient(),
|
||||
aspects=chart.get("aspects"), per_object_limit=req.limit,
|
||||
)
|
||||
except httpx.HTTPError as e:
|
||||
data_error = f"Warstwa danych niedostępna: {e}"
|
||||
out = build_natal_prompt(chart, report, req.budget, label)
|
||||
|
||||
elif req.profile == "period":
|
||||
if not (req.from_date and req.to_date):
|
||||
raise HTTPException(422, "profile=period wymaga from_date i to_date")
|
||||
from app.engine import houses as H
|
||||
from app.engine.timeline import build_timeline
|
||||
from app.significators import interpret_events
|
||||
|
||||
ramc, eps = engine.sidereal(moment)
|
||||
points = {"Asc": H.compute_asc(ramc, eps, moment.lat), "MC": H.compute_mc(ramc, eps)}
|
||||
for p in engine.positions(moment):
|
||||
points[p.name] = p.longitude
|
||||
events = build_timeline(engine, moment, points, req.from_date, req.to_date,
|
||||
req.techniques)
|
||||
try:
|
||||
interpret_events(events, DataClient())
|
||||
except httpx.HTTPError as e:
|
||||
data_error = f"Warstwa danych niedostępna: {e}" # oś czasu zostaje
|
||||
out = build_period_prompt(chart, events, req.from_date, req.to_date,
|
||||
req.budget, label)
|
||||
else:
|
||||
raise HTTPException(422, f"Nieznany profil: {req.profile!r} (natal | period)")
|
||||
except ValueError as e: # nieznany budżet
|
||||
raise HTTPException(422, str(e))
|
||||
|
||||
out["engine"] = engine.name
|
||||
if data_error:
|
||||
out["data_error"] = data_error
|
||||
return out
|
||||
|
||||
|
||||
class HoroscopeRequest(PromptRequest):
|
||||
"""Jak PromptRequest + wybór dostawcy modelu (LOG-31)."""
|
||||
provider: str | None = None # local (dom.) | openai | anthropic
|
||||
max_tokens: int | None = None
|
||||
|
||||
|
||||
@app.post("/chart/horoscope")
|
||||
def chart_horoscope(req: HoroscopeRequest) -> dict:
|
||||
"""Napisany horoskop (LOG-31): prompt z LOG-29/30 → model → gotowy tekst.
|
||||
|
||||
Zwraca TAKŻE użyty prompt — również gdy wywołanie modelu padnie — żeby dało się
|
||||
go obejrzeć i użyć ręcznie. `leaves_lan` mówi, czy treść baz opuściła naszą sieć
|
||||
(LOG-32); prezentacja ma na tej podstawie ostrzegać.
|
||||
"""
|
||||
from app.llm.base import LLMError
|
||||
from app.llm.factory import build_provider, max_tokens
|
||||
|
||||
out = chart_prompt(req) # ten sam prompt co w podglądzie
|
||||
try:
|
||||
provider = build_provider(req.provider)
|
||||
result = provider.generate(out["prompt"], req.max_tokens or max_tokens())
|
||||
except LLMError as e:
|
||||
# prompt zostaje — użytkownik moze go skopiowac i uzyc recznie
|
||||
out["llm_error"] = str(e)
|
||||
return out
|
||||
|
||||
out.update(
|
||||
horoscope=result.text,
|
||||
provider=result.provider,
|
||||
model=result.model,
|
||||
leaves_lan=result.leaves_lan,
|
||||
usage=result.usage,
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
@app.get("/llm/health")
|
||||
def llm_health(provider: str | None = None) -> dict:
|
||||
"""Czy model jest osiągalny i skonfigurowany (bez generowania czegokolwiek)."""
|
||||
from app.llm.base import LLMError
|
||||
from app.llm.factory import build_provider
|
||||
|
||||
try:
|
||||
return build_provider(provider).health()
|
||||
except LLMError as e:
|
||||
return {"status": f"blad konfiguracji: {e}"}
|
||||
|
||||
|
||||
class ProfectionsRequest(BaseModel):
|
||||
when_utc: datetime # moment urodzenia (UTC)
|
||||
lat: float = 0.0
|
||||
|
||||
@@ -1,339 +0,0 @@
|
||||
"""Generator promptów do LLM (LOG-29) + budżetowanie rozmiaru (LOG-30).
|
||||
|
||||
Składa z naszych wyliczeń gotowe, profesjonalne polecenie po polsku:
|
||||
- profil **natal** — horoskop urodzeniowy (na bazie /chart/report),
|
||||
- profil **period** — horoskop na wybrany okres (na bazie /chart/timeline).
|
||||
|
||||
Zasada naczelna promptu: model ma pisać WYŁĄCZNIE na podstawie dostarczonych danych
|
||||
i przy każdej tezie wskazać konkretny sygnifikator, z którego ona wynika. To odróżnia
|
||||
wynik od ogólnikowej wróżby i pozwala go zweryfikować.
|
||||
|
||||
Budżetowanie (LOG-30) — kolejność redukcji:
|
||||
1. deduplikacja (ten sam sygnifikator ORAZ ten sam opis),
|
||||
2. grupowanie identycznych opisów z licznikiem wystąpień,
|
||||
3. sortowanie malejąco wg punktacji siły (LOG-21),
|
||||
4. obcięcie ogona do budżetu — jednostką obcięcia jest CAŁE wskazanie,
|
||||
5. skracanie nadmiernie długich opisów (z jawnym oznaczeniem).
|
||||
Zawsze raportujemy, ile wskazań weszło i ile pominięto — użytkownik ma wiedzieć,
|
||||
czego brakuje, i móc zwiększyć budżet.
|
||||
|
||||
Generowanie jest deterministyczne: ten sam horoskop + ten sam budżet = ten sam prompt.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
# budżety w znakach (całego promptu); dobrane pod wklejanie do ChatGPT/Claude
|
||||
BUDGETS: dict[str, int] = {
|
||||
"concise": 4_000,
|
||||
"medium": 12_000,
|
||||
"extensive": 30_000,
|
||||
}
|
||||
DEFAULT_BUDGET = "medium"
|
||||
|
||||
MAX_EFFECT_CHARS = 320 # dłuższe opisy skracamy (krok 5 redukcji)
|
||||
CHARS_PER_TOKEN = 4.0 # zgrubny szacunek tokenów do podglądu w UI
|
||||
|
||||
DISCLAIMER = (
|
||||
"Zakończ krótką notą: treść jest interpretacją astrologiczną i nie stanowi porady "
|
||||
"medycznej, prawnej ani finansowej."
|
||||
)
|
||||
|
||||
|
||||
def _norm(text: str) -> str:
|
||||
return re.sub(r"\s+", " ", (text or "").strip().lower())
|
||||
|
||||
|
||||
def est_tokens(text: str) -> int:
|
||||
return int(len(text) / CHARS_PER_TOKEN)
|
||||
|
||||
|
||||
def _shorten(text: str, limit: int = MAX_EFFECT_CHARS) -> tuple[str, bool]:
|
||||
text = (text or "").strip()
|
||||
if len(text) <= limit:
|
||||
return text, False
|
||||
return text[:limit].rstrip() + " […skrócono]", True
|
||||
|
||||
|
||||
# ----------------------------------------------------------------- wskazania
|
||||
|
||||
def natal_indications(report: dict) -> list[dict]:
|
||||
"""Płaska lista wskazań z raportu natalnego: obiekt + faseta + opis + waga."""
|
||||
out: list[dict] = []
|
||||
for obj in report.get("objects") or []:
|
||||
name = obj.get("object")
|
||||
for facet in obj.get("facets") or []:
|
||||
score = float(facet.get("score") or 0.0)
|
||||
for s in facet.get("samples") or []:
|
||||
out.append({
|
||||
"context": f"{name} — {facet.get('label')}",
|
||||
"sort_key": (name or "", str(facet.get("label") or "")),
|
||||
"score": score,
|
||||
"significator": s.get("expanded") or s.get("significator") or "",
|
||||
"effect": s.get("effect") or "",
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def period_indications(events: list[dict]) -> list[dict]:
|
||||
"""Wskazania z osi czasu — waga rośnie z liczbą trafień w bazie."""
|
||||
out: list[dict] = []
|
||||
for ev in events or []:
|
||||
head = f"{ev.get('technique')} · {ev.get('significator')} · {ev.get('exact')}"
|
||||
total = float(ev.get("interpretations_count") or 0)
|
||||
for s in ev.get("interpretations") or []:
|
||||
out.append({
|
||||
"context": head,
|
||||
"sort_key": (str(ev.get("exact") or ""), str(ev.get("technique") or "")),
|
||||
"score": total,
|
||||
"significator": s.get("expanded") or s.get("significator") or "",
|
||||
"effect": s.get("effect") or "",
|
||||
"date": ev.get("exact"),
|
||||
"start": ev.get("start"),
|
||||
"end": ev.get("end"),
|
||||
"technique": ev.get("technique"),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def reduce_indications(items: list[dict], budget_chars: int) -> tuple[list[dict], dict]:
|
||||
"""Dedup → grupowanie → sortowanie wg wagi → obcięcie do budżetu.
|
||||
|
||||
Zwraca (wybrane wskazania, statystyki). Jednostką obcięcia jest całe wskazanie.
|
||||
"""
|
||||
# 1+2: dedup i grupowanie identycznych (kontekst, opis) z licznikiem
|
||||
grouped: dict[tuple[str, str], dict] = {}
|
||||
order: list[tuple[str, str]] = []
|
||||
for it in items:
|
||||
key = (_norm(it["context"]), _norm(it["effect"]))
|
||||
g = grouped.get(key)
|
||||
if g is None:
|
||||
g = {**it, "count": 0, "significators": []}
|
||||
grouped[key] = g
|
||||
order.append(key)
|
||||
g["count"] += 1
|
||||
if it["significator"] not in g["significators"]:
|
||||
g["significators"].append(it["significator"])
|
||||
merged = [grouped[k] for k in order]
|
||||
deduped = len(items) - len(merged)
|
||||
|
||||
# 3: sortowanie malejąco wg wagi; remis rozstrzygany deterministycznie
|
||||
merged.sort(key=lambda g: (-g["score"], -g["count"], g["sort_key"], g["significator"]))
|
||||
|
||||
# 5 (przed obcięciem, bo wpływa na rozmiar): skracanie długich opisów
|
||||
shortened = 0
|
||||
for g in merged:
|
||||
g["effect"], was = _shorten(g["effect"])
|
||||
shortened += 1 if was else 0
|
||||
|
||||
# 4: obcięcie ogona do budżetu — cała pozycja albo nic
|
||||
chosen: list[dict] = []
|
||||
used = 0
|
||||
for g in merged:
|
||||
cost = len(_render_indication(g)) + 1
|
||||
if used + cost > budget_chars and chosen:
|
||||
break
|
||||
chosen.append(g)
|
||||
used += cost
|
||||
|
||||
omitted = len(merged) - len(chosen)
|
||||
stats = {
|
||||
"source_rows": len(items),
|
||||
"after_grouping": len(merged),
|
||||
"deduplicated": deduped,
|
||||
"included": len(chosen),
|
||||
"omitted": omitted,
|
||||
"shortened": shortened,
|
||||
"min_score_included": round(chosen[-1]["score"], 3) if chosen else None,
|
||||
"max_score_omitted": round(merged[len(chosen)]["score"], 3) if omitted else None,
|
||||
}
|
||||
return chosen, stats
|
||||
|
||||
|
||||
def _render_indication(g: dict) -> str:
|
||||
times = f" (×{g['count']})" if g.get("count", 1) > 1 else ""
|
||||
sig = g.get("significator") or ""
|
||||
return f"- {sig} → {g.get('effect')}{times}"
|
||||
|
||||
|
||||
def _render_grouped(chosen: list[dict]) -> str:
|
||||
"""Renderuje wskazania pogrupowane po kontekście, zachowując kolejność wagi."""
|
||||
blocks: list[str] = []
|
||||
seen: dict[str, list[dict]] = {}
|
||||
order: list[str] = []
|
||||
for g in chosen:
|
||||
ctx = g["context"]
|
||||
if ctx not in seen:
|
||||
seen[ctx] = []
|
||||
order.append(ctx)
|
||||
seen[ctx].append(g)
|
||||
for ctx in order:
|
||||
rows = seen[ctx]
|
||||
blocks.append(f"## {ctx} [waga {rows[0]['score']:.2f}]")
|
||||
blocks.extend(_render_indication(g) for g in rows)
|
||||
blocks.append("")
|
||||
return "\n".join(blocks).rstrip()
|
||||
|
||||
|
||||
# ------------------------------------------------------------- sekcje danych
|
||||
|
||||
def _chart_section(chart: dict, moment_label: str | None) -> str:
|
||||
lines: list[str] = ["# DANE HOROSKOPU"]
|
||||
meta = []
|
||||
if moment_label:
|
||||
meta.append(f"moment: {moment_label}")
|
||||
if chart.get("zodiac"):
|
||||
z = chart["zodiac"]
|
||||
if chart.get("ayanamsha") is not None:
|
||||
z += f" (ayanamsa {chart['ayanamsha']:.4f}°)"
|
||||
meta.append(f"zodiak: {z}")
|
||||
if chart.get("house_system"):
|
||||
meta.append(f"system domów: {chart['house_system']}")
|
||||
if chart.get("sect"):
|
||||
meta.append(f"sekta: {'dzienna' if chart['sect'] == 'day' else 'nocna'}")
|
||||
if meta:
|
||||
lines.append(" · ".join(meta))
|
||||
|
||||
if chart.get("positions"):
|
||||
lines.append("\n## Pozycje")
|
||||
for p in chart["positions"]:
|
||||
house = f"dom {p['house']}" if p.get("house") else "—"
|
||||
lines.append(f"{p['name']:<12} {p.get('in_sign',''):<16} {house:<8} {p.get('direction','')}")
|
||||
|
||||
angles = chart.get("angles") or {}
|
||||
if angles:
|
||||
lines.append("\n## Osie")
|
||||
for key in ("Asc", "MC", "Dsc", "IC"):
|
||||
a = angles.get(key)
|
||||
if a:
|
||||
lines.append(f"{a['name']:<5} {a.get('in_sign','')}")
|
||||
|
||||
if chart.get("lots"):
|
||||
lines.append("\n## Lots (punkty arabskie)")
|
||||
for lot in chart["lots"]:
|
||||
lines.append(
|
||||
f"{lot['name']:<10} {lot.get('in_sign',''):<16} dom {lot.get('house','—')}"
|
||||
f" ({lot.get('formula','')})"
|
||||
)
|
||||
|
||||
if chart.get("aspects"):
|
||||
lines.append("\n## Aspekty (orb; A = aplikacyjny, S = separacyjny)")
|
||||
for a in chart["aspects"]:
|
||||
mark = a.get("as") or ""
|
||||
lines.append(
|
||||
f"{a['obj1']} {a['aspect']} {a['obj2']} orb {a.get('orb', 0):.2f}° {mark}"
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _indications_section(chosen: list[dict], stats: dict) -> str:
|
||||
if not chosen:
|
||||
return (
|
||||
"# WSKAZANIA Z BAZ\n"
|
||||
"(brak trafień w bazach dla tego horoskopu — oprzyj interpretację wyłącznie "
|
||||
"na danych horoskopu powyżej)"
|
||||
)
|
||||
head = [
|
||||
"# WSKAZANIA Z BAZ INTERPRETACYJNYCH",
|
||||
"Wskazania dopasowane do tego horoskopu, uporządkowane od najsilniejszych.",
|
||||
"Zapis: `- sygnifikator → opis (×ile razy wystąpiło w bazach)`.",
|
||||
]
|
||||
if stats.get("omitted"):
|
||||
head.append(
|
||||
f"UWAGA: pokazano {stats['included']} najsilniejszych wskazań, pominięto "
|
||||
f"{stats['omitted']} słabszych (limit długości)."
|
||||
)
|
||||
return "\n".join(head) + "\n\n" + _render_grouped(chosen)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ prompty
|
||||
|
||||
_NATAL_TASK = """# ZADANIE
|
||||
Jesteś doświadczonym astrologiem. Napisz profesjonalną interpretację HOROSKOPU
|
||||
URODZENIOWEGO po polsku, opierając się wyłącznie na danych podanych niżej."""
|
||||
|
||||
_PERIOD_TASK = """# ZADANIE
|
||||
Jesteś doświadczonym astrologiem. Napisz profesjonalną prognozę astrologiczną
|
||||
NA WYBRANY OKRES po polsku, opierając się wyłącznie na danych podanych niżej."""
|
||||
|
||||
_NATAL_OUTPUT = """# JAK MA WYGLĄDAĆ ODPOWIEDŹ
|
||||
1. Struktura: (a) portret ogólny, (b) temperament i sekta, (c) obszary życia według domów,
|
||||
(d) napięcia i wyzwania, (e) zasoby i mocne strony, (f) zwięzłe podsumowanie.
|
||||
2. KAŻDĄ tezę oprzyj na konkretnym wskazaniu i podaj je w nawiasie, np. „(Moon in 12th house)".
|
||||
Teza bez wskazania jest niedopuszczalna.
|
||||
3. Nie dodawaj twierdzeń, których nie da się wywieść z powyższych danych. Nie zmyślaj
|
||||
pozycji, aspektów ani wskazań; nie korzystaj z wiedzy spoza tego promptu.
|
||||
4. Gdy wskazania są sprzeczne, powiedz to wprost i wskaż obie strony, zamiast wybierać jedną.
|
||||
5. Wagę wskazania traktuj jako siłę świadectwa — mocniejsze mają pierwszeństwo w syntezie.
|
||||
6. Ton rzeczowy i profesjonalny: bez wróżbiarstwa, bez straszenia, bez diagnoz medycznych."""
|
||||
|
||||
_PERIOD_OUTPUT = """# JAK MA WYGLĄDAĆ ODPOWIEDŹ
|
||||
1. Uporządkuj prognozę CHRONOLOGICZNIE; przy każdym okresie podaj daty (start / dokładna / koniec).
|
||||
2. Dla każdej daty napisz, która technika ją wyznacza (profekcja, solariusz, dyrekcja solar-arc,
|
||||
Firdaria) i co z niej wynika.
|
||||
3. KAŻDĄ tezę oprzyj na konkretnym wskazaniu i podaj je w nawiasie. Teza bez wskazania jest
|
||||
niedopuszczalna.
|
||||
4. Nie zmyślaj dat ani zdarzeń spoza podanych. Nie korzystaj z wiedzy spoza tego promptu.
|
||||
5. Rozróżniaj okresy o mocnym świadectwie (wysoka waga, kilka technik zbieżnych w czasie)
|
||||
od słabych — i powiedz wprost, które są które.
|
||||
6. Na końcu dodaj krótkie zestawienie: najważniejsze okresy w kolejności ważności.
|
||||
7. Ton rzeczowy i profesjonalny: bez wróżbiarstwa, bez straszenia, bez diagnoz medycznych."""
|
||||
|
||||
|
||||
def _assemble(task: str, chart_sec: str, ind_sec: str, output: str) -> str:
|
||||
return "\n\n".join([task, chart_sec, ind_sec, output, f"# ZASTRZEŻENIE\n{DISCLAIMER}"])
|
||||
|
||||
|
||||
def _budget_chars(budget: str) -> int:
|
||||
if budget not in BUDGETS:
|
||||
raise ValueError(
|
||||
f"Nieznany budżet: {budget!r} (dostępne: {', '.join(BUDGETS)})"
|
||||
)
|
||||
return BUDGETS[budget]
|
||||
|
||||
|
||||
def _finish(prompt: str, budget: str, limit: int, stats: dict, profile: str) -> dict:
|
||||
stats = {
|
||||
**stats,
|
||||
"profile": profile,
|
||||
"budget": budget,
|
||||
"limit_chars": limit,
|
||||
"chars": len(prompt),
|
||||
"est_tokens": est_tokens(prompt),
|
||||
}
|
||||
return {"profile": profile, "prompt": prompt, "stats": stats}
|
||||
|
||||
|
||||
def build_natal_prompt(chart: dict, report: dict, budget: str = DEFAULT_BUDGET,
|
||||
moment_label: str | None = None) -> dict:
|
||||
"""Prompt na horoskop urodzeniowy (ekran „Interpretacje")."""
|
||||
limit = _budget_chars(budget)
|
||||
chart_sec = _chart_section(chart, moment_label)
|
||||
fixed = len(_NATAL_TASK) + len(chart_sec) + len(_NATAL_OUTPUT) + len(DISCLAIMER) + 200
|
||||
chosen, stats = reduce_indications(natal_indications(report), max(limit - fixed, 500))
|
||||
prompt = _assemble(_NATAL_TASK, chart_sec, _indications_section(chosen, stats), _NATAL_OUTPUT)
|
||||
return _finish(prompt, budget, limit, stats, "natal")
|
||||
|
||||
|
||||
def build_period_prompt(chart: dict, events: list[dict], from_date: str, to_date: str,
|
||||
budget: str = DEFAULT_BUDGET, moment_label: str | None = None) -> dict:
|
||||
"""Prompt na horoskop okresowy (ekran „Kalendarz")."""
|
||||
limit = _budget_chars(budget)
|
||||
chart_sec = _chart_section(chart, moment_label)
|
||||
task = f"{_PERIOD_TASK}\nZakres prognozy: **{from_date} — {to_date}**."
|
||||
|
||||
ev_lines = ["# OŚ CZASU (techniki predykcyjne)",
|
||||
"Zapis: technika · sygnifikator · start → dokładna → koniec."]
|
||||
for ev in events or []:
|
||||
ev_lines.append(
|
||||
f"- {ev.get('technique')} · {ev.get('significator')} · "
|
||||
f"{ev.get('start')} → {ev.get('exact')} → {ev.get('end')}"
|
||||
)
|
||||
events_sec = "\n".join(ev_lines)
|
||||
|
||||
fixed = len(task) + len(chart_sec) + len(events_sec) + len(_PERIOD_OUTPUT) + len(DISCLAIMER) + 200
|
||||
chosen, stats = reduce_indications(period_indications(events), max(limit - fixed, 500))
|
||||
body = events_sec + "\n\n" + _indications_section(chosen, stats)
|
||||
prompt = _assemble(task, chart_sec, body, _PERIOD_OUTPUT)
|
||||
stats["events"] = len(events or [])
|
||||
return _finish(prompt, budget, limit, stats, "period")
|
||||
@@ -1,50 +0,0 @@
|
||||
"""Uwierzytelnianie międzywarstwowe (LOG-32).
|
||||
|
||||
Warstwa logiczna oddaje treść baz interpretacyjnych, więc samo zalogowanie w
|
||||
prezentacji nie wystarczy — bez tego kontrolera wystarczyłoby uderzyć w logikę
|
||||
z pominięciem UI. Gdy ustawiono INTERNAL_TOKEN, każde żądanie (poza /health)
|
||||
musi go przynieść w nagłówku X-Astrololo-Token.
|
||||
|
||||
Bez INTERNAL_TOKEN kontrola jest wyłączona (dev / zgodność wstecz) — wtedy przy
|
||||
starcie leci ostrzeżenie.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import secrets
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
log = logging.getLogger("astrololo.security")
|
||||
|
||||
HEADER = "X-Astrololo-Token"
|
||||
PUBLIC_PATHS = frozenset({"/health"})
|
||||
|
||||
|
||||
def token() -> str:
|
||||
"""Czytany leniwie — konfiguracja może się zmienić bez importu modułu."""
|
||||
return os.getenv("INTERNAL_TOKEN", "")
|
||||
|
||||
|
||||
def enabled() -> bool:
|
||||
return bool(token())
|
||||
|
||||
|
||||
def install(app, layer: str) -> None:
|
||||
if not enabled():
|
||||
log.warning(
|
||||
"UWAGA: INTERNAL_TOKEN nie ustawiony — warstwa %s przyjmuje żądania od "
|
||||
"kogokolwiek, kto ma do niej dostęp sieciowy.", layer,
|
||||
)
|
||||
|
||||
@app.middleware("http")
|
||||
async def _guard(request: Request, call_next):
|
||||
if request.url.path in PUBLIC_PATHS or not enabled():
|
||||
return await call_next(request)
|
||||
got = request.headers.get(HEADER, "")
|
||||
if not secrets.compare_digest(got, token()):
|
||||
return JSONResponse({"detail": "Brak lub błędny token międzywarstwowy."},
|
||||
status_code=401)
|
||||
return await call_next(request)
|
||||
@@ -22,8 +22,6 @@ PLANET_ABBR = {
|
||||
"Jupiter": "Ju", "Saturn": "Sa", "Uranus": "Ur", "Neptune": "Ne", "Pluto": "Pl",
|
||||
# punkty wirtualne — tokeny wg SIGNIFICATORS KEY ([NN, [SN, [Lilith)
|
||||
"North Node": "NN", "South Node": "SN", "Lilith": "Lilith",
|
||||
# Lots — w bazie Fortuna występuje jako [PF (Part of Fortune)
|
||||
"Fortune": "PF", "Spirit": "Spirit",
|
||||
}
|
||||
SIGN_TO_ABBR = dict(zip(SIGNS, SIGN_ABBR))
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""Testy aspektów (LOG-06) — czysta matematyka."""
|
||||
from app.engine.aspects import RIGID_PAIRS, find_aspects, separation
|
||||
from app.engine.aspects import find_aspects, separation
|
||||
|
||||
|
||||
def test_separation_wraparound():
|
||||
@@ -55,28 +55,6 @@ def test_separating_when_moving_apart():
|
||||
assert a["applying"] is False and a["as"] == "S"
|
||||
|
||||
|
||||
def test_rigid_pair_nodes_not_reported():
|
||||
# SN = NN + 180° z definicji — trywialna opozycja, nie aspekt
|
||||
pos = [
|
||||
{"name": "North Node", "decimal": 42.0, "speed": -0.053},
|
||||
{"name": "South Node", "decimal": 222.0, "speed": -0.053},
|
||||
{"name": "Sun", "decimal": 42.5, "speed": 0.96},
|
||||
]
|
||||
pairs = {frozenset((a["obj1"], a["obj2"])) for a in find_aspects(pos)}
|
||||
assert frozenset({"North Node", "South Node"}) not in pairs
|
||||
# aspekty węzłów do innych obiektów zostają nietknięte
|
||||
assert frozenset({"Sun", "North Node"}) in pairs
|
||||
assert frozenset({"Sun", "South Node"}) in pairs
|
||||
|
||||
|
||||
def test_no_rigid_pairs_in_real_chart(own_engine, reference_moment):
|
||||
from app.engine.chart import build_chart
|
||||
|
||||
chart = build_chart(own_engine, reference_moment)
|
||||
found = [a for a in chart["aspects"] if frozenset((a["obj1"], a["obj2"])) in RIGID_PAIRS]
|
||||
assert not found, f"trywialne aspekty par sztywnych w horoskopie: {found}"
|
||||
|
||||
|
||||
def test_no_as_flag_without_speeds():
|
||||
pos = [{"name": "Sun", "decimal": 0.0}, {"name": "Moon", "decimal": 2.0}]
|
||||
assert "as" not in find_aspects(pos)[0]
|
||||
|
||||
@@ -1,212 +0,0 @@
|
||||
"""Dostawcy LLM (LOG-31) — bez wołania jakiegokolwiek prawdziwego modelu.
|
||||
|
||||
Transport podstawiamy przez httpx.MockTransport, więc testy są szybkie,
|
||||
deterministyczne i nic nie wychodzi na zewnątrz.
|
||||
"""
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from app.llm import factory
|
||||
from app.llm.base import Completion, LLMError
|
||||
from app.llm.providers import AnthropicProvider, ChatCompletionsProvider
|
||||
|
||||
|
||||
def _mock_client(handler):
|
||||
"""Podmienia httpx.Client na wersję z transportem testowym."""
|
||||
class _C(httpx.Client):
|
||||
def __init__(self, *a, **kw):
|
||||
kw["transport"] = httpx.MockTransport(handler)
|
||||
super().__init__(*a, **kw)
|
||||
return _C
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def chat_ok(monkeypatch):
|
||||
def handler(request):
|
||||
assert request.url.path.endswith("/chat/completions")
|
||||
return httpx.Response(200, json={
|
||||
"model": "test-model",
|
||||
"choices": [{"message": {"content": "Horoskop testowy."}}],
|
||||
"usage": {"prompt_tokens": 100, "completion_tokens": 50},
|
||||
})
|
||||
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
|
||||
|
||||
|
||||
# ------------------------------------------------------- protokół chat/completions
|
||||
|
||||
def test_local_provider_generates(chat_ok):
|
||||
p = ChatCompletionsProvider("local", "http://localhost:11434/v1", "m", leaves_lan=False)
|
||||
out = p.generate("prompt", 500)
|
||||
assert isinstance(out, Completion)
|
||||
assert out.text == "Horoskop testowy."
|
||||
assert out.usage["completion_tokens"] == 50
|
||||
|
||||
|
||||
def test_local_provider_does_not_leave_lan(chat_ok):
|
||||
p = ChatCompletionsProvider("local", "http://localhost:11434/v1", "m", leaves_lan=False)
|
||||
assert p.generate("prompt", 100).leaves_lan is False
|
||||
|
||||
|
||||
def test_cloud_provider_marks_leaving_lan(chat_ok):
|
||||
p = ChatCompletionsProvider("openai", "https://api.openai.com/v1", "m", "klucz",
|
||||
leaves_lan=True)
|
||||
assert p.generate("prompt", 100).leaves_lan is True
|
||||
|
||||
|
||||
def test_api_key_sent_only_when_set(monkeypatch):
|
||||
seen = {}
|
||||
|
||||
def handler(request):
|
||||
seen["auth"] = request.headers.get("authorization")
|
||||
return httpx.Response(200, json={"choices": [{"message": {"content": "x"}}]})
|
||||
|
||||
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
|
||||
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10)
|
||||
assert seen["auth"] is None
|
||||
ChatCompletionsProvider("openai", "http://x/v1", "m", "tajny").generate("p", 10)
|
||||
assert seen["auth"] == "Bearer tajny"
|
||||
|
||||
|
||||
def test_http_error_becomes_readable_message(monkeypatch):
|
||||
monkeypatch.setattr(httpx, "Client",
|
||||
_mock_client(lambda r: httpx.Response(400, text="zly model")))
|
||||
with pytest.raises(LLMError, match="400"):
|
||||
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10)
|
||||
|
||||
|
||||
def test_malformed_response_reported(monkeypatch):
|
||||
monkeypatch.setattr(httpx, "Client",
|
||||
_mock_client(lambda r: httpx.Response(200, json={"nonsens": 1})))
|
||||
with pytest.raises(LLMError, match="kształt"):
|
||||
ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10)
|
||||
|
||||
|
||||
def test_retries_then_succeeds(monkeypatch):
|
||||
calls = {"n": 0}
|
||||
|
||||
def handler(request):
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 3:
|
||||
return httpx.Response(429, text="za duzo")
|
||||
return httpx.Response(200, json={"choices": [{"message": {"content": "ok"}}]})
|
||||
|
||||
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
|
||||
monkeypatch.setattr("app.llm.providers.time.sleep", lambda s: None) # bez czekania
|
||||
assert ChatCompletionsProvider("local", "http://x/v1", "m").generate("p", 10).text == "ok"
|
||||
assert calls["n"] == 3
|
||||
|
||||
|
||||
# ------------------------------------------------------------------- Anthropic
|
||||
|
||||
def test_anthropic_generates(monkeypatch):
|
||||
def handler(request):
|
||||
assert request.url.path.endswith("/v1/messages")
|
||||
assert request.headers.get("x-api-key") == "klucz"
|
||||
assert request.headers.get("anthropic-version")
|
||||
return httpx.Response(200, json={
|
||||
"model": "claude-x",
|
||||
"content": [{"type": "text", "text": "Prognoza."}],
|
||||
"usage": {"input_tokens": 10, "output_tokens": 5},
|
||||
})
|
||||
|
||||
monkeypatch.setattr(httpx, "Client", _mock_client(handler))
|
||||
out = AnthropicProvider("https://api.anthropic.com", "claude-x", "klucz").generate("p", 100)
|
||||
assert out.text == "Prognoza." and out.leaves_lan is True
|
||||
|
||||
|
||||
def test_anthropic_requires_key():
|
||||
with pytest.raises(LLMError, match="LLM_API_KEY"):
|
||||
AnthropicProvider("https://api.anthropic.com", "m", "").generate("p", 10)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- fabryka
|
||||
|
||||
def test_default_provider_is_local(monkeypatch):
|
||||
monkeypatch.delenv("LLM_PROVIDER", raising=False)
|
||||
monkeypatch.delenv("LLM_MODEL", raising=False)
|
||||
monkeypatch.delenv("LLM_BASE_URL", raising=False)
|
||||
p = factory.build_provider()
|
||||
assert p.name == "local"
|
||||
# domyślnie NIC nie opuszcza sieci — prompt niesie opisy z baz (LOG-32)
|
||||
assert p.leaves_lan is False
|
||||
|
||||
|
||||
def test_openai_requires_key(monkeypatch):
|
||||
monkeypatch.delenv("LLM_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||||
# komunikat wskazuje ZMIENNĄ DO USTAWIENIA dla tego dostawcy, nie ogólne LLM_API_KEY
|
||||
with pytest.raises(LLMError, match="OPENAI_API_KEY"):
|
||||
factory.build_provider("openai")
|
||||
|
||||
|
||||
def test_unknown_provider_rejected():
|
||||
with pytest.raises(LLMError, match="Nieznany dostawca"):
|
||||
factory.build_provider("bzdura")
|
||||
|
||||
|
||||
def test_env_overrides_model_and_url(monkeypatch):
|
||||
monkeypatch.setenv("LLM_MODEL", "moj-model")
|
||||
monkeypatch.setenv("LLM_BASE_URL", "http://serwer:8000/v1")
|
||||
p = factory.build_provider("local")
|
||||
assert p.model == "moj-model" and p.base_url == "http://serwer:8000/v1"
|
||||
|
||||
|
||||
# ---------------------------------------- konfiguracja per dostawca (regresja LOG-31)
|
||||
# UI pozwala przelaczac dostawce przy kazdym zadaniu, wiec ustawienia JEDNEGO nie moga
|
||||
# przeciekac na pozostalych. Wczesniej wspolne LLM_BASE_URL/LLM_MODEL kierowaly zadania
|
||||
# do OpenAI na adres lokalnej Ollamy i prosily Anthropic o model llama.
|
||||
|
||||
def _clear(monkeypatch):
|
||||
for v in ("LLM_PROVIDER", "LLM_MODEL", "LLM_BASE_URL", "LLM_API_KEY",
|
||||
"LOCAL_MODEL", "LOCAL_BASE_URL", "LOCAL_API_KEY",
|
||||
"OPENAI_MODEL", "OPENAI_BASE_URL", "OPENAI_API_KEY",
|
||||
"ANTHROPIC_MODEL", "ANTHROPIC_BASE_URL", "ANTHROPIC_API_KEY"):
|
||||
monkeypatch.delenv(v, raising=False)
|
||||
|
||||
|
||||
def test_local_config_does_not_leak_to_cloud(monkeypatch):
|
||||
"""Sedno bledu: skonfigurowany model lokalny przejmowal zadania do chmury."""
|
||||
_clear(monkeypatch)
|
||||
monkeypatch.setenv("LLM_PROVIDER", "local")
|
||||
monkeypatch.setenv("LLM_BASE_URL", "http://ollama:11434/v1") # konfiguracja lokalnego
|
||||
monkeypatch.setenv("LLM_MODEL", "llama3.1:8b")
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
|
||||
|
||||
local = factory.build_provider("local")
|
||||
assert local.base_url == "http://ollama:11434/v1" and local.model == "llama3.1:8b"
|
||||
|
||||
openai = factory.build_provider("openai")
|
||||
assert openai.base_url == "https://api.openai.com/v1", "zadanie do OpenAI poszloby do Ollamy"
|
||||
assert openai.model == "gpt-4o-mini", "OpenAI dostalby nazwe modelu llama"
|
||||
|
||||
|
||||
def test_provider_specific_settings_win(monkeypatch):
|
||||
_clear(monkeypatch)
|
||||
monkeypatch.setenv("LLM_PROVIDER", "local")
|
||||
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant")
|
||||
monkeypatch.setenv("ANTHROPIC_MODEL", "claude-opus-4-8")
|
||||
p = factory.build_provider("anthropic")
|
||||
assert p.model == "claude-opus-4-8" and p.api_key == "sk-ant"
|
||||
|
||||
|
||||
def test_generic_vars_apply_only_to_default_provider(monkeypatch):
|
||||
"""Zgodnosc wstecz: wspolne LLM_* konfiguruja dostawce domyslnego i tylko jego."""
|
||||
_clear(monkeypatch)
|
||||
monkeypatch.setenv("LLM_PROVIDER", "openai")
|
||||
monkeypatch.setenv("LLM_API_KEY", "sk-generic")
|
||||
monkeypatch.setenv("LLM_MODEL", "gpt-4o")
|
||||
assert factory.build_provider("openai").model == "gpt-4o"
|
||||
assert factory.build_provider("local").model == "llama3.1:8b" # nie dziedziczy
|
||||
|
||||
|
||||
def test_cloud_without_key_is_rejected_clearly(monkeypatch):
|
||||
_clear(monkeypatch)
|
||||
monkeypatch.setenv("LLM_PROVIDER", "local")
|
||||
for name in ("openai", "anthropic"):
|
||||
with pytest.raises(LLMError, match=f"{name.upper()}_API_KEY"):
|
||||
factory.build_provider(name)
|
||||
|
||||
|
||||
def test_local_needs_no_key(monkeypatch):
|
||||
_clear(monkeypatch)
|
||||
assert factory.build_provider("local").api_key == ""
|
||||
@@ -1,71 +0,0 @@
|
||||
"""Lots / punkty arabskie (LOG-08).
|
||||
|
||||
Wyrocznie z notes3 (astro-seek, horoskop referencyjny 30.04.1984, urodzenie DZIENNE):
|
||||
- Fortuna wprost w tabeli obiektów: Cancer 12°35'24" = 102.5900°
|
||||
- Ducha (Spirit) potwierdza jego antyscja: Taurus 28°14' -> Spirit = 180 − 58.2333
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from app.engine.formats import sign_index
|
||||
from app.engine.lots import LOT_DEFS, compute_lots
|
||||
|
||||
# natalne długości horoskopu referencyjnego
|
||||
NATAL = {"Asc": 112.1786, "Sun": 40.1392, "Moon": 30.5514, "Mercury": 27.3839,
|
||||
"Venus": 27.6769, "Mars": 234.5411, "Jupiter": 282.9617, "Saturn": 223.3108}
|
||||
|
||||
FORTUNE_REF = 102.5900 # Cancer 12°35'24"
|
||||
SPIRIT_REF = 180.0 - 58.2333 # z antyscji Taurus 28°14'
|
||||
|
||||
|
||||
def _by_name(lots):
|
||||
return {lot["name"]: lot["longitude"] for lot in lots}
|
||||
|
||||
|
||||
def _arcmin(a, b):
|
||||
return abs(((a - b + 180.0) % 360.0) - 180.0) * 60.0
|
||||
|
||||
|
||||
def test_fortune_matches_astroseek():
|
||||
lots = _by_name(compute_lots(NATAL, is_day=True))
|
||||
assert _arcmin(lots["Fortune"], FORTUNE_REF) < 2.0
|
||||
assert sign_index(lots["Fortune"]) == 3 # Cancer
|
||||
|
||||
|
||||
def test_spirit_matches_antiscia_reference():
|
||||
lots = _by_name(compute_lots(NATAL, is_day=True))
|
||||
assert _arcmin(lots["Spirit"], SPIRIT_REF) < 2.0
|
||||
|
||||
|
||||
def test_fortune_and_spirit_reverse_at_night():
|
||||
day = _by_name(compute_lots(NATAL, is_day=True))
|
||||
night = _by_name(compute_lots(NATAL, is_day=False))
|
||||
# nocą formuła się odwraca: Fortuna nocna = Duch dzienny (i odwrotnie)
|
||||
assert _arcmin(night["Fortune"], day["Spirit"]) < 0.01
|
||||
assert _arcmin(night["Spirit"], day["Fortune"]) < 0.01
|
||||
|
||||
|
||||
def test_all_seven_hermetic_lots_present():
|
||||
lots = compute_lots(NATAL, is_day=True)
|
||||
assert [lot["name"] for lot in lots] == [d[0] for d in LOT_DEFS]
|
||||
assert len(lots) == 7
|
||||
|
||||
|
||||
def test_derived_lots_use_fortune_and_spirit():
|
||||
lots = _by_name(compute_lots(NATAL, is_day=True))
|
||||
# Necessity = Asc + Fortune − Mercury
|
||||
expected = (NATAL["Asc"] + lots["Fortune"] - NATAL["Mercury"]) % 360.0
|
||||
assert _arcmin(lots["Necessity"], expected) < 0.01
|
||||
# Eros = Asc + Venus − Spirit
|
||||
expected_eros = (NATAL["Asc"] + NATAL["Venus"] - lots["Spirit"]) % 360.0
|
||||
assert _arcmin(lots["Eros"], expected_eros) < 0.01
|
||||
|
||||
|
||||
def test_by_sign_method_lands_on_sign_start():
|
||||
lots = compute_lots(NATAL, is_day=True, method="sign")
|
||||
for lot in lots:
|
||||
assert lot["longitude"] % 30.0 == 0.0 # 0° wyliczonego znaku
|
||||
|
||||
|
||||
def test_unknown_method_rejected():
|
||||
with pytest.raises(ValueError):
|
||||
compute_lots(NATAL, is_day=True, method="bzdura")
|
||||
@@ -1,218 +0,0 @@
|
||||
"""Generator promptów (LOG-29) i budżetowanie (LOG-30).
|
||||
|
||||
Testy nie wołają żadnego modelu — sprawdzają skład promptu i niezmienniki redukcji.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from app.prompt import (
|
||||
BUDGETS,
|
||||
build_natal_prompt,
|
||||
build_period_prompt,
|
||||
natal_indications,
|
||||
period_indications,
|
||||
reduce_indications,
|
||||
)
|
||||
|
||||
CHART = {
|
||||
"zodiac": "tropical",
|
||||
"house_system": "whole_sign",
|
||||
"sect": "day",
|
||||
"positions": [
|
||||
{"name": "Sun", "sign": "Taurus", "in_sign": "Tau 10°12'37\"", "house": 11, "direction": "D"},
|
||||
{"name": "Moon", "sign": "Aries", "in_sign": "Ari 7°48'40\"", "house": 10, "direction": "D"},
|
||||
],
|
||||
"angles": {"Asc": {"name": "Asc", "in_sign": "Can 16°00'54\""},
|
||||
"MC": {"name": "MC", "in_sign": "Pis 25°50'50\""}},
|
||||
"lots": [{"name": "Fortune", "in_sign": "Can 7°15'14\"", "house": 1,
|
||||
"formula": "Asc + Moon − Sun"}],
|
||||
"aspects": [{"obj1": "Sun", "obj2": "Moon", "aspect": "conjunction", "orb": 2.34, "as": "A"}],
|
||||
}
|
||||
|
||||
|
||||
def _report(n_samples=3, score=2.0):
|
||||
return {"objects": [{
|
||||
"object": "Sun", "sign": "Taurus", "house": 11,
|
||||
"facets": [{
|
||||
"type": "sign", "label": "w znaku Taurus", "score": score,
|
||||
"samples": [{"significator": f"[Su in [Tau {i}", "expanded": f"Sun in Taurus {i}",
|
||||
"effect": f"efekt numer {i}"} for i in range(n_samples)],
|
||||
}],
|
||||
}]}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- budżetowanie
|
||||
|
||||
def test_budget_limits_prompt_size():
|
||||
# opisy MUSZĄ być różne — identyczne zlałoby grupowanie w jedną pozycję
|
||||
big = {"objects": [{
|
||||
"object": "Sun", "sign": "Taurus", "house": 11,
|
||||
"facets": [{"type": "sign", "label": "w znaku Taurus", "score": 1.0,
|
||||
"samples": [{"significator": f"[Su {i}", "expanded": f"Sun {i}",
|
||||
"effect": f"opis {i} " + "x" * 200} for i in range(500)]}],
|
||||
}]}
|
||||
out = build_natal_prompt(CHART, big, budget="concise")
|
||||
assert out["stats"]["chars"] <= BUDGETS["concise"] * 1.1 # z marginesem na sekcje stałe
|
||||
assert out["stats"]["omitted"] > 0
|
||||
|
||||
|
||||
def test_bigger_budget_includes_more():
|
||||
big = _report(n_samples=300, score=1.0)
|
||||
small = build_natal_prompt(CHART, big, budget="concise")["stats"]
|
||||
large = build_natal_prompt(CHART, big, budget="extensive")["stats"]
|
||||
assert large["included"] > small["included"]
|
||||
assert large["omitted"] < small["omitted"]
|
||||
|
||||
|
||||
def test_grouping_merges_identical_effects():
|
||||
items = [
|
||||
{"context": "Sun — w znaku", "sort_key": ("Sun", "a"), "score": 1.0,
|
||||
"significator": "Sun in Taurus", "effect": "ten sam opis"},
|
||||
{"context": "Sun — w znaku", "sort_key": ("Sun", "a"), "score": 1.0,
|
||||
"significator": "Sun in Taurus (inny zapis)", "effect": "Ten Sam Opis "},
|
||||
]
|
||||
chosen, stats = reduce_indications(items, 10_000)
|
||||
assert len(chosen) == 1 # zlane w jedno
|
||||
assert chosen[0]["count"] == 2 # z licznikiem wystąpień
|
||||
assert stats["after_grouping"] == 1
|
||||
assert stats["deduplicated"] == 1
|
||||
|
||||
|
||||
def test_sorted_by_score_desc():
|
||||
items = [
|
||||
{"context": f"c{i}", "sort_key": (f"c{i}", ""), "score": float(i),
|
||||
"significator": f"s{i}", "effect": f"e{i}"} for i in range(5)
|
||||
]
|
||||
chosen, _ = reduce_indications(items, 10_000)
|
||||
scores = [c["score"] for c in chosen]
|
||||
assert scores == sorted(scores, reverse=True)
|
||||
|
||||
|
||||
def test_weakest_are_dropped_first():
|
||||
items = [
|
||||
{"context": f"c{i}", "sort_key": (f"c{i}", ""), "score": float(i),
|
||||
"significator": f"s{i}", "effect": "opis " * 20} for i in range(20)
|
||||
]
|
||||
chosen, stats = reduce_indications(items, 400)
|
||||
assert stats["omitted"] > 0
|
||||
# to, co weszło, ma wagę nie niższą niż to, co odpadło
|
||||
assert stats["min_score_included"] >= stats["max_score_omitted"]
|
||||
|
||||
|
||||
def test_long_effects_are_shortened():
|
||||
items = [{"context": "c", "sort_key": ("c", ""), "score": 1.0,
|
||||
"significator": "s", "effect": "x" * 1000}]
|
||||
chosen, stats = reduce_indications(items, 10_000)
|
||||
assert stats["shortened"] == 1
|
||||
assert "skrócono" in chosen[0]["effect"]
|
||||
|
||||
|
||||
def test_never_truncates_mid_indication():
|
||||
items = [{"context": "c", "sort_key": ("c", ""), "score": 1.0,
|
||||
"significator": "s", "effect": "opis " * 50} for _ in range(10)]
|
||||
chosen, _ = reduce_indications(items, 200)
|
||||
assert chosen # zawsze co najmniej jedno całe
|
||||
for c in chosen:
|
||||
assert c["effect"].endswith(("opis", "[…skrócono]")) # nie urwane w pół słowa
|
||||
|
||||
|
||||
def test_unknown_budget_rejected():
|
||||
with pytest.raises(ValueError):
|
||||
build_natal_prompt(CHART, _report(), budget="gigantyczny")
|
||||
|
||||
|
||||
# ----------------------------------------------------------------- skład promptu
|
||||
|
||||
def test_natal_prompt_contains_required_sections():
|
||||
p = build_natal_prompt(CHART, _report(), moment_label="1984-04-30 09:20 UTC")["prompt"]
|
||||
for section in ["# ZADANIE", "# DANE HOROSKOPU", "# WSKAZANIA Z BAZ",
|
||||
"# JAK MA WYGLĄDAĆ ODPOWIEDŹ", "# ZASTRZEŻENIE"]:
|
||||
assert section in p
|
||||
assert "HOROSKOPU\nURODZENIOWEGO" in p or "URODZENIOWEGO" in p
|
||||
|
||||
|
||||
def test_natal_prompt_carries_chart_data():
|
||||
p = build_natal_prompt(CHART, _report(), moment_label="1984-04-30 09:20 UTC")["prompt"]
|
||||
assert "Tau 10°12'37\"" in p and "dom 11" in p # pozycja + dom
|
||||
assert "Can 16°00'54\"" in p # Asc
|
||||
assert "Fortune" in p # Lots
|
||||
assert "Sun conjunction Moon" in p or "conjunction" in p
|
||||
assert "sekta: dzienna" in p
|
||||
assert "1984-04-30 09:20 UTC" in p
|
||||
|
||||
|
||||
def test_prompt_demands_citing_significators():
|
||||
p = build_natal_prompt(CHART, _report())["prompt"]
|
||||
assert "Teza bez wskazania jest niedopuszczalna" in p
|
||||
assert "Nie zmyślaj" in p
|
||||
|
||||
|
||||
def test_prompt_has_disclaimer():
|
||||
p = build_natal_prompt(CHART, _report())["prompt"]
|
||||
assert "nie stanowi porady" in p
|
||||
|
||||
|
||||
def test_prompt_is_deterministic():
|
||||
a = build_natal_prompt(CHART, _report(n_samples=50), budget="medium")["prompt"]
|
||||
b = build_natal_prompt(CHART, _report(n_samples=50), budget="medium")["prompt"]
|
||||
assert a == b
|
||||
|
||||
|
||||
def test_omission_is_reported_in_prompt():
|
||||
out = build_natal_prompt(CHART, _report(n_samples=400, score=1.0), budget="concise")
|
||||
assert out["stats"]["omitted"] > 0
|
||||
assert "pominięto" in out["prompt"] # użytkownik/model wie, że coś odpadło
|
||||
|
||||
|
||||
def test_empty_report_still_builds_prompt():
|
||||
out = build_natal_prompt(CHART, {"objects": []})
|
||||
assert out["stats"]["included"] == 0
|
||||
assert "brak trafień" in out["prompt"]
|
||||
assert "# DANE HOROSKOPU" in out["prompt"]
|
||||
|
||||
|
||||
# ------------------------------------------------------------------- okresowy
|
||||
|
||||
EVENTS = [
|
||||
{"technique": "profection", "significator": "Lord of Year: Mars", "start": "2026-04-30",
|
||||
"exact": "2026-04-30", "end": "2027-04-30", "interpretations_count": 2,
|
||||
"interpretations": [{"significator": "[Ma", "expanded": "Mars", "effect": "opis marsowy"}]},
|
||||
{"technique": "solar_arc", "significator": "Sun conj Saturn", "start": "2026-01-01",
|
||||
"exact": "2026-06-15", "end": "2026-12-31", "interpretations_count": 1,
|
||||
"interpretations": [{"significator": "[Su [conj [Sa", "expanded": "Sun conjunction Saturn",
|
||||
"effect": "opis saturniczny"}]},
|
||||
]
|
||||
|
||||
|
||||
def test_period_prompt_has_dates_and_timeline():
|
||||
out = build_period_prompt(CHART, EVENTS, "2026-01-01", "2027-01-01")
|
||||
p = out["prompt"]
|
||||
assert "2026-01-01 — 2027-01-01" in p
|
||||
assert "# OŚ CZASU" in p
|
||||
assert "profection" in p and "solar_arc" in p
|
||||
assert "2026-06-15" in p # data dokładna zdarzenia
|
||||
assert "CHRONOLOGICZNIE" in p
|
||||
assert out["stats"]["events"] == 2
|
||||
assert out["profile"] == "period"
|
||||
|
||||
|
||||
def test_period_prompt_keeps_timeline_without_interpretations():
|
||||
"""Padnięta warstwa danych zabiera wskazania, ale NIE oś czasu — ta jest czysto
|
||||
obliczeniowa i bez niej prognoza okresowa jest bezużyteczna."""
|
||||
bare = [{k: v for k, v in ev.items()
|
||||
if k not in ("interpretations", "interpretations_count")} for ev in EVENTS]
|
||||
out = build_period_prompt(CHART, bare, "2026-01-01", "2027-01-01")
|
||||
assert out["stats"]["events"] == 2
|
||||
assert "profection" in out["prompt"] and "2026-06-15" in out["prompt"]
|
||||
assert out["stats"]["included"] == 0 # brak wskazań, ale oś czasu jest
|
||||
|
||||
|
||||
def test_period_indications_weight_by_hit_count():
|
||||
items = period_indications(EVENTS)
|
||||
assert [i["score"] for i in items] == [2.0, 1.0]
|
||||
|
||||
|
||||
def test_natal_indications_flatten_facets():
|
||||
items = natal_indications(_report(n_samples=4))
|
||||
assert len(items) == 4
|
||||
assert all(i["context"].startswith("Sun —") for i in items)
|
||||
assert all(i["score"] == 2.0 for i in items)
|
||||
@@ -1,65 +0,0 @@
|
||||
"""Token międzywarstwowy (LOG-32).
|
||||
|
||||
Warstwa logiczna oddaje treść baz, więc musi odrzucać żądania z pominięciem UI.
|
||||
"""
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from app import security
|
||||
|
||||
TOKEN = "tajny-token-testowy"
|
||||
|
||||
|
||||
def _app() -> FastAPI:
|
||||
app = FastAPI()
|
||||
security.install(app, "testowa")
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return {"status": "ok"}
|
||||
|
||||
@app.get("/secret")
|
||||
def secret():
|
||||
return {"rows": ["treść z bazy"]}
|
||||
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def guarded(monkeypatch):
|
||||
monkeypatch.setenv("INTERNAL_TOKEN", TOKEN)
|
||||
return TestClient(_app())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def open_app(monkeypatch):
|
||||
monkeypatch.delenv("INTERNAL_TOKEN", raising=False)
|
||||
return TestClient(_app())
|
||||
|
||||
|
||||
def test_rejects_request_without_token(guarded):
|
||||
assert guarded.get("/secret").status_code == 401
|
||||
|
||||
|
||||
def test_rejects_wrong_token(guarded):
|
||||
r = guarded.get("/secret", headers={security.HEADER: "zly"})
|
||||
assert r.status_code == 401
|
||||
assert "treść z bazy" not in r.text
|
||||
|
||||
|
||||
def test_accepts_correct_token(guarded):
|
||||
r = guarded.get("/secret", headers={security.HEADER: TOKEN})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["rows"] == ["treść z bazy"]
|
||||
|
||||
|
||||
def test_health_stays_public(guarded):
|
||||
"""Sonda k8s nie zna tokenu — /health musi działać bez niego."""
|
||||
assert guarded.get("/health").status_code == 200
|
||||
|
||||
|
||||
def test_disabled_when_token_unset(open_app):
|
||||
"""Brak konfiguracji = zgodność wstecz (dev), nie blokada."""
|
||||
assert not security.enabled()
|
||||
assert open_app.get("/secret").status_code == 200
|
||||
@@ -1,127 +0,0 @@
|
||||
"""Systemy zodiaku (LOG-04): ayanamsy, syderyczny, draconic, RA.
|
||||
|
||||
Wartości referencyjne ayanams pochodzą ze **Swiss Ephemeris** (get_ayanamsa_ut)
|
||||
użytego jako wyrocznia — silnika B nie importujemy tu (izolacja AGPL), więc stałe
|
||||
są wpięte tak jak referencje astro-seek w innych testach. Nasz model odtwarza je
|
||||
z dokładnością do ~0,02" w latach 1900–2100.
|
||||
"""
|
||||
import math
|
||||
|
||||
import pytest
|
||||
|
||||
from app.engine import zodiac as Z
|
||||
from app.engine.chart import build_chart
|
||||
from app.engine.formats import sign_index
|
||||
|
||||
# wyrocznia: swisseph get_ayanamsa_ut (JD UT -> ayanamsa °)
|
||||
ORACLE = {
|
||||
"sidereal_lahiri": {
|
||||
2451545.0: 23.857092, 2445820.88889: 23.638183,
|
||||
2415021.0: 22.460550, 2433283.0: 23.158744,
|
||||
2469808.0: 24.555632, 2488070.0: 25.254287,
|
||||
},
|
||||
"sidereal_fagan_bradley": {
|
||||
2451545.0: 24.740300, 2445820.88889: 24.521391,
|
||||
2415021.0: 23.343757, 2433283.0: 24.041952,
|
||||
2469808.0: 25.438840, 2488070.0: 26.137495,
|
||||
},
|
||||
"sidereal_krishnamurti": {
|
||||
2451545.0: 23.760240, 2415021.0: 22.363697,
|
||||
2469808.0: 24.458780, 2488070.0: 25.157435,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _arcsec(a, b):
|
||||
return abs(a - b) * 3600.0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(ORACLE))
|
||||
def test_ayanamsha_matches_swisseph_oracle(name):
|
||||
for jd, ref in ORACLE[name].items():
|
||||
got = Z.ayanamsha(name, jd)
|
||||
assert _arcsec(got, ref) < 0.1, f"{name} @ JD{jd}: {got} vs {ref} ({_arcsec(got, ref):.3f}\")"
|
||||
|
||||
|
||||
def test_julian_day_j2000():
|
||||
from datetime import datetime, timezone
|
||||
jd = Z.julian_day(datetime(2000, 1, 1, 12, 0, tzinfo=timezone.utc))
|
||||
assert abs(jd - 2451545.0) < 1e-6
|
||||
|
||||
|
||||
def test_offset_tropical_is_zero():
|
||||
assert Z.offset("tropical", 2451545.0) == 0.0
|
||||
|
||||
|
||||
def test_offset_draconic_is_node_longitude():
|
||||
assert Z.offset("draconic", 2451545.0, node_lon=68.0) == 68.0
|
||||
# normalizacja
|
||||
assert Z.offset("draconic", 2451545.0, node_lon=428.0) == pytest.approx(68.0)
|
||||
|
||||
|
||||
def test_offset_draconic_requires_node():
|
||||
with pytest.raises(ValueError):
|
||||
Z.offset("draconic", 2451545.0)
|
||||
|
||||
|
||||
def test_unknown_zodiac_rejected():
|
||||
with pytest.raises(ValueError):
|
||||
Z.offset("bzdura", 2451545.0)
|
||||
with pytest.raises(ValueError):
|
||||
Z.ayanamsha("sidereal_nieistnieje", 2451545.0)
|
||||
|
||||
|
||||
def test_apply_subtracts_and_normalizes():
|
||||
assert Z.apply(10.0, 20.0) == pytest.approx(350.0)
|
||||
assert Z.apply(40.0, 24.74) == pytest.approx(15.26)
|
||||
|
||||
|
||||
def test_to_equatorial_reference_points():
|
||||
# punkt równonocy: λ=0,β=0 -> RA=0, dec=0
|
||||
ra, dec = Z.to_equatorial(0.0, 0.0, 23.4392911)
|
||||
assert ra == pytest.approx(0.0, abs=1e-9) and dec == pytest.approx(0.0, abs=1e-9)
|
||||
# przesilenie letnie: λ=90,β=0 -> RA=90, dec=+ε
|
||||
ra, dec = Z.to_equatorial(90.0, 0.0, 23.4392911)
|
||||
assert ra == pytest.approx(90.0, abs=1e-6)
|
||||
assert dec == pytest.approx(23.4392911, abs=1e-6)
|
||||
|
||||
|
||||
# ---- integracja z pełnym horoskopem ----
|
||||
|
||||
def test_sidereal_chart_shifts_signs_back(own_engine, reference_moment):
|
||||
trop = build_chart(own_engine, reference_moment, "whole_sign")
|
||||
sid = build_chart(own_engine, reference_moment, "whole_sign", zodiac="sidereal_lahiri")
|
||||
assert sid["zodiac"] == "sidereal_lahiri"
|
||||
assert "ayanamsha" in sid and 23.0 < sid["ayanamsha"] < 24.5
|
||||
tp = {p["name"]: p for p in trop["positions"]}
|
||||
sp = {p["name"]: p for p in sid["positions"]}
|
||||
# syderyczna długość = tropikalna − ayanamsa (mod 360) dla każdego obiektu
|
||||
ay = sid["ayanamsha"]
|
||||
for name in tp:
|
||||
expect = (tp[name]["decimal"] - ay) % 360.0
|
||||
assert abs(((sp[name]["decimal"] - expect + 180) % 360) - 180) < 1e-4
|
||||
|
||||
|
||||
def test_sidereal_preserves_house_numbers(own_engine, reference_moment):
|
||||
# obrót zodiaku nie zmienia numerów domów (geometria niezmiennicza)
|
||||
trop = build_chart(own_engine, reference_moment, "whole_sign")
|
||||
sid = build_chart(own_engine, reference_moment, "whole_sign", zodiac="sidereal_lahiri")
|
||||
tp = {p["name"]: p["house"] for p in trop["positions"]}
|
||||
sp = {p["name"]: p["house"] for p in sid["positions"]}
|
||||
assert tp == sp
|
||||
|
||||
|
||||
def test_draconic_puts_node_at_zero_aries(own_engine, reference_moment):
|
||||
drac = build_chart(own_engine, reference_moment, "whole_sign", zodiac="draconic")
|
||||
assert drac["zodiac"] == "draconic"
|
||||
nn = next(p for p in drac["positions"] if p["name"] == "North Node")
|
||||
# węzeł wznoszący = 0° Barana w draconic
|
||||
assert sign_index(nn["decimal"]) == 0
|
||||
assert nn["decimal"] < 0.001 or nn["decimal"] > 359.999
|
||||
|
||||
|
||||
def test_tropical_default_unchanged(own_engine, reference_moment):
|
||||
# domyślnie tropikalny: brak ayanamshy, zodiac=tropical
|
||||
chart = build_chart(own_engine, reference_moment, "whole_sign")
|
||||
assert chart["zodiac"] == "tropical"
|
||||
assert "ayanamsha" not in chart
|
||||
@@ -5,7 +5,6 @@ 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
|
||||
@@ -13,12 +12,6 @@ import httpx
|
||||
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 {}
|
||||
|
||||
|
||||
class LogicClient:
|
||||
def __init__(self, base_url: str | None = None) -> None:
|
||||
self.base_url = (base_url or settings.logic_url).rstrip("/")
|
||||
@@ -26,7 +19,7 @@ class LogicClient:
|
||||
def query(self, query: str, field: str, exact: bool, limit: int) -> dict[str, Any]:
|
||||
payload = {"query": query, "field": field, "exact": exact, "limit": limit}
|
||||
with httpx.Client(timeout=settings.http_timeout) as client:
|
||||
r = client.post(f"{self.base_url}/api/query", json=payload, headers=_auth_headers())
|
||||
r = client.post(f"{self.base_url}/api/query", json=payload)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
@@ -38,7 +31,6 @@ class LogicClient:
|
||||
objects: list[str] | None = None,
|
||||
house_system: str = "whole_sign",
|
||||
stations: bool = False,
|
||||
zodiac: str = "tropical",
|
||||
) -> dict[str, Any]:
|
||||
"""Pełny horoskop dla danego momentu — woła logic /chart/positions."""
|
||||
payload = {
|
||||
@@ -48,11 +40,10 @@ class LogicClient:
|
||||
"objects": objects,
|
||||
"house_system": house_system,
|
||||
"stations": stations,
|
||||
"zodiac": zodiac,
|
||||
}
|
||||
# stacje wymagają root-findów — dłuższy timeout
|
||||
with httpx.Client(timeout=max(settings.http_timeout, 60.0) if stations else settings.http_timeout) as client:
|
||||
r = client.post(f"{self.base_url}/chart/positions", json=payload, headers=_auth_headers())
|
||||
r = client.post(f"{self.base_url}/chart/positions", json=payload)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
@@ -62,45 +53,7 @@ class LogicClient:
|
||||
"""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}
|
||||
with httpx.Client(timeout=max(settings.http_timeout, 30.0)) as client:
|
||||
r = client.post(f"{self.base_url}/chart/report", json=payload, headers=_auth_headers())
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
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,
|
||||
) -> dict[str, Any]:
|
||||
"""Gotowy prompt do LLM z wyliczeń (LOG-29/30) — woła logic /chart/prompt."""
|
||||
payload: dict[str, Any] = {
|
||||
"profile": profile, "when_utc": when_utc_iso,
|
||||
"lat": lat, "lon": lon, "budget": budget,
|
||||
}
|
||||
if from_date and to_date:
|
||||
payload["from_date"], payload["to_date"] = from_date, to_date
|
||||
with httpx.Client(timeout=max(settings.http_timeout, 60.0)) as client:
|
||||
r = client.post(f"{self.base_url}/chart/prompt", json=payload, headers=_auth_headers())
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
def horoscope(
|
||||
self, profile: str, when_utc_iso: str, lat: float, lon: float,
|
||||
budget: str = "medium", provider: 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 from_date and to_date:
|
||||
payload["from_date"], payload["to_date"] = from_date, to_date
|
||||
with httpx.Client(timeout=max(settings.http_timeout, 300.0)) as client:
|
||||
r = client.post(f"{self.base_url}/chart/horoscope", json=payload, headers=_auth_headers())
|
||||
r = client.post(f"{self.base_url}/chart/report", json=payload)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
@@ -114,6 +67,6 @@ class LogicClient:
|
||||
"from_date": from_date, "to_date": to_date, "interpret": interpret,
|
||||
}
|
||||
with httpx.Client(timeout=max(settings.http_timeout, 60.0)) as client:
|
||||
r = client.post(f"{self.base_url}/chart/timeline", json=payload, headers=_auth_headers())
|
||||
r = client.post(f"{self.base_url}/chart/timeline", json=payload)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
@@ -17,7 +17,7 @@ from fastapi.responses import HTMLResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from app import geocode, security
|
||||
from app import geocode
|
||||
from app.clients.logic_client import LogicClient
|
||||
from app.config import DEFAULT_LOCATION_LABEL, default_form
|
||||
|
||||
@@ -25,7 +25,6 @@ app = FastAPI(title="astrololo · warstwa prezentacji")
|
||||
app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
||||
templates = Jinja2Templates(directory="app/templates")
|
||||
logic = LogicClient()
|
||||
security.install(app) # logowanie + limit żądań (LOG-32)
|
||||
|
||||
|
||||
def _build_utc(date: str, time: str, tz_offset: float) -> tuple[str, str]:
|
||||
@@ -67,18 +66,16 @@ def chart_compute(
|
||||
lon: float = Form(0.0),
|
||||
house_system: str = Form("whole_sign"),
|
||||
stations: bool = Form(False),
|
||||
zodiac: str = Form("tropical"),
|
||||
):
|
||||
form = {"date": date, "time": time, "tz_offset": tz_offset,
|
||||
"lat": lat, "lon": lon, "house_system": house_system, "stations": stations,
|
||||
"zodiac": zodiac}
|
||||
"lat": lat, "lon": lon, "house_system": house_system, "stations": stations}
|
||||
ctx: dict = {"form": form, "result": None, "error": None, "moment": None}
|
||||
try:
|
||||
iso_utc, label = _build_utc(date, time, tz_offset)
|
||||
ctx["moment"] = label
|
||||
ctx["result"] = logic.positions(
|
||||
when_utc_iso=iso_utc, lat=lat, lon=lon,
|
||||
house_system=house_system, stations=stations, zodiac=zodiac,
|
||||
house_system=house_system, stations=stations,
|
||||
)
|
||||
except (httpx.HTTPError,) as e:
|
||||
ctx["error"] = _logic_error(e)
|
||||
@@ -128,28 +125,14 @@ def interpret_run(
|
||||
lat: float = Form(0.0),
|
||||
lon: float = Form(0.0),
|
||||
group: bool = Form(False),
|
||||
action: str = Form("report"),
|
||||
prompt_budget: str = Form("medium"),
|
||||
llm_provider: str = Form("local"),
|
||||
):
|
||||
form = {"date": date, "time": time, "tz_offset": tz_offset,
|
||||
"lat": lat, "lon": lon, "group": group, "prompt_budget": prompt_budget,
|
||||
"llm_provider": llm_provider}
|
||||
"lat": lat, "lon": lon, "group": group}
|
||||
ctx: dict = {"form": form, "result": None, "error": None, "moment": None}
|
||||
try:
|
||||
iso_utc, label = _build_utc(date, time, tz_offset)
|
||||
ctx["moment"] = label
|
||||
if action == "prompt":
|
||||
ctx["prompt_result"] = logic.prompt(
|
||||
profile="natal", when_utc_iso=iso_utc, lat=lat, lon=lon, budget=prompt_budget,
|
||||
)
|
||||
elif action == "horoscope":
|
||||
ctx["prompt_result"] = logic.horoscope(
|
||||
profile="natal", when_utc_iso=iso_utc, lat=lat, lon=lon,
|
||||
budget=prompt_budget, provider=llm_provider,
|
||||
)
|
||||
else:
|
||||
ctx["result"] = logic.report(when_utc_iso=iso_utc, lat=lat, lon=lon, group=group)
|
||||
ctx["result"] = logic.report(when_utc_iso=iso_utc, lat=lat, lon=lon, group=group)
|
||||
except httpx.HTTPError as e:
|
||||
ctx["error"] = _logic_error(e)
|
||||
except ValueError as e:
|
||||
@@ -176,33 +159,17 @@ def timeline_run(
|
||||
lon: float = Form(0.0),
|
||||
from_date: str = Form(...),
|
||||
to_date: str = Form(...),
|
||||
action: str = Form("timeline"),
|
||||
prompt_budget: str = Form("medium"),
|
||||
llm_provider: str = Form("local"),
|
||||
):
|
||||
form = {"date": date, "time": time, "tz_offset": tz_offset, "lat": lat, "lon": lon,
|
||||
"from_date": from_date, "to_date": to_date, "prompt_budget": prompt_budget,
|
||||
"llm_provider": llm_provider}
|
||||
"from_date": from_date, "to_date": to_date}
|
||||
ctx: dict = {"form": form, "result": None, "error": None, "moment": None}
|
||||
try:
|
||||
iso_utc, label = _build_utc(date, time, tz_offset)
|
||||
ctx["moment"] = label
|
||||
if action == "prompt":
|
||||
ctx["prompt_result"] = logic.prompt(
|
||||
profile="period", when_utc_iso=iso_utc, lat=lat, lon=lon,
|
||||
budget=prompt_budget, from_date=from_date, to_date=to_date,
|
||||
)
|
||||
elif action == "horoscope":
|
||||
ctx["prompt_result"] = logic.horoscope(
|
||||
profile="period", when_utc_iso=iso_utc, lat=lat, lon=lon,
|
||||
budget=prompt_budget, provider=llm_provider,
|
||||
from_date=from_date, to_date=to_date,
|
||||
)
|
||||
else:
|
||||
ctx["result"] = logic.timeline(
|
||||
when_utc_iso=iso_utc, lat=lat, lon=lon,
|
||||
from_date=from_date, to_date=to_date, interpret=True,
|
||||
)
|
||||
ctx["result"] = logic.timeline(
|
||||
when_utc_iso=iso_utc, lat=lat, lon=lon,
|
||||
from_date=from_date, to_date=to_date, interpret=True,
|
||||
)
|
||||
except httpx.HTTPError as e:
|
||||
ctx["error"] = _logic_error(e)
|
||||
except ValueError as e:
|
||||
|
||||
@@ -1,120 +0,0 @@
|
||||
"""Ochrona dostępu do aplikacji (LOG-32).
|
||||
|
||||
Rdzeniem produktu są oryginalne bazy interpretacyjne. Aplikacja podaje ich treść
|
||||
na wielu ścieżkach (`/significators`, `/interpret`, generator promptu), więc BRAK
|
||||
uwierzytelnienia oznacza, że każdy w sieci może je wypompować — bez udziału
|
||||
jakiegokolwiek modelu językowego. Ten moduł zamyka tę drogę.
|
||||
|
||||
Dwa mechanizmy:
|
||||
* **HTTP Basic** — wejście do aplikacji; włącza się, gdy ustawiono APP_PASSWORD.
|
||||
* **limit żądań** — hamuje masowe odpytywanie (eksfiltrację przez pętlę zapytań).
|
||||
|
||||
Świadomie NIE logujemy treści żądań ani promptów — logi to kolejny nośnik wycieku.
|
||||
|
||||
UWAGA: bez APP_PASSWORD ochrona jest WYŁĄCZONA (zgodność wstecz i wygoda dev).
|
||||
Wtedy przy starcie leci głośne ostrzeżenie — żeby nikt nie wdrożył tego w
|
||||
przekonaniu, że jest chroniony.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import logging
|
||||
import os
|
||||
import secrets
|
||||
import time
|
||||
from collections import deque
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.responses import HTMLResponse, JSONResponse
|
||||
|
||||
log = logging.getLogger("astrololo.security")
|
||||
|
||||
MAX_TRACKED_CLIENTS_DEFAULT = 4096
|
||||
MAX_TRACKED_CLIENTS = MAX_TRACKED_CLIENTS_DEFAULT # zabezpieczenie przed puchnięciem pamięci
|
||||
|
||||
|
||||
# konfiguracja czytana leniwie — testy i restart mogą ją zmienić bez importu modułu
|
||||
def app_user() -> str:
|
||||
return os.getenv("APP_USER", "astrololo")
|
||||
|
||||
|
||||
def app_password() -> str:
|
||||
return os.getenv("APP_PASSWORD", "")
|
||||
|
||||
|
||||
def rate_limit_per_min() -> int:
|
||||
return int(os.getenv("RATE_LIMIT_PER_MIN", "120"))
|
||||
|
||||
PUBLIC_PATHS = frozenset({"/health"})
|
||||
PUBLIC_PREFIXES = ("/static/",)
|
||||
|
||||
_hits: dict[str, deque[float]] = {}
|
||||
|
||||
|
||||
def auth_enabled() -> bool:
|
||||
return bool(app_password())
|
||||
|
||||
|
||||
def _is_public(path: str) -> bool:
|
||||
return path in PUBLIC_PATHS or path.startswith(PUBLIC_PREFIXES)
|
||||
|
||||
|
||||
def _authorized(header: str | None) -> bool:
|
||||
if not header or not header.lower().startswith("basic "):
|
||||
return False
|
||||
try:
|
||||
raw = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
|
||||
user, _, password = raw.partition(":")
|
||||
except (binascii.Error, UnicodeDecodeError, IndexError):
|
||||
return False
|
||||
# porównanie odporne na atak czasowy; oba pola muszą się zgadzać
|
||||
ok_user = secrets.compare_digest(user, app_user())
|
||||
ok_pass = secrets.compare_digest(password, app_password())
|
||||
return ok_user and ok_pass
|
||||
|
||||
|
||||
def _rate_limited(client: str) -> bool:
|
||||
cap = rate_limit_per_min()
|
||||
if cap <= 0:
|
||||
return False
|
||||
now = time.monotonic()
|
||||
window = _hits.get(client)
|
||||
if window is None:
|
||||
if len(_hits) >= MAX_TRACKED_CLIENTS:
|
||||
_hits.clear() # prosty reset zamiast nieograniczonego wzrostu
|
||||
window = _hits[client] = deque()
|
||||
while window and now - window[0] > 60.0:
|
||||
window.popleft()
|
||||
if len(window) >= cap:
|
||||
return True
|
||||
window.append(now)
|
||||
return False
|
||||
|
||||
|
||||
def install(app) -> None:
|
||||
"""Podpina ochronę pod wszystkie ścieżki poza /health i /static."""
|
||||
if not auth_enabled():
|
||||
log.warning(
|
||||
"UWAGA: APP_PASSWORD nie ustawione — aplikacja jest OTWARTA dla każdego, "
|
||||
"kto ma do niej dostęp sieciowy, wraz z treścią baz interpretacyjnych."
|
||||
)
|
||||
|
||||
@app.middleware("http")
|
||||
async def _guard(request: Request, call_next):
|
||||
if _is_public(request.url.path):
|
||||
return await call_next(request)
|
||||
|
||||
client = request.client.host if request.client else "?"
|
||||
if _rate_limited(client):
|
||||
return JSONResponse(
|
||||
{"detail": "Zbyt wiele żądań — spróbuj za chwilę."},
|
||||
status_code=429, headers={"Retry-After": "60"},
|
||||
)
|
||||
|
||||
if auth_enabled() and not _authorized(request.headers.get("authorization")):
|
||||
return HTMLResponse(
|
||||
"<h1>401 — wymagane logowanie</h1>", status_code=401,
|
||||
headers={"WWW-Authenticate": 'Basic realm="astrololo"'},
|
||||
)
|
||||
return await call_next(request)
|
||||
@@ -1,31 +0,0 @@
|
||||
// Kopiowanie do schowka dla przycisków [data-copy="#selektor"].
|
||||
// clipboard API wymaga secure context (https/localhost), a aplikacja bywa serwowana
|
||||
// po http://<ip> — dlatego jest awaryjne przejście na zaznaczenie + execCommand.
|
||||
document.addEventListener('click', function (e) {
|
||||
const btn = e.target.closest('[data-copy]');
|
||||
if (!btn) return;
|
||||
|
||||
const src = document.querySelector(btn.getAttribute('data-copy'));
|
||||
if (!src) return;
|
||||
|
||||
const done = ok => {
|
||||
const label = btn.textContent;
|
||||
btn.textContent = ok ? 'Skopiowano ✓' : 'Nie udało się — zaznacz i skopiuj ręcznie';
|
||||
setTimeout(() => { btn.textContent = label; }, 2500);
|
||||
};
|
||||
|
||||
const text = src.value !== undefined ? src.value : src.textContent;
|
||||
|
||||
if (navigator.clipboard && window.isSecureContext) {
|
||||
navigator.clipboard.writeText(text).then(() => done(true), () => done(false));
|
||||
return;
|
||||
}
|
||||
// awaryjnie: zaznacz zawartość i spróbuj starym poleceniem
|
||||
try {
|
||||
src.focus();
|
||||
if (src.select) src.select();
|
||||
done(document.execCommand('copy'));
|
||||
} catch (err) {
|
||||
done(false);
|
||||
}
|
||||
});
|
||||
@@ -81,10 +81,3 @@ tr:last-child td { border-bottom: none; }
|
||||
.geo-pin { font-size: 20px; line-height: 24px; text-align: center; }
|
||||
/* kafelki OSM są jasne — trzymamy kontrolki czytelne na ciemnym tle */
|
||||
.leaflet-control-attribution { font-size: .68rem; }
|
||||
|
||||
/* Generator promptu do LLM (LOG-29/30) */
|
||||
textarea.prompt { width: 100%; margin-top: .5rem; padding: .7rem .8rem; box-sizing: border-box;
|
||||
background: #12132a; color: var(--ink); border: 1px solid var(--line);
|
||||
border-radius: 10px; font-family: ui-monospace, SFMono-Regular, Menlo, monospace;
|
||||
font-size: .82rem; line-height: 1.45; resize: vertical; white-space: pre; }
|
||||
textarea.prompt:focus { outline: 2px solid var(--accent); outline-offset: 1px; }
|
||||
|
||||
@@ -1,86 +0,0 @@
|
||||
{# Wspólny blok: sterowanie generowaniem + prompt (LOG-29/30) + horoskop (LOG-31).
|
||||
Używany na ekranach Interpretacje i Kalendarz. #}
|
||||
<div class="opts">
|
||||
<label>Budżet promptu
|
||||
<select name="prompt_budget">
|
||||
{% set pb = form.prompt_budget or 'medium' %}
|
||||
<option value="concise" {{ 'selected' if pb == 'concise' else '' }}>zwięzły (~4 tys. znaków)</option>
|
||||
<option value="medium" {{ 'selected' if pb == 'medium' else '' }}>średni (~12 tys.)</option>
|
||||
<option value="extensive" {{ 'selected' if pb == 'extensive' else '' }}>obszerny (~30 tys.)</option>
|
||||
</select>
|
||||
</label>
|
||||
<label>Model
|
||||
<select name="llm_provider">
|
||||
{% set lp = form.llm_provider or 'local' %}
|
||||
<option value="local" {{ 'selected' if lp == 'local' else '' }}>lokalny — nic nie opuszcza sieci</option>
|
||||
<option value="anthropic" {{ 'selected' if lp == 'anthropic' else '' }}>Anthropic — dane wychodzą</option>
|
||||
<option value="openai" {{ 'selected' if lp == 'openai' else '' }}>OpenAI — dane wychodzą</option>
|
||||
</select>
|
||||
</label>
|
||||
</div>
|
||||
<p class="muted small">
|
||||
Prompt zawiera <strong>oryginalne opisy z baz</strong> oraz dane urodzeniowe. Model lokalny
|
||||
przetwarza je u nas; wybór dostawcy w chmurze oznacza, że ta treść <strong>opuszcza naszą
|
||||
sieć</strong>. Możesz najpierw obejrzeć prompt, a dopiero potem wysłać.
|
||||
</p>
|
||||
|
||||
{% if prompt_result %}
|
||||
{% set st = prompt_result.stats %}
|
||||
|
||||
{# --- wynik: napisany horoskop (LOG-31) + transparentność (PRE-15) --- #}
|
||||
{% if prompt_result.horoscope %}
|
||||
<div class="meta">
|
||||
Horoskop napisany przez: <strong>{{ prompt_result.provider }}</strong> ·
|
||||
model: {{ prompt_result.model }}
|
||||
{% if prompt_result.usage and prompt_result.usage.completion_tokens %}
|
||||
· tokeny odpowiedzi: {{ prompt_result.usage.completion_tokens }}
|
||||
{% endif %}
|
||||
{% if prompt_result.leaves_lan %}
|
||||
· <strong class="retro">dane opuściły sieć</strong>
|
||||
{% else %}
|
||||
· dane nie opuściły sieci
|
||||
{% endif %}
|
||||
</div>
|
||||
<div class="actions">
|
||||
<button type="button" class="ghost" data-copy="#horoscopeText">Kopiuj horoskop</button>
|
||||
</div>
|
||||
<textarea id="horoscopeText" class="prompt" rows="20" readonly>{{ prompt_result.horoscope }}</textarea>
|
||||
<p class="muted small">
|
||||
Treść wygenerował model językowy na podstawie {{ st.included }} wskazań z baz.
|
||||
<strong>Nie stanowi porady medycznej, prawnej ani finansowej.</strong>
|
||||
</p>
|
||||
{% endif %}
|
||||
|
||||
{% if prompt_result.llm_error %}
|
||||
<div class="error">
|
||||
Nie udało się napisać horoskopu: {{ prompt_result.llm_error }}<br>
|
||||
Prompt poniżej jest gotowy — możesz go skopiować i użyć ręcznie.
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{# --- prompt: zawsze dostępny do podglądu i skopiowania --- #}
|
||||
<div class="meta">
|
||||
Prompt · {{ st.chars }} znaków (~{{ st.est_tokens }} tokenów) ·
|
||||
budżet: {{ st.budget }} ·
|
||||
wskazań: <strong>{{ st.included }}</strong>
|
||||
{% if st.omitted %}· pominięto: <strong>{{ st.omitted }}</strong>{% endif %}
|
||||
{% if st.deduplicated %}· scalono powtórek: {{ st.deduplicated }}{% endif %}
|
||||
</div>
|
||||
|
||||
{% if st.omitted %}
|
||||
<p class="muted small">
|
||||
Pominięto {{ st.omitted }} najsłabszych wskazań (próg wagi {{ st.min_score_included }}).
|
||||
Chcesz komplet — wybierz obszerniejszy budżet i wygeneruj ponownie.
|
||||
</p>
|
||||
{% endif %}
|
||||
|
||||
{% if prompt_result.data_error %}
|
||||
<div class="error">{{ prompt_result.data_error }} — prompt złożony z samych wyliczeń.</div>
|
||||
{% endif %}
|
||||
|
||||
<div class="actions">
|
||||
<button type="button" class="ghost" data-copy="#promptText">Kopiuj prompt</button>
|
||||
<span class="muted small">Możesz też wkleić go samodzielnie do ChatGPT lub Claude.</span>
|
||||
</div>
|
||||
<textarea id="promptText" class="prompt" rows="14" readonly>{{ prompt_result.prompt }}</textarea>
|
||||
{% endif %}
|
||||
@@ -32,16 +32,6 @@
|
||||
<option value="porphyry" {{ 'selected' if hs == 'porphyry' else '' }}>Porphyry</option>
|
||||
</select>
|
||||
</label>
|
||||
<label>Zodiak
|
||||
<select name="zodiac">
|
||||
{% set zd = form.zodiac or 'tropical' %}
|
||||
<option value="tropical" {{ 'selected' if zd == 'tropical' else '' }}>Tropikalny</option>
|
||||
<option value="sidereal_lahiri" {{ 'selected' if zd == 'sidereal_lahiri' else '' }}>Syderyczny (Lahiri)</option>
|
||||
<option value="sidereal_fagan_bradley" {{ 'selected' if zd == 'sidereal_fagan_bradley' else '' }}>Syderyczny (Fagan-Bradley)</option>
|
||||
<option value="sidereal_krishnamurti" {{ 'selected' if zd == 'sidereal_krishnamurti' else '' }}>Syderyczny (Krishnamurti)</option>
|
||||
<option value="draconic" {{ 'selected' if zd == 'draconic' else '' }}>Draconic</option>
|
||||
</select>
|
||||
</label>
|
||||
</div>
|
||||
{% include "_location_picker.html" %}
|
||||
<div class="opts">
|
||||
@@ -65,7 +55,6 @@
|
||||
Silnik: <strong>{{ result.engine }}</strong> ·
|
||||
obiektów: {{ result.positions | length }}
|
||||
{% if result.house_system %}· domy: {{ result.house_system }}{% endif %}
|
||||
{% if result.zodiac %}· zodiak: {{ result.zodiac }}{% if result.ayanamsha is defined %} (ayanamsa {{ '%.4f'|format(result.ayanamsha) }}°){% endif %}{% endif %}
|
||||
{% if moment %}· moment: {{ moment }}{% endif %}
|
||||
</div>
|
||||
|
||||
@@ -101,18 +90,6 @@
|
||||
</tbody>
|
||||
</table>
|
||||
|
||||
{% if result.lots %}
|
||||
<div class="meta">Lots hermetyczne ({{ result.lots | length }}) · sekta: <strong>{{ result.sect }}</strong></div>
|
||||
<table class="angles">
|
||||
<thead><tr><th>Lot</th><th>Znak</th><th>W znaku</th><th>Dom</th><th>Formuła</th></tr></thead>
|
||||
<tbody>
|
||||
{% for l in result.lots %}
|
||||
<tr><td>{{ l.name }}</td><td>{{ l.sign }}</td><td class="mono">{{ l.in_sign }}</td><td>{{ l.house }}</td><td class="muted small">{{ l.formula }}</td></tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
{% endif %}
|
||||
|
||||
{% set with_stations = result.positions | selectattr('stations', 'defined') | list %}
|
||||
{% if with_stations %}
|
||||
<div class="meta">Stacje planet (poprzednia / następna; <span class="badge">blisko</span> = mniej niż 7 dni)</div>
|
||||
|
||||
@@ -28,12 +28,9 @@
|
||||
<label><input type="checkbox" name="group" value="true" {{ 'checked' if form.group else '' }}> grupuj identyczne opisy</label>
|
||||
</div>
|
||||
{% if location_label %}<p class="muted small">Wstępnie wpisano lokalizację: <strong>{{ location_label }}</strong> ({{ form.lat }}, {{ form.lon }}). Zmień pola lub kliknij „Tu i teraz".</p>{% endif %}
|
||||
{% include "_prompt_block.html" %}
|
||||
<div class="actions">
|
||||
<button type="button" id="nowBtn" class="ghost">Tu i teraz</button>
|
||||
<button type="submit" name="action" value="report">Szukaj interpretacji</button>
|
||||
<button type="submit" name="action" value="prompt" class="ghost">Generuj prompt (AI)</button>
|
||||
<button type="submit" name="action" value="horoscope">Napisz horoskop (AI)</button>
|
||||
<button type="submit">Szukaj interpretacji</button>
|
||||
<span id="geoNote" class="muted small"></span>
|
||||
</div>
|
||||
</form>
|
||||
@@ -89,5 +86,4 @@
|
||||
{% endif %}
|
||||
|
||||
<script src="/static/now.js"></script>
|
||||
<script src="/static/copy.js"></script>
|
||||
{% endblock %}
|
||||
|
||||
@@ -33,12 +33,9 @@
|
||||
<input type="date" name="to_date" value="{{ form.to_date or '' }}" required>
|
||||
</label>
|
||||
</div>
|
||||
{% include "_prompt_block.html" %}
|
||||
<div class="actions">
|
||||
<button type="button" id="nowBtn" class="ghost">Tu i teraz</button>
|
||||
<button type="submit" name="action" value="timeline">Pokaż kalendarz</button>
|
||||
<button type="submit" name="action" value="prompt" class="ghost">Generuj prompt (AI)</button>
|
||||
<button type="submit" name="action" value="horoscope">Napisz horoskop (AI)</button>
|
||||
<button type="submit">Pokaż kalendarz</button>
|
||||
<span id="geoNote" class="muted small"></span>
|
||||
</div>
|
||||
</form>
|
||||
@@ -79,5 +76,4 @@
|
||||
{% endif %}
|
||||
|
||||
<script src="/static/now.js"></script>
|
||||
<script src="/static/copy.js"></script>
|
||||
{% endblock %}
|
||||
|
||||
@@ -1,2 +0,0 @@
|
||||
-r requirements.txt
|
||||
pytest>=8.0
|
||||
@@ -1,62 +0,0 @@
|
||||
"""Niezmiennik: KAŻDE wyjście HTTP w dół niesie token międzywarstwowy (LOG-32).
|
||||
|
||||
Powód istnienia tego testu: token dodano do klienta na gałęzi, która odbiła się od
|
||||
mastera zanim powstały metody `prompt()` i `horoscope()`. Git zmergował obie zmiany
|
||||
czysto (różne linie), ale nowe metody wyszły BEZ tokenu — i dostawały 401 dopiero na
|
||||
produkcji. Zwykły test jednej metody by tego nie złapał, więc sprawdzamy regułę
|
||||
strukturalnie: nie ma wywołania bez `headers=`.
|
||||
"""
|
||||
import ast
|
||||
import pathlib
|
||||
|
||||
CLIENT = pathlib.Path(__file__).resolve().parents[1] / "app" / "clients" / "logic_client.py"
|
||||
|
||||
|
||||
def _http_calls(path: pathlib.Path) -> list[tuple[str, int, bool]]:
|
||||
"""(nazwa_metody_http, linia, czy_ma_headers) dla każdego client.post/get."""
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
out = []
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute):
|
||||
continue
|
||||
if node.func.attr not in ("post", "get", "put", "patch", "delete"):
|
||||
continue
|
||||
if not (isinstance(node.func.value, ast.Name) and node.func.value.id == "client"):
|
||||
continue
|
||||
has_headers = any(kw.arg == "headers" for kw in node.keywords)
|
||||
out.append((node.func.attr, node.lineno, has_headers))
|
||||
return out
|
||||
|
||||
|
||||
def test_client_module_exists():
|
||||
assert CLIENT.is_file()
|
||||
|
||||
|
||||
def test_every_outbound_call_sends_auth_header():
|
||||
calls = _http_calls(CLIENT)
|
||||
assert calls, "nie znaleziono żadnego wywołania HTTP — test przestał cokolwiek pilnować"
|
||||
missing = [f"{CLIENT.name}:{line} client.{verb}()" for verb, line, ok in calls if not ok]
|
||||
assert not missing, (
|
||||
"Wywołania w dół bez tokenu międzywarstwowego (dostaną 401 przy włączonej "
|
||||
"ochronie): " + ", ".join(missing)
|
||||
)
|
||||
|
||||
|
||||
def test_auth_headers_helper_is_lazy():
|
||||
"""Token czytany przy wywołaniu, nie przy imporcie — inaczej pod wystartowałby
|
||||
z pustym tokenem, gdyby zmienna pojawiła się później."""
|
||||
import os
|
||||
|
||||
from app.clients.logic_client import _auth_headers
|
||||
|
||||
old = os.environ.get("INTERNAL_TOKEN")
|
||||
try:
|
||||
os.environ["INTERNAL_TOKEN"] = "abc"
|
||||
assert _auth_headers() == {"X-Astrololo-Token": "abc"}
|
||||
os.environ.pop("INTERNAL_TOKEN")
|
||||
assert _auth_headers() == {} # ochrona wyłączona = brak nagłówka
|
||||
finally:
|
||||
if old is not None:
|
||||
os.environ["INTERNAL_TOKEN"] = old
|
||||
else:
|
||||
os.environ.pop("INTERNAL_TOKEN", None)
|
||||
@@ -1,133 +0,0 @@
|
||||
"""Logowanie do aplikacji i limit żądań (LOG-32).
|
||||
|
||||
To jest brama chroniąca oryginalne bazy interpretacyjne — bez niej każdy w sieci
|
||||
mógł je wypompować przez `/significators` czy generator promptu. Testy pilnują,
|
||||
że brama faktycznie zamyka, a nie tylko wygląda na zamkniętą.
|
||||
"""
|
||||
import base64
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.responses import HTMLResponse
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from app import security
|
||||
|
||||
USER, PASSWORD = "astrololo", "haslo-testowe"
|
||||
|
||||
|
||||
def _basic(user: str, password: str) -> dict[str, str]:
|
||||
raw = base64.b64encode(f"{user}:{password}".encode()).decode()
|
||||
return {"Authorization": f"Basic {raw}"}
|
||||
|
||||
|
||||
def _app() -> FastAPI:
|
||||
app = FastAPI()
|
||||
security.install(app)
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return {"status": "ok"}
|
||||
|
||||
@app.get("/significators", response_class=HTMLResponse)
|
||||
def significators():
|
||||
return "<p>treść z bazy interpretacyjnej</p>"
|
||||
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_rate_limit():
|
||||
security._hits.clear()
|
||||
yield
|
||||
security._hits.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def guarded(monkeypatch):
|
||||
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
|
||||
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "120")
|
||||
return TestClient(_app())
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ logowanie
|
||||
|
||||
def test_blocks_anonymous_access(guarded):
|
||||
r = guarded.get("/significators")
|
||||
assert r.status_code == 401
|
||||
assert "treść z bazy" not in r.text
|
||||
|
||||
|
||||
def test_challenges_with_basic_realm(guarded):
|
||||
"""Bez nagłówka WWW-Authenticate przeglądarka nie pokaże okna logowania."""
|
||||
assert "Basic" in guarded.get("/significators").headers.get("WWW-Authenticate", "")
|
||||
|
||||
|
||||
def test_rejects_wrong_password(guarded):
|
||||
assert guarded.get("/significators", headers=_basic(USER, "zle")).status_code == 401
|
||||
|
||||
|
||||
def test_rejects_wrong_user(guarded):
|
||||
assert guarded.get("/significators", headers=_basic("obcy", PASSWORD)).status_code == 401
|
||||
|
||||
|
||||
def test_rejects_malformed_header(guarded):
|
||||
for bad in ("Basic !!!niebase64!!!", "Bearer cokolwiek", "", "Basic"):
|
||||
assert guarded.get("/significators", headers={"Authorization": bad}).status_code == 401
|
||||
|
||||
|
||||
def test_allows_correct_credentials(guarded):
|
||||
r = guarded.get("/significators", headers=_basic(USER, PASSWORD))
|
||||
assert r.status_code == 200
|
||||
assert "treść z bazy" in r.text
|
||||
|
||||
|
||||
def test_health_stays_public(guarded):
|
||||
assert guarded.get("/health").status_code == 200
|
||||
|
||||
|
||||
def test_open_when_password_unset(monkeypatch):
|
||||
"""Brak hasła = zgodność wstecz; ochrona wyłączona (i ostrzegamy przy starcie)."""
|
||||
monkeypatch.delenv("APP_PASSWORD", raising=False)
|
||||
assert not security.auth_enabled()
|
||||
assert TestClient(_app()).get("/significators").status_code == 200
|
||||
|
||||
|
||||
# --------------------------------------------------------------- limit żądań
|
||||
|
||||
def test_rate_limit_blocks_flood(monkeypatch):
|
||||
"""Masowe odpytywanie to droga eksfiltracji nawet po zalogowaniu."""
|
||||
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
|
||||
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "5")
|
||||
client = TestClient(_app())
|
||||
auth = _basic(USER, PASSWORD)
|
||||
codes = [client.get("/significators", headers=auth).status_code for _ in range(8)]
|
||||
assert codes[:5] == [200] * 5
|
||||
assert 429 in codes[5:]
|
||||
|
||||
|
||||
def test_rate_limited_response_has_retry_after(monkeypatch):
|
||||
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
|
||||
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "1")
|
||||
client = TestClient(_app())
|
||||
auth = _basic(USER, PASSWORD)
|
||||
client.get("/significators", headers=auth)
|
||||
r = client.get("/significators", headers=auth)
|
||||
assert r.status_code == 429 and r.headers.get("Retry-After") == "60"
|
||||
|
||||
|
||||
def test_rate_limit_precedes_auth(monkeypatch):
|
||||
"""Limit musi działać także dla niezalogowanych — inaczej zgadywanie hasła
|
||||
i sondowanie API jest darmowe."""
|
||||
monkeypatch.setenv("APP_PASSWORD", PASSWORD)
|
||||
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "3")
|
||||
client = TestClient(_app())
|
||||
codes = [client.get("/significators").status_code for _ in range(6)]
|
||||
assert 429 in codes
|
||||
|
||||
|
||||
def test_rate_limit_disabled_when_zero(monkeypatch):
|
||||
monkeypatch.setenv("RATE_LIMIT_PER_MIN", "0")
|
||||
monkeypatch.delenv("APP_PASSWORD", raising=False)
|
||||
client = TestClient(_app())
|
||||
assert all(client.get("/significators").status_code == 200 for _ in range(30))
|
||||
Reference in New Issue
Block a user