164 lines
5.4 KiB
Python
164 lines
5.4 KiB
Python
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from app.api.deps import get_db, get_current_active_user
|
|
from app.models.ppr import Aircraft, UserAircraft
|
|
from app.schemas.ppr import Aircraft as AircraftSchema, UserAircraftCreate
|
|
from app.models.ppr import User
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/lookup/{registration}", response_model=List[AircraftSchema])
|
|
async def lookup_aircraft_by_registration(
|
|
registration: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""
|
|
Lookup aircraft by registration (clean match).
|
|
Removes non-alphanumeric characters from input for matching.
|
|
Checks user_aircraft table first, then aircraft table.
|
|
"""
|
|
# Clean the input registration (remove non-alphanumeric characters)
|
|
clean_input = ''.join(c for c in registration if c.isalnum()).upper()
|
|
|
|
if len(clean_input) < 4:
|
|
return []
|
|
|
|
# First check user_aircraft table
|
|
user_aircraft = db.query(UserAircraft).filter(
|
|
UserAircraft.clean_reg.like(f"{clean_input}%")
|
|
).limit(10).all()
|
|
|
|
if user_aircraft:
|
|
# Convert UserAircraft to Aircraft-like objects
|
|
result = []
|
|
for ua in user_aircraft:
|
|
# Create a mock Aircraft object with the user data
|
|
result.append({
|
|
'id': ua.id,
|
|
'registration': ua.registration,
|
|
'type_code': ua.type_code,
|
|
'clean_reg': ua.clean_reg,
|
|
'icao24': None,
|
|
'manufacturer_icao': None,
|
|
'manufacturer_name': None,
|
|
'model': None
|
|
})
|
|
return result
|
|
|
|
# If no user aircraft found, check main aircraft table
|
|
aircraft_list = db.query(Aircraft).filter(
|
|
Aircraft.clean_reg.like(f"{clean_input}%")
|
|
).limit(10).all()
|
|
|
|
return aircraft_list
|
|
|
|
|
|
@router.get("/public/lookup/{registration}", response_model=List[AircraftSchema])
|
|
async def public_lookup_aircraft_by_registration(
|
|
registration: str,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Public lookup aircraft by registration (clean match).
|
|
Removes non-alphanumeric characters from input for matching.
|
|
No authentication required.
|
|
Checks user_aircraft table first, then aircraft table.
|
|
"""
|
|
# Clean the input registration (remove non-alphanumeric characters)
|
|
clean_input = ''.join(c for c in registration if c.isalnum()).upper()
|
|
|
|
if len(clean_input) < 4:
|
|
return []
|
|
|
|
# First check user_aircraft table
|
|
user_aircraft = db.query(UserAircraft).filter(
|
|
UserAircraft.clean_reg.like(f"{clean_input}%")
|
|
).limit(10).all()
|
|
|
|
if user_aircraft:
|
|
# Convert UserAircraft to Aircraft-like objects
|
|
result = []
|
|
for ua in user_aircraft:
|
|
result.append({
|
|
'id': ua.id,
|
|
'registration': ua.registration,
|
|
'type_code': ua.type_code,
|
|
'clean_reg': ua.clean_reg,
|
|
'icao24': None,
|
|
'manufacturer_icao': None,
|
|
'manufacturer_name': None,
|
|
'model': None
|
|
})
|
|
return result
|
|
|
|
# If no user aircraft found, check main aircraft table
|
|
aircraft_list = db.query(Aircraft).filter(
|
|
Aircraft.clean_reg.like(f"{clean_input}%")
|
|
).limit(10).all()
|
|
|
|
return aircraft_list
|
|
|
|
|
|
@router.get("/search", response_model=List[AircraftSchema])
|
|
async def search_aircraft(
|
|
q: Optional[str] = Query(None, description="Search query for registration, type, or manufacturer"),
|
|
limit: int = Query(10, ge=1, le=100, description="Maximum number of results"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""
|
|
Search aircraft by registration, type code, or manufacturer name.
|
|
"""
|
|
if not q or len(q) < 2:
|
|
return []
|
|
|
|
# Clean search term
|
|
clean_query = ''.join(c for c in q if c.isalnum()).upper()
|
|
|
|
# Search across multiple fields
|
|
aircraft_list = db.query(Aircraft).filter(
|
|
(Aircraft.clean_reg.like(f"%{clean_query}%")) |
|
|
(Aircraft.type_code.like(f"%{q.upper()}%")) |
|
|
(Aircraft.manufacturer_name.like(f"%{q}%")) |
|
|
(Aircraft.model.like(f"%{q}%"))
|
|
).limit(limit).all()
|
|
|
|
return aircraft_list
|
|
|
|
|
|
@router.post("/user-aircraft", response_model=dict)
|
|
async def save_user_aircraft(
|
|
aircraft: UserAircraftCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""
|
|
Save a user-defined aircraft type for a registration.
|
|
"""
|
|
# Clean the registration
|
|
clean_reg = ''.join(c for c in aircraft.registration if c.isalnum()).upper()
|
|
|
|
# Check if already exists
|
|
existing = db.query(UserAircraft).filter(
|
|
UserAircraft.registration == aircraft.registration.upper()
|
|
).first()
|
|
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Aircraft registration already exists in user database")
|
|
|
|
# Create new user aircraft
|
|
user_aircraft = UserAircraft(
|
|
registration=aircraft.registration.upper(),
|
|
type_code=aircraft.type_code.upper(),
|
|
clean_reg=clean_reg,
|
|
created_by=current_user.username
|
|
)
|
|
|
|
db.add(user_aircraft)
|
|
db.commit()
|
|
db.refresh(user_aircraft)
|
|
|
|
return {"message": "Aircraft saved successfully", "id": user_aircraft.id} |