import time from generic.gn_request import Request # from generic.gn_intelipost import gn_intelipost # from datetime import datetime import json import urllib3 import unidecode import requests from urllib.parse import urlparse import os urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class intelipost(gn_intelipost): def __init__(self, argv:list): self.response = None self.db = False self.output_count = 0 self.suffix = "app_generic" self.unparsed_data = {} self.setup(argv) self.rote_handler() environment = 'homolog' if self.environment_config.get('ambiente') == 'H' else '' urls = self.get_url_environment(environment=environment) apiKey = self.get_token_data('apiKey') self.intelipost_request = Request( base_url = urls.get('base_url'), params = {}, verify= False, headers={ 'api-key': f"{apiKey}", 'platform': 'SOFTDIB', "platform-version": '1.0', 'Content-Type': 'application/json' } ) super(intelipost, self, ).__init__(argv) print('Fim') def get_url_environment(self, environment='homolog'): return { "base_url" :f"https://api.intelipost.com.br/api/v1/" } ##### ##### ######## ##### ####### ## ## ######## ###### ## #### ###### ####### ####### ###### ####### ######## ####### ## ## ## ## # ## # ## ## ## ## ### ## # ## # ## ## #### ## ## ## ## ## ## ## ## ## ## ## # ## # ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## #### ## ##### ###### ## ## ## #### #### ##### #### ## #### ## ## ## ## ## ## ## ### ## #### ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##### ##### #### ##### ####### ## ## #### ### ## ## ## ###### ###### ####### #### ### ## ####### #### ####### """ https://api.intelipost.com.br/api/v1/freight_hub Criar cotação utilizando a Central de Frete """ def post_route_handler_freight_hub(self): # JSON de entrada (novo formato) # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/freight_hub.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # exit() # return data = self.txt_to_dict() for request in data.get('C'): self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_freight_hub(self, data): data['additional_information']['rule_tags'] = self.unparsed_data.get('add_inf_rule_tags', {}).split(",") data['products'] = [] if self.unparsed_data: for product in self.unparsed_data: if product.get('TIPO_REG', {}).get('valor') == 'P': product_data = { "weight": product.get('weight/valor', 0), "cost_of_goods": product.get('cost_of_goods/valor', 0), "width": product.get('width/valor', 0), "height": product.get('height/valor', 0), "length": product.get('length/valor', 0), "quantity": product.get('quantity/valor', 0), "sku_id": product.get('sku_id/valor', 0), "product_category": product.get('product_category/valor', 0), } data['products'].append(product_data) return data def handle_output_freight_hub(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações básicas da resposta content = response_data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" # Formata a linha base c = [ "OK", "", "C", content_get('id', ''), content_get('origin_zip_code', ''), content_get('destination_zip_code', ''), content_get('platform', ''), ",".join(content_get('additional_information.delivery_method_ids', [])), content_get('additional_information.client_type', ''), content_get('additional_information.sales_channel', ''), ",".join(content_get('additional_information.rule_tags', [])), content_get('quoting_mode', ''), content_get('skip_return_modes', False), content_get('skip_quote_rules', False), content_get('client_id', 0), content_get('client_name', ''), content_get('created', 0), content_get('total_weight', 0), content_get('cubic_volume', 0), content_get('cost_of_goods', 0), content_get('cached', False) ] retorno_cobol += "|".join(c) + "|\n" # Processa as opções de entrega for option in content.get('delivery_options', []): option_get = self.create_getter(option) d = [ "OK", "", "D", option_get('delivery_method_id', ''), option_get('delivery_estimate_business_days', 0), option_get('delivery_estimate_transit_time_business_days', ''), option_get('delivery_processing_time_business_days', ''), option_get('delivery_additional_transit_time_business_days', ''), option_get('provider_shipping_cost', 0), option_get('final_shipping_cost', 0), option_get('description', ''), option_get('removed_by_return_modes', False), option_get('removed_by_quote_rules', False), option_get('cubic_weight', ''), ",".join(map(str, option_get('pickup_addresses', []))), option_get('delivery_method_type', ''), option_get('delivery_method_name', ''), option_get('shown_to_client', True), option_get('pickup_enabled', False), option_get('scheduling_enabled', False) ] retorno_cobol += "|".join(d) + "|\n" # Processa os volumes for volume in content.get('volumes', []): volume_get = self.create_getter(volume) v = [ "OK", "", "V", volume_get('volume_type', ''), volume_get('weight', 0), volume_get('cost_of_goods', 0), volume_get('width', 0), volume_get('height', 0), volume_get('length', 0), volume_get('description', ''), ",".join(map(str, volume_get('sku_groups_ids', []))), volume_get('product_category', ''), volume_get('quantity_of_items', 0), volume_get('volume_cm3', 0) ] retorno_cobol += v.join("|") + "|\n" # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') ##### ##### ######## ###### ###### ##### ###### ###### ###### ##### ### ## ## ## ###### ###### ##### ###### ## ## ######## ## ## ## ## # ## # ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ### ## # ## # ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ##### ## ## ## ## ## ## ## ## ### ###### ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ##### #### ## ## ## ## ##### ## ## #### ## ## ##### ##### ## ## ## ## ### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##### ##### #### ### ### ## ##### ###### ### ###### ##### ### ## #### ### ### ##### ###### ## ## #### """ https://api.intelipost.com.br/api/v1/quote/pickup_quote_by_product Este método assemelhase com o de cotação por produtos, porém com a possibilidade de retorno de seus pudos previamente cadastrados. """ def post_route_handler_quote_pickup_quote_by_product(self): # JSON de entrada (novo formato expandido) # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/quote_pickup_quote_by_product.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', {}) # exit() # return data = self.txt_to_dict() dados = [] produtos = [] for request in data.get('C'): if request.get('TIPO_REG').get('valor') == 'C': dados.append(request) elif request.get('TIPO_REG').get('valor') == 'P': produtos.append(request) self.unparsed_products = produtos for request in dados: self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_quote_pickup_quote_by_product(self, data): data['additional_information']['rule_tags'] = self.unparsed_data.get('add_inf_rule_tags', {}).split(",") data['products'] = [] if self.unparsed_products: for product in self.unparsed_products: product_data = { "quantity": product.get('quantity/valor', 0), "length": product.get('length/valor', 0), "can_group": product.get('can_group/valor', ""), "weight": product.get('weight/valor', 0), "width": product.get('width/valor', 0), } data['products'].append(product_data) return data def handle_output_quote_pickup_quote_by_product(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações básicas da resposta content = response_data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" # Formata a linha base c = [ "OK", "", "C", content_get('id', ''), content_get('origin_zip_code', ''), content_get('destination_zip_code', ''), content_get('platform', ''), content_get('additional_information', {}).get('extra_cost_absolute', ''), content_get('additional_information', {}).get('lead_time_business_days', ''), content_get('additional_information', {}).get('free_shipping', ''), ",".join(map(str, content_get('additional_information', {}).get('delivery_method_ids', []))), content_get('additional_information', {}).get('extra_cost_percentage', ''), content_get('additional_information', {}).get('tax_id', ''), content_get('additional_information', {}).get('client_type', ''), content_get('additional_information', {}).get('sales_channel', ''), content_get('additional_information', {}).get('payment_type', ''), content_get('additional_information', {}).get('is_state_tax_payer', ''), content_get('additional_information', {}).get('shipped_date', ''), content_get('additional_information', {}).get('calculate_estimated_date_exact', ''), content_get('additional_information', {}).get('logistic_contract_ids', ''), ",".join(content_get('additional_information', {}).get('rule_tags', [])), content_get('identification', ''), content_get('quoting_mode', ''), content_get('shipment_type', ''), content_get('shipment_sub_type', ''), content_get('skip_return_modes', ''), content_get('skip_quote_id_rules', ''), content_get('client_id', ''), content_get('client_name', ''), content_get('created', ''), content_get('created_iso', ''), content_get('cached', '') ] retorno_cobol += c.join("|") + "\n" # Processa as opções de entrega for option in content.get('delivery_options', []): option_get = self.create_getter(option) d = [ "OK", "", "D", option_get('id', ''), option_get('delivery_method_id', ''), option_get('delivery_estimate_business_days', ''), option_get('delivery_estimate_transit_time_business_days', ''), option_get('delivery_processing_time_business_days', ''), option_get('delivery_additional_transit_time_business_days', ''), option_get('warehouse_handling_time_end', ''), option_get('warehouse_handling_time_end_iso', ''), option_get('warehouse_handling_time', ''), option_get('provider_shipping_cost', ''), option_get('final_shipping_cost', ''), option_get('description', ''), option_get('delivery_note', ''), option_get('removed_by_return_modes', ''), option_get('removed_by_quote_id_rules', ''), option_get('cubic_weight', ''), option_get('delivery_method_external_id', ''), len(option_get('pickup_addresses', [])), option_get('delivery_estimate_date_min', ''), option_get('delivery_estimate_date_min_iso', ''), option_get('delivery_estimate_date_exact', ''), option_get('delivery_estimate_date_exact_iso', ''), option_get('delivery_estimate_date_max', ''), option_get('delivery_estimate_date_max_iso', ''), option_get('delivery_method_type', ''), option_get('delivery_method_name', ''), option_get('shown_to_client', ''), option_get('logistic_provider_name', ''), option_get('pickup_enabled', ''), option_get('scheduling_enabled', '') ] retorno_cobol += d.join("|") + "\n" # Processa os endereços de pickup para esta opção de entrega for pickup_addr in option.get('pickup_addresses', []): pickup_addr_get = self.create_getter(pickup_addr) pa = [ "OK", "", "PA", option_get('delivery_method_id', ''), pickup_addr_get('name', ''), pickup_addr_get('pudo_id', ''), pickup_addr_get('pudo_external_id', ''), pickup_addr_get('pudo_type', ''), pickup_addr_get('street', ''), pickup_addr_get('number', ''), pickup_addr_get('reference', ''), pickup_addr_get('additional', ''), pickup_addr_get('country', ''), pickup_addr_get('state_code', ''), pickup_addr_get('city', ''), pickup_addr_get('quarter', ''), pickup_addr_get('zip_code', ''), pickup_addr_get('latitude', ''), pickup_addr_get('longitude', '') ] retorno_cobol += pa.join("|") + "\n" # Processa os volumes for volume in content.get('volumes', []): volume_get = self.create_getter(volume) v = [ "OK", "", "V", volume_get('volume_type', ''), volume_get('weight', ''), volume_get('cost_of_goods', ''), volume_get('width', ''), volume_get('height', ''), volume_get('length', ''), volume_get('description', ''), ",".join(map(str, volume_get('sku_groups_ids', []))), volume_get('product_category', ''), volume_get('quantity_of_items', '') ] retorno_cobol += v.join("|") + "\n" # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') ##### ##### ######## ###### ##### ###### ## ## ##### #### ## ## ## ## ####### ## ## ## ## # ## # ## ## ## ## ## ## ## ## ## ## ## ## ## ### ### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ####### ## ## ## ## ## ## ## ## ## ##### ## ## ## ## ## ## ## ## # ## #### ## ## ## ## ##### ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ##### ##### #### ### ##### ### ## ## ##### ###### #### ## ## ####### """ https://api.intelipost.com.br/api/v1/quote Este método cria uma nova cotação e retorna os custos de frete e o prazo de entrega de todas as transportadoras retornadas na cotação. O método de cotação pode ser solicitado durante o checkout ou em qualquer ponto do site em que deseje verificar o valor do frete. Para garantir que o valor de frete cobrado está correto, tenha sempre certeza que a cotação foi feita com os mesmos itens que serão enviados. Nossa recomendação é que a última cotação seja feita no momento do checkout. """ # def beforeGet_quote_id(self): # # retorno = "" # # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/get_quote_id.json", 'r', encoding='utf-8') as file: # # retorno = file.read() # # data = json.loads(retorno) # # exit() # # return def format_rote_quote_id(self, rote): return rote.format(**self.rote_data['path']) def quote_id_handler(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações básicas da resposta content = response_data.get('content', {}) content = data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" c = [ "OK", "", "C", content_get('id', ''), content_get('client_id', ''), self.extract_date_ymd(content_get('created_iso', '')), self.extract_time_hms(content_get('created_iso', '')), content_get('additional_information.extra_cost_absolute', ''), content_get('additional_information.extra_cost_percentage', '') ] retorno_cobol += "|".join(c) + "|\n" for option in content.get('delivery_options', []): option_get = self.create_getter(option) d = [ "OK", "", "D", option_get('delivery_method_id', ''), option_get('delivery_estimate_business_days', ''), option_get('delivery_estimate_transit_time_business_days', ''), option_get('delivery_processing_time_business_days', ''), option_get('delivery_additional_transit_time_business_days', ''), "", "", "", option_get('provider_shipping_cost', ''), option_get('final_shipping_cost', ''), option_get('description', ''), option_get('delivery_note', ''), option_get('cubic_weight', ''), option_get('delivery_method_type', ''), option_get('delivery_method_name', ''), option_get('logistic_provider_name', ''), "", "", "", option_get('scheduling_enabled', ''), "", option_get('shown_to_client', ''), option_get('pickup_enabled', ''), option_get('removed_by_return_modes'), option_get('removed_by_quote_rules', ''), option_get('delivery_method_external_id','') ] retorno_cobol += "|".join(d) + "|\n" for volume in content.get('volumes', []): volume_get = self.create_getter(volume) v = [ "OK", "", "V", volume_get('volume_type', ''), volume_get('weight', ''), volume_get('cost_of_goods', ''), volume_get('width', ''), volume_get('height', ''), volume_get('length', ''), volume_get('description', ''), volume_get('sku_groups_ids', ''), volume_get('product_category', ''), volume_get('quantity_of_items', ''), volume_get('volume_cm3','') ] retorno_cobol += "|".join(v) + "|\n" # SIMULAÇÃO DE RETORNO PARA TESTES # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') def post_route_handler_quote(self): # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/quote.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # exit() # return data = self.txt_to_dict() items = [] volumes = [] for request in data.get('C'): if request.get('TIPO_REG', {}).get('valor') == 'C': items.append(request) elif request.get('TIPO_REG', {}).get('valor') == 'V': volumes.append(request) for request in items: self.unparsed_data = request self.unparsed_volumes = volumes request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_quote(self, data): data['additional_information'] = {} data['additional_information']['delivery_method_ids'] = self.unparsed_data.get('add_inf_delivery_method_ids', {}).get('valor').split(",") data['volumes'] = [] if self.unparsed_volumes: for volume in self.unparsed_volumes: volume_data = { "weight": volume.get('weight').get('valor'), "volume_type": volume.get('volume_type',).get('valor'), "cost_of_goods": volume.get('cost_of_goods').get('valor'), "width": volume.get('width').get('valor'), "height": volume.get('height').get('valor'), "length": volume.get('length').get('valor'), "description": volume.get('description').get('valor'), } data['volumes'].append(volume_data) return data def handle_output_quote(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} req_transp = self.intelipost_request.doRequest( rote="info" ) transportadora = [] if req_transp.ok: response_transp = req_transp.json() if req_transp.text else {} content_transp = response_transp.get('content', {}) for transp in content_transp.get('delivery_methods', []): transp_get = self.create_getter(transp) transportadora.append({ "id": transp_get('id', ''), "cnpj": transp_get('logistics_provider_tax_id', '') }) # Extrai informações básicas da resposta status = response_data.get('status', 'UNKNOWN') messages = response_data.get('messages', []) content = response_data.get('content', {}) retorno_cobol = "" content_get = self.create_getter(content) # Formata a linha base c = [ "OK", "", "C" , content_get('id', '') , content_get('platform', '') , content_get('additional_information.extra_cost_absolute', '') , content_get('additional_information.lead_time_business_days', '') , content_get('additional_information.free_shipping', '') , ",".join(map(str, content_get('additional_information.delivery_method_ids', []))) , content_get('additional_information.extra_cost_percentage', '') , content_get('additional_information.tax_id', '') , content_get('additional_information.client_type', '') , content_get('additional_information.sales_channel', '') , content_get('quoting_mode', '') , content_get('client_id', '') , self.extract_date_ymd(content_get('created_iso', '')) , self.extract_time_hms(content_get('created_iso', '')) ] retorno_cobol += "|".join(c) + "|\n" # Processa as opções de entrega for option in content.get('delivery_options', []): option_get = self.create_getter(option) transportadora_cnpj = '' for transp in transportadora: if transp.get('id') == option_get('delivery_method_id', ''): transportadora_cnpj = transp.get('cnpj', '') d = [ "OK", "", "D" , option_get('delivery_method_id', '') , option_get('delivery_estimate_business_days', '') , option_get('delivery_estimate_transit_time_business_days', '') , option_get('delivery_processing_time_business_days', '') , option_get('delivery_additional_transit_time_business_days', '') , self.extract_date_ymd(option_get('warehouse_handling_time_end_iso', '')) , self.extract_time_hms(option_get('warehouse_handling_time_end_iso', '')) , option_get('warehouse_handling_time', '') , option_get('provider_shipping_cost', '') , option_get('final_shipping_cost', '') , option_get('description', '') , option_get('delivery_note', '') , option_get('cubic_weight', '') , option_get('delivery_method_type', '') , option_get('delivery_method_name', '') , option_get('logistic_provider_name', '') , transportadora_cnpj , self.extract_date_ymd(option_get('delivery_estimated_date_min_iso', '')) #, self.extract_time_hms(option_get('delivery_estimated_date_min_iso', '')) , self.extract_date_ymd(option_get('delivery_estimated_date_exact_iso', '')) #, self.extract_time_hms(option_get('delivery_estimated_date_exact_iso', '')) , self.extract_date_ymd(option_get('delivery_estimated_date_max_iso', '')) #, self.extract_time_hms(option_get('delivery_estimated_date_max_iso', '')) , option_get('scheduling_enabled', '') , option_get('collect_time', '') , option_get('shown_to_client','') , option_get('pickup_enabled','') , option_get('removed_by_return_modes','') , option_get('removed_by_quote_id_rules','') , option_get('delivery_method_external_id','') ] retorno_cobol += "|".join(d) + "|\n" # Processa os volumes for volume in content.get('volumes', []): volume_get = self.create_getter(volume) v = [ "OK", "", "V" , volume_get('volume_type', '') , volume_get('weight', '') , volume_get('cost_of_goods', '') , volume_get('width', '') , volume_get('height', '') , volume_get('length', '') , volume_get('description', '') , ",".join(volume_get('sku_groups_ids', [])) , volume_get('quantity_of_items', '') , volume_get('volume_cm3','') ] retorno_cobol += "|".join(v) + "|\n" flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') else: response_data = self.api_request.req.json() if self.api_request.req.text else {} saida_cobol = "ER|{messages}|" messages = response_data.get('messages', []) if messages: messages_list = [msg.get('text', '') for msg in messages if isinstance(msg, dict) and 'text' in msg] saida_cobol = saida_cobol.format(messages=",".join(messages_list)) else: saida_cobol = saida_cobol.format(messages="Erro desconhecido") flag = "w+" if self.output_count == 0 else "a+" self.write_file(unidecode.unidecode(saida_cobol), self.saida_cobol, flag=flag, encoding='iso-8859-1') ##### ##### ######## ###### ###### ##### ###### ## ## ######## ##### ## ## ## ## # ## # ## ## ## ## ## ## ## ## ## ## # ## # ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##### ## ## ## ## ## ## ## ## ## ## ## ## ## ##### #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##### ##### #### ### ### ## ##### ###### #### #### ##### """ https://api.intelipost.com.br/api/v1/quote_by_product Criar cotação por produto """ def post_route_handler_quote_by_product(self): # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/quote_by_product.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', {}) # exit() # return data = self.txt_to_dict() request = None products = [] for dado in data.get('C'): if dado.get('TIPO_REG').get('valor') == 'C': request = dado elif dado.get('TIPO_REG').get('valor') == 'P': products.append(dado) self.unparsed_products = products self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_quote_by_product(self, data): data['additional_information'] = {} data['additional_information']['rule_tags'] = self.unparsed_data.get('add_inf_rule_tags', {}).get('valor').split(",") data['products'] = [] if self.unparsed_products: for products in self.unparsed_products: products_data = { "weight": products.get("weight").get('valor'), "cost_of_goods": products.get("cost_of_goods").get('valor'), "width": products.get("width").get('valor'), "height": products.get("height").get('valor'), "length": products.get("length").get('valor'), "quantity": products.get("quantity").get('valor'), "sku_id": products.get("sku_id").get('valor'), "product_category": products.get("product_category").get('valor'), } data['products'].append(products_data) self.write_file("OK||C|123213213\nOK||T|123123123", self.saida_cobol, flag="w+", encoding='iso-8859-1') exit(0) return data def handle_output_quote_by_product(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações básicas da resposta content = response_data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" # Formata a linha base c = [ "OK", "", "C", content_get('id', ''), content_get('origin_zip_code', ''), content_get('destination_zip_code', ''), content_get('platform', ''), content_get('additional_information.extra_cost_absolute', ''), content_get('additional_information.lead_time_business_days', ''), content_get('additional_information.free_shipping', ''), ",".join(map(str, content_get('additional_information.delivery_method_ids', []))), content_get('additional_information.extra_cost_percentage', ''), content_get('additional_information.tax_id', ''), content_get('additional_information.client_type', ''), content_get('additional_information.sales_channel', ''), content_get('additional_information.payment_type', ''), content_get('additional_information.is_state_tax_payer', ''), content_get('quoting_mode', ''), content_get('client_id', ''), self.extract_date_ymd(content_get('created_iso', '')), self.extract_time_hms(content_get('created_iso', '')) ] retorno_cobol += "|".join(c) + "|\n" # Processa as opções de entrega for option in content.get('delivery_options', []): option_get = self.create_getter(option) d = [ "OK", "", "D", option_get('delivery_method_id', ''), option_get('delivery_estimate_business_days', ''), option_get('delivery_estimate_transit_time_business_days', ''), option_get('delivery_processing_time_business_days', ''), option_get('delivery_additional_transit_time_business_days', ''), self.treat_value(self.extract_date_ymd(option_get('warehouse_handling_time_end_iso', ''))), self.treat_value(self.extract_time_hms(option_get('warehouse_handling_time_end_iso', ''))), option_get('warehouse_handling_time', ''), option_get('provider_shipping_cost', ''), option_get('final_shipping_cost', ''), option_get('description', ''), option_get('delivery_note', ''), option_get('removed_by_return_modes', ''), option_get('removed_by_quote_id_rules', ''), option_get('cubic_weight', ''), self.extract_date_ymd(option_get('delivery_estimate_date_min_iso', '')), self.extract_time_hms(option_get('delivery_estimate_date_min_iso', '')), self.extract_date_ymd(option_get('delivery_estimate_date_exact_iso', '')), self.extract_time_hms(option_get('delivery_estimate_date_exact_iso', '')), self.extract_date_ymd(option_get('delivery_estimate_date_max_iso', '')), self.extract_time_hms(option_get('delivery_estimate_date_max_iso', '')), option_get('delivery_method_type', ''), option_get('delivery_method_name', ''), option_get('scheduling_enabled', ''), option_get('logistic_provider_name', ''), option_get('shown_to_client', ''), option_get('pickup_enabled', ''), option_get('collect_time', ''), ] retorno_cobol += "|".join(d) + "|\n" # Processa os volumes for volume in content.get('volumes', []): volume_get = self.create_getter(volume) v = [ "OK", "", "V", volume_get('weight', ''), volume_get('cost_of_goods', ''), volume_get('width', ''), volume_get('height', ''), volume_get('length', ''), volume_get('description', ''), ",".join(map(str, volume_get('sku_groups_ids', []))), volume_get('product_category', ''), volume_get('volume_type', '') ] retorno_cobol += "|".join(v) + "|\n" flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') ##### ###### ###### ## ###### ###### ####### ###### ###### ###### ##### ###### ####### ####### ## ## ######## ###### ####### ##### ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ### ## # ## # ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ##### ## ###### ##### ## ## #### ## ## ## ## ## ## ## ## ## #### #### ## #### ## ##### #### ## ###### ## #### ## ## ## #### ##### ## ## ## ## ## ## ## ## ## ## ## ## ## ### ## #### ## ## ### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##### ### ## ###### ## ## ### ## ### ####### ###### ###### ###### ##### ###### ####### ####### ## ## #### ### ## ####### ##### ## ## def post_route_handler_shipment_order(self): # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/shipment_order.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', {}) # exit() # return data = self.txt_to_dict() items = [] volumes = [] for request in data.get('C'): if request.get('TIPO_REG', {}).get('valor') == 'P': items.append(request) elif request.get('TIPO_REG', {}).get('valor') == 'V': volumes.append(request) for request in items: self.unparsed_data = request self.unparsed_volumes = volumes request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_shipment_order(self, data): data['shipment_order_volume_array'] = [] if self.unparsed_volumes: for volume in self.unparsed_volumes: volume_data = { "name": volume.get('ship_ord_vol_array_name').get('valor'), "shipment_order_volume_number": volume.get('ship_ord_vol_array_shipment_order_volume_number').get('valor'), "weight": volume.get('ship_ord_vol_array_weight').get('valor'), "volume_type_code": volume.get('ship_ord_vol_array_volume_type_code').get('valor'), "width": volume.get('ship_ord_vol_array_width').get('valor'), "height": volume.get('ship_ord_vol_array_height').get('valor'), "length": volume.get('ship_ord_vol_array_length').get('valor'), "products_nature": volume.get('ship_ord_vol_array_products_nature').get('valor'), "is_icms_exempt": self.to_boolean(volume.get('ship_ord_vol_array_is_icms_exempt', False).get('valor'), False), "tracking_code": volume.get('ship_ord_vol_array_tracking_code').get('valor'), } volume_data['shipment_order_volume_invoce'] = { "invoice_series": volume.get("ship_ord_vol_array_inv_invoice_series").get("valor"), "invoice_number": volume.get("ship_ord_vol_array_inv_invoice_number").get("valor"), "invoice_key": volume.get("ship_ord_vol_array_inv_invoice_key").get("valor"), "invoice_date": volume.get("ship_ord_vol_array_inv_invoice_date").get("valor"), "invoice_total_value": volume.get("ship_ord_vol_array_inv_invoice_total_value").get("valor"), "invoice_products_value": volume.get("ship_ord_vol_array_inv_invoice_products_value").get("valor"), "invoice_cfop": volume.get("ship_ord_vol_array_inv_invoice_cfop").get("valor") } volume['content_declaration'] = { "content_declaration_number": volume.get("ship_ord_vol_array_cont_decl_content_declaration_number").get("valor"), "content_declaration_total_value": volume.get("ship_ord_vol_array_cont_decl_content_declaration_total_value").get("valor"), "content_declaration_date": self.to_boolean(volume.get("ship_ord_vol_array_cont_decl_content_declaration_date").get("valor"), False) } data['shipment_order_volume_array'].append(volume_data) return data def handle_output_shipment_order(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} content = response_data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" p = [ "OK" , "" , "P" , content_get("id", "") , content_get("client_id", "") , content_get("quote_id", "") , content_get("delivery_method_id", "") , content_get("estimated_delivery_days_lp", "") , content_get("provider_shipping_costs", "") , content_get("customer_shipping_costs", "") #, content.get("order_number", "") #, content.get("sales_order_number", "") , self.extract_date_ymd(content_get("estimated_delivery_date_iso", "")) , self.extract_time_hms(content_get("estimated_delivery_date_iso", "")) , self.extract_date_ymd(content_get("estimated_delivery_date_lp_iso", "")) , self.extract_time_hms(content_get("estimated_delivery_date_lp_iso", "")) , self.extract_date_ymd(content_get("shipped_date_iso", "")) , self.extract_time_hms(content_get("shipped_date_iso", "")) , self.extract_date_ymd(content_get("created_iso", "")) , self.extract_time_hms(content_get("created_iso", "")) , self.extract_date_ymd(content_get("modified_iso", "")) , self.extract_time_hms(content_get("modified_iso", "")) , content_get("carrier.driver.federal_tax_id", "") , content_get("carrier.driver.license", "") , content_get("carrier.driver.first_name", "") , content_get("carrier.driver.last_name", "") , content_get("carrier.driver.email", "") , content_get("carrier.driver.cell_phone", "") , content_get("carrier.vehicle.licence_plate", "") , content_get("carrier.vehicle.manufacturer", "") , content_get("carrier.vehicle.model", "") , content_get("carrier.vehicle.color", "") , content_get("carrier.vehicle.category", "") , content_get("end_customer.first_name", "") , content_get("end_customer.last_name", "") , content_get("end_customer.email", "") , content_get("end_customer.phone", "") , content_get("end_customer.cellphone", "") , content_get("end_customer.is_company", "") , content_get("end_customer.federal_tax_payer_id", "") , content_get("end_customer.state_tax_payer_id", "") , content_get("end_customer.shipping_address", "") , content_get("end_customer.shipping_number", "") , content_get("end_customer.shipping_additional", "") , content_get("end_customer.shipping_reference", "") , content_get("end_customer.shipping_quarter", "") , content_get("end_customer.shipping_city", "") , content_get("end_customer.shipping_zip_code", "") #, self.treat_value( end_customer.get("shipping_state", "")) , content_get("end_customer.shipping_state_code", "") , content_get("end_customer.shipping_country", "") , content_get("origin_warehouse_code", "") , content_get("origin_name", "") , content_get("origin_federal_tax_payer_id", "") , content_get("origin_customer_phone", "") , content_get("origin_customer_email", "") , content_get("origin_street", "") , content_get("origin_number", "") #, content_get("origin_additional", "") #, content_get("origin_reference", "") , content_get("origin_zip_code", "") , content_get("origin_city", "") , content_get("origin_quarter", "") , content_get("origin_state_code", "") , content_get("shipment_order_type", "") , content_get("parent_shipment_order_number", "") #, content_get("sales_channel", "") , content_get("scheduling_window_start", "") , content_get("scheduling_window_end", "") , content_get("scheduled", "") , content_get("shipment_order_volume_state", "") , content_get("delivered_date", "") , content_get("additional_information.key", "") , content_get("additional_information.key2", "") , content_get("delivery_address.name", "") , content_get("delivery_address.street", "") , content_get("delivery_address.number", "") , content_get("delivery_address.reference", "") , content_get("delivery_address.additional", "") , content_get("delivery_address.country", "") , content_get("delivery_address.state_code", "") , content_get("delivery_address.city", "") , content_get("delivery_address.quarter", "") , content_get("delivery_address.zip_code", "") , content_get("delivery_address.latitude", "") , content_get("delivery_address.longitude", "") ] retorno_cobol += "|".join(p) + "|\n" # shipment_order_volume_array for volume in content.get('shipment_order_volume_array', []): history = volume.get('shipment_order_volume_state_history_array', []) volume_get = self.create_getter(volume) v = [ "OK" , "" , "V" , volume_get("shipment_order_volume_number", "") , volume_get("shipment_order_id", "") , volume_get("shipment_order_volume_state", "") , volume_get("weight", "") , volume_get("volume_type_code", "") , volume_get("width", "") , volume_get("height", "") , volume_get("length", "") , volume_get("products_nature", "") , volume_get("products_quantity", "") , volume_get("is_icms_exempt", "") , self.extract_date_ymd(volume_get("created_iso", "")) , self.extract_time_hms(volume_get("created_iso", "")) , self.extract_date_ymd(volume_get("modified_iso", "")) , self.extract_time_hms(volume_get("modified_iso", "")) , volume_get("logistics_provider_data", "") , volume_get("logistic_provider_tracking_code", "") , volume_get("estimated_delivery_date_lp", "") , volume_get("estimated_delivery_date_lp_iso", "") , volume_get("shipment_order_volume_invoice.shipment_order_volume_number", "") , volume_get("shipment_order_volume_invoice.shipment_order_volume_id", "") , volume_get("shipment_order_volume_invoice.invoice_series", "") , volume_get("shipment_order_volume_invoice.invoice_number", "") , volume_get("shipment_order_volume_invoice.invoice_key", "") , self.extract_date_ymd(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , self.extract_time_hms(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , volume_get("shipment_order_volume_invoice.invoice_total_value", "") , volume_get("shipment_order_volume_invoice.invoice_products_value", "") , volume_get("shipment_order_volume_invoice.invoice_cfop", "") , volume_get("pre_shipment_list_state", "") , volume_get("has_clarify_delivery_fail", "") , volume_get("delivered_late", "") , volume_get("delivered_late_lp", "") , self.extract_date_ymd(volume_get("estimated_delivery_date_iso", "")) , self.extract_time_hms(volume_get("estimated_delivery_date_iso", "")) , volume_get("delivered", "") , volume_get("pre_shipment_list_id", "") , volume_get("logistic_provider_pre_shipment_list_id", "") , volume_get("name", "") , self.extract_date_ymd(volume_get("original_estimated_delivery_date_iso", "")) , self.extract_time_hms(volume_get("original_estimated_delivery_date_iso", "")) , self.extract_date_ymd(volume_get("original_estimated_delivery_date_lp_iso", "")) , self.extract_time_hms(volume_get("original_estimated_delivery_date_lp_iso", "")) , self.extract_date_ymd(volume_get("shipped_date_iso", "")) , self.extract_time_hms(volume_get("shipped_date_iso", "")) , volume_get("delivered_date", "") , volume_get("shipment_order_volume_state_localized", "") , volume_get("shipment_order_volume_id", "") , volume_get("tracking_code", "") ] retorno_cobol += "|".join(v) + "|\n" for product in volume.get('products', []): product_get = self.create_getter(product) vp = [ "OK" , "" , "VP" , product_get("weight") , product_get("width") , product_get("height") , product_get("length") , product_get("price") , product_get("description") , product_get("sku") , product_get("category") , product_get("quantity") , product_get("image_url") ] retorno_cobol += "|".join(vp) + "|\n" # shipment_order_volume_state_history_array for history in content.get('shipment_order_volume_state_history_array', []): history_get = self.create_getter(history) vh = [ "VH" , "" , history_get("shipment_order_volume_id", "") , history_get("shipment_order_volume_state", "") , history_get("tracking_state", "") , self.extract_date_ymd(history_get("created_iso", "")) , self.extract_time_hms(history_get("created_iso", "")) , history_get("provider_message", "") , history_get("provider_state", "") , history_get("esprinter_message", "") , history_get("shipment_volume_micro_state", "") , history_get("shipment_order_volume_state_localized", "") , history_get("shipment_order_volume_state_history", "") , self.extract_date_ymd(history_get("event_date_iso", "")) , self.extract_time_hms(history_get("event_date_iso", "")) ] retorno_cobol += "|".join(vh) + "|\n" # Escreve o resultado no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') else: response_data = self.api_request.req.json() if self.api_request.req.text else {} saida_cobol = "ER|{messages}|" messages = response_data.get('messages', []) # {'type': 'ERROR', 'text': 'Número do pedido já existe 123456.', 'key': 'shipmentOrder.save.already.existing.order.number'} if messages: messages_list = [msg.get('text', '') for msg in messages if isinstance(msg, dict) and 'text' in msg] saida_cobol = saida_cobol.format(messages=",".join(messages_list)) else: saida_cobol = saida_cobol.format(messages="Erro desconhecido") flag = "w+" if self.output_count == 0 else "a+" self.write_file(unidecode.unidecode(saida_cobol), self.saida_cobol, flag=flag, encoding='iso-8859-1') def format_rote_shipment_order_id(self,rote): return rote.format(**self.rote_data['path']) def shipment_order_id_handler(self, data): # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/shipment_order.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', {}) if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} content = response_data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" p = [ "OK" , "" , "P" , content_get("id", "") , content_get("client_id", "") # não existe na estrutura , content_get("quote_id", "") # não existe na estrutura , content_get("delivery_method_id", "") , content_get("estimated_delivery_days_lp", "") , "" # provider_shipping_costs - não existe na estrutura 2 , content_get("customer_shipping_costs", "") , self.extract_date_ymd(content_get("estimated_delivery_date_iso", "")) , self.extract_time_hms(content_get("estimated_delivery_date_iso", "")) , "" # estimated_delivery_date_lp_iso date - não existe na estrutura 2 , "" # estimated_delivery_date_lp_iso time - não existe na estrutura 2 , self.extract_date_ymd(content_get("shipped_date_iso", "")) , self.extract_time_hms(content_get("shipped_date_iso", "")) , self.extract_date_ymd(content_get("created_iso", "")) , self.extract_time_hms(content_get("created_iso", "")) , self.extract_date_ymd(content_get("modified_iso", "")) , self.extract_time_hms(content_get("modified_iso", "")) , content_get("carrier.driver.federal_tax_id", "") , content_get("carrier.driver.license", "") , content_get("carrier.driver.first_name", "") , content_get("carrier.driver.last_name", "") , content_get("carrier.driver.email", "") , content_get("carrier.driver.cell_phone", "") , content_get("carrier.vehicle.licence_plate", "") , content_get("carrier.vehicle.manufacturer", "") , content_get("carrier.vehicle.model", "") , content_get("carrier.vehicle.color", "") , content_get("carrier.vehicle.category", "") , content_get("end_customer.first_name", "") , content_get("end_customer.last_name", "") , content_get("end_customer.email", "") , content_get("end_customer.phone", "") , content_get("end_customer.cellphone", "") , content_get("end_customer.is_company", "") , content_get("end_customer.federal_tax_payer_id", "") , content_get("end_customer.state_tax_payer_id", "") , content_get("end_customer.shipping_address", "") , content_get("end_customer.shipping_number", "") , content_get("end_customer.shipping_additional", "") , content_get("end_customer.shipping_reference", "") , content_get("end_customer.shipping_quarter", "") , content_get("end_customer.shipping_city", "") , content_get("end_customer.shipping_zip_code", "") , content_get("end_customer.shipping_state_code", "") , content_get("end_customer.shipping_country", "") , "" # origin_warehouse_code - não existe na estrutura 2 , content_get("origin_name", "") , content_get("origin_federal_tax_payer_id", "") , content_get("origin_customer_phone", "") , content_get("origin_customer_email", "") , content_get("origin_street", "") , content_get("origin_number", "") , content_get("origin_zip_code", "") , content_get("origin_city", "") , content_get("origin_quarter", "") , content_get("origin_state_code", "") , content_get("shipment_order_type", "") , "" # parent_shipment_order_number - não existe na estrutura 2 , content_get("scheduling_window_start", "") # não existe na estrutura , content_get("scheduling_window_end", "") # não existe na estrutura , content_get("scheduled", "") , "" # shipment_order_volume_state - não existe na estrutura 2 , "" # delivered_date - não existe na estrutura 2 , content_get("additional_information.key", "") , content_get("additional_information.key2", "") , "" # delivery_address.name - não existe na estrutura 2 , "" # delivery_address.street - não existe na estrutura 2 , "" # delivery_address.number - não existe na estrutura 2 , "" # delivery_address.reference - não existe na estrutura 2 , "" # delivery_address.additional - não existe na estrutura 2 , "" # delivery_address.country - não existe na estrutura 2 , "" # delivery_address.state_code - não existe na estrutura 2 , "" # delivery_address.city - não existe na estrutura 2 , "" # delivery_address.quarter - não existe na estrutura 2 , "" # delivery_address.zip_code - não existe na estrutura 2 , "" # delivery_address.latitude - não existe na estrutura 2 , "" # delivery_address.longitude - não existe na estrutura 2 # Campos adicionais da estrutura 2 que não existem na primeira: , content_get("platform", "") , content_get("delivery_method_external_id", "") , content_get("order_number", "") , content_get("end_customer.shipping_state", "") , content_get("warehouse_address_id", "") , content_get("sales_channel", "") , content_get("delivery_method_name", "") , content_get("sales_order_number", "") , content_get("logistic_provider_name", "") , content_get("external_order_numbers.erp", "") , content_get("external_order_numbers.marketplace", "") , content_get("external_order_numbers.plataforma", "") , content_get("external_order_numbers.sales_channel", "") , content_get("tracking_url", "") , content_get("verification_pickup.code", "") ] retorno_cobol += "|".join(p) + "|\n" for volume in content.get('shipment_order_volume_array', []): volume_get = self.create_getter(volume) v = [ "OK" , "" , "V" , volume_get("shipment_order_volume_number", "") , volume_get("shipment_order_id", "") , volume_get("shipment_order_volume_state", "") , volume_get("weight", "") , volume_get("volume_type_code", "") , volume_get("width", "") , volume_get("height", "") , volume_get("length", "") , volume_get("products_nature", "") , volume_get("products_quantity", "") , volume_get("is_icms_exempt", "") , self.extract_date_ymd(volume_get("created_iso", "")) , self.extract_time_hms(volume_get("created_iso", "")) , self.extract_date_ymd(volume_get("modified_iso", "")) , self.extract_time_hms(volume_get("modified_iso", "")) , volume_get("logistics_provider_data", "") , volume_get("logistic_provider_tracking_code", "") , volume_get("estimated_delivery_date_lp", "") , "" # estimated_delivery_date_lp_iso - não existe na estrutura 2 , volume_get("shipment_order_volume_invoice.shipment_order_volume_number", "") , volume_get("shipment_order_volume_invoice.shipment_order_volume_id", "") , volume_get("shipment_order_volume_invoice.invoice_series", "") , volume_get("shipment_order_volume_invoice.invoice_number", "") , volume_get("shipment_order_volume_invoice.invoice_key", "") , self.extract_date_ymd(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , self.extract_time_hms(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , volume_get("shipment_order_volume_invoice.invoice_total_value", "") , volume_get("shipment_order_volume_invoice.invoice_products_value", "") , volume_get("shipment_order_volume_invoice.invoice_cfop", "") , volume_get("pre_shipment_list_state", "") , volume_get("has_clarify_delivery_fail", "") , volume_get("delivered_late", "") , volume_get("delivered_late_lp", "") , self.extract_date_ymd(volume_get("estimated_delivery_date_iso", "")) , self.extract_time_hms(volume_get("estimated_delivery_date_iso", "")) , volume_get("delivered", "") , volume_get("pre_shipment_list_id", "") , volume_get("logistic_provider_pre_shipment_list_id", "") , volume_get("name", "") , self.extract_date_ymd(volume_get("original_estimated_delivery_date_iso", "")) , self.extract_time_hms(volume_get("original_estimated_delivery_date_iso", "")) , "" # original_estimated_delivery_date_lp_iso date - não existe na estrutura 2 , "" # original_estimated_delivery_date_lp_iso time - não existe na estrutura 2 , self.extract_date_ymd(volume_get("shipped_date_iso", "")) , self.extract_time_hms(volume_get("shipped_date_iso", "")) , volume_get("delivered_date", "") , volume_get("shipment_order_volume_state_localized", "") , volume_get("shipment_order_volume_id", "") , volume_get("tracking_code", "") # Campos adicionais da estrutura 2 que não existem na primeira: , volume_get("logistic_provider_label_hash", "") , volume_get("packaging_code", "") , volume_get("original_estimated_delivery_date_lp", "") , volume_get("estimated_delivery_days_lp", "") , volume_get("client_pre_shipment_list", "") , volume_get("content_declaration", "") ] retorno_cobol += "|".join(v) + "|\n" for attachment in volume.get('attachments', []): va = [ "OK" , "" , "VA" # , attachment.get("id", "") # , attachment.get("name", "") # , attachment.get("url", "") # , attachment.get("type", "") # , attachment.get("size", "") ] retorno_cobol += "|".join(va) + "|\n" for product in volume.get('products', []): product_get = self.create_getter(product) vp = [ "OK" , "" , "VP" , product_get("weight", "") , product_get("width", "") , product_get("height", "") , product_get("length", "") , product_get("price", "") , product_get("description", "") , product_get("sku", "") , product_get("category", "") , product_get("quantity", "") , product_get("image_url", "") ] retorno_cobol += "|".join(vp) + "|\n" for vh in volume.get('shipment_order_volume_state_history_array', []): vh_get = self.create_getter(vh) vh_data = [ "OK" , "" , "VH" , vh_get("shipment_order_volume_id", "") , vh_get("shipment_order_volume_state", "") , vh_get("tracking_state", "") , self.extract_date_ymd(vh_get("created_iso", "")) , self.extract_time_hms(vh_get("created_iso", "")) , vh_get("provider_message", "") , vh_get("provider_state", "") , vh_get("esprinter_message", "") , "" # shipment_volume_micro_state - não existe mais como campo direto na estrutura 2 , vh_get("shipment_order_volume_state_localized", "") , vh_get("shipment_order_volume_state_history", "") , self.extract_date_ymd(vh_get("event_date_iso", "")) , self.extract_time_hms(vh_get("event_date_iso", "")) # Campos adicionais da estrutura 2 que não existem na primeira: , vh_get("shipper_provider_state", "") , vh_get("shipment_order_volume_micro_state.id", "") , vh_get("shipment_order_volume_micro_state.code", "") , vh_get("shipment_order_volume_micro_state.default_name", "") , vh_get("shipment_order_volume_micro_state.i18n_name", "") , vh_get("shipment_order_volume_micro_state.description", "") , vh_get("shipment_order_volume_micro_state.shipment_order_volume_state_id", "") , vh_get("shipment_order_volume_micro_state.shipment_volume_state_source_id", "") , vh_get("shipment_order_volume_micro_state.shipment_volume_state", "") , vh_get("shipment_order_volume_micro_state.shipment_volume_state_localized", "") , vh_get("shipment_order_volume_micro_state.name", "") , vh_get("location.address", "") , vh_get("location.number", "") , vh_get("location.additional", "") , vh_get("location.reference", "") , vh_get("location.city", "") , vh_get("location.state_code", "") , vh_get("location.quarter", "") , vh_get("location.zip_code", "") , vh_get("location.description", "") , vh_get("location.latitude", "") , vh_get("location.longitude", "") , vh_get("request_hash", "") , vh_get("request_origin", "") ] retorno_cobol += "|".join(vh_data) + "|\n" for vha in vh.get('attachments', []): vha_data = [ "OK" , "" , "VA" # , vha.get("id", "") # , vha.get("name", "") # , vha.get("url", "") # , vha.get("type", "") # , vha.get("size", "") ] retorno_cobol += "|".join(vha_data) + "|\n" # Escreve o resultado no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(unidecode.unidecode(retorno_cobol), self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() def put_route_handler_shipment_order_volume_pedido_volume(self): # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/shipment_order_volume_pedido_volume.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', {}) # exit() # return data = self.txt_to_dict() for request in data.get('C'): self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_shipment_order_volume_pedido_volume(self, data): # rota para considerar o mesmo pedido data['shipment_order_volume_numbers'] = self.rote_data['path']['volume'] self.current_rote = self.current_rote.format(**self.rote_data['path']) return data def format_rote_shipment_order_volume_pedido_volume(self, rote): return rote.format(**self.rote_data['path']) def handle_output_shipment_order_volume_pedido_volume(self, data): """ Método para tratar a resposta do pedido de volume na Intelipost. Retorna status, mensagens e número do volume. """ if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações da resposta content = response_data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" p = [ "OK" , "" , "P" , content_get("shipment_order_volume_id", "") ] retorno_cobol += "|".join(p) + "|\n" flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() def post_route_handler_shipment_order_set_volume(self): # retorno_cobol = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/shipment_order_set_volume.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # exit() # return data = self.txt_to_dict() items = [] volumes = [] for request in data.get('C', []): if request.get('TIPO_REG', {}).get('valor') == 'P': items.append(request) elif request.get('TIPO_REG', {}).get('valor') == 'V': volumes.append(request) for request in items: self.unparsed_data = request self.unparsed_volumes = volumes request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request #data_parsed = self.parser(request, request_structure) self.handle_send(data_parsed) def beforeSend_shipment_order_set_volume(self, data): data['shipment_order_volume_array'] = [] if self.unparsed_volumes: for volume in self.unparsed_volumes: volume_data = { "shipment_order_volume_number": volume.get("ship_ord_vol_array_shipment_order_volume_number").get("valor"), "weight": volume.get("ship_ord_vol_array_weight").get("valor"), "volume_type_code": volume.get("ship_ord_vol_array_volume_type_code").get("valor"), "width": volume.get("ship_ord_vol_array_width").get("valor"), "height": volume.get("ship_ord_vol_array_height").get("valor"), "length": volume.get("ship_ord_vol_array_length").get("valor"), "products_nature": volume.get("ship_ord_vol_array_products_nature").get("valor"), "products_quantity": volume.get("ship_ord_vol_array_products_quantity").get("valor"), "is_icms_exempt": volume.get("ship_ord_vol_array_is_icms_exempt").get("valor"), "tracking_code": volume.get("ship_ord_vol_array_tracking_code").get("valor"), } data['shipment_order_volume_array'].append(volume_data) return data def handle_output_shipment_order_set_volume(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} content = response_data.get('content', {}) content = data.get('content', []) for volume in content: volume_get = self.create_getter(volume) v = [ "OK" , "" , "V" , volume_get("shipment_order_volume_number", "") , volume_get("shipment_order_id", "") , volume_get("shipment_order_volume_state", "") , volume_get("weight", "") , volume_get("volume_type_code", "") , volume_get("width", "") , volume_get("height", "") , volume_get("length", "") , volume_get("products_nature", "") , volume_get("products_quantity", "") , volume_get("is_icms_exempt", "") , self.extract_date_ymd(volume_get("created_iso", "")) , self.extract_time_hms(volume_get("created_iso", "")) , self.extract_date_ymd(volume_get("modified_iso", "")) , self.extract_time_hms(volume_get("modified_iso", "")) , volume_get("logistics_provider_data", "") , volume_get("logistic_provider_tracking_code", "") , volume_get("estimated_delivery_date_lp", "") , volume_get("shipment_order_volume_invoice.shipment_order_volume_number", "") , volume_get("shipment_order_volume_invoice.shipment_order_volume_id", "") , volume_get("shipment_order_volume_invoice.invoice_series", "") , volume_get("shipment_order_volume_invoice.invoice_number", "") , volume_get("shipment_order_volume_invoice.invoice_key", "") , self.extract_date_ymd(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , self.extract_time_hms(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , volume_get("shipment_order_volume_invoice.invoice_total_value", "") , volume_get("shipment_order_volume_invoice.invoice_products_value", "") , volume_get("shipment_order_volume_invoice.invoice_cfop", "") , volume_get("pre_shipment_list_state", "") , volume_get("has_clarify_delivery_fail", "") , volume_get("delivered_late", "") , volume_get("delivered_late_lp", "") , volume_get("estimated_delivery_date", "") , volume_get("delivered", "") , volume_get("pre_shipment_list_id", "") , volume_get("logistic_provider_pre_shipment_list_id", "") , volume_get("name", "") , volume_get("original_estimated_delivery_date", "") , volume_get("original_estimated_delivery_date_lp", "") , volume_get("shipped_date", "") , volume_get("delivered_date", "") , volume_get("shipment_order_volume_state_localized", "") , volume_get("shipment_order_volume_id", "") , volume_get("tracking_code", "") ] retorno_cobol += "|".join(v) + "|\n" for vh in volume.get('shipment_order_volume_state_history_array', []): vh_get = self.create_getter(vh) vh_data = [ "OK" , "" , "VH" , vh_get("shipment_order_volume_id", "") , vh_get("shipment_order_volume_state", "") , vh_get("tracking_state", "") , self.extract_date_ymd(vh_get("created_iso", "")) , self.extract_time_hms(vh_get("created_iso", "")) , vh_get("provider_message", "") , vh_get("provider_state", "") , vh_get("esprinter_message", "") , vh_get("shipment_volume_micro_state", "") , vh_get("shipment_order_volume_state_localized", "") , vh_get("shipment_order_volume_state_history", "") , self.extract_date_ymd(vh_get("event_date_iso", "")) , self.extract_time_hms(vh_get("event_date_iso", "")) ] retorno_cobol += "|".join(vh_data) + "|\n" # Escreve o resultado no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() ##### ## ## ## ##### ####### #### ## ###### ###### ####### ###### ####### ## ## ######## ###### ####### ##### ## ## ## #### ### ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ### ## # ## # ## ## ## ## ## ## #### ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ###### ## #### ## #### ## ###### ##### ## ## #### ## ## #### ## #### ## ##### #### ## ###### ## ## ## ## ### ## ## ## ## ## #### ##### ## ## ## ## ## ### ## #### ## ## ### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##### ## ## ## ## ##### ####### ###### ## ## ### ## ### ####### ###### ####### ## ## #### ### ## ####### ##### ## ## def post_route_handler_shipment_order_cancel_shipment_order(self): # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/shipment_order_cancel.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', '') # retorno_cobol = "" # mensagens = "" # list_msgs = data.get('messages', []) # for msg in list_msgs: # mensagens += f"{msg.get('text', 'Sem mensagem')}," # # Template para status da resposta (tipo R) # response_status = "OK|{mensagens}|C|{content}" # # Formata a linha de status da resposta # retorno_cobol = response_status.format( # content=content, # mensagens=mensagens.rstrip(',') # ) # retorno_cobol += "\n" # # Escreve o retorno no arquivo de saída # flag = "w+" if self.output_count == 0 else "a+" # self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') # exit() # return data = self.txt_to_dict() for request in data.get('C'): self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request self.handle_send(data_parsed) def beforeSend_shipment_order_cancel_shipment_order(self, data): """ Método chamado antes de enviar a requisição para a Intelipost. Aqui você pode manipular os dados ou fazer validações adicionais. """ return data def handle_output_shipment_order_cancel_shipment_order(self, data): """ Método para tratar a resposta do cancelamento de pedido na Intelipost. Retorna status, mensagens e número do pedido cancelado. """ # Tratamento da resposta baseado no retorno da API if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações da resposta content = response_data.get('content', {}) retorno_cobol = "" mensagens = "" list_msgs = data.get('messages', []) for msg in list_msgs: mensagens += f"{msg.get('text', 'Sem mensagem')}," # Template para status da resposta (tipo R) response_status = "OK|{mensagens}|C|{content}" # Formata a linha de status da resposta retorno_cobol = response_status.format( content=content, mensagens=mensagens.rstrip(',') ) retorno_cobol += "\n" # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() ##### ##### ## ## ####### ###### ###### ## ## ## ###### ####### ## ## ## ## ###### ##### ## ## ## ## ### ## ## ## ## ## ## ### ### #### ## ## ## ## ### ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ####### ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## #### #### ## ##### ## # ## ###### ##### #### ## #### ## ## ## ## ## ## ## ## ## ### ## ## #### ## ## ## ## #### ## ## ### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ##### ##### ## ## #### ###### ### ## ## ## ## ## ### ## ####### ## ## ## ###### ##### """ https://api.intelipost.com.br/api/v1/shipment_order/multi/ready_for_shipment/with_date Essa API tem como objetivo marcar diversos pedidos com o status "Pronto para envio". Deve ser informado nessa API o número do pedido que se deseja alterar e a data/hora correspondente ao novo status. """ def post_route_handler_shipment_order_multi_ready_for_shipment_with_date(self): # retorno = "OK||C|" # # salvar o retorno no arquivo de saída # flag = "w+" if self.output_count == 0 else "a+" # self.write_file(retorno + "\n", self.saida_cobol, flag=flag, encoding='iso-8859-1') # exit() data = self.txt_to_dict() for request in data.get('C'): self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request self.handle_send(data_parsed) def beforeSend_shipment_order_multi_ready_for_shipment_with_date(self, data): return [ data ] def handle_output_shipment_order_multi_ready_for_shipment_with_date(self, data): # Tratamento da resposta baseado no retorno da API if self.api_request.req.ok: retorno_cobol = "OK||C|" flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() ## ###### ###### ###### ## ###### ##### ##### ####### ###### ##### ##### ## ###### ##### #### ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ###### ## ## ## ## ## ## ###### ## ## ## ## ##### #### ## ##### ## ###### ## ##### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ###### ###### ###### ## ## ###### ##### ##### #### ###### ##### ##### ## ## ###### ##### """ https://api.intelipost.com.br/api/v1/shipment_order/set_invoice Essa API adiciona dados da nota fiscal em um volume. Assim, o embarcador terá duas possibilidades de envio de documento fiscal: nota fiscal ou declaração de conteúdo (https://docs.intelipost.com.br/v1/pedido-de-entrega/atualizar-declaracao-de-conteudo-em-um-volume). Um volume só poderá ter um dos dois tipos de documento fiscal. """ def post_route_handler_shipment_order_set_invoice(self): data = self.txt_to_dict() self.unparsed_volumes = [] for request in data.get('C', []): if request.get('TIPO_REG').get('valor') == 'V': self.unparsed_volumes.append(request) for request in data.get('C'): if request.get('TIPO_REG').get('valor') == 'P': self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request self.handle_send(data_parsed) def beforeSend_shipment_order_set_invoice(self, data): data['shipment_order_volume_invoice_array'] = [] for volume in self.unparsed_volumes: volume_data = { "shipment_order_volume_number": volume.get('ship_ord_vol_array_shipment_order_volume_number', {}).get('valor'), "invoice_series": volume.get('ship_ord_vol_array_inv_invoice_series', {}).get('valor'), "invoice_number": volume.get('ship_ord_vol_array_inv_invoice_number', {}).get('valor'), "invoice_key": volume.get('ship_ord_vol_array_inv_invoice_key', {}).get('valor'), "invoice_date": self.convert_datetime(volume.get('ship_ord_vol_array_inv_invoice_date', {}).get('valor'),''), "invoice_total_value": volume.get('ship_ord_vol_array_inv_invoice_total_value', {}).get('valor'), "invoice_products_value": volume.get('ship_ord_vol_array_inv_invoice_products_value', {}).get('valor'), "invoice_cfop": volume.get('ship_ord_vol_array_inv_invoice_cfop', {}).get('valor'), } data['shipment_order_volume_invoice_array'].append(volume_data) return data def handle_output_shipment_order_set_invoice(self, data): # Tratamento da resposta baseado no retorno da API if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações da resposta messages = response_data.get('messages', []) content = response_data.get('content', {}) content = data.get('content', {}) messages = data.get('messages', []) retorno_cobol = "" # Formata a linha de informações do pedido p = [ "OK", "", "N", content.get('order_number', ''), ] retorno_cobol += "|".join(p) + "|\n" # Processa as notas fiscais for invoice in content.get('shipment_order_volume_invoice_array', []): invoice_get = self.create_getter(invoice) i = [ "OK", "", "I", invoice_get('shipment_order_volume_number', ''), invoice_get('shipment_order_volume_id', ''), invoice_get('invoice_series', ''), invoice_get('invoice_number', ''), invoice_get('invoice_key', ''), invoice_get('invoice_date', ''), invoice_get('invoice_date_iso', ''), invoice_get('invoice_total_value', ''), invoice_get('invoice_products_value', ''), invoice_get('invoice_cfop', '') ] retorno_cobol += "|".join(i) + "|\n" # Salvar arquivos flag = "w+" if self.output_count == 0 else "a+" self.write_file(unidecode.unidecode(retorno_cobol), self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() ## ###### ###### ##### ##### ###### ##### ## ##### ######## ###### ####### ###### ##### #### ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## # ## # ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ###### ## ## ## ## ## ## ## ## ## ##### ###### ##### ## ##### #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ## ## ## #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ###### ###### ##### ##### ###### ## ## ## ## ##### #### ### ## ####### ###### ##### """ https://api.intelipost.com.br/api/v1/shipment_order/set_tracking_data Atualiza o código de rastreamento de um pedido. Caso o pedido seja enviado usando uma etiqueta separada com o código, a exemplo das utilizadas pelo Correio, é necessário atualizar o código de rastreamento no pedido para que o controle do envio, a página de rastreamento e as notificações por email funcionem corretamente. """ def post_route_handler_shipment_order_set_tracking_data(self): # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/shipment_order_set_tracking_data.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # exit() # return data = self.txt_to_dict() self.unparsed_volumes = [] for request in data.get('C', []): if request.get('TIPO_REG').get('valor') == 'V': self.unparsed_volumes.append(request) for request in data.get('C'): if request.get('TIPO_REG').get('valor') == 'P': self.unparsed_data = request request_structure = self.atualWs.get('request') data_parsed = self.parser(request, request_structure) self.dados_requisicao = request self.handle_send(data_parsed) def beforeSend_shipment_order_set_tracking_data(self, data): data['tracking_data_array'] = [] if self.unparsed_volumes: for volume in self.unparsed_volumes: tracking_data = { "shipment_order_volume_number": volume.get("ship_ord_vol_array_shipment_order_volume_number").get("valor"), "tracking_code": volume.get("ship_ord_vol_array_tracking_code").get("valor"), } data['tracking_data_array'].append(tracking_data) return data def handle_output_shipment_order_set_tracking_data(self, data): """ Método chamado após o envio da requisição para a Intelipost. Aqui você pode manipular a resposta ou fazer validações adicionais. """ # Tratamento da resposta baseado no retorno da API if self.api_request.req.ok: # Salvar flag = "w+" if self.output_count == 0 else "a+" self.write_file("OK||P", self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() ###### ##### ## ## ## #### ##### #### ##### ###### ## #### #### ##### ## #### ## ## ## #### ### ## ## ## ## ## ## ## ## ## #### ## ## ## ## ## ## #### ## ## ## ## ## ## ###### #### ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #### ## ##### ## ## ###### ## ##### ## ## ##### ## ## ## ## ## ## ## ##### ## ## ## ## ## ## ###### ## ### ## ## ## ## ## ## ## ###### ## ## ## ## ## ## ###### ## ## ## ## ## ## ## ## #### ## #### ## ## ## ## ## #### #### ## ## ## ## #### def info_handler(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações da resposta status = response_data.get('status', 'UNKNOWN') messages = response_data.get('messages', []) content = response_data.get('content', {}) # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/info.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" # Formata a linha de base me = [ "OK", "", "ME", content_get('id', ''), content_get('common_name', '') ] retorno_cobol += "|".join(me) + "|\n" for option in content.get('delivery_methods', []): option_get = self.create_getter(option) mt = [ "OK", "", "MT", option_get('id', ''), option_get('name', ''), option_get('service_code', ''), option_get('logistics_provider_name', ''), option_get('logistics_provider_tax_id', ''), option_get('logistics_provider_code', ''), option_get('logistics_provider_id', ''), option_get('logistics_provider_license_code', ''), option_get('delivery_method_type', '') ] retorno_cobol += "|".join(mt) + "|\n" # SIMULAÇÃO DE RETORNO PARA TESTES # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(unidecode.unidecode(retorno_cobol), self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() def format_rote_shipment_order_get_label_pedido_volume(self,rote): return rote.format(**self.rote_data['path']) def shipment_order_get_label_pedido_volume_handler(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} # Extrai informações da resposta status = response_data.get('status', 'UNKNOWN') messages = response_data.get('messages', []) content = response_data.get('content', {}) # retorno = "" # with open("/var/www/html/webservice/gn_integracao/intelipost/templates_retorno/shipment_order_get_label_pedido_envio_numero_volume.json", 'r', encoding='utf-8') as file: # retorno = file.read() # data = json.loads(retorno) # content = data.get('content', {}) content_get = self.create_getter(content) retorno_cobol = "" saida_pdf = self.rote_data['path']['saida_pdf'] filename = "" if (content_get('label_url', '') and content_get('label_url', '') != 'null'): pdf_url = content_get('label_url', '') try: parsed_url = urlparse(pdf_url) filename = os.path.basename(parsed_url.path) filename = filename.replace(f"{self.rote_data['path']['pedido']}_", f"{self.rote_data['path']['pedido']}_{self.rote_data['path']['volume']}_") os.makedirs(os.path.dirname(saida_pdf), exist_ok=True) response = requests.get(pdf_url, stream=True, verify=False) response.raise_for_status() with open(f"{saida_pdf}/{filename}", 'wb') as file: for chunk in response.iter_content(chunk_size=8192): if chunk: file.write(chunk) except Exception as e: retorno_cobol += f"ER||Download de PDF falhou: {str(e)}\n" # Formata a linha de informações do pedido p = [ "OK", "", "P", content_get('order_number', ''), content_get('shipment_order_volume_number', ''), content_get('label_url', ''), f"{saida_pdf}", f"{filename}" ] retorno_cobol += "|".join(p) + "|\n" # SIMULAÇÃO DE RETORNO PARA TESTES # Escreve o retorno no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(retorno_cobol, self.saida_cobol, flag=flag, encoding='iso-8859-1') self.defaultErrorResponseCobol() def format_rote_shipment_order_tracking_code_tracking_code(self, rote): return rote.format(**self.rote_data['path']) def shipment_order_tracking_code_tracking_code_handler(self, data): if self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} for content in response_data['content']: content_get = self.create_getter(content) retorno_cobol = "" p = [ "OK" , "" , "P" , content_get("id", "") , content_get("client_id", "") # não existe na estrutura , content_get("quote_id", "") # não existe na estrutura , content_get("delivery_method_id", "") , content_get("estimated_delivery_days_lp", "") , "" # provider_shipping_costs - não existe na estrutura 2 , content_get("customer_shipping_costs", "") , self.extract_date_ymd(content_get("estimated_delivery_date_iso", "")) , self.extract_time_hms(content_get("estimated_delivery_date_iso", "")) , "" # estimated_delivery_date_lp_iso date - não existe na estrutura 2 , "" # estimated_delivery_date_lp_iso time - não existe na estrutura 2 , self.extract_date_ymd(content_get("shipped_date_iso", "")) , self.extract_time_hms(content_get("shipped_date_iso", "")) , self.extract_date_ymd(content_get("created_iso", "")) , self.extract_time_hms(content_get("created_iso", "")) , self.extract_date_ymd(content_get("modified_iso", "")) , self.extract_time_hms(content_get("modified_iso", "")) , content_get("carrier.driver.federal_tax_id", "") , content_get("carrier.driver.license", "") , content_get("carrier.driver.first_name", "") , content_get("carrier.driver.last_name", "") , content_get("carrier.driver.email", "") , content_get("carrier.driver.cell_phone", "") , content_get("carrier.vehicle.licence_plate", "") , content_get("carrier.vehicle.manufacturer", "") , content_get("carrier.vehicle.model", "") , content_get("carrier.vehicle.color", "") , content_get("carrier.vehicle.category", "") , content_get("end_customer.first_name", "") , content_get("end_customer.last_name", "") , content_get("end_customer.email", "") , content_get("end_customer.phone", "") , content_get("end_customer.cellphone", "") , content_get("end_customer.is_company", "") , content_get("end_customer.federal_tax_payer_id", "") , content_get("end_customer.state_tax_payer_id", "") , content_get("end_customer.shipping_address", "") , content_get("end_customer.shipping_number", "") , content_get("end_customer.shipping_additional", "") , content_get("end_customer.shipping_reference", "") , content_get("end_customer.shipping_quarter", "") , content_get("end_customer.shipping_city", "") , content_get("end_customer.shipping_zip_code", "") , content_get("end_customer.shipping_state_code", "") , content_get("end_customer.shipping_country", "") , "" # origin_warehouse_code - não existe na estrutura 2 , content_get("origin_name", "") , content_get("origin_federal_tax_payer_id", "") , content_get("origin_customer_phone", "") , content_get("origin_customer_email", "") , content_get("origin_street", "") , content_get("origin_number", "") , content_get("origin_zip_code", "") , content_get("origin_city", "") , content_get("origin_quarter", "") , content_get("origin_state_code", "") , content_get("shipment_order_type", "") , "" # parent_shipment_order_number - não existe na estrutura 2 , content_get("scheduling_window_start", "") # não existe na estrutura , content_get("scheduling_window_end", "") # não existe na estrutura , content_get("scheduled", "") , "" # shipment_order_volume_state - não existe na estrutura 2 , "" # delivered_date - não existe na estrutura 2 , content_get("additional_information.key", "") , content_get("additional_information.key2", "") , "" # delivery_address.name - não existe na estrutura 2 , "" # delivery_address.street - não existe na estrutura 2 , "" # delivery_address.number - não existe na estrutura 2 , "" # delivery_address.reference - não existe na estrutura 2 , "" # delivery_address.additional - não existe na estrutura 2 , "" # delivery_address.country - não existe na estrutura 2 , "" # delivery_address.state_code - não existe na estrutura 2 , "" # delivery_address.city - não existe na estrutura 2 , "" # delivery_address.quarter - não existe na estrutura 2 , "" # delivery_address.zip_code - não existe na estrutura 2 , "" # delivery_address.latitude - não existe na estrutura 2 , "" # delivery_address.longitude - não existe na estrutura 2 # Campos adicionais da estrutura 2 que não existem na primeira: , content_get("platform", "") , content_get("delivery_method_external_id", "") , content_get("order_number", "") , content_get("end_customer.shipping_state", "") , content_get("warehouse_address_id", "") , content_get("sales_channel", "") , content_get("delivery_method_name", "") , content_get("sales_order_number", "") , content_get("logistic_provider_name", "") , content_get("external_order_numbers.erp", "") , content_get("external_order_numbers.marketplace", "") , content_get("external_order_numbers.plataforma", "") , content_get("external_order_numbers.sales_channel", "") , content_get("tracking_url", "") , content_get("verification_pickup.code", "") ] retorno_cobol += "|".join(p) + "|\n" for volume in content.get('shipment_order_volume_array', []): volume_get = self.create_getter(volume) v = [ "OK" , "" , "V" , volume_get("shipment_order_volume_number", "") , volume_get("shipment_order_id", "") , volume_get("shipment_order_volume_state", "") , volume_get("weight", "") , volume_get("volume_type_code", "") , volume_get("width", "") , volume_get("height", "") , volume_get("length", "") , volume_get("products_nature", "") , volume_get("products_quantity", "") , volume_get("is_icms_exempt", "") , self.extract_date_ymd(volume_get("created_iso", "")) , self.extract_time_hms(volume_get("created_iso", "")) , self.extract_date_ymd(volume_get("modified_iso", "")) , self.extract_time_hms(volume_get("modified_iso", "")) , volume_get("logistics_provider_data", "") , volume_get("logistic_provider_tracking_code", "") , volume_get("estimated_delivery_date_lp", "") , "" # estimated_delivery_date_lp_iso - não existe na estrutura 2 , volume_get("shipment_order_volume_invoice.shipment_order_volume_number", "") , volume_get("shipment_order_volume_invoice.shipment_order_volume_id", "") , volume_get("shipment_order_volume_invoice.invoice_series", "") , volume_get("shipment_order_volume_invoice.invoice_number", "") , volume_get("shipment_order_volume_invoice.invoice_key", "") , self.extract_date_ymd(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , self.extract_time_hms(volume_get("shipment_order_volume_invoice.invoice_date_iso", "")) , volume_get("shipment_order_volume_invoice.invoice_total_value", "") , volume_get("shipment_order_volume_invoice.invoice_products_value", "") , volume_get("shipment_order_volume_invoice.invoice_cfop", "") , volume_get("pre_shipment_list_state", "") , volume_get("has_clarify_delivery_fail", "") , volume_get("delivered_late", "") , volume_get("delivered_late_lp", "") , self.extract_date_ymd(volume_get("estimated_delivery_date_iso", "")) , self.extract_time_hms(volume_get("estimated_delivery_date_iso", "")) , volume_get("delivered", "") , volume_get("pre_shipment_list_id", "") , volume_get("logistic_provider_pre_shipment_list_id", "") , volume_get("name", "") , self.extract_date_ymd(volume_get("original_estimated_delivery_date_iso", "")) , self.extract_time_hms(volume_get("original_estimated_delivery_date_iso", "")) , "" # original_estimated_delivery_date_lp_iso date - não existe na estrutura 2 , "" # original_estimated_delivery_date_lp_iso time - não existe na estrutura 2 , self.extract_date_ymd(volume_get("shipped_date_iso", "")) , self.extract_time_hms(volume_get("shipped_date_iso", "")) , volume_get("delivered_date", "") , volume_get("shipment_order_volume_state_localized", "") , volume_get("shipment_order_volume_id", "") , volume_get("tracking_code", "") # Campos adicionais da estrutura 2 que não existem na primeira: , volume_get("logistic_provider_label_hash", "") , volume_get("packaging_code", "") , volume_get("original_estimated_delivery_date_lp", "") , volume_get("estimated_delivery_days_lp", "") , volume_get("client_pre_shipment_list", "") , volume_get("content_declaration", "") ] retorno_cobol += "|".join(v) + "|\n" # Processa os anexos do volume if volume.get('attachments'): for attachment in volume.get('attachments', []): va = [ "OK" , "" , "VA" # , attachment.get("id", "") # , attachment.get("name", "") # , attachment.get("url", "") # , attachment.get("type", "") # , attachment.get("size", "") ] retorno_cobol += "|".join(va) + "|\n" # Processa os produtos do volume if volume.get('products'): for product in volume.get('products', []): product_get = self.create_getter(product) vp = [ "OK" , "" , "VP" , product_get("weight", "") , product_get("width", "") , product_get("height", "") , product_get("length", "") , product_get("price", "") , product_get("description", "") , product_get("sku", "") , product_get("category", "") , product_get("quantity", "") , product_get("image_url", "") ] retorno_cobol += "|".join(vp) + "|\n" if volume.get('shipment_order_volume_state_history_array'): for vh in volume.get('shipment_order_volume_state_history_array', []): vh_get = self.create_getter(vh) vh_data = [ "OK" , "" , "VH" , vh_get("shipment_order_volume_id", "") , vh_get("shipment_order_volume_state", "") , vh_get("tracking_state", "") , self.extract_date_ymd(vh_get("created_iso", "")) , self.extract_time_hms(vh_get("created_iso", "")) , vh_get("provider_message", "") , vh_get("provider_state", "") , vh_get("esprinter_message", "") , "" # shipment_volume_micro_state - não existe mais como campo direto na estrutura 2 , vh_get("shipment_order_volume_state_localized", "") , vh_get("shipment_order_volume_state_history", "") , self.extract_date_ymd(vh_get("event_date_iso", "")) , self.extract_time_hms(vh_get("event_date_iso", "")) # Campos adicionais da estrutura 2 que não existem na primeira: , vh_get("shipper_provider_state", "") , vh_get("shipment_order_volume_micro_state.id", "") , vh_get("shipment_order_volume_micro_state.code", "") , vh_get("shipment_order_volume_micro_state.default_name", "") , vh_get("shipment_order_volume_micro_state.i18n_name", "") , vh_get("shipment_order_volume_micro_state.description", "") , vh_get("shipment_order_volume_micro_state.shipment_order_volume_state_id", "") , vh_get("shipment_order_volume_micro_state.shipment_volume_state_source_id", "") , vh_get("shipment_order_volume_micro_state.shipment_volume_state", "") , vh_get("shipment_order_volume_micro_state.shipment_volume_state_localized", "") , vh_get("shipment_order_volume_micro_state.name", "") , vh_get("location.address", "") , vh_get("location.number", "") , vh_get("location.additional", "") , vh_get("location.reference", "") , vh_get("location.city", "") , vh_get("location.state_code", "") , vh_get("location.quarter", "") , vh_get("location.zip_code", "") , vh_get("location.description", "") , vh_get("location.latitude", "") , vh_get("location.longitude", "") , vh_get("request_hash", "") , vh_get("request_origin", "") ] retorno_cobol += "|".join(vh_data) + "|\n" if vh.get('attachments'): for vha in vh.get('attachments', []): vha_data = [ "OK" , "" , "VA" # , vha.get("id", "") # , vha.get("name", "") # , vha.get("url", "") # , vha.get("type", "") # , vha.get("size", "") ] retorno_cobol += "|".join(vha_data) + "|\n" # Escreve o resultado no arquivo de saída flag = "w+" if self.output_count == 0 else "a+" self.write_file(unidecode.unidecode(retorno_cobol), self.saida_cobol, flag=flag, encoding='iso-8859-1') self.output_count += 1 self.defaultErrorResponseCobol() def defaultErrorResponseCobol(self): if not self.api_request.req.ok: response_data = self.api_request.req.json() if self.api_request.req.text else {} saida_cobol = "ER|{messages}|" messages = response_data.get('messages', []) # {'type': 'ERROR', 'text': 'Número do pedido já existe 123456.', 'key': 'shipmentOrder.save.already.existing.order.number'} if messages: messages_list = [msg.get('text', '') for msg in messages if isinstance(msg, dict) and 'text' in msg] saida_cobol = saida_cobol.format(messages=",".join(messages_list)) else: saida_cobol = saida_cobol.format(messages="Erro desconhecido") flag = "w+" if self.output_count == 0 else "a+" self.write_file(unidecode.unidecode(saida_cobol), self.saida_cobol, flag=flag, encoding='iso-8859-1') """ .########..########.##.......########.########.######## .##.....##.##.......##.......##..........##....##...... .##.....##.##.......##.......##..........##....##...... .##.....##.######...##.......######......##....######.. .##.....##.##.......##.......##..........##....##...... .##.....##.##.......##.......##..........##....##...... .########..########.########.########....##....######## """ def delete_handler(self) : if not self.hasAttr("deleteFromApi",self) : return self.deleteFromApi() def deleteFromApi(self): rote = self.atualWs.get('rote') def extract_date_ymd(self, iso_date_string): """ Extracts date in format Ymd from ISO date string Example: "2016-06-30T16:30:41.000-03:00" -> "20160630" """ if not iso_date_string or not isinstance(iso_date_string, str): return "" try: # Extract date part (YYYY-MM-DD) date_part = iso_date_string.split('T')[0] # Remove hyphens return date_part.replace('-', '') except Exception: return "" def extract_time_hms(self, iso_date_string): """ Extracts time in format Hms from ISO date string Example: "2016-06-30T16:30:41.000-03:00" -> "163041" """ if not iso_date_string or not isinstance(iso_date_string, str): return "" try: # Split at T to get time part time_part = iso_date_string.split('T')[1] # Extract HH:MM:SS time_hms = time_part.split('.')[0] # Remove colons return time_hms.replace(':', '') except Exception: return "" def treat_value(self, value): if isinstance(value, (bool)): return "S" if value else "N" elif value is None: return "" elif value == False and not isinstance(value, (int, float)): return "N" elif value == True and not isinstance(value, (int, float)): return "S" elif value == 'False': return "N" elif value == 'True': return "S" elif isinstance(value, (int, float)): return str(value) elif isinstance(value, str): return value elif isinstance(value, (list, tuple)): return value if value else "" else: return str(value) def create_getter(self, obj): """ Cria uma função getter personalizada para um objeto específico """ def get(path, default=None): try: keys = path.split('.') current = obj for key in keys: current = current[key] return self.treat_value(current) except (KeyError, TypeError, AttributeError): return default return get def cep(self, cep, path): """ Formata o CEP para o padrão XXXXX-XXX """ if not cep or not isinstance(cep, str): return "" cep = cep.replace("-", "").replace(".", "").replace(" ", "") if len(cep) == 8: return f"{cep[:5]}-{cep[5:]}" return cep # 2025-07-30 17:00:00 > 2025-07-30T17:00:00-03:00 def convert_datetime(self, value, path): """ Converte uma string de data no formato 'YYYY-MM-DD HH:MM:SS' para o formato ISO 8601 com fuso horário. Exemplo: '2025-07-30 17:00:00' -> '2025-07-30T17:00:00-03:00' """ if not value or not isinstance(value, str): return "" try: # Tenta converter a string para um objeto datetime dt = datetime.strptime(value, '%Y-%m-%d %H:%M:%S') # Formata para o padrão ISO 8601 com fuso horário -03:00 return dt.strftime('%Y-%m-%dT%H:%M:%S-03:00') except ValueError: return "" def to_boolean(self, value, path): if value == "S": return True elif value == "N": return False def origin_warehouse_code(self, value, path): if value == "U01001": return "002" if value == "U01002": return "003" return value def handle_send(self, data) : """ Metodo para manipular o envio """ self.current_data = data self.prepare_send(data) # Faz um tratamento no dados antes de enviar func_rote = self.fixed_rote() func = f"beforeSend_{func_rote}"#.format(func_rote) # Tratamento de dados antes de enviar beforeSend = self.call(func, data, is_required= self.atualWs.get('need_beforeSend',False)) if beforeSend is not None: data = beforeSend response = self.send(data) self.current_data = {} return self.handle_response(response)