from selenium import webdriver
from selenium.webdriver.common.by import By
import time
import pandas as pd
import numpy as np
import re
import pymysql
from datetime import datetime, date, timedelta
from random import randint
import os
import sys, pathlib, pymupdf
import requests
import traceback
#conectarse a la bd consultar e insertar data
def ConsultarData(sql):
db = pymysql.connect(host='82.163.176.111',
user='mgmrenso_voldemort',
password='RHd1&n%=@V2@',
db='mgmrenso_dobby')
cursor = db.cursor()
try:
cursor.execute(sql)
result = cursor.fetchone()
# Creación del DataFrame y visualización de los resultados
# obtener nombres de las columnas
columns = [col[0] for col in cursor.description]
df = pd.DataFrame([dict(zip(columns, result))])
return df
except Exception as e:
db.rollback()
# print("Exception Ocurred: ", e)
finally:
db.close()
#funcion de insertardata
def loadtosql(table, data):
start_time2 = time.time()
connection = pymysql.connect(host='82.163.176.111',
user='mgmrenso_voldemort',
password='RHd1&n%=@V2@',
db='mgmrenso_dobby')
cursor = connection.cursor()
cols = ",".join([str(i) for i in data.columns.tolist()])
sql = "INSERT INTO "+table + \
"(" + cols + ") VALUES (" + "%s,"*(len(data.values[0])-1) + "%s)"
lista = list(data.itertuples(index=False, name=None))
try:
cursor.executemany(sql, lista)
connection.commit()
print("Datos copiados correctamente: ", data.shape[0])
print("-En- %s seconds ---" % (time.time() - start_time2))
connection.close()
except Exception as e:
connection.rollback()
print("Exception Ocurred: ", e)
def obtener_url_Factura_Anterior(ContratoAConsultar):
options= webdriver.ChromeOptions()
#options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:99.0) Gecko/20100101 Firefox/99.0")
#driver_path = 'C:\\Program Files (x86)\\chromedriver.exe'
options.add_argument('--headless')
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
driver= webdriver.Chrome(options=options)
#driver = webdriver.Chrome(driver_path,options=options)
#-----
driver.get("https://www.epm.com.co/clientesyusuarios/servicio-al-cliente/tu-factura/")
time.sleep(randint(10, 15))
try:
Chactbox= driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[2]/ul/li[2]')
Chactbox.click()
print("presionó")
except:
print("no presionó error")
print("Presionando de nuevo")
try:
time.sleep(randint(1, 5))
Chactbox= driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[2]/ul/li[2]')
Chactbox.click()
except:
print("no presionó error 2")
#----------------------
time.sleep(randint(10, 15))
#------------------------
try:
IniciarChact=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[5]/div[1]/div[1]/div/div/ul/li[1]/div/button')
IniciarChact.click()
print("presionó IniciarChact")
except:
print("error al presionar IniciarChact")
try:
IniciarChact=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[5]/div[1]/div[1]/div/div/ul/li[1]/div/button')
IniciarChact.click()
except:
print("error al presionar IniciarChact 2")
#-----------------------------
time.sleep(randint(10, 15))
#-----------------------------
try:
BotonFactura=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[4]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[3]/div[1]')
BotonFactura.click()
print("presionó BotonFactura")
except:
print("error al presionar BotonFactura")
try:
BotonFactura=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[4]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[3]/div[1]')
BotonFactura.click()
except:
print("error al presionar BotonFactura2")
#---------------------------
time.sleep(randint(10, 15))
#---------------------------
try:
BotonDuplicado=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[6]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[3]/div[3]')
BotonDuplicado.click()
print("presionó BotonDuplicado")
except:
print("error al presionar BotonDuplicado")
try:
BotonDuplicado=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[6]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[3]/div[3]')
BotonDuplicado.click()
except:
print("error al presionar BotonDuplicado2")
time.sleep(randint(10, 15))
try:
Input=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[5]/div/form/input')
Input.send_keys(str(ContratoAConsultar))#12802914
#presionar el boton de enviar
BotonEnviar=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[5]/div/button')
BotonEnviar.click()
print("presionó BotonEnviar")
except:
print("error al presionar BotonEnviar")
try:
Input=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[5]/div/form/input')
Input.send_keys(str(ContratoAConsultar))#12802914
#presionar el boton de enviar
BotonEnviar=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[5]/div/button')
BotonEnviar.click()
print("presionó BotonEnviar2")
except:
print("error al presionar BotonEnviar2")
#-----------------------------
time.sleep(randint(10, 15))
#------------------------------
try:
BotonFacturaAnterior=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[10]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[3]/div[1]/div/button[2]')
BotonFacturaAnterior.click()
print("presionó BotonFacturaAnterior")
except:
print("error al presionar BotonFacturaAnterior")
try:
time.sleep(randint(5, 10))
BotonFacturaAnterior=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[10]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[3]/div[1]/div/button[2]')
BotonFacturaAnterior.click()
print("presionó BotonFacturaAnterior2")
except:
print("error al presionar BotonFacturaAnterior2")
#-------------------------
time.sleep(randint(10, 15))
#------------------------------
try:
DescargarDuplicado=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[12]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[2]/div[1]/div/button[1]')
DescargarDuplicado.click()
print("presionó DescargarDuplicado")
except:
print("error al presionar DescargarDuplicado")
try:
DescargarDuplicado=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[12]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/div/div[2]/div[1]/div/button[1]')
DescargarDuplicado.click()
print("presionó DescargarDuplicado2")
except:
print("error al presionar DescargarDuplicado2")
time.sleep(randint(10, 15))
try:
LinkPdffactura=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[13]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/p/a')
hrefpdf=LinkPdffactura.get_attribute("href")
print("presionó LinkPdffactura")
except:
print("error al presionar LinkPdffactura")
try:
LinkPdffactura=driver.find_element(By.XPATH, '/html/body/div[1]/div/div/div[1]/div/div/div/div[2]/div/div/div/div[1]/div/div/div/div/div/div/div/div[2]/div/div/div/div/div/div/div[3]/div[3]/section/article[13]/div[3]/div/div/div[1]/div[2]/div/div[2]/div[2]/div/p/a')
hrefpdf=LinkPdffactura.get_attribute("href")
print("presionó LinkPdffactura2")
except:
print("error al presionar LinkPdffactura2")
driver.quit()
print("hrefpdf")
print(hrefpdf)
return hrefpdf
#url=obtener_url_Factura_Anterior()
#print(url)
#-----------------------------------------------guardar y buscar pdf-pdf --------------------------------------guardar y buscar---pdf-----------------------------------------
def descargar_pdf_desde_url(url, ruta_archivo):
# Verificar si el archivo ya existe en la ruta especificada
if os.path.exists(ruta_archivo):
print("El archivo ya existe en la ruta especificada:", ruta_archivo)
else:
# Realizar la solicitud GET a la URL
response = requests.get(url)
# Verificar si la solicitud fue exitosa (código 200)
if response.status_code == 200:
# Guardar el contenido de la respuesta en un archivo en la ruta especificada
with open(ruta_archivo, 'wb') as f:
f.write(response.content)
print("PDF descargado exitosamente en", ruta_archivo)
else:
print("Error al descargar el PDF. Código de estado:", response.status_code)
def LeerYExtraerPdf(fecha_actualPrincipal,ContratoAConsultar,url_pdf, ruta_archivo,idProyectoAConsultar,FechaConsultaAConsultar):
#----------------
# URL del PDF
#url_pdf = "https://facturawebepm.cadenaportalgestion.com/PDF/Show?id=Jemb/qwi6EGTVk0OrMbnD4HjWcxZVcgKna61bGHDkoj7MCQjIJmQxfSXAAclOSPS"
# Llamar a la función para descargar el PDF desde la URL
descargar_pdf_desde_url(url_pdf, ruta_archivo)
#---------------------------------------------------------------------------------
#leer pdf
doc = pymupdf.open(ruta_archivo) # Abrir el documento PDF
out = open("/var/www/Facturas/{}_{}.txt".format(ContratoAConsultar, fecha_actualPrincipal.strftime("%Y-%m-%d_%H-%M-%S")), "wb")
for page in doc: # Iterar las páginas del documento
text = page.get_text().encode("utf8") # Obtener el texto plano (está en UTF-8)
out.write(text) # Escribir el texto en el archivo de salida
out.close() # Cerrar el archivo de salida
# Leer el texto del archivo de salida
with open("/var/www/Facturas/{}_{}.txt".format(ContratoAConsultar, fecha_actualPrincipal.strftime("%Y-%m-%d_%H-%M-%S")), "r", encoding="utf-8") as file:
texto = file.read()
#Patron para sacar el consumo y costo de la energia
try:
patron_consumoEnergia = r"Valores Facturados\skWh x Costo \(\$\)\sValor \(\$\)\sEnergía ((ene|feb|mar|abr|may|jun|jul|ago|sep|oct|nov|dic)-\d{2})\s^(\d{3})$\s^(\d{1,3}(,\d{3})*)$"
# Buscar los valores
Energia_consumo = re.findall(patron_consumoEnergia, texto, flags=re.MULTILINE)
if Energia_consumo == []:
print("No hay servicios de energia
Array Vacio")
periodo = 0
consumo_energia = 0
costo_energia = 0
else:
# Seleccionar los grupos
for match in Energia_consumo:
periodo = match[0]
consumo_energia = match[2]
costo_energia = match[3]
except:
print("No hay servicios de energia")
periodo = 0
consumo_energia = 0
costo_energia = 0
#---------------------------------------M3---------Acueducto----------------------------
try:
patronAcueducto=r"Valores Facturados\sm3\sx Costo \(\$\)\sValor \(\$\)\sConsumo ((ene|feb|mar|abr|may|jun|jul|ago|sep|oct|nov|dic)-\d{2})\s(-?\d+(\.\d+)?)\s(-?\d{1,3}(?:\.\d{3})*,\d+)"
Acueducto_consumo = re.findall(patronAcueducto, texto, flags=re.MULTILINE)
if Acueducto_consumo ==[]:
print("No hay servicios de Acueducto array vacio")
periodoAcueducto = 0
consumo_Acueducto = 0
costo_Acueducto = 0
else:
# Seleccionar los grupos
for match in Acueducto_consumo:
periodoAcueducto = match[0]
consumo_Acueducto = match[2]
costo_Acueducto = match[4]
except:
print("No hay servicios de Acueducto")
periodoAcueducto = 0
consumo_Acueducto = 0
costo_Acueducto = 0
#--------------------------------------Gas-----------------------------Gas-----------
try:
patronGas=r"Valores Facturados\sm3 x Costo \(\$\)\sValor \(\$\)\sConsumo ((ene|feb|mar|abr|may|jun|jul|ago|sep|oct|nov|dic)-\d{2})\s(-?\d+\,\d+?)\s(-?\d{1,3}(?:\.\d{3})*,\d+)"
Gas_consumo = re.findall(patronGas, texto, flags=re.MULTILINE)
if Gas_consumo == []:
print("La lista está vacía
no hay servicio de gas")
periodoGas = 0
consumo_Gas = 0
costo_Gas = 0
else:
print("La lista no está vacía si hay servicio de gas")
# Seleccionar los grupos
for match in Gas_consumo:
periodoGas = match[0]
consumo_Gas = match[2]
costo_Gas = match[3]
except:
print("no hay servicio de gas")
periodoGas = 0
consumo_Gas = 0
costo_Gas = 0
if periodoAcueducto==0:
PeriodoFactura= periodoGas
else:
if periodoGas==0:
PeriodoFactura=periodo
else:
if periodo == 0:
PeriodoFactura=0
df=pd.DataFrame({'Periodo':PeriodoFactura,'Contrato':ContratoAConsultar,'urlpdf':url_pdf,'ConsumoAcueducto': consumo_Acueducto,'CostoAcueducto': costo_Acueducto,'ConsumoGas': consumo_Gas,'CostoGas': costo_Gas,'ConsumoEnergia': consumo_energia,'CostoEnergia': costo_energia}, index=[0])
df["idProyecto"]=idProyectoAConsultar
df["FechaGeneracion"]=datetime.now()
if len(df) > 0:
df = pd.DataFrame(df, columns=[
"Periodo", "FechaGeneracion", "Contrato", "urlpdf", "ConsumoAcueducto", "CostoAcueducto", "ConsumoGas", "CostoGas", "ConsumoEnergia", "CostoEnergia", "idProyecto"])
if df.size > 0:
print("df")
print(df)
loadtosql('informacionfacturas', df)
#=========================================Inicio api==========================================
def application(environ, start_response):
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
output = ""
try:
sqlconsulta = "SELECT * FROM consultafacturas WHERE 1"
DatosAconsultar = ConsultarData(sqlconsulta)
ContratoAConsultar = DatosAconsultar["Contrato"].iloc[0]
idProyectoAConsultar = DatosAconsultar["idProyecto"].iloc[0]
FechaConsultaAConsultar = DatosAconsultar["FechaConsulta"].iloc[0]
sqlconsultaFacturasactuales="SELECT * FROM informacionfacturas WHERE 1"
Facturasactuales=ConsultarData(sqlconsultaFacturasactuales)
def FormatoFecha(df):
# Extraer el mes y el año de la cadena de texto en la columna 'Periodo'
df['Mes_Periodo'] = df['Periodo'].str.split(' ').str[1].str.split('-').str[0]
df['Anio_Periodo'] = '20' + df['Periodo'].str.split(' ').str[1].str.split('-').str[1]
# Mapear los nombres de los meses abreviados a su forma completa en inglés
meses = {
'ene': 'Jan',
'feb': 'Feb',
'mar': 'Mar',
'abr': 'Apr',
'may': 'May',
'jun': 'Jun',
'jul': 'Jul',
'ago': 'Aug',
'sep': 'Sep',
'oct': 'Oct',
'nov': 'Nov',
'dic': 'Dec'
}
df['Mes_Periodo'] = df['Mes_Periodo'].map(meses)
# Construir la cadena de fecha completa en formato 'YYYY-MMM' (por ejemplo, '2024-Apr')
df['Fecha_Periodo'] = df['Anio_Periodo'] + '-' + df['Mes_Periodo']
# Convertir la cadena de texto a tipo datetime
df['Fecha_Periodo'] = pd.to_datetime(df['Fecha_Periodo'], format='%Y-%b', errors='coerce')
return df
# Aplicar la función FormatoFecha al DataFrame
Facturasactuales = FormatoFecha(Facturasactuales)
# Obtener el mes actual y el año actual
mes_actual = datetime.now().month
anio_actual = datetime.now().year
mes_contrato = Facturasactuales.loc[Facturasactuales['Contrato'] == str(ContratoAConsultar), 'Fecha_Periodo'].dt.month.values
if len(mes_contrato) > 0:
mes_contrato = mes_contrato[0] # Tomamos el primer valor si existe
else:
mes_contrato = "" # Asignamos un valor vacío si no hay entradas para el contrato específico
# Comparar si el mes de 'Fecha_Periodo' es igual al mes actual
if mes_contrato == mes_actual and mes_contrato != "":
print("El contrato "+str(ContratoAConsultar)+" está en el mes actual, siginifica que todo esta actualizado")
else:
print("El contrato "+str(ContratoAConsultar)+" no está en el mes actual. Procederemos con la extracción")
url_pdf_anterior=obtener_url_Factura_Anterior(ContratoAConsultar)
#url_pdf_anterior="https://facturawebepm.cadenaportalgestion.com/PDF/Show?id=2I4X2pa41yO3khy1JMOpZr1b5vr21Vq2ERKrRfKfzRWXFdsqTM7SBNGC+k/ZwJ/R"
fecha_actualPrincipal = datetime.now().date()
# Ruta completa del archivo donde se guardará el PDF (incluyendo el nombre del archivo)
#ruta_archivo = "./Facturas/"+str(ContratoAConsultar)+"_"+str(fecha_actualPrincipal)+".pdf"
ruta_archivo = "/var/www/Facturas/{}_{}.pdf".format(ContratoAConsultar, fecha_actualPrincipal.strftime("%Y-%m-%d_%H-%M-%S"))
LeerYExtraerPdf(fecha_actualPrincipal,ContratoAConsultar,url_pdf_anterior, ruta_archivo,idProyectoAConsultar,FechaConsultaAConsultar)
if url_pdf_anterior:
# output = f"PDF URL: {url_pdf_anterior}\n"
output = f"ok"
else:
output = "Error: No se pudo obtener la URL del PDF\n"
except Exception as e:
output = f"Error occurred: {e}\n"
output += f"Traceback:\n{traceback.format_exc()}"
return [output.encode('utf-8')]