from __future__ import annotations import os from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from sqlalchemy import select from sqlalchemy.orm import Session from db.channel_exports import CHANNEL_CODES, build_channel_product_payloads, resolve_pipeline_channel_code from db.magento_repositories import SqlAlchemyMagentoSyncQueueRepository from db.models import ChannelJob, ChannelSchedule, MagentoSyncJob, MagentoSyncQueue TERMINAL_STATUSES = {"completed", "failed", "cancelled", "not_implemented"} # Job types that coalesce on enqueue: newer queued work replaces older queued work. SUPERSEDE_ON_ENQUEUE_JOB_TYPES = frozenset({"push", "push_images"}) # Handler result statuses normalized to channel_job statuses. RESULT_STATUS_MAP = { "ok": "completed", "success": "completed", "disabled": "failed", } def enqueue_shopify_product_push_job( session: Session, *, channel_connection_id: int, native_connection_id: int, channel_code: str, skus: Optional[List[str]] = None, limit: Optional[int] = None, force: bool = False, dry_run: bool = False, shop_code: Optional[str] = None, notes: Optional[str] = None, supersede_queued: bool = True, ) -> ChannelJob: """Enqueue a Shopify product create/update job (executed by channel_job_worker).""" options: dict[str, Any] = { "skus": skus, "force": force, "connection_id": channel_connection_id, } if limit is not None: options["limit"] = limit if shop_code: options["shop_code"] = shop_code return enqueue_channel_job( session, channel_connection_id=channel_connection_id, channel_type="shopify", native_connection_id=native_connection_id, channel_code=channel_code, job_type="push", dry_run=dry_run, mode="full" if force else "incremental", notes=notes, options=options, supersede_queued=supersede_queued, ) def enqueue_shopify_product_push_jobs_batched( session: Session, *, channel_connection_id: int, native_connection_id: int, channel_code: str, skus: List[str], batch_size: int = 200, force: bool = False, dry_run: bool = False, shop_code: Optional[str] = None, notes: Optional[str] = None, supersede_queued: bool = True, ) -> List[ChannelJob]: """Enqueue one or more Shopify push jobs, splitting large SKU lists into batches.""" if not skus: return [] if supersede_queued: cancel_queued_channel_jobs( session, channel_type="shopify", native_connection_id=native_connection_id, job_type="push", channel_connection_id=channel_connection_id, reason=notes or "superseded by newer shopify push", ) batch_size = max(1, int(batch_size)) if len(skus) <= batch_size: return [ enqueue_shopify_product_push_job( session, channel_connection_id=channel_connection_id, native_connection_id=native_connection_id, channel_code=channel_code, skus=skus, force=force, dry_run=dry_run, shop_code=shop_code, notes=notes, supersede_queued=False, ) ] batch_count = (len(skus) + batch_size - 1) // batch_size jobs: List[ChannelJob] = [] for index in range(batch_count): batch = skus[index * batch_size : (index + 1) * batch_size] batch_notes = f"{notes or 'shopify_push'} batch {index + 1}/{batch_count}" jobs.append( enqueue_shopify_product_push_job( session, channel_connection_id=channel_connection_id, native_connection_id=native_connection_id, channel_code=channel_code, skus=batch, force=force, dry_run=dry_run, shop_code=shop_code, notes=batch_notes, supersede_queued=False, ) ) return jobs def enqueue_collection_bootstrap_job( session: Session, *, channel_connection_id: int, native_connection_id: int, channel_code: str, channel_type: str, dry_run: bool = False, notes: Optional[str] = None, ) -> ChannelJob: """Enqueue collection landing field bootstrap (metafields / Magento attribute discovery).""" return enqueue_channel_job( session, channel_connection_id=channel_connection_id, channel_type=channel_type, native_connection_id=native_connection_id, channel_code=channel_code, job_type="collection_bootstrap", dry_run=dry_run, mode="full", notes=notes or "collection_bootstrap", ) def enqueue_collection_push_job( session: Session, *, channel_connection_id: int, native_connection_id: int, channel_code: str, channel_type: str, path_slug: Optional[str] = None, category_l1: Optional[str] = None, only_with_landing_content: bool = True, dry_run: bool = False, provision_missing_remote: bool = False, apply_listing_paths: bool = True, sync_products: bool = True, sync_shopping_icons: bool = True, limit: int = 200, bootstrap_first: bool = True, notes: Optional[str] = None, ) -> ChannelJob: """Enqueue collection landing content push (optionally bootstrap remote fields first).""" options: dict[str, Any] = { "path_slug": path_slug, "category_l1": category_l1, "only_with_landing_content": only_with_landing_content, "provision_missing_remote": provision_missing_remote, "apply_listing_paths": apply_listing_paths, "sync_products": sync_products, "sync_shopping_icons": sync_shopping_icons, "limit": limit, "bootstrap_first": bootstrap_first, } return enqueue_channel_job( session, channel_connection_id=channel_connection_id, channel_type=channel_type, native_connection_id=native_connection_id, channel_code=channel_code, job_type="collection_push", dry_run=dry_run, mode="full", notes=notes or "collection_push", options=options, ) def enqueue_collection_reconcile_job( session: Session, *, channel_connection_id: int, native_connection_id: int, channel_code: str, channel_type: str, path_slug: str, category_l1: Optional[str] = None, collection: Optional[str] = None, cleanup_other_paths: bool = True, sync_magento: bool = False, sync_shopify: bool = False, magento_native_connection_id: Optional[int] = None, shopify_native_connection_id: Optional[int] = None, dry_run: bool = False, notes: Optional[str] = None, ) -> ChannelJob: """Enqueue collection SKU reconcile (membership cleanup + optional channel sync).""" options: dict[str, Any] = { "path_slug": path_slug, "category_l1": category_l1, "collection": collection, "cleanup_other_paths": cleanup_other_paths, "sync_magento": sync_magento, "sync_shopify": sync_shopify, "magento_native_connection_id": magento_native_connection_id, "shopify_native_connection_id": shopify_native_connection_id, } return enqueue_channel_job( session, channel_connection_id=channel_connection_id, channel_type=channel_type, native_connection_id=native_connection_id, channel_code=channel_code, job_type="collection_reconcile", dry_run=dry_run, mode="full", notes=notes or f"collection_reconcile:{path_slug}", options=options, ) def enqueue_collection_reconcile_all_job( session: Session, *, channel_connection_id: int, native_connection_id: int, channel_code: str, channel_type: str, category_l1: Optional[str] = None, dry_run: bool = False, notes: Optional[str] = None, ) -> ChannelJob: """Enqueue reconcile-all for every taxonomy-backed collection.""" options: dict[str, Any] = {"category_l1": category_l1} return enqueue_channel_job( session, channel_connection_id=channel_connection_id, channel_type=channel_type, native_connection_id=native_connection_id, channel_code=channel_code, job_type="collection_reconcile_all", dry_run=dry_run, mode="full", notes=notes or "collection_reconcile_all", options=options, ) def enqueue_start_shopping_magento_product_sync( session: Session, *, channel_connection_id: int, native_connection_id: int, collection_path_slugs: Optional[List[str]] = None, category_l1: Optional[str] = None, dry_run: bool = False, limit: int = 200, notes: Optional[str] = None, supersede_queued: bool = False, ) -> dict[str, Any]: from channel.url_canonical import normalize_plp_path from db.collection_channel_export import export_collections_for_channel from db.collection_landing_pages import get_collection_landing_page from db.collection_sku_mapping import resolve_collection_skus from magento.sku_policy import filter_magento_push_skus normalized_slugs = [ normalize_plp_path(str(path or "")) for path in (collection_path_slugs or []) if normalize_plp_path(str(path or "")) ] if normalized_slugs: rows: List[dict[str, Any]] = [] missing: List[str] = [] for slug in normalized_slugs: row = get_collection_landing_page(session, slug) if not row: missing.append(slug) continue rows.append(row) if missing: raise ValueError(f"Collection not found: {', '.join(missing[:10])}") else: payload = export_collections_for_channel( session, channel_code="magento", category_l1=category_l1, connection_id=channel_connection_id, include_inferred=False, include_skus=False, limit=limit, ) rows = list(payload.get("collections") or []) resolved_slugs: List[str] = [] master_sku_set: set[str] = set() sku_counts: dict[str, int] = {} for row in rows: slug = normalize_plp_path(row.get("path_slug") or "") if not slug: continue resolved = resolve_collection_skus( session, path_slug=slug, category_l1=row.get("category_l1"), collection=row.get("collection"), sku_prefixes=row.get("sku_prefixes"), ) master_skus = [ str(item.get("master_sku") or "").strip() for item in (resolved.get("skus") or []) if str(item.get("master_sku") or "").strip() ] if not master_skus: continue resolved_slugs.append(slug) sku_counts[slug] = len(master_skus) master_sku_set.update(master_skus) master_skus = filter_magento_push_skus(sorted(master_sku_set)) if not master_skus: return { "status": "skipped", "reason": "no_pushable_skus", "collection_count": len(resolved_slugs), "collection_path_slugs": resolved_slugs, "master_sku_count": 0, "queue_id": None, } label = f"start-shopping-products-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}" queue_options: dict[str, Any] = { "dry_run": dry_run, "use_master_catalog": True, "products_only": True, "force_upsert": True, "force_upsert_skus": list(master_skus), "limit_skus": list(master_skus), "force_relations": True, "expand_relations": True, "selected_attribute_codes": [ "shopping_collection", "shopping_l1", "shopping_l2", ], "sync_options_from_master": False, "run_pull_first": False, "run_pull_after": True, } queue_id = SqlAlchemyMagentoSyncQueueRepository(session).enqueue( native_connection_id, label, options=queue_options, supersede_queued=supersede_queued, ) return { "status": "queued", "backend": "magento_sync_queue", "queue_id": queue_id, "label": label, "collection_count": len(resolved_slugs), "collection_path_slugs": resolved_slugs, "master_sku_count": len(master_skus), "sku_counts": sku_counts, "notes": notes, } def enqueue_channel_job( session: Session, *, channel_connection_id: int, channel_type: str, native_connection_id: int, channel_code: str, job_type: str, dry_run: bool = True, mode: str = "incremental", schedule_id: Optional[int] = None, notes: Optional[str] = None, options: Optional[dict[str, Any]] = None, supersede_queued: Optional[bool] = None, ) -> ChannelJob: should_supersede = ( supersede_queued if supersede_queued is not None else job_type in SUPERSEDE_ON_ENQUEUE_JOB_TYPES ) if should_supersede: cancel_queued_channel_jobs( session, channel_type=channel_type, native_connection_id=native_connection_id, job_type=job_type, channel_connection_id=channel_connection_id, reason=notes or f"superseded by newer {job_type} job", ) row = ChannelJob( schedule_id=schedule_id, channel_connection_id=channel_connection_id, channel_type=channel_type, native_connection_id=native_connection_id, channel_code=channel_code, job_type=job_type, status="queued", dry_run=dry_run, mode=mode, notes=notes, result={"_options": options or {}} if options else None, ) session.add(row) session.flush() return row def cancel_queued_channel_jobs( session: Session, *, channel_type: Optional[str] = None, native_connection_id: Optional[int] = None, job_type: Optional[str] = None, channel_connection_id: Optional[int] = None, reason: str = "cancelled", ) -> int: """Mark queued channel jobs as cancelled. Running jobs are left untouched.""" stmt = select(ChannelJob).where(ChannelJob.status == "queued") if channel_type: stmt = stmt.where(ChannelJob.channel_type == channel_type) if native_connection_id is not None: stmt = stmt.where(ChannelJob.native_connection_id == native_connection_id) if job_type: stmt = stmt.where(ChannelJob.job_type == job_type) if channel_connection_id is not None: stmt = stmt.where(ChannelJob.channel_connection_id == channel_connection_id) rows = list(session.scalars(stmt).all()) now = datetime.utcnow() message = str(reason or "cancelled")[:2000] for row in rows: row.status = "cancelled" row.finished_at = now row.notes = message if rows: session.flush() return len(rows) def claim_next_channel_job(session: Session, *, worker_id: Optional[str] = None) -> Optional[ChannelJob]: worker = worker_id or f"channel-worker-{os.getpid()}" cutoff = datetime.utcnow() - timedelta(minutes=120) stale_rows = session.scalars( select(ChannelJob) .where(ChannelJob.status == "running") .where(ChannelJob.locked_at < cutoff) .with_for_update(skip_locked=True) ).all() for row in stale_rows: if row.attempts >= row.max_attempts: row.status = "failed" row.finished_at = datetime.utcnow() row.notes = "Exhausted attempts after stale running reset" else: row.status = "queued" row.locked_at = None row.locked_by = None row = session.scalars( select(ChannelJob) .where(ChannelJob.status == "queued") .where(ChannelJob.attempts < ChannelJob.max_attempts) .order_by(ChannelJob.requested_at, ChannelJob.id) .limit(1) .with_for_update(skip_locked=True) ).first() if row is None: return None row.status = "running" row.started_at = datetime.utcnow() row.locked_at = datetime.utcnow() row.locked_by = worker row.attempts = (row.attempts or 0) + 1 session.flush() return row def run_channel_job(session: Session, job: ChannelJob) -> dict[str, Any]: """Execute a claimed job and persist the outcome. Never raises for job-level failures: on error the session is rolled back first so the failure can be recorded in a clean transaction (the claim itself must already be committed by the caller). """ job_id = job.id try: result = _execute_channel_job(session, job) except Exception as exc: session.rollback() job = session.get(ChannelJob, job_id) result = {"status": "failed", "error": str(exc)} if job is None: return result job.status = "failed" job.error_count = 1 job.notes = str(exc) job.result = result job.finished_at = datetime.utcnow() job.locked_at = None job.locked_by = None _update_schedule_after_job(session, job) session.flush() return result outcome = summarize_result(result) job.status = outcome["status"] job.result = result job.total_count = outcome["total_count"] job.success_count = outcome["success_count"] job.error_count = outcome["error_count"] job.notes = outcome["notes"] or job.notes if outcome["status"] != "delegated": job.finished_at = datetime.utcnow() job.locked_at = None job.locked_by = None _update_schedule_after_job(session, job) session.flush() return result def summarize_result(result: dict[str, Any]) -> dict[str, Any]: """Pure normalization of a handler result dict into channel_job fields.""" raw_status = str(result.get("status") or "completed") status = RESULT_STATUS_MAP.get(raw_status, raw_status) total = _first_int(result, "total_count", "total", "count") success = _first_int(result, "success_count", "pushed") if success is None: success = (total or 0) if status == "completed" else 0 errors = _first_int(result, "error_count", "failed") if errors is None: errors = 1 if status == "failed" else 0 return { "status": status, "total_count": total, "success_count": success, "error_count": errors, "notes": result.get("message") or result.get("detail") or result.get("notes"), } def _shopify_shop_code(session: Session, job: ChannelJob) -> Optional[str]: """Resolve Shopify shop_code from the native connection row (not the pipeline channel code).""" from db.models import ShopifyConnection row = session.get(ShopifyConnection, job.native_connection_id) if row is not None: return row.shop_code legacy = str(job.channel_code or "").strip().lower() if legacy and legacy not in CHANNEL_CODES: return legacy return None def _first_int(result: dict[str, Any], *keys: str) -> Optional[int]: for key in keys: value = result.get(key) if value is not None: try: return int(value) except (TypeError, ValueError): continue return None def _job_options(job: ChannelJob) -> dict[str, Any]: return dict((job.result or {}).get("_options") or {}) def _execute_collection_bootstrap(session: Session, job: ChannelJob) -> dict[str, Any]: from db.collection_landing_push import bootstrap_collection_landing_channel result = bootstrap_collection_landing_channel( session, channel_code=job.channel_type, connection_id=job.native_connection_id, dry_run=job.dry_run, ) errors = result.get("errors") or [] status = "failed" if errors and not (result.get("created") or result.get("skipped")) else "completed" return {"backend": "collection_landing_bootstrap", "status": status, **result} def _execute_collection_push(session: Session, job: ChannelJob) -> dict[str, Any]: from db.collection_landing_push import bootstrap_collection_landing_channel, push_collection_landing_channel options = _job_options(job) bootstrap_result = None if bool(options.get("bootstrap_first", False)) and not job.dry_run: bootstrap_result = bootstrap_collection_landing_channel( session, channel_code=job.channel_type, connection_id=job.native_connection_id, dry_run=False, ) bootstrap_errors = bootstrap_result.get("errors") or [] if bootstrap_errors and not (bootstrap_result.get("created") or bootstrap_result.get("skipped")): return { "backend": "collection_landing_push", "status": "failed", "bootstrap": bootstrap_result, "error": "Collection bootstrap failed before push", } result = push_collection_landing_channel( session, channel_code=job.channel_type, connection_id=job.native_connection_id, path_slug=options.get("path_slug"), category_l1=options.get("category_l1"), only_with_landing_content=bool(options.get("only_with_landing_content", True)), dry_run=job.dry_run, provision_missing_remote=bool(options.get("provision_missing_remote", False)), apply_listing_paths=bool(options.get("apply_listing_paths", True)), sync_products=bool(options.get("sync_products", True)), sync_shopping_icons=bool(options.get("sync_shopping_icons", True)), limit=int(options.get("limit") or 200), ) if bootstrap_result is not None: result["bootstrap"] = bootstrap_result errors = result.get("errors") or [] pushed = int(result.get("pushed") or 0) status = "failed" if errors and pushed == 0 else "completed" return {"backend": "collection_landing_push", "status": status, **result} def _execute_collection_reconcile(session: Session, job: ChannelJob) -> dict[str, Any]: from db.collection_sku_mapping import ( reconcile_collection_sku_assignments, resolve_collection_skus, sync_collection_reconcile_channels, ) options = _job_options(job) path_slug = str(options.get("path_slug") or "").strip() if not path_slug: return {"backend": "collection_reconcile", "status": "failed", "error": "path_slug is required"} on_progress = _channel_job_progress_callback(job.id) on_progress( { "stage": "local_reconcile", "completed": 0, "notes": f"collection_reconcile:{path_slug} — assigning local listing paths", } ) result = reconcile_collection_sku_assignments( session, path_slug=path_slug, category_l1=options.get("category_l1"), collection=options.get("collection"), dry_run=job.dry_run, cleanup_other_paths=bool(options.get("cleanup_other_paths", True)), ) sku_count = int(result.get("sku_count") or result.get("assignment_count") or 0) on_progress( { "stage": "local_reconcile", "completed": sku_count, "total": sku_count, "notes": f"Local assignments complete ({sku_count} SKU(s))", } ) channel_sync: dict[str, Any] = {} if not job.dry_run and (options.get("sync_magento") or options.get("sync_shopify")): resolved = resolve_collection_skus( session, path_slug=result.get("path_slug") or path_slug, category_l1=result.get("category_l1"), collection=result.get("collection"), ) master_skus = [item["master_sku"] for item in (resolved.get("skus") or [])] channel_sync = sync_collection_reconcile_channels( session, path_slug=result.get("path_slug") or path_slug, master_skus=master_skus, magento_connection_id=options.get("magento_native_connection_id"), shopify_connection_id=options.get("shopify_native_connection_id"), sync_magento=bool(options.get("sync_magento")), sync_shopify=bool(options.get("sync_shopify")), on_progress=on_progress, ) if channel_sync: result["channel_sync"] = channel_sync shopify_result = channel_sync.get("shopify") or {} magento_incomplete = ( shopify_result.get("skipped") is True and shopify_result.get("reason") == "magento_sync_incomplete" ) job_status = "failed" if magento_incomplete else "completed" return { "backend": "collection_reconcile", "status": job_status, "total_count": sku_count, "success_count": sku_count if job_status == "completed" else 0, "error_count": 1 if magento_incomplete else 0, **result, } def _execute_collection_reconcile_all(session: Session, job: ChannelJob) -> dict[str, Any]: from db.collection_sku_mapping import reconcile_all_collection_assignments options = _job_options(job) result = reconcile_all_collection_assignments( session, category_l1=options.get("category_l1"), dry_run=job.dry_run, ) return {"backend": "collection_reconcile_all", "status": "completed", **result} def report_channel_image_push_outcome( session: Session, queue_id: int, *, assigned_sku_count: int = 0, skus_with_images: int = 0, planned_count: int = 0, success_count: int = 0, error_count: int = 0, message: str, extra: Optional[dict[str, Any]] = None, ) -> None: """Copy Magento image-push stats onto the delegated channel_job for dashboard polling.""" from sqlalchemy import select job = session.scalar( select(ChannelJob).where(ChannelJob.backend_queue_id == queue_id).limit(1) ) if job is None: return job.total_count = skus_with_images or assigned_sku_count or planned_count job.success_count = success_count job.error_count = error_count job.notes = message merged = dict(job.result or {}) completed = success_count + error_count merged.update( { "mode": "images_only", "assigned_sku_count": assigned_sku_count, "skus_with_images": skus_with_images, "planned_count": planned_count, "success_count": success_count, "error_count": error_count, "summary": message, "progress": { "total": planned_count or skus_with_images or assigned_sku_count, "completed": completed, "success_count": success_count, "error_count": error_count, "percent": round(100 * completed / planned_count, 1) if planned_count else None, }, **(extra or {}), } ) job.result = merged session.add(job) def _channel_job_progress_callback(job_id: int): """Persist in-flight counts on a separate DB session so dashboard polling works.""" def on_progress(stats: dict[str, Any]) -> None: _persist_channel_job_progress(job_id, stats) return on_progress def _persist_channel_job_progress(job_id: int, stats: dict[str, Any]) -> None: from datetime import datetime from db.session import get_session with get_session() as prog_session: row = prog_session.get(ChannelJob, job_id) if row is None: return if row.status == "running": row.locked_at = datetime.utcnow() total = stats.get("total") completed = stats.get("completed") if total is not None: row.total_count = int(total) or row.total_count if completed is not None: row.success_count = int(completed) if stats.get("pushed") is not None: row.success_count = int(stats.get("pushed") or 0) if stats.get("failed") is not None: row.error_count = int(stats.get("failed") or 0) if stats.get("notes"): row.notes = str(stats["notes"]) prior = dict(row.result or {}) total_count = row.total_count completed_count = int(completed if completed is not None else row.success_count or 0) row.result = { **prior, "_options": prior.get("_options") or _job_options(row), "stage": stats.get("stage") or prior.get("stage"), "progress": { "total": total_count, "completed": completed_count, "success_count": row.success_count, "error_count": row.error_count, "percent": round(100 * completed_count / total_count, 1) if total_count else None, }, } prog_session.commit() def _execute_magento_push_images(session: Session, job: ChannelJob) -> dict[str, Any]: options = _job_options(job) label = f"channel-job-{job.id}-images-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}" queue_options: dict[str, Any] = { "dry_run": job.dry_run, "use_master_catalog": True, "images_only": True, } limit_skus = options.get("limit_skus") if limit_skus: queue_options["limit_skus"] = limit_skus queue_id = SqlAlchemyMagentoSyncQueueRepository(session).enqueue( job.native_connection_id, label, options=queue_options, supersede_queued=True, ) job.backend_queue_id = queue_id return { "status": "delegated", "backend": "magento_sync_queue", "backend_status": "queued", "queue_id": queue_id, "label": label, "mode": "images_only", } def enqueue_magento_media_cleanup_job( session: Session, *, channel_connection_id: int, native_connection_id: int, channel_code: str, skus: Optional[List[str]] = None, all_assigned: bool = False, batch_size: int = 250, offset: int = 0, limit: Optional[int] = None, purge_unmapped: bool = True, dry_run: bool = False, notes: Optional[str] = None, supersede_queued: bool = True, ) -> ChannelJob: """Enqueue Magento orphan-only gallery cleanup (batched; no wipe / no re-upload).""" options: dict[str, Any] = { "skus": skus, "all_assigned": bool(all_assigned), "batch_size": max(1, int(batch_size or 250)), "offset": max(0, int(offset or 0)), "limit": limit, "purge_unmapped": bool(purge_unmapped), "connection_id": native_connection_id, } return enqueue_channel_job( session, channel_connection_id=channel_connection_id, channel_type="magento", native_connection_id=native_connection_id, channel_code=channel_code, job_type="media_cleanup", dry_run=dry_run, mode="orphans_only", notes=notes or "magento media orphan cleanup", options=options, supersede_queued=supersede_queued, ) def _execute_magento_media_cleanup(session: Session, job: ChannelJob) -> dict[str, Any]: from db.magento_repositories import ( SqlAlchemyMagentoCatalogStateRepository, SqlAlchemyMagentoConnectionRepository, SqlAlchemyMagentoMediaMapRepository, ) from magento.magento_api import MagentoRestClient from magento.media_cleanup import ( cleanup_orphaned_media_batch, resolve_orphan_cleanup_skus, ) from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs options = _job_options(job) connection_id = int(options.get("connection_id") or job.native_connection_id) conn = SqlAlchemyMagentoConnectionRepository(session).get_for_sync(connection_id) if not conn: return {"status": "failed", "error": f"Magento connection {connection_id} not found"} skus = resolve_orphan_cleanup_skus( session, skus=options.get("skus"), all_assigned=bool(options.get("all_assigned")), offset=int(options.get("offset") or 0), limit=options.get("limit"), ) if not skus: return { "status": "completed", "backend": "magento_media_cleanup", "mode": "orphans_only", "sku_count": 0, "total_count": 0, "success_count": 0, "error_count": 0, "message": "No SKUs matched for Magento orphan media cleanup.", } oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) api = MagentoRestClient(oauth) summary = cleanup_orphaned_media_batch( connection_id=connection_id, skus=skus, api=api, media_repo=SqlAlchemyMagentoMediaMapRepository(session), catalog_repo=SqlAlchemyMagentoCatalogStateRepository(session), session=session, dry_run=bool(job.dry_run), purge_unmapped=bool(options.get("purge_unmapped", True)), batch_size=int(options.get("batch_size") or 250), on_progress=_channel_job_progress_callback(job.id), include_sku_results=bool(job.dry_run) and len(skus) <= 100, ) raw_status = str(summary.get("status") or "completed") # Job finished even when some SKUs were partial — keep counts in result. if raw_status in {"ok", "dry_run", "partial"}: status = "completed" elif raw_status == "failed": status = "failed" else: status = "completed" return { "backend": "magento_media_cleanup", "mode": "orphans_only", **summary, "status": status, "cleanup_status": raw_status, } def _execute_shopify_push_images(session: Session, job: ChannelJob) -> dict[str, Any]: from shopify.media_sync import push_product_images options = _job_options(job) summary = push_product_images( session, dry_run=job.dry_run, limit=options.get("limit"), skus=options.get("skus"), shop_code=options.get("shop_code") or _shopify_shop_code(session, job), connection_id=options.get("connection_id") or job.channel_connection_id, force=bool(options.get("force")), on_progress=_channel_job_progress_callback(job.id), ) if summary.get("status") == "disabled": return {"status": "failed", "detail": summary.get("detail"), **summary} return {"backend": "shopify_image_push", "mode": "images_only", **summary} def _maybe_auto_link_after_pull(session: Session, job: ChannelJob, result: dict[str, Any]) -> dict[str, Any]: options = _job_options(job) if options.get("auto_link_skus") is False: return result if job.dry_run or result.get("status") == "failed": return result if job.job_type != "pull": return result from db.channel_sku_mapping import auto_link_skus_after_pull link_result = auto_link_skus_after_pull( session, channel_type=job.channel_type, channel_code=job.channel_code, compat_connection_id=job.channel_connection_id, native_connection_id=job.native_connection_id, ) result["auto_link"] = link_result return result def _execute_channel_job(session: Session, job: ChannelJob) -> dict[str, Any]: if job.channel_type == "magento" and job.job_type == "attribute_pull": from app.jobs.magento_baseline_sync import run_baseline_sync result = run_baseline_sync(job.native_connection_id, force=True, session=session) if result.get("status") == "failed": return { "backend": "magento_baseline_sync", "status": "failed", "error": result.get("error", "Magento attribute refresh failed"), **result, } counts = result.get("counts", {}) return { "backend": "magento_baseline_sync", "status": "completed", "refreshed_attributes": counts.get("attributes", 0), "refreshed_options": counts.get("options", 0), "total_count": (counts.get("attributes") or 0) + (counts.get("options") or 0), "message": ( f"Magento attributes refreshed: {counts.get('attributes', 0)} attributes, " f"{counts.get('options', 0)} options." ), "baseline": result, } if job.channel_type == "shopify" and job.job_type == "attribute_pull": from app.jobs.shopify_attribute_sync import run_shopify_attribute_sync result = run_shopify_attribute_sync( connection_id=job.native_connection_id, shop_code=_shopify_shop_code(session, job), session=session, ) if result.get("status") != "success": return { "backend": "shopify_attribute_sync", "status": "failed", "error": result.get("error", "Shopify attribute refresh failed"), **result, } total = ( result.get("attributes") or result.get("attribute_count") or result.get("metafields") or result.get("metafield_count") or result.get("count") ) return { "backend": "shopify_attribute_sync", "status": "completed", "total_count": total, "message": "Shopify attributes refreshed.", **result, } if job.job_type == "sku_link": from db.channel_sku_mapping import ( fetch_shopify_remote_catalog, link_mappings_from_remote, reconcile_connection_id_for_channel, ) options = _job_options(job) pipeline_code = resolve_pipeline_channel_code(job.channel_code, channel_type=job.channel_type) reconcile_conn_id = reconcile_connection_id_for_channel( pipeline_code, compat_connection_id=job.channel_connection_id, native_connection_id=job.native_connection_id, ) remote_catalog = None remote_source = None if pipeline_code == "shopify": remote_catalog = fetch_shopify_remote_catalog( session, connection_id=job.channel_connection_id, native_connection_id=job.native_connection_id, ) remote_source = "shopify_admin_api" result = link_mappings_from_remote( session, pipeline_code, connection_id=reconcile_conn_id, only_assigned=bool(options.get("only_assigned", True)), remote_catalog=remote_catalog, remote_source=remote_source, dry_run=job.dry_run, ) linked = result.get("upserted") or result.get("linked") or result.get("applied") or 0 total = ( result.get("remote_sku_count") or result.get("suggestion_count") or result.get("unmapped_master_count") or linked ) message = ( f"Linked {linked} SKU mapping(s) from {result.get('remote_source') or remote_source or 'remote catalog'}." ) if linked == 0 and int(result.get("suggestion_count") or 0) == 0: unmapped = int(result.get("unmapped_master_count") or 0) if unmapped: message += ( f" {unmapped} master SKU(s) have no remote match" " (configurable parents are created on push, not via link-from-remote)." ) return { "backend": "channel_sku_mapping", "status": "completed", "total_count": total, "success_count": linked, "message": message, **result, } if job.channel_type == "magento" and job.job_type == "push_images": return _execute_magento_push_images(session, job) if job.channel_type == "magento" and job.job_type == "media_cleanup": return _execute_magento_media_cleanup(session, job) if job.channel_type == "magento" and job.job_type == "push": label = f"channel-job-{job.id}-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}" options = _job_options(job) queue_options: Dict[str, Any] = { "dry_run": job.dry_run, "force_full_sync": job.mode == "full", "use_master_catalog": True, } if options.get("limit_skus"): queue_options["limit_skus"] = options["limit_skus"] queue_id = SqlAlchemyMagentoSyncQueueRepository(session).enqueue( job.native_connection_id, label, options=queue_options, supersede_queued=True, ) job.backend_queue_id = queue_id # The real sync happens in magento_worker.py; this job stays "delegated" # until refresh_delegated_job sees the backend queue row finish. return { "status": "delegated", "backend": "magento_sync_queue", "backend_status": "queued", "queue_id": queue_id, "label": label, } if job.channel_type == "magento" and job.job_type == "pull": from datetime import date, timedelta from app.jobs.magento_baseline_sync import run_baseline_sync from app.jobs.magento_pull import run_pull from settings import load_magento_pull_config full_pull = job.mode == "full" baseline = run_baseline_sync(job.native_connection_id, force=full_pull, session=session) cfg = load_magento_pull_config() since = date(1970, 1, 1) if full_pull else date.today() - timedelta(days=cfg.lookback_days) product_pull = run_pull( job.native_connection_id, since, full_pull=full_pull, dry_run=job.dry_run, persist_source_snapshot=not job.dry_run, ) status = "completed" if product_pull.get("status") == "failed" or baseline.get("status") == "failed": status = "failed" return _maybe_auto_link_after_pull( session, job, { "backend": "magento_pull", "status": status, "baseline": baseline, "product_pull": product_pull, "pulled": product_pull.get("pulled") or product_pull.get("matched_products") or 0, "total_count": product_pull.get("pulled") or product_pull.get("matched_products") or product_pull.get("api_total"), }, ) if job.channel_type == "shopify" and job.job_type == "push_images": return _execute_shopify_push_images(session, job) if job.job_type == "remove_products": from db.channel_product_removal import run_remove_products_job options = _job_options(job) return run_remove_products_job( session, channel_code=job.channel_type, connection_id=job.native_connection_id, options=options, dry_run=job.dry_run, on_progress=_channel_job_progress_callback(job.id), ) if job.channel_type == "shopify" and job.job_type == "push": from shopify.product_sync import push_products options = _job_options(job) result = push_products( session, dry_run=job.dry_run, force=bool(options.get("force")) or job.mode == "full", limit=options.get("limit"), skus=options.get("skus"), shop_code=options.get("shop_code") or _shopify_shop_code(session, job), connection_id=options.get("connection_id") or job.channel_connection_id, on_progress=_channel_job_progress_callback(job.id), ) return {"backend": "shopify_product_sync", **result} if job.channel_type == "shopify" and job.job_type == "pull": from app.jobs.shopify_attribute_sync import run_shopify_attribute_sync from app.jobs.shopify_pull import run_shopify_pull attrs = run_shopify_attribute_sync( connection_id=job.native_connection_id, shop_code=_shopify_shop_code(session, job), session=session, ) products = run_shopify_pull( connection_id=job.native_connection_id, shop_code=_shopify_shop_code(session, job), dry_run=job.dry_run, session=session, ) collections = {"status": "skipped", "dry_run": job.dry_run} if not job.dry_run: from db.models import ShopifyConnection from shopify.collection_sync import pull_shopify_collections from shopify.connections import build_client shop_code = _shopify_shop_code(session, job) connection = session.get(ShopifyConnection, job.native_connection_id) client = build_client(connection) collections = pull_shopify_collections( client, shop_code=shop_code or (connection.shop_code if connection else ""), connection_id=job.native_connection_id, session=session, ) status = "completed" if attrs.get("status") != "success" or products.get("status") != "success": status = "failed" return _maybe_auto_link_after_pull( session, job, { "backend": "shopify_pull", "status": status, "attributes": attrs, "products": products, "collections": collections, "pulled": products.get("distinct_skus") or products.get("inserted") or 0, "total_count": products.get("distinct_skus") or products.get("variant_rows") or products.get("input_rows"), }, ) if job.channel_type == "plytix" and job.job_type == "pull": from app.jobs.plytix_pull import run_plytix_pull result = run_plytix_pull(dry_run=job.dry_run) status = "completed" if result.get("status") == "ok" else "failed" return _maybe_auto_link_after_pull( session, job, { "backend": "plytix_pull", **result, "status": status, "pulled": result.get("distinct_skus") or result.get("inserted") or 0, "total_count": result.get("distinct_skus") or result.get("input_rows"), }, ) if job.job_type == "transform": pipeline_code = resolve_pipeline_channel_code(job.channel_code, channel_type=job.channel_type) payloads = build_channel_product_payloads( session, pipeline_code, only_assigned=False, connection_id=job.channel_connection_id, ) return {"status": "completed", "count": len(payloads), "backend": "channel_transform"} if job.job_type == "collection_bootstrap": return _execute_collection_bootstrap(session, job) if job.job_type == "collection_push": return _execute_collection_push(session, job) if job.job_type == "collection_reconcile": return _execute_collection_reconcile(session, job) if job.job_type == "collection_reconcile_all": return _execute_collection_reconcile_all(session, job) return { "status": "not_implemented", "message": f"{job.channel_type} {job.job_type} has no worker implementation yet", } def _job_progress(job: ChannelJob) -> dict[str, Any]: total = job.total_count success = job.success_count or 0 errors = job.error_count or 0 completed = success + errors stored = (job.result or {}).get("progress") if isinstance(job.result, dict) else None # In-flight Shopify jobs write explicit completed; delegated Magento jobs use sync_job counts. if isinstance(stored, dict) and job.status == "running" and stored.get("completed") is not None: completed = int(stored.get("completed") or completed) if stored.get("total") is not None: total = int(stored.get("total")) elif isinstance(stored, dict) and job.status == "delegated": if stored.get("total") is not None and not total: total = int(stored.get("total")) if stored.get("completed") is not None: completed = max(completed, int(stored.get("completed") or 0)) percent = round(100 * completed / total, 1) if total and total > 0 else None stage = None if isinstance(stored, dict): stage = stored.get("stage") elif isinstance(job.result, dict): stage = job.result.get("stage") return { "total": total, "completed": completed, "success_count": success, "error_count": errors, "percent": percent, "stage": stage, } def refresh_delegated_job(session: Session, job: ChannelJob) -> ChannelJob: """Finalize a delegated job from its backend queue row (called from the job GET endpoint). Magento push jobs hand off to magento_sync_queue; once that row reaches done/failed, copy the outcome (and the linked sync-job counts) onto the channel_job so the dashboard polling sees the true status. """ if job.status != "delegated" or not job.backend_queue_id: return job queue_row = session.get(MagentoSyncQueue, job.backend_queue_id) if queue_row is None: return job result = dict(job.result or {}) result["backend_status"] = queue_row.status job.result = result if queue_row.status not in {"done", "failed"}: if queue_row.job_id: sync_job = session.get(MagentoSyncJob, queue_row.job_id) if sync_job: job.backend_job_id = sync_job.id job.total_count = sync_job.total_count job.success_count = sync_job.success_count or 0 job.error_count = sync_job.error_count or 0 if sync_job.started_at and not job.started_at: job.started_at = sync_job.started_at result["progress"] = _job_progress(job) job.result = result session.flush() return job job.status = "completed" if queue_row.status == "done" else "failed" if queue_row.status == "failed": job.notes = queue_row.last_error or job.notes elif not job.notes and queue_row.status == "done": job.notes = "Magento image push finished (see counts)." if queue_row.job_id: sync_job = session.get(MagentoSyncJob, queue_row.job_id) if sync_job: job.backend_job_id = sync_job.id if job.total_count is None and sync_job.total_count is not None: job.total_count = sync_job.total_count if (job.success_count or 0) == 0 and (sync_job.success_count or 0) > 0: job.success_count = sync_job.success_count or 0 elif job.success_count is None: job.success_count = sync_job.success_count or 0 if job.error_count is None: job.error_count = sync_job.error_count or 0 if sync_job.notes and not job.notes: job.notes = sync_job.notes result = dict(job.result or {}) result["backend_status"] = queue_row.status if job.notes: result["summary"] = job.notes job.result = result job.finished_at = datetime.utcnow() _update_schedule_after_job(session, job) session.flush() return job def _update_schedule_after_job(session: Session, job: ChannelJob) -> None: if job.schedule_id is None: return schedule = session.get(ChannelSchedule, job.schedule_id) if not schedule: return schedule.last_run_at = datetime.utcnow() schedule.last_status = job.status schedule.last_error = job.notes if job.status in {"failed", "not_implemented"} else None schedule.last_enqueued_job_id = job.id def job_to_dict(job: ChannelJob) -> dict[str, Any]: progress = _job_progress(job) return { "id": job.id, "schedule_id": job.schedule_id, "channel_connection_id": job.channel_connection_id, "channel_type": job.channel_type, "native_connection_id": job.native_connection_id, "channel_code": job.channel_code, "job_type": job.job_type, "status": job.status, "dry_run": job.dry_run, "mode": job.mode, "requested_at": job.requested_at.isoformat() if job.requested_at else None, "started_at": job.started_at.isoformat() if job.started_at else None, "finished_at": job.finished_at.isoformat() if job.finished_at else None, "total_count": job.total_count, "success_count": job.success_count, "error_count": job.error_count, "notes": job.notes, "result": job.result or {}, "backend_queue_id": job.backend_queue_id, "backend_job_id": job.backend_job_id, "progress": progress, "terminal": job.status in TERMINAL_STATUSES, }