astrololo/services/logic/app/prompt.py

"""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,
    "huge": 120_000,
    # „maksymalny kontekst modelu" — wyliczany dynamicznie z okna kontekstu
    # wybranego modelu, ZAWSZE po odjęciu miejsca zarezerwowanego na odpowiedź.
    # Wartość poniżej jest tylko zapasem, gdy limity modelu są nieznane.
    "max": 400_000,
}
DEFAULT_BUDGET = "medium"
MAX_BUDGET = "max"

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, budget_chars: int | None = None) -> int:
    """Limit znaków promptu. `budget_chars` nadpisuje tabelę — używane dla opcji
    „maksymalny kontekst modelu", gdzie limit zależy od wybranego modelu i musi
    być policzony po odjęciu miejsca zarezerwowanego na odpowiedź."""
    if budget not in BUDGETS:
        raise ValueError(
            f"Nieznany budżet: {budget!r} (dostępne: {', '.join(BUDGETS)})"
        )
    if budget_chars and budget_chars > 0:
        return budget_chars
    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,
                       budget_chars: int | None = None) -> dict:
    """Prompt na horoskop urodzeniowy (ekran „Interpretacje")."""
    limit = _budget_chars(budget, budget_chars)
    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,
                        budget_chars: int | None = None) -> dict:
    """Prompt na horoskop okresowy (ekran „Kalendarz")."""
    limit = _budget_chars(budget, budget_chars)
    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")