""" Force a fresh ad-platform creative fetch and direct ClickHouse ad_creative push. Default mode is dry-run. To run: APPLY=1 UV_CACHE_DIR=/tmp/uv-cache uv run python manage.py shell < /tmp/refetch_non_auth_creatives_direct_push.py Optional env: MAX_PER_ACCOUNT=1 limit rows per account while testing ONLY_ACCOUNT_ID= restrict to one account What this does: - Targets only the non-auth creative blockers from the failed onboarding check. - Includes remaining PENDING rows. - Also includes the one-row test rows that were flipped to success/empty_url. - Clears cached ad_metadata before fetching so fetch_ad_creative_data_task must call the ad platform API again. - Runs fetch_ad_creative_data_task locally/synchronously. - Pushes successful rows directly to ClickHouse via ad_creative_sync helpers. - Does not dispatch overview generation, AI reports, or label assignment. """ import os import traceback from django.db import close_old_connections from django.utils import timezone from creative_module.models import AccountSyncLog, AdCreativeData from creative_module.tasks.ad_creative_task import fetch_ad_creative_data_task from creative_module.types import UpdateStatus from data_pipeline_module.services.ad_creative_sync import ( build_ad_creative_row, push_ad_creatives_to_clickhouse, ) from organization_auth.models import Account APPLY = 1 ONLY_ACCOUNT_ID = os.environ.get("ONLY_ACCOUNT_ID") MAX_PER_ACCOUNT = 1 TARGETS = [ { "name": "CF_EMEA_ APEX.DE_ Stellantis-Opel_DE_Automotive_CL*", "account_id": "ddc8c9a0-9cbb-4160-8fe5-e1c445da4439", "run_id": "e4880251-5ecd-4efa-9e3d-53406ec551e4", "test_ad_id": "120236024957050730", }, { "name": "CF_EMEA_ APEX.IT_GSK_IT_CPG*CL 2", "account_id": "326578b0-fd74-44eb-a25c-06c6ad23b26a", "run_id": "787fdb0b-3f3c-4bd3-8de6-f1551c1c938c", "test_ad_id": "120234520857230676", }, { "name": "CF EMEA_Brightfish.BE_Brightfish.BE_Belgium_Arts & Entertainment*CL", "account_id": "2c46b0eb-217f-4dca-912b-8288aab015a3", "run_id": "702112d6-532b-46f3-88c5-6bfcae75ba08", "test_ad_id": "120235323586060191", }, { "name": "CF_EMEA_Dentsu.HU_Zwack Inc_Hungary_Alcohol/LDA*CL", "account_id": "55b54c1d-213f-42be-9557-4fe199a2ae0c", "run_id": "7d23af7e-025d-49de-99eb-89fcc4fbabe1", "test_ad_id": "120231837700040571", }, { "name": "CF_EMEA_HandicapNL_IGM Network_Charitable Foundations*CL", "account_id": "aab443ee-7f13-404f-91d9-004afdd20d81", "run_id": "ade22646-896b-48d8-8af0-b1a8db16f949", "test_ad_id": "120233034287310500", }, { "name": "CF_EMEA_OMD.QA (974)_Qatar Central Bank_Travel & Tourism*CL", "account_id": "dadfc906-8169-4b17-931c-1cd60aed79f1", "run_id": "e8859667-436e-4bfa-b3c5-760806506762", "test_ad_id": "120240584487750136", }, { "name": "Meta_IN_Publicis-Groupe-IN_Cathay-Pacific_IN_Airline*AX-1022", "account_id": "ec4cfcf6-0a97-4c9e-ba94-af3b41ba390b", "run_id": "b65e559e-277f-4f94-8cbc-567dc4b1e410", "test_ad_id": "120229098671130773", }, { "name": "Meta_IN_Publicis-Groupe-IN_EFL_IN_Home-Appliances*AX-1022", "account_id": "bceff4f0-d150-4d68-a0d6-8c23a5bc6403", "run_id": "667de63a-72bb-474e-b275-c83f5493480d", "test_ad_id": "120230380531630674", }, { "name": "Platzi MX", "account_id": "29ec4a5f-facd-466f-83d9-2fcd14488078", "run_id": "5ae66125-74de-4654-aee9-ed62f7475b15", "test_ad_id": "1862914546399265", }, ] def rows_for_target(target): pending_qs = ( AdCreativeData.objects .filter(account_id=target["account_id"], sync_status=UpdateStatus.PENDING.value) .exclude(ad_id__isnull=True) .order_by("updated_at", "id") ) if MAX_PER_ACCOUNT: pending_rows = list(pending_qs[:MAX_PER_ACCOUNT]) else: pending_rows = list(pending_qs) test_rows = list( AdCreativeData.objects.filter( account_id=target["account_id"], ad_id=target["test_ad_id"], ) ) rows_by_id = {row.id: row for row in pending_rows} for row in test_rows: rows_by_id[row.id] = row return list(rows_by_id.values()) def reset_for_platform_refetch(row): AdCreativeData.objects.filter(id=row.id).update( ad_metadata={}, ad_overview=None, error_message=None, error_type=None, pixel_hash=None, cropped_hash=None, text_hash=None, creative_id=None, task_id=None, ch_pushed_at=None, sync_status=UpdateStatus.PENDING.value, updated_at=timezone.now(), ) def push_row_to_clickhouse(row_id): row = AdCreativeData.objects.select_related("creative").get(id=row_id) if row.sync_status != UpdateStatus.SUCCESS.value: return 0, "not_success" ch_row = build_ad_creative_row(row) if not ch_row: return 0, "no_buildable_row" inserted = push_ad_creatives_to_clickhouse([ch_row]) if inserted: AdCreativeData.objects.filter(id=row.id).update(ch_pushed_at=timezone.now()) return inserted, "pushed" return 0, "push_returned_0" total_targeted = 0 total_success = 0 total_pushed = 0 total_failed = 0 print(f"mode={'APPLY' if APPLY else 'DRY_RUN'}") print("direct_fetch=1 direct_ch_push=1 labels=0 overviews=0") for target in TARGETS: if ONLY_ACCOUNT_ID and target["account_id"] != ONLY_ACCOUNT_ID: continue account = Account.objects.get(id=target["account_id"]) run = AccountSyncLog.objects.get(id=target["run_id"], account=account) rows = rows_for_target(target) total_targeted += len(rows) print() print(f"{target['name']} ({account.id})") print(f" run={run.id} platform={account.platform} rows={len(rows)}") print(f" ad_ids={', '.join(str(row.ad_id) for row in rows[:12])}") if not APPLY: continue for row in rows: print(f" refetch ad_id={row.ad_id}") try: reset_for_platform_refetch(row) close_old_connections() result = fetch_ad_creative_data_task.apply( args=[ str(row.ad_id), str(account.id), str(run.triggered_by_id), account.platform, ], throw=False, ) if result.failed(): total_failed += 1 print(f" fetch_failed={result.result}") continue refreshed = AdCreativeData.objects.get(id=row.id) if refreshed.sync_status == UpdateStatus.SUCCESS.value: total_success += 1 inserted, push_status = push_row_to_clickhouse(row.id) total_pushed += inserted refreshed = AdCreativeData.objects.get(id=row.id) print( " fetch_status=%s creative_id=%s error_type=%s push_status=%s ch_pushed_at=%s" % ( refreshed.sync_status, refreshed.creative_id, refreshed.error_type, push_status, refreshed.ch_pushed_at, ) ) except Exception as exc: total_failed += 1 print(f" exception={type(exc).__name__}: {exc}") traceback.print_exc() print() print(f"total_targeted={total_targeted}") if APPLY: print(f"total_success={total_success}") print(f"total_pushed={total_pushed}") print(f"total_failed={total_failed}") else: print("dry_run_done=1")