Code Guide for Implementing Gemini AI and ScrapeGraph to Create an Automated and Scalable Competitive Intelligence Workflow.

%pip install --quiet -U langchain-scrapegraph langchain-google-genai pandas matplotlib seaborn

import getpass
Import os
Download json
import pandas as pd
Import List, Dict or Any
Datetime can be imported from another datetime
Matplotlib.pyplot can be imported as a plt
Buy Seaborn as sns

If not os.environ.get"SGAI_API_KEY"):
    os.environ["SGAI_API_KEY"] = getpass.getpass("ScrapeGraph AI API key:n")


If not, os.environ.get ("GOOGLE_API_KEY"):
    os.environ["GOOGLE_API_KEY"] = getpass.getpass("Google API key for Gemini:n")

from langchain_scrapegraph.tools import (
    SmartScraperTool,
    SearchScraperTool,
    MarkdownifyTool,
    GetCreditsTool,
)
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig, chain
from langchain_core.output_parsers import JsonOutputParser


SmartScraper Tool = smartscraper()
searchscraper = SearchScraperTool()
Markdownify Tool = markdownify()
GetCreditsTool()


llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-flash",
    temperature=0.1,
    convert_system_message_to_human=True
)

class CompetitiveAnalyzer:
    def __init__(self):
        self.results = []
        self.analysis_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
   
    def scrape_competitor_data(self, url: str, company_name: str = None) -> Dict[str, Any]:
        """Scrape comprehensive data from a competitor website"""
       
        extraction_prompt = """
 Extract the following information:
        1. Name of the company and its tagline
        2. Products/services that are offered
        3. If available, pricing information
        4. Target audience/market
        5. Benefits and features highlighted
        6. Technology stack mentioned
        7. Contact Information
        8. Social media presence
        9. News or Announcements Recent
        10. Team size indicators
        11. Funding information (if mentioned)
        12. Case studies or testimonials from customers
        13. Partner information
        14. Geographical presence/markets serviced
       
 Information is returned as a JSON-formatted structured data with a clearly defined categorization.
 Mark the information as "Not available" if it is unavailable.
        """
       
        try:
            result = smartscraper.invoke({
                "user_prompt": extraction_prompt,
                "website_url": url,
            })
           
            markdown_content = markdownify.invoke({"website_url": url})
           
            competitor_data = {
                "company_name"Name of company or "Unknown",
                "url": url,
                "scraped_data": result,
                "markdown_length": len(markdown_content),
                "analysis_date": self.analysis_timestamp,
                "success": True,
                "error"No
            }
           
 Return competitor_data
           
 Except Exception As e.Return
            return {
                "company_name"Name of company or "Unknown",
                "url": url,
                "scraped_data": None,
                "error": str(e),
                "success": False,
                "analysis_date": self.analysis_timestamp
            }
   
    def analyze_competitor_landscape(self, competitors: List[Dict[str, str]]) -> Dict[str, Any]:
        """Analyze multiple competitors and generate insights"""
       
        print(f"🔍 Starting competitive analysis for {len(competitors)} companies...")
       
        for i, competitor in enumerate(competitors, 1):
            print(f"📊 Analyzing {competitor['name']} ({i}/{len(competitors)})...")
           
            data = self.scrape_competitor_data(
 Competitor['url'],
 Due to the high demand for qualified candidates,['name']
            )
            self.results.append(data)
       
        analysis_prompt = ChatPromptTemplate.from_messages([
            ("system", """
            You are a senior business analyst specializing in competitive intelligence.
            Analyze the scraped competitor data and provide comprehensive insights including:
           
            1. Market positioning analysis
            2. Pricing strategy comparison
            3. Feature gap analysis  
            4. Target audience overlap
            5. Technology differentiation
            6. Market opportunities
            7. Competitive threats
            8. Strategic recommendations
           
            Provide actionable insights in JSON format with clear categories and recommendations.
            """),
            ("human", "Analyze this competitive data: {competitor_data}")
        ])
       
        clean_data = []
 For result in oneself.
 If result['success']:
                clean_data.append({
 'Company': result['company_name'],
 'url:' resultReturn['url'],
                    'data': result['scraped_data']
                })
       
        analysis_chain = analysis_prompt | llm | JsonOutputParser()
       
        try:
            competitive_analysis = analysis_chain.invoke({
                "competitor_data": json.dumps(clean_data, indent=2)
            })
        except:
            analysis_chain_text = analysis_prompt | llm
            competitive_analysis = analysis_chain_text.invoke({
                "competitor_data": json.dumps(clean_data, indent=2)
            })
       
        return {
            "analysis": competitive_analysis,
            "raw_data": self.results,
            "summary_stats": self.generate_summary_stats()
        }
   
    def generate_summary_stats(self) -> Dict[str, Any]:
        """Generate summary statistics from the analysis"""
 If r is a positive number, then successful_scrapes equals sum(1.Return['success'])
        failed_scrapes = len(self.results) - successful_scrapes
       
        return {
            "total_companies_analyzed": len(self.results),
            "successful_scrapes": successful_scrapes,
            "failed_scrapes""ailed_scrapes,
            "success_rate": f"{(successful_scrapes/len(self.results)*100):.1f}%" if self.results else "0%",
            "analysis_timestamp": self.analysis_timestamp
        }
   
 Export results by def (self, filename str = None).
        """Export results to JSON and CSV files"""
 Filename if it does not exist:
 Filename = F"competitive_analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
       
 With open(f)"{filename}.json""w" as f
            json.dump({
                "results": self.results,
                "summary": self.generate_summary_stats()
            }, f, indent=2)
       
        df_data = []
 For result in oneself.
 If result['success']:
                df_data.append({
 Results for 'Company:'['company_name'],
 "URL": results['url'],
 "Success": the result['success'],
                    'Data_Length': len(str(result['scraped_data']If result, then () is the condition['scraped_data'] Other than 0,
 Results of 'Analysis_Date:'['analysis_date']
                })
       
        if df_data:
            df = pd.DataFrame(df_data)
            df.to_csv(f"{filename}.csv", index=False)
           
        print(f"✅ Results exported to {filename}.json and {filename}.csv")

def run_ai_saas_analysis():
    """Run a comprehensive analysis of AI/SaaS competitors"""
   
    analyzer = CompetitiveAnalyzer()
   
    ai_saas_competitors = [
        {"name": "OpenAI", "url": "https://openai.com"},
        {"name": "Anthropic", "url": "https://anthropic.com"},
        {"name": "Hugging Face", "url": "https://huggingface.co"},
        {"name": "Cohere", "url": "https://cohere.ai"},
        {"name": "Scale AI", "url": "https://scale.com"},
    ]
   
    results = analyzer.analyze_competitor_landscape(ai_saas_competitors)
   
    print("n" + "="*80)
    print("🎯 COMPETITIVE ANALYSIS RESULTS")
    print("="*80)
   
    print(f"n📊 Summary Statistics:")
    stats = results['summary_stats']
 Stats.items for the key value():
        print(f"   {key.replace('_', ' ').title()}: {value}")
   
    print(f"n🔍 Strategic Analysis:")
 If isinstance (results)['analysis'], dict):
 Content in section['analysis'].items():
            print(f"n   {section.replace('_', ' ').title()}:")
 if content, list is instance:
 The following is the content of this article:
                    print(f"     • {item}")
            else:
                print(f"     {content}")
    else:
        print(results['analysis'])
   
    analyzer.export_results("ai_saas_competitive_analysis")
   
 Return results

def run_ecommerce_analysis():
    """Analyze e-commerce platform competitors"""
   
    analyzer = CompetitiveAnalyzer()
   
    ecommerce_competitors = [
        {"name": "Shopify", "url": "https://shopify.com"},
        {"name": "WooCommerce", "url": "https://woocommerce.com"},
        {"name": "BigCommerce", "url": "https://bigcommerce.com"},
        {"name": "Magento", "url": "https://magento.com"},
    ]
   
    results = analyzer.analyze_competitor_landscape(ecommerce_competitors)
    analyzer.export_results("ecommerce_competitive_analysis")
   
 Return results

@chain
def social_media_monitoring_chain(company_urls: List[str], config: RunnableConfig):
    """Monitor social media presence and engagement strategies of competitors"""
   
    social_media_prompt = ChatPromptTemplate.from_messages([
        ("system", """
        You are a social media strategist. Analyze the social media presence and strategies
        of these companies. Focus on:
        1. Platform presence (LinkedIn, Twitter, Instagram, etc.)
        2. Content strategy patterns
        3. Engagement tactics
        4. Community building approaches
        5. Brand voice and messaging
        6. Posting frequency and timing
        Provide actionable insights for improving social media strategy.
        """),
        ("human", "Analyze social media data for: {urls}")
    ])
   
    social_data = []
 For urls in company_urls
        try:
            result = smartscraper.invoke({
                "user_prompt": "Extract all social media links, community engagement features, and social proof elements",
                "website_url": url,
            })
            social_data.append({"url": url, "social_data": result})
 Except Exception As e."analysis = chain.invoke"Return
            social_data.append({"url": url, "error": str(e)})
   
    chain = social_media_prompt | llm
    analysis = chain.invoke({"urls": json.dumps(social_data, indent=2)}, config=config)
   
    return {
        "social_analysis": analysis,
        "raw_social_data": social_data
    }

def check_credits():
    """Check available credits"""
    try:
        credits_info = credits.invoke({})
        print(f"💳 Available Credits: {credits_info}")
 Return Credits_info
 "except Exception" is e.
        print(f"⚠️  Could not check credits: {e}")
 Return No

If __name__ is equal to "__main__":
    print("🚀 Advanced Competitive Analysis Tool with Gemini AI")
    print("="*60)
   
    check_credits()
   
    print("n🤖 Running AI/SaaS Competitive Analysis...")
    ai_results = run_ai_saas_analysis()
   
    run_additional = input("n❓ Run e-commerce analysis as well? (y/n): ").lower().strip()
 If run_additional is equal to 'y,' then:
        print("n🛒 Running E-commerce Platform Analysis...")
        ecom_results = run_ecommerce_analysis()
   
    print("n✨ Analysis complete! Check the exported files for detailed results.")