import asyncio
import logging
import math
import re
import threading
import weakref
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal, InvalidOperation
from html import unescape
from typing import Any
from zoneinfo import ZoneInfo

import httpx

from .lunar_support import solar_to_lunar_text

SHANGHAI_TZ = ZoneInfo("Asia/Shanghai")
WEATHER_URL = "https://api.open-meteo.com/v1/forecast"
GOLD_URL = "https://www.sge.com.cn/sjzx/yshqbg"
GOLD_HISTORY_URL = (
    "https://www.sge.com.cn/sjzx/quotation_daily_new"
    "?start_date={start_date}&end_date={end_date}"
)
AIR_URL = "https://air-quality-api.open-meteo.com/v1/air-quality"
OIL_BASE_URL = "https://cx.sinopecsales.com"
OIL_MAIN_PATH = "/yjkqiantai/core/main"
OIL_SWITCH_PATH = "/yjkqiantai/data/switchProvince"
OIL_INIT_PATH = "/yjkqiantai/data/initMainData"
ASHARE_URL = "https://hq.sinajs.cn/list=sh000001"
WEATHER_SOURCE = "Open-Meteo"
GOLD_SOURCE = "上海黄金交易所"
AIR_SOURCE = "Open-Meteo 空气质量"
OIL_SOURCE = "中国石化"
ASHARE_SOURCE = "新浪财经"
ASHARE_HEADERS = {"Referer": "https://finance.sina.com.cn", "User-Agent": "Mozilla/5.0"}
REQUEST_HEADERS = {"User-Agent": "dodo/0.1 (+https://dodo.bboy.app)"}
WEEKDAYS = ("星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日")
SOURCE_TIMEOUT_SECONDS = 3.0
GOLD_SOURCE_TIMEOUT_SECONDS = 6.0
OIL_SOURCE_TIMEOUT_SECONDS = 6.0
TOTAL_TIMEOUT_SECONDS = 6.5
WEATHER_TTL = timedelta(minutes=15)
WEATHER_STALE_TTL = timedelta(hours=6)
GOLD_TTL = timedelta(minutes=5)
GOLD_STALE_TTL = timedelta(days=7)
AIR_TTL = timedelta(hours=1)
AIR_STALE_TTL = timedelta(days=3)
OIL_TTL = timedelta(hours=6)
OIL_STALE_TTL = timedelta(days=30)
ASHARE_TTL = timedelta(minutes=10)
ASHARE_STALE_TTL = timedelta(days=7)

logger = logging.getLogger(__name__)

_TAG_RE = re.compile(r"<[^>]+>")
_DATE_RE = re.compile(r"上海黄金交易所\s*(\d{4})年(\d{2})月(\d{2})日\s*延时行情")
_ROW_RE = re.compile(r"<tr\b[^>]*>(.*?)</tr>", re.IGNORECASE | re.DOTALL)
_CELL_RE = re.compile(r"<t[dh]\b[^>]*>(.*?)</t[dh]>", re.IGNORECASE | re.DOTALL)


@dataclass
class CacheEntry:
    value: dict[str, Any]
    fetched_at: datetime


@dataclass
class LoopRuntime:
    inflight: dict[tuple[int, str], asyncio.Task[dict[str, Any]]]


_cache: dict[str, CacheEntry] = {}
_generation = 0
_state_guard = threading.Lock()
_runtimes: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, LoopRuntime] = (
    weakref.WeakKeyDictionary()
)


def _runtime(loop: asyncio.AbstractEventLoop | None = None) -> LoopRuntime:
    active_loop = loop or asyncio.get_running_loop()
    with _state_guard:
        runtime = _runtimes.get(active_loop)
        if runtime is None:
            runtime = LoopRuntime(inflight={})
            _runtimes[active_loop] = runtime
        return runtime


def clear_cache() -> None:
    global _generation
    with _state_guard:
        _generation += 1
        _cache.clear()


def _text(fragment: str) -> str:
    return " ".join(unescape(_TAG_RE.sub(" ", fragment)).split())


def parse_sge_au9999(html: str) -> tuple[Decimal, str]:
    date_match = _DATE_RE.search(_text(html))
    if date_match is None:
        raise ValueError("SGE market date not found")
    try:
        year, month, day = map(int, date_match.groups())
        market_date = date(year, month, day).isoformat()
    except ValueError as exc:
        raise ValueError("invalid SGE market date") from exc

    rows = [[_text(cell) for cell in _CELL_RE.findall(row)] for row in _ROW_RE.findall(html)]
    header = next((cells for cells in rows if "合约" in cells and "最新价" in cells), None)
    if header is None:
        raise ValueError("SGE quote table header not found")
    contract_index = header.index("合约")
    price_index = header.index("最新价")
    required_length = max(contract_index, price_index) + 1
    for cells in rows[rows.index(header) + 1 :]:
        if len(cells) < required_length or cells[contract_index] != "Au99.99":
            continue
        try:
            price = Decimal(cells[price_index].replace(",", ""))
        except InvalidOperation as exc:
            raise ValueError("invalid Au99.99 latest price") from exc
        if not price.is_finite() or price <= 0:
            raise ValueError("invalid Au99.99 latest price")
        return price, market_date
    raise ValueError("Au99.99 quote not found")


def parse_sge_au9999_history(html: str) -> tuple[Decimal, str]:
    rows = [[_text(cell) for cell in _CELL_RE.findall(row)] for row in _ROW_RE.findall(html)]
    header = next(
        (cells for cells in rows if "日期" in cells and "合约" in cells and "收盘价" in cells),
        None,
    )
    if header is None:
        raise ValueError("SGE history table header not found")
    date_index = header.index("日期")
    contract_index = header.index("合约")
    price_index = header.index("收盘价")
    required_length = max(date_index, contract_index, price_index) + 1
    quotes = []
    for cells in rows[rows.index(header) + 1 :]:
        if len(cells) < required_length or cells[contract_index] != "Au99.99":
            continue
        try:
            market_date = date.fromisoformat(cells[date_index]).isoformat()
            price = Decimal(cells[price_index].replace(",", ""))
        except (InvalidOperation, ValueError):
            continue
        if not price.is_finite() or price <= 0:
            continue
        quotes.append((market_date, price))
    if not quotes:
        raise ValueError("Au99.99 history quote not found")
    market_date, price = max(quotes)
    return price, market_date


async def fetch_weather(request: Callable[..., Awaitable[dict[str, Any]]]) -> dict[str, Any]:
    payload = await request(
        WEATHER_URL,
        params={
            "latitude": 29.88,
            "longitude": 121.55,
            "current": "temperature_2m,apparent_temperature,weather_code",
            "hourly": "precipitation_probability",
            "timezone": "Asia/Shanghai",
        },
    )
    if payload.get("timezone") != "Asia/Shanghai":
        raise ValueError("weather timezone must be Asia/Shanghai")
    current = payload["current"]
    if not isinstance(current, dict):
        raise TypeError("weather current must be an object")

    def finite_number(name: str, minimum: float, maximum: float) -> int | float:
        value = current[name]
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            raise TypeError(f"weather {name} must be numeric")
        if not math.isfinite(value) or not minimum <= value <= maximum:
            raise ValueError(f"weather {name} out of range")
        return value

    temperature = finite_number("temperature_2m", -100, 100)
    apparent_temperature = finite_number("apparent_temperature", -100, 100)
    weather_code = current["weather_code"]
    if isinstance(weather_code, bool) or not isinstance(weather_code, int):
        raise TypeError("weather weather_code must be an integer")
    if not 0 <= weather_code <= 99:
        raise ValueError("weather weather_code out of range")
    raw_time = current["time"]
    if not isinstance(raw_time, str):
        raise TypeError("weather time must be a string")
    observed_at = datetime.fromisoformat(raw_time)
    if observed_at.tzinfo is not None:
        if observed_at.utcoffset() != SHANGHAI_TZ.utcoffset(observed_at):
            raise ValueError("weather time must use Asia/Shanghai")
        observed_at = observed_at.astimezone(SHANGHAI_TZ)
    else:
        observed_at = observed_at.replace(tzinfo=SHANGHAI_TZ)
    precipitation_probability = _precipitation_probability(payload, observed_at)
    return {
        "temperature_c": temperature,
        "apparent_temperature_c": apparent_temperature,
        "weather_code": weather_code,
        "precipitation_probability": precipitation_probability,
        "observed_at": observed_at.isoformat(),
        "source": WEATHER_SOURCE,
    }


def _precipitation_probability(payload: dict[str, Any], observed_at: datetime) -> int | None:
    hourly = payload.get("hourly")
    if not isinstance(hourly, dict):
        raise TypeError("weather hourly must be an object")
    times = hourly.get("time")
    probabilities = hourly.get("precipitation_probability")
    if not isinstance(times, list) or not isinstance(probabilities, list):
        raise TypeError("weather hourly precipitation_probability must be a list")
    if len(times) != len(probabilities) or not times or not all(isinstance(value, str) for value in times):
        raise ValueError("weather hourly precipitation_probability must align with time")
    current_hour = observed_at.strftime("%Y-%m-%dT%H:00")
    try:
        index = times.index(current_hour)
    except ValueError:
        return None
    window = probabilities[index : index + 3]
    valid: list[float] = []
    for value in window:
        if value is None:
            continue
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            raise TypeError("weather precipitation_probability must be numeric or null")
        if not math.isfinite(value) or not 0 <= value <= 100:
            raise ValueError("weather precipitation_probability out of range")
        valid.append(float(value))
    return round(max(valid)) if valid else None


async def fetch_gold(request: Callable[..., Awaitable[str]]) -> dict[str, Any]:
    html = await request(GOLD_URL)
    try:
        price, market_date = parse_sge_au9999(html)
    except ValueError:
        date_match = _DATE_RE.search(_text(html))
        if date_match is None:
            raise
        try:
            quote_date = date(*map(int, date_match.groups()))
        except ValueError as exc:
            raise ValueError("invalid SGE market date") from exc
        start_date = quote_date - timedelta(days=31)
        history_url = GOLD_HISTORY_URL.format(
            start_date=start_date.isoformat(), end_date=quote_date.isoformat()
        )
        price, market_date = parse_sge_au9999_history(await request(history_url))
    return {
        "contract": "Au99.99",
        "latest_price": price,
        "currency": "CNY",
        "unit": "gram",
        "market_date": market_date,
        "delayed": True,
        "source": GOLD_SOURCE,
    }


_PM25_BREAKPOINTS = (
    (0.0, 35.0, 0.0, 50.0),
    (35.0, 75.0, 50.0, 100.0),
    (75.0, 115.0, 100.0, 150.0),
    (115.0, 150.0, 150.0, 200.0),
    (150.0, 250.0, 200.0, 300.0),
    (250.0, 350.0, 300.0, 400.0),
    (350.0, 500.0, 400.0, 500.0),
)
_PM10_BREAKPOINTS = (
    (0.0, 50.0, 0.0, 50.0),
    (50.0, 150.0, 50.0, 100.0),
    (150.0, 250.0, 100.0, 150.0),
    (250.0, 350.0, 150.0, 200.0),
    (350.0, 420.0, 200.0, 300.0),
    (420.0, 500.0, 300.0, 400.0),
    (500.0, 600.0, 400.0, 500.0),
)


def _sub_index(concentration: float, breakpoints: tuple[tuple[float, float, float, float], ...]) -> float:
    if concentration <= 0:
        return 0.0
    for low, high, index_low, index_high in breakpoints:
        if concentration <= high:
            return (index_high - index_low) / (high - low) * (concentration - low) + index_low
    return 500.0


def china_aqi(pm2_5: float, pm10: float) -> int:
    """HJ 633-2012 AQI from PM2.5 / PM10 concentrations (μg/m³), capped at 500."""
    return round(max(_sub_index(pm2_5, _PM25_BREAKPOINTS), _sub_index(pm10, _PM10_BREAKPOINTS)))


def aqi_level(aqi: int) -> str:
    if aqi <= 50:
        return "优"
    if aqi <= 100:
        return "良"
    if aqi <= 150:
        return "轻度污染"
    if aqi <= 200:
        return "中度污染"
    if aqi <= 300:
        return "重度污染"
    return "严重污染"


async def fetch_air(request: Callable[..., Awaitable[dict[str, Any]]]) -> dict[str, Any]:
    payload = await request(
        AIR_URL,
        params={
            "latitude": 29.88,
            "longitude": 121.55,
            "current": "pm2_5,pm10",
            "timezone": "Asia/Shanghai",
        },
    )
    if payload.get("timezone") != "Asia/Shanghai":
        raise ValueError("air quality timezone must be Asia/Shanghai")
    current = payload["current"]
    if not isinstance(current, dict):
        raise TypeError("air quality current must be an object")

    def concentration(name: str) -> float:
        value = current[name]
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            raise TypeError(f"air quality {name} must be numeric")
        if not math.isfinite(value) or not 0 <= value <= 2000:
            raise ValueError(f"air quality {name} out of range")
        return float(value)

    pm2_5 = concentration("pm2_5")
    pm10 = concentration("pm10")
    raw_time = current["time"]
    if not isinstance(raw_time, str):
        raise TypeError("air quality time must be a string")
    observed_at = datetime.fromisoformat(raw_time)
    if observed_at.tzinfo is not None:
        if observed_at.utcoffset() != SHANGHAI_TZ.utcoffset(observed_at):
            raise ValueError("air quality time must use Asia/Shanghai")
        observed_at = observed_at.astimezone(SHANGHAI_TZ)
    else:
        observed_at = observed_at.replace(tzinfo=SHANGHAI_TZ)
    aqi = china_aqi(pm2_5, pm10)
    return {
        "aqi": aqi,
        "level": aqi_level(aqi),
        "pm2_5": pm2_5,
        "pm10": pm10,
        "observed_at": observed_at.isoformat(),
        "source": AIR_SOURCE,
    }


async def _request_json(client: httpx.AsyncClient, url: str, **kwargs: Any) -> dict[str, Any]:
    response = await client.get(url, **kwargs)
    response.raise_for_status()
    return response.json()


async def _request_text(client: httpx.AsyncClient, url: str, **kwargs: Any) -> str:
    response = await client.get(url, **kwargs)
    response.raise_for_status()
    return response.text


async def _request_bytes(client: httpx.AsyncClient, url: str, **kwargs: Any) -> bytes:
    response = await client.get(url, **kwargs)
    response.raise_for_status()
    return response.content


def parse_sina_index(text: str) -> dict[str, Any]:
    match = re.search(r'"([^"]*)"', text)
    if match is None:
        raise ValueError("sina quote not found")
    fields = match.group(1).split(",")
    if len(fields) < 5:
        raise ValueError("sina quote is too short")
    if fields[0] != "上证指数":
        raise ValueError("unexpected sina symbol")

    def number(index: int) -> Decimal:
        try:
            value = Decimal(fields[index].replace(",", ""))
        except InvalidOperation as exc:
            raise ValueError("invalid sina number") from exc
        if not value.is_finite() or value <= 0:
            raise ValueError("invalid sina number")
        return value

    previous_close = number(2)
    price = number(3)
    as_of = next((field for field in fields if re.fullmatch(r"\d{4}-\d{2}-\d{2}", field)), None)
    if as_of is None:
        raise ValueError("sina quote date not found")
    try:
        date.fromisoformat(as_of)
    except ValueError as exc:
        raise ValueError("invalid sina quote date") from exc
    change_percent = float((price - previous_close) / previous_close * 100)
    return {
        "name": fields[0],
        "price": price,
        "prev_close": previous_close,
        "change_percent": round(change_percent, 2),
        "as_of": as_of,
        "source": ASHARE_SOURCE,
    }


async def fetch_ashare(request: Callable[..., Awaitable[bytes]]) -> dict[str, Any]:
    raw = await request(ASHARE_URL, headers=ASHARE_HEADERS)
    try:
        text = raw.decode("gbk")
    except UnicodeDecodeError as exc:
        raise ValueError("sina quote is not decodable as GBK") from exc
    return parse_sina_index(text)


def parse_oil_payload(payload: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(payload, dict):
        raise TypeError("sinopec payload must be an object")
    data = payload.get("data")
    if not isinstance(data, dict):
        raise TypeError("sinopec data must be an object")
    province = data.get("provinceData")
    if not isinstance(province, dict):
        raise TypeError("sinopec provinceData must be an object")

    def price(key: str) -> Decimal:
        if key not in province:
            raise ValueError(f"sinopec {key} missing")
        try:
            value = Decimal(str(province[key]))
        except InvalidOperation as exc:
            raise ValueError(f"sinopec {key} invalid") from exc
        if not value.is_finite() or not Decimal("0.01") <= value <= Decimal(100):
            raise ValueError(f"sinopec {key} out of range")
        return value

    effective_at = None
    raw_effective = province.get("START_DATE")
    if isinstance(raw_effective, str) and raw_effective:
        try:
            parsed = datetime.strptime(raw_effective, "%Y-%m-%d %H:%M:%S").replace(tzinfo=SHANGHAI_TZ)
        except ValueError:
            effective_at = None
        else:
            effective_at = parsed.isoformat()
    return {
        "gas_92": price("GAS_92"),
        "gas_95": price("GAS_95"),
        "gas_98": price("AIPAO_GAS_98"),
        "diesel_0": price("CHECHAI_0"),
        "effective_at": effective_at,
        "source": OIL_SOURCE,
    }


async def fetch_oil(client: httpx.AsyncClient) -> dict[str, Any]:
    main = await client.get(f"{OIL_BASE_URL}{OIL_MAIN_PATH}")
    main.raise_for_status()
    if client.cookies.get("SESSION") is None:
        raise ValueError("sinopec session cookie missing")
    switch = await client.post(f"{OIL_BASE_URL}{OIL_SWITCH_PATH}", json={"provinceId": "33"})
    switch.raise_for_status()
    init = await client.get(
        f"{OIL_BASE_URL}{OIL_INIT_PATH}",
        headers={"Referer": f"{OIL_BASE_URL}{OIL_MAIN_PATH}"},
    )
    init.raise_for_status()
    return parse_oil_payload(init.json())


def _serializable(value: dict[str, Any]) -> dict[str, Any]:
    return {key: str(item) if isinstance(item, Decimal) else item for key, item in value.items()}


def _source_timeout(name: str) -> float:
    if name == "gold":
        return GOLD_SOURCE_TIMEOUT_SECONDS
    if name == "oil":
        return OIL_SOURCE_TIMEOUT_SECONDS
    return SOURCE_TIMEOUT_SECONDS


async def _refresh_source(
    name: str,
    fetcher: Callable[[], Awaitable[dict[str, Any]]],
    fetched_at: datetime,
    generation: int,
) -> dict[str, Any]:
    value = await asyncio.wait_for(fetcher(), timeout=_source_timeout(name))
    with _state_guard:
        if generation == _generation:
            newest = _cache.get(name)
            if newest is None or newest.fetched_at <= fetched_at:
                _cache[name] = CacheEntry(value=value, fetched_at=fetched_at)
    return value


def _source_error_code(exc: BaseException) -> str:
    if isinstance(exc, TimeoutError):
        return "timeout"
    if isinstance(exc, (KeyError, TypeError, ValueError, InvalidOperation)):
        return "invalid_upstream_response"
    return "upstream_unavailable"


def _finish_inflight(
    runtime: LoopRuntime,
    key: tuple[int, str],
    task: asyncio.Task[dict[str, Any]],
) -> None:
    # Calling exception() marks failures as retrieved even when every shielded waiter
    # was cancelled. This synchronous callback cannot create an untracked cleanup task.
    if not task.cancelled():
        try:
            task.exception()
        except asyncio.CancelledError:
            pass
    if runtime.inflight.get(key) is task:
        runtime.inflight.pop(key, None)


async def _cached_source(
    name: str,
    fetcher: Callable[[], Awaitable[dict[str, Any]]],
    now: datetime,
    fresh_for: timedelta,
    stale_for: timedelta,
) -> tuple[dict[str, Any] | None, str | None]:
    runtime = _runtime()
    with _state_guard:
        generation = _generation
    key = (generation, name)
    with _state_guard:
        cached = _cache.get(name) if generation == _generation else None
    if cached is not None and now - cached.fetched_at <= fresh_for:
        return {**_serializable(cached.value), "stale": False}, None
    task = runtime.inflight.get(key)
    if task is None:
        task = asyncio.create_task(_refresh_source(name, fetcher, now, generation))
        runtime.inflight[key] = task
        task.add_done_callback(
            lambda completed, active=runtime, task_key=key: _finish_inflight(
                active, task_key, completed
            )
        )

    try:
        value = await asyncio.shield(task)
    except Exception as exc:
        error_code = _source_error_code(exc)
        logger.warning("Today environment source %s failed (%s)", name, error_code, exc_info=exc)
        with _state_guard:
            generation_is_current = generation == _generation
        if generation_is_current and cached is not None and now - cached.fetched_at <= stale_for:
            return {**_serializable(cached.value), "stale": True}, error_code
        return None, error_code
    return {**_serializable(value), "stale": False}, None


async def _default_weather_fetcher() -> dict[str, Any]:
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(SOURCE_TIMEOUT_SECONDS),
        headers={"User-Agent": "dodo/0.1 (+https://dodo.bboy.app)"},
    ) as client:
        return await fetch_weather(lambda url, **kwargs: _request_json(client, url, **kwargs))


async def _default_gold_fetcher() -> dict[str, Any]:
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(SOURCE_TIMEOUT_SECONDS),
        headers={"User-Agent": "dodo/0.1 (+https://dodo.bboy.app)"},
    ) as client:
        return await fetch_gold(lambda url, **kwargs: _request_text(client, url, **kwargs))


async def _default_air_fetcher() -> dict[str, Any]:
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(SOURCE_TIMEOUT_SECONDS),
        headers=REQUEST_HEADERS,
    ) as client:
        return await fetch_air(lambda url, **kwargs: _request_json(client, url, **kwargs))


async def _default_oil_fetcher() -> dict[str, Any]:
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(OIL_SOURCE_TIMEOUT_SECONDS),
        headers={"User-Agent": "Mozilla/5.0"},
    ) as client:
        return await fetch_oil(client)


async def _default_ashare_fetcher() -> dict[str, Any]:
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(SOURCE_TIMEOUT_SECONDS),
        headers=REQUEST_HEADERS,
    ) as client:
        return await fetch_ashare(lambda url, **kwargs: _request_bytes(client, url, **kwargs))


async def get_environment(
    *,
    now: datetime | None = None,
    weather_fetcher: Callable[[], Awaitable[dict[str, Any]]] | None = None,
    gold_fetcher: Callable[[], Awaitable[dict[str, Any]]] | None = None,
) -> dict[str, Any]:
    current = now or datetime.now(UTC)
    local_date = current.astimezone(SHANGHAI_TZ).date()
    weather_task = _cached_source(
        "weather", weather_fetcher or _default_weather_fetcher, current, WEATHER_TTL, WEATHER_STALE_TTL
    )
    gold_task = _cached_source(
        "gold", gold_fetcher or _default_gold_fetcher, current, GOLD_TTL, GOLD_STALE_TTL
    )
    weather_result, gold_result = await asyncio.wait_for(
        asyncio.gather(weather_task, gold_task), timeout=TOTAL_TIMEOUT_SECONDS
    )
    weather, weather_error = weather_result
    gold, gold_error = gold_result
    errors = {}
    if weather_error:
        errors["weather"] = weather_error
    if gold_error:
        errors["gold"] = gold_error
    return {
        "date": {
            "solar_date": local_date.isoformat(),
            "weekday": WEEKDAYS[local_date.weekday()],
            "lunar": solar_to_lunar_text(local_date),
            "timezone": "Asia/Shanghai",
        },
        "weather": weather,
        "gold": gold,
        "errors": errors,
    }


async def get_environment_extras(
    *,
    now: datetime | None = None,
    air_fetcher: Callable[[], Awaitable[dict[str, Any]]] | None = None,
    oil_fetcher: Callable[[], Awaitable[dict[str, Any]]] | None = None,
    ashare_fetcher: Callable[[], Awaitable[dict[str, Any]]] | None = None,
) -> dict[str, Any]:
    """Air quality, Zhejiang oil prices, and the SSE index for the Today strip."""
    current = now or datetime.now(UTC)
    sources = (
        ("air", air_fetcher or _default_air_fetcher, AIR_TTL, AIR_STALE_TTL),
        ("oil", oil_fetcher or _default_oil_fetcher, OIL_TTL, OIL_STALE_TTL),
        ("ashare", ashare_fetcher or _default_ashare_fetcher, ASHARE_TTL, ASHARE_STALE_TTL),
    )
    results = await asyncio.wait_for(
        asyncio.gather(
            *(
                _cached_source(name, fetcher, current, fresh_for, stale_for)
                for name, fetcher, fresh_for, stale_for in sources
            )
        ),
        timeout=TOTAL_TIMEOUT_SECONDS,
    )
    values: dict[str, Any] = {}
    errors: dict[str, str] = {}
    for (name, *_), (value, error) in zip(sources, results):
        values[name] = value
        if error:
            errors[name] = error
    return {**values, "errors": errors}