fosadiaria.py
3.87 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from datetime import datetime
from time import sleep
import os, subprocess
from configuracion import *
import pandas as pd
class FosaDiaria():
"""Esta clase almacenara todos los datos de finalizacion de
una planilla de foza, se busca se baja el excel se abra
y se guarda en ran los resultados"""
def __init__(self) -> None:
self.Cache = {}
self.Archivo = "data/fosa.json"
self.cabezaera = cabezaeraDescargas
self.CarpetaDescarga = carpetaDescargas
def BuscarDatosDominio(self,dominio,selenium,dia=False):
if dia == False:
now = datetime.now()
dia = now.strftime("%d/%m/%Y")
try:
data = self.Cache[dia][dominio]
except:
data = False
if not data == False:
return data
try:
self.ActualizarDatosDominios(selenium,dia)
except Exception as E:
print(f"""Error al actualizar dominio {str(E)}""")
return self.LimpiarData(data,dominio,dia)
def LimpiarData(self,data,dominio,dia):
return(data)
def ActualizarDatosDominios(self,selenium,dia):
self.BajarExcel(selenium)
archivo = self.BuscarExcelDescargado()
df = pd.read_excel(archivo)
#print(df)
#print(df.to_dict())
self.ProcesarDatos(df,dia)
def ProcesarDatos(self,df,dia):
data = df.to_dict()
print(data["domino"], type(data["domino"]))
try:
test = self.Cache[dia]
print(test)
except:
self.Cache[dia] = {}
for n in data["domino"].keys():
print("key", n)
dominio = data["domino"][n]
print(dominio,"dominio")
print(type(n))
self.Cache[dia][dominio] ={
"dominio":dominio,
"tipo":data["tipo"][n],
"inicio":data["inicio"][n],
"fin":data["fin"][n],
"resultado":data["resultado"][n],
"cuit":data["cuit"][n],
"vence":data["vence"][n],
"hora":self.CorreccionDeHora(data["fin"][n])
}
n = n + 1 #arranco desde 0
print(self.Cache)
def CorreccionDeHora(self,fin):
"""Hora de finalizacion menos 10 minutos"""
return(fin)
def BajarExcel(self,selenium):
self.LimpiarCarpertaDescarga()
selenium.get("https://rto.cent.gov.ar/rto/RTO/planillaDiaria")
now = datetime.now()
dia = now.strftime("%d")
mes = now.strftime("%m")
ano = now.strftime("%Y")
self.LimpiarCarpertaDescarga()
URL = f"""https://rto.cent.gov.ar/rto/report/planillaDiariaEXCEL/planillaDiariaEXCEL?codTaller=54&fechaConsulta={dia}%2F{mes}%2F{ano}&agruparPorLinea=false&format=excel&extension=xls"""
selenium.get(URL)
def BuscarExcelDescargado(self,tmp="/dev/shm/tmpscdecas.txt"):
test = True
n = 0
while test:
arc = self._BuscarExcelDescargado(tmp)
n = n + 1
if not arc == False:
print(arc, n)
test = False
if n > 70:
return " "
return(arc)
def _BuscarExcelDescargado(self,tmp):
rta = False
os.system(grepfile)
sleep(0.1)
file1 = open(tmp, 'r')
Lines = file1.readlines()
for line in Lines:
der = str(line).split(" ")
archivo = f"""{self.CarpetaDescarga}{der[-1]}"""
rta = archivo.replace("\n","")
file1.close()
os.system(f"""rm {tmp}""")
return rta
def LimpiarCarpertaDescarga(self):
try:
os.system(f"""rm {self.CarpetaDescarga}{self.cabezaera}*""")
sleep(0.1)
except Exception as E:
print(f"Error al intentar borrar planillas diarias anteriores: {str(E)}")