| from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine |
| from database_interaction.models import User, Base |
| from sqlalchemy.orm import sessionmaker |
| from geopy.geocoders import Nominatim |
| from dotenv import load_dotenv |
| import os |
|
|
| load_dotenv() |
| |
| DATABASE_URL = "sqlite+aiosqlite:///./database_files/main.db" |
| engine = create_async_engine(DATABASE_URL, echo=False) |
| AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) |
| geolocator = Nominatim(user_agent="ai_assistant") |
|
|
| def get_location_name(lat: float, lon: float) -> str: |
| """Get location name from coordinates""" |
| try: |
| location = geolocator.reverse((lat, lon), language='en') |
| if location and location.raw and 'address' in location.raw: |
| address = location.raw['address'] |
| return ( |
| address.get('city') |
| or address.get('town') |
| or address.get('village') |
| or address.get('municipality') |
| or address.get('county') |
| or "Unknown" |
| ) |
| return "Unknown" |
| except Exception as e: |
| print(f"[Geocoding Error] {e}") |
| return "Unknown" |
|
|
| async def init_user_db(): |
| """Initialize user database tables""" |
| async with engine.begin() as conn: |
| await conn.run_sync(Base.metadata.create_all) |
|
|
| async def create_or_update_user(user_id: str, first_name: str = None, last_name: str = None, |
| latitude: float = None, longitude: float = None): |
| """Create or update user information |
| :rtype: None |
| """ |
| location_name = get_location_name(latitude, longitude) if latitude and longitude else None |
| async with AsyncSessionLocal() as session: |
| async with session.begin(): |
| user = await session.get(User, user_id) |
| if user: |
| if first_name is not None: |
| user.first_name = first_name |
| if last_name is not None: |
| user.last_name = last_name |
| if latitude is not None: |
| user.latitude = latitude |
| if longitude is not None: |
| user.longitude = longitude |
| if location_name is not None: |
| user.location = location_name |
| else: |
| user = User( |
| user_id=user_id, |
| first_name=first_name, |
| last_name=last_name, |
| latitude=latitude, |
| longitude=longitude, |
| location=location_name |
| ) |
| session.add(user) |
| await session.commit() |
|
|
| async def get_user_by_id(user_id: str): |
| """Get user by ID""" |
| async with AsyncSessionLocal() as session: |
| result = await session.get(User, user_id) |
| if result: |
| return { |
| "user_id": user_id, |
| "first_name": result.first_name, |
| "last_name": result.last_name, |
| "latitude": result.latitude, |
| "longitude": result.longitude, |
| "location": result.location |
| } |
| return { |
| "user_id": user_id, |
| "first_name": None, |
| "last_name": None, |
| "latitude": None, |
| "longitude": None, |
| "location": None |
| } |
|
|
