# app/modules/marketplace/routes/api/public.py """ Public Letzshop vendor lookup API endpoints. Allows potential vendors to find themselves in the Letzshop marketplace and claim their shop during signup. All endpoints are public (no authentication required). """ import logging import re from typing import Annotated from fastapi import APIRouter, Depends, Query from pydantic import BaseModel from sqlalchemy.orm import Session from app.core.database import get_db from app.exceptions import ResourceNotFoundException from app.modules.marketplace.services.letzshop import LetzshopVendorSyncService from app.modules.marketplace.models import LetzshopVendorCache router = APIRouter(prefix="/letzshop-vendors") logger = logging.getLogger(__name__) # ============================================================================= # Response Schemas # ============================================================================= class LetzshopVendorInfo(BaseModel): """Letzshop vendor information for display.""" letzshop_id: str | None = None slug: str name: str company_name: str | None = None description: str | None = None email: str | None = None phone: str | None = None website: str | None = None address: str | None = None city: str | None = None categories: list[str] = [] background_image_url: str | None = None social_media_links: list[str] = [] letzshop_url: str is_claimed: bool = False @classmethod def from_cache(cls, cache: LetzshopVendorCache, lang: str = "en") -> "LetzshopVendorInfo": """Create from cache entry.""" return cls( letzshop_id=cache.letzshop_id, slug=cache.slug, name=cache.name, company_name=cache.company_name, description=cache.get_description(lang), email=cache.email, phone=cache.phone, website=cache.website, address=cache.get_full_address(), city=cache.city, categories=cache.categories or [], background_image_url=cache.background_image_url, social_media_links=cache.social_media_links or [], letzshop_url=cache.letzshop_url, is_claimed=cache.is_claimed, ) class LetzshopVendorListResponse(BaseModel): """Paginated list of Letzshop vendors.""" vendors: list[LetzshopVendorInfo] total: int page: int limit: int has_more: bool class LetzshopLookupRequest(BaseModel): """Request to lookup a Letzshop vendor by URL.""" url: str # e.g., https://letzshop.lu/vendors/my-shop or just "my-shop" class LetzshopLookupResponse(BaseModel): """Response from Letzshop vendor lookup.""" found: bool vendor: LetzshopVendorInfo | None = None error: str | None = None # ============================================================================= # Helper Functions # ============================================================================= def extract_slug_from_url(url_or_slug: str) -> str: """ Extract vendor slug from Letzshop URL or return as-is if already a slug. Handles: - https://letzshop.lu/vendors/my-shop - https://letzshop.lu/en/vendors/my-shop - letzshop.lu/vendors/my-shop - my-shop """ # Clean up the input url_or_slug = url_or_slug.strip() # If it looks like a URL, extract the slug if "letzshop" in url_or_slug.lower() or "/" in url_or_slug: # Remove protocol if present url_or_slug = re.sub(r"^https?://", "", url_or_slug) # Match pattern like letzshop.lu/[lang/]vendors/SLUG[/...] match = re.search(r"letzshop\.lu/(?:[a-z]{2}/)?vendors?/([^/?#]+)", url_or_slug, re.IGNORECASE) if match: return match.group(1).lower() # If just a path like vendors/my-shop match = re.search(r"vendors?/([^/?#]+)", url_or_slug) if match: return match.group(1).lower() # Return as-is (assume it's already a slug) return url_or_slug.lower() # ============================================================================= # Endpoints # ============================================================================= @router.get("", response_model=LetzshopVendorListResponse) # public async def list_letzshop_vendors( search: Annotated[str | None, Query(description="Search by name")] = None, category: Annotated[str | None, Query(description="Filter by category")] = None, city: Annotated[str | None, Query(description="Filter by city")] = None, only_unclaimed: Annotated[bool, Query(description="Only show unclaimed vendors")] = False, lang: Annotated[str, Query(description="Language for descriptions")] = "en", page: Annotated[int, Query(ge=1)] = 1, limit: Annotated[int, Query(ge=1, le=50)] = 20, db: Session = Depends(get_db), ) -> LetzshopVendorListResponse: """ List Letzshop vendors from cached directory. The cache is periodically synced from Letzshop's public GraphQL API. Run the sync task manually or wait for scheduled sync if cache is empty. """ sync_service = LetzshopVendorSyncService(db) vendors, total = sync_service.search_cached_vendors( search=search, city=city, category=category, only_unclaimed=only_unclaimed, page=page, limit=limit, ) return LetzshopVendorListResponse( vendors=[LetzshopVendorInfo.from_cache(v, lang) for v in vendors], total=total, page=page, limit=limit, has_more=(page * limit) < total, ) @router.post("/lookup", response_model=LetzshopLookupResponse) # public async def lookup_letzshop_vendor( request: LetzshopLookupRequest, lang: Annotated[str, Query(description="Language for descriptions")] = "en", db: Session = Depends(get_db), ) -> LetzshopLookupResponse: """ Lookup a Letzshop vendor by URL or slug. This endpoint: 1. Extracts the slug from the provided URL 2. Looks up vendor in local cache (or fetches from Letzshop if not cached) 3. Checks if the vendor is already claimed on our platform 4. Returns vendor info for signup pre-fill """ try: slug = extract_slug_from_url(request.url) if not slug: return LetzshopLookupResponse( found=False, error="Could not extract vendor slug from URL", ) sync_service = LetzshopVendorSyncService(db) # First try cache cache_entry = sync_service.get_cached_vendor(slug) # If not in cache, try to fetch from Letzshop if not cache_entry: logger.info(f"Vendor {slug} not in cache, fetching from Letzshop...") cache_entry = sync_service.sync_single_vendor(slug) if not cache_entry: return LetzshopLookupResponse( found=False, error="Vendor not found on Letzshop", ) return LetzshopLookupResponse( found=True, vendor=LetzshopVendorInfo.from_cache(cache_entry, lang), ) except Exception as e: logger.error(f"Error looking up Letzshop vendor: {e}") return LetzshopLookupResponse( found=False, error="Failed to lookup vendor", ) @router.get("/stats") # public async def get_letzshop_vendor_stats( db: Session = Depends(get_db), ) -> dict: """ Get statistics about the Letzshop vendor cache. Returns total, active, claimed, and unclaimed vendor counts. """ sync_service = LetzshopVendorSyncService(db) return sync_service.get_sync_stats() @router.get("/{slug}", response_model=LetzshopVendorInfo) # public async def get_letzshop_vendor( slug: str, lang: Annotated[str, Query(description="Language for descriptions")] = "en", db: Session = Depends(get_db), ) -> LetzshopVendorInfo: """ Get a specific Letzshop vendor by slug. Returns 404 if vendor not found in cache or on Letzshop. """ slug = slug.lower() sync_service = LetzshopVendorSyncService(db) # First try cache cache_entry = sync_service.get_cached_vendor(slug) # If not in cache, try to fetch from Letzshop if not cache_entry: logger.info(f"Vendor {slug} not in cache, fetching from Letzshop...") cache_entry = sync_service.sync_single_vendor(slug) if not cache_entry: raise ResourceNotFoundException("LetzshopVendor", slug) return LetzshopVendorInfo.from_cache(cache_entry, lang)