# app/services/letzshop_export_service.py """ Service for exporting products to Letzshop CSV format. Generates Google Shopping compatible CSV files for Letzshop marketplace. """ import csv import io import logging from sqlalchemy.orm import Session, joinedload from models.database.marketplace_product import MarketplaceProduct from models.database.product import Product logger = logging.getLogger(__name__) # Letzshop CSV columns in order LETZSHOP_CSV_COLUMNS = [ "id", "title", "description", "link", "image_link", "additional_image_link", "availability", "price", "sale_price", "brand", "gtin", "mpn", "google_product_category", "product_type", "condition", "adult", "multipack", "is_bundle", "age_group", "color", "gender", "material", "pattern", "size", "size_type", "size_system", "item_group_id", "custom_label_0", "custom_label_1", "custom_label_2", "custom_label_3", "custom_label_4", "identifier_exists", "unit_pricing_measure", "unit_pricing_base_measure", "shipping", "atalanda:tax_rate", "atalanda:quantity", "atalanda:boost_sort", "atalanda:delivery_method", ] class LetzshopExportService: """Service for exporting products to Letzshop CSV format.""" def __init__(self, default_tax_rate: float = 17.0): """ Initialize the export service. Args: default_tax_rate: Default VAT rate for Luxembourg (17%) """ self.default_tax_rate = default_tax_rate def export_vendor_products( self, db: Session, vendor_id: int, language: str = "en", include_inactive: bool = False, ) -> str: """ Export all products for a vendor in Letzshop CSV format. Args: db: Database session vendor_id: Vendor ID to export products for language: Language for title/description (en, fr, de) include_inactive: Whether to include inactive products Returns: CSV string content """ # Query products for this vendor with their marketplace product data query = ( db.query(Product) .filter(Product.vendor_id == vendor_id) .options( joinedload(Product.marketplace_product).joinedload( MarketplaceProduct.translations ) ) ) if not include_inactive: query = query.filter(Product.is_active == True) products = query.all() logger.info( f"Exporting {len(products)} products for vendor {vendor_id} in {language}" ) return self._generate_csv(products, language) def export_marketplace_products( self, db: Session, marketplace: str = "Letzshop", language: str = "en", limit: int | None = None, ) -> str: """ Export marketplace products directly (admin use). Args: db: Database session marketplace: Filter by marketplace source language: Language for title/description limit: Optional limit on number of products Returns: CSV string content """ query = ( db.query(MarketplaceProduct) .filter(MarketplaceProduct.is_active == True) .options(joinedload(MarketplaceProduct.translations)) ) if marketplace: query = query.filter( MarketplaceProduct.marketplace.ilike(f"%{marketplace}%") ) if limit: query = query.limit(limit) products = query.all() logger.info( f"Exporting {len(products)} marketplace products for {marketplace} in {language}" ) return self._generate_csv_from_marketplace_products(products, language) def _generate_csv(self, products: list[Product], language: str) -> str: """Generate CSV from vendor Product objects.""" output = io.StringIO() writer = csv.DictWriter( output, fieldnames=LETZSHOP_CSV_COLUMNS, delimiter="\t", quoting=csv.QUOTE_MINIMAL, ) writer.writeheader() for product in products: if product.marketplace_product: row = self._product_to_row(product, language) writer.writerow(row) return output.getvalue() def _generate_csv_from_marketplace_products( self, products: list[MarketplaceProduct], language: str ) -> str: """Generate CSV from MarketplaceProduct objects directly.""" output = io.StringIO() writer = csv.DictWriter( output, fieldnames=LETZSHOP_CSV_COLUMNS, delimiter="\t", quoting=csv.QUOTE_MINIMAL, ) writer.writeheader() for mp in products: row = self._marketplace_product_to_row(mp, language) writer.writerow(row) return output.getvalue() def _product_to_row(self, product: Product, language: str) -> dict: """Convert a Product (with MarketplaceProduct) to a CSV row.""" mp = product.marketplace_product return self._marketplace_product_to_row( mp, language, vendor_sku=product.vendor_sku ) def _marketplace_product_to_row( self, mp: MarketplaceProduct, language: str, vendor_sku: str | None = None, ) -> dict: """Convert a MarketplaceProduct to a CSV row dict.""" # Get localized title and description title = mp.get_title(language) or "" description = mp.get_description(language) or "" # Format price with currency price = "" if mp.price_numeric: price = f"{mp.price_numeric:.2f} {mp.currency or 'EUR'}" elif mp.price: price = mp.price # Format sale price sale_price = "" if mp.sale_price_numeric: sale_price = f"{mp.sale_price_numeric:.2f} {mp.currency or 'EUR'}" elif mp.sale_price: sale_price = mp.sale_price # Additional images - join with comma if multiple additional_images = "" if mp.additional_images: additional_images = ",".join(mp.additional_images) elif mp.additional_image_link: additional_images = mp.additional_image_link # Determine identifier_exists identifier_exists = mp.identifier_exists if not identifier_exists: identifier_exists = "yes" if (mp.gtin or mp.mpn) else "no" return { "id": vendor_sku or mp.marketplace_product_id, "title": title, "description": description, "link": mp.link or mp.source_url or "", "image_link": mp.image_link or "", "additional_image_link": additional_images, "availability": mp.availability or "in stock", "price": price, "sale_price": sale_price, "brand": mp.brand or "", "gtin": mp.gtin or "", "mpn": mp.mpn or "", "google_product_category": mp.google_product_category or "", "product_type": mp.product_type_raw or "", "condition": mp.condition or "new", "adult": mp.adult or "no", "multipack": str(mp.multipack) if mp.multipack else "", "is_bundle": mp.is_bundle or "no", "age_group": mp.age_group or "", "color": mp.color or "", "gender": mp.gender or "", "material": mp.material or "", "pattern": mp.pattern or "", "size": mp.size or "", "size_type": mp.size_type or "", "size_system": mp.size_system or "", "item_group_id": mp.item_group_id or "", "custom_label_0": mp.custom_label_0 or "", "custom_label_1": mp.custom_label_1 or "", "custom_label_2": mp.custom_label_2 or "", "custom_label_3": mp.custom_label_3 or "", "custom_label_4": mp.custom_label_4 or "", "identifier_exists": identifier_exists, "unit_pricing_measure": mp.unit_pricing_measure or "", "unit_pricing_base_measure": mp.unit_pricing_base_measure or "", "shipping": mp.shipping or "", "atalanda:tax_rate": str(self.default_tax_rate), "atalanda:quantity": "", # Would need inventory data "atalanda:boost_sort": "", "atalanda:delivery_method": "", } # Singleton instance letzshop_export_service = LetzshopExportService()