iOSAI/Utils/LogManager.py

# import logging
# import os
# import sys
# import shutil
# from pathlib import Path
#
#
# class LogManager:
#     # 运行根目录：打包后取 exe 目录；源码运行取项目目录
#     if getattr(sys, "frozen", False):
#         projectRoot = os.path.dirname(sys.executable)
#     else:
#         projectRoot = os.path.dirname(os.path.dirname(__file__))
#
#     logDir = os.path.join(projectRoot, "log")
#     _loggers = {}
#
#     @classmethod
#     def _setupLogger(cls, udid, name, logName, level=logging.INFO):
#         """创建或获取 logger，并绑定到文件"""
#         deviceLogDir = os.path.join(cls.logDir, udid)
#         os.makedirs(deviceLogDir, exist_ok=True)
#         logFile = os.path.join(deviceLogDir, logName)
#
#         logger_name = f"{udid}_{name}"
#         logger = logging.getLogger(logger_name)
#         logger.setLevel(level)
#
#         # 避免重复添加 handler
#         if not any(
#             isinstance(h, logging.FileHandler) and h.baseFilename == os.path.abspath(logFile)
#             for h in logger.handlers
#         ):
#             fileHandler = logging.FileHandler(logFile, mode="a", encoding="utf-8")
#             formatter = logging.Formatter(
#                 "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
#                 datefmt="%Y-%m-%d %H:%M:%S"
#             )
#             fileHandler.setFormatter(formatter)
#             logger.addHandler(fileHandler)
#
#         return logger
#
#     @classmethod
#     def info(cls, text, udid="system"):
#         cls._setupLogger(udid, "infoLogger", "info.log", level=logging.INFO).info(f"[{udid}] {text}")
#
#     @classmethod
#     def warning(cls, text, udid="system"):
#         cls._setupLogger(udid, "warningLogger", "warning.log", level=logging.WARNING).warning(f"[{udid}] {text}")
#
#     @classmethod
#     def error(cls, text, udid="system"):
#         cls._setupLogger(udid, "errorLogger", "error.log", level=logging.ERROR).error(f"[{udid}] {text}")
#
#     # 清空日志
#     @classmethod
#     def clearLogs(cls):
#         """启动时清空 log 目录下所有文件"""
#         print("开始清空日志...")
#
#         # 关闭所有 handler
#         for name, logger in logging.Logger.manager.loggerDict.items():
#             if isinstance(logger, logging.Logger):
#                 for handler in logger.handlers[:]:
#                     try:
#                         handler.close()
#                     except Exception:
#                         pass
#                     logger.removeHandler(handler)
#
#         # 仅删除目录里的所有文件和子目录
#         log_path = Path(cls.logDir)
#         if log_path.exists():
#             for item in log_path.iterdir():
#                 if item.is_file():
#                     item.unlink()
#                 elif item.is_dir():
#                     import shutil
#                     shutil.rmtree(item)
#
#         print("日志清空完成")


import logging
import os
import re
import sys
import shutil
from pathlib import Path

import requests

from Entity.Variables import token


class LogManager:
    # 运行根目录：打包后取 exe 目录；源码运行取项目目录
    if getattr(sys, "frozen", False):
        projectRoot = os.path.dirname(sys.executable)
    else:
        projectRoot = os.path.dirname(os.path.dirname(__file__))

    logDir = os.path.join(projectRoot, "log")
    _loggers = {}
    _method_loggers = {}  # 新增：缓存“设备+方法”的 logger

    # ---------- 工具函数 ----------
    @classmethod
    def _safe_filename(cls, name: str, max_len: int = 80) -> str:
        """
        将方法名/udid等转成安全文件名：
        - 允许字母数字、点、下划线、连字符
        - 允许常见 CJK 字符（中日韩）
        - 其他非法字符替换为下划线
        - 合并多余下划线，裁剪长度
        """
        if not name:
            return "unknown"
        name = str(name).strip()

        # 替换 Windows 非法字符和控制符
        name = re.sub(r'[\\/:*?"<>|\r\n\t]+', '_', name)

        # 只保留 ① 英数._- ② CJK 统一表意文字、日文平/片假名、韩文音节
        name = re.sub(rf'[^a-zA-Z0-9_.\-'
                      r'\u4e00-\u9fff'  # 中
                      r'\u3040-\u30ff'  # 日
                      r'\uac00-\ud7a3'  # 韩
                      r']+', '_', name)
        # 合并多余下划线，去两端空白与下划线
        name = re.sub(r'_+', '_', name).strip(' _.')
        # 避免空
        name = name or "unknown"
        # Windows 预留名避免（CON/PRN/AUX/NUL/COM1…）
        if re.fullmatch(r'(?i)(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])', name):
            name = f"_{name}"
        # 限长
        return name[:max_len] or "unknown"

    # ---------- 旧的：按级别写固定文件 ----------
    @classmethod
    def _setupLogger(cls, udid, name, logName, level=logging.INFO):
        """创建或获取 logger，并绑定到设备目录下的固定文件（info.log / warning.log / error.log）"""
        deviceLogDir = os.path.join(cls.logDir, cls._safe_filename(udid))
        os.makedirs(deviceLogDir, exist_ok=True)
        logFile = os.path.join(deviceLogDir, logName)

        logger_name = f"{udid}_{name}"
        logger = logging.getLogger(logger_name)
        logger.setLevel(level)

        # 避免重复添加 handler
        if not any(
                isinstance(h, logging.FileHandler) and h.baseFilename == os.path.abspath(logFile)
                for h in logger.handlers
        ):
            fileHandler = logging.FileHandler(logFile, mode="a", encoding="utf-8")
            formatter = logging.Formatter(
                "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S"
            )
            fileHandler.setFormatter(formatter)
            logger.addHandler(fileHandler)

        return logger

    @classmethod
    def info(cls, text, udid="system"):
        cls._setupLogger(udid, "infoLogger", "info.log", level=logging.INFO).info(f"[{udid}] {text}")

    @classmethod
    def warning(cls, text, udid="system"):
        cls._setupLogger(udid, "warningLogger", "warning.log", level=logging.WARNING).warning(f"[{udid}] {text}")

    @classmethod
    def error(cls, text, udid="system"):
        cls._setupLogger(udid, "errorLogger", "error.log", level=logging.ERROR).error(f"[{udid}] {text}")

    # ---------- 新增：按“设备+方法”分别写独立日志文件 ----------
    @classmethod
    def _setupMethodLogger(cls, udid: str, method: str, level=logging.INFO):
        """
        为某设备的某个方法单独创建 logger：
        log/<udid>/<method>.log
        """
        udid_key = cls._safe_filename(udid or "system")
        method_key = cls._safe_filename(method or "general")
        cache_key = (udid_key, method_key)

        # 命中缓存
        if cache_key in cls._method_loggers:
            return cls._method_loggers[cache_key]

        deviceLogDir = os.path.join(cls.logDir, udid_key)
        os.makedirs(deviceLogDir, exist_ok=True)
        logFile = os.path.join(deviceLogDir, f"{method_key}.log")

        logger_name = f"{udid_key}.{method_key}"
        logger = logging.getLogger(logger_name)
        logger.setLevel(level)
        logger.propagate = False  # 避免向根 logger 传播导致控制台重复打印

        # 避免重复添加 handler
        if not any(
                isinstance(h, logging.FileHandler) and h.baseFilename == os.path.abspath(logFile)
                for h in logger.handlers
        ):
            fileHandler = logging.FileHandler(logFile, mode="a", encoding="utf-8")
            formatter = logging.Formatter(
                "%(asctime)s - %(levelname)s - %(name)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S"
            )
            fileHandler.setFormatter(formatter)
            logger.addHandler(fileHandler)

        cls._method_loggers[cache_key] = logger
        return logger

    @classmethod
    def method_info(cls, text, method, udid="system"):
        """按设备+方法写 INFO 到 log/<udid>/<method>.log"""
        cls._setupMethodLogger(udid, method, level=logging.INFO).info(f"[{udid}][{method}] {text}")

    @classmethod
    def method_warning(cls, text, method, udid="system"):
        cls._setupMethodLogger(udid, method, level=logging.WARNING).warning(f"[{udid}][{method}] {text}")

    @classmethod
    def method_error(cls, text, method, udid="system"):
        cls._setupMethodLogger(udid, method, level=logging.ERROR).error(f"[{udid}][{method}] {text}")

    # 清空日志
    @classmethod
    def clearLogs(cls):
        """启动时清空 log 目录下所有文件"""
        print("开始清空日志...")

        # 关闭所有 handler
        for name, logger in logging.Logger.manager.loggerDict.items():
            if isinstance(logger, logging.Logger):
                for handler in logger.handlers[:]:
                    try:
                        handler.close()
                    except Exception:
                        pass
                    logger.removeHandler(handler)

        # 删除 log 目录
        log_path = Path(cls.logDir)
        if log_path.exists():
            for item in log_path.iterdir():
                if item.is_file():
                    item.unlink()
                elif item.is_dir():
                    shutil.rmtree(item)

        # 清缓存
        cls._method_loggers.clear()
        print("日志清空完成")

    @classmethod
    def upload_all_logs(cls, server_url: str, extra_data: dict = None):
        """
        上传 log/ 目录下所有文件和子目录中的日志
        """
        log_path = Path(cls.logDir)
        if not log_path.exists():
            print("[upload_all_logs] 日志目录不存在")
            return False

        success_files, failed_files = [], []

        for file in log_path.rglob("*.log"):  # 递归找到所有 .log 文件
            try:

                files = {"file": open(file, "rb")}
                headers = {
                    "Authorization": f"Bearer {token}",
                }

                data = {
                    # "relative_path": str(file.relative_to(log_path))
                    "tenantId": 1,
                    "userId": 1,
                }

                if extra_data:
                    data.update(extra_data)
                resp = requests.post(server_url, files=files, data=data, headers=headers ,timeout=15)

                print(resp.text)
                print(resp.status_code)

                if resp.status_code == 200:
                    success_files.append(file.name)
                else:
                    failed_files.append((file.name, resp.status_code))
            except Exception as e:
                failed_files.append((file.name, str(e)))
            finally:
                try:
                    files["file"].close()
                except:
                    pass

        print(f"[upload_all_logs] 成功上传: {success_files}")
        if failed_files:
            print(f"[upload_all_logs] 失败文件: {failed_files}")

        return len(failed_files) == 0