Version 1.0

Connect_CC2.py

+import requests
+
+class Callcenter:
+    def __init__(self,api_key,url):
+        self.url=url
+        # Paso 1: Hacer un POST para obtener el token
+        requests.packages.urllib3.disable_warnings()
+        url_token = f"{self.url}/api/authentication_apikey_token"
+        payload = {"api_key": f"{api_key}"}
+        response = requests.post(url_token, json=payload, verify=False)
+        self.token = response.json()["token"]
+
+        # Paso 2: Enviar un GET con el token obtenido en el encabezado y obtener la Cookie
+        self.user_center_id = 42
+        url_center = f"{self.url}/api/centers/select/{self.user_center_id}"
+        headers = {"CCAuthorization": f"Bearer {self.token}"}
+        response = requests.get(url_center, headers=headers, verify=False)
+        self.cookie = response.headers["Set-Cookie"]
+        print(self.cookie)
+
+    def search_campaing(self,id_campaign):
+        url_center = f"{self.url}/api/outgoingcampaigns/{id_campaign}"
+        print(f"url: {url_center}")
+        headers = {
+            "Content-Type": "application/json",
+			"Accept": "application/json",
+            "CCAuthorization": f"Bearer {self.token}",
+            "Cookie": self.cookie
+        }
+        response = requests.get(url_center, headers=headers,  verify=False)
+        return response
+    def insert_number_campaing(self,id_campaign,campaign_fields,campaign_data):
+        url_center = f"{self.url}/api/outgoingcampaigns/{id_campaign}/add-calls"
+        headers = {
+            "Content-Type": "application/json",
+			"Accept": "application/json",
+            "CCAuthorization": f"Bearer {self.token}",
+            "Cookie": self.cookie
+        }      
+        payload = {
+           "fields": campaign_fields,
+           "data": campaign_data
+        }
+        response = requests.post(url_center, headers=headers, json=payload, verify=False)
+        return response.content
\ No newline at end of file

Connect_Db.py

+import mysql.connector
+
+class MySQLDatabase:
+    def __init__(self, host, user, password, database):
+        try:
+            self.connection = mysql.connector.connect(
+                host=host,
+                user=user,
+                password=password,
+                database=database
+            )
+            self.cursor = self.connection.cursor(dictionary=True)
+            print("Conexión exitosa a la base de datos ",database)
+        except mysql.connector.Error as err:
+            print(f"Error al conectar: {err}")
+
+    def consultar(self, query, params=None):
+        try:
+            self.cursor.execute(query, params)
+            return self.cursor.fetchall()
+        except mysql.connector.Error as err:
+            print(f"Error en la consulta: {err}")
+            return None
+
+    def insertar(self, query, params=None):
+        try:
+            self.cursor.execute(query, params)
+            self.connection.commit()
+            print("Inserción exitosa")
+            return self.cursor.lastrowid
+        except mysql.connector.Error as err:
+            print(f"Error al insertar: {err}")
+            self.connection.rollback()
+            return None
+
+    def actualizar(self, query, params=None):
+        result=False
+        try:
+            self.cursor.execute(query, params)
+            self.connection.commit()
+            print("Actualizar OK: ",params)
+            result=True
+            return result
+        except mysql.connector.Error as err:
+            print(f"Error al insertar: {err}")
+            self.connection.rollback()
+            result=False
+            return result
+
+    def cerrar(self):
+        self.cursor.close()
+        self.connection.close()
+        print("Conexión cerrada")

README.md

 # Webservice Marcimex
 
+Webservice para la inserción de númros a una campaña existente.
+Chequeos de ingreso de datos en la misma
+Primer test sobre base local e inserción de números en una campaña
\ No newline at end of file

app.py

-from functools import wraps
+import os
+import json
+from Connect_CC2 import Callcenter
+from functions import convertir_fecha_hora,check_date_in_out,limpiar_texto,status_camp,valid_array_fields,valid_array_data,convertir_hora
+from datetime import datetime, timedelta
+from dotenv import load_dotenv
 from flask import Flask, request, jsonify, render_template, redirect, url_for, session, flash
-import pandas as pd
-from mysql.connector import Error
+from Connect_Db import MySQLDatabase
+from flask_jwt_extended import JWTManager, create_access_token, create_refresh_token, jwt_required, get_jwt_identity
+
+db_host = os.getenv("DB_HOST")
+db_user = os.getenv("DB_USER")
+db_pass = os.getenv("DB_PASS")
+db_call_center = os.getenv("DB_NAME_CALL_CENTER")
+db_name_asterisk = os.getenv("DB_NAME_ASTERISK")
+db_name_cdr = os.getenv("DB_NAME_CDR")
+db_name_hiper_service = os.getenv("DB_NAME_HIPER_SERVICE")
+
+cc_url = 'https://mxcc-gruposerviandina.hiperpbx.com'
+api_key  = '07j3FHMnDNrIV43A2eB8BlNEDEf1FVXQQtDuNm82jaPANIetnYTiL97QzenOdlbGKEqyd8rtD0HQWtaQ'
+
+
 
 
 app = Flask(__name__) 
+app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET_KEY", "clave_por_defecto")
+app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(minutes=int(os.getenv("JWT_ACCESS_TOKEN_EXPIRES", 5)))
+app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=int(os.getenv("JWT_REFRESH_TOKEN_EXPIRES", 1)))
+app.config["JWT_TOKEN_LOCATION"] = os.getenv("JWT_TOKEN_LOCATION", "headers").split(",")
+app.config["JWT_HEADER_NAME"] = os.getenv("JWT_HEADER_NAME", "Authorization")
+app.config["JWT_HEADER_TYPE"] = os.getenv("JWT_HEADER_TYPE", "Bearer")
+jwt = JWTManager(app)
+USUARIOS = {"admin": "1234"}
+fecha_hora = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+#formato_convertir_fecha = "%d/%m/%Y %H:%M:%S"
 
 @app.route('/')
 def home(): 
-    return "¡Hola, mundo!"
+    namecampaign='Camp_Agente_1007'
+    db = MySQLDatabase(host="localhost", user="root", password="", database="call_center")
+    resultado = db.consultar("select id, estatus from campaign where name=%s limit 1", (f'{namecampaign}',))
+    return jsonify(resultado)
+
+@app.route('/login', methods=['POST'])
+def login():
+    datos = request.json
+    usuario = datos.get("usuario")
+    contrasena = datos.get("contrasena")
+    db = MySQLDatabase(host=db_host, user=db_user, password="", database=db_name_hiper_service)
+    resultado = db.consultar("SELECT * FROM hiper_service_usuarios WHERE users = %s AND password = %s",(f'{usuario}',f'{contrasena}'))
+    if resultado:
+        access_token = create_access_token(identity=usuario)
+        refresh_token = create_refresh_token(identity=usuario)
+        nuevo_id = db.insertar("INSERT INTO hiper_service_accesos (usuario, token ,fecha_hora, status, message) VALUES (%s, %s, %s, %s, %s)",(f'{usuario}',f'{access_token}',f'{fecha_hora}','AccessToken','OK'))
+        db.cerrar()
+        return jsonify({
+            "access_token": access_token,
+            "refresh_token": refresh_token
+        }), 200
+    else:
+        nuevo_id = db.insertar("INSERT INTO hiper_service_accesos (usuario, token ,fecha_hora, status, message) VALUES (%s, %s, %s, %s, %s)",(f'{usuario}','',f'{fecha_hora}','AccessToken','Error'))
+        db.cerrar()
+        return jsonify({"error": "Access Token Error"}), 401
+
+@app.route('/Generate_Campaign', methods=['GET'])
+@jwt_required()
+def Generate_Campaign():
+    usuario = get_jwt_identity()
+    db = MySQLDatabase(host=db_host, user=db_user, password="", database=db_name_hiper_service)
+    nuevo_id = db.insertar("INSERT INTO hiper_service_accesos (usuario, token ,fecha_hora, status, message) VALUES (%s, %s, %s, %s, %s)",(f'{usuario}','',f'{fecha_hora}','Generate Campaing','Access'))
+    db.cerrar()
+    request_data = request.get_json()
+    dataCollection=request_data["dataCollection"]
+    campaign_id=request_data["campaign_id"]
+    campaign_name=request_data["campaign_name"]
+    campaign_date_ini=request_data["campaign_date_ini"]
+    campaign_date_fin=request_data["campaign_date_fin"]
+    campaign_queue=request_data["campaign_queue"]
+    campaign_status=request_data["campaign_status"]
+    campaign_observ=request_data["campaign_observ"]
+    campaign_fields=request_data["fields"]
+    campaign_data=request_data["data"]
+    #Validar campos de fecha y hora y diferencias de tiempos entre ambas
+    date_ini_check = convertir_fecha_hora(campaign_date_ini)
+    date_ini_result = date_ini_check[0]
+    date_ini_camp = date_ini_check[1]
+    time_ini_camp = date_ini_check[2]
+    date_fin_check = convertir_fecha_hora(campaign_date_fin)
+    date_fin_result = date_fin_check[0]
+    date_fin_camp = date_fin_check[1]
+    time_fin_camp = date_fin_check[2]
+    if (date_ini_result==True & date_fin_result==True):
+        #Chequear diferencias de Tiempos entre ambas fechas y horas
+        date_in=str(date_ini_camp) + ' ' + str(time_ini_camp) 
+        date_out=str(date_fin_camp) + ' ' + str(time_fin_camp) 
+        check_date = check_date_in_out(date_in,date_out)
+        if(check_date==False):
+            return jsonify({"mensaje": f"Error al generar Campaing , {campaign_name}! Fecha y Hora Inicial debe ser menor a Fecha y Hora Final"}), 401
+    else:
+        return jsonify({"mensaje": f"Error al generar Campaing , {campaign_name}! Formato Invalido Fecha y Hora"}), 401
+    #Validar nombres permitidos en campaña
+    check_name_camp=limpiar_texto(campaign_name)
+    check_name_camp_result=check_name_camp[0]
+    if(check_name_camp_result==True):
+        camp_name=check_name_camp[1]
+    else:
+        return jsonify({"mensaje": f"Error al generar Campaing , {campaign_name}! Formato Invalido Nombre de Campaña"}), 401
+    #Validar Status Campaña
+    valid_status_camp=status_camp(campaign_status)
+    valid_status_camp_result=valid_status_camp[0]
+    if (valid_status_camp_result==True):
+        camp_status=valid_status_camp[1]
+    else:
+        return jsonify({"mensaje": f"Error al generar Campaing , {campaign_name}!  Status Invalido de Campaña"}), 401
+    #Validar nombres permitidos en observaciones
+    check_name_obsv=limpiar_texto(campaign_observ)
+    check_name_obsv_result=check_name_obsv[0]       
+    if(check_name_obsv_result==True):
+        camp_obsv=check_name_obsv[1]
+    else:
+        return jsonify({"mensaje": f"Error al generar Campaing , {campaign_name}! Formato Invalido Observaciones"}), 401
+    #Validar Campo Fields campo 0 =phone resto string
+    check_fields=valid_array_fields(campaign_fields) 
+    if(check_fields[0]==True):
+        check_field_cant = check_fields[1]
+        check_field_data = check_fields[2]
+    else:
+        return jsonify({"mensaje": f"Error al generar Campaing , {campaign_name}! Formato Invalido Fileds"}), 401
+    #check_data=valid_array_data(campaign_data,check_field_cant)
+    #Modificacion debido que primer campo debe ser numerico
+    check_data=True
+    if(check_data==True):
+        return_data='ok'
+    else:
+        return jsonify({"mensaje": f"Error al generar Campaing , {campaign_name}! Formato Invalido Data"}), 401
+    #Fin de validaciones
+    #Buscar el nombre de la campaña en la base de datos y tomar el valor del ID
+    db_cc = MySQLDatabase(host=db_host, user=db_user, password="", database=db_call_center)
+    resultado_campaign = db_cc.consultar("SELECT * FROM campaign WHERE name = %s limit 1",(f'{camp_name}',))
+    if resultado_campaign:
+        for registro in resultado_campaign:
+            camp_id_search=registro['id']
+            camp_name_search=registro['name']
+            camp_status_search=registro['estatus']
+            camp_datetime_init_search=registro['datetime_init']
+            camp_datetime_end_search=registro['datetime_end']
+            camp_daytime_init_search=str(registro['daytime_init'])
+            camp_daytime_end_search=str(registro['daytime_end'])
+            camp_queue_search=registro['queue']
+            time_ini = convertir_hora(camp_daytime_init_search)
+            time_out = convertir_hora(camp_daytime_end_search)
+            camp_data_search=[camp_status_search,str(camp_datetime_init_search),str(camp_datetime_end_search),time_ini,time_out]
+            camp_data_input=[camp_status,str(date_ini_camp),str(date_fin_camp),str(time_ini_camp),str(time_fin_camp)]
+            if (camp_data_search==camp_data_input):
+                result="Datos Iguales"
+            else:
+                result='Datos Distintos'
+                #Update de registros en la campaña
+                actualizar_campaign = db_cc.actualizar("update campaign set datetime_init = %s, datetime_end = %s, daytime_init = %s, daytime_end = %s, estatus = %s where id = %s ",(f'{str(date_ini_camp)}',f'{str(date_fin_camp)}',f'{str(time_ini_camp)}',f'{str(time_fin_camp)}',f'{camp_status}',f'{camp_id_search}'))
+            #Conectar al CC2 para insertar datos en la la campaña
+        
+        
+        payload = {
+           "fields": campaign_fields,
+           "data": campaign_data
+        }
+        print(payload)
+        callcenter = Callcenter(api_key, cc_url)
+        id_campaign= 87
+        response = callcenter.search_campaing(id_campaign)
+        data = response.json()['id']
+        print(data)
+        
+        insert=callcenter.insert_number_campaing(data,campaign_fields,campaign_data)
+        print(insert)
+        
+        db_cc.cerrar()
+        #return jsonify({{payload}}), 200
+        return jsonify({"mensaje": f"Camp Ingreso Datos: {camp_data_input}  Camp Resultado: {camp_data_search}  Comparacion: {result} Resgistros: {payload}"}), 200
+        #return jsonify({"mensaje": f"Encontrado Campaing ,ID: {camp_id_search} Name: {camp_name_search} Status:  {camp_status_search} Fecha Comienzo: {camp_datetime_init_search} Hora Comienzo: {camp_daytime_init_search} Fecha Fin: {camp_datetime_end_search} Hora Fin: {camp_daytime_end_search} Queue: {camp_queue_search} ! Validar Registros de Ingreso"}), 200
+    else:
+        return jsonify({"mensaje": f"No Encontrado Campaing , {camp_name}! Se debe crear una campaña"}), 401
+
+
+
+
+
+@jwt.expired_token_loader
+def token_caducado(header, payload):
+    return jsonify({
+        "error": "El token ha caducado. Inicia sesión nuevamente."
+    }), 401    
+@jwt.unauthorized_loader
+def token_no_proporcionado(callback):
+    return jsonify({
+        "error": "Acceso denegado. No se proporcionó un token."
+    }), 401
+@jwt.invalid_token_loader
+def token_invalido(error):
+    return jsonify({
+        "error": "El token proporcionado es inválido o está mal formado."
+    }), 401
+@jwt.needs_fresh_token_loader
+def token_no_fresco():
+    return jsonify({
+        "error": "Se requiere un token fresco para esta operación."
+    }), 401
+
+
 
-@app.route('/prueba')
-def acceso(): 
-    return "¡Hola, Prueba!"
+@app.route('/prueba', methods=['POST'])
+def acceso():
+    headers = request.headers
+    auth = headers.get("Authorization")
+    if auth == 'Bearer cGJ4OnMzWFRnM0hTQ1Y1R':
+        request_data = request.get_json()
+        if 'dataCollection' in request_data:
+            data_collection = request_data['dataCollection']
+            return jsonify({"message": "DataCollection: data_collection"}), 200
+        else:
+            return jsonify({"message": "DataCollection: Not Found"}), 401
+    else:
+       return jsonify({"message": "ERROR: Unauthorized"}), 401 
+    
+    
+    
+    
+#   if auth == 'Bearer cGJ4OnMzWFRnM0hTQ1Y1R':
+#        request_data = request.get_json()
+#        if 'dataCollection' in request_data:
+#            data_collection = request_data['dataCollection']        
+#        
+#        if (data_collection=='crm_insert_campaing'):
+#            id_campaign = request_data['id_campaign']
+#            fields = request_data['fields']
+#            data = request_data['data']
+#            #return jsonify({"message": "OK: Authorized"}), 200
+#            return jsonify({"message": "OK: data_collection"}), 200
+#        else:
+#            return 'Invalid Data Coellections'
+#    else:
+#        return jsonify({"message": "ERROR: Unauthorized"}), 401
+ 
+ 
+    #return auth 
 
 
 if __name__ == '__main__': 

functions.py

+import re
+from datetime import datetime, timedelta
+
+def convertir_fecha_hora(fecha_hora_str):
+    result=False
+    fecha_formato=False
+    hora_formato=False
+    #validar el tipo de formato de hora 7/11/2024 08:05:00(j/m/Y H:i:s) o 07/11/2024 08:05:00(d/m/Y H:i:s)
+    format_date = ["%d/%m/%Y %H:%M:%S", "%-d/%m/%Y %H:%M:%S"]
+    for formato in format_date:
+        try:
+            fecha_hora = datetime.strptime(fecha_hora_str, formato)
+            result=True
+            fecha_formato = fecha_hora.strftime("%Y-%m-%d")
+            hora_formato = fecha_hora.strftime("%H:%M:%S")
+            return result, fecha_formato, hora_formato
+        except ValueError:
+            continue
+    return result, fecha_formato, hora_formato
+      
+def convertir_hora(time_ing):
+    time_ini = datetime.strptime(time_ing, "%H:%M:%S")
+    time_ini = str(time_ini)
+    time_ini = time_ini.split()
+    time_ini = time_ini[1]
+    return time_ini
+
+def check_date_in_out(date_in,date_out):
+    # Convertir a objetos datetime
+    result=False
+    date_ing = datetime.strptime(date_in, '%Y-%m-%d %H:%M:%S')
+    date_fin = datetime.strptime(date_out, '%Y-%m-%d %H:%M:%S')
+    time_ing = str(date_ing)
+    time_ing = time_ing.split()
+    time_ing = time_ing[1]
+    time_fin = str(date_fin)
+    time_fin = time_fin.split()
+    time_fin = time_fin[1]   
+    
+    
+    # Validar fechas y horas
+    if date_ing >= date_fin:
+        result=False
+    else:
+        result=True
+        if time_ing >= time_fin:
+            result=False
+        else:
+            result=True
+    return result
+
+def limpiar_texto(string):
+    result=True
+    texto_limpio = re.sub(r'[^a-zA-Z0-9-_\s]', '', string)
+    #texto_limpio = texto_limpio.encode('utf-8')
+    
+    if(texto_limpio==''):
+        result=False
+        texto_limpio=''    
+    return result,texto_limpio
+
+def status_camp(status_num):
+    #status_num = re.sub(r'[0-9]', '', status)
+    result=False
+    status_data=False  
+    if (status_num=='0'):
+        result=True
+        status_data='A'
+    elif (status_num=='1'):
+        result=True
+        status_data='T'
+    elif (status_num=='9'):
+        result=True
+        status_data='I'
+    else:
+        result=False
+        status_data=False
+    return result,status_data
+
+def valid_array_fields(array_in):
+    result=False
+    cantidad=0
+    datos_limpios=[]
+    
+    # Validar si es una lista
+    if isinstance(array_in, list):
+        cantidad = len(array_in)
+        # Verificar si el primer campo es 'phone'
+        if array_in[0] == "phone":
+            datos_limpios = [re.sub(r'[^a-zA-Z0-9-_\s]', '', item) for item in array_in]
+            result=True
+        else:
+            result=False
+    else:
+        result=False
+    return result,cantidad,datos_limpios
+
+def valid_array_data(array_in,count):
+    result=False
+    cantidad=0
+    if isinstance(array_in, list):
+        patron = re.compile(r'[\w\s.,]+')  # Acepta letras, dígitos, espacios, puntos y comas
+        for i, registro in enumerate(array_in, 1):
+            elementos_validos = all(patron.fullmatch(campo) for campo in registro)
+            if elementos_validos:
+                if(len(registro)==count):
+                    result=True
+                else:
+                    result=False  
+            else:
+                result=False
+    else:
+        result=False
+    return result
\ No newline at end of file

requirements.txt

 blinker==1.9.0