gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix chart refresh and bot command surface

Browse Source
This commit is contained in:
2569718930@qq.com
2026-06-13 01:34:09 +08:00
parent cf35e55da4
commit f115be1fd3
26 changed files with 328 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+1
View File
@@ -75,6 +75,7 @@ OPEN_METEO_MIN_CALL_INTERVAL_SEC=5
POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS=1
POLYWEATHER_SCAN_TERMINAL_PAYLOAD_TTL_SEC=600
POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC=45
POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS=3000
POLYWEATHER_HTTP_TIMEOUT_SEC=8
POLYWEATHER_HTTP_RETRY_COUNT=0
POLYWEATHER_HTTP_RETRY_BACKOFF_SEC=0.2
README.md
+1 -3
View File
@@ -25,7 +25,7 @@ Public docs center: `/docs/intro` on the main site (bilingual product documentat
- Subscription live: `Pro Monthly 29.9 USDC / 30 days` and `Pro Quarterly 79.9 USDC / 90 days`.
- Referral pricing live: invited users can get the first monthly Pro at `20 USDC`; inviters receive `3500` points after a valid first Pro payment, capped at 10 paid invites per month.
- `/city` and `/deb` now free (daily cap 10 each); points redeemable for payment discount (`500 pts = 1 USDC`, monthly max `3 USDC`, quarterly max `8 USDC`). Useful user feedback can also receive manual point rewards through ops.
- Points are redeemable for payment discounts (`500 pts = 1 USDC`, monthly max `3 USDC`, quarterly max `8 USDC`). Useful user feedback can also receive manual point rewards through ops.
- Onchain checkout live: Polygon contract checkout (USDC / USDC.e) plus Ethereum mainnet USDC direct-transfer confirmation.
- Auto-reconciliation live: event listener + periodic confirm loop.
- Ops dashboard live: `/ops` for memberships, leaderboard, user feedback triage, manual point grants, and payment incident triage.
@@ -195,8 +195,6 @@ Production payment routes are configured by the backend. Polygon remains the def
| Command | Purpose |
| :-- | :-- |
| `/city <name>` | City real-time analysis |
| `/deb <name>` | DEB historical reconciliation |
| `/top` | User leaderboard |
| `/id` | Show current chat ID |
| `/diag` | Startup diagnostics |
README_ZH.md
+1 -3
View File
@@ -18,7 +18,7 @@
- 已上线订阅制：`Pro 月付 29.9 USDC / 30 天`，`Pro 季度 79.9 USDC / 90 天`。
- 积分获取已切换为邀请制度：被邀请人完成首次 Pro 付款后，邀请人获得 `3500` 积分；Telegram 群发言不再获得积分。
- `/city` 与 `/deb` 已改为免费（每日各 10 次）；积分可用于支付抵扣（`500 分 = 1 USDC`，月付最多抵 `3 USDC`，季度最多抵 `8 USDC`）。真实、有上下文、有价值的用户反馈也可通过运营后台人工奖励积分。
- 积分可用于支付抵扣（`500 分 = 1 USDC`，月付最多抵 `3 USDC`，季度最多抵 `8 USDC`）。真实、有上下文、有价值的用户反馈也可通过运营后台人工奖励积分。
- 邀请首月价：被邀请人首次月付 `20 USDC`；每个邀请人每月最多 10 个有效付费邀请奖励。
- 已上线链上支付：Polygon 合约支付（USDC / USDC.e）+ Ethereum 主网 USDC 直转确认。
- 已上线自动补单：事件监听 + 周期确认双链路。
@@ -205,8 +205,6 @@ POLYWEATHER_OPS_ADMIN_EMAILS=yhrsc30@gmail.com
| 指令 | 用途 |
| :-- | :-- |
| `/city <name>` | 城市实时分析 |
| `/deb <name>` | DEB 历史对账 |
| `/top` | 用户积分排行 |
| `/id` | 查看聊天 Chat ID |
| `/diag` | Bot 启动诊断 |
docker-compose.yml
+1
View File
@@ -103,6 +103,7 @@ services:
POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''
POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-3}
POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-2}
POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS:-3000}
POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-8000}
POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60}
POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-60}
frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx
+1 -1
View File
@@ -923,7 +923,7 @@ export function LiveTemperatureThresholdChart({
lastPatchAtRef.current = now;
markDetailRequest("network");
fetchHourlyForecastForCity(city, { resolution: targetResolution })
fetchHourlyForecastForCity(city, { bypassLocalCache: true, resolution: targetResolution })
.then((data) => {
if (cancelled) return;
if (!data) {
frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts
+58 -2
View File
@@ -20,6 +20,7 @@ import {
__resetHourlyDetailRequestQueueForTest,
__runQueuedHourlyDetailRequestForTest,
clearCityDetailCache,
fetchHourlyForecastForCity,
readCityDetailBatchDiagnostics,
} from "@/components/dashboard/scan-terminal/temperature-chart-logic";
@@ -64,6 +65,13 @@ export async function runTests() {
!querySource.includes("window.setInterval"),
"scan list should subscribe to SSE patch state instead of running a 5-minute interval",
);
assert(
querySource.includes("handleForegroundScanRefresh") &&
querySource.includes('document.visibilityState !== "visible"') &&
querySource.includes("SCAN_CACHE_TTL_MS") &&
querySource.includes("fetchScanTerminal({ forceRefresh: false, showLoading: false })"),
"scan list should silently revalidate stale rows when a long-hidden browser tab returns to the foreground",
);
assert(
chartLogicSource.includes("DASHBOARD_REFRESH_POLICY_MS.metar") &&
!chartSource.includes("window.setInterval"),
@@ -79,8 +87,10 @@ export async function runTests() {
assert(
chartSource.includes("NO_PATCH_CACHED_DETAIL_REFRESH_MS = DASHBOARD_REFRESH_POLICY_MS.observation") &&
chartSource.includes("refreshCachedDetail") &&
chartSource.includes("fetchHourlyForecastForCity(city, { resolution: targetResolution })"),
"visible charts should do a lightweight cached detail refresh every observation cadence when no SSE patch arrives",
chartSource.includes("fetchHourlyForecastForCity(city, { bypassLocalCache: true, resolution: targetResolution })") &&
chartLogicSource.includes("options.bypassLocalCache") &&
chartLogicSource.includes("const forceRefresh = Boolean(options.ignoreCache)"),
"visible charts should bypass the five-minute browser detail cache every observation cadence without force-refreshing backend sources",
);
assert(
chartSource.includes("preloadTemperatureChartCanvas"),
@@ -442,6 +452,7 @@ export async function runTests() {
const originalWindow = (globalThis as any).window;
const originalSessionStorage = (globalThis as any).sessionStorage;
const originalFetch = (globalThis as any).fetch;
const store = new Map<string, string>();
(globalThis as any).window = {};
(globalThis as any).sessionStorage = {
@@ -462,6 +473,50 @@ export async function runTests() {
},
};
try {
clearCityDetailCache();
const revalidationUrls: string[] = [];
let revalidationFetches = 0;
(globalThis as any).fetch = async (url: string) => {
revalidationFetches += 1;
revalidationUrls.push(String(url));
return {
ok: true,
json: async () => ({
cities: ["fallback-revalidate"],
details: {
"fallback-revalidate": {
city: "fallback-revalidate",
hourly: {
times: ["00:00"],
temps: [revalidationFetches],
},
},
},
errors: {},
missing: [],
partial: false,
}),
};
};
const firstDetail = await fetchHourlyForecastForCity("fallback-revalidate", { resolution: "10m" });
const cachedDetail = await fetchHourlyForecastForCity("fallback-revalidate", { resolution: "10m" });
const revalidatedDetail = await fetchHourlyForecastForCity("fallback-revalidate", {
bypassLocalCache: true,
resolution: "10m",
});
assert(
firstDetail?.temps[0] === 1 &&
cachedDetail?.temps[0] === 1 &&
revalidatedDetail?.temps[0] === 2 &&
revalidationFetches === 2,
"bypassLocalCache should revalidate cached chart detail through the network",
);
assert(
revalidationUrls.every((url) => url.includes("force_refresh=false")),
"bypassLocalCache must not force-refresh backend observation sources",
);
clearCityDetailCache();
const cacheKey = "paris:10m";
store.set(
@@ -490,6 +545,7 @@ export async function runTests() {
clearCityDetailCache();
(globalThis as any).window = originalWindow;
(globalThis as any).sessionStorage = originalSessionStorage;
(globalThis as any).fetch = originalFetch;
}
}
frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts
+2 -2
View File
@@ -231,10 +231,10 @@ export function runTests() {
);
const fallbackRefreshBlock = chart.match(/const refreshCachedDetail = \(\) => \{[\s\S]*?\n \};/)?.[0] || "";
assert(
fallbackRefreshBlock.includes("fetchHourlyForecastForCity(city, { resolution: targetResolution })") &&
fallbackRefreshBlock.includes("fetchHourlyForecastForCity(city, { bypassLocalCache: true, resolution: targetResolution })") &&
!fallbackRefreshBlock.includes("ignoreCache: true") &&
!fallbackRefreshBlock.includes("setIsHourlyLoading(true)"),
"no-patch fallback refresh should update the chart through cached batch detail without force-refreshing or showing the loading overlay",
"no-patch fallback refresh should revalidate through cached backend detail without force-refreshing sources or showing the loading overlay",
);
const resyncBlock = chart.match(/useEffect\(\(\) => \{\s*if \(!resyncVersion \|\| !city\) return;[\s\S]*?\}, \[resyncVersion, city, targetResolution, applySuccessfulHourlyDetail\]\);/)?.[0] || "";
assert(
frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts
+9 -3
View File
@@ -1321,6 +1321,7 @@ function mergeRowObservationIntoHourly(
}
type HourlyForecastFetchOptions = {
bypassLocalCache?: boolean;
ignoreCache?: boolean;
resolution?: string;
};
@@ -1561,13 +1562,14 @@ async function fetchHourlyForecastForCity(
const resParam = options.resolution || "10m";
const cacheKey = `${city}:${resParam}`;
const forceRefresh = Boolean(options.ignoreCache);
const bypassLocalCache = forceRefresh || Boolean(options.bypassLocalCache);
if (!forceRefresh) {
if (!bypassLocalCache) {
const cached = readHourlyCacheEntry(cacheKey);
if (cached) {
return cached.data;
}
} else {
} else if (forceRefresh) {
const recentlyRefreshed = readHourlyCacheEntry(cacheKey, {
maxAgeMs: HOURLY_FORCE_REFRESH_DEDUP_MS,
});
@@ -1576,7 +1578,11 @@ async function fetchHourlyForecastForCity(
}
}
const requestKey = options.ignoreCache ? `${city}:${resParam}:live` : `${city}:${resParam}`;
const requestKey = forceRefresh
? `${city}:${resParam}:live`
: bypassLocalCache
? `${city}:${resParam}:revalidate`
: `${city}:${resParam}`;
const pending = _hourlyRequestCache.get(requestKey);
if (pending) return pending;
frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts
+35 -1
View File
@@ -103,6 +103,8 @@ export function useScanTerminalQuery({
} = useRemoteDataQuery<ScanTerminalResponse>();
const lastForcedScanRefreshAtRef = useRef(0);
const lastForegroundScanRefreshAtRef = useRef(0);
const lastScanSuccessAtRef = useRef(0);
const patchVersion = useSsePatchVersion();
const [cachedRows, setCachedRows] = useState<ScanTerminalResponse | null>(() => {
if (typeof window !== "undefined") {
@@ -140,7 +142,11 @@ export function useScanTerminalQuery({
tradingRegion,
}),
showLoading,
onSuccess: (data) => { writeScanCache(data, tradingRegion || ""); setCachedRows(data); },
onSuccess: (data) => {
lastScanSuccessAtRef.current = Date.now();
writeScanCache(data, tradingRegion || "");
setCachedRows(data);
},
});
},
[isPro, proAccessLoading, run, timezoneOffsetSeconds, tradingRegion],
@@ -172,6 +178,34 @@ export function useScanTerminalQuery({
void fetchScanTerminal({ forceRefresh: true, showLoading: true });
}, [fetchScanTerminal, terminalData]);
useEffect(() => {
if (typeof window === "undefined" || typeof document === "undefined") return;
if (proAccessLoading || !isPro) return;
const handleForegroundScanRefresh = () => {
if (document.visibilityState !== "visible") return;
if (scanRemote.status === "loading") return;
const now = Date.now();
if (now - lastForegroundScanRefreshAtRef.current < 30_000) return;
if (
lastScanSuccessAtRef.current > 0 &&
now - lastScanSuccessAtRef.current < SCAN_CACHE_TTL_MS
) {
return;
}
lastForegroundScanRefreshAtRef.current = now;
void fetchScanTerminal({ forceRefresh: false, showLoading: false });
};
document.addEventListener("visibilitychange", handleForegroundScanRefresh);
window.addEventListener("focus", handleForegroundScanRefresh);
return () => {
document.removeEventListener("visibilitychange", handleForegroundScanRefresh);
window.removeEventListener("focus", handleForegroundScanRefresh);
};
}, [fetchScanTerminal, isPro, proAccessLoading, scanRemote.status]);
// Preload adjacent regions in idle time for instant tab switches
useEffect(() => {
if (typeof window === "undefined" || !tradingRegion || !isPro) return;
src/bot/handlers/basic.py
-31
View File
@@ -17,7 +17,6 @@ from src.auth.telegram_group_pricing import TelegramGroupPricing, TELEGRAM_MEMBE
from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env
_BASIC_COMMANDS = {"start", "help", "id", "top", "diag", "bind", "unbind"}
_BASIC_COMMANDS = {"start", "help", "id", "top", "diag", "bind", "unbind", "markets"}
class BasicCommandHandler:
@@ -60,10 +59,6 @@ class BasicCommandHandler:
def _unbind(message):
self._dispatch(message)
@self.bot.message_handler(commands=["markets"])
def _markets(message):
self._dispatch(message)
if hasattr(self.bot, "chat_join_request_handler"):
@self.bot.chat_join_request_handler(func=lambda request: True)
def _chat_join_request(request):
@@ -123,9 +118,6 @@ class BasicCommandHandler:
if command == "unbind":
self.handle_unbind(message)
return
if command == "markets":
self.handle_markets(message)
return
def handle_start_help(self, message: Any) -> None:
trace = CommandTrace("/start", message)
@@ -141,7 +133,6 @@ class BasicCommandHandler:
trace.set_status("ok")
finally:
trace.emit()
@staticmethod
def _is_private_text_fallback(message: Any) -> bool:
text = str(getattr(message, "text", "") or "").strip()
@@ -553,25 +544,3 @@ class BasicCommandHandler:
trace.set_status("ok", "unbound")
finally:
trace.emit()
def handle_markets(self, message: Any) -> None:
trace = CommandTrace("/markets", message)
try:
chat_type = str(getattr(getattr(message, "chat", None), "type", "") or "").strip().lower()
if chat_type and chat_type != "private":
self.bot.reply_to(
message,
"ℹ️ `/markets` 仅支持私聊机器人查询。",
parse_mode="Markdown",
)
trace.set_status("blocked", f"unsupported_chat_type:{chat_type}")
return
self.bot.reply_to(
message,
"ℹ️ 市场概览 (Focus Digest) 功能已移除。\n频道继续接收关键市场警报推送；如需查看当前市场状态，请访问 https://polyweather.top/",
disable_web_page_preview=True,
)
trace.set_status("ok", "removed")
finally:
trace.emit()
src/bot/handlers/city.py
-111
View File
@@ -1,111 +0,0 @@
from __future__ import annotations
from typing import Any
from loguru import logger
from src.bot.command_parser import extract_command_name
from src.bot.command_parser import split_command_and_args
from src.bot.command_guard import CommandGuard
from src.bot.io_layer import BotIOLayer
from src.bot.observability import CommandTrace
from src.bot.services.city_command_service import CityCommandService
from src.bot.settings import CITY_DAILY_FREE_LIMIT, CITY_QUERY_COST
def _is_city_command(message: Any) -> bool:
command = extract_command_name(
getattr(message, "text", None),
getattr(message, "entities", None),
)
return command in {"city", "pwcity"}
class CityCommandHandler:
def __init__(
self,
bot: Any,
guard: CommandGuard,
city_service: CityCommandService,
io_layer: BotIOLayer,
):
self.bot = bot
self.guard = guard
self.city_service = city_service
self.io_layer = io_layer
def register(self) -> None:
@self.bot.message_handler(commands=["city", "pwcity"])
def _city_command(message):
self.handle(message)
@self.bot.message_handler(
func=lambda message: _is_city_command(message),
content_types=["text"],
)
def _city_text(message):
self.handle(message)
def handle(self, message: Any) -> None:
if getattr(message, "_pw_city_handled", False):
return
if not _is_city_command(message):
return
setattr(message, "_pw_city_handled", True)
setattr(message, "_pw_command_handled", True)
trace = CommandTrace("/city", message)
try:
_, args = split_command_and_args(getattr(message, "text", None))
if not args:
trace.set_status("bad_request", "missing_city")
self.io_layer.send_query_message(
message,
"❌ 请输入城市名称\n\n用法: <code>/city chicago</code>",
parse_mode="HTML",
)
return
city_input = args.strip().lower()
resolved = self.city_service.resolve_city(city_input)
if not resolved.ok:
city_list = ", ".join(resolved.supported_cities or [])
trace.set_status("bad_request", "city_not_supported")
self.io_layer.send_query_message(
message,
f"❌ 未找到城市: <b>{city_input}</b>\n\n支持的城市: {city_list}",
parse_mode="HTML",
)
return
city_name = str(resolved.city_name)
if not self.guard.check_daily_query_limit(message, "city", CITY_DAILY_FREE_LIMIT):
trace.set_status("blocked", "daily_query_limit")
return
if not self.guard.ensure_access_and_points(message, CITY_QUERY_COST, "/city"):
trace.set_status("blocked", "guard_rejected")
return
self.io_layer.send_query_message(
message,
f"🔍 正在查询 {city_name.title()} 的天气数据...",
)
report_result = self.city_service.build_report(city_name, CITY_QUERY_COST)
if not report_result.ok:
trace.set_status("failed", report_result.error or "city_report_failed")
self.io_layer.send_query_message(
message,
f"❌ 查询失败: {report_result.error}",
)
return
self.io_layer.send_query_message(
message,
str(report_result.report),
parse_mode="HTML",
)
trace.set_status("ok", city_name)
except Exception as exc:
trace.set_status("failed", "unexpected_error")
logger.exception("查询 /city 失败")
self.io_layer.send_query_message(message, f"❌ 查询失败: {exc}")
finally:
trace.emit()
src/bot/handlers/deb.py
-105
View File
@@ -1,105 +0,0 @@
from __future__ import annotations
from typing import Any
from loguru import logger
from src.bot.command_parser import extract_command_name
from src.bot.command_parser import split_command_and_args
from src.bot.command_guard import CommandGuard
from src.bot.io_layer import BotIOLayer
from src.bot.observability import CommandTrace
from src.bot.services.deb_command_service import DebCommandService
from src.bot.settings import DEB_DAILY_FREE_LIMIT, DEB_QUERY_COST
def _is_deb_command(message: Any) -> bool:
command = extract_command_name(
getattr(message, "text", None),
getattr(message, "entities", None),
)
return command in {"deb", "pwdeb"}
class DebCommandHandler:
def __init__(
self,
bot: Any,
guard: CommandGuard,
deb_service: DebCommandService,
io_layer: BotIOLayer,
):
self.bot = bot
self.guard = guard
self.deb_service = deb_service
self.io_layer = io_layer
def register(self) -> None:
@self.bot.message_handler(commands=["deb", "pwdeb"])
def _deb_command(message):
self.handle(message)
@self.bot.message_handler(
func=lambda message: _is_deb_command(message),
content_types=["text"],
)
def _deb_text(message):
self.handle(message)
def handle(self, message: Any) -> None:
if getattr(message, "_pw_deb_handled", False):
return
if not _is_deb_command(message):
return
setattr(message, "_pw_deb_handled", True)
setattr(message, "_pw_command_handled", True)
trace = CommandTrace("/deb", message)
try:
_, args = split_command_and_args(getattr(message, "text", None))
if not args:
trace.set_status("bad_request", "missing_city")
self.io_layer.send_query_message(
message,
"❌ 用法: <code>/deb ankara</code>",
parse_mode="HTML",
)
return
city_input = args.strip().lower()
city_name = self.deb_service.resolve_city(city_input)
if not self.deb_service.has_history(city_name):
trace.set_status("bad_request", "history_missing")
self.io_layer.send_query_message(
message,
f"❌ 暂无 {city_name} 的历史数据。",
parse_mode="HTML",
)
return
if not self.guard.check_daily_query_limit(message, "deb", DEB_DAILY_FREE_LIMIT):
trace.set_status("blocked", "daily_query_limit")
return
if not self.guard.ensure_access_and_points(message, DEB_QUERY_COST, "/deb"):
trace.set_status("blocked", "guard_rejected")
return
report_result = self.deb_service.build_report(city_name, DEB_QUERY_COST)
if not report_result.ok:
trace.set_status("failed", report_result.error or "deb_report_failed")
self.io_layer.send_query_message(
message,
f"❌ 查询失败: {report_result.error}",
)
return
self.io_layer.send_query_message(
message,
str(report_result.report),
parse_mode="HTML",
)
trace.set_status("ok", city_name)
except Exception as exc:
trace.set_status("failed", "unexpected_error")
logger.exception("查询 /deb 失败")
self.io_layer.send_query_message(message, f"❌ 查询失败: {exc}")
finally:
trace.emit()
src/bot/io_layer.py
+3 -19
View File
@@ -1,14 +1,11 @@
from __future__ import annotations
import os
from datetime import datetime
from typing import Any, Dict, Optional, Tuple
from loguru import logger
from src.bot.settings import (
CITY_DAILY_FREE_LIMIT,
DEB_DAILY_FREE_LIMIT,
GROUP_MESSAGE_POINTS_ENABLED,
MESSAGE_COOLDOWN_SEC,
MESSAGE_DAILY_CAP,
@@ -194,28 +191,20 @@ class BotIOLayer:
def build_welcome_text(self) -> str:
return (
"🚀 <b>PolyWeather 天气查询机器人</b>\n\n"
"🚀 <b>PolyWeather 机器人</b>\n\n"
"可用指令:\n"
f"/city [城市名] 或 /pwcity [城市名] - 查询城市天气预测与实测 (免费, 每日 {CITY_DAILY_FREE_LIMIT} 次)\n"
f"/deb [城市名] 或 /pwdeb [城市名] - 查看 DEB 融合预测准确率 (免费, 每日 {DEB_DAILY_FREE_LIMIT} 次)\n"
"/markets - 私聊机器人查看当前市场监控摘要\n"
"/top - 查看积分排行榜\n"
"/id - 获取当前聊天的 Chat ID\n\n"
"/diag - 查看 Bot 启动诊断\n\n"
"/bind - 绑定 Supabase 账号（可选）\n"
"/unbind - 解除当前 Telegram 与网页账号绑定\n\n"
"🔗 机器人: <a href=\"https://t.me/polyyuanbot\">@polyyuanbot</a>\n"
"👥 社群: <a href=\"https://t.me/+Io5H9oVHFmVjOTQ5\">加入 Telegram 群组</a>\n\n"
"📌 <i>私有频道用于接收自动推送；手动查看市场概览请私聊机器人发送 <code>/markets</code>。</i>\n\n"
"示例: <code>/city 伦敦</code> 或 <code>/pwcity 伦敦</code>\n"
"💡 <i>提示: 积分现在通过邀请制度获得；有效邀请完成首次 Pro 付款后，邀请人获得积分奖励。</i>"
"👥 社群: <a href=\"https://t.me/+Io5H9oVHFmVjOTQ5\">加入 Telegram 群组</a>"
)
def build_points_rank_text(self, user: Any) -> str:
self.db.upsert_user(user.id, self.display_name(user))
user_info = self.db.get_user(user.id)
now = datetime.now()
today_str = now.strftime("%Y-%m-%d")
leaderboard = self.db.get_leaderboard(limit=5)
rank_text = "🏆 <b>PolyWeather 用户积分排行</b>\n"
@@ -227,17 +216,12 @@ class BotIOLayer:
rank_text += f"{medal} {username}: <b>{points}</b> 分\n"
if user_info:
daily_queries_date = str(user_info.get("daily_queries_date") or "")
city_used = int(user_info.get("daily_city_queries") or 0) if daily_queries_date == today_str else 0
deb_used = int(user_info.get("daily_deb_queries") or 0) if daily_queries_date == today_str else 0
rank_text += "────────────────────\n"
rank_text += (
"👤 <b>我的状态：</b>\n"
f"┣ 累计积分: <code>{user_info['points']}</code>\n"
"┣ 积分获取: <code>邀请付费用户</code>\n"
"┣ 抵扣规则: <code>500分 = 1 USDC，单笔最多抵3U</code>\n"
f"┗ /city 免费 ({city_used}/{CITY_DAILY_FREE_LIMIT}) | /deb 免费 ({deb_used}/{DEB_DAILY_FREE_LIMIT})"
"┗ 抵扣规则: <code>500分 = 1 USDC，单笔最多抵3U</code>"
)
return rank_text
src/bot/orchestrator.py
+6 -36
View File
@@ -5,32 +5,21 @@ from typing import Any
from loguru import logger # type: ignore
from src.bot.analysis.city_analysis_service import CityAnalysisService
from src.bot.analysis.deb_analysis_service import DebAnalysisService
from src.bot.command_guard import CommandGuard
from src.bot.handlers.activity import ActivityHandler
from src.bot.handlers.basic import BasicCommandHandler
from src.bot.handlers.city import CityCommandHandler
from src.bot.handlers.deb import DebCommandHandler
from src.bot.io_layer import BotIOLayer
from src.bot.runtime_coordinator import StartupCoordinator
from src.bot.services.city_command_service import CityCommandService
from src.bot.services.deb_command_service import DebCommandService
from src.utils.config_validation import validate_or_raise
from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env
def _project_root() -> str:
return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def _register_handlers(
bot: Any,
config: dict[str, Any],
io_layer: BotIOLayer,
guard: CommandGuard,
city_service: CityCommandService,
deb_service: DebCommandService,
guard: Any | None,
city_service: Any | None,
deb_service: Any | None,
startup_coordinator: StartupCoordinator,
) -> None:
BasicCommandHandler(
@@ -39,25 +28,12 @@ def _register_handlers(
runtime_status_provider=startup_coordinator.get_runtime_status,
config=config,
).register()
CityCommandHandler(
bot=bot,
guard=guard,
city_service=city_service,
io_layer=io_layer,
).register()
DebCommandHandler(
bot=bot,
guard=guard,
deb_service=deb_service,
io_layer=io_layer,
).register()
ActivityHandler(bot=bot, io_layer=io_layer).register()
def start_bot() -> None:
import telebot # type: ignore
from src.data_collection.weather_sources import WeatherDataCollector
from src.database.db_manager import DBManager
from src.utils.config_loader import load_config
@@ -70,14 +46,8 @@ def start_bot() -> None:
bot = telebot.TeleBot(token)
db = DBManager()
weather = WeatherDataCollector(config)
io_layer = BotIOLayer(bot=bot, db=db)
city_analysis = CityAnalysisService(weather=weather)
deb_analysis = DebAnalysisService(project_root=_project_root())
guard = CommandGuard(io_layer=io_layer)
city_service = CityCommandService(analysis=city_analysis)
deb_service = DebCommandService(analysis=deb_analysis)
startup_coordinator = StartupCoordinator(
bot=bot,
config=config,
@@ -90,9 +60,9 @@ def start_bot() -> None:
bot=bot,
config=config,
io_layer=io_layer,
guard=guard,
city_service=city_service,
deb_service=deb_service,
guard=None,
city_service=None,
deb_service=None,
startup_coordinator=startup_coordinator,
)
runtime_status = startup_coordinator.start_all()
src/bot/services/city_command_service.py
-38
View File
@@ -1,38 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Optional
from src.bot.analysis.city_analysis_service import CityAnalysisService
@dataclass
class CityResolveResult:
ok: bool
city_name: Optional[str] = None
supported_cities: List[str] | None = None
@dataclass
class CityReportResult:
ok: bool
report: Optional[str] = None
error: Optional[str] = None
class CityCommandService:
def __init__(self, analysis: CityAnalysisService):
self.analysis = analysis
def resolve_city(self, city_input: str) -> CityResolveResult:
city_name, supported = self.analysis.resolve_city(city_input)
if not city_name:
return CityResolveResult(ok=False, supported_cities=supported)
return CityResolveResult(ok=True, city_name=city_name, supported_cities=supported)
def build_report(self, city_name: str, city_query_cost: int) -> CityReportResult:
try:
report = self.analysis.build_city_report(city_name, city_query_cost)
return CityReportResult(ok=True, report=report)
except Exception as exc:
return CityReportResult(ok=False, error=str(exc))
src/bot/services/deb_command_service.py
-31
View File
@@ -1,31 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
from src.bot.analysis.deb_analysis_service import DebAnalysisService
@dataclass
class DebReportResult:
ok: bool
report: Optional[str] = None
error: Optional[str] = None
class DebCommandService:
def __init__(self, analysis: DebAnalysisService):
self.analysis = analysis
def resolve_city(self, city_input: str) -> str:
return self.analysis.resolve_city(city_input)
def has_history(self, city_name: str) -> bool:
return self.analysis.has_history(city_name)
def build_report(self, city_name: str, deb_query_cost: int) -> DebReportResult:
try:
report = self.analysis.build_deb_accuracy_report(city_name, deb_query_cost)
return DebReportResult(ok=True, report=report)
except Exception as exc:
return DebReportResult(ok=False, error=str(exc))
src/bot/services/entitlement_service.py
+1 -1
View File
@@ -41,7 +41,7 @@ class BotEntitlementService:
"POLYWEATHER_BOT_USE_SUPABASE_ENTITLEMENT",
SUPABASE_ENTITLEMENT.enabled,
)
commands = protected_commands or ("/city", "/deb")
commands = protected_commands or ()
self.protected_commands: Set[str] = {str(c).strip().lower() for c in commands if str(c).strip()}
def check(self, user_id: int, command_label: str) -> EntitlementDecision:
src/bot/settings.py
-4
View File
@@ -31,10 +31,6 @@ MESSAGE_MIN_LENGTH = _env_int("POLYWEATHER_BOT_MESSAGE_MIN_LENGTH", 3, min_value
MESSAGE_COOLDOWN_SEC = _env_int("POLYWEATHER_BOT_MESSAGE_COOLDOWN_SEC", 30, min_value=0)
# Optional per-chat override map, parsed in BotIOLayer:
# POLYWEATHER_BOT_MESSAGE_COOLDOWN_BY_CHAT="-1003586303099:10,-1003539418691:20"
CITY_QUERY_COST = _env_int("POLYWEATHER_BOT_CITY_QUERY_COST", 0, min_value=0)
DEB_QUERY_COST = _env_int("POLYWEATHER_BOT_DEB_QUERY_COST", 0, min_value=0)
CITY_DAILY_FREE_LIMIT = _env_int("POLYWEATHER_BOT_CITY_DAILY_FREE_LIMIT", 10, min_value=1)
DEB_DAILY_FREE_LIMIT = _env_int("POLYWEATHER_BOT_DEB_DAILY_FREE_LIMIT", 10, min_value=1)
FIRST_MESSAGE_BONUS = _env_int("POLYWEATHER_BOT_FIRST_MESSAGE_BONUS", 2, min_value=0)
WELCOME_BONUS = _env_int("POLYWEATHER_BOT_WELCOME_BONUS", 20, min_value=0)
tests/test_bot_basic_handler.py
+16 -44
View File
@@ -79,7 +79,7 @@ def test_basic_handler_diag_returns_html():
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
)
bot = DummyBot()
@@ -131,7 +131,7 @@ def test_start_bind_token_binds_telegram_to_web_account():
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
),
)
@@ -173,7 +173,7 @@ def test_confirm_bind_callback_consumes_token_and_binds_account():
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
),
)
@@ -204,7 +204,7 @@ def test_private_text_fallback_replies_with_binding_help():
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
),
)
@@ -230,7 +230,7 @@ def test_private_text_fallback_ignores_slash_commands():
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
),
)
@@ -241,7 +241,7 @@ def test_private_text_fallback_ignores_slash_commands():
assert bot.replies == []
def test_basic_handler_markets_returns_summary():
def test_basic_handler_ignores_removed_markets_command():
bot = DummyBot()
io_layer = SimpleNamespace(
build_welcome_text=lambda: "WELCOME",
@@ -254,44 +254,16 @@ def test_basic_handler_markets_returns_summary():
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
),
config={},
)
handler.handle_markets(_message("/markets"))
handler._dispatch(_message("/markets"))
assert len(bot.replies) == 1
assert "市场概览" in bot.replies[0]["text"]
assert "已移除" in bot.replies[0]["text"]
def test_basic_handler_markets_rejects_channel_chat():
bot = DummyBot()
io_layer = SimpleNamespace(
build_welcome_text=lambda: "WELCOME",
build_points_rank_text=lambda _user: "TOP",
)
handler = BasicCommandHandler(
bot=bot,
io_layer=io_layer,
runtime_status_provider=lambda: RuntimeStatus(
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
required_group_chat_id="-1001234567890",
),
config={},
)
msg = _message("/markets")
msg.chat = SimpleNamespace(id=-1001, type="channel")
handler.handle_markets(msg)
assert len(bot.replies) == 1
assert "仅支持私聊机器人查询" in bot.replies[0]["text"]
assert bot.replies == []
assert bot.sent_messages == []
def _join_request(user_id: int = 12345, chat_id: int = -100123):
@@ -324,7 +296,7 @@ def test_join_request_auto_approves_bound_active_pro_user(monkeypatch):
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-100123",
),
entitlement_service=entitlement,
@@ -354,7 +326,7 @@ def test_join_request_keeps_unbound_user_pending_by_default(monkeypatch):
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-100123",
),
entitlement_service=entitlement,
@@ -389,7 +361,7 @@ def test_join_request_keeps_trial_user_pending(monkeypatch):
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-100123",
),
entitlement_service=entitlement,
@@ -429,7 +401,7 @@ def test_join_request_approves_trial_user_with_queued_paid_subscription(monkeypa
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-100123",
),
entitlement_service=entitlement,
@@ -469,7 +441,7 @@ def test_join_request_uses_subscription_window_without_latest_fallback(monkeypat
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-100123",
),
entitlement_service=entitlement,
@@ -499,7 +471,7 @@ def test_join_request_can_decline_ineligible_user_when_configured(monkeypatch):
started_at="2026-03-12 00:00:00 UTC",
loops=[],
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-100123",
),
entitlement_service=entitlement,
tests/test_bot_command_guard.py
+14 -1
View File
@@ -2,6 +2,7 @@ from types import SimpleNamespace
from unittest.mock import Mock
from src.bot.command_guard import CommandGuard
from src.bot.services.entitlement_service import BotEntitlementService
def _message():
@@ -19,8 +20,20 @@ def test_guard_charges_points():
io_layer = SimpleNamespace(bot=fake_bot, ensure_query_points=Mock(return_value=True))
guard = CommandGuard(io_layer=io_layer, group_chat_id="-100123")
ok = guard.ensure_access_and_points(_message(), 1, "/city")
ok = guard.ensure_access_and_points(_message(), 1, "/top")
assert ok is True
assert io_layer.ensure_query_points.call_count == 1
assert fake_bot.reply_to.call_count == 0
def test_entitlement_service_has_no_removed_protected_commands_by_default(monkeypatch):
monkeypatch.setenv("POLYWEATHER_BOT_USE_SUPABASE_ENTITLEMENT", "false")
db = SimpleNamespace(get_user=lambda _user_id: {})
service = BotEntitlementService(db=db, enabled=True)
decision = service.check(123, "/city")
assert service.protected_commands == set()
assert decision.allowed is True
assert decision.reason == "command_not_protected"
tests/test_bot_handlers.py
+55 -86
View File
@@ -1,98 +1,67 @@
from types import SimpleNamespace
from unittest.mock import Mock
from src.bot.handlers.city import CityCommandHandler
from src.bot.handlers.deb import DebCommandHandler
from src.bot.services.city_command_service import CityReportResult, CityResolveResult
from src.bot.services.deb_command_service import DebReportResult
from src.bot.orchestrator import _register_handlers
from src.bot.runtime_coordinator import RuntimeStatus
def _message(text: str):
return SimpleNamespace(
text=text,
from_user=SimpleNamespace(id=123, username="tester", first_name="Tester"),
chat=SimpleNamespace(id=999),
entities=[],
class RecordingBot:
def __init__(self):
self.command_handlers = []
def message_handler(self, *args, **kwargs):
commands = kwargs.get("commands")
if commands:
self.command_handlers.extend(commands)
def _decorator(func):
return func
return _decorator
def chat_join_request_handler(self, *args, **kwargs):
def _decorator(func):
return func
return _decorator
def callback_query_handler(self, *args, **kwargs):
def _decorator(func):
return func
return _decorator
def test_register_handlers_does_not_expose_removed_query_commands():
bot = RecordingBot()
io_layer = SimpleNamespace(
build_welcome_text=Mock(return_value="WELCOME"),
build_points_rank_text=Mock(return_value="TOP"),
track_group_text_activity=Mock(),
)
def test_city_handler_missing_city_shows_usage():
bot = SimpleNamespace()
guard = SimpleNamespace(
check_daily_query_limit=Mock(return_value=True),
ensure_access_and_points=Mock(return_value=True),
)
city_service = SimpleNamespace(resolve_city=Mock(), build_report=Mock())
io_layer = SimpleNamespace(send_query_message=Mock())
handler = CityCommandHandler(
bot=bot,
guard=guard,
city_service=city_service,
io_layer=io_layer,
)
handler.handle(_message("/city"))
assert io_layer.send_query_message.call_count == 1
assert "请输入城市名称" in io_layer.send_query_message.call_args[0][1]
assert guard.ensure_access_and_points.call_count == 0
def test_city_handler_happy_path_pushes_progress_and_report():
bot = SimpleNamespace()
guard = SimpleNamespace(
check_daily_query_limit=Mock(return_value=True),
ensure_access_and_points=Mock(return_value=True),
)
city_service = SimpleNamespace(
resolve_city=Mock(
return_value=CityResolveResult(
ok=True,
city_name="london",
supported_cities=["london"],
startup_coordinator = SimpleNamespace(
get_runtime_status=Mock(
return_value=RuntimeStatus(
started_at="2026-06-13 00:00:00 UTC",
loops=[],
command_access_mode="public",
protected_commands=[],
required_group_chat_id="",
)
),
build_report=Mock(return_value=CityReportResult(ok=True, report="CITY_REPORT")),
)
)
io_layer = SimpleNamespace(send_query_message=Mock())
handler = CityCommandHandler(
_register_handlers(
bot=bot,
guard=guard,
city_service=city_service,
config={},
io_layer=io_layer,
guard=SimpleNamespace(),
city_service=SimpleNamespace(),
deb_service=SimpleNamespace(),
startup_coordinator=startup_coordinator,
)
handler.handle(_message("/city london"))
assert io_layer.send_query_message.call_count == 2
first_call = io_layer.send_query_message.call_args_list[0]
second_call = io_layer.send_query_message.call_args_list[1]
assert "正在查询 London 的天气数据" in first_call.args[1]
assert second_call.args[1] == "CITY_REPORT"
assert second_call.kwargs["parse_mode"] == "HTML"
def test_deb_handler_history_missing_returns_hint():
bot = SimpleNamespace()
guard = SimpleNamespace(
check_daily_query_limit=Mock(return_value=True),
ensure_access_and_points=Mock(return_value=True),
)
deb_service = SimpleNamespace(
resolve_city=Mock(return_value="ankara"),
has_history=Mock(return_value=False),
build_report=Mock(return_value=DebReportResult(ok=True, report="DEB_REPORT")),
)
io_layer = SimpleNamespace(send_query_message=Mock())
handler = DebCommandHandler(
bot=bot,
guard=guard,
deb_service=deb_service,
io_layer=io_layer,
)
handler.handle(_message("/deb ankara"))
assert io_layer.send_query_message.call_count == 1
assert "暂无 ankara 的历史数据" in io_layer.send_query_message.call_args[0][1]
assert guard.ensure_access_and_points.call_count == 0
assert "city" not in bot.command_handlers
assert "pwcity" not in bot.command_handlers
assert "deb" not in bot.command_handlers
assert "pwdeb" not in bot.command_handlers
assert "markets" not in bot.command_handlers
tests/test_bot_io_layer_points.py
+42
View File
@@ -9,14 +9,30 @@ class DummyDB:
def __init__(self):
self.upserts = []
self.activities = []
self.users = {}
def upsert_user(self, telegram_id, username):
self.upserts.append((telegram_id, username))
self.users.setdefault(
telegram_id,
{
"points": 0,
"daily_queries_date": "",
"daily_city_queries": 0,
"daily_deb_queries": 0,
},
)
def add_message_activity(self, telegram_id, **kwargs):
self.activities.append({"telegram_id": telegram_id, **kwargs})
return {"awarded": True, "points_added": kwargs.get("points_to_add", 0)}
def get_user(self, telegram_id):
return self.users.get(telegram_id)
def get_leaderboard(self, limit=5):
return []
def _message(chat_id: int | str, text: str = "有效发言"):
return SimpleNamespace(
@@ -51,3 +67,29 @@ def test_group_message_points_skip_unconfigured_chat_when_allowlist_exists(monke
assert db.upserts == []
assert db.activities == []
def test_welcome_text_hides_removed_query_commands():
io_layer = BotIOLayer(bot=SimpleNamespace(), db=DummyDB())
text = io_layer.build_welcome_text()
assert "/city" not in text
assert "/pwcity" not in text
assert "/deb" not in text
assert "/pwdeb" not in text
assert "/markets" not in text
assert "私有频道" not in text
assert "示例" not in text
assert "积分现在通过邀请制度" not in text
def test_points_rank_text_hides_removed_query_usage():
db = DummyDB()
io_layer = BotIOLayer(bot=SimpleNamespace(), db=db)
user = SimpleNamespace(id=123, username="alice", first_name="Alice")
text = io_layer.build_points_rank_text(user)
assert "/city" not in text
assert "/deb" not in text
tests/test_bot_runtime_coordinator.py
+3 -3
View File
@@ -12,7 +12,7 @@ def test_startup_coordinator_respects_disable_flags(monkeypatch):
bot=DummyBot(),
config={},
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
)
runtime = coordinator.start_all()
@@ -27,7 +27,7 @@ def test_render_runtime_status_html_contains_key_fields():
runtime = RuntimeStatus(
started_at="2026-03-12 00:00:00 UTC",
command_access_mode="group_member",
protected_commands=["/city", "/deb"],
protected_commands=[],
required_group_chat_id="-1001234567890",
loops=[],
)
@@ -35,4 +35,4 @@ def test_render_runtime_status_html_contains_key_fields():
assert "Bot 启动诊断" in html
assert "命令准入" in html
assert "/city, /deb" in html
assert "受保护命令: <code>--</code>" in html
tests/test_deployment_runtime_config.py
+1
View File
@@ -100,6 +100,7 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services():
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true'" in collector_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-3}" in web_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-2}" in web_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS:-3000}" in web_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-8000}" in web_block
assert "UVICORN_WORKERS: ${UVICORN_WORKERS:-2}" in web_block
assert "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''" in bot_block
tests/test_web_observability.py
+55 -6
View File
@@ -925,9 +925,11 @@ def test_chart_scope_overlays_collector_runway_history_from_db(monkeypatch):
assert history[-1] == {"time": "2026-06-06T05:28:00+00:00", "temp": 24.8}
def test_chart_data_cache_hit_does_not_start_full_stale_refresh(monkeypatch):
def test_chart_data_cache_hit_starts_full_stale_refresh(monkeypatch):
import asyncio
refresh_calls = []
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
@@ -951,7 +953,7 @@ def test_chart_data_cache_hit_does_not_start_full_stale_refresh(monkeypatch):
monkeypatch.setattr(
city_api,
"_start_city_full_stale_refresh",
lambda city: (_ for _ in ()).throw(AssertionError("chart scope must not start full stale refresh")),
refresh_calls.append,
)
monkeypatch.setattr(
city_api.legacy_routes,
@@ -962,6 +964,7 @@ def test_chart_data_cache_hit_does_not_start_full_stale_refresh(monkeypatch):
payload = asyncio.run(city_api._get_city_chart_data("paris", force_refresh=False))
assert payload["hourly"]["temps"] == [25.0]
assert refresh_calls == ["paris"]
def test_chart_data_returns_cached_payload_when_optional_overlay_times_out(monkeypatch):
@@ -1058,12 +1061,57 @@ def test_city_detail_batch_partial_timeout_default_stays_below_proxy_budget(monk
monkeypatch.delenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", raising=False)
monkeypatch.delenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", raising=False)
monkeypatch.delenv("POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY", raising=False)
monkeypatch.delenv("POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS", raising=False)
assert city_api._city_detail_batch_concurrency() == 3
assert city_api._city_detail_batch_global_concurrency() == 2
assert city_api._city_detail_batch_queue_wait_seconds() == 3.0
assert city_api._city_detail_batch_partial_timeout_seconds() == 8.0
def test_city_detail_batch_waits_briefly_for_global_builder_slot(monkeypatch):
import asyncio
import threading
build_calls = 0
async def build_batch_item(city, **kwargs):
nonlocal build_calls
build_calls += 1
return city, {"city": city}
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY", "1")
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS", "200")
monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE", None)
monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE", 0)
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api, "_build_city_detail_batch_item_async", build_batch_item)
semaphore = city_api._city_detail_batch_build_semaphore()
assert semaphore.acquire(blocking=False) is True
release_timer = threading.Timer(0.02, semaphore.release)
release_timer.start()
try:
payload = asyncio.run(
city_api.get_city_detail_batch_payload(
object(),
cities="Wait-Paris,Wait-Shanghai",
resolution="10m",
limit=2,
)
)
finally:
release_timer.join(timeout=1)
assert payload["partial"] is False
assert payload.get("busy") is not True
assert sorted(payload["details"]) == ["wait-paris", "wait-shanghai"]
assert payload["missing"] == []
assert payload["diagnostics"]["response_source"] == "fresh_build"
assert build_calls == 2
def test_city_detail_batch_returns_busy_when_global_builder_slot_is_full(monkeypatch):
import asyncio
@@ -1075,6 +1123,7 @@ def test_city_detail_batch_returns_busy_when_global_builder_slot_is_full(monkeyp
return city, {"city": city}
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY", "1")
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS", "10")
monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE", None)
monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE", 0)
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
@@ -1087,7 +1136,7 @@ def test_city_detail_batch_returns_busy_when_global_builder_slot_is_full(monkeyp
payload = asyncio.run(
city_api.get_city_detail_batch_payload(
object(),
cities="Paris,Shanghai",
cities="Busy-Paris,Busy-Shanghai",
resolution="10m",
limit=2,
)
@@ -1098,14 +1147,14 @@ def test_city_detail_batch_returns_busy_when_global_builder_slot_is_full(monkeyp
assert payload["partial"] is True
assert payload["busy"] is True
assert payload["details"] == {}
assert payload["missing"] == ["paris", "shanghai"]
assert payload["missing"] == ["busy-paris", "busy-shanghai"]
assert payload["diagnostics"]["partial_reason"] == "busy"
assert payload["diagnostics"]["response_source"] == "busy"
assert payload["diagnostics"]["requested_count"] == 2
assert payload["diagnostics"]["completed_count"] == 0
assert payload["diagnostics"]["missing_count"] == 2
assert payload["diagnostics"]["city_status"]["paris"]["status"] == "busy"
assert payload["diagnostics"]["city_status"]["shanghai"]["status"] == "busy"
assert payload["diagnostics"]["city_status"]["busy-paris"]["status"] == "busy"
assert payload["diagnostics"]["city_status"]["busy-shanghai"]["status"] == "busy"
assert build_calls == 0
web/services/city_api.py
+23 -1
View File
@@ -397,6 +397,8 @@ async def _get_city_chart_data(city: str, *, force_refresh: bool) -> Dict[str, A
if cached_entry:
payload = cached_entry.get("payload") or {}
if payload:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
_start_city_full_stale_refresh(city)
payload = await _run_optional_city_chart_overlay(
city=city,
overlay_name="runway_history",
@@ -999,6 +1001,17 @@ def _city_detail_batch_global_concurrency() -> int:
return max(1, min(4, value))
def _city_detail_batch_queue_wait_seconds() -> float:
try:
wait_ms = int(
os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS", "3000")
or "3000"
)
except ValueError:
wait_ms = 3000
return max(0.0, min(5.0, wait_ms / 1000.0))
def _city_detail_batch_build_semaphore() -> threading.BoundedSemaphore:
global _CITY_DETAIL_BATCH_BUILD_SEMAPHORE, _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE
size = _city_detail_batch_global_concurrency()
@@ -1146,7 +1159,16 @@ async def get_city_detail_batch_payload(
async def _build_uncached_payload() -> Dict[str, Any]:
build_semaphore = _city_detail_batch_build_semaphore()
if not build_semaphore.acquire(blocking=False):
queue_wait_seconds = _city_detail_batch_queue_wait_seconds()
acquired = await timer.measure_async(
"wait_builder_slot",
lambda: run_in_threadpool(
build_semaphore.acquire,
True,
queue_wait_seconds,
),
)
if not acquired:
missing = list(city_names)
errors: Dict[str, str] = {}
details: Dict[str, Any] = {}