feat: add export job tracking to Letzshop jobs tab
- Export operations now create LetzshopSyncLog entries with type 'product_export' - Added log_export method to LetzshopExportService (follows architecture rules) - Updated list_letzshop_jobs to include export jobs - Added export filter option back to jobs dropdown - Export jobs display with blue badge and cloud-upload icon 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -708,6 +708,34 @@ class LetzshopOrderService:
|
||||
}
|
||||
)
|
||||
|
||||
# Product exports from letzshop_sync_logs
|
||||
if job_type in (None, "export"):
|
||||
export_query = self.db.query(LetzshopSyncLog).filter(
|
||||
LetzshopSyncLog.vendor_id == vendor_id,
|
||||
LetzshopSyncLog.operation_type == "product_export",
|
||||
)
|
||||
if status:
|
||||
export_query = export_query.filter(LetzshopSyncLog.status == status)
|
||||
|
||||
export_logs = export_query.order_by(
|
||||
LetzshopSyncLog.created_at.desc()
|
||||
).all()
|
||||
|
||||
for log in export_logs:
|
||||
jobs.append(
|
||||
{
|
||||
"id": log.id,
|
||||
"type": "export",
|
||||
"status": log.status,
|
||||
"created_at": log.created_at,
|
||||
"started_at": log.started_at,
|
||||
"completed_at": log.completed_at,
|
||||
"records_processed": log.records_processed or 0,
|
||||
"records_succeeded": log.records_succeeded or 0,
|
||||
"records_failed": log.records_failed or 0,
|
||||
}
|
||||
)
|
||||
|
||||
# Sort all jobs by created_at descending
|
||||
jobs.sort(key=lambda x: x["created_at"], reverse=True)
|
||||
|
||||
|
||||
@@ -8,9 +8,11 @@ Generates Google Shopping compatible CSV files for Letzshop marketplace.
|
||||
import csv
|
||||
import io
|
||||
import logging
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy.orm import Session, joinedload
|
||||
|
||||
from models.database.letzshop import LetzshopSyncLog
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.product import Product
|
||||
|
||||
@@ -279,6 +281,53 @@ class LetzshopExportService:
|
||||
"atalanda:delivery_method": "",
|
||||
}
|
||||
|
||||
def log_export(
|
||||
self,
|
||||
db: Session,
|
||||
vendor_id: int,
|
||||
started_at: datetime,
|
||||
completed_at: datetime,
|
||||
records_processed: int,
|
||||
languages_succeeded: int,
|
||||
languages_failed: int,
|
||||
triggered_by: str,
|
||||
error_details: dict | None = None,
|
||||
) -> LetzshopSyncLog:
|
||||
"""
|
||||
Log an export operation to the sync log.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
vendor_id: Vendor ID
|
||||
started_at: When the export started
|
||||
completed_at: When the export completed
|
||||
records_processed: Number of products exported
|
||||
languages_succeeded: Number of languages successfully exported
|
||||
languages_failed: Number of languages that failed
|
||||
triggered_by: Who triggered the export (e.g., "admin:123")
|
||||
error_details: Optional error details if any failures
|
||||
|
||||
Returns:
|
||||
Created LetzshopSyncLog entry
|
||||
"""
|
||||
sync_log = LetzshopSyncLog(
|
||||
vendor_id=vendor_id,
|
||||
operation_type="product_export",
|
||||
direction="outbound",
|
||||
status="completed" if languages_failed == 0 else "partial",
|
||||
records_processed=records_processed,
|
||||
records_succeeded=languages_succeeded,
|
||||
records_failed=languages_failed,
|
||||
started_at=started_at,
|
||||
completed_at=completed_at,
|
||||
duration_seconds=int((completed_at - started_at).total_seconds()),
|
||||
triggered_by=triggered_by,
|
||||
error_details=error_details,
|
||||
)
|
||||
db.add(sync_log)
|
||||
db.flush()
|
||||
return sync_log
|
||||
|
||||
|
||||
# Singleton instance
|
||||
letzshop_export_service = LetzshopExportService()
|
||||
|
||||
Reference in New Issue
Block a user