feat: add internal diagnostic log & console/file logging

swanlab/sdk/cmd/login.py

@@ -12,9 +12,10 @@
 from swanlab.sdk.internal import apikey
 from swanlab.sdk.internal.context import RunConfig, RunContext, get_context, has_context, use_temp_context
 from swanlab.sdk.internal.core_python import client
+from swanlab.sdk.internal.pkg import console
 from swanlab.sdk.internal.pkg.scope import Scope
 from swanlab.sdk.internal.settings import Settings, settings
-from swanlab.sdk.pkg import console, helper
+from swanlab.sdk.pkg import helper
 from swanlab.sdk.typings.core_python.api.bootstrap import LoginResponse
 
 

swanlab/sdk/internal/apikey/__init__.py

@@ -11,9 +11,10 @@
 
 from rich.text import Text
 
+from swanlab.sdk.internal.pkg import console
 from swanlab.sdk.internal.pkg.netrc import get_nrc_path, remove_host_suffix, write_netrc
 from swanlab.sdk.internal.pkg.settings import get_current_settings
-from swanlab.sdk.pkg import console, helper
+from swanlab.sdk.pkg import helper
 
 __all__ = ["get", "save", "exists"]
 

swanlab/sdk/internal/core_python/client/__init__.py

@@ -14,10 +14,10 @@
 from swanlab.sdk.internal.core_python.api.bootstrap import login_by_api_key
 from swanlab.sdk.internal.pkg.netrc import remove_host_suffix
 from swanlab.sdk.internal.pkg.scope import set_context
-from swanlab.sdk.pkg import console
 from swanlab.sdk.pkg.version import get_swanlab_version
 from swanlab.sdk.typings.core_python.api.bootstrap import LoginResponse
 
+from ...pkg import console
 from . import session
 from .helper import decode_response
 

swanlab/sdk/internal/core_python/client/session.py

@@ -8,20 +8,23 @@
 
 import contextvars
 import copy
-from typing import Optional
+import time
+from typing import Optional, Union
 
 from requests import Session
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 
 from swanlab.exceptions import ApiError
 from swanlab.sdk.internal.core_python.client.helper import decode_error_response
+from swanlab.sdk.internal.pkg import log
+from swanlab.sdk.pkg import helper
 from swanlab.sdk.pkg.version import get_swanlab_version
 
 __all__ = ["create", "TimeoutHTTPAdapter", "SessionWithRetry"]
 VERSION_HEADER = "X-SwanLab-SDK-Version"
 # 用于存储当前请求的重试次数，避免在请求中传递 retries 参数
-request_retries_ctx = contextvars.ContextVar("request_retries", default=None)
+request_retries_ctx: contextvars.ContextVar[Optional[int]] = contextvars.ContextVar("request_retries", default=None)
 
 
 class TimeoutHTTPAdapter(HTTPAdapter):
@@ -51,6 +54,25 @@ def send(self, request, *args, **kwargs):
         return super().send(request, *args, **kwargs)
 
 
+@helper.catch_and_return_none()
+def format_body_preview(body: Optional[Union[bytes, str]], max_len: int = 1000) -> str:
+    """
+    统一格式化并截断请求/响应体，安全处理二进制乱码。
+    如果是 bytes，尝试用 utf-8 解码；超过 max_len 则截断防刷屏。
+    """
+    if not body:
+        return ""
+
+    if isinstance(body, bytes):
+        body_str = body.decode("utf-8", errors="replace")
+    else:
+        body_str = str(body)
+
+    if len(body_str) > max_len:
+        return body_str[:max_len] + " ... (truncated)"
+    return body_str
+
+
 class SessionWithRetry(Session):
     """
     支持在请求级别自定义重试次数的 Session。
@@ -73,18 +95,52 @@ def request(self, method, url, *args, **kwargs):
 
     def send(self, request, **kwargs):
         """
-        重写底层发送方法，统一处理所有响应的校验逻辑
+        重写底层发送方法，统一处理所有响应的校验逻辑和网络日志记录
         """
+        method = (request.method or "unknown").upper()
+
+        # --- [DEBUG] 记录请求详情 ---
+        if helper.env.DEBUG:
+            # 这里的 request.url 已经包含了 params 拼接后的完整 query 字符串
+            log.debug("[HTTP-REQ] %s %s | Headers: %s", method, request.url, request.headers)
+            body = request.body
+            if body:
+                body_preview = format_body_preview(body) or "<unknown binary data>"
+                log.debug("[HTTP-REQ-BODY] %s", body_preview)
+        # ---------------------------
+
+        start = time.perf_counter()
+
         # 调用父类（或 Adapter）获取响应
         response = super().send(request, **kwargs)
 
-        # 1. 2xx 响应直接放行
+        elapsed_ms = (time.perf_counter() - start) * 1000
+        trace_id = response.headers.get("traceid", "unknown")
+
+        # 1. 2xx 响应：记录正常日志后放行
         if response.ok:
+            log.debug(
+                "[HTTP] %s %s -> %s (%.0fms) trace:%s",
+                method,
+                request.url,
+                response.status_code,
+                elapsed_ms,
+                trace_id,
+            )
+
+            # --- [DEBUG] 记录成功响应详情 ---
+            if helper.env.DEBUG:
+                log.debug("[HTTP-RES] Headers: %s", response.headers)
+                # 直接传 response.text (str) 给格式化函数
+                text = response.text
+                if text:
+                    resp_preview = format_body_preview(text) or "<unknown data>"
+                    log.debug("[HTTP-RES-BODY] %s", resp_preview)
+            # -------------------------------
+
             return response
 
-        # 2. 准备 Fallback 默认值
-        method = (request.method or "unknown").upper()
-        trace_id = response.headers.get("traceid", "unknown")
+        # 2. 非 2xx 响应：准备 Fallback 默认值
         error_code = "unknown code"
         error_message = "unknown error"
 
@@ -93,7 +149,29 @@ def send(self, request, **kwargs):
         if decoded is not None:
             error_code, error_message = decoded
 
-        # 4. 抛出友好的自定义 ApiError
+        # 4. 记录错误日志（附带响应体，方便排查）
+        log.error(
+            "[HTTP] %s %s -> %s (%.0fms) trace:%s | [ERR] code=%s message=%s",
+            method,
+            request.url,
+            response.status_code,
+            elapsed_ms,
+            trace_id,
+            error_code,
+            error_message,
+        )
+
+        # --- [DEBUG] 记录失败响应详情 ---
+        if helper.env.DEBUG:
+            log.debug("[HTTP-RES-ERR] Headers: %s", response.headers)
+            text = response.text
+            if text and not decoded:
+                # 只有当解码失败时，才额外把原始错误 body 打印出来
+                err_preview = format_body_preview(text) or "<unknown data>"
+                log.debug("[HTTP-RES-ERR-BODY] %s", err_preview)
+        # -------------------------------
+
+        # 5. 抛出友好的自定义 ApiError
         raise ApiError(response, method=method, trace_id=trace_id, code=error_code, message=error_message)
 
     # ---------------------------------- 类型提示占位符，保留以保证 IDE 友好 ----------------------------------

swanlab/sdk/pkg/console/__init__.py → swanlab/sdk/internal/pkg/console/__init__.py

@@ -3,17 +3,17 @@
 @file: __init__.py
 @time: 2026/3/6 13:20
 @description: SwanLab SDK 控制台日志模块，负责打印日志
+当 SWANLAB_DEBUG=true 时，终端输出会同步镜像到诊断日志文件（通过 log 模块）
 """
 
-import os
+from typing import Any
 
 from rich.console import Console
 from rich.markup import escape
 from rich.text import Text
 
-# 设计上 console 独立于其他模块，包括settings。但是我们又希望有一个环境变量去控制是否打印debug信息，所以这里额外绑定一个环境变量
-DEBUG = os.getenv("SWANLAB_DEBUG", "false").lower() in ["true", "1", "yes", "on"]
-
+from swanlab.sdk.internal.pkg import log
+from swanlab.sdk.pkg import helper
 
 __all__ = ["debug", "info", "warning", "error", "critical", "disable_log", "enable_log"]
 
@@ -62,6 +62,20 @@ def _print_formatted(level_name: str, *args, **kwargs):
     _console.print(prefix_text, *safe_args, **kwargs)
 
 
+def _to_plain_text(*args: Any) -> str:
+    """
+    将 console 的 *args 转为纯文本字符串，用于写入诊断日志文件。
+    去除 rich markup，保留原始语义。
+    """
+    parts = []
+    for arg in args:
+        if isinstance(arg, Text):
+            parts.append(arg.plain)
+        else:
+            parts.append(str(arg))
+    return " ".join(parts)
+
+
 # -----------------------------------------------------------------------------
 # 导出的模块级日志函数
 # -----------------------------------------------------------------------------
@@ -77,34 +91,43 @@ def print(*args, **kwargs):  # noqa: A001
 
 def debug(*args, **kwargs):
     """发送调试消息"""
-    if not _can_log or not DEBUG:
+    if not _can_log or not helper.env.DEBUG:
         return
     _print_formatted("debug", *args, **kwargs)
+    log.debug("[CONSOLE] %s", _to_plain_text(*args))
 
 
 def info(*args, **kwargs):
     """发送常规通知"""
     if not _can_log:
         return
     _print_formatted("info", *args, **kwargs)
+    if helper.env.DEBUG:
+        log.info("[CONSOLE] %s", _to_plain_text(*args))
 
 
 def warning(*args, **kwargs):
     """发生警告"""
     if not _can_log:
         return
     _print_formatted("warning", *args, **kwargs)
+    if helper.env.DEBUG:
+        log.warning("[CONSOLE] %s", _to_plain_text(*args))
 
 
 def error(*args, **kwargs):
     """发生错误"""
     if not _can_log:
         return
     _print_formatted("error", *args, **kwargs)
+    if helper.env.DEBUG:
+        log.error("[CONSOLE] %s", _to_plain_text(*args))
 
 
 def critical(*args, **kwargs):
     """致命错误"""
     if not _can_log:
         return
     _print_formatted("critical", *args, **kwargs)
+    if helper.env.DEBUG:
+        log.critical("[CONSOLE] %s", _to_plain_text(*args))