Test Failed
Push — main ( d5344d...b56e1d )
by Sat CFDI
03:19
created

satdigitalinvoice.local.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
from .log_tools import LogAdapter
10
from .file_data_managers import ConfigManager
11
12
PUE_PAGADA = 0
13
PPD_IGNORAR = 1
14
EMAIL_NOTIFICADA = 2
15
STATUS_SAT = 3
16
17
config = ConfigManager()
18
sat_manager = sat.SAT()
19
20
logger = LogAdapter(logging.getLogger())
21
DATA_DIR = '.data'
22
23
24
class LocalDB:
25
    def __init__(self):
26
        self.local_storage = diskcache.Cache(os.path.join(DATA_DIR, 'cache'))
27
28
    def pue_pagada(self, uuid: UUID):
29
        return self.local_storage.get(
30
            (PUE_PAGADA, uuid), False
31
        )
32
33
    def pue_pagada_set(self, uuid: UUID, value: bool):
34
        if value:
35
            self.local_storage[(PUE_PAGADA, uuid)] = value
36
        else:
37
            try:
38
                del self.local_storage[(PUE_PAGADA, uuid)]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable PUE_PAGADA does not seem to be defined.
Loading history...
39
            except KeyError:
40
                pass
41
42
    def ppd_ignorar(self, uuid: UUID):
43
        return self.local_storage.get(
44
            (PPD_IGNORAR, uuid), False
45
        )
46
47
    def ppd_ignorar_set(self, uuid: UUID, value: bool):
48
        if value:
49
            self.local_storage[(PPD_IGNORAR, uuid)] = value
50
        else:
51
            try:
52
                del self.local_storage[(PPD_IGNORAR, uuid)]
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable PPD_IGNORAR does not seem to be defined.
Loading history...
53
            except KeyError:
54
                pass
55
56
    def email_notificada(self, uuid: UUID):
57
        return self.local_storage.get(
58
            (EMAIL_NOTIFICADA, uuid), False
59
        )
60
61
    def email_notificada_set(self, uuid: UUID, value: bool):
62
        if value:
63
            self.local_storage[(EMAIL_NOTIFICADA, uuid)] = value
64
        else:
65
            try:
66
                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...
67
            except KeyError:
68
                pass
69
70
    def status_sat(self, uuid: UUID):
71
        return self.local_storage.get(
72
            (STATUS_SAT, uuid), {}
73
        )
74
75
    def status_sat_set(self, uuid: UUID, value: dict):
76
        if value:
77
            self.local_storage[(STATUS_SAT, uuid)] = value
78
        else:
79
            try:
80
                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...
81
            except KeyError:
82
                pass
83
84
85
class LocalDBSatCFDI(LocalDB):
86
    def email_notificada(self, cfdi: SatCFDI):
87
        if cfdi["Fecha"] <= config['email_notificada_hasta']:
88
            return True
89
        return super().email_notificada(cfdi.uuid)
90
91
    def pue_pagada(self, cfdi: SatCFDI):
92
        if cfdi["Fecha"] <= config['pue_pagada_hasta']:
93
            return True
94
        return super().pue_pagada(cfdi.uuid)
95
96
    def ppd_ignorar(self, cfdi: SatCFDI):
97
        if cfdi["Fecha"] <= config['ppd_ignorar_hasta']:
98
            return True
99
        return super().ppd_ignorar(cfdi.uuid)
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
        logger.info_yaml({
112
            'pue_pagada': self.pue_pagada(cfdi),
113
            'ppd_ignorar': self.ppd_ignorar(cfdi),
114
            'email_notificada': self.email_notificada(cfdi),
115
            'status_sat': self.status_sat(cfdi)
116
        })
117
118
119
local_db = LocalDBSatCFDI()
120
121
122
def save_data(file, data):
123
    with open(os.path.join(DATA_DIR, file), 'wb') as f:
124
        pickle.dump(data, f)
125
126
127
def load_data(file, default=None):
128
    try:
129
        with open(os.path.join(DATA_DIR, file), 'rb') as f:
130
            return pickle.load(f)
131
    except FileNotFoundError:
132
        return default
133