341 lines
9.9 KiB
Python
341 lines
9.9 KiB
Python
# import ssl
|
|
import os
|
|
import base64
|
|
import tempfile
|
|
import logging
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from http.client import HTTPConnection, HTTPSConnection
|
|
import orjson as json
|
|
try:
|
|
from app.models import MODELS_RESTAURANT, MODELS_RETAIL
|
|
except Exception:
|
|
from models import MODELS_RESTAURANT, MODELS_RETAIL
|
|
|
|
try:
|
|
from app.commons.common import slugify, file_open
|
|
except Exception:
|
|
from commons.common import slugify, file_open
|
|
from app.commons.dblogin import context_http
|
|
|
|
|
|
HEADERS = {
|
|
"Content-type": "application/json",
|
|
"Accept": "text/plain"
|
|
}
|
|
|
|
|
|
def encoder(obj):
|
|
# FIXME: add datetime, buffer, bytes
|
|
if isinstance(obj, date):
|
|
return {
|
|
'__class__': 'date',
|
|
'year': obj.year,
|
|
'month': obj.month,
|
|
'day': obj.day,
|
|
}
|
|
elif isinstance(obj, Decimal):
|
|
return str(obj)
|
|
raise TypeError(repr(obj) + " is not JSON serializable")
|
|
|
|
|
|
def logout(ctx):
|
|
port = ctx['params']['port']
|
|
host = ctx['params']['server']
|
|
db = ctx['params']['database']
|
|
if ctx['params']['mode'] == 'http':
|
|
conn = HTTPConnection(host, port=port, timeout=30)
|
|
else:
|
|
conn = HTTPSConnection(host, port=port, timeout=30, context=context_http)
|
|
url = '/' + db + '/logout'
|
|
payload = json.dumps({
|
|
'context': {
|
|
'session': ctx.get('session'),
|
|
'login': ctx.get('login')}
|
|
}, default=encoder)
|
|
conn.request('POST', url, body=payload, headers=HEADERS)
|
|
response = conn.getresponse()
|
|
print(response.status, 'status logout')
|
|
if response.status != 200:
|
|
res = response.read()
|
|
logging.error(f'error request {res}')
|
|
else:
|
|
res = json.loads(response.read())
|
|
conn.close()
|
|
return res
|
|
|
|
|
|
class Model(object):
|
|
|
|
def __init__(self, model, ctx, fields=None, main_window=None):
|
|
self.model = model
|
|
self.ctx = ctx
|
|
self.main_window = main_window
|
|
self.port = ctx['params']['port']
|
|
self.host = ctx['params']['server']
|
|
self.db = ctx['params']['database']
|
|
HEADERS['Authorization'] = "Bearer " + ctx.get('session')
|
|
# self.api = 'http://localhost:8010/ZIRUS'
|
|
if ctx['params']['mode'] == 'http':
|
|
self.conn = HTTPConnection
|
|
self.context_http = None
|
|
else:
|
|
self.conn = HTTPSConnection
|
|
self.context_http = context_http
|
|
|
|
environment = ctx['params']['environment']
|
|
if environment == 'retail':
|
|
_model = MODELS_RETAIL.get(model)
|
|
else:
|
|
_model = MODELS_RESTAURANT.get(model)
|
|
self.fields = None
|
|
if fields:
|
|
self.fields = fields
|
|
elif _model.get('fields'):
|
|
self.fields = _model['fields']
|
|
|
|
def __getattr__(self, name, *args):
|
|
'Return attribute value'
|
|
self.method = name
|
|
return self
|
|
|
|
def find(self, domain, order=None, limit=1000, ctx=None, fields=None):
|
|
if ctx:
|
|
self.ctx.update(ctx)
|
|
if not fields:
|
|
fields = self.fields
|
|
args_ = {
|
|
'model': self.model,
|
|
'domain': domain,
|
|
'order': order,
|
|
'limit': limit,
|
|
'fields_names': fields,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/search', args_)
|
|
return res
|
|
|
|
def fields_get(self, fields=[]):
|
|
if not fields:
|
|
fields = self.fields
|
|
args_ = {
|
|
'model': self.model,
|
|
'fields_names': fields,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/fields_get', args_)
|
|
return res
|
|
|
|
# deprecated
|
|
def write_many(self, ids, values, fields=None):
|
|
if not fields:
|
|
fields = self.fields
|
|
args_ = {
|
|
'model': self.model,
|
|
'ids': ids,
|
|
'values': values,
|
|
'fields': fields,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/save_many', args_)
|
|
return res
|
|
|
|
def write(self, ids, values, fields=None):
|
|
if not fields:
|
|
fields = self.fields
|
|
if values.get('rec_name'):
|
|
_ = values.pop('rec_name')
|
|
args_ = {
|
|
'model': self.model,
|
|
'ids': ids,
|
|
'values': values,
|
|
'fields': fields,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/write', args_)
|
|
return res
|
|
|
|
def create(self, values):
|
|
if values.get('rec_name'):
|
|
_ = values.pop('rec_name')
|
|
args_ = {
|
|
'model': self.model,
|
|
'record': values,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/create', args_)
|
|
return res
|
|
|
|
def search_count(self, domain):
|
|
args_ = {
|
|
'model': self.model,
|
|
'domain': domain,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/search_count', args_)
|
|
return res
|
|
|
|
def method_instance(self, method, instance):
|
|
args_ = {
|
|
'model': self.model,
|
|
'method': method,
|
|
'instance': instance,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/method_instance', args_)
|
|
return res
|
|
|
|
def delete(self, ids):
|
|
args_ = {
|
|
'model': self.model,
|
|
'ids': ids,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/delete', args_)
|
|
return res
|
|
|
|
def get_connection(self, method, target, data=None):
|
|
url = '/' + self.db + target
|
|
if data:
|
|
payload = json.dumps(data, default=encoder)
|
|
else:
|
|
payload = None
|
|
if self.context_http:
|
|
conn = self.conn(
|
|
self.host,
|
|
port=self.port,
|
|
timeout=30,
|
|
context=self.context_http)
|
|
else:
|
|
conn = self.conn(self.host, port=self.port, timeout=30)
|
|
res = 'error_network'
|
|
try:
|
|
conn.request(method, url, body=payload, headers=HEADERS)
|
|
response = conn.getresponse()
|
|
res = json.loads(response.read())
|
|
if response.status != 200:
|
|
logging.error(f'error request {str(res)}')
|
|
if self.main_window and hasattr(self.main_window, 'statusbar'):
|
|
self.main_window.dialog(
|
|
'error_server',
|
|
extra_message=res['detail']['error'])
|
|
res = res['detail']
|
|
conn.close()
|
|
except Exception:
|
|
logging.exception('error de conexion')
|
|
return res
|
|
|
|
def __call__(self, values=None, args=None, ctx=None):
|
|
if ctx:
|
|
self.ctx.update(ctx)
|
|
args_ = {
|
|
'model': self.model,
|
|
'method': self.method,
|
|
'args': [],
|
|
'context': self.ctx,
|
|
}
|
|
if values:
|
|
logging.warning(f'deprecate option values call method with list instance. \n model: {self.model} method: {self.method}')
|
|
args_['args'] = [values]
|
|
if args:
|
|
args_['args'] = args
|
|
res = self.get_connection('POST', '/method', args_)
|
|
return res
|
|
|
|
def button_method(self, method, ids=[], ctx=None):
|
|
if ctx:
|
|
self.ctx.update(ctx)
|
|
args_ = {
|
|
'model': self.model,
|
|
'method': method,
|
|
'ids': ids,
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/button_method', args_)
|
|
return res
|
|
|
|
|
|
class Report(object):
|
|
|
|
def __init__(self, ctx):
|
|
self.ctx = ctx
|
|
self.port = ctx['params']['port']
|
|
self.host = ctx['params']['server']
|
|
self.db = ctx['params']['database']
|
|
if ctx['params']['mode'] == 'http':
|
|
self.conn = HTTPConnection
|
|
self.context_http = None
|
|
else:
|
|
self.conn = HTTPSConnection
|
|
self.context_http = context_http
|
|
|
|
def get_connection(self, method, target, data=None):
|
|
url = '/' + self.db + target
|
|
if data:
|
|
payload = json.dumps(data, default=encoder)
|
|
else:
|
|
payload = None
|
|
conn = self.conn(self.host, port=self.port)
|
|
conn.request(method, url, body=payload, headers=HEADERS)
|
|
response = conn.getresponse()
|
|
res = json.loads(response.read())
|
|
conn.close()
|
|
return res
|
|
|
|
def get(self, values):
|
|
args_ = {
|
|
'report': values['report_name'],
|
|
'data': values['args'],
|
|
'records': values.get('records', []),
|
|
'record': values.get('record', -1),
|
|
'context': self.ctx,
|
|
}
|
|
res = self.get_connection('POST', '/report', args_)
|
|
return res
|
|
|
|
def open(self, args):
|
|
oext, content, direct_print, file_name = args
|
|
|
|
dtemp = tempfile.mkdtemp(prefix='tryton_')
|
|
fp_name = os.path.join(
|
|
dtemp, slugify(file_name) + os.extsep + slugify(oext)
|
|
)
|
|
print('archivo :', dtemp, fp_name)
|
|
|
|
content_data = base64.b64decode(content)
|
|
with open(fp_name, 'wb') as file_d:
|
|
file_d.write(content_data)
|
|
file_open(fp_name, type, direct_print=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_model = False
|
|
test_report = True
|
|
ctx = {
|
|
'company': 1,
|
|
'user': 1,
|
|
'params': {'api_url': 'localhost:5070', 'database': 'DEMO50'}
|
|
}
|
|
if test_model:
|
|
model = {'model': 'sale.sale'}
|
|
test_model = Model(model, ctx)
|
|
id = 180
|
|
data = {
|
|
'reference': 'OC-02874',
|
|
}
|
|
res = test_model.write([id], data)
|
|
|
|
if test_report:
|
|
data = {
|
|
'report_name': 'sale_pos_frontend.sale_square_box_report',
|
|
'args': {
|
|
'date': '2021-01-12',
|
|
'turn': 1,
|
|
'company': 1,
|
|
'shop': 1,
|
|
},
|
|
}
|
|
report = Report(ctx)
|
|
res = report.get(data)
|
|
report.open(res)
|