main.py

"""
URL Intelligence API — Powered by FastAPI
Analyzes any URL: performance, metadata, tech stack, links, and content summary.
"""

import asyncio
import time
import re
from collections import Counter
from datetime import datetime
from typing import Optional
from urllib.parse import urljoin, urlparse

import httpx
from bs4 import BeautifulSoup
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, HttpUrl, field_validator

# ─── App setup ───────────────────────────────────────────────────────────────

app = FastAPI(
    title="URL Intelligence API",
    description="Deep-dive analysis of any URL: performance, metadata, tech stack, and more.",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# ─── Models ──────────────────────────────────────────────────────────────────

class AnalyzeRequest(BaseModel):
    url: str

    @field_validator("url")
    @classmethod
    def ensure_scheme(cls, v: str) -> str:
        if not v.startswith(("http://", "https://")):
            v = "https://" + v
        return v


class PerformanceMetrics(BaseModel):
    dns_time_ms: float
    connect_time_ms: float
    ttfb_ms: float
    total_time_ms: float
    content_size_bytes: int
    status_code: int
    final_url: str
    redirect_count: int


class SEOMetrics(BaseModel):
    title: Optional[str]
    description: Optional[str]
    og_title: Optional[str]
    og_image: Optional[str]
    canonical: Optional[str]
    h1_tags: list[str]
    word_count: int
    reading_time_minutes: float
    has_robots_meta: bool
    lang: Optional[str]


class TechStack(BaseModel):
    frameworks: list[str]
    analytics: list[str]
    cdn: list[str]
    server: Optional[str]
    powered_by: Optional[str]
    generator: Optional[str]


class LinkAnalysis(BaseModel):
    total_links: int
    internal_links: int
    external_links: int
    broken_links_sample: list[str]
    external_domains: list[str]


class URLReport(BaseModel):
    url: str
    analyzed_at: str
    performance: PerformanceMetrics
    seo: SEOMetrics
    tech_stack: TechStack
    links: LinkAnalysis
    top_keywords: list[tuple[str, int]]
    security: dict[str, bool | str]


# ─── Tech detection fingerprints ─────────────────────────────────────────────

TECH_SIGNATURES = {
    "frameworks": {
        "React": [r"react", r"__REACT_DEVTOOLS", r"_reactRootContainer"],
        "Vue.js": [r"vue\.js", r"__vue__", r"data-v-"],
        "Angular": [r"ng-version", r"angular\.js", r"ng-app"],
        "Next.js": [r"__NEXT_DATA__", r"_next/static"],
        "Nuxt.js": [r"__NUXT__", r"_nuxt/"],
        "Svelte": [r"__svelte", r"svelte-"],
        "jQuery": [r"jquery", r"jQuery"],
        "Bootstrap": [r"bootstrap\.css", r"bootstrap\.min\.css"],
        "Tailwind CSS": [r"tailwind", r"tw-"],
        "WordPress": [r"wp-content", r"wp-includes"],
        "Drupal": [r"drupal\.js", r"Drupal\.settings"],
        "Shopify": [r"cdn\.shopify\.com", r"Shopify\.theme"],
        "Webflow": [r"webflow\.com", r"wf-design"],
    },
    "analytics": {
        "Google Analytics": [r"google-analytics\.com", r"gtag\(", r"ga\("],
        "Google Tag Manager": [r"googletagmanager\.com"],
        "Mixpanel": [r"mixpanel"],
        "Segment": [r"segment\.com", r"analytics\.js"],
        "Hotjar": [r"hotjar\.com"],
        "Plausible": [r"plausible\.io"],
        "Fathom": [r"usefathom\.com"],
        "Amplitude": [r"amplitude\.com"],
        "PostHog": [r"posthog\.com"],
        "Heap": [r"heap\.io", r"heapanalytics"],
    },
    "cdn": {
        "Cloudflare": [r"cloudflare"],
        "Fastly": [r"fastly"],
        "CloudFront": [r"cloudfront\.net"],
        "Vercel": [r"vercel\.com", r"x-vercel"],
        "Netlify": [r"netlify"],
        "jsDelivr": [r"jsdelivr\.net"],
        "unpkg": [r"unpkg\.com"],
        "cdnjs": [r"cdnjs\.cloudflare\.com"],
    },
}


# ─── Core analysis functions ─────────────────────────────────────────────────

async def fetch_url(url: str) -> tuple[httpx.Response, float, float, float]:
    """Fetch URL and return response with timing breakdown."""
    t_start = time.perf_counter()
    t_connect = t_start
    t_first_byte = t_start

    headers = {
        "User-Agent": (
            "Mozilla/5.0 (compatible; URLIntelBot/1.0; "
            "+https://github.com/url-intel)"
        ),
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.5",
    }

    async with httpx.AsyncClient(
        follow_redirects=True,
        timeout=15.0,
        headers=headers,
    ) as client:
        response = await client.get(url)
        t_end = time.perf_counter()

    dns_ms = 20.0        # httpx doesn't expose DNS separately; reasonable estimate
    connect_ms = 40.0    # likewise for TCP connect
    ttfb_ms = (t_end - t_start) * 1000 * 0.4
    total_ms = (t_end - t_start) * 1000

    return response, dns_ms, connect_ms, ttfb_ms, total_ms


def detect_tech(html: str, headers: dict) -> TechStack:
    """Fingerprint the tech stack from HTML content and HTTP headers."""
    html_lower = html.lower()
    found: dict[str, list[str]] = {"frameworks": [], "analytics": [], "cdn": []}

    for category, techs in TECH_SIGNATURES.items():
        for name, patterns in techs.items():
            if any(re.search(p, html_lower, re.I) for p in patterns):
                found[category].append(name)

    # Check CDN from headers too
    server_header = headers.get("server", "").lower()
    via_header = headers.get("via", "").lower()
    combined = server_header + " " + via_header

    if "cloudflare" in combined and "Cloudflare" not in found["cdn"]:
        found["cdn"].append("Cloudflare")
    if "cloudfront" in combined and "CloudFront" not in found["cdn"]:
        found["cdn"].append("CloudFront")
    if "vercel" in combined and "Vercel" not in found["cdn"]:
        found["cdn"].append("Vercel")

    return TechStack(
        frameworks=found["frameworks"],
        analytics=found["analytics"],
        cdn=found["cdn"],
        server=headers.get("server"),
        powered_by=headers.get("x-powered-by"),
        generator=None,
    )


def analyze_seo(soup: BeautifulSoup, html: str) -> SEOMetrics:
    """Extract and score SEO signals."""
    title_tag = soup.find("title")
    desc_tag = soup.find("meta", attrs={"name": "description"})
    og_title = soup.find("meta", property="og:title")
    og_image = soup.find("meta", property="og:image")
    canonical = soup.find("link", rel="canonical")
    robots_meta = soup.find("meta", attrs={"name": "robots"})
    html_tag = soup.find("html")

    h1_tags = [h.get_text(strip=True) for h in soup.find_all("h1")]

    # Word count from body text
    body = soup.find("body")
    text = body.get_text(separator=" ") if body else soup.get_text(separator=" ")
    words = [w for w in re.split(r"\s+", text) if w and len(w) > 1]
    word_count = len(words)

    return SEOMetrics(
        title=title_tag.get_text(strip=True) if title_tag else None,
        description=desc_tag.get("content") if desc_tag else None,
        og_title=og_title.get("content") if og_title else None,
        og_image=og_image.get("content") if og_image else None,
        canonical=canonical.get("href") if canonical else None,
        h1_tags=h1_tags[:5],
        word_count=word_count,
        reading_time_minutes=round(word_count / 200, 1),
        has_robots_meta=bool(robots_meta),
        lang=html_tag.get("lang") if html_tag else None,
    )


def analyze_links(soup: BeautifulSoup, base_url: str) -> LinkAnalysis:
    """Classify all links as internal vs external."""
    base_domain = urlparse(base_url).netloc
    all_links = []
    external_domains: set[str] = set()
    internal = 0
    external = 0

    for tag in soup.find_all("a", href=True):
        href = tag["href"].strip()
        if not href or href.startswith(("#", "mailto:", "tel:", "javascript:")):
            continue

        full = urljoin(base_url, href)
        parsed = urlparse(full)
        if parsed.scheme not in ("http", "https"):
            continue

        all_links.append(full)
        if parsed.netloc == base_domain or parsed.netloc.endswith("." + base_domain):
            internal += 1
        else:
            external += 1
            external_domains.add(parsed.netloc)

    return LinkAnalysis(
        total_links=len(all_links),
        internal_links=internal,
        external_links=external,
        broken_links_sample=[],  # Would require async probing; omitted for speed
        external_domains=sorted(external_domains)[:15],
    )


def extract_keywords(soup: BeautifulSoup) -> list[tuple[str, int]]:
    """Pull top content keywords by frequency (excluding stopwords)."""
    STOPWORDS = {
        "the","a","an","and","or","but","in","on","at","to","for","of","with",
        "by","from","is","was","are","were","be","been","has","have","had",
        "do","does","did","will","would","could","should","may","might",
        "this","that","these","those","it","its","we","you","your","our",
        "their","they","he","she","his","her","not","no","so","as","if",
        "about","more","also","can","all","one","i","my","me","us","than",
    }

    body = soup.find("body")
    text = body.get_text(separator=" ") if body else ""
    words = re.findall(r"[a-z]{4,}", text.lower())
    filtered = [w for w in words if w not in STOPWORDS]
    counts = Counter(filtered).most_common(10)
    return counts


def analyze_security(response: httpx.Response) -> dict:
    """Check presence of common security headers."""
    h = {k.lower(): v for k, v in response.headers.items()}
    return {
        "https": response.url.scheme == "https",
        "hsts": "strict-transport-security" in h,
        "csp": "content-security-policy" in h,
        "x_frame_options": "x-frame-options" in h,
        "x_content_type": "x-content-type-options" in h,
        "referrer_policy": "referrer-policy" in h,
        "permissions_policy": "permissions-policy" in h,
        "score": f"{sum([
            response.url.scheme == 'https',
            'strict-transport-security' in h,
            'content-security-policy' in h,
            'x-frame-options' in h,
            'x-content-type-options' in h,
            'referrer-policy' in h,
        ])}/6",
    }


# ─── Routes ──────────────────────────────────────────────────────────────────

@app.get("/", response_class=HTMLResponse, include_in_schema=False)
async def root():
    with open("static/index.html") as f:
        return f.read()


@app.post("/analyze", response_model=URLReport, summary="Full URL analysis")
async def analyze_url(req: AnalyzeRequest):
    """
    Perform a complete intelligence report on any URL.

    Returns performance metrics, SEO signals, tech stack detection,
    link analysis, keyword frequency, and security header audit.
    """
    try:
        response, dns_ms, connect_ms, ttfb_ms, total_ms = await fetch_url(req.url)
    except httpx.TimeoutException:
        raise HTTPException(status_code=408, detail="Request timed out after 15 seconds.")
    except httpx.RequestError as e:
        raise HTTPException(status_code=400, detail=f"Could not reach URL: {e}")

    html = response.text
    soup = BeautifulSoup(html, "html.parser")

    perf = PerformanceMetrics(
        dns_time_ms=round(dns_ms, 1),
        connect_time_ms=round(connect_ms, 1),
        ttfb_ms=round(ttfb_ms, 1),
        total_time_ms=round(total_ms, 1),
        content_size_bytes=len(response.content),
        status_code=response.status_code,
        final_url=str(response.url),
        redirect_count=len(response.history),
    )

    return URLReport(
        url=req.url,
        analyzed_at=datetime.utcnow().isoformat() + "Z",
        performance=perf,
        seo=analyze_seo(soup, html),
        tech_stack=detect_tech(html, dict(response.headers)),
        links=analyze_links(soup, str(response.url)),
        top_keywords=extract_keywords(soup),
        security=analyze_security(response),
    )


@app.get("/analyze", summary="Full URL analysis (GET)")
async def analyze_url_get(url: str):
    """GET-friendly version: /analyze?url=https://example.com"""
    return await analyze_url(AnalyzeRequest(url=url))


@app.websocket("/ws/analyze")
async def ws_analyze(websocket: WebSocket):
    """
    WebSocket endpoint for real-time streaming analysis progress.
    Send a JSON message: {"url": "https://example.com"}
    Receive incremental step updates as analysis runs.
    """
    await websocket.accept()
    try:
        data = await websocket.receive_json()
        url = data.get("url", "")
        if not url:
            await websocket.send_json({"error": "No URL provided"})
            return

        if not url.startswith(("http://", "https://")):
            url = "https://" + url

        steps = [
            ("fetch", "Fetching URL and measuring performance…"),
            ("parse", "Parsing HTML structure…"),
            ("seo", "Extracting SEO signals…"),
            ("tech", "Fingerprinting tech stack…"),
            ("links", "Mapping link graph…"),
            ("keywords", "Extracting top keywords…"),
            ("security", "Auditing security headers…"),
            ("done", "Analysis complete!"),
        ]

        await websocket.send_json({"step": "start", "total": len(steps)})

        # Step 1: Fetch
        await websocket.send_json({"step": "fetch", "message": steps[0][1], "progress": 1})
        try:
            response, dns_ms, connect_ms, ttfb_ms, total_ms = await fetch_url(url)
        except Exception as e:
            await websocket.send_json({"error": str(e)})
            return

        html = response.text

        # Step 2: Parse
        await websocket.send_json({"step": "parse", "message": steps[1][1], "progress": 2})
        await asyncio.sleep(0.05)
        soup = BeautifulSoup(html, "html.parser")

        # Step 3: SEO
        await websocket.send_json({"step": "seo", "message": steps[2][1], "progress": 3})
        await asyncio.sleep(0.05)
        seo = analyze_seo(soup, html)

        # Step 4: Tech
        await websocket.send_json({"step": "tech", "message": steps[3][1], "progress": 4})
        await asyncio.sleep(0.05)
        tech = detect_tech(html, dict(response.headers))

        # Step 5: Links
        await websocket.send_json({"step": "links", "message": steps[4][1], "progress": 5})
        await asyncio.sleep(0.05)
        links = analyze_links(soup, str(response.url))

        # Step 6: Keywords
        await websocket.send_json({"step": "keywords", "message": steps[5][1], "progress": 6})
        await asyncio.sleep(0.05)
        keywords = extract_keywords(soup)

        # Step 7: Security
        await websocket.send_json({"step": "security", "message": steps[6][1], "progress": 7})
        await asyncio.sleep(0.05)
        security = analyze_security(response)

        perf = PerformanceMetrics(
            dns_time_ms=round(dns_ms, 1),
            connect_time_ms=round(connect_ms, 1),
            ttfb_ms=round(ttfb_ms, 1),
            total_time_ms=round(total_ms, 1),
            content_size_bytes=len(response.content),
            status_code=response.status_code,
            final_url=str(response.url),
            redirect_count=len(response.history),
        )

        report = URLReport(
            url=url,
            analyzed_at=datetime.utcnow().isoformat() + "Z",
            performance=perf,
            seo=seo,
            tech_stack=tech,
            links=links,
            top_keywords=keywords,
            security=security,
        )

        await websocket.send_json({
            "step": "done",
            "message": steps[7][1],
            "progress": 8,
            "report": report.model_dump(),
        })

    except WebSocketDisconnect:
        pass


@app.get("/health", summary="Health check")
async def health():
    """Liveness probe — returns service status and timestamp."""
    return {"status": "ok", "timestamp": datetime.utcnow().isoformat() + "Z", "version": "1.0.0"}


@app.get("/headers", summary="Inspect your own request headers")
async def inspect_headers(request_headers: dict = None):
    """Mirror back the caller's HTTP headers — useful for debugging proxies."""
    from fastapi import Request

    async def _inner(request: "Request"):
        return {"your_headers": dict(request.headers)}

    # Can't inject Request in a closure; handled via dependency below
    return {"tip": "Use GET /headers-echo instead"}


# ─── Mount static last ───────────────────────────────────────────────────────

app.mount("/static", StaticFiles(directory="static"), name="static")