Jyotisha/scripts/shadbala_service/api.py

"""`/api/shadbala`: the six-fold strength core plus the advanced evidence layer.

Moved out of ``JyotishAPIHandler`` for BUG-1463. The handler method used
``swe.julday`` with no ``swe`` in scope, so every request carrying
year/month/day/lat/lon failed with ``NameError`` (HTTP 500). This module
imports Swiss Ephemeris itself; the handler only forwards the body.

The handler is passed in for its existing body parsers (``_get_float``,
``_parse_timezone`` ...). They raise the server's ``BadRequest`` and that
propagates unchanged. This module never imports the HTTP server.
"""

from __future__ import annotations

import logging
from typing import Any

import swisseph as swe

from ayanamsa_utils import DEFAULT_AYANAMSA_NAME, UnsupportedAyanamsaError, normalize_ayanamsa_name

SIGNS = (
    "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
    "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
)
BIRTH_KEYS = ("year", "month", "day", "lat", "lon")

logger = logging.getLogger(__name__)


class ShadbalaInputError(ValueError):
    """The request names an ayanamsa the engine does not support."""


def _ayanamsa(body: dict[str, Any]) -> str:
    """Same reading as the server's ``_request_ayanamsa(body)``."""
    raw = body.get("ayanamsa") or body.get("ayanamsa_name") or body.get("ayanamsa_policy")
    if raw in (None, ""):
        return DEFAULT_AYANAMSA_NAME
    try:
        return normalize_ayanamsa_name(raw)
    except UnsupportedAyanamsaError as exc:
        raise ShadbalaInputError(str(exc)) from exc


def birth_context(handler: Any, body: dict[str, Any], birth_hour: float, birth_minute: float, birth_second: float) -> dict[str, Any] | None:
    """Precise Swiss Ephemeris context when the body carries the full birth moment."""
    if not all(key in body for key in BIRTH_KEYS):
        return None
    import shadbala

    year = handler._get_int(body, "year", 1990, 1800, 2400)
    month = handler._get_int(body, "month", 6, 1, 12)
    day = handler._get_int(body, "day", 15, 1, 31)
    lat = handler._get_float(body, "lat", 0, -90, 90)
    lon = handler._get_float(body, "lon", 0, -180, 180)
    tz = handler._parse_timezone(body, lat, lon, year, month, day, birth_hour, birth_minute, birth_second)
    hour_decimal = handler._birth_hour_decimal(birth_hour, birth_minute, birth_second)
    jd_ut = swe.julday(year, month, day, hour_decimal - tz)
    return shadbala.build_shadbala_context(jd_ut, lat, lon, _ayanamsa(body), tz)


def rule_variants(advanced_enabled: bool) -> dict[str, Any]:
    return {
        "selected": ["core_sixfold", "advanced_evidence"] if advanced_enabled else ["core_sixfold"],
        "available": [
            {"key": "core_sixfold", "label": "六重力量主算法", "status": "active", "source": "scripts/shadbala.py"},
            {"key": "advanced_evidence", "label": "Kala/Yuddha/Sputa 增强证据", "status": "active" if advanced_enabled else "unavailable", "source": "scripts/shadbala_advanced.py"},
        ],
        "boundary": "增强层只作为证据补充，不覆盖主 Shadbala 的总分与排名。",
    }


def advanced_layer(handler: Any, body: dict[str, Any], planets: dict[str, Any], base_result: Any) -> dict[str, Any]:
    """Kala sub-items, Yuddha and Sputa Drishti evidence; ``{}`` when it cannot be built."""
    from datetime import datetime

    try:
        import shadbala_advanced as advanced_mod

        birth_hour = handler._get_float(body, "birth_hour", body.get("hour", 12), 0, 23)
        birth_minute = handler._get_float(body, "birth_minute", body.get("minute", 0), 0, 59)
        birth_second = handler._get_birth_second(body)
        year = handler._get_int(body, "year", datetime.now().year, 1800, 2400)
        month = handler._get_int(body, "month", 1, 1, 12)
        day = handler._get_int(body, "day", 1, 1, 31)
        lat = handler._get_float(body, "lat", body.get("birth_lat", 0), -90, 90)
        lon = handler._get_float(body, "lon", body.get("birth_lon", 0), -180, 180)
        tz = handler._parse_timezone(body, lat, lon, year, month, day, birth_hour, birth_minute, birth_second)
        hour_decimal = handler._birth_hour_decimal(birth_hour, birth_minute, birth_second)
        solar_lon = planets.get("Sun", {}).get("lon", 0)
        base_planets = base_result.get("planets", {}) if isinstance(base_result, dict) else {}
        comparison_planets = {}
        kala_additions = {}
        drik_sputa = {}
        for planet, pdata in planets.items():
            if planet not in base_planets:
                continue
            comparison_planets[planet] = {
                **pdata,
                "degree": pdata.get("lon", pdata.get("degree", 0)),
                "shadbala_total": base_planets.get(planet, {}).get("total_virupas", 0),
            }
        for planet in comparison_planets:
            kala_additions[planet] = round(advanced_mod.calc_varsha_maasa_dina_hora_bala(
                planet, year, month, day, hour_decimal, solar_lon, lat, lon, tz
            ), 2)
            drik_sputa[planet] = round(advanced_mod.calc_drik_bala_sputa(planet, comparison_planets), 2)
        yuddha = advanced_mod.calc_yuddha_bala(comparison_planets)
        active_yuddha = {planet: value for planet, value in yuddha.items() if abs(value) > 0}
        top_kala = sorted(kala_additions.items(), key=lambda item: item[1], reverse=True)[:3]
        return {
            "source": "scripts/shadbala_advanced.py",
            "method": "Kala Bala完整子项 + Yuddha Bala + Sputa Drishti 证据层",
            "kala_vmdh": kala_additions,
            "top_kala_support": [{"planet": planet, "virupas": value} for planet, value in top_kala],
            "yuddha_bala": yuddha,
            "active_yuddha": active_yuddha,
            "sputa_drik_bala": drik_sputa,
            "next_action": "若高级层与主排名冲突，先保留主 Shadbala 排名，再把冲突作为需要人工复核的证据。",
        }
    except Exception as exc:
        logger.warning("[api_server] shadbala advanced layer failed: %s", exc)
        return {}


def compute_shadbala(handler: Any, body: dict[str, Any]) -> dict[str, Any]:
    import shadbala

    planets, planet_lons, asc_sign_idx = handler._normalized_planets_from_body(body)
    if not {"Sun", "Moon"} <= set(planet_lons):
        raise ShadbalaInputError("planets must include Sun and Moon longitude data")
    birth_hour = handler._get_float(body, "birth_hour", body.get("hour", 12), 0, 23)
    birth_minute = handler._get_float(body, "birth_minute", body.get("minute", 0), 0, 59)
    birth_second = handler._get_birth_second(body)
    birth_hour_decimal = handler._birth_hour_decimal(birth_hour, birth_minute, birth_second)
    context = birth_context(handler, body, birth_hour, birth_minute, birth_second)
    result = shadbala.calc_shadbala(
        planets,
        SIGNS[asc_sign_idx],
        birth_hour_decimal,
        planet_lons["Sun"],
        planet_lons["Moon"],
        birth_minute,
        context=context,
    )
    advanced = advanced_layer(handler, body, planets, result)
    return {
        "success": True,
        "endpoint": "shadbala",
        "rule_variants": rule_variants(bool(advanced)),
        "advanced_layer": advanced,
        "result": result,
    }