""" SERP Features Optimization Node Optimizes content for featured snippets, PAA, and AI Overviews """ import os import uuid import httpx from datetime import datetime from typing import Any, Dict, List from state.campaign_state import ( SEOCampaignState, CampaignPhase, transition_phase, add_error, add_warning, OptimizationQueueItem, SERPFeatureData, ) NEXT_API_URL = os.getenv("NEXT_API_URL", "http://127.0.0.1:3000") # Optimization templates for different SERP features SNIPPET_OPTIMIZATION = { "definition": { "format": "What is X? X is [40-60 word direct answer]. This [benefit/context].", "structure": ["direct_answer", "expansion", "context"], "word_count": (40, 60), }, "how_to": { "format": "Numbered list with clear action verbs", "structure": ["intro", "numbered_steps", "conclusion"], "step_count": (5, 8), }, "comparison": { "format": "Table with clear headers and concise cells", "structure": ["intro", "comparison_table", "recommendation"], "columns": ["Feature", "Option A", "Option B"], }, "list": { "format": "Bulleted list with consistent structure", "structure": ["intro_sentence", "bullet_items", "summary"], "item_count": (5, 10), }, } PAA_OPTIMIZATION = { "answer_length": (50, 100), # Words "structure": "Question -> Direct Answer -> Brief Expansion", "max_questions": 5, } AI_OVERVIEW_OPTIMIZATION = { "answer_length": (40, 60), # Words - concise for citation "format": "Direct statement -> Evidence/Data -> Brand mention", "differentiation": [ "Unique data point", "Proprietary methodology", "Expert authority signal", ], } def get_optimization_type_for_feature( feature: SERPFeatureData, keyword: str ) -> str: """Determine the best optimization type for a SERP feature""" keyword_lower = keyword.lower() if feature.get("feature_type") == "snippet": # Detect query intent if keyword_lower.startswith(("what is", "what are", "define", "meaning of")): return "snippet_definition" elif keyword_lower.startswith(("how to", "how do", "steps to", "guide to")): return "snippet_how_to" elif "vs" in keyword_lower or "compared to" in keyword_lower or "difference" in keyword_lower: return "snippet_comparison" elif keyword_lower.startswith(("best", "top", "list of", "types of")): return "snippet_list" else: return "snippet_paragraph" elif feature.get("feature_type") == "paa": return "paa_expansion" elif feature.get("feature_type") == "ai_overview": return "ai_overview_citation" return "general_optimization" async def fetch_existing_pages(site_id: str) -> Dict[str, dict]: """Fetch existing pages for optimization""" pages = {} try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get( f"{NEXT_API_URL}/api/admin/seo/pages", headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_TOKEN', '')}"}, params={"status": "PUBLISHED", "limit": 100} ) if response.status_code == 200: data = response.json() for page in data.get("pages", []): # Index by slug and keywords pages[page.get("slug", "")] = page # Also index by target keyword if available if page.get("targetKeyword"): pages[page["targetKeyword"].lower()] = page except Exception as e: print(f"[SERPFeaturesNode] Fetch pages error: {e}") return pages async def queue_optimization( page_id: str, optimization_type: str, keyword: str, priority: int, site_id: str ) -> str | None: """Queue a SERP feature optimization job""" try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( f"{NEXT_API_URL}/api/admin/seo/optimize/queue", headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_TOKEN', '')}"}, json={ "siteId": site_id, "pageId": page_id, "optimizationType": optimization_type, "targetKeyword": keyword, "priority": priority, "source": "langgraph_serp_optimization", } ) if response.status_code in (200, 201): data = response.json() return data.get("jobId") except Exception as e: print(f"[SERPFeaturesNode] Queue optimization error: {e}") return None async def serp_optimization_phase(state: SEOCampaignState) -> SEOCampaignState: """ SERP Features Optimization phase node - Analyzes existing content for SERP feature optimization opportunities - Queues optimizations for snippets, PAA, and AI Overviews - Tracks optimization queue in state """ print(f"[SERPFeaturesNode] Starting SERP optimization for site {state['site_id']}") site_id = state["site_id"] now = datetime.utcnow().isoformat() optimization_queue: List[OptimizationQueueItem] = [] optimized_count = 0 try: # Fetch existing pages existing_pages = await fetch_existing_pages(site_id) print(f"[SERPFeaturesNode] Found {len(existing_pages)} existing pages") # Process SERP feature opportunities for feature in state.get("serp_features_data", []): keyword = feature.get("keyword", "").lower() # Skip if already captured if feature.get("captured"): continue # Skip if already optimized if feature.get("optimized"): continue # Find matching page page = existing_pages.get(keyword) if not page: # Try partial match for slug, p in existing_pages.items(): if keyword in slug.lower() or slug.lower() in keyword: page = p break # Determine optimization type regardless of page existence optimization_type = get_optimization_type_for_feature(feature, keyword) # Determine priority feature_type = feature.get("feature_type", "") if feature_type == "ai_overview": priority = 10 # Highest - AI Overviews are valuable elif feature_type == "snippet": priority = 9 elif feature_type == "paa": priority = 8 else: priority = 7 # Create optimization queue item (even without page match for tracking) page_id = page.get("id") if page else None opt_item = OptimizationQueueItem( id=str(uuid.uuid4()), page_id=page_id, optimization_type=optimization_type, target_keyword=keyword, priority=priority, status="pending" if page_id else "needs_content", completed_at=None, ) optimization_queue.append(opt_item) # Queue the optimization job if we have a page if page_id: max_to_queue = state.get("config", {}).get("max_optimizations_per_run", 20) if optimized_count < max_to_queue: try: job_id = await queue_optimization( page_id, optimization_type, keyword, priority, site_id ) if job_id: opt_item["status"] = "queued" optimized_count += 1 except Exception as e: print(f"[SERPFeaturesNode] Failed to queue optimization for '{keyword}': {e}") # If no API available, report the opportunities found if optimized_count == 0 and optimization_queue: state = add_warning(state, "serp_features", f"SERP optimization API unavailable - {len(optimization_queue)} opportunities identified") print(f"[SERPFeaturesNode] Queued {optimized_count} optimizations, {len(optimization_queue)} total opportunities") except Exception as e: print(f"[SERPFeaturesNode] Error during SERP optimization phase: {e}") state = add_error(state, "serp_features", f"SERP optimization phase failed: {str(e)}") # Update state state["optimization_queue"] = optimization_queue state["pages_optimized"] = optimized_count # Transition to link discovery phase state = transition_phase(state, CampaignPhase.LINK_OPPS) state["updated_at"] = now return state