bboysoul
273cd7aac7
- 后端新增三源：中石化浙江油价(3步会话)、新浪上证行情、Open-Meteo 空气质量(国标AQI换算)，按源独立超时与状态 - 天气行改为 降雨概率% · AQI 等级（无雨无AQI时保留观测时间），stale 补缓存标记 - 前端 strip 六格两行(桌面)/四行(移动)，倒数日取本地缓存最近一条，超12字截断 - 修复两处 AppSheet 动画竞态存量 e2e 失败(modal contract footer 时序、scrollable 重开残留 dialog)
684 lines
24 KiB
Python
684 lines
24 KiB
Python
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}
|