Miguel
/
AutoBackups
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor project management and backup functionalities; add API tests and update configuration handling

This commit is contained in:
Miguel 2025-09-02 09:20:29 +02:00
parent b9aed346ab
commit b5ec940868
9 changed files with 559 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.github/copilot-instructions.md vendored Normal file
View File

@ -0,0 +1,6 @@
- we are using conda environment: autobackups.
- so to test use: conda activate autobackups ; python src/app.py
- do not use fallbacks if there is not requested
- all comments in the software please in english

12
config.json
View File

@ -5,18 +5,6 @@
"type": "siemens_s7",
"enabled": true,
"description": "Directorio principal de proyectos Siemens"
},
{
"path": "D:\\Engineering\\Projects",
"type": "siemens_s7",
"enabled": true,
"description": "Proyectos de ingeniería adicionales"
},
{
"path": "C:\\Important\\Documentation",
"type": "manual",
"enabled": true,
"description": "Documentación importante para backup manual"
}
],
"backup_destination": "D:\\Backups\\AutoBackups",

103
projects.json
View File

@ -1,39 +1,39 @@
{
"metadata": {
"version": "1.0",
"last_updated": "2025-09-01T15:49:25.929290+00:00",
"total_projects": 2
"last_updated": "2025-09-02T07:15:16.655495+00:00",
"total_projects": 3
},
"projects": [
{
"id": "example_project_001",
"name": "PLC_MainLine_Example",
"path": "C:\\Projects\\Siemens\\LineA\\Project1",
"id": "project_cbe604d1_20250901_180135",
"name": "Ssae0452 Last Version Walter",
"path": "C:\\Users\\migue\\Downloads\\TestBackups\\Ssae0452 Last Version Walter",
"type": "siemens_s7",
"s7p_file": "C:\\Projects\\Siemens\\LineA\\Project1\\project.s7p",
"observation_directory": "C:\\Projects\\Siemens",
"relative_path": "LineA\\Project1",
"backup_path": "LineA\\Project1",
"s7p_file": "C:\\Users\\migue\\Downloads\\TestBackups\\Ssae0452 Last Version Walter\\Ssae0452.s7p",
"observation_directory": "C:\\Users\\migue\\Downloads\\TestBackups",
"relative_path": "Ssae0452 Last Version Walter",
"backup_path": "Ssae0452 Last Version Walter",
"schedule_config": {
"schedule": "daily",
"schedule_time": "02:00",
"enabled": true,
"next_scheduled_backup": "2025-09-02T02:00:00Z"
"next_scheduled_backup": ""
},
"backup_history": {
"last_backup_date": "2025-09-01T02:15:30Z",
"last_backup_file": "D:\\Backups\\AutoBackups\\LineA\\Project1\\2025-09-01\\02-15-30_projects.zip",
"backup_count": 5,
"last_successful_backup": "2025-09-01T02:15:30Z"
"last_backup_date": "",
"last_backup_file": "",
"backup_count": 0,
"last_successful_backup": ""
},
"hash_info": {
"last_s7p_hash": "abc123def456789",
"last_full_hash": "def789ghi012345",
"last_s7p_timestamp": "2025-08-31T14:30:00Z",
"last_s7p_size": 2048576,
"last_scan_timestamp": "2025-09-01T02:10:00Z",
"file_count": 1247,
"total_size_bytes": 125847296
"last_s7p_hash": "",
"last_full_hash": "",
"last_s7p_timestamp": "",
"last_s7p_size": 0,
"last_scan_timestamp": "",
"file_count": 0,
"total_size_bytes": 0
},
"status": {
"current_status": "ready",
@ -42,16 +42,61 @@
"next_retry": null,
"files_in_use": false,
"exclusivity_check_passed": true,
"last_status_update": "2025-09-01T02:15:35Z"
"last_status_update": ""
},
"discovery_info": {
"discovered_date": "2025-09-01T08:30:15Z",
"discovery_method": "everything_api",
"discovered_date": "",
"discovery_method": "",
"auto_discovered": true
}
},
{
"id": "",
"id": "project_e63eea24_20250901_180135",
"name": "Ssae04_11 - compiled",
"path": "C:\\Users\\migue\\Downloads\\TestBackups\\LineaB\\Ssae04_11 - compiled",
"type": "siemens_s7",
"s7p_file": "C:\\Users\\migue\\Downloads\\TestBackups\\LineaB\\Ssae04_11 - compiled\\Ssae0452.s7p",
"observation_directory": "C:\\Users\\migue\\Downloads\\TestBackups",
"relative_path": "LineaB\\Ssae04_11 - compiled",
"backup_path": "LineaB\\Ssae04_11 - compiled",
"schedule_config": {
"schedule": "daily",
"schedule_time": "02:00",
"enabled": true,
"next_scheduled_backup": ""
},
"backup_history": {
"last_backup_date": "2025-09-02T07:19:34.159415+00:00",
"last_backup_file": "D:\\Backups\\AutoBackups\\LineaB\\Ssae04_11 - compiled\\2025-09-02\\09-19-32_projects.zip",
"backup_count": 1,
"last_successful_backup": "2025-09-02T07:19:34.159415+00:00"
},
"hash_info": {
"last_s7p_hash": "",
"last_full_hash": "",
"last_s7p_timestamp": "",
"last_s7p_size": 0,
"last_scan_timestamp": "",
"file_count": 0,
"total_size_bytes": 0
},
"status": {
"current_status": "ready",
"last_error": null,
"retry_count": 0,
"next_retry": null,
"files_in_use": false,
"exclusivity_check_passed": true,
"last_status_update": "2025-09-02T07:19:34.159415+00:00"
},
"discovery_info": {
"discovered_date": "",
"discovery_method": "",
"auto_discovered": true
}
},
{
"id": "project_c0be5aea_20250901_180135",
"name": "Ssae04_14 - TIA",
"path": "C:\\Users\\migue\\Downloads\\TestBackups\\LineaB\\Ssae04_14 - TIA",
"type": "siemens_s7",
@ -97,10 +142,10 @@
}
],
"statistics": {
"total_backups_created": 15,
"total_backup_size_mb": 2450.5,
"average_backup_time_seconds": 45.2,
"last_global_scan": "2025-09-01T08:30:15Z",
"total_backups_created": 0,
"total_backup_size_mb": 0.0,
"average_backup_time_seconds": 0.0,
"last_global_scan": "",
"projects_with_errors": 0,
"projects_pending_retry": 0
}

25
src/app.py
View File

@ -165,14 +165,29 @@ class AutoBackupsFlaskApp:
else:
projects = []
# Agregar proyectos al manager
for project_info in projects:
self.project_manager.add_or_update_project(project_info)
# Obtener proyectos existentes para evitar duplicados
existing_projects = self.project_manager.get_all_projects()
existing_paths = {project.path for project in existing_projects}
msg = f"Descubrimiento completado: {len(projects)} proyectos"
# Agregar solo proyectos nuevos al manager
new_projects_count = 0
for project_info in projects:
project_path = project_info.get("path")
if project_path not in existing_paths:
self.project_manager.add_or_update_project(project_info)
new_projects_count += 1
self.logger.debug(
f"Nuevo proyecto agregado: {project_info.get('name')}"
)
else:
self.logger.debug(
f"Proyecto ya existe, omitido: {project_info.get('name')}"
)
msg = f"Descubrimiento completado: {len(projects)} proyectos encontrados, {new_projects_count} nuevos agregados"
self.logger.info(msg)
return len(projects)
return new_projects_count
except Exception as e:
self.logger.error(f"Error en descubrimiento de proyectos: {e}")

42
src/models/project_model.py
View File

@ -300,3 +300,45 @@ class ProjectManager:
self.get_projects_by_status(ProjectStatus.RETRY_PENDING)
)
self.save_projects()
def update_project_config(
self, project_id: str, config_data: Dict[str, Any]
) -> bool:
"""Actualizar configuración de un proyecto específico"""
try:
project = self.get_project(project_id)
if not project:
return False
# Actualizar configuración de schedule si se proporciona
if "schedule_config" in config_data:
schedule_config = config_data["schedule_config"]
if "schedule" in schedule_config:
project.schedule = schedule_config["schedule"]
if "schedule_time" in schedule_config:
project.schedule_time = schedule_config["schedule_time"]
if "enabled" in schedule_config:
project.enabled = schedule_config["enabled"]
# Actualizar next_scheduled_backup si es necesario
if "next_scheduled_backup" in schedule_config:
project.next_scheduled_backup = schedule_config[
"next_scheduled_backup"
]
# Actualizar otros campos si se proporcionan
for key, value in config_data.items():
if hasattr(project, key) and key != "schedule_config":
setattr(project, key, value)
# Actualizar metadata y guardar
self.metadata["last_updated"] = datetime.now(timezone.utc).isoformat()
self.save_projects()
return True
except Exception as e:
print(
f"Error actualizando configuración del proyecto " f"{project_id}: {e}"
)
return False

50
src/routes/api_routes.py
View File

@ -44,15 +44,31 @@ def register_api_routes(app, autobackups_instance):
f"Backup manual solicitado para: {project_id}"
)
# TODO: Implementar backup manual real
# Por ahora solo simulamos la respuesta
# Obtener el proyecto
project = autobackups_instance.project_manager.get_project(project_id)
if not project:
return jsonify({"error": f"Proyecto {project_id} no encontrado"}), 404
# Ejecutar backup usando el servicio
backup_result = autobackups_instance.backup_service.backup_project(project)
if backup_result["success"]:
# Actualizar información del proyecto si el backup fue exitoso
if "backup_file" in backup_result:
project.update_backup_info(backup_result["backup_file"])
autobackups_instance.project_manager.save_projects()
return jsonify(
{
"status": "success",
"message": backup_result["message"],
"backup_file": backup_result.get("backup_file"),
"file_size_mb": backup_result.get("file_size_mb"),
}
)
else:
return jsonify({"error": backup_result["error"]}), 500
return jsonify(
{
"status": "success",
"message": f"Backup iniciado para proyecto {project_id}",
}
)
except Exception as e:
autobackups_instance.logger.error(f"Error in manual backup: {e}")
return jsonify({"error": str(e)}), 500
@ -66,11 +82,21 @@ def register_api_routes(app, autobackups_instance):
if not data:
return jsonify({"error": "No data provided"}), 400
# TODO: Implementar actualización real de configuración
autobackups_instance.logger.info(f"Config actualizada para: {project_id}")
autobackups_instance.logger.info(f"Nueva config: {data}")
# Usar el método real del ProjectManager
success = autobackups_instance.project_manager.update_project_config( # noqa: E501
project_id, data
)
if success:
autobackups_instance.logger.info(
f"Config actualizada para: {project_id}"
)
return jsonify({"status": "success"})
else:
error_msg = f"No se pudo actualizar el proyecto {project_id}"
autobackups_instance.logger.error(error_msg)
return jsonify({"error": error_msg}), 404
return jsonify({"status": "success"})
except Exception as e:
autobackups_instance.logger.error(f"Error updating project config: {e}")
return jsonify({"error": str(e)}), 500

136
src/services/basic_backup_service.py
View File

@ -5,6 +5,8 @@ Versión simplificada para Fase 1 sin Everything API
import os
import logging
import hashlib
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Any
@ -31,7 +33,9 @@ class BasicBackupService:
if obs_dir.get("enabled", True):
path = obs_dir["path"]
if not os.path.exists(path):
self.logger.warning(f"Directorio de observación no existe: {path}")
self.logger.warning(
f"Directorio de observación no existe: {path}"
)
self.logger.info("Requerimientos básicos del sistema verificados")
return True
@ -46,7 +50,8 @@ class BasicBackupService:
try:
obs_dirs = [
obs_dir for obs_dir in self.config.observation_directories
obs_dir
for obs_dir in self.config.observation_directories
if obs_dir.get("enabled", True) and obs_dir.get("type") == "siemens_s7"
]
@ -63,14 +68,18 @@ class BasicBackupService:
projects.append(project_info)
self.logger.info(f"Proyecto encontrado: {project_info['name']}")
self.logger.info(f"Descubrimiento completado. {len(projects)} proyectos encontrados")
self.logger.info(
f"Descubrimiento completado. {len(projects)} proyectos encontrados"
)
return projects
except Exception as e:
self.logger.error(f"Error en descubrimiento de proyectos: {e}")
return []
def _create_project_info(self, s7p_file: Path, obs_dir: Dict[str, Any]) -> Dict[str, Any]:
def _create_project_info(
self, s7p_file: Path, obs_dir: Dict[str, Any]
) -> Dict[str, Any]:
"""Crear información de proyecto desde archivo .s7p"""
project_path = s7p_file.parent
obs_path = Path(obs_dir["path"])
@ -81,7 +90,11 @@ class BasicBackupService:
except ValueError:
relative_path = project_path.name
# Generar ID único para el proyecto
project_id = self._generate_project_id(str(project_path))
return {
"id": project_id,
"name": project_path.name,
"path": str(project_path),
"s7p_file": str(s7p_file),
@ -90,5 +103,118 @@ class BasicBackupService:
"backup_path": str(relative_path),
"type": "siemens_s7",
"auto_discovered": True,
"discovery_method": "filesystem_search"
"discovery_method": "filesystem_search",
}
def _generate_project_id(self, path: str) -> str:
"""Generar ID único para un proyecto basado en su ruta"""
# Usar hash de la ruta + timestamp para garantizar unicidad
path_hash = hashlib.md5(path.encode()).hexdigest()[:8]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return f"project_{path_hash}_{timestamp}"
def backup_project(self, project) -> Dict[str, Any]:
"""Ejecutar backup de un proyecto específico"""
try:
self.logger.info(f"Iniciando backup del proyecto: {project.name}")
# Verificar que el directorio del proyecto existe
if not os.path.exists(project.path):
error_msg = f"Directorio del proyecto no existe: " f"{project.path}"
self.logger.error(error_msg)
return {"success": False, "error": error_msg}
# Verificar que el archivo .s7p existe
if hasattr(project, "s7p_file") and project.s7p_file:
if not os.path.exists(project.s7p_file):
error_msg = f"Archivo .s7p no existe: {project.s7p_file}"
self.logger.error(error_msg)
return {"success": False, "error": error_msg}
# Verificar espacio en disco
backup_dest = self.config.backup_destination
min_space_mb = self.config.global_settings.get("min_free_space_mb", 100)
try:
import shutil
free_space_bytes = shutil.disk_usage(backup_dest).free
free_space_mb = free_space_bytes / (1024 * 1024)
if free_space_mb < min_space_mb:
error_msg = (
f"Espacio insuficiente en destino. "
f"Disponible: {free_space_mb:.1f}MB, "
f"Requerido: {min_space_mb}MB"
)
self.logger.error(error_msg)
return {"success": False, "error": error_msg}
except Exception as e:
self.logger.warning(f"No se pudo verificar espacio en disco: {e}")
# Crear directorio de backup
backup_path = self._create_backup_path(project)
os.makedirs(backup_path, exist_ok=True)
# Crear archivo ZIP
timestamp = datetime.now().strftime("%H-%M-%S")
backup_filename = f"{timestamp}_projects.zip"
backup_filepath = os.path.join(backup_path, backup_filename)
# Comprimir proyecto
import zipfile
with zipfile.ZipFile(backup_filepath, "w", zipfile.ZIP_DEFLATED) as zipf:
self._add_directory_to_zip(zipf, project.path, project.backup_path)
# Verificar que el archivo se creó correctamente
if os.path.exists(backup_filepath):
file_size = os.path.getsize(backup_filepath)
file_size_mb = file_size / (1024 * 1024)
self.logger.info(
f"Backup completado: {backup_filename} " f"({file_size_mb:.2f} MB)"
)
return {
"success": True,
"backup_file": backup_filepath,
"file_size_mb": file_size_mb,
"message": f"Backup creado exitosamente: {backup_filename}",
}
else:
error_msg = "Error: el archivo de backup no se creó"
self.logger.error(error_msg)
return {"success": False, "error": error_msg}
except Exception as e:
error_msg = f"Error durante el backup: {str(e)}"
self.logger.error(error_msg)
return {"success": False, "error": error_msg}
def _create_backup_path(self, project) -> str:
"""Crear la ruta de backup para un proyecto"""
backup_base = Path(self.config.backup_destination)
# Usar backup_path del proyecto si está disponible
if hasattr(project, "backup_path") and project.backup_path:
project_backup_path = project.backup_path
else:
# Fallback al nombre del proyecto
project_backup_path = project.name
# Crear estructura: backup_destination/project_path/YYYY-MM-DD/
date_folder = datetime.now().strftime("%Y-%m-%d")
full_backup_path = backup_base / project_backup_path / date_folder
return str(full_backup_path)
def _add_directory_to_zip(self, zipf, source_dir, archive_name):
"""Agregar directorio completo al archivo ZIP"""
source_path = Path(source_dir)
for file_path in source_path.rglob("*"):
if file_path.is_file():
# Calcular ruta relativa para el archivo en el ZIP
relative_path = file_path.relative_to(source_path.parent)
zipf.write(file_path, relative_path)

91
test_api.py Normal file
View File

@ -0,0 +1,91 @@
#!/usr/bin/env python3
"""
Script para probar backup manual vía API
"""
import requests
import json
def test_manual_backup_api():
"""Probar backup manual vía API HTTP"""
base_url = "http://127.0.0.1:5120"
try:
# 1. Obtener lista de proyectos
print("1. Obteniendo lista de proyectos...")
response = requests.get(f"{base_url}/api/projects")
if response.status_code == 200:
data = response.json()
projects = data.get("projects", [])
print(f" ✅ Proyectos encontrados: {len(projects)}")
if projects:
project = projects[0]
project_id = project["id"]
project_name = project["name"]
print(f" Probando con: {project_name}")
# 2. Ejecutar backup manual
print("2. Ejecutando backup manual...")
backup_response = requests.post(
f"{base_url}/api/projects/{project_id}/backup",
headers={"Content-Type": "application/json"},
)
if backup_response.status_code == 200:
backup_data = backup_response.json()
print(" ✅ BACKUP EXITOSO!")
print(f" Mensaje: {backup_data.get('message')}")
print(f" Archivo: {backup_data.get('backup_file')}")
print(f" Tamaño: {backup_data.get('file_size_mb', 'N/A')} MB")
else:
print(" ❌ BACKUP FALLÓ!")
print(f" Status: {backup_response.status_code}")
print(f" Error: {backup_response.text}")
# 3. Probar actualización de configuración
print("3. Probando actualización de configuración...")
config_data = {
"schedule_config": {
"schedule": "hourly",
"schedule_time": "15:30",
"enabled": False,
}
}
config_response = requests.put(
f"{base_url}/api/projects/{project_id}/config",
headers={"Content-Type": "application/json"},
json=config_data,
)
if config_response.status_code == 200:
print(" ✅ CONFIGURACIÓN ACTUALIZADA!")
config_result = config_response.json()
print(f" Status: {config_result.get('status')}")
else:
print(" ❌ ACTUALIZACIÓN FALLÓ!")
print(f" Status: {config_response.status_code}")
print(f" Error: {config_response.text}")
else:
print(" ❌ No hay proyectos para probar")
else:
print(f" ❌ Error obteniendo proyectos: {response.status_code}")
print(f" {response.text}")
except requests.exceptions.ConnectionError:
print("❌ ERROR: No se pudo conectar al servidor")
print(" Asegúrate de que la aplicación Flask esté ejecutándose")
except Exception as e:
print(f"❌ ERROR INESPERADO: {e}")
if __name__ == "__main__":
print("AutoBackups - Test API via HTTP")
print("=" * 40)
test_manual_backup_api()
print("=" * 40)

142
test_functions.py Normal file
View File

@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""
Script de prueba para verificar las funcionalidades de backup manual
y configuración de proyectos implementadas
"""
import sys
import os
from pathlib import Path
# Agregar el directorio src al path
current_dir = Path(__file__).parent
src_dir = current_dir / "src"
sys.path.insert(0, str(src_dir))
# Imports de módulos propios
from models.config_model import Config
from models.project_model import ProjectManager
from services.basic_backup_service import BasicBackupService
def test_backup_manual():
"""Probar funcionalidad de backup manual"""
print("=== TEST BACKUP MANUAL ===")
try:
# Inicializar servicios
config = Config()
project_manager = ProjectManager()
backup_service = BasicBackupService(config)
# Obtener proyectos
projects = project_manager.get_all_projects()
print(f"Proyectos encontrados: {len(projects)}")
if not projects:
print("No hay proyectos para probar")
return
# Probar backup del primer proyecto
project = projects[0]
print(f"Probando backup del proyecto: {project.name}")
print(f"Ruta del proyecto: {project.path}")
# Ejecutar backup
result = backup_service.backup_project(project)
if result["success"]:
print("✅ BACKUP EXITOSO!")
print(f"Archivo creado: {result['backup_file']}")
print(f"Tamaño: {result['file_size_mb']:.2f} MB")
else:
print("❌ BACKUP FALLÓ!")
print(f"Error: {result['error']}")
except Exception as e:
print(f"❌ ERROR EN TEST: {e}")
import traceback
traceback.print_exc()
def test_config_update():
"""Probar funcionalidad de actualización de configuración"""
print("\n=== TEST ACTUALIZACIÓN DE CONFIGURACIÓN ===")
try:
# Inicializar ProjectManager
project_manager = ProjectManager()
# Obtener proyectos
projects = project_manager.get_all_projects()
if not projects:
print("No hay proyectos para probar")
return
project = projects[0]
print(f"Probando configuración del proyecto: {project.name}")
print(
f"Estado inicial - Habilitado: {project.enabled}, Schedule: {project.schedule}"
)
# Preparar nueva configuración
new_config = {
"schedule_config": {
"schedule": "hourly",
"schedule_time": "10:30",
"enabled": False,
}
}
# Actualizar configuración
success = project_manager.update_project_config(project.id, new_config)
if success:
print("✅ CONFIGURACIÓN ACTUALIZADA!")
# Verificar cambios
updated_project = project_manager.get_project(project.id)
print(
f"Estado actualizado - Habilitado: {updated_project.enabled}, Schedule: {updated_project.schedule}"
)
# Restaurar configuración original
restore_config = {
"schedule_config": {
"schedule": "daily",
"schedule_time": "02:00",
"enabled": True,
}
}
project_manager.update_project_config(project.id, restore_config)
print("✅ Configuración restaurada")
else:
print("❌ FALLO AL ACTUALIZAR CONFIGURACIÓN!")
except Exception as e:
print(f"❌ ERROR EN TEST: {e}")
import traceback
traceback.print_exc()
def main():
"""Función principal"""
print("AutoBackups - Test de Funcionalidades")
print("=" * 50)
# Test 1: Backup manual
test_backup_manual()
# Test 2: Actualización de configuración
test_config_update()
print("\n" + "=" * 50)
print("Tests completados")
if __name__ == "__main__":
main()