Passed
Push — main ( 57aa61...6a9ca4 )
by Sat CFDI
01:44
created

satdigitalinvoice.localdb.load_data()   A

Complexity

Conditions 3

Size

Total Lines 6
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Importance

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