# Optimized Python Code # Archivo: DMControllerBattCharger.py # Ruta: src\App\Ventas_Module\VentasEUA\DataMaster\DMControllerBattCharger.py # Lenguaje: Python con Flask from flask import Flask, render_template, send_file, abort, url_for, jsonify, request,session import os from dotenv import load_dotenv import ftplib from werkzeug.utils import secure_filename import tempfile from App.Security_Module.UserPassword import user_has_access from App.Subir_Archivo import subir_archivo_ftp_desde_request,view_pdf_file,verificar_actualizar_archivos from Consultas_SQL.Ventas.VentasEUA.DataMaster.DMControllerBattChargerSQL import ( get_Catalogo_de_cargadores_de_baterias, insert_Catalogo_de_cargadores_de_baterias,update_Catalogo_de_cargadores_de_baterias,delete_Catalogo_de_cargadores_de_baterias,item_exists,searchitems,update_Catalogo_de_cargadores_de_baterias_TechnicalSheet ) import requests import os # Cargar las variables de entorno desde el archivo .env load_dotenv() # ///////////////////////////////// # Variables Generales # //////////////////////////////// ruta_pdf = "/domains/sycelephant.com/public_html/file/Ventas/EUA/DMITEMS/CONTROLLER_BATTCHARGER" carpeta = "/Ventas/EUA/DMITEMS/CONTROLLER_BATTCHARGER/" # ///////////////////////////////// # Funciones Genericas # //////////////////////////////// # ///////////////////////////////// # Varifica items Repetidos # //////////////////////////////// def existen_items(item): result = item_exists(item) if result: # Verifica si result no es None (puede ser None si la consulta no encuentra nada) count = int(result[0]) # Accede al primer elemento de la tupla (índice 0) y lo convierte a entero. else: count = 0 return count # ///////////////////////////////// # Funciones dedicadas # //////////////////////////////// def register_dmCatalogo_de_cargadores_de_baterias_routes(app): """ Registra las rutas para la gestión de departamentos directamente en la aplicación Flask. """ # ///////////////////////////////// # Consultar Breakers # //////////////////////////////// @app.route("/ventas/data_master/Catalogo_de_cargadores_de_baterias/listCatalogo_de_cargadores_de_baterias", methods=["GET"]) def list_Catalogo_de_cargadores_de_baterias(): """ user_id = session.get('user_id') if not user_has_access(user_id, "Ventas","read"): message = "No tiene permisos para visualizar este archivo" return render_template("Security/AccessDened.html",message) """ """Retorna la lista de get_Catalogo_de_cargadores_de_baterias.""" try: TIPOSCONTROLES = get_Catalogo_de_cargadores_de_baterias() return jsonify(TIPOSCONTROLES), 200 except Exception as e: return jsonify({"error": str(e)}), 500 # Función para verificar si un item ya existe en la base de datos @app.route("/ventas/pdf_viewerCatalogo_de_cargadores_de_baterias/view/") def ver_pdfCatalogo_de_cargadores_de_baterias(filename_or_url): """Endpoint para visualizar un archivo PDF específico.""" return view_pdf_file(filename_or_url) # ///////////////////////////////// # Guardar Breakers # //////////////////////////////// @app.route("/ventas/data_master/brakers/addCatalogo_de_cargadores_de_baterias", methods=["POST"]) def add_Catalogo_de_cargadores_de_baterias(): user_id = session.get('user_id') try: item = request.form.get("item") description = request.form.get("description") price = request.form.get("price") technicalSheet = request.files.get("TechnicalSheet") if not all([item, description, price]): return jsonify({"error": "Todos los campos son obligatorios."}), 400 # Verificar si el item ya existe if existen_items(item) == 1: return jsonify({"error": "El item ya existe en el catálogo."}), 400 filename = None if technicalSheet: # Definir la ruta donde guardar el PDF # Usar la función para guardar el archivo #resultado = guardar_archivo_pdf(technicalSheet, item, ruta_pdf) # Usar la función para subir el archivo via SFTP resultado = subir_archivo_ftp_desde_request(technicalSheet, item, ruta_pdf,carpeta) if resultado["exito"]: filename = resultado["url_web"] else: return jsonify({"error": resultado["mensaje"]}), 400 # Database Insert result = insert_Catalogo_de_cargadores_de_baterias(item, description, price, filename) return jsonify({"message": "Catálogo de cargadores de baterías agregado exitosamente."}), 201 except Exception as e: return jsonify({"error": str(e)}), 500 # ///////////////////////////////// # Actualizar inputs de Breakers # //////////////////////////////// @app.route("/ventas/data_master/updateCatalogo_de_cargadores_de_baterias", methods=["POST"]) def update_Catalogo_de_cargadores_de_baterias_route(): """Actualiza un breaker en la base de datos.""" try: data = request.get_json() item = data.get("item") description = data.get("description") price = data.get("price") if not all([item, description, price]): return jsonify({"error": "Todos los campos son obligatorios."}), 400 update_Catalogo_de_cargadores_de_baterias(item, description, price) return jsonify({"message": "Catálogo de cargadores de baterías actualizado exitosamente."}), 200 except Exception as e: return jsonify({"error": str(e)}), 500 # ///////////////////////////////// # Eliminar Breakers # //////////////////////////////// @app.route("/ventas/data_master/Catalogo_de_cargadores_de_baterias/deleteCatalogo_de_cargadores_de_baterias", methods=["POST"]) def delete_Catalogo_de_cargadores_de_baterias_route(): """Elimina un control de la base de datos.""" try: data = request.get_json() item = data.get("item") if not item: return jsonify({"error": "El campo Item es obligatorio."}), 400 delete_Catalogo_de_cargadores_de_baterias(item) return jsonify({"message": "Catálogo de cargadores de baterías eliminado exitosamente."}), 200 except Exception as e: return jsonify({"error": str(e)}), 500 # ///////////////////////////////// # Revisar la tabla CZE_BreakersCatalog y verificar si estan en archivos de fichas tecnicas # //////////////////////////////// @app.route("/ventas/data_master/Catalogo_de_cargadores_de_baterias/actualizarCatalogo_de_cargadores_de_baterias", methods=["GET"]) def updated_Catalogo_de_cargadores_de_baterias_route(): """Actualiza todas las Fichas técnicas verificando su existencia en el servidor.""" try: # Configuramos la ruta al módulo y el nombre de la función module_path = "Consultas_SQL.Ventas.VentasEUA.DataMaster.DMControllerBattChargerSQL" function_name = "searchitems" # Llamamos a verificar_actualizar_archivos con la configuración dinámica resultado = verificar_actualizar_archivos( tabla="CZE_ControllerBattCharger", campo_id="Item_batterycharger", campo_url="TechnicalSheet", update_function=update_Catalogo_de_cargadores_de_baterias_TechnicalSheet, ruta_remota=ruta_pdf, module_path=module_path, # Pasamos la ruta al módulo search_function_name=function_name, # Pasamos el nombre de la función base_url="https://file.sycelephant.com", carpeta = carpeta ) if resultado["exito"]: return jsonify({"message": resultado["mensaje"]}), 200 else: print(f"Error en actualización: {resultado['mensaje']}") return jsonify({"error": resultado["mensaje"]}), 500 except Exception as e: import traceback print(f"Error en la actualización: {str(e)}") print(traceback.format_exc()) return jsonify({"error": str(e)}), 500