#!/usr/bin/env python3
# pyright: reportAttributeAccessIssue=false, reportIndexIssue=false
"""
Google Ads API 클라이언트 모듈

Usage:
    from utils.google_ads_client import GoogleAdsClient
    client = GoogleAdsClient()
    campaigns = client.list_campaigns()
"""

import os
from typing import Any

from google.ads.googleads.client import GoogleAdsClient as _GoogleAdsClient
from google.ads.googleads.errors import GoogleAdsException

from utils.env_loader import load_env_keys  # type: ignore[import-not-found]
from utils.logger import get_logger  # type: ignore[import-not-found]

logger = get_logger(__name__)

_ENV_KEYS_PATH = "/home/jay/workspace/.env.keys"

_DEFAULT_INSIGHT_METRICS = [
    "metrics.impressions",
    "metrics.clicks",
    "metrics.cost_micros",
    "metrics.ctr",
    "metrics.average_cpc",
    "metrics.conversions",
]


class GoogleAdsClient:
    """Google Ads API 클라이언트"""

    def __init__(self, env_keys_path: str = _ENV_KEYS_PATH) -> None:
        """
        GoogleAdsClient 초기화.

        Args:
            env_keys_path: .env.keys 파일 경로 (기본값: /home/jay/workspace/.env.keys)

        Raises:
            ValueError: 필수 환경변수가 누락된 경우
        """
        self._env_keys_path = env_keys_path
        load_env_keys(env_keys_path)

        required_keys = [
            "GOOGLE_ADS_DEVELOPER_TOKEN",
            "GOOGLE_ADS_CLIENT_ID",
            "GOOGLE_ADS_CLIENT_SECRET",
            "GOOGLE_ADS_REFRESH_TOKEN",
            "GOOGLE_ADS_CUSTOMER_ID",
        ]
        missing = [k for k in required_keys if not os.environ.get(k)]
        if missing:
            raise ValueError(f"필수 환경변수 누락: {missing}")

        self._developer_token = os.environ["GOOGLE_ADS_DEVELOPER_TOKEN"]
        self._client_id = os.environ["GOOGLE_ADS_CLIENT_ID"]
        self._client_secret = os.environ["GOOGLE_ADS_CLIENT_SECRET"]
        self._refresh_token = os.environ["GOOGLE_ADS_REFRESH_TOKEN"]
        self._customer_id = os.environ["GOOGLE_ADS_CUSTOMER_ID"]

        self._client = _GoogleAdsClient.load_from_dict(
            {
                "developer_token": self._developer_token,
                "client_id": self._client_id,
                "client_secret": self._client_secret,
                "refresh_token": self._refresh_token,
                "use_proto_plus": True,
            }
        )

        logger.info("GoogleAdsClient 초기화 완료 (customer_id=%s)", self._customer_id)

    # ------------------------------------------------------------------
    # 내부 헬퍼
    # ------------------------------------------------------------------

    @staticmethod
    def _to_dict(obj: Any) -> dict:
        """SDK proto-plus 객체를 순수 dict로 변환."""
        if hasattr(obj, "__class__") and hasattr(obj, "_pb"):
            # proto-plus 객체는 type(obj).to_json() 또는 직접 dict 변환
            from google.protobuf.json_format import MessageToDict

            return MessageToDict(obj._pb, preserving_proto_field_name=True)
        if hasattr(obj, "__dict__"):
            return {k: v for k, v in obj.__dict__.items() if not k.startswith("_")}
        return dict(obj)

    def _get_ga_service(self):
        """GoogleAdsService 반환."""
        return self._client.get_service("GoogleAdsService")

    # ------------------------------------------------------------------
    # 계정 정보
    # ------------------------------------------------------------------

    def get_account_info(self) -> dict:
        """
        광고 계정 상태 및 예산 정보를 반환한다.

        GAQL로 customer 리소스를 조회하여 계정 이름, 통화, 시간대 등을 반환한다.

        Returns:
            dict: id, descriptive_name, currency_code, time_zone, auto_tagging_enabled 포함 dict

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        query = """
            SELECT
                customer.id,
                customer.descriptive_name,
                customer.currency_code,
                customer.time_zone,
                customer.auto_tagging_enabled,
                customer.status
            FROM customer
            LIMIT 1
        """
        try:
            ga_service = self._get_ga_service()
            response = ga_service.search(customer_id=self._customer_id, query=query)
            rows = list(response)
            if not rows:
                return {}
            row = rows[0]
            customer = row.customer
            result = {
                "id": str(customer.id),
                "descriptive_name": customer.descriptive_name,
                "currency_code": customer.currency_code,
                "time_zone": customer.time_zone,
                "auto_tagging_enabled": customer.auto_tagging_enabled,
                "status": customer.status.name,
            }
            logger.debug("계정 정보 조회 완료 (customer_id=%s)", self._customer_id)
            return result
        except GoogleAdsException as e:
            logger.error("계정 정보 조회 실패: %s", e)
            raise

    # ------------------------------------------------------------------
    # 캠페인 CRUD
    # ------------------------------------------------------------------

    def list_campaigns(self, limit: int = 25) -> list[dict]:
        """
        캠페인 목록을 반환한다.

        Args:
            limit: 최대 반환 개수 (기본값 25)

        Returns:
            list[dict]: 캠페인 정보 목록 (id, name, status, advertising_channel_type,
                        start_date, end_date 포함)

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        query = f"""
            SELECT
                campaign.id,
                campaign.name,
                campaign.status,
                campaign.advertising_channel_type,
                campaign.start_date,
                campaign.end_date,
                campaign_budget.amount_micros
            FROM campaign
            ORDER BY campaign.id
            LIMIT {limit}
        """
        try:
            ga_service = self._get_ga_service()
            response = ga_service.search(customer_id=self._customer_id, query=query)
            result = []
            for row in response:
                campaign = row.campaign
                result.append(
                    {
                        "id": str(campaign.id),
                        "name": campaign.name,
                        "status": campaign.status.name,
                        "advertising_channel_type": campaign.advertising_channel_type.name,
                        "start_date": campaign.start_date,
                        "end_date": campaign.end_date,
                        "budget_amount_micros": row.campaign_budget.amount_micros,
                    }
                )
            logger.debug("캠페인 목록 조회 완료: %d건", len(result))
            return result
        except GoogleAdsException as e:
            logger.error("캠페인 목록 조회 실패: %s", e)
            raise

    def get_campaign(self, campaign_id: str) -> dict:
        """
        단일 캠페인 정보를 반환한다.

        Args:
            campaign_id: 캠페인 ID

        Returns:
            dict: 캠페인 상세 정보

        Raises:
            GoogleAdsException: API 호출 실패 시
            ValueError: 해당 캠페인을 찾을 수 없는 경우
        """
        query = f"""
            SELECT
                campaign.id,
                campaign.name,
                campaign.status,
                campaign.advertising_channel_type,
                campaign.start_date,
                campaign.end_date,
                campaign_budget.amount_micros
            FROM campaign
            WHERE campaign.id = {campaign_id}
            LIMIT 1
        """
        try:
            ga_service = self._get_ga_service()
            response = ga_service.search(customer_id=self._customer_id, query=query)
            rows = list(response)
            if not rows:
                raise ValueError(f"캠페인을 찾을 수 없습니다: id={campaign_id}")
            row = rows[0]
            campaign = row.campaign
            result = {
                "id": str(campaign.id),
                "name": campaign.name,
                "status": campaign.status.name,
                "advertising_channel_type": campaign.advertising_channel_type.name,
                "start_date": campaign.start_date,
                "end_date": campaign.end_date,
                "budget_amount_micros": row.campaign_budget.amount_micros,
            }
            logger.debug("캠페인 조회 완료: id=%s", campaign_id)
            return result
        except GoogleAdsException as e:
            logger.error("캠페인 조회 실패 (id=%s): %s", campaign_id, e)
            raise

    def create_campaign(
        self,
        name: str,
        budget_amount: int,
        status: str = "PAUSED",
    ) -> dict:
        """
        새 캠페인을 생성한다. (검색 캠페인 기본값)

        예산을 먼저 생성한 뒤 캠페인에 연결한다.

        Args:
            name: 캠페인 이름
            budget_amount: 일일 예산 (마이크로 단위, 1원 = 1,000,000 micros)
            status: 초기 상태 (기본값: PAUSED)

        Returns:
            dict: 생성된 캠페인 resource_name 및 id 포함 dict

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            # 1) 캠페인 예산 생성
            budget_service = self._client.get_service("CampaignBudgetService")
            budget_operation = self._client.get_type("CampaignBudgetOperation")
            budget = budget_operation.create
            budget.name = f"{name}_budget"
            budget.amount_micros = budget_amount
            budget.delivery_method = self._client.enums.BudgetDeliveryMethodEnum.STANDARD
            budget_response = budget_service.mutate_campaign_budgets(
                customer_id=self._customer_id,
                operations=[budget_operation],
            )
            budget_resource_name = budget_response.results[0].resource_name

            # 2) 캠페인 생성
            campaign_service = self._client.get_service("CampaignService")
            campaign_operation = self._client.get_type("CampaignOperation")
            campaign = campaign_operation.create
            campaign.name = name
            campaign.status = self._client.enums.CampaignStatusEnum[status]
            campaign.advertising_channel_type = self._client.enums.AdvertisingChannelTypeEnum.SEARCH
            campaign.campaign_budget = budget_resource_name
            campaign.network_settings.target_google_search = True
            campaign.network_settings.target_search_network = True

            campaign_response = campaign_service.mutate_campaigns(
                customer_id=self._customer_id,
                operations=[campaign_operation],
            )
            resource_name = campaign_response.results[0].resource_name
            campaign_id = resource_name.split("/")[-1]

            result = {
                "id": campaign_id,
                "resource_name": resource_name,
                "name": name,
                "status": status,
                "budget_amount_micros": budget_amount,
            }
            logger.info("캠페인 생성 완료: id=%s name=%s", campaign_id, name)
            return result
        except GoogleAdsException as e:
            logger.error("캠페인 생성 실패: %s", e)
            raise

    def update_campaign(self, campaign_id: str, **params: Any) -> dict:
        """
        기존 캠페인을 업데이트한다.

        Args:
            campaign_id: 캠페인 ID
            **params: 업데이트할 필드 (name, status 등)

        Returns:
            dict: 업데이트된 resource_name 포함 dict

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            campaign_service = self._client.get_service("CampaignService")
            campaign_operation = self._client.get_type("CampaignOperation")
            campaign = campaign_operation.update
            campaign.resource_name = campaign_service.campaign_path(self._customer_id, campaign_id)

            field_mask_paths = []
            if "name" in params:
                campaign.name = params["name"]
                field_mask_paths.append("name")
            if "status" in params:
                campaign.status = self._client.enums.CampaignStatusEnum[params["status"]]
                field_mask_paths.append("status")

            from google.protobuf import field_mask_pb2

            campaign_operation.update_mask.CopyFrom(field_mask_pb2.FieldMask(paths=field_mask_paths))

            response = campaign_service.mutate_campaigns(
                customer_id=self._customer_id,
                operations=[campaign_operation],
            )
            resource_name = response.results[0].resource_name
            result = {
                "id": campaign_id,
                "resource_name": resource_name,
                "updated_fields": field_mask_paths,
            }
            logger.info("캠페인 업데이트 완료: id=%s fields=%s", campaign_id, field_mask_paths)
            return result
        except GoogleAdsException as e:
            logger.error("캠페인 업데이트 실패 (id=%s): %s", campaign_id, e)
            raise

    def delete_campaign(self, campaign_id: str) -> bool:
        """
        캠페인을 삭제한다 (상태를 REMOVED로 변경).

        Args:
            campaign_id: 삭제할 캠페인 ID

        Returns:
            bool: 삭제 성공 여부

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            campaign_service = self._client.get_service("CampaignService")
            campaign_operation = self._client.get_type("CampaignOperation")
            resource_name = campaign_service.campaign_path(self._customer_id, campaign_id)
            campaign_operation.remove = resource_name
            campaign_service.mutate_campaigns(
                customer_id=self._customer_id,
                operations=[campaign_operation],
            )
            logger.info("캠페인 삭제 완료 (REMOVED): id=%s", campaign_id)
            return True
        except GoogleAdsException as e:
            logger.error("캠페인 삭제 실패 (id=%s): %s", campaign_id, e)
            raise

    # ------------------------------------------------------------------
    # 광고그룹 CRUD
    # ------------------------------------------------------------------

    def list_ad_groups(
        self,
        campaign_id: str | None = None,
        limit: int = 25,
    ) -> list[dict]:
        """
        광고그룹 목록을 반환한다.

        Args:
            campaign_id: 지정 시 해당 캠페인의 광고그룹만 반환.
                         None이면 계정 전체 광고그룹 반환.
            limit: 최대 반환 개수 (기본값 25)

        Returns:
            list[dict]: 광고그룹 정보 목록 (id, name, status, campaign_id, cpc_bid_micros 포함)

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        where_clause = ""
        if campaign_id:
            where_clause = f"WHERE campaign.id = {campaign_id}"

        query = f"""
            SELECT
                ad_group.id,
                ad_group.name,
                ad_group.status,
                ad_group.type,
                ad_group.cpc_bid_micros,
                campaign.id,
                campaign.name
            FROM ad_group
            {where_clause}
            ORDER BY ad_group.id
            LIMIT {limit}
        """
        try:
            ga_service = self._get_ga_service()
            response = ga_service.search(customer_id=self._customer_id, query=query)
            result = []
            for row in response:
                ag = row.ad_group
                result.append(
                    {
                        "id": str(ag.id),
                        "name": ag.name,
                        "status": ag.status.name,
                        "type": ag.type_.name,
                        "cpc_bid_micros": ag.cpc_bid_micros,
                        "campaign_id": str(row.campaign.id),
                        "campaign_name": row.campaign.name,
                    }
                )
            logger.debug("광고그룹 목록 조회 완료: %d건", len(result))
            return result
        except GoogleAdsException as e:
            logger.error("광고그룹 목록 조회 실패: %s", e)
            raise

    def create_ad_group(
        self,
        campaign_id: str,
        name: str,
        cpc_bid: int,
        status: str = "PAUSED",
    ) -> dict:
        """
        광고그룹을 생성한다.

        Args:
            campaign_id: 상위 캠페인 ID
            name: 광고그룹 이름
            cpc_bid: CPC 입찰가 (마이크로 단위)
            status: 초기 상태 (기본값: PAUSED)

        Returns:
            dict: 생성된 광고그룹 id, resource_name 포함 dict

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            campaign_service = self._client.get_service("CampaignService")
            ad_group_service = self._client.get_service("AdGroupService")
            ad_group_operation = self._client.get_type("AdGroupOperation")
            ad_group = ad_group_operation.create
            ad_group.name = name
            ad_group.status = self._client.enums.AdGroupStatusEnum[status]
            ad_group.campaign = campaign_service.campaign_path(self._customer_id, campaign_id)
            ad_group.type_ = self._client.enums.AdGroupTypeEnum.SEARCH_STANDARD
            ad_group.cpc_bid_micros = cpc_bid

            response = ad_group_service.mutate_ad_groups(
                customer_id=self._customer_id,
                operations=[ad_group_operation],
            )
            resource_name = response.results[0].resource_name
            ad_group_id = resource_name.split("/")[-1]

            result = {
                "id": ad_group_id,
                "resource_name": resource_name,
                "name": name,
                "status": status,
                "campaign_id": campaign_id,
                "cpc_bid_micros": cpc_bid,
            }
            logger.info("광고그룹 생성 완료: id=%s name=%s", ad_group_id, name)
            return result
        except GoogleAdsException as e:
            logger.error("광고그룹 생성 실패: %s", e)
            raise

    def update_ad_group(self, ad_group_id: str, **params: Any) -> dict:
        """
        기존 광고그룹을 업데이트한다.

        Args:
            ad_group_id: 광고그룹 ID
            **params: 업데이트할 필드 (name, status, cpc_bid_micros 등)

        Returns:
            dict: 업데이트된 resource_name 및 수정 필드 포함 dict

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            ad_group_service = self._client.get_service("AdGroupService")
            ad_group_operation = self._client.get_type("AdGroupOperation")
            ad_group = ad_group_operation.update
            ad_group.resource_name = ad_group_service.ad_group_path(self._customer_id, ad_group_id)

            field_mask_paths = []
            if "name" in params:
                ad_group.name = params["name"]
                field_mask_paths.append("name")
            if "status" in params:
                ad_group.status = self._client.enums.AdGroupStatusEnum[params["status"]]
                field_mask_paths.append("status")
            if "cpc_bid_micros" in params:
                ad_group.cpc_bid_micros = params["cpc_bid_micros"]
                field_mask_paths.append("cpc_bid_micros")

            from google.protobuf import field_mask_pb2

            ad_group_operation.update_mask.CopyFrom(field_mask_pb2.FieldMask(paths=field_mask_paths))

            response = ad_group_service.mutate_ad_groups(
                customer_id=self._customer_id,
                operations=[ad_group_operation],
            )
            resource_name = response.results[0].resource_name
            result = {
                "id": ad_group_id,
                "resource_name": resource_name,
                "updated_fields": field_mask_paths,
            }
            logger.info("광고그룹 업데이트 완료: id=%s fields=%s", ad_group_id, field_mask_paths)
            return result
        except GoogleAdsException as e:
            logger.error("광고그룹 업데이트 실패 (id=%s): %s", ad_group_id, e)
            raise

    def delete_ad_group(self, ad_group_id: str) -> bool:
        """
        광고그룹을 삭제한다 (상태를 REMOVED로 변경).

        Args:
            ad_group_id: 삭제할 광고그룹 ID

        Returns:
            bool: 삭제 성공 여부

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            ad_group_service = self._client.get_service("AdGroupService")
            ad_group_operation = self._client.get_type("AdGroupOperation")
            resource_name = ad_group_service.ad_group_path(self._customer_id, ad_group_id)
            ad_group_operation.remove = resource_name

            ad_group_service.mutate_ad_groups(
                customer_id=self._customer_id,
                operations=[ad_group_operation],
            )
            logger.info("광고그룹 삭제 완료 (REMOVED): id=%s", ad_group_id)
            return True
        except GoogleAdsException as e:
            logger.error("광고그룹 삭제 실패 (id=%s): %s", ad_group_id, e)
            raise

    # ------------------------------------------------------------------
    # 키워드 관리
    # ------------------------------------------------------------------

    def list_keywords(self, ad_group_id: str, limit: int = 25) -> list[dict]:
        """
        광고그룹의 키워드 목록을 반환한다.

        Args:
            ad_group_id: 광고그룹 ID
            limit: 최대 반환 개수 (기본값 25)

        Returns:
            list[dict]: 키워드 정보 목록 (criterion_id, keyword_text, match_type, status 포함)

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        query = f"""
            SELECT
                ad_group_criterion.criterion_id,
                ad_group_criterion.keyword.text,
                ad_group_criterion.keyword.match_type,
                ad_group_criterion.status,
                ad_group_criterion.cpc_bid_micros,
                ad_group.id
            FROM ad_group_criterion
            WHERE ad_group_criterion.type = KEYWORD
              AND ad_group.id = {ad_group_id}
            ORDER BY ad_group_criterion.criterion_id
            LIMIT {limit}
        """
        try:
            ga_service = self._get_ga_service()
            response = ga_service.search(customer_id=self._customer_id, query=query)
            result = []
            for row in response:
                criterion = row.ad_group_criterion
                result.append(
                    {
                        "criterion_id": str(criterion.criterion_id),
                        "keyword_text": criterion.keyword.text,
                        "match_type": criterion.keyword.match_type.name,
                        "status": criterion.status.name,
                        "cpc_bid_micros": criterion.cpc_bid_micros,
                        "ad_group_id": str(row.ad_group.id),
                    }
                )
            logger.debug("키워드 목록 조회 완료: %d건 (ad_group_id=%s)", len(result), ad_group_id)
            return result
        except GoogleAdsException as e:
            logger.error("키워드 목록 조회 실패 (ad_group_id=%s): %s", ad_group_id, e)
            raise

    def add_keywords(
        self,
        ad_group_id: str,
        keywords: list[dict],
    ) -> list[dict]:
        """
        광고그룹에 키워드를 추가한다.

        Args:
            ad_group_id: 광고그룹 ID
            keywords: 추가할 키워드 목록. 각 항목은 아래 키를 포함하는 dict:
                      - text (str): 키워드 텍스트
                      - match_type (str): BROAD | PHRASE | EXACT (기본값: BROAD)
                      - cpc_bid_micros (int, optional): CPC 입찰가

        Returns:
            list[dict]: 생성된 키워드 resource_name 목록

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            ad_group_service = self._client.get_service("AdGroupService")
            ad_group_criterion_service = self._client.get_service("AdGroupCriterionService")
            ad_group_resource = ad_group_service.ad_group_path(self._customer_id, ad_group_id)

            operations = []
            for kw in keywords:
                operation = self._client.get_type("AdGroupCriterionOperation")
                criterion = operation.create
                criterion.ad_group = ad_group_resource
                criterion.status = self._client.enums.AdGroupCriterionStatusEnum.ENABLED
                criterion.keyword.text = kw["text"]
                match_type_str = kw.get("match_type", "BROAD")
                criterion.keyword.match_type = self._client.enums.KeywordMatchTypeEnum[match_type_str]
                if "cpc_bid_micros" in kw:
                    criterion.cpc_bid_micros = kw["cpc_bid_micros"]
                operations.append(operation)

            response = ad_group_criterion_service.mutate_ad_group_criteria(
                customer_id=self._customer_id,
                operations=operations,
            )
            result = [{"resource_name": r.resource_name} for r in response.results]
            logger.info("키워드 추가 완료: %d건 (ad_group_id=%s)", len(result), ad_group_id)
            return result
        except GoogleAdsException as e:
            logger.error("키워드 추가 실패 (ad_group_id=%s): %s", ad_group_id, e)
            raise

    def update_keyword_status(
        self,
        keyword_criterion_id: str,
        ad_group_id: str,
        status: str,
    ) -> dict:
        """
        키워드 상태를 업데이트한다.

        Args:
            keyword_criterion_id: 키워드 criterion ID
            ad_group_id: 키워드가 속한 광고그룹 ID
            status: 변경할 상태 (ENABLED | PAUSED | REMOVED)

        Returns:
            dict: 업데이트된 resource_name 포함 dict

        Raises:
            GoogleAdsException: API 호출 실패 시
        """
        try:
            ad_group_criterion_service = self._client.get_service("AdGroupCriterionService")
            resource_name = ad_group_criterion_service.ad_group_criterion_path(
                self._customer_id, ad_group_id, keyword_criterion_id
            )

            operation = self._client.get_type("AdGroupCriterionOperation")
            criterion = operation.update
            criterion.resource_name = resource_name
            criterion.status = self._client.enums.AdGroupCriterionStatusEnum[status]

            from google.protobuf import field_mask_pb2

            operation.update_mask.CopyFrom(field_mask_pb2.FieldMask(paths=["status"]))

            response = ad_group_criterion_service.mutate_ad_group_criteria(
                customer_id=self._customer_id,
                operations=[operation],
            )
            result = {
                "resource_name": response.results[0].resource_name,
                "criterion_id": keyword_criterion_id,
                "ad_group_id": ad_group_id,
                "status": status,
            }
            logger.info(
                "키워드 상태 업데이트 완료: criterion_id=%s status=%s",
                keyword_criterion_id,
                status,
            )
            return result
        except GoogleAdsException as e:
            logger.error(
                "키워드 상태 업데이트 실패 (criterion_id=%s): %s",
                keyword_criterion_id,
                e,
            )
            raise

    # ------------------------------------------------------------------
    # 성과 리포트 (인사이트)
    # ------------------------------------------------------------------

    def get_insights(
        self,
        entity_id: str,
        entity_type: str = "campaign",
        date_range: str = "LAST_7_DAYS",
        metrics: list[str] | None = None,
        since: str | None = None,
        until: str | None = None,
    ) -> list[dict]:
        """
        광고 성과 리포트를 반환한다.

        Args:
            entity_id: 조회할 캠페인/광고그룹/키워드 ID
            entity_type: 'campaign', 'ad_group', 'keyword' 중 하나 (기본값: 'campaign')
            date_range: GAQL 날짜 범위 프리셋 (기본값: LAST_7_DAYS).
                        LAST_7_DAYS, LAST_30_DAYS, THIS_MONTH, LAST_MONTH 등 지원.
            metrics: 조회할 metrics 필드 목록. None이면 기본 metrics 사용.

        Returns:
            list[dict]: 성과 리포트 레코드 목록

        Raises:
            ValueError: 지원하지 않는 entity_type인 경우
            GoogleAdsException: API 호출 실패 시
        """
        if metrics is None:
            metrics = _DEFAULT_INSIGHT_METRICS

        metrics_clause = ", ".join(metrics)
        entity_lower = entity_type.lower()

        if entity_lower == "campaign":
            resource = "campaign"
            where_field = "campaign.id"
            select_fields = "campaign.id, campaign.name, segments.date"
        elif entity_lower == "ad_group":
            resource = "ad_group"
            where_field = "ad_group.id"
            select_fields = "ad_group.id, ad_group.name, campaign.id, segments.date"
        elif entity_lower == "keyword":
            resource = "keyword_view"
            where_field = "ad_group_criterion.criterion_id"
            select_fields = (
                "ad_group_criterion.criterion_id, "
                "ad_group_criterion.keyword.text, "
                "ad_group.id, campaign.id, segments.date"
            )
        else:
            raise ValueError(
                f"지원하지 않는 entity_type: {entity_type}. " "'campaign', 'ad_group', 'keyword' 중 하나여야 합니다."
            )

        # 날짜 조건 구성
        if since and until:
            date_condition = f"segments.date BETWEEN '{since}' AND '{until}'"
        else:
            date_condition = f"segments.date DURING {date_range}"

        query = f"""
            SELECT
                {select_fields},
                {metrics_clause}
            FROM {resource}
            WHERE {where_field} = {entity_id}
              AND {date_condition}
            ORDER BY segments.date DESC
        """
        try:
            ga_service = self._get_ga_service()
            response = ga_service.search(customer_id=self._customer_id, query=query)
            result = []
            for row in response:
                record: dict[str, Any] = {
                    "entity_id": entity_id,
                    "entity_type": entity_type,
                    "date": row.segments.date,
                    "impressions": row.metrics.impressions,
                    "clicks": row.metrics.clicks,
                    "cost_micros": row.metrics.cost_micros,
                    "ctr": row.metrics.ctr,
                    "average_cpc": row.metrics.average_cpc,
                    "conversions": row.metrics.conversions,
                }
                result.append(record)
            logger.debug(
                "인사이트 조회 완료: %d건 (entity_type=%s, id=%s, date_range=%s, since=%s, until=%s)",
                len(result),
                entity_type,
                entity_id,
                date_range,
                since,
                until,
            )
            return result
        except GoogleAdsException as e:
            logger.error(
                "인사이트 조회 실패 (entity_type=%s, id=%s): %s",
                entity_type,
                entity_id,
                e,
            )
            raise

    # ------------------------------------------------------------------
    # 반응형 검색 광고 생성
    # ------------------------------------------------------------------

    def create_responsive_search_ad(
        self,
        ad_group_id: str,
        headlines: list[str],
        descriptions: list[str],
        final_url: str,
    ) -> dict:
        """
        반응형 검색 광고(RSA)를 생성한다.

        Args:
            ad_group_id: 광고그룹 ID
            headlines: 헤드라인 목록 (최소 3개, 최대 15개)
            descriptions: 설명 목록 (최소 2개, 최대 4개)
            final_url: 광고 클릭 시 연결할 최종 URL

        Returns:
            dict: 생성된 광고 resource_name, ad_group_id 포함 dict

        Raises:
            ValueError: 헤드라인 또는 설명 개수가 유효 범위를 벗어난 경우
            GoogleAdsException: API 호출 실패 시
        """
        if not (3 <= len(headlines) <= 15):
            raise ValueError(f"헤드라인은 3~15개여야 합니다. (현재: {len(headlines)}개)")
        if not (2 <= len(descriptions) <= 4):
            raise ValueError(f"설명은 2~4개여야 합니다. (현재: {len(descriptions)}개)")

        try:
            ad_group_service = self._client.get_service("AdGroupService")
            ad_group_ad_service = self._client.get_service("AdGroupAdService")
            operation = self._client.get_type("AdGroupAdOperation")

            ad_group_ad = operation.create
            ad_group_ad.status = self._client.enums.AdGroupAdStatusEnum.PAUSED
            ad_group_ad.ad_group = ad_group_service.ad_group_path(self._customer_id, ad_group_id)

            # 반응형 검색 광고 설정
            ad = ad_group_ad.ad
            ad.final_urls.append(final_url)

            rsa = ad.responsive_search_ad
            for text in headlines:
                asset = self._client.get_type("AdTextAsset")
                asset.text = text
                rsa.headlines.append(asset)
            for text in descriptions:
                asset = self._client.get_type("AdTextAsset")
                asset.text = text
                rsa.descriptions.append(asset)

            response = ad_group_ad_service.mutate_ad_group_ads(
                customer_id=self._customer_id,
                operations=[operation],
            )
            resource_name = response.results[0].resource_name

            result = {
                "resource_name": resource_name,
                "ad_group_id": ad_group_id,
                "final_url": final_url,
                "headline_count": len(headlines),
                "description_count": len(descriptions),
            }
            logger.info(
                "반응형 검색 광고 생성 완료: ad_group_id=%s resource_name=%s",
                ad_group_id,
                resource_name,
            )
            return result
        except GoogleAdsException as e:
            logger.error("반응형 검색 광고 생성 실패 (ad_group_id=%s): %s", ad_group_id, e)
            raise