Arch/services/document_service.py

440 lines
14 KiB
Python
Raw Normal View History

2025-03-03 15:35:24 -03:00
import os
import json
from datetime import datetime
import pytz
import mimetypes
from werkzeug.utils import secure_filename
from flask import current_app
from utils.file_utils import (
load_json_file, save_json_file, ensure_dir_exists,
get_next_id, format_document_directory_name,
format_version_filename
)
from utils.security import calculate_checksum, check_file_type
from services.project_service import find_project_directory
def get_allowed_filetypes():
"""
Obtener los tipos de archivo permitidos.
Returns:
dict: Diccionario de tipos de archivo
"""
storage_path = current_app.config['STORAGE_PATH']
filetypes_file = os.path.join(storage_path, 'filetypes', 'filetypes.json')
return load_json_file(filetypes_file, {})
def add_document(project_id, document_data, file, creator_username):
"""
Añadir un nuevo documento a un proyecto.
Args:
project_id (int): ID del proyecto
document_data (dict): Datos del documento
file: Objeto de archivo (de Flask)
creator_username (str): Usuario que crea el documento
Returns:
tuple: (success, message, document_id)
"""
# Buscar directorio del proyecto
project_dir = find_project_directory(project_id)
if not project_dir:
return False, f"No se encontró el proyecto con ID {project_id}.", None
# Validar datos obligatorios
if 'nombre' not in document_data or not document_data['nombre']:
return False, "El nombre del documento es obligatorio.", None
# Validar tipo de archivo
allowed_filetypes = get_allowed_filetypes()
filename = secure_filename(file.filename)
extension = filename.rsplit('.', 1)[1].lower() if '.' in filename else ''
if extension not in allowed_filetypes:
return False, f"Tipo de archivo no permitido: {extension}", None
# Verificar MIME type
if not check_file_type(file.stream, [allowed_filetypes[extension]['mime_type']]):
return False, "El tipo de archivo no coincide con su extensión.", None
# Obtener siguiente ID de documento
document_id = get_next_id('document')
# Preparar directorio del documento
documents_dir = os.path.join(project_dir, 'documents')
dir_name = format_document_directory_name(document_id, document_data['nombre'])
document_dir = os.path.join(documents_dir, dir_name)
# Verificar si ya existe
if os.path.exists(document_dir):
return False, "Ya existe un documento con ese nombre en este proyecto.", None
# Crear directorio
ensure_dir_exists(document_dir)
# Preparar primera versión
version = 1
version_filename = format_version_filename(version, document_data['nombre'], extension)
version_path = os.path.join(document_dir, version_filename)
# Guardar archivo
file.seek(0)
file.save(version_path)
# Calcular checksum
checksum = calculate_checksum(version_path)
# Obtener tamaño del archivo
file_size = os.path.getsize(version_path)
# Preparar metadatos del documento
document_meta = {
'document_id': f"{document_id:03d}_{document_data['nombre'].lower().replace(' ', '_')}",
'original_filename': filename,
'versions': [
{
'version': version,
'filename': version_filename,
'created_at': datetime.now(pytz.UTC).isoformat(),
'created_by': creator_username,
'description': document_data.get('description', 'Versión inicial'),
'file_size': file_size,
'mime_type': allowed_filetypes[extension]['mime_type'],
'checksum': checksum,
'downloads': []
}
]
}
# Guardar metadatos
meta_file = os.path.join(document_dir, 'meta.json')
save_json_file(meta_file, document_meta)
return True, "Documento añadido correctamente.", document_id
def add_version(project_id, document_id, version_data, file, creator_username):
"""
Añadir una nueva versión a un documento existente.
Args:
project_id (int): ID del proyecto
document_id (int): ID del documento
version_data (dict): Datos de la versión
file: Objeto de archivo (de Flask)
creator_username (str): Usuario que crea la versión
Returns:
tuple: (success, message, version_number)
"""
# Buscar directorio del proyecto
project_dir = find_project_directory(project_id)
if not project_dir:
return False, f"No se encontró el proyecto con ID {project_id}.", None
# Buscar documento
document_dir = find_document_directory(project_dir, document_id)
if not document_dir:
return False, f"No se encontró el documento con ID {document_id}.", None
# Cargar metadatos del documento
meta_file = os.path.join(document_dir, 'meta.json')
document_meta = load_json_file(meta_file)
if not document_meta:
return False, "Error al cargar metadatos del documento.", None
# Validar tipo de archivo
allowed_filetypes = get_allowed_filetypes()
filename = secure_filename(file.filename)
extension = filename.rsplit('.', 1)[1].lower() if '.' in filename else ''
if extension not in allowed_filetypes:
return False, f"Tipo de archivo no permitido: {extension}", None
# Verificar MIME type
if not check_file_type(file.stream, [allowed_filetypes[extension]['mime_type']]):
return False, "El tipo de archivo no coincide con su extensión.", None
# Determinar número de versión
last_version = max([v['version'] for v in document_meta['versions']])
new_version = last_version + 1
# Preparar nombre de archivo
doc_name = document_meta['document_id'].split('_', 1)[1] if '_' in document_meta['document_id'] else 'document'
version_filename = format_version_filename(new_version, doc_name, extension)
version_path = os.path.join(document_dir, version_filename)
# Guardar archivo
file.seek(0)
file.save(version_path)
# Calcular checksum
checksum = calculate_checksum(version_path)
# Obtener tamaño del archivo
file_size = os.path.getsize(version_path)
# Preparar metadatos de la versión
version_meta = {
'version': new_version,
'filename': version_filename,
'created_at': datetime.now(pytz.UTC).isoformat(),
'created_by': creator_username,
'description': version_data.get('description', f'Versión {new_version}'),
'file_size': file_size,
'mime_type': allowed_filetypes[extension]['mime_type'],
'checksum': checksum,
'downloads': []
}
# Añadir versión a metadatos
document_meta['versions'].append(version_meta)
# Guardar metadatos actualizados
save_json_file(meta_file, document_meta)
return True, "Nueva versión añadida correctamente.", new_version
def get_document(project_id, document_id):
"""
Obtener información de un documento.
Args:
project_id (int): ID del proyecto
document_id (int): ID del documento
Returns:
dict: Datos del documento o None si no existe
"""
# Buscar directorio del proyecto
project_dir = find_project_directory(project_id)
if not project_dir:
return None
# Buscar documento
document_dir = find_document_directory(project_dir, document_id)
if not document_dir:
return None
# Cargar metadatos
meta_file = os.path.join(document_dir, 'meta.json')
document_meta = load_json_file(meta_file)
# Agregar ruta del directorio
if document_meta:
document_meta['directory'] = os.path.basename(document_dir)
return document_meta
def get_document_version(project_id, document_id, version):
"""
Obtener información de una versión específica de un documento.
Args:
project_id (int): ID del proyecto
document_id (int): ID del documento
version (int): Número de versión
Returns:
tuple: (dict, str) - (Metadatos de la versión, ruta al archivo)
"""
document = get_document(project_id, document_id)
if not document:
return None, None
# Buscar versión específica
version_meta = None
for v in document['versions']:
if v['version'] == int(version):
version_meta = v
break
if not version_meta:
return None, None
# Preparar ruta al archivo
project_dir = find_project_directory(project_id)
document_dir = find_document_directory(project_dir, document_id)
file_path = os.path.join(document_dir, version_meta['filename'])
return version_meta, file_path
def get_latest_version(project_id, document_id):
"""
Obtener la última versión de un documento.
Args:
project_id (int): ID del proyecto
document_id (int): ID del documento
Returns:
tuple: (dict, str) - (Metadatos de la versión, ruta al archivo)
"""
document = get_document(project_id, document_id)
if not document or not document['versions']:
return None, None
# Encontrar la versión más reciente
latest_version = max(document['versions'], key=lambda v: v['version'])
# Preparar ruta al archivo
project_dir = find_project_directory(project_id)
document_dir = find_document_directory(project_dir, document_id)
file_path = os.path.join(document_dir, latest_version['filename'])
return latest_version, file_path
def register_download(project_id, document_id, version, username):
"""
Registrar una descarga de documento.
Args:
project_id (int): ID del proyecto
document_id (int): ID del documento
version (int): Número de versión
username (str): Usuario que descarga
Returns:
bool: True si se registró correctamente, False en caso contrario
"""
# Buscar documento
project_dir = find_project_directory(project_id)
if not project_dir:
return False
document_dir = find_document_directory(project_dir, document_id)
if not document_dir:
return False
# Cargar metadatos
meta_file = os.path.join(document_dir, 'meta.json')
document_meta = load_json_file(meta_file)
if not document_meta:
return False
# Buscar versión
for v in document_meta['versions']:
if v['version'] == int(version):
# Registrar descarga
download_info = {
'user_id': username,
'downloaded_at': datetime.now(pytz.UTC).isoformat()
}
v['downloads'].append(download_info)
# Guardar metadatos actualizados
save_json_file(meta_file, document_meta)
return True
return False
def find_document_directory(project_dir, document_id):
"""
Encontrar el directorio de un documento por su ID.
Args:
project_dir (str): Ruta al directorio del proyecto
document_id (int): ID del documento
Returns:
str: Ruta al directorio o None si no se encuentra
"""
documents_dir = os.path.join(project_dir, 'documents')
if not os.path.exists(documents_dir):
return None
# Prefijo a buscar en nombres de directorios
prefix = f"@{int(document_id):03d}_@"
for dir_name in os.listdir(documents_dir):
if dir_name.startswith(prefix):
return os.path.join(documents_dir, dir_name)
return None
def get_project_documents(project_id):
"""
Obtener todos los documentos de un proyecto.
Args:
project_id (int): ID del proyecto
Returns:
list: Lista de documentos
"""
project_dir = find_project_directory(project_id)
if not project_dir:
return []
documents_dir = os.path.join(project_dir, 'documents')
if not os.path.exists(documents_dir):
return []
documents = []
# Iterar sobre directorios de documentos
for dir_name in os.listdir(documents_dir):
if dir_name.startswith('@') and os.path.isdir(os.path.join(documents_dir, dir_name)):
document_dir = os.path.join(documents_dir, dir_name)
meta_file = os.path.join(document_dir, 'meta.json')
if os.path.exists(meta_file):
document_meta = load_json_file(meta_file)
if document_meta:
# Extraer ID del documento del nombre del directorio
try:
doc_id = int(dir_name.split('_', 1)[0].replace('@', ''))
document_meta['id'] = doc_id
document_meta['directory'] = dir_name
documents.append(document_meta)
except (ValueError, IndexError):
pass
return documents
def delete_document(project_id, document_id):
"""
Eliminar un documento.
Args:
project_id (int): ID del proyecto
document_id (int): ID del documento
Returns:
tuple: (success, message)
"""
# Buscar documento
project_dir = find_project_directory(project_id)
if not project_dir:
return False, f"No se encontró el proyecto con ID {project_id}."
document_dir = find_document_directory(project_dir, document_id)
if not document_dir:
return False, f"No se encontró el documento con ID {document_id}."
# Eliminar directorio y contenido
try:
import shutil
shutil.rmtree(document_dir)
return True, "Documento eliminado correctamente."
except Exception as e:
return False, f"Error al eliminar el documento: {str(e)}"