370 lines
11 KiB
Python
370 lines
11 KiB
Python
import os
|
|
import json
|
|
from datetime import datetime
|
|
import pytz
|
|
from flask import current_app
|
|
from utils.file_utils import (
|
|
load_json_file,
|
|
save_json_file,
|
|
ensure_dir_exists,
|
|
get_next_id,
|
|
format_project_directory_name,
|
|
)
|
|
|
|
|
|
def create_project(project_data, creator_username):
|
|
"""
|
|
Crear un nuevo proyecto.
|
|
|
|
Args:
|
|
project_data (dict): Datos del proyecto
|
|
creator_username (str): Usuario que crea el proyecto
|
|
|
|
Returns:
|
|
tuple: (success, message, project_id)
|
|
"""
|
|
# Validar datos obligatorios
|
|
required_fields = ["descripcion", "cliente", "esquema"]
|
|
for field in required_fields:
|
|
if field not in project_data or not project_data[field]:
|
|
return False, f"El campo '{field}' es obligatorio.", None
|
|
|
|
# Obtener siguiente ID de proyecto
|
|
project_id = get_next_id("project")
|
|
|
|
# Crear código de proyecto (PROJ001, etc.)
|
|
project_code = f"PROJ{project_id:03d}"
|
|
|
|
# Preparar directorio del proyecto
|
|
storage_path = current_app.config["STORAGE_PATH"]
|
|
dir_name = format_project_directory_name(project_id, project_data["descripcion"])
|
|
project_dir = os.path.join(storage_path, "projects", dir_name)
|
|
|
|
# Verificar si ya existe
|
|
if os.path.exists(project_dir):
|
|
return False, "Ya existe un proyecto con esa descripción.", None
|
|
|
|
# Crear directorios
|
|
ensure_dir_exists(project_dir)
|
|
ensure_dir_exists(os.path.join(project_dir, "documents"))
|
|
|
|
# Crear metadatos del proyecto
|
|
project_meta = {
|
|
"codigo": project_code,
|
|
"proyecto_padre": project_data.get("proyecto_padre"),
|
|
"esquema": project_data["esquema"],
|
|
"descripcion": project_data["descripcion"],
|
|
"cliente": project_data["cliente"],
|
|
"destinacion": project_data.get("destinacion", ""),
|
|
"ano_creacion": datetime.now().year,
|
|
"fecha_creacion": datetime.now(pytz.UTC).isoformat(),
|
|
"creado_por": creator_username,
|
|
"estado": "activo",
|
|
"ultima_modificacion": datetime.now(pytz.UTC).isoformat(),
|
|
"modificado_por": creator_username,
|
|
}
|
|
|
|
# Guardar metadatos
|
|
meta_file = os.path.join(project_dir, "project_meta.json")
|
|
save_json_file(meta_file, project_meta)
|
|
|
|
# Guardar permisos del proyecto (inicialmente vacío)
|
|
permissions_file = os.path.join(project_dir, "permissions.json")
|
|
save_json_file(permissions_file, {})
|
|
|
|
# Copiar el esquema seleccionado
|
|
schema_file = os.path.join(storage_path, "schemas", "schema.json")
|
|
schemas = load_json_file(schema_file, {})
|
|
|
|
if project_data["esquema"] in schemas:
|
|
project_schema_file = os.path.join(project_dir, "schema.json")
|
|
save_json_file(project_schema_file, schemas[project_data["esquema"]])
|
|
|
|
return True, "Proyecto creado correctamente.", project_id
|
|
|
|
|
|
def update_project(project_id, project_data, modifier_username):
|
|
"""
|
|
Actualizar un proyecto existente.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
project_data (dict): Datos actualizados
|
|
modifier_username (str): Usuario que modifica
|
|
|
|
Returns:
|
|
tuple: (success, message)
|
|
"""
|
|
# Buscar el proyecto
|
|
project_dir = find_project_directory(project_id)
|
|
|
|
if not project_dir:
|
|
return False, f"No se encontró el proyecto con ID {project_id}."
|
|
|
|
# Cargar metadatos actuales
|
|
meta_file = os.path.join(project_dir, "project_meta.json")
|
|
current_meta = load_json_file(meta_file)
|
|
|
|
# Actualizar campos
|
|
for key, value in project_data.items():
|
|
if key in current_meta and key not in [
|
|
"codigo",
|
|
"fecha_creacion",
|
|
"creado_por",
|
|
"ano_creacion",
|
|
]:
|
|
current_meta[key] = value
|
|
|
|
# Actualizar metadatos de modificación
|
|
current_meta["ultima_modificacion"] = datetime.now(pytz.UTC).isoformat()
|
|
current_meta["modificado_por"] = modifier_username
|
|
|
|
# Guardar metadatos actualizados
|
|
save_json_file(meta_file, current_meta)
|
|
|
|
return True, "Proyecto actualizado correctamente."
|
|
|
|
|
|
def get_project(project_id):
|
|
"""
|
|
Obtener información de un proyecto.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
|
|
Returns:
|
|
dict: Datos del proyecto o None si no existe
|
|
"""
|
|
project_dir = find_project_directory(project_id)
|
|
|
|
if not project_dir:
|
|
return None
|
|
|
|
# Cargar metadatos
|
|
meta_file = os.path.join(project_dir, "project_meta.json")
|
|
project_meta = load_json_file(meta_file)
|
|
|
|
# Agregar la ruta del directorio
|
|
project_meta["directory"] = os.path.basename(project_dir)
|
|
|
|
return project_meta
|
|
|
|
|
|
def delete_project(project_id):
|
|
"""
|
|
Eliminar un proyecto (marcar como inactivo).
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
|
|
Returns:
|
|
tuple: (success, message)
|
|
"""
|
|
project_dir = find_project_directory(project_id)
|
|
|
|
if not project_dir:
|
|
return False, f"No se encontró el proyecto con ID {project_id}."
|
|
|
|
# Cargar metadatos
|
|
meta_file = os.path.join(project_dir, "project_meta.json")
|
|
project_meta = load_json_file(meta_file)
|
|
|
|
# Marcar como inactivo (no eliminar físicamente)
|
|
project_meta["estado"] = "inactivo"
|
|
project_meta["ultima_modificacion"] = datetime.now(pytz.UTC).isoformat()
|
|
|
|
# Guardar metadatos actualizados
|
|
save_json_file(meta_file, project_meta)
|
|
|
|
return True, "Proyecto marcado como inactivo."
|
|
|
|
|
|
def get_all_projects(include_inactive=False):
|
|
"""
|
|
Obtener todos los proyectos.
|
|
|
|
Args:
|
|
include_inactive (bool): Incluir proyectos inactivos
|
|
|
|
Returns:
|
|
list: Lista de proyectos
|
|
"""
|
|
storage_path = current_app.config["STORAGE_PATH"]
|
|
projects_dir = os.path.join(storage_path, "projects")
|
|
projects = []
|
|
|
|
# Iterar sobre directorios de proyectos
|
|
for dir_name in os.listdir(projects_dir):
|
|
if dir_name.startswith("@"): # Formato de directorio de proyecto
|
|
project_dir = os.path.join(projects_dir, dir_name)
|
|
|
|
if os.path.isdir(project_dir):
|
|
meta_file = os.path.join(project_dir, "project_meta.json")
|
|
|
|
if os.path.exists(meta_file):
|
|
project_meta = load_json_file(meta_file)
|
|
|
|
# Incluir solo si está activo o se solicitan inactivos
|
|
if include_inactive or project_meta.get("estado") == "activo":
|
|
# Agregar la ruta del directorio
|
|
project_meta["directory"] = dir_name
|
|
projects.append(project_meta)
|
|
|
|
return projects
|
|
|
|
|
|
def find_project_directory(project_id):
|
|
"""
|
|
Encontrar el directorio de un proyecto por su ID.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
|
|
Returns:
|
|
str: Ruta al directorio o None si no se encuentra
|
|
"""
|
|
storage_path = current_app.config["STORAGE_PATH"]
|
|
projects_dir = os.path.join(storage_path, "projects")
|
|
|
|
# Prefijo a buscar en nombres de directorios
|
|
prefix = f"@{int(project_id):03d}_@"
|
|
|
|
for dir_name in os.listdir(projects_dir):
|
|
if dir_name.startswith(prefix):
|
|
return os.path.join(projects_dir, dir_name)
|
|
|
|
return None
|
|
|
|
|
|
def get_project_children(project_id):
|
|
"""
|
|
Obtener proyectos hijos de un proyecto.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto padre
|
|
|
|
Returns:
|
|
list: Lista de proyectos hijos
|
|
"""
|
|
all_projects = get_all_projects()
|
|
project_code = f"PROJ{int(project_id):03d}"
|
|
|
|
# Filtrar proyectos con este padre
|
|
children = [p for p in all_projects if p.get("proyecto_padre") == project_code]
|
|
|
|
return children
|
|
|
|
|
|
def get_project_document_count(project_id):
|
|
"""
|
|
Contar documentos en un proyecto.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
|
|
Returns:
|
|
int: Número de documentos
|
|
"""
|
|
project_dir = find_project_directory(project_id)
|
|
|
|
if not project_dir:
|
|
return 0
|
|
|
|
documents_dir = os.path.join(project_dir, "documents")
|
|
|
|
if not os.path.exists(documents_dir):
|
|
return 0
|
|
|
|
# Contar directorios de documentos
|
|
count = 0
|
|
for item in os.listdir(documents_dir):
|
|
if os.path.isdir(os.path.join(documents_dir, item)) and item.startswith("@"):
|
|
count += 1
|
|
|
|
return count
|
|
|
|
|
|
def filter_projects(filter_params):
|
|
"""
|
|
Filtrar proyectos según los parámetros proporcionados.
|
|
|
|
Args:
|
|
filter_params (dict): Diccionario con parámetros de filtrado
|
|
- cliente: Nombre de cliente
|
|
- estado: Estado del proyecto ('activo', 'inactivo')
|
|
- ano_inicio: Año de inicio para filtrar
|
|
- ano_fin: Año final para filtrar
|
|
- descripcion: Término de búsqueda en descripción
|
|
|
|
Returns:
|
|
list: Lista de proyectos que cumplen los criterios
|
|
"""
|
|
# Obtener todos los proyectos (incluyendo inactivos si se solicitan)
|
|
include_inactive = filter_params.get("estado") == "inactivo"
|
|
all_projects = get_all_projects(include_inactive)
|
|
filtered_projects = []
|
|
|
|
for project in all_projects:
|
|
# Filtrar por cliente
|
|
if "cliente" in filter_params and filter_params["cliente"]:
|
|
if project["cliente"] != filter_params["cliente"]:
|
|
continue
|
|
|
|
# Filtrar por estado
|
|
if "estado" in filter_params and filter_params["estado"]:
|
|
if project["estado"] != filter_params["estado"]:
|
|
continue
|
|
|
|
# Filtrar por año de creación (rango)
|
|
if "ano_inicio" in filter_params and filter_params["ano_inicio"]:
|
|
if project["ano_creacion"] < int(filter_params["ano_inicio"]):
|
|
continue
|
|
|
|
if "ano_fin" in filter_params and filter_params["ano_fin"]:
|
|
if project["ano_creacion"] > int(filter_params["ano_fin"]):
|
|
continue
|
|
|
|
# Filtrar por término en descripción
|
|
if "descripcion" in filter_params and filter_params["descripcion"]:
|
|
if (
|
|
filter_params["descripcion"].lower()
|
|
not in project["descripcion"].lower()
|
|
):
|
|
continue
|
|
|
|
# Si pasó todos los filtros, agregar a la lista
|
|
filtered_projects.append(project)
|
|
|
|
return filtered_projects
|
|
|
|
|
|
def archive_project(project_id, archiver_username):
|
|
"""
|
|
Archivar un proyecto (marcar como archivado).
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
archiver_username (str): Usuario que archiva el proyecto
|
|
|
|
Returns:
|
|
tuple: (success, message)
|
|
"""
|
|
project_dir = find_project_directory(project_id)
|
|
|
|
if not project_dir:
|
|
return False, f"No se encontró el proyecto con ID {project_id}."
|
|
|
|
# Cargar metadatos
|
|
meta_file = os.path.join(project_dir, "project_meta.json")
|
|
project_meta = load_json_file(meta_file)
|
|
|
|
# Marcar como archivado
|
|
project_meta["estado"] = "archivado"
|
|
project_meta["ultima_modificacion"] = datetime.now(pytz.UTC).isoformat()
|
|
project_meta["modificado_por"] = archiver_username
|
|
|
|
# Guardar metadatos actualizados
|
|
save_json_file(meta_file, project_meta)
|
|
|
|
return True, "Proyecto archivado correctamente."
|