110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
import json
|
|
import os
|
|
import sys
|
|
|
|
import httpx
|
|
from flask import Flask, request, jsonify
|
|
|
|
import contabilidad.pdf as pdf
|
|
import contabilidad.passwords as passwords
|
|
import contabilidad.text_handler as th
|
|
from contabilidad.log import Log
|
|
|
|
|
|
app = Flask(__name__)
|
|
log = Log('/var/log/python/contabilidad.log')
|
|
api_key = os.environ.get('PYTHON_KEY')
|
|
|
|
|
|
def validate_key(request_obj):
|
|
if 'Authorization' in request_obj.headers:
|
|
auth = request_obj.headers.get('Authorization')
|
|
if isinstance(auth, list):
|
|
auth = auth[0]
|
|
if 'Bearer' in auth:
|
|
try:
|
|
auth = auth.split(' ')[1]
|
|
except:
|
|
return False
|
|
return auth == api_key
|
|
if 'API_KEY' in request_obj.values:
|
|
return request_obj.values.get('API_KEY') == api_key
|
|
if 'api_key' in request_obj.values:
|
|
return request_obj.values.get('api_key') == api_key
|
|
return False
|
|
|
|
|
|
@app.route('/pdf/parse', methods=['POST'])
|
|
def pdf_parse():
|
|
if not validate_key(request):
|
|
return jsonify({'message': 'Not Authorized'})
|
|
data = request.get_json()
|
|
if not isinstance(data['files'], list):
|
|
data['files'] = [data['files']]
|
|
password_file = '/app/config/.passwords.yml'
|
|
pwds = passwords.get_passwords(password_file)
|
|
output = []
|
|
for file in data['files']:
|
|
filename = os.path.realpath(os.path.join('/app/data', file['filename']))
|
|
t = file['filename'].split('.')
|
|
temp = os.path.realpath(os.path.join('/app/data', t[0] + '-temp.pdf'))
|
|
for p in pwds:
|
|
if not pdf.check_password(filename, p):
|
|
continue
|
|
pdf.remove_encryption(filename, p, temp)
|
|
obj = pdf.get_data(temp)
|
|
try:
|
|
text = th.text_cleanup(pdf.get_text(temp))
|
|
except IndexError as ie:
|
|
print(ie, file=sys.stderr)
|
|
continue
|
|
outputs = []
|
|
for o in obj:
|
|
out = json.loads(o.df.to_json(orient='records'))
|
|
if out[0]['0'] == 'FECHA':
|
|
for i, line in enumerate(out):
|
|
if 'FECHA' in line['0'] or 'ACTUALICE' in line['0']:
|
|
continue
|
|
if line['0'] == '':
|
|
spl = line['1'].split(' ')
|
|
else:
|
|
spl = line['0'].split(' ')
|
|
line['0'] = ' '.join(spl[:3])
|
|
line['1'] = ' '.join(spl[3:])
|
|
out[i] = line
|
|
outputs.append(out)
|
|
os.remove(temp)
|
|
output.append({'bank': text['bank'], 'filename': file['filename'], 'tables': outputs, 'text': text['text']})
|
|
return jsonify(output)
|
|
|
|
|
|
@app.route('/cambio/get', methods=['POST'])
|
|
def cambios():
|
|
if not validate_key(request):
|
|
return jsonify({'message': 'Not Authorized'})
|
|
data = request.get_json()
|
|
valid = {
|
|
"CLF": "uf",
|
|
"IVP": "ivp",
|
|
"USD": "dolar",
|
|
"USDo": "dolar_intercambio",
|
|
"EUR": "euro",
|
|
"IPC": "ipc",
|
|
"UTM": "utm",
|
|
"IMACEC": "imacec",
|
|
"TPM": "tpm",
|
|
"CUP": "libra_cobre",
|
|
"TZD": "tasa_desempleo",
|
|
"BTC": "bitcoin"
|
|
}
|
|
base_url = 'https://mindicador.cl/api/'
|
|
url = f"{base_url}{valid[data['desde']]}/{'-'.join(list(reversed(data['fecha'].split('-'))))}"
|
|
res = httpx.get(url)
|
|
if res.status_code != httpx.codes.OK:
|
|
return jsonify({'error': 'Valor no encontrado.'})
|
|
return jsonify(res.json())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', debug=True)
|