diff --git a/app/api/v1/admin/vendors.py b/app/api/v1/admin/vendors.py index e239ca17..01a450e8 100644 --- a/app/api/v1/admin/vendors.py +++ b/app/api/v1/admin/vendors.py @@ -308,3 +308,56 @@ def delete_vendor( message = admin_service.delete_vendor(db, vendor.id) db.commit() return {"message": message} + + +# ============================================================================ +# LETZSHOP EXPORT +# ============================================================================ + + +@router.get("/{vendor_identifier}/export/letzshop") +def export_vendor_products_letzshop( + vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"), + language: str = Query("en", description="Language for title/description (en, fr, de)"), + include_inactive: bool = Query(False, description="Include inactive products"), + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """ + Export vendor products in Letzshop CSV format (Admin only). + + Generates a Google Shopping compatible CSV file for Letzshop marketplace. + The file uses tab-separated values and includes all required Letzshop fields. + + **Supported languages:** en, fr, de + + **CSV Format:** + - Delimiter: Tab (\\t) + - Encoding: UTF-8 + - Fields: id, title, description, price, availability, image_link, etc. + + Returns: + CSV file as attachment (vendor_code_letzshop_export.csv) + """ + from fastapi.responses import Response + + from app.services.letzshop_export_service import letzshop_export_service + + vendor = vendor_service.get_vendor_by_identifier(db, vendor_identifier) + + csv_content = letzshop_export_service.export_vendor_products( + db=db, + vendor_id=vendor.id, + language=language, + include_inactive=include_inactive, + ) + + filename = f"{vendor.vendor_code.lower()}_letzshop_export.csv" + + return Response( + content=csv_content, + media_type="text/csv; charset=utf-8", + headers={ + "Content-Disposition": f'attachment; filename="{filename}"', + }, + ) diff --git a/app/api/v1/vendor/letzshop.py b/app/api/v1/vendor/letzshop.py index 3795a163..373a245b 100644 --- a/app/api/v1/vendor/letzshop.py +++ b/app/api/v1/vendor/letzshop.py @@ -690,3 +690,57 @@ def list_fulfillment_queue( skip=skip, limit=limit, ) + + +# ============================================================================ +# Product Export +# ============================================================================ + + +@router.get("/export") +def export_products_letzshop( + language: str = Query("en", description="Language for title/description (en, fr, de)"), + include_inactive: bool = Query(False, description="Include inactive products"), + current_user: User = Depends(get_current_vendor_api), + db: Session = Depends(get_db), +): + """ + Export vendor products in Letzshop CSV format. + + Generates a Google Shopping compatible CSV file for Letzshop marketplace. + The file uses tab-separated values and includes all required Letzshop fields. + + **Supported languages:** en, fr, de + + **CSV Format:** + - Delimiter: Tab (\\t) + - Encoding: UTF-8 + - Fields: id, title, description, price, availability, image_link, etc. + + Returns: + CSV file as attachment (vendor_code_letzshop_export.csv) + """ + from fastapi.responses import Response + + from app.services.letzshop_export_service import letzshop_export_service + from app.services.vendor_service import vendor_service + + vendor_id = current_user.token_vendor_id + vendor = vendor_service.get_vendor_by_id(db, vendor_id) + + csv_content = letzshop_export_service.export_vendor_products( + db=db, + vendor_id=vendor_id, + language=language, + include_inactive=include_inactive, + ) + + filename = f"{vendor.vendor_code.lower()}_letzshop_export.csv" + + return Response( + content=csv_content, + media_type="text/csv; charset=utf-8", + headers={ + "Content-Disposition": f'attachment; filename="{filename}"', + }, + ) diff --git a/app/services/letzshop_export_service.py b/app/services/letzshop_export_service.py new file mode 100644 index 00000000..9723aaa4 --- /dev/null +++ b/app/services/letzshop_export_service.py @@ -0,0 +1,280 @@ +# 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 typing import BinaryIO + +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() diff --git a/tests/integration/api/v1/admin/test_letzshop.py b/tests/integration/api/v1/admin/test_letzshop.py index 1377d5f2..082520b2 100644 --- a/tests/integration/api/v1/admin/test_letzshop.py +++ b/tests/integration/api/v1/admin/test_letzshop.py @@ -359,3 +359,197 @@ class TestAdminLetzshopAccessControl: response = client.get("/api/v1/admin/letzshop/vendors") assert response.status_code == 401 + + +@pytest.mark.integration +@pytest.mark.api +@pytest.mark.admin +@pytest.mark.letzshop +class TestAdminLetzshopExportAPI: + """Test admin Letzshop product export endpoints.""" + + def test_export_vendor_products_empty( + self, client, admin_headers, test_vendor + ): + """Test exporting products when vendor has no products.""" + response = client.get( + f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop", + headers=admin_headers, + ) + + assert response.status_code == 200 + assert response.headers["content-type"] == "text/csv; charset=utf-8" + # Should have header row at minimum + content = response.text + assert "id\ttitle\tdescription" in content + + def test_export_vendor_products_with_data( + self, client, db, admin_headers, test_vendor + ): + """Test exporting products with actual data.""" + from models.database.product import Product + from models.database.marketplace_product import MarketplaceProduct + from models.database.marketplace_product_translation import ( + MarketplaceProductTranslation, + ) + + # Create marketplace product + mp = MarketplaceProduct( + marketplace_product_id="EXPORT-TEST-001", + price="29.99", + price_numeric=29.99, + currency="EUR", + brand="TestBrand", + availability="in stock", + is_active=True, + ) + db.add(mp) + db.flush() + + # Add translation + translation = MarketplaceProductTranslation( + marketplace_product_id=mp.id, + language="en", + title="Export Test Product", + description="A product for testing exports", + ) + db.add(translation) + db.flush() + + # Create product linked to vendor + product = Product( + vendor_id=test_vendor.id, + vendor_sku="EXP-001", + marketplace_product_id=mp.id, + is_active=True, + ) + db.add(product) + db.commit() + + response = client.get( + f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop", + headers=admin_headers, + ) + + assert response.status_code == 200 + content = response.text + assert "EXP-001" in content + assert "Export Test Product" in content + assert "29.99 EUR" in content + + def test_export_vendor_products_french( + self, client, db, admin_headers, test_vendor + ): + """Test exporting products in French.""" + from models.database.product import Product + from models.database.marketplace_product import MarketplaceProduct + from models.database.marketplace_product_translation import ( + MarketplaceProductTranslation, + ) + + mp = MarketplaceProduct( + marketplace_product_id="EXPORT-FR-001", + price="19.99", + is_active=True, + ) + db.add(mp) + db.flush() + + # Add French translation + translation_fr = MarketplaceProductTranslation( + marketplace_product_id=mp.id, + language="fr", + title="Produit Test Export", + description="Un produit pour tester les exportations", + ) + db.add(translation_fr) + db.flush() + + product = Product( + vendor_id=test_vendor.id, + vendor_sku="EXP-FR-001", + marketplace_product_id=mp.id, + is_active=True, + ) + db.add(product) + db.commit() + + response = client.get( + f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop?language=fr", + headers=admin_headers, + ) + + assert response.status_code == 200 + content = response.text + assert "Produit Test Export" in content + + def test_export_vendor_products_include_inactive( + self, client, db, admin_headers, test_vendor + ): + """Test exporting including inactive products.""" + from models.database.product import Product + from models.database.marketplace_product import MarketplaceProduct + from models.database.marketplace_product_translation import ( + MarketplaceProductTranslation, + ) + + # Create inactive product + mp = MarketplaceProduct( + marketplace_product_id="EXPORT-INACTIVE-001", + price="5.99", + is_active=True, + ) + db.add(mp) + db.flush() + + translation = MarketplaceProductTranslation( + marketplace_product_id=mp.id, + language="en", + title="Inactive Product", + ) + db.add(translation) + db.flush() + + product = Product( + vendor_id=test_vendor.id, + vendor_sku="INACTIVE-001", + marketplace_product_id=mp.id, + is_active=False, # Inactive + ) + db.add(product) + db.commit() + + # Without include_inactive + response = client.get( + f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop", + headers=admin_headers, + ) + assert "INACTIVE-001" not in response.text + + # With include_inactive + response = client.get( + f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop?include_inactive=true", + headers=admin_headers, + ) + assert "INACTIVE-001" in response.text + + def test_export_vendor_products_by_vendor_code( + self, client, admin_headers, test_vendor + ): + """Test exporting products using vendor_code identifier.""" + response = client.get( + f"/api/v1/admin/vendors/{test_vendor.vendor_code}/export/letzshop", + headers=admin_headers, + ) + + assert response.status_code == 200 + assert response.headers["content-type"] == "text/csv; charset=utf-8" + + def test_export_vendor_not_found(self, client, admin_headers): + """Test exporting for non-existent vendor.""" + response = client.get( + "/api/v1/admin/vendors/999999/export/letzshop", + headers=admin_headers, + ) + + assert response.status_code == 404 diff --git a/tests/integration/api/v1/vendor/test_letzshop.py b/tests/integration/api/v1/vendor/test_letzshop.py index 8e0cb93e..3d974e52 100644 --- a/tests/integration/api/v1/vendor/test_letzshop.py +++ b/tests/integration/api/v1/vendor/test_letzshop.py @@ -497,3 +497,234 @@ class TestVendorLetzshopFulfillmentAPI: data = response.json() assert data["success"] is True assert data["tracking_number"] == "1Z999AA1" + + +@pytest.mark.integration +@pytest.mark.api +@pytest.mark.vendor +@pytest.mark.letzshop +class TestVendorLetzshopExportAPI: + """Test vendor Letzshop product export endpoints.""" + + def test_export_products_empty( + self, client, vendor_user_headers, test_vendor_with_vendor_user + ): + """Test exporting products when vendor has no products.""" + response = client.get( + "/api/v1/vendor/letzshop/export", headers=vendor_user_headers + ) + + assert response.status_code == 200 + assert response.headers["content-type"] == "text/csv; charset=utf-8" + # Should have header row at minimum + content = response.text + assert "id\ttitle\tdescription" in content + + def test_export_products_with_data( + self, client, db, vendor_user_headers, test_vendor_with_vendor_user + ): + """Test exporting products with actual data.""" + from models.database.product import Product + from models.database.marketplace_product import MarketplaceProduct + from models.database.marketplace_product_translation import ( + MarketplaceProductTranslation, + ) + + # Create marketplace product + mp = MarketplaceProduct( + marketplace_product_id="VENDOR-EXPORT-001", + price="49.99", + price_numeric=49.99, + currency="EUR", + brand="VendorBrand", + availability="in stock", + is_active=True, + ) + db.add(mp) + db.flush() + + # Add translation + translation = MarketplaceProductTranslation( + marketplace_product_id=mp.id, + language="en", + title="Vendor Export Product", + description="A product for testing vendor exports", + ) + db.add(translation) + db.flush() + + # Create product linked to vendor + product = Product( + vendor_id=test_vendor_with_vendor_user.id, + vendor_sku="VEXP-001", + marketplace_product_id=mp.id, + is_active=True, + ) + db.add(product) + db.commit() + + response = client.get( + "/api/v1/vendor/letzshop/export", headers=vendor_user_headers + ) + + assert response.status_code == 200 + content = response.text + assert "VEXP-001" in content + assert "Vendor Export Product" in content + assert "49.99 EUR" in content + + def test_export_products_french_language( + self, client, db, vendor_user_headers, test_vendor_with_vendor_user + ): + """Test exporting products in French.""" + from models.database.product import Product + from models.database.marketplace_product import MarketplaceProduct + from models.database.marketplace_product_translation import ( + MarketplaceProductTranslation, + ) + + mp = MarketplaceProduct( + marketplace_product_id="VENDOR-FR-001", + price="39.99", + is_active=True, + ) + db.add(mp) + db.flush() + + # Add French translation + translation_fr = MarketplaceProductTranslation( + marketplace_product_id=mp.id, + language="fr", + title="Produit Vendeur Export", + description="Un produit pour tester", + ) + db.add(translation_fr) + db.flush() + + product = Product( + vendor_id=test_vendor_with_vendor_user.id, + vendor_sku="VFR-001", + marketplace_product_id=mp.id, + is_active=True, + ) + db.add(product) + db.commit() + + response = client.get( + "/api/v1/vendor/letzshop/export?language=fr", + headers=vendor_user_headers, + ) + + assert response.status_code == 200 + content = response.text + assert "Produit Vendeur Export" in content + + def test_export_products_german_language( + self, client, db, vendor_user_headers, test_vendor_with_vendor_user + ): + """Test exporting products in German.""" + from models.database.product import Product + from models.database.marketplace_product import MarketplaceProduct + from models.database.marketplace_product_translation import ( + MarketplaceProductTranslation, + ) + + mp = MarketplaceProduct( + marketplace_product_id="VENDOR-DE-001", + price="59.99", + is_active=True, + ) + db.add(mp) + db.flush() + + translation_de = MarketplaceProductTranslation( + marketplace_product_id=mp.id, + language="de", + title="Deutsches Exportprodukt", + ) + db.add(translation_de) + db.flush() + + product = Product( + vendor_id=test_vendor_with_vendor_user.id, + vendor_sku="VDE-001", + marketplace_product_id=mp.id, + is_active=True, + ) + db.add(product) + db.commit() + + response = client.get( + "/api/v1/vendor/letzshop/export?language=de", + headers=vendor_user_headers, + ) + + assert response.status_code == 200 + content = response.text + assert "Deutsches Exportprodukt" in content + + def test_export_products_include_inactive( + self, client, db, vendor_user_headers, test_vendor_with_vendor_user + ): + """Test exporting including inactive products.""" + from models.database.product import Product + from models.database.marketplace_product import MarketplaceProduct + from models.database.marketplace_product_translation import ( + MarketplaceProductTranslation, + ) + + mp = MarketplaceProduct( + marketplace_product_id="VENDOR-INACTIVE-001", + price="9.99", + is_active=True, + ) + db.add(mp) + db.flush() + + translation = MarketplaceProductTranslation( + marketplace_product_id=mp.id, + language="en", + title="Inactive Vendor Product", + ) + db.add(translation) + db.flush() + + product = Product( + vendor_id=test_vendor_with_vendor_user.id, + vendor_sku="VINACTIVE-001", + marketplace_product_id=mp.id, + is_active=False, # Inactive + ) + db.add(product) + db.commit() + + # Without include_inactive + response = client.get( + "/api/v1/vendor/letzshop/export", + headers=vendor_user_headers, + ) + assert "VINACTIVE-001" not in response.text + + # With include_inactive + response = client.get( + "/api/v1/vendor/letzshop/export?include_inactive=true", + headers=vendor_user_headers, + ) + assert "VINACTIVE-001" in response.text + + def test_export_unauthenticated(self, client): + """Test that unauthenticated requests are rejected.""" + response = client.get("/api/v1/vendor/letzshop/export") + assert response.status_code == 401 + + def test_export_filename_uses_vendor_code( + self, client, vendor_user_headers, test_vendor_with_vendor_user + ): + """Test that export filename uses vendor code.""" + response = client.get( + "/api/v1/vendor/letzshop/export", headers=vendor_user_headers + ) + + assert response.status_code == 200 + content_disposition = response.headers.get("content-disposition", "") + assert "letzshop_export.csv" in content_disposition