144 lines
5.0 KiB
Python
144 lines
5.0 KiB
Python
import os
|
||
import platform
|
||
import subprocess
|
||
from config_loader import CONFIG
|
||
|
||
# ========================================================================
|
||
def get_current_os():
|
||
if os.path.exists("/.dockerenv"):
|
||
return "docker"
|
||
|
||
system = platform.system().lower()
|
||
if system == "darwin":
|
||
return "mac"
|
||
if system == "windows":
|
||
return "windows"
|
||
return "linux"
|
||
|
||
# ========================================================================
|
||
def get_mount_point():
|
||
return CONFIG.smb.mount_points.get(get_current_os(), "")
|
||
|
||
|
||
def esta_montado(mount_point):
|
||
if not mount_point:
|
||
return False
|
||
if get_current_os() == "windows":
|
||
return os.path.exists(mount_point)
|
||
return os.path.ismount(mount_point) or os.path.exists(
|
||
os.path.join(mount_point, ".mounted_check")
|
||
)
|
||
|
||
# ========================================================================
|
||
def montar_smb():
|
||
mount_point = get_mount_point()
|
||
if not mount_point:
|
||
print(" ⚠️ No hay punto de montaje configurado para este SO.")
|
||
return False
|
||
|
||
if esta_montado(mount_point):
|
||
return True
|
||
|
||
current_os = get_current_os()
|
||
|
||
if current_os != "windows" and not os.path.exists(mount_point):
|
||
os.makedirs(mount_point)
|
||
|
||
smb = CONFIG.smb
|
||
|
||
try:
|
||
if current_os == "mac":
|
||
cmd = (
|
||
f"mount_smbfs //'{smb.username}':'{smb.password}'"
|
||
f"@{smb.server}/{smb.share} {mount_point}"
|
||
)
|
||
subprocess.run(cmd, shell=True, check=True, capture_output=True)
|
||
elif current_os == "linux":
|
||
cmd = (
|
||
f"sudo mount -t cifs //{smb.server}/{smb.share} {mount_point}"
|
||
f" -o username={smb.username},password={smb.password}"
|
||
)
|
||
subprocess.run(cmd, shell=True, check=True, capture_output=True)
|
||
|
||
elif current_os == "docker":
|
||
print("ℹ️ Docker: SMB debe estar montado por el host")
|
||
return os.path.exists(mount_point)
|
||
|
||
elif current_os == "windows":
|
||
ruta_smb = f"\\\\{smb.server}\\{smb.share}"
|
||
cmd = [
|
||
"net", "use", mount_point, ruta_smb,
|
||
smb.password, f"/user:{smb.username}", "/persistent:no",
|
||
]
|
||
subprocess.run(cmd, check=True, capture_output=True)
|
||
|
||
print(f" ✅ SMB montado en: {mount_point}")
|
||
return True
|
||
except Exception as e:
|
||
print(f" ⚠️ Error montando SMB: {e}")
|
||
return False
|
||
|
||
# ========================================================================
|
||
def sincronizar_a_smb(ruta_local, subfolder_cuenta):
|
||
if not os.path.isdir(ruta_local):
|
||
print(f" ⚠️ La carpeta local no existe: {ruta_local}")
|
||
return False
|
||
|
||
if not montar_smb():
|
||
print(" ❌ No se pudo sincronizar: SMB no montado.")
|
||
return False
|
||
|
||
mount_point = get_mount_point()
|
||
current_os = get_current_os()
|
||
|
||
if current_os == "windows" and mount_point.endswith(":") and len(mount_point) == 2:
|
||
mount_point += "\\"
|
||
|
||
subfolder = subfolder_cuenta.lstrip("/")
|
||
ruta_destino = os.path.normpath(os.path.join(mount_point, subfolder))
|
||
|
||
if not os.path.exists(ruta_destino):
|
||
os.makedirs(ruta_destino)
|
||
print(f" 📁 Carpeta creada en SMB: {ruta_destino}")
|
||
|
||
source = os.path.abspath(ruta_local)
|
||
destination = os.path.abspath(ruta_destino)
|
||
|
||
print(f" 🚀 Sincronizando {current_os.upper()}: {source} -> {destination}")
|
||
|
||
try:
|
||
if current_os in ("mac", "linux", "docker"):
|
||
# Solo sincronizar PDF y XML
|
||
cmd = (
|
||
f"find {source} -maxdepth 1 \\( -name '*.pdf' -o -name '*.xml' \\) "
|
||
f"-exec rsync -av --remove-source-files {{}} {destination}/ \\;"
|
||
)
|
||
subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
|
||
# Limpiar temporales restantes (json, zip)
|
||
subprocess.run(
|
||
f"find {source} -type f \\( -name '*.json' -o -name '*.zip' \\) -delete",
|
||
shell=True, check=True, capture_output=True, text=True,
|
||
)
|
||
|
||
elif current_os == "windows":
|
||
import shutil
|
||
for archivo in os.listdir(source):
|
||
if not archivo.lower().endswith((".pdf", ".xml")):
|
||
continue
|
||
src_path = os.path.join(source, archivo)
|
||
dst_path = os.path.join(destination, archivo)
|
||
if os.path.isfile(src_path):
|
||
if os.path.exists(dst_path):
|
||
os.remove(dst_path)
|
||
shutil.move(src_path, dst_path)
|
||
|
||
for archivo in os.listdir(source):
|
||
if archivo.lower().endswith((".json", ".zip")):
|
||
os.remove(os.path.join(source, archivo))
|
||
|
||
print(f" ✅ Sincronización completada.")
|
||
return True
|
||
except Exception as e:
|
||
print(f" ❌ Error en sincronización: {e}")
|
||
return False
|
||
# ======================================================================== |