import os import re import time import shutil import requests from pathlib import Path from datetime import datetime from pydantic import BaseModel from kerykeion import AstrologicalSubject, KerykeionChartSVG from chart_utils import extract_synastry_wheel class SynastryPersonInput(BaseModel): name: str year: int month: int day: int hour: int minute: int place: str class SynastryChartInput(BaseModel): person1: SynastryPersonInput person2: SynastryPersonInput def safe_file_name(name: str) -> str: name = name.strip().lower() name = re.sub(r"[^a-z0-9]+", "-", name) name = re.sub(r"-+", "-", name).strip("-") return name or "user" def get_place_details(place: str) -> dict: """ Place name to latitude/longitude. First try Photon. If Photon fails, fallback to Nominatim. Timezone fixed: Asia/Kolkata """ headers = { "User-Agent": "AstrologyApp/1.0 (contact: harsh@example.com)", "Accept": "application/json", } # 1. Photon geocoder try: photon_url = "https://photon.komoot.io/api/" photon_params = { "q": place, "limit": 1, } response = requests.get( photon_url, params=photon_params, headers=headers, timeout=15, ) response.raise_for_status() data = response.json() features = data.get("features", []) if features: feature = features[0] coordinates = feature.get("geometry", {}).get("coordinates", []) if len(coordinates) >= 2: longitude = float(coordinates[0]) latitude = float(coordinates[1]) props = feature.get("properties", {}) city = ( props.get("city") or props.get("name") or props.get("county") or place ) country = props.get("country", "India") return { "latitude": latitude, "longitude": longitude, "city": city, "country": country, "timezone_str": "Asia/Kolkata", } except Exception as photon_error: print(f"Photon geocoder failed for '{place}': {photon_error}") time.sleep(1) # 2. Nominatim fallback try: nominatim_url = "https://nominatim.openstreetmap.org/search" nominatim_params = { "q": place, "format": "json", "limit": 1, "addressdetails": 1, "countrycodes": "in", } response = requests.get( nominatim_url, params=nominatim_params, headers=headers, timeout=15, ) response.raise_for_status() data = response.json() if not data: raise ValueError(f"Place not found: {place}") item = data[0] latitude = float(item["lat"]) longitude = float(item["lon"]) address = item.get("address", {}) city = ( address.get("city") or address.get("town") or address.get("village") or address.get("county") or address.get("state_district") or place ) country = address.get("country", "India") return { "latitude": latitude, "longitude": longitude, "city": city, "country": country, "timezone_str": "Asia/Kolkata", } except Exception as nominatim_error: raise ValueError( f"Location error for '{place}'. " f"Photon and Nominatim both failed. " f"Error: {str(nominatim_error)}" ) def find_latest_svg(folder_path: str, max_retries: int = 5, delay: float = 0.5) -> Path: """Find latest SVG with retry logic to wait for file to be written.""" folder = Path(folder_path) for attempt in range(max_retries): svg_files = list(folder.rglob("*.svg")) if svg_files: latest = max(svg_files, key=lambda file: file.stat().st_mtime) # Verify file is readable try: latest.stat().st_size return latest except (OSError, FileNotFoundError): pass if attempt < max_retries - 1: time.sleep(delay) raise FileNotFoundError("Kerykeion SVG chart was not generated.") def create_subject(person: SynastryPersonInput, geo: dict) -> AstrologicalSubject: return AstrologicalSubject( person.name, person.year, person.month, person.day, person.hour, person.minute, lng=geo["longitude"], lat=geo["latitude"], tz_str=geo["timezone_str"], city=geo["city"], nation=geo["country"], online=False, ) def generate_synastry_chart_svg(data: SynastryChartInput, charts_dir: str) -> dict: """ Generate full Kerykeion Synastry chart, then extract only wheel like birth chart. Save final extracted chart in: charts/synastry-chart/ Return: { "chart": "/charts/synastry-chart/file.svg" } """ synastry_folder = os.path.join(charts_dir, "synastry-chart") os.makedirs(synastry_folder, exist_ok=True) temp_dir = None try: # 1. Convert places to lat/lon person1_geo = get_place_details(data.person1.place) person2_geo = get_place_details(data.person2.place) # 2. Create Kerykeion subjects person1 = create_subject(data.person1, person1_geo) person2 = create_subject(data.person2, person2_geo) # 3. Temporary output folder timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") temp_dir = os.path.join(synastry_folder, f"temp_{timestamp}") os.makedirs(temp_dir, exist_ok=True) # 4. Generate full Synastry SVG chart = KerykeionChartSVG( person1, chart_type="Synastry", second_obj=person2, new_output_directory=temp_dir, ) chart.makeSVG() # Wait for file to be written to disk time.sleep(1) # 5. Find generated full SVG full_chart_svg = find_latest_svg(temp_dir) p1_name = safe_file_name(data.person1.name) p2_name = safe_file_name(data.person2.name) final_file_name = ( f"{p1_name}_and_{p2_name}_synastry_chart_{timestamp}.svg" ) final_path = os.path.join(synastry_folder, final_file_name) # 6. Extract only wheel from full Synastry SVG extracted = extract_synastry_wheel( str(full_chart_svg), final_path ) if not extracted: # If extraction fails, save full chart as fallback shutil.move(str(full_chart_svg), final_path) # 7. Cleanup temp folder shutil.rmtree(temp_dir, ignore_errors=True) return { "chart": f"/charts/synastry-chart/{final_file_name}" } except Exception as e: if temp_dir and os.path.exists(temp_dir): shutil.rmtree(temp_dir, ignore_errors=True) raise Exception(f"Synastry chart generation failed: {str(e)}")