#!/usr/bin/env python3 """ Meta Marketing API - 크리에이티브 9개 + 광고 18개 생성 스크립트 기존 캠페인/광고세트/이미지 해시를 활용하여 크리에이티브와 광고를 생성한다. 담당: 엔키 (개발5팀) 생성일: 2026-04-06 관련 태스크: task-1491.2 (task-1491.1에서 생성된 광고세트/이미지해시 사용) 실행: python meta_creative_ad_setup.py python meta_creative_ad_setup.py --dry-run """ import argparse import json import os import sys from datetime import datetime from pathlib import Path # 프로젝트 루트를 sys.path에 추가 _WORKSPACE = Path("/home/jay/workspace") if str(_WORKSPACE) not in sys.path: sys.path.insert(0, str(_WORKSPACE)) from utils.env_loader import load_env_keys # type: ignore[import-not-found] from utils.meta_ads_client import MetaAdsClient # type: ignore[import-not-found] # facebook_business SDK (Ad 직접 생성에 필요) try: import requests from facebook_business.adobjects.ad import Ad from facebook_business.adobjects.adaccount import AdAccount except ImportError as e: print(f"[ERROR] 필수 패키지를 찾을 수 없습니다: {e}") sys.exit(1) # --------------------------------------------------------------------------- # 상수 정의 # --------------------------------------------------------------------------- _ENV_KEYS_PATH = "/home/jay/workspace/.env.keys" _GRAPH_API_BASE = "https://graph.facebook.com/v25.0" OUTPUT_DIR = Path("/home/jay/workspace/teams/dev5-team/output") OUTPUT_FILE = OUTPUT_DIR / "meta_creative_ad_result.json" LANDING_URL = "https://incar-top1.tistory.com" # --------------------------------------------------------------------------- # 기존 캠페인 ID (이미 생성됨 — 새로 생성하지 않음) # --------------------------------------------------------------------------- CAMPAIGN_IDS = { "리드": "120244493213800104", # 리쿠르팅_리드_2026Q2 "잠재고객": "120244493213880104", # 리쿠르팅_잠재고객_2026Q2 } # --------------------------------------------------------------------------- # 기존 광고세트 ID (이미 생성됨 — 새로 생성하지 않음) # --------------------------------------------------------------------------- # (캠페인유형, 브랜드) → 광고세트 ID ADSET_MAP: dict[tuple[str, str], str] = { ("리드", "인카금융서비스"): "120244493234600104", ("리드", "GA"): "120244493237840104", ("리드", "서울대보험쌤"): "120244493238350104", ("잠재고객", "인카금융서비스"): "120244493218590104", ("잠재고객", "GA"): "120244493230370104", ("잠재고객", "서울대보험쌤"): "120244493230950104", } # --------------------------------------------------------------------------- # 이미지 해시 9개 (이미 업로드됨) # --------------------------------------------------------------------------- # (셀번호, 셀디렉토리명, 브랜드, 앵글한글) CELL_INFO: list[tuple[int, str, str, str]] = [ (1, "cell-1-incar-fair", "인카금융서비스", "공정한수수료"), (2, "cell-2-incar-leader", "인카금융서비스", "리더지원"), (3, "cell-3-incar-support", "인카금융서비스", "업무지원"), (4, "cell-4-ga-fair", "GA", "공정한수수료"), (5, "cell-5-ga-leader", "GA", "리더지원"), (6, "cell-6-ga-support", "GA", "업무지원"), (7, "cell-7-snu-fair", "서울대보험쌤", "공정한수수료"), (8, "cell-8-snu-leader", "서울대보험쌤", "리더지원"), (9, "cell-9-snu-support", "서울대보험쌤", "업무지원"), ] IMAGE_HASH_MAP: dict[str, str] = { "cell-1-incar-fair": "431704f93354d64849ca9b950b1a58f8", "cell-2-incar-leader": "860a2df81739d8311146d028bb43aa8b", "cell-3-incar-support": "2639214e727e766249d3d72c7745201b", "cell-4-ga-fair": "b358438a4a136d6173dc4dacad1c7aa4", "cell-5-ga-leader": "d0ed5f2ebdea82adf9bcc4b6d99b938b", "cell-6-ga-support": "03762a14491ff58b4d8a4ccbdff978c5", "cell-7-snu-fair": "d7b07745b21c842034957a737de0ad89", "cell-8-snu-leader": "8e43730baea0377fdbfde0bb9fef06d0", "cell-9-snu-support": "0d3c288465dd9e6a93003c9bf7d19aad", } # --------------------------------------------------------------------------- # 페이지 ID 매핑 (env에서 로드) # --------------------------------------------------------------------------- # META_PAGE_ID_RECRUIT → 인카금융서비스(Cell 1,2,3) + GA(Cell 4,5,6) # META_PAGE_ID_SNU → 서울대보험쌤(Cell 7,8,9) BRAND_TO_PAGE_ENV: dict[str, str] = { "인카금융서비스": "META_PAGE_ID_RECRUIT", "GA": "META_PAGE_ID_RECRUIT", "서울대보험쌤": "META_PAGE_ID_SNU", } # --------------------------------------------------------------------------- # 브랜드별 광고 메시지 # --------------------------------------------------------------------------- BRAND_MESSAGES: dict[str, str] = { "인카금융서비스": "인카금융서비스와 함께 성공적인 보험 커리어를 시작하세요! 경력직/신입 모두 환영합니다.", "GA": "GA(General Agency)에서 당신의 보험 전문가 역량을 키워보세요. 지금 바로 지원하세요!", "서울대보험쌤": "서울대 출신 보험 전문가들과 함께하는 성장의 기회! 서울대보험쌤 팀에 합류하세요.", } CAMPAIGN_TYPES = ["리드", "잠재고객"] # --------------------------------------------------------------------------- # 유틸리티 # --------------------------------------------------------------------------- def log(msg: str) -> None: """타임스탬프 포함 콘솔 로그.""" ts = datetime.now().strftime("%H:%M:%S") print(f"[{ts}] {msg}") # --------------------------------------------------------------------------- # 페이지 ID 유효성 검증 (Graph API 직접 호출) # --------------------------------------------------------------------------- def validate_page_id(page_id: str, access_token: str) -> bool: """ Graph API로 페이지 존재 여부를 확인한다. GET https://graph.facebook.com/v25.0/{page_id}?fields=id,name&access_token=... Args: page_id: 검증할 Facebook 페이지 ID access_token: Meta 액세스 토큰 Returns: bool: 페이지 유효 여부 """ url = f"{_GRAPH_API_BASE}/{page_id}" params = { "fields": "id,name", "access_token": access_token, } try: resp = requests.get(url, params=params, timeout=30) data = resp.json() if "error" in data: log(f" [ERROR] 페이지 검증 실패 (page_id={page_id}): {data['error'].get('message', data['error'])}") return False page_name = data.get("name", "(이름 없음)") log(f" 페이지 확인: id={data.get('id')} name={page_name}") return True except Exception as e: log(f" [ERROR] 페이지 검증 중 예외 발생 (page_id={page_id}): {e}") return False # --------------------------------------------------------------------------- # 광고세트 존재 여부 확인 # --------------------------------------------------------------------------- def verify_adsets(access_token: str) -> dict[tuple[str, str], bool]: """ 기존 광고세트 6개의 존재 여부를 확인한다. GET /v25.0/{adset_id}?fields=id,name,status Returns: dict: (campaign_type, brand) → 유효 여부 """ validity: dict[tuple[str, str], bool] = {} for (campaign_type, brand), adset_id in ADSET_MAP.items(): url = f"{_GRAPH_API_BASE}/{adset_id}" params = { "fields": "id,name,status", "access_token": access_token, } try: resp = requests.get(url, params=params, timeout=30) data = resp.json() if "error" in data: log( f" [WARN] 광고세트 없음 ({campaign_type}_{brand}, id={adset_id}): " f"{data['error'].get('message', data['error'])}" ) validity[(campaign_type, brand)] = False else: adset_name = data.get("name", "(이름 없음)") adset_status = data.get("status", "UNKNOWN") log(f" 광고세트 확인: {adset_name} id={data.get('id')} status={adset_status}") validity[(campaign_type, brand)] = True except Exception as e: log(f" [WARN] 광고세트 확인 중 예외 발생 ({campaign_type}_{brand}): {e}") validity[(campaign_type, brand)] = False return validity # --------------------------------------------------------------------------- # 드라이런 모드 # --------------------------------------------------------------------------- def dry_run_report() -> None: """실제 API 호출 없이 실행 구조를 출력한다.""" # .env.keys 로드 (환경변수 확인용) try: load_env_keys(_ENV_KEYS_PATH) except Exception: pass page_recruit = os.environ.get("META_PAGE_ID_RECRUIT", "(META_PAGE_ID_RECRUIT 미설정)") page_snu = os.environ.get("META_PAGE_ID_SNU", "(META_PAGE_ID_SNU 미설정)") pixel_id = os.environ.get("META_PIXEL_ID", "(META_PIXEL_ID 미설정)") print("\n" + "=" * 65) print(" DRY-RUN 모드 — 실제 API 호출 없음") print("=" * 65) print("\n[1] MetaAdsClient 초기화 + 토큰/계정 유효성 확인") print("\n[2] 페이지 ID 유효성 검증 (Graph API)") print(f" META_PAGE_ID_RECRUIT ({page_recruit})") print(f" GET {_GRAPH_API_BASE}/{{page_id}}?fields=id,name") print(f" META_PAGE_ID_SNU ({page_snu})") print(f" GET {_GRAPH_API_BASE}/{{page_id}}?fields=id,name") print(" ※ 검증 실패 시 스크립트 중단") print("\n[3] 기존 광고세트 6개 존재 확인 (GET /v25.0/{adset_id}?fields=id,name,status)") for (campaign_type, brand), adset_id in ADSET_MAP.items(): print(f" {campaign_type}_{brand} id={adset_id}") print("\n[4] 크리에이티브 9개 생성 (기존 이미지 해시 사용)") for cell_no, cell_dir, brand, angle in CELL_INFO: page_env = BRAND_TO_PAGE_ENV[brand] page_id = os.environ.get(page_env, f"({page_env} 미설정)") creative_name = f"리쿠르팅_Cell{cell_no}_{brand}_{angle}" img_hash = IMAGE_HASH_MAP[cell_dir] print(f" Cell{cell_no} {creative_name}") print(f" image_hash={img_hash} page_id={page_id}") print("\n[5] 광고 18개 생성 (각 광고세트 × 3개 크리에이티브)") for campaign_type in CAMPAIGN_TYPES: for brand in ["인카금융서비스", "GA", "서울대보험쌤"]: adset_id = ADSET_MAP[(campaign_type, brand)] brand_cells = [(no, d, b, a) for no, d, b, a in CELL_INFO if b == brand] for cell_no, _, _, _ in brand_cells: ad_name = f"{campaign_type}_{brand}_Cell{cell_no}" print(f" {ad_name} adset_id={adset_id}") print("\n[6] 결과 JSON 저장") print(f" → {OUTPUT_FILE}") print("\n환경변수 현황:") print(f" META_PAGE_ID_RECRUIT = {page_recruit}") print(f" META_PAGE_ID_SNU = {page_snu}") print(f" META_PIXEL_ID = {pixel_id}") print("\n" + "=" * 65 + "\n") # --------------------------------------------------------------------------- # 헬퍼: Ad 직접 생성 (AdAccount.create_ad 사용) # --------------------------------------------------------------------------- def _create_ad( ad_account: AdAccount, adset_id: str, creative_id: str, name: str, pixel_id: str, ) -> dict: """ facebook_business SDK의 AdAccount.create_ad를 사용하여 광고를 생성한다. MetaAdsClient에 create_ad 메서드가 없으므로 SDK를 직접 사용한다. """ params = { Ad.Field.name: name, Ad.Field.adset_id: adset_id, Ad.Field.creative: {"creative_id": creative_id}, Ad.Field.status: "PAUSED", Ad.Field.tracking_specs: [ { "action.type": ["offsite_conversion"], "fb_pixel": [pixel_id], } ], } ad_result = ad_account.create_ad(fields=[], params=params) data: dict = getattr(ad_result, "export_all_data", lambda: dict(ad_result))() # type: ignore[arg-type] return data # --------------------------------------------------------------------------- # 결과 저장 # --------------------------------------------------------------------------- def _save(result: dict) -> None: """결과를 JSON 파일로 저장한다.""" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) with open(OUTPUT_FILE, "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) log(f"결과 저장 완료: {OUTPUT_FILE}") # --------------------------------------------------------------------------- # 메인 로직 # --------------------------------------------------------------------------- def run() -> None: """크리에이티브 9개 + 광고 18개를 생성한다.""" result: dict = { "timestamp": datetime.now().isoformat(), "account_id": "", "page_validation": {}, "adset_validation": {}, "creatives": [], "ads": [], "errors": [], } # ------------------------------------------------------------------ # 1. MetaAdsClient 초기화 + 토큰/계정 유효성 확인 # ------------------------------------------------------------------ log("[1] MetaAdsClient 초기화 중...") try: client = MetaAdsClient() except Exception as e: log(f"[ERROR] 클라이언트 초기화 실패: {e}") result["errors"].append({"step": "init", "error": str(e)}) _save(result) sys.exit(1) # 토큰 유효성 확인 access_token = os.environ.get("META_ACCESS_TOKEN", "") try: token_info = client.check_token() if not token_info.get("is_valid", False): log("[ERROR] 액세스 토큰이 유효하지 않습니다.") result["errors"].append({"step": "check_token", "error": "is_valid=False"}) _save(result) sys.exit(1) log(f" 토큰 유효 확인 (app_id={token_info.get('app_id')})") except Exception as e: log(f"[WARN] 토큰 유효성 확인 실패 (계속 진행): {e}") result["errors"].append({"step": "check_token", "error": str(e)}) # 계정 정보 확인 try: account_info = client.get_account_info() account_id = account_info.get("id", os.environ.get("META_AD_ACCOUNT_ID", "")) if not account_id.startswith("act_"): account_id = f"act_{account_id}" result["account_id"] = account_id log( f" 계정 확인: {account_id} 통화={account_info.get('currency', 'USD')} " f"상태={account_info.get('account_status')}" ) except Exception as e: log(f"[ERROR] 계정 정보 조회 실패: {e}") result["errors"].append({"step": "get_account_info", "error": str(e)}) _save(result) sys.exit(1) # SDK AdAccount 인스턴스 (Ad 직접 생성용) ad_account = AdAccount(result["account_id"]) # 환경변수 로드 pixel_id = os.environ.get("META_PIXEL_ID", "1461062562329883") page_recruit = os.environ.get("META_PAGE_ID_RECRUIT", "") page_snu = os.environ.get("META_PAGE_ID_SNU", "") if not page_recruit: log("[WARN] META_PAGE_ID_RECRUIT 환경변수가 설정되지 않았습니다.") if not page_snu: log("[WARN] META_PAGE_ID_SNU 환경변수가 설정되지 않았습니다.") # ------------------------------------------------------------------ # 2. 페이지 ID 유효성 검증 (Graph API) # ------------------------------------------------------------------ log("[2] 페이지 ID 유효성 검증 중...") page_ids_to_check: dict[str, str] = {} if page_recruit: page_ids_to_check["META_PAGE_ID_RECRUIT"] = page_recruit if page_snu: page_ids_to_check["META_PAGE_ID_SNU"] = page_snu page_valid: dict[str, bool] = {} for env_key, pid in page_ids_to_check.items(): is_valid = validate_page_id(pid, access_token) page_valid[env_key] = is_valid result["page_validation"][env_key] = {"page_id": pid, "valid": is_valid} if not is_valid: log(f"[ERROR] 페이지 ID 검증 실패 ({env_key}={pid}). 스크립트를 중단합니다.") result["errors"].append( {"step": "validate_page", "env_key": env_key, "page_id": pid, "error": "페이지 검증 실패"} ) _save(result) sys.exit(1) # 브랜드 → 페이지 ID 런타임 매핑 brand_page_map: dict[str, str] = { "인카금융서비스": page_recruit, "GA": page_recruit, "서울대보험쌤": page_snu, } # ------------------------------------------------------------------ # 3. 기존 광고세트 6개 존재 여부 확인 # ------------------------------------------------------------------ log("[3] 기존 광고세트 존재 여부 확인 중 (6개)...") adset_validity = verify_adsets(access_token) result["adset_validation"] = { f"{ct}_{brand}": {"adset_id": ADSET_MAP[(ct, brand)], "valid": valid} for (ct, brand), valid in adset_validity.items() } invalid_adsets = [(k, v) for k, v in adset_validity.items() if not v] if invalid_adsets: for (ct, brand), _ in invalid_adsets: log(f" [WARN] 광고세트 조회 실패: {ct}_{brand} (id={ADSET_MAP[(ct, brand)]})") # ------------------------------------------------------------------ # 4. 크리에이티브 9개 생성 # ------------------------------------------------------------------ log("[4] 크리에이티브 생성 (9개)...") # cell_dir → creative_id creative_map: dict[str, str] = {} for cell_no, cell_dir, brand, angle in CELL_INFO: image_hash = IMAGE_HASH_MAP[cell_dir] page_id = brand_page_map.get(brand, "") creative_name = f"리쿠르팅_Cell{cell_no}_{brand}_{angle}" message = BRAND_MESSAGES[brand] if not page_id: log(f" [SKIP] 크리에이티브 스킵 (Cell{cell_no}): page_id 없음 (브랜드={brand})") result["errors"].append( { "step": "create_creative", "cell": cell_dir, "error": f"page_id 없음 (브랜드={brand})", } ) continue try: creative = client.create_creative( name=creative_name, image_hash=image_hash, page_id=page_id, message=message, link=LANDING_URL, ) creative_id = creative.get("id", "") creative_map[cell_dir] = creative_id result["creatives"].append( { "name": creative_name, "id": creative_id, "cell": cell_dir, "brand": brand, "angle": angle, "image_hash": image_hash, "page_id": page_id, } ) log(f" 크리에이티브 생성 완료: {creative_name} id={creative_id}") except Exception as e: log(f" [ERROR] 크리에이티브 생성 실패 ({creative_name}): {e}") result["errors"].append( { "step": "create_creative", "cell": cell_dir, "name": creative_name, "error": str(e), } ) # ------------------------------------------------------------------ # 5. 광고 18개 생성 (각 광고세트 × 3개 크리에이티브) # ------------------------------------------------------------------ log("[5] 광고 생성 (18개 — 각 광고세트에 3개씩)...") ad_count = 0 for campaign_type in CAMPAIGN_TYPES: for brand in ["인카금융서비스", "GA", "서울대보험쌤"]: adset_id = ADSET_MAP.get((campaign_type, brand)) if not adset_id: log(f" [SKIP] 광고 스킵: 광고세트 없음 ({campaign_type}_{brand})") continue # 해당 브랜드의 셀 목록 (Cell 순서대로 순차 처리) brand_cells = [(no, cell_dir, b, angle) for no, cell_dir, b, angle in CELL_INFO if b == brand] for cell_no, cell_dir, _, _ in brand_cells: creative_id = creative_map.get(cell_dir) if not creative_id: log(f" [SKIP] 광고 스킵: 크리에이티브 없음 ({cell_dir})") result["errors"].append( { "step": "create_ad", "cell": cell_dir, "campaign_type": campaign_type, "brand": brand, "error": "creative_id 없음 (크리에이티브 생성 실패)", } ) continue ad_name = f"{campaign_type}_{brand}_Cell{cell_no}" try: ad_result = _create_ad( ad_account=ad_account, adset_id=adset_id, creative_id=creative_id, name=ad_name, pixel_id=pixel_id, ) ad_id = ad_result.get("id", "") result["ads"].append( { "name": ad_name, "id": ad_id, "campaign_type": campaign_type, "brand": brand, "cell_no": cell_no, "adset_id": adset_id, "creative_id": creative_id, } ) ad_count += 1 log(f" 광고 생성 완료: {ad_name} id={ad_id}") except Exception as e: log(f" [ERROR] 광고 생성 실패 ({ad_name}): {e}") result["errors"].append( { "step": "create_ad", "name": ad_name, "adset_id": adset_id, "creative_id": creative_id, "error": str(e), } ) log(f" 총 {ad_count}개 광고 생성 완료") # ------------------------------------------------------------------ # 6. 결과 JSON 저장 # ------------------------------------------------------------------ log("[6] 결과 저장 중...") _save(result) # 요약 출력 log("=" * 55) log("실행 완료 요약") log(f" 계정: {result['account_id']}") log(f" 크리에이티브: {len(result['creatives'])}개 생성") log(f" 광고: {len(result['ads'])}개 생성") log(f" 오류: {len(result['errors'])}건") log(f" 결과 파일: {OUTPUT_FILE}") log("=" * 55) # --------------------------------------------------------------------------- # 진입점 # --------------------------------------------------------------------------- def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Meta Marketing API 크리에이티브 9개 + 광고 18개 생성 스크립트", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument( "--dry-run", action="store_true", default=False, help="실제 API 호출 없이 실행 구조만 출력한다.", ) return parser.parse_args() if __name__ == "__main__": args = parse_args() if args.dry_run: dry_run_report() sys.exit(0) run()