""" 🔮 东玄 Ziwei Doushu API v4.0 Professional Purple Star Astrology service with AI-powered analysis Developed by Sisi & TH Studio Updated: 2026-02-03 (🌟 v4.0 巴纳姆效果+六亲分析+10步批命流程！让解读从"机械断语"变成"温暖有共鸣的分析") """ from fastapi import FastAPI, HTTPException, Header, Depends from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import Literal, Optional, Dict, Any import subprocess import json import os import secrets import threading from datetime import datetime from sihua_analyzer import SihuaAnalyzer from ai_interpreter import AIInterpreter from knowledge_enhanced_interpreter import KnowledgeEnhancedInterpreter from daxun_analyzer import DaxunAnalyzer from resolution_advisor import ResolutionAdvisor from evidence import build_meta, build_strength, build_timing, build_evidence, build_lu_ji_flow, build_flow_year_sihua from liuyue_analyzer import LiuyueAnalyzer app = FastAPI( title="东玄 Ziwei Doushu API", version="4.0.0", description="🔮 Professional Purple Star Astrology API - 🆕v4.0巴纳姆温暖解读 + 六亲分析 + 10步批命流程 + 流月运势预测 + 1000+条星曜规则 + AI解读 (v4.0.0! 🌟✨)" ) # CORS configuration app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) # 🔐 API Key Configuration # Protected endpoints require x-api-key. Health/warmup/root stay public. API_KEY = os.environ.get("API_KEY") if not API_KEY: print("🔒 API_KEY is not configured. Protected endpoints will return 503.") def verify_api_key(x_api_key: Optional[str] = Header(None)): """Verify API key from x-api-key header; fail closed when unconfigured.""" if not API_KEY: raise HTTPException( status_code=503, detail="🔒 Server API key is not configured. Service unavailable." ) if not x_api_key or not secrets.compare_digest(x_api_key, API_KEY): raise HTTPException( status_code=403, detail="🔒 Invalid or missing API key. Please contact SisiTheFox for access." ) return x_api_key MAX_CONCURRENT_ZIWEI = int(os.environ.get("ZIWEI_MAX_CONCURRENT", "2")) _ziwei_subprocess_slots = threading.BoundedSemaphore(MAX_CONCURRENT_ZIWEI) _node_dependency_lock = threading.Lock() def ensure_node_dependencies() -> None: """Install Node dependencies on runtimes that do not run Docker build steps.""" if os.path.isdir("node_modules/iztro"): return with _node_dependency_lock: if os.path.isdir("node_modules/iztro"): return result = subprocess.run( ["npm", "install", "--production"], capture_output=True, text=True, timeout=180 ) if result.returncode != 0: raise HTTPException( status_code=500, detail=f"Node dependency installation failed: {result.stderr or result.stdout}" ) class ZiweiRequest(BaseModel): """紫微斗数计算请求""" birth_date: str = Field(..., description="出生日期 YYYY-MM-DD", example="2000-01-01") birth_time: str = Field(..., description="出生时间 HH:MM", example="14:30") gender: Literal['男', '女'] = Field(..., description="性别: 男/女", example="男") is_leap_month: bool = Field(default=False, description="是否闰月") fix_leap: bool = Field(default=True, description="是否修正闰月") def run_ziwei_calculation(request: BaseModel) -> Dict[str, Any]: """调用Node.js iztro服务进行基础排盘,返回命盘dict 接受ZiweiRequest或MonthlyFortuneRequest(共享排盘字段)。 统一处理subprocess调用、returncode检查和JSON解析, 失败时直接抛出HTTPException。 """ ensure_node_dependencies() input_data = { "birth_date": request.birth_date, "birth_time": request.birth_time, "gender": request.gender, "is_leap_month": request.is_leap_month, "fix_leap": request.fix_leap } if not _ziwei_subprocess_slots.acquire(blocking=False): raise HTTPException( status_code=429, detail="Ziwei service is busy. Please retry shortly." ) try: result = subprocess.run( ['node', 'ziwei_service.js'], input=json.dumps(input_data), capture_output=True, text=True, timeout=10 ) finally: _ziwei_subprocess_slots.release() if result.returncode != 0: raise HTTPException( status_code=500, detail=f"Ziwei calculation error: {result.stderr}" ) return json.loads(result.stdout) class ZiweiResponse(BaseModel): """紫微斗数计算响应""" status: str data: Optional[Dict[str, Any]] = None error: Optional[str] = None @app.get("/") def root(): """API健康检查""" return { "status": "online", "service": "东玄 Ziwei Doushu API - 天機玄鑑", "version": "4.0.0", "updated": "2026-02-03 (🌟 v4.0 巴纳姆温暖效果+六亲分析+10步批命流程！)", "documentation": "/docs", "website": "https://astrologybazi.com", "developed_by": "Sisi & TH Studio 🦊💕", "endpoints": { "basic": "/calculate-ziwei - 基础排盘", "v2_enhanced": "/analyze-ziwei-enhanced - 四化飞星 + AI解读", "v3_knowledge": "/analyze-ziwei-knowledge-enhanced - 知识库增强版", "v4_barnum": "/analyze-ziwei-v4 - 🆕 v4.0巴纳姆温暖版 ⭐⭐⭐ RECOMMENDED!", "monthly_fortune": "/analyze-monthly-fortune - 流月运势预测（未来12个月）", "shen_sha_lookup": "/shen-sha-lookup/{命宫地支}/{年份} - 十二宫神煞速查", "health": "/health - 健康检查（轻量级）", "warmup": "/warmup - 预热端点（保持模型活跃，推荐UptimeRobot使用）" }, "v4_0_features": [ "🌟🌟🌟 v4.0 巴纳姆温暖版上线！让命理解读不再冰冷！", "💕 巴纳姆冷读效果 - 温暖有共鸣的解读语言", "👨‍👩‍👧 六亲分析 - 父母/兄弟/夫妻/子女运完整批断", "📋 10步标准批命流程 - 专业系统化解读", "🔥 核心秘诀 - 福德宫化忌→夫妻死别判断", "📚 1000+星曜组合规则（从916条扩展）", "🎭 36条核心口诀自动应用", "✨ 敏感内容温和处理（避免恐吓式语言）" ], "v3_20_features": [ "🗓️ 流月运势预测系统 - 未来12个月运势", "🔮 十二宫位神煞速查表 - 快速判断吉凶", "🎯 焦点宫分析法 - 找出最凶险/最吉利月份", "📊 月度运势评分（0-100分）- 量化预测" ], "v3_18_features": [ "🎯🎯🎯 大运计算准确性修复（v3.18.0）- 和文墨天机100%一致！", "✅ 使用iztro返回的decadal原始数据，不再自己计算", "✅ 消除不同软件间的偏差问题", "✅ 提升用户信任度和专业性" ], "v3_17_features": [ "🎯🎯🎯 大运分析系统上线（v3.17.0）- 用户最关心的功能！", "📅 当前大运分析（宫位主题、年龄范围、星曜配置、强度评分）", "🔄 前五年/后五年区分（天干主导 vs 地支主导）", "⚠️ 大运交接期检测（末年动荡期 + 首年适应期）", "🌟 下一个大运预告（未来运势趋势、上升/下降/持平）", "💀 空宫大运危险评估（空宫 × 化忌流年 × 末年系数）", "✨ 基于772KB专业紫微知识库（5大核心模块）", "🎯 集成13颗主星完整特性分析", "📚 融合十二宫位深度解读", "🔮 整合十天干四化理论（THE CORE）", "💡 应用南北派138页实战断诀", "💕 温暖专业的Sisi风格解读", "🤖 知识库增强AI引擎（解读质量提升3倍）", "🌟 星曜亮度完整显示（庙旺陷系统）- v3.5", "📊 宫位强弱评分系统（0-100分+等级）- v3.5", "🎲 三方四正综合分析（相关宫位力量评估）- v3.6", "🔗 星曜组合汇总（四正宫位星曜全览）- v3.6", "🎭 格局自动识别（15个经典格局）- v3.7", "⚖️ 格局吉凶评分（0-100分智能匹配）- v3.7", "🏆 格局质量总结（整体命盘质量评估）- v3.7", "🎉 星曜组合断语库5.0（916条专业规则）⭐⭐⭐⭐⭐", "🚀 Google Notebook半自动提取（效率提升55倍！）", "🔬 质量检查系统（零重复+格式验证+ID冲突解决）", "🎯 7大分类组合解读（性格/事业/感情/健康/运势/才华/人际）", "📊 智能匹配评分系统（0-100分+置信度）", "🔮 按分类整理的专业断语（快速定位关键信息）", "🌟 三星同宫组合（包含紫微系/财星/桃花/杀破狼等）", "💑 婚姻感情深化（桃花/夫妻宫/婚姻运全面解读）", "🏥 健康状况深化（各系统疾病倾向+长寿格局）", "🎓 才华能力深化（文昌文曲/艺术/学术/技术全覆盖）", "👥 人际交往深化（贵人/人缘/口才/声望）", "🎭 特殊格局系列（夹命格/冲破格等特殊配置）", "✅ 对标专业软件数据丰富度（超越文墨天机！）" ], "knowledge_modules": { "foundations": [ "01-main-stars.md (26KB) - 13颗主星特性", "02-twelve-palaces.md (20KB) - 12宫位分析法", "03-four-transformations.md (28KB) - 四化飞星理论" ], "timing": [ "01-annual-fortune.md (191KB) - 流年大运推算" ], "interpretations": [ "01-practical-divination.md (359KB) - 南北派断诀" ], "total_size": "772KB professional knowledge" }, "upgrade_highlights_v3": [ "知识库增强：解读质量提升300%", "专业度提升：融合13主星+12宫位+四化理论", "实战性提升：应用138页南北派断诀", "温暖度提升：Sisi风格，避免恐吓式语言", "可信度提升：基于经典书籍，非AI臆想" ], "sisi_note": "💕💕💕💕 v4.0 巴纳姆温暖版震撼上线！让命理解读从'机械断语'变成'温暖有共鸣的分析'！六亲分析+10步批命+核心秘诀，让用户感受到被真正理解！命理不再冰冷，解读充满共鸣！🦊💕✨ Barnum Warm Edition Launched!" } @app.post("/calculate-ziwei", response_model=ZiweiResponse) def calculate_ziwei(request: ZiweiRequest, api_key: str = Depends(verify_api_key)): """ 计算紫微斗数命盘 Complete Ziwei Doushu chart calculation with AI-powered analysis 完整的紫微斗数排盘与AI智能分析 """ try: ziwei_data = run_ziwei_calculation(request) return ZiweiResponse( status="success", data=ziwei_data ) except subprocess.TimeoutExpired: raise HTTPException(status_code=504, detail="Calculation timeout") except json.JSONDecodeError as e: raise HTTPException(status_code=500, detail=f"Invalid response from ziwei service: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail=f"Calculation error: {str(e)}") # ⚠️ DEPRECATED (2026-07): 2026-07-12全仓库排查确认astrologybazi前端零调用此endpoint # (app/lib/hf/hf-api-config.ts里声明了这个路径字符串，但从没被实际import/调用)。 # 保留仅为兼容，不删除代码——见 API_UNIFICATION_PLAN.md §1.2/§6。三代/四代(v4)才是活的。 @app.post("/analyze-ziwei-enhanced", response_model=ZiweiResponse) def analyze_ziwei_enhanced(request: ZiweiRequest, api_key: str = Depends(verify_api_key)): """ 增强版紫微斗数分析 - 包含四化飞星和AI解读 ⚠️ DEPRECATED: 零前端调用，详见函数上方模块级注释。 Enhanced Ziwei Doushu Analysis with Four Transformations and AI Interpretation 完整的紫微斗数排盘 + 四化飞星分析 + AI智能解读 """ try: # 1. 首先调用基础排盘功能 ziwei_chart = run_ziwei_calculation(request) # 2. 四化飞星分析 sihua_analyzer = SihuaAnalyzer(ziwei_chart) sihua_analysis = sihua_analyzer.analyze_all_sihua() # 3. AI智能解读 ai_interpreter = AIInterpreter() ai_interpretation = ai_interpreter.interpret_full_chart( sihua_analysis, ziwei_chart ) # 4. 生成文字总结 summary_text = ai_interpreter.generate_summary_text(ai_interpretation) # 5. 组合完整结果 enhanced_data = { **ziwei_chart, # 包含原始命盘数据 'sihua_analysis': sihua_analysis, # 四化飞星分析 'ai_interpretation': ai_interpretation, # AI解读 'summary_text': summary_text, # 文字总结 'service_version': '2.0_enhanced_sihua_ai' } return ZiweiResponse( status="success", data=enhanced_data ) except subprocess.TimeoutExpired: raise HTTPException(status_code=504, detail="Calculation timeout") except json.JSONDecodeError as e: raise HTTPException(status_code=500, detail=f"Invalid response from ziwei service: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail=f"Enhanced analysis error: {str(e)}") @app.post("/analyze-ziwei-knowledge-enhanced", response_model=ZiweiResponse) def analyze_ziwei_knowledge_enhanced(request: ZiweiRequest, api_key: str = Depends(verify_api_key)): """ 知识库增强版紫微斗数分析 v4.0 🔮✨💕 Knowledge-Enhanced Ziwei Doushu Analysis - 温暖有共鸣的命理解读 完整的紫微斗数排盘 + 四化飞星分析 + 巴纳姆效果 + 六亲分析 + 10步批命流程 v4.0 新特性 (2026-02-03): - 💕 巴纳姆冷读效果 - 让解读温暖有共鸣 - 👨‍👩‍👧 六亲分析 - 父母/兄弟/夫妻/子女运完整批断 - 📋 10步标准批命流程 - 专业系统化解读 - 🔥 核心秘诀 - 福德宫化忌判断 - 📚 951条星曜组合规则 - 🎭 36条核心口诀自动应用 - ✨ 敏感内容温和处理 保留v3.x特性: - ✨ 基于772KB专业知识库（5大核心模块） - 🎯 大运分析 + 化解建议系统 - 💕 温暖专业的Sisi风格解读 """ try: # 1. 基础排盘 ziwei_chart = run_ziwei_calculation(request) # 2. 四化飞星分析 sihua_analyzer = SihuaAnalyzer(ziwei_chart) sihua_analysis = sihua_analyzer.analyze_all_sihua() # 3. 大运分析 birth_year = int(request.birth_date.split('-')[0]) current_age = datetime.now().year - birth_year daxun_analyzer = DaxunAnalyzer(ziwei_chart) current_daxun = daxun_analyzer.analyze_current_daxun(current_age) next_daxun = daxun_analyzer.predict_next_daxun(current_age) daxun_analysis = { 'current': current_daxun, 'next': next_daxun } # 4. 化解建议分析 resolution_advisor = ResolutionAdvisor(ziwei_chart) resolution_suggestions = resolution_advisor.analyze_resolution_needs( sihua_analysis, daxun_analysis ) # 5. v4.0 知识库增强AI解读 (巴纳姆+六亲+10步流程) knowledge_interpreter = KnowledgeEnhancedInterpreter() # 添加gender和birth_info到chart_data供v4.0引擎使用 ziwei_chart['gender'] = request.gender ziwei_chart['birth_info'] = { 'year': birth_year, 'date': request.birth_date, 'time': request.birth_time } # v4.1: 使用统一的enhanced方法（已包含庚干双化忌） knowledge_interpretation = knowledge_interpreter.interpret_full_chart_enhanced( sihua_analysis, ziwei_chart ) # 6. v4.0 生成巴纳姆温暖版文字总结 summary_text = knowledge_interpreter.generate_summary_text_v4( knowledge_interpretation ) # 7. 组合完整结果 enhanced_data = { **ziwei_chart, # 原始命盘数据 'sihua_analysis': sihua_analysis, # 四化飞星分析 'daxun_analysis': daxun_analysis, # 大运分析 'resolution_suggestions': resolution_suggestions, # 化解建议 'knowledge_interpretation': knowledge_interpretation, # v4.0知识库增强解读 'summary_text': summary_text, # v4.0巴纳姆温暖版文字总结 'service_version': '4.0.0_barnum_six_relations', 'knowledge_base_date': '2026-02-03', 'interpreter_type': 'KnowledgeEnhancedInterpreter_v4', 'barnum_enhanced': True, # v4.0标识 'six_relations_included': True, # 六亲分析标识 'reading_methodology': '10步标准批命流程', 'daxun_accuracy': '100% - Verified with Wenmotianji', 'resolution_strategies': '5大化解策略 - 避灾/医疗/精神/健康/财富', 'sisi_signature': '🦊 Generated with warmth & wisdom by Sisi v4.0' } # 2026-07-12 Phase 3: 追加 meta/strength/timing/evidence 四层字段，旧字段一个不改。 enhanced_data['meta'] = build_meta() enhanced_data['strength'] = build_strength(ziwei_chart.get('palaces', [])) enhanced_data['timing'] = build_timing(daxun_analysis) enhanced_data['evidence'] = build_evidence( ziwei_chart.get('chart', {}), ziwei_chart.get('palaces', []), sihua_analysis, daxun_analysis ) # 禄因忌果结构化字段(2026-07-13)：化禄=资源入口,化忌=仓库(资源沉淀处/一生课题) enhanced_data['lu_ji_flow'] = build_lu_ji_flow(sihua_analysis) # 流年四化(2026-07-13)：查询年天干四化+本命盘落宫/亮度,补齐Phase 3规范缺口 enhanced_data['flow_year_sihua'] = build_flow_year_sihua(ziwei_chart.get('palaces', [])) return ZiweiResponse( status="success", data=enhanced_data ) except subprocess.TimeoutExpired: raise HTTPException(status_code=504, detail="Calculation timeout") except json.JSONDecodeError as e: raise HTTPException(status_code=500, detail=f"Invalid response from ziwei service: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail=f"v4.0 Barnum-enhanced analysis error: {str(e)}") @app.post("/analyze-ziwei-v4", response_model=ZiweiResponse) def analyze_ziwei_v4(request: ZiweiRequest, api_key: str = Depends(verify_api_key)): """ v4.0 巴纳姆温暖版 - 别名端点 等同于 /analyze-ziwei-knowledge-enhanced (已升级到v4.0) 保留此端点是为了向后兼容 """ return analyze_ziwei_knowledge_enhanced(request, api_key) @app.get("/debug/modules") def debug_modules(api_key: str = Depends(verify_api_key)): """调试：检查模块导入状态""" debug_info = { "timestamp": datetime.now().isoformat(), "modules": {}, "interpreter_info": {} } # 检查geju模块 try: import geju_patterns debug_info["modules"]["geju_patterns"] = { "imported": True, "pattern_count": len(geju_patterns.GEJU_PATTERNS) } except Exception as e: debug_info["modules"]["geju_patterns"] = { "imported": False, "error": str(e) } try: import geju_matcher debug_info["modules"]["geju_matcher"] = { "imported": True, "has_GejuMatcher": hasattr(geju_matcher, 'GejuMatcher') } except Exception as e: debug_info["modules"]["geju_matcher"] = { "imported": False, "error": str(e) } # 检查interpreter try: interpreter = KnowledgeEnhancedInterpreter() debug_info["interpreter_info"]["has_geju_matcher"] = hasattr(interpreter, 'geju_matcher') debug_info["interpreter_info"]["class_name"] = interpreter.__class__.__name__ except Exception as e: debug_info["interpreter_info"]["error"] = str(e) return debug_info class MonthlyFortuneRequest(BaseModel): """流月运势请求""" birth_date: str = Field(..., description="出生日期 YYYY-MM-DD", example="2000-01-01") birth_time: str = Field(..., description="出生时间 HH:MM", example="14:30") gender: Literal['男', '女'] = Field(..., description="性别: 男/女", example="男") is_leap_month: bool = Field(default=False, description="是否闰月") fix_leap: bool = Field(default=True, description="是否修正闰月") target_year: int = Field(default_factory=lambda: datetime.now().year, description="目标年份(默认当前年)", example=2026) @app.post("/analyze-monthly-fortune", response_model=ZiweiResponse) def analyze_monthly_fortune(request: MonthlyFortuneRequest, api_key: str = Depends(verify_api_key)): """ 流月运势预测 v1.0 🌟✨ (NEW!) Monthly Fortune Prediction - 未来12个月运势预测 基于十二宫位神煞断吉凶 + 焦点宫分析法 新特性 (2026-01-20): - 🗓️ 流月命宫计算（斗君定位法） - 🔮 十二宫位神煞速查表 - 🎯 焦点宫分析（重病点/良药点） - 📊 月度运势评分（0-100分） - 💡 个性化趋吉避凶建议 """ try: # 1. 基础排盘 ziwei_chart = run_ziwei_calculation(request) # 2. 流月分析 liuyue_analyzer = LiuyueAnalyzer(ziwei_chart) # 3. 分析12个月运势 monthly_fortunes = liuyue_analyzer.analyze_12_months(request.target_year) # 4. 焦点宫分析 focal_points = liuyue_analyzer.find_focal_points(request.target_year) # 5. 组合结果 monthly_data = { 'target_year': request.target_year, 'ming_gong_zhi': liuyue_analyzer.ming_gong_zhi, 'monthly_fortunes': monthly_fortunes, 'focal_points': focal_points, 'summary': { 'best_months': focal_points['medicine_points'], 'worst_months': focal_points['disease_points'], 'yearly_summary': focal_points['summary'] }, 'service_version': '3.20.0_monthly_fortune', 'knowledge_base_date': '2026-01-20', 'sisi_signature': '🦊 Generated with knowledge & love by Sisi' } return ZiweiResponse( status="success", data=monthly_data ) except subprocess.TimeoutExpired: raise HTTPException(status_code=504, detail="Calculation timeout") except json.JSONDecodeError as e: raise HTTPException(status_code=500, detail=f"Invalid response from ziwei service: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail=f"Monthly fortune analysis error: {str(e)}") @app.get("/shen-sha-lookup/{ming_gong_zhi}/{year}") def shen_sha_lookup(ming_gong_zhi: str, year: int): """ 十二宫位神煞速查 🔮 Quick Divine Sha Lookup - 根据命宫地支快速查询流年流月神煞 Args: ming_gong_zhi: 命宫地支（子丑寅卯辰巳午未申酉戌亥） year: 目标年份（如2026） Returns: 年度神煞 + 12个月神煞分布 """ from liuyue_analyzer import get_annual_shen_sha, get_monthly_shen_sha, SHEN_SHA_MEANINGS, EARTHLY_BRANCHES if ming_gong_zhi not in EARTHLY_BRANCHES: raise HTTPException(status_code=400, detail=f"Invalid earthly branch: {ming_gong_zhi}") # 计算流年地支 liu_nian_zhi = EARTHLY_BRANCHES[(year - 1984) % 12] # 年度神煞 annual_sha = get_annual_shen_sha(ming_gong_zhi, liu_nian_zhi) annual_info = SHEN_SHA_MEANINGS.get(annual_sha, {}) # 12个月神煞 monthly_shas = [] for month in range(1, 13): sha = get_monthly_shen_sha(ming_gong_zhi, month) sha_info = SHEN_SHA_MEANINGS.get(sha, {}) monthly_shas.append({ 'month': month, 'shen_sha': sha, 'meaning': sha_info.get('meaning', ''), 'fortune': sha_info.get('fortune', 'neutral') }) return { 'ming_gong_zhi': ming_gong_zhi, 'year': year, 'liu_nian_zhi': liu_nian_zhi, 'annual_shen_sha': { 'name': annual_sha, 'meaning': annual_info.get('meaning', ''), 'meaning_en': annual_info.get('en', ''), 'fortune': annual_info.get('fortune', 'neutral') }, 'monthly_shen_shas': monthly_shas, 'sisi_note': '🦊 十二宫神煞速查 - 补四柱之不足！' } @app.get("/health") @app.head("/health") def health_check(): """健康检查端点""" return { "status": "healthy", "timestamp": datetime.now().isoformat() } @app.get("/warmup") @app.head("/warmup") def warmup(): """ 预热端点 - 保持服务处于活跃状态 Warmup endpoint for UptimeRobot - keeps the service active 支持GET和HEAD两种方法（和其他正常运行的API保持一致） """ return { "status": "warm", "timestamp": datetime.now().isoformat(), "message": "Service is active and ready" } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)