proyecto cargado en Docker y ejecutándose con crontrab en macpro
This commit is contained in:
George committed
1 parent
1503d6a2ec
commit
2c2c88bfd6
21 files changed
+1068
-754
No files matched your search
+182
@@ -0,0 +1,182 @@
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from dotenv import load_dotenv
|
||||
from logger_manager import get_logger
|
||||
|
||||
# ============================================================================
|
||||
load_dotenv()
|
||||
|
||||
logger = get_logger("smb")
|
||||
|
||||
# ============================================================================
|
||||
def get_mount_base() -> str:
|
||||
"""Base de montaje SMB según la plataforma (.env)."""
|
||||
if sys.platform == "darwin":
|
||||
return os.getenv("SMB_MOUNT_MAC", "/Volumes/FileSAP")
|
||||
return os.getenv("SMB_MOUNT_LINUX", "/mnt/filesap")
|
||||
|
||||
|
||||
def _creds() -> dict:
|
||||
return {
|
||||
"server": os.getenv("SMB_SERVER", ""),
|
||||
"share": os.getenv("SMB_SHARE", ""),
|
||||
"user": os.getenv("SMB_USER", ""),
|
||||
"password": os.getenv("SMB_PASSWORD", ""),
|
||||
}
|
||||
|
||||
# ============================================================================
|
||||
def detectar_montaje(mount_base: str) -> bool:
|
||||
"""Verifica que el punto de montaje exista y esté montado de verdad."""
|
||||
p = Path(mount_base)
|
||||
if not p.is_dir():
|
||||
return False
|
||||
if os.path.ismount(mount_base):
|
||||
return True
|
||||
try:
|
||||
return any(p.iterdir())
|
||||
except PermissionError:
|
||||
return False
|
||||
|
||||
# ============================================================================
|
||||
def convertir_unc_a_local(ruta_unc: str, mount_base: str | None = None) -> str | None:
|
||||
"""Convierte una ruta UNC (\\servidor\\recurso\\carpeta\\archivo.pdf) a ruta
|
||||
local sobre el montaje SMB actual."""
|
||||
ruta_unc = (ruta_unc or "").strip()
|
||||
if len(ruta_unc) < 5:
|
||||
return None
|
||||
base = mount_base or get_mount_base()
|
||||
if not base:
|
||||
return None
|
||||
path = ruta_unc.replace("\\", "/").lstrip("/")
|
||||
partes = path.split("/")
|
||||
if len(partes) < 2:
|
||||
return None
|
||||
ruta_relativa = "/".join(partes[2:])
|
||||
return os.path.join(base, ruta_relativa)
|
||||
|
||||
# ============================================================================
|
||||
def montar_smb(mount_base: str) -> tuple[bool, str]:
|
||||
"""Intenta montar la unidad SMB sin sudo (usuario con privilegios)."""
|
||||
creds = _creds()
|
||||
if not creds["server"] or not creds["share"]:
|
||||
return False, "Faltan SMB_SERVER/SMB_SHARE en .env."
|
||||
|
||||
Path(mount_base).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
try:
|
||||
if sys.platform == "darwin":
|
||||
url = f"//{creds['user']}:{creds['password']}@{creds['server']}/{creds['share']}"
|
||||
result = subprocess.run(
|
||||
["mount_smbfs", url, mount_base],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
else:
|
||||
opts = f"username={creds['user']},password={creds['password']}"
|
||||
result = subprocess.run(
|
||||
["mount", "-t", "cifs", f"//{creds['server']}/{creds['share']}", mount_base, "-o", opts],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
return False, "Comando de montaje no disponible en el sistema."
|
||||
|
||||
if result.returncode != 0:
|
||||
return False, result.stderr.strip() or "Error montando SMB (código %d)." % result.returncode
|
||||
|
||||
return True, f"Unidad montada en {mount_base}."
|
||||
|
||||
# ============================================================================
|
||||
def _rsync_cmd(destino: str) -> list[str]:
|
||||
return [
|
||||
"rsync", "-rtv", "--progress",
|
||||
"--remove-source-files",
|
||||
"--include=*.pdf",
|
||||
"--include=*.xml",
|
||||
"--exclude=*",
|
||||
"./",
|
||||
f"{destino}/",
|
||||
]
|
||||
|
||||
# ============================================================================
|
||||
def transferir_empresa(empresa: dict, mount_base: str, auto_mount: bool = False) -> dict:
|
||||
"""Transfiere downloads/<nombre>/ hacia <mount_base>/<smb_subfolder>/ con rsync.
|
||||
|
||||
Si el SMB no está montado: si auto_mount está habilitado se intenta montar;
|
||||
si falla (o no está habilitado), se conservan los archivos localmente.
|
||||
"""
|
||||
nombre = empresa["nombre"]
|
||||
origen = Path("downloads") / nombre
|
||||
|
||||
if not origen.is_dir():
|
||||
return {"empresa": nombre, "estado": "sin_descargas", "movidos": 0}
|
||||
|
||||
if not detectar_montaje(mount_base):
|
||||
if auto_mount:
|
||||
ok, msg = montar_smb(mount_base)
|
||||
if ok:
|
||||
logger.info(f" 🔌 {msg}")
|
||||
else:
|
||||
logger.warning(f" ⚠️ No se pudo montar SMB ({mount_base}): {msg}. "
|
||||
f"Archivos conservados en {origen}.")
|
||||
return {"empresa": nombre, "estado": "smb_no_montado", "movidos": 0}
|
||||
else:
|
||||
logger.warning(f" ⚠️ Empresa '{nombre}': SMB NO montado ({mount_base}). "
|
||||
f"Transferencia omitida, archivos conservados en {origen}.")
|
||||
return {"empresa": nombre, "estado": "smb_no_montado", "movidos": 0}
|
||||
|
||||
destino = Path(mount_base) / empresa["smb_subfolder"]
|
||||
try:
|
||||
destino.mkdir(parents=True, exist_ok=True)
|
||||
except Exception as e:
|
||||
logger.warning(f" ⚠️ Empresa '{nombre}': no se pudo crear {destino}: {e}. "
|
||||
f"Transferencia omitida, archivos conservados.")
|
||||
return {"empresa": nombre, "estado": "error_destino", "movidos": 0}
|
||||
|
||||
logger.info(f" 🔁 Empresa '{nombre}': rsync {origen}/ -> {destino}/")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
_rsync_cmd(str(destino)),
|
||||
cwd=str(origen),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
logger.error(" ❌ rsync no está instalado en el sistema.")
|
||||
return {"empresa": nombre, "estado": "rsync_faltante", "movidos": 0}
|
||||
|
||||
if result.returncode != 0:
|
||||
logger.error(f" ❌ Error rsync ({nombre}):\n{result.stderr.strip()}")
|
||||
return {"empresa": nombre, "estado": "error_rsync", "movidos": 0}
|
||||
|
||||
restantes = [f for f in origen.iterdir() if f.is_file()]
|
||||
movidos = len(restantes) == 0
|
||||
estado = "ok" if movidos else "parcial"
|
||||
logger.info(f" ✅ Transferencia {estado} ({nombre}). Quedan {len(restantes)} archivos locales.")
|
||||
return {"empresa": nombre, "estado": estado, "movidos": movidos}
|
||||
|
||||
# ============================================================================
|
||||
def transferir_a_smb(empresas: list[dict], mount_base: str, auto_mount: bool = False) -> list[dict]:
|
||||
resultados = []
|
||||
for emp in empresas:
|
||||
if emp.get("locked", False):
|
||||
logger.info(f" ⏭️ Empresa '{emp['nombre']}' bloqueada (LOCKED), omitiendo transferencia.")
|
||||
continue
|
||||
resultados.append(transferir_empresa(emp, mount_base, auto_mount))
|
||||
return resultados
|
||||
|
||||
# ============================================================================
|
||||
if __name__ == "__main__":
|
||||
import json
|
||||
|
||||
base = get_mount_base()
|
||||
if "--mount" in sys.argv:
|
||||
ok, msg = montar_smb(base)
|
||||
print(f"{'✅' if ok else '❌'} {msg}")
|
||||
else:
|
||||
cfg = json.load(open(Path(__file__).parent / "config.json", encoding="utf-8"))
|
||||
activas = [e for e in cfg.get("empresas", []) if not e.get("locked", False)]
|
||||
auto = os.getenv("SMB_AUTO_MOUNT", "false").lower() == "true"
|
||||
for r in transferir_a_smb(activas, base, auto):
|
||||
print(r)
|
||||
# ============================================================================
|
||||
Reference in new issue
Block a user