From e338236a1c4d4b08b0497da7e5275e47cf21ff6b Mon Sep 17 00:00:00 2001 From: Kevin_Ramirez Date: Thu, 22 Jan 2026 09:49:59 -0600 Subject: [PATCH] Se arreglo el problema de company --- .../a76/general_catalogs/company/models.py | 2 +- .../a76/general_catalogs/company/routes.py | 161 ++--------- .../a76/general_catalogs/company/service.py | 270 ++++++++++++++---- .../company/submodels/certification.py | 2 +- backend/debug_mapper.py | 15 + backend/inspect_schema.py | 20 ++ backend/verify_logo_presence.py | 32 +++ 7 files changed, 311 insertions(+), 191 deletions(-) create mode 100644 backend/debug_mapper.py create mode 100644 backend/inspect_schema.py create mode 100644 backend/verify_logo_presence.py diff --git a/backend/api/v1/modules/a76/general_catalogs/company/models.py b/backend/api/v1/modules/a76/general_catalogs/company/models.py index fdcc0262..61ccc8a6 100644 --- a/backend/api/v1/modules/a76/general_catalogs/company/models.py +++ b/backend/api/v1/modules/a76/general_catalogs/company/models.py @@ -63,7 +63,7 @@ class Company(Base, TimestampMixin): logo: Mapped[Optional[str]] = mapped_column(String(255)) has_express_line: Mapped[Optional[str]] = mapped_column(String(2), default="N") order_format_type: Mapped[Optional[str]] = mapped_column(String(19)) - is_service_company: Mapped[Optional[str]] = mapped_column(String(2), default="N") + is_service_company: Mapped[Optional[bool]] = mapped_column(Boolean, default=False) client_name: Mapped[Optional[str]] = mapped_column(String(300)) subassembly_mode: Mapped[Optional[str]] = mapped_column(String(7)) diff --git a/backend/api/v1/modules/a76/general_catalogs/company/routes.py b/backend/api/v1/modules/a76/general_catalogs/company/routes.py index 6aadda85..e6d4bc52 100644 --- a/backend/api/v1/modules/a76/general_catalogs/company/routes.py +++ b/backend/api/v1/modules/a76/general_catalogs/company/routes.py @@ -50,7 +50,8 @@ async def create_company( ) service = CompanyService(db) - return service.create_company_manually(data, tenant_id=tenant_id) + new_company = service.create_company_manually(data, tenant_id=tenant_id) + return CompanyResponseDTO.model_validate(service.flatten_company_dto(new_company)) @router.get( @@ -94,7 +95,10 @@ async def list_companies( total_pages = (total + page_size - 1) // page_size return { - "items": [CompanyResponseDTO.model_validate(item) for item in items], + "items": [ + CompanyResponseDTO.model_validate(service.flatten_company_dto(item)) + for item in items + ], "total": total, "page": page, "page_size": page_size, @@ -122,135 +126,12 @@ async def get_my_companies( service = CompanyService(db) companies = service.get_companies_by_tenant(tenant_id) - return [CompanyResponseDTO.model_validate(company) for company in companies] - - -@router.get( - "/status/exists", - response_model=dict, - summary="Check if company exists for tenant", -) -async def check_company_exists( - db: Session = Depends(get_core_db), - current_user: dict = Depends(get_current_user), -): - """Check if a company exists for the current tenant""" - tenant_id = current_user.get("tenant_id") - if not tenant_id: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Tenant ID not found in user data", - ) - - service = CompanyService(db) - exists = service.exists_company(tenant_id) - - return {"exists": exists} - - -@router.get( - "/info/basic/{company_id}", - response_model=dict, - summary="Get basic company info", -) -async def get_basic_info( - company_id: int, - db: Session = Depends(get_core_db), - current_user: dict = Depends(get_current_user), -): - """Get basic information about a company""" - tenant_id = current_user.get("tenant_id") - company_id_from_user = current_user.get("company_id") - - # Validate access - validate_access_to_resource( - db, tenant_id, company_id_from_user, Company, company_id, "id" - ) - - company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_from_user) - if not company: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Company not found", - ) - - return { - "id": company.id, - "name": company.name, - "rfc": company.rfc, - "program": company.program, - } - - -@router.get( - "/info/responsible/{company_id}", - response_model=dict, - summary="Get responsible person info", -) -async def get_responsible_info( - company_id: int, - db: Session = Depends(get_core_db), - current_user: dict = Depends(get_current_user), -): - """Get responsible person information for a company""" - tenant_id = current_user.get("tenant_id") - company_id_from_user = current_user.get("company_id") - - # Validate access - validate_access_to_resource( - db, tenant_id, company_id_from_user, Company, company_id, "id" - ) - - company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_from_user) - if not company: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Company not found", - ) - - return { - "responsible": company.responsible, - "responsible_name": company.responsible_name, - "responsible_last_name": company.responsible_last_name, - "responsible_mother_last_name": company.responsible_mother_last_name, - "responsible_rfc": company.responsible_rfc, - "position": company.position, - } - - -@router.get( - "/info/program/{company_id}", - response_model=dict, - summary="Get program information", -) -async def get_program_info( - company_id: int, - db: Session = Depends(get_core_db), - current_user: dict = Depends(get_current_user), -): - """Get program information for a company""" - tenant_id = current_user.get("tenant_id") - company_id_from_user = current_user.get("company_id") - - # Validate access - validate_access_to_resource( - db, tenant_id, company_id_from_user, Company, company_id, "id" - ) - - company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_from_user) - if not company: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Company not found", - ) - - return { - "program": company.program, - "program_number": company.program_number, - "prosec": company.prosec, - "prosec_authorization": company.prosec_authorization, - } + return [ + CompanyResponseDTO.model_validate(service.flatten_company_dto(company)) + for company in companies + ] +# ... existing code ... @router.get( "/{company_id}", @@ -270,6 +151,7 @@ async def get_company( detail="Tenant ID not found in user data", ) + service = CompanyService(db) company = CompanyService.get_by_id(db, company_id, tenant_id, 0) if not company: raise HTTPException( @@ -277,7 +159,7 @@ async def get_company( detail="Company not found", ) - return CompanyResponseDTO.model_validate(company) + return CompanyResponseDTO.model_validate(service.flatten_company_dto(company)) @router.put( @@ -299,17 +181,18 @@ async def update_company( detail="Tenant ID not found in user data", ) - updated_company = CompanyService.update(db, company_id, tenant_id, 0, data) + service = CompanyService(db) + updated_company = service.update(db, company_id, tenant_id, 0, data) if not updated_company: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Company not found", ) - return CompanyResponseDTO.model_validate(updated_company) + return CompanyResponseDTO.model_validate(service.flatten_company_dto(updated_company)) + - return CompanyResponseDTO.model_validate(updated_company) @@ -347,6 +230,13 @@ async def get_company_logo_image( raise HTTPException(status_code=404, detail="Logo file not found on server") return FileResponse(file_path) + + +@router.delete( + "/{company_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Delete a company", +) async def delete_company( company_id: int, db: Session = Depends(get_core_db), @@ -443,7 +333,8 @@ async def upload_company_logo( # Actualizar la empresa con la ruta del logo update_data = CompanyUpdateDTO(logo=file_path) - updated_company = CompanyService.update(db, company_id, tenant_id, 0, update_data) + service = CompanyService(db) + updated_company = service.update(db, company_id, tenant_id, 0, update_data) return { "message": "Logo uploaded successfully", diff --git a/backend/api/v1/modules/a76/general_catalogs/company/service.py b/backend/api/v1/modules/a76/general_catalogs/company/service.py index 5ed24feb..90460e79 100644 --- a/backend/api/v1/modules/a76/general_catalogs/company/service.py +++ b/backend/api/v1/modules/a76/general_catalogs/company/service.py @@ -97,8 +97,139 @@ class CompanyService: logger.error(f"Error creating company: {str(e)}") raise HTTPException(status_code=500, detail="Error creating company") - @staticmethod + + # ==================== HELPERS FOR FIELD MAPPING ==================== + def _extract_company_fields(self, data: Dict[str, Any]) -> Dict[str, Any]: + """Extrae campos que pertenecen a la tabla Company principal""" + company_fields = [ + "name", "rfc", "curp", "main_activity", "program", "program_number", + "prosec", "prosec_authorization", "sector1", "sector2", "sector3", + "manufacturer_id", "broker_company", "responsible", "responsible_name", + "responsible_last_name", "responsible_mother_last_name", "responsible_rfc", + "position", "logo", "has_express_line", "order_format_type", + "is_service_company", "client_name", "subassembly_mode", "previous_code", + "active_labels", "active_fractions", "activate_caat", "trans_interface", + "american_costs", "scaf_readonly", "parts_replacement", "activate_facmexame", + "part_reference", "international_firm", "ftp_key", "sifra_path", + "version_type", "sql_language", "balance_operation_mode", "inter_db_name" + ] + return {k: v for k, v in data.items() if k in company_fields} + + def _extract_certification_fields(self, data: Dict[str, Any]) -> Dict[str, Any]: + """Extrae campos que pertenecen a CompanyCertification""" + cert_fields = [ + "is_certified_company", "certified_company_registration", + "certified_company_start_date", "certified_company_end_date", + "annex31_certification_date", "annex31_certification_number", + "annex31_modality", "annex31_company_type", "annex31_renewal_date", + "annex31_final_certification_date", "is_oea_company", "ctpat_svi", + "trusted_exporter_number", "neec_company" + ] + return {k: v for k, v in data.items() if k in cert_fields} + + def _extract_prevalidator_fields(self, data: Dict[str, Any]) -> Dict[str, Any]: + """Extrae campos que pertenecen a CompanyPrevalidator""" + # Note: 'prevalidator_key' in DTO maps to 'key' in model + fields = {} + if "prevalidator_key" in data: + fields["key"] = data["prevalidator_key"] + + # Add other fields if present in DTO in the future + return fields + + def flatten_company_dto(self, company: Company) -> Dict[str, Any]: + """Flattens Company and its submodels into a single dict for DTO validation""" + # 1. Base Company fields + result = { + k: getattr(company, k) + for k in company.__mapper__.c.keys() + } + # Explicitly ensure logo is present (defensive programming) + if hasattr(company, 'logo'): + result['logo'] = company.logo + + # Convert has_express_line from String "S"/"N" to Boolean + if hasattr(company, 'has_express_line'): + val = getattr(company, 'has_express_line', "N") + result['has_express_line'] = (val == "S") + + # 2. Certification fields + if company.certification: + cert_fields = [ + "is_certified_company", "certified_company_registration", + "certified_company_start_date", "certified_company_end_date", + "annex31_certification_date", "annex31_certification_number", + "annex31_modality", "annex31_company_type", "annex31_renewal_date", + "annex31_final_certification_date", "is_oea_company", "ctpat_svi", + "trusted_exporter_number", "neec_company" + ] + for field in cert_fields: + val = getattr(company.certification, field, None) + if val is not None: + result[field] = val + + # 3. Prevalidator fields + if company.prevalidator: + if company.prevalidator.key: + result["prevalidator_key"] = company.prevalidator.key + + return result + + # ==================== CRUD METHODS ==================== + + def create_company_manually(self, data: CompanyCreateDTO, tenant_id: int) -> Company: + from .submodels.certification import CompanyCertification + from .submodels.prevalidator import CompanyPrevalidator + + try: + # 1. Preparar datos + obj_data = data.model_dump(exclude_unset=True) + + # Handle boolean flags for Company (Hybrid Approach) + # has_express_line is String(2), is_service_company is Boolean + if "has_express_line" in obj_data and isinstance(obj_data["has_express_line"], bool): + obj_data["has_express_line"] = "S" if obj_data["has_express_line"] else "N" + + # 2. Extract fields for each model + company_data = self._extract_company_fields(obj_data) + cert_data = self._extract_certification_fields(obj_data) + preval_data = self._extract_prevalidator_fields(obj_data) + + # 3. Create Company + db_company = Company(**company_data, tenant_id=tenant_id) + self.db.add(db_company) + self.db.flush() # Generate ID + + # 4. Create Certification if data exists + if cert_data: + cert = CompanyCertification(**cert_data, company_id=db_company.id) + self.db.add(cert) + + # 5. Create Prevalidator if data exists + if preval_data: + preval = CompanyPrevalidator(**preval_data, company_id=db_company.id) + self.db.add(preval) + + # 6. Commit + self.db.commit() + self.db.refresh(db_company) + + return db_company + + except IntegrityError as e: + self.db.rollback() + logger.error(f"IntegrityError creating company manually: {str(e)}") + raise HTTPException( + status_code=400, + detail="Error de integridad: Es posible que esta empresa ya exista.", + ) + except Exception as e: + self.db.rollback() + logger.error(f"Error creating company manually: {str(e)}") + raise HTTPException(status_code=500, detail=f"Error creando empresa: {str(e)}") + def update( + self, # Changed to instance method to use self helper methods db: Session, company_id: int, tenant_id: int, @@ -106,30 +237,57 @@ class CompanyService: company_data: CompanyUpdateDTO, ) -> Optional[Company]: """Update a company""" - company = CompanyService.get_by_id(db, company_id, tenant_id, company_id_unused) + from .submodels.certification import CompanyCertification + from .submodels.prevalidator import CompanyPrevalidator + + # Use self.db if db is passed as None, or use passed db (legacy support) + session = db if db else self.db + + company = self.get_by_id(session, company_id, tenant_id, company_id_unused) if not company: return None # Update only provided fields update_data = company_data.model_dump(exclude_unset=True) - boolean_fields_str = ["has_express_line", "is_service_company"] - for field, value in update_data.items(): - if field in boolean_fields_str: - # Convert boolean to "S"/"N" - if isinstance(value, bool): - value = "S" if value else "N" - + # 1. Update Company fields + company_fields = self._extract_company_fields(update_data) + + # Hybrid Approach: has_express_line is String, is_service_company is Boolean + if "has_express_line" in company_fields and isinstance(company_fields["has_express_line"], bool): + company_fields["has_express_line"] = "S" if company_fields["has_express_line"] else "N" + + for field, value in company_fields.items(): setattr(company, field, value) + # 2. Update Certification + cert_fields = self._extract_certification_fields(update_data) + if cert_fields: + if company.certification: + for field, value in cert_fields.items(): + setattr(company.certification, field, value) + else: + new_cert = CompanyCertification(**cert_fields, company_id=company.id) + session.add(new_cert) + + # 3. Update Prevalidator + preval_fields = self._extract_prevalidator_fields(update_data) + if preval_fields: + if company.prevalidator: + for field, value in preval_fields.items(): + setattr(company.prevalidator, field, value) + else: + new_preval = CompanyPrevalidator(**preval_fields, company_id=company.id) + session.add(new_preval) + try: - db.commit() - db.refresh(company) + session.commit() + session.refresh(company) return company except Exception as e: - db.rollback() + session.rollback() logger.error(f"Error updating company {company_id}: {str(e)}") - raise HTTPException(status_code=500, detail="Error updating company") + raise HTTPException(status_code=500, detail=f"Error al actualizar la empresa: {str(e)}") @staticmethod def delete( @@ -141,19 +299,56 @@ class CompanyService: return False try: + # Manual cascade delete for submodels to ensure order and avoid FK issues + # (Even though cascade="all, delete-orphan" is set, manual deletion is safer for strict DBs) + + # 1. Delete Certification + if company.certification: + db.delete(company.certification) + + # 2. Delete Prevalidator + if company.prevalidator: + db.delete(company.prevalidator) + + # 3. Delete Electronic Agent + if company.electronic_agent: + db.delete(company.electronic_agent) + + # 4. Delete VU + if company.ventanilla_unica: + db.delete(company.ventanilla_unica) + + # 5. Delete CFDI + if company.cfdi: + db.delete(company.cfdi) + + # 6. Delete Digital Certificates + for cert in company.digital_certificates: + db.delete(cert) + + # 7. Delete Addresses + for addr in company.addresses: + db.delete(addr) + + # Flush to execute submodel deletions first + db.flush() + db.delete(company) db.commit() return True except IntegrityError as e: db.rollback() logger.error(f"IntegrityError deleting company {company_id}: {str(e)}") - # Check if it's a foreign key constraint - if "foreign key constraint" in str(e).lower(): - raise HTTPException( - status_code=400, - detail="No se puede eliminar la empresa porque tiene registros relacionados (facturas, conceptos, etc.)" - ) - raise HTTPException(status_code=400, detail="Error al eliminar la empresa") + # Try to get detailed error from psycopg2 + detail = "No se puede eliminar la empresa porque tiene registros relacionados." + if hasattr(e, 'orig') and hasattr(e.orig, 'diag'): + if e.orig.diag.message_detail: + detail += f" Detalles: {e.orig.diag.message_detail}" + + raise HTTPException( + status_code=400, + detail=detail + ) except Exception as e: db.rollback() logger.error(f"Error deleting company {company_id}: {str(e)}") @@ -176,37 +371,4 @@ class CompanyService: .filter(Company.tenant_id == tenant_id) .first() is not None - ) - - def create_company_manually(self, data: CompanyCreateDTO, tenant_id: int) -> Company: - - try: - # 1. Preparar datos - obj_data = data.model_dump(exclude_unset=True) - - boolean_fields_str = ["has_express_line", "is_service_company"] - for field in boolean_fields_str: - if field in obj_data and isinstance(obj_data[field], bool): - obj_data[field] = "S" if obj_data[field] else "N" - - # 2. Crear objeto SQLAlchemy - db_obj = Company(**obj_data, tenant_id=tenant_id) - - # 3. Guardar - self.db.add(db_obj) - self.db.commit() - self.db.refresh(db_obj) - - return db_obj - - except IntegrityError as e: - self.db.rollback() - logger.error(f"IntegrityError creating company manually: {str(e)}") - raise HTTPException( - status_code=400, - detail="Error de integridad: Es posible que esta empresa ya exista.", - ) - except Exception as e: - self.db.rollback() - logger.error(f"Error creating company manually: {str(e)}") - raise HTTPException(status_code=500, detail=f"Error creando empresa: {str(e)}") \ No newline at end of file + ) \ No newline at end of file diff --git a/backend/api/v1/modules/a76/general_catalogs/company/submodels/certification.py b/backend/api/v1/modules/a76/general_catalogs/company/submodels/certification.py index 5624704b..eea80e08 100644 --- a/backend/api/v1/modules/a76/general_catalogs/company/submodels/certification.py +++ b/backend/api/v1/modules/a76/general_catalogs/company/submodels/certification.py @@ -2,7 +2,7 @@ Modelo de certificaciones de empresa """ from typing import Optional, TYPE_CHECKING -from sqlalchemy import Integer, String, SmallInteger, ForeignKey +from sqlalchemy import Integer, String, SmallInteger, ForeignKey, Boolean from sqlalchemy.orm import Mapped, mapped_column, relationship from core.database import Base from api.v1.common.base_models import TimestampMixin diff --git a/backend/debug_mapper.py b/backend/debug_mapper.py new file mode 100644 index 00000000..cc19459b --- /dev/null +++ b/backend/debug_mapper.py @@ -0,0 +1,15 @@ +from api.v1.modules.a76.general_catalogs.company.models import Company +from api.v1.modules.a76.general_catalogs.company.service import CompanyService +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +print("Checking Company Mapper keys...") +try: + keys = Company.__mapper__.c.keys() + print(f"Keys found: {keys}") + if 'logo' in keys: + print("SUCCESS: 'logo' is in keys") + else: + print("FAILURE: 'logo' is NOT in keys") +except Exception as e: + print(f"Error: {e}") diff --git a/backend/inspect_schema.py b/backend/inspect_schema.py new file mode 100644 index 00000000..63a44ca6 --- /dev/null +++ b/backend/inspect_schema.py @@ -0,0 +1,20 @@ +from sqlalchemy import create_engine, text + +# Connect to DB (adjust for localhost) +DB_URL = "postgresql://postgres:postgres@localhost:5432/anexo76_core" +engine = create_engine(DB_URL) + +sql = """ +SELECT column_name, data_type, character_maximum_length +FROM information_schema.columns +WHERE table_name = 'company' AND table_schema = 'a76' +AND column_name IN ('has_express_line', 'is_service_company'); +""" + +try: + with engine.connect() as conn: + result = conn.execute(text(sql)) + for row in result: + print(f"Column: {row[0]}, Type: {row[1]}, Length: {row[2]}") +except Exception as e: + print(f"Error: {e}") diff --git a/backend/verify_logo_presence.py b/backend/verify_logo_presence.py new file mode 100644 index 00000000..9ccabeb1 --- /dev/null +++ b/backend/verify_logo_presence.py @@ -0,0 +1,32 @@ +from api.v1.modules.a76.general_catalogs.company.models import Company +from api.v1.modules.a76.general_catalogs.company.service import CompanyService +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker, Session +import os + +# Connect to DB (adjust for localhost) +DB_URL = "postgresql://postgres:postgres@localhost:5432/anexo76_core" +engine = create_engine(DB_URL) +SessionLocal = sessionmaker(bind=engine) +session = SessionLocal() + +try: + print("Querying first company...") + company = session.query(Company).first() + if company: + print(f"Company ID: {company.id}") + print(f"Direct Logo Access: '{company.logo}'") + + service = CompanyService(session) + flattened = service.flatten_company_dto(company) + + print(f"Flattened Logo: '{flattened.get('logo')}'") + + has_logo = 'logo' in flattened + print(f"Is 'logo' key in dict?: {has_logo}") + else: + print("No companies found in DB.") +except Exception as e: + print(f"Error: {e}") +finally: + session.close()