src/api_logic.py

from pydantic import BaseModel
from typing import List, Optional
from google.ads.googleads.client import GoogleAdsClient
from google.ads.googleads.errors import GoogleAdsException
import numpy as np
import os

# Read yaml from Secrets in HF
yaml_str = os.environ["GOOGLE_ADS_YAML"]
# Write YAML to a temporary file
with open("google-ads.yaml", "w") as f:
    f.write(yaml_str)

client = GoogleAdsClient.load_from_storage("google-ads.yaml")
keyword_plan_idea_service = client.get_service("KeywordPlanIdeaService")

class KeywordRequestModel(BaseModel):
    customer_id: str
    keywords: List[str]
    geo_target: Optional[str] = None
    language: Optional[str] = None

# --- Call Google Ads API
def fetch_keyword_ideas(request: KeywordRequestModel):
    """
    Fetch keyword ideas from the Google Ads API based on the given parameters.

    Args:
        request (KeywordRequestModel):
            - customer_id (str): Google Ads customer ID.
            - keywords (List[str]): A list of seed keywords to generate ideas from.
            - geo_target (Optional[str]): (Optional) Geo target constant ID.
            - language (Optional[str]): (Optional) Language constant ID.

    Returns:
        List[dict]: A list of keyword idea results, where each item contains:
            - "keyword" (str): The suggested keyword text.
            - "avg_monthly_searches" (int): Average monthly search volume.
            - "competition" (str): Competition level (LOW / MEDIUM / HIGH).
            - "competition_index" (int): Numerical competition index.
        If an error occurs, an empty list is returned.
    """

    req = client.get_type("GenerateKeywordIdeasRequest")
    req.customer_id = request.customer_id
    req.keyword_seed.keywords.extend(request.keywords)

    if request.geo_target:
        req.geo_target_constants.append(
            client.get_service("GoogleAdsService").geo_target_constant_path(request.geo_target)
        )
    if request.language:
        req.language = client.get_service("GoogleAdsService").language_constant_path(request.language)

    try:
        response = keyword_plan_idea_service.generate_keyword_ideas(request=req)

        result_items = []

        # --- Loop results ---
        for idea in response.results:
            metrics = idea.keyword_idea_metrics

            # Monthly search volumes
            monthly_volumes = []
            if metrics.monthly_search_volumes:
                for mv in metrics.monthly_search_volumes:
                    monthly_volumes.append({
                        "month": mv.month.name,
                        "year": mv.year,
                        "monthly_searches": mv.monthly_searches
                    })


            # According To "Rule Book For GKP With Code" Transform Some Values
            raw_cpc = (((metrics.low_top_of_page_bid_micros) + (metrics.high_top_of_page_bid_micros)) / 2) / 1000000
            log_cpc = np.log(raw_cpc + 1)


            result_items.append({
                "keyword": idea.text,
                "avg_monthly_searches": metrics.avg_monthly_searches,
                "competition": metrics.competition.name,
                "competition_index": metrics.competition_index,
                "low_top_of_page_bid_micros": metrics.low_top_of_page_bid_micros / 1000000,
                "high_top_of_page_bid_micros": metrics.high_top_of_page_bid_micros / 1000000,
                # "average_cpc_micros": metrics.average_cpc_micros,
                "monthly_search_volumes": monthly_volumes,
                ##Created Features
                "log_cpc": log_cpc

            })

        #Normalize the Log CPC Values
        log_values = [item["log_cpc"] for item in result_items]
        min_log = min(log_values)
        max_log = max(log_values)

        
        def scale_0_100(x, min_val, max_val):
            if max_val == min_val:
                return 0
            return ((x - min_val) / (max_val - min_val)) * 100

        #Adding "log_cpc_norm_0_100" to Results_Items
        for item in result_items:
            item["log_cpc_norm_0_100"] = scale_0_100(item["log_cpc"], min_log, max_log)


        # --- Aggregate results ---
        aggregate = response.aggregate_metric_results
        aggregate_device = []

        if aggregate and aggregate.device_searches:
            for d in aggregate.device_searches:
                aggregate_device.append({
                    "device": d.device.name,
                    "search_count": d.search_count
                })
        # return response        

        return {
            "results": result_items,
            "aggregateMetricResults": {
                "device_searches": aggregate_device
            }
        }

    except Exception as ex:
        print(f"Error: {ex}")
        return {
            "results": [],
            "aggregateMetricResults": {}
        }


# --- Keyword selection
def rank_and_filter_keywords(api_response: dict, limit: int = 20):
    """
    Rank and filter keyword results based on average monthly searches.

    Args:
        api_response (dict):
            A dict returned from fetch_keyword_ideas(), expected to have a "results" key.
        limit (int, optional):
            Maximum number of top keywords to return. Defaults to 20.

    Returns:
        list:
            A list of up to `limit` keyword dictionaries, sorted in descending
            order by their avg_monthly_searches value.
    """

    # Check if it includs results key
    if "results" not in api_response or not isinstance(api_response["results"], list):
        print("Error: No results found in API response")
        return []

    keyword_list = api_response["results"]

    # Filter out entries where avg_monthly_searches is None (just in case)
    filtered = [k for k in keyword_list if k.get("avg_monthly_searches") is not None]

    # Sort by avg_monthly_searches in descending order
    sorted_keywords = sorted(
        filtered,
        key=lambda x: x["avg_monthly_searches"],
        reverse=True
    )

    return sorted_keywords[:limit]