| | 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 |
| | } |
| |
|
