server-core/app/im/client.py

# -*- 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。

    出口统一把 ``*avatar*`` 字段 resolve 为可访问直链（im-service 存的是
    users 表原始对象路径 /oss/...，桌面/Web 无法直接渲染）。
    """
    headers = {"Authorization": auth_header} if auth_header else {}
    data = await _request(method=method, path=f"/api/v1{api_path}", headers=headers, json_body=body, params=params)
    return resolve_avatar_fields(data)


def resolve_avatar_fields(obj):
    """递归把 dict/list 中所有键含 ``avatar`` 的字符串字段补全为 CDN 直链。

    覆盖 im-service 返回的：conversation.avatar / peer.avatar / members[].avatar /
    message.sender_avatar / friend.from_avatar 等。非 /oss 与完整 URL 由 resolve_url 原样处理。
    """
    from ..infrastructure.oss import resolve_url
    if isinstance(obj, dict):
        out = {}
        for k, v in obj.items():
            if isinstance(k, str) and "avatar" in k.lower() and isinstance(v, str):
                out[k] = resolve_url(v)
            elif isinstance(v, (dict, list)):
                out[k] = resolve_avatar_fields(v)
            else:
                out[k] = v
        return out
    if isinstance(obj, list):
        return [resolve_avatar_fields(i) for i in obj]
    return obj


# ── 内部同步（业务路由调用，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)}