ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/development' into feature/fraction-catalogs

Browse Source
This commit is contained in:
acazares
2026-02-19 17:22:37 -06:00
parent 2c3a2735db e41a381d0a
commit dc37f458f5
58 changed files with 8857 additions and 8230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/api/v1/modules/a24/fa/fa_item_lines/models.py
View File

@@ -16,7 +16,7 @@ from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
if TYPE_CHECKING:
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.models import LineItem
class FaLineItem(Base, TenantScopedMixin, TimestampMixin):

2
backend/api/v1/modules/a24/fa/fa_item_lines/service.py
View File

@@ -96,7 +96,7 @@ class FaLineItemService:
"""Crear una nueva línea de activo fijo"""
try:
# Verificar que la línea base existe en a76.item_lines
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.models import LineItem
base_line_item = (
db.query(LineItem)

50
backend/api/v1/modules/a76/audit_log/events.py
View File

@@ -1,11 +1,14 @@
"""
Audit Log Events
"""
from sqlalchemy import event, inspect
from sqlalchemy.orm import Session
from .services.service import AuditService
from .utils.serialization import serialize_for_json
from core.context import get_user_context
def register_audit_listeners(models_to_audit):
"""
Register SQLAlchemy listeners for given models
@@ -15,16 +18,23 @@ def register_audit_listeners(models_to_audit):
event.listen(model, "after_update", after_update_listener)
event.listen(model, "after_delete", after_delete_listener)
def _get_current_username():
try:
context = get_user_context()
if context:
# Token usually has 'preferred_username' or 'name' or 'sub'
return context.get("preferred_username") or context.get("email") or context.get("sub") or "System"
return (
context.get("preferred_username")
or context.get("email")
or context.get("sub")
or "System"
)
except:
pass
return "System"
def after_insert_listener(mapper, connection, target):
"""
Listener for INSERT operations
@@ -32,8 +42,9 @@ def after_insert_listener(mapper, connection, target):
table_name = target.__tablename__
record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
username = _get_current_username()
company_id = getattr(target, "company_id", None)
company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
tenant_id = getattr(target, "tenant_id", None)
# Create a session bound to the connection
session = Session(bind=connection)
try:
@@ -44,24 +55,26 @@ def after_insert_listener(mapper, connection, target):
record_data=record_data,
username=username,
record_id=str(getattr(target, "id", "")),
company_id=company_id
company_id=company_id,
tenant_id=tenant_id,
)
except Exception as e:
print(f"Error logging insert: {e}")
finally:
session.close()
def after_update_listener(mapper, connection, target):
"""
Listener for UPDATE operations
"""
table_name = target.__tablename__
state = inspect(target)
changes = {}
old_values = {}
new_values = {}
for attr in state.attrs:
hist = attr.history
if hist.has_changes():
@@ -74,6 +87,8 @@ def after_update_listener(mapper, connection, target):
record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
username = _get_current_username()
company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
tenant_id = getattr(target, "tenant_id", None)
session = Session(bind=connection)
try:
@@ -84,13 +99,16 @@ def after_update_listener(mapper, connection, target):
record_data=record_data,
username=username,
record_id=str(getattr(target, "id", "")),
old_values=old_values,
new_values=new_values
old_values=serialize_for_json(old_values),
new_values=serialize_for_json(new_values),
company_id=company_id,
tenant_id=tenant_id,
)
except Exception as e:
print(f"Error logging update: {e}")
print(f"Error logging update: {e}")
finally:
session.close()
session.close()
def after_delete_listener(mapper, connection, target):
"""
@@ -99,7 +117,9 @@ def after_delete_listener(mapper, connection, target):
table_name = target.__tablename__
record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
username = _get_current_username()
company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
tenant_id = getattr(target, "tenant_id", None)
session = Session(bind=connection)
try:
AuditService.log_crud_operation(
@@ -108,9 +128,11 @@ def after_delete_listener(mapper, connection, target):
operation_type="DELETE",
record_data=record_data,
username=username,
record_id=str(getattr(target, "id", ""))
record_id=str(getattr(target, "id", "")),
company_id=company_id,
tenant_id=tenant_id,
)
except Exception as e:
print(f"Error logging delete: {e}")
print(f"Error logging delete: {e}")
finally:
session.close()
session.close()

7
backend/api/v1/modules/a76/audit_log/middleware.py
View File

@@ -18,5 +18,10 @@ class UserContextMiddleware(BaseHTTPMiddleware):
# Log error or ignore
pass
response = await call_next(request)
try:
response = await call_next(request)
except Exception:
# Re-raise the exception to let other middleware and handlers deal with it
raise
return response

26
backend/api/v1/modules/a76/audit_log/models.py
View File

@@ -4,10 +4,18 @@ Audit Log Models
from sqlalchemy import Column, Integer, String, Date, Time, DateTime, Text, Index, func
from sqlalchemy.dialects.postgresql import JSONB, ARRAY
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class AuditLog(Base):
class AuditLog(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "audit_logs"
__table_args__ = (
Index('idx_audit_username_date', 'username', 'date'),
Index('idx_audit_procedure_date', 'procedure', 'date'),
Index('idx_audit_system_timestamp', 'system', 'timestamp'),
Index('idx_audit_table_record', 'table_name', 'record_id'),
{"schema": "a76"} # Use the a76 schema for audit logs
)
# Primary Key
spec_id = Column(Integer, primary_key=True, autoincrement=True)
@@ -23,8 +31,6 @@ class AuditLog(Base):
# Technical Columns
timestamp = Column(DateTime(timezone=True), nullable=False, index=True) # Combined for queries
system = Column(String(20), nullable=False, index=True, default="SCAF")
company_id = Column(Integer, nullable=True, index=True)
tenant_id = Column(Integer, nullable=True, index=True)
# Traceability
table_name = Column(String(100), nullable=True, index=True)
@@ -42,16 +48,4 @@ class AuditLog(Base):
endpoint = Column(String(500), nullable=True)
request_method = Column(String(10), nullable=True)
session_id = Column(String(50), nullable=True, index=True)
execution_time_ms = Column(Integer, nullable=True)
# Metadata
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
# Composite Indexes for common filters
__table_args__ = (
Index('idx_audit_username_date', 'username', 'date'),
Index('idx_audit_procedure_date', 'procedure', 'date'),
Index('idx_audit_system_timestamp', 'system', 'timestamp'),
Index('idx_audit_table_record', 'table_name', 'record_id'),
{'extend_existing': True}
)
execution_time_ms = Column(Integer, nullable=True)

6
backend/api/v1/modules/a76/audit_log/services/service.py
View File

@@ -86,7 +86,8 @@ class AuditService:
# Context
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
company_id: Optional[int] = None
company_id: Optional[int] = None,
tenant_id: Optional[int] = None
):
"""
High-level wrapper to log CRUD operations automatically mapping to Legacy format
@@ -149,7 +150,8 @@ class AuditService:
changed_fields=changed_fields,
ip_address=ip_address,
user_agent=user_agent,
company_id=company_id
company_id=company_id,
tenant_id=tenant_id
)
@staticmethod

7
backend/api/v1/modules/a76/audit_log/utils/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
"""
Audit Log Utilities
"""
from .serialization import serialize_for_json
__all__ = ["serialize_for_json"]

44
backend/api/v1/modules/a76/audit_log/utils/serialization.py Normal file
View File

@@ -0,0 +1,44 @@
"""
Serialization utilities for audit logs
"""
from datetime import date, datetime, time
from decimal import Decimal
from uuid import UUID
from typing import Any, Dict
def serialize_value(value: Any) -> Any:
"""
Convert a Python value to a JSON-serializable type
"""
if value is None:
return None
elif isinstance(value, (date, datetime)):
return value.isoformat()
elif isinstance(value, time):
return value.isoformat()
elif isinstance(value, Decimal):
return float(value)
elif isinstance(value, UUID):
return str(value)
elif isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
elif isinstance(value, (list, tuple)):
return [serialize_value(item) for item in value]
elif isinstance(value, dict):
return {key: serialize_value(val) for key, val in value.items()}
else:
# For any other type, try to return as-is (str, int, float, bool, None)
# If it fails JSON serialization later, at least we tried
return value
def serialize_for_json(data: Dict[str, Any]) -> Dict[str, Any]:
"""
Recursively serialize a dictionary for JSON storage
"""
if not data:
return data
return {key: serialize_value(value) for key, value in data.items()}

29
backend/api/v1/modules/a76/general_catalogs/fractions/historical_tariff_fractions/dto.py Normal file
View File

@@ -0,0 +1,29 @@
"""
DTOs for historical tariff fractions.
"""
from datetime import datetime
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, ConfigDict
class HistoricalTariffFractionResponseDTO(BaseModel):
id: int
historical_fraction: Optional[str] = None
unit_of_measure_code: Optional[str] = None
country: Optional[str] = None
fraction_type: Optional[str] = None
sector: Optional[str] = None
import_tax_rate: Optional[Decimal] = None
export_tax_rate: Optional[Decimal] = None
publication_date: Optional[datetime] = None
is_immex: Optional[bool] = None
normal_temporality: Optional[bool] = None
services_temporality: Optional[bool] = None
certified_temporality: Optional[bool] = None
by_log: Optional[bool] = None
end_date: Optional[datetime] = None
model_config = ConfigDict(from_attributes=True)

1
backend/api/v1/modules/a76/general_catalogs/fractions/historical_tariff_fractions/models.py
View File

@@ -18,6 +18,7 @@ class HistoricalTariffFraction(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(Integer, primary_key=True, nullable=False)
historical_fraction: Mapped[Optional[str]] = mapped_column(String(8), nullable=True)
nico: Mapped[Optional[str]] = mapped_column(String(2), nullable=True)
unit_of_measure_code: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.unit_of_measure_customs.code"), nullable=True)
country: Mapped[Optional[str]] = mapped_column(ForeignKey("public.countries.m3_key"), nullable=True)
fraction_type: Mapped[Optional[str]] = mapped_column(String(7), nullable=True)

13180
backend/api/v1/modules/a76/general_catalogs/fractions/historical_tariff_fractions/seed.py
View File

File diff suppressed because it is too large Load Diff

6
backend/api/v1/modules/a76/general_catalogs/router.py
View File

@@ -28,7 +28,7 @@ from .electronic_notices.routes import router as electronic_notices_router
router = APIRouter()
router.include_router(company_router, tags=["a76 / company"])
router.include_router(company_router, tags=["a76 / company"])
router.include_router(package_router)
router.include_router(ports_router)
router.include_router(tariff_fractions_router)
@@ -37,7 +37,7 @@ router.include_router(historical_tariff_fractions_router, prefix="/fractions/his
router.include_router(canadian_tariff_fractions_router, prefix="/fractions/canadian-tariff-fractions", tags=["a76 / canadian_tariff_fractions"])
router.include_router(depreciation_catalog_router)
router.include_router(fda_catalog_router)
router.include_router(seal_router, tags=["a76 / seal"])
router.include_router(seal_router, tags=["a76 / seal"])
router.include_router(units_of_measure_router)
router.include_router(identifiers_router)
router.include_router(exchange_rate_router, tags=["a76 / exchange_rate"])
@@ -53,4 +53,4 @@ router.include_router(signatures_router)
router.include_router(error_catalogs_router)
router.include_router(doda_router)
router.include_router(prevalidators_router)
router.include_router(electronic_notices_router)
router.include_router(electronic_notices_router)

3
backend/api/v1/modules/a76/imports/tasks.py
View File

@@ -345,8 +345,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.models import LineItem
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.line_customs.models import LineCustom

10
backend/api/v1/modules/a76/invoices/imports/temporary/validators/common.py
View File

@@ -7,7 +7,7 @@ from api.v1.modules.a76.clients_and_providers.models import ClientProvider
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
from ....models import InvoiceComplianceMx
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
from ....models import TransportType, Currency, WeightUnit
@@ -435,11 +435,11 @@ def validate_common(
# Only check for existing items during update operations (when invoice has an id)
if hasattr(invoice, "id"):
has_items = (
db.query(Item)
db.query(LineItem)
.filter(
Item.invoice_id == invoice.id,
Item.tenant_id == tenant_id,
Item.company_id == company_id,
LineItem.invoice_id == invoice.id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
.first()
)

4
backend/api/v1/modules/a76/items/__init__.py
View File

@@ -4,16 +4,14 @@ Items module - Annex 76 Compliance
# Import models in correct order to avoid circular dependencies
# LineItem must be imported before models that reference it
from .line_items.models import LineItem
from .line_financials.models import LineFinancial
from .line_quantities.models import LineQuantity
from .line_customs.models import LineCustom
from .line_descriptions.models import LineDescription
from .line_references.models import LineReference
from .models import Item, CTMReceipt, SubassemblyEntry
from .models import LineItem, CTMReceipt, SubassemblyEntry
__all__ = [
"Item",
"LineItem",
"LineFinancial",
"LineQuantity",

10
backend/api/v1/modules/a76/items/common/common_validators.py
View File

@@ -1,6 +1,6 @@
from sqlalchemy import func
from core.exceptions import ErrorCollector
from ..line_items import models
from .. import models
from sqlalchemy.orm import Session
@@ -21,11 +21,11 @@ def item_exists(db: Session, item_line: int, tenant_id: int, company_id: int):
def count_items(db: Session, invoice_id: int, tenant_id: int, company_id: int):
count = (
db.query(func.count())
.select_from(models.Item)
.select_from(models.LineItem)
.filter(
models.Item.invoice_id == invoice_id,
models.Item.tenant_id == tenant_id,
models.Item.company_id == company_id,
models.LineItem.invoice_id == invoice_id,
models.LineItem.tenant_id == tenant_id,
models.LineItem.company_id == company_id,
)
.scalar()
)

19
backend/api/v1/modules/a76/items/imports/temporary/validators/common.py
View File

@@ -6,8 +6,7 @@ from sqlalchemy import func
from ....common.fractions import search_fraction_preference
from ....common.common_validators import item_exists
from ....models import Item
from ....line_items.models import LineItem
from ....models import LineItem
from ....line_customs.models import FractionType, LineCustom
from api.v1.modules.a76.items.schemas import LineItemCreate
from api.v1.modules.a76.invoices.models import InvoiceHeader
@@ -26,21 +25,13 @@ from api.v1.modules.a76.general_catalogs.company.models import Company
def validate_common(
db: Session,
line: LineItemCreate,
invoice_id: int, # Para creación, se pasa directamente; para update, se consulta del item
tenant_id: int,
company_id: int,
errors: ErrorCollector,
line_number: int,
):
# Para updates, line.item_id existe; para creates, es None
item_header = None
if line.item_id:
item_header = db.query(Item).filter(Item.id == line.item_id).first()
if item_header:
invoice_id = item_header.invoice_id
invoice: InvoiceHeader = invoice_exists_by_id(
db, invoice_id, tenant_id, company_id, errors
db, line.invoice_id, tenant_id, company_id, errors
)
line_item: LineItem = item_exists(db, line.line_number, tenant_id, company_id)
@@ -164,7 +155,7 @@ def validate_common(
code="PACKAGE_QUANTITY_MUST_BE_GREATER_THAN_ZERO",
)
else:
if (line.quantity.package_quantity or line.quantity.package_quantity > 0) and not line.quantity.package_id:
if line.quantity.package_quantity and (line.quantity.package_quantity > 0 and not line.quantity.package_id):
errors.add_error(
field=f"line[{line_number}].quantity.package_id",
message="El paquete es obligatorio cuando se proporciona la cantidad de paquetes.",
@@ -301,8 +292,8 @@ def validate_common(
code="AMERICAN_FRACTION_NOT_FOUND",
)
if item_header and item_header.order:
if len(item_header.order) > 20:
if line.order:
if len(line.order) > 20:
errors.add_error(
field=f"item.order",
message="El campo orden no debe exceder los 20 caracteres.",

26
backend/api/v1/modules/a76/items/imports/temporary/validators/create.py
View File

@@ -4,7 +4,7 @@ from sqlalchemy.orm import Session
from ....common.common_validators import count_items
from core.exceptions import ErrorCollector
from ....line_items.models import LineItem
from ....models import LineItem
from ....line_financials.models import LineFinancial
from ....line_financials.schemas import LineFinancialCreate
from ....line_quantities.models import LineQuantity
@@ -15,7 +15,7 @@ from ....line_descriptions.models import LineDescription
from ....line_descriptions.schemas import LineDescriptionCreate
from ....line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from ....models import Item
from ....models import LineItem
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.packages.models import Package
@@ -27,8 +27,7 @@ from .common import validate_common
def validate_create(
db: Session,
line, # LineItemCreate schema (Pydantic)
invoice_id: int, # Passed from service
line: LineItem, # LineItemCreate schema (Pydantic)
tenant_id: int,
company_id: int,
errors: ErrorCollector,
@@ -42,16 +41,7 @@ def validate_create(
line: LineItemCreate schema with nested data (financial, quantity, customs, etc.)
invoice_id: ID of the invoice this line belongs to
fa_data: FaLineItemCreateDTO or None (None for INV system)
"""
# Inicializar nested schemas si no existen (para poder validar y modificar)
if not line.financial:
line.financial = LineFinancialCreate()
if not line.quantity:
line.quantity = LineQuantityCreate()
if not line.customs:
line.customs = LineCustomCreate()
if not line.description:
line.description = LineDescriptionCreate()
"""
# Access fa_data safely
fa_data = getattr(line, "fa_data", None)
@@ -101,8 +91,8 @@ def validate_create(
principal_item_exists = db.query(
exists().where(
(LineItem.id == FaLineItem.id)
& (LineItem.item_id == Item.id)
& (Item.invoice_id == invoice_id)
& (LineItem.id == LineItem.id)
& (LineItem.invoice_id == line.invoice_id)
& (LineItem.line_number == line_number)
& (FaLineItem.is_subitem == False)
& (FaLineItem.contains_subitems == True)
@@ -131,14 +121,14 @@ def validate_create(
code="SUBITEM_NUMBER_INVALID",
)
validate_common(db, line, invoice_id, tenant_id, company_id, errors, line_number)
validate_common(db, line, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.id == invoice_id,
InvoiceHeader.id == line.invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)

20
backend/api/v1/modules/a76/items/imports/temporary/validators/update.py
View File

@@ -3,8 +3,7 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists
from core.exceptions import ErrorCollector
from ....line_items.models import LineItem
from ....models import Item
from ....models import LineItem
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import (
USTariffFraction,
@@ -15,8 +14,7 @@ from .common import validate_common
def validate_update(
db: Session,
line: LineItem,
existing_line: LineItem,
invoice_id: int, # Passed from service
existing_line: LineItem,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
@@ -26,14 +24,14 @@ def validate_update(
Validar y procesar actualización parcial de línea de importación temporal.
Si un campo no se proporciona, se mantiene el valor existente.
"""
validate_common(db, line, invoice_id, tenant_id, company_id, errors, line_number)
validate_common(db, line, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.id == invoice_id,
InvoiceHeader.id == line.invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id
)
@@ -81,6 +79,8 @@ def validate_update(
# Mantener peso existente
line.quantity.net_weight = existing_line.quantity.net_weight
print(f"After weight conversion: net_weight={line.quantity.net_weight}, gross_weight={line.quantity.gross_weight}, weight_type={invoice_weight_type}")
# Convertir peso bruto si se proporcionó
if line.quantity.gross_weight is not None:
gross_weight_input = line.quantity.gross_weight
@@ -141,8 +141,8 @@ def validate_update(
line.customs.advalorem_american = existing_line.customs.advalorem_american
# Orden de compra
if not line.reference.purchase_order:
line.reference.purchase_order = existing_line.reference.purchase_order
if not line.order:
line.order = existing_line.order
# Descripciones
if not line.description.description_spanish:
@@ -176,8 +176,8 @@ def validate_update(
# Número de parte
if not line.part_number:
line.part_number = existing_line.part_number
if not line.part_number_id:
line.part_number_id = existing_line.part_number_id
# Pago de impuesto
if line.tax_payment is None:

2
backend/api/v1/modules/a76/items/line_customs/models.py
View File

@@ -5,7 +5,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
from core.database import Base
if TYPE_CHECKING:
from ..line_items.models import LineItem
from ..models import LineItem
class FractionType:
"""Enumeration for fraction types"""

2
backend/api/v1/modules/a76/items/line_descriptions/models.py
View File

@@ -4,7 +4,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
from core.database import Base
if TYPE_CHECKING:
from ..line_items.models import LineItem
from ..models import LineItem
class LineDescription(Base):
"""

2
backend/api/v1/modules/a76/items/line_financials/models.py
View File

@@ -5,7 +5,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
from core.database import Base
if TYPE_CHECKING:
from ..line_items.models import LineItem
from ..models import LineItem
class LineFinancial(Base):
"""

16
backend/api/v1/modules/a76/items/line_items/__init__.py
View File

@@ -1,16 +0,0 @@
"""Line items module"""
from .models import LineItem
from .schemas import (
LineItemBase,
LineItemCreate,
LineItemUpdate,
LineItemResponse,
)
__all__ = [
"LineItem",
"LineItemBase",
"LineItemCreate",
"LineItemUpdate",
"LineItemResponse",
]

200
backend/api/v1/modules/a76/items/line_items/models.py
View File

@@ -1,200 +0,0 @@
from decimal import Decimal
from typing import Optional, TYPE_CHECKING
from sqlalchemy import (
Boolean,
String,
Integer,
Numeric,
SmallInteger,
ForeignKey,
ForeignKeyConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
if TYPE_CHECKING:
from ..models import Item
from ..line_financials.models import LineFinancial
from ..line_quantities.models import LineQuantity
from ..line_customs.models import LineCustom
from ..line_descriptions.models import LineDescription
from ..line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
class LineItem(Base, TenantScopedMixin, TimestampMixin):
"""
Unified line items for all items
Consolidates all line-level data from Q and S tables
"""
__tablename__ = "item_lines"
__table_args__ = ({"schema": "a76"},)
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
item_id: Mapped[int] = mapped_column(ForeignKey("a76.items.id"))
line_number: Mapped[int] = mapped_column(Integer) # LINEAIMPO/LINEAEXPO/LINEA
# Part identification
part_number: Mapped[Optional[int]] = mapped_column(
Integer, ForeignKey("a76.parts.id")
) # NUMPARTE
component_part_number: Mapped[Optional[int]] = mapped_column(
Integer, ForeignKey("a76.parts.id")
) # NUMPARTECOM
class_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.classes.id")
) # CLASE
# Unit of measure
unit_of_measure: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.units_of_measure.id")
) # UNIDADMEDIDA/UNIMED
alternate_unit: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.units_of_measure.id")
) # UNIMEDALTERNA
uma_key: Mapped[Optional[str]] = mapped_column(String(2)) # CLAVEUMA
auxiliary_unit: Mapped[Optional[str]] = mapped_column(String(5)) # UNIMEDAUXILIAR
# Permits and certificates
permit_number: Mapped[Optional[str]] = mapped_column(String(20)) # NUMPERMISO
page_line: Mapped[Optional[str]] = mapped_column(String(10)) # PAGRENGLON
has_certificate: Mapped[Optional[bool]] = mapped_column(
Boolean
) # TIENECO/CERTORIGEN
certificate_number: Mapped[Optional[str]] = mapped_column(
String(10)
) # NOCERTIFICADO
octave_permit: Mapped[Optional[str]] = mapped_column(String(20)) # PERMISOROCTAVA
permits_ped: Mapped[Optional[str]] = mapped_column(String(500)) # PERMISOSPED
# FDA
has_fda_code: Mapped[Optional[bool]] = mapped_column(Boolean) # LLEVACODFDA
fda_key: Mapped[Optional[str]] = mapped_column(String(10)) # CLAVEFDA
# Special flags
is_military_mcia: Mapped[Optional[bool]] = mapped_column(Boolean) # ESMCIAMILITAR
# IV32 (Tax identification)
iv32_type_key: Mapped[Optional[str]] = mapped_column(String(5)) # CLAVETIPOIV32
iv32_number: Mapped[Optional[str]] = mapped_column(String(35)) # NUMEROIV32
# IN CASE OF EXPO
scrap_invoice: Mapped[Optional[str]] = mapped_column(String(15)) # FACTURASCRAP
consecutive_destination: Mapped[Optional[int]] = mapped_column(
Integer
) # CONSECUTIVODES
ctm_section: Mapped[Optional[str]] = mapped_column(String(3)) # APARTADOCTM
# Tax payment
tax_payment: Mapped[Optional[bool]] = mapped_column(Boolean) # PAGOIMPUESTO
payment_method: Mapped[Optional[str]] = mapped_column(
String(9)
) # FORMAPAGO/FORMAPAGOTIGI
igi_amount: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # MONTOIGI
igi_payment_method: Mapped[Optional[str]] = mapped_column(
String(9)
) # FORMAPAGOTIGI
# FCC
fcc_key: Mapped[Optional[str]] = mapped_column(String(30)) # CLAVEFCC
# Valuation method
valuation_method: Mapped[Optional[str]] = mapped_column(String(2)) # METVALOR
valuation_determined_value: Mapped[Optional[Decimal]] = mapped_column(
Numeric(29, 8)
) # METVALORVALORDETERMINADO/METVALORACIONVALORDETERMINADO
valuation_reason: Mapped[Optional[str]] = mapped_column(
String(500)
) # METVALORMOTIVODEUSO/METVALORACIONMOTIVODEUSO
# Container rules
container_rule: Mapped[Optional[str]] = mapped_column(String(50)) # CONTENEDORREGLA
container_parts_ii: Mapped[Optional[str]] = mapped_column(
String(50)
) # CONTENEDORPARTESII
# APHIS
consecutive_aphis: Mapped[Optional[int]] = mapped_column(
Integer
) # CONSECUTIVOAPHIS
# BOM/Commercial
bom_version: Mapped[Optional[int]] = mapped_column(Integer) # VERSIONBOM
bill_version: Mapped[Optional[int]] = mapped_column(Integer) # VERSIONBILL
# TLCAN value
tlcan_value: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # VALORTLCAN
# Identifier
identifier: Mapped[Optional[str]] = mapped_column(String(2)) # IDENTIFICADOR
# Validation fields
validation_zero: Mapped[Optional[int]] = mapped_column(Integer) # VALIDACIONZERO
validation_one: Mapped[Optional[int]] = mapped_column(Integer) # VALIDACIONUNO
# Material type
material_type: Mapped[Optional[str]] = mapped_column(
String(50)
) # TIPOMAT/TIPODENUMPARTE
# Order concept
order_type: Mapped[Optional[str]] = mapped_column(String(50)) # TIPODEORDEN
line_concept: Mapped[Optional[str]] = mapped_column(
String(50)
) # CONCEPTODELAPARTIDA
# Review dispatch
review_dispatch: Mapped[Optional[str]] = mapped_column(String(10)) # REVISARDESP
# Take component from PT
take_component_pt: Mapped[Optional[int]] = mapped_column(Integer) # TOMARCOMOPT
# Pallet
pallet2: Mapped[Optional[int]] = mapped_column(SmallInteger) # PALLET2
# Wildcard field
wildcard_field: Mapped[Optional[str]] = mapped_column(String(100)) # CAMPOCOMODIN
# Relationships
item: Mapped["Item"] = relationship(back_populates="lines")
financial: Mapped[Optional["LineFinancial"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
quantity: Mapped[Optional["LineQuantity"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
customs: Mapped[Optional["LineCustom"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
description: Mapped[Optional["LineDescription"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
reference: Mapped[Optional["LineReference"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
class_info: Mapped[Optional["Class"]] = relationship(
"api.v1.modules.a76.classes.models.Class",
foreign_keys=[class_id],
viewonly=True,
)
unit_of_measure_info: Mapped[Optional["UnitOfMeasure"]] = relationship(
"api.v1.modules.a76.general_catalogs.units_of_measure.models.UnitOfMeasure",
foreign_keys=[unit_of_measure],
viewonly=True,
)
fa_data: Mapped[Optional["FaLineItem"]] = relationship(
"FaLineItem",
back_populates="master_info",
cascade="all, delete-orphan",
uselist=False,
)
part_info: Mapped[Optional["api.v1.modules.a76.parts.models.Part"]] = relationship(
"api.v1.modules.a76.parts.models.Part",
foreign_keys=[part_number],
viewonly=True,
)

293
backend/api/v1/modules/a76/items/line_items/schemas.py
View File

@@ -1,293 +0,0 @@
from decimal import Decimal
from typing import Optional, Any
from pydantic import BaseModel, Field, ConfigDict, field_validator, model_validator
# Import nested schemas
from ..line_customs.schemas import (
LineCustomCreate,
LineCustomUpdate,
LineCustomResponse,
)
from ..line_descriptions.schemas import (
LineDescriptionCreate,
LineDescriptionUpdate,
LineDescriptionResponse,
)
from ..line_quantities.schemas import (
LineQuantityCreate,
LineQuantityUpdate,
LineQuantityResponse,
)
from ..line_financials.schemas import (
LineFinancialCreate,
LineFinancialUpdate,
LineFinancialResponse,
)
from ..line_references.schemas import (
LineReferenceCreate,
LineReferenceUpdate,
LineReferenceResponse,
)
from api.v1.modules.a24.fa.fa_item_lines.dto import (
FaLineItemCreateDTO,
FaLineItemUpdateDTO,
FaLineItemResponseDTO,
)
# ============================================================================
# LINE ITEM SCHEMAS
# ============================================================================
class LineItemBase(BaseModel):
"""Base schema for line items"""
model_config = ConfigDict(populate_by_name=True)
item_id: int = Field(..., description="ID of the parent item")
line_number: int = Field(..., description="Line number")
# Part identification
part_number_id: Optional[int] = Field(
None,
description="Part number",
alias="part_number",
serialization_alias="part_number_id",
)
component_part_number_id: Optional[int] = Field(
None,
description="Component part number",
alias="component_part_number",
serialization_alias="component_part_number_id",
)
class_id: Optional[int] = Field(None, description="Class code")
# Unit of measure
unit_of_measure: Optional[int] = Field(None, description="Unit of measure")
alternate_unit: Optional[int] = Field(None, description="Alternate unit")
uma_key: Optional[str] = Field(None, max_length=2, description="UMA key")
auxiliary_unit: Optional[str] = Field(
None, max_length=5, description="Auxiliary unit"
)
# Permits and certificates
permit_number: Optional[str] = Field(
None, max_length=20, description="Permit number"
)
page_line: Optional[str] = Field(None, max_length=10, description="Page line")
has_certificate: Optional[bool] = Field(None, description="Has certificate")
certificate_number: Optional[str] = Field(
None, max_length=10, description="Certificate number"
)
octave_permit: Optional[str] = Field(
None, max_length=20, description="Octave permit"
)
permits_ped: Optional[str] = Field(None, max_length=500, description="PED permits")
# FDA
has_fda_code: Optional[bool] = Field(None, description="Has FDA code")
fda_key: Optional[str] = Field(None, max_length=10, description="FDA key")
# Special flags
is_military_mcia: Optional[bool] = Field(
None, description="Is military merchandise"
)
# IV32
iv32_type_key: Optional[str] = Field(
None, max_length=5, description="IV32 type key"
)
iv32_number: Optional[str] = Field(None, max_length=35, description="IV32 number")
# Export specific
scrap_invoice: Optional[str] = Field(
None, max_length=15, description="Scrap invoice"
)
consecutive_destination: Optional[int] = Field(
None, description="Consecutive destination"
)
ctm_section: Optional[str] = Field(None, max_length=3, description="CTM section")
# Tax payment
tax_payment: Optional[bool] = Field(None, description="Tax payment")
payment_method: Optional[str] = Field(
None, max_length=9, description="Payment method"
)
igi_amount: Optional[Decimal] = Field(None, description="IGI amount")
igi_payment_method: Optional[str] = Field(
None, max_length=9, description="IGI payment method"
)
# FCC
fcc_key: Optional[str] = Field(None, max_length=30, description="FCC key")
# Valuation method
valuation_method: Optional[str] = Field(
None, max_length=2, description="Valuation method"
)
valuation_determined_value: Optional[Decimal] = Field(
None, description="Valuation determined value"
)
valuation_reason: Optional[str] = Field(
None, max_length=500, description="Valuation reason"
)
# Container rules
container_rule: Optional[str] = Field(
None, max_length=50, description="Container rule"
)
container_parts_ii: Optional[str] = Field(
None, max_length=50, description="Container parts II"
)
# APHIS
consecutive_aphis: Optional[int] = Field(None, description="Consecutive APHIS")
# BOM/Commercial
bom_version: Optional[int] = Field(None, description="BOM version")
bill_version: Optional[int] = Field(None, description="Bill version")
# TLCAN value
tlcan_value: Optional[Decimal] = Field(None, description="TLCAN value")
# Identifier
identifier: Optional[str] = Field(None, max_length=2, description="Identifier")
# Validation fields
validation_zero: Optional[int] = Field(None, description="Validation zero")
validation_one: Optional[int] = Field(None, description="Validation one")
# Material type
material_type: Optional[str] = Field(
None, max_length=50, description="Material type"
)
# Order concept
order_type: Optional[str] = Field(None, max_length=50, description="Order type")
line_concept: Optional[str] = Field(None, max_length=50, description="Line concept")
# Review dispatch
review_dispatch: Optional[str] = Field(
None, max_length=10, description="Review dispatch"
)
# Take component from PT
take_component_pt: Optional[int] = Field(None, description="Take component from PT")
# Pallet
pallet2: Optional[int] = Field(None, description="Pallet 2")
# Wildcard field
wildcard_field: Optional[str] = Field(
None, max_length=100, description="Wildcard field"
)
class LineItemCreate(LineItemBase):
"""Schema for creating line item with all nested data"""
# Override base fields - estos se asignan automáticamente en el service
item_id: Optional[int] = Field(
None, description="ID of the parent item (auto-assigned)"
)
line_number: Optional[int] = Field(None, description="Line number (auto-assigned)")
financial: Optional[LineFinancialCreate] = Field(
None, description="Financial data for this line"
)
quantity: Optional[LineQuantityCreate] = Field(
None, description="Quantity data for this line"
)
customs: Optional[LineCustomCreate] = Field(
None, description="Customs data for this line"
)
description: Optional[LineDescriptionCreate] = Field(
None, description="Description data for this line"
)
reference: Optional[LineReferenceCreate] = Field(
None, description="Reference data for this line"
)
fa_data: Optional[FaLineItemCreateDTO] = Field(
None, description="Fixed Asset data for this line"
)
class LineItemUpdate(LineItemBase):
"""Schema for updating line item with all nested data"""
# Override base fields - todos opcionales en updates
item_id: Optional[int] = Field(None, description="ID of the parent item")
line_number: Optional[int] = Field(None, description="Line number")
financial: Optional[LineFinancialUpdate] = Field(
None, description="Financial data for this line"
)
quantity: Optional[LineQuantityUpdate] = Field(
None, description="Quantity data for this line"
)
customs: Optional[LineCustomUpdate] = Field(
None, description="Customs data for this line"
)
description: Optional[LineDescriptionUpdate] = Field(
None, description="Description data for this line"
)
reference: Optional[LineReferenceUpdate] = Field(
None, description="Reference data for this line"
)
fa_data: Optional[FaLineItemUpdateDTO] = Field(
None, description="Fixed Asset data for this line"
)
class LineItemResponse(LineItemBase):
"""Schema for line item response with all nested data"""
id: int
item_id: int
financial: Optional[LineFinancialResponse] = None
quantity: Optional[LineQuantityResponse] = None
customs: Optional[LineCustomResponse] = None
description: Optional[LineDescriptionResponse] = None
reference: Optional[LineReferenceResponse] = None
fa_data: Optional[FaLineItemResponseDTO] = None
# Fields populated from relationships
class_code: Optional[str] = None
class_description: Optional[str] = None
unit_of_measure_code: Optional[str] = None
model_config = ConfigDict(from_attributes=True)
@model_validator(mode="before")
@classmethod
def extract_relationship_info(cls, data: Any) -> Any:
"""Extract class_code, class_description and unit_of_measure_code from relationships"""
if isinstance(data, dict):
return data
# It's an ORM object
result = {}
for key in cls.model_fields.keys():
if hasattr(data, key):
result[key] = getattr(data, key)
# Map model field names to schema field names for aliased fields
if hasattr(data, "part_number"):
result["part_number_id"] = data.part_number
if hasattr(data, "component_part_number"):
result["component_part_number_id"] = data.component_part_number
# Extract class info
if hasattr(data, "class_info") and data.class_info is not None:
result["class_code"] = data.class_info.class_code
result["class_description"] = data.class_info.description_es
# Extract unit of measure code
if (
hasattr(data, "unit_of_measure_info")
and data.unit_of_measure_info is not None
):
result["unit_of_measure_code"] = data.unit_of_measure_info.code
return result

2
backend/api/v1/modules/a76/items/line_quantities/models.py
View File

@@ -7,7 +7,7 @@ from core.database import Base
from api.v1.modules.a76.general_catalogs.packages.models import Package
if TYPE_CHECKING:
from ..line_items.models import LineItem
from ..models import LineItem
class LineQuantity(Base):
"""

2
backend/api/v1/modules/a76/items/line_references/models.py
View File

@@ -4,7 +4,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
from core.database import Base
if TYPE_CHECKING:
from ..line_items.models import LineItem
from ..models import LineItem
class LineReference(Base):
"""

235
backend/api/v1/modules/a76/items/models.py
View File

@@ -3,58 +3,229 @@ Normalized Database Schema for SCAF (Fixed Assets) and SCAII (Parts Inventory)
SQLAlchemy v2 - Annex 24 Compliance
"""
from typing import Optional, TYPE_CHECKING, List
from sqlalchemy import Boolean, String, Integer, ForeignKey
from typing import Optional, TYPE_CHECKING
from core.database import Base
from decimal import Decimal
from sqlalchemy import Boolean, String, Integer, Numeric, SmallInteger, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
if TYPE_CHECKING:
from .line_items.models import LineItem
from .line_financials.models import LineFinancial
from .line_quantities.models import LineQuantity
from .line_customs.models import LineCustom
from .line_descriptions.models import LineDescription
from .line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.invoices.models import InvoiceHeader
# ============================================================================
# CORE ENTITIES
# ============================================================================
class Item(Base, TenantScopedMixin, TimestampMixin):
class LineItem(Base, TenantScopedMixin, TimestampMixin):
"""
Unified item header table for all import/export operations
Consolidates headers from both SCAF and SCAII systems
"""
__tablename__ = "items"
__tablename__ = "item_lines"
__table_args__ = {
"schema": "a76",
}
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
invoice_id: Mapped[int] = mapped_column(ForeignKey("a76.invoice_header.id")) # CONSECUTIVO
invoice_id: Mapped[int] = mapped_column(
ForeignKey("a76.invoice_header.id")
) # CONSECUTIVO
line_number: Mapped[int] = mapped_column(Integer) # LINEAIMPO/LINEAEXPO/LINEA
# Part identification
part_number_id: Mapped[Optional[int]] = mapped_column(
Integer, ForeignKey("a76.parts.id")
) # NUMPARTE
component_part_number_id: Mapped[Optional[int]] = mapped_column(
Integer, ForeignKey("a76.parts.id")
) # NUMPARTECOM
class_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.classes.id")
) # CLASE
# Unit of measure
unit_of_measure: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.units_of_measure.id")
) # UNIDADMEDIDA/UNIMED
alternate_unit: Mapped[Optional[int]] = mapped_column(
ForeignKey("a76.units_of_measure.id")
) # UNIMEDALTERNA
uma_key: Mapped[Optional[str]] = mapped_column(String(2)) # CLAVEUMA
auxiliary_unit: Mapped[Optional[str]] = mapped_column(String(5)) # UNIMEDAUXILIAR
# Permits and certificates
permit_number: Mapped[Optional[str]] = mapped_column(String(20)) # NUMPERMISO
page_line: Mapped[Optional[str]] = mapped_column(String(10)) # PAGRENGLON
has_certificate: Mapped[Optional[bool]] = mapped_column(
Boolean
) # TIENECO/CERTORIGEN
certificate_number: Mapped[Optional[str]] = mapped_column(
String(10)
) # NOCERTIFICADO
octave_permit: Mapped[Optional[str]] = mapped_column(String(20)) # PERMISOROCTAVA
permits_ped: Mapped[Optional[str]] = mapped_column(String(500)) # PERMISOSPED
# FDA
has_fda_code: Mapped[Optional[bool]] = mapped_column(Boolean) # LLEVACODFDA
fda_key: Mapped[Optional[str]] = mapped_column(String(10)) # CLAVEFDA
# Special flags
is_military_mcia: Mapped[Optional[bool]] = mapped_column(Boolean) # ESMCIAMILITAR
# IV32 (Tax identification)
iv32_type_key: Mapped[Optional[str]] = mapped_column(String(5)) # CLAVETIPOIV32
iv32_number: Mapped[Optional[str]] = mapped_column(String(35)) # NUMEROIV32
# IN CASE OF EXPO
scrap_invoice: Mapped[Optional[str]] = mapped_column(String(15)) # FACTURASCRAP
consecutive_destination: Mapped[Optional[int]] = mapped_column(
Integer
) # CONSECUTIVODES
ctm_section: Mapped[Optional[str]] = mapped_column(String(3)) # APARTADOCTM
# Tax payment
tax_payment: Mapped[Optional[bool]] = mapped_column(Boolean) # PAGOIMPUESTO
payment_method: Mapped[Optional[str]] = mapped_column(
String(9)
) # FORMAPAGO/FORMAPAGOTIGI
igi_amount: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # MONTOIGI
igi_payment_method: Mapped[Optional[str]] = mapped_column(
String(9)
) # FORMAPAGOTIGI
# FCC
fcc_key: Mapped[Optional[str]] = mapped_column(String(30)) # CLAVEFCC
# Valuation method
valuation_method: Mapped[Optional[str]] = mapped_column(String(2)) # METVALOR
valuation_determined_value: Mapped[Optional[Decimal]] = mapped_column(
Numeric(29, 8)
) # METVALORVALORDETERMINADO/METVALORACIONVALORDETERMINADO
valuation_reason: Mapped[Optional[str]] = mapped_column(
String(500)
) # METVALORMOTIVODEUSO/METVALORACIONMOTIVODEUSO
# Container rules
container_rule: Mapped[Optional[str]] = mapped_column(String(50)) # CONTENEDORREGLA
container_parts_ii: Mapped[Optional[str]] = mapped_column(
String(50)
) # CONTENEDORPARTESII
# APHIS
consecutive_aphis: Mapped[Optional[int]] = mapped_column(
Integer
) # CONSECUTIVOAPHIS
# BOM/Commercial
bom_version: Mapped[Optional[int]] = mapped_column(Integer) # VERSIONBOM
bill_version: Mapped[Optional[int]] = mapped_column(Integer) # VERSIONBILL
# TLCAN value
tlcan_value: Mapped[Optional[Decimal]] = mapped_column(Numeric(23, 8)) # VALORTLCAN
# Identifier
identifier: Mapped[Optional[str]] = mapped_column(String(2)) # IDENTIFICADOR
# Validation fields
validation_zero: Mapped[Optional[int]] = mapped_column(Integer) # VALIDACIONZERO
validation_one: Mapped[Optional[int]] = mapped_column(Integer) # VALIDACIONUNO
# Material type
material_type: Mapped[Optional[str]] = mapped_column(
String(50)
) # TIPOMAT/TIPODENUMPARTE
# Order concept
order_type: Mapped[Optional[str]] = mapped_column(String(50)) # TIPODEORDEN
line_concept: Mapped[Optional[str]] = mapped_column(
String(50)
) # CONCEPTODELAPARTIDA
# Review dispatch
review_dispatch: Mapped[Optional[str]] = mapped_column(String(10)) # REVISARDESP
# Take component from PT
take_component_pt: Mapped[Optional[int]] = mapped_column(Integer) # TOMARCOMOPT
# Pallet
pallet2: Mapped[Optional[int]] = mapped_column(SmallInteger) # PALLET2
# Wildcard field
wildcard_field: Mapped[Optional[str]] = mapped_column(String(100)) # CAMPOCOMODIN
# Item references
reference_number: Mapped[Optional[str]] = mapped_column(
String(20)) # NUMREFERENCIA
order: Mapped[Optional[str]] = mapped_column(
String(50)) # ORDENCOMPRA / ORDENVENTA
reference_number: Mapped[Optional[str]] = mapped_column(String(20)) # NUMREFERENCIA
order: Mapped[Optional[str]] = mapped_column(String(50)) # ORDENCOMPRA / ORDENVENTA
guide_number: Mapped[Optional[str]] = mapped_column(
String(50)) # NUMEROGUIA/NUMERODEGUIA
String(50)
) # NUMEROGUIA/NUMERODEGUIA
# Dates
depreciation_date: Mapped[Optional[int]] = mapped_column(
Integer) # FECHADEPRECIACION
Integer
) # FECHADEPRECIACION
# Administrative fields
rectification: Mapped[Optional[bool]] = mapped_column(
Boolean) # RECTIFICACION
rectification: Mapped[Optional[bool]] = mapped_column(Boolean) # RECTIFICACION
warehouse: Mapped[Optional[str]] = mapped_column(String(30)) # BODEGA
location: Mapped[Optional[str]] = mapped_column(
String(200)) # LOCALIZACION
location: Mapped[Optional[str]] = mapped_column(String(200)) # LOCALIZACION
# Relationships (one-to-many)
lines: Mapped[List["LineItem"]] = relationship(
"LineItem", back_populates="item", cascade="all, delete-orphan")
invoice: Mapped["InvoiceHeader"] = relationship("InvoiceHeader")
# Relationships
financial: Mapped[Optional["LineFinancial"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
quantity: Mapped[Optional["LineQuantity"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
customs: Mapped[Optional["LineCustom"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
description: Mapped[Optional["LineDescription"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
reference: Mapped[Optional["LineReference"]] = relationship(
back_populates="line", cascade="all, delete-orphan", uselist=False
)
class_info: Mapped[Optional["Class"]] = relationship(
"api.v1.modules.a76.classes.models.Class",
foreign_keys=[class_id],
viewonly=True,
)
unit_of_measure_info: Mapped[Optional["UnitOfMeasure"]] = relationship(
"api.v1.modules.a76.general_catalogs.units_of_measure.models.UnitOfMeasure",
foreign_keys=[unit_of_measure],
viewonly=True,
)
fa_data: Mapped[Optional["FaLineItem"]] = relationship(
"FaLineItem",
back_populates="master_info",
cascade="all, delete-orphan",
uselist=False,
)
part_info: Mapped[Optional["Part"]] = relationship(
"Part",
foreign_keys=[part_number_id],
viewonly=True,
)
# ============================================================================
# SUPPORTING TABLES
# ============================================================================
@@ -65,6 +236,7 @@ class PackingList(Base, TenantScopedMixin, TimestampMixin):
Packing list items
From: SPartidasPackingList
"""
__tablename__ = "packing_lists"
__table_args__ = {
"schema": "a76",
@@ -73,7 +245,8 @@ class PackingList(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
item_line_id: Mapped[int] = mapped_column(Integer) # LINEA
packing_list_number: Mapped[Optional[str]] = mapped_column(
String(100)) # NUMPACKINGLIST
String(100)
) # NUMPACKINGLIST
class CTMReceipt(Base, TenantScopedMixin, TimestampMixin):
@@ -81,6 +254,7 @@ class CTMReceipt(Base, TenantScopedMixin, TimestampMixin):
CTM Receipt lines (temporary manufacturing)
From: SPartidasReciboCTM
"""
__tablename__ = "ctm_receipts"
__table_args__ = {
"schema": "a76",
@@ -88,11 +262,11 @@ class CTMReceipt(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
receipt_line: Mapped[int] = mapped_column(
ForeignKey("a76.item_lines.id")) # LINEARECIBO
ForeignKey("a76.item_lines.id")
) # LINEARECIBO
option: Mapped[Optional[str]] = mapped_column(String(3)) # OPCION
exit_invoice: Mapped[Optional[str]] = mapped_column(
String(19)) # FACTURASALIDA
exit_invoice: Mapped[Optional[str]] = mapped_column(String(19)) # FACTURASALIDA
class SubassemblyEntry(Base, TenantScopedMixin, TimestampMixin):
@@ -100,6 +274,7 @@ class SubassemblyEntry(Base, TenantScopedMixin, TimestampMixin):
Subassembly/Submanufacturing Entry lines
From: SPartidasEntradaSM
"""
__tablename__ = "subassembly_entries"
__table_args__ = {
"schema": "a76",
@@ -107,10 +282,10 @@ class SubassemblyEntry(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
remission_line: Mapped[int] = mapped_column(Integer) # LINEAREMISION
exit_invoice: Mapped[Optional[str]] = mapped_column(
String(15)) # FACTURASALIDA
exit_invoice: Mapped[Optional[str]] = mapped_column(String(15)) # FACTURASALIDA
exit_line: Mapped[Optional[int]] = mapped_column(Integer) # LINEASALIDA
# ============================================================================
# INDEXES AND CONSTRAINTS
# ============================================================================
@@ -189,8 +364,8 @@ MIGRATION STRATEGY FROM ORIGINAL TABLES TO NORMALIZED SCHEMA:
7. QUERYING EXAMPLES:
```python
# Get all imports (both systems)
session.query(Item).filter(
Item.item_type.in_(['IMPORT', 'EQUIPMENT_IMPORT_TEMP', 'EQUIPMENT_IMPORT_DEF'])
session.query(LineItem).filter(
LineItem.item_type.in_(['IMPORT', 'EQUIPMENT_IMPORT_TEMP', 'EQUIPMENT_IMPORT_DEF'])
)
# Get all lines for a specific part across all items
@@ -199,8 +374,8 @@ MIGRATION STRATEGY FROM ORIGINAL TABLES TO NORMALIZED SCHEMA:
)
# Get SCAF equipment with depreciation
session.query(Item).join(LineItem).filter(
Item.system_origin == 'SCAF',
session.query(LineItem).join(LineItem).filter(
LineItem.system_origin == 'SCAF',
LineItem.value_depreciated_usd.isnot(None)
)
```

27
backend/api/v1/modules/a76/items/routes.py
View File

@@ -11,10 +11,10 @@ from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from .schemas import (
ItemCreate,
ItemUpdate,
ItemResponse,
ItemListResponse,
LineItemCreate,
LineItemUpdate,
LineItemResponse,
LineItemListResponse,
)
from .service import ItemService
@@ -25,9 +25,9 @@ router = APIRouter(prefix="/items", tags=["Items"])
# ITEM CRUD ENDPOINTS
# ============================================================================
@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
@router.post("/", response_model=LineItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(
item_data: ItemCreate,
item_data: LineItemCreate,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
@@ -36,7 +36,6 @@ async def create_item(
Create a new item with multiple line items and their nested data
The item follows a one-to-many relationship structure:
- Item has many LineItems
- Each LineItem has one LineFinancial
- Each LineItem has one LineQuantity
- Each LineItem has one LineCustoms
@@ -49,7 +48,7 @@ async def create_item(
return service.create(db, item_data, tenant_id, company_id)
@router.get("/{item_id}", response_model=ItemResponse)
@router.get("/{item_id}", response_model=LineItemResponse)
async def get_item(
item_id: int = Path(..., description="Item ID"),
company_id: int = Query(..., description="Company ID"),
@@ -70,7 +69,7 @@ async def get_item(
return item
@router.get("/", response_model=ItemListResponse)
@router.get("/", response_model=LineItemListResponse)
async def list_items(
company_id: int = Query(..., description="Company ID"),
skip: int = Query(0, ge=0, description="Number of records to skip"),
@@ -109,7 +108,7 @@ async def list_items(
items, total = service.get_all(
db, tenant_id, company_id, skip, limit, filters)
return ItemListResponse(
return LineItemListResponse(
total=total,
items=items,
skip=skip,
@@ -117,10 +116,10 @@ async def list_items(
)
@router.put("/{item_id}", response_model=ItemResponse)
@router.put("/{item_id}", response_model=LineItemResponse)
async def update_item(
item_id: int = Path(..., description="Item ID"),
item_data: ItemUpdate = ...,
item_data: LineItemUpdate = ...,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
@@ -168,7 +167,7 @@ async def delete_item(
# ADDITIONAL ENDPOINTS FOR INVOICE
# ============================================================================
@router.get("/invoice/{invoice_id}/items", response_model=ItemListResponse)
@router.get("/invoice/{invoice_id}/items", response_model=LineItemListResponse)
async def get_items_by_invoice(
invoice_id: int = Path(..., description="Invoice ID"),
company_id: int = Query(..., description="Company ID"),
@@ -186,7 +185,7 @@ async def get_items_by_invoice(
items, total = service.get_by_invoice(
db, invoice_id, tenant_id, company_id, skip, limit)
return ItemListResponse(
return LineItemListResponse(
total=total,
items=items,
skip=skip,

328
backend/api/v1/modules/a76/items/schemas.py
View File

@@ -1,19 +1,46 @@
"""
Schemas for Items and related entities
Complete nested one-to-one structure:
Item -> LineItem -> LineFinancial -> LineQuantity -> LineCustoms -> LineDescription -> LineReference
LineItem -> LineFinancial -> LineQuantity -> LineCustoms -> LineDescription -> LineReference
"""
from typing import Optional
from typing import Any, Optional
from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field, ConfigDict
from pydantic import BaseModel, Field, ConfigDict, model_validator
# Import schemas from individual modules
from .line_items.schemas import (
LineItemCreate,
LineItemUpdate,
LineItemResponse
# Import nested schemas
from .line_customs.schemas import (
LineCustomCreate,
LineCustomUpdate,
LineCustomResponse,
)
from .line_descriptions.schemas import (
LineDescriptionCreate,
LineDescriptionUpdate,
LineDescriptionResponse,
)
from .line_quantities.schemas import (
LineQuantityCreate,
LineQuantityUpdate,
LineQuantityResponse,
)
from .line_financials.schemas import (
LineFinancialCreate,
LineFinancialUpdate,
LineFinancialResponse,
)
from .line_references.schemas import (
LineReferenceCreate,
LineReferenceUpdate,
LineReferenceResponse,
)
from api.v1.modules.a24.fa.fa_item_lines.dto import (
FaLineItemCreateDTO,
FaLineItemUpdateDTO,
FaLineItemResponseDTO,
)
@@ -21,53 +48,294 @@ from .line_items.schemas import (
# ITEM SCHEMAS
# ============================================================================
class ItemBase(BaseModel):
class LineItemBase(BaseModel):
"""Base schema for items"""
model_config = ConfigDict(populate_by_name=True)
invoice_id: int = Field(..., description="Invoice ID")
line_number: int = Field(..., description="Line number")
# Part identification
part_number_id: Optional[int] = Field(
None,
description="Part number",
alias="part_number",
serialization_alias="part_number_id",
)
component_part_number_id: Optional[int] = Field(
None,
description="Component part number",
alias="component_part_number",
serialization_alias="component_part_number_id",
)
class_id: Optional[int] = Field(None, description="Class code")
# Unit of measure
unit_of_measure: Optional[int] = Field(None, description="Unit of measure")
alternate_unit: Optional[int] = Field(None, description="Alternate unit")
uma_key: Optional[str] = Field(None, max_length=2, description="UMA key")
auxiliary_unit: Optional[str] = Field(
None, max_length=5, description="Auxiliary unit"
)
# Permits and certificates
permit_number: Optional[str] = Field(
None, max_length=20, description="Permit number"
)
page_line: Optional[str] = Field(None, max_length=10, description="Page line")
has_certificate: Optional[bool] = Field(None, description="Has certificate")
certificate_number: Optional[str] = Field(
None, max_length=10, description="Certificate number"
)
octave_permit: Optional[str] = Field(
None, max_length=20, description="Octave permit"
)
permits_ped: Optional[str] = Field(None, max_length=500, description="PED permits")
# FDA
has_fda_code: Optional[bool] = Field(None, description="Has FDA code")
fda_key: Optional[str] = Field(None, max_length=10, description="FDA key")
# Special flags
is_military_mcia: Optional[bool] = Field(
None, description="Is military merchandise"
)
# IV32
iv32_type_key: Optional[str] = Field(
None, max_length=5, description="IV32 type key"
)
iv32_number: Optional[str] = Field(None, max_length=35, description="IV32 number")
# Export specific
scrap_invoice: Optional[str] = Field(
None, max_length=15, description="Scrap invoice"
)
consecutive_destination: Optional[int] = Field(
None, description="Consecutive destination"
)
ctm_section: Optional[str] = Field(None, max_length=3, description="CTM section")
# Tax payment
tax_payment: Optional[bool] = Field(None, description="Tax payment")
payment_method: Optional[str] = Field(
None, max_length=9, description="Payment method"
)
igi_amount: Optional[Decimal] = Field(None, description="IGI amount")
igi_payment_method: Optional[str] = Field(
None, max_length=9, description="IGI payment method"
)
# FCC
fcc_key: Optional[str] = Field(None, max_length=30, description="FCC key")
# Valuation method
valuation_method: Optional[str] = Field(
None, max_length=2, description="Valuation method"
)
valuation_determined_value: Optional[Decimal] = Field(
None, description="Valuation determined value"
)
valuation_reason: Optional[str] = Field(
None, max_length=500, description="Valuation reason"
)
# Container rules
container_rule: Optional[str] = Field(
None, max_length=50, description="Container rule"
)
container_parts_ii: Optional[str] = Field(
None, max_length=50, description="Container parts II"
)
# APHIS
consecutive_aphis: Optional[int] = Field(None, description="Consecutive APHIS")
# BOM/Commercial
bom_version: Optional[int] = Field(None, description="BOM version")
bill_version: Optional[int] = Field(None, description="Bill version")
# TLCAN value
tlcan_value: Optional[Decimal] = Field(None, description="TLCAN value")
# Identifier
identifier: Optional[str] = Field(None, max_length=2, description="Identifier")
# Validation fields
validation_zero: Optional[int] = Field(None, description="Validation zero")
validation_one: Optional[int] = Field(None, description="Validation one")
# Material type
material_type: Optional[str] = Field(
None, max_length=50, description="Material type"
)
# Order concept
order_type: Optional[str] = Field(None, max_length=50, description="Order type")
line_concept: Optional[str] = Field(None, max_length=50, description="Line concept")
# Review dispatch
review_dispatch: Optional[str] = Field(
None, max_length=10, description="Review dispatch"
)
# Take component from PT
take_component_pt: Optional[int] = Field(None, description="Take component from PT")
# Pallet
pallet2: Optional[int] = Field(None, description="Pallet 2")
# Wildcard field
wildcard_field: Optional[str] = Field(
None, max_length=100, description="Wildcard field"
)
reference_number: Optional[str] = Field(
None, max_length=20, description="Reference number")
None, max_length=20, description="Reference number"
)
order: Optional[str] = Field(None, max_length=50, description="Order")
guide_number: Optional[str] = Field(
None, max_length=50, description="Guide number")
guide_number: Optional[str] = Field(None, max_length=50, description="Guide number")
# Dates
depreciation_date: Optional[int] = Field(
None, description="Depreciation date")
depreciation_date: Optional[int] = Field(None, description="Depreciation date")
# Administrative fields
rectification: Optional[int] = Field(None, description="Rectification")
warehouse: Optional[str] = Field(
None, max_length=30, description="Warehouse")
location: Optional[str] = Field(
None, max_length=200, description="Location")
warehouse: Optional[str] = Field(None, max_length=30, description="Warehouse")
location: Optional[str] = Field(None, max_length=200, description="Location")
class ItemCreate(ItemBase):
class LineItemCreate(LineItemBase):
"""Schema for creating item with nested lines (one-to-many)"""
lines: Optional[list[LineItemCreate]] = Field(
default=[], description="List of line items")
# Override base fields - estos se asignan automáticamente en el service
line_number: Optional[int] = Field(None, description="Line number (auto-assigned)")
financial: Optional[LineFinancialCreate] = Field(
None, description="Financial data for this line"
)
quantity: Optional[LineQuantityCreate] = Field(
None, description="Quantity data for this line"
)
customs: Optional[LineCustomCreate] = Field(
None, description="Customs data for this line"
)
description: Optional[LineDescriptionCreate] = Field(
None, description="Description data for this line"
)
reference: Optional[LineReferenceCreate] = Field(
None, description="Reference data for this line"
)
fa_data: Optional[FaLineItemCreateDTO] = Field(
None, description="Fixed Asset data for this line"
)
class ItemUpdate(ItemBase):
class LineItemUpdate(LineItemBase):
"""Schema for updating item"""
invoice_id: Optional[int] = Field(None, description="Invoice ID")
lines: Optional[list[LineItemUpdate]] = Field(
None, description="List of line items to update")
# Override base fields - todos opcionales en updates
line_number: Optional[int] = Field(None, description="Line number")
financial: Optional[LineFinancialUpdate] = Field(
None, description="Financial data for this line"
)
quantity: Optional[LineQuantityUpdate] = Field(
None, description="Quantity data for this line"
)
customs: Optional[LineCustomUpdate] = Field(
None, description="Customs data for this line"
)
description: Optional[LineDescriptionUpdate] = Field(
None, description="Description data for this line"
)
reference: Optional[LineReferenceUpdate] = Field(
None, description="Reference data for this line"
)
fa_data: Optional[FaLineItemUpdateDTO] = Field(
None, description="Fixed Asset data for this line"
)
class ItemResponse(ItemBase):
"""Schema for item response with nested data (one-to-many)"""
class LineItemResponse(LineItemBase):
"""Schema for single item response"""
id: int
lines: list[LineItemResponse] = Field(
default=[], description="List of line items")
invoice_id: int
line_number: int
model_config = ConfigDict(from_attributes=True)
# Part identification
part_number_id: Optional[int] = Field(
None, alias="part_number", serialization_alias="part_number_id"
)
component_part_number_id: Optional[int] = Field(
None,
alias="component_part_number",
serialization_alias="component_part_number_id",
)
class_id: Optional[int] = None
# Nested data
financial: Optional[LineFinancialResponse] = None
quantity: Optional[LineQuantityResponse] = None
customs: Optional[LineCustomResponse] = None
description: Optional[LineDescriptionResponse] = None
reference: Optional[LineReferenceResponse] = None
fa_data: Optional[FaLineItemResponseDTO] = None
# Fields populated from relationships
class_code: Optional[str] = None
class_description: Optional[str] = None
unit_of_measure_code: Optional[str] = None
# Additional fields that might be present
reference_number: Optional[str] = None
order: Optional[str] = None
warehouse: Optional[str] = None
location: Optional[str] = None
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
@model_validator(mode="before")
@classmethod
def extract_relationship_info(cls, data: Any) -> Any:
"""Extract class_code, class_description and unit_of_measure_code from relationships"""
if isinstance(data, dict):
return data
# It's an ORM object
result = {}
for key in cls.model_fields.keys():
if hasattr(data, key):
result[key] = getattr(data, key)
# Map model field names to schema field names for aliased fields
if hasattr(data, "part_number"):
result["part_number_id"] = data.part_number
if hasattr(data, "component_part_number"):
result["component_part_number_id"] = data.component_part_number
# Extract class info
if hasattr(data, "class_info") and data.class_info is not None:
result["class_code"] = data.class_info.class_code
result["class_description"] = data.class_info.description_es
# Extract unit of measure code
if (
hasattr(data, "unit_of_measure_info")
and data.unit_of_measure_info is not None
):
result["unit_of_measure_code"] = data.unit_of_measure_info.code
return result
class ItemListResponse(BaseModel):
class LineItemListResponse(BaseModel):
"""Schema for paginated item list"""
total: int = Field(..., description="Total number of items")
items: list[ItemResponse] = Field(..., description="List of items")
items: list[LineItemResponse] = Field(..., description="List of items")
skip: int = Field(..., description="Number of skipped items")
limit: int = Field(..., description="Maximum items per page")

446
backend/api/v1/modules/a76/items/service.py
View File

@@ -1,12 +1,15 @@
"""
Service layer for Items business logic
Handles CRUD operations for Item with complete one-to-one relationships:
Item -> LineItem -> LineFinancial
-> LineQuantity
-> LineCustoms
-> LineDescription
-> LineReference
-> FaLineItem (Fixed Assets - a24)
Handles CRUD operations for LineItem with complete one-to-one relationships:
LineItem -> LineFinancial
-> LineQuantity
-> LineCustoms
-> LineDescription
-> LineReference
-> FaLineItem (Fixed Assets - a24)
After refactoring: LineItem is the main entity, representing a single line item in an invoice.
There is no intermediate Item entity anymore. Each LineItem belongs directly to an InvoiceHeader.
"""
import logging
@@ -24,17 +27,14 @@ from core.exceptions import ErrorCollector
from .imports.temporary.validators.create import validate_create
from .imports.temporary.validators.update import validate_update
from api.v1.modules.a76.items.line_items.schemas import LineItemCreate, LineItemUpdate
from .schemas import ItemCreate, ItemUpdate
from .line_items.models import LineItem
from .schemas import LineItemCreate, LineItemUpdate
from .line_financials.models import LineFinancial
from .line_quantities.models import LineQuantity
from .line_customs.models import LineCustom
from .line_descriptions.models import LineDescription
from .line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from .models import Item
from .models import LineItem
from api.v1.modules.a76.invoices.models import InvoiceHeader
logger = logging.getLogger(__name__)
@@ -52,8 +52,7 @@ class ItemService:
max_line = (
db.query(func.max(LineItem.line_number))
.join(Item, LineItem.item_id == Item.id)
.filter(Item.invoice_id == invoice_id)
.filter(LineItem.invoice_id == invoice_id)
.scalar()
)
@@ -62,12 +61,15 @@ class ItemService:
@staticmethod
def _renumber_all_invoice_lines(db: Session, invoice_id: int) -> None:
"""Renumber all line_items for a given invoice to be consecutive (1, 2, 3, ...)."""
items = db.query(Item).filter(Item.invoice_id == invoice_id).all()
all_lines = [line for item in items for line in item.lines]
all_lines.sort(key=lambda x: x.line_number if x.line_number else 0)
items = (
db.query(LineItem)
.filter(LineItem.invoice_id == invoice_id)
.order_by(LineItem.line_number)
.all()
)
for idx, line in enumerate(all_lines, start=1):
line.line_number = idx
for idx, item in enumerate(items, start=1):
item.line_number = idx
@staticmethod
def _lock_invoice(
@@ -143,24 +145,24 @@ class ItemService:
@staticmethod
def get_by_id(
db: Session, item_id: int, tenant_id: int, company_id: int
) -> Optional[Item]:
) -> Optional[LineItem]:
"""Get an item by ID with tenant/company validation"""
return (
db.query(Item)
db.query(LineItem)
.options(
joinedload(Item.lines).joinedload(LineItem.financial),
joinedload(Item.lines).joinedload(LineItem.quantity),
joinedload(Item.lines).joinedload(LineItem.customs),
joinedload(Item.lines).joinedload(LineItem.description),
joinedload(Item.lines).joinedload(LineItem.reference),
joinedload(Item.lines).joinedload(LineItem.class_info),
joinedload(Item.lines).joinedload(LineItem.unit_of_measure_info),
joinedload(Item.lines).joinedload(LineItem.fa_data),
joinedload(LineItem.financial),
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.description),
joinedload(LineItem.reference),
joinedload(LineItem.class_info),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.fa_data),
)
.filter(
Item.id == item_id,
Item.tenant_id == tenant_id,
Item.company_id == company_id,
LineItem.id == item_id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
.first()
)
@@ -173,42 +175,42 @@ class ItemService:
skip: int = 0,
limit: int = 100,
filters: Optional[dict] = None,
) -> Tuple[List[Item], int]:
) -> Tuple[List[LineItem], int]:
"""Get all items for a tenant/company with pagination and optional filters"""
query = (
db.query(Item)
db.query(LineItem)
.options(
joinedload(Item.lines).joinedload(LineItem.financial),
joinedload(Item.lines).joinedload(LineItem.quantity),
joinedload(Item.lines).joinedload(LineItem.customs),
joinedload(Item.lines).joinedload(LineItem.description),
joinedload(Item.lines).joinedload(LineItem.reference),
joinedload(Item.lines).joinedload(LineItem.class_info),
joinedload(Item.lines).joinedload(LineItem.unit_of_measure_info),
joinedload(Item.lines).joinedload(LineItem.fa_data),
joinedload(LineItem.financial),
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.description),
joinedload(LineItem.reference),
joinedload(LineItem.class_info),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.fa_data),
)
.filter(
Item.tenant_id == tenant_id,
Item.company_id == company_id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
)
# Apply filters if provided
if filters:
if filters.get("invoice_id"):
query = query.filter(Item.invoice_id == filters["invoice_id"])
query = query.filter(LineItem.invoice_id == filters["invoice_id"])
if filters.get("item_type"):
query = query.filter(Item.item_type == filters["item_type"])
query = query.filter(LineItem.item_type == filters["item_type"])
if filters.get("system_origin"):
query = query.filter(Item.system_origin == filters["system_origin"])
query = query.filter(LineItem.system_origin == filters["system_origin"])
if filters.get("search"):
search_term = f"%{filters['search']}%"
query = query.filter(
or_(
Item.invoice_id.ilike(search_term),
Item.reference_number.ilike(search_term),
Item.order.ilike(search_term),
Item.guide_number.ilike(search_term),
LineItem.invoice_id.ilike(search_term),
LineItem.reference_number.ilike(search_term),
LineItem.order.ilike(search_term),
LineItem.guide_number.ilike(search_term),
)
)
@@ -224,22 +226,22 @@ class ItemService:
company_id: int,
skip: int = 0,
limit: int = 100,
) -> Tuple[List[Item], int]:
) -> Tuple[List[LineItem], int]:
"""Get all items for a specific invoice"""
query = (
db.query(Item)
db.query(LineItem)
.options(
joinedload(Item.lines).joinedload(LineItem.financial),
joinedload(Item.lines).joinedload(LineItem.quantity),
joinedload(Item.lines).joinedload(LineItem.customs),
joinedload(Item.lines).joinedload(LineItem.description),
joinedload(Item.lines).joinedload(LineItem.reference),
joinedload(Item.lines).joinedload(LineItem.fa_data),
joinedload(LineItem.financial),
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.description),
joinedload(LineItem.reference),
joinedload(LineItem.fa_data),
)
.filter(
Item.invoice_id == invoice_id,
Item.tenant_id == tenant_id,
Item.company_id == company_id,
LineItem.invoice_id == invoice_id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
)
@@ -250,10 +252,10 @@ class ItemService:
@staticmethod
def create(
db: Session,
item_data: ItemCreate,
item_data: LineItemCreate,
tenant_id: int,
company_id: int,
) -> Item:
) -> LineItem:
"""Create a new item with all related nested data (multiple lines)"""
# Validaciones con ErrorCollector
@@ -271,109 +273,67 @@ class ItemService:
if not invoice_updated(db, item_data.invoice_id, tenant_id, company_id, errors):
errors.raise_if_errors("Error al crear el item")
# Lock invoice and pre-calculate line_numbers
# Lock invoice and calculate line number
if not ItemService._lock_invoice(
db, item_data.invoice_id, tenant_id, company_id, errors
):
errors.raise_if_errors("Error al crear el item")
line_numbers = []
if item_data.lines:
starting_line = ItemService._get_next_line_number(db, item_data.invoice_id)
line_numbers = [starting_line + i for i in range(len(item_data.lines))]
# Calculate the next line number for this single item
line_number = ItemService._get_next_line_number(db, item_data.invoice_id)
# Validar cada line item que se va a crear
if item_data.lines:
for idx, line_data in enumerate(item_data.lines):
line_number = line_numbers[idx] # Usar el line_number calculado
# Validar el item
validate_create(
db,
item_data, # Schema Pydantic completo
tenant_id,
company_id,
errors,
line_number,
)
validate_create(
db,
line_data, # Schema Pydantic completo
item_data.invoice_id, # invoice_id
tenant_id,
company_id,
errors,
line_number,
)
# Validaciones adicionales específicas del negocio
if item_data.fa_data and item_data.fa_data.is_subitem is None:
errors.add_required_error(field=f"fa_data.is_subitem")
# Validaciones adicionales específicas del negocio
if line_data.fa_data and line_data.fa_data.is_subitem is None:
errors.add_required_error(
field=f"lines[{line_number}].fa_data.is_subitem"
)
if item_data.fa_data and item_data.fa_data.subitem_number is None:
errors.add_required_error(field=f"fa_data.subitem_number")
if line_data.fa_data and line_data.fa_data.subitem_number is None:
errors.add_required_error(
field=f"lines[{line_number}].fa_data.subitem_number"
)
# Validar apóstrofes en número de parte
if line_data.part_number_id and "'" in str(line_data.part_number_id):
errors.add_error(
field=f"lines[{line_number}].part_number",
message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos",
code="WARNING_APOSTROPHE",
)
# Si hay errores, lanzar excepción ANTES de intentar crear
errors.raise_if_errors("Error al crear el item")
try:
# Extract lines data
lines_data = item_data.lines or []
item_dict = item_data.model_dump(exclude={"lines"})
# Prepare item data
item_dict = item_data.model_dump(
exclude={
"financial",
"quantity",
"customs",
"description",
"reference",
"fa_data",
}
)
# Add tenant and company
item_dict["tenant_id"] = tenant_id
item_dict["company_id"] = company_id
# Add tenant, company and line number
item_dict.update(
{
"tenant_id": tenant_id,
"company_id": company_id,
"line_number": line_number,
}
)
# Create the item
db_item = Item(**item_dict)
db_item = LineItem(**item_dict)
db.add(db_item)
db.flush() # Get the item ID
# Create line items if provided
for idx, line_data in enumerate(lines_data):
line_dict = line_data.model_dump(
exclude={
"financial",
"quantity",
"customs",
"description",
"reference",
"fa_data",
}
)
line_dict.update(
{
"item_id": db_item.id,
"tenant_id": tenant_id,
"company_id": company_id,
"line_number": (
line_numbers[idx]
if line_numbers
else ItemService._get_next_line_number(
db, item_data.invoice_id
)
),
}
)
# Map schema field names to model field names
line_dict["part_number"] = line_dict.pop("part_number_id", None)
line_dict["component_part_number"] = line_dict.pop(
"component_part_number_id", None
)
db_line = LineItem(**line_dict)
db.add(db_line)
db.flush()
# Create all nested data
ItemService._create_line_nested_data(
db, db_line, line_data, tenant_id, company_id
)
# Create all nested data
ItemService._create_line_nested_data(
db, db_item, item_data, tenant_id, company_id
)
db.commit()
db.refresh(db_item)
@@ -384,27 +344,27 @@ class ItemService:
logger.error(f"Error creating item: {e}")
raise HTTPException(
status_code=400,
detail="Item creation failed - integrity constraint violated",
detail="LineItem creation failed - integrity constraint violated",
)
except Exception as e:
db.rollback()
logger.error(f"Unexpected error creating item: {e}")
raise HTTPException(status_code=500, detail="Error creating item")
logger.error(f"Unexpected error creating LineItem: {e}")
raise HTTPException(status_code=500, detail="Error creating LineItem")
@staticmethod
def update(
db: Session,
item_id: int,
item_data: ItemUpdate,
item_data: LineItemUpdate,
tenant_id: int,
company_id: int,
) -> Item:
) -> LineItem:
"""Update an item and optionally its nested data (multiple lines)"""
# Get existing item
db_item = ItemService.get_by_id(db, item_id, tenant_id, company_id)
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
raise HTTPException(status_code=404, detail="LineItem not found")
# Validaciones con ErrorCollector
errors = ErrorCollector()
@@ -418,143 +378,85 @@ class ItemService:
):
errors.raise_if_errors("Error al actualizar el item")
# Pre-calcular line_numbers para cada línea (en update, las líneas se renumeran desde 1)
line_numbers = []
if item_data.lines:
line_numbers = [i + 1 for i in range(len(item_data.lines))]
# Validar el item que se va a actualizar
validate_update(
db,
item_data, # Schema de update
db_item, # LineItem existente en DB
tenant_id,
company_id,
errors,
db_item.line_number,
)
# Validar cada line item que se va a actualizar
if item_data.lines:
for idx, line_data in enumerate(item_data.lines):
line_number = line_numbers[idx] # Usar el line_number calculado
# Si el line tiene ID, es actualización; si no, es creación
if hasattr(line_data, "id") and line_data.id:
# Buscar el line item existente
existing_line = next(
(line for line in db_item.lines if line.id == line_data.id),
None,
)
if existing_line:
# Validar update con línea existente
validate_update(
db,
line_data, # Schema de update
existing_line, # LineItem existente en DB
invoice_id_to_lock, # invoice_id
tenant_id,
company_id,
errors,
line_number,
)
else:
# Es un nuevo line item, validar como creación
validate_create(
db,
line_data, # Schema Pydantic completo
invoice_id_to_lock, # invoice_id
tenant_id,
company_id,
errors,
line_number,
)
# Validar tipo de partida
if hasattr(item_data, "item_type") and item_data.item_type:
tipo_partida = item_data.item_type
if tipo_partida and tipo_partida not in ["N", "S"]:
errors.add_error(
field=f"item_type",
message=f"Tipo de partida debe ser 'N' (Normal) o 'S' (Subpartida), recibido: '{tipo_partida}'",
solution=None,
code="INVALID_ITEM_TYPE",
value=str(tipo_partida),
)
# Validaciones adicionales específicas del negocio
# (Aplican tanto para crear como actualizar)
# Validar apóstrofes en número de parte
if line_data.part_number_id and "'" in str(line_data.part_number_id):
# Si es subpartida (S), debe tener partida principal
if tipo_partida == "S":
if not hasattr(item_data, "main_line_id") or not item_data.main_line_id:
errors.add_error(
field=f"lines[{line_number}].part_number",
message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos",
field=f"main_line_id",
message="Las subpartidas (tipo 'S') deben tener una partida principal",
solution=None,
code="WARNING_APOSTROPHE",
code="MISSING_MAIN_LINE",
)
# Validar tipo de partida
if hasattr(line_data, "item_type"):
tipo_partida = line_data.item_type
if tipo_partida and tipo_partida not in ["N", "S"]:
errors.add_error(
field=f"lines[{line_number}].item_type",
message=f"Tipo de partida debe ser 'N' (Normal) o 'S' (Subpartida), recibido: '{tipo_partida}'",
solution=None,
code="INVALID_ITEM_TYPE",
value=str(tipo_partida),
)
# Si es subpartida (S), debe tener partida principal
if tipo_partida == "S":
if (
not hasattr(line_data, "main_line_id")
or not line_data.main_line_id
):
errors.add_error(
field=f"lines[{line_number}].main_line_id",
message="Las subpartidas (tipo 'S') deben tener una partida principal",
solution=None,
code="MISSING_MAIN_LINE",
)
# Si hay errores, lanzar excepción ANTES de actualizar
errors.raise_if_errors("Error al actualizar el item")
try:
# Extract lines data
lines_data = item_data.lines
item_dict = item_data.model_dump(exclude={"lines"}, exclude_unset=True)
# Get item data excluding nested objects
item_dict = item_data.model_dump(
exclude={
"financial",
"quantity",
"customs",
"description",
"reference",
"fa_data",
},
exclude_unset=True,
)
# Update item fields
for key, value in item_dict.items():
setattr(db_item, key, value)
# Update lines if provided (replace all lines)
if lines_data is not None:
# Delete existing lines (cascade will handle nested data)
for existing_line in db_item.lines:
db.delete(existing_line)
db.flush()
# Delete existing nested data
db.query(LineFinancial).filter(
LineFinancial.item_line_id == db_item.id
).delete()
db.query(LineQuantity).filter(
LineQuantity.item_line_id == db_item.id
).delete()
db.query(LineCustom).filter(LineCustom.item_line_id == db_item.id).delete()
db.query(LineDescription).filter(
LineDescription.item_line_id == db_item.id
).delete()
db.query(LineReference).filter(
LineReference.item_line_id == db_item.id
).delete()
db.query(FaLineItem).filter(FaLineItem.id == db_item.id).delete()
db.flush()
# Create new lines
for idx, line_data in enumerate(lines_data):
line_dict = line_data.model_dump(
exclude={
"financial",
"quantity",
"customs",
"description",
"reference",
"fa_data",
},
exclude_unset=True,
)
line_dict.update(
{
"item_id": db_item.id,
"tenant_id": tenant_id,
"company_id": company_id,
"line_number": idx + 1,
}
)
# Create new nested data
ItemService._create_line_nested_data(
db, db_item, item_data, tenant_id, company_id
)
# Map schema field names to model field names
line_dict["part_number"] = line_dict.pop("part_number_id", None)
line_dict["component_part_number"] = line_dict.pop(
"component_part_number_id", None
)
db_line = LineItem(**line_dict)
db.add(db_line)
db.flush()
# Create all nested data
ItemService._create_line_nested_data(
db, db_line, line_data, tenant_id, company_id
)
# Renumber all lines for this invoice to ensure consecutive numbering
ItemService._renumber_all_invoice_lines(db, db_item.invoice_id)
# Renumber all lines for this invoice to ensure consecutive numbering
ItemService._renumber_all_invoice_lines(db, db_item.invoice_id)
db.commit()
db.refresh(db_item)

45
backend/api/v1/modules/a76/reports/exportacion/aviso_consolidado/service.py
View File

@@ -17,7 +17,7 @@ from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
from api.v1.modules.a76.pedmientos.models.pedimento_validation import PedimentoValidation
from api.v1.modules.a76.customs_brokers.models import CustomsBroker, CustomsBrokerPersonnel
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import ClientProvider, ClientProviderAddress, ClientProviderPrograms
# --- SCHEMAS FOR TEMPLATE CONTEXT ---
@@ -186,7 +186,7 @@ class AvisoConsolidadoExportacionService:
logistics = header.logistics
# Fetch Items associated with this invoice (MOVED UP FOR WEIGHT CALCULATION)
items = db.query(Item).filter(Item.invoice_id == invoice_id).all()
items = db.query(LineItem).filter(LineItem.invoice_id == invoice_id).all()
# Peso Bruto
peso_bruto_val = "0.0"
@@ -194,14 +194,12 @@ class AvisoConsolidadoExportacionService:
# Calculate sum from items first
if items:
for item in items:
if item.lines:
for line in item.lines:
if line.quantity and line.quantity.gross_weight:
try:
calculated_gross_weight += float(line.quantity.gross_weight)
except (ValueError, TypeError):
pass
for item in items:
if item.quantity and item.quantity.gross_weight:
try:
calculated_gross_weight += float(item.quantity.gross_weight)
except (ValueError, TypeError):
pass
if financials and financials.gross_weight and float(financials.gross_weight) > 0:
peso_bruto_val = f"{financials.gross_weight:,.2f}"
@@ -379,20 +377,19 @@ class AvisoConsolidadoExportacionService:
cant_total = 0.0
if items:
for item in items:
if item.lines:
for line in item.lines:
# Priority: Quantity (UMA or Standard)
q = 0.0
if line.quantity:
try:
if line.quantity.quantity_uma is not None:
q = float(line.quantity.quantity_uma)
elif line.quantity.quantity is not None:
q = float(line.quantity.quantity)
except (ValueError, TypeError):
q = 0.0
cant_total += q
for item in items:
for line in item:
# Priority: Quantity (UMA or Standard)
q = 0.0
if line.quantity:
try:
if line.quantity.quantity_uma is not None:
q = float(line.quantity.quantity_uma)
elif line.quantity.quantity is not None:
q = float(line.quantity.quantity)
except (ValueError, TypeError):
q = 0.0
cant_total += q
# Format: 15 chars, 3 decimals? Actually Clarion LINEPRINT usually just prints the text.
# Clarion 'CLIP(FORMAT(Loc:CantTotal,@n015.3))' removes spaces.

22
backend/api/v1/modules/a76/reports/exportacion/descargo/service.py
View File

@@ -15,10 +15,9 @@ from datetime import datetime
# --- MODELOS (Imported from system for Header info) ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates
from api.v1.modules.core.tenants.models import Tenant
@@ -40,8 +39,8 @@ class FIFOAssignmentService:
Returns a list of calculated discharges.
"""
# 1. Get Export Lines
export_lines = db.query(LineItem).join(Item).filter(
Item.invoice_id == invoice_id
export_lines = db.query(LineItem).join(LineItem).filter(
LineItem.invoice_id == invoice_id
).options(
joinedload(LineItem.quantity),
joinedload(LineItem.description),
@@ -58,7 +57,7 @@ class FIFOAssignmentService:
if qty_needed <= 0:
continue
part_number = exp_line.part_number
part_number = exp_line.part_number_id
if not part_number:
self._log(f"Skipping line {exp_line.id}, no part number")
continue
@@ -79,10 +78,10 @@ class FIFOAssignmentService:
# 2. Find Import Candidates (FIFO order by payment date)
# Use outerjoin for pedimento dates to avoid filtering out candidates with missing dates
candidates = db.query(LineItem).join(Item).join(InvoiceHeader)\
candidates = db.query(LineItem).join(InvoiceHeader)\
.join(InvoiceComplianceMx).join(InvoiceComplianceMx.pedimento).outerjoin(Pedimentos.pedimento_dates)\
.filter(
LineItem.part_number == part_number,
LineItem.part_number_id == part_number,
InvoiceHeader.operation_type == 'imp', # Assuming 'imp' is the value for Import based on Enum
).order_by(
PedimentoDates.payment_date.asc()
@@ -90,7 +89,7 @@ class FIFOAssignmentService:
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.financial),
joinedload(LineItem.item).joinedload(Item.invoice).joinedload(InvoiceHeader.compliance_mx).joinedload(InvoiceComplianceMx.pedimento).joinedload(Pedimentos.pedimento_dates)
joinedload(LineItem.item).joinedload(LineItem.invoice).joinedload(InvoiceHeader.compliance_mx).joinedload(InvoiceComplianceMx.pedimento).joinedload(Pedimentos.pedimento_dates)
).all()
self._log(f"Found {len(candidates)} candidates for {part_number}")
@@ -259,16 +258,15 @@ class DescargaReportService:
# --- 2. Obtener Líneas de Exportación (Lo que necesitamos cubrir) ---
if progress_callback: progress_callback(20, "Obteniendo items a exportar...")
export_lines = db.query(LineItem).filter(
LineItem.item_id == Item.id,
Item.invoice_id == invoice_id
export_lines = db.query(LineItem).filter(
LineItem.invoice_id == invoice_id
).options(
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.description),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.part_info)
).join(Item).all()
).join(LineItem).all()
items_reporte = []

11
backend/api/v1/modules/a76/reports/importacion/consolidados/mex/service.py
View File

@@ -17,7 +17,7 @@ from api.v1.modules.a76.invoices.models import (
)
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.line_items.models import LineItem
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider,
ClientProviderAddress,
@@ -27,7 +27,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -482,8 +482,7 @@ class ConsolidadoImportacionMexService:
lines = (
db.query(LineItem)
.join(Item, LineItem.item_id == Item.id)
.filter(Item.invoice_id.in_(target_invoice_ids))
.filter(LineItem.invoice_id.in_(target_invoice_ids))
.all()
)
@@ -532,7 +531,7 @@ class ConsolidadoImportacionMexService:
.filter(LineFinancial.item_line_id == line.id)
.first()
)
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
# --- Resolver Identificadores (MOVED INSIDE MAIN LOOP) ---
us_fraction_raw = ""
@@ -559,7 +558,7 @@ class ConsolidadoImportacionMexService:
# --- Multi-Currency Normalization Logic ---
# Determine Line Currency context
# Use manual lookup instead of specific attribute
invoice_id = line.item.invoice_id if line.item else None
invoice_id = line.invoice_id
line_invoice = invoice_map.get(invoice_id) if invoice_id else None
line_currency_is_mxn = False

11
backend/api/v1/modules/a76/reports/importacion/consolidados/temporary/mex/service.py
View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics, InvoiceComplianceMx
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.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
)
@@ -21,7 +20,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -314,8 +313,8 @@ class ConsolidadoImportacionMexService:
# NOT consolidating all invoices from the same Pedimento.
target_invoice_ids = [header.id]
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(
Item.invoice_id.in_(target_invoice_ids)
lines = db.query(LineItem).filter(
LineItem.invoice_id.in_(target_invoice_ids)
).all()
partidas_list = []
@@ -350,7 +349,7 @@ class ConsolidadoImportacionMexService:
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
fin = db.query(LineFinancial).filter(LineFinancial.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
# --- Resolver Identificadores (MOVED INSIDE MAIN LOOP) ---
us_fraction_raw = ""
@@ -371,7 +370,7 @@ class ConsolidadoImportacionMexService:
# --- Multi-Currency Normalization Logic ---
# Determine Line Currency context
# Use manual lookup instead of specific attribute
invoice_id = line.item.invoice_id if line.item else None
invoice_id = line.invoice_id
line_invoice = invoice_map.get(invoice_id) if invoice_id else None
line_currency_is_mxn = False

17
backend/api/v1/modules/a76/reports/importacion/facturas/mex/service.py
View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
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.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider,
ClientProviderAddress,
@@ -23,7 +22,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -36,6 +35,7 @@ from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models impor
# --- MODELO DE UNIDADES DE MEDIDA ---
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.packages.models import Package
# --- SCHEMAS ---
from .schemas import (
@@ -422,9 +422,8 @@ class FacturaImportacionMexService:
if progress_callback:
progress_callback(50, "Procesando partidas...")
lines = (
db.query(LineItem)
.join(Item, LineItem.item_id == Item.id)
.filter(Item.invoice_id == header.id)
db.query(LineItem)
.filter(LineItem.invoice_id == header.id)
.all()
)
partidas_list = []
@@ -440,10 +439,10 @@ class FacturaImportacionMexService:
.filter(LineFinancial.item_line_id == line.id)
.first()
)
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
desc_final = "S/D"
num_parte_final = str(line.part_number or "S/N")
num_parte_final = str(line.part_number_id or "S/N")
fraccion_raw = ""
origen_final = "MEX"
@@ -533,7 +532,7 @@ class FacturaImportacionMexService:
if uom:
unidad_desc = uom.description or uom.code
else:
unidad_desc = ""
unidad_desc = ""
partidas_list.append(
PartidaSchema(
@@ -552,7 +551,7 @@ class FacturaImportacionMexService:
if qty and qty.package_quantity
else 0
),
clave_bultos=(qty.package_key or "") if qty else "",
clave_bultos=(qty.package_info.key if (qty and qty.package_info) else ""),
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(
qty.gross_weight if qty else 0

11
backend/api/v1/modules/a76/reports/importacion/facturas/temporary/mex/service.py
View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
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.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
)
@@ -21,7 +20,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -227,16 +226,16 @@ class FacturaImportacionMexService:
)
if progress_callback: progress_callback(50, "Procesando partidas...")
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(Item.invoice_id == header.id).all()
lines = db.query(LineItem).filter(LineItem.invoice_id == header.id).all()
partidas_list = []
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
fin = db.query(LineFinancial).filter(LineFinancial.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
desc_final = "S/D"
num_parte_final = str(line.part_number or "S/N")
num_parte_final = str(line.part_number_id or "S/N")
fraccion_raw = ""
origen_final = "MEX"
@@ -316,7 +315,7 @@ class FacturaImportacionMexService:
cantidad_importacion=self.formatear_numero(qty.quantity if qty else 0),
unidad_medida=qty.weight_unit if qty else "PZA",
cantidad_bultos=int(qty.package_quantity) if qty and qty.package_quantity else 0,
clave_bultos=(qty.package_key or "") if qty else "",
clave_bultos=(qty.package_info.key if (qty and qty.package_info) else ""),
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(qty.gross_weight if qty else 0),
valor_costo_unitario=self.formatear_numero(v_unitario),

544
backend/api/v1/modules/a76/reports/importacion/facturas/usa/service.py
View File

@@ -10,18 +10,19 @@ from fastapi import HTTPException
from sqlalchemy.orm import Session
# --- MODELOS ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
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.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
ClientProvider,
ClientProviderAddress,
ClientProviderPrograms,
)
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -30,32 +31,38 @@ from api.v1.modules.a76.transportation.trailers.models import Trailer
from api.v1.modules.a76.transportation.drivers.models import Driver
# --- MODELO DE FRACCIONES ---
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import TariffFraction
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import (
TariffFraction,
)
# --- SCHEMAS ---
# Reuse schemas from neighbor package as they fit the same data structure
from ..mex.schemas import (
ClienteSchema, PartidaSchema, TotalesSchema,
FacturaSchema, FacturaImportacionCompleta
ClienteSchema,
PartidaSchema,
TotalesSchema,
FacturaSchema,
FacturaImportacionCompleta,
)
class FacturaImportacionUsaService:
def __init__(self):
self.template_dir = Path(__file__).parent.parent / "templates"
self.jinja_env = Environment(
loader=FileSystemLoader(self.template_dir),
autoescape=select_autoescape(['html', 'xml'])
autoescape=select_autoescape(["html", "xml"]),
)
self.template = self.jinja_env.get_template('factura_usa_ver.html')
self.template = self.jinja_env.get_template("factura_usa_ver.html")
def _get_document_title(self, invoice_type: str, is_american: bool = True) -> str:
"""
Determina el título del documento basado en el tipo de factura.
Args:
invoice_type: Tipo de factura (TEM, DEF, MEX, CR)
is_american: Si es factura americana (True) o mexicana (False)
Returns:
Título formateado para la factura
"""
@@ -66,7 +73,7 @@ class FacturaImportacionUsaService:
"TEM": "Importación Temporal",
"CR": "Importación de Cambio de Régimen",
}
# Mapeo para facturas americanas
american_titles = {
"MEX": "Mexican Purchases Import Invoice",
@@ -74,18 +81,18 @@ class FacturaImportacionUsaService:
"TEM": "Temporary Importation",
"CR": "Regime Change Importation",
}
# Seleccionar el mapa correcto
titles = american_titles if is_american else mexican_titles
# Obtener el título (normalizar a mayúsculas)
invoice_type_upper = invoice_type.upper() if invoice_type else ""
title = titles.get(invoice_type_upper, "")
# Fallback a genéricos si no se encuentra
if not title:
return "Commercial Invoice" if is_american else "Factura de Importación"
return title
def _get_wkhtmltopdf_config(self):
@@ -95,28 +102,49 @@ class FacturaImportacionUsaService:
return pdfkit.configuration(wkhtmltopdf=path)
def formatear_numero(self, valor, decimales: int = 2):
if valor is None: return 0.0
if valor is None:
return 0.0
try:
return round(float(valor), decimales)
except: return 0.0
except:
return 0.0
def _format_fraccion_fallback(self, fraccion_raw: str) -> str:
if not fraccion_raw or len(fraccion_raw) < 8:
return fraccion_raw
return f"{fraccion_raw[:4]}.{fraccion_raw[4:6]}.{fraccion_raw[6:]}"
def _obtener_datos_cliente(self, db: Session, client_id: int, rol: str) -> ClienteSchema:
def _obtener_datos_cliente(
self, db: Session, client_id: int, rol: str
) -> ClienteSchema:
main = db.query(ClientProvider).filter(ClientProvider.id == client_id).first()
if not main:
return ClienteSchema(header=rol, nombre="Unknown", direccion="", tax_id="", codigo_postal="", ciudad="", estado="", pais="USA")
addr = db.query(ClientProviderAddress).filter(ClientProviderAddress.client_id == client_id).first()
prog = db.query(ClientProviderPrograms).filter(ClientProviderPrograms.client_id == client_id).first()
return ClienteSchema(
header=rol,
nombre="Unknown",
direccion="",
tax_id="",
codigo_postal="",
ciudad="",
estado="",
pais="USA",
)
addr = (
db.query(ClientProviderAddress)
.filter(ClientProviderAddress.client_id == client_id)
.first()
)
prog = (
db.query(ClientProviderPrograms)
.filter(ClientProviderPrograms.client_id == client_id)
.first()
)
return ClienteSchema(
header=rol,
nombre=(main.name or main.short_name) or "N/A",
direccion=(addr.streets or "") if addr else "",
direccion=(addr.streets or "") if addr else "",
num_exterior=(addr.exterior_number or "") if addr else "",
num_interior=(addr.interior_number or "") if addr else "",
colonia=(addr.neighborhood or "") if addr else "",
@@ -124,58 +152,128 @@ class FacturaImportacionUsaService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "USA") if addr else "USA",
tax_id=prog.tax_id if (prog and prog.tax_id) else (getattr(main, 'rfc', "") or ""),
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=prog.prosec_authorization if (prog and prog.prosec and prog.prosec_authorization) else "",
reg_emp=prog.val_certified_company_registry if (prog and hasattr(prog, 'val_certified_company_registry')) else (
prog.certified_company_registry if (prog and prog.certified_company_registry) else ""
tax_id=(
prog.tax_id
if (prog and prog.tax_id)
else (getattr(main, "rfc", "") or "")
),
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=(
prog.prosec_authorization
if (prog and prog.prosec and prog.prosec_authorization)
else ""
),
reg_emp=(
prog.val_certified_company_registry
if (prog and hasattr(prog, "val_certified_company_registry"))
else (
prog.certified_company_registry
if (prog and prog.certified_company_registry)
else ""
)
),
cert=(
prog.is_certified_company
if (prog and prog.is_certified_company)
else ""
),
cert=prog.is_certified_company if (prog and prog.is_certified_company) else ""
)
def obtener_datos(self, db: Session, invoice_id: int, company_id: int, progress_callback: Optional[Callable] = None, currency_code: str = 'ORIGINAL') -> FacturaImportacionCompleta:
def obtener_datos(
self,
db: Session,
invoice_id: int,
company_id: int,
progress_callback: Optional[Callable] = None,
currency_code: str = "ORIGINAL",
) -> FacturaImportacionCompleta:
try:
if progress_callback: progress_callback(10, "Searching invoice...")
header = db.query(InvoiceHeader).filter(InvoiceHeader.id == invoice_id, InvoiceHeader.company_id == company_id).first()
if not header: raise HTTPException(status_code=404, detail="Invoice not found")
if progress_callback:
progress_callback(10, "Searching invoice...")
header = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.id == invoice_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if not header:
raise HTTPException(status_code=404, detail="Invoice not found")
compliance = header.compliance_mx
compliance = header.compliance_mx
logistics = header.logistics if header.logistics else None
financials = header.financials if header.financials else None
if progress_callback: progress_callback(20, "Fetching entry data...")
pedimento_id = compliance.pedimento_id if (compliance and compliance.pedimento_id) else header.related_doc_id
pedimento = db.query(Pedimentos).filter(Pedimentos.id == pedimento_id).first() if pedimento_id else None
if progress_callback: progress_callback(30, "Fetching client and supplier...")
if progress_callback:
progress_callback(20, "Fetching entry data...")
pedimento_id = (
compliance.pedimento_id
if (compliance and compliance.pedimento_id)
else header.related_doc_id
)
pedimento = (
db.query(Pedimentos).filter(Pedimentos.id == pedimento_id).first()
if pedimento_id
else None
)
if progress_callback:
progress_callback(30, "Fetching client and supplier...")
proveedor_id = compliance.provider_id if compliance else None
cliente_proveedor = self._obtener_datos_cliente(db, proveedor_id, "Supplier:") if proveedor_id else ClienteSchema(header="Supplier", nombre="Unassigned", direccion="", tax_id="", codigo_postal="", ciudad="", estado="", pais="")
cliente_proveedor = (
self._obtener_datos_cliente(db, proveedor_id, "Supplier:")
if proveedor_id
else ClienteSchema(
header="Supplier",
nombre="Unassigned",
direccion="",
tax_id="",
codigo_postal="",
ciudad="",
estado="",
pais="",
)
)
nombre_agente = ""
if compliance and compliance.customs_broker_id:
broker = db.query(CustomsBroker).filter(CustomsBroker.id == compliance.customs_broker_id).first()
if broker: nombre_agente = broker.name
broker = (
db.query(CustomsBroker)
.filter(CustomsBroker.id == compliance.customs_broker_id)
.first()
)
if broker:
nombre_agente = broker.name
company = db.query(Company).filter(Company.id == header.company_id).first()
# Datos Default (Company/Importer)
cliente_default = ClienteSchema(
header="Importer / Consignee:",
nombre=getattr(company, 'name', "Local Company"),
nombre=getattr(company, "name", "Local Company"),
direccion="FISCAL ADDRESS",
num_exterior="", colonia="", codigo_postal="", ciudad="", estado="", pais="MEX",
tax_id=getattr(company, 'rfc', ""),
programa=getattr(company, 'program', "IMMEX"), autorizacion=getattr(company, 'program_number', "")
num_exterior="",
colonia="",
codigo_postal="",
ciudad="",
estado="",
pais="MEX",
tax_id=getattr(company, "rfc", ""),
programa=getattr(company, "program", "IMMEX"),
autorizacion=getattr(company, "program_number", ""),
)
# Left Side Logic (Sold To)
cliente_vendido = cliente_default
if compliance and compliance.sold_to_id:
# Force English header for American Invoice
clean_header = "Sold To:"
clean_header = "Sold To:"
# raw_header = compliance.sold_to_header or "SOLD_TO"
# clean_header = raw_header.replace("_", " ").title() + ":"
cliente_vendido = self._obtener_datos_cliente(db, compliance.sold_to_id, clean_header)
cliente_vendido = self._obtener_datos_cliente(
db, compliance.sold_to_id, clean_header
)
# Right Side Logic (Shipped To)
cliente_enviado = cliente_default
if compliance and compliance.shipped_to_id:
@@ -183,26 +281,37 @@ class FacturaImportacionUsaService:
clean_header_shipped = "Shipped To:"
# raw_header_shipped = compliance.shipped_to_header or "SHIPPED_TO"
# clean_header_shipped = raw_header_shipped.replace("_", " ").title() + ":"
# Fetch client data
cliente_enviado = self._obtener_datos_cliente(db, compliance.shipped_to_id, clean_header_shipped)
remesa_valor = str(compliance.remesa) if (compliance and compliance.remesa) else ""
acuse_valor = str(compliance.edocument) if (compliance and compliance.edocument) else "N/A"
# Fetch client data
cliente_enviado = self._obtener_datos_cliente(
db, compliance.shipped_to_id, clean_header_shipped
)
remesa_valor = (
str(compliance.remesa) if (compliance and compliance.remesa) else ""
)
acuse_valor = (
str(compliance.edocument)
if (compliance and compliance.edocument)
else "N/A"
)
patente_val = ""
if pedimento and pedimento.license:
patente_val = pedimento.license
elif 'broker' in locals() and broker and broker.license:
elif "broker" in locals() and broker and broker.license:
patente_val = broker.license
# --- Transport Data Fetching ---
transporte_txt = str(logistics.transport_type) if (logistics and logistics.transport_type) else ""
transporte_txt = (
str(logistics.transport_type)
if (logistics and logistics.transport_type)
else ""
)
num_transporte_val = (logistics.trailer_num or "") if logistics else ""
# Init values
placas_val = (logistics.license_plate or "") if logistics else "" # Plates
placas_val = (logistics.license_plate or "") if logistics else "" # Plates
placas_remolque_val = ""
transportista_val = (logistics.carrier_id or "") if logistics else ""
caat_val = ""
@@ -212,53 +321,89 @@ class FacturaImportacionUsaService:
if logistics:
# 1. Transporter (CAAT / SCAC)
if logistics.carrier_id:
transporter_obj = db.query(Transporter).filter(Transporter.transporter_key == logistics.carrier_id).first()
transporter_obj = (
db.query(Transporter)
.filter(Transporter.transporter_key == logistics.carrier_id)
.first()
)
if transporter_obj:
caat_val = transporter_obj.caat_code or ""
scac_val = transporter_obj.transport_code or "" # Mapping transport_code to SCAC
scac_val = (
transporter_obj.transport_code or ""
) # Mapping transport_code to SCAC
transportista_val = transporter_obj.name or logistics.carrier_id
# 2. Vehicle (Plates)
if logistics.transport_id:
veh_obj = db.query(Vehicle).filter(Vehicle.vehicle_key == logistics.transport_id).first()
veh_obj = (
db.query(Vehicle)
.filter(Vehicle.vehicle_key == logistics.transport_id)
.first()
)
if veh_obj:
placas_val = veh_obj.plate_number or placas_val
elif logistics.vehicle_num:
veh_obj = db.query(Vehicle).filter(Vehicle.vehicle_key == logistics.vehicle_num).first()
if veh_obj:
placas_val = veh_obj.plate_number or placas_val
placas_val = veh_obj.plate_number or placas_val
elif logistics.vehicle_num:
veh_obj = (
db.query(Vehicle)
.filter(Vehicle.vehicle_key == logistics.vehicle_num)
.first()
)
if veh_obj:
placas_val = veh_obj.plate_number or placas_val
# 3. Trailer
if logistics.trailer_num:
trl_obj = db.query(Trailer).filter(Trailer.trailer_number == logistics.trailer_num).first()
trl_obj = (
db.query(Trailer)
.filter(Trailer.trailer_number == logistics.trailer_num)
.first()
)
if trl_obj:
placas_remolque_val = trl_obj.plate_number or ""
# 4. Driver (License)
if logistics.carrier_id and logistics.driver_name:
drv_obj = db.query(Driver).filter(
Driver.transporter_key == logistics.carrier_id,
Driver.driver_name == logistics.driver_name
).first()
drv_obj = (
db.query(Driver)
.filter(
Driver.transporter_key == logistics.carrier_id,
Driver.driver_name == logistics.driver_name,
)
.first()
)
if drv_obj:
licencia_cond_val = drv_obj.license_number or ""
licencia_cond_val = drv_obj.license_number or ""
# Determine Currency
moneda_final = getattr(header, 'currency', "USD") or "USD"
if currency_code == 'MXN':
moneda_final = 'MXN'
elif currency_code == 'USD':
moneda_final = 'USD'
moneda_final = getattr(header, "currency", "USD") or "USD"
if currency_code == "MXN":
moneda_final = "MXN"
elif currency_code == "USD":
moneda_final = "USD"
factura_schema = FacturaSchema(
numero=header.invoice_number or "N/A",
titulo_documento=self._get_document_title(header.invoice_type or "", is_american=True),
titulo_documento=self._get_document_title(
header.invoice_type or "", is_american=True
),
fecha=str(header.invoice_date) if header.invoice_date else "",
tipo_cambio=float(financials.exchange_rate) if (financials and financials.exchange_rate) else (float(pedimento.exchange_rate) if pedimento and pedimento.exchange_rate else 1.0),
tipo_cambio=(
float(financials.exchange_rate)
if (financials and financials.exchange_rate)
else (
float(pedimento.exchange_rate)
if pedimento and pedimento.exchange_rate
else 1.0
)
),
moneda=moneda_final,
incoterm=(logistics.incoterm or "") if logistics else "",
observaciones=header.observation_es or header.observation_en or "",
pedimento=f"{pedimento.year} {pedimento.customs_office[:2] if pedimento.customs_office else ''} {pedimento.license} {pedimento.pedimento_number}" if pedimento else "",
pedimento=(
f"{pedimento.year} {pedimento.customs_office[:2] if pedimento.customs_office else ''} {pedimento.license} {pedimento.pedimento_number}"
if pedimento
else ""
),
clave_pedimento=pedimento.pedimento_code if pedimento else "",
regimen=header.document_type or "",
patente=patente_val,
@@ -271,62 +416,87 @@ class FacturaImportacionUsaService:
caat=caat_val,
scac=scac_val,
licencia_conductor=licencia_cond_val,
aduana=compliance.aduana if (compliance and compliance.aduana) else (pedimento.customs_office[:2] if (pedimento and pedimento.customs_office) else ""),
aduana=(
compliance.aduana
if (compliance and compliance.aduana)
else (
pedimento.customs_office[:2]
if (pedimento and pedimento.customs_office)
else ""
)
),
precinto=(logistics.seal_number or "") if logistics else "",
destino=(logistics.destination_goods or "") if logistics else "",
remesa=remesa_valor, acuse_electronico=acuse_valor
remesa=remesa_valor,
acuse_electronico=acuse_valor,
)
if progress_callback: progress_callback(50, "Processing items...")
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(Item.invoice_id == header.id).all()
if progress_callback:
progress_callback(50, "Processing items...")
lines = db.query(LineItem).filter(LineItem.invoice_id == header.id).all()
partidas_list = []
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
fin = db.query(LineFinancial).filter(LineFinancial.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
qty = (
db.query(LineQuantity)
.filter(LineQuantity.item_line_id == line.id)
.first()
)
fin = (
db.query(LineFinancial)
.filter(LineFinancial.item_line_id == line.id)
.first()
)
part_master = (
db.query(Part).filter(Part.id == line.part_number_id).first()
)
desc_final = "N/D"
num_parte_final = str(line.part_number or "N/A")
fraccion_raw = ""
num_parte_final = str(line.part_number_id or "N/A")
fraccion_raw = ""
origen_final = "MEX"
if part_master:
# Prefer English description if available, else Spanish
desc_final = part_master.description_english or part_master.description_spanish or "No Desc."
desc_final = (
part_master.description_english
or part_master.description_spanish
or "No Desc."
)
num_parte_final = part_master.part_number
# Prefer US Fraction (HTS) if available
fraccion_raw = part_master.us_fraction if part_master.us_fraction else ""
fraccion_raw = (
part_master.us_fraction if part_master.us_fraction else ""
)
if part_master.fa_data and part_master.fa_data.origin_country:
origen_final = part_master.fa_data.origin_country
# FRACTION LOGIC: Use US Fraction (us_fraction) if available, otherwise blank
fraccion_imprimir = ""
# Check part master US fraction
if part_master and part_master.us_fraction:
fraccion_imprimir = part_master.us_fraction.strip()
# Optional: Format if needed, but raw is usually fine for US HTS
# If valid US fraction logic requires looking up in DB, we could add that here.
# For now, per requirement: "Si no tiene, pues de queda en blanco"
# Default "General" and "0%" if no specific logic for US duties yet
preferencia_txt = "General"
preferencia_txt = "General"
advalorem_txt = "0%"
# Prioritize USD for American Invoice logic if available?
# Sticking to same logic as Mex for now but could prioritize USD columns.
# Actually, duplicate logic from mex service for now to ensure consistency.
v_unitario = 0.0
v_total = 0.0
if fin:
is_mxn = (factura_schema.moneda == 'MXN')
is_mxn = factura_schema.moneda == "MXN"
if is_mxn:
v_unitario = float(fin.unit_cost_commercial_mxn or 0.0)
v_total = float(fin.value_commercial_mxn or 0.0)
@@ -335,13 +505,13 @@ class FacturaImportacionUsaService:
v_total = float(fin.value_commercial_usd or 0.0)
if not v_unitario:
v_unitario = float(fin.commercial_unit_cost or 0.0)
v_unitario = float(fin.commercial_unit_cost or 0.0)
if not v_total:
v_total = float(fin.total_commercial_value or 0.0)
v_total = float(fin.total_commercial_value or 0.0)
cantidad = float(qty.quantity) if (qty and qty.quantity) else 0.0
if cantidad > 0:
if v_unitario > 0 and v_total == 0:
v_total = v_unitario * cantidad
@@ -349,39 +519,59 @@ class FacturaImportacionUsaService:
v_unitario = v_total / cantidad
# UOM Mapping for English context
uom_raw = qty.weight_unit if qty else "PCS"
if uom_raw == "PZA": uom_raw = "PCS"
uom_raw = line.unit_of_measure_info.code if line.unit_of_measure_info else "PCS"
if uom_raw == "PZA":
uom_raw = "PCS"
partidas_list.append(PartidaSchema(
numero_parte=num_parte_final,
descripcion=desc_final,
fraccion=fraccion_imprimir,
origen=origen_final,
advalorem=advalorem_txt,
preferencia=preferencia_txt,
cantidad_importacion=self.formatear_numero(qty.quantity if qty else 0),
unidad_medida=uom_raw,
cantidad_bultos=int(qty.package_quantity) if qty and qty.package_quantity else 0,
clave_bultos=(qty.package_key or "") if qty else "",
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(qty.gross_weight if qty else 0),
valor_costo_unitario=self.formatear_numero(v_unitario),
valor_total=self.formatear_numero(v_total)
))
partidas_list.append(
PartidaSchema(
numero_parte=num_parte_final,
descripcion=desc_final,
fraccion=fraccion_imprimir,
origen=origen_final,
advalorem=advalorem_txt,
preferencia=preferencia_txt,
cantidad_importacion=self.formatear_numero(
qty.quantity if qty else 0
),
unidad_medida=uom_raw,
cantidad_bultos=(
int(qty.package_quantity)
if qty and qty.package_quantity
else 0
),
clave_bultos=(
qty.package_info.key if (qty and qty.package_info) else ""
),
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(
qty.gross_weight if qty else 0
),
valor_costo_unitario=self.formatear_numero(v_unitario),
valor_total=self.formatear_numero(v_total),
)
)
totales = self.calcular_totales(partidas_list, Decimal(factura_schema.tipo_cambio))
totales = self.calcular_totales(
partidas_list, Decimal(factura_schema.tipo_cambio)
)
return FacturaImportacionCompleta(
cliente_proveedor=cliente_proveedor, cliente_vendido=cliente_vendido,
cliente_enviado=cliente_enviado, factura=factura_schema,
partidas=partidas_list, totales=totales
cliente_proveedor=cliente_proveedor,
cliente_vendido=cliente_vendido,
cliente_enviado=cliente_enviado,
factura=factura_schema,
partidas=partidas_list,
totales=totales,
)
except Exception as e:
print(f"Error Service A76 USA: {e}")
raise HTTPException(status_code=500, detail=f"Error: {str(e)}")
def calcular_totales(self, partidas: List[PartidaSchema], tipo_cambio: Decimal) -> TotalesSchema:
def calcular_totales(
self, partidas: List[PartidaSchema], tipo_cambio: Decimal
) -> TotalesSchema:
cant = sum(p.cantidad_importacion for p in partidas)
valor = sum(p.valor_total for p in partidas)
peso_n = sum(p.peso_neto for p in partidas)
@@ -389,22 +579,38 @@ class FacturaImportacionUsaService:
bultos = sum(p.cantidad_bultos for p in partidas)
claves = [p.clave_bultos for p in partidas if p.clave_bultos]
clave_comun = max(set(claves), key=claves.count) if claves else ""
# if bultos > 1 and clave_comun and not clave_comun.endswith("S"): clave_comun += "S"
# if bultos > 1 and clave_comun and not clave_comun.endswith("S"): clave_comun += "S"
# Don't pluralize strictly in English without logic, kept simple.
tc = float(tipo_cambio) if tipo_cambio else 1.0
return TotalesSchema(
cantidad_total=self.formatear_numero(cant), bultos_total=bultos, clave_bultos=clave_comun,
peso_neto_total=self.formatear_numero(peso_n), peso_bruto_total=self.formatear_numero(peso_b),
valor_total_total=self.formatear_numero(valor), valor_total_dolares=self.formatear_numero(valor / tc if tc > 0 else 0)
return TotalesSchema(
cantidad_total=self.formatear_numero(cant),
bultos_total=bultos,
clave_bultos=clave_comun,
peso_neto_total=self.formatear_numero(peso_n),
peso_bruto_total=self.formatear_numero(peso_b),
valor_total_total=self.formatear_numero(valor),
valor_total_dolares=self.formatear_numero(valor / tc if tc > 0 else 0),
)
def generar_factura_completa(self, db: Session, invoice_id: int, company_id: int, formato: str = "pdf", progress_callback: Optional[Callable] = None, currency_code: str = 'ORIGINAL') -> Tuple[bytes, str, str]:
if progress_callback: progress_callback(5, "Starting report service...")
datos = self.obtener_datos(db, invoice_id, company_id, progress_callback, currency_code)
if progress_callback: progress_callback(80, "Rendering template...")
def generar_factura_completa(
self,
db: Session,
invoice_id: int,
company_id: int,
formato: str = "pdf",
progress_callback: Optional[Callable] = None,
currency_code: str = "ORIGINAL",
) -> Tuple[bytes, str, str]:
if progress_callback:
progress_callback(5, "Starting report service...")
datos = self.obtener_datos(
db, invoice_id, company_id, progress_callback, currency_code
)
if progress_callback:
progress_callback(80, "Rendering template...")
# LOGO LOGIC
logo_b64 = None
try:
@@ -419,26 +625,48 @@ class FacturaImportacionUsaService:
if target_path.exists():
with open(target_path, "rb") as image_file:
encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
encoded_string = base64.b64encode(image_file.read()).decode(
"utf-8"
)
mime = "image/png"
if target_path.suffix.lower() in ['.jpg', '.jpeg']: mime = "image/jpeg"
if target_path.suffix.lower() in [".jpg", ".jpeg"]:
mime = "image/jpeg"
logo_b64 = f"data:{mime};base64,{encoded_string}"
except Exception as e:
print(f"Error loading logo: {e}")
context = {
'cliente_proveedor': datos.cliente_proveedor.model_dump(), 'cliente_vendido': datos.cliente_vendido.model_dump(),
'cliente_enviado': datos.cliente_enviado.model_dump(), 'factura': datos.factura.model_dump(),
'partidas': [p.model_dump() for p in datos.partidas], 'totales': datos.totales.model_dump(),
'logo_b64': logo_b64
"cliente_proveedor": datos.cliente_proveedor.model_dump(),
"cliente_vendido": datos.cliente_vendido.model_dump(),
"cliente_enviado": datos.cliente_enviado.model_dump(),
"factura": datos.factura.model_dump(),
"partidas": [p.model_dump() for p in datos.partidas],
"totales": datos.totales.model_dump(),
"logo_b64": logo_b64,
}
html_content = self.template.render(**context)
nombre = f"Commercial_Invoice_{datos.factura.numero}.{formato}"
if formato == "html": return html_content.encode('utf-8'), nombre, "text/html"
if progress_callback: progress_callback(90, "Generating PDF...")
options = {'page-size': 'Letter', 'margin-top': '0.5in', 'margin-right': '0.5in', 'margin-bottom': '0.5in', 'margin-left': '0.5in', 'encoding': "UTF-8", 'enable-local-file-access': None}
pdf = pdfkit.from_string(html_content, False, options=options, configuration=self._get_wkhtmltopdf_config())
if progress_callback: progress_callback(100, "Completed")
if formato == "html":
return html_content.encode("utf-8"), nombre, "text/html"
if progress_callback:
progress_callback(90, "Generating PDF...")
options = {
"page-size": "Letter",
"margin-top": "0.5in",
"margin-right": "0.5in",
"margin-bottom": "0.5in",
"margin-left": "0.5in",
"encoding": "UTF-8",
"enable-local-file-access": None,
}
pdf = pdfkit.from_string(
html_content,
False,
options=options,
configuration=self._get_wkhtmltopdf_config(),
)
if progress_callback:
progress_callback(100, "Completed")
return pdf, nombre, "application/pdf"

43
backend/api/v1/modules/a76/reports/importacion/packing_list/service.py
View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
# --- MODELOS ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.line_customs.models import LineCustom
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
@@ -22,7 +21,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -238,40 +237,40 @@ class PackingListService:
)
if progress_callback: progress_callback(50, "Procesando partidas...")
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(Item.invoice_id == header.id).all()
lines = db.query(LineItem).filter(LineItem.invoice_id == header.id).all()
partidas_list = []
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
weight_type = db.query(InvoiceLogistics.weight_type).filter(InvoiceLogistics.invoice_id == line.invoice_id).scalar()
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
# --- WEIGHT CALCULATION LOGIC ---
peso_neto_kg = 0.0
peso_bruto_kg = 0.0
peso_neto_lb = 0.0
peso_bruto_lb = 0.0
peso_bruto_lb = 0.0
if qty:
raw_net = float(qty.net_weight or 0)
raw_gross = float(qty.gross_weight or 0)
unit = (qty.weight_unit or "KG").upper()
if unit == "LB" or unit == "LBS":
peso_neto_lb = raw_net
peso_bruto_lb = raw_gross
peso_neto_kg = raw_net / 2.20462
peso_bruto_kg = raw_gross / 2.20462
else: # Default KG
peso_neto_kg = raw_net
peso_bruto_kg = raw_gross
peso_neto_lb = raw_net * 2.20462
peso_bruto_lb = raw_gross * 2.20462
raw_net = float(qty.net_weight or 0)
raw_gross = float(qty.gross_weight or 0)
unit = (weight_type or "KGS").upper()
if unit == "LBS":
peso_neto_lb = raw_net
peso_bruto_lb = raw_gross
peso_neto_kg = raw_net / 2.20462
peso_bruto_kg = raw_gross / 2.20462
else: # Default KG
peso_neto_kg = raw_net
peso_bruto_kg = raw_gross
peso_neto_lb = raw_net * 2.20462
peso_bruto_lb = raw_gross * 2.20462
# --------------------------------
custom_obj = db.query(LineCustom).filter(LineCustom.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
desc_final = "S/D"
num_parte_final = str(line.part_number or "S/N")
num_parte_final = str(line.part_number_id or "S/N")
fraccion_raw = ""
origen_final = "MEX"
uom_comercial = "PZA" # Default UOM

6
backend/api/v1/modules/core/dashboard/service.py
View File

@@ -12,7 +12,7 @@ from api.v1.modules.a76.invoices.models import (
InvoiceComplianceMx,
OperationType,
)
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider,
ClientOrProviderEnum,
@@ -244,9 +244,9 @@ class DashboardService:
def get_items_metrics(self) -> KPIMetric:
"""Obtiene métricas de items/productos"""
total_items = (
self.db.query(func.count(Item.id))
self.db.query(func.count(LineItem.id))
.filter(
Item.tenant_id == self.tenant_id, Item.company_id == self.company_id
LineItem.tenant_id == self.tenant_id, LineItem.company_id == self.company_id
)
.scalar()
or 0