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

Passed

Push — main ( 31e698...c64486 )

by Sat CFDI

created 2023-04-24 06:29 UTC

satdigitalinvoice.localdb.StatusState.str() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	11
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	11
dl	0
loc	11
rs	8.6666
c	0
b	0
f	0
cc	6
nop	1

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 EstadoComprobante
from satcfdi.pacs import sat

from . import PPD
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_merge(self, uuid: str | UUID, estatus: str, es_cancelable: str = None,
                     estatus_cancelacion: str = None, fecha_cancelacion: str = None, fecha_ultima_consulta: str = None):
        if isinstance(uuid, str):
            uuid = UUID(uuid)

        value = {
            "Estatus": estatus,
        }
        if es_cancelable:
            value["EsCancelable"] = es_cancelable

        if estatus_cancelacion:
            value["EstatusCancelacion"] = estatus_cancelacion

        if fecha_cancelacion:
            value["FechaCancelacion"] = datetime.fromisoformat(fecha_cancelacion)

        if fecha_ultima_consulta:
            value["UltimaConsulta"] = datetime.fromisoformat(fecha_ultima_consulta)
        else:
            value["UltimaConsulta"] = datetime.now().replace(microsecond=0)

        current_value = self.status(uuid)
        if current_value:
            if current_value["UltimaConsulta"] > value["UltimaConsulta"]:
                return

        self[(STATUS_SAT, uuid)] = current_value | value

    def status_export(self, uuid: UUID):
        if i := self.status(uuid):
            return [
                uuid,
                i["Estatus"],
                i.get("EsCancelable"),
                i.get("EstatusCancelacion"),
                i.get("FechaCancelacion"),
                i.get("UltimaConsulta"),
            ]

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

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

    def solicitud_merge(self, solicitud_id, response, request=None):
        solicitudes = self.get_solicitudes()
        solicitud = solicitudes.setdefault(solicitud_id, {})

        solicitud['response'] = solicitud.get('response', {}) | response
        if request:
            solicitud['request'] = solicitud.get('request', {}) | request
        solicitud['last_update'] = datetime.now().replace(microsecond=0)

        self.set_solicitudes(solicitudes)
        return solicitud

    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(
                    uuid=cfdi.uuid,
                    estatus=estado_to_estatus(res["Estatus"]),
                    es_cancelable=res["EsCancelable"],
                    estatus_cancelacion=res["EstatusCancelacion"]
                )
            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 == EstadoComprobante.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


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

SAT-CFDI / python-satdigitalinvoice

Push — main ( 31e698...c64486 )

satdigitalinvoice.localdb.StatusState.__str__() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

satdigitalinvoice.localdb.StatusState.str() B