""" Content Phase Node Generates and queues content based on research findings """ 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, ContentQueueItem, ) NEXT_API_URL = os.getenv("NEXT_API_URL", "http://127.0.0.1:3000") def prioritize_content_opportunities(state: SEOCampaignState) -> List[ContentQueueItem]: """ Analyze research data and prioritize content opportunities. Priority factors: - Keywords with SERP features we don't own - High-volume keywords from GSC with low position - Content gaps from competitor analysis - Keywords where we have AI Overview opportunity """ opportunities = [] now = datetime.utcnow().isoformat() # 1. SERP feature opportunities (highest priority) for feature in state.get("serp_features_data", []): if not feature.get("captured", False): priority = 10 # Highest priority for uncaptured features # AI Overview opportunities get extra priority if feature.get("feature_type") == "ai_overview": priority = 10 elif feature.get("feature_type") == "snippet": priority = 9 elif feature.get("feature_type") == "paa": priority = 8 opportunities.append(ContentQueueItem( id=str(uuid.uuid4()), keyword=feature.get("keyword", ""), content_type="serp_optimized_article", target_serp_features=[feature.get("feature_type", "")], priority=priority, status="pending", page_id=None, created_at=now, )) # 2. GSC opportunities (low position, high impressions) gsc_data = state.get("gsc_data") if gsc_data: for query in gsc_data.get("top_queries", [])[:20]: keyword = query.get("query", "") position = query.get("position", 100) impressions = query.get("impressions", 0) # Already have content for this keyword? Skip if any(o["keyword"] == keyword for o in opportunities): continue # Priority based on position and impressions if position > 10 and impressions > 100: priority = 7 # Striking distance - can reach page 1 elif position > 20 and impressions > 50: priority = 5 # Longer term opportunity else: continue # Skip low-opportunity keywords opportunities.append(ContentQueueItem( id=str(uuid.uuid4()), keyword=keyword, content_type="seo_article", target_serp_features=["snippet", "paa"], # Target common features priority=priority, status="pending", page_id=None, created_at=now, )) # 3. Content gaps from competitors for comp in state.get("competitor_data", []): for gap in comp.get("content_gaps", [])[:10]: keyword = gap.get("keyword", "") if any(o["keyword"] == keyword for o in opportunities): continue # Priority based on search volume volume = gap.get("searchVolume", 0) if volume > 1000: priority = 6 elif volume > 500: priority = 5 else: priority = 4 opportunities.append(ContentQueueItem( id=str(uuid.uuid4()), keyword=keyword, content_type="competitive_content", target_serp_features=["snippet", "paa"], priority=priority, status="pending", page_id=None, created_at=now, )) # Sort by priority (highest first) and limit opportunities.sort(key=lambda x: x["priority"], reverse=True) return opportunities[:50] # Limit to top 50 async def queue_content_generation( opportunity: ContentQueueItem, site_id: str ) -> str | None: """Queue a content generation job via the Next.js API""" try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( f"{NEXT_API_URL}/api/admin/seo/content/queue", headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_TOKEN', '')}"}, json={ "siteId": site_id, "keyword": opportunity["keyword"], "contentType": opportunity["content_type"], "targetFeatures": opportunity["target_serp_features"], "priority": opportunity["priority"], "source": "langgraph_campaign", } ) if response.status_code in (200, 201): data = response.json() return data.get("jobId") except Exception as e: print(f"[ContentNode] Queue error for '{opportunity['keyword']}': {e}") return None async def content_phase(state: SEOCampaignState) -> SEOCampaignState: """ Content phase node - Analyzes research data to prioritize content opportunities - Queues content generation jobs - Updates state with content queue """ print(f"[ContentNode] Starting content phase for site {state['site_id']}") site_id = state["site_id"] now = datetime.utcnow().isoformat() try: # Prioritize content opportunities opportunities = prioritize_content_opportunities(state) print(f"[ContentNode] Prioritized {len(opportunities)} content opportunities") # Update content queue in state state["content_queue"] = opportunities # Queue top priority items for generation queued_count = 0 max_to_queue = state.get("config", {}).get("max_content_per_run", 10) for opp in opportunities[:max_to_queue]: try: job_id = await queue_content_generation(opp, site_id) if job_id: # Update opportunity status opp["status"] = "queued" queued_count += 1 except Exception as e: print(f"[ContentNode] Failed to queue '{opp.get('keyword', '')}': {e}") # Continue with other opportunities continue # If no content API available, just mark opportunities as pending if queued_count == 0 and opportunities: state = add_warning(state, "content", "Content API unavailable - opportunities identified but not queued") # Mark first few as simulated queued for demo purposes for opp in opportunities[:5]: opp["status"] = "pending" queued_count = min(5, len(opportunities)) print(f"[ContentNode] Queued {queued_count} content generation jobs") state["pages_generated"] = queued_count except Exception as e: print(f"[ContentNode] Error during content phase: {e}") state = add_error(state, "content", f"Content phase failed: {str(e)}") state["content_queue"] = [] state["pages_generated"] = 0 # Transition to SERP features phase state = transition_phase(state, CampaignPhase.SERP_FEATURES) state["updated_at"] = now return state