Merge pull request 'feature/process-invoices' (#202) from feature/process-invoices into development

Reviewed-on: ADUANASOFT/anexo76#202
This commit is contained in:
2026-03-13 16:06:44 +00:00
43 changed files with 30496 additions and 442 deletions

View File

@@ -0,0 +1,345 @@
"""
Annex 24 - Balance Management Core
SQLAlchemy v2
Plugs into the existing schema:
invoice_header (a76.invoice_header) ← already exists
item_lines (a76.item_lines) ← already exists
These 4 tables are ALL you need for balances:
a24.balance_movement ← the ledger (append-only, never UPDATE)
a24.discharge_header ← one discharge per export/SM/CTM event
a24.discharge_detail ← one row per (export line × import lot consumed)
a24.discharge_scrap ← mermas, desperdicios, destrucciones
Design rules:
1. NEVER update balance_movement rows — only INSERT
2. Balance = SUM of movements. No cached balance columns.
3. Every discharge_detail row MUST reference a balance_movement row
4. PEPS order is enforced via order_peps (monotonic, set on INSERT)
"""
from __future__ import annotations
import datetime
from decimal import Decimal
from enum import Enum
from typing import TYPE_CHECKING, List, Optional
from ..discharges.models import DischargeDetail
from sqlalchemy import (
BigInteger,
CheckConstraint,
Date,
ForeignKey,
Index,
Integer,
Numeric,
String,
UniqueConstraint,
text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
if TYPE_CHECKING:
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.parts.models import Part
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class MovementType(str, Enum):
# ── Positive (add to balance) ──────────────────────────────────────────
ENTRY = "entry" # Normal import entry
RETURN = "return" # Material returned to balance
POSITIVE_ADJUSTMENT = "pos_adj" # Physical inventory adjustment (+)
TRANSFER_IN = "transfer_in" # CTM/SM received (the receiving side)
REGIME_CHANGE_IN = "regime_chg_in" # Eg. temporary → definitive (entry side)
# ── Negative (consume from balance) ───────────────────────────────────
CONSUMPTION = "consumption" # Consumed in export (most common)
WASTE = "waste" # Merma de proceso
SCRAP = "scrap" # Desperdicio / scrap
DESTRUCTION = "destruction" # Destrucción oficial ante aduana
NEGATIVE_ADJUSTMENT = "neg_adj" # Physical inventory adjustment (-)
TRANSFER_OUT = "transfer_out" # CTM/SM sent (the sending side)
EXPIRATION = "expiration" # Balance cancelled due to deadline
REGIME_CHANGE_OUT = "regime_chg_out" # Eg. temporary → definitive (exit side)
# Which movement types reduce the balance (sign = -1)
NEGATIVE_MOVEMENTS = {
MovementType.CONSUMPTION,
MovementType.WASTE,
MovementType.SCRAP,
MovementType.DESTRUCTION,
MovementType.NEGATIVE_ADJUSTMENT,
MovementType.TRANSFER_OUT,
MovementType.EXPIRATION,
MovementType.REGIME_CHANGE_OUT,
}
# Which types count toward "used" (CANTUSADA in Anexo 24 report)
USED_MOVEMENTS = {
MovementType.CONSUMPTION,
MovementType.WASTE,
MovementType.SCRAP,
MovementType.DESTRUCTION,
}
# ---------------------------------------------------------------------------
# 1. BalanceMovement (the ledger — APPEND ONLY)
#
# One row per atomic change to a specific import lot.
# Current balance of any lot = SUM of (signed quantity) over its rows.
#
# import_item_line_id → a76.item_lines.id (the import line = the "lot")
# import_invoice_id → a76.invoice_header.id (the import invoice)
# part_number_id → a76.parts.id (denormalized for PEPS index)
# source_item_line_id → a76.item_lines.id (export/SM/CTM line, if any)
# source_invoice_id → a76.invoice_header.id (export/SM/CTM invoice, if any)
#
# NOTE: No discharge_detail_id column. Navigate the other direction via
# DischargeDetail.movement_id to avoid a circular FK and keep this
# table truly append-only (no UPDATE ever needed).
# ---------------------------------------------------------------------------
class BalanceMovement(Base, TenantScopedMixin, TimestampMixin):
"""
Core ledger table. NEVER update existing rows — only INSERT.
Balance of a lot = SUM(quantity) WHERE sign=+1 (entries)
- SUM(quantity) WHERE sign=-1 (exits)
Append-only rule is enforced at the application layer.
"""
__tablename__ = "balance_movement"
__table_args__ = (
UniqueConstraint(
"import_item_line_id", "order_peps",
name="uq_balance_movement_lot_peps",
),
CheckConstraint("quantity > 0", name="ck_balance_movement_qty_positive"),
# PEPS lookup: "give me available lots for this part+regime, oldest first"
# part_number_id is denormalized here so this index is self-contained.
Index(
"ix_balmov_peps_lookup",
"tenant_id", "part_number_id", "movement_type", "order_peps",
postgresql_include=["import_item_line_id", "quantity", "value_me", "value_mn"],
),
# Balance calculation per lot
Index("ix_balmov_lot", "import_item_line_id"),
# "What did this export consume?"
Index("ix_balmov_source", "source_invoice_id", "source_item_line_id"),
# Anexo 24 period reports
Index("ix_balmov_operation_date", "tenant_id", "operation_date", "movement_type"),
{"schema": "a24"},
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
# ── The import lot this movement belongs to ────────────────────────────
import_invoice_id: Mapped[int] = mapped_column(
ForeignKey("a76.invoice_header.id"),
comment="Import invoice (cabecera de importación)",
)
import_item_line_id: Mapped[int] = mapped_column(
ForeignKey("a76.item_lines.id"),
comment="Import line item = the PEPS lot",
)
# ── Denormalized part reference — enables efficient PEPS index ─────────
# Must equal import_line.part_number_id. Set on INSERT, never changed.
part_number_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.parts.id"),
comment="Denormalized from item_lines.part_number_id. Enables PEPS index without joins.",
)
# ── What kind of movement ─────────────────────────────────────────────
movement_type: Mapped[MovementType] = mapped_column(
String(20),
comment="See MovementType enum. Determines sign and whether qty counts as used.",
)
# ── Quantity and value (always stored positive) ───────────────────────
quantity: Mapped[Decimal] = mapped_column(
Numeric(19, 8),
comment="Always positive. Sign is inferred from movement_type via NEGATIVE_MOVEMENTS.",
)
value_me: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8), comment="USD")
value_mn: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8), comment="MXN")
net_weight: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))
# ── Document that caused this movement ────────────────────────────────
# NULL for ENTRY movements (the import invoice itself is the cause)
source_invoice_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.invoice_header.id"),
comment="Export / SM / CTM invoice. NULL for entries.",
)
source_item_line_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.item_lines.id"),
comment="Specific line in the export / SM / CTM invoice.",
)
# ── PEPS ordering ─────────────────────────────────────────────────────
# Simplest approach: set order_peps = id (globally monotonic).
# Finer approach: use a per-part sequence or epoch-based value.
order_peps: Mapped[int] = mapped_column(
BigInteger,
comment="PEPS order within this lot. Lower = older = consumed first.",
)
# ── Business date of the operation ───────────────────────────────────
operation_date: Mapped[datetime.date] = mapped_column(
Date, comment="Date of the actual business event, not DB insert."
)
notes: Mapped[Optional[str]] = mapped_column(String(300))
# ── Relationships ─────────────────────────────────────────────────────
import_invoice: Mapped["InvoiceHeader"] = relationship(
foreign_keys=[import_invoice_id],
)
import_line: Mapped["LineItem"] = relationship(
foreign_keys=[import_item_line_id],
)
part: Mapped[Optional["Part"]] = relationship(
foreign_keys=[part_number_id],
)
source_invoice: Mapped[Optional["InvoiceHeader"]] = relationship(
foreign_keys=[source_invoice_id],
)
source_line: Mapped[Optional["LineItem"]] = relationship(
foreign_keys=[source_item_line_id],
)
# Back-reference: navigate to the detail that consumed this movement.
# Use viewonly=True — ownership lives on DischargeDetail.movement_id.
discharge_detail: Mapped[Optional["DischargeDetail"]] = relationship(
back_populates="movement",
foreign_keys="[DischargeDetail.movement_id]",
primaryjoin="BalanceMovement.id == DischargeDetail.movement_id",
viewonly=True,
)
# ---------------------------------------------------------------------------
# Repository helpers (copy to your service/repository layer)
# ---------------------------------------------------------------------------
#
#
# ── Current balance of a lot ─────────────────────────────────────────────
#
# from sqlalchemy import case, func, select
#
# def current_balance(session, import_item_line_id: int):
# sign = case(
# (BalanceMovement.movement_type.in_(NEGATIVE_MOVEMENTS), -1),
# else_=1,
# )
# affects_used = BalanceMovement.movement_type.in_(USED_MOVEMENTS)
# return session.execute(
# select(
# func.sum(sign * BalanceMovement.quantity).label("current_balance"),
# func.sum(
# case((affects_used, BalanceMovement.quantity), else_=0)
# ).label("quantity_used"),
# func.sum(
# case((affects_used, BalanceMovement.value_me), else_=0)
# ).label("value_used_me"),
# ).where(BalanceMovement.import_item_line_id == import_item_line_id)
# ).one()
#
#
# ── PEPS resolver — call BEFORE inserting a CONSUMPTION movement ─────────
#
# def peps_lots_for(session, tenant_id, part_number_id, operation_type, qty_needed):
# """
# Returns import lots in FIFO order with their available balance.
# Walk the list and consume until qty_needed is satisfied.
# """
# sign = case(
# (BalanceMovement.movement_type.in_(NEGATIVE_MOVEMENTS), -1),
# else_=1,
# )
# lot_balances = (
# select(
# BalanceMovement.import_item_line_id,
# func.sum(sign * BalanceMovement.quantity).label("available"),
# func.min(BalanceMovement.order_peps).label("oldest_peps"),
# )
# .where(
# BalanceMovement.tenant_id == tenant_id,
# BalanceMovement.part_number_id == part_number_id,
# )
# .group_by(BalanceMovement.import_item_line_id)
# .having(func.sum(sign * BalanceMovement.quantity) > 0)
# .order_by("oldest_peps")
# .subquery()
# )
# return session.execute(select(lot_balances)).all()
#
#
# ── Transaction flow for a new discharge ─────────────────────────────────
#
# def apply_discharge(session, export_invoice_id, lines):
# """
# lines = [{"export_line_id": X, "part_number_id": Y, "quantity": Z}, ...]
#
# Two-step INSERT: movement first, then detail referencing it.
# No UPDATE on balance_movement — design rule 1 is preserved.
# """
# header = DischargeHeader(source_invoice_id=export_invoice_id, ...)
# session.add(header)
# session.flush() # get header.id
#
# for line in lines:
# lots = peps_lots_for(session, ..., line["part_number_id"], qty_needed=line["quantity"])
# remaining = line["quantity"]
#
# for lot in lots:
# consume = min(lot.available, remaining)
#
# # Step 1: insert movement
# mov = BalanceMovement(
# import_item_line_id = lot.import_item_line_id,
# part_number_id = line["part_number_id"],
# movement_type = MovementType.CONSUMPTION,
# quantity = consume,
# source_invoice_id = export_invoice_id,
# source_item_line_id = line["export_line_id"],
# order_peps = <next_sequence>,
# operation_date = datetime.date.today(),
# )
# session.add(mov)
# session.flush() # get mov.id
#
# # Step 2: insert detail referencing the movement
# det = DischargeDetail(
# discharge_header_id = header.id,
# export_item_line_id = line["export_line_id"],
# import_item_line_id = lot.import_item_line_id,
# movement_id = mov.id, # NOT NULL — set immediately
# quantity_discharged = consume,
# )
# session.add(det)
#
# remaining -= consume
# if remaining <= 0:
# break
#
# session.commit()

View File

@@ -0,0 +1,370 @@
"""
Annex 24 - Balance Management Core
SQLAlchemy v2
Plugs into the existing schema:
invoice_header (a76.invoice_header) ← already exists
item_lines (a76.item_lines) ← already exists
These 4 tables are ALL you need for balances:
a24.balance_movement ← the ledger (append-only, never UPDATE)
a24.discharge_header ← one discharge per export/SM/CTM event
a24.discharge_detail ← one row per (export line × import lot consumed)
a24.discharge_scrap ← mermas, desperdicios, destrucciones
Design rules:
1. NEVER update balance_movement rows — only INSERT
2. Balance = SUM of movements. No cached balance columns.
3. Every discharge_detail row MUST reference a balance_movement row
4. PEPS order is enforced via order_peps (monotonic, set on INSERT)
"""
from __future__ import annotations
import datetime
from decimal import Decimal
from enum import Enum
from typing import TYPE_CHECKING, List, Optional
from ..discharges.models import DischargeDetail
from sqlalchemy import (
BigInteger,
CheckConstraint,
Date,
ForeignKey,
Index,
Integer,
Numeric,
String,
text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
if TYPE_CHECKING:
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.items.models import LineItem
from ..balance_movements.models import BalanceMovement
class DischargeType(str, Enum):
TEMPORARY = "temporary" # SDescargaT — export against temp import
DEFINITIVE = "definitive" # SDescargaD — export against def import
REPAIR = "repair" # SDescargaR — repair / return
DIRECTED = "directed" # SDescargaM — Directed discharge
CTM = "ctm" # SDescargaCTM — between plants (same group)
SUBASSEMBLY = "subassembly" # SDescargaSM — to external maquiladora
WASTE_SCRAP = "waste_scrap" # SDescargaMerDes — merma / desperdicio
class DischargeStatus(str, Enum):
PENDING = "pending"
APPLIED = "applied"
PARTIAL = "partial"
CANCELLED = "cancelled"
# ---------------------------------------------------------------------------
# 2. DischargeHeader
#
# One row per discharge event (export, SM batch, CTM transfer…).
# Groups all DischargeDetail rows for the same business event.
#
# source_invoice_id → a76.invoice_header.id (the export/SM/CTM invoice)
# def_import_invoice_id → a76.invoice_header.id (DEF discharges only)
# ---------------------------------------------------------------------------
class DischargeHeader(Base, TenantScopedMixin, TimestampMixin):
"""
One discharge per export event.
Groups DischargeDetail rows (one per import lot consumed).
"""
__tablename__ = "discharge_header"
__table_args__ = (
Index("ix_dischdr_source", "source_invoice_id", "status"),
Index("ix_dischdr_date", "tenant_id", "discharge_date", "discharge_type"),
{"schema": "a24"},
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
# ── Export / SM / CTM document that triggers this discharge ───────────
source_invoice_id: Mapped[int] = mapped_column(
ForeignKey("a76.invoice_header.id"),
comment="Export, SM-out or CTM-send invoice that owns this discharge.",
)
# ── For DEFINITIVE discharges only ────────────────────────────────────
def_import_invoice_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.invoice_header.id"),
comment="Populated only for discharge_type=DEFINITIVE.",
)
discharge_type: Mapped[DischargeType] = mapped_column(String(15))
status: Mapped[DischargeStatus] = mapped_column(
String(15), server_default=text("'applied'")
)
discharge_date: Mapped[datetime.date] = mapped_column(Date)
# ── Control fields from legacy SDescarga* tables ──────────────────────
reference_invoice: Mapped[Optional[str]] = mapped_column(
String(19), comment="FACREFERENCIA — for rectifications"
)
discharge_subtype: Mapped[Optional[str]] = mapped_column(
String(10), comment="TIPODESC: NORMAL, PARCIAL, REPARACION, UTILERIA"
)
partial_sequence: Mapped[Optional[int]] = mapped_column(
Integer, comment="CONSECPARCIAL — for partial discharges"
)
sales_order: Mapped[Optional[str]] = mapped_column(String(20)) # ORDENVENTA
ctm_section: Mapped[Optional[str]] = mapped_column(String(3)) # APARTADOCTM
is_tooling: Mapped[bool] = mapped_column(
server_default=text("false"), comment="PORUTILERIA"
)
discharge_sm: Mapped[Optional[str]] = mapped_column(String(4)) # DESCARGASM
is_repair_update: Mapped[bool] = mapped_column(
server_default=text("false"), comment="ACTUALREPARACION"
)
material_type_expo: Mapped[Optional[str]] = mapped_column(
String(10), comment="TIPOMATEXPO"
)
# ── Cancellation trail ────────────────────────────────────────────────
cancelled_by: Mapped[Optional[str]] = mapped_column(String(20))
cancellation_reason: Mapped[Optional[str]] = mapped_column(String(300))
# ── Relationships ─────────────────────────────────────────────────────
source_invoice: Mapped["InvoiceHeader"] = relationship(
foreign_keys=[source_invoice_id],
)
def_import_invoice: Mapped[Optional["InvoiceHeader"]] = relationship(
foreign_keys=[def_import_invoice_id],
)
details: Mapped[List["DischargeDetail"]] = relationship(
back_populates="header", cascade="all, delete-orphan"
)
scraps: Mapped[List["DischargeScrap"]] = relationship(
back_populates="header", cascade="all, delete-orphan"
)
# ---------------------------------------------------------------------------
# 3. DischargeDetail
#
# The critical traceability link:
# "Export line X consumed Y units from import lot Z"
#
# One row per (export_line × import_lot) pair.
# A single export line can span multiple rows when PEPS pulls from
# more than one import lot.
#
# discharge_header_id → a24.discharge_header.id
# export_item_line_id → a76.item_lines.id (the export line)
# import_item_line_id → a76.item_lines.id (the import lot consumed)
# movement_id → a24.balance_movement.id (the ledger row)
# ---------------------------------------------------------------------------
class DischargeDetail(Base, TenantScopedMixin, TimestampMixin):
"""
One row per (export line × import lot consumed).
This is the traceability record the SAT asks for:
"Show me which import pedimento covered this export line."
"""
__tablename__ = "discharge_detail"
__table_args__ = (
CheckConstraint("movement_id IS NOT NULL", name="ck_dischdet_movement_required"),
Index("ix_dischdet_header", "discharge_header_id"),
Index("ix_dischdet_import_lot", "import_item_line_id"),
Index("ix_dischdet_export_line", "export_item_line_id"),
Index("ix_dischdet_part", "tenant_id", "part_number"),
{"schema": "a24"},
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
discharge_header_id: Mapped[int] = mapped_column(
ForeignKey("a24.discharge_header.id"),
)
# ── Export side ───────────────────────────────────────────────────────
export_item_line_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.item_lines.id"),
comment="NULL for waste-only discharges.",
)
# Denormalized for report performance
part_number: Mapped[Optional[str]] = mapped_column(
String(70), comment="NUMPARTE of the export line (denormalized)"
)
export_part_number: Mapped[Optional[str]] = mapped_column(
String(70), comment="NUMPARTEEXPO — as it appears in the pedimento"
)
export_line_ref: Mapped[Optional[int]] = mapped_column(
Integer, comment="LINEAEXPOREF — for rectification references"
)
# ── Import lot side (PEPS lot consumed) ───────────────────────────────
import_item_line_id: Mapped[int] = mapped_column(
ForeignKey("a76.item_lines.id"),
)
# ── Ledger row created for this consumption (NOT NULL — design rule 3) ─
movement_id: Mapped[int] = mapped_column(
BigInteger,
ForeignKey("a24.balance_movement.id"),
comment="The BalanceMovement that records this consumption. Required.",
)
# ── Consumed quantities and values ────────────────────────────────────
quantity_discharged: Mapped[Decimal] = mapped_column(Numeric(19, 8)) # CANTDESC
unit_of_measure: Mapped[Optional[str]] = mapped_column(String(5))
value_mn: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8))
value_me: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8))
net_weight: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))
gross_weight: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))
# ── Tariff classification of the consumed input ───────────────────────
tariff_fraction: Mapped[Optional[str]] = mapped_column(String(10)) # FRACCIONIMPO
fraction_type: Mapped[Optional[str]] = mapped_column(String(7)) # TIPOFRACCION
ad_valorem: Mapped[Optional[str]] = mapped_column(String(10)) # ADVALOREMIMPO
country_of_origin: Mapped[Optional[str]] = mapped_column(String(3)) # PAISMERCANCIA
sector: Mapped[Optional[str]] = mapped_column(String(8)) # SECTOR
# ── Repair-specific ───────────────────────────────────────────────────
original_part: Mapped[Optional[str]] = mapped_column(
String(70), comment="PARTEORIGINAL"
)
equivalent_quantity: Mapped[Optional[Decimal]] = mapped_column(
Numeric(19, 8), comment="CANTEQUIVALENTE"
)
equivalent_unit: Mapped[Optional[str]] = mapped_column(String(5))
returned_quantity_sm: Mapped[Optional[Decimal]] = mapped_column(
Numeric(19, 8), comment="CANTRETORNADASAM"
)
# ── Waste / scrap ─────────────────────────────────────────────────────
waste_type: Mapped[Optional[str]] = mapped_column(
String(1), comment="M=merma, D=desperdicio, S=scrap"
)
take_balance_base_pt: Mapped[Optional[str]] = mapped_column(
String(2), comment="TOMARSALDOBASEALPT"
)
# ── Tax fields (from SDescargaT) ──────────────────────────────────────
igi_amount: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # MONTOIGI
tax_payment: Mapped[Optional[str]] = mapped_column(String(1)) # PAGOIMPUESTO
has_certificate: Mapped[Optional[str]] = mapped_column(String(1)) # TIENECERT
iva_mn: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # VALORIVAMN
iva_me: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # VALORIVAME
# ── SM origin traceability ────────────────────────────────────────────
origin_import_invoice: Mapped[Optional[str]] = mapped_column(
String(15), comment="FACTURAIMPO original (denorm for SM)"
)
procedence: Mapped[Optional[str]] = mapped_column(String(3)) # PROCEDENCIA
# ── Relationships ─────────────────────────────────────────────────────
header: Mapped["DischargeHeader"] = relationship(back_populates="details")
export_line: Mapped[Optional["LineItem"]] = relationship(
foreign_keys=[export_item_line_id],
)
import_line: Mapped["LineItem"] = relationship(
foreign_keys=[import_item_line_id],
)
movement: Mapped["BalanceMovement"] = relationship(
foreign_keys=[movement_id],
back_populates="discharge_detail",
)
# ---------------------------------------------------------------------------
# 4. DischargeScrap
#
# Mermas, desperdicios and destrucciones.
# Can be standalone (no export invoice) or linked to a DischargeHeader.
# Always generates a BalanceMovement of type WASTE / SCRAP / DESTRUCTION.
#
# Replaces: SDescargaS + SDescargaMerDes
# ---------------------------------------------------------------------------
class DischargeScrap(Base, TenantScopedMixin, TimestampMixin):
"""
Waste / scrap / destruction records.
Replaces SDescargaS and the waste portion of SDescargaMerDes.
"""
__tablename__ = "discharge_scrap"
__table_args__ = (
Index("ix_dischscrap_header", "discharge_header_id"),
Index("ix_dischscrap_import_lot", "import_item_line_id"),
Index("ix_dischscrap_date", "tenant_id", "scrap_date"),
{"schema": "a24"},
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
# ── Optional parent discharge ─────────────────────────────────────────
discharge_header_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a24.discharge_header.id"),
comment="NULL when scrap is registered independently (not tied to an export).",
)
# ── The import lot being scrapped ──────────────────────────────────────
import_item_line_id: Mapped[int] = mapped_column(
ForeignKey("a76.item_lines.id"),
)
# ── Ledger row recording this scrap ───────────────────────────────────
movement_id: Mapped[Optional[int]] = mapped_column(
BigInteger, ForeignKey("a24.balance_movement.id"),
)
# ── Type ──────────────────────────────────────────────────────────────
scrap_type: Mapped[str] = mapped_column(
String(1), comment="M=merma, D=desperdicio, S=scrap, X=destrucción"
)
# ── Finished-good export line that generated this scrap ───────────────
finished_good_line_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.item_lines.id"),
comment="Export line of the product whose manufacture created this scrap.",
)
finished_good_part: Mapped[Optional[str]] = mapped_column(String(70))
# ── Scrap export pedimento (if scrap is exported separately) ──────────
scrap_export_invoice_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.invoice_header.id"),
comment="If desperdicio has its own export pedimento.",
)
# ── The scrapped material ─────────────────────────────────────────────
part_number: Mapped[str] = mapped_column(String(70))
item_class: Mapped[Optional[str]] = mapped_column(String(8))
quantity: Mapped[Decimal] = mapped_column(Numeric(19, 8))
unit_of_measure: Mapped[str] = mapped_column(String(5))
value_mn: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8))
value_me: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8))
net_weight: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))
procedence: Mapped[Optional[str]] = mapped_column(String(3))
scrap_date: Mapped[datetime.date] = mapped_column(Date)
# ── Relationships ─────────────────────────────────────────────────────
header: Mapped[Optional["DischargeHeader"]] = relationship(
back_populates="scraps",
)
import_line: Mapped["LineItem"] = relationship(
foreign_keys=[import_item_line_id],
)
finished_good_line: Mapped[Optional["LineItem"]] = relationship(
foreign_keys=[finished_good_line_id],
)
scrap_export_invoice: Mapped[Optional["InvoiceHeader"]] = relationship(
foreign_keys=[scrap_export_invoice_id],
)
movement: Mapped[Optional["BalanceMovement"]] = relationship(
foreign_keys=[movement_id],
)

View File

@@ -43,6 +43,9 @@ class ClassCreateDTO(BaseModel):
iva_exempt_fraction: Optional[str] = Field(
None, max_length=4, description="IVA exempt fraction"
)
is_active: Optional[bool] = Field(
True, description="Indicates if the class is active (default: true)"
)
class Config:
from_attributes = True
@@ -76,6 +79,9 @@ class ClassCreateDTOFA(ClassCreateDTO):
class_enabled: Optional[bool] = Field(
True, description="Indica si la clase está habilitada"
)
is_active: Optional[bool] = Field(
True, description="Indicates if the class is active (default: true)"
)
class Config:
from_attributes = True
@@ -116,6 +122,9 @@ class ClassUpdateDTO(BaseModel):
iva_exempt_fraction: Optional[str] = Field(
None, max_length=4, description="IVA exempt fraction"
)
is_active: Optional[bool] = Field(
True, description="Indicates if the class is active (default: true)"
)
model_config = ConfigDict(from_attributes=True, extra='forbid') # Explicitly forbid extra fields
@@ -136,6 +145,7 @@ class ClassResponseDTO(BaseModel):
sub_key: Optional[str] = None
physical_review: Optional[int] = None
iva_exempt_fraction: Optional[str] = None
is_active: Optional[bool] = None
created_at: datetime
updated_at: datetime

View File

@@ -84,13 +84,17 @@ class Class(Base, TenantScopedMixin, TimestampMixin):
String(4)
) # FRACCIONEXENTAIVA
is_active: Mapped[bool] = mapped_column(
default=True, server_default="true", nullable=False
) # Campo para habilitar/deshabilitar clases sin eliminarlas
# Relationships
material_type: Mapped[Optional["MaterialType"]] = relationship(
foreign_keys=[material_key]
)
unit_of_measure_info: Mapped[Optional["UnitOfMeasure"]] = relationship(
foreign_keys=[unit_of_measure]
)
)
# Inverse relationship with GParts that have this class
parts: Mapped[list["Part"]] = relationship(

View File

@@ -1,32 +0,0 @@
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from sqlalchemy import (
ForeignKeyConstraint,
Integer,
PrimaryKeyConstraint,
String,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column
class FractionRuleOctave(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "fraction_rule_octave"
__table_args__ = (
PrimaryKeyConstraint("id", name="fraction_rule_octave_pkey"),
UniqueConstraint(
"tenant_id",
"company_id",
"permission",
"line",
"fraction",
name="uq_fraction_rule_octave_permission_line_fraction",
),
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
permission: Mapped[str] = mapped_column(String(20))
line: Mapped[int] = mapped_column(Integer)
fraction: Mapped[str] = mapped_column(String(10))

View File

@@ -15,6 +15,7 @@ from .models import Company
from ...audit_log.services.service import AuditService
from ..units_of_measure.seed import seed as units_of_measure_seed
from ..fractions.historical_tariff_fractions.seed import seed as historical_tariff_fractions_seed
from ..fractions.warning_fractions.seed import seed as warning_fractions_seed
from core.context import get_user_context
from sqlalchemy import text
@@ -755,6 +756,24 @@ class CompanyService:
"""))
db.execute(text("SET session_replication_role = DEFAULT;"))
# 3. Warning Fractions
values_warning = ", ".join(
[
f"({format_value(fraction)}, {format_value(description)}, {format_value(warning_type)}, {tenant_id}, {company_id})"
for fraction, description, warning_type in warning_fractions_seed
]
)
if values_warning:
db.execute(text("ALTER TABLE public.warning_fractions DISABLE TRIGGER ALL;"))
db.execute(text(f"""
INSERT INTO public.warning_fractions
(fraction, description, warning_type, tenant_id, company_id)
VALUES {values_warning}
ON CONFLICT (fraction, company_id) DO NOTHING;
"""))
db.execute(text("ALTER TABLE public.warning_fractions ENABLE TRIGGER ALL;"))
def get_companies_by_tenant(self, tenant_id: int) -> List[Company]:
"""Get all companies for a tenant"""
return (

View File

@@ -0,0 +1,24 @@
from typing import Optional
from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class PreviousFraction(Base, TenantScopedMixin, TimestampMixin):
"""
Tabla de correlación entre fracciones anteriores y actuales.
Paridad: SFraccionesAnterioresN (Clarion SCAII).
Se usa en REVISA_UM_REGLA_OCTAVA para aceptar fracciones cambiadas
cuando el permiso de RO tiene fecha de inicio anterior al corte (~2010-05-31).
"""
__tablename__ = "previous_fractions"
__table_args__ = {"schema": "a76"}
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
current_fraction: Mapped[Optional[str]] = mapped_column(String(50)) # FraccionActual
previous_fraction: Mapped[Optional[str]] = mapped_column(String(50)) # FraccionAnterior

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,47 @@
"""
Modelos ORM para fracciones de alerta
"""
from typing import Optional
from sqlalchemy import Boolean, Integer, PrimaryKeyConstraint, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class WarningFraction(Base, TenantScopedMixin, TimestampMixin):
"""
Modelo para fracciones arancelarias con alerta.
Fracciones que requieren atención especial o documentación adicional.
"""
__tablename__ = "warning_fractions"
__table_args__ = (
PrimaryKeyConstraint("id", name="warning_fractions_pkey"),
UniqueConstraint(
"fraction", "company_id", name="uq_warning_fractions_fraction_company"
),
{"schema": "public"},
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# Código de la fracción arancelaria (ej: 01012101)
fraction: Mapped[str] = mapped_column(String(15), nullable=False, index=True)
# Descripción de la alerta
description: Mapped[Optional[str]] = mapped_column(String(1000))
# Tipo de alerta (ej: PERMISO, RESTRICCION, NOM, etc.)
warning_type: Mapped[Optional[str]] = mapped_column(String(50), index=True)
# Indica si la alerta está activa
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true")
def __repr__(self) -> str:
return (
f"<WarningFraction(fraction='{self.fraction}', "
f"warning_type='{self.warning_type}', is_active={self.is_active})>"
)

View File

@@ -0,0 +1,141 @@
seed = [
("8701210100", "", "PROCESS"),
("8702100500", "", "PROCESS"),
("8702400600", "", "PROCESS"),
("8703310200", "", "PROCESS"),
("8703500200", "", "PROCESS"),
("8704210400", "", "PROCESS"),
("8704410200", "", "PROCESS"),
("8705400200", "", "PROCESS"),
("8711300401", "", "PROCESS"),
("8711409999", "", "PROCESS"),
("8711900100", "", "PROCESS"),
("8716200499", "", "PROCESS"),
("8716390400", "", "PROCESS"),
("8716399903", "", "PROCESS"),
("8716409100", "", "PROCESS"),
("8716900300", "", "PROCESS"),
("8716909901", "", "PROCESS"),
("8716909902", "", "PROCESS"),
("8716909999", "", "PROCESS"),
("8711100300", "", "PROCESS"),
("8711200501", "", "PROCESS"),
("8711200502", "", "PROCESS"),
("8711200599", "", "PROCESS"),
("8711300499", "", "PROCESS"),
("8711400100", "", "PROCESS"),
("8711400200", "", "PROCESS"),
("8711409901", "", "PROCESS"),
("8711500301", "", "PROCESS"),
("8711500302", "", "PROCESS"),
("8711500399", "", "PROCESS"),
("8711600100", "", "PROCESS"),
("8711909900", "", "PROCESS"),
("8716100100", "", "PROCESS"),
("8716200401", "", "PROCESS"),
("8716200402", "", "PROCESS"),
("8703210200", "", "PROCESS"),
("8703220200", "", "PROCESS"),
("8703230200", "", "PROCESS"),
("8703240200", "", "PROCESS"),
("8716310301", "", "PROCESS"),
("8716310302", "", "PROCESS"),
("8716310399", "", "PROCESS"),
("8716390300", "", "PROCESS"),
("8703320200", "", "PROCESS"),
("8703330200", "", "PROCESS"),
("8703400200", "", "PROCESS"),
("8703600200", "", "PROCESS"),
("8703700200", "", "PROCESS"),
("8703900200", "", "PROCESS"),
("8716390800", "", "PROCESS"),
("8716399901", "", "PROCESS"),
("8716399902", "", "PROCESS"),
("8701220100", "", "PROCESS"),
("8701230100", "", "PROCESS"),
("8701240100", "", "PROCESS"),
("8701290100", "", "PROCESS"),
("8704220700", "", "PROCESS"),
("8704230200", "", "PROCESS"),
("8704310500", "", "PROCESS"),
("8704320700", "", "PROCESS"),
("8704420200", "", "PROCESS"),
("8704430200", "", "PROCESS"),
("8704510300", "", "PROCESS"),
("8704520200", "", "PROCESS"),
("8716399904", "", "PROCESS"),
("8716399905", "", "PROCESS"),
("8716399999", "", "PROCESS"),
("8702200500", "", "PROCESS"),
("8702300500", "", "PROCESS")
]

View File

@@ -65,7 +65,8 @@ class UnitOfMeasureCustoms(Base, TimestampMixin):
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
code: Mapped[str] = mapped_column(String(10), nullable=False) # CLAVE
description: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)
description: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)
a76_unit_code: Mapped[Optional[str]] = mapped_column(String(5), nullable=True) # UnidadSCAII
# 5. GUniMedida (Main)

View File

@@ -1,95 +0,0 @@
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
from core.exceptions import ErrorCollector
from ...schemas import InvoiceHeaderCreate
from ...common.common_validators import validate_common, validate_required_fields_by_operation
def validate_create(db: Session, invoice: InvoiceHeaderCreate, tenant_id: int, company_id: int, errors: ErrorCollector) -> None:
""" Valida la creación de una nueva factura de importe temporal """
if not invoice.operation_type:
errors.add_required_error("operation_type")
if not invoice.invoice_type:
errors.add_required_error("invoice_type")
if not invoice.document_type and invoice.invoice_type != "MEX":
errors.add_required_error("document_type")
if not invoice.invoice_number:
errors.add_required_error("invoice_number")
if not invoice.invoice_date:
errors.add_required_error("invoice_date")
# Validar campos obligatorios según tipo de operación
invoice_data = {
'provider_header': invoice.compliance_mx.provider_header if invoice.compliance_mx else None,
'provider_id': invoice.compliance_mx.provider_id if invoice.compliance_mx else None,
'sold_to_header': invoice.compliance_mx.sold_to_header if invoice.compliance_mx else None,
'sold_to_id': invoice.compliance_mx.sold_to_id if invoice.compliance_mx else None,
'shipped_to_header': invoice.compliance_mx.shipped_to_header if invoice.compliance_mx else None,
'shipped_to_id': invoice.compliance_mx.shipped_to_id if invoice.compliance_mx else None,
'customs_broker_id': invoice.compliance_mx.customs_broker_id if invoice.compliance_mx else None,
'pedimento_id': invoice.compliance_mx.pedimento_id if invoice.compliance_mx else None,
}
validate_required_fields_by_operation(
invoice_data=invoice_data,
operation_type=invoice.operation_type,
errors=errors
)
if errors.has_errors():
"""Se retorna porque hay campos obligatorios según el tipo de operación que deben ser llenados"""
return
validate_common(db, invoice, tenant_id, company_id, errors)
if errors.has_errors():
"""Se retorna por que fallaron las validaciones generales"""
return
if invoice.compliance_mx:
if not invoice.compliance_mx.pedimento_id:
invoice.compliance_mx.remesa = None
if invoice.financials:
if not invoice.financials.exchange_rate:
invoice.financials.exchange_rate = db.query(ExchangeRate.value).filter(ExchangeRate.date == invoice.invoice_date).scalar()
invoice.document_type = (invoice.document_type or "").upper()
if invoice.logistics:
if not invoice.logistics.transport_type:
invoice.logistics.transport_type = "none"
if invoice.logistics.transport_type == "none" and invoice.logistics.transport_num:
invoice.logistics.transport_num = None
invoice.logistics.incoterm = (invoice.logistics.incoterm or "").upper()
if not invoice.logistics.weight_type:
invoice.logistics.weight_type = "kgs"
if invoice.financials:
if not invoice.financials.currency:
invoice.financials.currency = "foreign"
if invoice.financials.currency == "local":
invoice.financials.currency_type = "MXN"
elif invoice.financials.currency == "foreign":
invoice.financials.currency_type = "USD"
elif invoice.financials.currency == "manual":
invoice.financials.currency_type = (invoice.financials.currency_type or "").upper()

View File

@@ -1,284 +0,0 @@
from typing import Optional
from datetime import date
from decimal import Decimal
from sqlalchemy.orm import Session
from ...common.common_validators import validate_common, validate_required_fields_by_operation
from core.exceptions import ErrorCollector
from ...schemas import InvoiceHeaderUpdate
from ...models import InvoiceHeader
# Helper function para limpiar strings (equivalente a Clip())
def clean_str(value: Optional[str]) -> Optional[str]:
if value is None or value == "":
return None
return value.strip()
def validate_update(
db: Session,
invoice: InvoiceHeaderUpdate,
existing_invoice: InvoiceHeader,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
) -> None:
"""
Valida y procesa la actualización parcial de una factura de importación temporal.
Lógica: Si un campo viene con valor, se limpia/valida.
Si no, se mantiene el valor existente de la factura.
Args:
invoice: Datos de la factura a validar/actualizar (modificado in-place)
existing_invoice: Factura existente en la base de datos
errors: Colector de errores
Returns:
None (modifica invoice in-place y acumula errores en errors)
"""
# Validar campos requeridos según el tipo de operación
invoice_dict = {
'provider_id': invoice.compliance_mx.provider_id if invoice.compliance_mx else (existing_invoice.compliance_mx.provider_id if existing_invoice.compliance_mx else None),
'sold_to_id': invoice.compliance_mx.sold_to_id if invoice.compliance_mx else (existing_invoice.compliance_mx.sold_to_id if existing_invoice.compliance_mx else None),
'sold_to_header': invoice.compliance_mx.sold_to_header if invoice.compliance_mx else (existing_invoice.compliance_mx.sold_to_header if existing_invoice.compliance_mx else None),
'shipped_to_id': invoice.compliance_mx.shipped_to_id if invoice.compliance_mx else (existing_invoice.compliance_mx.shipped_to_id if existing_invoice.compliance_mx else None),
'customs_broker_id': invoice.compliance_mx.customs_broker_id if invoice.compliance_mx else (existing_invoice.compliance_mx.customs_broker_id if existing_invoice.compliance_mx else None),
'pedimento_id': invoice.compliance_mx.pedimento_id if invoice.compliance_mx else (existing_invoice.compliance_mx.pedimento_id if existing_invoice.compliance_mx else None),
}
validate_required_fields_by_operation(
invoice_data=invoice_dict,
operation_type=invoice.operation_type or (existing_invoice.operation_type or 'imp'),
errors=errors
)
# Primero ejecutar validaciones comunes
validate_common(db, invoice, tenant_id, company_id, errors)
# Mapeo de columnas CSV a campos de la factura
# Siguiendo la lógica del código Clarion original
# Columna A: Pedimento (si no viene en CSV, usar el existente)
if invoice.compliance_mx:
if invoice.compliance_mx.pedimento_id:
invoice.compliance_mx.pedimento_id = invoice.compliance_mx.pedimento_id
else:
invoice.compliance_mx.pedimento_id = existing_invoice.compliance_mx.pedimento_id if existing_invoice.compliance_mx else None
# Columna B: Remesa
if invoice.compliance_mx:
if invoice.compliance_mx.remesa:
invoice.compliance_mx.remesa = invoice.compliance_mx.remesa
else:
invoice.compliance_mx.remesa = existing_invoice.compliance_mx.remesa if existing_invoice.compliance_mx else None
# Columna C: Factura (OBLIGATORIO)
if invoice.invoice_number is not None:
invoice.invoice_number = clean_str(invoice.invoice_number)
if not invoice.invoice_number:
errors.add_required_error("invoice_number")
else:
invoice.invoice_number = existing_invoice.invoice_number
# Columna D: Fecha
if not invoice.invoice_date:
invoice.invoice_date = existing_invoice.invoice_date
# Columna E: Tipo Cambio
if invoice.financials:
if invoice.financials.exchange_rate is None:
if existing_invoice.financials:
invoice.financials.exchange_rate = existing_invoice.financials.exchange_rate
# Columna F: Régimen
if invoice.document_type:
invoice.document_type = clean_str(invoice.document_type).upper()
else:
invoice.document_type = existing_invoice.document_type
# Columna G: Clave Proveedor
if invoice.compliance_mx:
if invoice.compliance_mx.provider_id is None:
invoice.compliance_mx.provider_id = existing_invoice.compliance_mx.provider_id if existing_invoice.compliance_mx else None
# Columna H: Clave Vendido A
if invoice.compliance_mx:
if invoice.compliance_mx.sold_to_id is None:
invoice.compliance_mx.sold_to_id = existing_invoice.compliance_mx.sold_to_id if existing_invoice.compliance_mx else None
# Columna I: Clave Enviado A
if invoice.compliance_mx:
if invoice.compliance_mx.shipped_to_id is None:
invoice.compliance_mx.shipped_to_id = existing_invoice.compliance_mx.shipped_to_id if existing_invoice.compliance_mx else None
# Columna J: Clave A. Aduanal
if invoice.compliance_mx:
if invoice.compliance_mx.customs_broker_id is None:
invoice.compliance_mx.customs_broker_id = existing_invoice.compliance_mx.customs_broker_id if existing_invoice.compliance_mx else None
# Columna K: Clave Transportista
if invoice.logistics:
# Note: logistics in update schema seems to be a single object, but in model it's a list.
# This validator seems to expect a single object (InvoiceLogisticsUpdate).
# We'll stick to the existing logic but make it safe.
if hasattr(invoice.logistics, 'carrier_id') and invoice.logistics.carrier_id is None:
invoice.logistics.carrier_id = existing_invoice.logistics.carrier_id if existing_invoice.logistics else None
# Columna L: Nombre Conductor
if invoice.logistics:
if hasattr(invoice.logistics, 'driver_name') and not invoice.logistics.driver_name:
invoice.logistics.driver_name = existing_invoice.logistics.driver_name if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'driver_name'):
invoice.logistics.driver_name = clean_str(invoice.logistics.driver_name)
# Columna M: Tipo Transporte
if invoice.logistics:
if hasattr(invoice.logistics, 'transport_type') and not invoice.logistics.transport_type:
invoice.logistics.transport_type = existing_invoice.logistics.transport_type if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'transport_type'):
invoice.logistics.transport_type = clean_str(invoice.logistics.transport_type)
# Columna N: Número de Transporte
if invoice.logistics:
if hasattr(invoice.logistics, 'transport_num') and not invoice.logistics.transport_num:
invoice.logistics.transport_num = existing_invoice.logistics.transport_num if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'transport_num'):
invoice.logistics.transport_num = clean_str(invoice.logistics.transport_num)
# Columna O: Tipo de Moneda
if invoice.financials:
if not invoice.financials.currency:
invoice.financials.currency = existing_invoice.financials.currency if existing_invoice.financials else None
else:
invoice.financials.currency = clean_str(invoice.financials.currency).lower()
# Columna P: Clave Moneda
if invoice.financials:
if not invoice.financials.currency_type:
invoice.financials.currency_type = existing_invoice.financials.currency_type if existing_invoice.financials else None
else:
invoice.financials.currency_type = clean_str(invoice.financials.currency_type).upper()
# Columna Q: Flete
if invoice.financials:
if invoice.financials.freight is None:
invoice.financials.freight = existing_invoice.financials.freight if existing_invoice.financials else None
# Columna R: Val Seguros
if invoice.financials:
if invoice.financials.insurance_value is None:
invoice.financials.insurance_value = existing_invoice.financials.insurance_value if existing_invoice.financials else None
# Columna S: Seguros
if invoice.financials:
if invoice.financials.insurance is None:
invoice.financials.insurance = existing_invoice.financials.insurance if existing_invoice.financials else None
# Columna T: Embalaje
if invoice.financials:
if invoice.financials.packaging is None:
invoice.financials.packaging = existing_invoice.financials.packaging if existing_invoice.financials else None
# Columna U: Otros Incrementables
if invoice.financials:
if invoice.financials.other_increments is None:
invoice.financials.other_increments = existing_invoice.financials.other_increments if existing_invoice.financials else None
# Columna V: Incoterms
if invoice.logistics:
if hasattr(invoice.logistics, 'incoterm') and not invoice.logistics.incoterm:
invoice.logistics.incoterm = existing_invoice.logistics.incoterm if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'incoterm'):
invoice.logistics.incoterm = clean_str(invoice.logistics.incoterm).upper()
# Columna W: Precinto
if invoice.logistics:
if hasattr(invoice.logistics, 'seal_number') and not invoice.logistics.seal_number:
invoice.logistics.seal_number = existing_invoice.logistics.seal_number if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'seal_number'):
invoice.logistics.seal_number = clean_str(invoice.logistics.seal_number)
# Columna X: Fecha de Emisión
if not invoice.emission_date:
invoice.emission_date = existing_invoice.emission_date
# Columna Y: Tipo de Peso (Opcional)
if invoice.logistics:
if hasattr(invoice.logistics, 'weight_type') and not invoice.logistics.weight_type:
invoice.logistics.weight_type = existing_invoice.logistics.weight_type if existing_invoice.logistics else None
elif hasattr(invoice.logistics, 'weight_type'):
invoice.logistics.weight_type = clean_str(invoice.logistics.weight_type).upper()
# Columna Z: Número de Manifiesto (Opcional)
if invoice.compliance_mx.manifest_number:
if not invoice.compliance_mx.manifest_number:
invoice.compliance_mx.manifest_number = existing_invoice.compliance_mx.manifest_number if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.manifest_number = clean_str(invoice.compliance_mx.manifest_number)
# Columna AA: E-Document (Opcional)
if invoice.compliance_mx:
if not invoice.compliance_mx.edocument:
invoice.compliance_mx.edocument = existing_invoice.compliance_mx.edocument if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.edocument = clean_str(invoice.compliance_mx.edocument)
# Columna AB: Num. Operación (Opcional)
if invoice.compliance_mx:
if not invoice.compliance_mx.vucem_operation_num:
invoice.compliance_mx.vucem_operation_num = existing_invoice.compliance_mx.vucem_operation_num if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.vucem_operation_num = clean_str(invoice.compliance_mx.vucem_operation_num)
# Columna AB: Aduana (OBLIGATORIO)
if invoice.compliance_mx:
if not invoice.compliance_mx.aduana:
invoice.compliance_mx.aduana = existing_invoice.compliance_mx.aduana if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.aduana = clean_str(invoice.compliance_mx.aduana)
# Columna AC: Enviado Por (Obligatorio)
if invoice.compliance_mx:
if not invoice.compliance_mx.shipped_by_id:
invoice.compliance_mx.shipped_by_id = existing_invoice.compliance_mx.shipped_by_id if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.shipped_by_id = clean_str(invoice.compliance_mx.shipped_by_id)
# Columna AD: Aduana_Cruce (Obligatorio)
current_aduana = invoice.compliance_mx.aduana if invoice.compliance_mx else (existing_invoice.compliance_mx.aduana if existing_invoice.compliance_mx else None)
if not current_aduana:
errors.add_required_error("aduana")
# Columna AC: Sección de Despacho / Puerto de Entrada (Opcional)
if invoice.compliance_mx:
if not invoice.compliance_mx.port_of_entry:
invoice.compliance_mx.port_of_entry = existing_invoice.compliance_mx.port_of_entry if existing_invoice.compliance_mx else None
else:
invoice.compliance_mx.port_of_entry = clean_str(invoice.compliance_mx.port_of_entry)
# Columna AE: Observación en Español (Opcional)
if not invoice.observation_es:
invoice.observation_es = existing_invoice.observation_es
else:
invoice.observation_es = clean_str(invoice.observation_es)
# Columna AF: Observación en Inglés (Opcional)
if not invoice.observation_en:
invoice.observation_en = existing_invoice.observation_en
else:
invoice.observation_en = clean_str(invoice.observation_en)
# Columna AG: cfdi_uuid (Opcional)
if not invoice.cfdi_uuid:
invoice.cfdi_uuid = existing_invoice.cfdi_uuid if existing_invoice.compliance_mx else None
else:
invoice.cfdi_uuid = clean_str(invoice.cfdi_uuid)
# Columna AH: Localizacion (Opcional)
if invoice.compliance_mx.location:
if not invoice.compliance_mx.location:
invoice.compliance_mx.location = existing_invoice.compliance_mx.location if existing_invoice.compliance_mx.location else None
else:
invoice.compliance_mx.location = clean_str(invoice.compliance_mx.location)

View File

@@ -0,0 +1,285 @@
from datetime import date, datetime
from decimal import Decimal
from typing import List
from sqlalchemy.orm import Session
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.invoices.models import Currency, InvoiceHeader, InvoiceStatus
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.items.common.fractions import search_fraction_preference
from api.v1.modules.a76.general_catalogs.company.models import Company
from core.exceptions import ErrorCollector
from .pre_validators import pre_validators
from .sub_process.review_classes import review_classes
from .sub_process.review_exchange_rate import review_exchange_rate
from .sub_process.review_weights import review_weights_kgs, review_weights_lbs
from .sub_process.review_series import review_series
from .sub_process.review_rule_octave import (
llena_impo_permiso_regla_octava,
revpermiso_regla_octava,
valida_imp_regla_octava,
descuenta_cupo_r_octava,
)
from .sub_process.review_uma import revisa_uma
from .sub_process.assing_values import assign_values_lines, assign_values_invoice
def _validate_lines(
db: Session,
invoice: InvoiceHeader,
lines: List[LineItem],
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> tuple[list, dict]:
"""Recorre cada partida y ejecuta las validaciones individuales."""
company = db.get(Company, invoice.company_id)
company_rfc = (company.rfc or "").strip().upper() if company else ""
# Deduplicación de cupos disponibles: (permiso, ro_line, pais) → OctaveAvailableEntry
octave_available: dict = {}
# Partidas a descargar: se pasa a valida_imp_regla_octava
octave_desc: list = []
for line in lines:
# Validar costo unitario capturado en partidas principales
if (line.financial and line.fa_data) and not line.fa_data.is_subitem and (line.financial.unit_cost_capture or Decimal(0)) == 0:
errors.add_error(
field=f"line[{line.line_number}].unit_cost_capture",
message="No existe el costo unitario para la Partida.",
solution=[
f"Entrar a la partida: {line.line_number} y capturar el Costo Unitario."
],
code="UNIT_COST_REQUIRED",
)
# Validar clase habilitada/deshabilitada
if line.class_id is not None:
cls: Class | None = db.get(Class, line.class_id)
if cls and cls.is_active is False:
errors.add_error(
field=f"line[{line.line_number}].class",
message=(
f"El número de parte: {cls.class_code} esta desactivado, no se pueden hacer movimientos."
),
solution=["Seleccionar un número de parte activo."],
code="CLASS_DISABLED",
)
# Validar número de parte habilitado/deshabilitado
if line.part_number_id is not None:
part: Part | None = db.get(Part, line.part_number_id)
if part and part.is_active is False:
errors.add_error(
field=f"line[{line.line_number}].part_number",
message=(
f"El número de parte: {part.part_number} esta desactivado, no se pueden hacer movimientos."
),
solution=["Seleccionar un número de parte activo."],
code="PART_DISABLED",
)
search_fraction_preference(
db=db,
country=line.customs.origin_country or "",
fraccion=line.customs.fraction if line.customs else "",
fraction_type=line.customs.fraction_type if line.customs else "",
sector=line.customs.sector if line.customs else "",
invoice_date=invoice.invoice_date,
errors=errors,
)
review_series(db, line, company_rfc, errors)
# Validación de la Regla Octava
if line.octave_permit:
if not company.prosec:
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message="No se puede hacer uso de la Regla Octava, ...",
solution=["Borrar el permiso o dar de alta el permiso PROSEC..."],
code="OCTAVA_SIN_PROSEC",
)
else:
desc_entry = llena_impo_permiso_regla_octava(
db=db,
invoice=invoice,
line=line,
company_rfc=company_rfc,
tenant_id=tenant_id,
company_id=company_id,
errors=errors,
)
if desc_entry is not None:
octave_desc.append(desc_entry)
available = revpermiso_regla_octava(
db=db,
invoice=invoice,
line=line,
company_rfc=company_rfc,
tenant_id=tenant_id,
company_id=company_id,
errors=errors,
)
if available is not None:
key = (available.octave_permit, available.ro_line, available.country_code)
octave_available.setdefault(key, available)
revisa_uma(db=db, line=line, tenant_id=tenant_id, company_id=company_id, errors=errors)
return octave_desc, octave_available
def _validate_sisimp_limits(
invoice: InvoiceHeader,
errors: ErrorCollector,
) -> None:
"""
Valida los límites de cantidad, peso y valor configurados en SisImp.
TODO: Leer los parámetros SisImp desde la configuración del sistema:
- SisImp:CantLimiteMin / SisImp:CantLimite
- SisImp:PesoLimiteMin / SisImp:PesoLimite
- SisImp:ValorLimiteMin / SisImp:ValorLimite
Una vez disponibles, usar invoice.financials.total_quantity, net_weight y value_mc.
"""
# TODO: Implementar cuando SisImp esté disponible en la configuración del tenant
pass
def _update_invoice_totals(invoice: InvoiceHeader) -> None:
"""
Copia los totales calculados de financials/logistics al encabezado de la factura
y calcula IVA, incrementables y valores de aduanas.
TODO: Validar SSisGen:ActSeguridad para asignar el usuario que actualizó.
TODO: Validar SSisGen:CalValBaseTCPed para registrar el mensaje de procesamiento.
"""
tc = Decimal(str(invoice.financials.exchange_rate or 0))
tc_mm = Decimal(str(invoice.financials.exchange_rate_mm or 0))
# Calcular IVA sólo para facturas con fecha posterior al corte (78165 en Clarion = 2004-06-01 aprox.)
iva_factor = Decimal(str(invoice.financials.iva_factor or 0)) if invoice.financials.iva_factor else Decimal(0)
if invoice.invoice_date and invoice.invoice_date >= date(2014, 12, 31):
invoice.financials.iva_mn = float(Decimal(str(invoice.financials.value_mn or 0)) * iva_factor / 100)
invoice.financials.iva_me = float(Decimal(str(invoice.financials.value_me or 0)) * iva_factor / 100)
else:
invoice.financials.iva_mn = 0.0
invoice.financials.iva_me = 0.0
# Calcular total de incrementables por tipo de moneda
freight = Decimal(str(invoice.financials.freight or 0))
insurance = Decimal(str(invoice.financials.insurance or 0))
packaging = Decimal(str(invoice.financials.packaging or 0))
other = Decimal(str(invoice.financials.other_increments or 0))
base_increm = freight + insurance + packaging + other
if invoice.financials.currency == Currency.FOREIGN: # ME
val_seguro = Decimal(str(invoice.financials.total_increments_me or 0)) - base_increm
invoice.financials.total_increments_me = float(base_increm + val_seguro)
invoice.financials.total_increments_mn = float(Decimal(str(invoice.financials.total_increments_me)) * tc)
elif invoice.financials.currency == Currency.LOCAL: # MN
val_seguro = Decimal(str(invoice.financials.total_increments_mn or 0)) - base_increm
invoice.financials.total_increments_mn = float(base_increm + val_seguro)
invoice.financials.total_increments_me = float(
(Decimal(str(invoice.financials.total_increments_mn)) / tc) if tc else Decimal(0)
)
elif invoice.financials.currency == Currency.MANUAL: # MC
val_seguro = (
(Decimal(str(invoice.financials.total_increments_me or 0)) / tc_mm) if tc_mm else Decimal(0)
) - base_increm
invoice.financials.total_increments_me = float((base_increm + val_seguro) * tc_mm)
invoice.financials.total_increments_mn = float(Decimal(str(invoice.financials.total_increments_me)) * tc)
# Marcar la factura como procesada
invoice.status = InvoiceStatus.PROCESSED
invoice.party_count = len(invoice.financials.__dict__) # se sobreescribirá con el conteo real
# TODO: SSisGen:ActSeguridad = 1 → invoice.updated_by = current_user
# TODO: SSisGen:CalValBaseTCPed = 1 →
# invoice.process_log = "Se Actualizó con el Tipo de Cambio de la Fecha de Pago de Pedimento."
def main_process(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id: str) -> dict:
"""
Proceso principal para importar facturas.
Flujo (porta la rutina principal del legacy SCAII):
1. Validación previa de datos (pre_validators)
2. Tipo de cambio del pedimento (TODO: SSisGen:CalValBaseTCPed)
3. Revisión de clases, tipo de cambio y pesos
4. Asignación de valores por partida y totalización
5. Validaciones per-línea (costo, clase, número de parte, Regla Octava, UMA)
6. Validación de límites SisImp (TODO)
7. Si no hay errores: actualizar totales e incrementables en la factura y hacer commit
8. Si hay errores: rollback (SQLAlchemy lo maneja con la excepción)
"""
errors = ErrorCollector()
# Paso 1: Validación previa
lines = pre_validators(db, invoice, tenant_id, company_id, errors)
if not lines:
errors.add_error(
field="line_items",
message="La factura debe contener al menos una partida para ser importada",
solution=["Agregue partidas a la factura antes de intentar importarla"],
code="NO_LINE_ITEMS",
)
errors.raise_if_errors()
# TODO: SSisGen:CalValBaseTCPed = 1 → obtener tipo de cambio de la fecha de pago del pedimento
# y asignarlo a invoice.financials.exchange_rate antes de continuar.
# invoice.which_exchange_rate = 'TCPED' (o 'TCFAC' si CalValBaseTCPed = 0)
# Paso 2: Revisión de clases y fracciones
review_classes(db, invoice, lines, tenant_id, company_id, errors)
review_exchange_rate(db, invoice, errors)
if invoice.logistics and invoice.logistics.weight_type == "kgs":
review_weights_kgs(db, lines, tenant_id, company_id, errors)
elif invoice.logistics and invoice.logistics.weight_type == "lbs":
review_weights_lbs(db, lines, tenant_id, company_id, errors)
# Paso 3: Asignación de valores por partida y totalización de factura
assign_values_lines(invoice, lines)
assign_values_invoice(invoice, lines)
# Paso 4: Validaciones per-línea
octave_desc, octave_available = _validate_lines(db, invoice, lines, tenant_id, company_id, errors)
company = db.get(Company, invoice.company_id)
if company.prosec and octave_desc:
valida_imp_regla_octava(
db=db,
desc_list=octave_desc,
dis_dict=octave_available,
tenant_id=tenant_id,
company_id=company_id,
errors=errors,
)
# Paso 5: Límites de SisImp
_validate_sisimp_limits(invoice, errors)
errors.raise_if_errors()
# Paso 6: Descontar cupos de Regla Octava
sql_errors: list = []
if octave_desc:
descuenta_cupo_r_octava(
db=db,
desc_list=octave_desc,
tenant_id=tenant_id,
company_id=company_id,
sql_errors=sql_errors,
)
# Paso 7: Actualizar totales, IVA e incrementables y marcar como procesada
_update_invoice_totals(invoice)
db.flush()

View File

@@ -0,0 +1,107 @@
from sqlalchemy import func
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.general_catalogs.fractions.warning_fractions.models import WarningFraction
from core.exceptions import ErrorCollector
def pre_validators(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id: str, errors: ErrorCollector):
if invoice.status == InvoiceStatus.PROCESSED:
errors.add(
"status",
"La factura ya fue procesada y no puede ser exportada",
solution=["Verifique el estatus de la factura antes de intentar exportarla"],
code="ALREADY_PROCESSED",
value=invoice.status,
)
if not invoice.invoice_date:
errors.add_required_error("invoice_date")
if not invoice.document_type:
errors.add_required_error("document_type")
if not invoice.compliance_mx.provider_id:
errors.add_required_error("compliance_mx.provider_id")
if not invoice.compliance_mx.sold_to_id:
errors.add_required_error("compliance_mx.sold_to_id")
if not invoice.compliance_mx.shipped_to_id:
errors.add_required_error("compliance_mx.shipped_by_id")
if not invoice.compliance_mx.customs_broker_id:
errors.add_required_error("compliance_mx.customs_broker_id")
#TODO: SSISGEN: Seguridad Ejemplo en: BrowseQFacImp
# 2.- Existe tipo de cambio para la factura seleccionada
#TODO: SSISGEN: VALIDACION DEL TIPO DE CAMBIO EN BASE A LA FECHA DE PAGO DEL PEDIMENTO.
if not invoice.financials.exchange_rate or invoice.financials.exchange_rate <= 0:
errors.add_range_error(
"financials.exchange_rate",
min_value=0.0001,
)
if not invoice.financials.currency:
errors.add_required_error("El Tipo de Moneda esta vacio no se puede actualizar")
elif invoice.financials.currency == "manual" and not invoice.financials.currency_type:
errors.add_required_error("financials.currency_type")
# 3.- Validacion que deber de existir un pedimento cuando es requerido
if not invoice.compliance_mx.is_pedimento_pending and not invoice.compliance_mx.pedimento_id:
errors.add_required_error("compliance_mx.pedimento_number")
# 4.- Verificacion de que existan partidas para la factura, si no hay partidas no se puede procesar
item_count = (
db.query(func.count(LineItem.id))
.filter(
LineItem.invoice_id == invoice.id,
LineItem.tenant_id == invoice.tenant_id,
LineItem.company_id == invoice.company_id,
)
.scalar()
)
if item_count == 0:
errors.add_error(
field="items",
message="La factura no tiene partidas capturadas.",
solution=["Capture al menos una partida antes de procesar la factura."],
code="NO_ITEMS_FOUND",
)
# Advertencias para las fracciones y su horario
lines = db.query(LineItem).filter(
LineItem.invoice_id == invoice.id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
).all()
fractions = {line.fraction for line in lines if line.fraction}
if fractions:
warned_fractions = {
row.fraction
for row in db.query(WarningFraction.fraction)
.filter(WarningFraction.fraction.in_(fractions), WarningFraction.tenant_id == tenant_id)
.all()
}
for line in lines:
if line.fraction in warned_fractions:
errors.add_warning(
field="fraction",
message="Advertencia: Esta mercancía, sólo podrá entrar al territorio nacional por las aduanas del país, de lunes a sábado de 8:00 a 13:00 hrs. Ley 10, 18, LIGIE 1, Capítulo 87, RGCE 4.5.31., Anexo 4.",
solution=[""],
code="WARNING_FRACTION",
)
return lines

View File

@@ -0,0 +1,75 @@
from typing import Any, Dict
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from core.celery_app import celery_app
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from .task import process_invoice_task
router = APIRouter()
@router.post("/invoices/{invoice_id}/process")
def trigger_invoice_process(
invoice_id: int,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Inicia el procesamiento de una factura de importación como tarea Celery.
Retorna el task_id para hacer polling del progreso.
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
task = process_invoice_task.apply_async(
args=[invoice_id, str(tenant_id), str(company_id)]
)
return {"task_id": task.id}
@router.get("/invoices/process/{task_id}/status")
def get_invoice_process_status(task_id: str):
"""
Consulta el estado de progreso de una tarea de procesamiento de factura.
Retorna:
- state: 'PROCESSING' | 'SUCCESS' | 'FAILURE'
- info: { current: int, status: str } (cuando state == 'PROCESSING')
- result: dict (cuando state == 'SUCCESS' o 'FAILURE')
"""
task_result = celery_app.AsyncResult(task_id)
if task_result.state in ("PENDING", "STARTED"):
return {
"state": "PROCESSING",
"info": {"current": 0, "status": "Iniciando..."},
}
if task_result.state == "PROGRESS":
return {
"state": "PROCESSING",
"info": task_result.info or {"current": 0, "status": "Procesando..."},
}
if task_result.state == "SUCCESS":
return {
"state": "SUCCESS",
"result": task_result.result,
}
# FAILURE u otro estado de error
error_info = task_result.result
if isinstance(error_info, Exception):
error_msg = str(error_info)
else:
error_msg = str(error_info) if error_info else "Error desconocido"
return {
"state": "FAILURE",
"result": error_msg,
}

View File

@@ -0,0 +1,99 @@
from decimal import Decimal
from typing import List
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import Currency, InvoiceHeader
from api.v1.modules.a76.items.models import LineItem
def assign_values_lines(
invoice: InvoiceHeader,
lines: List[LineItem],
) -> None:
"""
Calculates and assigns unit costs and values in all currency types for every
line item, based on the invoice currency and exchange rates.
Also resets inventory counters (quantity_returned, quantity_returned_temp,
quantity_existence) to zero, as done in the legacy ASIGNAVALORES_PARTIDA routine.
Currency mapping (legacy -> current enum):
ME (moneda extranjera / foreign) -> Currency.FOREIGN
MN (moneda nacional / local) -> Currency.LOCAL
MC (moneda de cuenta / manual) -> Currency.MANUAL
"""
currency = invoice.financials.currency
tc = Decimal(str(invoice.financials.exchange_rate or 0))
tc_mm = Decimal(str(invoice.financials.exchange_rate_mm or 0))
for line in lines:
if line.financial is None:
continue
capture = line.financial.unit_cost_capture or Decimal(0)
qty = (line.quantity.quantity or Decimal(0)) if line.quantity else Decimal(0)
if currency == Currency.FOREIGN: # ME
line.financial.unit_cost_usd = capture
line.financial.value_usd = capture * qty
line.financial.unit_cost_mxn = capture * tc
line.financial.value_mxn = capture * tc * qty
line.financial.value_mc = capture * qty
elif currency == Currency.LOCAL: # MN
line.financial.unit_cost_mxn = capture
line.financial.value_mxn = capture * qty
line.financial.unit_cost_usd = (capture / tc) if tc else Decimal(0)
line.financial.value_usd = (capture / tc * qty) if tc else Decimal(0)
line.financial.value_mc = capture * qty
elif currency == Currency.MANUAL: # MC
line.financial.unit_cost_usd = capture * tc_mm
line.financial.value_usd = capture * tc_mm * qty
line.financial.unit_cost_mxn = capture * tc_mm * tc
line.financial.value_mxn = capture * tc_mm * tc * qty
line.financial.value_mc = capture * qty
# Reset inventory counters
if line.quantity is not None:
line.quantity.quantity_returned = Decimal(0)
line.quantity.quantity_returned_temp = Decimal(0)
line.quantity.quantity_existence = Decimal(0)
def assign_values_invoice(
invoice: InvoiceHeader,
lines: List[LineItem],
) -> None:
"""
Totals quantity, weight, and value fields from all line items and writes the
aggregated results to invoice.financials.
Ported from legacy ASIGNAVALORES_FACTURA routine.
"""
total_quantity = Decimal(0)
total_net_weight = Decimal(0)
total_gross_weight = Decimal(0)
total_packages = 0
total_value_mxn = Decimal(0)
total_value_usd = Decimal(0)
total_value_mc = Decimal(0)
for line in lines:
if line.quantity:
total_quantity += line.quantity.quantity or Decimal(0)
total_net_weight += line.quantity.net_weight or Decimal(0)
total_gross_weight += line.quantity.gross_weight or Decimal(0)
total_packages += line.quantity.package_quantity or 0
if line.financial:
total_value_mxn += line.financial.value_mxn or Decimal(0)
total_value_usd += line.financial.value_usd or Decimal(0)
total_value_mc += line.financial.value_mc or Decimal(0)
if invoice.financials is not None:
invoice.financials.value_mn = float(total_value_mxn)
invoice.financials.value_me = float(total_value_usd)
invoice.financials.value_mc = float(total_value_mc)
invoice.financials.total_quantity = float(total_quantity)
invoice.financials.net_weight = float(total_net_weight)
invoice.financials.gross_weight = float(total_gross_weight)
invoice.financials.total_packages = total_packages

View File

@@ -0,0 +1,163 @@
from typing import List
from sqlalchemy import func, or_
from sqlalchemy.orm import Session
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.fractions.historical_tariff_fractions.models import HistoricalTariffFraction
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import TariffFraction
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.items.models import LineItem
from core.exceptions import ErrorCollector
def _fraction_exists_in_catalog(db: Session, fraction_code: str) -> bool:
"""Returns True if the fraction exists in TariffFraction (SFracciones) or
HistoricalTariffFraction (GFraccionesHistorico).
Fraction format: first 8 chars = base fraction, chars 9-10 = NICO/country (optional).
"""
if not fraction_code:
return True
base_frac = fraction_code[:8]
nico = fraction_code[8:10] if len(fraction_code) > 8 else ""
# Check SFracciones (TariffFraction)
tariff_q = db.query(TariffFraction).filter(
func.left(TariffFraction.code, 8) == base_frac
)
if nico:
tariff_q = tariff_q.filter(TariffFraction.nico == nico)
else:
tariff_q = tariff_q.filter(
or_(TariffFraction.nico.is_(None), TariffFraction.nico == "")
)
if tariff_q.first() is not None:
return True
# Check GFraccionesHistorico (HistoricalTariffFraction)
hist_q = db.query(HistoricalTariffFraction).filter(
HistoricalTariffFraction.historical_fraction == base_frac
)
if nico:
hist_q = hist_q.filter(HistoricalTariffFraction.country == nico)
else:
hist_q = hist_q.filter(
or_(
HistoricalTariffFraction.country.is_(None),
HistoricalTariffFraction.country == "",
)
)
return hist_q.first() is not None
def _validate_line_fraction(
db: Session, line: LineItem, errors: ErrorCollector
) -> None:
"""Adds a FRACCION error if the line's fraction does not exist in either catalog."""
fraction = line.customs.fraction if line.customs else None
if not fraction:
return
if _fraction_exists_in_catalog(db, fraction):
return
class_code = line.class_info.class_code if line.class_info else ""
errors.add_error(
field="fraction",
message=(
f"La Factura contiene la fraccion: {fraction} asociada al Clase {class_code} "
"que no existe en el catálogo de fracciones"
),
solution=["Agregar la fracción a fracciones históricas."],
code="FRACCION",
)
def review_classes(
db: Session,
invoice: InvoiceHeader,
lines: List[LineItem],
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> None:
"""
Validates class and fraction integrity for all line items of an invoice.
Logic ported from legacy REVISA_CLASE routine:
1. Identify line items with no assigned class (class_id IS NULL).
2. If rule_3121_parties_ii: validate that every line has container_parts_ii set.
3a. If no class errors: validate fractions for ALL lines.
3b. If class errors exist: add a CLASE error per invalid line and validate
fractions only for those lines.
4. If the invoice has not been reviewed by the company, flag lines whose
class requires physical review (Class.physical_review == 1).
"""
# 1. Lines whose class is not assigned / not found in the catalog
invalid_class_lines = [line for line in lines if line.class_id is None]
has_class_errors = bool(invalid_class_lines)
# 2. Container validation (Regla 3.1.21 Partes II)
if invoice.compliance_mx.rule_3121_parties_ii:
for line in lines:
if not line.container_parts_ii:
class_code = line.class_info.class_code if line.class_info else ""
errors.add_error(
field="container_parts_ii",
message=f"La clase: {class_code} no tiene contenedor asignado",
solution=[f"Capturar el contenedor en la partida: {line.line_number}"],
code="CONTENEDOR",
)
# 3. Fraction and class validation
if not has_class_errors:
# All classes are valid — validate fractions for every line
for line in lines:
_validate_line_fraction(db, line, errors)
else:
# Some classes are missing — report CLASE errors and validate
# fractions only for the affected lines
for line in invalid_class_lines:
errors.add_error(
field="class",
message=f"La clase no existe en catálogo de clases (línea: {line.line_number})",
solution=[
f"Borrar la partida: {line.line_number}, "
"o dar de alta la clase en el catálogo de Clases"
],
code="CLASE",
)
_validate_line_fraction(db, line, errors)
# 4. Physical review check
if not invoice.compliance_mx.was_reviewed_by_company:
review_required_lines = (
db.query(LineItem)
.join(Class, LineItem.class_id == Class.id)
.filter(
LineItem.invoice_id == invoice.id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
Class.physical_review == 1,
)
.all()
)
for line in review_required_lines:
cls = db.get(Class, line.class_id)
class_code = cls.class_code if cls else ""
errors.add_error(
field="physical_review",
message=(
f"La Clase: {class_code} en la Linea: {line.line_number} "
"No ha sido Revisada."
),
solution=[
f"Revisar el Equipo de la Partida: {line.line_number} "
"y Asignar Como Revisado a Nivel Factura."
],
code="CLASE",
)

View File

@@ -0,0 +1,67 @@
from decimal import Decimal
from sqlalchemy import func
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
from api.v1.modules.a76.invoices.models import InvoiceHeader
from core.exceptions import ErrorCollector
def review_exchange_rate(
db: Session,
invoice: InvoiceHeader,
errors: ErrorCollector,
) -> None:
"""
Validates that the invoice's exchange rate matches the one registered in the
catalog for the invoice date (gtipocambio / ExchangeRate table).
Ported from legacy REVISA_TIPOCAMBIO routine.
Only runs when the system is NOT configured to use the pedimento's exchange
rate (SisGen:CalValBaseTCPed = 0), which corresponds to the TODO comment in
main_process: the caller is responsible for skipping this call when that flag
is active.
"""
if not invoice.invoice_date:
return
catalog_rate: ExchangeRate | None = (
db.query(ExchangeRate)
.filter(
ExchangeRate.tenant_id == invoice.tenant_id,
ExchangeRate.company_id == invoice.company_id,
func.date(ExchangeRate.date) == invoice.invoice_date,
)
.first()
)
if catalog_rate is None:
errors.add_error(
field="exchange_rate",
message="No existe tipo de cambio registrado para la fecha de la factura",
solution=[
"Capture el tipo de cambio correspondiente a la fecha de la factura "
"en el catálogo de Tipo de Cambio"
],
code="TIPO_DE_CAMBIO",
)
return
invoice_tc = (
Decimal(str(invoice.financials.exchange_rate))
if invoice.financials and invoice.financials.exchange_rate is not None
else None
)
catalog_tc = catalog_rate.value
if invoice_tc is None or catalog_tc is None or invoice_tc != catalog_tc:
errors.add_error(
field="exchange_rate",
message="No esta capturado correctamente el Tipo de Cambio",
solution=[
"Capture o modifique el tipo de cambio que corresponda a la factura "
"en el catálogo de Tipo de Cambio"
],
code="TIPO_DE_CAMBIO",
)

View File

@@ -0,0 +1,944 @@
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional, List
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.fractions.previous_fractions.models import PreviousFraction
from api.v1.modules.a76.general_catalogs.unit_conversions.models import UnitConversion
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.rule_octave.balances.models import OctaveBalance
from api.v1.modules.a76.rule_octave.country.models import CountryRuleOct
from api.v1.modules.a76.rule_octave.fractions.models import FractionRuleOctave
from api.v1.modules.a76.rule_octave.permissions.models import OctavePermission
from core.exceptions import ErrorCollector
# Clarion date 80354 ≈ 2010-05-31 (see previous_fractions/models.py)
_CLARION_DATE_CUTOFF = date(2010, 5, 31)
# RFCs where fraction comparison uses only the first 8 characters
_RFC_FRACTION_8_CHARS = {"CLA940831AZ5", "CTE9801305I8"}
@dataclass
class OctavePermitEntry:
"""
Entrada de permiso de Regla Octava para importación.
Paridad: registro en QueuePermReglaOctDesc / QROImp (Clarion SCAII).
"""
invoice_import: str # FacturaImpo — encabezado de la factura
import_line: int # LineaImpo
part_number: str # NumParte (número de parte o clase)
octave_permit: str # PermisoROctava
ro_line: int # LineaRO
origin_country: str # PaisOrigen
fraction_type: str # TipoFracImpo
import_fraction: str # FraccionImpo
sector: str # Sector
class_code: str # Clase
unit_of_measure: str # UniMed — se resuelve en _revisa_um
quantity: Decimal # Cantidad
value: Decimal # Valor
quantity_used: Decimal # CantUsada (inicia en 0)
value_used: Decimal # ValorUsado (inicia en 0)
# ─────────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────────
def _get_unit_equivalence(
db: Session,
from_unit: str,
to_unit: str,
tenant_id: str,
company_id: str,
) -> tuple[str, Decimal]:
"""
Busca una conversión entre dos unidades de medida.
Paridad: REVEQUIVALENCIA (Clarion SCAII).
Retorna (multi_divide, factor_conv):
- ('M', factor) → multiplicar cantidad por factor
- ('D', factor) → dividir cantidad por factor
- ('', 0) → no existe equivalencia
"""
conv = (
db.query(UnitConversion)
.filter(
UnitConversion.tenant_id == tenant_id,
UnitConversion.company_id == company_id,
UnitConversion.from_unit_code == from_unit,
UnitConversion.to_unit_code == to_unit,
)
.first()
)
if conv and conv.conversion_factor:
return "M", conv.conversion_factor
conv_inv = (
db.query(UnitConversion)
.filter(
UnitConversion.tenant_id == tenant_id,
UnitConversion.company_id == company_id,
UnitConversion.from_unit_code == to_unit,
UnitConversion.to_unit_code == from_unit,
)
.first()
)
if conv_inv and conv_inv.conversion_factor:
return "D", conv_inv.conversion_factor
return "", Decimal(0)
def _previous_fraction_exists(
db: Session,
tenant_id: str,
company_id: str,
previous_fraction: str,
current_fraction: str,
) -> bool:
"""
Verifica si existe un mapeo fracción-anterior → fracción-actual.
Paridad: consulta a SFraccionesAnterioresN (Clarion SCAII).
"""
return (
db.query(PreviousFraction)
.filter(
PreviousFraction.tenant_id == tenant_id,
PreviousFraction.company_id == company_id,
PreviousFraction.previous_fraction == previous_fraction,
PreviousFraction.current_fraction == current_fraction,
)
.count()
) > 0
# ─────────────────────────────────────────────────────────────────────────────
# REVISA_UM_REGLA_OCTAVA
# ─────────────────────────────────────────────────────────────────────────────
def _revisa_um_regla_octava(
db: Session,
line: LineItem,
permission: OctavePermission,
entry: OctavePermitEntry,
company_rfc: str,
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> None:
"""
Revisión de unidades de medida para Regla Octava.
Paridad: REVISA_UM_REGLA_OCTAVA (Clarion SCAII).
- Busca la fracción del permiso RO a nivel de línea.
- Valida que la fracción de la partida coincida con la del permiso.
- Determina la cantidad y UM a registrar en el entry, usando conversión si aplica.
"""
ro_line = line.reference.ro_line if line.reference else 0
if not ro_line:
return
line_fraction = (
db.query(FractionRuleOctave)
.filter(
FractionRuleOctave.tenant_id == tenant_id,
FractionRuleOctave.company_id == company_id,
FractionRuleOctave.permission == line.octave_permit,
FractionRuleOctave.line == ro_line,
)
.first()
)
if line_fraction is None:
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message=(
f"El Permiso de Regla Octava: {line.octave_permit}"
f" con línea: {ro_line} no existe."
),
solution=["Revisar el Permiso de Regla Octava en la Partida de Importación."],
code="RO_FRACTION_NOT_FOUND",
)
return
# ── Validar fracción ─────────────────────────────────────────────────────
line_fraccion = (line.customs.fraction or "") if line.customs else ""
ro_fraccion = line_fraction.fraction or ""
use_8_chars = company_rfc.upper() in _RFC_FRACTION_8_CHARS
frac_line = line_fraccion[:8] if use_8_chars else line_fraccion
frac_ro = ro_fraccion[:8] if use_8_chars else ro_fraccion
if frac_line != frac_ro:
start_date = permission.start_date
if start_date and hasattr(start_date, "date"):
start_date = start_date.date()
allow_previous = start_date is not None and start_date <= _CLARION_DATE_CUTOFF
fraction_ok = False
if allow_previous:
fraction_ok = _previous_fraction_exists(
db,
tenant_id,
company_id,
previous_fraction=ro_fraccion[:8],
current_fraction=line_fraccion[:8],
)
if not fraction_ok:
errors.add_error(
field=f"line[{line.line_number}].customs.fraction",
message=f"La fracción: {frac_line} es diferente a la Fraccion: {frac_ro}",
solution=["Revisar la Partida de Importación y el Permiso de Regla Octava."],
code="RO_FRACTION_MISMATCH",
)
return
# ── Determinar cantidad y valor según unidad de medida ───────────────────
entry.unit_of_measure = line_fraction.unit_of_measure or ""
line_um = line.unit_of_measure_info.code if line.unit_of_measure_info else ""
quantity = Decimal(str(line.quantity.quantity or 0)) if line.quantity else Decimal(0)
value_me = Decimal(str(line.financial.value_usd or 0)) if line.financial else Decimal(0)
if line_fraction.unit_of_measure == line_um:
entry.quantity = quantity
entry.value = value_me
return
multi_divide, factor_conv = _get_unit_equivalence(
db, line_um, line_fraction.unit_of_measure or "", tenant_id, company_id
)
if multi_divide == "M":
entry.quantity = quantity * factor_conv
entry.value = value_me
elif multi_divide == "D":
entry.quantity = quantity / factor_conv if factor_conv else Decimal(0)
entry.value = value_me
else:
if line_fraction.unit_of_measure == "KGS":
net_weight = (
Decimal(str(line.quantity.net_weight or 0)) if line.quantity else Decimal(0)
)
entry.quantity = net_weight
entry.value = value_me
else:
errors.add_error(
field=f"line[{line.line_number}].unit_of_measure",
message=(
f"Regla Octava: No hay equivalencia entre la U.M. del Permiso de RO: "
f"{line_fraction.unit_of_measure} y la U.M. de captura {line_um}."
),
solution=["Capturar su equivalencia en el Catálogo de Equivalencias."],
code="EQUIVALENCIA",
)
# ─────────────────────────────────────────────────────────────────────────────
# LLENA_IMPO_PERMISO_REGLA_OCTAVA
# ─────────────────────────────────────────────────────────────────────────────
def llena_impo_permiso_regla_octava(
db: Session,
invoice: InvoiceHeader,
line: LineItem,
company_rfc: str,
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> Optional[OctavePermitEntry]:
"""
Llena lo que se quiere importar con los permisos de Regla Octava.
Paridad: LLENA_IMPO_PERMISO_REGLA_OCTAVA (Clarion SCAII).
Retorna un OctavePermitEntry listo para agregar a la cola de permisos RO,
o None si el permiso indicado en la partida no existe en el catálogo.
"""
octave_permit_code = line.octave_permit or ""
permission = (
db.query(OctavePermission)
.filter(
OctavePermission.tenant_id == tenant_id,
OctavePermission.company_id == company_id,
OctavePermission.permission == octave_permit_code,
)
.first()
)
if permission is None:
return None
# NumParte: preferir número de parte, caer en código de clase
part_number = ""
if line.part_info and line.part_info.part_number:
part_number = line.part_info.part_number
elif line.class_info and line.class_info.class_code:
part_number = line.class_info.class_code
class_code = (line.class_info.class_code or "") if line.class_info else ""
# Sector: preferir el de la partida, caer en el del permiso
if line.customs and line.customs.sector:
sector = line.customs.sector
else:
sector = permission.sector or ""
ro_line = line.reference.ro_line if line.reference else 0
entry = OctavePermitEntry(
invoice_import=invoice.invoice_number or "",
import_line=line.line_number,
part_number=part_number,
octave_permit=octave_permit_code,
ro_line=ro_line or 0,
origin_country=(line.customs.origin_country or "") if line.customs else "",
fraction_type=(line.customs.fraction_type or "") if line.customs else "",
import_fraction=(line.customs.fraction or "") if line.customs else "",
sector=sector,
class_code=class_code,
unit_of_measure="", # resuelto en _revisa_um_regla_octava
quantity=Decimal(0),
value=Decimal(0),
quantity_used=Decimal(0),
value_used=Decimal(0),
)
_revisa_um_regla_octava(
db=db,
line=line,
permission=permission,
entry=entry,
company_rfc=company_rfc,
tenant_id=tenant_id,
company_id=company_id,
errors=errors,
)
return entry
# ─────────────────────────────────────────────────────────────────────────────
# REVPERMISO_REGLA_OCTAVA
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class OctaveAvailableEntry:
"""
Cupo disponible de una línea de permiso de Regla Octava.
Paridad: registro en QueuePermReglaOctDis / QRODis (Clarion SCAII).
"""
octave_permit: str # Permiso
fraction: str # Fraccion
ro_line: int # LineaRO
country_code: str # ClaveM3 (PaisOrigen)
sector: str # Sector (del permiso)
available_quantity: Decimal # CantidadDisponible = CantidadCupo - CantUsada
quantity_used: Decimal # CantUsada (inicia en 0)
available_value: Decimal # ValorDisponible = ValorCupo - ValorUsada
value_used: Decimal # ValorUsado (inicia en 0)
def _validate_fraction_ro(
db: Session,
line: LineItem,
permission: OctavePermission,
line_fraction: FractionRuleOctave,
company_rfc: str,
tenant_id: str,
company_id: str,
errors: ErrorCollector,
octave_permit_code: str,
) -> bool:
"""
Valida que la fracción de la partida coincida con la del permiso de RO.
Reutilizada por REVPERMISO_REGLA_OCTAVA.
Retorna True si es válida, False si se añadió un error.
"""
line_fraccion = (line.customs.fraction or "") if line.customs else ""
ro_fraccion = line_fraction.fraction or ""
use_8_chars = company_rfc.upper() in _RFC_FRACTION_8_CHARS
frac_line = line_fraccion[:8] if use_8_chars else line_fraccion
frac_ro = ro_fraccion[:8] if use_8_chars else ro_fraccion
if frac_line == frac_ro:
return True
start_date = permission.start_date
if start_date and hasattr(start_date, "date"):
start_date = start_date.date()
allow_previous = start_date is not None and start_date <= _CLARION_DATE_CUTOFF
if allow_previous and _previous_fraction_exists(
db, tenant_id, company_id,
previous_fraction=ro_fraccion[:8],
current_fraction=line_fraccion[:8],
):
return True
errors.add_error(
field=f"line[{line.line_number}].customs.fraction",
message=f"La fracción: {frac_line} es diferente a la Fraccion: {frac_ro}.",
solution=[f"Agregarla al Número de Permiso de Regla Octava: {octave_permit_code}."],
code="PERMISO_REG_8VA",
)
return False
def revpermiso_regla_octava(
db: Session,
invoice: InvoiceHeader,
line: LineItem,
company_rfc: str,
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> Optional[OctaveAvailableEntry]:
"""
Revisión del Permiso de Regla Octava.
Paridad: REVPERMISO_REGLA_OCTAVA (Clarion SCAII).
Valida:
- Existencia del permiso en el catálogo.
- Coherencia de sector (cuando fracción es PROSEC).
- Rango de fechas del permiso vs fecha de la factura.
- Existencia de la línea/fracción dentro del permiso.
- Concordancia de fracción arancelaria.
- Cupo disponible no agotado.
- Existencia del país de origen dentro de la línea del permiso.
Retorna un OctaveAvailableEntry si no hay errores (para ser agregado a la
cola QueuePermReglaOctDis del proceso principal), o None en caso contrario.
"""
octave_permit_code = (line.octave_permit or "").strip()
ro_line = line.reference.ro_line if line.reference else 0
origin_country = (line.customs.origin_country or "") if line.customs else ""
fraction_type = (line.customs.fraction_type or "") if line.customs else ""
line_sector = (line.customs.sector or "") if line.customs else ""
invoice_date = invoice.invoice_date
has_error = False
if ro_line:
if not octave_permit_code:
# Tiene LineaRO pero no hay permiso capturado
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message=(
"No existe capturado un Permiso de Regla Octava "
"y si una Línea de algún permiso de Regla Octava"
),
solution=[
"Capturar un Permiso de Regla Octava para la Linea "
"correspondiente del Permiso de Regla Octava."
],
code="PERMISO_REG_8VA",
)
return None
# ── Buscar el permiso ─────────────────────────────────────────────
permission = (
db.query(OctavePermission)
.filter(
OctavePermission.tenant_id == tenant_id,
OctavePermission.company_id == company_id,
OctavePermission.permission == octave_permit_code,
)
.first()
)
if permission is None:
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message=(
f"La Partida: {line.line_number} tiene el permiso "
f"{octave_permit_code} de Regla Octava y no esta dado de alta."
),
solution=[
f"Dar de alta el permiso de Regla Octava o Borrar "
f"la partida: {line.line_number}."
],
code="PERMISO_REG_8VA",
)
return None
# ── Validar sector cuando la preferencia es PROSEC ────────────────
if fraction_type == "PROSEC" and line_sector != (permission.sector or ""):
errors.add_error(
field=f"line[{line.line_number}].customs.sector",
message=(
f"El sector: {line_sector} de la partida es diferente al sector: "
f"{permission.sector} del permiso de Regla Octava."
),
solution=[
f"Cambiar de sector en la Partida o en el Permiso: {octave_permit_code}."
],
code="PERMISO_REG_8VA",
)
has_error = True
# ── Validar rango de fechas del permiso ───────────────────────────
perm_start = permission.start_date
perm_end = permission.end_date
if perm_start and hasattr(perm_start, "date"):
perm_start = perm_start.date()
if perm_end and hasattr(perm_end, "date"):
perm_end = perm_end.date()
inv_date = invoice_date
if inv_date and hasattr(inv_date, "date"):
inv_date = inv_date.date()
if inv_date and perm_start and perm_end:
if inv_date < perm_start or inv_date > perm_end:
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message=(
f"La fecha de la factura no corresponde al rango de fechas "
f"del Permiso: {octave_permit_code} de Regla Octava."
),
solution=[
"Modificar el Rango de fechas del Permiso de Regla Octava "
"o seleccionar uno que en su rango entre la fecha de la Factura."
],
code="PERMISO_REG_8VA",
)
has_error = True
# ── Buscar línea/fracción dentro del permiso ──────────────────────
line_fraction = (
db.query(FractionRuleOctave)
.filter(
FractionRuleOctave.tenant_id == tenant_id,
FractionRuleOctave.company_id == company_id,
FractionRuleOctave.permission == octave_permit_code,
FractionRuleOctave.line == ro_line,
)
.first()
)
if line_fraction is None:
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message=(
f"No existe la Línea: {ro_line} amparada en el permiso "
f"seleccionado {octave_permit_code} de Regla Octava."
),
solution=["Verificar los Permisos de Regla Octava"],
code="PERMISO_REG_8VA",
)
return None
# ── Validar fracción arancelaria ──────────────────────────────────
if not _validate_fraction_ro(
db, line, permission, line_fraction, company_rfc,
tenant_id, company_id, errors, octave_permit_code,
):
has_error = True
# ── Validar cupo disponible ───────────────────────────────────────
quota_qty = line_fraction.quota_quantity or Decimal(0)
qty_used = line_fraction.quantity_used or Decimal(0)
available_qty = quota_qty - qty_used
if available_qty == 0:
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message=(
"La Cantidad Cupo ya esta agotada, no se puede importar "
"más para este permiso de Regla Octava."
),
solution=["Tramitar otro permiso o seleccione otro Permiso."],
code="PERMISO_REG_8VA",
)
has_error = True
# ── Validar país de origen dentro de la línea del permiso ─────────
country = (
db.query(CountryRuleOct)
.filter(
CountryRuleOct.tenant_id == tenant_id,
CountryRuleOct.company_id == company_id,
CountryRuleOct.permission == octave_permit_code,
CountryRuleOct.line == ro_line,
CountryRuleOct.fraction == (line_fraction.fraction or ""),
CountryRuleOct.country_code == origin_country,
)
.first()
)
if country is None:
errors.add_error(
field=f"line[{line.line_number}].customs.origin_country",
message=(
f"No existe la Línea: {ro_line} con país: {origin_country} "
f"amparada en el permiso seleccionado"
),
solution=[
f"Agregar el País: {origin_country} al Número de Permiso "
f"de Regla Octava: {octave_permit_code}."
],
code="PERMISO_REG_8VA",
)
has_error = True
if has_error:
return None
# ── Construir entry de cupo disponible ────────────────────────────
quota_val = line_fraction.quota_value or Decimal(0)
val_used = line_fraction.value_used or Decimal(0)
return OctaveAvailableEntry(
octave_permit=octave_permit_code,
fraction=line_fraction.fraction or "",
ro_line=ro_line,
country_code=origin_country,
sector=permission.sector or "",
available_quantity=available_qty,
quantity_used=Decimal(0),
available_value=quota_val - val_used,
value_used=Decimal(0),
)
else:
# ro_line == 0: si hay permiso capturado es un error
if octave_permit_code:
errors.add_error(
field=f"line[{line.line_number}].octave_permit",
message=(
f"No existe una Línea capturada para esta partida de importación "
f"y se tiene el Permiso de Regla Octava: {octave_permit_code}."
),
solution=["Capturar una Linea de Permiso de Regla Octava."],
code="PERMISO_REG_8VA",
)
return None
# ─────────────────────────────────────────────────────────────────────────────
# VALIDA_IMP_REGLA_OCTAVA
# ─────────────────────────────────────────────────────────────────────────────
def valida_imp_regla_octava(
db: Session,
desc_list: List[OctavePermitEntry],
dis_dict: dict,
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> None:
"""
Valida lo que se quiere importar contra los cupos disponibles de Regla Octava.
Paridad: VALIDA_IMP_REGLA_OCTAVA (Clarion SCAII).
Fase 1 — Asignación de saldos:
Itera QueuePermReglaOctDesc (qué se quiere importar) contra
QueuePermReglaOctDis (cupo disponible) y distribuye la cantidad/valor
disponible entre las partidas, acumulando lo "usado" en cada lista.
Fase 2 — Validación final:
- Cantidad == 0: error (sin conversión de U.M.).
- Cantidad CantUsada != 0: error (cupo insuficiente).
Parámetros:
desc_list Lista de OctavePermitEntry construida por llena_impo_permiso_regla_octava.
dis_dict Diccionario {(permiso, ro_line, country_code): OctaveAvailableEntry}
construido por revpermiso_regla_octava.
"""
# ── Fase 1: Balancear saldos ──────────────────────────────────────────────
desc_sorted = sorted(
desc_list,
key=lambda e: (e.octave_permit, e.ro_line, e.origin_country),
)
for desc in desc_sorted:
key = (desc.octave_permit, desc.ro_line, desc.origin_country)
dis = dis_dict.get(key)
if dis is None:
continue
# BREAK: la partida ya está completamente consumida
if (
desc.quantity - desc.quantity_used == 0
or desc.value - desc.value_used == 0
):
continue
# CYCLE: el cupo disponible ya está agotado
if (
dis.available_quantity - dis.quantity_used == 0
or dis.available_value - dis.value_used == 0
):
continue
# ── Cantidad ──────────────────────────────────────────────────────
qty_needed = desc.quantity - desc.quantity_used
qty_avail = dis.available_quantity - dis.quantity_used
if qty_needed <= qty_avail:
desc.quantity_used += qty_needed
dis.quantity_used += qty_needed
else:
desc.quantity_used += qty_avail
dis.quantity_used += qty_avail
# ── Valor ─────────────────────────────────────────────────────────
val_needed = desc.value - desc.value_used
val_avail = dis.available_value - dis.value_used
if val_needed <= val_avail:
# Asignación directa al total (no acumulativa) — paridad Clarion
desc.value_used = desc.value
dis.value_used += val_needed
else:
# Asignación directa al disponible — paridad Clarion
desc.value_used = dis.available_value
dis.value_used += val_avail
# ── Fase 2: Validación final ──────────────────────────────────────────────
for desc in desc_list:
line_fraction = (
db.query(FractionRuleOctave)
.filter(
FractionRuleOctave.tenant_id == tenant_id,
FractionRuleOctave.company_id == company_id,
FractionRuleOctave.permission == desc.octave_permit,
FractionRuleOctave.line == desc.ro_line,
)
.first()
)
# line_fraction reservado para validación de costo unitario (comentada en Clarion)
_ = line_fraction
if desc.quantity == 0:
errors.add_error(
field=f"line[{desc.import_line}].octave_permit",
message=(
"La Cantidad a Descargar del Permiso de Regla Octava es Cero "
"debido a que No hay conversión entre la U.M. de la Partida "
"vs la del Permiso"
),
solution=[
"Revisar las U.M. de la Partida de Importación "
"y del Permiso de Regla Octava."
],
code="PERMISO_REG_8VA",
)
elif desc.quantity - desc.quantity_used != 0:
errors.add_error(
field=f"line[{desc.import_line}].octave_permit",
message=(
"No se puede importar más de este material "
"bajo el amparo de Regla Octava."
),
solution=[
"Revisar la cantidad a Importar de la línea "
"o solicitar otro permiso."
],
code="PERMISO_REG_8VA",
)
# ─────────────────────────────────────────────────────────────────────────────
# LLENASALDOS_REGLA_OCTAVA
# ─────────────────────────────────────────────────────────────────────────────
def _llena_fraccion_regla_oct(sector: str) -> str:
"""
Devuelve la fracción arancelaria de Regla Octava según el sector.
Paridad: LLENA_FRACCION_REGLA_OCT (Clarion SCAII).
El orden de comparación es deliberado: los prefijos más largos (XIX, XXIII)
se evalúan antes que los más cortos (XX, II) para evitar falsos positivos,
exactamente como advierte el comentario original del Clarion.
"""
s = (sector or "").strip()
if s == "I": return "98020001"
if s[:2] == "II": return "98020002"
if s == "III": return "98020003"
if s == "IV": return "98020004"
if s == "V": return "98020005"
if s == "VI": return "98020006"
if s == "VII": return "98020007"
if s == "VIII": return "98020008"
if s == "IX": return "98020009"
if s == "X": return "98020010"
if s == "XI": return "98020011"
if s == "XII": return "98020012"
if s == "XIII": return "98020013"
if s == "XIV": return "98020014"
if s == "XV": return "98020015"
if s == "XVI": return "98020016"
if s == "XVII": return "98020017"
if s == "XVIII": return "98020018"
if s[:3] == "XIX": return "98020019" # antes de XX
if s == "XXI": return "98020021"
if s == "XXII": return "98020022"
if s[:5] == "XXIII": return "98020023" # antes de XX
if s == "XXIV": return "98020024"
if s == "XXV": return "98020025"
if s[:2] == "XX": return "98020020" # al último — paridad comentario Clarion
return ""
def llenasaldos_regla_octava(
db: Session,
entry: "OctavePermitEntry",
unit_cost_me: Decimal,
tenant_id: str,
company_id: str,
sql_errors: list,
consecutive_ref: list,
) -> None:
"""
Llena / actualiza el saldo de Regla Octava en SSaldosReglaOctava.
Paridad: LLENASALDOS_REGLA_OCTAVA (Clarion SCAII).
- Si el registro no existe (ERRORCODE = 35 en Clarion → no encontrado):
inserta uno nuevo con CantExitencia = quantity_used y
ValorME = quantity_used * unit_cost_me.
- Si ya existe: acumula CantExitencia y ValorME.
consecutive_ref es una lista de un elemento [int] que actúa como
contador mutable compartido con descuenta_cupo_r_octava.
"""
fraction = _llena_fraccion_regla_oct(entry.sector)
valor_me = entry.quantity_used * unit_cost_me
existing = (
db.query(OctaveBalance)
.filter(
OctaveBalance.tenant_id == tenant_id,
OctaveBalance.company_id == company_id,
OctaveBalance.invoice_import == entry.invoice_import,
OctaveBalance.part_number == entry.part_number,
OctaveBalance.origin_country == entry.origin_country,
OctaveBalance.fraction_type == entry.fraction_type,
OctaveBalance.sector == entry.sector,
OctaveBalance.octave_permit == entry.octave_permit,
OctaveBalance.origin == "TEM",
OctaveBalance.system == "SCAF",
OctaveBalance.line == entry.ro_line,
)
.first()
)
try:
if existing is None:
new_balance = OctaveBalance(
tenant_id=tenant_id,
company_id=company_id,
invoice_import=entry.invoice_import,
part_number=entry.part_number,
origin_country=entry.origin_country,
fraction_type=entry.fraction_type,
sector=entry.sector,
octave_permit=entry.octave_permit,
origin="TEM",
system="SCAF",
line=entry.ro_line,
ro_fraction=fraction,
class_code=entry.class_code,
import_fraction=entry.import_fraction,
unit_of_measure=entry.unit_of_measure,
quantity_stock=entry.quantity_used,
value_me=valor_me,
)
db.add(new_balance)
db.flush([new_balance])
else:
existing.quantity_stock = (existing.quantity_stock or Decimal(0)) + entry.quantity_used
existing.value_me = (existing.value_me or Decimal(0)) + valor_me
db.flush([existing])
except Exception as exc:
consecutive_ref[0] += 1
action = "Agregar" if existing is None else "Actualizar"
sql_errors.append({
"consecutive": consecutive_ref[0],
"error": (
f"Error al {action} en (SSaldosReglaOctava) {exc}"
),
})
# ─────────────────────────────────────────────────────────────────────────────
# DESCUENTA_CUPO_R_OCTAVA
# ─────────────────────────────────────────────────────────────────────────────
def descuenta_cupo_r_octava(
db: Session,
desc_list: List[OctavePermitEntry],
tenant_id: str,
company_id: str,
sql_errors: list,
) -> None:
"""
Descuenta los cupos de la Regla Octava en GFracROctava.
Paridad: DESCUENTA_CUPO_R_OCTAVA (Clarion SCAII).
Por cada entrada en desc_list que tenga cantidad y valor usados distintos
de cero:
1. Llama a llenasaldos_regla_octava para registrar/actualizar SSaldosReglaOctava.
2. Actualiza FractionRuleOctave acumulando:
- quantity_used += entry.quantity_used
- value_used += entry.quantity_used * unit_cost_me
Los errores de actualización se acumulan en sql_errors como dicts con
las claves 'consecutive' y 'error'.
"""
consecutive_ref = [0]
for entry in desc_list:
if entry.quantity_used == 0 or entry.value_used == 0:
continue
fra_oct = (
db.query(FractionRuleOctave)
.filter(
FractionRuleOctave.tenant_id == tenant_id,
FractionRuleOctave.company_id == company_id,
FractionRuleOctave.permission == entry.octave_permit,
FractionRuleOctave.line == entry.ro_line,
)
.first()
)
if fra_oct is None:
continue
unit_cost_me = fra_oct.unit_cost_me or Decimal(0)
llenasaldos_regla_octava(
db=db,
entry=entry,
unit_cost_me=unit_cost_me,
tenant_id=tenant_id,
company_id=company_id,
sql_errors=sql_errors,
consecutive_ref=consecutive_ref,
)
fra_oct.quantity_used = (fra_oct.quantity_used or Decimal(0)) + entry.quantity_used
fra_oct.value_used = (fra_oct.value_used or Decimal(0)) + entry.quantity_used * unit_cost_me
try:
db.flush([fra_oct])
except Exception as exc:
consecutive_ref[0] += 1
sql_errors.append({
"consecutive": consecutive_ref[0],
"error": (
f"Error al regresar el Cupo en (Permiso de Regla Octava) {exc}"
),
})

View File

@@ -0,0 +1,93 @@
from typing import List
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.items.series.models import Serie
from core.exceptions import ErrorCollector
# Clarion: SisGen:CantvsCantSeries
# TODO: leer desde configuración del tenant cuando SisGen esté disponible
_SISIMP_CANT_VS_CANT_SERIES: int = 0 # 0 = desactivado
# RFCs donde la validación de cantidad vs series únicamente aplica a PZA (paridad Clarion)
_RFC_EXCEPCION_PZA = {
"IMS030409FZ0",
"TOP140430PB6",
"AMA7504258K2",
"BZG111091T9",
}
def review_series(
db: Session,
line: LineItem,
company_rfc: int,
errors: ErrorCollector,
) -> None:
"""
Revisa la cantidad de series vs la cantidad a importar, por partida.
Paridad: REVISA_CANT_SERIES (Clarion SCAII).
Reglas:
- Si la partida tiene LlevaSerie / has_serial activado:
1. Si no existen registros de series → error SERIES_VACIAS.
2. Si SisGen:CantvsCantSeries = 1:
- Para RFC excepcionales: solo valida la coincidencia si UnidadMedida = 'PZA'.
- Para el resto: valida siempre que count(series) != cantidad.
"""
# Resolver RFC de la empresa una sola vez
# Solo aplica a partidas que llevan serie
if not (line.description and line.description.has_serial):
return
series_count = (
db.query(Serie)
.filter(Serie.line_item_id == line.id)
.count()
)
if series_count == 0:
errors.add_error(
field=f"line[{line.line_number}].series",
message="La opción de contiene series esta activada y no existen registros de Series.",
solution=[
"Desactivar la opción de Lleva series o registrar las series a esta partida."
],
code="SERIES_VACIAS",
)
return
# GNiv:CantSerievsCant = 0 → el bloque series > cantidad estaba comentado en Clarion original
# TODO: leer SisGen:CantvsCantSeries desde la configuración del tenant
if _SISIMP_CANT_VS_CANT_SERIES != 1:
return
qty = line.quantity.quantity if line.quantity else None
if qty is None:
return
# Determinar si se debe validar series
validate_series = (
series_count != qty and
(company_rfc not in _RFC_EXCEPCION_PZA or line.unit_of_measure == "PZA")
)
if validate_series:
errors.add_error(
field=f"line[{line.line_number}].series",
message="La Cantidad de Series No Coincide con la Cantidad de la Partida.",
solution=[f"Nivelar las series de la Partida {line.line_number}."],
code="SERIES_VS_CANT",
)

View File

@@ -0,0 +1,129 @@
from decimal import Decimal
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasureCustoms
from api.v1.modules.a76.items.models import LineItem
from core.exceptions import ErrorCollector
from .review_rule_octave import _get_unit_equivalence
def revisa_uma(
db: Session,
line: LineItem,
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> None:
"""
Revisa la unidad de medida americana (UMA) de aduana para una partida.
Paridad: REVISA_UMA (Clarion SCAII).
Flujo:
1. Verifica que la UM de la partida exista en el catálogo GUnimedida.
2. Verifica que esa UM tenga asignada una Clave_AMex (UM aduana mexicana).
3. Verifica que la Clave_AMex exista en el catálogo GUMAduana.
4. Verifica que el registro de aduana tenga una UnidadSCAII (a76_unit_code).
5. Calcula la cantidad UMA:
- Si la UM de la partida coincide con UnidadSCAII → cantidad directa.
- Si no → busca conversión en el catálogo de equivalencias.
6. Escribe CantImpoUMA y ClaveUMA de regreso en la partida.
"""
uom = line.unit_of_measure_info
line_um_code = uom.code if uom else ""
# ── 1. Verificar existencia de la UM en el catálogo ──────────────────────
if uom is None:
errors.add_error(
field=f"line[{line.line_number}].unit_of_measure",
message=(
f"No existe la unidad de medida {line_um_code}"
" en el catálogo de Unidades de Medida."
),
solution=["Capturarla en el catálogo de Unidades de Medida."],
code="UNIMEDIDA",
)
return
# ── 2. Verificar que tenga Clave_AMex asignada ───────────────────────────
customs_code = uom.customs_code or ""
if not customs_code:
errors.add_error(
field=f"line[{line.line_number}].unit_of_measure",
message=(
f"No existe la unidad de medida de la aduana para la unidad de medida "
f"{line_um_code} en el catálogo de Unidades de Medida."
),
solution=["Asignar la U.M.A en el catálogo de Unidades de Medida."],
code="UNIMEDIDA",
)
return
# ── 3. Buscar la UM en el catálogo de Aduana Mex (GUMAduana) ─────────────
customs_uom = (
db.query(UnitOfMeasureCustoms)
.filter(UnitOfMeasureCustoms.code == customs_code)
.first()
)
if customs_uom is None:
errors.add_error(
field=f"line[{line.line_number}].unit_of_measure",
message=(
f"No existe la unidad de medida {customs_code}"
" en el catálogo de Unidades de Medida de la Aduana Mex."
),
solution=["Actualizar sus Catálogos Fijos o llamar a su proveedor."],
code="UNIMEDIDA",
)
return
# ── 4. Verificar que tenga UnidadSCAII asignada ──────────────────────────
scaii_unit = customs_uom.a76_unit_code or ""
if not scaii_unit:
errors.add_error(
field=f"line[{line.line_number}].unit_of_measure",
message=(
f"No tiene asignada la U.M. Equivalente SCAII la unidad de medida "
f"{customs_code} en el catálogo de Unidades de Medida de la Aduana Mex."
),
solution=["Actualizar sus Catálogos Fijos o llamar a su proveedor."],
code="UNIMEDIDA",
)
return
# ── 5. Calcular cantidad UMA ──────────────────────────────────────────────
line_qty = Decimal(str(line.quantity.quantity or 0)) if line.quantity else Decimal(0)
cant_uma = Decimal(0)
if line_um_code == scaii_unit:
cant_uma = line_qty
else:
multi_divide, factor_conv = _get_unit_equivalence(
db, line_um_code, scaii_unit, tenant_id, company_id
)
if multi_divide == "M":
cant_uma = line_qty * factor_conv
elif multi_divide == "D":
cant_uma = line_qty / factor_conv if factor_conv else Decimal(0)
else:
errors.add_error(
field=f"line[{line.line_number}].unit_of_measure",
message=(
f"No hay equivalencia entre la U.M. Partida: {line_um_code}"
f" y la U.M. Aduana {scaii_unit}."
),
solution=[
"Capturar su equivalencia en el Catálogo de Conversiones "
"o configurar la U.M.Aduana correcta en la U.M. Comercial."
],
code="EQUIVALENCIA",
)
return
# ── 6. Escribir resultados en la partida ──────────────────────────────────
if line.quantity:
line.quantity.quantity_uma = cant_uma
line.uma_key = customs_code

View File

@@ -0,0 +1,102 @@
from typing import List
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.models import LineItem
from core.exceptions import ErrorCollector
def review_weights_kgs(
db: Session,
lines: List[LineItem],
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> None:
"""
Validates that, for lines whose unit of measure is KGS, the net weight
equals the import quantity.
Ported from legacy REVISA_CANT_vs_PESONETO_KGS routine.
"""
# Resolve the id of the KGS unit of measure for this tenant/company
kgs_uom = (
db.query(UnitOfMeasure)
.filter(
UnitOfMeasure.code == "KGS",
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id,
)
.first()
)
if kgs_uom is None:
# No KGS unit defined for this tenant — nothing to validate
return
for line in lines:
if line.unit_of_measure != kgs_uom.id:
continue
if line.quantity is None:
continue
net_weight = line.quantity.net_weight
qty = line.quantity.quantity
if net_weight is None or qty is None or net_weight != qty:
errors.add_error(
field=f"line[{line.line_number}].net_weight",
message=(
f"Existe diferencia entre la cantidad y el Peso Neto, "
f"la cantidad es de {qty} KGS y el Peso Neto es de {net_weight} KGS."
),
solution=["Igualar el Peso Neto con la cantidad a importar."],
code="PESO_NETO_KGS",
)
def review_weights_lbs(
db: Session,
lines: List[LineItem],
tenant_id: str,
company_id: str,
errors: ErrorCollector,
) -> None:
"""
Validates that, for lines whose unit of measure is LBS, the net weight
equals the import quantity.
Ported from legacy REVISA_CANT_vs_PESONETO_LBS routine.
"""
lbs_uom = (
db.query(UnitOfMeasure)
.filter(
UnitOfMeasure.code == "LBS",
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id,
)
.first()
)
if lbs_uom is None:
return
for line in lines:
if line.unit_of_measure != lbs_uom.id:
continue
if line.quantity is None:
continue
net_weight = line.quantity.net_weight
qty = line.quantity.quantity
if net_weight is None or qty is None or net_weight != qty:
errors.add_error(
field=f"line[{line.line_number}].net_weight",
message=(
f"Existe diferencia entre la cantidad y el Peso Neto, "
f"la cantidad es de {qty} LBS y el Peso Neto es de {net_weight} LBS."
),
solution=["Igualar el Peso Neto con la cantidad a importar."],
code="PESO_NETO_LBS",
)

View File

@@ -0,0 +1,122 @@
from celery import Task
from core.celery_app import celery_app
from core.database import CoreSessionLocal
from core.exceptions import ErrorCollector, ValidationException
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.general_catalogs.company.models import Company
from .pre_validators import pre_validators
from .sub_process.review_classes import review_classes
from .sub_process.review_exchange_rate import review_exchange_rate
from .sub_process.review_weights import review_weights_kgs, review_weights_lbs
from .sub_process.review_rule_octave import valida_imp_regla_octava, descuenta_cupo_r_octava
from .sub_process.assing_values import assign_values_lines, assign_values_invoice
from .main_process import _validate_sisimp_limits, _update_invoice_totals, _validate_lines
def _progress(task: Task, current: int, status: str) -> None:
task.update_state(state="PROGRESS", meta={"current": current, "status": status})
@celery_app.task(bind=True, name="process_invoice_task")
def process_invoice_task(self: Task, invoice_id: int, tenant_id: str, company_id: str) -> dict:
"""
Procesa una factura de importación ejecutando todas las validaciones y
actualizaciones del proceso principal (main_process) con reporte de progreso.
"""
db = CoreSessionLocal()
try:
# ── Paso 1: Cargar factura ────────────────────────────────────────────
_progress(self, 5, "Cargando factura...")
invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id)
if invoice is None:
return {
"status": "error",
"message": f"Factura con id {invoice_id} no encontrada.",
"errors": [],
}
errors = ErrorCollector()
# ── Paso 2: Pre-validaciones ──────────────────────────────────────────
_progress(self, 10, "Validando datos de la factura...")
lines = pre_validators(db, invoice, tenant_id, company_id, errors)
if not lines:
errors.add_error(
field="line_items",
message="La factura debe contener al menos una partida para ser importada",
solution=["Agregue partidas a la factura antes de intentar importarla"],
code="NO_LINE_ITEMS",
)
errors.raise_if_errors()
# ── Paso 3: Revisión clases, tipo de cambio y pesos ──────────────────
_progress(self, 30, "Revisando clases y tipo de cambio...")
review_classes(db, invoice, lines, tenant_id, company_id, errors)
review_exchange_rate(db, invoice, errors)
if invoice.logistics and invoice.logistics.weight_type == "kgs":
review_weights_kgs(db, lines, tenant_id, company_id, errors)
elif invoice.logistics and invoice.logistics.weight_type == "lbs":
review_weights_lbs(db, lines, tenant_id, company_id, errors)
# ── Paso 4: Asignación de valores ─────────────────────────────────────
_progress(self, 50, "Calculando valores por partida...")
assign_values_lines(invoice, lines)
assign_values_invoice(invoice, lines)
# ── Paso 5: Validaciones por partida ──────────────────────────────────
_progress(self, 70, "Validando partidas...")
octave_desc, octave_available = _validate_lines(
db, invoice, lines, tenant_id, company_id, errors
)
# ── Paso 6: Regla Octava y límites SisImp ─────────────────────────────
_progress(self, 85, "Validando cupos de Regla Octava...")
company: Company | None = db.get(Company, invoice.company_id)
if company and company.prosec and octave_desc:
valida_imp_regla_octava(
db=db,
desc_list=octave_desc,
dis_dict=octave_available,
tenant_id=tenant_id,
company_id=company_id,
errors=errors,
)
_validate_sisimp_limits(invoice, errors)
errors.raise_if_errors()
# ── Paso 7: Descuento de cupos y actualización de totales ─────────────
_progress(self, 95, "Actualizando totales...")
sql_errors: list = []
if octave_desc:
descuenta_cupo_r_octava(
db=db,
desc_list=octave_desc,
tenant_id=tenant_id,
company_id=company_id,
sql_errors=sql_errors,
)
_update_invoice_totals(invoice)
db.flush()
db.commit()
return {
"status": "success",
"invoice_id": invoice_id,
"sql_errors": sql_errors,
}
except ValidationException as exc:
db.rollback()
return {
"status": "validation_error",
"message": exc.message,
"errors": exc.errors,
}
except Exception as exc:
db.rollback()
raise exc
finally:
db.close()

View File

@@ -63,6 +63,11 @@ class TransportType(str, Enum):
AIRPLANE = "airplane"
GONDOLA = "gondola"
FLATBED = "flatbed"
class InvoiceStatus(str, Enum):
PENDING = "pending"
PROCESSED = "processed"
REVERSED = "reversed"
# --- 1. Invoice Header (invoice_header) ---
@@ -108,17 +113,17 @@ class InvoiceHeader(Base, TenantScopedMixin, TimestampMixin):
emission_date: Mapped[Optional[datetime]] = mapped_column(Date) # FECHAEMISION
# Status & Control
is_updated: Mapped[bool] = mapped_column(Boolean) # ESTATUS
is_updated_rec: Mapped[Optional[bool]] = mapped_column(
Boolean
status: Mapped[InvoiceStatus] = mapped_column(String(10)) # ESTATUS
status_rec: Mapped[Optional[InvoiceStatus]] = mapped_column(
String(10)
) # ESTATUSREC / Estatus de recepción
is_updated_rep: Mapped[Optional[bool]] = mapped_column(
Boolean
status_rep: Mapped[Optional[InvoiceStatus]] = mapped_column(
String(10)
) # ESTATUSREP / Estatus de reporte
updated_date: Mapped[Optional[datetime]] = mapped_column(
processed_date: Mapped[Optional[datetime]] = mapped_column(
TIMESTAMP(timezone=False)
) # FECHAACTUALIZACION / FECHAACTUAL
who_updated: Mapped[Optional[str]] = mapped_column(
who_processed: Mapped[Optional[str]] = mapped_column(
String(20)
) # USUARIOACT / Quien actualizó
capture_user: Mapped[Optional[str]] = mapped_column(
@@ -526,6 +531,9 @@ class InvoiceFinancials(Base, TenantScopedMixin, TimestampMixin):
total_quantity: Mapped[Optional[float]] = mapped_column(
Numeric(19, 8)
) # CANTEXPO/CANTIMPO / Cantidad total
total_packages: Mapped[Optional[int]] = mapped_column(
Integer
) # CANTBULTOS / Cantidad de bultos
gross_weight: Mapped[Optional[float]] = mapped_column(
Numeric(19, 8)
) # PESOBRUTO / Peso bruto

View File

@@ -275,6 +275,7 @@ class InvoiceFinancialsBase(BaseModel):
)
seal_value_2500: Optional[bool] = Field(None, description="Seal value 2500")
total_quantity: Optional[Decimal] = Field(0.00, description="Total quantity")
total_packages: Optional[int] = Field(0, description="Total packages")
gross_weight: Optional[Decimal] = Field(0.00, description="Gross weight")
net_weight: Optional[Decimal] = Field(0.00, description="Net weight")
bundle_count: Optional[int] = Field(0, description="Bundle count")

View File

@@ -10,8 +10,9 @@ from .customs_brokers.routes import router as customs_broker_router
# Importar routers de módulos
from .general_catalogs.router import router as general_catalogs_router
from .invoices.routes import router as invoices_router
from .invoices.imports.process.routes import router as invoice_process_router
from .items.routes import router as items_router
from .classes import router as classes_router
from .classes.routes import router as classes_router
from .clients_and_providers import router as client_and_provider_router
from .layouts_csv.facturas.routes import router as imports_router
@@ -20,11 +21,10 @@ from .layouts_csv.cambio_regimen_regularizacion.routes import router as cambio_r
from .csv_templates.routes import router as csv_templates_router
from .invoice_settings.routes import router as invoice_settings_router
from .item_presets.routes import router as item_presets_router
from .general_catalogs.company import router as company_router
from .country_rule_oct.routes import router as country_rule_oct_router
from .transportation.drivers.routes import router as drivers_router
from .doc_types_dig.routes import router as doc_types_dig_router
from .fraction_rule_octave.routes import router as fraction_rule_octave_router
from .rule_octave.fractions.routes import router as fraction_rule_octave_router
from .rule_octave.country.routes import router as country_rule_oct_router
from .parts import router as parts_router
from .boms import router as boms_router
from .pedmientos.router import router as pedimentos_router
@@ -42,14 +42,15 @@ from .reports.exportacion.aviso_consolidado.routes import router as aviso_consol
from .reports.movements.invoices.routes import router as movement_invoices_router
from .reports.movements.saldos.routes import router as movement_saldos_router
from .reports.exportacion.descargo.routes import router as discharge_reports_router
from .manifests.manifest.routes import router as manifests_router
from .manifests.driver.routes import router as manifest_drivers_router
from .manifests.manifiesto_anexo.routes import router as manifest_anexos_router
from .reports.exportacion.transmission.MAINX30.routes import router as transmission_router
from .reports.importacion.transmission.temporal.MAINX30.routes import router as transmission_temporal_router
from .reports.importacion.transmission.definitive.MAINX30.routes import router as transmission_definitive_router
from .reports.importacion.winsaai.router import router as winsaai_router
from .manifests.manifest.routes import router as manifests_router
from .manifests.driver.routes import router as manifest_drivers_router
from .manifests.manifiesto_anexo.routes import router as manifest_anexos_router
# Router principal
router = APIRouter()
@@ -57,6 +58,7 @@ router = APIRouter()
# Registrar módulos
router.include_router(general_catalogs_router, prefix="/a76", tags=["a76 / general_catalogs"])
router.include_router(invoices_router, prefix="/a76", tags=["a76 / invoices"])
router.include_router(invoice_process_router, prefix="/a76", tags=["a76 / invoices"])
router.include_router(items_router, prefix="/a76", tags=["a76 / items"])
router.include_router(imports_router, prefix="/a76/imports", tags=["a76 / imports"])
router.include_router(exportacion_imports_router, prefix="/a76/imports/exportacion", tags=["a76 / imports / exportacion"])

View File

@@ -0,0 +1,56 @@
from decimal import Decimal
from typing import Optional
from sqlalchemy import Integer, Numeric, PrimaryKeyConstraint, String
from sqlalchemy.orm import Mapped, mapped_column
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class OctaveBalance(Base, TenantScopedMixin, TimestampMixin):
"""
Saldos de Regla Octava por factura/partida.
Paridad: SSaldosReglaOctava (Clarion SCAII).
La PK es compuesta natural, igual que ROctSal_PKFaParPaiTiSeROProSisLin.
Se agregan tenant_id y company_id al inicio para el scope multi-tenant.
"""
__tablename__ = "octave_balance"
__table_args__ = (
PrimaryKeyConstraint(
"tenant_id",
"company_id",
"invoice_import",
"part_number",
"origin_country",
"fraction_type",
"sector",
"octave_permit",
"origin",
"system",
"line",
name="pk_octave_balance",
),
{"schema": "a76"},
)
# ── PK compuesta ──────────────────────────────────────────────────────────
invoice_import: Mapped[str] = mapped_column(String(15)) # FACTURAIMPO
part_number: Mapped[str] = mapped_column(String(70)) # NUMPARTE
origin_country: Mapped[str] = mapped_column(String(3)) # PAISORIGEN
fraction_type: Mapped[str] = mapped_column(String(7)) # TIPOFRACIMPO
sector: Mapped[str] = mapped_column(String(8)) # SECTOR
octave_permit: Mapped[str] = mapped_column(String(20)) # PERMISOROCTAVA
origin: Mapped[str] = mapped_column(String(3)) # PROCEDENCIA ('TEM')
system: Mapped[str] = mapped_column(String(5)) # SISTEMA ('fixed_asset | inventory')
line: Mapped[int] = mapped_column(Integer) # LINEA
# ── Datos ─────────────────────────────────────────────────────────────────
quantity_stock: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8)) # CANTEXITENCIA
unit_of_measure: Mapped[Optional[str]] = mapped_column(String(5)) # UNIMED
class_code: Mapped[Optional[str]] = mapped_column(String(8)) # CLASE
import_fraction: Mapped[Optional[str]] = mapped_column(String(10)) # FRACCIONIMPO
ro_fraction: Mapped[Optional[str]] = mapped_column(String(10)) # FRACCIONROCTAVA
value_me: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # VALORME

View File

@@ -0,0 +1,40 @@
from decimal import Decimal
from typing import Optional
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from sqlalchemy import Integer, Numeric, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
class FractionRuleOctave(Base, TenantScopedMixin, TimestampMixin):
"""
Fracciones por permiso de Regla Octava.
Paridad: GFracROctava (Clarion SCAII).
PK natural: (permission, line, fraction) — se usa junto a tenant/company.
"""
__tablename__ = "fraction_rule_octave"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"company_id",
"permission",
"line",
"fraction",
name="uq_fraction_rule_octave_permission_line_fraction",
),
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
permission: Mapped[str] = mapped_column(String(20)) # PERMISO
line: Mapped[int] = mapped_column(Integer) # LINEA
fraction: Mapped[str] = mapped_column(String(10)) # FRACCION
quota_quantity: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8)) # CANTIDADCUPO
quantity_used: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # CANTUSADA
quota_value: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # VALORCUPO
value_used: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # VALORUSADA
unit_cost_me: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # COSTOUNITARIOME
unit_of_measure: Mapped[Optional[str]] = mapped_column(String(5)) # UNIMED

View File

@@ -0,0 +1,32 @@
from typing import Optional
from sqlalchemy import DateTime, Integer, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class OctavePermission(Base, TenantScopedMixin, TimestampMixin):
"""
Permisos para la Regla Octava.
Paridad: GPermisoReglaOct (Clarion SCAII).
"""
__tablename__ = "permission_rule_octave"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"company_id",
"permission",
name="uq_permissions_rule_octave_permission",
),
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
permission: Mapped[str] = mapped_column(String(20), nullable=False) # PERMISO
start_date: Mapped[Optional[DateTime]] = mapped_column(DateTime()) # FECHAINICIO
end_date: Mapped[Optional[DateTime]] = mapped_column(DateTime()) # FECHAFINAL
sector: Mapped[Optional[str]] = mapped_column(String(8)) # SECTOR
system: Mapped[Optional[str]] = mapped_column(String(5)) # SISTEMA

View File

@@ -54,7 +54,8 @@ celery_app.conf.update(
"api.v1.modules.a76.reports.importacion.winsaai.invoices.task",
"api.v1.modules.a76.reports.importacion.winsaai.pedimentos.task",
"api.v1.modules.core.help_center.tasks",
"api.v1.modules.core.help_center.tasks"
"api.v1.modules.core.help_center.tasks",
"api.v1.modules.a76.invoices.imports.process.task",
] # Ruta al módulo donde están las tareas
)

View File

@@ -491,5 +491,29 @@ export const invoicesApi = {
});
return api.delete(`/v1/a76/invoices/${invoiceId}/collections/${collectionId}/?${params.toString()}`);
}
},
processInvoice: (invoiceId: number, companyId: number) => {
const params = new URLSearchParams({
company_id: companyId.toString()
});
return api.post<{ task_id: string }>(
`/v1/a76/invoices/${invoiceId}/process?${params.toString()}`,
{}
);
},
getProcessStatus: (taskId: string) => {
return api.get<{
state: 'PROCESSING' | 'SUCCESS' | 'FAILURE';
info?: { current: number; status: string };
result?: {
status: 'success' | 'validation_error' | 'error';
invoice_id?: number;
message?: string;
errors?: Array<{ field: string; message: string; code?: string; solution?: string[] }>;
sql_errors?: Array<{ consecutive: number; error: string }>;
};
}>(`/v1/a76/invoices/process/${taskId}/status`);
}
};

View File

@@ -577,22 +577,37 @@
// Esta función se llama cuando el diálogo reporta SUCCESS
try {
if (result.status === 'success') {
const blob = base64ToBlob(result.content, result.media_type);
const url = window.URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
a.download = result.file_name;
document.body.appendChild(a);
a.click();
window.URL.revokeObjectURL(url);
document.body.removeChild(a);
toast.success('PDF Descargado exitosamente');
if (result.content) {
// Resultado de generación de PDF: descargar archivo
const blob = base64ToBlob(result.content, result.media_type);
const url = window.URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
a.download = result.file_name;
document.body.appendChild(a);
a.click();
window.URL.revokeObjectURL(url);
document.body.removeChild(a);
toast.success('PDF Descargado exitosamente');
} else {
// Resultado de procesamiento de factura
toast.success('Factura procesada correctamente');
reloadData();
}
} else if (result.status === 'validation_error') {
const errors: any[] = result.errors || [];
const preview = errors
.slice(0, 3)
.map((e: any) => `• ${e.message}`)
.join('\n');
const extra = errors.length > 3 ? `\n...y ${errors.length - 3} más` : '';
toast.error(`${errors.length} error(es) de validación:\n${preview}${extra}`);
} else {
toast.error('El worker reportó un error: ' + (result.message || 'Desconocido'));
}
} catch (e) {
console.error('Error al procesar descarga:', e);
toast.error('Error al procesar el archivo descargado');
console.error('Error al procesar resultado:', e);
toast.error('Error al procesar el resultado de la tarea');
} finally {
// Cerrar diálogo después de un breve momento
setTimeout(() => {
@@ -683,6 +698,33 @@
}
}
async function handleProcessInvoice() {
if (!selectedInvoice || !companyStore.activeCompany) {
toast.info('Selecciona una factura para procesar');
return;
}
try {
const response = await invoicesApi.processInvoice(
selectedInvoice.id,
companyStore.activeCompany.id
);
if (response.error) {
toast.error(`Error al iniciar el proceso: ${response.error}`);
return;
}
currentTaskId = response.data!.task_id;
currentStatusFunction = invoicesApi.getProcessStatus;
progressDialogTitle = 'Procesando factura';
showProgressDialog = true;
} catch (e) {
console.error('Error al iniciar proceso de factura:', e);
toast.error('No se pudo iniciar el proceso');
}
}
// Opciones de tipo de operación para el filtro
const operationTypeOptions = [
{ value: '', label: 'Todas' },
@@ -1030,12 +1072,12 @@
<div class="h-6 w-px bg-border"></div>
</div>
{/if}
<!-- Por ahora, Actualizar Estado funciona solo si hay ESTRICTAMENTE UNA seleccionada -->
<!-- Procesa la factura seleccionada mediante tarea Celery -->
<Button
variant="outline"
size="sm"
disabled={loading || selectedInvoiceIds.length !== 1}
onclick={() => handleUpdateStatus(true)}
onclick={handleProcessInvoice}
>
<RefreshCw class="mr-2 h-4 w-4" />
Actualizar