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

Merge pull request 'feature/data-entry' (#185) from feature/data-entry into development

Browse Source 
Reviewed-on: ADUANASOFT/anexo76#185
This commit is contained in:
acazares
2026-03-05 15:36:42 +00:00
parent 93407d1333 375df79da5
commit 88d6412bae
14 changed files with 890 additions and 735 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
backend/api/v1/modules/a76/general_catalogs/company/service.py
View File

@@ -3,6 +3,7 @@ Capa de servicio para lógica de negocio de empresa
"""
import logging
from datetime import datetime
from typing import List, Optional, Tuple, Dict, Any
from fastapi import HTTPException
@@ -12,7 +13,10 @@ from sqlalchemy.orm import Session
from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
from .models import Company
from ...audit_log.services.service import AuditService
from ..units_of_measure.seed import seed as units_of_measure_seed
from ..fractions.historical_tariff_fractions.seed import seed as historical_tariff_fractions_seed
from core.context import get_user_context
from sqlalchemy import text
logger = logging.getLogger(__name__)
@@ -34,7 +38,7 @@ class CompanyService:
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Company], int]:
"""Get all companies for a tenant with pagination"""
query = db.query(Company).filter(Company.tenant_id == tenant_id)
query = db.query(Company).filter(Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
# Apply filters if provided
if filters:
@@ -62,6 +66,7 @@ class CompanyService:
.filter(
Company.id == company_id,
Company.tenant_id == tenant_id,
Company.deleted_at.is_(None)
)
.first()
)
@@ -381,7 +386,10 @@ class CompanyService:
if addr_ind2:
self.db.add(CompanyAddress(**addr_ind2, address_type='industrial2', company_id=db_company.id))
# 7. Commit
# 7. Seed company data (tenant/company dependent)
self._seed_company_data(self.db, tenant_id, db_company.id)
# 8. Commit
self.db.commit()
self.db.refresh(db_company)
@@ -584,17 +592,9 @@ class CompanyService:
# ----------------------
try:
# Cascading deletes are handled by relationship settings, but manual is safer here
if company.certification: db.delete(company.certification)
if company.prevalidator: db.delete(company.prevalidator)
if company.electronic_agent: db.delete(company.electronic_agent)
if company.ventanilla_unica: db.delete(company.ventanilla_unica)
if company.cfdi: db.delete(company.cfdi)
for cert in company.digital_certificates: db.delete(cert)
for addr in company.addresses: db.delete(addr)
company.deleted_at = datetime.utcnow()
db.flush()
db.delete(company)
db.commit()
# --- Audit Log ---
@@ -698,11 +698,68 @@ class CompanyService:
# Custom methods
def _seed_company_data(self, db: Session, tenant_id: int, company_id: int):
"""Seeds tenant/company dependent data for a new company"""
def format_value(val):
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return f"'{str(val).replace(chr(39), chr(39)*2)}'"
# 1. Units of Measure
val_uom = ", ".join(
[
f"({format_value(code)}, {format_value(desc)}, {format_value(desc_en)}, "
f"{format_value(customs)}, {format_value(american)}, {format_value(ace)}, {format_value(oma)}, {tenant_id}, {company_id})"
for code, desc, desc_en, customs, american, ace, oma in units_of_measure_seed
]
)
db.execute(text("ALTER TABLE a76.units_of_measure DISABLE TRIGGER ALL;"))
db.execute(text(f"INSERT INTO a76.units_of_measure (code, description, description_en, customs_code, american_code, ace_code, oma_code, tenant_id, company_id) VALUES {val_uom} ON CONFLICT (code, tenant_id, company_id) DO NOTHING;"))
db.execute(text("ALTER TABLE a76.units_of_measure ENABLE TRIGGER ALL;"))
# 2. Historical Tariff Fractions
def format_bool(val):
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return "TRUE" if str(val).upper() == "TRUE" else "FALSE"
def format_timestamp(val):
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return f"'{str(val)}'"
values_historical = ", ".join(
[
f"({format_value(historical_fraction)}, {format_value(nico)}, {format_value(unit_measure)}, {format_value(country)}, "
f"{format_value(fraction_type)}, {format_value(sector)}, {format_value(import_tax)}, "
f"{format_value(export_tax)}, {format_timestamp(pub_date)}, {format_bool(is_immex)}, "
f"{format_bool(normal_temp)}, {format_bool(services_temp)}, {format_bool(certified_temp)}, "
f"{format_bool(by_log)}, {format_timestamp(end_date)}, "
f"{tenant_id}, {company_id})"
for (historical_fraction, nico, unit_measure, country, fraction_type, sector, import_tax,
export_tax, pub_date, is_immex, normal_temp, services_temp, certified_temp, by_log, end_date) in historical_tariff_fractions_seed
]
)
if values_historical:
db.execute(text("SET session_replication_role = replica;"))
db.execute(text(f"""
INSERT INTO a76.historical_tariff_fractions
(historical_fraction, nico, unit_of_measure_code, country, fraction_type, sector,
import_tax_rate, export_tax_rate, publication_date, is_immex,
normal_temporality, services_temporality, certified_temporality, by_log, end_date,
tenant_id, company_id)
VALUES {values_historical}
ON CONFLICT DO NOTHING;
"""))
db.execute(text("SET session_replication_role = DEFAULT;"))
def get_companies_by_tenant(self, tenant_id: int) -> List[Company]:
"""Get all companies for a tenant"""
return (
self.db.query(Company)
.filter(Company.tenant_id == tenant_id)
.filter(Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
.order_by(Company.name)
.all()
)
@@ -711,7 +768,7 @@ class CompanyService:
"""Check if a company exists for a tenant"""
return (
self.db.query(Company)
.filter(Company.tenant_id == tenant_id)
.filter(Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
.first()
is not None
)