src/server.py

"""AuspexIQ MCP server — live YouTube niche analysis.

Tools: scan_niche (niche saturation + outliers + ENTER/CROWDED/AVOID verdict)
and channel_outliers (which of a channel's videos overperformed its baseline).
"""

import asyncio
import hashlib
import json
import logging
import math
import os
import sys
import time
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
from urllib.parse import parse_qs, urlencode, urlparse

from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
from x402.http import (
    OKXAuthConfig,
    OKXFacilitatorClient,
    OKXFacilitatorConfig,
    PaymentOption,
)
from x402.http.middleware.fastapi import PaymentMiddlewareASGI
from x402.http.types import RouteConfig
from x402.mechanisms.evm.deferred.server import AggrDeferredEvmScheme
from x402.mechanisms.evm.exact.server import ExactEvmScheme
from x402.server import x402ResourceServer

import config
from src import analysis, youtube
from src.youtube import RequestMeter, ToolFault

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("auspex")

if not youtube.api_key_present():
    log.critical(
        "%s is not set — every tool call will return MISSING_API_KEY until it is configured",
        config.YT_API_KEY_ENV,
    )

mcp = FastMCP(
    name="AuspexIQ",
    instructions=(
        "Live YouTube niche analysis for agents. scan_niche returns saturation, "
        "breakout outlier videos, and an ENTER/CROWDED/AVOID verdict for a niche "
        "keyword. channel_outliers reveals which of a channel's videos overperformed "
        "its own baseline. Every number comes from the YouTube Data API v3 at "
        "request time (or a short-TTL cache of a previous live response)."
    ),
)


def _now():
    return datetime.now(timezone.utc)


def _iso(dt):
    return dt.strftime("%Y-%m-%dT%H:%M:%SZ")


def _watch_url(video_id):
    return "https://www.youtube.com/watch?" + urlencode({"v": video_id})


def _params_hash(values):
    return hashlib.sha256(repr(values).encode()).hexdigest()[:12]


def _meta(cache_state, fetched_at_iso, units_spent):
    return {
        "cache": cache_state,
        "fetched_at": fetched_at_iso,
        "quota_units_spent": units_spent,
        "quota_units_remaining_today": youtube.quota.remaining,
    }


def _log_request(tool, params_hash, cache_state, units, started, outcome):
    log.info(
        json.dumps(
            {
                "tool": tool,
                "params": params_hash,
                "cache": cache_state,
                "units": units,
                "latency_ms": round((time.monotonic() - started) * 1000),
                "outcome": outcome,
            }
        )
    )


def _dedupe(values):
    return list(dict.fromkeys(values))


def _chunks(values, size):
    for i in range(0, len(values), size):
        yield values[i : i + size]


async def _execute(tool, params_hash, cache_key, ttl_s, worst_case_units, run):
    """Shared request wrapper: cache lookup, quota precheck, pipeline, cache store,
    structured error conversion, one-line request log."""
    started = time.monotonic()
    meter = RequestMeter()
    try:
        cached = youtube.cache.get(cache_key)
        if cached is not None:
            payload, fetched_at = cached
            _log_request(tool, params_hash, "hit", 0, started, payload.get("verdict", "ok"))
            return {**payload, "meta": _meta("hit", fetched_at, 0)}
        youtube.quota.precheck(worst_case_units)
        payload = await run(meter)
        fetched_at = _iso(_now())
        youtube.cache.put(cache_key, payload, ttl_s, fetched_at)
        _log_request(tool, params_hash, "miss", meter.units, started, payload.get("verdict", "ok"))
        return {**payload, "meta": _meta("miss", fetched_at, meter.units)}
    except ToolFault as fault:
        _log_request(tool, params_hash, "miss", meter.units, started, fault.code)
        return fault.to_response()
    except Exception as exc:  # defensive: a tool must never raise to the client
        log.exception("unexpected failure in %s", tool)
        fault = ToolFault("YT_API_ERROR", f"Unexpected server error: {exc}. Retry may help.", True)
        _log_request(tool, params_hash, "miss", meter.units, started, fault.code)
        return fault.to_response()


# ---------------------------------------------------------------------------
# Tool 1: scan_niche
# ---------------------------------------------------------------------------


def _validate_scan(query, region_code, recency_days, max_results):
    if not isinstance(query, str) or not (
        config.QUERY_MIN_LEN <= len(query.strip()) <= config.QUERY_MAX_LEN
    ):
        raise ToolFault(
            "INVALID_INPUT",
            f"'query' must be a string of {config.QUERY_MIN_LEN}-{config.QUERY_MAX_LEN} "
            f"characters, e.g. a niche keyword like a topic phrase.",
            False,
        )
    region = (region_code or config.DEFAULT_REGION_CODE).strip().upper()
    if len(region) != 2 or not region.isalpha():
        raise ToolFault(
            "INVALID_INPUT",
            "'region_code' must be an ISO 3166-1 alpha-2 code such as 'US' or 'GB'.",
            False,
        )
    if not (config.RECENCY_DAYS_MIN <= recency_days <= config.RECENCY_DAYS_MAX):
        raise ToolFault(
            "INVALID_INPUT",
            f"'recency_days' must be between {config.RECENCY_DAYS_MIN} and "
            f"{config.RECENCY_DAYS_MAX}.",
            False,
        )
    if not (config.MAX_RESULTS_MIN <= max_results <= config.MAX_RESULTS_MAX):
        raise ToolFault(
            "INVALID_INPUT",
            f"'max_results' must be between {config.MAX_RESULTS_MIN} and "
            f"{config.MAX_RESULTS_MAX}.",
            False,
        )
    return query.strip(), region, int(recency_days), int(max_results)


async def _recent_uploads(channel, meter):
    """Fetch and normalize the channel's most recent uploads."""
    if not channel["uploads"]:
        return []
    page = await youtube.playlist_page(
        channel["uploads"], config.BASELINE_RECENT_UPLOADS, None, meter
    )
    upload_ids = [
        item["contentDetails"]["videoId"]
        for item in page.get("items", [])
        if item.get("contentDetails", {}).get("videoId")
    ]
    if not upload_ids:
        return []
    data = await youtube.list_videos(upload_ids, meter)
    return [youtube.normalize_video(item) for item in data.get("items", [])]


async def _deep_baseline(channel, now, meter):
    """Baseline from the channel's recent uploads; falls back to the lifetime
    average (real numbers from channels.list) when uploads can't qualify."""
    fallback = (
        analysis.lifetime_average(channel["view_count"], channel["video_count"]),
        "lifetime_avg",
    )
    try:
        uploads = await _recent_uploads(channel, meter)
        baseline = analysis.recent_median_baseline(uploads, now)
        if baseline is None:
            return fallback
        return baseline, "recent_median"
    except ToolFault as fault:
        if fault.code in ("QUOTA_EXHAUSTED", "MISSING_API_KEY"):
            raise
        return fallback


def _video_format(record):
    if record.get("stream"):
        return "livestream"
    if analysis.is_short(record.get("seconds", 0)):
        return "short"
    return "video"


def _format_outlier(record):
    method_label = (
        "recent median" if record["baseline_method"] == "recent_median" else "lifetime average"
    )
    return {
        "title": record["title"],
        "url": _watch_url(record["id"]),
        "channel": record["channel_title"],
        "channel_id": record["channel_id"],
        "channel_subs": record["subs"] if record["subs"] is not None else 0,
        "views": record["views"],
        "published_at": _iso(record["published_at"]),
        "format": _video_format(record),
        "channel_baseline": round(record["baseline"]),
        "baseline_method": record["baseline_method"],
        "outlier_multiple": round(record["multiple"], 2),
        "why": f"{record['multiple']:.1f}x this channel's {method_label}",
    }


async def _scan_pipeline(query, region, recency_days, max_results, meter):
    now = _now()
    published_after = _iso(now - timedelta(days=recency_days))

    search = await youtube.search_videos(query, region, published_after, max_results, meter)
    video_ids = _dedupe(
        item["id"]["videoId"]
        for item in search.get("items", [])
        if item.get("id", {}).get("videoId")
    )

    videos = []
    streams_filtered = 0
    if video_ids:
        data = await youtube.list_videos(video_ids, meter)
        for item in data.get("items", []):
            video = youtube.normalize_video(item)
            if video["views"] is None or video["published_at"] is None:
                continue  # hidden view count or missing metadata
            if analysis.is_short(video["seconds"]):
                continue
            if video["stream"]:
                streams_filtered += 1  # cumulative stream views skew comparisons
                continue
            if not video["channel_id"]:
                continue
            videos.append(video)

    channels_by_id = {}
    channel_ids = _dedupe(v["channel_id"] for v in videos)
    if channel_ids:
        data = await youtube.list_channels(channel_ids, meter)
        channels_by_id = {
            channel["id"]: channel
            for channel in (youtube.normalize_channel(item) for item in data.get("items", []))
        }
    videos = [v for v in videos if v["channel_id"] in channels_by_id]

    result_counts = Counter(v["channel_id"] for v in videos)
    views_in_set = defaultdict(int)
    for v in videos:
        views_in_set[v["channel_id"]] += v["views"]
    ranked = sorted(
        result_counts, key=lambda cid: (-result_counts[cid], -views_in_set[cid], cid)
    )
    deep_ids = ranked[: config.BASELINE_DEEP_CHANNELS]

    baselines = {}
    semaphore = asyncio.Semaphore(config.BASELINE_CONCURRENCY)

    async def deep(channel_id):
        async with semaphore:
            baselines[channel_id] = await _deep_baseline(channels_by_id[channel_id], now, meter)

    await asyncio.gather(*(deep(cid) for cid in deep_ids))
    for cid in result_counts:
        if cid not in baselines:
            channel = channels_by_id[cid]
            baselines[cid] = (
                analysis.lifetime_average(channel["view_count"], channel["video_count"]),
                "lifetime_avg",
            )

    outlier_records = []
    for v in videos:
        baseline, method = baselines[v["channel_id"]]
        multiple = analysis.outlier_multiple(v["views"], baseline)
        if multiple is not None and multiple >= config.SCAN_OUTLIER_MULTIPLE:
            outlier_records.append(
                {
                    **v,
                    "baseline": baseline,
                    "baseline_method": method,
                    "multiple": multiple,
                    "subs": channels_by_id[v["channel_id"]]["subs"],
                }
            )
    outlier_records.sort(key=lambda r: -r["multiple"])

    saturation, verdict, reasons, signals = analysis.assess_niche(videos, outlier_records, now)

    return {
        "ok": True,
        "query": query,
        "region_code": region,
        "analyzed": {
            "videos": len(videos),
            "channels": len(result_counts),
            "deep_baseline_channels": len(deep_ids),
            "livestreams_filtered": streams_filtered,
        },
        "saturation_score": saturation,
        "verdict": verdict,
        "verdict_reasons": reasons,
        "signals": signals,
        "outliers": [
            _format_outlier(r) for r in outlier_records[: config.SCAN_OUTLIERS_MAX]
        ],
    }


async def _scan_request(query, region_code, recency_days, max_results):
    """Full scan_niche request: validation, cache, quota, pipeline. Shared by
    the MCP tool and the paid REST endpoint."""
    params_hash = _params_hash((query, region_code, recency_days, max_results))
    started = time.monotonic()
    try:
        q, region, days, n_results = _validate_scan(
            query, region_code, recency_days, max_results
        )
    except ToolFault as fault:
        _log_request("scan_niche", params_hash, "-", 0, started, fault.code)
        return fault.to_response()
    cache_key = ("scan_niche", q.lower(), region, days, n_results)
    return await _execute(
        "scan_niche",
        params_hash,
        cache_key,
        config.SCAN_CACHE_TTL_S,
        config.SCAN_WORST_CASE_UNITS,
        lambda meter: _scan_pipeline(q, region, days, n_results, meter),
    )


@mcp.tool
async def scan_niche(
    query: str,
    region_code: str = config.DEFAULT_REGION_CODE,
    recency_days: int = config.RECENCY_DAYS_DEFAULT,
    max_results: int = config.MAX_RESULTS_DEFAULT,
) -> dict:
    """Assess a YouTube niche keyword using live YouTube data: who ranks, how
    concentrated the niche is, which videos are outliers relative to their own
    channel's baseline, and an ENTER / CROWDED / AVOID verdict for a new entrant.

    query: the niche keyword (2-80 chars). region_code: ISO 3166-1 alpha-2
    (default US). recency_days: only consider videos published in this window
    (30-1825, default 365). max_results: search results to analyze (10-50).
    """
    return await _scan_request(query, region_code, recency_days, max_results)


# ---------------------------------------------------------------------------
# Tool 2: channel_outliers
# ---------------------------------------------------------------------------


def _validate_channel_inputs(channel, lookback_videos, min_multiple):
    if not isinstance(channel, str) or not channel.strip():
        raise ToolFault(
            "INVALID_INPUT",
            "'channel' is required: a channel ID (UC...), an @handle, or a full "
            "youtube.com channel URL.",
            False,
        )
    if not (config.LOOKBACK_VIDEOS_MIN <= lookback_videos <= config.LOOKBACK_VIDEOS_MAX):
        raise ToolFault(
            "INVALID_INPUT",
            f"'lookback_videos' must be between {config.LOOKBACK_VIDEOS_MIN} and "
            f"{config.LOOKBACK_VIDEOS_MAX}.",
            False,
        )
    if not (config.MIN_MULTIPLE_MIN <= min_multiple <= config.MIN_MULTIPLE_MAX):
        raise ToolFault(
            "INVALID_INPUT",
            f"'min_multiple' must be between {config.MIN_MULTIPLE_MIN} and "
            f"{config.MIN_MULTIPLE_MAX}.",
            False,
        )
    return channel.strip(), int(lookback_videos), float(min_multiple)


def _not_found(reference):
    return ToolFault(
        "CHANNEL_NOT_FOUND",
        f"Could not resolve '{reference}' to a YouTube channel. Pass a channel ID "
        f"(UC...), an @handle, or a full youtube.com channel URL.",
        False,
    )


async def _resolve_channel(reference, meter):
    ref = reference.strip()
    channel_id = None
    handle = None
    if ref.startswith("UC") and len(ref) == 24 and " " not in ref:
        channel_id = ref
    elif ref.startswith(("http://", "https://")) or "youtube.com" in ref or "youtu.be" in ref:
        parsed = urlparse(ref if "://" in ref else "https://" + ref)
        segments = [s for s in parsed.path.split("/") if s]
        host = (parsed.hostname or "").lower()
        if host.endswith("youtu.be") or (
            segments and segments[0] in ("watch", "shorts", "embed", "live")
        ):
            raise ToolFault(
                "INVALID_INPUT",
                "That looks like a video URL, not a channel. Pass a channel ID "
                "(UC...), an @handle, or a channel URL — or use video_context to "
                "analyze a single video.",
                False,
            )
        if segments and segments[0] == "channel" and len(segments) > 1:
            channel_id = segments[1]
        elif segments:
            at_segments = [s for s in segments if s.startswith("@")]
            handle = at_segments[0][1:] if at_segments else segments[-1]
        else:
            raise _not_found(reference)
    elif ref.startswith("@"):
        handle = ref[1:]
    else:
        handle = ref
    if channel_id:
        data = await youtube.list_channels([channel_id], meter)
    else:
        if not handle:
            raise _not_found(reference)
        data = await youtube.channel_by_handle(handle, meter)
    items = data.get("items") or []
    if not items:
        raise _not_found(reference)
    return youtube.normalize_channel(items[0])


async def _channel_pipeline(channel_ref, lookback, min_multiple, meter):
    now = _now()
    channel = await _resolve_channel(channel_ref, meter)

    upload_ids = []
    if channel["uploads"]:
        page_token = None
        while len(upload_ids) < lookback:
            page_size = min(50, lookback - len(upload_ids))
            try:
                page = await youtube.playlist_page(
                    channel["uploads"], page_size, page_token, meter
                )
            except ToolFault as fault:
                # A brand-new channel's uploads playlist can 404; that is an
                # honest zero-uploads answer, not a server failure.
                if fault.code == "YT_API_ERROR" and "playlistNotFound" in fault.message:
                    break
                raise
            upload_ids.extend(
                item["contentDetails"]["videoId"]
                for item in page.get("items", [])
                if item.get("contentDetails", {}).get("videoId")
            )
            page_token = page.get("nextPageToken")
            if not page_token:
                break
    upload_ids = _dedupe(upload_ids)[:lookback]

    videos = []
    for batch in _chunks(upload_ids, 50):
        data = await youtube.list_videos(batch, meter)
        for item in data.get("items", []):
            video = youtube.normalize_video(item)
            if video["views"] is None or video["published_at"] is None:
                continue  # hidden view count or missing metadata
            videos.append(video)

    baseline = analysis.recent_median_baseline(videos, now)
    if baseline is None:
        baseline = analysis.lifetime_average(channel["view_count"], channel["video_count"])
        method = "lifetime_avg"
    else:
        method = "recent_median"

    outlier_records = []
    for v in videos:
        multiple = analysis.outlier_multiple(v["views"], baseline)
        if multiple is not None and multiple >= min_multiple:
            outlier_records.append(
                {
                    **v,
                    "channel_title": channel["title"],
                    "baseline": baseline,
                    "baseline_method": method,
                    "multiple": multiple,
                    "subs": channel["subs"],
                }
            )
    outlier_records.sort(key=lambda r: -r["multiple"])

    return {
        "ok": True,
        "channel": {
            "id": channel["id"],
            "title": channel["title"],
            "subscribers": channel["subs"] if channel["subs"] is not None else 0,
            "url": "https://www.youtube.com/channel/" + channel["id"],
        },
        "baseline": round(baseline),
        "baseline_method": method,
        "videos_considered": len(videos),
        "outliers": [_format_outlier(r) for r in outlier_records],
    }


async def _channel_request(channel, lookback_videos, min_multiple):
    """Full channel_outliers request: validation, cache, quota, pipeline.
    Shared by the MCP tool and the paid REST endpoint."""
    params_hash = _params_hash((channel, lookback_videos, min_multiple))
    started = time.monotonic()
    try:
        ref, lookback, multiple = _validate_channel_inputs(
            channel, lookback_videos, min_multiple
        )
    except ToolFault as fault:
        _log_request("channel_outliers", params_hash, "-", 0, started, fault.code)
        return fault.to_response()
    cache_key = ("channel_outliers", ref.lower(), lookback, multiple)
    worst_case = config.COST_LIST + 2 * math.ceil(lookback / 50) * config.COST_LIST
    return await _execute(
        "channel_outliers",
        params_hash,
        cache_key,
        config.CHANNEL_CACHE_TTL_S,
        worst_case,
        lambda meter: _channel_pipeline(ref, lookback, multiple, meter),
    )


@mcp.tool
async def channel_outliers(
    channel: str,
    lookback_videos: int = config.LOOKBACK_VIDEOS_DEFAULT,
    min_multiple: float = config.MIN_MULTIPLE_DEFAULT,
) -> dict:
    """Reveal which of a YouTube channel's recent videos overperformed the
    channel's own baseline (median views of recent long-form uploads), using
    live YouTube data.

    channel: a channel ID (UC...), an @handle, or a full youtube.com channel
    URL. lookback_videos: how many recent uploads to analyze (10-100, default
    30). min_multiple: views/baseline threshold to count as an outlier
    (1.5-10, default 2.5).
    """
    return await _channel_request(channel, lookback_videos, min_multiple)


# ---------------------------------------------------------------------------
# Tool 3: video_context — why did this video blow up (or not)
# ---------------------------------------------------------------------------


def _parse_video_ref(reference):
    """Extract an 11-char video ID from a raw ID or any YouTube video URL."""
    ref = (reference or "").strip()
    if not ref:
        raise ToolFault(
            "INVALID_INPUT", "'video' is required: a YouTube video URL or video ID.", False
        )
    if "/" not in ref and "?" not in ref and len(ref) == 11:
        return ref
    parsed = urlparse(ref if "://" in ref else "https://" + ref)
    host = (parsed.hostname or "").lower()
    segments = [s for s in parsed.path.split("/") if s]
    candidate = None
    if host.endswith("youtu.be") and segments:
        candidate = segments[0]
    elif "youtube.com" in host:
        query = parse_qs(parsed.query)
        if query.get("v"):
            candidate = query["v"][0]
        elif segments and segments[0] in ("shorts", "embed", "live") and len(segments) > 1:
            candidate = segments[1]
    if candidate and len(candidate) == 11:
        return candidate
    raise ToolFault(
        "INVALID_INPUT",
        f"Could not extract a video ID from '{reference}'. Pass a YouTube video "
        f"URL or the 11-character video ID.",
        False,
    )


async def _video_context_pipeline(video_id, meter):
    now = _now()
    data = await youtube.list_videos([video_id], meter)
    items = data.get("items") or []
    if not items:
        raise ToolFault(
            "VIDEO_NOT_FOUND",
            f"No YouTube video exists for id '{video_id}'. Check the URL/ID.",
            False,
        )
    video = youtube.normalize_video(items[0])
    if video["views"] is None or video["published_at"] is None:
        raise ToolFault(
            "VIDEO_NOT_FOUND",
            "This video's view count is hidden or metadata is unavailable, so it "
            "cannot be analyzed.",
            False,
        )

    channel_data = await youtube.list_channels([video["channel_id"]], meter)
    channel_items = channel_data.get("items") or []
    if not channel_items:
        raise ToolFault(
            "CHANNEL_NOT_FOUND", "The video's channel could not be fetched.", False
        )
    channel = youtube.normalize_channel(channel_items[0])

    percentile = None
    velocity_multiple = None
    channel_median_vpd = None
    try:
        uploads = await _recent_uploads(channel, meter)
    except ToolFault as fault:
        if fault.code in ("QUOTA_EXHAUSTED", "MISSING_API_KEY"):
            raise
        uploads = []
    baseline = analysis.recent_median_baseline(uploads, now)
    if baseline is None:
        baseline = analysis.lifetime_average(channel["view_count"], channel["video_count"])
        method = "lifetime_avg"
    else:
        method = "recent_median"

    peers = [
        u
        for u in uploads
        if u["views"] is not None and u["published_at"] is not None and u["id"] != video_id
    ]
    if peers:
        percentile = round(
            100 * sum(1 for u in peers if u["views"] < video["views"]) / len(peers)
        )
        peer_vpd = sorted(
            analysis.views_per_day(u["views"], u["published_at"], now) for u in peers
        )
        channel_median_vpd = peer_vpd[len(peer_vpd) // 2]
        if channel_median_vpd > 0:
            velocity_multiple = (
                analysis.views_per_day(video["views"], video["published_at"], now)
                / channel_median_vpd
            )

    multiple = analysis.outlier_multiple(video["views"], baseline)
    classification = analysis.classify_video(multiple)
    age_days = max((now - video["published_at"]).total_seconds() / 86400, 1.0)

    video_format = _video_format(video)
    why = []
    if multiple is not None:
        why.append(
            f"{multiple:.1f}x the channel's "
            f"{'recent median' if method == 'recent_median' else 'lifetime average'} "
            f"of {round(baseline):,} views"
        )
    else:
        why.append(
            f"channel baseline is below {config.BASELINE_FLOOR} views, too small "
            f"to compute a meaningful multiple"
        )
    if video_format != "video":
        why.append(
            f"note: this is a {video_format}, compared against the channel's "
            f"long-form baseline (streams and Shorts accumulate views differently)"
        )
    if percentile is not None:
        why.append(f"beats {percentile}% of the channel's recent uploads")
    if velocity_multiple is not None:
        why.append(
            f"has averaged {velocity_multiple:.1f}x more views per day over its "
            f"life than the channel's recent uploads"
        )

    return {
        "ok": True,
        "video": {
            "title": video["title"],
            "url": _watch_url(video_id),
            "views": video["views"],
            "published_at": _iso(video["published_at"]),
            "age_days": round(age_days, 1),
            "format": video_format,
        },
        "channel": {
            "id": channel["id"],
            "title": channel["title"],
            "subscribers": channel["subs"] if channel["subs"] is not None else 0,
            "baseline": round(baseline),
            "baseline_method": method,
        },
        "outlier_multiple": round(multiple, 2) if multiple is not None else None,
        "percentile_vs_recent_uploads": percentile,
        "views_per_day": round(analysis.views_per_day(video["views"], video["published_at"], now)),
        "channel_median_views_per_day": round(channel_median_vpd) if channel_median_vpd else None,
        "velocity_multiple": round(velocity_multiple, 2) if velocity_multiple else None,
        "classification": classification,
        "why": why,
    }


async def _video_context_request(video):
    params_hash = _params_hash((video,))
    started = time.monotonic()
    try:
        video_id = _parse_video_ref(video)
    except ToolFault as fault:
        _log_request("video_context", params_hash, "-", 0, started, fault.code)
        return fault.to_response()
    cache_key = ("video_context", video_id)
    return await _execute(
        "video_context",
        params_hash,
        cache_key,
        config.CHANNEL_CACHE_TTL_S,
        config.VIDEO_CONTEXT_WORST_CASE_UNITS,
        lambda meter: _video_context_pipeline(video_id, meter),
    )


@mcp.tool
async def video_context(video: str) -> dict:
    """Explain any YouTube video's performance relative to its own channel,
    using live YouTube data: outlier multiple vs the channel's recent-median
    baseline, percentile among recent uploads, views-per-day velocity, and a
    classification from MEGA_OUTLIER to UNDERPERFORMER.

    video: a YouTube video URL or the 11-character video ID.
    """
    return await _video_context_request(video)


# ---------------------------------------------------------------------------
# Tool 4: rising_channels — momentum radar for a niche
# ---------------------------------------------------------------------------


def _validate_radar(niche, region_code, recency_days, max_subs):
    query, region, days, _ = _validate_scan(
        niche, region_code, recency_days, config.RADAR_SEARCH_RESULTS
    )
    if not (config.RADAR_MAX_SUBS_MIN <= max_subs <= config.RADAR_MAX_SUBS_MAX):
        raise ToolFault(
            "INVALID_INPUT",
            f"'max_subs' must be between {config.RADAR_MAX_SUBS_MIN} and "
            f"{config.RADAR_MAX_SUBS_MAX}.",
            False,
        )
    return query, region, days, int(max_subs)


async def _radar_pipeline(niche, region, recency_days, max_subs, meter):
    now = _now()
    published_after = _iso(now - timedelta(days=recency_days))

    search = await youtube.search_videos(
        niche, region, published_after, config.RADAR_SEARCH_RESULTS, meter
    )
    video_ids = _dedupe(
        item["id"]["videoId"]
        for item in search.get("items", [])
        if item.get("id", {}).get("videoId")
    )
    videos = []
    if video_ids:
        data = await youtube.list_videos(video_ids, meter)
        for item in data.get("items", []):
            video = youtube.normalize_video(item)
            if video["views"] is None or video["published_at"] is None:
                continue
            if analysis.is_short(video["seconds"]) or not video["channel_id"]:
                continue
            videos.append(video)

    channels_by_id = {}
    channel_ids = _dedupe(v["channel_id"] for v in videos)
    if channel_ids:
        data = await youtube.list_channels(channel_ids, meter)
        channels_by_id = {
            channel["id"]: channel
            for channel in (youtube.normalize_channel(item) for item in data.get("items", []))
        }
    videos = [v for v in videos if v["channel_id"] in channels_by_id]

    # candidates: small enough, with a lifetime average above the floor
    candidates = []
    for cid, channel in channels_by_id.items():
        if channel["subs"] is None or channel["subs"] > max_subs:
            continue
        if (
            analysis.lifetime_average(channel["view_count"], channel["video_count"])
            < config.BASELINE_FLOOR
        ):
            continue
        candidates.append(cid)

    result_counts = Counter(v["channel_id"] for v in videos)
    views_in_set = defaultdict(int)
    for v in videos:
        views_in_set[v["channel_id"]] += v["views"]
    candidates.sort(key=lambda cid: (-result_counts[cid], -views_in_set[cid], cid))
    deep_ids = candidates[: config.RADAR_DEEP_CHANNELS]

    momentum_by_id = {}
    semaphore = asyncio.Semaphore(config.BASELINE_CONCURRENCY)

    async def measure(channel_id):
        channel = channels_by_id[channel_id]
        async with semaphore:
            try:
                uploads = await _recent_uploads(channel, meter)
            except ToolFault as fault:
                if fault.code in ("QUOTA_EXHAUSTED", "MISSING_API_KEY"):
                    raise
                return
        recent = analysis.recent_median_baseline(uploads, now)
        if recent is None:
            return  # not enough qualifying uploads to measure momentum honestly
        lifetime = analysis.lifetime_average(channel["view_count"], channel["video_count"])
        momentum_by_id[channel_id] = (recent, lifetime, recent / lifetime)

    await asyncio.gather(*(measure(cid) for cid in deep_ids))

    rising = []
    cooling_count = 0
    for cid, (recent, lifetime, momentum) in momentum_by_id.items():
        if momentum < config.RADAR_MIN_MOMENTUM:
            cooling_count += 1  # declining channels are never sold as "rising"
            continue
        channel = channels_by_id[cid]
        top_video = max(
            (v for v in videos if v["channel_id"] == cid), key=lambda v: v["views"]
        )
        sample_multiple = analysis.outlier_multiple(top_video["views"], recent)
        rising.append(
            {
                "channel": channel["title"],
                "channel_id": cid,
                "url": "https://www.youtube.com/channel/" + cid,
                "subscribers": channel["subs"],
                "videos_in_results": result_counts[cid],
                "recent_median_views": round(recent),
                "lifetime_avg_views": round(lifetime),
                "momentum_multiple": round(momentum, 2),
                "top_video_in_results": {
                    "title": top_video["title"],
                    "url": _watch_url(top_video["id"]),
                    "views": top_video["views"],
                    "outlier_multiple": round(sample_multiple, 2)
                    if sample_multiple is not None
                    else None,
                },
            }
        )
    rising.sort(key=lambda r: -r["momentum_multiple"])

    measured = len(momentum_by_id)
    if measured == 0:
        note = (
            "no channels in this niche could be measured for momentum (too few "
            "qualifying recent uploads); try a broader niche or a higher max_subs"
        )
    elif not rising:
        note = (
            f"none of the {measured} measured channels are genuinely rising — every "
            f"recent median sits below its lifetime average; this niche is cooling"
        )
    else:
        note = (
            f"{len(rising)} of {measured} measured channels have recent-median views "
            f"above their lifetime average ({cooling_count} cooling channels excluded)"
        )

    return {
        "ok": True,
        "niche": niche,
        "region_code": region,
        "analyzed": {
            "videos": len(videos),
            "channels": len(channels_by_id),
            "candidates_measured": measured,
            "cooling_excluded": cooling_count,
        },
        "note": note,
        "rising": rising[: config.RADAR_MAX_CHANNELS],
    }


async def _radar_request(niche, region_code, recency_days, max_subs):
    params_hash = _params_hash((niche, region_code, recency_days, max_subs))
    started = time.monotonic()
    try:
        q, region, days, subs_cap = _validate_radar(
            niche, region_code, recency_days, max_subs
        )
    except ToolFault as fault:
        _log_request("rising_channels", params_hash, "-", 0, started, fault.code)
        return fault.to_response()
    cache_key = ("rising_channels", q.lower(), region, days, subs_cap)
    return await _execute(
        "rising_channels",
        params_hash,
        cache_key,
        config.SCAN_CACHE_TTL_S,
        config.RADAR_WORST_CASE_UNITS,
        lambda meter: _radar_pipeline(q, region, days, subs_cap, meter),
    )


@mcp.tool
async def rising_channels(
    niche: str,
    region_code: str = config.DEFAULT_REGION_CODE,
    recency_days: int = config.RECENCY_DAYS_DEFAULT,
    max_subs: int = config.RADAR_MAX_SUBS_DEFAULT,
) -> dict:
    """Find the fastest-rising channels in a YouTube niche using live data:
    channels whose recent-median views far exceed their lifetime average.
    Built for sponsor scouting, collab targeting, and competitor detection.

    niche: the niche keyword (2-80 chars). region_code: ISO 3166-1 alpha-2.
    recency_days: search window (30-1825). max_subs: only return channels at
    or below this subscriber count (1000-10000000, default 500000).
    """
    return await _radar_request(niche, region_code, recency_days, max_subs)


# ---------------------------------------------------------------------------
# Paid REST endpoints (x402-gated; same pipelines, same no-mock rules)
# ---------------------------------------------------------------------------

# "enforced" (x402 middleware installed) | "free" (explicitly disabled, serve
# results directly) | "unconfigured" (payments expected but credentials absent)
_payment_mode = "unconfigured"


def _payment_gate_error():
    return JSONResponse(
        {
            "ok": False,
            "error": {
                "code": "PAYMENT_NOT_CONFIGURED",
                "message": "This paid endpoint is not accepting calls yet: the "
                "operator has not configured payment credentials. Retry after "
                "the operator completes setup.",
                "retryable": True,
            },
        },
        status_code=503,
    )


_REST_STATUS = {
    "INVALID_INPUT": 400,
    "CHANNEL_NOT_FOUND": 404,
    "VIDEO_NOT_FOUND": 404,
    "QUOTA_EXHAUSTED": 429,
    "YT_API_ERROR": 502,
    "MISSING_API_KEY": 503,
    "UPSTREAM_TIMEOUT": 504,
}


def _rest_response(result):
    """REST callers get real HTTP status codes; the body stays structured."""
    if result.get("ok"):
        return JSONResponse(result)
    code = result.get("error", {}).get("code", "")
    return JSONResponse(result, status_code=_REST_STATUS.get(code, 500))


def _int_param(params, name, default):
    raw = params.get(name)
    if raw is None or raw == "":
        return default
    try:
        return int(raw)
    except ValueError:
        raise ToolFault("INVALID_INPUT", f"'{name}' must be an integer.", False)


def _float_param(params, name, default):
    raw = params.get(name)
    if raw is None or raw == "":
        return default
    try:
        return float(raw)
    except ValueError:
        raise ToolFault("INVALID_INPUT", f"'{name}' must be a number.", False)


@mcp.custom_route(config.PAID_SCAN_PATH, methods=["GET"])
async def paid_scan_niche(request: Request) -> JSONResponse:
    if _payment_mode == "unconfigured":
        return _payment_gate_error()
    params = request.query_params
    try:
        recency_days = _int_param(params, "recency_days", config.RECENCY_DAYS_DEFAULT)
        max_results = _int_param(params, "max_results", config.MAX_RESULTS_DEFAULT)
    except ToolFault as fault:
        return _rest_response(fault.to_response())
    result = await _scan_request(
        params.get("query", ""),
        params.get("region_code", config.DEFAULT_REGION_CODE),
        recency_days,
        max_results,
    )
    return _rest_response(result)


@mcp.custom_route(config.PAID_CHANNEL_PATH, methods=["GET"])
async def paid_channel_outliers(request: Request) -> JSONResponse:
    if _payment_mode == "unconfigured":
        return _payment_gate_error()
    params = request.query_params
    try:
        lookback = _int_param(params, "lookback_videos", config.LOOKBACK_VIDEOS_DEFAULT)
        multiple = _float_param(params, "min_multiple", config.MIN_MULTIPLE_DEFAULT)
    except ToolFault as fault:
        return _rest_response(fault.to_response())
    result = await _channel_request(params.get("channel", ""), lookback, multiple)
    return _rest_response(result)


# ---------------------------------------------------------------------------
# Health check + ASGI app + entrypoint
# ---------------------------------------------------------------------------


@mcp.custom_route("/api/video_context", methods=["GET"])  # neutral alias while free
@mcp.custom_route(config.PAID_VIDEO_PATH, methods=["GET"])
async def paid_video_context(request: Request) -> JSONResponse:
    if _payment_mode == "unconfigured":
        return _payment_gate_error()
    result = await _video_context_request(request.query_params.get("video", ""))
    return _rest_response(result)


@mcp.custom_route(config.PAID_RADAR_PATH, methods=["GET"])
async def paid_rising_channels(request: Request) -> JSONResponse:
    if _payment_mode == "unconfigured":
        return _payment_gate_error()
    params = request.query_params
    try:
        recency_days = _int_param(params, "recency_days", config.RECENCY_DAYS_DEFAULT)
        max_subs = _int_param(params, "max_subs", config.RADAR_MAX_SUBS_DEFAULT)
    except ToolFault as fault:
        return _rest_response(fault.to_response())
    result = await _radar_request(
        params.get("niche", ""),
        params.get("region_code", config.DEFAULT_REGION_CODE),
        recency_days,
        max_subs,
    )
    return _rest_response(result)


@mcp.custom_route("/healthz", methods=["GET"])
async def healthz(request: Request) -> JSONResponse:
    return JSONResponse({"ok": True})


def _build_app():
    """Build the ASGI app; install the x402 payment middleware when the
    payment credentials are fully configured."""
    global _payment_mode
    application = mcp.http_app()

    if os.environ.get(config.PAYMENTS_ENABLED_ENV, "true").strip().lower() in (
        "false",
        "0",
        "no",
    ):
        _payment_mode = "free"
        log.info(
            "payments disabled via %s — the /paid endpoints serve results directly",
            config.PAYMENTS_ENABLED_ENV,
        )
        return application

    creds = {
        name: os.environ.get(name, "").strip()
        for name in (
            config.OKX_API_KEY_ENV,
            config.OKX_SECRET_KEY_ENV,
            config.OKX_PASSPHRASE_ENV,
            config.PAY_TO_ADDRESS_ENV,
        )
    }
    missing = [name for name, value in creds.items() if not value]
    if missing:
        log.critical(
            "x402 payment credentials missing (%s) — paid endpoints return "
            "PAYMENT_NOT_CONFIGURED until they are set",
            ", ".join(missing),
        )
        return application

    facilitator = OKXFacilitatorClient(
        OKXFacilitatorConfig(
            auth=OKXAuthConfig(
                api_key=creds[config.OKX_API_KEY_ENV],
                secret_key=creds[config.OKX_SECRET_KEY_ENV],
                passphrase=creds[config.OKX_PASSPHRASE_ENV],
            ),
            base_url=os.environ.get(config.OKX_BASE_URL_ENV, "").strip(),
        )
    )
    x402_server = x402ResourceServer(facilitator)
    x402_server.register(config.X402_NETWORK, ExactEvmScheme())
    x402_server.register(config.X402_NETWORK, AggrDeferredEvmScheme())

    pay_to = creds[config.PAY_TO_ADDRESS_ENV]

    def options(price_usdt):
        price = f"${price_usdt}"
        return [
            PaymentOption(
                scheme="exact", price=price, network=config.X402_NETWORK, pay_to=pay_to
            ),
            PaymentOption(
                scheme="aggr_deferred",
                price=price,
                network=config.X402_NETWORK,
                pay_to=pay_to,
            ),
        ]

    priced = {
        config.PAID_SCAN_PATH: (
            config.PAID_PRICE_SCAN_USDT,
            "Live YouTube niche scan: saturation, outliers, ENTER/CROWDED/AVOID verdict",
        ),
        config.PAID_CHANNEL_PATH: (
            config.PAID_PRICE_CHANNEL_USDT,
            "Live YouTube channel outlier audit vs the channel's own baseline",
        ),
        config.PAID_VIDEO_PATH: (
            config.PAID_PRICE_VIDEO_USDT,
            "Explain any YouTube video's performance vs its own channel's baseline",
        ),
        config.PAID_RADAR_PATH: (
            config.PAID_PRICE_RADAR_USDT,
            "Fastest-rising channels in a YouTube niche by recent-vs-lifetime momentum",
        ),
    }
    # a "0" price means the service is free: leave its route out of the payment
    # gate so it returns the result directly, per the marketplace contract
    routes = {
        f"GET {path}": RouteConfig(
            accepts=options(price), description=description, mime_type="application/json"
        )
        for path, (price, description) in priced.items()
        if float(price) > 0
    }
    application.add_middleware(PaymentMiddlewareASGI, routes=routes, server=x402_server)
    _payment_mode = "enforced"
    log.info(
        "x402 payments enabled on %s, pay-to %s (scan %s / channel %s / video %s / radar %s USDT)",
        config.X402_NETWORK,
        pay_to,
        config.PAID_PRICE_SCAN_USDT,
        config.PAID_PRICE_CHANNEL_USDT,
        config.PAID_PRICE_VIDEO_USDT,
        config.PAID_PRICE_RADAR_USDT,
    )
    return application


app = _build_app()

if __name__ == "__main__":
    if "--stdio" in sys.argv:
        mcp.run(transport="stdio")
    else:
        mcp.run(
            transport="http",
            host="0.0.0.0",
            port=int(os.environ.get("PORT", "8000")),
        )