Upload 10 files

Dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    curl \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Download Hugging Face models during build (cached in image)
+RUN python -c "from transformers import pipeline; pipeline('token-classification', model='ml6team/keyphrase-extraction-distilbert-inspec')"
+RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('all-MiniLM-L6-v2')"
+RUN python -c "from transformers import BartForConditionalGeneration, BartTokenizer; BartForConditionalGeneration.from_pretrained('facebook/bart-large-cnn'); BartTokenizer.from_pretrained('facebook/bart-large-cnn')"
+
+# Copy application code
+COPY . .
+
+# Expose port (default for HF Spaces)
+EXPOSE 7860
+
+# Run FastAPI server
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,35 @@
 ---
-title: SEO
-emoji: 🐢
+title: SEO Multi-Agent System
+emoji: 🚀
 colorFrom: blue
-colorTo: purple
+colorTo: green
 sdk: docker
+app_port: 7860
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# SEO Multi-Agent System
+
+Autonomous 24/7 SEO optimization with zero manual intervention.
+
+## Features
+
+- **Technical Auditing**: Automated crawl and quality checks.
+- **Content Optimization**: AI-driven keyword and meta tag generation.
+- **Competitor Intelligence**: SERP analysis (via Playwright).
+- **Automated Indexing**: Google Search Console & IndexNow submission.
+- **Performance Analytics**: GA4 & GSC monitoring.
+
+## Configuration
+
+This system is designed to be triggered by a Cloudflare Worker request.
+
+### Environment Variables (Secrets)
+Set these in your Space settings:
+- `GSC_CREDENTIALS_JSON`: Content of your Google Search Console JSON key.
+- `GA4_CREDENTIALS_JSON`: Content of your GA4 JSON key.
+- `INDEXNOW_KEY`: Your IndexNow API key.
+
+## Deployment
+
+This Space runs automatically on Hugging Face infrastructure using the Docker SDK.

agents/backlink_indexing.py

@@ -0,0 +1,79 @@
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+import requests
+import logging
+import os
+import xml.etree.ElementTree as ET
+from datetime import datetime, timedelta
+
+logger = logging.getLogger(__name__)
+
+class BacklinkIndexingAgent:
+    def __init__(self, gsc_credentials_path, site_url):
+        self.site_url = site_url
+        self.gsc = None
+        
+        if os.path.exists(gsc_credentials_path):
+            try:
+                credentials = service_account.Credentials.from_service_account_file(
+                    gsc_credentials_path,
+                    scopes=['https://www.googleapis.com/auth/webmasters']
+                )
+                self.gsc = build('searchconsole', 'v1', credentials=credentials)
+            except Exception as e:
+                logger.error(f"Failed to initialize GSC client: {e}")
+        else:
+             logger.warning(f"GSC credentials not found at {gsc_credentials_path}")
+
+    def submit_to_indexnow(self, urls, api_key="your-key", key_location=""):
+        try:
+            endpoint = "https://api.indexnow.org/IndexNow"
+            host = self.site_url.replace('https://', '').replace('http://', '')
+            payload = {
+                "host": host,
+                "key": api_key,
+                "keyLocation": key_location or f"{self.site_url}/indexnow-key.txt",
+                "urlList": urls
+            }
+            response = requests.post(endpoint, json=payload, timeout=10)
+            return response.status_code == 200
+        except Exception as e:
+            logger.error(f"IndexNow submission failed: {e}")
+            return False
+
+    def submit_to_google(self, url):
+        if not self.gsc: return False
+        try:
+            body = {'url': url, 'type': 'URL_UPDATED'}
+            self.gsc.urlInspection().index().inspect(
+                siteUrl=self.site_url,
+                body=body
+            ).execute()
+            return True
+        except Exception as e:
+            logger.error(f"GSC submission failed for {url}: {e}")
+            return False
+
+    def auto_submit_new_pages(self, new_urls):
+        results = {'indexnow': False, 'google': []}
+        
+        # IndexNow
+        if new_urls:
+            results['indexnow'] = self.submit_to_indexnow(new_urls)
+            
+        # Google (Limit 200/day normally)
+        for url in new_urls[:10]: 
+            full_url = url if url.startswith('http') else f"{self.site_url}{url}"
+            success = self.submit_to_google(full_url)
+            results['google'].append({'url': url, 'success': success})
+            
+        return results
+
+    def check_indexing_status(self):
+        if not self.gsc: return []
+        # In a real impl, query searchAnalytics or urlInspection
+        # Placeholder for demonstration
+        return []
+
+    def get_indexing_errors(self):
+        return []

agents/competitor_intelligence.py

@@ -0,0 +1,72 @@
+from playwright.sync_api import sync_playwright
+from sentence_transformers import SentenceTransformer, util
+import time
+import logging
+
+logger = logging.getLogger(__name__)
+
+class CompetitorIntelligenceAgent:
+    def __init__(self):
+        try:
+            self.semantic_model = SentenceTransformer('all-mpnet-base-v2')
+        except Exception as e:
+            logger.error(f"Failed to load semantic model: {e}")
+
+    def scrape_serp(self, keyword):
+        results = []
+        try:
+            with sync_playwright() as p:
+                browser = p.chromium.launch(headless=True)
+                page = browser.new_page()
+                
+                # Search Google (Note: frequent scraping might get blocked)
+                page.goto(f"https://www.google.com/search?q={keyword}&num=10")
+                time.sleep(2) # Be polite
+                
+                # Extract results (Selectors might need maintenance)
+                links = page.locator('div.g a').all()
+                for link in links:
+                    url = link.get_attribute('href')
+                    if url and 'google' not in url:
+                        results.append(url)
+                        if len(results) >= 5: break
+                
+                browser.close()
+        except Exception as e:
+            logger.error(f"SERP scraping failed: {e}")
+        
+        return results
+
+    def analyze_content_gaps(self, your_content, competitor_contents):
+        if not hasattr(self, 'semantic_model'): return []
+        
+        your_emb = self.semantic_model.encode(your_content, convert_to_tensor=True)
+        comp_embs = self.semantic_model.encode(competitor_contents, convert_to_tensor=True)
+        
+        similarities = util.cos_sim(your_emb, comp_embs)
+        
+        gaps = []
+        for idx, sim in enumerate(similarities[0]):
+            if sim < 0.6:
+                gaps.append({
+                   "competitor_idx": idx,
+                   "similarity": float(sim),
+                   "note": "Topic gap detected"
+                })
+        return gaps
+
+    def generate_competitive_report(self, keyword):
+        logger.info(f"Generating report for keyword: {keyword}")
+        
+        competitor_urls = self.scrape_serp(keyword)
+        competitor_contents = []
+        
+        # In a real run, we would visit each URL to get content
+        # For this skeleton, we'll placeholder
+        
+        return {
+            "keyword": keyword,
+            "top_competitors": competitor_urls,
+            "analysis_status": "completed_basic",
+            "message": "Deep content analysis requires full scraping infrastructure which is rate-limit sensitive."
+        }

agents/content_optimizer.py

@@ -0,0 +1,80 @@
+from transformers import pipeline, BartForConditionalGeneration, BartTokenizer, T5ForConditionalGeneration, T5Tokenizer
+from sentence_transformers import SentenceTransformer, util
+import requests
+from bs4 import BeautifulSoup
+import logging
+
+logger = logging.getLogger(__name__)
+
+class ContentOptimizationAgent:
+    def __init__(self):
+        try:
+            self.keyword_extractor = pipeline(
+                "token-classification",
+                model="ml6team/keyphrase-extraction-distilbert-inspec",
+                aggregation_strategy="simple"
+            )
+            self.meta_model = BartForConditionalGeneration.from_pretrained("facebook/bart-large-cnn")
+            self.meta_tokenizer = BartTokenizer.from_pretrained("facebook/bart-large-cnn")
+            self.semantic_model = SentenceTransformer('all-MiniLM-L6-v2')
+            
+            # Load T5 only if needed to save memory, or here if sufficient RAM (Space has 16GB)
+            self.title_model = T5ForConditionalGeneration.from_pretrained("google/flan-t5-large")
+            self.title_tokenizer = T5Tokenizer.from_pretrained("google/flan-t5-large")
+        except Exception as e:
+            logger.error(f"Failed to load models for Content Optimizer: {e}")
+
+    def extract_keywords(self, text):
+        if not hasattr(self, 'keyword_extractor'): return []
+        results = self.keyword_extractor(text[:512]) # Limit for speed
+        keywords = [r['word'] for r in results if r['score'] > 0.7]
+        return list(set(keywords))
+
+    def generate_meta_description(self, text, max_length=160):
+        if not hasattr(self, 'meta_model'): return ""
+        inputs = self.meta_tokenizer(text[:1024], return_tensors="pt", truncation=True)
+        summary_ids = self.meta_model.generate(
+            inputs["input_ids"],
+            max_length=max_length,
+            min_length=100,
+            length_penalty=2.0,
+            num_beams=4,
+            early_stopping=True
+        )
+        return self.meta_tokenizer.decode(summary_ids, skip_special_tokens=True)
+
+    def optimize_title(self, content, keyword):
+        if not hasattr(self, 'title_model'): return ""
+        prompt = f"Write an SEO-optimized page title under 60 characters for content about '{keyword}': {content[:500]}"
+        inputs = self.title_tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512)
+        outputs = self.title_model.generate(inputs["input_ids"], max_length=60, num_beams=5)
+        return self.title_tokenizer.decode(outputs, skip_special_tokens=True)
+
+    def analyze_page(self, url):
+        logger.info(f"Optimizing content for: {url}")
+        try:
+            response = requests.get(url, timeout=10)
+            soup = BeautifulSoup(response.text, 'html.parser')
+            content = soup.get_text(separator=' ', strip=True)
+            
+            keywords = self.extract_keywords(content)
+            suggested_meta = self.generate_meta_description(content)
+            
+            # Detect main keyword (simple frequency for now, or use first extracted)
+            main_keyword = keywords[0] if keywords else "general"
+            suggested_title = self.optimize_title(content, main_keyword)
+
+            recommendations = []
+            if len(keywords) < 3:
+                recommendations.append("Content might be too thin; few keywords detected.")
+            
+            return {
+                "url": url,
+                "extracted_keywords": keywords,
+                "suggested_title": suggested_title,
+                "suggested_meta_description": suggested_meta,
+                "recommendations": recommendations
+            }
+        except Exception as e:
+            logger.error(f"Content optimization failed: {e}")
+            return {"error": str(e)}

agents/orchestrator.py

@@ -0,0 +1,45 @@
+import redis
+import json
+import logging
+from datetime import datetime
+
+logger = logging.getLogger(__name__)
+
+class OrchestratorAgent:
+    def __init__(self, redis_host='localhost', redis_port=6379, agents=None):
+        self.agents = agents or {}
+        try:
+            self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
+            # Test connection
+            self.redis.ping()
+        except Exception as e:
+            logger.warning(f"Redis connection failed (Orchestrator): {e}")
+            self.redis = None
+
+    def create_workflow(self, workflow_name, payload):
+        workflow_id = f"workflow_{workflow_name}_{datetime.now().timestamp()}"
+        logger.info(f"Creating workflow {workflow_id}")
+        
+        # In a real system, you'd decompose this into steps and enqueue them
+        # For this prototype, we just log it and maybe store state
+        
+        if self.redis:
+            try:
+                self.redis.set(workflow_id, json.dumps({
+                    "name": workflow_name,
+                    "payload": payload,
+                    "status": "created",
+                    "created_at": datetime.now().isoformat()
+                }))
+                self.redis.expire(workflow_id, 86400) # 1 day expiry
+            except Exception as e:
+                logger.error(f"Failed to save workflow state: {e}")
+                
+        return workflow_id
+
+    def monitor_agent_health(self):
+        # Check system health
+        return {
+            "orchestrator": "healthy",
+            "redis": "connected" if self.redis else "disconnected"
+        }

agents/performance_analytics.py

@@ -0,0 +1,113 @@
+from google.analytics.data_v1beta import BetaAnalyticsDataClient
+from google.analytics.data_v1beta.types import RunReportRequest, DateRange, Metric, Dimension
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+import pandas as pd
+from datetime import datetime, timedelta
+import requests
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+class PerformanceAnalyticsAgent:
+    def __init__(self, ga4_property_id, ga4_credentials_path, gsc_credentials_path, site_url):
+        self.property_id = ga4_property_id
+        self.site_url = site_url
+        self.ga4_client = None
+        self.gsc = None
+
+        if os.path.exists(ga4_credentials_path):
+            try:
+                creds = service_account.Credentials.from_service_account_file(ga4_credentials_path)
+                self.ga4_client = BetaAnalyticsDataClient(credentials=creds)
+            except Exception as e: logger.error(f"GA4 init failed: {e}")
+        
+        if os.path.exists(gsc_credentials_path):
+            try:
+                creds = service_account.Credentials.from_service_account_file(gsc_credentials_path)
+                self.gsc = build('searchconsole', 'v1', credentials=creds)
+            except Exception as e: logger.error(f"GSC init failed: {e}")
+
+    def get_underperforming_pages(self, days=30):
+        if not self.gsc: return []
+        try:
+            response = self.gsc.searchanalytics().query(
+                siteUrl=self.site_url,
+                body={
+                    'startDate': (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d'),
+                    'endDate': datetime.now().strftime('%Y-%m-%d'),
+                    'dimensions': ['page'],
+                    'rowLimit': 100
+                }
+            ).execute()
+            
+            underperforming = []
+            for row in response.get('rows', []):
+                ctr = row['ctr']
+                if row['impressions'] > 100 and ctr < 0.02:
+                    underperforming.append({
+                        'url': row['keys'][0],
+                        'impressions': row['impressions'],
+                        'ctr': ctr
+                    })
+            return underperforming
+        except Exception as e:
+            logger.error(f"GSC query failed: {e}")
+            return []
+
+    def detect_algorithm_update(self):
+        # Simplified volatility check
+        # In production this would compare day-over-day ranking distributions
+        return {
+            'volatility_detected': False,
+            'avg_position_change': 0.0
+        }
+
+    def generate_weekly_report(self):
+        report = {
+            'summary': {'total_pageviews': 0},
+            'top_pages': [],
+            'underperforming_pages': []
+        }
+        
+        if self.ga4_client:
+            try:
+                request = RunReportRequest(
+                    property=f"properties/{self.property_id}",
+                    date_ranges=[DateRange(start_date="7daysAgo", end_date="today")],
+                    dimensions=[Dimension(name="pagePath")],
+                    metrics=[Metric(name="screenPageViews")]
+                )
+                response = self.ga4_client.run_report(request)
+                
+                total_pv = 0
+                for row in response.rows:
+                    pv = int(row.metric_values[0].value)
+                    total_pv += pv
+                    report['top_pages'].append({
+                        'path': row.dimension_values[0].value,
+                        'views': pv
+                    })
+                report['summary']['total_pageviews'] = total_pv
+            except Exception as e:
+                logger.error(f"GA4 report failed: {e}")
+
+        return report
+
+    def monitor_core_web_vitals(self, url):
+        # Uses public PageSpeed Insights API
+        try:
+            psi_url = "https://www.googleapis.com/pagespeedonline/v5/runPagespeed"
+            params = {'url': url, 'strategy': 'mobile', 'category': ['performance', 'seo']}
+            resp = requests.get(psi_url, params=params)
+            data = resp.json()
+            
+            lighthouse = data.get('lighthouseResult', {})
+            return {
+                'url': url,
+                'performance_score': lighthouse.get('categories', {}).get('performance', {}).get('score')
+            }
+        except Exception as e:
+            logger.error(f"CWV check failed: {e}")
+            return {'error': str(e)}

agents/technical_auditor.py

@@ -0,0 +1,75 @@
+from transformers import pipeline
+import requests
+from bs4 import BeautifulSoup
+import logging
+
+logger = logging.getLogger(__name__)
+
+class TechnicalAuditorAgent:
+    def __init__(self):
+        try:
+            self.classifier = pipeline(
+                "zero-shot-classification",
+                model="facebook/bart-large-mnli",
+                device=-1  # CPU inference
+            )
+        except Exception as e:
+             logger.warning(f"Could not load ML models for Technical Auditor: {e}")
+             self.classifier = None
+
+    def assess_content_quality(self, page_text):
+        if not self.classifier:
+            return ["error-model-not-loaded"]
+            
+        result = self.classifier(
+            page_text[:1024], # Limit text length for speed
+            candidate_labels=["high-quality", "thin-content", "spam", "keyword-stuffing"]
+        )
+        return result['labels']
+
+    def audit_page(self, url):
+        logger.info(f"Auditing page: {url}")
+        results = {
+            "url": url,
+            "broken_links": [],
+            "missing_alt": [],
+            "content_quality": "unknown"
+        }
+        
+        try:
+            # Note: In a real scenario, you might need to handle headers/user-agents
+            response = requests.get(url, timeout=10)
+            if response.status_code != 200:
+                logger.error(f"Page returned status {response.status_code}")
+                return results
+
+            soup = BeautifulSoup(response.text, 'html.parser')
+            
+            # Check content quality
+            text_content = soup.get_text(separator=' ', strip=True)
+            results['content_quality'] = self.assess_content_quality(text_content)
+
+            # Check images
+            images = soup.find_all('img')
+            for img in images:
+                if not img.get('alt'):
+                    results['missing_alt'].append(img.get('src'))
+
+            # Check links (simplified)
+            links = soup.find_all('a')
+            for link in links:
+                href = link.get('href')
+                if href and href.startswith('http'):
+                    try:
+                        head = requests.head(href, timeout=5)
+                        if head.status_code >= 400:
+                            results['broken_links'].append(href)
+                    except:
+                         results['broken_links'].append(href)
+
+        except Exception as e:
+            logger.error(f"Audit failed for {url}: {e}")
+        
+        logger.info(f"Audit complete for {url}: {results}")
+        # In a real system, you would post these results back to a webhook or database
+        return results

main.py

@@ -0,0 +1,142 @@
+from fastapi import FastAPI, BackgroundTasks, HTTPException, Request
+from pydantic import BaseModel
+from typing import List, Optional, Dict, Any
+import logging
+import os
+import json
+
+# Import agents (we will create these files next)
+from agents.technical_auditor import TechnicalAuditorAgent
+from agents.content_optimizer import ContentOptimizationAgent
+from agents.competitor_intelligence import CompetitorIntelligenceAgent
+from agents.backlink_indexing import BacklinkIndexingAgent
+from agents.performance_analytics import PerformanceAnalyticsAgent
+from agents.orchestrator import OrchestratorAgent
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(title="SEO Multi-Agent System", version="1.0.0")
+
+# Initialize agents
+# Note: In a real deployment, credentials would be loaded from env vars or mounted secrets
+technical_agent = TechnicalAuditorAgent()
+content_agent = ContentOptimizationAgent()
+competitor_agent = CompetitorIntelligenceAgent()
+indexing_agent = BacklinkIndexingAgent(
+    gsc_credentials_path="/app/credentials/gsc-credentials.json",
+    site_url="https://yoursite.pages.dev" # Replace with actual site
+)
+performance_agent = PerformanceAnalyticsAgent(
+    ga4_property_id="your-property-id", # Replace
+    ga4_credentials_path="/app/credentials/ga4-credentials.json",
+    gsc_credentials_path="/app/credentials/gsc-credentials.json",
+    site_url="https://yoursite.pages.dev" # Replace
+)
+orchestrator = OrchestratorAgent(redis_host="redis", redis_port=6379, agents={}) # Redis host depends on service name in docker-compose or similar
+
+# --- MODELS ---
+
+class PageAuditRequest(BaseModel):
+    url: str
+    checks: Optional[List[str]] = ["all"]
+
+class ContentOptimizeRequest(BaseModel):
+    url: str
+    tasks: Optional[List[str]] = ["all"]
+
+class CompetitorAnalysisRequest(BaseModel):
+    keyword: str
+
+class IndexingRequest(BaseModel):
+    urls: List[str]
+
+class WorkflowTrigger(BaseModel):
+    workflow_name: str
+    payload: Dict[str, Any]
+
+# --- ENDPOINTS ---
+
+@app.get("/")
+async def root():
+    return {
+        "service": "SEO Multi-Agent System",
+        "status": "running",
+        "agents": ["technical", "content", "competitor", "indexing", "performance"]
+    }
+
+@app.get("/health")
+async def health_check():
+    # Basic check, can expand to check agents
+    return {"status": "healthy", "orchestrator": "online"}
+
+# TECHNICAL AUDITOR
+@app.post("/audit/execute")
+async def execute_audit(request: PageAuditRequest, background_tasks: BackgroundTasks):
+    background_tasks.add_task(technical_agent.audit_page, request.url)
+    return {"status": "started", "url": request.url}
+
+# CONTENT OPTIMIZER
+@app.post("/optimize")
+async def optimize_content(request: ContentOptimizeRequest):
+    result = content_agent.analyze_page(request.url)
+    return result
+
+# COMPETITOR INTELLIGENCE
+@app.post("/competitor/analyze")
+async def analyze_competitor(request: CompetitorAnalysisRequest, background_tasks: BackgroundTasks):
+    background_tasks.add_task(competitor_agent.generate_competitive_report, request.keyword)
+    return {"status": "started", "keyword": request.keyword}
+
+# INDEXING
+@app.post("/index/submit")
+async def submit_indexing(request: IndexingRequest):
+    results = indexing_agent.auto_submit_new_pages(request.urls)
+    return results
+
+@app.get("/index/status")
+async def indexing_status():
+    status = indexing_agent.check_indexing_status()
+    errors = indexing_agent.get_indexing_errors()
+    return {"indexed_pages": status, "errors": errors}
+
+# PERFORMANCE & ANALYTICS
+@app.get("/analytics/underperforming")
+async def get_underperforming():
+    try:
+        pages = performance_agent.get_underperforming_pages(days=30)
+        return pages
+    except Exception as e:
+        logger.error(f"Error getting underperforming pages: {e}")
+        return []
+
+@app.get("/report/weekly")
+async def weekly_report():
+    try:
+        report = performance_agent.generate_weekly_report()
+        return report
+    except Exception as e:
+         logger.error(f"Error generating weekly report: {e}")
+         return {"error": str(e)}
+
+@app.get("/monitor/algorithm")
+async def check_algorithm():
+    try:
+        status = performance_agent.detect_algorithm_update()
+        return status
+    except Exception as e:
+        return {"volatility_detected": False, "error": str(e)}
+
+@app.get("/monitor/cwv")
+async def check_cwv(url: str):
+    result = performance_agent.monitor_core_web_vitals(url)
+    return result
+
+# ORCHESTRATOR
+@app.post("/workflow/trigger")
+async def trigger_workflow(request: WorkflowTrigger):
+    workflow_id = orchestrator.create_workflow(
+        request.workflow_name,
+        request.payload
+    )
+    return {"workflow_id": workflow_id, "status": "started"}

requirements.txt

@@ -0,0 +1,21 @@
+fastapi==0.109.0
+uvicorn[standard]==0.27.0
+transformers==4.36.0
+torch==2.1.0
+sentence-transformers==2.3.1
+beautifulsoup4==4.12.3
+scrapy==2.11.0
+playwright==1.41.0
+pandas==2.2.0
+numpy==1.26.3
+scikit-learn==1.4.0
+redis==5.0.1
+psycopg2-binary==2.9.9
+google-api-python-client==2.116.0
+google-auth==2.27.0
+google-analytics-data==0.18.5
+apscheduler==3.10.4
+requests==2.31.0
+pillow==10.2.0
+lxml==5.1.0
+keybert==0.8.4