Hua/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: compress DingTalk helpers

Browse Source
This commit is contained in:
Re-bin
2026-02-28 16:23:43 +00:00
parent cfc55d626a
commit 73a708770e
1 changed files with 16 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
nanobot/channels/dingtalk.py
View File

@@ -200,34 +200,18 @@ class DingTalkChannel(BaseChannel):
@staticmethod @staticmethod
def _is_http_url(value: str) -> bool: def _is_http_url(value: str) -> bool:
low = value.lower() return urlparse(value).scheme in ("http", "https")
return low.startswith("http://") or low.startswith("https://")
def _guess_upload_type(self, media_ref: str) -> str: def _guess_upload_type(self, media_ref: str) -> str:
parsed = urlparse(media_ref) ext = Path(urlparse(media_ref).path).suffix.lower()
path = parsed.path if parsed.scheme else media_ref if ext in self._IMAGE_EXTS: return "image"
ext = Path(path).suffix.lower() if ext in self._AUDIO_EXTS: return "voice"
if ext in self._IMAGE_EXTS: if ext in self._VIDEO_EXTS: return "video"
return "image"
if ext in self._AUDIO_EXTS:
return "voice"
if ext in self._VIDEO_EXTS:
return "video"
return "file" return "file"
def _guess_filename(self, media_ref: str, upload_type: str) -> str: def _guess_filename(self, media_ref: str, upload_type: str) -> str:
parsed = urlparse(media_ref) name = os.path.basename(urlparse(media_ref).path)
path = parsed.path if parsed.scheme else media_ref return name or {"image": "image.jpg", "voice": "audio.amr", "video": "video.mp4"}.get(upload_type, "file.bin")
name = os.path.basename(path)
if name:
return name
fallback = {
"image": "image.jpg",
"voice": "audio.amr",
"video": "video.mp4",
"file": "file.bin",
}
return fallback.get(upload_type, "file.bin")
async def _read_media_bytes( async def _read_media_bytes(
self, self,
@@ -288,33 +272,16 @@ class DingTalkChannel(BaseChannel):
try: try:
resp = await self._http.post(url, files=files) resp = await self._http.post(url, files=files)
text = resp.text text = resp.text
try: result = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
result = resp.json()
except Exception:
result = {}
if resp.status_code >= 400: if resp.status_code >= 400:
logger.error( logger.error("DingTalk media upload failed status={} type={} body={}", resp.status_code, media_type, text[:500])
"DingTalk media upload failed status={} type={} body={}",
resp.status_code,
media_type,
text[:500],
)
return None return None
errcode = result.get("errcode", 0) errcode = result.get("errcode", 0)
if errcode != 0: if errcode != 0:
logger.error( logger.error("DingTalk media upload api error type={} errcode={} body={}", media_type, errcode, text[:500])
"DingTalk media upload api error type={} errcode={} body={}",
media_type,
errcode,
text[:500],
)
return None return None
media_id = ( sub = result.get("result") or {}
result.get("media_id") media_id = result.get("media_id") or result.get("mediaId") or sub.get("media_id") or sub.get("mediaId")
or result.get("mediaId")
or (result.get("result") or {}).get("media_id")
or (result.get("result") or {}).get("mediaId")
)
if not media_id: if not media_id:
logger.error("DingTalk media upload missing media_id body={}", text[:500]) logger.error("DingTalk media upload missing media_id body={}", text[:500])
return None return None
@@ -347,25 +314,13 @@ class DingTalkChannel(BaseChannel):
resp = await self._http.post(url, json=payload, headers=headers) resp = await self._http.post(url, json=payload, headers=headers)
body = resp.text body = resp.text
if resp.status_code != 200: if resp.status_code != 200:
logger.error( logger.error("DingTalk send failed msgKey={} status={} body={}", msg_key, resp.status_code, body[:500])
"DingTalk send failed msgKey={} status={} body={}",
msg_key,
resp.status_code,
body[:500],
)
return False return False
try: try: result = resp.json()
result = resp.json() except Exception: result = {}
except Exception:
result = {}
errcode = result.get("errcode") errcode = result.get("errcode")
if errcode not in (None, 0): if errcode not in (None, 0):
logger.error( logger.error("DingTalk send api error msgKey={} errcode={} body={}", msg_key, errcode, body[:500])
"DingTalk send api error msgKey={} errcode={} body={}",
msg_key,
errcode,
body[:500],
)
return False return False
logger.debug("DingTalk message sent to {} with msgKey={}", chat_id, msg_key) logger.debug("DingTalk message sent to {} with msgKey={}", chat_id, msg_key)
return True return True