"""Most: policzony horoskop → tokeny sygnifikatorów → wyszukiwanie w bazie. Zalążek LOG-15/16: z pozycji obiektów (wraz z domami) generujemy tokeny w składni bazy i wyszukujemy pasujące rekordy. Dla każdego obiektu tworzymy kilka **faset**: - „w znaku" — planeta + token znaku (`[Su` + `[Tau`) - „w domu" — planeta + token domu (`[Su` + `11th H.`) Aspekty (`[conj`,`[sq`,`[opp`) wymagają policzenia aspektów (LOG-06) — na później. Format skrótów odczytany z realnej bazy: planety `[Su`,`[Mo`,…; znaki `[Ari`,`[Tau`,…; domy `12th H.`; np. `[Sa [conj [Su in 6th H.`. """ from __future__ import annotations from typing import Any, Protocol from app.abbreviations import expand from app.engine.formats import SIGN_ABBR, SIGNS PLANET_ABBR = { "Sun": "Su", "Moon": "Mo", "Mercury": "Me", "Venus": "Ve", "Mars": "Ma", "Jupiter": "Ju", "Saturn": "Sa", "Uranus": "Ur", "Neptune": "Ne", "Pluto": "Pl", } SIGN_TO_ABBR = dict(zip(SIGNS, SIGN_ABBR)) class DataSource(Protocol): def search( self, key: str, value: str, exact: bool, limit: int, fields: list[str] | None = None ) -> dict[str, Any]: ... def _effect(row: dict) -> str: for col in ("actioneffect", "topicresult", "bodypart"): v = row.get(col) if v and str(v).strip().lower() not in ("", "nan"): return str(v).strip() return "" def _is_noise(sig: str, effect: str) -> bool: s = sig.strip().lower() e = effect.strip().lower() return ( e in ("", "nan", "x", "x?", "?", "-") # efekt pusty/zastępczy or s.startswith("significator") # wiersz-legenda/nagłówek or "header" in s or s.startswith("*") # *MARKER / *header or s in ("x", "x?", "nan") # znacznik „pomiń rekord" ) def _ordinal(n: int) -> str: if 10 <= n % 100 <= 20: suffix = "th" else: suffix = {1: "st", 2: "nd", 3: "rd"}.get(n % 10, "th") return f"{n}{suffix}" def _facet_samples(rows: list[dict], token: str, limit: int = 4) -> list[dict]: """Rekordy, których sygnifikator zawiera token — bez szumu.""" out: list[dict] = [] tok = token.lower() for r in rows: sig = str(r.get("significator") or "") if tok not in sig.lower(): continue eff = _effect(r) if _is_noise(sig, eff): continue out.append({"significator": sig.strip(), "expanded": expand(sig.strip()), "effect": eff}) return out def build_report(positions: list[dict], data: DataSource, per_object_limit: int = 5000) -> dict: """positions: pozycje z build_chart (name, sign, direction, house). Dla każdego obiektu: jedno zapytanie o token planety, potem faseta „w znaku" i „w domu" (jeśli dom policzony). """ items: list[dict] = [] provider = None for p in positions: name = p.get("name") if name not in PLANET_ABBR: continue planet_tok = "[" + PLANET_ABBR[name] raw = data.search( key="significator", value=planet_tok, exact=False, limit=per_object_limit, fields=["significator", "actioneffect", "topicresult", "bodypart"], ) provider = raw.get("provider", provider) rows = raw.get("rows", []) facets: list[dict] = [] sign = p.get("sign") sign_tok = "[" + SIGN_TO_ABBR.get(sign, "") sign_samples = _facet_samples(rows, sign_tok) facets.append({ "type": "sign", "label": f"w znaku {sign}", "token": sign_tok, "count": len(sign_samples), "samples": sign_samples, }) house = p.get("house") if house: ordn = _ordinal(int(house)) house_samples = _facet_samples(rows, f"{ordn} h") # matcuje '12th H.' facets.append({ "type": "house", "label": f"w {ordn} domu", "token": f"{ordn} H.", "count": len(house_samples), "samples": house_samples, }) items.append({ "object": name, "sign": sign, "house": house, "direction": p.get("direction"), "planet_token": planet_tok, "planet_total": raw.get("total", 0), "facets": facets, }) return {"provider": provider, "objects": items}