""" LiuYao analysis utilities for AI processing. This module provides functions to prepare data for AI analysis of LiuYao divination. """ import logging import json import re from typing import Dict, List, Optional, Union, Any from django.utils import timezone from django.conf import settings from ai.services.factory import LLMServiceFactory from iching.utils.utils import filter_prompt_for_response from ai.utils import get_active_template from ai.utils.i18n import normalize_language try: import iching.utils as iutils except ImportError: logger = logging.getLogger(__name__) logger.warning("Could not import iching.utils") iutils = None logger = logging.getLogger(__name__) # Constants - moved to class for better organization class LiuYaoConstants: """Constants for LiuYao analysis.""" # Hexagram names and numbers (King Wen sequence) HEXAGRAM_NAMES = { 1: "乾 (qián) - The Creative", 2: "坤 (kūn) - The Receptive", 3: "屯 (zhūn) - Difficulty at the Beginning", 4: "蒙 (méng) - Youthful Folly", 5: "需 (xū) - Waiting", 6: "讼 (sòng) - Conflict", 7: "师 (shī) - The Army", 8: "比 (bǐ) - Holding Together", 9: "小畜 (xiǎo chù) - Small Accumulation", 10: "履 (lǚ) - Treading", 11: "泰 (tài) - Peace", 12: "否 (pǐ) - Standstill", 13: "同人 (tóng rén) - Fellowship", 14: "大有 (dà yǒu) - Great Possession", 15: "谦 (qiān) - Modesty", 16: "豫 (yù) - Enthusiasm", 17: "随 (suí) - Following", 18: "蛊 (gǔ) - Decay", 19: "临 (lín) - Approach", 20: "观 (guān) - Contemplation", 21: "噬嗑 (shì kè) - Biting Through", 22: "贲 (bì) - Grace", 23: "剥 (bō) - Splitting Apart", 24: "复 (fù) - Return", 25: "无妄 (wú wàng) - Innocence", 26: "大畜 (dà chù) - Great Accumulation", 27: "颐 (yí) - Nourishment", 28: "大过 (dà guò) - Great Excess", 29: "坎 (kǎn) - The Abysmal Water", 30: "离 (lí) - The Clinging Fire", 31: "咸 (xián) - Influence", 32: "恒 (héng) - Duration", 33: "遁 (dùn) - Retreat", 34: "大壮 (dà zhuàng) - Great Power", 35: "晋 (jìn) - Progress", 36: "明夷 (míng yí) - Darkening of the Light", 37: "家人 (jiā rén) - Family", 38: "睽 (kuí) - Opposition", 39: "蹇 (jiǎn) - Obstruction", 40: "解 (xiè) - Deliverance", 41: "损 (sǔn) - Decrease", 42: "益 (yì) - Increase", 43: "夬 (guài) - Breakthrough", 44: "姤 (gòu) - Coming to Meet", 45: "萃 (cuì) - Gathering Together", 46: "升 (shēng) - Pushing Upward", 47: "困 (kùn) - Oppression", 48: "井 (jǐng) - The Well", 49: "革 (gé) - Revolution", 50: "鼎 (dǐng) - The Cauldron", 51: "震 (zhèn) - Thunder", 52: "艮 (gèn) - Mountain", 53: "渐 (jiàn) - Development", 54: "归妹 (guī mèi) - The Marrying Maiden", 55: "丰 (fēng) - Abundance", 56: "旅 (lǚ) - The Wanderer", 57: "巽 (xùn) - The Gentle Wind", 58: "兑 (duì) - The Joyous Lake", 59: "涣 (huàn) - Dispersion", 60: "节 (jié) - Limitation", 61: "中孚 (zhōng fú) - Inner Truth", 62: "小过 (xiǎo guò) - Small Excess", 63: "既济 (jì jì) - After Completion", 64: "未济 (wèi jì) - Before Completion" } # Trigram names TRIGRAM_NAMES = { "111": "乾 (qián) - Heaven", "000": "坤 (kūn) - Earth", "001": "震 (zhèn) - Thunder", "010": "坎 (kǎn) - Water", "100": "艮 (gèn) - Mountain", "110": "巽 (xùn) - Wind/Wood", "101": "离 (lí) - Fire", "011": "兑 (duì) - Lake" } # Line positions and meanings LINE_POSITIONS = { 1: "初爻 - Initial line (foundation)", 2: "二爻 - Second line (inner growth)", 3: "三爻 - Third line (transition)", 4: "四爻 - Fourth line (approaching the outer world)", 5: "五爻 - Fifth line (position of power)", 6: "上爻 - Sixth line (transition away from the situation)" } # Six Gods names SIX_GOD_NAMES = { 0: "青龙", # qing long - Azure Dragon 1: "朱雀", # zhu que - Vermilion Bird 2: "勾陈", # gou chen - Great Yokeback 3: "腾蛇", # teng she - Flying Serpent 4: "白虎", # bai hu - White Tiger 5: "玄武" # xuan wu - Black Tortoise } # Relationship mapping for wangshui RELATIONSHIP_MAPPING = { "same": "临", "same5e": "同五行", "empty": "空", "6harmony": "六合", "clash": "冲", "produce": "生", "counter": "克", "harm": "害", "punish": "刑", "counter-6harmony": "克合", "produce-6harmony": "生合", "direct-1": "化进神", "direct--1": "化退神", "direct-0": "化自身", "changed-sameday": "化日" } # Utility functions def get_hexagram_name(hexagram_number: int) -> str: """Get hexagram name based on its number.""" return LiuYaoConstants.HEXAGRAM_NAMES.get(hexagram_number, f"Unknown hexagram {hexagram_number}") def get_trigram_name(trigram_code: str) -> str: """Get trigram name based on its binary code.""" return LiuYaoConstants.TRIGRAM_NAMES.get(trigram_code, f"Unknown trigram {trigram_code}") def identify_changing_lines(liuyao) -> List[int]: """Identify changing lines in the hexagram.""" changing_lines = [] for i in range(1, 7): line_value = getattr(liuyao, f'y{i}') if line_value in ('000', '111'): # Moving lines (老阴老阳) changing_lines.append(i) return changing_lines # Consolidated formatter class class DataFormatter: """Consolidated formatter for various data types.""" @staticmethod def format_data(data: Any, data_type: str = "generic") -> str: """Generic data formatter that handles multiple data types.""" if not data: return f"无{data_type}数据 (No {data_type} data)" try: if isinstance(data, str): return data elif isinstance(data, dict): return "\n".join([f"{key}：{value}" for key, value in data.items()]) elif isinstance(data, list): return "\n".join([str(item) for item in data]) else: return str(data) except Exception as e: logger.error(f"Error formatting {data_type} data: {e}") return f"数据格式化错误 (Error formatting {data_type} data): {str(e)}" @staticmethod def format_six_gods(god_index: Union[int, None]) -> str: """Format Six Gods information.""" if god_index is None: return "无六神数据 (No Six Gods data)" if not isinstance(god_index, int) or god_index not in LiuYaoConstants.SIX_GOD_NAMES: return f"六神: 未知 (Unknown): {god_index}" god_name = LiuYaoConstants.SIX_GOD_NAMES[god_index] return f"六神: {god_name}" @staticmethod def format_wangshui_line(rel_data: Dict[str, Any]) -> str: """Format wangshui (旺衰) information for a single line.""" if not rel_data or not isinstance(rel_data, dict): return "" result = [] for period in ['day', 'month']: if period in rel_data and rel_data[period]: period_info = [] for item in rel_data[period]: if item is None: continue if isinstance(item, str) and item.startswith("cs12-"): try: cs_index = int(item.replace("cs12-", "")) cs_text = iutils.bz.g12ChangSheng.get(cs_index, str(cs_index)) period_info.append(f"十二长生: {cs_text}") except (ValueError, TypeError): continue elif isinstance(item, str): mapped_text = LiuYaoConstants.RELATIONSHIP_MAPPING.get(item) if mapped_text: period_info.append(mapped_text) if period_info: result.append(f"{'日' if period == 'day' else '月'}: {', '.join(period_info)}") return ", ".join(result) # Response validation class class ResponseValidator: """Handles validation and extraction of LLM responses.""" @staticmethod def extract_json_from_text(text: str) -> Optional[Dict[str, Any]]: """Extract JSON from various text formats.""" # Method 1: Try direct JSON parsing try: return json.loads(text) except json.JSONDecodeError: pass # Method 2: JSON in markdown code blocks json_pattern = r'```json\s*([\s\S]*?)\s*```' match = re.search(json_pattern, text) if match: try: json_content = match.group(1).strip() return json.loads(json_content) except json.JSONDecodeError: pass # Method 3: Find JSON objects in text json_object_pattern = r'\{[\s\S]*?\}' matches = re.findall(json_object_pattern, text) for match in matches: try: json_data = json.loads(match) if 'think' in json_data and 'analysis' in json_data: return json_data except json.JSONDecodeError: continue return None @staticmethod def extract_fields_manually(text: str) -> Dict[str, str]: """Extract think and analysis fields using regex patterns.""" extracted = {"think": "", "analysis": ""} patterns = { "think": r'"think"\s*:\s*"([^"]*(?:\\.[^"]*)*)"', "analysis": r'"analysis"\s*:\s*"([^"]*(?:\\.[^"]*)*)"' } for field, pattern in patterns.items(): match = re.search(pattern, text) if match: extracted[field] = match.group(1) return extracted @staticmethod def validate_liuyao_response(response_text: str) -> Dict[str, str]: """Validate and extract structured data from the LLM response.""" logger.debug(f"Response text (first 100 chars): {response_text[:100]}...") # Try JSON extraction first json_data = ResponseValidator.extract_json_from_text(response_text) if json_data and 'think' in json_data and 'analysis' in json_data: return { "liuyao_analysis": json_data.get('analysis', ''), "think": json_data.get('think', '') } # Try manual extraction manual_data = ResponseValidator.extract_fields_manually(response_text) if manual_data["think"] or manual_data["analysis"]: return { "liuyao_analysis": manual_data["analysis"] or response_text, "think": manual_data["think"] } # Fallback - return raw text return { "liuyao_analysis": response_text, "think": "" } def prepare_liuyao_prompt(liuyao, custom_template=None, language: str = 'zh-hans'): """ Prepare a prompt for LiuYao analysis. Args: liuyao: LiuYao instance custom_template: Optional custom template to use language: Language code for template retrieval and filler text Returns: str: Formatted prompt """ language = normalize_language(language) # Get the active template or use custom template template_content = custom_template if not template_content: try: template_content = get_active_template('liuyao', language=language) except Exception as e: logger.error(f"Error retrieving template: {e}") return None # Format data for template replacements = {} # Basic information replacements['{{question}}'] = liuyao.question # *** Recalculate the data for accurate, up-to-date information *** # Use the shared recalculate_liuyao_data function try: # Convert Django model to dictionary liuyao_dict = { 'y1': liuyao.y1, 'y2': liuyao.y2, 'y3': liuyao.y3, 'y4': liuyao.y4, 'y5': liuyao.y5, 'y6': liuyao.y6, 'data': liuyao.data if liuyao.data else {} } # If BaZi data is not available, use the current qdate if 'bz' not in liuyao_dict['data']: date = liuyao.qdate liuyao_dict = iutils.liuyao.recalculate_liuyao_data(liuyao_dict, date) else: liuyao_dict = iutils.liuyao.recalculate_liuyao_data(liuyao_dict) # Extract data for template formatting data = liuyao_dict.get('data', {}) ly = data.get('ly', {}) bz_data = data.get('bz', {}) # Format date with BaZi information date_formatted = liuyao.qdate.strftime("%Y/%m/%d %H:%M:%S") # Add BaZi stems and branches if bz_data and isinstance(bz_data, dict): year_stem = iutils.bz.gGodstem.get(bz_data.get('year', {}).get('g'), '') year_branch = iutils.bz.gEarthstem.get(bz_data.get('year', {}).get('e'), '') month_stem = iutils.bz.gGodstem.get(bz_data.get('month', {}).get('g'), '') month_branch = iutils.bz.gEarthstem.get(bz_data.get('month', {}).get('e'), '') day_stem = iutils.bz.gGodstem.get(bz_data.get('day', {}).get('g'), '') day_branch = iutils.bz.gEarthstem.get(bz_data.get('day', {}).get('e'), '') hour_stem = iutils.bz.gGodstem.get(bz_data.get('hour', {}).get('g'), '') hour_branch = iutils.bz.gEarthstem.get(bz_data.get('hour', {}).get('e'), '') # Get empty pillars (空亡) empty_branches = [] if 'empty' in bz_data and isinstance(bz_data['empty'], dict): empty_dict = bz_data['empty'] # Handle both string and integer keys (database uses string keys, tests use integer keys) empty1 = empty_dict.get(0) or empty_dict.get('0') empty2 = empty_dict.get(1) or empty_dict.get('1') if empty1 is not None and empty2 is not None: empty_branches = [ iutils.bz.gEarthstem.get(empty1, ''), iutils.bz.gEarthstem.get(empty2, '') ] bazi_str = f" {year_stem}{year_branch}年 {month_stem}{month_branch}月 {day_stem}{day_branch}日 {hour_stem}{hour_branch}时" if empty_branches and all(empty_branches): bazi_str += f" （空亡：{''.join(empty_branches)}）" date_formatted += bazi_str replacements['{{qdate}}'] = date_formatted logger.debug("Recalculated LiuYao data successfully") except Exception as e: logger.error(f"Error recalculating LiuYao data: {e}") return f"处理六爻数据时发生错误: {str(e)}" # Format the hexagram calculation data hexagram_calculation = format_hexagram_calculation(data) replacements['{{hexagram_calculation}}'] = hexagram_calculation # Identify changing lines changing_lines = identify_changing_lines(liuyao) changing_lines_text = ", ".join([LiuYaoConstants.LINE_POSITIONS.get(line, f"Line {line}") for line in changing_lines]) if not changing_lines: changing_lines_text = "无动爻 (No changing lines)" replacements['{{changing_lines}}'] = changing_lines_text replacements['{{changing_lines_count}}'] = str(len(changing_lines)) # Changed hexagram information changed_hexagram_info = "" changed_hexagram_number = "" # Safely check for cgua data - note that cgua may be None in some cases if ly is not None and isinstance(ly, dict) and 'cgua' in ly and ly['cgua'] is not None: if 'gua' in ly['cgua']: # Add 1 to gua index because HEXAGRAM_NAMES is 1-indexed but calc6Yao returns 0-indexed values changed_num = ly['cgua']['gua'] + 1 changed_hexagram_number = str(changed_num) changed_hexagram_info = get_hexagram_name(changed_num) replacements['{{changed_hexagram_info}}'] = changed_hexagram_info replacements['{{changed_hexagram_number}}'] = changed_hexagram_number # Replace all placeholders in the template for placeholder, value in replacements.items(): template_content = template_content.replace(placeholder, str(value)) # Strip content after the last "---" (triple dash) separator # This removes format instructions (like JSON format requirements) that are # not needed for conversation prompts. The filter_prompt_for_response function # handles edge cases like "---" on its own line. original_length = len(template_content) template_content = filter_prompt_for_response(template_content) if len(template_content) < original_length: logger.debug(f"Stripped content after last '---' separator, new length: {len(template_content)} characters (was {original_length})") return template_content def format_hexagram_calculation(liuyao_dict): """Format the hexagram calculation data for display. Args: liuyao_dict: Complete LiuYao data dictionary with all components Returns: str: Formatted display string """ if not liuyao_dict: return "无数据 (No data)" try: # Extract components from the liuyao_dict ly_data = liuyao_dict.get('ly', {}) god6_index = liuyao_dict.get('god6', 0) rel_data = liuyao_dict.get('rel', {}) if not ly_data: return "无六爻数据 (No LiuYao data)" # Create a formatted output based on the data structure result = [] # Process original gua data if 'ogua' in ly_data and ly_data['ogua'] is not None: result.append("原卦信息:") ogua = ly_data['ogua'] if 'gua' in ogua: # Get hexagram name from g64Gua (0-indexed) gua_index = ogua['gua'] hexagram_name = iutils.liuyao.g64Gua.get(gua_index, f"未知卦象 {gua_index}") # Get adjusted index for HEXAGRAM_NAMES (1-indexed) adjusted_index = gua_index + 1 hexagram_info = get_hexagram_name(adjusted_index) # Check for special properties special_properties = [] if ogua.get('clash6', False): special_properties.append("六冲") if ogua.get('harmony6', False): special_properties.append("六合") if ogua.get('ws', False): special_properties.append("游魂") if ogua.get('rs', False): special_properties.append("归魂") special_properties_text = "" if special_properties: special_properties_text = f" ({'/'.join(special_properties)})" # Extract trigrams if 'yao' in ogua: yao = ogua['yao'] upper_trigram_code = str(yao['y6']) + str(yao['y5']) + str(yao['y4']) lower_trigram_code = str(yao['y3']) + str(yao['y2']) + str(yao['y1']) upper_trigram = get_trigram_name(upper_trigram_code) lower_trigram = get_trigram_name(lower_trigram_code) result.append(f" 本卦：{hexagram_name} - {hexagram_info.split(' - ')[1]}{special_properties_text}") result.append(f" 卦序：{adjusted_index}") result.append(f" 上卦：{upper_trigram}") result.append(f" 下卦：{lower_trigram}") # Process palace information if 'palace' in ogua: palace = ogua['palace'] palace_info = [" 卦宫:"] if 'palace' in palace: palace_index = palace['palace'] palace_name = iutils.liuyao.g8GuaData.get(palace_index, {}).get('n', f"宫{palace_index}") pos = palace.get('pos', "") palace_info.append(f" 卦宫: {palace_name}宫（第{pos}卦）") if 'shi' in palace: palace_info.append(f" 世爻: {palace['shi']}爻") if 'yin' in palace: palace_info.append(f" 应爻: {palace['yin']}爻") result.append("\n".join(palace_info)) # Add six yao information six_yao_info = [" 六爻:"] # Process six lines from top to bottom has_yao_data = False if 'g' in ogua and 'e' in ogua and 'q' in ogua: has_yao_data = True for i in range(6, 0, -1): yao_num = {1: "一", 2: "二", 3: "三", 4: "四", 5: "五", 6: "六"}[i] yao_key = f'y{i}' # Calculate the god for this line based on the day's god6 value # The sequence should use the correct formula for six gods ordering god_index = (god6_index + i - 1) % 6 god_name = LiuYaoConstants.SIX_GOD_NAMES.get(god_index, "未知") # Get stem and branch for this line stem_idx = ogua['g'].get(yao_key) branch_idx = ogua['e'].get(yao_key) # Convert indices to actual names using gGodstem and gEarthstem from bz stem_name = iutils.bz.gGodstem.get(stem_idx, str(stem_idx)) if isinstance(stem_idx, int) else stem_idx branch_name = iutils.bz.gEarthstem.get(branch_idx, str(branch_idx)) if isinstance(branch_idx, int) else branch_idx # Get relationship (六亲) if available qin_index = ogua['q'].get(yao_key) qin_name = iutils.liuyao.g6Qin.get(qin_index, str(qin_index)) if isinstance(qin_index, int) else qin_index # Check for hidden gods (伏神) hidden_info = "" if 'hg' in ogua and 'he' in ogua and yao_key in ogua.get('hg', {}) and yao_key in ogua.get('he', {}): hidden_stem_idx = ogua['hg'].get(yao_key) hidden_branch_idx = ogua['he'].get(yao_key) # Convert hidden stem/branch indices to names hidden_stem_name = iutils.bz.gGodstem.get(hidden_stem_idx, str(hidden_stem_idx)) if isinstance(hidden_stem_idx, int) else hidden_stem_idx hidden_branch_name = iutils.bz.gEarthstem.get(hidden_branch_idx, str(hidden_branch_idx)) if isinstance(hidden_branch_idx, int) else hidden_branch_idx # Get hidden relationship if available hidden_rel = "" if 'hq' in ogua and yao_key in ogua.get('hq', {}): hidden_qin_idx = ogua['hq'].get(yao_key) hidden_qin_name = iutils.liuyao.g6Qin.get(hidden_qin_idx, str(hidden_qin_idx)) if isinstance(hidden_qin_idx, int) else hidden_qin_idx if hidden_qin_name: hidden_rel = f"，{hidden_qin_name}" if hidden_stem_name and hidden_branch_name: # Check if we should display this hidden god (based on hq) should_display = yao_key in ogua.get('hq', {}) if should_display: hidden_info = f" (伏神：{hidden_stem_name}{hidden_branch_name}{hidden_rel})" # Build the yao information line (no trailing comma if there are no more items) yao_line = f" {yao_num}爻：{god_name}, {stem_name}{branch_name}" if qin_name: yao_line += f"，{qin_name}" # Add hidden gods info if available if hidden_info: yao_line += hidden_info six_yao_info.append(yao_line) # Add 旺衰 (wangshui) information on a separate line if rel_data and isinstance(rel_data, dict) and 'ogua' in rel_data: ogua_rel = rel_data['ogua'] qkey = f'q{i}' if qkey in ogua_rel and ogua_rel[qkey]: # Process and format the relationship data wangshui_info = DataFormatter.format_wangshui_line(ogua_rel[qkey]) if wangshui_info: six_yao_info.append(f" 旺衰: {wangshui_info}") # Add hidden god wangshui info if available hgkey = f'hg{i}' if hgkey in ogua_rel and ogua_rel[hgkey] and yao_key in ogua.get('hq', {}): hidden_wangshui = DataFormatter.format_wangshui_line(ogua_rel[hgkey]) if hidden_wangshui: six_yao_info.append(f" 伏神旺衰: {hidden_wangshui}") if has_yao_data: result.append("\n" + "\n".join(six_yao_info)) # Add five element relationships section for ogua if has_yao_data and 'e' in ogua: five_element_relationships = [ " 六爻五行关系:", " 提示：AI 不需要自行计算五行生克关系，可直接使用本结构中已给出的关系。", " 描述：该部分表示六个爻位之间的五行生克关系，每个爻位与其他爻的相互关系，用五行的视角进行表示。" ] # Process each yao from 6 to 1 for current_yao in range(6, 0, -1): current_yao_key = f'y{current_yao}' # Get the current yao's element from earth branch (地支) if current_yao_key in ogua['e']: current_earth_idx = ogua['e'][current_yao_key] current_element_idx = iutils.bz.getEarthElementI(current_earth_idx) if current_element_idx is not None: # Build relationships with other yaos relationships = {} for other_yao in range(1, 7): if other_yao != current_yao: # Skip self other_yao_key = f'y{other_yao}' if other_yao_key in ogua['e']: other_earth_idx = ogua['e'][other_yao_key] other_element_idx = iutils.bz.getEarthElementI(other_earth_idx) if other_element_idx is not None: # Calculate relationship rel = iutils.bz.calcElementRel(current_element_idx, other_element_idx) # Map relationship to display text if rel == 0: # same rel_text = "同" elif rel == 1: # iproduce rel_text = "生" elif rel == 2: # icounter rel_text = "克" else: # counterme (3) or produceme (4) rel_text = "-" relationships[other_yao] = rel_text # Format the relationship line yao_num_chinese = {6: "六", 5: "五", 4: "四", 3: "三", 2: "二", 1: "一"}[current_yao] rel_parts = [] for yao_num in range(1, 7): if yao_num != current_yao: rel_text = relationships.get(yao_num, "-") rel_parts.append(f"{yao_num}:{rel_text}") rel_line = f" {yao_num_chinese}爻: {{{', '.join(rel_parts)}}}" five_element_relationships.append(rel_line) if len(five_element_relationships) > 1: # Only add if we have content result.append("\n" + "\n".join(five_element_relationships)) # Add hidden gods relationships sections if has_yao_data and 'e' in ogua and 'he' in ogua and 'hq' in ogua: # Find all yaos with hidden gods hidden_god_yaos = {} for i in range(1, 7): yao_key = f'y{i}' if (yao_key in ogua.get('he', {}) and yao_key in ogua.get('hq', {}) and ogua['he'][yao_key] is not None): # Get hidden god info hidden_earth_idx = ogua['he'][yao_key] hidden_qin_idx = ogua['hq'][yao_key] hidden_qin_name = iutils.liuyao.g6Qin.get(hidden_qin_idx, str(hidden_qin_idx)) hidden_god_yaos[i] = { 'earth_idx': hidden_earth_idx, 'qin_name': hidden_qin_name } if hidden_god_yaos: # Section 1: 六爻对伏神的五行关系 yao_to_hidden_relationships = [ " 六爻对伏神的五行关系:", " 提示：AI 不需要自行计算五行生克关系，可直接使用本结构中已给出的关系。", " 描述：该部分记录主卦六个爻位对\"伏神\"的五行生克关系。伏神是被压制在动爻之下的爻，虽不显但有一定影响。" ] for current_yao in range(6, 0, -1): current_yao_key = f'y{current_yao}' if current_yao_key in ogua['e']: current_earth_idx = ogua['e'][current_yao_key] current_element_idx = iutils.bz.getEarthElementI(current_earth_idx) if current_element_idx is not None: # Build relationships with hidden gods hidden_relationships = {} for hidden_yao, hidden_info in hidden_god_yaos.items(): hidden_element_idx = iutils.bz.getEarthElementI(hidden_info['earth_idx']) if hidden_element_idx is not None: # Calculate relationship rel = iutils.bz.calcElementRel(current_element_idx, hidden_element_idx) # Map relationship to display text if rel == 0: # same rel_text = "同" elif rel == 1: # iproduce rel_text = "生" elif rel == 2: # icounter rel_text = "克" else: # counterme (3) or produceme (4) rel_text = "-" hidden_relationships[hidden_yao] = rel_text if hidden_relationships: # Format the relationship line yao_num_chinese = {6: "六", 5: "五", 4: "四", 3: "三", 2: "二", 1: "一"}[current_yao] rel_parts = [] for hidden_yao in sorted(hidden_relationships.keys()): rel_text = hidden_relationships[hidden_yao] rel_parts.append(f"{hidden_yao}:{rel_text}") rel_line = f" {yao_num_chinese}爻: {{{', '.join(rel_parts)}}}" yao_to_hidden_relationships.append(rel_line) if len(yao_to_hidden_relationships) > 1: result.append("\n" + "\n".join(yao_to_hidden_relationships)) # Section 2: 伏神对六爻的五行关系 hidden_to_yao_relationships = [ " 伏神对六爻的五行关系:", " 提示：AI 不需要自行计算五行生克关系，可直接使用本结构中已给出的关系。", " 描述：表示伏神本身对主爻的五行作用关系，属于\"伏神为主动者，六爻为被动者\"。" ] for hidden_yao in sorted(hidden_god_yaos.keys()): hidden_info = hidden_god_yaos[hidden_yao] hidden_element_idx = iutils.bz.getEarthElementI(hidden_info['earth_idx']) if hidden_element_idx is not None: # Build relationships with all yaos yao_relationships = {} for target_yao in range(1, 7): target_yao_key = f'y{target_yao}' if target_yao_key in ogua['e']: target_earth_idx = ogua['e'][target_yao_key] target_element_idx = iutils.bz.getEarthElementI(target_earth_idx) if target_element_idx is not None: # Calculate relationship rel = iutils.bz.calcElementRel(hidden_element_idx, target_element_idx) # Map relationship to display text if rel == 0: # same rel_text = "同" elif rel == 1: # iproduce rel_text = "生" elif rel == 2: # icounter rel_text = "克" else: # counterme (3) or produceme (4) rel_text = "-" yao_relationships[target_yao] = rel_text if yao_relationships: # Format the relationship line yao_num_chinese = {1: "一", 2: "二", 3: "三", 4: "四", 5: "五", 6: "六"}[hidden_yao] rel_parts = [] for target_yao in range(1, 7): rel_text = yao_relationships.get(target_yao, "-") rel_parts.append(f"{target_yao}:{rel_text}") rel_line = f" {yao_num_chinese}爻{hidden_info['qin_name']}: {{{', '.join(rel_parts)}}}" hidden_to_yao_relationships.append(rel_line) if len(hidden_to_yao_relationships) > 1: result.append("\n" + "\n".join(hidden_to_yao_relationships)) # Only include other nested dictionaries that are useful if 'changeyao' in ogua and isinstance(ogua['changeyao'], dict): # Check if there are any changing yao has_changing_yao = any(value == 1 for value in ogua['changeyao'].values()) if has_changing_yao: # Get relationship data for changed gua if available cgua_rel_data = {} if rel_data and isinstance(rel_data, dict) and 'cgua' in rel_data: cgua_rel_data = rel_data['cgua'] or {} changing_info = [" 动爻:"] # Track which lines are already moving for later use with 暗动 moving_lines = set() for subkey, value in ogua['changeyao'].items(): if value == 1: yao_num = subkey.replace('y', '') moving_lines.add(int(yao_num)) # Get relationship information relationship_info = "" cy_key = f'cy{yao_num}' # Extra defensive check for None values anywhere in the chain if cgua_rel_data is None or not isinstance(cgua_rel_data, dict): continue if cy_key in cgua_rel_data: # Skip if value is None if cgua_rel_data.get(cy_key) is None: continue # Only proceed if it's a list if isinstance(cgua_rel_data[cy_key], list) and cgua_rel_data[cy_key]: # Map relationships to their Chinese names with updated mappings relationship_mapping = { "direct-1": "化进神", "direct--1": "化退神", "changed-sameday": "化日", "empty": "化空", "6harmony": "化六合", "clash": "化冲", "produce": "化回头生", "counter": "化回头克", "harm": "化害", "punish": "化刑", "counter-6harmony": "化克合", "produce-6harmony": "化生合", } rel_list = cgua_rel_data[cy_key] rel_texts = [] # Process each relationship - extra safe check for rel_item in rel_list if rel_list else []: if rel_item is None: continue if isinstance(rel_item, str) and rel_item.startswith("cs12-"): try: # Extract the cs12 index with error handling cs_index_str = rel_item.replace("cs12-", "") cs_index = int(cs_index_str) # Map to the corresponding text from g12ChangSheng cs_text = iutils.bz.g12ChangSheng.get(cs_index, str(cs_index)) rel_texts.append(f"化{cs_text}(十二长生)") except (ValueError, TypeError): # Skip invalid cs12 formats logger.warning(f"Invalid cs12 format: {rel_item}") continue else: # Use the mapping for other relationships (with extra type check) if isinstance(rel_item, str): # Make sure rel_item is a string mapped_text = relationship_mapping.get(rel_item, None) if mapped_text: # Only add if not empty rel_texts.append(mapped_text) if rel_texts: relationship_info = f" ({', '.join(rel_texts)})" changing_info.append(f" 第{yao_num}爻{relationship_info}") result.append("\n" + "\n".join(changing_info)) # Add 暗动 (hidden moving lines) section for lines with "clash" relationship to day BaZi hidden_moving_lines = [] # Check if the relationship data is available if rel_data and isinstance(rel_data, dict) and 'ogua' in rel_data: ogua_rel = rel_data['ogua'] # Check each line for clash relationship with day for i in range(1, 7): # Skip if this line is already a moving line if i in moving_lines: continue qkey = f'q{i}' if qkey in ogua_rel and ogua_rel[qkey] and isinstance(ogua_rel[qkey], dict): # Check for 'day' relationship data that contains 'clash' if 'day' in ogua_rel[qkey] and isinstance(ogua_rel[qkey]['day'], list): if 'clash' in ogua_rel[qkey]['day']: hidden_moving_lines.append(i) # Add 暗动 section if we found any hidden moving lines if hidden_moving_lines: hidden_changing_info = [" 暗动:"] for hidden_line in hidden_moving_lines: hidden_changing_info.append(f" 第{hidden_line}爻") result.append("\n" + "\n".join(hidden_changing_info)) else: result.append("\n 动爻: 无") # Track which lines are already moving for later use with 暗动 moving_lines = set() # Add 暗动 (hidden moving lines) section for lines with "clash" relationship to day BaZi hidden_moving_lines = [] # Check if the relationship data is available if rel_data and isinstance(rel_data, dict) and 'ogua' in rel_data: ogua_rel = rel_data['ogua'] # Check each line for clash relationship with day for i in range(1, 7): # Skip if this line is already a moving line if i in moving_lines: continue qkey = f'q{i}' if qkey in ogua_rel and ogua_rel[qkey] and isinstance(ogua_rel[qkey], dict): # Check for 'day' relationship data that contains 'clash' if 'day' in ogua_rel[qkey] and isinstance(ogua_rel[qkey]['day'], list): if 'clash' in ogua_rel[qkey]['day']: hidden_moving_lines.append(i) # Add 暗动 section if we found any hidden moving lines if hidden_moving_lines: hidden_changing_info = [" 暗动:"] for hidden_line in hidden_moving_lines: hidden_changing_info.append(f" 第{hidden_line}爻") result.append("\n" + "\n".join(hidden_changing_info)) # No five element relationships section needed for ogua # Process changed gua data if 'cgua' in ly_data and ly_data['cgua'] is not None: result.append("\n变卦信息:") cgua = ly_data['cgua'] if 'gua' in cgua: # Get hexagram name from g64Gua (0-indexed) gua_index = cgua['gua'] hexagram_name = iutils.liuyao.g64Gua.get(gua_index, f"未知卦象 {gua_index}") # Get adjusted index for HEXAGRAM_NAMES (1-indexed) adjusted_index = gua_index + 1 hexagram_info = get_hexagram_name(adjusted_index) # Check for special properties for changed hexagram special_properties = [] if cgua.get('clash6', False): special_properties.append("六冲") if cgua.get('harmony6', False): special_properties.append("六合") if cgua.get('ws', False): special_properties.append("游魂") if cgua.get('rs', False): special_properties.append("归魂") special_properties_text = "" if special_properties: special_properties_text = f" ({'/'.join(special_properties)})" # Extract trigrams if 'yao' in cgua: yao = cgua['yao'] upper_trigram_code = str(yao['y6']) + str(yao['y5']) + str(yao['y4']) lower_trigram_code = str(yao['y3']) + str(yao['y2']) + str(yao['y1']) upper_trigram = get_trigram_name(upper_trigram_code) lower_trigram = get_trigram_name(lower_trigram_code) result.append(f" 变卦：{hexagram_name} - {hexagram_info.split(' - ')[1]}{special_properties_text}") result.append(f" 卦序：{adjusted_index}") result.append(f" 上卦：{upper_trigram}") result.append(f" 下卦：{lower_trigram}") # Process palace information if 'palace' in cgua: palace = cgua['palace'] palace_info = [" 卦宫:"] # For changing gua, use orig_palace if available palace_index = palace.get('orig_palace', palace.get('palace', 0)) palace_name = iutils.liuyao.g8GuaData.get(palace_index, {}).get('n', f"宫{palace_index}") pos = palace.get('pos', "") palace_info.append(f" 卦宫: {palace_name}宫（第{pos}卦）") if 'shi' in palace: palace_info.append(f" 世爻: {palace['shi']}爻") if 'yin' in palace: palace_info.append(f" 应爻: {palace['yin']}爻") result.append("\n".join(palace_info)) # Add six yao information for changed hexagram six_yao_info = [" 六爻:"] # Process six lines from top to bottom has_yao_data = False if 'g' in cgua and 'e' in cgua and 'q' in cgua: has_yao_data = True for i in range(6, 0, -1): yao_num = {1: "一", 2: "二", 3: "三", 4: "四", 5: "五", 6: "六"}[i] yao_key = f'y{i}' # Calculate the god for this line god_index = (god6_index + i - 1) % 6 god_name = LiuYaoConstants.SIX_GOD_NAMES.get(god_index, "未知") # Get stem and branch for this line stem_idx = cgua['g'].get(yao_key) branch_idx = cgua['e'].get(yao_key) # Convert indices to actual names using gGodstem and gEarthstem from bz stem_name = iutils.bz.gGodstem.get(stem_idx, str(stem_idx)) if isinstance(stem_idx, int) else stem_idx branch_name = iutils.bz.gEarthstem.get(branch_idx, str(branch_idx)) if isinstance(branch_idx, int) else branch_idx # Get relationship (六亲) if available qin_index = cgua['q'].get(yao_key) qin_name = iutils.liuyao.g6Qin.get(qin_index, str(qin_index)) if isinstance(qin_index, int) else qin_index # Check for hidden gods (伏神) hidden_info = "" if 'hg' in cgua and 'he' in cgua and yao_key in cgua.get('hg', {}) and yao_key in cgua.get('he', {}): hidden_stem_idx = cgua['hg'].get(yao_key) hidden_branch_idx = cgua['he'].get(yao_key) # Convert hidden stem/branch indices to names hidden_stem_name = iutils.bz.gGodstem.get(hidden_stem_idx, str(hidden_stem_idx)) if isinstance(hidden_stem_idx, int) else hidden_stem_idx hidden_branch_name = iutils.bz.gEarthstem.get(hidden_branch_idx, str(hidden_branch_idx)) if isinstance(hidden_branch_idx, int) else hidden_branch_idx # Get hidden relationship if available hidden_rel = "" if 'hq' in cgua and yao_key in cgua.get('hq', {}): hidden_qin_idx = cgua['hq'].get(yao_key) hidden_qin_name = iutils.liuyao.g6Qin.get(hidden_qin_idx, str(hidden_qin_idx)) if isinstance(hidden_qin_idx, int) else hidden_qin_idx if hidden_qin_name: hidden_rel = f"，{hidden_qin_name}" if hidden_stem_name and hidden_branch_name: # Check if we should display this hidden god (based on hq) should_display = yao_key in cgua.get('hq', {}) if should_display: hidden_info = f" (伏神：{hidden_stem_name}{hidden_branch_name}{hidden_rel})" # Build the yao information line (no trailing comma if there are no more items) yao_line = f" {yao_num}爻：{god_name}, {stem_name}{branch_name}" if qin_name: yao_line += f"，{qin_name}" # Add hidden gods info if available if hidden_info: yao_line += hidden_info six_yao_info.append(yao_line) # Add 旺衰 (wangshui) information on a separate line if rel_data and isinstance(rel_data, dict) and 'cgua' in rel_data and rel_data['cgua'] is not None: cgua_rel = rel_data['cgua'] qkey = f'q{i}' if qkey in cgua_rel and cgua_rel[qkey]: # Process and format the relationship data wangshui_info = DataFormatter.format_wangshui_line(cgua_rel[qkey]) if wangshui_info: six_yao_info.append(f" 旺衰: {wangshui_info}") # Add hidden god wangshui info if available hgkey = f'hg{i}' if hgkey in cgua_rel and cgua_rel[hgkey] and yao_key in cgua.get('hq', {}): hidden_wangshui = DataFormatter.format_wangshui_line(cgua_rel[hgkey]) if hidden_wangshui: six_yao_info.append(f" 伏神旺衰: {hidden_wangshui}") if has_yao_data: result.append("\n" + "\n".join(six_yao_info)) # No five element relationships section needed for cgua return "\n".join(result) except Exception as e: logger.error(f"Error formatting hexagram calculation: {e}") return f"数据格式化错误 (Error formatting data): {str(e)}" def analyze_liuyao(liuyao, model_key=None, provider=None, language=None): """ Generate AI analysis for LiuYao divination. Args: liuyao: LiuYao instance to analyze model_key: Optional model key to use provider: Optional provider to use Returns: dict: Analysis result """ try: language = normalize_language(language) # Get AI configuration for liuyao if not provided if not provider or not model_key: from ai.utils.config import get_ai_config config = get_ai_config('liuyao') provider = provider or config['provider'] model_key = model_key or config['model'] # Prepare the prompt prompt = prepare_liuyao_prompt(liuyao, language=language) if not prompt: logger.error(f"Failed to prepare prompt for LiuYao {liuyao.id}") liuyao.analysis_status = 'error' liuyao.save(update_fields=['analysis_status']) return None # Get LLM service llm_service = LLMServiceFactory.get_service(provider) if model_key: llm_service.change_model(model_key) # Call LLM API response = llm_service.get_completion( prompt=prompt, temperature=0.7, max_tokens=4096 ) if not response: logger.error(f"Empty response from LLM for LiuYao {liuyao.id}") liuyao.analysis_status = 'error' liuyao.save(update_fields=['analysis_status']) return None # Extract thinking process if available thinking_process = "" think_pattern = r'([\s\S]*?)' think_match = re.search(think_pattern, response) if think_match: thinking_process = think_match.group(1).strip() # Remove the thinking section from the response response = re.sub(think_pattern, '', response).strip() # Validate and structure the response analysis_data = ResponseValidator.validate_liuyao_response(response) # Add metadata result = { "prompt": filter_prompt_for_response(prompt), "provider": provider, "model": model_key, "liuyao_analysis": analysis_data.get("liuyao_analysis", response), "think": analysis_data.get("think", thinking_process), "analysis": analysis_data.get("liuyao_analysis", response), # Add explicit analysis field for frontend "language": language, } # Update the LiuYao instance with the result liuyao.ai_analysis = result liuyao.analysis_status = 'completed' liuyao.analysis_timestamp = timezone.now() liuyao.save(update_fields=['ai_analysis', 'analysis_status', 'analysis_timestamp']) return result except Exception as e: logger.error(f"Error analyzing LiuYao {liuyao.id}: {e}") liuyao.analysis_status = 'error' liuyao.save(update_fields=['analysis_status']) return None