diff --git a/frontend/app/api/scan/terminal/ai-city/stream/route.ts b/frontend/app/api/scan/terminal/ai-city/stream/route.ts new file mode 100644 index 00000000..76391b9c --- /dev/null +++ b/frontend/app/api/scan/terminal/ai-city/stream/route.ts @@ -0,0 +1,70 @@ +import { NextRequest, NextResponse } from "next/server"; +import { + applyAuthResponseCookies, + buildBackendRequestHeaders, +} from "@/lib/backend-auth"; + +const API_BASE = process.env.POLYWEATHER_API_BASE_URL; + +export const dynamic = "force-dynamic"; +export const maxDuration = 90; + +export async function POST(req: NextRequest) { + if (!API_BASE) { + return NextResponse.json( + { error: "POLYWEATHER_API_BASE_URL is not configured" }, + { status: 500 }, + ); + } + + let body: unknown = {}; + try { + body = await req.json(); + } catch { + body = {}; + } + + const requestBody = + body && typeof body === "object" ? (body as Record) : {}; + const auth = await buildBackendRequestHeaders(req); + const headers = new Headers(auth.headers); + headers.set("Content-Type", "application/json"); + headers.set("Accept", "text/event-stream"); + + try { + const res = await fetch(`${API_BASE}/api/scan/terminal/ai-city/stream`, { + method: "POST", + headers, + cache: "no-store", + body: JSON.stringify(requestBody), + }); + if (!res.ok || !res.body) { + const raw = await res.text(); + const response = NextResponse.json( + { error: `Backend returned ${res.status}`, detail: raw.slice(0, 300) }, + { status: res.status === 402 || res.status === 403 ? res.status : 502 }, + ); + return applyAuthResponseCookies(response, auth.response); + } + + const response = new NextResponse(res.body, { + status: 200, + headers: { + "Content-Type": "text/event-stream; charset=utf-8", + "Cache-Control": "no-store, no-transform", + "X-Accel-Buffering": "no", + }, + }); + return applyAuthResponseCookies(response, auth.response); + } catch (error) { + const response = NextResponse.json( + { + error: "Failed to stream city AI data", + detail: String(error), + city: requestBody.city, + }, + { status: 500 }, + ); + return applyAuthResponseCookies(response, auth.response); + } +} diff --git a/frontend/components/dashboard/ScanTerminalDashboard.tsx b/frontend/components/dashboard/ScanTerminalDashboard.tsx index af66e665..c75146c7 100644 --- a/frontend/components/dashboard/ScanTerminalDashboard.tsx +++ b/frontend/components/dashboard/ScanTerminalDashboard.tsx @@ -91,22 +91,155 @@ type AiCityForecastState = { status: "idle" | "loading" | "ready" | "failed"; payload?: AiCityForecastPayload | null; error?: string | null; + streamText?: string | null; + streamRaw?: string | null; }; -let aiCityFetchQueue: Promise = Promise.resolve(); +const AI_CITY_MAX_CONCURRENT = 2; +let aiCityActiveCount = 0; +const aiCityPendingQueue: Array<() => void> = []; + +function createAiCityAbortError() { + return new DOMException("The AI city request was aborted.", "AbortError"); +} + +function drainAiCityFetchQueue() { + while (aiCityActiveCount < AI_CITY_MAX_CONCURRENT && aiCityPendingQueue.length) { + const next = aiCityPendingQueue.shift(); + next?.(); + } +} function enqueueAiCityFetch( task: () => Promise, signal: AbortSignal, + callbacks?: { + onQueued?: () => void; + onStart?: () => void; + }, ): Promise { - const run = aiCityFetchQueue.then(async () => { - if (signal.aborted) { - throw new DOMException("The AI city request was aborted.", "AbortError"); + return new Promise((resolve, reject) => { + let started = false; + let queuedStart: (() => void) | null = null; + + const cleanup = () => { + signal.removeEventListener("abort", handleAbort); + }; + const removeQueuedStart = () => { + if (!queuedStart) return; + const index = aiCityPendingQueue.indexOf(queuedStart); + if (index >= 0) { + aiCityPendingQueue.splice(index, 1); + } + queuedStart = null; + }; + const finishActive = () => { + aiCityActiveCount = Math.max(0, aiCityActiveCount - 1); + cleanup(); + drainAiCityFetchQueue(); + }; + const handleAbort = () => { + if (started) return; + removeQueuedStart(); + cleanup(); + reject(createAiCityAbortError()); + drainAiCityFetchQueue(); + }; + const start = () => { + queuedStart = null; + if (signal.aborted) { + cleanup(); + reject(createAiCityAbortError()); + drainAiCityFetchQueue(); + return; + } + started = true; + aiCityActiveCount += 1; + callbacks?.onStart?.(); + task() + .then(resolve, reject) + .finally(finishActive); + }; + + signal.addEventListener("abort", handleAbort, { once: true }); + queuedStart = start; + if (aiCityActiveCount < AI_CITY_MAX_CONCURRENT) { + start(); + } else { + callbacks?.onQueued?.(); + aiCityPendingQueue.push(start); } - return task(); }); - aiCityFetchQueue = run.catch(() => undefined); - return run; +} + +function parseSseBlock(block: string): { event: string; data: unknown } | null { + const lines = block.split(/\r?\n/); + let event = "message"; + const dataLines: string[] = []; + for (const line of lines) { + if (line.startsWith("event:")) { + event = line.slice("event:".length).trim() || "message"; + } else if (line.startsWith("data:")) { + dataLines.push(line.slice("data:".length).trimStart()); + } + } + if (!dataLines.length) return null; + const raw = dataLines.join("\n"); + try { + return { event, data: JSON.parse(raw) }; + } catch { + return { event, data: raw }; + } +} + +function decodeJsonStringFragment(fragment: string) { + const safe = fragment.replace(/\\$/g, ""); + try { + return JSON.parse(`"${safe.replace(/"/g, '\\"')}"`) as string; + } catch { + return safe + .replace(/\\"/g, '"') + .replace(/\\n/g, "\n") + .replace(/\\r/g, "\r") + .replace(/\\t/g, "\t") + .replace(/\\\\/g, "\\"); + } +} + +function extractStreamingJsonField(raw: string, field: string) { + const keyIndex = raw.indexOf(`"${field}"`); + if (keyIndex < 0) return ""; + const colonIndex = raw.indexOf(":", keyIndex); + if (colonIndex < 0) return ""; + const quoteIndex = raw.indexOf('"', colonIndex + 1); + if (quoteIndex < 0) return ""; + let end = raw.length; + let escaped = false; + for (let i = quoteIndex + 1; i < raw.length; i += 1) { + const char = raw[i]; + if (escaped) { + escaped = false; + continue; + } + if (char === "\\") { + escaped = true; + continue; + } + if (char === '"') { + end = i; + break; + } + } + return decodeJsonStringFragment(raw.slice(quoteIndex + 1, end)).trim(); +} + +function extractStreamingAirportRead(raw: string, locale: string) { + const primaryField = locale === "en-US" ? "metar_read_en" : "metar_read_zh"; + const fallbackField = locale === "en-US" ? "metar_read_zh" : "metar_read_en"; + return ( + extractStreamingJsonField(raw, primaryField) || + extractStreamingJsonField(raw, fallbackField) + ); } function formatShortDate(value?: string | null, locale = "zh-CN") { @@ -684,13 +817,13 @@ function AiPinnedCityCard({ if (!aiForecastKey) return; let cancelled = false; const controller = new AbortController(); - setAiForecast({ status: "loading" }); + setAiForecast({ status: "loading", streamText: null, streamRaw: "" }); enqueueAiCityFetch( () => - fetch("/api/scan/terminal/ai-city", { + fetch("/api/scan/terminal/ai-city/stream", { method: "POST", headers: { - Accept: "application/json", + Accept: "text/event-stream", "Content-Type": "application/json", }, cache: "no-store", @@ -698,18 +831,18 @@ function AiPinnedCityCard({ body: JSON.stringify({ city: detailCityName, force_refresh: aiRefreshToken > 0, - locale: "zh-CN", + locale, }), }).then(async (response) => { if (!response.ok) { - let detail = ""; + let detailMessage = ""; try { const errorPayload = await response.json(); const message = String(errorPayload?.error || "").trim(); const rawDetail = String(errorPayload?.detail || "").trim(); const elapsed = Number(errorPayload?.elapsed_ms); const timeout = Number(errorPayload?.timeout_ms); - detail = [ + detailMessage = [ message, rawDetail, Number.isFinite(elapsed) && Number.isFinite(timeout) @@ -719,13 +852,108 @@ function AiPinnedCityCard({ .filter(Boolean) .join(" · "); } catch { - detail = ""; + detailMessage = ""; } - throw new Error(detail ? `HTTP ${response.status} · ${detail}` : `HTTP ${response.status}`); + throw new Error(detailMessage ? `HTTP ${response.status} · ${detailMessage}` : `HTTP ${response.status}`); } - return response.json() as Promise; + const contentType = response.headers.get("content-type") || ""; + if (!response.body || !contentType.includes("text/event-stream")) { + return response.json() as Promise; + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let rawStream = ""; + let finalPayload: AiCityForecastPayload | null = null; + const handleBlock = (block: string) => { + const message = parseSseBlock(block); + if (!message || !message.data || typeof message.data !== "object") { + return; + } + const data = message.data as Record; + if (message.event === "progress") { + const progressText = + String(locale === "en-US" ? data.message_en || "" : data.message_zh || "").trim() || + String(data.message || "").trim(); + if (progressText && !cancelled) { + setAiForecast((current) => + current.status === "loading" + ? { ...current, streamText: current.streamText || progressText } + : current, + ); + } + } else if (message.event === "delta") { + const content = String(data.content || ""); + if (!content) return; + rawStream += content; + const airportRead = extractStreamingAirportRead(rawStream, locale); + if (!cancelled) { + setAiForecast((current) => + current.status === "loading" + ? { + ...current, + streamRaw: rawStream, + streamText: airportRead || current.streamText || null, + } + : current, + ); + } + } else if (message.event === "final") { + finalPayload = data as AiCityForecastPayload; + } + }; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const blocks = buffer.split(/\n\n|\r\n\r\n/); + buffer = blocks.pop() || ""; + for (const block of blocks) { + handleBlock(block); + } + } + buffer += decoder.decode(); + if (buffer.trim()) { + handleBlock(buffer); + } + if (!finalPayload) { + throw new Error("AI stream ended before final payload"); + } + return finalPayload; }), controller.signal, + { + onQueued: () => { + if (cancelled) return; + setAiForecast((current) => + current.status === "loading" + ? { + ...current, + streamText: isEn + ? "Waiting for the AI airport read queue..." + : "正在等待 AI 机场报文解读队列...", + } + : current, + ); + }, + onStart: () => { + if (cancelled) return; + setAiForecast((current) => + current.status === "loading" + ? { + ...current, + streamText: current.streamRaw + ? current.streamText + : isEn + ? "Connecting to DeepSeek V4-Pro for airport bulletin streaming..." + : "正在连接 DeepSeek V4-Pro，准备流式解读机场报文...", + } + : current, + ); + }, + }, ) .then((payload) => { if (!cancelled) { @@ -742,7 +970,7 @@ function AiPinnedCityCard({ cancelled = true; controller.abort(); }; - }, [aiForecastKey, aiRefreshToken, detailCityName]); + }, [aiForecastKey, aiRefreshToken, detailCityName, locale]); const aiCityForecast = aiForecast.payload?.city_forecast || null; const localizedFinalJudgment = @@ -911,11 +1139,21 @@ function AiPinnedCityCard({ {isEn ? "AI airport weather read" : "AI 机场报文解读"} {aiForecast.status === "loading" ? ( -

- {isEn - ? "Deepseek V4 pro is reading the latest airport bulletin..." - : "Deepseek V4 pro 正在解读最新机场报文..."} -

+ <> +

+ {aiForecast.streamText || + (isEn + ? "Deepseek V4 pro is reading the latest airport bulletin..." + : "Deepseek V4 pro 正在解读最新机场报文...")} +

+ {aiForecast.streamText ? ( +

+ {isEn + ? "Streaming airport read; final forecast is still being completed..." + : "机场报文解读正在流式输出，最终预报结论仍在补全..."} +

+ ) : null} + ) : aiForecast.status === "ready" && aiCityForecast ? ( <>

diff --git a/web/routes.py b/web/routes.py index 59c4345f..1317f27e 100644 --- a/web/routes.py +++ b/web/routes.py @@ -7,7 +7,7 @@ from typing import Optional from fastapi.concurrency import run_in_threadpool from fastapi import APIRouter, BackgroundTasks, HTTPException, Request -from fastapi.responses import PlainTextResponse +from fastapi.responses import PlainTextResponse, StreamingResponse from loguru import logger from src.analysis.deb_algorithm import load_history @@ -36,6 +36,7 @@ from web.scan_terminal_service import ( build_scan_city_ai_forecast_payload, build_scan_terminal_ai_payload, build_scan_terminal_payload, + stream_scan_city_ai_forecast_payload, ) from web.core import ( AnalyticsEventRequest, @@ -1777,3 +1778,36 @@ async def scan_terminal_ai_city(request: Request): locale=locale, ) + +@router.post("/api/scan/terminal/ai-city/stream") +async def scan_terminal_ai_city_stream(request: Request): + _assert_entitlement(request) + try: + body = await request.json() + except Exception: + body = {} + if not isinstance(body, dict): + raise HTTPException(status_code=400, detail="Invalid JSON body") + city = str(body.get("city") or "").strip() + if not city: + raise HTTPException(status_code=400, detail="city is required") + force_refresh = str(body.get("force_refresh") or "false").lower() in { + "1", + "true", + "yes", + "on", + } + locale = str(body.get("locale") or "zh-CN").strip() + return StreamingResponse( + stream_scan_city_ai_forecast_payload( + city, + force_refresh=force_refresh, + locale=locale, + ), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-store", + "X-Accel-Buffering": "no", + }, + ) + diff --git a/web/scan_terminal_service.py b/web/scan_terminal_service.py index d82a3f59..65491ff5 100644 --- a/web/scan_terminal_service.py +++ b/web/scan_terminal_service.py @@ -9,7 +9,7 @@ import hashlib from concurrent.futures import TimeoutError as FutureTimeoutError from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Iterator, List, Optional import httpx from loguru import logger @@ -385,6 +385,18 @@ def _extract_provider_content(data: Any) -> str: return str(message.get("content") or "") +def _extract_provider_stream_delta(data: Any) -> str: + if not isinstance(data, dict): + return "" + choices = data.get("choices") or [] + if not choices or not isinstance(choices[0], dict): + return "" + delta = choices[0].get("delta") or {} + if not isinstance(delta, dict): + return "" + return str(delta.get("content") or "") + + def _provider_response_meta(data: Any) -> Dict[str, Any]: if not isinstance(data, dict): return {} @@ -1492,6 +1504,355 @@ def _scan_city_ai_cache_key(ai_input: Dict[str, Any]) -> str: return "city-ai:" + hashlib.sha256(raw.encode("utf-8")).hexdigest() +def _city_forecast_cache_key(city_name: str) -> str: + return f"city_forecast:{SCAN_CITY_AI_PROMPT_VERSION}:{city_name.lower()}" + + +def _sse_event(event: str, payload: Dict[str, Any]) -> str: + return ( + f"event: {event}\n" + f"data: {json.dumps(payload, ensure_ascii=False, default=str)}\n\n" + ) + + +def _build_city_ai_stream_request( + ai_input: Dict[str, Any], + *, + locale: str, +) -> Dict[str, Any]: + normalized_locale = _normalize_locale(locale) + system_prompt = ( + "你是 PolyWeather 的城市最高温与机场 METAR 解读员。" + "只返回一个紧凑 JSON object，不要 Markdown。" + "必须先写 metar_read_zh 和 metar_read_en 字段，便于前端流式显示机场报文解读；" + "然后写 final_judgment_zh/final_judgment_en、predicted_max、range_low、range_high、unit、confidence、" + "reasoning_zh/reasoning_en、risks_zh/risks_en、model_cluster_note_zh/model_cluster_note_en。" + "METAR 解读必须具体说明报文时间、温度、风向风速、云量/天气/能见度/露点中与温度路径相关的因素；" + "涉及风时要说明当前风向对机场最高温路径倾向增温、降温还是中性，并给出理由。" + "所有 *_zh 字段写简体中文，所有 *_en 字段写英文，不得留空。" + "不要写交易建议、BUY/SELL、Kelly 或套利。" + ) + return { + "model": SCAN_AI_MODEL, + "temperature": 0.2, + "max_tokens": SCAN_CITY_AI_MAX_TOKENS, + "response_format": {"type": "json_object"}, + "stream": True, + "messages": [ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": json.dumps( + { + "locale": normalized_locale, + "task": ( + "Return JSON keys in this exact order: metar_read_zh, metar_read_en, " + "final_judgment_zh, final_judgment_en, predicted_max, range_low, range_high, " + "unit, confidence, reasoning_zh, reasoning_en, risks_zh, risks_en, " + "model_cluster_note_zh, model_cluster_note_en. Keep it compact." + ), + "city_snapshot": ai_input, + }, + ensure_ascii=False, + ), + }, + ], + } + + +def _cache_city_ai_payload( + cache_key: str, + *, + data: Dict[str, Any], + generated_at: str, + ai_raw: Dict[str, Any], +) -> None: + with _SCAN_CITY_AI_CACHE_LOCK: + _SCAN_CITY_AI_CACHE[cache_key] = { + "expires_at": time.time() + SCAN_AI_CACHE_TTL_SEC, + "generated_at": generated_at, + "city": data.get("name"), + "city_display_name": data.get("display_name"), + "payload": ai_raw, + } + + +def _build_city_ai_result_payload( + *, + data: Dict[str, Any], + generated_at: str, + started_at: float, + ai_raw: Dict[str, Any], + cached: bool = False, + degraded: bool = False, + reason: Optional[str] = None, + reason_zh: Optional[str] = None, + reason_en: Optional[str] = None, +) -> Dict[str, Any]: + payload: Dict[str, Any] = { + "status": "ready", + "cached": cached, + "model": SCAN_AI_MODEL, + "provider": "deepseek", + "city": data.get("name"), + "city_display_name": data.get("display_name"), + "generated_at": generated_at, + "duration_ms": int((time.time() - started_at) * 1000), + "city_forecast": ai_raw, + } + if degraded: + payload["degraded"] = True + if reason: + payload["reason"] = reason + if reason_zh: + payload["reason_zh"] = reason_zh + if reason_en: + payload["reason_en"] = reason_en + return payload + + +def stream_scan_city_ai_forecast_payload( + city: str, + *, + force_refresh: bool = False, + locale: str = "zh-CN", +) -> Iterator[str]: + started_at = time.time() + city_name = _normalize_city_key(city) + normalized_locale = _normalize_locale(locale) + if not city_name: + yield _sse_event("final", {"status": "failed", "reason": "city is required"}) + return + if city_name not in CITIES: + reason_en = f"Unknown city: {city_name}" + reason_zh = f"未知城市：{city_name}" + yield _sse_event( + "final", + { + "status": "failed", + "model": SCAN_AI_MODEL, + "provider": "deepseek", + "city": city_name, + "city_display_name": str(city or "").strip() or city_name, + "reason": reason_en if normalized_locale == "en-US" else reason_zh, + "reason_en": reason_en, + "reason_zh": reason_zh, + }, + ) + return + + cache_key = _city_forecast_cache_key(city_name) + if not force_refresh: + with _SCAN_CITY_AI_CACHE_LOCK: + cached = _SCAN_CITY_AI_CACHE.get(cache_key) + if cached and cached.get("expires_at", 0) >= time.time(): + yield _sse_event( + "final", + { + "status": "ready", + "cached": True, + "model": SCAN_AI_MODEL, + "provider": "deepseek", + "city": cached.get("city") or city_name, + "city_display_name": cached.get("city_display_name") or city_name, + "generated_at": cached.get("generated_at"), + "duration_ms": 0, + "city_forecast": cached.get("payload"), + }, + ) + return + + yield _sse_event( + "progress", + { + "stage": "loading_city", + "message_zh": "正在读取城市实况、模型和最新机场报文…", + "message_en": "Loading city observations, model cluster and latest airport bulletin…", + }, + ) + data = _analyze( + city_name, + force_refresh=False, + include_llm_commentary=False, + detail_mode="full", + ) + ai_input = _build_city_ai_prompt(data) + yield _sse_event( + "progress", + { + "stage": "calling_ai", + "city": data.get("name") or city_name, + "city_display_name": data.get("display_name") or city_name, + "message_zh": "DeepSeek V4-Pro 开始流式解读机场报文…", + "message_en": "DeepSeek V4-Pro is streaming the airport bulletin read…", + }, + ) + + if not SCAN_AI_ENABLED: + yield _sse_event( + "final", + { + "status": "disabled", + "model": SCAN_AI_MODEL, + "provider": "deepseek", + "city": data.get("name") or city_name, + "city_display_name": data.get("display_name") or city_name, + "reason": "POLYWEATHER_SCAN_AI_ENABLED is not enabled", + }, + ) + return + if not str(os.getenv("POLYWEATHER_DEEPSEEK_API_KEY") or "").strip(): + yield _sse_event( + "final", + { + "status": "missing_key", + "model": SCAN_AI_MODEL, + "provider": "deepseek", + "city": data.get("name") or city_name, + "city_display_name": data.get("display_name") or city_name, + "reason": "POLYWEATHER_DEEPSEEK_API_KEY is not configured", + }, + ) + return + + request_json = _build_city_ai_stream_request(ai_input, locale=normalized_locale) + timeout = httpx.Timeout( + timeout=float(SCAN_CITY_AI_TIMEOUT_SEC), + connect=min(8.0, float(SCAN_CITY_AI_TIMEOUT_SEC)), + read=float(SCAN_CITY_AI_TIMEOUT_SEC), + write=10.0, + pool=5.0, + ) + headers = { + "Authorization": f"Bearer {os.getenv('POLYWEATHER_DEEPSEEK_API_KEY')}", + "Content-Type": "application/json", + } + accumulated = "" + last_meta: Dict[str, Any] = {} + try: + logger.info( + "scan city AI stream request city={} locale={} input_bytes={} timeout_sec={}", + ai_input.get("city"), + normalized_locale, + len(json.dumps(request_json, ensure_ascii=False, default=str).encode("utf-8")), + SCAN_CITY_AI_TIMEOUT_SEC, + ) + with httpx.Client(timeout=timeout) as client: + with client.stream( + "POST", + f"{SCAN_AI_BASE_URL}/chat/completions", + headers=headers, + json=request_json, + ) as response: + response.raise_for_status() + for line in response.iter_lines(): + text = str(line or "").strip() + if not text or not text.startswith("data:"): + continue + payload_text = text[5:].strip() + if payload_text == "[DONE]": + break + try: + chunk = json.loads(payload_text) + except Exception: + continue + last_meta = _provider_response_meta(chunk) or last_meta + delta = _extract_provider_stream_delta(chunk) + if delta: + accumulated += delta + yield _sse_event( + "delta", + { + "content": delta, + "raw_length": len(accumulated), + }, + ) + try: + ai_raw = _extract_ai_json_object(accumulated) + if isinstance(ai_raw, dict): + ai_raw["_polyweather_meta"] = { + **last_meta, + "streamed": True, + } + except Exception as exc: + ai_raw = _build_city_ai_fallback( + ai_input, + locale=normalized_locale, + reason=str(exc), + raw_content=accumulated, + ) + generated_at = datetime.utcnow().isoformat() + "Z" + _cache_city_ai_payload( + cache_key, + data=data, + generated_at=generated_at, + ai_raw=ai_raw, + ) + yield _sse_event( + "final", + _build_city_ai_result_payload( + data=data, + generated_at=generated_at, + started_at=started_at, + ai_raw=ai_raw, + ), + ) + except httpx.TimeoutException as exc: + duration_ms = int((time.time() - started_at) * 1000) + reason_en = f"DeepSeek V4-Pro timed out after {SCAN_CITY_AI_TIMEOUT_SEC}s" + reason_zh = f"DeepSeek V4-Pro 在 {SCAN_CITY_AI_TIMEOUT_SEC} 秒内未返回" + logger.warning( + "scan city AI stream timeout fallback city={} duration_ms={} model={} error={}", + data.get("name") or city_name, + duration_ms, + SCAN_AI_MODEL, + exc, + ) + ai_raw = _build_city_ai_fallback( + ai_input, + locale=normalized_locale, + reason=reason_en if normalized_locale == "en-US" else reason_zh, + raw_content=accumulated, + ) + generated_at = datetime.utcnow().isoformat() + "Z" + yield _sse_event( + "final", + _build_city_ai_result_payload( + data=data, + generated_at=generated_at, + started_at=started_at, + ai_raw=ai_raw, + degraded=True, + reason=reason_en if normalized_locale == "en-US" else reason_zh, + reason_en=reason_en, + reason_zh=reason_zh, + ), + ) + except Exception as exc: + reason = str(exc) + logger.warning( + "scan city AI stream failed city={} model={} error={}", + data.get("name") or city_name, + SCAN_AI_MODEL, + reason, + ) + yield _sse_event( + "final", + { + "status": "failed", + "model": SCAN_AI_MODEL, + "provider": "deepseek", + "city": data.get("name") or city_name, + "city_display_name": data.get("display_name") or city_name, + "duration_ms": int((time.time() - started_at) * 1000), + "reason": reason, + "reason_en": reason, + "reason_zh": reason, + "raw_reason": reason, + }, + ) + + def build_scan_city_ai_forecast_payload( city: str, *, @@ -1525,7 +1886,7 @@ def build_scan_city_ai_forecast_payload( normalized_locale, SCAN_AI_MODEL, ) - cache_key = f"city_forecast:{SCAN_CITY_AI_PROMPT_VERSION}:{city_name.lower()}" + cache_key = _city_forecast_cache_key(city_name) if not force_refresh: with _SCAN_CITY_AI_CACHE_LOCK: cached = _SCAN_CITY_AI_CACHE.get(cache_key) @@ -1662,14 +2023,12 @@ def build_scan_city_ai_forecast_payload( "raw_reason": raw_reason, } generated_at = datetime.utcnow().isoformat() + "Z" - with _SCAN_CITY_AI_CACHE_LOCK: - _SCAN_CITY_AI_CACHE[cache_key] = { - "expires_at": time.time() + SCAN_AI_CACHE_TTL_SEC, - "generated_at": generated_at, - "city": data.get("name") or city_name, - "city_display_name": data.get("display_name") or city_name, - "payload": ai_raw, - } + _cache_city_ai_payload( + cache_key, + data=data, + generated_at=generated_at, + ai_raw=ai_raw, + ) logger.info( "scan city AI forecast complete city={} duration_ms={} model={} confidence={}", data.get("name") or city_name,