fix(ci): 修复格式检查与告警

Author: RhoninSeiei

astrbot/core/provider/manager.py

@@ -496,8 +496,12 @@ def get_merged_provider_config(self, provider_config: dict) -> dict:
                 merged_config = {**provider_source, **pc}
                 # 保持 id 为 provider 的 id，而不是 source 的 id
                 merged_config["id"] = pc["id"]
-                merged_config["type"] = provider_source.get("type", merged_config.get("type"))
-                merged_config["provider"] = provider_source.get("provider", merged_config.get("provider"))
+                merged_config["type"] = provider_source.get(
+                    "type", merged_config.get("type")
+                )
+                merged_config["provider"] = provider_source.get(
+                    "provider", merged_config.get("provider")
+                )
                 merged_config["provider_type"] = provider_source.get(
                     "provider_type", merged_config.get("provider_type")
                 )
@@ -506,7 +510,9 @@ def get_merged_provider_config(self, provider_config: dict) -> dict:
                     and merged_config.get("type") == "openai_oauth_chat_completion"
                     and merged_config.get("auth_mode") == "openai_oauth"
                 ):
-                    access_token = (merged_config.get("oauth_access_token") or "").strip()
+                    access_token = (
+                        merged_config.get("oauth_access_token") or ""
+                    ).strip()
                     if access_token:
                         merged_config["key"] = [access_token]
                 pc = merged_config
@@ -527,7 +533,7 @@ def _resolve_env_key_list(self, provider_config: dict) -> dict:
                     if env_val is None:
                         provider_id = provider_config.get("id")
                         logger.warning(
-                            f"Provider {provider_id} 配置项 key[{idx}] 使用环境变量 {env_key} 但未设置。",
+                            f"Provider {provider_id} 配置项 key[{idx}] 使用的环境变量未设置。",
                         )
                         resolved_keys.append("")
                     else:

astrbot/core/provider/oauth/openai_oauth.py

@@ -20,9 +20,11 @@
 def create_pkce_flow() -> dict[str, str]:
     state = secrets.token_hex(16)
     verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode().rstrip("=")
-    challenge = base64.urlsafe_b64encode(
-        hashlib.sha256(verifier.encode()).digest()
-    ).decode().rstrip("=")
+    challenge = (
+        base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())
+        .decode()
+        .rstrip("=")
+    )
     return {
         "state": state,
         "verifier": verifier,
@@ -57,7 +59,9 @@ def parse_authorization_input(raw: str) -> tuple[str, str]:
         parsed = urlparse(value)
         if parsed.query:
             query = parse_qs(parsed.query)
-            return query.get("code", [""])[0].strip(), query.get("state", [""])[0].strip()
+            return query.get("code", [""])[0].strip(), query.get("state", [""])[
+                0
+            ].strip()
         query = parse_qs(value)
         return query.get("code", [""])[0].strip(), query.get("state", [""])[0].strip()
     if "#" in value:
@@ -83,7 +87,9 @@ def parse_oauth_credential_json(raw: str) -> dict[str, Any] | None:
     expires_at = _normalize_expires_at(
         data.get("expired") or data.get("expires_at") or data.get("expires"),
     )
-    account_id = str(data.get("account_id") or "").strip() or extract_account_id_from_jwt(access_token)
+    account_id = str(
+        data.get("account_id") or ""
+    ).strip() or extract_account_id_from_jwt(access_token)
     email = str(data.get("email") or "").strip() or extract_email_from_jwt(access_token)
     return {
         "access_token": access_token,
@@ -122,8 +128,12 @@ async def refresh_access_token(
     return await _request_token(payload, proxy_url)
 
 
-async def _request_token(payload: dict[str, str], proxy_url: str = "") -> dict[str, Any]:
-    async with httpx.AsyncClient(proxy=proxy_url or None, timeout=OPENAI_OAUTH_TIMEOUT) as client:
+async def _request_token(
+    payload: dict[str, str], proxy_url: str = ""
+) -> dict[str, Any]:
+    async with httpx.AsyncClient(
+        proxy=proxy_url or None, timeout=OPENAI_OAUTH_TIMEOUT
+    ) as client:
         response = await client.post(
             OPENAI_OAUTH_TOKEN_URL,
             data=payload,
@@ -134,7 +144,9 @@ async def _request_token(payload: dict[str, str], proxy_url: str = "") -> dict[s
         )
     data = response.json()
     if response.status_code < 200 or response.status_code >= 300:
-        raise ValueError(f"oauth token request failed: status={response.status_code}, body={data}")
+        raise ValueError(
+            f"oauth token request failed: status={response.status_code}, body={data}"
+        )
     access_token = (data.get("access_token") or "").strip()
     refresh_token = (data.get("refresh_token") or "").strip()
     expires_in = int(data.get("expires_in") or 0)

astrbot/core/provider/sources/openai_oauth_source.py

@@ -25,15 +25,16 @@ def __init__(self, provider_config, provider_settings) -> None:
         super().__init__(patched_config, provider_settings)
         self.provider_config = patched_config
         self.api_keys = [access_token] if access_token else self.api_keys
-        self.chosen_api_key = access_token or (self.api_keys[0] if self.api_keys else "")
+        self.chosen_api_key = access_token or (
+            self.api_keys[0] if self.api_keys else ""
+        )
         self.account_id = (
             patched_config.get("oauth_account_id")
             or patched_config.get("account_id")
             or ""
         ).strip()
         self.base_url = (
-            patched_config.get("api_base")
-            or "https://chatgpt.com/backend-api/codex"
+            patched_config.get("api_base") or "https://chatgpt.com/backend-api/codex"
         ).rstrip("/")
 
     async def get_models(self):
@@ -44,12 +45,18 @@ async def get_models(self):
         return []
 
     async def _request_backend(self, payload: dict[str, Any]) -> dict[str, Any]:
-        access_token = (self.provider_config.get("oauth_access_token") or self.chosen_api_key or "").strip()
-        account_id = (self.provider_config.get("oauth_account_id") or self.account_id or "").strip()
+        access_token = (
+            self.provider_config.get("oauth_access_token") or self.chosen_api_key or ""
+        ).strip()
+        account_id = (
+            self.provider_config.get("oauth_account_id") or self.account_id or ""
+        ).strip()
         if not access_token:
             raise Exception("当前 OAuth Source 尚未绑定 access token")
         if not account_id:
-            raise Exception("当前 OAuth Source 缺少 chatgpt_account_id，请重新绑定或导入完整 JSON 凭据")
+            raise Exception(
+                "当前 OAuth Source 缺少 chatgpt_account_id，请重新绑定或导入完整 JSON 凭据"
+            )
 
         headers = {
             "Authorization": f"Bearer {access_token}",
@@ -88,7 +95,9 @@ def _format_backend_error(self, status_code: int, text: str) -> str:
             data = json.loads(stripped)
             return f"Codex backend request failed: status={status_code}, body={data}"
         except Exception:
-            return f"Codex backend request failed: status={status_code}, body={stripped}"
+            return (
+                f"Codex backend request failed: status={status_code}, body={stripped}"
+            )
 
     def _parse_backend_response(self, text: str) -> dict[str, Any]:
         completed_response: dict[str, Any] | None = None
@@ -123,10 +132,14 @@ def _parse_backend_response(self, text: str) -> dict[str, Any]:
         if stripped.startswith("{"):
             data = json.loads(stripped)
             if isinstance(data, dict):
-                if data.get("type") == "response.completed" and isinstance(data.get("response"), dict):
+                if data.get("type") == "response.completed" and isinstance(
+                    data.get("response"), dict
+                ):
                     return data["response"]
                 return data
-        raise Exception("Codex backend response did not contain response.completed event")
+        raise Exception(
+            "Codex backend response did not contain response.completed event"
+        )
 
     def _convert_message_content(self, raw_content: Any) -> str | list[dict[str, Any]]:
         if isinstance(raw_content, str):
@@ -235,7 +248,9 @@ def _convert_messages_to_backend_input(
                     if not name or not call_id:
                         continue
                     if not isinstance(arguments, str):
-                        arguments = json.dumps(arguments, ensure_ascii=False, default=str)
+                        arguments = json.dumps(
+                            arguments, ensure_ascii=False, default=str
+                        )
                     response_items.append(
                         {
                             "type": "function_call",
@@ -244,7 +259,9 @@ def _convert_messages_to_backend_input(
                             "arguments": arguments,
                         }
                     )
-        return "\n\n".join(part for part in instructions_parts if part).strip(), response_items
+        return "\n\n".join(
+            part for part in instructions_parts if part
+        ).strip(), response_items
 
     def _extract_response_usage(self, usage: Any) -> TokenUsage | None:
         if usage is None:
@@ -265,7 +282,9 @@ def _extract_response_usage(self, usage: Any) -> TokenUsage | None:
             output=output_tokens,
         )
 
-    def _convert_tools_to_backend_format(self, tool_list: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    def _convert_tools_to_backend_format(
+        self, tool_list: list[dict[str, Any]]
+    ) -> list[dict[str, Any]]:
         backend_tools: list[dict[str, Any]] = []
         for tool in tool_list:
             if not isinstance(tool, dict):
@@ -283,7 +302,8 @@ def _convert_tools_to_backend_format(self, tool_list: list[dict[str, Any]]) -> l
                 "type": "function",
                 "name": name,
                 "description": str(function.get("description") or "").strip(),
-                "parameters": function.get("parameters") or {"type": "object", "properties": {}},
+                "parameters": function.get("parameters")
+                or {"type": "object", "properties": {}},
             }
             backend_tools.append(backend_tool)
         return backend_tools
@@ -298,43 +318,95 @@ async def _parse_responses_completion(self, response: Any, tools) -> LLMResponse
         if output_text:
             llm_response.result_chain = MessageChain().message(output_text)
 
-        output_items = list(response.get("output", []) if isinstance(response, dict) else getattr(response, "output", []) or [])
+        output_items = list(
+            response.get("output", [])
+            if isinstance(response, dict)
+            else getattr(response, "output", []) or []
+        )
         reasoning_parts: list[str] = []
         tool_args: list[dict[str, Any]] = []
         tool_names: list[str] = []
         tool_ids: list[str] = []
 
         for item in output_items:
-            item_type = item.get("type") if isinstance(item, dict) else getattr(item, "type", None)
+            item_type = (
+                item.get("type")
+                if isinstance(item, dict)
+                else getattr(item, "type", None)
+            )
             if item_type == "reasoning":
-                summaries = item.get("summary", []) if isinstance(item, dict) else getattr(item, "summary", []) or []
+                summaries = (
+                    item.get("summary", [])
+                    if isinstance(item, dict)
+                    else getattr(item, "summary", []) or []
+                )
                 for summary in summaries:
-                    text = summary.get("text") if isinstance(summary, dict) else getattr(summary, "text", None)
+                    text = (
+                        summary.get("text")
+                        if isinstance(summary, dict)
+                        else getattr(summary, "text", None)
+                    )
                     if text:
                         reasoning_parts.append(str(text))
             elif item_type == "function_call" and tools is not None:
-                arguments = item.get("arguments", "{}") if isinstance(item, dict) else getattr(item, "arguments", "{}")
+                arguments = (
+                    item.get("arguments", "{}")
+                    if isinstance(item, dict)
+                    else getattr(item, "arguments", "{}")
+                )
                 try:
-                    parsed_args = json.loads(arguments) if isinstance(arguments, str) else arguments
+                    parsed_args = (
+                        json.loads(arguments)
+                        if isinstance(arguments, str)
+                        else arguments
+                    )
                 except Exception:
                     parsed_args = {}
                 tool_args.append(parsed_args if isinstance(parsed_args, dict) else {})
-                tool_names.append(str(item.get("name", "") if isinstance(item, dict) else getattr(item, "name", "") or ""))
-                tool_ids.append(str(item.get("call_id", "") if isinstance(item, dict) else getattr(item, "call_id", "") or ""))
+                tool_names.append(
+                    str(
+                        item.get("name", "")
+                        if isinstance(item, dict)
+                        else getattr(item, "name", "") or ""
+                    )
+                )
+                tool_ids.append(
+                    str(
+                        item.get("call_id", "")
+                        if isinstance(item, dict)
+                        else getattr(item, "call_id", "") or ""
+                    )
+                )
             elif item_type == "message" and not output_text:
-                content_items = item.get("content", []) if isinstance(item, dict) else getattr(item, "content", []) or []
+                content_items = (
+                    item.get("content", [])
+                    if isinstance(item, dict)
+                    else getattr(item, "content", []) or []
+                )
                 item_text_parts: list[str] = []
                 for content in content_items:
-                    ctype = content.get("type") if isinstance(content, dict) else getattr(content, "type", None)
+                    ctype = (
+                        content.get("type")
+                        if isinstance(content, dict)
+                        else getattr(content, "type", None)
+                    )
                     if ctype in {"output_text", "text"}:
-                        text = content.get("text") if isinstance(content, dict) else getattr(content, "text", None)
+                        text = (
+                            content.get("text")
+                            if isinstance(content, dict)
+                            else getattr(content, "text", None)
+                        )
                         if text:
                             item_text_parts.append(str(text))
                 if item_text_parts:
-                    llm_response.result_chain = MessageChain().message("".join(item_text_parts).strip())
+                    llm_response.result_chain = MessageChain().message(
+                        "".join(item_text_parts).strip()
+                    )
 
         if reasoning_parts:
-            llm_response.reasoning_content = "\n".join(part for part in reasoning_parts if part)
+            llm_response.reasoning_content = "\n".join(
+                part for part in reasoning_parts if part
+            )
 
         if tool_args:
             llm_response.role = "tool"
@@ -346,10 +418,18 @@ async def _parse_responses_completion(self, response: Any, tools) -> LLMResponse
             raise Exception(f"账号态 responses 响应无法解析：{response}。")
 
         llm_response.raw_completion = response
-        response_id = response.get("id") if isinstance(response, dict) else getattr(response, "id", None)
+        response_id = (
+            response.get("id")
+            if isinstance(response, dict)
+            else getattr(response, "id", None)
+        )
         if response_id:
             llm_response.id = response_id
-        usage = self._extract_response_usage(response.get("usage") if isinstance(response, dict) else getattr(response, "usage", None))
+        usage = self._extract_response_usage(
+            response.get("usage")
+            if isinstance(response, dict)
+            else getattr(response, "usage", None)
+        )
         if usage is not None:
             llm_response.usage = usage
         return llm_response