增加验收设备管理与动态资源调度
This commit is contained in:
@@ -152,6 +152,24 @@ def list_evidences(params: dict[str, Any]) -> dict[str, Any]:
|
||||
return data
|
||||
|
||||
|
||||
def list_configured_notes() -> list[dict[str, Any]]:
|
||||
"""读取凭证后台中由用户配置的多篇展示笔记。"""
|
||||
if not Config.EVIDENCE_ADMIN_BASE_URL:
|
||||
return []
|
||||
try:
|
||||
response = SESSION.get(
|
||||
f"{Config.EVIDENCE_ADMIN_BASE_URL}/api/notes",
|
||||
timeout=min(Config.CRAWLER_TIMEOUT_SECONDS, 15),
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
except (requests.RequestException, ValueError) as exc:
|
||||
raise TaskError("OCR_CONFIGURED_NOTES_UNAVAILABLE", "配置笔记服务暂时不可用", 502) from exc
|
||||
if payload.get("error"):
|
||||
raise TaskError("OCR_CONFIGURED_NOTES_ERROR", str(payload["error"].get("message") or "配置笔记读取失败"), 502)
|
||||
return payload.get("data") or []
|
||||
|
||||
|
||||
def export_evidences(params: dict[str, Any]) -> tuple[io.BytesIO, str, int]:
|
||||
query = dict(params)
|
||||
query["limit"] = 100
|
||||
|
||||
Reference in New Issue
Block a user