This repository has been archived by the owner on May 31, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 0
/
quart_app.py
126 lines (93 loc) · 3.52 KB
/
quart_app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
import os
import pystray
import asyncio
import threading
from PIL import Image
from Ajio import Ajio
from Amazon import Amazon
from hypercorn.asyncio import serve
from hypercorn.config import Config
from quart import Quart, jsonify, request
app = Quart(__name__)
global ajio_instance
@app.route('/ajio_run', methods=['POST'])
async def ajio_run():
request_data = await request.get_json()
requester_ip = request.remote_addr
print(f"Route: /ajio_run")
print(f"Requester IP Address: {requester_ip}")
print(f"Requester data: {request_data}")
try:
if 'url' in request_data:
url = request_data['url']
if '/p/' in url:
global ajio_instance
ajio_instance = Ajio()
result = await ajio_instance.main(url)
return jsonify(result), 200
else:
return jsonify({'error': "Enter full product url. Short url not accepted"}), 400
except:
return jsonify({'error': "Enter valid url"}), 400
@app.route('/ajio_login', methods=['POST'])
async def ajio_login():
request_data = await request.get_json()
requester_ip = request.remote_addr
print(f"Route: /ajio_login")
print(f"Requester IP Address: {requester_ip}")
print(f"Requester data: {request_data}")
if 'mobileNumber' in request_data:
mobile_number = request_data['mobileNumber']
global ajio_instance
ajio_instance = Ajio()
result = await ajio_instance.login(mobile_number)
return jsonify(result), 200
return jsonify({"Error": "Enter mobile number"})
@app.route('/ajio_otp', methods=['POST'])
async def ajio_OTP():
request_data = await request.get_json()
requester_ip = request.remote_addr
print(f"Route: /ajio_login")
print(f"Requester IP Address: {requester_ip}")
print(f"Requester data: {request_data}")
if 'OTP' in request_data:
OTP = request_data['OTP']
global ajio_instance
result = await ajio_instance.OTP(OTP)
return jsonify(result), 200
return jsonify({"Error": "Enter OTP"})
@app.route("/amazon_run", methods=['POST'])
async def amazon_run():
request_data = await request.get_json()
requester_ip = request.remote_addr
print(f"Requester IP Address: {requester_ip}")
print(f"Requester data: {request_data}")
try:
if 'url' in request_data:
url = request_data['url']
if '/dp/' or '/gp/' in url:
amazon_instance = Amazon()
result = await amazon_instance.main(url)
return jsonify(result), 200
else:
return jsonify({'error': "Enter full product url. Short url not accepted"}), 400
else:
return jsonify({'error': "Enter valid url"}), 400
except TypeError:
return jsonify({'error': "Enter url"}), 400
def background_task():
icon = Image.open('icon.ico')
menu = pystray.Menu(pystray.MenuItem('Exit', lambda: os._exit(0)))
tray = pystray.Icon("name", icon, "Title", menu)
tray.run()
async def run_server():
asyncio.set_event_loop(asyncio.new_event_loop())
config = Config()
config.bind = "127.0.0.1:5123",
config.workers = 1
await serve(app, config)
if __name__ == "__main__":
background_thread = threading.Thread(target=background_task)
background_thread.daemon = True
background_thread.start()
asyncio.run(run_server())