""" 轻量日志模块，通过 HTTP POST 上报到阿里云 SLS。 仅依赖 Python 标准库（3.8+）。 用法： from log import logger, set_log_info set_log_info(brand_id="12345", version="1.0.0") log = logger.child(tag="my_module") log.info("操作成功", extra={"count": 42}) log.error("操作失败", extra={"reason": "网络超时"}) """ import json import os import platform import socket import sys import threading import traceback import uuid from typing import Any, Callable, Dict, Optional, Union from urllib.request import Request, urlopen from urllib.error import URLError LOG_URL = "https://tnnm-alsc-saas-merchant-spider-fy21.cn-wulanchabu.log.aliyuncs.com/logstores/kry-cli/track?APIVersion=0.6.0" _ENV_MAP = {"daily": "gld", "pre": "pre", "prod": "prod"} _SESSION_ID = os.environ.get("REWIND_SESSION_ID", "") _ROBOT_ID = os.environ.get("ROBOT_UID", "") _TRACE_ID = uuid.uuid4().hex # 全局上下文（set_log_info 写入） _ctx = { "brand_id": None, "brand_name": None, "brand_count": 0, "shop_count": 0, "version": None, "user_id": None, "source": None, "mobile": None, "pkg": None, "call_path": None, "call_skill": None, } # ────────────────────────────────────────────── # 环境 # ────────────────────────────────────────────── def get_env() -> str: """获取当前环境标识，与 JS 版 getEnv() 逻辑一致。""" claw_env = os.environ.get("CLIENT_CLAW_ENV", "") if claw_env and claw_env in _ENV_MAP: return _ENV_MAP[claw_env] return "pre" def is_test_runner() -> bool: """检测是否运行在 eval 测试环境中。""" session_id = os.environ.get("REWIND_SESSION_ID", "") return session_id.startswith("eval-sess-eval") # ────────────────────────────────────────────── # 全局上下文设置 # ────────────────────────────────────────────── def set_log_info( brand_id: Optional[str] = None, brand_name: Optional[str] = None, brand_count: Optional[int] = None, shop_count: Optional[int] = None, version: Optional[str] = None, user_id: Optional[str] = None, source: Optional[str] = None, mobile: Optional[str] = None, pkg: Optional[str] = None, ) -> None: """设置日志全局上下文，与 JS 版 setLogInfo() 对应。""" if brand_id is not None: _ctx["brand_id"] = brand_id if brand_name is not None: _ctx["brand_name"] = brand_name if brand_count is not None: _ctx["brand_count"] = brand_count if shop_count is not None: _ctx["shop_count"] = shop_count if version is not None: _ctx["version"] = version if user_id is not None: _ctx["user_id"] = user_id if source is not None: _ctx["source"] = source if mobile is not None: _ctx["mobile"] = mobile if pkg is not None: _ctx["pkg"] = pkg def get_caller(argv: list) -> Dict[str, str]: """从命令行参数中提取 kry-* skill 调用路径，与 JS 版 getCaller() 对应。""" result = {"path": "", "skill": ""} file_path = "" for arg in argv or []: if "kry-" in arg: file_path = arg break if not file_path: return result segments = file_path.replace("\\", "/").split("/") skill_index = -1 for i, seg in enumerate(segments): if "kry-" in seg and seg != "kry-claw-skills": skill_index = i break if skill_index == -1: return result result["skill"] = segments[skill_index] result["path"] = "/".join(segments[skill_index + 1:]) return result def set_caller(argv: Optional[list] = None) -> Optional[Dict[str, str]]: """从 sys.argv 或指定参数中提取调用者信息并写入全局上下文。""" try: call = get_caller(argv if argv is not None else sys.argv) if call["skill"]: _ctx["call_skill"] = call["skill"] if call["path"]: _ctx["call_path"] = call["path"] return call except Exception: return None # ────────────────────────────────────────────── # 序列化辅助 # ────────────────────────────────────────────── def _safe_value(v: Any) -> Any: """将值转为可 JSON 序列化的形式。""" if isinstance(v, str): return v if isinstance(v, BaseException): err: Dict[str, Any] = { "type": type(v).__name__, "message": str(v), "traceback": "".join(traceback.format_exception(type(v), v, v.__traceback__)), } # 附加自定义属性 for k, val in vars(v).items(): if k not in err: err[k] = val return err try: json.dumps(v) return v except (TypeError, ValueError): return str(v) def _stringify(data: Dict[str, Any]) -> Dict[str, Any]: """将 dict 中的非字符串值序列化为 JSON 字符串，与 JS 版 stringify() 对应。""" result = {} for k, v in data.items(): result[k] = _safe_value(v) if isinstance(v, str) else json.dumps(_safe_value(v), ensure_ascii=False) return result # ────────────────────────────────────────────── # 核心上报 # ────────────────────────────────────────────── def _send_log(level: str, message: str, extra: Optional[Dict[str, Any]] = None) -> None: """异步上报日志到阿里云 SLS。""" env = get_env() if env == "prod" and level == "debug": return common = {"traceId": _TRACE_ID, "level": level, "message": message} tags = { "env": env, "brandId": _ctx.get("brand_id"), "brandName": _ctx.get("brand_name"), "brandCount": str(_ctx.get("brand_count", 0)), "shopCount": str(_ctx.get("shop_count", 0)), "version": _ctx.get("version"), "sessionId": _SESSION_ID, "userId": _ctx.get("user_id"), "hostname": socket.gethostname(), "platform": sys.platform, "arch": platform.machine(), "source": _ctx.get("source"), "mobile": _ctx.get("mobile"), "pkg": _ctx.get("pkg"), "callPath": _ctx.get("call_path"), "callSkill": _ctx.get("call_skill"), } if is_test_runner(): tags["test"] = "TestRunner" try: extra = extra or {} tag = extra.pop("tag", None) logs = {**common, "tag": tag} if extra: logs["data"] = extra body = json.dumps({"__logs__": [_stringify(logs)], "__tags__": tags}, ensure_ascii=False) except Exception: body = json.dumps( { "__logs__": [json.dumps({**common, "level": "error", "message": message, "data": "[Unserializable]"})], "__tags__": tags, } ) def _post() -> None: try: req = Request(LOG_URL, data=body.encode("utf-8"), method="POST") req.add_header("Content-Type", "application/json") with urlopen(req, timeout=5) as _: pass except (URLError, OSError, Exception): pass # 异步上报，不阻塞主线程 threading.Thread(target=_post, daemon=True).start() # ────────────────────────────────────────────── # Logger # ────────────────────────────────────────────── class Logger: """ 轻量日志记录器，支持 child() 创建带固定 tag 的子 logger。 API 与 JS 版 logger 保持一致。 """ def __init__(self, default_extra: Optional[Dict[str, Any]] = None): self._default_extra: Dict[str, Any] = default_extra or {} def info(self, message: str, extra: Optional[Dict[str, Any]] = None) -> None: merged = {**self._default_extra, **(extra or {})} _send_log("info", message, merged) def warn(self, message: str, extra: Optional[Dict[str, Any]] = None) -> None: merged = {**self._default_extra, **(extra or {})} _send_log("warn", message, merged) def error(self, message: str, extra: Optional[Union[Dict[str, Any], BaseException]] = None) -> None: if isinstance(extra, BaseException): err_dict = _safe_value(extra) merged = {**self._default_extra, **err_dict} else: merged = {**self._default_extra, **(extra or {})} _send_log("error", message, merged) def debug(self, message: str, extra: Optional[Dict[str, Any]] = None) -> None: merged = {**self._default_extra, **(extra or {})} _send_log("debug", message, merged) def child(self, extra: Optional[Dict[str, Any]] = None) -> "Logger": """创建子 logger，合并固定的 extra 字段（通常传 tag）。""" return Logger({**self._default_extra, **(extra or {})}) # 全局 logger 实例 logger = Logger() # ────────────────────────────────────────────── # CLI 入口（用于测试） # ────────────────────────────────────────────── if __name__ == "__main__": import time set_log_info(version="1.0.0", source="test") set_caller() log = logger.child({"tag": "test_module"}) log.info("Python 日志模块测试", {"env": get_env(), "trace_id": _TRACE_ID}) log.warn("这是一条警告日志") try: raise ValueError("模拟错误") except Exception as e: log.error("捕获到异常", e) # 等待异步上报完成 time.sleep(2) print("日志测试完成")