import time import uuid from django.db import close_old_connections from django.utils import timezone from common.exceptions import RateLimitExceeded from creative_module.models import ( AccountSyncLog, AdCreativeData, CreativeOverview, CreativeOverviewPrompt, ) from creative_module.tasks.ad_creative_task import fetch_ad_creative_data_task from creative_module.tasks.overview_task import generate_creative_overviews_batch from creative_module.services.onboarding_service import refresh_onboarding_run from creative_module.types import UpdateStatus from organization_auth.models import Account ACCOUNT_ID = "0a61564f-6d44-4f04-946a-30575a6c2590" STALE_RUN_ID = "e04601a7-1a68-4de1-a8a5-0f85ce92b83c" LATEST_FAILED_RUN_ID = "86d75712-2c28-4f52-97e0-1e382761b003" TARGET_AD_IDS = [ "120227548866390710", "120230898681030710", "120233128477530710", "120233514266200710", "120233902613540710", "120234316277530710", "120234635607080710", "120234952733400710", "120235098359550710", ] MAX_TOTAL_ATTEMPTS = 4 # initial try + 3 retries, matching Celery intent GENERATE_DEFAULT_OVERVIEWS_SYNCHRONOUSLY = True def run_fetch_sync(row, account, user_id): """Run the Celery task body in-process and emulate final failure bookkeeping.""" AdCreativeData.objects.filter(id=row.id).update( account_sync_log_id=LATEST_FAILED_RUN_ID, ad_metadata={}, error_message=None, error_type=None, task_id=f"manual-sync-{uuid.uuid4()}", sync_status=UpdateStatus.PENDING.value, ) last_exc = None for attempt in range(1, MAX_TOTAL_ATTEMPTS + 1): close_old_connections() try: print(f"[creative] {row.ad_id}: attempt {attempt}/{MAX_TOTAL_ATTEMPTS}") result = fetch_ad_creative_data_task.run( row.ad_id, str(account.id), user_id, account.platform, "manual-sync", ) print(f"[creative] {row.ad_id}: {result}") return result except RateLimitExceeded as exc: last_exc = exc wait_seconds = max(float(getattr(exc, "retry_after", 30) or 30), 1.0) print(f"[creative] {row.ad_id}: rate limited, sleeping {wait_seconds:.1f}s") time.sleep(wait_seconds) except Exception as exc: last_exc = exc if attempt < MAX_TOTAL_ATTEMPTS: wait_seconds = 10 * attempt print(f"[creative] {row.ad_id}: error {type(exc).__name__}: {exc}; sleeping {wait_seconds}s") time.sleep(wait_seconds) else: print(f"[creative] {row.ad_id}: final error {type(exc).__name__}: {exc}") AdCreativeData.objects.filter(id=row.id).update( error_message=f"Manual sync error after {MAX_TOTAL_ATTEMPTS - 1} retries: {last_exc}", ad_overview="-", sync_status=UpdateStatus.FAILED.value, ) return { "account_id": str(account.id), "ad_id": row.ad_id, "status": "FAILED", "error_message": str(last_exc), } def generate_default_overviews(account, synced_ad_ids): rows = list( AdCreativeData.objects.filter( account=account, ad_id__in=synced_ad_ids, sync_status=UpdateStatus.SUCCESS.value, creative_id__isnull=False, ).select_related("creative") ) if not rows: print("[overview] no successfully materialized creatives with creative_id") return [] prompts = list( CreativeOverviewPrompt.objects.filter( account=account, is_active=True, is_default=True, ) ) if not prompts: print("[overview] no active default overview prompts") return [] results = [] seen_creative_ids = set() for row in rows: creative = row.creative if not creative or creative.id in seen_creative_ids: continue seen_creative_ids.add(creative.id) prompt_ids = [ str(prompt.id) for prompt in prompts if prompt.creative_type is None or prompt.creative_type == creative.type ] if not prompt_ids: print(f"[overview] {creative.id}: no compatible default prompts for type={creative.type}") continue for prompt_id in prompt_ids: overview, created = CreativeOverview.objects.get_or_create( creative=creative, overview_prompt_id=prompt_id, defaults={"sync_status": UpdateStatus.PENDING.value}, ) if not created and overview.sync_status != UpdateStatus.SUCCESS.value: overview.sync_status = UpdateStatus.PENDING.value overview.error_message = None overview.save(update_fields=["sync_status", "error_message"]) close_old_connections() print(f"[overview] {creative.id}: generating {len(prompt_ids)} default overviews") result = generate_creative_overviews_batch.run(str(creative.id), prompt_ids) print(f"[overview] {creative.id}: {result}") results.append(result) return results def main(): account = Account.objects.get(id=ACCOUNT_ID) latest_run = AccountSyncLog.objects.get(id=LATEST_FAILED_RUN_ID, account=account) user_id = latest_run.triggered_by_id qs = ( AdCreativeData.objects.filter( account=account, ad_id__in=TARGET_AD_IDS, account_sync_log_id=STALE_RUN_ID, sync_status=UpdateStatus.PENDING.value, ) .order_by("ad_id") ) rows = list(qs) found_ad_ids = [row.ad_id for row in rows] missing = sorted(set(TARGET_AD_IDS) - set(found_ad_ids)) extra = sorted(set(found_ad_ids) - set(TARGET_AD_IDS)) print(f"Account: {account.ad_account_name} ({account.id})") print(f"User: {user_id}") print(f"Rows found: {len(rows)}") if missing: raise RuntimeError(f"Missing expected pending rows: {missing}") if extra: raise RuntimeError(f"Unexpected rows selected: {extra}") if len(rows) != len(TARGET_AD_IDS): raise RuntimeError(f"Expected {len(TARGET_AD_IDS)} rows, found {len(rows)}") started_at = timezone.now() results = [] for row in rows: results.append(run_fetch_sync(row, account, user_id)) status_rows = list( AdCreativeData.objects.filter(account=account, ad_id__in=TARGET_AD_IDS) .order_by("ad_id") .values("ad_id", "sync_status", "error_type", "creative_id", "error_message") ) print("[creative] final rows:") for status_row in status_rows: print(status_row) if GENERATE_DEFAULT_OVERVIEWS_SYNCHRONOUSLY: generate_default_overviews(account, TARGET_AD_IDS) refreshed_run = refresh_onboarding_run(latest_run) print("[run] refreshed:") print( { "id": str(refreshed_run.id), "overall_status": refreshed_run.overall_status, "current_stage": refreshed_run.current_stage, "ad_creative_sync_status": refreshed_run.ad_creative_sync_status, "stats_sync_status": refreshed_run.stats_sync_status, "ai_reports_status": refreshed_run.ai_reports_status, "started_at": started_at.isoformat(), "finished_at": refreshed_run.finished_at.isoformat() if refreshed_run.finished_at else None, } ) main()