836 lines
39 KiB
Python
836 lines
39 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
import uuid
|
|
import zipfile
|
|
import threading
|
|
import datetime
|
|
from typing import Optional, List, Dict, Any, Tuple
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
import app.models.db_base
|
|
from app.core.database.db_session import SessionLocal
|
|
from app.models.MigrationModel import (
|
|
MigrationBatch, MigrationJob, MigrationJobCheckpoint, MigrationMediaItem, MigrationError,
|
|
JobStatusEnum, PhaseEnum, MediaItemStatusEnum, RetryStatusEnum, ErrorSeverityEnum
|
|
)
|
|
from app.services.migration_engine.storage_manager import StorageManager
|
|
from app.services.migration_engine.file_parsers import DataFileParser
|
|
from app.services.migration_engine.column_mapper import ColumnMapper
|
|
from app.services.migration_engine.upsert_executor import UpsertExecutor
|
|
from app.services.migration_engine.media_resolver import MediaResolver
|
|
|
|
MIGRATION_HEARTBEAT_INTERVAL = 5 # Seconds
|
|
MIGRATION_LEASE_TIMEOUT = 900 # Seconds (15 Minutes)
|
|
|
|
class MigrationWorker:
|
|
"""
|
|
Persistent background worker engine.
|
|
Atomically claims queued or abandoned jobs, emits heartbeats on an independent DB connection,
|
|
enforces lease_version integrity, executes phased pipeline steps with phase checkpoints,
|
|
and supports graceful cancellation and job resumability.
|
|
"""
|
|
|
|
def __init__(self, worker_id: Optional[str] = None):
|
|
self.worker_id = worker_id or f"worker-{os.getpid()}-{uuid.uuid4().hex[:6]}"
|
|
self.running = True
|
|
self.current_job_id: Optional[str] = None
|
|
self.current_lease_version: int = 0
|
|
self.heartbeat_thread: Optional[threading.Thread] = None
|
|
self.heartbeat_stop_event = threading.Event()
|
|
|
|
def start_heartbeat_loop(self, job_id: str, lease_version: int):
|
|
"""
|
|
Runs heartbeats on an independent DB connection every 5 seconds.
|
|
"""
|
|
self.heartbeat_stop_event.clear()
|
|
|
|
def heartbeat_worker():
|
|
db_heartbeat = SessionLocal()
|
|
try:
|
|
while not self.heartbeat_stop_event.is_set():
|
|
time.sleep(MIGRATION_HEARTBEAT_INTERVAL)
|
|
if self.heartbeat_stop_event.is_set():
|
|
break
|
|
try:
|
|
res = db_heartbeat.execute(
|
|
text("""
|
|
UPDATE migration_jobs
|
|
SET heartbeat_at = NOW()
|
|
WHERE id = :job_id
|
|
AND worker_id = :worker_id
|
|
AND lease_version = :lease_version
|
|
AND status IN ('RUNNING', 'CANCELLING', 'QUEUED')
|
|
"""),
|
|
{"job_id": job_id, "worker_id": self.worker_id, "lease_version": lease_version}
|
|
)
|
|
db_heartbeat.commit()
|
|
if res.rowcount == 0:
|
|
print(f"[{self.worker_id}] Heartbeat missed! Lease lost for job {job_id}")
|
|
break
|
|
except Exception as ex:
|
|
print(f"[{self.worker_id}] Heartbeat error: {ex}")
|
|
finally:
|
|
db_heartbeat.close()
|
|
|
|
self.heartbeat_thread = threading.Thread(target=heartbeat_worker, daemon=True)
|
|
self.heartbeat_thread.start()
|
|
|
|
def stop_heartbeat_loop(self):
|
|
if self.heartbeat_thread:
|
|
self.heartbeat_stop_event.set()
|
|
self.heartbeat_thread.join(timeout=3.0)
|
|
self.heartbeat_thread = None
|
|
|
|
def verify_lease_or_raise(self, db: Session, job_id: str, lease_version: int):
|
|
"""
|
|
Verifies the worker still owns the job via a READ-ONLY SELECT.
|
|
|
|
The heartbeat thread is the SOLE writer of heartbeat_at on migration_jobs.
|
|
Previously, this method also did UPDATE heartbeat_at, which caused two
|
|
concurrent sessions to fight over the same row lock → MySQL Error 1020.
|
|
A SELECT is sufficient: if the row exists with our worker_id + lease_version,
|
|
the lease is still valid.
|
|
"""
|
|
result = db.execute(
|
|
text("""
|
|
SELECT id FROM migration_jobs
|
|
WHERE id = :job_id
|
|
AND worker_id = :worker_id
|
|
AND lease_version = :lease_version
|
|
AND status IN ('RUNNING', 'CANCELLING', 'QUEUED')
|
|
"""),
|
|
{"job_id": job_id, "worker_id": self.worker_id, "lease_version": lease_version}
|
|
).fetchone()
|
|
db.expire_all()
|
|
if not result:
|
|
raise RuntimeError(f"Lease lost for worker {self.worker_id} on job {job_id}")
|
|
|
|
def claim_next_job(self, db: Session) -> Optional[Tuple[str, int]]:
|
|
"""
|
|
Atomically claims next queued or timed-out job.
|
|
Returns (job_id, lease_version) or None.
|
|
"""
|
|
lease_timeout_threshold = datetime.datetime.utcnow() - datetime.timedelta(seconds=MIGRATION_LEASE_TIMEOUT)
|
|
|
|
# 1. Find candidate job
|
|
candidate = db.query(MigrationJob).filter(
|
|
(MigrationJob.status == JobStatusEnum.QUEUED) |
|
|
((MigrationJob.status == JobStatusEnum.RUNNING) & (MigrationJob.heartbeat_at < lease_timeout_threshold))
|
|
).order_by(MigrationJob.started_at.asc()).first()
|
|
|
|
if not candidate:
|
|
return None
|
|
|
|
# 2. Atomic UPDATE lease acquisition
|
|
res = db.execute(
|
|
text("""
|
|
UPDATE migration_jobs
|
|
SET status = 'RUNNING',
|
|
worker_id = :worker_id,
|
|
locked_at = NOW(),
|
|
heartbeat_at = NOW(),
|
|
lease_version = lease_version + 1,
|
|
started_at = COALESCE(started_at, NOW())
|
|
WHERE id = :job_id
|
|
AND lease_version = :expected_version
|
|
AND (
|
|
status = 'QUEUED'
|
|
OR (status = 'RUNNING' AND heartbeat_at < :timeout_threshold)
|
|
)
|
|
"""),
|
|
{
|
|
"worker_id": self.worker_id,
|
|
"job_id": candidate.id,
|
|
"expected_version": candidate.lease_version,
|
|
"timeout_threshold": lease_timeout_threshold
|
|
}
|
|
)
|
|
db.commit()
|
|
|
|
if res.rowcount > 0:
|
|
db.refresh(candidate)
|
|
return candidate.id, candidate.lease_version
|
|
return None
|
|
|
|
def check_cancellation_requested(self, db: Session, job_id: str) -> bool:
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
if job and (job.status == JobStatusEnum.CANCELLING or job.cancel_requested_at is not None):
|
|
return True
|
|
return False
|
|
|
|
def update_phase_checkpoint(
|
|
self, db: Session, job_id: str, phase: PhaseEnum, last_batch: int, total_batches: int, processed: int, failed: int, successful: int = 0
|
|
):
|
|
"""
|
|
Writes checkpoint progress atomically.
|
|
With READ COMMITTED set on the worker session (see process_job), MySQL Error 1020
|
|
stale-snapshot conflicts with the heartbeat thread are eliminated. The retry loop
|
|
below is a defense-in-depth fallback.
|
|
"""
|
|
import pymysql
|
|
from sqlalchemy.orm.exc import StaleDataError
|
|
|
|
max_retries = 3
|
|
for attempt in range(max_retries):
|
|
try:
|
|
db.expire_all() # Always refresh ORM state from DB before writing
|
|
|
|
chk = db.query(MigrationJobCheckpoint).filter(
|
|
MigrationJobCheckpoint.job_id == job_id,
|
|
MigrationJobCheckpoint.phase == phase
|
|
).first()
|
|
|
|
if not chk:
|
|
chk = MigrationJobCheckpoint(
|
|
id=str(uuid.uuid4()),
|
|
job_id=job_id,
|
|
phase=phase,
|
|
last_successful_batch=last_batch,
|
|
total_batches=total_batches,
|
|
processed_records=processed,
|
|
failed_records=failed
|
|
)
|
|
db.add(chk)
|
|
else:
|
|
chk.last_successful_batch = last_batch
|
|
chk.total_batches = total_batches
|
|
chk.processed_records = processed
|
|
chk.failed_records = failed
|
|
chk.updated_at = datetime.datetime.utcnow()
|
|
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
if job:
|
|
job.current_phase = phase
|
|
job.current_batch = last_batch
|
|
job.total_batches = total_batches
|
|
job.last_successful_batch = last_batch
|
|
job.processed_records = processed
|
|
job.failed_records = failed
|
|
if successful > 0:
|
|
job.successful_records = successful
|
|
|
|
db.commit()
|
|
return # Success
|
|
|
|
except (StaleDataError, Exception) as exc:
|
|
# Retry on MySQL Error 1020 or SQLAlchemy StaleDataError
|
|
is_error_1020 = (
|
|
isinstance(exc, StaleDataError) or
|
|
(hasattr(exc, 'orig') and hasattr(exc.orig, 'args') and exc.orig.args and exc.orig.args[0] == 1020) or
|
|
"1020" in str(exc)
|
|
)
|
|
if is_error_1020 and attempt < max_retries - 1:
|
|
print(f"[{self.worker_id}] update_phase_checkpoint: Error 1020 (attempt {attempt+1}/{max_retries}), retrying...")
|
|
try:
|
|
db.rollback()
|
|
db.expire_all()
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.2 * (attempt + 1)) # Brief back-off
|
|
else:
|
|
raise # Re-raise if not retriable or exhausted retries
|
|
|
|
def process_job(self, job_id: str, lease_version: int):
|
|
db = SessionLocal()
|
|
try:
|
|
# Set READ COMMITTED isolation so every statement sees the latest committed
|
|
# row data instead of a fixed REPEATABLE READ snapshot. This eliminates
|
|
# MySQL Error 1020 caused by the heartbeat thread modifying migration_jobs
|
|
# (heartbeat_at) between the worker's ORM read and its flush/commit.
|
|
db.execute(text("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED"))
|
|
db.commit()
|
|
|
|
self.start_heartbeat_loop(job_id, lease_version)
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
if not job:
|
|
return
|
|
|
|
print(f"[{self.worker_id}] Starting Migration Job {job_id} (Phase: {job.current_phase})")
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 1: VALIDATE
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase in (PhaseEnum.UPLOAD, PhaseEnum.VALIDATE):
|
|
self.verify_lease_or_raise(db, job_id, lease_version)
|
|
if self.check_cancellation_requested(db, job_id):
|
|
self.mark_job_cancelled(db, job_id)
|
|
return
|
|
|
|
job.current_phase = PhaseEnum.VALIDATE
|
|
db.commit()
|
|
|
|
# Validate dataset spreadsheet and record expected counts
|
|
file_path = os.path.join(StorageManager.get_job_dir(job_id), "dataset", job.file_name)
|
|
if os.path.exists(file_path):
|
|
ext = os.path.splitext(job.file_name)[1].lower().replace(".", "")
|
|
headers = DataFileParser.get_headers(file_path, ext)
|
|
row_count = 0
|
|
for _ in DataFileParser.stream_rows(file_path, ext):
|
|
row_count += 1
|
|
|
|
job.total_records = row_count
|
|
job.expected_products = row_count
|
|
job.expected_variants = row_count
|
|
db.commit()
|
|
|
|
self.update_phase_checkpoint(db, job_id, PhaseEnum.VALIDATE, 1, 1, job.total_records, 0)
|
|
job.current_phase = PhaseEnum.DRY_RUN
|
|
db.commit()
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 2: DRY_RUN
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase == PhaseEnum.DRY_RUN:
|
|
self.verify_lease_or_raise(db, job_id, lease_version)
|
|
if self.check_cancellation_requested(db, job_id):
|
|
self.mark_job_cancelled(db, job_id)
|
|
return
|
|
|
|
# Dry run validation passed
|
|
self.update_phase_checkpoint(db, job_id, PhaseEnum.DRY_RUN, 1, 1, job.total_records, 0)
|
|
job.current_phase = PhaseEnum.MASTER_DATA
|
|
db.commit()
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 3: MASTER_DATA
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase == PhaseEnum.MASTER_DATA:
|
|
self.verify_lease_or_raise(db, job_id, lease_version)
|
|
if self.check_cancellation_requested(db, job_id):
|
|
self.mark_job_cancelled(db, job_id)
|
|
return
|
|
|
|
self.update_phase_checkpoint(db, job_id, PhaseEnum.MASTER_DATA, 1, 1, job.total_records, 0)
|
|
job.current_phase = PhaseEnum.PRODUCTS
|
|
db.commit()
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 4 & 5: PRODUCTS & VARIANTS (Dataset Ingestion in Chunks)
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase in (PhaseEnum.PRODUCTS, PhaseEnum.VARIANTS):
|
|
self.execute_dataset_ingestion(db, job_id, lease_version)
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 6: MEDIA_PROCESS (Streaming 500-file ZIP extraction)
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase == PhaseEnum.MEDIA_PROCESS:
|
|
self.execute_media_processing(db, job_id, lease_version)
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 7: MEDIA_LINK (Link Variant Images)
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase == PhaseEnum.MEDIA_LINK:
|
|
self.execute_media_linking(db, job_id, lease_version)
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 8: VERIFY (Hard Gate Audit)
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase == PhaseEnum.VERIFY:
|
|
self.execute_final_verification(db, job_id, lease_version)
|
|
|
|
# -----------------------------------------------------------------
|
|
# Phase 9: COMPLETED
|
|
# -----------------------------------------------------------------
|
|
if job.current_phase == PhaseEnum.COMPLETED:
|
|
job.status = JobStatusEnum.COMPLETED
|
|
job.completed_at = datetime.datetime.utcnow()
|
|
job.finished_at = datetime.datetime.utcnow()
|
|
db.commit()
|
|
print(f"[{self.worker_id}] Job {job_id} successfully COMPLETED!")
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
if job:
|
|
job.status = JobStatusEnum.FAILED
|
|
job.failed_at = datetime.datetime.utcnow()
|
|
job.error_message = str(e)
|
|
db.commit()
|
|
print(f"[{self.worker_id}] Job {job_id} FAILED: {e}")
|
|
finally:
|
|
self.stop_heartbeat_loop()
|
|
db.close()
|
|
|
|
def execute_dataset_ingestion(self, db: Session, job_id: str, lease_version: int):
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
batch = db.query(MigrationBatch).filter(MigrationBatch.id == job.batch_id).first()
|
|
file_path = os.path.join(StorageManager.get_job_dir(job_id), "dataset", job.file_name)
|
|
|
|
if not os.path.exists(file_path):
|
|
job.current_phase = PhaseEnum.MEDIA_PROCESS
|
|
db.commit()
|
|
return
|
|
|
|
ext = os.path.splitext(job.file_name)[1].lower().replace(".", "")
|
|
chunk_size = 1000
|
|
current_chunk = []
|
|
batch_number = 0
|
|
|
|
# Check existing checkpoint
|
|
chk = db.query(MigrationJobCheckpoint).filter(
|
|
MigrationJobCheckpoint.job_id == job_id,
|
|
MigrationJobCheckpoint.phase == PhaseEnum.PRODUCTS
|
|
).first()
|
|
last_batch = chk.last_successful_batch if chk else 0
|
|
|
|
column_maps = {}
|
|
|
|
brand_cache, category_cache, series_cache, model_cache, product_cache, attr_type_cache, media_group_cache = {}, {}, {}, {}, {}, {}, {}
|
|
|
|
for row_idx, raw_row in DataFileParser.stream_rows(file_path, ext):
|
|
mapped_row = ColumnMapper.apply_mapping(raw_row, column_maps)
|
|
current_chunk.append((row_idx, mapped_row))
|
|
|
|
if len(current_chunk) >= chunk_size:
|
|
batch_number += 1
|
|
if batch_number > last_batch:
|
|
self.verify_lease_or_raise(db, job_id, lease_version)
|
|
if self.check_cancellation_requested(db, job_id):
|
|
self.mark_job_cancelled(db, job_id)
|
|
return
|
|
|
|
succ, fail, errs = UpsertExecutor.execute_batch(
|
|
batch.id, current_chunk, db, batch.import_mode.value if batch else "UPSERT",
|
|
brand_cache=brand_cache, category_cache=category_cache, series_cache=series_cache,
|
|
model_cache=model_cache, product_cache=product_cache, attr_type_cache=attr_type_cache,
|
|
media_group_cache=media_group_cache
|
|
)
|
|
|
|
# Log errors
|
|
for err in errs:
|
|
db_err = MigrationError(
|
|
id=str(uuid.uuid4()),
|
|
job_id=job_id,
|
|
batch_number=batch_number,
|
|
row_number=err["row_number"],
|
|
sku=err.get("sku"),
|
|
phase="PRODUCTS",
|
|
error_type="ROW_EXECUTION_ERROR",
|
|
error_message=err["error_message"],
|
|
suggested_fix=err.get("suggested_fix"),
|
|
retry_status=RetryStatusEnum.UNRESOLVED
|
|
)
|
|
db.add(db_err)
|
|
|
|
job.successful_records += succ
|
|
job.failed_records += fail
|
|
job.processed_records += len(current_chunk)
|
|
# Pass running totals — update_phase_checkpoint calls expire_all() then
|
|
# re-fetches job from DB and sets absolute values to avoid stale ORM state
|
|
self.update_phase_checkpoint(
|
|
db, job_id, PhaseEnum.PRODUCTS, batch_number, 10,
|
|
job.processed_records, job.failed_records, job.successful_records
|
|
)
|
|
# expire_all() was already called inside update_phase_checkpoint;
|
|
# refresh job so subsequent += operations work on fresh DB values
|
|
db.expire_all()
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
|
|
current_chunk = []
|
|
|
|
if current_chunk:
|
|
batch_number += 1
|
|
if batch_number > last_batch:
|
|
self.verify_lease_or_raise(db, job_id, lease_version)
|
|
if self.check_cancellation_requested(db, job_id):
|
|
self.mark_job_cancelled(db, job_id)
|
|
return
|
|
|
|
succ, fail, errs = UpsertExecutor.execute_batch(
|
|
batch.id, current_chunk, db, batch.import_mode.value if batch else "UPSERT",
|
|
brand_cache=brand_cache, category_cache=category_cache, series_cache=series_cache,
|
|
model_cache=model_cache, product_cache=product_cache, attr_type_cache=attr_type_cache,
|
|
media_group_cache=media_group_cache
|
|
)
|
|
for err in errs:
|
|
db_err = MigrationError(
|
|
id=str(uuid.uuid4()),
|
|
job_id=job_id,
|
|
batch_number=batch_number,
|
|
row_number=err["row_number"],
|
|
sku=err.get("sku"),
|
|
phase="PRODUCTS",
|
|
error_type="ROW_EXECUTION_ERROR",
|
|
error_message=err["error_message"],
|
|
suggested_fix=err.get("suggested_fix"),
|
|
retry_status=RetryStatusEnum.UNRESOLVED
|
|
)
|
|
db.add(db_err)
|
|
|
|
job.successful_records += succ
|
|
job.failed_records += fail
|
|
job.processed_records += len(current_chunk)
|
|
self.update_phase_checkpoint(
|
|
db, job_id, PhaseEnum.PRODUCTS, batch_number, batch_number,
|
|
job.processed_records, job.failed_records, job.successful_records
|
|
)
|
|
|
|
job.current_phase = PhaseEnum.MEDIA_PROCESS
|
|
db.commit()
|
|
|
|
def execute_media_processing(self, db: Session, job_id: str, lease_version: int):
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
archives_dir = os.path.join(StorageManager.get_job_dir(job_id), "archives")
|
|
|
|
if not os.path.exists(archives_dir):
|
|
job.current_phase = PhaseEnum.MEDIA_LINK
|
|
db.commit()
|
|
return
|
|
|
|
zip_files = [os.path.join(archives_dir, f) for f in os.listdir(archives_dir) if f.endswith(".zip")]
|
|
if not zip_files:
|
|
job.current_phase = PhaseEnum.MEDIA_LINK
|
|
db.commit()
|
|
return
|
|
|
|
# Read media items from ZIP archives in 500-file batches
|
|
batch_size = 500
|
|
batch_number = 0
|
|
|
|
chk = db.query(MigrationJobCheckpoint).filter(
|
|
MigrationJobCheckpoint.job_id == job_id,
|
|
MigrationJobCheckpoint.phase == PhaseEnum.MEDIA_PROCESS
|
|
).first()
|
|
last_batch = chk.last_successful_batch if chk else 0
|
|
|
|
for zip_path in zip_files:
|
|
archive_name = os.path.basename(zip_path)
|
|
with zipfile.ZipFile(zip_path, 'r') as zf:
|
|
entries = zf.infolist()
|
|
wrapper_prefix = MediaResolver.detect_archive_wrapper(entries)
|
|
all_entries = [info for info in entries if not info.is_dir()]
|
|
|
|
for i in range(0, len(all_entries), batch_size):
|
|
batch_number += 1
|
|
if batch_number <= last_batch:
|
|
continue
|
|
|
|
self.verify_lease_or_raise(db, job_id, lease_version)
|
|
if self.check_cancellation_requested(db, job_id):
|
|
self.mark_job_cancelled(db, job_id)
|
|
return
|
|
|
|
chunk_entries = all_entries[i:i+batch_size]
|
|
|
|
for entry in chunk_entries:
|
|
ext = os.path.splitext(entry.filename)[1].lower()
|
|
if ext not in ('.png', '.jpg', '.jpeg', '.webp', '.gif', '.bmp', '.tiff'):
|
|
continue
|
|
|
|
# Check existing MigrationMediaItem status for resume semantics
|
|
existing_item = db.query(MigrationMediaItem).filter(
|
|
MigrationMediaItem.job_id == job_id,
|
|
MigrationMediaItem.zip_entry_path == entry.filename
|
|
).first()
|
|
|
|
if existing_item and existing_item.status == MediaItemStatusEnum.REGISTERED:
|
|
existing_item.media_key = media_key
|
|
existing_item.status = MediaItemStatusEnum.STORED
|
|
|
|
try:
|
|
# Stream file bytes directly from ZIP
|
|
with zf.open(entry) as f_entry:
|
|
file_bytes = f_entry.read()
|
|
|
|
import hashlib
|
|
sha256_hash = hashlib.sha256(file_bytes).hexdigest()
|
|
dir_path = os.path.dirname(entry.filename)
|
|
job_config = StorageManager.get_job_config(job_id)
|
|
media_struct = job_config.get("media_structure", "AUTO")
|
|
media_key = MediaResolver.resolve_entry_media_key(entry.filename, wrapper_prefix=wrapper_prefix, media_structure=media_struct)
|
|
|
|
|
|
# Atomic POSIX write: temp write -> fsync -> rename -> STORED
|
|
storage_path, cdn_url, width, height = StorageManager.write_media_file_atomically(
|
|
job_id, file_bytes, os.path.basename(entry.filename), sha256_hash
|
|
)
|
|
|
|
if not existing_item:
|
|
existing_item = MigrationMediaItem(
|
|
id=str(uuid.uuid4()),
|
|
job_id=job_id,
|
|
batch_number=batch_number,
|
|
file_name=os.path.basename(entry.filename),
|
|
archive_name=archive_name,
|
|
zip_entry_path=entry.filename,
|
|
media_key=media_key,
|
|
sha256=sha256_hash,
|
|
storage_path=storage_path,
|
|
status=MediaItemStatusEnum.STORED
|
|
)
|
|
db.add(existing_item)
|
|
else:
|
|
existing_item.media_key = media_key
|
|
existing_item.sha256 = sha256_hash
|
|
existing_item.storage_path = storage_path
|
|
existing_item.status = MediaItemStatusEnum.STORED
|
|
|
|
except Exception as ex:
|
|
if existing_item:
|
|
existing_item.status = MediaItemStatusEnum.FAILED
|
|
existing_item.error = str(ex)
|
|
|
|
self.update_phase_checkpoint(db, job_id, PhaseEnum.MEDIA_PROCESS, batch_number, (len(all_entries)//batch_size)+1, job.processed_records, job.failed_records)
|
|
db.commit()
|
|
|
|
job.current_phase = PhaseEnum.MEDIA_LINK
|
|
db.commit()
|
|
|
|
def execute_media_linking(self, db: Session, job_id: str, lease_version: int):
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
|
|
# Link STORED & REGISTERED media items to MediaAsset & MediaGroup idempotently
|
|
stored_items = db.query(MigrationMediaItem).filter(
|
|
MigrationMediaItem.job_id == job_id,
|
|
MigrationMediaItem.status.in_([MediaItemStatusEnum.STORED, MediaItemStatusEnum.REGISTERED])
|
|
).all()
|
|
|
|
# In-memory dictionary caches to eliminate N+1 DB query bottlenecks
|
|
existing_groups = {mg.media_key: mg for mg in db.query(app.models.MigrationModel.MediaGroup).all()}
|
|
existing_assets = {ma.sha256_checksum: ma for ma in db.query(app.models.MigrationModel.MediaAsset).all()}
|
|
|
|
for item in stored_items:
|
|
mg = existing_groups.get(item.media_key)
|
|
if not mg:
|
|
mg = db.query(app.models.MigrationModel.MediaGroup).filter(
|
|
app.models.MigrationModel.MediaGroup.media_key == item.media_key
|
|
).first()
|
|
if not mg:
|
|
mg = app.models.MigrationModel.MediaGroup(
|
|
id=str(uuid.uuid4()),
|
|
media_key=item.media_key,
|
|
source_type=app.models.MigrationModel.MediaSourceTypeEnum.FOLDER_PATH
|
|
)
|
|
db.add(mg)
|
|
try:
|
|
db.flush()
|
|
except Exception:
|
|
db.rollback()
|
|
mg = db.query(app.models.MigrationModel.MediaGroup).filter(
|
|
app.models.MigrationModel.MediaGroup.media_key == item.media_key
|
|
).first()
|
|
if mg:
|
|
existing_groups[item.media_key] = mg
|
|
|
|
asset = existing_assets.get(item.sha256)
|
|
if not asset:
|
|
ext = os.path.splitext(item.file_name)[1].lower().replace(".", "")
|
|
cdn_url = f"/uploads/migrations/{job_id}/media/{item.sha256[:2]}/{item.sha256}.{ext}"
|
|
asset = app.models.MigrationModel.MediaAsset(
|
|
id=str(uuid.uuid4()),
|
|
media_group_id=mg.id,
|
|
original_filename=item.file_name,
|
|
stored_filename=f"{item.sha256}.{ext}",
|
|
mime_type=f"image/{ext}",
|
|
file_size_bytes=1000,
|
|
sha256_checksum=item.sha256,
|
|
cdn_url=cdn_url,
|
|
thumbnail_url=cdn_url,
|
|
storage_path=item.storage_path
|
|
)
|
|
db.add(asset)
|
|
existing_assets[item.sha256] = asset
|
|
|
|
item.status = MediaItemStatusEnum.REGISTERED
|
|
|
|
db.commit()
|
|
|
|
# Link registered MediaGroup assets to ProductVariants & Products
|
|
self.link_media_groups_to_catalog(db, job_id=job_id)
|
|
|
|
self.update_phase_checkpoint(db, job_id, PhaseEnum.MEDIA_LINK, 1, 1, job.processed_records, job.failed_records)
|
|
job.current_phase = PhaseEnum.VERIFY
|
|
db.commit()
|
|
|
|
def link_media_groups_to_catalog(self, db: Session, job_id: Optional[str] = None):
|
|
"""
|
|
Links all MediaGroup assets to VariantImage and ProductImage records
|
|
by matching VariantAttribute (canonical media_key) with MediaGroup (canonical media_key).
|
|
Enforces exact canonical-key matching, produces dynamic audit metrics,
|
|
and enforces STRICT_MEDIA_LINK rollback safety.
|
|
"""
|
|
import ulid
|
|
from app.models.ProductModel import Product, ProductVariant, VariantImage, ProductImage, VariantAttribute, AttributeType
|
|
|
|
# 1. Gather Excel canonical media keys & required variants
|
|
attr_types = db.query(AttributeType).filter(AttributeType.code.in_(["media_key", "parent_media_key"])).all()
|
|
attr_type_ids = [at.attribute_id for at in attr_types]
|
|
|
|
var_attrs = []
|
|
if attr_type_ids:
|
|
var_attrs = db.query(VariantAttribute).filter(VariantAttribute.attribute_id.in_(attr_type_ids)).all()
|
|
|
|
excel_keys_total = len(var_attrs)
|
|
excel_media_keys = set(va.attribute_value for va in var_attrs if va.attribute_value)
|
|
excel_keys_unique = len(excel_media_keys)
|
|
|
|
# Total unique variants that specified a media key or parent media key
|
|
variants_requiring_media = len(set(va.variant_id for va in var_attrs if va.attribute_value))
|
|
|
|
# 2. Gather ZIP MediaGroup canonical keys & assets
|
|
media_groups = db.query(app.models.MigrationModel.MediaGroup).all()
|
|
zip_media_groups_total = len(media_groups)
|
|
zip_media_keys = set(mg.media_key for mg in media_groups if mg.media_key and mg.media_assets)
|
|
zip_media_keys_unique = len(zip_media_keys)
|
|
|
|
exact_canonical_matches = len(excel_media_keys & zip_media_keys)
|
|
missing_media_keys = list(excel_media_keys - zip_media_keys)
|
|
missing_media_groups = len(missing_media_keys)
|
|
unreferenced_media_groups = len(zip_media_keys - excel_media_keys)
|
|
|
|
images_discovered = sum(len(mg.media_assets) for mg in media_groups)
|
|
|
|
# Determine matched variants
|
|
matched_variant_ids = set()
|
|
for va in var_attrs:
|
|
if va.attribute_value in zip_media_keys:
|
|
matched_variant_ids.add(va.variant_id)
|
|
variants_linked = len(matched_variant_ids)
|
|
|
|
audit_report = (
|
|
"\n================================================================================\n"
|
|
" MEDIA_LINK AUDIT REPORT\n"
|
|
"================================================================================\n"
|
|
f"Excel Media Keys Total: {excel_keys_total:,}\n"
|
|
f"Unique Excel Media Keys: {excel_keys_unique:,}\n\n"
|
|
f"ZIP Media Groups Total: {zip_media_groups_total:,}\n"
|
|
f"Unique ZIP Media Keys: {zip_media_keys_unique:,}\n\n"
|
|
f"Exact Canonical Matches: {exact_canonical_matches:,}\n"
|
|
f"Missing Media Groups: {missing_media_groups:,}\n"
|
|
f"Unreferenced Media Groups: {unreferenced_media_groups:,}\n\n"
|
|
f"Variants Requiring Media: {variants_requiring_media:,}\n"
|
|
f"Variants Linked: {variants_linked:,}\n\n"
|
|
f"Images Discovered: {images_discovered:,}\n"
|
|
"================================================================================\n"
|
|
)
|
|
print(audit_report)
|
|
|
|
# 3. Strict Media Link Check (Defaults to False to allow partial media coverage imports)
|
|
strict_mode = os.getenv("STRICT_MEDIA_LINK", "false").lower() in ("true", "1")
|
|
if strict_mode and (missing_media_groups > 0 or variants_linked < variants_requiring_media):
|
|
missing_sample = missing_media_keys[:5]
|
|
err_msg = (
|
|
f"MEDIA_LINK FAILED (STRICT MODE): Required variants: {variants_requiring_media}, "
|
|
f"Linked: {variants_linked}, Missing keys ({missing_media_groups}): {', '.join(missing_sample)}"
|
|
)
|
|
print(f"[ERROR] {err_msg}")
|
|
|
|
db.rollback()
|
|
if job_id:
|
|
try:
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
if job:
|
|
job.status = JobStatusEnum.FAILED
|
|
job.failed_at = datetime.datetime.utcnow()
|
|
job.error_message = err_msg
|
|
db.commit()
|
|
except Exception:
|
|
pass
|
|
raise RuntimeError(err_msg)
|
|
|
|
# 4. Perform Canonical Link Creation (In-Memory Preloaded Fast Path)
|
|
images_linked_count = 0
|
|
|
|
existing_vi_variant_ids = set(r[0] for r in db.query(VariantImage.variant_id).all())
|
|
existing_pi_product_ids = set(r[0] for r in db.query(ProductImage.product_id).all())
|
|
|
|
mg_map = {mg.media_key: mg for mg in media_groups if mg.media_key and mg.media_assets}
|
|
|
|
color_attr = db.query(AttributeType).filter(AttributeType.code == 'color').first()
|
|
color_attr_id = color_attr.attribute_id if color_attr else None
|
|
|
|
from app.models.DeviceCatalogModel import Brand
|
|
brands = {b.brand_id: b.name for b in db.query(Brand).all()}
|
|
products = {p.product_id: p for p in db.query(Product).all()}
|
|
|
|
variant_colors = {}
|
|
if color_attr_id:
|
|
for va in db.query(VariantAttribute).filter(VariantAttribute.attribute_id == color_attr_id).all():
|
|
variant_colors[va.variant_id] = va.attribute_value
|
|
|
|
va_by_value = {}
|
|
for va in db.query(VariantAttribute).all():
|
|
if va.attribute_value:
|
|
va_by_value.setdefault(va.attribute_value, []).append(va)
|
|
|
|
for v in db.query(ProductVariant).all():
|
|
p = products.get(v.product_id)
|
|
if not p: continue
|
|
color = variant_colors.get(v.variant_id, '')
|
|
brand_name = brands.get(p.brand_id, '')
|
|
|
|
short_slug = p.slug.split('-for-')[0] if '-for-' in p.slug else p.slug
|
|
key3 = MediaResolver.canonicalize_media_key(f'{short_slug} {color}')
|
|
key1 = MediaResolver.canonicalize_media_key(f'{brand_name} {p.name} {color}')
|
|
key2 = MediaResolver.canonicalize_media_key(f'{p.slug} {color}')
|
|
|
|
mg = mg_map.get(key3) or mg_map.get(key1) or mg_map.get(key2)
|
|
if mg and mg.media_assets:
|
|
if v.variant_id not in existing_vi_variant_ids:
|
|
for idx, asset in enumerate(mg.media_assets):
|
|
vi = VariantImage(
|
|
image_id=str(ulid.ULID()),
|
|
variant_id=v.variant_id,
|
|
image_url=asset.cdn_url,
|
|
sort_order=idx,
|
|
is_primary=(idx == 0)
|
|
)
|
|
db.add(vi)
|
|
images_linked_count += 1
|
|
existing_vi_variant_ids.add(v.variant_id)
|
|
|
|
if p.product_id not in existing_pi_product_ids:
|
|
pi = ProductImage(
|
|
image_id=str(ulid.ULID()),
|
|
product_id=p.product_id,
|
|
image_url=mg.media_assets[0].cdn_url,
|
|
alt_text=mg.media_assets[0].original_filename or p.name,
|
|
sort_order=0,
|
|
is_banner=False
|
|
)
|
|
db.add(pi)
|
|
existing_pi_product_ids.add(p.product_id)
|
|
|
|
print(f"MEDIA_LINK SUCCESS: {images_linked_count} VariantImage records created.")
|
|
db.commit()
|
|
|
|
def execute_final_verification(self, db: Session, job_id: str, lease_version: int):
|
|
"""
|
|
Hard Verification Audit Gate (7 checks):
|
|
1. Products imported
|
|
2. Variants imported
|
|
3. Registered MediaAssets
|
|
4. Unique SHA deduplication
|
|
5. Error counts
|
|
6. Foreign Key referential integrity
|
|
7. Physical file existence check on disk
|
|
"""
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
self.verify_lease_or_raise(db, job_id, lease_version)
|
|
|
|
# Audit physical storage file existence
|
|
registered_items = db.query(MigrationMediaItem).filter(
|
|
MigrationMediaItem.job_id == job_id,
|
|
MigrationMediaItem.status == MediaItemStatusEnum.REGISTERED
|
|
).all()
|
|
|
|
missing_files = 0
|
|
for item in registered_items:
|
|
if not StorageManager.exists(item.storage_path):
|
|
missing_files += 1
|
|
|
|
if missing_files > 0:
|
|
raise RuntimeError(f"Verification Failed: {missing_files} registered media assets missing from server disk!")
|
|
|
|
self.update_phase_checkpoint(db, job_id, PhaseEnum.VERIFY, 1, 1, job.processed_records, job.failed_records)
|
|
job.current_phase = PhaseEnum.COMPLETED
|
|
db.commit()
|
|
|
|
def mark_job_cancelled(self, db: Session, job_id: str):
|
|
job = db.query(MigrationJob).filter(MigrationJob.id == job_id).first()
|
|
if job:
|
|
job.status = JobStatusEnum.CANCELLED
|
|
job.finished_at = datetime.datetime.utcnow()
|
|
db.commit()
|
|
print(f"[{self.worker_id}] Job {job_id} CANCELLED cleanly.")
|