Passed
Push — main ( 39980e...a63a26 )
by Sat CFDI
01:50
created

LocalDBSatCFDI.pendiente_flip()   A

Complexity

Conditions 1

Size

Total Lines 4
Code Lines 4

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 4
dl 0
loc 4
rs 10
c 0
b 0
f 0
cc 1
nop 2
1
import logging
2
import os
3
import pickle
4
from uuid import UUID
5
6
import diskcache
7
from satcfdi.accounting import SatCFDI
8
from satcfdi.pacs import sat
9
from .log_tools import print_yaml
10
11
EMAIL_NOTIFICADA = 2
12
STATUS_SAT = 3
13
PENDIENTE = 4
14
15
sat_manager = sat.SAT()
16
17
logger = logging.getLogger(__name__)
18
DATA_DIR = '.data'
19
20
21
class LocalDB:
22
    def __init__(self):
23
        self.local_storage = diskcache.Cache(os.path.join(DATA_DIR, 'cache'))
24
25
    def saldar(self, uuid: UUID):
26
        return self.local_storage.get(
27
            (PENDIENTE, uuid), True
28
        )
29
30
    def saldar_set(self, uuid: UUID, value: bool):
31
        if value:
32
            try:
33
                del self.local_storage[(PENDIENTE, uuid)]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable PENDIENTE does not seem to be defined.
Loading history...
34
            except KeyError:
35
                pass
36
        else:
37
            self.local_storage[(PENDIENTE, uuid)] = value
38
39
    def notificar(self, uuid: UUID):
40
        return self.local_storage.get(
41
            (EMAIL_NOTIFICADA, uuid), True
42
        )
43
44
    def notificar_set(self, uuid: UUID, value: bool):
45
        if value:
46
            try:
47
                del self.local_storage[(EMAIL_NOTIFICADA, uuid)]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable EMAIL_NOTIFICADA does not seem to be defined.
Loading history...
48
            except KeyError:
49
                pass
50
        else:
51
            self.local_storage[(EMAIL_NOTIFICADA, uuid)] = value
52
53
    def status_sat(self, uuid: UUID):
54
        return self.local_storage.get(
55
            (STATUS_SAT, uuid), {}
56
        )
57
58
    def status_sat_set(self, uuid: UUID, value: dict):
59
        if value:
60
            self.local_storage[(STATUS_SAT, uuid)] = value
61
        else:
62
            try:
63
                del self.local_storage[(STATUS_SAT, uuid)]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable STATUS_SAT does not seem to be defined.
Loading history...
64
            except KeyError:
65
                pass
66
67
68
class LocalDBSatCFDI(LocalDB):
69
    def __init__(self, config):
70
        super().__init__()
71
        self.config = config
72
73
    def notificar(self, cfdi: SatCFDI):
74
        if cfdi["Fecha"] >= self.config['notificar_a_partir']:
75
            return super().notificar(cfdi.uuid)
76
        return False
77
78
    def notificar_flip(self, cfdi: SatCFDI):
79
        v = not self.notificar(cfdi)
80
        self.notificar_set(cfdi.uuid, v)
81
        return v
82
83
    def saldar(self, cfdi: SatCFDI):
84
        if cfdi["TipoDeComprobante"] != "I":
85
            return None
86
        if cfdi["MetodoPago"] == "PPD" and cfdi.saldo_pendiente == 0:
87
            return 0
88
        if cfdi["Fecha"] >= self.config['saldar_a_partir'][cfdi["MetodoPago"]]:
89
            if super().saldar(cfdi.uuid):
90
                if cfdi["MetodoPago"] == "PPD":
91
                    return cfdi.saldo_pendiente
92
                else:
93
                    return cfdi["Total"]
94
        return 0
95
96
    def saldar_flip(self, cfdi: SatCFDI):
97
        v = not self.saldar(cfdi)
98
        self.saldar_set(cfdi.uuid, v)
99
        return v
100
101
    def status_sat(self, cfdi: SatCFDI, update=False):
102
        if update:
103
            res = sat_manager.status(cfdi)
104
            if res["ValidacionEFOS"] == "200":
105
                self.status_sat_set(cfdi.uuid, res)
106
            return res
107
        else:
108
            return super().status_sat(cfdi.uuid)
109
110
    def describe(self, cfdi: SatCFDI):
111
        print_yaml({
112
            'saldar': self.saldar(cfdi),
113
            'notificar': self.notificar(cfdi),
114
            'status_sat': self.status_sat(cfdi)
115
        })
116
117
118
def save_data(file, data):
119
    with open(os.path.join(DATA_DIR, file), 'wb') as f:
120
        pickle.dump(data, f)
121
122
123
def load_data(file, default=None):
124
    try:
125
        with open(os.path.join(DATA_DIR, file), 'rb') as f:
126
            return pickle.load(f)
127
    except FileNotFoundError:
128
        return default
129