satdigitalinvoice.localdb.LocalDB.status() - Code Metrics - Inspection of "Initial" - SAT-CFDI/python-satdigitalinvoice - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 89ce93...d5cc6d )

by Sat CFDI

created 2023-04-22 06:16 UTC

satdigitalinvoice.localdb.LocalDB.status() A

↳ Parent: satdigitalinvoice.localdb

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	2
rs	10
c	0
b	0
f	0
cc	1
nop	2

import logging
import os
import pickle
from datetime import datetime
from enum import Enum
from uuid import UUID

import diskcache
from satcfdi.accounting import SatCFDI
from satcfdi.accounting.models import CFDIEstatus
from satcfdi.pacs import sat

from . import PPD
from .log_tools import print_yaml
from .utils import estado_to_estatus

LIQUIDATED = 0
NOTIFIED = 2
STATUS_SAT = 3
FOLIO = 5
SERIE = 6
SOLICITUDES = 'solicitudes'

sat_manager = sat.SAT()

logger = logging.getLogger(__name__)


class LocalDB(diskcache.Cache):
    def __init__(self, base_path: str):
        super().__init__(directory=os.path.join(base_path, 'cache'))
        self.base_path = base_path

    def folio(self) -> int:
        return self.get(FOLIO, 1000)

    def folio_set(self, value: int):
        self[FOLIO] = value

    def serie(self) -> str:
        return self.get(SERIE, 'A')

    def serie_set(self, value: str):
        self[SERIE] = value

    def liquidated(self, uuid: UUID):
        return self.get((LIQUIDATED, uuid))

    def liquidated_set(self, uuid: UUID, value: bool):
        self[(LIQUIDATED, uuid)] = value

    def notified(self, uuid: UUID):
        return self.get((NOTIFIED, uuid))

    def notified_set(self, uuid: UUID, value: bool):
        self[(NOTIFIED, uuid)] = value

    def status(self, uuid: UUID):
        return self.get((STATUS_SAT, uuid), {})

    def status_set(self, uuid: UUID, value: dict):
        if value:
            self[(STATUS_SAT, uuid)] = value
        else:
            try:
                del self[(STATUS_SAT, uuid)]
            except KeyError:
                pass

    def status_merge(self, uuid: UUID, value: dict):
        self[(STATUS_SAT, uuid)] = self.status(uuid) | value

    def get_solicitudes(self):
        return self.load_data(SOLICITUDES, {})

    def set_solicitudes(self, solicitudes):
        self.save_data(SOLICITUDES, solicitudes)

    def save_data(self, file, data):
        with open(os.path.join(self.base_path, file), 'wb') as f:
            pickle.dump(data, f)

    def load_data(self, file, default=None):
        try:
            with open(os.path.join(self.base_path, file), 'rb') as f:
                return pickle.load(f)
        except FileNotFoundError:
            return default


class StatusState(Enum):
    NONE = 1
    PAID = 2
    PENDING = 3
    IGNORED = 4
    CANCELLED = 5

    def __str__(self):
        if self.name == "NONE":
            return "💰"
        if self.name == "IGNORED":
            return "🚫"
        if self.name == "PAID":
            return "✔"
        if self.name == "PENDING":
            return ""
        if self.name == "CANCELLED":
            return "❌"


class LocalDBSatCFDI(LocalDB):
    def __init__(self, base_path, enviar_a_partir, pagar_a_partir):
        super().__init__(base_path)
        self.enviar_a_partir = enviar_a_partir
        self.pagar_a_partir = pagar_a_partir

    def notified(self, cfdi: SatCFDI):
        v = super().notified(cfdi.uuid)
        if v is None and cfdi["Fecha"] < self.enviar_a_partir:
            return True
        return v

    def notified_flip(self, cfdi: SatCFDI):
        v = not self.notified(cfdi)
        self.notified_set(cfdi.uuid, v)
        return v

    def liquidated(self, cfdi: SatCFDI):
        v = super().liquidated(cfdi.uuid)
        if v is None and cfdi["Fecha"] < self.pagar_a_partir[cfdi["MetodoPago"]]:
            return True
        return v

    def liquidated_flip(self, cfdi: SatCFDI):
        v = not self.liquidated(cfdi)
        self.liquidated_set(cfdi.uuid, v)
        return v

    def status_sat(self, cfdi: SatCFDI, update=False):
        if update:
            res = sat_manager.status(cfdi)
            if res["ValidacionEFOS"] == "200":
                self.status_merge(cfdi.uuid, {
                    'Estatus': estado_to_estatus(res["Estado"]),
                    'EsCancelable': res["EsCancelable"],
                    'EstatusCancelacion': res["EstatusCancelacion"],
                    "last_update": datetime.now(),
                })
            else:
                raise ValueError("Error al actualizar estado de %s: %s", cfdi.uuid, res)
            return res
        else:
            return super().status(cfdi.uuid)

    def liquidated_state(self, cfdi: SatCFDI):
        if cfdi.estatus == CFDIEstatus.Cancelado:
            return StatusState.CANCELLED

        if cfdi["TipoDeComprobante"] != "I":
            return StatusState.NONE

        mpago = cfdi["MetodoPago"]
        if cfdi['Total'] == 0 or (mpago == PPD and cfdi.saldo_pendiente == 0):
            return StatusState.PAID

        if self.liquidated(cfdi):
            if mpago == PPD:
                return StatusState.IGNORED
            return StatusState.PAID

        return StatusState.PENDING

    def describe(self, cfdi: SatCFDI):
        print_yaml({
            'saldada': self.liquidated(cfdi),
            'enviada': self.notified(cfdi),
            'status_sat': self.status_sat(cfdi)
        })


1			import logging
2			import os
3			import pickle
4			from datetime import datetime
5			from enum import Enum
6			from uuid import UUID
7
8			import diskcache
9			from satcfdi.accounting import SatCFDI
10			from satcfdi.accounting.models import CFDIEstatus
11			from satcfdi.pacs import sat
12
13			from . import PPD
14			from .log_tools import print_yaml
15			from .utils import estado_to_estatus
16
17			LIQUIDATED = 0
18			NOTIFIED = 2
19			STATUS_SAT = 3
20			FOLIO = 5
21			SERIE = 6
22			SOLICITUDES = 'solicitudes'
23
24			sat_manager = sat.SAT()
25
26			logger = logging.getLogger(__name__)
27
28
29			class LocalDB(diskcache.Cache):
30			def __init__(self, base_path: str):
31			super().__init__(directory=os.path.join(base_path, 'cache'))
32			self.base_path = base_path
33
34			def folio(self) -> int:
35			return self.get(FOLIO, 1000)
36
37			def folio_set(self, value: int):
38			self[FOLIO] = value
39
40			def serie(self) -> str:
41			return self.get(SERIE, 'A')
42
43			def serie_set(self, value: str):
44			self[SERIE] = value
45
46			def liquidated(self, uuid: UUID):
47			return self.get((LIQUIDATED, uuid))
48
49			def liquidated_set(self, uuid: UUID, value: bool):
50			self[(LIQUIDATED, uuid)] = value
51
52			def notified(self, uuid: UUID):
53			return self.get((NOTIFIED, uuid))
54
55			def notified_set(self, uuid: UUID, value: bool):
56			self[(NOTIFIED, uuid)] = value
57
58			def status(self, uuid: UUID):
59			return self.get((STATUS_SAT, uuid), {})
60
61			def status_set(self, uuid: UUID, value: dict):
62			if value:
63			self[(STATUS_SAT, uuid)] = value
64			else:
65			try:
66			del self[(STATUS_SAT, uuid)]
67			except KeyError:
68			pass
69
70			def status_merge(self, uuid: UUID, value: dict):
71			self[(STATUS_SAT, uuid)] = self.status(uuid) \| value
72
73			def get_solicitudes(self):
74			return self.load_data(SOLICITUDES, {})
75
76			def set_solicitudes(self, solicitudes):
77			self.save_data(SOLICITUDES, solicitudes)
78
79			def save_data(self, file, data):
80			with open(os.path.join(self.base_path, file), 'wb') as f:
81			pickle.dump(data, f)
82
83			def load_data(self, file, default=None):
84			try:
85			with open(os.path.join(self.base_path, file), 'rb') as f:
86			return pickle.load(f)
87			except FileNotFoundError:
88			return default
89
90
91			class StatusState(Enum):
92			NONE = 1
93			PAID = 2
94			PENDING = 3
95			IGNORED = 4
96			CANCELLED = 5
97
98			def __str__(self):
99			if self.name == "NONE":
100			return "💰"
101			if self.name == "IGNORED":
102			return "🚫"
103			if self.name == "PAID":
104			return "✔"
105			if self.name == "PENDING":
106			return ""
107			if self.name == "CANCELLED":
108			return "❌"
109
110
111			class LocalDBSatCFDI(LocalDB):
112			def __init__(self, base_path, enviar_a_partir, pagar_a_partir):
113			super().__init__(base_path)
114			self.enviar_a_partir = enviar_a_partir
115			self.pagar_a_partir = pagar_a_partir
116
117			def notified(self, cfdi: SatCFDI):
118			v = super().notified(cfdi.uuid)
119			if v is None and cfdi["Fecha"] < self.enviar_a_partir:
120			return True
121			return v
122
123			def notified_flip(self, cfdi: SatCFDI):
124			v = not self.notified(cfdi)
125			self.notified_set(cfdi.uuid, v)
126			return v
127
128			def liquidated(self, cfdi: SatCFDI):
129			v = super().liquidated(cfdi.uuid)
130			if v is None and cfdi["Fecha"] < self.pagar_a_partir[cfdi["MetodoPago"]]:
131			return True
132			return v
133
134			def liquidated_flip(self, cfdi: SatCFDI):
135			v = not self.liquidated(cfdi)
136			self.liquidated_set(cfdi.uuid, v)
137			return v
138
139			def status_sat(self, cfdi: SatCFDI, update=False):
140			if update:
141			res = sat_manager.status(cfdi)
142			if res["ValidacionEFOS"] == "200":
143			self.status_merge(cfdi.uuid, {
144			'Estatus': estado_to_estatus(res["Estado"]),
145			'EsCancelable': res["EsCancelable"],
146			'EstatusCancelacion': res["EstatusCancelacion"],
147			"last_update": datetime.now(),
148			})
149			else:
150			raise ValueError("Error al actualizar estado de %s: %s", cfdi.uuid, res)
151			return res
152			else:
153			return super().status(cfdi.uuid)
154
155			def liquidated_state(self, cfdi: SatCFDI):
156			if cfdi.estatus == CFDIEstatus.Cancelado:
157			return StatusState.CANCELLED
158
159			if cfdi["TipoDeComprobante"] != "I":
160			return StatusState.NONE
161
162			mpago = cfdi["MetodoPago"]
163			if cfdi['Total'] == 0 or (mpago == PPD and cfdi.saldo_pendiente == 0):
164			return StatusState.PAID
165
166			if self.liquidated(cfdi):
167			if mpago == PPD:
168			return StatusState.IGNORED
169			return StatusState.PAID
170
171			return StatusState.PENDING
172
173			def describe(self, cfdi: SatCFDI):
174			print_yaml({
175			'saldada': self.liquidated(cfdi),
176			'enviada': self.notified(cfdi),
177			'status_sat': self.status_sat(cfdi)
178			})
179

SAT-CFDI / python-satdigitalinvoice

Push — main ( 89ce93...d5cc6d )

satdigitalinvoice.localdb.LocalDB.status() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like