trytonpsk-sale_web_channel/routes.py

101 lines
3.2 KiB
Python

# 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(
'/<database_name>/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'''<body onload="window.close()">
<h2>Login Success with code {code} and state <h2>
<h3> refresh token exchange <h3>
<p> {res} <p>
</body>''', 200, content_type='text/html')
@app.route(
'/<database_name>/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(
'/<database_name>/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