from fastapi import requests import psycopg2 import requests class IncidentConnection(): conn = None def __init__(self): self.dbname = "tcsdb" self.user = "tcsadmin" self.password = "tcsadmin.2012" self.host = "198.251.74.100" self.port = "5432" self.schema = "-c search_path=hawa_consume" def open_connection(self): """Abre una nueva conexión a la base de datos.""" try: conn = psycopg2.connect( dbname=self.dbname, user=self.user, password=self.password, host=self.host, port=self.port, options=self.schema ) return conn except psycopg2.OperationalError as err: print(f"Error de conexión: {err}") return None def close_connection(self, conn): print("""Cierra la conexión a la base de datos.""") if conn: conn.close() def read_all(self): print("""Ejecuta una consulta y devuelve los resultados.""") conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute(""" SELECT * FROM "incidentes" """) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return[] # endpoint para extraer los ticket en base al sla def read_all_tickets_by_sla(self, sla_prefix): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT * FROM "tickets_by_sla" WHERE upper(sla) LIKE %s""", (sla_prefix.upper() + '%',)) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] # endpoint para extraer los ticket en base al sla def read_all_tickets_noparams_sla(self): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT * FROM "tickets_by_sla" """) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] # endpoint para extraer los ticket en base al sla def read_all_tickets_no_params_sla_c_t_s(self): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT * FROM tickets_by_sla_c_t_s """) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #enpoint para devolver los tickets mas repetidos por mes y basado en la comparativa anual def read_ticket_rep_and_comp_year(self, anio1, anio2, mes): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT i.categoria, i.tipo, SUM(CASE WHEN EXTRACT(YEAR FROM i.fecha_de_registro) = %s THEN 1 ELSE 0 END) AS incidentes_2019, SUM(CASE WHEN EXTRACT(YEAR FROM i.fecha_de_registro) = %s THEN 1 ELSE 0 END) AS incidentes_2020 FROM incidentes i WHERE EXTRACT(MONTH FROM i.fecha_de_registro) = %s AND (EXTRACT(YEAR FROM i.fecha_de_registro) = %s OR EXTRACT(YEAR FROM i.fecha_de_registro) = %s) GROUP BY i.categoria, i.tipo ORDER BY i.categoria, i.tipo""", (anio1, anio2, mes, anio1, anio2)) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #enpoint para devolver los tickets mas repetidos por mes y basado en la comparativa anual def read_ticket_cad_y_nocad_year(self): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT * FROM tickets_cad_y_nocad_by_anios""") data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] # endpoint para extraer la tendencia de errores de incidentes en base al tipo def read_tendencia_by_tipo(self, anio1, anio2): conn = self.open_connection() if conn is not None: with self.conn.cursor() as cur: cur.execute("""SELECT tipo, titulo, SUM(CASE WHEN EXTRACT(YEAR FROM fecha_de_registro) = %s THEN 1 ELSE 0 END) AS cantidad_tickets_2019, SUM(CASE WHEN EXTRACT(YEAR FROM fecha_de_registro) = %s THEN 1 ELSE 0 END) AS cantidad_tickets_2020 FROM incidentes WHERE EXTRACT(YEAR FROM fecha_de_registro) IN (%s, %s) GROUP BY tipo, titulo""", (anio1, anio2, anio1, anio2)) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] # endpoint para extraer la tendencia de errores de incidentes en base al tipo y categoria def read_tendencias_tickets_all_anios(self): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT * FROM view_tendencia_tickets_all_anios""") data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #endpoint para extraer la data de calificaciones de por empresa def read_all_califications(self): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""select * from vista_calificaciones""") data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #endpoint para extraer la informacion de los incidentes en base a la categoria def read_by_category(self, categoryName): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT * FROM "incidentes" WHERE categoria = %s""", (categoryName,)) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #endpoint par extraer la infromacion en base al tecnico def read_by_technician(self, technician_name): conn = self.open_connection() if conn is not None: technician_name=technician_name+' ' nombres = technician_name.split(' ') with conn.cursor() as cur: cur.execute("""SELECT * FROM vista_promedio_tiempo_resolucion WHERE upper(tecnico_de_segunda_linea) LIKE %s AND upper(tecnico_de_segunda_linea) LIKE %s""", ('%' + nombres[0].upper() + '%','%'+nombres[1].upper()+'%')) data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #enpoint para extraer el grupo de soporte que resuelve más incidentes def read_total_incidents_by_group(self): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""SELECT * FROM incidentes_summary_view""") data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #enpoint para extraer el grupo de soporte que resuelve más incidentes def read_sla_cads_nocads_percents(self): conn = self.open_connection() if conn is not None: with conn.cursor() as cur: cur.execute("""select * from sla_cads_nocads_percents""") data = cur.fetchall() if cur.rowcount > 0 else [] self.close_connection(conn) return data return [] #-------------------------------------------------------------------------------------------------- @staticmethod def get_data_in_url(url, token): headers = {"Authorization": f"Bearer {token}"} response = requests.get(url, headers=headers) return response.json() #-------------------------------------------------------------------------------------------------- #endpoint de insercion de ejemplo def write(self, data): with self.conn.cursor()as cur: cur.execute("""INSERT INTO "incident"(name,phone) VALUES(%(name)s, %(phone)s)""",data) self.conn.commit() def __def__(self): self.conn.close() # def __init__(self): # try: # self.conn = psycopg2.connect(dbname="tcsdb", user="tcsadmin", password="tcsadmin.2012", host="198.251.74.100", port="5432", options="-c search_path=hawa_consume") # # self.conn = psycopg2.connect("dbname=hawadata user=root password=2058yrd7HToLT1MAQxfkpKmpXz1oFPHI host=dpg-cnr8kqmct0pc73cqpt3g-a.oregon-postgres.render.com port=5432") # except psycopg2.OperationalError as err: # print(err) # self.conn.close() # # endpoint para extraer todos los incidentes sin parametros # def read_all(self): # with self.conn.cursor() as cur: # cur.execute(""" SELECT * FROM "incidentes" """) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # # endpoint para extraer los ticket en base al sla # def read_all_tickets_by_sla(self, sla_prefix): # with self.conn.cursor() as cur: # cur.execute("""SELECT * FROM "tickets_by_sla" WHERE upper(sla) LIKE %s""", (sla_prefix.upper() + '%',)) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # # endpoint para extraer los ticket en base al sla # def read_all_tickets_noparams_sla(self): # with self.conn.cursor() as cur: # cur.execute("""SELECT * FROM "tickets_by_sla" """) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # # endpoint para extraer los ticket en base al sla # def read_all_tickets_no_params_sla_c_t_s(self): # with self.conn.cursor() as cur: # cur.execute("""SELECT * FROM tickets_by_sla_c_t_s """) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #enpoint para devolver los tickets mas repetidos por mes y basado en la comparativa anual # def read_ticket_rep_and_comp_year(self, anio1, anio2, mes): # with self.conn.cursor() as cur: # cur.execute("""SELECT # i.categoria, # i.tipo, # SUM(CASE WHEN EXTRACT(YEAR FROM i.fecha_de_registro) = %s THEN 1 ELSE 0 END) AS incidentes_2019, # SUM(CASE WHEN EXTRACT(YEAR FROM i.fecha_de_registro) = %s THEN 1 ELSE 0 END) AS incidentes_2020 # FROM # incidentes i # WHERE # EXTRACT(MONTH FROM i.fecha_de_registro) = %s # AND (EXTRACT(YEAR FROM i.fecha_de_registro) = %s OR EXTRACT(YEAR FROM i.fecha_de_registro) = %s) # GROUP BY # i.categoria, # i.tipo # ORDER BY # i.categoria, # i.tipo""", # (anio1, anio2, mes, anio1, anio2)) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #enpoint para devolver los tickets mas repetidos por mes y basado en la comparativa anual # def read_ticket_cad_y_nocad_year(self): # with self.conn.cursor() as cur: # cur.execute("""SELECT * # FROM # tickets_cad_y_nocad_by_anios""") # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # # endpoint para extraer la tendencia de errores de incidentes en base al tipo # def read_tendencia_by_tipo(self, anio1, anio2): # with self.conn.cursor() as cur: # cur.execute("""SELECT tipo, titulo, # SUM(CASE WHEN EXTRACT(YEAR FROM fecha_de_registro) = %s THEN 1 ELSE 0 END) AS cantidad_tickets_2019, # SUM(CASE WHEN EXTRACT(YEAR FROM fecha_de_registro) = %s THEN 1 ELSE 0 END) AS cantidad_tickets_2020 # FROM incidentes # WHERE EXTRACT(YEAR FROM fecha_de_registro) IN (%s, %s) # GROUP BY tipo, titulo""", # (anio1, anio2, anio1, anio2)) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # # endpoint para extraer la tendencia de errores de incidentes en base al tipo y categoria # def read_tendencias_tickets_all_anios(self): # with self.conn.cursor() as cur: # cur.execute("""SELECT * FROM view_tendencia_tickets_all_anios""") # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #endpoint para extraer la data de calificaciones de por empresa # def read_all_califications(self): # with self.conn.cursor() as cur: # cur.execute("""select * from vista_calificaciones""") # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #endpoint para extraer la informacion de los incidentes en base a la categoria # def read_by_category(self, categoryName): # with self.conn.cursor() as cur: # cur.execute("""SELECT * FROM "incidentes" WHERE categoria = %s""", (categoryName,)) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #endpoint par extraer la infromacion en base al tecnico # def read_by_technician(self, technician_name): # technician_name=technician_name+' ' # nombres = technician_name.split(' ') # with self.conn.cursor() as cur: # cur.execute("""SELECT * FROM vista_promedio_tiempo_resolucion WHERE upper(tecnico_de_segunda_linea) LIKE %s AND upper(tecnico_de_segunda_linea) LIKE %s""", ('%' + nombres[0].upper() + '%','%'+nombres[1].upper()+'%')) # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #enpoint para extraer el grupo de soporte que resuelve más incidentes # def read_total_incidents_by_group(self): # with self.conn.cursor() as cur: # cur.execute("""SELECT * FROM incidentes_summary_view""") # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #enpoint para extraer el grupo de soporte que resuelve más incidentes # def read_sla_cads_nocads_percents(self): # with self.conn.cursor() as cur: # cur.execute("""select * from sla_cads_nocads_percents""") # data = cur.fetchall() if cur.rowcount > 0 else [] # return data # #-------------------------------------------------------------------------------------------------- # @staticmethod # def get_data_in_url(url, token): # headers = {"Authorization": f"Bearer {token}"} # response = requests.get(url, headers=headers) # return response.json() # #-------------------------------------------------------------------------------------------------- # #endpoint de insercion de ejemplo # def write(self, data): # with self.conn.cursor()as cur: # cur.execute("""INSERT INTO "incident"(name,phone) VALUES(%(name)s, %(phone)s)""",data) # self.conn.commit() # def __def__(self): # self.conn.close()