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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
|
import base64
import datetime
import functools
import io
import json
from array import array
import jwt
from odoo import http
from odoo.http import request
from odoo.modules import get_module_resource
from odoo.tools.config import config
from PIL import Image
from pytz import timezone
class Controller(http.Controller):
jwt_secret_key = "NTNv7j0TuYARvmNMmWXo6fKvM4o6nvaUi9ryX38ZHL1bkrnD1ObOQ8JAUmHCBq7Iy7otZcyAagBLHVKvvYaIpmMuxmARQ97jUVG16Jkpkp1wXOPsrF9zwew6TpczyHkHgX5EuLg2MeBuiTqJACs1J0apruOOJCggOtkjB"
# jwt_secret_key = "8fc74567db6878107842b466a4d6622230167740"
@staticmethod
def must_authorized(private=False, private_key=''):
def wrapper(func):
@functools.wraps(func)
def inner_wrapper(*args, **kwargs):
self = args[0]
auth = self.authenticate()
if not auth:
return self.unauthorized_response()
self.set_user_pricelist_ctx()
if private:
auth_key = None if isinstance(auth, bool) else auth[private_key]
param_key = int(kwargs.get(private_key, -1))
if auth_key != param_key:
return self.unauthorized_response()
return func(*args, **kwargs)
return inner_wrapper
return wrapper
def authenticate(self):
wsgienv = request.httprequest.environ
try:
db = wsgienv['HTTP_DB']
username = wsgienv['HTTP_USERNAME']
password = wsgienv['HTTP_PASSWORD']
request.session.authenticate(db, username, password)
return True
except:
try:
authorization = wsgienv['HTTP_AUTHORIZATION']
except:
authorization = None
request.session.authenticate(config.get('db_name'), 'stephan@indoteknik.co.id', 'P@ssvv0rd755')
token = request.env['ir.config_parameter'].sudo().get_param('rest_api_token') or ''
result = False
if authorization == token:
result = True
user_token = self.verify_user_token()
if user_token:
result = user_token
return result
def set_user_pricelist_ctx(self):
user_token = self.authenticate()
pricelist = request.env['product.pricelist'].new()
if isinstance(user_token, dict):
partner = request.env['res.partner'].browse(user_token['partner_id'])
if partner: pricelist = partner.property_product_pricelist
context = request.env.context.copy()
context.update({'user_pricelist': pricelist})
request.env.context = context
def get_request_params(self, kw, queries):
result = {
'valid': True,
'reason': [],
'value': {},
'query': {}
}
for key, rules in queries.items():
is_number = 'number' in rules
is_exclude_if_null = 'exclude_if_null' in rules
alias = next((r.replace('alias:', '') for r in rules if r.startswith('alias:')), key)
default = next((r.replace('default:', '') for r in rules if r.startswith('default:')), None)
value = kw.get(key, '')
if value in ['null', 'undefined']:
value = ''
if 'required' in rules and not value:
result['reason'].append(f"{key} is required")
if 'number' in rules and value and not value.isdigit():
result['reason'].append(f"{key} must be a number")
result['query'][key] = value
if not value and default:
value = default
if is_number and value.isdigit():
value = int(value)
if not value and is_exclude_if_null:
continue
result['value'][alias] = value
result['valid'] = not result['reason']
return result
def time_to_str(self, object, format):
time = ''
if isinstance(object, datetime.datetime):
time = object.astimezone(timezone('Asia/Jakarta')).strftime(format)
return time
def response(self, data=[], code=200, description='OK', headers=[]):
response = {
'status': {
'code': code,
'description': description
}
}
if code == 200:
response.update({'result': data})
response = json.dumps(response)
return request.make_response(response, [
('Access-Control-Allow-Origin', '*'),
('Access-Control-Allow-Headers', '*'),
('Access-Control-Allow-Methods', '*'),
('Content-Type', 'application/json'),
] + headers)
def unauthorized_response(self):
return self.response(code=401, description='Unauthorized')
def search_filter(self, model: str, kw: dict, query: array = []):
""" To search data by default API Params if exist """
limit = kw.get('limit', 0)
offset = kw.get('offset', 0)
order = kw.get('order', '')
return request.env[model].search(query, limit=int(limit), offset=int(offset), order=order)
def create_user_token(self, user):
return jwt.encode({'id': user.id}, self.jwt_secret_key)
def verify_user_token(self):
try:
token = request.httprequest.environ['HTTP_TOKEN']
user_token = jwt.decode(token, self.jwt_secret_key, algorithms=['HS256'])
user = request.env['res.users'].browse([user_token['id']])
if not user:
return False
data = {
'user_id': user.id,
'partner_id': None
}
if user.partner_id:
data['partner_id'] = user.partner_id.id
return data
except:
return False
def get_partner_child_ids(self, partner_id):
partner = request.env['res.partner'].search([('id', '=', partner_id)], limit=1)
partner_child_ids = [x['id'] for x in partner.child_ids] + [partner.id]
if partner.parent_id:
partner_child_ids += [x['id'] for x in partner.parent_id.child_ids]
partner_child_ids += [partner.parent_id.id]
return partner_child_ids
@http.route('/api/token', auth='public', methods=['GET', 'OPTIONS'])
def get_api_token(self, **kw):
return self.response(request.env['ir.config_parameter'].sudo().get_param('rest_api_token') or '')
@http.route('/api/ip-address', auth='public', methods=['GET', 'OPTIONS'])
def get_ip_address(self):
address = request.httprequest.remote_addr
address = address if address != '127.0.0.1' else False
return self.response(address)
@http.route('/api/image/<model>/<field>/<id>', auth='public', methods=['GET'])
def get_image(self, model, field, id, **kw):
model_name = model
model = request.env[model].sudo().search([('id', '=', id)], limit=1)
image = model[field] if field in model else ''
if model_name in ['product.template']:
version = '1' if field in ['image_256', 'image_512', 'image_1024', 'image_1920'] else '2'
ratio = kw.get('ratio', '')
variant = kw.get('variant', False)
image = model[field] if version == '1' else model['image_256']
if not variant:
image = self.add_watermark_to_image(image, ratio, version)
response_headers = [
('Content-Type', 'image/jpg'),
('Cache-Control', 'public, max-age=3600')
]
return request.make_response(
base64.b64decode(image),
response_headers
)
def add_watermark_to_image(self, image, ratio, version='1'):
if not image: return ''
LOGO_FILENAME = {
'1': 'logo-indoteknik-gray.png',
'2': 'logo-indoteknik-link.png'
}
logo_path = get_module_resource('indoteknik_api', 'static', 'src', 'images', LOGO_FILENAME.get(version))
logo_img = Image.open(logo_path).convert('RGBA')
img_data = io.BytesIO(base64.b64decode(image))
img = Image.open(img_data).convert('RGBA')
img_width, img_height = img.size
longest_wh = max(img_height, img_width)
# Resize logo image
logo_img_w = img_width // 2.2
logo_img_h = img_height // 2.2
new_img = img
if ratio == 'square':
new_img = Image.new('RGBA', (longest_wh, longest_wh), (255, 255, 255, 255))
paste_x = (longest_wh - img_width) // 2
paste_y = (longest_wh - img_height) // 2
new_img.paste(img, (paste_x, paste_y), img)
if version == '2':
logo_img_w = img_width // 1.8
logo_img_h = img_height // 1.8
logo_img.thumbnail((logo_img_w, logo_img_h))
# Add watermark
new_img.paste(logo_img, (12, 10), logo_img)
buffered = io.BytesIO()
new_img.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode('utf-8')
|
