#!/usr/bin/env python3 """Meta Ads CLI - Meta Marketing API 관리 도구""" import argparse import json import sys from datetime import date, timedelta from typing import NoReturn sys.path.insert(0, "/home/jay/workspace") from utils.meta_ads_client import MetaAdsClient # type: ignore[import-not-found] # --------------------------------------------------------------------------- # 헬퍼 # --------------------------------------------------------------------------- def mask_token(token: str) -> str: """토큰 앞 8자 + '...' + 뒤 4자 마스킹""" if not token: return "(empty)" if len(token) <= 12: return "*" * len(token) return token[:8] + "..." + token[-4:] def error_exit(message: str) -> NoReturn: """에러 메시지를 stderr로 출력하고 exit 1""" print(f"Error: {message}", file=sys.stderr) sys.exit(1) def print_json(data) -> None: """JSON 형식으로 출력""" print(json.dumps(data, indent=2, ensure_ascii=False)) def days_to_time_range(days: int) -> dict: """--days 값을 Meta API time_range dict으로 변환""" until = date.today() since = until - timedelta(days=days - 1) return { "since": since.strftime("%Y-%m-%d"), "until": until.strftime("%Y-%m-%d"), } # --------------------------------------------------------------------------- # 명령어 핸들러 # --------------------------------------------------------------------------- def cmd_status(client: MetaAdsClient, args: argparse.Namespace) -> None: """status: 토큰 유효성 + 계정 정보 출력""" try: token_info = client.check_token() except Exception as e: error_exit(f"토큰 정보 조회 실패: {e}") try: account_info = client.get_account_info() except Exception as e: error_exit(f"계정 정보 조회 실패: {e}") # 토큰 값 노출 방지: user_id, app_id 등 식별자는 노출해도 되지만 # access_token 필드가 혹시 포함된 경우 마스킹 safe_token = { k: (mask_token(v) if k == "access_token" and isinstance(v, str) else v) for k, v in token_info.items() } data = { "token": safe_token, "account": account_info, } print_json(data) # --------------------------------------------------------------------------- def cmd_campaigns_list(client: MetaAdsClient, args: argparse.Namespace) -> None: """campaigns list: 캠페인 목록 출력""" limit: int = getattr(args, "limit", 25) try: campaigns = client.list_campaigns(limit=limit) except Exception as e: error_exit(f"캠페인 목록 조회 실패: {e}") print_json(campaigns) def cmd_campaigns_create(client: MetaAdsClient, args: argparse.Namespace) -> None: """campaigns create: 캠페인 생성 (status=PAUSED 고정)""" try: result = client.create_campaign( name=args.name, objective=args.objective, status="PAUSED", daily_budget=args.budget, ) except Exception as e: error_exit(f"캠페인 생성 실패: {e}") print_json(result) def cmd_campaigns_get(client: MetaAdsClient, args: argparse.Namespace) -> None: """campaigns get: 단일 캠페인 조회""" try: result = client.get_campaign(campaign_id=args.id) except Exception as e: error_exit(f"캠페인 조회 실패: {e}") print_json(result) def cmd_campaigns_delete(client: MetaAdsClient, args: argparse.Namespace) -> None: """campaigns delete: 캠페인 삭제""" try: success = client.delete_campaign(campaign_id=args.id) except Exception as e: error_exit(f"캠페인 삭제 실패: {e}") print_json({"success": success, "campaign_id": args.id}) # --------------------------------------------------------------------------- def cmd_upload_image(client: MetaAdsClient, args: argparse.Namespace) -> None: """upload-image: 이미지 업로드 후 hash 반환""" try: image_hash = client.upload_image(image_path=args.filepath) except FileNotFoundError as e: error_exit(str(e)) except Exception as e: error_exit(f"이미지 업로드 실패: {e}") print_json({"hash": image_hash, "filepath": args.filepath}) # --------------------------------------------------------------------------- def cmd_insights(client: MetaAdsClient, args: argparse.Namespace) -> None: """insights: 광고 성과 인사이트 조회""" time_range = days_to_time_range(args.days) try: results = client.get_insights( object_id=args.object_id, object_type=args.object_type, time_range=time_range, ) except Exception as e: error_exit(f"인사이트 조회 실패: {e}") print_json(results) # --------------------------------------------------------------------------- def cmd_token_refresh(client: MetaAdsClient, args: argparse.Namespace) -> None: """token refresh: 장기 토큰 교환 후 .env.keys 업데이트""" try: new_token = client.exchange_token() except Exception as e: error_exit(f"토큰 교환 실패: {e}") try: client.update_env_token(new_token) except Exception as e: error_exit(f".env.keys 토큰 업데이트 실패: {e}") print("장기 토큰 교환 완료") print_json({"status": "success", "token": mask_token(new_token)}) # --------------------------------------------------------------------------- # 파서 구성 # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Meta Ads CLI - Meta Marketing API 관리 도구", formatter_class=argparse.RawDescriptionHelpFormatter, ) subparsers = parser.add_subparsers(dest="command", metavar="COMMAND") # ---- status ---- subparsers.add_parser( "status", help="토큰 유효성 및 계정 정보 확인", ) # ---- campaigns ---- campaigns_parser = subparsers.add_parser( "campaigns", help="캠페인 관리", ) campaigns_sub = campaigns_parser.add_subparsers(dest="action", metavar="ACTION") # campaigns list list_p = campaigns_sub.add_parser("list", help="캠페인 목록 조회") list_p.add_argument("--limit", type=int, default=25, help="최대 결과 수 (기본: 25)") # campaigns create create_p = campaigns_sub.add_parser("create", help="캠페인 생성") create_p.add_argument("--name", required=True, help="캠페인 이름") create_p.add_argument( "--objective", required=True, help="캠페인 목표 (예: OUTCOME_TRAFFIC, OUTCOME_AWARENESS)", ) create_p.add_argument("--budget", type=int, required=True, help="일일 예산 (센트/원 단위)") # campaigns get get_p = campaigns_sub.add_parser("get", help="단일 캠페인 조회") get_p.add_argument("--id", required=True, help="캠페인 ID") # campaigns delete delete_p = campaigns_sub.add_parser("delete", help="캠페인 삭제") delete_p.add_argument("--id", required=True, help="캠페인 ID") # ---- upload-image ---- upload_parser = subparsers.add_parser( "upload-image", help="이미지 파일 업로드", ) upload_parser.add_argument("filepath", help="업로드할 이미지 파일 경로") # ---- insights ---- insights_parser = subparsers.add_parser( "insights", help="광고 성과 인사이트 조회", ) insights_parser.add_argument( "--object-id", dest="object_id", required=True, help="조회할 캠페인/광고세트/광고 ID", ) insights_parser.add_argument( "--object-type", dest="object_type", default="campaign", choices=["campaign", "adset", "ad"], help="오브젝트 타입 (기본: campaign)", ) insights_parser.add_argument( "--days", type=int, default=7, help="조회 기간(일수), since/until 으로 변환됨 (기본: 7)", ) # ---- token ---- token_parser = subparsers.add_parser( "token", help="토큰 관리", ) token_sub = token_parser.add_subparsers(dest="action", metavar="ACTION") token_sub.add_parser("refresh", help="장기 액세스 토큰으로 교환 및 .env.keys 업데이트") return parser # --------------------------------------------------------------------------- # 디스패처 # --------------------------------------------------------------------------- def dispatch(client: MetaAdsClient, args: argparse.Namespace) -> None: command = args.command if command == "status": cmd_status(client, args) elif command == "campaigns": action = getattr(args, "action", None) if action == "list": cmd_campaigns_list(client, args) elif action == "create": cmd_campaigns_create(client, args) elif action == "get": cmd_campaigns_get(client, args) elif action == "delete": cmd_campaigns_delete(client, args) else: print( "campaigns 서브 명령어를 지정하세요: list | create | get | delete", file=sys.stderr, ) sys.exit(1) elif command == "upload-image": cmd_upload_image(client, args) elif command == "insights": cmd_insights(client, args) elif command == "token": action = getattr(args, "action", None) if action == "refresh": cmd_token_refresh(client, args) else: print("token 서브 명령어를 지정하세요: refresh", file=sys.stderr) sys.exit(1) else: print("명령어를 지정하세요. --help 로 도움말을 확인하세요.", file=sys.stderr) sys.exit(1) # --------------------------------------------------------------------------- # 진입점 # --------------------------------------------------------------------------- def main() -> None: parser = build_parser() args = parser.parse_args() if not args.command: parser.print_help() sys.exit(0) try: client = MetaAdsClient() except Exception as e: msg = str(e) print(f"Error: MetaAdsClient 초기화 실패 - {msg}", file=sys.stderr) sys.exit(1) dispatch(client, args) if __name__ == "__main__": main()