Add chart freshness diagnostics
This commit is contained in:
2569718930@qq.com
+129
-12
@@ -973,6 +973,90 @@ def _city_detail_batch_partial_timeout_seconds() -> Optional[float]:
|
||||
return max(0.001, min(60.0, timeout_ms / 1000.0))
|
||||
|
||||
|
||||
def _city_detail_batch_partial_timeout_ms() -> Optional[int]:
|
||||
timeout_sec = _city_detail_batch_partial_timeout_seconds()
|
||||
if timeout_sec is None:
|
||||
return None
|
||||
return int(round(timeout_sec * 1000.0))
|
||||
|
||||
|
||||
def _city_detail_batch_partial_reason(
|
||||
*,
|
||||
busy: bool,
|
||||
missing: List[str],
|
||||
errors: Dict[str, str],
|
||||
) -> Optional[str]:
|
||||
if busy:
|
||||
return "busy"
|
||||
if missing and errors:
|
||||
return "timeout_error"
|
||||
if missing:
|
||||
return "timeout"
|
||||
if errors:
|
||||
return "error"
|
||||
return None
|
||||
|
||||
|
||||
def _build_city_detail_batch_diagnostics(
|
||||
*,
|
||||
city_names: List[str],
|
||||
details: Dict[str, Any],
|
||||
errors: Dict[str, str],
|
||||
missing: List[str],
|
||||
resolution: Optional[str],
|
||||
detail_scope: str,
|
||||
force_refresh: bool,
|
||||
response_source: str,
|
||||
busy: bool = False,
|
||||
city_durations_ms: Optional[Dict[str, float]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
city_durations_ms = city_durations_ms or {}
|
||||
missing_set = set(missing)
|
||||
error_set = set(errors)
|
||||
detail_set = set(details)
|
||||
partial_reason = _city_detail_batch_partial_reason(
|
||||
busy=busy,
|
||||
missing=missing,
|
||||
errors=errors,
|
||||
)
|
||||
city_status: Dict[str, Dict[str, Any]] = {}
|
||||
for city in city_names:
|
||||
if busy:
|
||||
status = "busy"
|
||||
elif city in missing_set:
|
||||
status = "timeout"
|
||||
elif city in error_set:
|
||||
status = "error"
|
||||
elif city in detail_set:
|
||||
status = "ok"
|
||||
else:
|
||||
status = "missing"
|
||||
city_status[city] = {
|
||||
"status": status,
|
||||
"duration_ms": city_durations_ms.get(city),
|
||||
}
|
||||
if city in errors:
|
||||
city_status[city]["error"] = errors[city]
|
||||
|
||||
return {
|
||||
"version": 1,
|
||||
"response_source": response_source,
|
||||
"partial": bool(partial_reason),
|
||||
"partial_reason": partial_reason,
|
||||
"requested_count": len(city_names),
|
||||
"completed_count": len(details),
|
||||
"missing_count": len(missing),
|
||||
"error_count": len(errors),
|
||||
"batch_concurrency": _city_detail_batch_concurrency(),
|
||||
"global_concurrency": _city_detail_batch_global_concurrency(),
|
||||
"partial_timeout_ms": _city_detail_batch_partial_timeout_ms(),
|
||||
"force_refresh": force_refresh,
|
||||
"resolution": resolution,
|
||||
"scope": detail_scope,
|
||||
"city_status": city_status,
|
||||
}
|
||||
|
||||
|
||||
async def get_city_detail_batch_payload(
|
||||
request: Request,
|
||||
*,
|
||||
@@ -1014,30 +1098,52 @@ 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):
|
||||
missing = list(city_names)
|
||||
errors: Dict[str, str] = {}
|
||||
details: Dict[str, Any] = {}
|
||||
return {
|
||||
"cities": city_names,
|
||||
"details": {},
|
||||
"errors": {},
|
||||
"missing": list(city_names),
|
||||
"details": details,
|
||||
"errors": errors,
|
||||
"missing": missing,
|
||||
"partial": True,
|
||||
"busy": True,
|
||||
"stale_reason": "city detail batch builder is busy",
|
||||
"diagnostics": _build_city_detail_batch_diagnostics(
|
||||
city_names=city_names,
|
||||
details=details,
|
||||
errors=errors,
|
||||
missing=missing,
|
||||
resolution=resolution,
|
||||
detail_scope=detail_scope,
|
||||
force_refresh=force_refresh,
|
||||
response_source="busy",
|
||||
busy=True,
|
||||
),
|
||||
}
|
||||
|
||||
try:
|
||||
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
|
||||
city_durations_ms: Dict[str, float] = {}
|
||||
|
||||
async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
|
||||
async with semaphore:
|
||||
return await _build_city_detail_batch_item_async(
|
||||
city,
|
||||
force_refresh=force_refresh,
|
||||
market_slug=market_slug,
|
||||
target_date=target_date,
|
||||
resolution=resolution,
|
||||
detail_scope=detail_scope,
|
||||
timing_recorder=timer,
|
||||
)
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
return await _build_city_detail_batch_item_async(
|
||||
city,
|
||||
force_refresh=force_refresh,
|
||||
market_slug=market_slug,
|
||||
target_date=target_date,
|
||||
resolution=resolution,
|
||||
detail_scope=detail_scope,
|
||||
timing_recorder=timer,
|
||||
)
|
||||
finally:
|
||||
city_durations_ms[city] = round(
|
||||
(time.perf_counter() - started) * 1000.0,
|
||||
1,
|
||||
)
|
||||
|
||||
task_by_city = {
|
||||
city: asyncio.create_task(_build_with_limit(city))
|
||||
@@ -1076,6 +1182,17 @@ async def get_city_detail_batch_payload(
|
||||
"errors": errors,
|
||||
"missing": missing,
|
||||
"partial": bool(missing or errors),
|
||||
"diagnostics": _build_city_detail_batch_diagnostics(
|
||||
city_names=city_names,
|
||||
details=details,
|
||||
errors=errors,
|
||||
missing=missing,
|
||||
resolution=resolution,
|
||||
detail_scope=detail_scope,
|
||||
force_refresh=force_refresh,
|
||||
response_source="fresh_build",
|
||||
city_durations_ms=city_durations_ms,
|
||||
),
|
||||
}
|
||||
finally:
|
||||
build_semaphore.release()
|
||||
|
||||
@@ -162,6 +162,11 @@ class SseManager:
|
||||
|
||||
@staticmethod
|
||||
def _format_event(event: dict[str, Any]) -> str:
|
||||
if str(event.get("type") or "").startswith("city_observation_patch"):
|
||||
event = {
|
||||
**event,
|
||||
"sse_emitted_at_ms": int(time.time() * 1000),
|
||||
}
|
||||
return f"data: {json.dumps(event, ensure_ascii=False, separators=(',', ':'))}\n\n"
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user