""" Performance and Reporting Node Generates campaign reports and tracks metrics """ import os import httpx from datetime import datetime from typing import Any, Dict, List from state.campaign_state import ( SEOCampaignState, CampaignPhase, transition_phase, add_error, ) NEXT_API_URL = os.getenv("NEXT_API_URL", "http://127.0.0.1:3000") async def save_campaign_to_db(state: SEOCampaignState) -> bool: """Save campaign state to database via API""" try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( f"{NEXT_API_URL}/api/admin/seo/campaigns/save", headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_TOKEN', '')}"}, json={ "campaignId": state["campaign_id"], "siteId": state["site_id"], "campaignType": state["campaign_type"], "currentPhase": state["current_phase"], "stateSnapshot": state, # Full state "status": "completed" if state["current_phase"] == CampaignPhase.COMPLETED.value else "active", "pagesGenerated": state.get("pages_generated", 0), "snippetsCaptured": state.get("snippets_captured", 0), "linksApproved": state.get("links_approved", 0), "linksAcquired": state.get("links_acquired", 0), } ) return response.status_code in (200, 201) except Exception as e: print(f"[PerformanceNode] Save error: {e}") return False def generate_metrics_summary(state: SEOCampaignState) -> Dict[str, Any]: """Generate a metrics summary for the campaign""" # Count SERP features by type serp_features = state.get("serp_features_data", []) snippet_opps = len([f for f in serp_features if f.get("feature_type") == "snippet"]) paa_opps = len([f for f in serp_features if f.get("feature_type") == "paa"]) ai_overview_opps = len([f for f in serp_features if f.get("feature_type") == "ai_overview"]) # Count content queue status content_queue = state.get("content_queue", []) content_queued = len([c for c in content_queue if c.get("status") == "queued"]) content_completed = len([c for c in content_queue if c.get("status") == "completed"]) # Count optimization queue status opt_queue = state.get("optimization_queue", []) opt_queued = len([o for o in opt_queue if o.get("status") == "queued"]) opt_completed = len([o for o in opt_queue if o.get("status") == "completed"]) # Count link opportunities link_opps = state.get("link_opportunities", []) links_approved = len([l for l in link_opps if l.get("status") == "approved"]) links_rejected = len([l for l in link_opps if l.get("status") == "rejected"]) links_pending = len([l for l in link_opps if l.get("status") == "pending"]) links_acquired = len([l for l in link_opps if l.get("link_acquired")]) # GSC summary gsc_data = state.get("gsc_data") gsc_summary = None if gsc_data: gsc_summary = { "total_impressions": gsc_data.get("total_impressions", 0), "total_clicks": gsc_data.get("total_clicks", 0), "average_ctr": round(gsc_data.get("average_ctr", 0) * 100, 2), "average_position": round(gsc_data.get("average_position", 0), 1), } # Competitor summary competitor_summary = [] for comp in state.get("competitor_data", []): competitor_summary.append({ "domain": comp.get("domain"), "domain_authority": comp.get("domain_authority"), "content_gaps_found": len(comp.get("content_gaps", [])), }) return { "campaign_id": state["campaign_id"], "site_id": state["site_id"], "campaign_type": state["campaign_type"], "phases_completed": state.get("phases_completed", []), "duration_seconds": calculate_duration(state), "serp_features": { "snippet_opportunities": snippet_opps, "paa_opportunities": paa_opps, "ai_overview_opportunities": ai_overview_opps, "total_opportunities": len(serp_features), }, "content": { "pages_queued": content_queued, "pages_completed": content_completed, "pages_generated": state.get("pages_generated", 0), }, "optimization": { "optimizations_queued": opt_queued, "optimizations_completed": opt_completed, "pages_optimized": state.get("pages_optimized", 0), }, "links": { "opportunities_found": len(link_opps), "pending_approval": links_pending, "approved": links_approved, "rejected": links_rejected, "acquired": links_acquired, }, "gsc": gsc_summary, "competitors_analyzed": competitor_summary, "errors": len(state.get("errors", [])), "warnings": len(state.get("warnings", [])), "generated_at": datetime.utcnow().isoformat(), } def calculate_duration(state: SEOCampaignState) -> int: """Calculate campaign duration in seconds""" try: started = datetime.fromisoformat(state.get("started_at", "")) ended = datetime.fromisoformat(state.get("updated_at", "")) return int((ended - started).total_seconds()) except Exception: return 0 def generate_recommendations(state: SEOCampaignState) -> List[str]: """Generate actionable recommendations based on campaign results""" recommendations = [] # SERP feature recommendations serp_features = state.get("serp_features_data", []) uncaptured_snippets = [f for f in serp_features if f.get("feature_type") == "snippet" and not f.get("captured")] if uncaptured_snippets: recommendations.append( f"Optimize {len(uncaptured_snippets)} pages for featured snippets - " f"focus on 'What is' and 'How to' query formats with 40-60 word answers" ) ai_overview_opps = [f for f in serp_features if f.get("feature_type") == "ai_overview"] if ai_overview_opps: recommendations.append( f"Target {len(ai_overview_opps)} AI Overview opportunities - " f"ensure content has unique data points and clear authority signals" ) # Content recommendations content_queue = state.get("content_queue", []) high_priority = [c for c in content_queue if c.get("priority", 0) >= 8] if high_priority: recommendations.append( f"Prioritize {len(high_priority)} high-priority content pieces - " f"these have the best SERP feature potential" ) # Link recommendations link_opps = state.get("link_opportunities", []) pending = [l for l in link_opps if l.get("status") == "pending"] if pending: recommendations.append( f"Review {len(pending)} pending link opportunities - " f"all opportunities require human approval before outreach" ) # GSC recommendations gsc_data = state.get("gsc_data") if gsc_data: if gsc_data.get("average_ctr", 0) < 0.03: recommendations.append( "CTR is below 3% - consider optimizing title tags and meta descriptions " "for click-through improvement" ) if gsc_data.get("average_position", 100) > 15: recommendations.append( "Average position is outside top 15 - focus on content quality " "and internal linking to improve rankings" ) # Default recommendation if none generated if not recommendations: recommendations.append( "Continue monitoring SERP features and competitor movements - " "run this campaign weekly for best results" ) return recommendations async def generate_report_phase(state: SEOCampaignState) -> SEOCampaignState: """ Report generation phase node - Generates comprehensive campaign metrics - Creates actionable recommendations - Saves final state to database - Marks campaign as completed """ print(f"[PerformanceNode] Generating report for campaign {state['campaign_id']}") now = datetime.utcnow().isoformat() try: # Generate metrics summary metrics = generate_metrics_summary(state) print(f"[PerformanceNode] Metrics: {metrics['content']['pages_generated']} pages, " f"{metrics['links']['opportunities_found']} link opportunities") # Generate recommendations recommendations = generate_recommendations(state) print(f"[PerformanceNode] Generated {len(recommendations)} recommendations") # Add report to state config for retrieval state["config"] = state.get("config", {}) state["config"]["final_report"] = { "metrics": metrics, "recommendations": recommendations, "errors": state.get("errors", []), "warnings": state.get("warnings", []), } # Try to save to database (optional - campaign still completes if this fails) try: saved = await save_campaign_to_db(state) if not saved: print("[PerformanceNode] Warning: Could not save campaign to database API") except Exception as e: print(f"[PerformanceNode] Database save error (non-fatal): {e}") except Exception as e: print(f"[PerformanceNode] Error generating report: {e}") state = add_error(state, "report", f"Report generation error: {str(e)}") # Still set a minimal report state["config"] = state.get("config", {}) state["config"]["final_report"] = { "metrics": {"error": str(e)}, "recommendations": ["Check system configuration and retry"], "errors": state.get("errors", []), "warnings": state.get("warnings", []), } # Mark campaign as completed state = transition_phase(state, CampaignPhase.COMPLETED) state["completed_at"] = now state["updated_at"] = now print(f"[PerformanceNode] Campaign completed") return state