528 lines
19 KiB
Python
528 lines
19 KiB
Python
#
|
||
# import datetime
|
||
# import io
|
||
# import logging
|
||
# import os
|
||
# import re
|
||
# import sys
|
||
# import shutil
|
||
# import zipfile
|
||
# from pathlib import Path
|
||
# import requests
|
||
#
|
||
#
|
||
# 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 目录下所有文件"""
|
||
#
|
||
# # 关闭所有 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()
|
||
#
|
||
# @classmethod
|
||
# def upload_all_logs(cls, server_url, token, userId, tenantId):
|
||
# log_path = Path(cls.logDir)
|
||
# if not log_path.exists():
|
||
# return False
|
||
#
|
||
# timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
# filename = f"{timestamp}_logs.zip"
|
||
# print(filename)
|
||
# zip_buf = io.BytesIO()
|
||
# with zipfile.ZipFile(zip_buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||
# for p in log_path.rglob("*"):
|
||
# if p.is_file():
|
||
# arcname = str(p.relative_to(log_path))
|
||
# zf.write(p, arcname=arcname)
|
||
#
|
||
# zip_bytes = zip_buf.getvalue()
|
||
#
|
||
# headers = {"vvtoken": token}
|
||
# data = {"tenantId": tenantId, "userId": userId}
|
||
#
|
||
#
|
||
# files = {
|
||
# "file": (filename, io.BytesIO(zip_bytes), "application/zip")
|
||
# }
|
||
#
|
||
# # 3) 上传
|
||
# resp = requests.post(server_url, headers=headers, data=data, files=files)
|
||
# if resp.json()['data']:
|
||
# return True
|
||
# return False
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# -*- coding: utf-8 -*-
|
||
import datetime
|
||
import io
|
||
import logging
|
||
import os
|
||
import re
|
||
import sys
|
||
import shutil
|
||
import zipfile
|
||
from pathlib import Path
|
||
import requests
|
||
|
||
# ========= 全局：强制 UTF-8（打包 EXE / 无控制台也生效） =========
|
||
def _force_utf8_everywhere():
|
||
os.environ.setdefault("PYTHONUTF8", "1")
|
||
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
||
# windowed 模式下 stdout/stderr 可能没有 buffer，这里做保护包装
|
||
try:
|
||
if getattr(sys.stdout, "buffer", None):
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if getattr(sys.stderr, "buffer", None):
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
# _force_utf8_everywhere()
|
||
|
||
|
||
# ========= 全局：强制 UTF-8 + 关闭缓冲（运行期立刻生效） =========
|
||
def _force_utf8_everywhere():
|
||
os.environ.setdefault("PYTHONUTF8", "1")
|
||
# 等价于 -u：让 stdout/stderr 无缓冲
|
||
os.environ.setdefault("PYTHONUNBUFFERED", "1")
|
||
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
||
|
||
# 若是 3.7+，优先用 reconfigure 实时改流
|
||
try:
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace",
|
||
line_buffering=True, write_through=True)
|
||
elif getattr(sys.stdout, "buffer", None):
|
||
# 退路：重新包一层，启用行缓冲 + 直写
|
||
sys.stdout = io.TextIOWrapper(
|
||
sys.stdout.buffer, encoding="utf-8",
|
||
errors="replace", line_buffering=True
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
if hasattr(sys.stderr, "reconfigure"):
|
||
sys.stderr.reconfigure(encoding="utf-8", errors="replace",
|
||
line_buffering=True, write_through=True)
|
||
elif getattr(sys.stderr, "buffer", None):
|
||
sys.stderr = io.TextIOWrapper(
|
||
sys.stderr.buffer, encoding="utf-8",
|
||
errors="replace", line_buffering=True
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
# ===========================================================
|
||
|
||
|
||
class LogManager:
|
||
"""
|
||
设备级与“设备+方法”级日志管理：
|
||
- log/<udid>/info.log | warning.log | error.log
|
||
- log/<udid>/<method>.log
|
||
- 文件统一 UTF-8 编码，避免 GBK/CP936 导致的 emoji 报错
|
||
- 提供 clearLogs() 与 upload_all_logs()
|
||
"""
|
||
# 运行根目录：打包后取 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")
|
||
_method_loggers = {} # 缓存“设备+方法”的 logger
|
||
|
||
# ---------- 工具：安全文本/文件名 ----------
|
||
|
||
@staticmethod
|
||
def _safe_text(obj) -> str:
|
||
"""把任意对象安全转为可写字符串（避免因编码问题再次抛异常）"""
|
||
try:
|
||
if isinstance(obj, bytes):
|
||
return obj.decode("utf-8", "replace")
|
||
s = str(obj)
|
||
# 确保解码上屏不再出错
|
||
_ = s.encode("utf-8", "replace")
|
||
return s
|
||
except Exception:
|
||
try:
|
||
return repr(obj)
|
||
except Exception:
|
||
return "<unprintable>"
|
||
|
||
@classmethod
|
||
def _safe_filename(cls, name: str, max_len: int = 80) -> str:
|
||
"""
|
||
将方法名/udid等转成安全文件名：
|
||
- 允许字母数字、点、下划线、连字符
|
||
- 保留常见 CJK 字符（中日韩）
|
||
- 其余替换为下划线；合并下划线；避免保留名；限长
|
||
"""
|
||
if not name:
|
||
return "unknown"
|
||
name = str(name).strip()
|
||
name = re.sub(r'[\\/:*?"<>|\r\n\t]+', '_', name) # Windows 非法字符
|
||
name = re.sub(
|
||
r'[^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"
|
||
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"
|
||
|
||
# ---------- 设备级固定文件：info/warning/error ----------
|
||
|
||
@classmethod
|
||
def _setupLogger(cls, udid, name, logName, level=logging.INFO):
|
||
"""创建或获取 logger，并绑定到设备目录下的固定文件（info.log / warning.log / error.log）"""
|
||
udid_key = cls._safe_filename(udid or "system")
|
||
deviceLogDir = os.path.join(cls.logDir, udid_key)
|
||
os.makedirs(deviceLogDir, exist_ok=True)
|
||
logFile = os.path.join(deviceLogDir, logName)
|
||
|
||
logger_name = f"{udid_key}_{name}"
|
||
logger = logging.getLogger(logger_name)
|
||
logger.setLevel(level)
|
||
logger.propagate = False # 不向根 logger 传播，避免重复
|
||
|
||
# 避免重复添加同一路径的 file handler
|
||
abs_target = os.path.abspath(logFile)
|
||
for h in logger.handlers:
|
||
if isinstance(h, logging.FileHandler) and getattr(h, "baseFilename", "") == abs_target:
|
||
return logger
|
||
|
||
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"):
|
||
msg = cls._safe_text(f"[{udid}] {text}")
|
||
cls._setupLogger(udid, "infoLogger", "info.log", level=logging.INFO).info(msg)
|
||
|
||
@classmethod
|
||
def warning(cls, text, udid="system"):
|
||
msg = cls._safe_text(f"[{udid}] {text}")
|
||
cls._setupLogger(udid, "warningLogger", "warning.log", level=logging.WARNING).warning(msg)
|
||
|
||
@classmethod
|
||
def error(cls, text, udid="system"):
|
||
msg = cls._safe_text(f"[{udid}] {text}")
|
||
cls._setupLogger(udid, "errorLogger", "error.log", level=logging.ERROR).error(msg)
|
||
|
||
# ---------- “设备+方法”独立文件：<udid>/<method>.log ----------
|
||
|
||
@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)
|
||
|
||
# 命中缓存
|
||
logger = cls._method_loggers.get(cache_key)
|
||
if logger:
|
||
return logger
|
||
|
||
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
|
||
|
||
abs_target = os.path.abspath(logFile)
|
||
for h in logger.handlers:
|
||
if isinstance(h, logging.FileHandler) and getattr(h, "baseFilename", "") == abs_target:
|
||
cls._method_loggers[cache_key] = logger
|
||
return logger
|
||
|
||
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"):
|
||
msg = cls._safe_text(f"[{udid}][{method}] {text}")
|
||
cls._setupMethodLogger(udid, method, level=logging.INFO).info(msg)
|
||
|
||
@classmethod
|
||
def method_warning(cls, text, method, udid="system"):
|
||
msg = cls._safe_text(f"[{udid}][{method}] {text}")
|
||
cls._setupMethodLogger(udid, method, level=logging.WARNING).warning(msg)
|
||
|
||
@classmethod
|
||
def method_error(cls, text, method, udid="system"):
|
||
msg = cls._safe_text(f"[{udid}][{method}] {text}")
|
||
cls._setupMethodLogger(udid, method, level=logging.ERROR).error(msg)
|
||
|
||
# ---------- 清空日志 ----------
|
||
|
||
@classmethod
|
||
def clearLogs(cls):
|
||
"""启动时清空 log 目录下所有文件"""
|
||
# 先关闭所有 logger 的文件句柄
|
||
for _, logger in logging.Logger.manager.loggerDict.items():
|
||
if isinstance(logger, logging.Logger):
|
||
for handler in list(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():
|
||
try:
|
||
if item.is_file():
|
||
item.unlink(missing_ok=True) # py>=3.8
|
||
elif item.is_dir():
|
||
shutil.rmtree(item, ignore_errors=True)
|
||
except Exception:
|
||
# 不阻塞清理
|
||
pass
|
||
|
||
cls._method_loggers.clear()
|
||
|
||
# ---------- 上传所有日志（内存打包 zip） ----------
|
||
|
||
@classmethod
|
||
def upload_all_logs(cls, server_url, token, userId, tenantId):
|
||
"""
|
||
将 log/ 目录下所有日志打包为 zip（内存），上传至服务器：
|
||
- headers: {"vvtoken": <token>}
|
||
- form: {"tenantId": <tenantId>, "userId": <userId>, "file": <zip>}
|
||
返回 True/False
|
||
"""
|
||
try:
|
||
log_path = Path(cls.logDir)
|
||
if not log_path.exists():
|
||
logging.info("[upload_all_logs] 日志目录不存在：%s", log_path)
|
||
return False
|
||
|
||
# 文件名仅用于表单，不落盘，可包含安全字符
|
||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
filename = f"{timestamp}_logs.zip"
|
||
logging.info("[upload_all_logs] 打包文件名：%s", filename)
|
||
|
||
# 1) 内存中打包 zip
|
||
zip_buf = io.BytesIO()
|
||
with zipfile.ZipFile(zip_buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||
for p in log_path.rglob("*"):
|
||
if p.is_file():
|
||
arcname = str(p.relative_to(log_path))
|
||
zf.write(p, arcname=arcname)
|
||
zip_bytes = zip_buf.getvalue()
|
||
|
||
# 2) 组织请求
|
||
headers = {"vvtoken": token} if token else {}
|
||
data = {"tenantId": tenantId, "userId": userId}
|
||
files = {"file": (filename, io.BytesIO(zip_bytes), "application/zip")}
|
||
|
||
# 3) 上传
|
||
resp = requests.post(server_url, headers=headers, data=data, files=files, timeout=120)
|
||
ok = False
|
||
try:
|
||
js = resp.json()
|
||
ok = bool(js.get("data"))
|
||
except Exception:
|
||
# 响应不是 JSON，也许是纯文本；降级按状态码判断
|
||
ok = resp.ok
|
||
|
||
if ok:
|
||
logging.info("[upload_all_logs] 上传成功：%s", server_url)
|
||
return True
|
||
else:
|
||
logging.error("[upload_all_logs] 上传失败，status=%s, text=%s", resp.status_code, LogManager._safe_text(resp.text))
|
||
return False
|
||
|
||
except Exception as e:
|
||
logging.error("[upload_all_logs] 异常：%s", LogManager._safe_text(e))
|
||
return False
|