AlexeerCT
c812f5214c
- Introduced a new helper function `_ensure_client_provider_address` to manage the assignment of addresses for ClientProvider instances. - Refactored the `create_business_catalogs` function to utilize the new helper, ensuring that existing addresses are reused if available, preventing duplicate entries. - This change enhances the integrity of address data during business catalog creation by ensuring proper address assignment based on existing records.
426 lines
12 KiB
Python
426 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
|
|
from api.v1.modules.a76.clients_and_providers.models import (
|
|
ClientOrProviderEnum,
|
|
ClientProvider,
|
|
ClientProviderAddress,
|
|
)
|
|
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
|
|
from api.v1.modules.a76.general_catalogs.company.models import Company
|
|
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
|
|
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
|
|
from api.v1.modules.a76.invoices.models import (
|
|
Currency,
|
|
InvoiceComplianceMx,
|
|
InvoiceFinancials,
|
|
InvoiceHeader,
|
|
InvoiceLogistics,
|
|
InvoiceStatus,
|
|
OperationType,
|
|
)
|
|
from api.v1.modules.a76.items.line_customs.models import LineCustom
|
|
from api.v1.modules.a76.items.line_descriptions.models import LineDescription
|
|
from api.v1.modules.a76.items.line_financials.models import LineFinancial
|
|
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
|
|
from api.v1.modules.a76.items.models import LineItem
|
|
from api.v1.modules.core.tenants.models import Tenant, TenantType
|
|
from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType
|
|
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
|
|
|
|
|
|
def ensure_reference_data(db: Session) -> None:
|
|
for key, desc in [
|
|
("TEM", "IMPORTACION TEMPORAL"),
|
|
("DEF", "IMPORTACION DEFINITIVA"),
|
|
("MEX", "COMPRAS MEXICANAS"),
|
|
("DONAC", "DONACION"),
|
|
]:
|
|
if db.get(InvoiceType, key) is None:
|
|
operation = "exp" if key == "DONAC" else "imp"
|
|
db.add(
|
|
InvoiceType(
|
|
key=key,
|
|
description=desc,
|
|
note="seed for tests",
|
|
type="both",
|
|
operation=operation,
|
|
)
|
|
)
|
|
|
|
if db.get(RegimenPedimento, "A1") is None:
|
|
db.add(
|
|
RegimenPedimento(
|
|
code="A1",
|
|
description="Regimen de prueba",
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
|
|
def ensure_tenant_company(db: Session, tenant_id: int = 1, company_id: int = 1) -> Company:
|
|
tenant = db.get(Tenant, tenant_id)
|
|
if tenant is None:
|
|
tenant = Tenant(
|
|
id=tenant_id,
|
|
name=f"Tenant {tenant_id}",
|
|
slug=f"tenant-{tenant_id}",
|
|
type=TenantType.SHARED,
|
|
keycloak_realm="test",
|
|
is_active=True,
|
|
)
|
|
db.add(tenant)
|
|
|
|
company = db.get(Company, company_id)
|
|
if company is None:
|
|
company = Company(
|
|
id=company_id,
|
|
tenant_id=tenant_id,
|
|
name=f"Company {company_id}",
|
|
rfc="TST010101AAA",
|
|
prosec=False,
|
|
)
|
|
db.add(company)
|
|
db.flush()
|
|
return company
|
|
|
|
|
|
def _ensure_client_provider_address(
|
|
db: Session,
|
|
client: ClientProvider,
|
|
*,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
country: str,
|
|
) -> None:
|
|
"""
|
|
Garantiza que el cliente tenga fila de dirección. Si el ORM no tiene `address`
|
|
cargada pero en BD ya existe (datos previos en CI / DB compartida), reutiliza
|
|
esa fila en lugar de INSERT duplicado (pkey id = client_id).
|
|
"""
|
|
if client.address is not None:
|
|
return
|
|
existing = (
|
|
db.query(ClientProviderAddress)
|
|
.filter(ClientProviderAddress.client_id == client.id)
|
|
.first()
|
|
)
|
|
if existing is not None:
|
|
client.address = existing
|
|
return
|
|
client.address = ClientProviderAddress(
|
|
id=client.id,
|
|
client_id=client.id,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
country=country,
|
|
)
|
|
|
|
|
|
def create_business_catalogs(db: Session, tenant_id: int, company_id: int) -> dict:
|
|
provider = (
|
|
db.query(ClientProvider)
|
|
.filter(
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
ClientProvider.name == "Proveedor Test",
|
|
)
|
|
.first()
|
|
)
|
|
if provider is None:
|
|
provider = ClientProvider(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
name="Proveedor Test",
|
|
client_or_provider=ClientOrProviderEnum.BOTH,
|
|
is_active=True,
|
|
)
|
|
db.add(provider)
|
|
db.flush()
|
|
_ensure_client_provider_address(
|
|
db,
|
|
provider,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
country="MEX",
|
|
)
|
|
|
|
sold_to = (
|
|
db.query(ClientProvider)
|
|
.filter(
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
ClientProvider.name == "Cliente Test",
|
|
)
|
|
.first()
|
|
)
|
|
if sold_to is None:
|
|
sold_to = ClientProvider(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
name="Cliente Test",
|
|
client_or_provider=ClientOrProviderEnum.BOTH,
|
|
is_active=True,
|
|
)
|
|
db.add(sold_to)
|
|
db.flush()
|
|
_ensure_client_provider_address(
|
|
db,
|
|
sold_to,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
country="USA",
|
|
)
|
|
|
|
shipped_to = (
|
|
db.query(ClientProvider)
|
|
.filter(
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
ClientProvider.name == "Destinatario Test",
|
|
)
|
|
.first()
|
|
)
|
|
if shipped_to is None:
|
|
shipped_to = ClientProvider(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
name="Destinatario Test",
|
|
client_or_provider=ClientOrProviderEnum.BOTH,
|
|
is_active=True,
|
|
)
|
|
db.add(shipped_to)
|
|
db.flush()
|
|
_ensure_client_provider_address(
|
|
db,
|
|
shipped_to,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
country="USA",
|
|
)
|
|
|
|
broker = (
|
|
db.query(CustomsBroker)
|
|
.filter(
|
|
CustomsBroker.tenant_id == tenant_id,
|
|
CustomsBroker.company_id == company_id,
|
|
CustomsBroker.broker_key == "A1234",
|
|
)
|
|
.first()
|
|
)
|
|
if broker is None:
|
|
broker = CustomsBroker(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
broker_key="A1234",
|
|
name="Agente Test",
|
|
)
|
|
db.add(broker)
|
|
db.flush()
|
|
|
|
uom = (
|
|
db.query(UnitOfMeasure)
|
|
.filter(
|
|
UnitOfMeasure.tenant_id == tenant_id,
|
|
UnitOfMeasure.company_id == company_id,
|
|
UnitOfMeasure.code == "PZA",
|
|
)
|
|
.first()
|
|
)
|
|
if uom is None:
|
|
uom = UnitOfMeasure(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
code="PZA",
|
|
description="Pieza",
|
|
)
|
|
db.add(uom)
|
|
db.flush()
|
|
|
|
exr = (
|
|
db.query(ExchangeRate)
|
|
.filter(
|
|
ExchangeRate.tenant_id == tenant_id,
|
|
ExchangeRate.company_id == company_id,
|
|
ExchangeRate.date == datetime(2026, 3, 17, 0, 0, 0),
|
|
)
|
|
.first()
|
|
)
|
|
if exr is None:
|
|
exr = ExchangeRate(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
date=datetime(2026, 3, 17, 0, 0, 0),
|
|
value=Decimal("17.250000"),
|
|
local_currency="MN",
|
|
foreign_currency="ME",
|
|
)
|
|
db.add(exr)
|
|
|
|
db.flush()
|
|
return {
|
|
"provider_id": provider.id,
|
|
"sold_to_id": sold_to.id,
|
|
"shipped_to_id": shipped_to.id,
|
|
"broker_id": broker.id,
|
|
"uom_id": uom.id,
|
|
}
|
|
|
|
|
|
def create_import_invoice_with_line(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
catalogs: dict,
|
|
invoice_type: str = "TEM",
|
|
invoice_number: str = "IMP-0001",
|
|
qty: Decimal = Decimal("10"),
|
|
) -> tuple[InvoiceHeader, LineItem]:
|
|
invoice = InvoiceHeader(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
system="fixed_asset",
|
|
operation_type=OperationType.IMP,
|
|
invoice_type=invoice_type,
|
|
document_type="A1",
|
|
invoice_number=invoice_number,
|
|
invoice_date=date(2026, 3, 17),
|
|
status=InvoiceStatus.PENDING,
|
|
)
|
|
invoice.compliance_mx = InvoiceComplianceMx(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
provider_id=catalogs["provider_id"],
|
|
sold_to_id=catalogs["sold_to_id"],
|
|
shipped_to_id=catalogs["shipped_to_id"],
|
|
customs_broker_id=catalogs["broker_id"],
|
|
is_pedimento_pending=True,
|
|
was_reviewed_by_company=True,
|
|
)
|
|
invoice.financials = InvoiceFinancials(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
currency=Currency.FOREIGN,
|
|
exchange_rate=Decimal("17.250000"),
|
|
iva_factor="16",
|
|
)
|
|
invoice.logistics = InvoiceLogistics(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
weight_type="kgs",
|
|
)
|
|
db.add(invoice)
|
|
db.flush()
|
|
|
|
line = LineItem(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
invoice_id=invoice.id,
|
|
line_number=1,
|
|
unit_of_measure=catalogs["uom_id"],
|
|
)
|
|
line.quantity = LineQuantity(
|
|
quantity=qty,
|
|
net_weight=Decimal("50"),
|
|
gross_weight=Decimal("55"),
|
|
package_quantity=1,
|
|
)
|
|
line.financial = LineFinancial(
|
|
unit_cost_capture=Decimal("10"),
|
|
value_usd=Decimal("100"),
|
|
value_mxn=Decimal("1725"),
|
|
)
|
|
line.customs = LineCustom(origin_country="MEX", fraction_type="GENERAL")
|
|
line.description = LineDescription(has_serial=False)
|
|
db.add(line)
|
|
db.flush()
|
|
return invoice, line
|
|
|
|
|
|
def create_export_invoice_with_line(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
catalogs: dict,
|
|
source_import_invoice: InvoiceHeader,
|
|
source_import_line: LineItem,
|
|
qty: Decimal = Decimal("4"),
|
|
) -> tuple[InvoiceHeader, LineItem]:
|
|
invoice = InvoiceHeader(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
system="fixed_asset",
|
|
operation_type=OperationType.EXP,
|
|
invoice_type="DONAC",
|
|
document_type="A1",
|
|
invoice_number="EXP-0001",
|
|
invoice_date=date(2026, 3, 18),
|
|
status=InvoiceStatus.PENDING,
|
|
)
|
|
invoice.compliance_mx = InvoiceComplianceMx(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
provider_id=catalogs["provider_id"],
|
|
sold_to_id=catalogs["sold_to_id"],
|
|
shipped_to_id=catalogs["shipped_to_id"],
|
|
customs_broker_id=catalogs["broker_id"],
|
|
is_pedimento_pending=True,
|
|
was_reviewed_by_company=True,
|
|
)
|
|
invoice.financials = InvoiceFinancials(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
currency=Currency.FOREIGN,
|
|
exchange_rate=Decimal("17.250000"),
|
|
iva_factor="16",
|
|
)
|
|
invoice.logistics = InvoiceLogistics(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
weight_type="kgs",
|
|
)
|
|
db.add(invoice)
|
|
db.flush()
|
|
|
|
line = LineItem(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
invoice_id=invoice.id,
|
|
line_number=1,
|
|
unit_of_measure=catalogs["uom_id"],
|
|
)
|
|
line.quantity = LineQuantity(
|
|
quantity=qty,
|
|
net_weight=Decimal("20"),
|
|
gross_weight=Decimal("22"),
|
|
package_quantity=1,
|
|
)
|
|
line.financial = LineFinancial(
|
|
unit_cost_capture=Decimal("12"),
|
|
value_usd=Decimal("48"),
|
|
value_mxn=Decimal("828"),
|
|
)
|
|
line.customs = LineCustom(origin_country="MEX", origin_procedure="TEM", fraction_type="GENERAL")
|
|
line.description = LineDescription(has_serial=False)
|
|
db.add(line)
|
|
db.flush()
|
|
|
|
db.add(
|
|
FaLineItem(
|
|
id=line.id,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
discharge=True,
|
|
search_invoice=source_import_invoice.invoice_number,
|
|
search_line=source_import_line.line_number,
|
|
is_subitem=False,
|
|
)
|
|
)
|
|
db.flush()
|
|
return invoice, line
|