transdesk-importer-python-sdk 1.2.0

SDK de leitura/transformacao de planilhas de apolices Transdesk para o formato interno 77seg

transdesk-importer-python-sdk

SDK Python para leitura e transformacao de planilhas de apolices da Transdesk no formato interno consumido pela 77seg-core-api.

O SDK e puro: ele apenas le, transforma e valida a planilha, devolvendo modelos tipados. Ele nao grava no banco, nao cadastra leads/corretoras e nao gera link de assinatura -- essas responsabilidades ficam na core-api (ver Referencia de enrichment).

Instalacao

pip install transdesk-importer-python-sdk

Requer Python >= 3.11.

Uso

A entrada pode ser bytes (upload em memoria), um objeto file-like binario ou um caminho de arquivo. O formato (.xls/.xlsx) e detectado automaticamente por magic bytes -- nao depende da extensao.

from transdesk.importer import TransdeskImporter

# 1) a partir de bytes (ex.: upload de um endpoint)
file_bytes = request.files["file"].file.read()
result = TransdeskImporter().import_policies(file_bytes)

# 2) a partir de um caminho
result = TransdeskImporter().import_policies("apolices.xls")

if not result.is_valid:
    # result.errors -> list[ValidationError]
    for err in result.errors:
        print(err)
else:
    for policy in result.policies:        # list[InternalPolicy]
        print(policy.reference_id, len(policy.data.items))

# serializacao pronta para JSON
payload = result.to_dict()   # dict
as_json = result.to_json()   # str (ensure_ascii=False)

Retorno

import_policies(file) devolve um ImportResult:

Campo	Tipo	Descricao
policies	list[InternalPolicy]	Apolices ja no formato interno
errors	list[ValidationError]	Divergencias encontradas na validacao
is_valid	bool	True quando errors esta vazio

Helpers: result.to_dict() e result.to_json(**kwargs).

Principais modelos

• Saida (EN): InternalPolicy, PolicyData, ResellerData, Broker, Lead, CustomerInfo, Telephone, Mobile, Address, InternalItem.
• Canonico (PT, em InternalPolicy.original_data): CanonicalPolicy, CanonicalItem, Subestipulante, Cliente, Endereco, DadosBancarios, UnidadeVenda, Cobertura, Complemento.

InternalItem tem um schema "achatado" e consistente: todos os campos de cobertura e de risco estao sempre presentes; os que nao se aplicam ao tipo do item (ex.: campos de vida num veiculo) vem como null.

Campos adicionais de veiculo/reboque (colunas apos CHAVE_PIX na planilha):

Coluna planilha	Campo InternalItem	Observacao
EMPRESA_RASTREAMENTO	vehicle_tracker_company	Somente quando has_vehicle_tracker
CATEGORIA	vehicle_classification	Classificacao do veiculo (int)
MARCA	brand_name / brand	Veiculo; reboque usa MARCA_REBOQUE
MODELO	model_description / model	Veiculo; reboque usa MODELO_REBOQUE
RENAVAM	renavam
COR	color
ANO	model_year, manufacture_year	Mesmo valor nos dois campos
EIXOS	axles
CAMBIO	transmission
ALIENADO	is_alienated	N/A ou vazio = false

Validacao

O ValidationError cobre tres tipos:

• policy_count -- divergencia na contagem de apolices;
• item_count -- divergencia na contagem de itens de uma apolice;
• premium -- divergencia (apos arredondamento) entre o premio somado no canonico e o premio somado no formato interno de um item.

Arquitetura

planilha (bytes/file-like/path)
        |
        v
   ExcelReader            -> DataFrame
        |
        v
   ApoliceBuilder         -> list[CanonicalPolicy]   (camada canonica, PT)
        |
        v
   InternalMapper         -> list[InternalPolicy]    (camada interna, EN)
        |
        v
   InternalPolicyValidator-> list[ValidationError]
        |
        v
   ImportResult (policies + errors)

Desenvolvimento

python -m venv venv && source venv/bin/activate
pip install -e ".[dev]"
pytest

Publicacao: criar uma tag vX.Y.Z dispara o workflow de publish no PyPI (.github/workflows/publish.yml).

Referencia de enrichment (core-api)

Esta secao nao faz parte do SDK. Ela preserva, como referencia, a logica de "enrichment" e persistencia que saiu do importer e deve ser portada para a 77seg-core-api.

Apos obter as InternalPolicy do SDK, a core-api deve:

1. Completar os dados de cada apolice (cadastrar lead e unidade/corretora, injetar enterprise_id/reseller_id e os defaults da venda);
2. Persistir os registros das vendas no banco;
3. Gerar o link de assinatura (Paperless) e atualizar o status para pending_signature.

Codigo original (removido do SDK), que serve de base para a implementacao na core-api:

class InternalImportPipeline:

    def __init__(self, enterprise_id, reseller_id):
        self.enterprise_id = enterprise_id
        self.reseller_id = reseller_id

    def execute(self, policies):
        policies = [self._complete_data(policy) for policy in policies]
        policies = self._create_db_records(policies)
        policies = self._generate_signature_link(policies)
        return policies

    def _complete_data(self, policy):
        lead = policy.get('data').get('lead')
        # TODO: gerar o cadastro do lead no BD

        broker = policy.get('data').get('reseller').get('broker')
        # TODO: gerar o cadastro da unidade (corretora) no BD
        # TODO: gerar o cadastro do vendedor padrao da unidade (corretora) no BD

        # preencher os dados faltantes
        policy.update({
            'enterprise': {'id': self.enterprise_id},
            'reseller': {'id': self.reseller_id, 'person': broker},
            'lead': lead,
            'status': 'quotation_completed',
            'payment_method': 'bank_slip',
            'multi': True,
            'active': True,
            'deleted': False
        })
        return policy

    def _create_db_records(self, policies):
        # TODO: insere os registros no BD
        return policies

    def _generate_signature_link(self, policies):
        # TODO: gera o link de assinatura (via Paperless) para cada venda
        # TODO: atualiza o status da venda p/ pending_signature
        return policies

Ponto de integracao na core-api (esboco):

from transdesk.importer import TransdeskImporter

result = TransdeskImporter().import_policies(file_bytes)
if not result.is_valid:
    ...  # retornar result.errors

policies = [p.to_dict() if hasattr(p, "to_dict") else p for p in result.policies]
# aplicar _complete_data / _create_db_records / _generate_signature_link aqui