Spaces:
Sleeping
Sleeping
File size: 6,072 Bytes
1585962 9ba72a4 54b64b8 1585962 e502067 3508c55 e502067 1585962 833a381 1585962 833a381 1585962 833a381 1585962 ed0d186 1585962 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | 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]
|