feat: add Letzshop vendor directory with sync and admin management

- Add LetzshopVendorCache model to store cached vendor data from Letzshop API
- Create LetzshopVendorSyncService for syncing vendor directory
- Add Celery task for background vendor sync
- Create admin page at /admin/letzshop/vendor-directory with:
  - Stats dashboard (total, claimed, unclaimed vendors)
  - Searchable/filterable vendor list
  - "Sync Now" button to trigger sync
  - Ability to create platform vendors from Letzshop cache
- Add API endpoints for vendor directory management
- Add Pydantic schemas for API responses

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-13 20:35:46 +01:00
parent 78b14a4b00
commit ccfbbcb804
13 changed files with 2571 additions and 46 deletions

View File

@@ -172,6 +172,153 @@ class LetzshopSyncLog(Base, TimestampMixin):
return f"<LetzshopSyncLog(id={self.id}, type='{self.operation_type}', status='{self.status}')>"
class LetzshopVendorCache(Base, TimestampMixin):
"""
Cache of Letzshop marketplace vendor directory.
This table stores vendor data fetched from Letzshop's public GraphQL API,
allowing users to browse and claim existing Letzshop shops during signup.
Data is periodically synced from Letzshop (e.g., daily via Celery task).
"""
__tablename__ = "letzshop_vendor_cache"
id = Column(Integer, primary_key=True, index=True)
# Letzshop identifiers
letzshop_id = Column(String(50), unique=True, nullable=False, index=True)
"""Unique ID from Letzshop (e.g., 'lpkedYMRup')."""
slug = Column(String(200), unique=True, nullable=False, index=True)
"""URL slug (e.g., 'nicks-diecast-corner')."""
# Basic info
name = Column(String(255), nullable=False)
"""Vendor display name."""
company_name = Column(String(255), nullable=True)
"""Legal company name."""
is_active = Column(Boolean, default=True)
"""Whether vendor is active on Letzshop."""
# Descriptions (multilingual)
description_en = Column(Text, nullable=True)
description_fr = Column(Text, nullable=True)
description_de = Column(Text, nullable=True)
# Contact information
email = Column(String(255), nullable=True)
phone = Column(String(50), nullable=True)
fax = Column(String(50), nullable=True)
website = Column(String(500), nullable=True)
# Location
street = Column(String(255), nullable=True)
street_number = Column(String(50), nullable=True)
city = Column(String(100), nullable=True)
zipcode = Column(String(20), nullable=True)
country_iso = Column(String(5), default="LU")
latitude = Column(String(20), nullable=True)
longitude = Column(String(20), nullable=True)
# Categories (stored as JSON array of names)
categories = Column(JSON, default=list)
"""List of category names, e.g., ['Fashion', 'Shoes']."""
# Images
background_image_url = Column(String(500), nullable=True)
# Social media (stored as JSON array of URLs)
social_media_links = Column(JSON, default=list)
"""List of social media URLs."""
# Opening hours (multilingual text)
opening_hours_en = Column(Text, nullable=True)
opening_hours_fr = Column(Text, nullable=True)
opening_hours_de = Column(Text, nullable=True)
# Representative
representative_name = Column(String(255), nullable=True)
representative_title = Column(String(100), nullable=True)
# Claiming status (linked to our platform)
claimed_by_vendor_id = Column(
Integer, ForeignKey("vendors.id"), nullable=True, index=True
)
"""If claimed, links to our Vendor record."""
claimed_at = Column(DateTime(timezone=True), nullable=True)
"""When the vendor was claimed on our platform."""
# Sync metadata
last_synced_at = Column(DateTime(timezone=True), nullable=False)
"""When this record was last updated from Letzshop."""
raw_data = Column(JSON, nullable=True)
"""Full raw response from Letzshop API for reference."""
# Relationship to claimed vendor
claimed_vendor = relationship("Vendor", foreign_keys=[claimed_by_vendor_id])
__table_args__ = (
Index("idx_vendor_cache_city", "city"),
Index("idx_vendor_cache_claimed", "claimed_by_vendor_id"),
Index("idx_vendor_cache_active", "is_active"),
)
def __repr__(self):
return f"<LetzshopVendorCache(id={self.id}, slug='{self.slug}', name='{self.name}')>"
@property
def is_claimed(self) -> bool:
"""Check if this vendor has been claimed on our platform."""
return self.claimed_by_vendor_id is not None
@property
def letzshop_url(self) -> str:
"""Get the Letzshop profile URL."""
return f"https://letzshop.lu/vendors/{self.slug}"
def get_description(self, lang: str = "en") -> str | None:
"""Get description in specified language with fallback."""
descriptions = {
"en": self.description_en,
"fr": self.description_fr,
"de": self.description_de,
}
# Try requested language, then fallback order
for try_lang in [lang, "en", "fr", "de"]:
if descriptions.get(try_lang):
return descriptions[try_lang]
return None
def get_opening_hours(self, lang: str = "en") -> str | None:
"""Get opening hours in specified language with fallback."""
hours = {
"en": self.opening_hours_en,
"fr": self.opening_hours_fr,
"de": self.opening_hours_de,
}
for try_lang in [lang, "en", "fr", "de"]:
if hours.get(try_lang):
return hours[try_lang]
return None
def get_full_address(self) -> str | None:
"""Get formatted full address."""
parts = []
if self.street:
addr = self.street
if self.street_number:
addr += f" {self.street_number}"
parts.append(addr)
if self.zipcode or self.city:
parts.append(f"{self.zipcode or ''} {self.city or ''}".strip())
return ", ".join(parts) if parts else None
class LetzshopHistoricalImportJob(Base, TimestampMixin):
"""
Track progress of historical order imports from Letzshop.