Passed
Push — main ( b85a21...c81107 )
by Sat CFDI
01:47
created

LocalDBSatCFDI.notificar_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 pendiente(self, uuid: UUID):
26
        return self.local_storage.get(
27
            (PENDIENTE, uuid), True
28
        )
29
30
    def pendiente_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 pendiente(self, cfdi: SatCFDI):
84
        if cfdi["MetodoPago"] == "PPD" and cfdi.saldo_pendiente == 0:
85
            return False
86
        if cfdi["Fecha"] >= self.config['pendiente_a_partir'][cfdi["MetodoPago"]]:
87
            return super().pendiente(cfdi.uuid)
88
        return False
89
90
    def pendiente_flip(self, cfdi: SatCFDI):
91
        v = not self.pendiente(cfdi)
92
        self.pendiente_set(cfdi.uuid, v)
93
        return v
94
95
    def status_sat(self, cfdi: SatCFDI, update=False):
96
        if update:
97
            res = sat_manager.status(cfdi)
98
            if res["ValidacionEFOS"] == "200":
99
                self.status_sat_set(cfdi.uuid, res)
100
            return res
101
        else:
102
            return super().status_sat(cfdi.uuid)
103
104
    def describe(self, cfdi: SatCFDI):
105
        print_yaml({
106
            'pendiente': self.pendiente(cfdi),
107
            'email_notificada': self.notificar(cfdi),
108
            'status_sat': self.status_sat(cfdi)
109
        })
110
111
112
def save_data(file, data):
113
    with open(os.path.join(DATA_DIR, file), 'wb') as f:
114
        pickle.dump(data, f)
115
116
117
def load_data(file, default=None):
118
    try:
119
        with open(os.path.join(DATA_DIR, file), 'rb') as f:
120
            return pickle.load(f)
121
    except FileNotFoundError:
122
        return default
123