""" Daily Magento pull: fetch updated products and persist to magento_catalog_state. CLI: python -m app.jobs.magento_pull --connection-id X --since YYYY-MM-DD """ from __future__ import annotations import argparse import hashlib import json import logging import sys from datetime import date, datetime, timedelta, timezone from typing import Optional logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _canonical_json(obj) -> str: if obj is None: return "null" return json.dumps(obj, sort_keys=True, default=str) def _media_hash(media_files: list) -> str: blob = _canonical_json(media_files) return hashlib.sha256(blob.encode()).hexdigest() def run_pull( connection_id: int, since: date, *, page_size: int = 100, persist_source_snapshot: bool = True, full_pull: bool = False, dry_run: bool = False, ) -> dict: from db.session import get_session from db.magento_repositories import ( SqlAlchemyMagentoConnectionRepository, SqlAlchemyMagentoCatalogStateRepository, ) from db.repositories import SqlAlchemyIngestRepository from db.source_imports import import_magento_source_dataframe from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs from magento.magento_api import MagentoRestClient since_dt = datetime.combine(since, datetime.min.time(), tzinfo=timezone.utc) pulled = 0 pages = 0 errors = 0 seen = 0 api_total: Optional[int] = None with get_session() as session: conn_repo = SqlAlchemyMagentoConnectionRepository(session) catalog_repo = SqlAlchemyMagentoCatalogStateRepository(session) ingest_repo = SqlAlchemyIngestRepository(session) conn = conn_repo.get_for_sync(connection_id) if not conn: return {"status": "failed", "error": "Connection not found", "pulled": 0} oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) api = MagentoRestClient(oauth) current_page = 1 source_rows = [] # Use list_products (same as /api/magento/connections/{id}/products) - no filter_groups, # so OAuth signature works. Filter by updated_at client-side. while True: status, items, total, err = api.list_products( page_size=page_size, current_page=current_page ) if status != 200 or err: logger.error("Pull failed: HTTP %s %s", status, err) return {"status": "failed", "error": err or f"HTTP {status}", "pulled": pulled} if api_total is None and total is not None: api_total = int(total) if not items: break pages += 1 for p in items: if not isinstance(p, dict): continue sku = p.get("sku") if not sku or not str(sku).strip(): continue updated_at_raw = p.get("updated_at") updated_at = None if updated_at_raw: try: updated_at = datetime.fromisoformat( str(updated_at_raw).replace("Z", "+00:00") ) if updated_at.tzinfo is None: updated_at = updated_at.replace(tzinfo=timezone.utc) except ValueError: pass if not full_pull and (updated_at is None or updated_at < since_dt): continue seen += 1 if dry_run: continue try: if persist_source_snapshot: source_rows.append(p) price_val = None if "price" in p and p["price"] is not None: try: price_val = float(p["price"]) except (TypeError, ValueError): pass special_val = None custom = p.get("custom_attributes") or [] for attr in custom: if isinstance(attr, dict) and attr.get("attribute_code") == "special_price": try: special_val = float(attr.get("value", 0)) except (TypeError, ValueError): pass break media_entries = p.get("media_gallery_entries") or [] media_files = [] for m in media_entries: if isinstance(m, dict): media_files.append({ "id": m.get("id"), "file": m.get("file"), "types": m.get("types") or [], "position": m.get("position", 0), "disabled": m.get("disabled", False), }) media_files.sort(key=lambda x: (x.get("position", 0), x.get("id") or 0)) m_hash = _media_hash(media_files) if media_files else None catalog_repo.upsert_catalog_state( connection_id, sku, magento_product_id=p.get("id"), updated_at_magento=updated_at, price=price_val, special_price=special_val, media_files=media_files if media_files else None, media_hash=m_hash, ) pulled += 1 except Exception as e: logger.exception("Failed to process %s: %s", sku, e) errors += 1 if total is not None and current_page * page_size >= total: break current_page += 1 if dry_run: session.rollback() return { "status": "success", "dry_run": True, "full_pull": full_pull, "since": since.isoformat(), "matched_products": seen, "api_total": api_total, "pages_scanned": pages, } source_ingest_id = None source_stats = None if persist_source_snapshot and source_rows: import pandas as pd df = pd.DataFrame(source_rows) source_ingest_id, source_stats = import_magento_source_dataframe( ingest_repo, df, file_name=f"magento_api_pull_{connection_id}_{datetime.now(timezone.utc).isoformat()}.json", file_hash=hashlib.sha256(_canonical_json(source_rows).encode()).hexdigest(), connection_id=connection_id, ) session.commit() result = { "status": "success", "dry_run": False, "full_pull": full_pull, "pulled": pulled, "pages": pages, "errors": errors, } if source_ingest_id is not None: result["source_ingest_id"] = source_ingest_id result["source_snapshot"] = source_stats return result def main() -> int: parser = argparse.ArgumentParser(description="Magento daily pull: fetch updated products") parser.add_argument("--connection-id", type=int, required=True, help="Magento connection ID") parser.add_argument( "--since", type=str, help="Since date YYYY-MM-DD (default: lookback from today)", ) parser.add_argument("--page-size", type=int, default=100, help="Page size for API") parser.add_argument( "--no-source-snapshot", action="store_true", help="Do not persist raw Magento product payloads into source snapshot table", ) parser.add_argument( "--full", action="store_true", help="Pull all products regardless of updated_at (full catalog snapshot)", ) parser.add_argument( "--dry-run", action="store_true", help="Scan remote catalog and report counts without persisting", ) args = parser.parse_args() if args.since: try: since = date.fromisoformat(args.since) except ValueError: print("Invalid --since format, use YYYY-MM-DD", file=sys.stderr) return 1 else: from settings import load_magento_pull_config cfg = load_magento_pull_config() since = date.today() - timedelta(days=cfg.lookback_days) result = run_pull( args.connection_id, since, page_size=args.page_size, persist_source_snapshot=not args.no_source_snapshot, full_pull=args.full, dry_run=args.dry_run, ) if result.get("status") == "failed": print(result.get("error", "unknown"), file=sys.stderr) return 1 print(f"Pulled {result.get('pulled', 0)} products, {result.get('pages', 0)} pages") return 0 if __name__ == "__main__": sys.exit(main())