""" Zero-Click SERP Optimization Graph Focused workflow for capturing featured snippets, PAA, and AI Overviews """ from typing import Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.base import BaseCheckpointSaver from state.campaign_state import ( SEOCampaignState, CampaignPhase, SERPFeatureType, create_initial_state, transition_phase, ) from nodes.research_node import research_phase, detect_serp_features from nodes.serp_features_node import serp_optimization_phase async def keyword_research(state: SEOCampaignState) -> SEOCampaignState: """ Zero-click focused keyword research Prioritizes: - Question keywords (What, How, Why, When) - Definition keywords (What is, Define, Meaning) - Comparison keywords (vs, compared to, difference) - List keywords (Best, Top, Types of) """ print(f"[ZeroClickResearch] Starting keyword research for site {state['site_id']}") try: # Run base research state = await research_phase(state) # Ensure config exists if state.get("config") is None: state["config"] = {} # Extract and categorize keywords by intent keywords = [] gsc_data = state.get("gsc_data") if gsc_data: keywords = [q.get("query", "") for q in gsc_data.get("top_queries", [])] # If no GSC keywords, use SERP feature keywords if not keywords: serp_features = state.get("serp_features_data", []) keywords = [f.get("keyword", "") for f in serp_features if f.get("keyword")] # If still no keywords, use real sports betting keywords if not keywords: keywords = [ # Definition queries (for featured snippets) "what is sharp money in sports betting", "what are player props", "what is public fade strategy", # How-to queries "how to analyze mlb totals", "how to find value bets", "how to read betting odds", # Comparison queries "mlb moneyline vs spread betting", "nba totals vs player props", # List queries "best mlb betting analytics tools", "top sports betting strategies 2026", ] # Categorize by zero-click potential categorized = { "definition": [], # What is, define, meaning "how_to": [], # How to, steps, guide "comparison": [], # vs, compared, difference "list": [], # Best, top, types "other": [], } for kw in keywords: kw_lower = kw.lower() if any(kw_lower.startswith(p) for p in ["what is", "what are", "define", "meaning of"]): categorized["definition"].append(kw) elif any(kw_lower.startswith(p) for p in ["how to", "how do", "steps to", "guide to"]): categorized["how_to"].append(kw) elif any(p in kw_lower for p in [" vs ", "compared to", "difference between", "versus"]): categorized["comparison"].append(kw) elif any(kw_lower.startswith(p) for p in ["best ", "top ", "list of", "types of"]): categorized["list"].append(kw) else: categorized["other"].append(kw) # Store categorization in config state["config"]["keyword_categories"] = categorized state["config"]["zero_click_keywords"] = ( categorized["definition"] + categorized["how_to"] + categorized["comparison"] + categorized["list"] ) print(f"[ZeroClickResearch] Categorized keywords: " f"Definition={len(categorized['definition'])}, " f"How-to={len(categorized['how_to'])}, " f"Comparison={len(categorized['comparison'])}, " f"List={len(categorized['list'])}") except Exception as e: print(f"[ZeroClickResearch] Error during keyword research: {e}") # Ensure config exists and set empty defaults if state.get("config") is None: state["config"] = {} state["config"]["keyword_categories"] = {"definition": [], "how_to": [], "comparison": [], "list": [], "other": []} state["config"]["zero_click_keywords"] = [] return state async def snippet_optimization(state: SEOCampaignState) -> SEOCampaignState: """ Optimize content for featured snippets Applies optimization patterns: - Definition snippets: 40-60 word direct answers - How-to snippets: Numbered lists with clear steps - Comparison snippets: Tables with structured data - List snippets: Bullet points with consistent format """ print(f"[SnippetOptimization] Optimizing for featured snippets") try: # Ensure config exists if state.get("config") is None: state["config"] = {} categories = state.get("config", {}).get("keyword_categories", {}) # Track optimization targets optimizations = [] # Definition snippet targets for kw in categories.get("definition", [])[:10]: optimizations.append({ "keyword": kw, "feature_type": SERPFeatureType.FEATURED_SNIPPET.value, "optimization_type": "snippet_definition", "format": "40-60 word direct answer starting with '[Subject] is...'", "status": "pending", }) # How-to snippet targets for kw in categories.get("how_to", [])[:10]: optimizations.append({ "keyword": kw, "feature_type": SERPFeatureType.FEATURED_SNIPPET.value, "optimization_type": "snippet_how_to", "format": "Numbered list with 5-8 clear action steps", "status": "pending", }) # Comparison snippet targets for kw in categories.get("comparison", [])[:5]: optimizations.append({ "keyword": kw, "feature_type": SERPFeatureType.FEATURED_SNIPPET.value, "optimization_type": "snippet_comparison", "format": "Comparison table with clear headers", "status": "pending", }) # List snippet targets for kw in categories.get("list", [])[:10]: optimizations.append({ "keyword": kw, "feature_type": SERPFeatureType.FEATURED_SNIPPET.value, "optimization_type": "snippet_list", "format": "Bullet list with 5-10 items", "status": "pending", }) state["config"]["snippet_optimizations"] = optimizations print(f"[SnippetOptimization] Queued {len(optimizations)} snippet optimizations") except Exception as e: print(f"[SnippetOptimization] Error: {e}") state["config"]["snippet_optimizations"] = [] return state async def paa_expansion(state: SEOCampaignState) -> SEOCampaignState: """ Expand content for People Also Ask features For each target keyword: - Find 3-5 related PAA questions - Create FAQ schema with 2-3 sentence answers - Add internal links between related questions """ print(f"[PAAExpansion] Expanding content for PAA") try: # Ensure config exists if state.get("config") is None: state["config"] = {} zero_click_keywords = state.get("config", {}).get("zero_click_keywords", []) paa_targets = [] for kw in zero_click_keywords[:20]: # Generate related questions based on keyword paa_targets.append({ "keyword": kw, "feature_type": SERPFeatureType.PEOPLE_ALSO_ASK.value, "related_questions": generate_related_questions(kw), "answer_format": "2-3 sentences, direct answer first", "schema_type": "FAQPage", "status": "pending", }) state["config"]["paa_targets"] = paa_targets print(f"[PAAExpansion] Queued {len(paa_targets)} PAA expansions") except Exception as e: print(f"[PAAExpansion] Error: {e}") state["config"]["paa_targets"] = [] return state def generate_related_questions(keyword: str) -> list[str]: """Generate related PAA questions for a keyword""" questions = [] kw_lower = keyword.lower() # Generic question patterns base_topic = kw_lower.replace("how to ", "").replace("what is ", "").replace("best ", "") questions.append(f"What is {base_topic}?") questions.append(f"How does {base_topic} work?") questions.append(f"Why is {base_topic} important?") questions.append(f"What are the benefits of {base_topic}?") questions.append(f"How do you use {base_topic}?") return questions[:5] async def ai_overview_targeting(state: SEOCampaignState) -> SEOCampaignState: """ Optimize for AI Overview citations AI Overviews prefer: - Direct answers (40-60 words) - Unique data or proprietary insights - Clear authority signals - Brand mention co-occurrence """ print(f"[AIOverviewTargeting] Targeting AI Overviews") try: # Ensure config exists if state.get("config") is None: state["config"] = {} zero_click_keywords = state.get("config", {}).get("zero_click_keywords", []) ai_overview_targets = [] for kw in zero_click_keywords[:15]: ai_overview_targets.append({ "keyword": kw, "feature_type": SERPFeatureType.AI_OVERVIEW.value, "optimization_strategy": { "direct_answer": "40-60 words, definitive statement", "unique_data": "Include proprietary EventheOdds data point", "authority_signal": "Cite sample size or methodology", "brand_mention": "Natural brand reference in context", }, "format": "Direct Statement -> Data/Evidence -> Context/Brand", "status": "pending", }) state["config"]["ai_overview_targets"] = ai_overview_targets print(f"[AIOverviewTargeting] Queued {len(ai_overview_targets)} AI Overview targets") except Exception as e: print(f"[AIOverviewTargeting] Error: {e}") state["config"]["ai_overview_targets"] = [] return state async def apply_schema_markup(state: SEOCampaignState) -> SEOCampaignState: """ Generate and queue schema markup for SERP features Schema types: - FAQPage for PAA targeting - HowTo for how-to snippets - Article with speakable for AI assistants """ print(f"[SchemaMarkup] Generating schema markup") try: # Ensure config exists if state.get("config") is None: state["config"] = {} schema_queue = [] # FAQ schema for PAA targets for target in state.get("config", {}).get("paa_targets", []): schema_queue.append({ "keyword": target["keyword"], "schema_type": "FAQPage", "questions": target.get("related_questions", []), "status": "pending", }) # HowTo schema for how-to snippets for opt in state.get("config", {}).get("snippet_optimizations", []): if opt.get("optimization_type") == "snippet_how_to": schema_queue.append({ "keyword": opt["keyword"], "schema_type": "HowTo", "status": "pending", }) # Article schema with speakable for AI Overview targets for target in state.get("config", {}).get("ai_overview_targets", []): schema_queue.append({ "keyword": target["keyword"], "schema_type": "Article", "speakable": True, "status": "pending", }) state["config"]["schema_queue"] = schema_queue print(f"[SchemaMarkup] Queued {len(schema_queue)} schema implementations") except Exception as e: print(f"[SchemaMarkup] Error: {e}") state["config"]["schema_queue"] = [] return state async def generate_zero_click_report(state: SEOCampaignState) -> SEOCampaignState: """ Generate zero-click optimization report """ print(f"[ZeroClickReport] Generating report") try: # Ensure config exists if state.get("config") is None: state["config"] = {} config = state.get("config", {}) report = { "campaign_type": "zero_click_authority", "keyword_analysis": { "definition_keywords": len(config.get("keyword_categories", {}).get("definition", [])), "how_to_keywords": len(config.get("keyword_categories", {}).get("how_to", [])), "comparison_keywords": len(config.get("keyword_categories", {}).get("comparison", [])), "list_keywords": len(config.get("keyword_categories", {}).get("list", [])), "total_zero_click_keywords": len(config.get("zero_click_keywords", [])), }, "optimizations_queued": { "snippet_optimizations": len(config.get("snippet_optimizations", [])), "paa_expansions": len(config.get("paa_targets", [])), "ai_overview_targets": len(config.get("ai_overview_targets", [])), "schema_implementations": len(config.get("schema_queue", [])), }, "recommendations": [ "Focus on definition keywords with 40-60 word direct answers", "Create numbered step guides for how-to keywords", "Add comparison tables for vs. keywords", "Implement FAQPage schema for all PAA targets", "Add unique EventheOdds data points for AI Overview citations", ], } state["config"]["zero_click_report"] = report except Exception as e: print(f"[ZeroClickReport] Error generating report: {e}") state["config"]["zero_click_report"] = {"error": str(e)} state = transition_phase(state, CampaignPhase.COMPLETED) state["completed_at"] = __import__("datetime").datetime.utcnow().isoformat() print(f"[ZeroClickReport] Campaign completed") return state def create_zero_click_graph(checkpointer: BaseCheckpointSaver | None = None): """ Create the zero-click SERP optimization graph Flow: 1. Keyword Research - Categorize by zero-click intent 2. Snippet Optimization - Target featured snippets 3. PAA Expansion - Expand for People Also Ask 4. AI Overview Targeting - Optimize for AI citations 5. Schema Markup - Generate structured data 6. Report - Generate optimization report This is a non-interrupt graph focused purely on SERP feature optimization. """ graph = StateGraph(SEOCampaignState) # Add nodes graph.add_node("keyword_research", keyword_research) graph.add_node("snippet_optimization", snippet_optimization) graph.add_node("paa_expansion", paa_expansion) graph.add_node("ai_overview_targeting", ai_overview_targeting) graph.add_node("schema_markup", apply_schema_markup) graph.add_node("report", generate_zero_click_report) # Linear flow graph.add_edge(START, "keyword_research") graph.add_edge("keyword_research", "snippet_optimization") graph.add_edge("snippet_optimization", "paa_expansion") graph.add_edge("paa_expansion", "ai_overview_targeting") graph.add_edge("ai_overview_targeting", "schema_markup") graph.add_edge("schema_markup", "report") graph.add_edge("report", END) # Compile (no interrupts needed for this workflow) compiled = graph.compile(checkpointer=checkpointer) return compiled