import os import json import re 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 from services.project_service import get_all_projects from services.document_service import get_project_documents def build_search_index(): """ Construir un índice de búsqueda para proyectos y documentos. Returns: tuple: (success, message) """ try: # Obtener todos los proyectos (incluyendo inactivos) all_projects = get_all_projects(include_inactive=True) # Inicializar índice index = { 'projects': {}, 'documents': {}, 'clients': {}, 'years': {}, 'users': {}, 'last_updated': datetime.now(pytz.UTC).isoformat() } # Indexar proyectos for project in all_projects: project_id = int(project.get('codigo', '').replace('PROJ', '')) if not project_id: continue # Datos básicos del proyecto project_data = { 'id': project_id, 'code': project.get('codigo'), 'description': project.get('descripcion', ''), 'client': project.get('cliente', ''), 'destination': project.get('destinacion', ''), 'year': project.get('ano_creacion'), 'status': project.get('estado', 'inactivo'), 'created_by': project.get('creado_por', ''), 'modified_by': project.get('modificado_por', ''), 'directory': project.get('directory', '') } # Añadir al índice de proyectos index['projects'][str(project_id)] = project_data # Añadir a índice de clientes client = project.get('cliente', 'Sin cliente') if client not in index['clients']: index['clients'][client] = [] index['clients'][client].append(project_id) # Añadir a índice de años year = str(project.get('ano_creacion', 'Sin año')) if year not in index['years']: index['years'][year] = [] index['years'][year].append(project_id) # Añadir a índice de usuarios creator = project.get('creado_por', 'desconocido') if creator not in index['users']: index['users'][creator] = {'created': [], 'modified': []} index['users'][creator]['created'].append(project_id) modifier = project.get('modificado_por') if modifier and modifier not in index['users']: index['users'][modifier] = {'created': [], 'modified': []} if modifier: index['users'][modifier]['modified'].append(project_id) # Indexar documentos del proyecto documents = get_project_documents(project_id) for doc in documents: doc_id = doc.get('id') if not doc_id: continue # Extraer nombre del documento doc_name = doc.get('document_id', '').split('_', 1)[1] if '_' in doc.get('document_id', '') else f'doc_{doc_id}' # Datos básicos del documento doc_data = { 'id': doc_id, 'name': doc_name, 'project_id': project_id, 'original_filename': doc.get('original_filename', ''), 'versions': len(doc.get('versions', [])), 'latest_version': max([v.get('version', 0) for v in doc.get('versions', [])]) if doc.get('versions') else 0, 'directory': doc.get('directory', '') } # Añadir al índice de documentos index['documents'][f"{project_id}_{doc_id}"] = doc_data # Guardar índice storage_path = current_app.config['STORAGE_PATH'] index_file = os.path.join(storage_path, 'indices.json') save_json_file(index_file, index) return True, "Índice construido correctamente." except Exception as e: current_app.logger.error(f"Error al construir índice: {str(e)}") return False, f"Error al construir índice: {str(e)}" def search_projects(query, filters=None): """ Buscar proyectos en el índice. Args: query (str): Término de búsqueda filters (dict, optional): Filtros adicionales - client: Cliente específico - year: Año de creación - status: Estado del proyecto - created_by: Usuario creador Returns: list: Lista de proyectos que coinciden con la búsqueda """ # Cargar índice storage_path = current_app.config['STORAGE_PATH'] index_file = os.path.join(storage_path, 'indices.json') index = load_json_file(index_file, {}) if not index or 'projects' not in index: return [] # Preparar resultados results = [] # Convertir query a minúsculas para búsqueda insensible a mayúsculas query = query.lower() # Buscar en proyectos for project_id, project_data in index.get('projects', {}).items(): # Aplicar filtros si existen if filters: # Filtrar por cliente if 'client' in filters and filters['client'] and project_data.get('client') != filters['client']: continue # Filtrar por año if 'year' in filters and filters['year'] and str(project_data.get('year')) != str(filters['year']): continue # Filtrar por estado if 'status' in filters and filters['status'] and project_data.get('status') != filters['status']: continue # Filtrar por creador if 'created_by' in filters and filters['created_by'] and project_data.get('created_by') != filters['created_by']: continue # Si no hay query, incluir todos los proyectos que pasen los filtros if not query: results.append(project_data) continue # Buscar coincidencias en campos relevantes if (query in project_data.get('code', '').lower() or query in project_data.get('description', '').lower() or query in project_data.get('client', '').lower() or query in project_data.get('destination', '').lower()): results.append(project_data) return results def search_documents(query, project_id=None): """ Buscar documentos en el índice. Args: query (str): Término de búsqueda project_id (int, optional): ID del proyecto para limitar la búsqueda Returns: list: Lista de documentos que coinciden con la búsqueda """ # Cargar índice storage_path = current_app.config['STORAGE_PATH'] index_file = os.path.join(storage_path, 'indices.json') index = load_json_file(index_file, {}) if not index or 'documents' not in index: return [] # Preparar resultados results = [] # Convertir query a minúsculas para búsqueda insensible a mayúsculas query = query.lower() # Buscar en documentos for doc_key, doc_data in index.get('documents', {}).items(): # Si se especifica proyecto, filtrar por él if project_id is not None and doc_data.get('project_id') != int(project_id): continue # Si no hay query, incluir todos los documentos del proyecto if not query: results.append(doc_data) continue # Buscar coincidencias en campos relevantes if (query in doc_data.get('name', '').lower() or query in doc_data.get('original_filename', '').lower()): results.append(doc_data) return results def get_index_stats(): """ Obtener estadísticas del índice. Returns: dict: Estadísticas del índice """ # Cargar índice storage_path = current_app.config['STORAGE_PATH'] index_file = os.path.join(storage_path, 'indices.json') index = load_json_file(index_file, {}) if not index: return { 'projects': 0, 'documents': 0, 'clients': 0, 'years': 0, 'users': 0, 'last_updated': None } # Calcular estadísticas stats = { 'projects': len(index.get('projects', {})), 'documents': len(index.get('documents', {})), 'clients': len(index.get('clients', {})), 'years': len(index.get('years', {})), 'users': len(index.get('users', {})), 'last_updated': index.get('last_updated') } return stats def schedule_index_update(): """ Programar actualización periódica del índice. """ from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler() # Programar actualización diaria a las 3:00 AM scheduler.add_job( build_search_index, 'cron', hour=3, minute=0, id='index_update' ) scheduler.start() current_app.logger.info("Actualización de índice programada.")