""" Research Phase Node Gathers GSC data, competitor analysis, and SERP feature opportunities """ import os import httpx from datetime import datetime from typing import Any, Dict from state.campaign_state import ( SEOCampaignState, CampaignPhase, transition_phase, add_error, add_warning, GSCData, CompetitorData, SERPFeatureData, ) # Next.js API base URL NEXT_API_URL = os.getenv("NEXT_API_URL", "http://127.0.0.1:3000") MCP_URL = os.getenv("MCP_URL", "http://127.0.0.1:3001") async def fetch_gsc_data(site_id: str) -> GSCData | None: """Fetch Google Search Console data for a site""" try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get( f"{NEXT_API_URL}/api/admin/seo/gsc/{site_id}/performance", headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_TOKEN', '')}"} ) if response.status_code == 200: data = response.json() return GSCData( property_url=data.get("propertyUrl", ""), date_range=data.get("dateRange", {}), total_impressions=data.get("totalImpressions", 0), total_clicks=data.get("totalClicks", 0), average_ctr=data.get("averageCtr", 0.0), average_position=data.get("averagePosition", 0.0), top_queries=data.get("topQueries", [])[:50], top_pages=data.get("topPages", [])[:50], fetched_at=datetime.utcnow().isoformat(), ) except Exception as e: print(f"[ResearchNode] GSC fetch error: {e}") return None async def fetch_competitor_data(site_id: str) -> list[CompetitorData]: """Fetch competitor analysis data""" competitors = [] try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get( f"{NEXT_API_URL}/api/admin/seo/competitors", headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_TOKEN', '')}"} ) if response.status_code == 200: data = response.json() for comp in data.get("competitors", [])[:5]: # Top 5 competitors.append(CompetitorData( domain=comp.get("domain", ""), domain_authority=comp.get("domainAuthority", 0), estimated_traffic=comp.get("estimatedTraffic", 0), top_keywords=comp.get("topKeywords", [])[:20], content_gaps=comp.get("contentGaps", [])[:20], backlink_sources=comp.get("backlinkSources", [])[:20], analyzed_at=datetime.utcnow().isoformat(), )) except Exception as e: print(f"[ResearchNode] Competitor fetch error: {e}") return competitors async def detect_serp_features(keywords: list[str]) -> list[SERPFeatureData]: """Detect SERP feature opportunities for keywords""" features = [] # Call MCP serp_feature_checker tool if available try: async with httpx.AsyncClient(timeout=60.0) as client: for keyword in keywords[:30]: # Limit to 30 keywords try: response = await client.post( f"{MCP_URL}/rpc", json={ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "serp_feature_checker", "arguments": {"keyword": keyword} } } ) if response.status_code == 200: result = response.json().get("result", {}) content = result.get("content", []) if content: import json data = json.loads(content[0].get("text", "{}")) for feature_type in ["snippet", "paa", "ai_overview"]: if data.get(f"has_{feature_type}"): features.append(SERPFeatureData( keyword=keyword, feature_type=feature_type, current_position=data.get("our_position"), captured=data.get(f"we_own_{feature_type}", False), captured_at=None, content_hash=None, ai_overview_mention=data.get("ai_overview_mentions_us", False), citation_context=data.get("citation_context"), optimized=False, optimization_applied=None, )) except Exception as e: print(f"[ResearchNode] SERP check error for '{keyword}': {e}") continue except Exception as e: print(f"[ResearchNode] SERP detection error: {e}") # Fallback: create placeholder opportunities for top keywords if not features: for keyword in keywords[:10]: for feature_type in ["snippet", "paa"]: features.append(SERPFeatureData( keyword=keyword, feature_type=feature_type, current_position=None, captured=False, captured_at=None, content_hash=None, ai_overview_mention=False, citation_context=None, optimized=False, optimization_applied=None, )) return features async def research_phase(state: SEOCampaignState) -> SEOCampaignState: """ Research phase node Gathers: - Google Search Console data - Competitor analysis - SERP feature opportunities """ print(f"[ResearchNode] Starting research for site {state['site_id']}") site_id = state["site_id"] now = datetime.utcnow().isoformat() try: # Fetch GSC data gsc_data = await fetch_gsc_data(site_id) if gsc_data: state["gsc_data"] = gsc_data print(f"[ResearchNode] GSC data: {gsc_data['total_impressions']} impressions, {gsc_data['total_clicks']} clicks") else: state = add_warning(state, "research", "Could not fetch GSC data - using fallback keywords") # Use sample fallback keywords for testing gsc_data = None # Fetch competitor data competitor_data = await fetch_competitor_data(site_id) state["competitor_data"] = competitor_data print(f"[ResearchNode] Analyzed {len(competitor_data)} competitors") # Extract keywords for SERP feature detection keywords = [] if gsc_data: keywords = [q.get("query", "") for q in gsc_data.get("top_queries", [])] for comp in competitor_data: keywords.extend([k.get("keyword", "") for k in comp.get("top_keywords", [])]) # If no keywords from APIs, use real sports betting keywords from existing content if not keywords: keywords = [ # MLB betting keywords (from existing content) "mlb props betting strategy", "mlb moneyline value picks", "mlb totals sharp money", "mlb spread betting analytics", "mlb injury impact betting", # NBA betting keywords "nba player props predictions", "nba spread betting tips", "nba totals over under", "nba sharp money indicators", # General sports betting "what is sharp money betting", "how to fade the public", "best sports betting analytics", "sports betting value picks", "live odds comparison tools", ] state = add_warning(state, "research", "Using sports betting keywords - connect GSC for personalized data") keywords = list(set(keywords))[:50] # Dedupe and limit # Detect SERP feature opportunities serp_features = await detect_serp_features(keywords) state["serp_features_data"] = serp_features print(f"[ResearchNode] Found {len(serp_features)} SERP feature opportunities") # Transition to next phase state = transition_phase(state, CampaignPhase.CONTENT) state["updated_at"] = now except Exception as e: print(f"[ResearchNode] Error during research: {e}") state = add_error(state, "research", f"Research phase failed: {str(e)}") # Still transition to allow workflow to continue state = transition_phase(state, CampaignPhase.CONTENT) state["updated_at"] = now return state