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(): """Get all allowed filetypes from storage.""" storage_path = current_app.config["STORAGE_PATH"] filetypes_path = os.path.join(storage_path, "filetypes", "filetypes.json") if os.path.exists(filetypes_path): with open(filetypes_path, "r", encoding="utf-8") as f: return json.load(f) return {} 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)}"