from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from typing import Optional from astrology import calculate_planets from db import get_db_connection from auth_routes import get_current_user router = APIRouter(prefix="/user", tags=["User Settings"]) def convert_to_24_hour(hour: int, minute: int, ampm: str): h = int(hour) m = int(minute) ampm_upper = ampm.upper().strip() if ampm else "PM" if ampm_upper == "AM": if h == 12: h = 0 elif ampm_upper == "PM": if h != 12: h += 12 return h, m def calculate_user_birth_signs(details: dict): try: if not details: return None, None, None lat = details.get("birth_lat") lon = details.get("birth_lon") if lat is None or lon is None: print("Cannot calculate user signs: birth_lat/birth_lon missing") return None, None, None hour = details.get("birth_hour") or 12 minute = details.get("birth_minute") or 0 ampm = details.get("birth_ampm") or "PM" if details.get("unknown_birth_time"): hour = 12 minute = 0 ampm = "PM" calc_hour, calc_minute = convert_to_24_hour(hour, minute, ampm) result = calculate_planets( year=int(details["birth_year"]), month=int(details["birth_month"]), day=int(details["birth_day"]), hour=calc_hour, minute=calc_minute, second=0, latitude=float(lat), longitude=float(lon), tz=0.0, ) sun_sign = result.get("planets", {}).get("sun", {}).get("sign") moon_sign = result.get("planets", {}).get("moon", {}).get("sign") rising_sign = result.get("ascendant", {}).get("sign") print("User settings signs:", { "sun": sun_sign, "moon": moon_sign, "rising": rising_sign, }) return sun_sign, moon_sign, rising_sign except Exception as e: print("Error calculating user birth signs:", e) return None, None, None def attach_signs_to_details(details: dict): if not details: return details sun_sign, moon_sign, rising_sign = calculate_user_birth_signs(details) details["sun_sign"] = sun_sign details["moon_sign"] = moon_sign details["rising_sign"] = rising_sign return details class BirthDetailsUpdateRequest(BaseModel): name: str birth_year: int birth_month: int birth_day: int birth_hour: int birth_minute: int birth_ampm: str unknown_birth_time: bool = False birth_place: str birth_lat: Optional[float] = None birth_lon: Optional[float] = None timezone: Optional[str] = None @router.get("/birth-details") def get_user_birth_details(current_user=Depends(get_current_user)): conn = get_db_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute( """ SELECT ubd.id, ubd.user_id, ubd.name, ubd.birth_year, ubd.birth_month, ubd.birth_day, ubd.birth_hour, ubd.birth_minute, ubd.birth_ampm, ubd.unknown_birth_time, ubd.birth_place, ubd.birth_lat, ubd.birth_lon, ubd.timezone, ubd.created_at, ubd.updated_at, u.email FROM user_birth_details ubd LEFT JOIN users u ON u.id = ubd.user_id WHERE ubd.user_id = %s ORDER BY ubd.id DESC LIMIT 1 """, (current_user["id"],) ) details = cursor.fetchone() if not details: raise HTTPException(status_code=404, detail="Birth details not found.") details = attach_signs_to_details(details) return details finally: cursor.close() conn.close() @router.put("/birth-details") def update_user_birth_details( req: BirthDetailsUpdateRequest, current_user=Depends(get_current_user) ): if not req.name.strip(): raise HTTPException(status_code=400, detail="Name is required.") if req.birth_year < 1900 or req.birth_year > 2100: raise HTTPException(status_code=400, detail="Invalid birth year.") if req.birth_month < 1 or req.birth_month > 12: raise HTTPException(status_code=400, detail="Invalid birth month.") if req.birth_day < 1 or req.birth_day > 31: raise HTTPException(status_code=400, detail="Invalid birth day.") if req.birth_hour < 1 or req.birth_hour > 12: raise HTTPException(status_code=400, detail="Birth hour must be between 1 and 12.") if req.birth_minute < 0 or req.birth_minute > 59: raise HTTPException(status_code=400, detail="Birth minute must be between 0 and 59.") if req.birth_ampm.upper() not in {"AM", "PM"}: raise HTTPException(status_code=400, detail="Birth AM/PM must be AM or PM.") if not req.birth_place.strip(): raise HTTPException(status_code=400, detail="Birth place is required.") birth_hour = req.birth_hour birth_minute = req.birth_minute birth_ampm = req.birth_ampm.upper() if req.unknown_birth_time: birth_hour = 12 birth_minute = 0 birth_ampm = "PM" conn = get_db_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute( """ SELECT id FROM user_birth_details WHERE user_id = %s ORDER BY id DESC LIMIT 1 """, (current_user["id"],) ) existing = cursor.fetchone() if not existing: raise HTTPException(status_code=404, detail="Birth details not found.") cursor.execute( """ UPDATE user_birth_details SET name = %s, birth_year = %s, birth_month = %s, birth_day = %s, birth_hour = %s, birth_minute = %s, birth_ampm = %s, unknown_birth_time = %s, birth_place = %s, birth_lat = %s, birth_lon = %s, timezone = %s WHERE id = %s AND user_id = %s """, ( req.name.strip(), req.birth_year, req.birth_month, req.birth_day, birth_hour, birth_minute, birth_ampm, req.unknown_birth_time, req.birth_place.strip(), req.birth_lat, req.birth_lon, req.timezone, existing["id"], current_user["id"], ) ) conn.commit() cursor.execute( """ SELECT ubd.id, ubd.user_id, ubd.name, ubd.birth_year, ubd.birth_month, ubd.birth_day, ubd.birth_hour, ubd.birth_minute, ubd.birth_ampm, ubd.unknown_birth_time, ubd.birth_place, ubd.birth_lat, ubd.birth_lon, ubd.timezone, ubd.created_at, ubd.updated_at, u.email FROM user_birth_details ubd LEFT JOIN users u ON u.id = ubd.user_id WHERE ubd.id = %s AND ubd.user_id = %s """, (existing["id"], current_user["id"]) ) updated_details = cursor.fetchone() updated_details = attach_signs_to_details(updated_details) return { "success": True, "message": "Birth details updated successfully.", "data": updated_details } finally: cursor.close() conn.close()