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

feature/copias-envoices

Browse Source
This commit is contained in:
hreyes
2026-05-29 12:47:40 -06:00
parent ff14bd3645
commit 0448f84e80
4 changed files with 171 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/api/v1/modules/a76/invoices/routes.py
View File

@@ -271,6 +271,7 @@ def delete_invoice(
def copy_invoice(
invoice_id: int = Path(..., description="ID de la factura a copiar"),
company_id: int = Query(..., description="Company ID"),
new_invoice_number: Optional[str] = Query(None, description="Número de factura para la copia"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
@@ -285,7 +286,7 @@ def copy_invoice(
)
try:
return services.InvoiceService.copy(db, invoice_id, tenant_id, company_id)
return services.InvoiceService.copy(db, invoice_id, tenant_id, company_id, new_invoice_number)
except HTTPException:
raise
except Exception as e:

129
backend/api/v1/modules/a76/invoices/services.py
View File

@@ -13,6 +13,13 @@ from .exports.validators.create import validate_create as validate_create_export
from .exports.validators.update import validate_update as validate_update_export
from .common.common_validators import invoice_exists
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
from api.v1.modules.a76.items.line_descriptions.models import LineDescription
from api.v1.modules.a76.items.line_references.models import LineReference
from api.v1.modules.a76.items.series.models import Serie
from api.v1.modules.a76.general_catalogs.identifiers.models import IdentifierDetail
from . import models, schemas
@@ -777,12 +784,15 @@ class InvoiceService:
invoice_id: int,
tenant_id: int,
company_id: int,
new_invoice_number: Optional[str] = None,
) -> models.InvoiceHeader:
"""Duplica una factura con todas sus tablas relacionadas.
"""Duplica una factura con todas sus tablas relacionadas, incluyendo partidas.
El número de factura de la copia lleva sufijo '-COPIA' (o '-COPIA-N' si ya existe).
Si se provee new_invoice_number, se usa ese número (falla con 409 si ya existe).
Si no, genera sufijo '-COPIA' / '-COPIA-N' automáticamente.
Estado reseteado a 'pending', sin pedimento ni datos de procesamiento.
"""
from fastapi import HTTPException
original = (
db.query(models.InvoiceHeader)
.filter(
@@ -793,21 +803,31 @@ class InvoiceService:
.first()
)
if not original:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Factura no encontrada")
# Generar número de factura único para la copia
base_number = original.invoice_number or ""
candidate = f"{base_number}-COPIA"
counter = 1
while db.query(models.InvoiceHeader).filter(
models.InvoiceHeader.invoice_number == candidate,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
).first():
counter += 1
candidate = f"{base_number}-COPIA-{counter}"
new_invoice_number = candidate
if new_invoice_number:
exists = db.query(models.InvoiceHeader).filter(
models.InvoiceHeader.invoice_number == new_invoice_number,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
).first()
if exists:
raise HTTPException(
status_code=409,
detail=f"Ya existe una factura con el número '{new_invoice_number}'"
)
else:
base_number = original.invoice_number or ""
candidate = f"{base_number}-COPIA"
counter = 1
while db.query(models.InvoiceHeader).filter(
models.InvoiceHeader.invoice_number == candidate,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
).first():
counter += 1
candidate = f"{base_number}-COPIA-{counter}"
new_invoice_number = candidate
username = _get_current_username()
@@ -839,7 +859,7 @@ class InvoiceService:
).first()
if comp:
EXCLUDE_COMP = {
"invoice_id", "pedimento_id", "vucem_operation_num",
"invoice_id", "vucem_operation_num",
"electronic_signature", "certificate_number", "niu_number",
"code_signature", "edocument", "created_at", "updated_at",
}
@@ -915,6 +935,83 @@ class InvoiceService:
col_data["company_id"] = company_id
db.add(models.InvoiceCollections(**col_data))
# Copiar partidas (LineItem) con todas sus sub-tablas
EXCLUDE_LINE = {"id", "invoice_id", "created_at", "updated_at"}
EXCLUDE_SUB = {"id", "item_line_id", "created_at", "updated_at"}
EXCLUDE_SERIE = {"id", "line_item_id", "created_at", "updated_at"}
EXCLUDE_IDENT = {"id", "item_line_id", "created_at", "updated_at"}
line_items = db.query(LineItem).filter(
LineItem.invoice_id == invoice_id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
).order_by(LineItem.line_number).all()
for item in line_items:
line_data = {
c.name: getattr(item, c.name)
for c in LineItem.__table__.columns
if c.name not in EXCLUDE_LINE
}
line_data["invoice_id"] = new_invoice.id
new_item = LineItem(**line_data)
db.add(new_item)
db.flush()
# Copiar series primero para reusar sus IDs en LineReference
serie_id_map: dict[int, int] = {}
for serie in db.query(Serie).filter(Serie.line_item_id == item.id).all():
s_data = {
c.name: getattr(serie, c.name)
for c in Serie.__table__.columns
if c.name not in EXCLUDE_SERIE
}
s_data["line_item_id"] = new_item.id
new_serie = Serie(**s_data)
db.add(new_serie)
db.flush()
serie_id_map[serie.id] = new_serie.id
fin = db.query(LineFinancial).filter(LineFinancial.item_line_id == item.id).first()
if fin:
f_data = {c.name: getattr(fin, c.name) for c in LineFinancial.__table__.columns if c.name not in EXCLUDE_SUB}
f_data["item_line_id"] = new_item.id
db.add(LineFinancial(**f_data))
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == item.id).first()
if qty:
q_data = {c.name: getattr(qty, c.name) for c in LineQuantity.__table__.columns if c.name not in EXCLUDE_SUB}
q_data["item_line_id"] = new_item.id
db.add(LineQuantity(**q_data))
cus = db.query(LineCustom).filter(LineCustom.item_line_id == item.id).first()
if cus:
cu_data = {c.name: getattr(cus, c.name) for c in LineCustom.__table__.columns if c.name not in EXCLUDE_SUB}
cu_data["item_line_id"] = new_item.id
db.add(LineCustom(**cu_data))
desc = db.query(LineDescription).filter(LineDescription.item_line_id == item.id).first()
if desc:
d_data = {c.name: getattr(desc, c.name) for c in LineDescription.__table__.columns if c.name not in EXCLUDE_SUB}
d_data["item_line_id"] = new_item.id
db.add(LineDescription(**d_data))
ref = db.query(LineReference).filter(LineReference.item_line_id == item.id).first()
if ref:
r_data = {
c.name: getattr(ref, c.name)
for c in LineReference.__table__.columns
if c.name not in {"id", "item_line_id", "serie_id", "created_at", "updated_at"}
}
r_data["item_line_id"] = new_item.id
r_data["serie_id"] = serie_id_map.get(ref.serie_id) if ref.serie_id else None
db.add(LineReference(**r_data))
for ident in db.query(IdentifierDetail).filter(IdentifierDetail.item_line_id == item.id).all():
i_data = {c.name: getattr(ident, c.name) for c in IdentifierDetail.__table__.columns if c.name not in EXCLUDE_IDENT}
i_data["item_line_id"] = new_item.id
db.add(IdentifierDetail(**i_data))
db.commit()
db.refresh(new_invoice)
return new_invoice