app/services/metadata_service.py

"""
MetadataService — Modular client tracking service.
Extracts and enriches client metadata from HTTP requests for fraud detection & analytics.

Components:
  1. parse_user_agent()  — Browser / OS / Device detection from User-Agent string
  2. get_geoip_info()    — Async GeoIP lookup via ip-api.com (free tier)
  3. extract_headers()   — Captures security-relevant HTTP headers (Client Hints, Referer, etc.)
  4. build_metadata()    — Merges frontend browser_meta with server-side headers into one JSON blob
"""

import json
from typing import Optional, Dict, Any

import httpx
from fastapi import Request


# ─── 1. User-Agent Parsing ────────────────────────────────────────────────────

def parse_user_agent(ua: Optional[str]) -> Dict[str, str]:
    """
    Parse a User-Agent string into browser, OS, and device type.
    Returns {"browser": ..., "os": ..., "device_type": ...}
    """
    result = {"browser": "Unknown", "os": "Unknown", "device_type": "Desktop"}

    if not ua:
        return result

    ua_lower = ua.lower()

    # ── Device ──
    if "mobi" in ua_lower:
        result["device_type"] = "Mobile"
    elif "tablet" in ua_lower or "ipad" in ua_lower:
        result["device_type"] = "Tablet"

    # ── Browser (order matters: Edge contains "chrome", so check Edge first) ──
    if "edg" in ua_lower:
        result["browser"] = "Edge"
    elif "opr" in ua_lower or "opera" in ua_lower:
        result["browser"] = "Opera"
    elif "chrome" in ua_lower and "chromium" not in ua_lower:
        result["browser"] = "Chrome"
    elif "firefox" in ua_lower:
        result["browser"] = "Firefox"
    elif "safari" in ua_lower:
        result["browser"] = "Safari"

    # ── OS (order matters: iPhone/iPad UAs contain "Mac OS X", so check them first) ──
    if "iphone" in ua_lower:
        result["os"] = "iOS"
        result["device_type"] = "Mobile"
    elif "ipad" in ua_lower:
        result["os"] = "iPadOS"
        result["device_type"] = "Tablet"
    elif "android" in ua_lower:
        result["os"] = "Android"
    elif "windows" in ua_lower:
        result["os"] = "Windows"
    elif "mac os" in ua_lower or "macintosh" in ua_lower:
        result["os"] = "MacOS"
    elif "cros" in ua_lower:
        result["os"] = "ChromeOS"
    elif "linux" in ua_lower:
        result["os"] = "Linux"

    return result


# ─── 2. GeoIP Lookup ──────────────────────────────────────────────────────────

async def get_geoip_info(ip: Optional[str]) -> Optional[str]:
    """
    Look up IP geolocation via ip-api.com. Returns JSON string or None.
    Skips localhost / private-range IPs.
    """
    if not ip or ip in ("127.0.0.1", "localhost", "::1", "0.0.0.0"):
        return None

    # Skip common private ranges
    if ip and ip.startswith(("10.", "172.16.", "192.168.")):
        return None

    try:
        async with httpx.AsyncClient(timeout=3.0) as client:
            resp = await client.get(f"http://ip-api.com/json/{ip}?fields=66846719")
            if resp.status_code == 200:
                data = resp.json()
                if data.get("status") == "success":
                    return json.dumps({
                        "country":     data.get("country"),
                        "countryCode": data.get("countryCode"),
                        "region":      data.get("regionName"),
                        "city":        data.get("city"),
                        "zip":         data.get("zip"),
                        "lat":         data.get("lat"),
                        "lon":         data.get("lon"),
                        "timezone":    data.get("timezone"),
                        "isp":         data.get("isp"),
                        "org":         data.get("org"),
                        "as":          data.get("as"),
                        "mobile":      data.get("mobile"),
                        "proxy":       data.get("proxy"),
                        "hosting":     data.get("hosting"),
                    })
    except Exception:
        pass  # Non-critical — order proceeds without geo data
    return None


# ─── 3. Header Extraction ─────────────────────────────────────────────────────

def extract_headers(request: Request) -> Dict[str, Any]:
    """
    Pull security/analytics-relevant headers from the incoming HTTP request.
    """
    h = request.headers
    return {
        "referer":             h.get("Referer"),
        "accept_language":     h.get("Accept-Language"),
        "sec_ch_ua":           h.get("Sec-CH-UA"),
        "sec_ch_ua_mobile":    h.get("Sec-CH-UA-Mobile"),
        "sec_ch_ua_platform":  h.get("Sec-CH-UA-Platform"),
        "sec_fetch_site":      h.get("Sec-Fetch-Site"),
        "x_forwarded_for":     h.get("X-Forwarded-For"),
    }


# ─── 4. Build Merged Metadata ─────────────────────────────────────────────────

def build_metadata(frontend_json: Optional[str], request: Request) -> str:
    """
    Merge frontend-supplied browser metadata with server-extracted headers
    into a single JSON string for storage.
    """
    browser_meta = {}
    if frontend_json:
        try:
            browser_meta = json.loads(frontend_json)
        except (json.JSONDecodeError, TypeError):
            browser_meta = {"raw": frontend_json}

    return json.dumps({
        "browser_meta": browser_meta,
        "headers":      extract_headers(request),
    })