Script process csv drive

core/apps/base/resources/api_calls.py

@@ -68,6 +68,29 @@ def call_api_eps(num_aut: int) -> dict:
     return request_api(url, headers, payload)
 
 
+def update_status_factura_cajacopi(factura: str, valor_factura: int, num_aut: str) -> dict:
+    """
+    Cambia el estado de una factura.
+    Incialmente llamado desde core.apps.tasks.utils.pipeline.Send2Cajacopi.cajacopi_calls.
+    :param factura: Valor que representa la factura de una autorización.
+    :param valor_factura: Valor total de una factura.
+    :param num_aut: Número de autorización.
+    :return: Respuesta de la api de Cajacopi:
+            Ej.:
+                - {}
+                - {"codigo": "1", "mensaje": "Error!. La autorizacion no tiene detalles. AutProCode:2083"}
+    """
+    url = "https://genesis.cajacopieps.com/api/api_qr.php"
+    payload = {
+        "function": "p_inserta_factura_aut",
+        "factura": factura,
+        "valorFactura": valor_factura,
+        "autorizacion": num_aut
+    }
+    headers = {'Content-Type': 'text/plain'}
+    return request_api(url, headers, payload)
+
+
 @logtime('API')
 def auth_api_medicar():
     """
@@ -326,6 +349,7 @@ def check_med_bd(codcum: str):
         cursor.close()
         conn.close()
 
+
 def find_cums(articulos: Tuple) -> Dict[str, str]:
     """
     Busca en base de datos todos los articulos con
@@ -360,6 +384,7 @@ def find_cums(articulos: Tuple) -> Dict[str, str]:
         cursor.close()
         conn.close()
 
+
 def make_dbconn() -> Connection:
     import pymssql
     server = config('SQL_SERVER_HOST')
@@ -370,7 +395,6 @@ def make_dbconn() -> Connection:
                            user=username, password=password)
 
 
-
 if __name__ == '__main__':
     ...
     # print(call_api_eps(1))

core/apps/base/resources/decorators.py

@@ -36,13 +36,34 @@ def wrapper(*fargs, **fkwargs):
                     title = f"{title} {fargs[2]['serial']}"
             logger.info(f"{title or tag} {func.__name__!r} tardó {format(time() - start, '.4f')}s.")
             return value
+
         return wrapper
+
     return decorator
 
+
+def ignore_unhashable(func):
+    uncached = func.__wrapped__
+    attributes = functools.WRAPPER_ASSIGNMENTS + ('cache_info', 'cache_clear')
+
+    @functools.wraps(func, assigned=attributes)
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except TypeError as error:
+            if 'unhashable type' in str(error):
+                return uncached(*args, **kwargs)
+            raise
+
+    wrapper.__uncached__ = uncached
+    return wrapper
+
+
 def hash_dict(func):
     """Transform mutable dictionary into immutable
     Useful to be compatible with cache
     """
+
     class HDict(dict):
         def __hash__(self):
             return hash(frozenset(self.items()))
@@ -52,8 +73,10 @@ def wrapped(*args, **kwargs):
         args = tuple([HDict(arg) if isinstance(arg, dict) else arg for arg in args])
         kwargs = {k: HDict(v) if isinstance(v, dict) else v for k, v in kwargs.items()}
         return func(*args, **kwargs)
+
     return wrapped
 
+
 def timed_lru_cache(seconds: int, maxsize: int = 4):
     def wrapper_cache(func):
         func = functools.lru_cache(maxsize=maxsize)(func)
@@ -70,4 +93,4 @@ def wrapped_func(*args, **kwargs):
 
         return wrapped_func
 
-    return wrapper_cache
+    return wrapper_cache

core/apps/base/resources/tools.py

@@ -293,6 +293,7 @@ def notify(reason: str, subject: str, body: str, to=None, bcc: list = []):
                 'check-datosgov': 'Correo enviado por problema al consultar datos.gov.co.',
                 'expd-no-encontrado': '{} Correo enviado por expediente no encontrado.',
                 'med-control': '{} Correo enviado por medicamento controlado.',
+                'task-fill-data': '{} Correo enviado por facturas no procesadas en rutina.',
             }
     except Exception as e:
         logger.error(f"{rad} Correo de {reason} no fue enviado. Error: {e}")

core/apps/tasks/admin.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

core/apps/tasks/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class TasksConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'core.apps.tasks'

core/apps/tasks/management/commands/actualiza_facturas.py

@@ -0,0 +1,13 @@
+from django.core.management import BaseCommand
+
+from core.apps.tasks.utils.main import Dispensacion
+from core.settings import logger as log
+
+
+class Command(BaseCommand):
+    help = 'Ejecuta script de dispensación, actualizando las facturas de procesadas activas.'
+
+    def handle(self, *args, **options):
+        log.info(f"{' ACTUALIZANDO FACTURAS INICIO ':▼^50}")
+        Dispensacion().run()
+        log.info(f"{' ACTUALIZANDO FACTURAS INICIO ':▲^50}")

core/apps/tasks/migrations/0001_initial.py

@@ -0,0 +1,33 @@
+# Generated by Django 4.1.3 on 2023-09-23 11:38
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='FacturasProcesadas',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('agregado', models.DateTimeField(auto_now_add=True)),
+                ('actualizado', models.DateTimeField(auto_now=True)),
+                ('factura', models.CharField(max_length=24, unique=True)),
+                ('fecha_factura', models.DateTimeField()),
+                ('acta', models.CharField(max_length=24)),
+                ('numero_autorizacion', models.CharField(max_length=24, unique=True)),
+                ('valor_total', models.IntegerField(blank=True, null=True)),
+                ('link_soporte', models.CharField(blank=True, max_length=254, null=True)),
+                ('estado', models.CharField(blank=True, max_length=128, null=True)),
+                ('resp_cajacopi', models.JSONField(blank=True, null=True)),
+            ],
+            options={
+                'db_table': 'facturas_procesadas',
+            },
+        ),
+    ]

core/apps/tasks/models.py

@@ -0,0 +1,21 @@
+from django.db.models import Model
+from django.db.models import IntegerField, DateTimeField, CharField, JSONField
+
+
+class FacturasProcesadas(Model):
+    agregado = DateTimeField(auto_now_add=True)
+    actualizado = DateTimeField(auto_now=True)
+    factura = CharField(max_length=24, unique=True)
+    fecha_factura = DateTimeField()
+    acta = CharField(max_length=24)
+    numero_autorizacion = CharField(unique=True, max_length=24)
+    valor_total = IntegerField(blank=True, null=True)
+    link_soporte = CharField(max_length=254, blank=True, null=True)
+    estado = CharField(max_length=128, blank=True, null=True)
+    resp_cajacopi = JSONField(blank=True, null=True)
+
+    def __str__(self):
+        return f"{self.factura} estado={self.estado!r}"
+
+    class Meta:
+        db_table = 'facturas_procesadas'

core/apps/tasks/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.