# This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. import json import logging import http.client from urllib.parse import urlparse, parse_qs import requests from werkzeug.exceptions import abort from werkzeug.wrappers import Response from trytond.wsgi import app from trytond.protocols.wrappers import with_pool, with_transaction logger = logging.getLogger(__name__) @app.route( '//sale_web_channel/mercadolibre/login_app', methods=['GET']) @with_pool @with_transaction(context={'_skip_warnings': True}) def login_app(request, pool): parse_result = urlparse(request.url) dict_result = parse_qs(parse_result.query) code = dict_result.get('code') if code: Channel = pool.get('sale.web_channel.mercado_libre') channel, = Channel.search(['state', '=', 'draft']) code = code[0] url = 'https://api.mercadolibre.com/oauth/token' headers = { 'accept': 'application/json', 'content-type': 'application/x-www-form-urlencoded', } data = { 'grant_type': 'authorization_code', 'client_id': channel.app_id, 'client_secret': channel.secret_key, 'code': code, 'redirect_uri': channel.redirect_uri, } response = requests.post(url, headers=headers, data=data) if response.status_code == 200: result = response.json() res = f""" Acces Token: {result['access_token']} \n User Id: {result['user_id']} \n Refresh Token: {result['refresh_token']} \n Expires In: {result['expires_in']} \n """ print(json_response) else: res = f"Error en la solicitud {response.status_code}" print("Error en la solicitud:", response.status_code) else: try: print(request) except: pass return Response( f'''

Login Success with code {code} and state

refresh token exchange

{res}
''', 200, content_type='text/html') @app.route( '//sale_web_channel/mercadolibre/webhook', methods=['POST']) @with_pool @with_transaction(context={'_skip_warnings': True}) def webhooks_endpoint_mercadolibre(request, pool): request_body = request.get_data(as_text=True) payload = json.loads(request_body) Channel = pool.get('sale.web_channel.mercado_libre') Channel.create_sale(payload) print(payload, 'validate payload') return True @app.route( '//sale_web_channel/shopify/webhook', methods=['POST']) @with_pool @with_transaction(context={'_skip_warnings': True}) def webhooks_endpoint_shopify(request, pool): request_body = request.get_data(as_text=True) payload = json.loads(request_body) Channel = pool.get('sale.web_channel.shopify') res = Channel.request_api(payload) print(res, 'validate') return True