Jyotisha/scripts/pancha_mahapurusha.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Pancha Mahapurusha Yoga（五王瑜伽）完整检测模块 v1.0
基于BPHS和dashaflow (MIT) detect_yogas核心算法

五种瑜伽：
- Ruchaka Yoga (Mars)
- Bhadra Yoga (Mercury)
- Hamsa Yoga (Jupiter)
- Malavya Yoga (Venus)
- Shasha Yoga (Saturn)

破格口径按占星师定稿 2026-10-03 甲5（BUG-1207）：逆行不破格；燃烧、与凶星
同宫或被照只记「部分减弱」；D9 落陷判不成立（现状，待问）。本模块是五大人格的
唯一主检测器，`yoga_rules.json` 同名规则与 `yogas_doshas.calc_pancha_mahapurusha_yoga`
为 reference_only。
"""

from typing import Dict, List, Optional

SIGNS = ['Aries', 'Taurus', 'Gemini', 'Cancer', 'Leo', 'Virgo',
         'Libra', 'Scorpio', 'Sagittarius', 'Capricorn', 'Aquarius', 'Pisces']

KENDRA_HOUSES = {1, 4, 7, 10}
MAHAPURUSHA_PLANETS = {"Mars", "Mercury", "Jupiter", "Venus", "Saturn"}

MAHAPURUSHA_NAMES = {
    "Mars": "Ruchaka Yoga",
    "Mercury": "Bhadra Yoga",
    "Jupiter": "Hamsa Yoga",
    "Venus": "Malavya Yoga",
    "Saturn": "Shasha Yoga",
}

MAHAPURUSHA_EFFECTS = {
    "Ruchaka": "军事领袖、建筑大师、体魄强健、勇敢果断",
    "Bhadra": "演说家、学者、机智聪慧、商业头脑",
    "Hamsa": "精神导师、哲学家、富足慷慨、智慧卓越",
    "Malavya": "艺术家、享乐者、魅力超凡、审美敏锐",
    "Shasha": "政治家、长者、坚韧执着、权威地位",
}

EXALTATION = {'Sun':('Aries',10),'Moon':('Taurus',3),'Mars':('Capricorn',28),'Mercury':('Virgo',15),'Jupiter':('Cancer',5),'Venus':('Pisces',27),'Saturn':('Libra',20)}
OWN_SIGNS = {'Sun':{'Leo'},'Moon':{'Cancer'},'Mars':{'Aries','Scorpio'},'Mercury':{'Gemini','Virgo'},'Jupiter':{'Sagittarius','Pisces'},'Venus':{'Taurus','Libra'},'Saturn':{'Capricorn','Aquarius'}}

# Combustion orb (degrees from Sun)
COMBUSTION_ORB = {
    'Moon': {'direct': 12, 'retrograde': 12},
    'Mars': {'direct': 17, 'retrograde': 10},
    'Mercury': {'direct': 14, 'retrograde': 12},
    'Jupiter': {'direct': 11, 'retrograde': 10},
    'Venus': {'direct': 10, 'retrograde': 8},
    'Saturn': {'direct': 15, 'retrograde': 10},
}

NATURAL_MALEFICS = {"Saturn", "Mars", "Sun", "Rahu", "Ketu"}


def _is_exalted_or_own(planet: str, sign: str) -> bool:
    if planet in EXALTATION and EXALTATION[planet][0] == sign:
        return True
    if planet in OWN_SIGNS and sign in OWN_SIGNS[planet]:
        return True
    return False


def _is_combust(planet: str, planet_degree: float, sun_degree: float, retrograde: bool = False) -> bool:
    """检查行星是否被太阳燃烧"""
    profile = COMBUSTION_ORB.get(planet, {'direct': 10})
    orb = profile.get('retrograde' if retrograde else 'direct', profile.get('direct', 10))
    diff = abs(planet_degree - sun_degree)
    if diff > 180:
        diff = 360 - diff
    return diff <= orb


def _has_malefic_aspect(planet_sign: str, all_planets: Dict, planet: str = None) -> List[str]:
    """Natural malefics in the same sign as the yoga planet (the planet itself excluded, BUG-1180)."""
    afflicting = []
    for pn, pd in all_planets.items():
        if pn == planet or pn not in NATURAL_MALEFICS:
            continue
        # 简单合相检查
        if pd.get('sign') == planet_sign:
            afflicting.append(pn)
    return afflicting


# Parashari whole-sign graha drishti, as aspects.calc_house_aspects: every graha
# the 7th; Mars also 4 / 8, Saturn 3 / 10; Rahu / Ketu cast none.
MALEFIC_DRISHTI = {'Sun': (7,), 'Mars': (4, 7, 8), 'Saturn': (3, 7, 10)}


def _malefic_aspectors(planet_sign: str, all_planets: Dict, planet: str = None) -> List[str]:
    """Natural malefics whose graha drishti falls on the yoga planet's sign."""
    if planet_sign not in SIGNS:
        return []
    target = SIGNS.index(planet_sign)
    casters = []
    for pn, offsets in MALEFIC_DRISHTI.items():
        pd = all_planets.get(pn)
        if pn == planet or not isinstance(pd, dict) or pd.get('sign') not in SIGNS:
            continue
        origin = SIGNS.index(pd['sign'])
        if any((origin + offset - 1) % 12 == target for offset in offsets):
            casters.append(pn)
    return casters


def _longitude(pd: Dict) -> Optional[float]:
    """Sidereal longitude of a row; `degree` may be a longitude or a degree in sign."""
    for key in ('lon', 'degree_raw', 'longitude'):
        value = pd.get(key)
        if isinstance(value, (int, float)):
            return float(value) % 360.0
    degree = pd.get('degree')
    if not isinstance(degree, (int, float)):
        return None
    if degree >= 30 or pd.get('sign') not in SIGNS:
        return float(degree) % 360.0
    return SIGNS.index(pd['sign']) * 30.0 + float(degree)


MAHAPURUSHA_ORDER = ('Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn')
PLANET_ZH = {'Sun': '太阳', 'Mars': '火星', 'Mercury': '水星', 'Jupiter': '木星', 'Venus': '金星',
             'Saturn': '土星', 'Rahu': '罗睺', 'Ketu': '计都'}


def detect_pancha_mahapurusha(planets: Dict, sun_degree: float = None) -> List[Dict]:
    """
    检测所有Pancha Mahapurusha Yoga（主检测器）。

    占星师定稿 2026-10-03 甲5（BUG-1207）：
    - 只从命宫（Lagna）角宫计算：行星在上升起第 1/4/7/10 宫且入庙或入旺；
    - 逆行不破格；
    - 燃烧不破格，只记「部分减弱」（`strength: reduced`）；
    - 与凶星同宫或被凶星照只降低强度，不消灭格局；
    - D9 落陷判不成立（`is_valid: False`）保留现状：只在传入 `navamsa_sign` 时检查，
      产品路径不传；定稿未提及，待问占星师。

    Args:
        planets: {planet: {'sign': str, 'house': int, 'lon'|'degree': float, 'navamsa_sign'?: str}}
            `house` 从上升数；经度取 `lon` / `degree_raw` / `longitude`，否则 `degree`
            （≥30 视为经度，否则按星座内度数）。
        sun_degree: 太阳黄道经度（燃烧检测）；缺省时取 planets['Sun'] 的经度。

    Returns:
        检测到的PMC Yoga列表（含 strength: full / reduced / cancelled）
    """
    yogas = []
    sun_deg = sun_degree
    if sun_deg is None and isinstance(planets.get('Sun'), dict):
        sun_deg = _longitude(planets['Sun'])

    for planet in MAHAPURUSHA_ORDER:
        pd = planets.get(planet)
        if not pd:
            continue

        sign = pd.get('sign', '')
        house = pd.get('house', 0)

        # 基本条件：在Kendra + own/exalted
        if house not in KENDRA_HOUSES:
            continue
        if not _is_exalted_or_own(planet, sign):
            continue

        yoga_name = MAHAPURUSHA_NAMES[planet]
        effect = MAHAPURUSHA_EFFECTS.get(yoga_name.replace(" Yoga", ""), "")

        cancellations = []
        reductions = []
        reductions_en = []
        is_valid = True

        # 1. 燃烧：部分减弱（容许度表与 jyotish_engine.COMBUSTION_ORBS 一致，
        #    tests/test_combustion_orb_tables_agree.py）
        planet_deg = _longitude(pd)
        retro = bool(pd.get('retrograde'))
        if sun_deg is not None and planet_deg is not None and _is_combust(planet, planet_deg, sun_deg, retro):
            profile = COMBUSTION_ORB.get(planet, {'direct': 10})
            threshold = profile.get('retrograde' if retro else 'direct', profile.get('direct', 10))
            reductions.append(f"{PLANET_ZH[planet]}燃烧（距太阳 {threshold}° 以内）")
            reductions_en.append(f"{planet} combust (within {threshold}° of the Sun)")

        # 2. 逆行：不破格，不减弱（定稿甲5）

        # 3. 凶星同宫 / 凶星照：只降低强度
        malefics = _has_malefic_aspect(sign, planets, planet)
        if malefics:
            reductions.append(f"与凶星同宫：{'、'.join(PLANET_ZH.get(m, m) for m in malefics)}")
            reductions_en.append(f"conjunct malefic(s): {', '.join(malefics)}")
        aspectors = _malefic_aspectors(sign, planets, planet)
        if aspectors:
            reductions.append(f"受凶星照：{'、'.join(PLANET_ZH.get(m, m) for m in aspectors)}")
            reductions_en.append(f"aspected by malefic(s): {', '.join(aspectors)}")

        # 4. Navamsa弱势（D9中落陷）：不成立。现状保留——只在调用方给出
        #    `navamsa_sign` 时检查；产品路径（chart.yogas、规则引擎、报告）都不传，
        #    所以这一条目前不生效。定稿未提及，记为下轮问题，不在本单启用。
        navamsa_sign = pd.get('navamsa_sign', '')
        if navamsa_sign in OWN_SIGNS.get(planet, set()) or (planet in EXALTATION and EXALTATION[planet][0] == navamsa_sign):
            pass  # D9 also strong → yoga reinforced
        elif navamsa_sign:
            deb_sign_idx = (SIGNS.index(EXALTATION[planet][0]) + 6) % 12 if planet in EXALTATION else -1
            if deb_sign_idx >= 0 and navamsa_sign == SIGNS[deb_sign_idx]:
                cancellations.append(f"{planet}在Navamsa落陷 — Yoga根基不稳")
                is_valid = False

        strength = 'cancelled' if not is_valid else ('reduced' if reductions else 'full')
        yogas.append({
            'name': yoga_name,
            'planet': planet,
            'sign': sign,
            'house': house,
            'is_valid': is_valid,
            'strength': strength,
            'strength_cn': {'full': '完整', 'reduced': '部分减弱', 'cancelled': '不成立'}[strength],
            'effect': effect,
            'reductions': reductions,
            'reductions_en': reductions_en,
            'cancellations': cancellations,
        })

    return yogas


def assess_pmc_strength(planets: Dict, sun_degree: float = None) -> Dict:
    """
    评估Pancha Mahapurusha Yoga总体强度。

    Returns:
        PMC评估结果
    """
    yogas = detect_pancha_mahapurusha(planets, sun_degree)
    valid = [y for y in yogas if y['is_valid']]
    invalid = [y for y in yogas if not y['is_valid']]

    reduced = [y['name'] for y in valid if y['strength'] == 'reduced']
    if len(valid) >= 2:
        assessment = "多重大王瑜伽 — 命主在不同领域均有杰出天赋"
    elif len(valid) == 1:
        assessment = f"{valid[0]['name']} — 命主在特定领域有卓越潜力"
    elif len(invalid) > 0:
        st = [y['planet'] for y in invalid]
        assessment = f"瑜伽潜质存在但因{', '.join(st)}在 D9 落陷而不成立"
    else:
        assessment = "无Pancha Mahapurusha Yoga"

    return {
        'method': 'Pancha Mahapurusha Yoga 完整检测 v1.0',
        'yogas': yogas,
        'valid_count': len(valid),
        'reduced': reduced,
        'invalid_count': len(invalid),
        'assessment': assessment,
    }