Biblioteca em Python para lidar com os processos do eSocial:
- Validação dos XML's dos eventos;
- Comunicação com o Webservices do eSocial para envio e consulta de lotes;
- Assinatura dos XML's (e conexão com o webservices) com certificado tipo
A1.
Apesar desta biblioteca ter sido desenvolvida para lidar especialmente com os eventos de SST (Saúde e Segurança do Trabalho), nada impede que ela possa ser utilizada para enviar/validar quaisquer dos eventos disponíveis no projeto eSocial.
No momento só é possível utilizar assinaturas do tipo A1 em arquivos no formato PKCS#12 (geralmente arquivos com extensão .pfx ou .p12).
PyPi:
pip install libesocial
A versão mais recente diretamente do repositório:
pip install https://github.com/qualitaocupacional/libesocial/archive/main.zip
Ou você pode clonar este repositório:
git clone https://github.com/qualitaocupacional/libesocial
Entrar na pasta do repositório recém clonado:
> cd libesocial
> python setup.py install
Montando um Lote e transmitindo para o eSocial
import esocial.xml
import esocial.client
ide_empregador = {
'tpInsc': 1,
'nrInsc': '12345678901234' # CNPJ/CPF completo (com 14/11 dígitos)
}
ide_transmissor = {
'tpInsc': 1,
'nrInsc': '43210987654321' # CNPJ/CPF completo (com 14/11 dígitos)
}
esocial_ws = esocial.client.WSClient(
pfx_file='caminho/para/o/arquivo/certificado/A1',
pfx_passw='senha do arquivo de certificado',
employer_id=ide_empregador,
# Se o transmissor é o próprio empregador, não precisa informar o "sender_id"
sender_id=ide_transmissor,
)
evento1_grupo1 = esocial.xml.load_fromfile('evento1.xml')
evento2_grupo1 = esocial.xml.load_fromfile('evento2.xml')
# Adicionando eventos ao lote. O evento já vai ser assinado usando o certificado fornecido e validado contra o XSD do evento
# Se gen_event_id == True, o Id do evento é gerado pela lib (default = False)
evento1_id, evento1_assinado = esocial_ws.add_event(evento1_grupo1, gen_event_id=True)
evento2_id, evento2_assinado = esocial_ws.add_event(evento2_grupo1, gen_event_id=True)
result, batch_xml = esocial_ws.send(group_id=1)
# result vai ser um Element object
# <Element {http://www.esocial.gov.br/schema/lote/eventos/envio/retornoEnvio/v1_1_0}eSocial at 0x>
print(esocial.xml.dump_tostring(result, xml_declaration=False, pretty_print=True))
# batch_xml vai ser um Element object com o XML de envio de lote
# <Element {http://www.esocial.gov.br/schema/lote/eventos/envio/v1_1_1}eSocial at 0x>
print(esocial.xml.dump_tostring(batch_xml, xml_declaration=False, pretty_print=True))Consultando o resultado do processamento de um Lote
import esocial.xml
import esocial.client
ide_empregador = {
'tpInsc': 1,
'nrInsc': '12345678901234' # CNPJ/CPF completo (com 14/11 dígitos)
}
ide_transmissor = {
'tpInsc': 1,
'nrInsc': '43210987654321' # CNPJ/CPF completo (com 14/11 dígitos)
}
esocial_ws = esocial.client.WSClient(
pfx_file='caminho/para/o/arquivo/certificado/A1',
pfx_passw='senha do arquivo de certificado',
employer_id=ide_empregador,
# Se o transmissor é o próprio empregador, não precisa informar o "sender_id"
sender_id=ide_transmissor,
)
# De posse do número do protocolo de envio
response = esocial_ws.retrieve('1.2.202109.0000000000000000001')
# response vai ser um Element object
#<Element {http://www.esocial.gov.br/schema/lote/eventos/envio/retornoProcessamento/v1_3_0}eSocial at 0x>
print(esocial.xml.dump_tostring(result, xml_declaration=False, pretty_print=True))Para obter algumas informações relevantes da resposta, use o método decode_response(response):
import esocial.xml
import json
response_decoded = esocial.xml.decode_response(response)
print(json.dumps(response_decoded.toDict(), indent=4))Exemplo de Saída
{
"status": {
"ocorrencias": [],
"cdResposta": "201",
"descResposta": "Lote processado com sucesso."
},
"lote": {
"dhRecepcao": "2021-10-04T11:45:44.16",
"versaoAplicativoRecepcao": "0.1.105",
"protocoloEnvio": "1.1.202110.0000000000011111111"
},
"eventos": [
{
"id": "ID1123456780000002021100411454300001",
"processamento": {
"ocorrencias": [],
"cdResposta": "201",
"descResposta": "Sucesso.",
"versaoAppProcessamento": "13.3.1",
"dhProcessamento": "2021-10-04T11:45:50.923"
},
"recibo": {
"nrRecibo": "1.1.0000000000111111111",
"hash": "GeGBSm+RjCxk53xh1oLQ22FDIR2Je3SQ6emcYGDm0Bo="
}
},
{
"id": "ID1123456780000002021100411454300002",
"processamento": {
"ocorrencias": [],
"cdResposta": "201",
"descResposta": "Sucesso.",
"versaoAppProcessamento": "13.3.1",
"dhProcessamento": "2021-10-04T11:45:51.56"
},
"recibo": {
"nrRecibo": "1.1.0000000000111111112",
"hash": "EqjMGQU5vPfT1qu24HIO/yn06DrLwA5IFJKP04mNedE="
}
},
{
"id": "ID1123456780000002021100411454300003",
"processamento": {
"ocorrencias": [],
"cdResposta": "201",
"descResposta": "Sucesso.",
"versaoAppProcessamento": "13.3.1",
"dhProcessamento": "2021-10-04T11:45:52.243"
},
"recibo": {
"nrRecibo": "1.2.0000000000111111113",
"hash": "Lf9tQsGezML23RmWQYQg4Y+qzwn9BDAtyfyGiMfadfE="
}
},
{
"id": "ID1123456780000002021100411454300004",
"processamento": {
"ocorrencias": [],
"cdResposta": "201",
"descResposta": "Sucesso.",
"versaoAppProcessamento": "13.3.1",
"dhProcessamento": "2021-10-04T11:45:52.9"
},
"recibo": {
"nrRecibo": "1.1.0000000000111111114",
"hash": "zp8AJYm0uOoNTW+2oQEitCm0f6tIK8LbxqT8+Jel4rg="
}
}
]
}O retorno vai ser um objeto do tipo DotMap, que pode ser acessado assim:
print(response_decoded.status.cdResposta, '-', response_decoded.status.descResposta)
for evt in response_decoded.eventos:
print('ID:', evt.id)
print('Recibo:', evt.recibo.nrRecibo)
print('-'*10)Por padrão, o webservice de envio/consulta de lotes é o de "Produção Restrita", para enviar para o ambiente de "Produção Empresas", onde as coisas são para valer:
import esocial.client
esocial_ws = esocial.client.WSClient(
pfx_file='caminho/para/o/arquivo/certificado/A1',
pfx_passw='senha do arquivo de certificado',
employer_id=ide_empregador,
sender_id=ide_empregador,
target='production'
)
# OU usar os códigos do atributo "tpAmb", de acordo com a documentação:
# 1 = Produção
# 2 = Produção Restrita
esocial_ws = esocial.client.WSClient(
pfx_file='caminho/para/o/arquivo/certificado/A1',
pfx_passw='senha do arquivo de certificado',
employer_id=ide_empregador,
sender_id=ide_empregador,
target=1
)
...Assinando um evento
Se por algum motivo você precisar assinar algum arquivo XML separadamente, pode usar as funções utilitárias da LIBeSocial. Lembrando que o método "add_event(xml_element)" já faz a assinatura do evento antes de adicioná-lo ao lote.
import esocial.xml
import esocial.utils
cert_data = esocial.utils.pkcs12_data('my_cert_file.pfx', 'my password')
evt2220 = esocial.xml.load_fromfile('S2220.xml')
# Assina o XML com os algoritmos descritos na documentação do eSocial
evt2220_signed = esocial.xml.sign(evt2220, cert_data)Validando um evento
import esocial.xml
evt2220 = esocial.xml.load_fromfile('S2220.xml')
try:
esocial.xml.XMLValidate(evt2220).validate()
except esocial.xml.XMLValidateError as e:
print('O XML do evento S-2220 é inválido!')
print(e)
for err in e.errors:
print(' ->', err)ou
import esocial.xml
evt2220 = esocial.xml.load_fromfile('S2220.xml')
xmlschema = esocial.xml.XMLValidate(evt2220)
if xmlschema.isvalid():
print('XML do evento é válido! :-D.')
else:
print('O XML do evento S-2220 é inválido!')
print(str(xmlschema.last_error))De acordo com o manual do desenvolvedor do eSocial, versão 1.10 (página 114), é necessário instalar a cadeia de certificação do eSocial para poder utilizar os Webservices. Que são:
Raiz
Primeiro, o manual está desatualizado, sendo que os servidores do eSocial estão utilizando a versão 10 do certificado Raiz do ICP-Brasil:
Segundo, utilizando a LIBeSocial não há necessidade de instalar nenhum desses certificados na máquina que vai enviar os eventos. Os respectivos certificados já estão "agrupados" no arquivo "serpro_full_chain.pem" na pasta certs que acompanha a LIBeSocial.
Entretando, certificados expiram e/ou são trocados. Se necessário, para criar um novo arquivo com a cadeia de certificados novos, após baixar os devidos arquivos .crt (que devem estar no formato PEM), é só concatenar os arquivos em um único. Exemplo em Linux/Unix:
$ cat ICP-Brasilv10.crt AC_Secretaria_da_Receita_Federal_do_Brasil_v3.crt Autoridade_Certificadora_do_SERPRO_RFB_SSL.crt > novo_arquivo_certificados.pem
E informar esse arquivo ao instanciar o cliente esocial:
import esocial.client
esocial_ws = esocial.client.WSClient(
pfx_file='caminho/para/o/arquivo/certificado/A1',
pfx_passw='senha do arquivo de certificado',
employer_id=ide_empregador,
target='production',
ca_file='/caminho/para/novo_arquivo_certificados.pem',
)
...Instalar a pytest (ver requirements-dev.txt).
Na raíz do projeto, executar:
$ pytest
A LIBeSocial é um projeto de código aberto, desenvolvido pelo departamento de Pesquisa e Desenvolvimento e Tecnologia da Informação da Qualitá Segurança e Saúde Ocupacional e está licenciada pela Apache License 2.0.