Pine
111 lines
3.9 KiB
Python
111 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""IM 微服务客户端（opc-im-service，loopback）。
|
||
|
||
- 对外 REST 转发：server-core /im/{path} → im-service /api/v1/{path}（透传用户 JWT）
|
||
- 内部同步：任务群 / 园区群 / 客服通知（带 IM_INTERNAL_TOKEN）
|
||
- WS 桥接在 router.py 实现
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
import httpx
|
||
|
||
from .. import config
|
||
|
||
|
||
class IMError(Exception):
|
||
"""IM 服务调用异常。"""
|
||
|
||
|
||
def _base() -> str:
|
||
return config.IM_BASE_URL.rstrip("/")
|
||
|
||
|
||
def _internal_headers() -> dict[str, str]:
|
||
if not config.IM_INTERNAL_TOKEN:
|
||
return {}
|
||
return {"X-IM-Token": config.IM_INTERNAL_TOKEN}
|
||
|
||
|
||
async def _request(*, method: str, path: str, headers: dict[str, str] | None = None,
|
||
json_body: dict | None = None, params: dict | None = None) -> dict:
|
||
"""带超时/重试的 IM 服务请求；2xx 返回 JSON，否则抛 IMError。"""
|
||
url = f"{_base()}{path}"
|
||
last: Exception | None = None
|
||
for _ in range(max(1, config.IM_RETRIES + 1)):
|
||
try:
|
||
async with httpx.AsyncClient(timeout=config.IM_TIMEOUT) as client:
|
||
resp = await client.request(method, url, json=json_body, headers=headers, params=params)
|
||
if resp.status_code == 200:
|
||
try:
|
||
return resp.json() if resp.content else {}
|
||
except Exception: # noqa: BLE001
|
||
return {}
|
||
if resp.status_code < 500:
|
||
raise IMError(f"im-service {resp.status_code}: {resp.text[:200]}")
|
||
last = IMError(f"im-service {resp.status_code}: {resp.text[:200]}")
|
||
except httpx.HTTPError as exc:
|
||
last = IMError(f"im-service request failed: {exc}")
|
||
except IMError as exc:
|
||
last = exc
|
||
if " 4" in str(exc) or " 3" in str(exc):
|
||
break
|
||
await asyncio.sleep(0.2)
|
||
raise last or IMError("im-service unavailable")
|
||
|
||
|
||
async def forward(api_path: str, *, auth_header: str, method: str = "GET",
|
||
body: dict | None = None, params: dict | None = None) -> dict:
|
||
"""把用户 REST 请求转发到 im-service /api/v1/*，透传 JWT。"""
|
||
headers = {"Authorization": auth_header} if auth_header else {}
|
||
return await _request(method=method, path=f"/api/v1{api_path}", headers=headers, json_body=body, params=params)
|
||
|
||
|
||
# ── 内部同步（业务路由调用，fire-and-forget 不阻断主流程） ─────
|
||
|
||
async def sync_task_group(task_id: str, task_title: str = "") -> bool:
|
||
"""任务发布/接单/中标后自动建/刷新任务群。失败仅记录，不抛给主流程。"""
|
||
try:
|
||
await _request(
|
||
method="POST", path="/internal/task-sync",
|
||
headers=_internal_headers(),
|
||
json_body={"task_id": task_id, "task_title": task_title},
|
||
)
|
||
return True
|
||
except IMError:
|
||
return False
|
||
|
||
|
||
async def sync_park_group(park_id: str, park_name: str = "") -> bool:
|
||
"""确保园区公共群存在。"""
|
||
try:
|
||
await _request(
|
||
method="POST", path="/internal/park-sync",
|
||
headers=_internal_headers(),
|
||
json_body={"park_id": park_id, "park_name": park_name},
|
||
)
|
||
return True
|
||
except IMError:
|
||
return False
|
||
|
||
|
||
async def internal_mqtt_credentials(user_id: str) -> dict:
|
||
"""登录时生成/更新用户 MQTT 凭证并同步 EMQX（随登录响应下发）。
|
||
|
||
生成新的一次性密码 → im-service 写 im_mqtt_credentials → 同步 EMQX。
|
||
返回 dict 含 broker_ws_url/username/password/client_id/topic_prefix/presence_topic。
|
||
"""
|
||
return await _request(
|
||
method="POST", path="/internal/mqtt-credentials",
|
||
headers=_internal_headers(),
|
||
json_body={"user_id": str(user_id)},
|
||
)
|
||
|
||
|
||
async def health() -> dict:
|
||
try:
|
||
return await _request(method="GET", path="/health")
|
||
except IMError as exc:
|
||
return {"ok": False, "detail": str(exc)}
|