81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import hmac
|
||
import base64
|
||
import logging
|
||
import time
|
||
from urllib.parse import quote_plus
|
||
|
||
import httpx
|
||
from urllib.parse import urlparse
|
||
|
||
from app.config import settings
|
||
from app.schemas import IMPublishRequest, PublishResponse
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class IMPublisher:
|
||
async def publish(self, req: IMPublishRequest, request_id: str = "") -> PublishResponse:
|
||
rid = request_id or "-"
|
||
if not settings.im_webhook_url:
|
||
logger.warning("im_skipped rid=%s reason=empty_webhook_url", rid)
|
||
return PublishResponse(ok=False, detail="缺少 IM_WEBHOOK_URL 配置")
|
||
|
||
parsed = urlparse(settings.im_webhook_url)
|
||
host = parsed.netloc or "(invalid_url)"
|
||
logger.info(
|
||
"im_publish_start rid=%s webhook_host=%s sign_enabled=%s title_chars=%d body_truncated_to=3800",
|
||
rid,
|
||
host,
|
||
bool(settings.im_secret),
|
||
len(req.title or ""),
|
||
)
|
||
|
||
webhook = self._with_signature(settings.im_webhook_url, settings.im_secret)
|
||
payload = {
|
||
"msg_type": "post",
|
||
"content": {
|
||
"post": {
|
||
"zh_cn": {
|
||
"title": req.title,
|
||
"content": [[{"tag": "text", "text": req.body_markdown[:3800]}]],
|
||
}
|
||
}
|
||
},
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
logger.info("im_http_post rid=%s method=POST timeout_s=20", rid)
|
||
r = await client.post(webhook, json=payload)
|
||
try:
|
||
data = r.json()
|
||
except Exception:
|
||
data = {"status_code": r.status_code, "text": r.text}
|
||
|
||
logger.info(
|
||
"im_http_response rid=%s status=%s body_preview=%s",
|
||
rid,
|
||
r.status_code,
|
||
str(data)[:500],
|
||
)
|
||
|
||
if r.status_code >= 400:
|
||
logger.warning("im_push_failed rid=%s http_status=%s", rid, r.status_code)
|
||
return PublishResponse(ok=False, detail=f"IM 推送失败: {data}", data=data)
|
||
|
||
logger.info("im_push_ok rid=%s", rid)
|
||
return PublishResponse(ok=True, detail="IM 推送成功", data=data)
|
||
|
||
def _with_signature(self, webhook: str, secret: str | None) -> str:
|
||
# 兼容飞书 webhook 签名参数：timestamp/sign
|
||
if not secret:
|
||
return webhook
|
||
|
||
ts = str(int(time.time()))
|
||
string_to_sign = f"{ts}\n{secret}".encode("utf-8")
|
||
sign = base64.b64encode(hmac.new(string_to_sign, b"", hashlib.sha256).digest()).decode("utf-8")
|
||
connector = "&" if "?" in webhook else "?"
|
||
return f"{webhook}{connector}timestamp={ts}&sign={quote_plus(sign)}"
|