Passed
Push — main ( 76adad...7d83f6 )
by Sat CFDI
01:50
created

satdigitalinvoice.localdb.LocalDB.folio()   A

Complexity

Conditions 1

Size

Total Lines 2
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Importance

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