"""
Utility functions for BaZi analysis using AI.
"""
import logging
import json
from typing import Dict, Any, Optional, List
from bazi.models import Person
from iching.utils.bz import (
gGodstem, gEarthstem, gGodstem5Element, gEarthElement,
findEarthstemGod, findGodstem5E
)
from iching.utils.bzshagod import gShagodNames
from django.utils import timezone
import re
from ai.utils import get_active_template
from ai.services.factory import LLMServiceFactory
from iching.utils.utils import filter_prompt_for_response
from ai.utils.i18n import (
normalize_language,
translate_gender,
translate_element,
get_text,
none_text,
error_text,
unknown_text,
join_list,
format_luck_description,
is_missing,
translate_heavenly_stem,
translate_earthly_branch,
translate_ten_god,
translate_special_star,
)
logger = logging.getLogger(__name__)
def load_prompt_template(template_name: str) -> str:
"""
Load a prompt template from the file system or database.
This is a compatibility wrapper around get_active_template.
Args:
template_name: Name of the template file (without path)
Returns:
The template content as a string
"""
# Extract divination type from filename (e.g., bazi_analysis.txt -> bazi)
divination_type = template_name.split('_')[0]
return get_active_template(divination_type, language='zh-hans', filename=template_name)
def prepare_bazi_prompt(person: Person, custom_template: str = None, language: str = 'zh-hans') -> str:
"""
Prepare a prompt for BaZi analysis based on a Person object.
Args:
person: Person object with BaZi data
custom_template: Optional custom template string to use instead of loading from file/db
language: Language code for template retrieval and placeholder text
Returns:
Formatted prompt string with BaZi data
"""
language = normalize_language(language)
none_value = none_text(language)
error_value = error_text(language)
unknown_value = unknown_text(language)
# Ensure we have the latest BaZi calculation
try:
bazi_data = person.calculate_bazi()
if not bazi_data:
logger.error(f"Person id={person.id} returned empty BaZi data")
raise ValueError("Person has no BaZi data to analyze")
# Validate critical BaZi data components
required_pillars = ['year', 'month', 'day']
missing_pillars = [pillar for pillar in required_pillars if pillar not in bazi_data]
if missing_pillars:
logger.error(f"Person id={person.id} is missing required pillars: {', '.join(missing_pillars)}")
raise ValueError(f"BaZi data is missing required pillars: {', '.join(missing_pillars)}")
# Validate that each required pillar has the necessary data
for pillar in required_pillars:
pillar_data = bazi_data[pillar]
if pillar_data is None:
logger.error(f"Person id={person.id} has None for {pillar} pillar")
raise ValueError(f"BaZi data has None for {pillar} pillar")
if not isinstance(pillar_data, dict):
logger.error(f"Person id={person.id} has invalid {pillar} pillar: {type(pillar_data)}")
raise ValueError(f"BaZi data has invalid {pillar} pillar: {type(pillar_data)}")
if 'god_idx' not in pillar_data and 'g' not in pillar_data:
logger.error(f"Person id={person.id} {pillar} pillar missing god_idx/g")
raise ValueError(f"BaZi data {pillar} pillar missing god_idx/g")
if 'earth_idx' not in pillar_data and 'e' not in pillar_data:
logger.error(f"Person id={person.id} {pillar} pillar missing earth_idx/e")
raise ValueError(f"BaZi data {pillar} pillar missing earth_idx/e")
logger.debug(f"BaZi data keys: {', '.join(bazi_data.keys())}")
except Exception as e:
logger.error(f"Error calculating BaZi for person id={person.id}: {str(e)}", exc_info=True)
raise ValueError(f"Failed to calculate BaZi data: {str(e)}")
# Load the template
try:
if custom_template:
template = custom_template
logger.debug("Using provided custom template")
else:
# Use the get_active_template function to get the current active template
logger.info(f"Requesting BaZi template for language: {language}")
template = get_active_template('bazi', language=language)
logger.info(f"Successfully loaded BaZi analysis template for language: {language}")
# Log first 200 characters of template to verify language
logger.debug(f"Template preview (first 200 chars): {template[:200]}")
except Exception as e:
logger.error(f"Error loading template: {str(e)}", exc_info=True)
raise
# Basic person info
replacements = {
'{{name}}': person.name,
'{{gender}}': translate_gender(person.gender, language),
'{{birth_date}}': str(person.birth_date),
'{{birth_time}}': str(person.birth_time) if person.birth_time else none_value,
}
# BaZi chart pillars
pillars = ['year', 'month', 'day', 'hour']
pillar_fields = ['stem', 'branch', 'stem_element', 'hidden_elements', 'hidden_relationships', 'stem_relationship', 'sha_gods']
for pillar in pillars:
for field in pillar_fields:
replacements[f'{{{{{pillar}_{field}}}}}'] = none_value
for pillar in pillars:
try:
pillar_data = bazi_data.get(pillar)
if not pillar_data:
logger.warning(f"{pillar} pillar not found or empty in BaZi data")
continue
stem_idx = pillar_data.get('god_idx')
branch_idx = pillar_data.get('earth_idx')
if stem_idx is None or branch_idx is None:
logger.warning(f"{pillar} pillar missing stem ({stem_idx}) or branch ({branch_idx}) index")
continue
stem = gGodstem.get(stem_idx)
branch = gEarthstem.get(branch_idx)
stem_element = gGodstem5Element[stem_idx]['e'] if stem_idx in gGodstem5Element else None
# Translate stems and branches for English
translated_stem = translate_heavenly_stem(stem, language) if stem else none_value
translated_branch = translate_earthly_branch(branch, language) if branch else none_value
replacements[f'{{{{{pillar}_stem}}}}'] = translated_stem
replacements[f'{{{{{pillar}_branch}}}}'] = translated_branch
replacements[f'{{{{{pillar}_stem_element}}}}'] = translate_element(stem_element, language)
# Hidden elements in the branch
hidden_gods: List[str] = []
hidden_relationships: List[str] = []
try:
earth_gods = findEarthstemGod(branch_idx)
if earth_gods:
for god_idx in earth_gods.values():
god = gGodstem.get(god_idx)
god_element = gGodstem5Element.get(god_idx, {}).get('e')
if god:
# Translate heavenly stem for English
translated_god = translate_heavenly_stem(god, language)
hidden_gods.append(f"{translated_god} ({translate_element(god_element, language)})")
if pillar_data.get('hidden_gods'):
for hidden_god in pillar_data['hidden_gods']:
if hidden_god.get('god') == god and hidden_god.get('ten_god'):
# Translate ten god for English
translated_ten_god = translate_ten_god(hidden_god['ten_god'], language)
translated_god_for_rel = translate_heavenly_stem(god, language)
hidden_relationships.append(f"{translated_god_for_rel} ({translated_ten_god})")
break
except Exception as exc:
logger.error(f"Error processing hidden gods for {pillar}: {str(exc)}", exc_info=True)
replacements[f'{{{{{pillar}_hidden_elements}}}}'] = join_list(hidden_gods, language) if hidden_gods else none_value
replacements[f'{{{{{pillar}_hidden_relationships}}}}'] = join_list(hidden_relationships, language) if hidden_relationships else none_value
# Ten Gods relationship with Day Master
ten_god_value = pillar_data.get('ten_god')
if pillar == 'day':
replacements[f'{{{{{pillar}_stem_relationship}}}}'] = get_text('self_label', language)
elif ten_god_value:
# Translate ten god for English
translated_ten_god = translate_ten_god(ten_god_value, language)
replacements[f'{{{{{pillar}_stem_relationship}}}}'] = translated_ten_god
else:
replacements[f'{{{{{pillar}_stem_relationship}}}}'] = unknown_value
# Sha Gods (special stars)
try:
sha_gods_data = bazi_data.get('sha_gods', {}).get(pillar)
if sha_gods_data and sha_gods_data.get('gods') is not None:
names = [god.get('name', '') for god in sha_gods_data.get('gods', []) if god]
# Translate special star names for English
translated_names = [translate_special_star(name, language) for name in names if name]
replacements[f'{{{{{pillar}_sha_gods}}}}'] = join_list(translated_names, language) if translated_names else none_value
else:
replacements[f'{{{{{pillar}_sha_gods}}}}'] = none_value
except Exception as exc:
logger.error(f"Error processing sha_gods for {pillar}: {str(exc)}", exc_info=True)
replacements[f'{{{{{pillar}_sha_gods}}}}'] = error_value
except Exception as exc:
logger.error(f"Error processing {pillar} pillar: {str(exc)}", exc_info=True)
for field in pillar_fields:
replacements[f'{{{{{pillar}_{field}}}}}'] = error_value
# Element strengths
try:
elements = ['wood', 'fire', 'earth', 'metal', 'water']
strengths = bazi_data.get('element_strengths')
if strengths is None:
logger.warning("element_strengths is None or missing")
for element in elements:
replacements[f'{{{{{element}_strength}}}}'] = none_value
else:
for element in elements:
value = strengths.get(element)
replacements[f'{{{{{element}_strength}}}}'] = str(value) if value is not None else none_value
except Exception as exc:
logger.error(f"Error processing element strengths: {str(exc)}", exc_info=True)
for element in ['wood', 'fire', 'earth', 'metal', 'water']:
replacements[f'{{{{{element}_strength}}}}'] = error_value
# Additional data
try:
logger.debug(f"Processing additional BaZi data with keys: {', '.join(bazi_data.keys() if bazi_data else [])}")
chart_structure = bazi_data.get('chart_structure')
replacements['{{chart_structure}}'] = str(chart_structure) if chart_structure is not None else none_value
life_number = bazi_data.get('life_number')
if life_number is not None:
try:
replacements['{{life_number}}'] = str(life_number)
except Exception as exc:
logger.warning(f"Error converting life_number to string: {str(exc)}")
replacements['{{life_number}}'] = none_value
else:
logger.warning("life_number not found in BaZi data")
replacements['{{life_number}}'] = none_value
life_palace = bazi_data.get('life_palace')
replacements['{{life_palace}}'] = str(life_palace) if life_palace is not None else none_value
nominal_age = bazi_data.get('nominal_age')
if nominal_age is not None:
try:
replacements['{{current_age}}'] = str(nominal_age)
except Exception as exc:
logger.warning(f"Error converting nominal_age to string: {str(exc)}")
replacements['{{current_age}}'] = none_value
else:
logger.warning("nominal_age not found in BaZi data")
replacements['{{current_age}}'] = none_value
current_luck_god: Optional[str] = None
current_luck_earth: Optional[str] = None
current_luck_age_range: Optional[str] = None
replacements['{{current_luck_pillar}}'] = none_value
# Initialize luck pillar placeholders (up to 3 luck pillars)
replacements['{{luck_pillar_1}}'] = none_value
replacements['{{luck_pillar_2}}'] = none_value
replacements['{{luck_pillar_3}}'] = none_value
ten_year_fate = bazi_data.get('ten_year_fate')
current_column_index: Optional[int] = None
if isinstance(ten_year_fate, dict):
columns = ten_year_fate.get('columns')
current_column_index = ten_year_fate.get('current_column_index')
if isinstance(columns, list):
# Find current luck pillar - first try is_current flag
for column in columns:
if column and isinstance(column, dict) and column.get('is_current'):
god_val = column.get('god')
earth_val = column.get('earth')
age_val = column.get('age_range')
# Convert empty strings, None, or string "None" to None
current_luck_god = god_val if god_val and god_val != "None" and str(god_val).strip() else None
current_luck_earth = earth_val if earth_val and earth_val != "None" and str(earth_val).strip() else None
current_luck_age_range = age_val if age_val and age_val != "None" and str(age_val).strip() else None
break
# Fallback: if not found by is_current, use current_column_index
if current_luck_god is None and current_luck_earth is None:
if current_column_index is not None and isinstance(current_column_index, int):
if 0 <= current_column_index < len(columns):
column = columns[current_column_index]
if column and isinstance(column, dict):
god_val = column.get('god')
earth_val = column.get('earth')
age_val = column.get('age_range')
# Convert empty strings, None, or string "None" to None
current_luck_god = god_val if god_val and god_val != "None" and str(god_val).strip() else None
current_luck_earth = earth_val if earth_val and earth_val != "None" and str(earth_val).strip() else None
current_luck_age_range = age_val if age_val and age_val != "None" and str(age_val).strip() else None
# If still not found and we have nominal_age, calculate current luck pillar based on age
# This handles cases where the person is beyond the 10th pillar
if current_luck_god is None and current_luck_earth is None and nominal_age is not None:
try:
start_year = ten_year_fate.get('start_year')
direction = ten_year_fate.get('direction')
if start_year is not None and direction is not None and len(columns) > 0:
# Calculate which luck pillar the person is in based on their age
# Each luck pillar covers 10 years, starting from start_year
# pillar_offset is how many pillars away from the first column (index 0)
pillar_offset = (nominal_age - start_year) // 10
# Get the first column to use as base for calculation
first_column = columns[0]
if first_column and isinstance(first_column, dict):
first_god = first_column.get('god')
first_earth = first_column.get('earth')
if first_god and first_earth:
# Find the indices of the first column's god and earth
first_god_idx = None
first_earth_idx = None
for idx, stem in gGodstem.items():
if stem == first_god:
first_god_idx = idx
break
for idx, branch in gEarthstem.items():
if branch == first_earth:
first_earth_idx = idx
break
if first_god_idx is not None and first_earth_idx is not None:
# Calculate the current pillar's god and earth based on direction and pillar_offset
# The first column already has 1 increment applied, so we need to add (pillar_offset) more increments
from iching.utils.bz import gGLen, gELen
if direction == 1: # forward
current_god_idx = (first_god_idx + pillar_offset) % gGLen
current_earth_idx = (first_earth_idx + pillar_offset) % gELen
else: # backward
current_god_idx = (first_god_idx - pillar_offset) % gGLen
current_earth_idx = (first_earth_idx - pillar_offset) % gELen
current_luck_god = gGodstem.get(current_god_idx)
current_luck_earth = gEarthstem.get(current_earth_idx)
# Calculate age range for this pillar
age_start = start_year + (pillar_offset * 10)
age_end = age_start + 9
current_luck_age_range = f"{age_start}-{age_end}"
except Exception as exc:
logger.warning(f"Error calculating current luck pillar from age: {str(exc)}", exc_info=True)
# Extract up to 3 luck pillars starting from current, but only if they don't exceed the 10th pillar
# There are 10 luck pillars total (indices 0-9), so we only include if index <= 9
# However, the current luck pillar should always be shown if found, even if it exceeds the 10th pillar
if current_column_index is not None and isinstance(current_column_index, int):
day_god_idx = None
if bazi_data.get('day'):
day_god_idx = bazi_data['day'].get('god_idx') or bazi_data['day'].get('g')
# Import ten god calculation functions once
from iching.utils.bz import calc10God, calcEarthstem10God
# Extract up to 3 luck pillars (luck_pillar_1, luck_pillar_2, luck_pillar_3)
# These are the next pillars after the current one
# Only include them if they don't exceed the 10th pillar (index <= 9)
for i in range(3): # Extract 3 luck pillars
pillar_index = current_column_index + i + 1 # Start from next pillar (current + 1)
if pillar_index < len(columns) and pillar_index <= 9: # Don't exceed 10th pillar (index 9)
column = columns[pillar_index]
if column and isinstance(column, dict):
pillar_god = column.get('god') or None
pillar_earth = column.get('earth') or None
pillar_age_range = column.get('age_range') or None
# Calculate ten god relationships for this luck pillar
pillar_stem_relationship = None
pillar_branch_relationships = None
if day_god_idx is not None and pillar_god:
try:
# Find stem index
pillar_stem_idx = None
for idx, stem in gGodstem.items():
if stem == pillar_god:
pillar_stem_idx = idx
break
if pillar_stem_idx is not None:
stem_rel = calc10God(day_god_idx, pillar_stem_idx)
pillar_stem_relationship = translate_ten_god(stem_rel, language) if stem_rel else None
# Find branch index and calculate branch ten gods
if pillar_earth:
pillar_branch_idx = None
for idx, branch in gEarthstem.items():
if branch == pillar_earth:
pillar_branch_idx = idx
break
if pillar_branch_idx is not None:
branch_ten_gods = calcEarthstem10God(day_god_idx, pillar_branch_idx)
if branch_ten_gods:
descriptions = [ten_god for ten_god in branch_ten_gods.values() if ten_god]
translated_descriptions = [translate_ten_god(ten_god, language) for ten_god in descriptions if ten_god]
pillar_branch_relationships = join_list(translated_descriptions, language) if translated_descriptions else None
except Exception as exc:
logger.warning(f"Error calculating ten god relationships for luck pillar {i+1}: {str(exc)}")
# Format the luck pillar description
formatted_pillar = format_luck_description(
pillar_god,
pillar_earth,
pillar_age_range,
None if is_missing(pillar_stem_relationship, language) else pillar_stem_relationship,
None if is_missing(pillar_branch_relationships, language) else pillar_branch_relationships,
language,
)
replacements[f'{{{{luck_pillar_{i+1}}}}}'] = formatted_pillar
else:
replacements[f'{{{{luck_pillar_{i+1}}}}}'] = none_value
else:
# Pillar index exceeds 10th pillar or is out of range, leave empty
replacements[f'{{{{luck_pillar_{i+1}}}}}'] = none_value
else:
logger.warning("ten_year_fate columns missing or invalid")
elif ten_year_fate is not None:
logger.warning("ten_year_fate has unexpected format")
# Ensure current luck pillar is always formatted, even if current_column_index is None or > 9
# This ensures the current luck pillar is always shown according to the person's age
# Only format if we have at least one valid non-empty value (god or earth)
if (current_luck_god is not None and str(current_luck_god).strip()) or (current_luck_earth is not None and str(current_luck_earth).strip()):
# Format current luck pillar with basic info (ten god relationships will be added later if available)
# This is a fallback to ensure it's always shown
try:
day_god_idx = None
if bazi_data.get('day'):
day_god_idx = bazi_data['day'].get('god_idx') or bazi_data['day'].get('g')
# Try to calculate ten god relationships for current luck pillar
current_luck_stem_relationship = None
current_luck_branch_relationships = None
if day_god_idx is not None and current_luck_god:
try:
from iching.utils.bz import calc10God, calcEarthstem10God
# Find stem index
pillar_stem_idx = None
for idx, stem in gGodstem.items():
if stem == current_luck_god:
pillar_stem_idx = idx
break
if pillar_stem_idx is not None:
stem_rel = calc10God(day_god_idx, pillar_stem_idx)
current_luck_stem_relationship = translate_ten_god(stem_rel, language) if stem_rel else None
# Find branch index and calculate branch ten gods
if current_luck_earth:
pillar_branch_idx = None
for idx, branch in gEarthstem.items():
if branch == current_luck_earth:
pillar_branch_idx = idx
break
if pillar_branch_idx is not None:
branch_ten_gods = calcEarthstem10God(day_god_idx, pillar_branch_idx)
if branch_ten_gods:
descriptions = [ten_god for ten_god in branch_ten_gods.values() if ten_god]
translated_descriptions = [translate_ten_god(ten_god, language) for ten_god in descriptions if ten_god]
current_luck_branch_relationships = join_list(translated_descriptions, language) if translated_descriptions else None
except Exception as exc:
logger.warning(f"Error calculating ten god relationships for current luck pillar: {str(exc)}")
# Format the current luck pillar
# Convert None to empty string to avoid "None" appearing in output
replacements['{{current_luck_pillar}}'] = format_luck_description(
current_luck_god or "",
current_luck_earth or "",
current_luck_age_range or "",
None if is_missing(current_luck_stem_relationship, language) else current_luck_stem_relationship,
None if is_missing(current_luck_branch_relationships, language) else current_luck_branch_relationships,
language,
)
except Exception as exc:
logger.warning(f"Error formatting current luck pillar: {str(exc)}")
# Fallback to simple format if formatting fails
if (current_luck_god and str(current_luck_god).strip()) or (current_luck_earth and str(current_luck_earth).strip()):
replacements['{{current_luck_pillar}}'] = format_luck_description(
current_luck_god or "",
current_luck_earth or "",
current_luck_age_range or "",
None,
None,
language,
)
else:
replacements['{{current_luck_pillar}}'] = none_value
current_year = timezone.now().year
replacements['{{current_year}}'] = str(current_year)
next_year = current_year + 1
next2_year = current_year + 2
next3_year = current_year + 3
next4_year = current_year + 4
next5_year = current_year + 5
next6_year = current_year + 6
next7_year = current_year + 7
next8_year = current_year + 8
next9_year = current_year + 9
current_year_stem: Optional[str] = None
current_year_branch: Optional[str] = None
next_year_stem: Optional[str] = None
next_year_branch: Optional[str] = None
next2_year_stem: Optional[str] = None
next2_year_branch: Optional[str] = None
next3_year_stem: Optional[str] = None
next3_year_branch: Optional[str] = None
next4_year_stem: Optional[str] = None
next4_year_branch: Optional[str] = None
next5_year_stem: Optional[str] = None
next5_year_branch: Optional[str] = None
next6_year_stem: Optional[str] = None
next6_year_branch: Optional[str] = None
next7_year_stem: Optional[str] = None
next7_year_branch: Optional[str] = None
next8_year_stem: Optional[str] = None
next8_year_branch: Optional[str] = None
next9_year_stem: Optional[str] = None
next9_year_branch: Optional[str] = None
year_fate = bazi_data.get('year_fate')
if isinstance(year_fate, dict):
year_columns = year_fate.get('columns')
if isinstance(year_columns, list):
for column in year_columns:
if not column or not isinstance(column, dict):
continue
year = column.get('year')
god = column.get('god') or None
earth = column.get('earth') or None
if year == current_year:
current_year_stem = god
current_year_branch = earth
elif year == next_year:
next_year_stem = god
next_year_branch = earth
elif year == next2_year:
next2_year_stem = god
next2_year_branch = earth
elif year == next3_year:
next3_year_stem = god
next3_year_branch = earth
elif year == next4_year:
next4_year_stem = god
next4_year_branch = earth
elif year == next5_year:
next5_year_stem = god
next5_year_branch = earth
elif year == next6_year:
next6_year_stem = god
next6_year_branch = earth
elif year == next7_year:
next7_year_stem = god
next7_year_branch = earth
elif year == next8_year:
next8_year_stem = god
next8_year_branch = earth
elif year == next9_year:
next9_year_stem = god
next9_year_branch = earth
if current_year_stem is None and current_year_branch is None:
for column in year_columns:
if column and isinstance(column, dict) and column.get('is_current'):
current_year_stem = column.get('god') or None
current_year_branch = column.get('earth') or None
break
else:
logger.warning("year_fate columns missing or invalid")
elif year_fate is not None:
logger.warning("year_fate has unexpected format")
# Translate year stems and branches for English
replacements['{{current_year_stem}}'] = translate_heavenly_stem(current_year_stem, language) if current_year_stem else none_value
replacements['{{current_year_branch}}'] = translate_earthly_branch(current_year_branch, language) if current_year_branch else none_value
replacements['{{next_year}}'] = str(next_year)
replacements['{{next_year_stem}}'] = translate_heavenly_stem(next_year_stem, language) if next_year_stem else none_value
replacements['{{next_year_branch}}'] = translate_earthly_branch(next_year_branch, language) if next_year_branch else none_value
replacements['{{next2_year}}'] = str(next2_year)
replacements['{{next2_year_stem}}'] = translate_heavenly_stem(next2_year_stem, language) if next2_year_stem else none_value
replacements['{{next2_year_branch}}'] = translate_earthly_branch(next2_year_branch, language) if next2_year_branch else none_value
replacements['{{next3_year}}'] = str(next3_year)
replacements['{{next3_year_stem}}'] = translate_heavenly_stem(next3_year_stem, language) if next3_year_stem else none_value
replacements['{{next3_year_branch}}'] = translate_earthly_branch(next3_year_branch, language) if next3_year_branch else none_value
replacements['{{next4_year}}'] = str(next4_year)
replacements['{{next4_year_stem}}'] = translate_heavenly_stem(next4_year_stem, language) if next4_year_stem else none_value
replacements['{{next4_year_branch}}'] = translate_earthly_branch(next4_year_branch, language) if next4_year_branch else none_value
replacements['{{next5_year}}'] = str(next5_year)
replacements['{{next5_year_stem}}'] = translate_heavenly_stem(next5_year_stem, language) if next5_year_stem else none_value
replacements['{{next5_year_branch}}'] = translate_earthly_branch(next5_year_branch, language) if next5_year_branch else none_value
replacements['{{next6_year}}'] = str(next6_year)
replacements['{{next6_year_stem}}'] = translate_heavenly_stem(next6_year_stem, language) if next6_year_stem else none_value
replacements['{{next6_year_branch}}'] = translate_earthly_branch(next6_year_branch, language) if next6_year_branch else none_value
replacements['{{next7_year}}'] = str(next7_year)
replacements['{{next7_year_stem}}'] = translate_heavenly_stem(next7_year_stem, language) if next7_year_stem else none_value
replacements['{{next7_year_branch}}'] = translate_earthly_branch(next7_year_branch, language) if next7_year_branch else none_value
replacements['{{next8_year}}'] = str(next8_year)
replacements['{{next8_year_stem}}'] = translate_heavenly_stem(next8_year_stem, language) if next8_year_stem else none_value
replacements['{{next8_year_branch}}'] = translate_earthly_branch(next8_year_branch, language) if next8_year_branch else none_value
replacements['{{next9_year}}'] = str(next9_year)
replacements['{{next9_year_stem}}'] = translate_heavenly_stem(next9_year_stem, language) if next9_year_stem else none_value
replacements['{{next9_year_branch}}'] = translate_earthly_branch(next9_year_branch, language) if next9_year_branch else none_value
try:
day_god_idx = None
if bazi_data.get('day'):
day_god_idx = bazi_data['day'].get('god_idx') or bazi_data['day'].get('g')
def calculate_year_ten_gods(year_stem: Optional[str], year_branch: Optional[str]):
stem_relationship = none_value
branch_relationships = none_value
if day_god_idx is None or not year_stem:
return stem_relationship, branch_relationships
year_stem_idx = None
for idx, stem in gGodstem.items():
if stem == year_stem:
year_stem_idx = idx
break
if year_stem_idx is None:
return stem_relationship, branch_relationships
from iching.utils.bz import calc10God
stem_rel = calc10God(day_god_idx, year_stem_idx)
# Translate ten god for English
stem_relationship = translate_ten_god(stem_rel, language) if stem_rel else none_value
if not year_branch:
return stem_relationship, branch_relationships
year_branch_idx = None
for idx, branch in gEarthstem.items():
if branch == year_branch:
year_branch_idx = idx
break
if year_branch_idx is None:
return stem_relationship, branch_relationships
from iching.utils.bz import calcEarthstem10God
branch_ten_gods = calcEarthstem10God(day_god_idx, year_branch_idx)
if branch_ten_gods:
descriptions = [ten_god for ten_god in branch_ten_gods.values() if ten_god]
# Translate ten gods for English
translated_descriptions = [translate_ten_god(ten_god, language) for ten_god in descriptions if ten_god]
branch_relationships = join_list(translated_descriptions, language) if translated_descriptions else none_value
return stem_relationship, branch_relationships
current_year_stem_relationship, current_year_branch_relationships = calculate_year_ten_gods(current_year_stem, current_year_branch)
replacements['{{current_year_stem_relationship}}'] = current_year_stem_relationship
replacements['{{current_year_branch_relationships}}'] = current_year_branch_relationships
next_year_stem_relationship, next_year_branch_relationships = calculate_year_ten_gods(next_year_stem, next_year_branch)
replacements['{{next_year_stem_relationship}}'] = next_year_stem_relationship
replacements['{{next_year_branch_relationships}}'] = next_year_branch_relationships
next2_year_stem_relationship, next2_year_branch_relationships = calculate_year_ten_gods(next2_year_stem, next2_year_branch)
replacements['{{next2_year_stem_relationship}}'] = next2_year_stem_relationship
replacements['{{next2_year_branch_relationships}}'] = next2_year_branch_relationships
next3_year_stem_relationship, next3_year_branch_relationships = calculate_year_ten_gods(next3_year_stem, next3_year_branch)
replacements['{{next3_year_stem_relationship}}'] = next3_year_stem_relationship
replacements['{{next3_year_branch_relationships}}'] = next3_year_branch_relationships
next4_year_stem_relationship, next4_year_branch_relationships = calculate_year_ten_gods(next4_year_stem, next4_year_branch)
replacements['{{next4_year_stem_relationship}}'] = next4_year_stem_relationship
replacements['{{next4_year_branch_relationships}}'] = next4_year_branch_relationships
next5_year_stem_relationship, next5_year_branch_relationships = calculate_year_ten_gods(next5_year_stem, next5_year_branch)
replacements['{{next5_year_stem_relationship}}'] = next5_year_stem_relationship
replacements['{{next5_year_branch_relationships}}'] = next5_year_branch_relationships
next6_year_stem_relationship, next6_year_branch_relationships = calculate_year_ten_gods(next6_year_stem, next6_year_branch)
replacements['{{next6_year_stem_relationship}}'] = next6_year_stem_relationship
replacements['{{next6_year_branch_relationships}}'] = next6_year_branch_relationships
next7_year_stem_relationship, next7_year_branch_relationships = calculate_year_ten_gods(next7_year_stem, next7_year_branch)
replacements['{{next7_year_stem_relationship}}'] = next7_year_stem_relationship
replacements['{{next7_year_branch_relationships}}'] = next7_year_branch_relationships
next8_year_stem_relationship, next8_year_branch_relationships = calculate_year_ten_gods(next8_year_stem, next8_year_branch)
replacements['{{next8_year_stem_relationship}}'] = next8_year_stem_relationship
replacements['{{next8_year_branch_relationships}}'] = next8_year_branch_relationships
next9_year_stem_relationship, next9_year_branch_relationships = calculate_year_ten_gods(next9_year_stem, next9_year_branch)
replacements['{{next9_year_stem_relationship}}'] = next9_year_stem_relationship
replacements['{{next9_year_branch_relationships}}'] = next9_year_branch_relationships
# Only update current_luck_pillar if we have valid data
if (current_luck_god is not None and str(current_luck_god).strip()) or (current_luck_earth is not None and str(current_luck_earth).strip()):
current_luck_stem_relationship, current_luck_branch_relationships = calculate_year_ten_gods(current_luck_god, current_luck_earth)
replacements['{{current_luck_pillar}}'] = format_luck_description(
current_luck_god or "",
current_luck_earth or "",
current_luck_age_range or "",
None if is_missing(current_luck_stem_relationship, language) else current_luck_stem_relationship,
None if is_missing(current_luck_branch_relationships, language) else current_luck_branch_relationships,
language,
)
except Exception as exc:
logger.error(f"Error calculating ten god relationships for yearly data: {str(exc)}", exc_info=True)
replacements['{{current_year_stem_relationship}}'] = none_value
replacements['{{current_year_branch_relationships}}'] = none_value
replacements['{{next_year_stem_relationship}}'] = none_value
replacements['{{next_year_branch_relationships}}'] = none_value
replacements['{{next2_year_stem_relationship}}'] = none_value
replacements['{{next2_year_branch_relationships}}'] = none_value
replacements['{{next3_year_stem_relationship}}'] = none_value
replacements['{{next3_year_branch_relationships}}'] = none_value
replacements['{{next4_year_stem_relationship}}'] = none_value
replacements['{{next4_year_branch_relationships}}'] = none_value
replacements['{{next5_year_stem_relationship}}'] = none_value
replacements['{{next5_year_branch_relationships}}'] = none_value
replacements['{{next6_year_stem_relationship}}'] = none_value
replacements['{{next6_year_branch_relationships}}'] = none_value
replacements['{{next7_year_stem_relationship}}'] = none_value
replacements['{{next7_year_branch_relationships}}'] = none_value
replacements['{{next8_year_stem_relationship}}'] = none_value
replacements['{{next8_year_branch_relationships}}'] = none_value
replacements['{{next9_year_stem_relationship}}'] = none_value
replacements['{{next9_year_branch_relationships}}'] = none_value
# Always format current luck pillar, even if there's an error in ten god calculations
# Use format_luck_description to ensure consistent formatting
# Only format if we have valid data
if (current_luck_god is not None and str(current_luck_god).strip()) or (current_luck_earth is not None and str(current_luck_earth).strip()):
replacements['{{current_luck_pillar}}'] = format_luck_description(
current_luck_god or "",
current_luck_earth or "",
current_luck_age_range or "",
None, # No ten god relationships available due to error
None, # No branch relationships available due to error
language,
)
else:
replacements['{{current_luck_pillar}}'] = none_value
except Exception as exc:
logger.error(f"Error processing additional BaZi data: {str(exc)}", exc_info=True)
replacements['{{chart_structure}}'] = none_value
replacements['{{life_number}}'] = none_value
replacements['{{life_palace}}'] = none_value
replacements['{{current_age}}'] = none_value
replacements['{{current_luck_pillar}}'] = none_value
replacements['{{luck_pillar_1}}'] = none_value
replacements['{{luck_pillar_2}}'] = none_value
replacements['{{luck_pillar_3}}'] = none_value
# Replace all placeholders in the template
try:
prompt = template
for placeholder, value in replacements.items():
prompt = prompt.replace(placeholder, 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(prompt)
prompt = filter_prompt_for_response(prompt)
if len(prompt) < original_length:
logger.debug(f"Stripped content after last '---' separator, new length: {len(prompt)} characters (was {original_length})")
logger.debug(f"Successfully prepared prompt, length: {len(prompt)} characters")
return prompt
except Exception as e:
logger.error(f"Error replacing placeholders in template: {str(e)}", exc_info=True)
raise ValueError(f"Error formatting BaZi prompt: {str(e)}")
def analyze_bazi(
person: Person,
model_key: Optional[str] = None,
provider: Optional[str] = None,
language: Optional[str] = None,
) -> Dict[str, Any]:
"""
Analyze a person's BaZi chart using AI.
This function uses the active template from the database if available,
otherwise falls back to the file system template.
Args:
person: Person object to analyze
model_key: Optional model key to use (from settings.GROQ_MODELS or settings.OPENAI_MODELS)
provider: Optional LLM provider to use ('groq' or 'openai')
Returns:
Dictionary with analysis results
"""
try:
language = normalize_language(language)
# Get AI configuration for bazi if not provided
if not provider or not model_key:
from ai.utils.config import get_ai_config
config = get_ai_config('bazi')
provider = provider or config['provider']
model_key = model_key or config['model']
# Log analysis attempt
logger.info(f"Starting BaZi analysis for person id={person.id} using provider={provider}, model={model_key}")
try:
# Create LLM service using factory
llm_service = LLMServiceFactory.get_service(provider)
logger.debug(f"Created LLM service: {llm_service.__class__.__name__}")
except Exception as e:
logger.error(f"Failed to create LLM service: {str(e)}")
raise ValueError(f"Could not initialize LLM service: {str(e)}")
# Use specified model if provided
if model_key:
try:
llm_service.change_model(model_key)
logger.debug(f"Changed model to {model_key}")
except Exception as e:
logger.error(f"Failed to change model to {model_key}: {str(e)}")
raise ValueError(f"Could not change model to {model_key}: {str(e)}")
# Prepare the prompt (using active template from database if available)
try:
prompt = prepare_bazi_prompt(person, language=language)
logger.debug(f"Prepared BaZi prompt using active template, length: {len(prompt)} characters")
except Exception as e:
logger.error(f"Failed to prepare BaZi prompt: {str(e)}")
raise ValueError(f"Could not prepare BaZi prompt: {str(e)}")
# Get completion from service
try:
logger.debug(f"Requesting LLM completion")
raw_analysis = llm_service.get_completion(
prompt=prompt,
temperature=0.7,
max_tokens=4096
)
logger.debug(f"Received LLM response, length: {len(raw_analysis) if raw_analysis else 0} characters")
if not raw_analysis:
logger.error("LLM returned empty response")
raise ValueError("Empty response from LLM service")
except Exception as e:
logger.error(f"Failed to get LLM completion: {str(e)}")
raise ValueError(f"Error getting completion from LLM service: {str(e)}")
# Initialize analysis and thinking content
analysis = raw_analysis.strip()
thinking_content = None
try:
# Extract thinking content if present
thinking_match = re.search(r'(.*?)', analysis, re.DOTALL)
if thinking_match:
thinking_content = thinking_match.group(1).strip()
# Remove the thinking section from the analysis
analysis = re.sub(r'.*?', '', analysis, flags=re.DOTALL).strip()
logger.debug(f"Extracted thinking content, length: {len(thinking_content)} characters")
# Extract content from triple backticks if present
backticks_match = re.search(r'```(?:json)?(.*?)```', analysis, re.DOTALL)
if backticks_match:
# Get the content inside backticks
backticks_content = backticks_match.group(1).strip()
analysis = backticks_content
logger.debug("Extracted content from backticks")
# Try to parse as JSON to ensure it's valid
try:
json_data = json.loads(analysis)
# If successful parsing, use the parsed JSON object
analysis = json_data
logger.debug("Successfully parsed analysis as JSON")
except json.JSONDecodeError as json_err:
# If not valid JSON, keep as string
logger.warning(f"Analysis not valid JSON, using as string: {str(json_err)}")
except Exception as e:
logger.error(f"Failed to process LLM response: {str(e)}")
logger.error(f"Raw analysis: {raw_analysis[:100]}...") # Log the beginning of the raw analysis
raise ValueError(f"Error processing LLM response: {str(e)}")
# Get the actual model used
try:
used_model = llm_service.model
used_provider = provider
logger.debug(f"Used model: {used_model}, provider: {used_provider}")
except Exception as e:
logger.error(f"Failed to get model information: {str(e)}")
used_model = model_key or "unknown"
used_provider = provider or "unknown"
# Prepare the ai_analysis dictionary
try:
ai_analysis = {
'bazi_analysis': analysis,
'prompt': filter_prompt_for_response(prompt),
'provider': used_provider,
'model': used_model,
'language': language,
}
# Add thinking content if it was found
if thinking_content:
ai_analysis['think'] = thinking_content
logger.debug(f"Prepared AI analysis dictionary with keys: {', '.join(ai_analysis.keys())}")
except Exception as e:
logger.error(f"Failed to prepare ai_analysis dictionary: {str(e)}")
raise ValueError(f"Error preparing analysis dictionary: {str(e)}")
# Save the analysis to the person object
try:
person.ai_analysis = ai_analysis
person.analysis_timestamp = timezone.now()
person.analysis_status = 'completed'
person.save(update_fields=['ai_analysis', 'analysis_timestamp', 'analysis_status'])
logger.info(f"Successfully saved analysis for person id={person.id}")
except Exception as e:
logger.error(f"Failed to save analysis to person object: {str(e)}")
raise ValueError(f"Error saving analysis to database: {str(e)}")
return person.ai_analysis
except Exception as e:
logger.error(f"Error analyzing BaZi for person {person.id}: {str(e)}", exc_info=True)
raise