Ferramentas/AjustaFundos/actions/file/json_file_saver.py

133 lines
4.4 KiB
Python

import json
import traceback
from datetime import datetime, date
from decimal import Decimal
from types import SimpleNamespace
from pathlib import Path
from typing import Any, Optional, Literal, Union
from pydantic import BaseModel
# ✔️ agora usando sua UI
from actions.ui.ui import ok, fail
class JsonFileSaver:
"""
Classe utilitária para salvar qualquer tipo de dado em disco,
automaticamente convertido para JSON se aplicável.
Suporta: dict, list, str, bytes, BaseModel, SimpleNamespace, Decimal, datetime, etc.
"""
def __init__(
self,
encoding: str = "utf-8",
indent: int = 4,
default_suffix: str = ".json",
):
self.encoding = encoding
self.indent = indent
self.default_suffix = default_suffix
# ------------------------------------------------------------
# Método público principal
# ------------------------------------------------------------
def save(
self,
data: Union[str, bytes, dict, list, Any],
file_path: str,
as_json: Optional[bool] = None,
add_timestamp: bool = False,
mode: Literal["overwrite", "append"] = "overwrite",
) -> dict:
"""
Salva o conteúdo em disco com serialização segura.
"""
try:
path = Path(file_path)
path.parent.mkdir(parents=True, exist_ok=True)
# Adiciona timestamp no nome, se solicitado
if add_timestamp:
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
stem = path.stem
suffix = path.suffix or self.default_suffix
path = path.with_name(f"{stem}_{timestamp}{suffix}")
# Detecta automaticamente se é JSON
if as_json is None:
as_json = isinstance(data, (dict, list, SimpleNamespace, BaseModel))
# Define o modo de escrita
write_mode = (
"wb" if isinstance(data, bytes) else "a" if mode == "append" else "w"
)
# Salva arquivo
if isinstance(data, bytes):
with open(path, write_mode) as f:
f.write(data)
elif as_json:
safe_data = self._convert(data)
with open(path, write_mode, encoding=self.encoding) as f:
json.dump(safe_data, f, ensure_ascii=False, indent=self.indent)
else:
if not isinstance(data, str):
data = str(data)
with open(path, write_mode, encoding=self.encoding) as f:
f.write(data)
file_size = path.stat().st_size if path.exists() else 0
# ✔️ substitui print() por UI
ok(f"Arquivo salvo: {path} ({file_size} bytes)")
return {
"success": True,
"path": str(path.resolve()),
"size": file_size,
"error": None,
}
except Exception as e:
error_message = (
f"Falha ao salvar arquivo '{file_path}': {e}\n"
f"{traceback.format_exc()}"
)
# ✔️ substitui print de erro por UI
fail(error_message)
return {
"success": False,
"path": str(file_path),
"size": 0,
"error": str(e),
}
# ------------------------------------------------------------
# Conversor recursivo interno
# ------------------------------------------------------------
def _convert(self, obj: Any) -> Any:
"""Converte qualquer tipo para formato serializável JSON."""
if obj is None:
return None
if isinstance(obj, (str, int, float, bool)):
return obj
if isinstance(obj, bytes):
return obj.decode("utf-8", errors="ignore")
if isinstance(obj, Decimal):
return float(obj)
if isinstance(obj, (datetime, date)):
return obj.isoformat()
if isinstance(obj, SimpleNamespace):
return {k: self._convert(v) for k, v in vars(obj).items()}
if isinstance(obj, BaseModel):
return obj.model_dump()
if isinstance(obj, list):
return [self._convert(i) for i in obj]
if isinstance(obj, dict):
return {k: self._convert(v) for k, v in obj.items()}
return str(obj)