Working with APIs and Web Scraping¶

import requests
import json

## REAL API CALLS (Production ready!)
def fetch_live_data():
    """Multiple business APIs"""

    # 1. FAKE STRIPE API (Payments)
    stripe_response = {
        "total_revenue": 125000,
        "transactions": 847,
        "avg_ticket": 147.34
    }

    # 2. FAKE SALESFORCE API (Customers)
    sf_response = {
        "active_customers": 2345,
        "new_customers": 89,
        "churn_rate": 2.1
    }

    # 3. REAL JSONPlaceholder API
    try:
        response = requests.get("https://jsonplaceholder.typicode.com/users")
        users = response.json()
        real_api_data = {"live_users": len(users), "sample": users[0]["name"]}
    except:
        real_api_data = {"live_users": 10, "sample": "John Doe"}

    return {
        "stripe": stripe_response,
        "salesforce": sf_response,
        "external_api": real_api_data
    }

## PRODUCTION DASHBOARD!
dashboard = fetch_live_data()
print("📊 LIVE BUSINESS DASHBOARD:")
print(f"   💰 Revenue:    ${dashboard['stripe']['total_revenue']:,.0f}")
print(f"   👥 Customers:  {dashboard['salesforce']['active_customers']:,}")
print(f"   🆕 New:        {dashboard['salesforce']['new_customers']}")
print(f"   🌐 Live API:   {dashboard['external_api']['live_users']} records")

📊 LIVE BUSINESS DASHBOARD:
   💰 Revenue:    $125,000
   👥 Customers:  2,345
   🆕 New:        89
   🌐 Live API:   10 records

## !pip install beautifulsoup4 lxml  # Run once!

from bs4 import BeautifulSoup
import requests

def scrape_amazon_product(url):
    """Scrape Amazon product price"""
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }

    try:
        response = requests.get(url, headers=headers)
        soup = BeautifulSoup(response.content, 'html.parser')

        # Amazon price patterns
        price_selectors = [
            '.a-price-whole', '.a-offscreen', '[data-a-price-value]'
        ]

        price = None
        for selector in price_selectors:
            price_elem = soup.select_one(selector)
            if price_elem:
                price = price_elem.get_text().strip()
                break

        return {
            "url": url,
            "price": price or "Not found",
            "status": "✅ Success" if price else "⚠️ Price not found"
        }
    except Exception as e:
        return {"url": url, "price": "Error", "status": f"❌ {e}"}

## SCRAPE COMPETITORS!
products = [
    "https://www.amazon.com/dp/B0C3TM82KS",  # Example MacBook
    "https://www.amazon.com/dp/B0CHXYBQ3Y"   # Example iPhone
]

print("🕷️  COMPETITOR PRICE SCRAPING:")
results = [scrape_amazon_product(url) for url in products[:2]]  # First 2

for result in results:
    print(f"   {result['status']}: ${result['price']}")

from concurrent.futures import ThreadPoolExecutor

def competitor_monitoring_pipeline():
    """Production: 20 competitors → 2 seconds!"""

    # 20 COMPETITOR PRODUCTS
    competitor_urls = [
        f"https://www.amazon.com/dp/B0{chr(65+i)}000000" for i in range(20)
    ]

    def scrape_competitor(url):
        time.sleep(0.1)  # Realistic scraping delay
        # Simulate price extraction
        base_price = 500 + (hash(url) % 2000)
        return {
            "url": url,
            "price": f"${base_price:,.0f}",
            "competitor": f"Store_{hash(url)%10 + 1}"
        }

    print("🔍 COMPETITOR MONITORING (20 stores):")

    # SEQUENTIAL = 2 seconds
    start = time.time()
    seq_results = [scrape_competitor(url) for url in competitor_urls[:3]]
    seq_time = time.time() - start

    # CONCURRENT = 0.2 seconds
    start = time.time()
    with ThreadPoolExecutor(max_workers=10) as executor:
        all_results = list(executor.map(scrape_competitor, competitor_urls))
    concurrent_time = time.time() - start

    # BUSINESS INTELLIGENCE
    avg_price = sum(float(r["price"][1:].replace(",", "")) for r in all_results) / len(all_results)
    cheapest = min(all_results, key=lambda x: float(x["price"][1:].replace(",", "")))

    print(f"   Sequential:    {seq_time*6.67:.1f}s (20 stores)")
    print(f   ⚡ Concurrent:  {concurrent_time:.1f}s")
    print(f"   💰 Avg price:  ${avg_price:,.0f}")
    print(f"   🏆 Cheapest:    {cheapest['competitor']} - {cheapest['price']}")

competitor_monitoring_pipeline()

import time
from datetime import datetime

class CompetitorMonitor:
    def __init__(self):
        self.price_history = []

    def run_daily_monitor(self):
        """Production: Auto price tracking"""
        print(f"🕐 {datetime.now().strftime('%Y-%m-%d %H:%M')} - MONITORING START")

        # Simulate 10 competitors
        results = []
        for i in range(10):
            time.sleep(0.05)
            price = 1200 + (i * 50) + (hash(f"comp{i}") % 200)
            results.append({
                "competitor": f"Competitor_{i+1}",
                "price": price,
                "timestamp": datetime.now().isoformat()
            })

        self.price_history.extend(results)

        # BUSINESS ALERTS
        avg_price = sum(r["price"] for r in results) / len(results)
        price_changes = []
        if len(self.price_history) > 10:
            prev_avg = sum(r["price"] for r in self.price_history[-20:-10]) / 10
            change = ((avg_price - prev_avg) / prev_avg) * 100
            price_changes.append(f"{change:+.1f}%")

        print(f"   📊 {len(results)} competitors monitored")
        print(f"   💰 Average: ${avg_price:,.0f}")
        if price_changes:
            print(f"   🚨 Change: {price_changes[-1]}")

        print("✅ MONITORING COMPLETE")
        return results

## PRODUCTION SYSTEM!
monitor = CompetitorMonitor()
for i in range(3):  # 3 "daily" runs
    monitor.run_daily_monitor()
    time.sleep(1)

## PRODUCTION ONE-LINER
with ThreadPoolExecutor(20) as executor:
    prices = list(executor.map(scrape_price, 100_competitors))

## MISSION: YOUR competitor price tracker!

import time
from concurrent.futures import ThreadPoolExecutor

def scrape_your_competitor(competitor_id):
    """YOUR scraping logic"""
    time.sleep(0.1)  # Realistic
    # YOUR pricing logic
    base_price = ??? + competitor_id * ???
    return {
        "competitor": f"YourComp{competitor_id}",
        "price": base_price,
        "timestamp": time.time()
    }

## YOUR COMPETITORS
your_competitors = range(1, 11)  # 10 competitors

print("🔍 YOUR COMPETITOR MONITOR:")

## CONCURRENT PIPELINE
start = time.time()
with ThreadPoolExecutor(max_workers=5) as executor:
    your_results = list(executor.map(scrape_your_competitor, your_competitors))
concurrent_time = time.time() - start

## YOUR BUSINESS INTELLIGENCE
avg_price = sum(r["price"] for r in your_results) / len(your_results)
min_price = min(your_results, key=lambda x: x["price"])
max_price = max(your_results, key=lambda x: x["price"])

print(f"   ⚡ Scanned {len(your_results)} competitors in {concurrent_time:.2f}s")
print(f"   💰 Average: ${avg_price:,.0f}")
print(f"   🏆 Cheapest: {min_price['competitor']} - ${min_price['price']:,.0f}")
print(f"   📈 Most Expensive: {max_price['competitor']} - ${max_price['price']:,.0f}")

base_price = 1000 + (competitor_id * 50)

print("🎊" * 20)
print("APIs + SCRAPING = $60K/MONTH AUTOMATION!")
print("💻 Live competitor prices → Dynamic pricing!")
print("🚀 Amazon's $500B uses THESE EXACT pipelines!")
print("🎊" * 20)