155 lines
5.1 KiB
Python
155 lines
5.1 KiB
Python
import os
|
|
import re
|
|
import platform
|
|
import subprocess
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
SMB_SERVER = os.getenv("SMB_SERVER", "").strip()
|
|
SMB_SHARE = os.getenv("SMB_SHARE", "").strip()
|
|
SMB_USER = os.getenv("SMB_USER", "").strip()
|
|
SMB_PASSWORD = os.getenv("SMB_PASSWORD", "").strip()
|
|
SMB_SUBFOLDER = os.getenv("SMB_SUBFOLDER", "").strip()
|
|
SMB_MOUNT_MAC = os.getenv("SMB_MOUNT_MAC", "").strip()
|
|
SMB_MOUNT_LINUX = os.getenv("SMB_MOUNT_LINUX", "").strip()
|
|
SMB_MOUNT_WINDOWS = os.getenv("SMB_MOUNT_WINDOWS", "").strip()
|
|
|
|
|
|
def get_current_os():
|
|
system = platform.system().lower()
|
|
if system == "darwin":
|
|
return "mac"
|
|
if system == "windows":
|
|
return "windows"
|
|
return "linux"
|
|
|
|
|
|
def get_mount_point():
|
|
current_os = get_current_os()
|
|
mapping = {
|
|
"mac": SMB_MOUNT_MAC,
|
|
"linux": SMB_MOUNT_LINUX,
|
|
"windows": SMB_MOUNT_WINDOWS,
|
|
}
|
|
return mapping.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)
|
|
|
|
try:
|
|
if current_os == "mac":
|
|
cmd = (
|
|
f"mount_smbfs //'{SMB_USER}':'{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_USER},password={SMB_PASSWORD}"
|
|
)
|
|
subprocess.run(cmd, shell=True, check=True, capture_output=True)
|
|
elif current_os == "windows":
|
|
ruta_smb = f"\\\\{SMB_SERVER}\\{SMB_SHARE}"
|
|
cmd = [
|
|
"net", "use", mount_point, ruta_smb,
|
|
SMB_PASSWORD, f"/user:{SMB_USER}", "/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):
|
|
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 = SMB_SUBFOLDER.lstrip("/")
|
|
if subfolder:
|
|
ruta_destino = os.path.normpath(os.path.join(mount_point, subfolder))
|
|
else:
|
|
ruta_destino = mount_point
|
|
|
|
if not os.path.exists(ruta_destino):
|
|
os.makedirs(ruta_destino)
|
|
print(f" 📁 Carpeta creada: {ruta_destino}")
|
|
|
|
source = os.path.abspath(ruta_local)
|
|
destination = os.path.abspath(ruta_destino)
|
|
|
|
print(f" 🚀 Sincronizando a {current_os.upper()}: {destination}")
|
|
|
|
try:
|
|
if current_os in ("mac", "linux"):
|
|
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)
|
|
print(f" ✅ Archivos copiados a NAS")
|
|
subprocess.run(
|
|
f"find {source} -type f \\( -name '*.json' -o -name '*.zip' \\) -delete",
|
|
shell=True, check=True, capture_output=True, text=True,
|
|
)
|
|
print(f" 🧹 Limpieza de temporales completada")
|
|
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)
|
|
print(f" ✅ Archivos copiados a NAS")
|
|
for archivo in os.listdir(source):
|
|
if archivo.lower().endswith((".json", ".zip")):
|
|
os.remove(os.path.join(source, archivo))
|
|
print(f" 🧹 Limpieza de temporales completada")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f" ❌ Error en sincronización: {e}")
|
|
return False
|