318 lines
13 KiB
Python
318 lines
13 KiB
Python
import os
|
|
import json
|
|
import zipfile
|
|
import tempfile
|
|
from datetime import datetime
|
|
import pytz
|
|
from flask import current_app
|
|
|
|
from services.project_service import get_project, find_project_directory
|
|
from services.document_service import get_project_documents, get_latest_version
|
|
from utils.file_utils import ensure_dir_exists
|
|
|
|
def export_project(project_id, document_ids=None, include_metadata=True):
|
|
"""
|
|
Exportar un proyecto completo o documentos seleccionados a un archivo ZIP.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto a exportar
|
|
document_ids (list, optional): Lista de IDs de documentos a incluir.
|
|
Si es None, se incluyen todos los documentos.
|
|
include_metadata (bool, optional): Incluir metadatos del proyecto y documentos.
|
|
Por defecto es True.
|
|
|
|
Returns:
|
|
tuple: (success, message, zip_path)
|
|
- success (bool): True si la exportación fue exitosa
|
|
- message (str): Mensaje descriptivo
|
|
- zip_path (str): Ruta al archivo ZIP generado
|
|
"""
|
|
# Verificar que el proyecto existe
|
|
project = get_project(project_id)
|
|
|
|
if not project:
|
|
return False, f"No se encontró el proyecto con ID {project_id}.", None
|
|
|
|
# Obtener directorio del proyecto
|
|
project_dir = find_project_directory(project_id)
|
|
|
|
if not project_dir:
|
|
return False, f"No se encontró el directorio del proyecto con ID {project_id}.", None
|
|
|
|
# Crear directorio temporal para la exportación
|
|
temp_dir = tempfile.mkdtemp()
|
|
|
|
try:
|
|
# Obtener documentos del proyecto
|
|
all_documents = get_project_documents(project_id)
|
|
|
|
# Filtrar documentos si se especificaron IDs
|
|
if document_ids:
|
|
documents = [doc for doc in all_documents if doc.get('id') in document_ids]
|
|
else:
|
|
documents = all_documents
|
|
|
|
if not documents:
|
|
return False, "No hay documentos para exportar.", None
|
|
|
|
# Crear nombre para el archivo ZIP
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
project_code = project.get('codigo', f'PROJ{project_id}')
|
|
zip_filename = f"{project_code}_export_{timestamp}.zip"
|
|
|
|
# Ruta completa al archivo ZIP
|
|
storage_path = current_app.config['STORAGE_PATH']
|
|
exports_dir = os.path.join(storage_path, 'exports')
|
|
ensure_dir_exists(exports_dir)
|
|
zip_path = os.path.join(exports_dir, zip_filename)
|
|
|
|
# Crear archivo ZIP
|
|
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
|
# Incluir metadatos del proyecto si se solicita
|
|
if include_metadata:
|
|
# Crear archivo JSON con metadatos del proyecto
|
|
project_meta = {
|
|
'codigo': project.get('codigo'),
|
|
'descripcion': project.get('descripcion'),
|
|
'cliente': project.get('cliente'),
|
|
'destinacion': project.get('destinacion'),
|
|
'ano_creacion': project.get('ano_creacion'),
|
|
'fecha_creacion': project.get('fecha_creacion'),
|
|
'creado_por': project.get('creado_por'),
|
|
'fecha_exportacion': datetime.now(pytz.UTC).isoformat(),
|
|
'documentos_incluidos': len(documents)
|
|
}
|
|
|
|
# Guardar metadatos en archivo temporal
|
|
meta_file = os.path.join(temp_dir, 'project_info.json')
|
|
with open(meta_file, 'w', encoding='utf-8') as f:
|
|
json.dump(project_meta, f, ensure_ascii=False, indent=2)
|
|
|
|
# Añadir al ZIP
|
|
zipf.write(meta_file, arcname='project_info.json')
|
|
|
|
# Añadir documentos (última versión de cada uno)
|
|
for document in documents:
|
|
doc_id = document.get('id')
|
|
doc_name = document.get('document_id', '').split('_', 1)[1] if '_' in document.get('document_id', '') else f'doc_{doc_id}'
|
|
|
|
# Obtener última versión
|
|
version_meta, file_path = get_latest_version(project_id, doc_id)
|
|
|
|
if not version_meta or not file_path or not os.path.exists(file_path):
|
|
continue
|
|
|
|
# Nombre para el archivo en el ZIP
|
|
version_num = version_meta.get('version', 1)
|
|
original_filename = document.get('original_filename', os.path.basename(file_path))
|
|
|
|
# Usar nombre original o generar uno basado en metadatos
|
|
if '.' in original_filename:
|
|
base_name, extension = original_filename.rsplit('.', 1)
|
|
zip_filename = f"{doc_name}_v{version_num}.{extension}"
|
|
else:
|
|
zip_filename = f"{doc_name}_v{version_num}"
|
|
|
|
# Añadir archivo al ZIP
|
|
zipf.write(file_path, arcname=f'documents/{zip_filename}')
|
|
|
|
# Incluir metadatos del documento si se solicita
|
|
if include_metadata:
|
|
# Crear metadatos simplificados
|
|
doc_meta = {
|
|
'nombre': doc_name,
|
|
'version': version_num,
|
|
'fecha_creacion': version_meta.get('created_at'),
|
|
'creado_por': version_meta.get('created_by'),
|
|
'descripcion': version_meta.get('description', ''),
|
|
'tamano': version_meta.get('file_size', 0)
|
|
}
|
|
|
|
# Guardar metadatos en archivo temporal
|
|
doc_meta_file = os.path.join(temp_dir, f'{doc_name}_meta.json')
|
|
with open(doc_meta_file, 'w', encoding='utf-8') as f:
|
|
json.dump(doc_meta, f, ensure_ascii=False, indent=2)
|
|
|
|
# Añadir al ZIP
|
|
zipf.write(doc_meta_file, arcname=f'documents/{doc_name}_meta.json')
|
|
|
|
return True, f"Proyecto exportado correctamente. {len(documents)} documentos incluidos.", zip_path
|
|
|
|
except Exception as e:
|
|
return False, f"Error al exportar proyecto: {str(e)}", None
|
|
|
|
finally:
|
|
# Limpiar directorio temporal
|
|
import shutil
|
|
shutil.rmtree(temp_dir)
|
|
|
|
def export_document_versions(project_id, document_id, include_all_versions=False):
|
|
"""
|
|
Exportar todas las versiones de un documento específico.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
document_id (int): ID del documento
|
|
include_all_versions (bool, optional): Incluir todas las versiones.
|
|
Si es False, solo se incluye la última versión.
|
|
|
|
Returns:
|
|
tuple: (success, message, zip_path)
|
|
"""
|
|
from services.document_service import get_document, find_document_directory
|
|
|
|
# Verificar que el proyecto existe
|
|
project = get_project(project_id)
|
|
|
|
if not project:
|
|
return False, f"No se encontró el proyecto con ID {project_id}.", None
|
|
|
|
# Obtener directorio del proyecto
|
|
project_dir = find_project_directory(project_id)
|
|
|
|
if not project_dir:
|
|
return False, f"No se encontró el directorio del proyecto con ID {project_id}.", None
|
|
|
|
# Obtener documento
|
|
document = get_document(project_id, document_id)
|
|
|
|
if not document:
|
|
return False, f"No se encontró el documento con ID {document_id}.", None
|
|
|
|
# Obtener directorio del documento
|
|
document_dir = find_document_directory(project_dir, document_id)
|
|
|
|
if not document_dir:
|
|
return False, f"No se encontró el directorio del documento con ID {document_id}.", None
|
|
|
|
# Crear nombre para el archivo ZIP
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
doc_name = document.get('document_id', '').split('_', 1)[1] if '_' in document.get('document_id', '') else f'doc_{document_id}'
|
|
zip_filename = f"{doc_name}_export_{timestamp}.zip"
|
|
|
|
# Ruta completa al archivo ZIP
|
|
storage_path = current_app.config['STORAGE_PATH']
|
|
exports_dir = os.path.join(storage_path, 'exports')
|
|
ensure_dir_exists(exports_dir)
|
|
zip_path = os.path.join(exports_dir, zip_filename)
|
|
|
|
try:
|
|
# Crear archivo ZIP
|
|
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
|
# Obtener versiones
|
|
versions = document.get('versions', [])
|
|
|
|
if not versions:
|
|
return False, "El documento no tiene versiones.", None
|
|
|
|
# Si solo se quiere la última versión
|
|
if not include_all_versions:
|
|
latest_version = max(versions, key=lambda v: v['version'])
|
|
versions = [latest_version]
|
|
|
|
# Añadir cada versión al ZIP
|
|
for version in versions:
|
|
version_num = version.get('version', 1)
|
|
version_filename = version.get('filename')
|
|
|
|
if not version_filename:
|
|
continue
|
|
|
|
file_path = os.path.join(document_dir, version_filename)
|
|
|
|
if not os.path.exists(file_path):
|
|
continue
|
|
|
|
# Añadir archivo al ZIP
|
|
zipf.write(file_path, arcname=f'v{version_num}_{version_filename}')
|
|
|
|
# Añadir metadatos
|
|
meta_data = {
|
|
'document_id': document.get('document_id'),
|
|
'original_filename': document.get('original_filename'),
|
|
'versions_included': len(versions),
|
|
'export_date': datetime.now(pytz.UTC).isoformat()
|
|
}
|
|
|
|
# Crear archivo temporal para metadatos
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as temp:
|
|
json.dump(meta_data, temp, ensure_ascii=False, indent=2)
|
|
temp_path = temp.name
|
|
|
|
# Añadir metadatos al ZIP
|
|
zipf.write(temp_path, arcname='document_info.json')
|
|
|
|
# Eliminar archivo temporal
|
|
os.unlink(temp_path)
|
|
|
|
return True, f"Documento exportado correctamente. {len(versions)} versiones incluidas.", zip_path
|
|
|
|
except Exception as e:
|
|
return False, f"Error al exportar documento: {str(e)}", None
|
|
|
|
def create_project_report(project_id, report_type='summary'):
|
|
"""
|
|
Crear un informe del proyecto en formato JSON.
|
|
|
|
Args:
|
|
project_id (int): ID del proyecto
|
|
report_type (str, optional): Tipo de informe ('summary', 'detailed').
|
|
Por defecto es 'summary'.
|
|
|
|
Returns:
|
|
tuple: (success, message, report_data)
|
|
"""
|
|
# Verificar que el proyecto existe
|
|
project = get_project(project_id)
|
|
|
|
if not project:
|
|
return False, f"No se encontró el proyecto con ID {project_id}.", None
|
|
|
|
# Obtener documentos del proyecto
|
|
documents = get_project_documents(project_id)
|
|
|
|
# Crear informe básico
|
|
report = {
|
|
'project_code': project.get('codigo'),
|
|
'description': project.get('descripcion'),
|
|
'client': project.get('cliente'),
|
|
'destination': project.get('destinacion'),
|
|
'creation_year': project.get('ano_creacion'),
|
|
'creation_date': project.get('fecha_creacion'),
|
|
'created_by': project.get('creado_por'),
|
|
'status': project.get('estado'),
|
|
'last_modified': project.get('ultima_modificacion'),
|
|
'modified_by': project.get('modificado_por'),
|
|
'document_count': len(documents),
|
|
'report_generated': datetime.now(pytz.UTC).isoformat(),
|
|
'report_type': report_type
|
|
}
|
|
|
|
# Si es un informe detallado, incluir información de documentos
|
|
if report_type == 'detailed':
|
|
report['documents'] = []
|
|
|
|
for doc in documents:
|
|
doc_info = {
|
|
'id': doc.get('id'),
|
|
'name': doc.get('document_id', '').split('_', 1)[1] if '_' in doc.get('document_id', '') else '',
|
|
'original_filename': doc.get('original_filename'),
|
|
'version_count': len(doc.get('versions', [])),
|
|
}
|
|
|
|
# Incluir información de la última versión
|
|
if doc.get('versions'):
|
|
latest_version = max(doc.get('versions', []), key=lambda v: v['version'])
|
|
doc_info['latest_version'] = {
|
|
'version': latest_version.get('version'),
|
|
'created_at': latest_version.get('created_at'),
|
|
'created_by': latest_version.get('created_by'),
|
|
'description': latest_version.get('description'),
|
|
'file_size': latest_version.get('file_size'),
|
|
'download_count': len(latest_version.get('downloads', []))
|
|
}
|
|
|
|
report['documents'].append(doc_info)
|
|
|
|
return True, "Informe generado correctamente.", report
|