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

Passed

Push — main ( 8315de...0350f8 )

by Sat CFDI

created 2023-04-22 18:00 UTC

LocalDB.solicitud_merge() A

↳ Parent: satdigitalinvoice.localdb

Complexity

Conditions

Size

Total Lines	10
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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 .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_merge(self, uuid: UUID, estatus: EstadoComprobante, es_cancelable=None,
                     estatus_cancelacion=None, fecha_cancelacion: datetime = None):
        value = {
            "Estatus": estatus,
        }
        if es_cancelable:
            value["EsCancelable"] = es_cancelable
        if estatus_cancelacion:
            value["EstatusCancelacion"] = estatus_cancelacion
        if fecha_cancelacion:
            value["FechaCancelacion"] = fecha_cancelacion
        value["UltimaConsulta"] = datetime.now().replace(microsecond=0),
        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 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)

    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["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 .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_merge(self, uuid: UUID, estatus: EstadoComprobante, es_cancelable=None,
62			estatus_cancelacion=None, fecha_cancelacion: datetime = None):
63			value = {
64			"Estatus": estatus,
65			}
66			if es_cancelable:
67			value["EsCancelable"] = es_cancelable
68			if estatus_cancelacion:
69			value["EstatusCancelacion"] = estatus_cancelacion
70			if fecha_cancelacion:
71			value["FechaCancelacion"] = fecha_cancelacion
72			value["UltimaConsulta"] = datetime.now().replace(microsecond=0),
73			self[(STATUS_SAT, uuid)] = self.status(uuid) \| value
74
75			def get_solicitudes(self):
76			return self.load_data(SOLICITUDES, {})
77
78			def set_solicitudes(self, solicitudes):
79			self.save_data(SOLICITUDES, solicitudes)
80
81			def solicitud_merge(self, solicitud_id, response, request=None):
82			solicitudes = self.get_solicitudes()
83			solicitud = solicitudes.setdefault(solicitud_id, {})
84
85			solicitud['response'] = solicitud.get('response', {}) \| response
86			if request:
87			solicitud['request'] = solicitud.get('request', {}) \| request
88			solicitud['last_update'] = datetime.now().replace(microsecond=0)
89
90			self.set_solicitudes(solicitudes)
91
92			def save_data(self, file, data):
93			with open(os.path.join(self.base_path, file), 'wb') as f:
94			pickle.dump(data, f)
95
96			def load_data(self, file, default=None):
97			try:
98			with open(os.path.join(self.base_path, file), 'rb') as f:
99			return pickle.load(f)
100			except FileNotFoundError:
101			return default
102
103
104			class StatusState(Enum):
105			NONE = 1
106			PAID = 2
107			PENDING = 3
108			IGNORED = 4
109			CANCELLED = 5
110
111			def __str__(self):
112			if self.name == "NONE":
113			return "💰"
114			if self.name == "IGNORED":
115			return "🚫"
116			if self.name == "PAID":
117			return "✔"
118			if self.name == "PENDING":
119			return ""
120			if self.name == "CANCELLED":
121			return "❌"
122
123
124			class LocalDBSatCFDI(LocalDB):
125			def __init__(self, base_path, enviar_a_partir, pagar_a_partir):
126			super().__init__(base_path)
127			self.enviar_a_partir = enviar_a_partir
128			self.pagar_a_partir = pagar_a_partir
129
130			def notified(self, cfdi: SatCFDI):
131			v = super().notified(cfdi.uuid)
132			if v is None and cfdi["Fecha"] < self.enviar_a_partir:
133			return True
134			return v
135
136			def notified_flip(self, cfdi: SatCFDI):
137			v = not self.notified(cfdi)
138			self.notified_set(cfdi.uuid, v)
139			return v
140
141			def liquidated(self, cfdi: SatCFDI):
142			v = super().liquidated(cfdi.uuid)
143			if v is None and cfdi["Fecha"] < self.pagar_a_partir[cfdi["MetodoPago"]]:
144			return True
145			return v
146
147			def liquidated_flip(self, cfdi: SatCFDI):
148			v = not self.liquidated(cfdi)
149			self.liquidated_set(cfdi.uuid, v)
150			return v
151
152			def status_sat(self, cfdi: SatCFDI, update=False):
153			if update:
154			res = sat_manager.status(cfdi)
155			if res["ValidacionEFOS"] == "200":
156			self.status_merge(
157			cfdi.uuid,
158			estatus=estado_to_estatus(res["Estatus"]),
159			es_cancelable=res["EsCancelable"],
160			estatus_cancelacion=res["EstatusCancelacion"]
161			)
162			else:
163			raise ValueError("Error al actualizar estado de %s: %s", cfdi.uuid, res)
164			return res
165			else:
166			return super().status(cfdi.uuid)
167
168			def liquidated_state(self, cfdi: SatCFDI):
169			if cfdi.estatus == EstadoComprobante.Cancelado:
170			return StatusState.CANCELLED
171
172			if cfdi["TipoDeComprobante"] != "I":
173			return StatusState.NONE
174
175			mpago = cfdi["MetodoPago"]
176			if cfdi['Total'] == 0 or (mpago == PPD and cfdi.saldo_pendiente == 0):
177			return StatusState.PAID
178
179			if self.liquidated(cfdi):
180			if mpago == PPD:
181			return StatusState.IGNORED
182			return StatusState.PAID
183
184			return StatusState.PENDING
185

SAT-CFDI / python-satdigitalinvoice

Push — main ( 8315de...0350f8 )

LocalDB.solicitud_merge() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like