#!/usr/bin/env python3 """Assembler for the KP three-year monthly report packet.""" from __future__ import annotations import sys from datetime import datetime from pathlib import Path from types import SimpleNamespace try: # pragma: no cover - import path differs under CLI vs pytest from domain_calculation_service import compute_chart from kp_monthly_report_contract import build_kp_monthly_report_contract from kp_monthly_theme_support import build_kp_monthly_theme_support from kp_monthly_transits import build_monthly_transit_snapshot from kp_monthly_vimshottari import build_monthly_vimshottari_snapshot from kp_system import build_kp_western_support_surface, kp_maturity_profile except ImportError: # pragma: no cover from scripts.domain_calculation_service import compute_chart from scripts.kp_monthly_report_contract import build_kp_monthly_report_contract from scripts.kp_monthly_theme_support import build_kp_monthly_theme_support from scripts.kp_monthly_transits import build_monthly_transit_snapshot from scripts.kp_monthly_vimshottari import build_monthly_vimshottari_snapshot from scripts.kp_system import build_kp_western_support_surface, kp_maturity_profile _HUMAN_LABELS = { "Jupiter": "木星", "Saturn": "土星", "Rahu": "北交点", "Ketu": "南交点", "Aries": "白羊座", "Taurus": "金牛座", "Gemini": "双子座", "Cancer": "巨蟹座", "Leo": "狮子座", "Virgo": "处女座", "Libra": "天秤座", "Scorpio": "天蝎座", "Sagittarius": "射手座", "Capricorn": "摩羯座", "Aquarius": "水瓶座", "Pisces": "双鱼座", } def _add_months(start_year: int, start_month: int, offset: int) -> tuple[int, int]: month_index = (start_month - 1) + offset return start_year + (month_index // 12), (month_index % 12) + 1 def _label(value: str | None) -> str: if not value: return "-" return _HUMAN_LABELS.get(value, value) def _build_slow_planet_signals( current_transits: dict, next_transits: dict | None, ) -> list[str]: signals = [] current_planets = current_transits.get("planets") if isinstance(current_transits.get("planets"), dict) else {} next_planets = next_transits.get("planets") if isinstance(next_transits, dict) and isinstance(next_transits.get("planets"), dict) else {} for planet in ("Jupiter", "Saturn", "Rahu", "Ketu"): current = current_planets.get(planet) if isinstance(current_planets.get(planet), dict) else {} nxt = next_planets.get(planet) if isinstance(next_planets.get(planet), dict) else {} current_sign = current.get("sign") next_sign = nxt.get("sign") if current_sign and next_sign and current_sign != next_sign: signals.append(f"{_label(planet)}换座：{_label(current_sign)}→{_label(next_sign)}") elif current.get("retrograde"): signals.append(f"{_label(planet)}逆行强调") return signals def _highlight_reason(row: dict) -> str: signals = row.get("slow_planet_signals") if isinstance(row.get("slow_planet_signals"), list) else [] if signals: return "；".join(str(item) for item in signals[:2]) levels = row.get("vimshottari_five_levels") if isinstance(row.get("vimshottari_five_levels"), dict) else {} level_rows = levels.get("levels") if isinstance(levels.get("levels"), dict) else {} md = level_rows.get("mahadasha") if isinstance(level_rows.get("mahadasha"), dict) else {} ad = level_rows.get("antardasha") if isinstance(level_rows.get("antardasha"), dict) else {} return f"主运/次运焦点：{_label(md.get('lord'))} / {_label(ad.get('lord'))}" def _build_yearly_highlights(month_rows: list[dict]) -> list[dict]: years: dict[int, list[dict]] = {} for row in month_rows: month = str(row.get("month") or "") if len(month) < 4: continue year = int(month[:4]) years.setdefault(year, []).append(row) yearly = [] for year in sorted(years): ranked = sorted( years[year], key=lambda item: ( len(item.get("slow_planet_signals") or []), 1 if any("换座" in str(signal) for signal in (item.get("slow_planet_signals") or [])) else 0, ), reverse=True, ) picks = [] for row in ranked: picks.append({"month": row.get("month"), "reason": _highlight_reason(row), "status": "parameter_sensitive"}) if len(picks) == 4: break yearly.append({"year": year, "months": picks, "status": "parameter_sensitive"}) return yearly def _resolve_cmd_kp(): try: # pragma: no cover - script execution path from __main__ import cmd_kp as resolved return resolved except ImportError: pass scripts_dir = Path(__file__).resolve().parent if str(scripts_dir) not in sys.path: sys.path.insert(0, str(scripts_dir)) try: # pragma: no cover - pytest/module import path from jyotish_engine import cmd_kp as resolved return resolved except ImportError: # pragma: no cover from scripts.jyotish_engine import cmd_kp as resolved return resolved def build_kp_monthly_report_packet( *, birth_payload: dict, start_month: str, month_count: int, western_support: dict | None = None, ) -> dict: start_year, start_month_number = [int(part) for part in start_month.split("-", 1)] natal_chart = compute_chart({**birth_payload, "ayanamsa": "lahiri", "node_mode": "mean"}) moon_lon = natal_chart.get("planets", {}).get("Moon", {}).get("degree_raw") kp_args = SimpleNamespace( year=birth_payload["year"], month=birth_payload["month"], day=birth_payload["day"], hour=birth_payload["hour"], minute=birth_payload["minute"], second=birth_payload.get("second", 0), lat=birth_payload["lat"], lon=birth_payload["lon"], tz=birth_payload["tz"], node_mode="mean", ayanamsa="lahiri", ) natal_kp = _resolve_cmd_kp()(kp_args) packet = build_kp_monthly_report_contract( start_month=start_month, month_count=month_count, timezone_offset=float(birth_payload["tz"]), ) maturity_profile = kp_maturity_profile() western_support_surface = build_kp_western_support_surface( western_support, maturity_profile=maturity_profile, ) packet["supporting_systems"]["western_kp_support"] = western_support_surface packet["western_support"] = western_support_surface packet["months"] = [] month_rows = [] for offset in range(month_count): year, month = _add_months(start_year, start_month_number, offset) anchor_dt = datetime(year, month, 1, 12, 0, 0) monthly_levels = build_monthly_vimshottari_snapshot( birth_dt=datetime( birth_payload["year"], birth_payload["month"], birth_payload["day"], birth_payload["hour"], birth_payload["minute"], birth_payload.get("second", 0), ), moon_lon=moon_lon, anchor_dt=anchor_dt, ) monthly_transits = build_monthly_transit_snapshot( year=year, month=month, lat=float(birth_payload["lat"]), lon=float(birth_payload["lon"]), tz=float(birth_payload["tz"]), ) theme_support = build_kp_monthly_theme_support( natal_kp=natal_kp, monthly_levels=monthly_levels.get("levels") or {}, monthly_transits=monthly_transits, ) month_rows.append( { "month": f"{year:04d}-{month:02d}", "anchor_local": monthly_transits["anchor_local"], "vimshottari_five_levels": monthly_levels, "monthly_transits": monthly_transits, "theme_support": theme_support, "status": "parameter_sensitive", "must_not_claim": packet["must_not_claim"], } ) for index, row in enumerate(month_rows): next_transits = None if index + 1 < len(month_rows): next_transits = month_rows[index + 1].get("monthly_transits") row["slow_planet_signals"] = _build_slow_planet_signals( row.get("monthly_transits") or {}, next_transits, ) packet["months"] = month_rows packet["yearly_highlights"] = _build_yearly_highlights(month_rows) packet["natal_kp"] = natal_kp return packet