Blame for resolution_advisor.py

#!/usr/bin/env python3
"""
紫微斗数化解建议系统 - Resolution Advisor
基于妈妈案例总结的实战化解策略

提供5大类化解建议:
1. 避灾策略 - 行为规避
2. 医疗应劫 - 主动化解血光
3. 精神转移 - 利用吉星能量
4. 定期体检 - 健康监控
5. 财富平衡 - 空劫化解

Created: 2025-12-14
Author: Sisi 🦊
"""

from typing import Dict, List, Any, Optional
from datetime import datetime


class ResolutionAdvisor:
    """化解建议顾问"""

    def __init__(self, ziwei_chart: Dict[str, Any]):
        """
        初始化化解建议顾问

        Args:
            ziwei_chart: 完整紫微命盘数据
        """
        self.chart = ziwei_chart
        self.palaces = ziwei_chart.get('palaces', [])
        self.chart_info = ziwei_chart.get('chart', {})

    def analyze_resolution_needs(
        self,
        sihua_analysis: Dict[str, Any],
        daxun_analysis: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        分析化解需求并生成建议

        Args:
            sihua_analysis: 四化分析结果
            daxun_analysis: 大运分析结果

        Returns:
            化解建议字典
        """

        # 1. 识别风险因素
        risks = self._identify_risks(sihua_analysis, daxun_analysis)

        # 2. 识别保护因素（吉星）
        protections = self._identify_protections(sihua_analysis)

        # 3. 生成化解建议
        suggestions = self._generate_suggestions(risks, protections, daxun_analysis)

        # 4. 生成优先级清单
        priority_actions = self._generate_priority_actions(risks, suggestions)

        return {
            'risks_identified': risks,
            'protections_identified': protections,
            'resolution_strategies': suggestions,
            'priority_actions': priority_actions,
            'overall_advice': self._generate_overall_advice(risks, protections)
        }

    def _identify_risks(
        self,
        sihua_analysis: Dict[str, Any],
        daxun_analysis: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        """识别风险因素"""
        risks = []

        # 1. 化忌风险
        year_sihua = sihua_analysis.get('year_sihua', {})
        huaji = year_sihua.get('化忌', {})

        if huaji:
            star = huaji.get('star', '')
            palace = huaji.get('palace', '')
            impact = huaji.get('impact_score', 0)

            risk_level = '高' if impact >= 80 else '中' if impact >= 60 else '低'

            risk = {
                'type': '流年化忌',
                'star': star,
                'palace': palace,
                'impact_score': impact,
                'risk_level': risk_level,
                'description': f'{star}化忌在{palace}',
                'concerns': self._get_huaji_concerns(star, palace)
            }
            risks.append(risk)

        # 2. 大运交接期风险
        current_daxun = daxun_analysis.get('current', {})
        if current_daxun and 'error' not in current_daxun:
            daxun_year = current_daxun.get('daxun_year', 0)
            is_transition = current_daxun.get('is_transition', False)

            if is_transition or daxun_year >= 9:
                risks.append({
                    'type': '大运交接期',
                    'daxun_year': daxun_year,
                    'palace': current_daxun.get('palace_name', ''),
                    'risk_level': '高' if daxun_year == 10 else '中',
                    'description': f'大运第{daxun_year}年，即将交接',
                    'concerns': ['运势波动', '重大变动', '需要谨慎']
                })

        # 3. 疾厄空宫风险
        jiee_palace = self._find_palace('疾厄')
        if jiee_palace:
            is_empty = len(jiee_palace.get('major_stars', [])) == 0
            if is_empty:
                risks.append({
                    'type': '疾厄空宫',
                    'risk_level': '长期',
                    'description': '疾厄宫无主星，健康根基不稳',
                    'concerns': ['体弱多病', '易受流年冲击', '需持续监控']
                })

        # 4. 迁移空宫风险（如果有化忌相关）
        qianyi_palace = self._find_palace('迁移')
        if qianyi_palace:
            is_empty = len(qianyi_palace.get('major_stars', [])) == 0
            if is_empty and huaji.get('palace') == '迁移':
                risks.append({
                    'type': '迁移空宫+化忌',
                    'risk_level': '高',
                    'description': '迁移宫空宫受化忌冲击',
                    'concerns': ['外出不利', '变动风险', '搬迁谨慎']
                })

        return risks

    def _identify_protections(self, sihua_analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
        """识别保护因素（吉星、化禄、化权、化科）"""
        protections = []

        year_sihua = sihua_analysis.get('year_sihua', {})

        # 1. 化禄保护
        hulu = year_sihua.get('化禄', {})
        if hulu:
            protections.append({
                'type': '化禄',
                'star': hulu.get('star', ''),
                'palace': hulu.get('palace', ''),
                'power': hulu.get('impact_score', 0),
                'benefits': self._get_hulu_benefits(hulu.get('star', ''), hulu.get('palace', ''))
            })

        # 2. 化权保护
        huquan = year_sihua.get('化权', {})
        if huquan:
            protections.append({
                'type': '化权',
                'star': huquan.get('star', ''),
                'palace': huquan.get('palace', ''),
                'power': huquan.get('impact_score', 0),
                'benefits': [f'{huquan.get("star", "")}得权，能量增强']
            })

        # 3. 化科保护
        huke = year_sihua.get('化科', {})
        if huke:
            protections.append({
                'type': '化科',
                'star': huke.get('star', ''),
                'palace': huke.get('palace', ''),
                'power': huke.get('impact_score', 0),
                'benefits': [f'{huke.get("star", "")}化科，贵人相助']
            })

        return protections

    def _generate_suggestions(
        self,
        risks: List[Dict[str, Any]],
        protections: List[Dict[str, Any]],
        daxun_analysis: Dict[str, Any]
    ) -> Dict[str, Any]:
        """生成化解建议"""

        suggestions = {
            'avoidance_strategy': [],      # 避灾策略
            'medical_remedy': [],           # 医疗应劫
            'spiritual_elevation': [],      # 精神转移
            'health_monitoring': [],        # 定期体检
            'wealth_balance': []            # 财富平衡
        }

        # 基于风险类型生成建议
        for risk in risks:
            risk_type = risk['type']

            # 1. 避灾策略
            if risk_type in ['流年化忌', '迁移空宫+化忌', '大运交接期']:
                suggestions['avoidance_strategy'].extend(
                    self._get_avoidance_advice(risk)
                )

            # 2. 医疗应劫
            if risk_type == '流年化忌' and risk.get('star') in ['廉贞', '巨门', '擎羊', '陀罗']:
                suggestions['medical_remedy'].extend(
                    self._get_medical_remedy_advice(risk)
                )

            # 3. 健康监控
            if risk_type in ['疾厄空宫', '流年化忌']:
                suggestions['health_monitoring'].extend(
                    self._get_health_monitoring_advice(risk, daxun_analysis)
                )

        # 基于保护因素生成建议
        for protection in protections:
            if protection['type'] == '化禄':
                # 精神转移 - 利用化禄能量
                suggestions['spiritual_elevation'].extend(
                    self._get_spiritual_elevation_advice(protection)
                )

        # 财富平衡建议（通用）
        suggestions['wealth_balance'] = self._get_wealth_balance_advice()

        return suggestions

    def _get_huaji_concerns(self, star: str, palace: str) -> List[str]:
        """获取化忌星曜的关注点"""

        concerns_map = {
            '廉贞': ['健康问题', '血光之灾', '炎症肿瘤', '情绪波动'],
            '巨门': ['口舌是非', '沟通障碍', '暗疾隐患', '人际纠纷'],
            '太阴': ['情绪低落', '女性健康', '财务波动', '感情困扰'],
            '贪狼': ['过度消费', '感情纠葛', '健康透支', '冲动决策'],
            '天机': ['计划变动', '过度焦虑', '神经紧张', '决策犹豫'],
            '文昌': ['考试失利', '文书错误', '学习障碍', '名誉受损'],
            '文曲': ['文书问题', '合同纠纷', '学业不顺', '名声受损']
        }

        base_concerns = concerns_map.get(star, ['需要谨慎'])

        # 如果化忌在疾厄宫，强调健康问题
        if palace == '疾厄':
            return ['健康警示'] + base_concerns
        elif palace == '财帛':
            return ['财务风险'] + base_concerns
        elif palace == '夫妻':
            return ['感情波折'] + base_concerns
        else:
            return base_concerns

    def _get_hulu_benefits(self, star: str, palace: str) -> List[str]:
        """获取化禄星曜的益处"""

        benefits_map = {
            '天同': ['福星护体', '享乐能力强', '逢凶化吉', '心情愉悦', '贵人相助'],
            '天机': ['智慧增长', '贵人相助', '计划顺利', '灵活应变'],
            '太阳': ['光明正大', '贵人多助', '事业顺利', '能量充沛'],
            '武曲': ['财运亨通', '决断力强', '事业有成', '收入增加'],
            '紫微': ['帝星降临', '权威提升', '贵人帮扶', '事业发展'],
            '太阴': ['财运提升', '感情和睦', '女性贵人', '柔和处事']
        }

        base_benefits = benefits_map.get(star, ['好运加持'])

        # 如果化禄在命宫，强调全年保护
        if palace == '命宫':
            return ['全年受到强力保护'] + base_benefits
        elif palace == '财帛':
            return ['财运亨通'] + base_benefits
        else:
            return base_benefits

    def _get_avoidance_advice(self, risk: Dict[str, Any]) -> List[Dict[str, str]]:
        """获取避灾策略建议"""

        advice = []
        risk_type = risk['type']

        if risk_type == '大运交接期':
            advice.append({
                'category': '大运交接谨慎',
                'action': '不宜重大变动',
                'detail': '避免搬家、换工作、大额投资等重大决策',
                'priority': '高'
            })
            advice.append({
                'category': '静守为宜',
                'action': '保持稳定',
                'detail': '发挥守成特质，稳中求安，不宜冒险',
                'priority': '高'
            })

        if risk_type in ['流年化忌', '迁移空宫+化忌']:
            palace = risk.get('palace', '')
            if palace in ['迁移', '命宫']:
                advice.append({
                    'category': '减少外出',
                    'action': '避免远行',
                    'detail': '尽量减少长途旅行、搬迁等变动，外出需谨慎',
                    'priority': '中'
                })

        return advice

    def _get_medical_remedy_advice(self, risk: Dict[str, Any]) -> List[Dict[str, str]]:
        """获取医疗应劫建议"""

        star = risk.get('star', '')

        if star == '廉贞':
            return [
                {
                    'category': '主动验血',
                    'action': '年初精密体检',
                    'detail': '心血管指标、炎症指标、肿瘤标志物、甲状腺功能全面检查',
                    'priority': '高',
                    'timing': '农历一月或二月'
                },
                {
                    'category': '小手术应劫',
                    'action': '处理小问题',
                    'detail': '如有牙齿、皮肤、静脉曲张等小问题，主动预约处理（人工应劫）',
                    'priority': '中',
                    'timing': '农历一月或二月'
                },
                {
                    'category': '献血行善',
                    'action': '主动献血',
                    'detail': '如身体允许，小额献血200ml，既应劫又行善',
                    'priority': '低',
                    'timing': '全年适宜'
                }
            ]

        if star == '巨门':
            return [
                {
                    'category': '口腔咽喉检查',
                    'action': '口腔及消化系统检查',
                    'detail': '巨门主口舌，化忌时注意口腔溃疡、咽喉炎、肠胃功能紊乱，建议做胃镜或消化科检查',
                    'priority': '高',
                    'timing': '农历一月或流年化忌月份'
                },
                {
                    'category': '心理减压',
                    'action': '管理口舌是非带来的心理压力',
                    'detail': '巨门化忌易招口舌是非、被误解，长期压力影响消化和睡眠。建议定期心理疏导或冥想练习',
                    'priority': '中',
                    'timing': '全年持续'
                }
            ]

        if star == '擎羊':
            return [
                {
                    'category': '外伤血光预防',
                    'action': '年度全面体检 + 避免高风险活动',
                    'detail': '擎羊主刀伤血光，化忌加重。避免极限运动、注意交通安全、厨房用火用刀谨慎，年初做外科体检',
                    'priority': '高',
                    'timing': '农历一月体检，全年注意安全'
                },
                {
                    'category': '牙科眼科检查',
                    'action': '主动排查头面部问题',
                    'detail': '擎羊与头面部相关，化忌时注意牙齿、眼睛、头部问题，主动预约检查可人工应劫',
                    'priority': '中',
                    'timing': '农历一月或二月'
                }
            ]

        if star == '陀罗':
            return [
                {
                    'category': '慢性病排查',
                    'action': '针对性慢性病检查',
                    'detail': '陀罗主暗疾拖延，化忌时潜伏的慢性病容易加重。重点检查：关节、皮肤、慢性炎症、甲状腺等容易被忽视的问题',
                    'priority': '高',
                    'timing': '农历一月全面检查'
                },
                {
                    'category': '运动调理',
                    'action': '规律低强度运动',
                    'detail': '陀罗性质拖沓粘滞，通过规律运动（瑜伽、游泳、散步）促进气血流通，防止暗疾淤积',
                    'priority': '中',
                    'timing': '全年坚持，每周3-4次'
                }
            ]

        return []

    def _get_health_monitoring_advice(
        self,
        risk: Dict[str, Any],
        daxun_analysis: Dict[str, Any]
    ) -> List[Dict[str, str]]:
        """获取健康监控建议"""

        advice = []

        # 根据风险等级确定体检频率
        risk_level = risk.get('risk_level', '中')

        current_daxun = daxun_analysis.get('current', {})
        daxun_year = current_daxun.get('daxun_year', 0)

        # 大运末年或高风险 - 每3个月
        if risk_level == '高' or daxun_year >= 9:
            frequency = '每3个月一次'
            priority = '高'
        # 中等风险 - 每4-6个月
        elif risk_level == '中':
            frequency = '每4-6个月一次'
            priority = '中'
        # 低风险或长期监控 - 每6-12个月
        else:
            frequency = '每6-12个月一次'
            priority = '中'

        advice.append({
            'category': '定期体检',
            'action': f'全面体检{frequency}',
            'detail': '心血管、甲状腺、消化系统、肿瘤筛查、血液检查',
            'priority': priority
        })

        # 如果是疾厄空宫，添加长期监控建议
        if risk.get('type') == '疾厄空宫':
            advice.append({
                'category': '长期健康管理',
                'action': '建立健康档案',
                'detail': '记录体检结果，追踪健康指标变化，早发现早治疗',
                'priority': '高'
            })

        return advice

    def _get_spiritual_elevation_advice(self, protection: Dict[str, Any]) -> List[Dict[str, str]]:
        """获取精神转移建议（利用吉星能量）"""

        star = protection.get('star', '')
        palace = protection.get('palace', '')

        advice = []

        # 天同化禄特别建议
        if star == '天同' and palace == '命宫':
            advice.extend([
                {
                    'category': '多花钱买快乐',
                    'action': '提升生活品质',
                    'detail': '买好吃的、买好用的、买喜欢的，不要省钱！钱花在快乐上就是化解',
                    'priority': '高'
                },
                {
                    'category': '培养享乐习惯',
                    'action': '每天开心一件事',
                    'detail': '听音乐、看电影、逛公园、和朋友聊天、养花养鱼',
                    'priority': '高'
                },
                {
                    'category': '保持好心情',
                    'action': '心宽体胖',
                    'detail': '心情愉悦是抵抗病气的最好盾牌，避免生气焦虑担忧',
                    'priority': '高'
                }
            ])

        # 其他化禄建议
        elif protection['type'] == '化禄':
            advice.append({
                'category': f'{star}化禄能量',
                'action': '发挥正面能量',
                'detail': f'{star}化禄在{palace}，多从事相关领域活动，增强好运',
                'priority': '中'
            })

        return advice

    def _get_wealth_balance_advice(self) -> List[Dict[str, str]]:
        """获取财富平衡建议（空劫化解）"""

        # 检查福德宫是否有地空地劫
        fude_palace = self._find_palace('福德')
        has_kongxie = False

        if fude_palace:
            minor_stars = fude_palace.get('minor_stars', [])
            star_names = []
            for star in minor_stars:
                if isinstance(star, dict):
                    star_names.append(star.get('name', ''))
                else:
                    star_names.append(str(star))

            has_kongxie = '地空' in star_names or '地劫' in star_names

        if has_kongxie:
            return [
                {
                    'category': '公益捐赠',
                    'action': '持续小额捐赠',
                    'detail': '每月固定金额（如$50-100）支持健康、教育、扶贫等领域',
                    'priority': '中',
                    'principle': '最好的"劫"，化解空劫能量'
                },
                {
                    'category': '精神学习',
                    'action': '探索内在世界',
                    'detail': '学习命理、哲学、宗教，参加冥想、瑜伽、太极',
                    'priority': '中',
                    'principle': '最好的"空"，精神满足不能来自储蓄'
                },
                {
                    'category': '高雅享受',
                    'action': '提升幸福感',
                    'detail': '旅游、艺术品、茶道、园艺等非必需但提升生活品质的支出',
                    'priority': '低',
                    'principle': '财富流动才是真正富足'
                },
                {
                    'category': '财富平衡公式',
                    'action': '90%实际+10%虚空',
                    'detail': '财富管理 = 90%实际需求（守库）+ 10%虚空支出（化解空劫）',
                    'priority': '中',
                    'principle': '舍得舍得，有舍才有得'
                }
            ]

        # 如果没有地空地劫，返回通用财富建议
        return [
            {
                'category': '量入为出',
                'action': '合理理财',
                'detail': '避免投机、高风险投资，稳健理财为主',
                'priority': '中'
            }
        ]

    def _generate_priority_actions(
        self,
        risks: List[Dict[str, Any]],
        suggestions: Dict[str, Any]
    ) -> List[Dict[str, str]]:
        """生成优先级行动清单"""

        all_actions = []

        # 收集所有建议
        for category, items in suggestions.items():
            for item in items:
                if isinstance(item, dict):
                    all_actions.append({
                        **item,
                        'strategy_type': category
                    })

        # 按优先级排序
        priority_order = {'高': 0, '中': 1, '低': 2}
        all_actions.sort(key=lambda x: priority_order.get(x.get('priority', '中'), 1))

        # 返回前10个最重要的行动
        return all_actions[:10]

    def _generate_overall_advice(
        self,
        risks: List[Dict[str, Any]],
        protections: List[Dict[str, Any]]
    ) -> str:
        """生成总体建议"""

        total_risk_score = sum(r.get('impact_score', 0) for r in risks if 'impact_score' in r)
        total_protection_score = sum(p.get('power', 0) for p in protections)

        # 吉凶比例
        if total_protection_score > 0 and total_risk_score > 0:
            ratio = total_protection_score / total_risk_score

            if ratio >= 3.0:
                level = '可控'
                emoji = '✅'
                advice = '吉力远大于凶力，虽有风险但整体可控。保持警惕，主动化解，定期体检，保持好心情。'
            elif ratio >= 2.0:
                level = '中等'
                emoji = '⚠️'
                advice = '吉力大于凶力，需要重视但不必恐慌。积极采取化解措施，定期监控，谨慎行事。'
            else:
                level = '较高'
                emoji = '🔴'
                advice = '凶力较强，需要特别谨慎。务必采取全面化解措施，密切监控，寻求专业帮助。'

            return f"{emoji} 风险等级：{level}。{advice}吉凶比例：{ratio:.1f}:1（吉{total_protection_score}分 vs 凶{total_risk_score}分）"

        elif total_risk_score > 0:
            return f"⚠️ 有风险因素但缺少吉星保护，需要格外谨慎。建议全面采取化解措施。"

        else:
            return "✅ 整体运势平稳，保持良好心态，积极向上。"

    def _find_palace(self, palace_name: str) -> Optional[Dict[str, Any]]:
        """查找指定宫位"""
        for palace in self.palaces:
            if palace.get('name') == palace_name:
                return palace
        return None


def generate_resolution_advice(
    ziwei_chart: Dict[str, Any],
    sihua_analysis: Dict[str, Any],
    daxun_analysis: Dict[str, Any]
) -> Dict[str, Any]:
    """
    生成化解建议（便捷函数）

    Args:
        ziwei_chart: 紫微命盘数据
        sihua_analysis: 四化分析结果
        daxun_analysis: 大运分析结果

    Returns:
        完整的化解建议
    """
    advisor = ResolutionAdvisor(ziwei_chart)
    return advisor.analyze_resolution_needs(sihua_analysis, daxun_analysis)


if __name__ == "__main__":
    # 测试代码
    print("🔮 化解建议系统测试")
    print("=" * 80)

    # 使用妈妈案例测试
    import json

    try:
        with open('mama_chart_full_result.json', 'r', encoding='utf-8') as f:
            mama_chart = json.load(f)

        from sihua_analyzer import SihuaAnalyzer
        from daxun_analyzer import DaxunAnalyzer

        # 分析四化
        sihua_analyzer = SihuaAnalyzer(mama_chart)
        sihua_analysis = sihua_analyzer.analyze_all_sihua()

        # 分析大运（2026年，71岁）
        daxun_analyzer = DaxunAnalyzer(mama_chart)
        current_daxun = daxun_analyzer.analyze_current_daxun(71)
        next_daxun = daxun_analyzer.predict_next_daxun(71)

        daxun_analysis = {
            'current': current_daxun,
            'next': next_daxun
        }

        # 生成化解建议
        resolution = generate_resolution_advice(mama_chart, sihua_analysis, daxun_analysis)

        print("\n📊 风险识别:")
        for risk in resolution['risks_identified']:
            print(f"  - {risk['type']}: {risk['description']} (风险等级: {risk.get('risk_level', 'N/A')})")

        print("\n✨ 保护因素:")
        for protection in resolution['protections_identified']:
            print(f"  - {protection['type']}: {protection['star']}在{protection['palace']} (能量: {protection['power']}分)")

        print("\n💡 优先行动清单:")
        for i, action in enumerate(resolution['priority_actions'][:5], 1):
            print(f"  {i}. [{action.get('priority', '中')}] {action.get('category', '')}: {action.get('action', '')}")

        print(f"\n🎯 总体建议:")
        print(f"  {resolution['overall_advice']}")

        print("\n" + "=" * 80)
        print("✅ 测试完成！")

    except FileNotFoundError:
        print("❌ 找不到测试文件 mama_chart_full_result.json")
    except Exception as e:
        print(f"❌ 测试失败: {str(e)}")
        import traceback
        traceback.print_exc()


# 🦊 Generated with love by Sisi