from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import AllowAny from serp.models import Keyword, Groups from django.http import JsonResponse import requests import json import base64 from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime import tldextract def parse_domain(site_url): ext = tldextract.extract(site_url) domain_name = f"{ext.domain}.{ext.suffix}" return domain_name def get_last_search_volume(search_volume_data): try: last_search_volume = next((item["search_volume"] for item in search_volume_data if item.get("search_volume") is not None), 0) return last_search_volume except Exception as e: return "NA" def ranked_keywords(keyword_objs): keyword_list = [k.keyword for k in keyword_objs] location_list = [k.location_code for k in keyword_objs] site_url_list = [k.site_url for k in keyword_objs] dataseo_username = "ranjit.singh@richestsoft.com" dataseo_password = "38ea7274bacf72f0" auth_string = f"{dataseo_username}:{dataseo_password}" auth_base64 = base64.b64encode(auth_string.encode("utf-8")).decode("utf-8") url = "https://api.dataforseo.com/v3/keywords_data/google_ads/search_volume/live" headers = {"Authorization": f"Basic {auth_base64}", "Content-Type": "application/json"} payload = [{"location_code": location_list[0], "language_code": "en", "keywords": keyword_list, "sort_by": "search_volume"}] response = requests.post(url, headers=headers, data=json.dumps(payload)) json_data = response.json() print(json_data) if json_data.get("status_code") != 20000: print(f"Error: {json_data.get('status_code')} - {json_data.get('status_message')}") return [] difficulty_payload = [{"keywords": keyword_list[:1000], "location_code": location_list[0], "language_code": "en"}] # Ensure max 1000 difficulty_url = "https://api.dataforseo.com/v3/dataforseo_labs/google/bulk_keyword_difficulty/live" difficulty_response = requests.post(difficulty_url, headers=headers, data=json.dumps(difficulty_payload)) difficulty_data = difficulty_response.json() if difficulty_data.get("status_code") != 20000: print(f"Difficulty API Error: {difficulty_data.get('status_code')} - {difficulty_data.get('status_message')}") difficulty_map = {} for task in difficulty_data.get("tasks", []): for result in task.get("result", []): for item in result.get("items", []): keyword = item.get("keyword") difficulty = item.get("keyword_difficulty", "NA") if keyword: difficulty_map[keyword] = difficulty search_intent_payload = [{"keywords": keyword_list[:1000], "language_code": "en"}] # Ensure max 1000 search_intent_url = "https://api.dataforseo.com/v3/dataforseo_labs/google/search_intent/live" search_intent_response = requests.post(search_intent_url, headers=headers, data=json.dumps(search_intent_payload)) search_intent_data = search_intent_response.json() search_intent_map = {} for task in search_intent_data.get("tasks", []): for result in task.get("result", []): for item in result.get("items", []): keyword = item.get("keyword") intent = item.get("keyword_intent", {}) if keyword: search_intent_map[keyword] = intent.get("label", "NA") # Initialize the final result array extracted_data = [] for task in json_data.get("tasks", []): for item in task.get("result", []): keyword = item.get("keyword") or "" search_volume = get_last_search_volume(item.get("monthly_searches")) cpc = item.get("cpc") if item.get("cpc") is not None else 0.0 monthly_searches = item.get("monthly_searches") or [] keyword_difficulty = difficulty_map.get(keyword, 0) search_intent = search_intent_map.get(keyword, 0) if monthly_searches is not None: cleaned_monthly_searches = [{"year": m.get("year") or 0, "month": m.get("month") or 0, "search_volume": m.get("search_volume") or 0} for m in monthly_searches] extracted_data.append({"keyword": keyword, "search_volume": search_volume, "cpc": cpc, "keyword_difficulty": keyword_difficulty, "search_intent": search_intent, "monthly_searches": cleaned_monthly_searches}) return extracted_data def process_keyword_batch(keyword_objs): try: results = ranked_keywords(keyword_objs) except Exception as e: print(f"[Search Volume Metrics Error] {e}") return [] result_map = {item["keyword"]: item for item in results if item.get("keyword")} for keyword in keyword_objs: data = result_map.get(keyword.keyword) print("Keyword: ",data) if data: keyword.search_intent = data.get("search_intent") keyword.search_volume = data.get("search_volume") keyword.search_volume_data = data.get("monthly_searches") keyword.cpc = data.get("cpc") # keyword.traffic = data.get("traffic") keyword.keyword_difficulty = data.get("keyword_difficulty") else: keyword.search_intent = "NA" keyword.search_volume = "NA" keyword.cpc = "NA" # keyword.traffic = "NA" keyword.keyword_difficulty = "NA" return keyword_objs def process_group(group, sort_order): group_id = group.id Groups.objects.filter(id=group_id).update(metric_status="SCHD") sort_field = "created_date" if int(sort_order) % 2 == 1 else "-created_date" keywords = list(Keyword.objects.filter(fk_group_id=group_id, metric_status="INIT").order_by(sort_field)) keywords=keywords[:1] print(keywords) if not keywords: Groups.objects.filter(id=group_id).update(metric_status="COMP") return batch_size = 100 batches = [keywords[i : i + batch_size] for i in range(0, len(keywords), batch_size)] updated_keywords = [] # Using ThreadPoolExecutor to parallelize batch processing with ThreadPoolExecutor(max_workers=5) as executor: # adjust max_workers as needed futures = [executor.submit(process_keyword_batch, batch) for batch in batches] for future in as_completed(futures): result = future.result() if result: updated_keywords.extend(result) if updated_keywords: for keyword in updated_keywords: search_volume_data = keyword.search_volume_data if search_volume_data is not None: keyword.search_volume_data = search_volume_data[::-1] Keyword.objects.filter(id=keyword.id).update(cpc=keyword.cpc, keyword_difficulty=keyword.keyword_difficulty, search_intent=keyword.search_intent, search_volume=keyword.search_volume, search_volume_data=keyword.search_volume_data, metric_status="COMP") Groups.objects.filter(id=group_id).update(metric_status="COMP") @api_view(["GET"]) @permission_classes((AllowAny,)) def insightTracker(request, sort_order): """Cron job to process one group at a time using metric_status.""" group = Groups.objects.filter(metric_status="SCHD").first() if not group: group = Groups.objects.filter(metric_status="INIT").first() if group: Groups.objects.filter(id=group.id).update(metric_status="SCHD") if not group: return JsonResponse({"status": "true", "message": "No groups to process"}) try: process_group(group, sort_order) except Exception as e: print(f"[Group Error] {group.id} - {str(e)}") Groups.objects.filter(id=group.id).update(metric_status="FAIL") return JsonResponse({"status": "true", "message": "Group processing completed"})