mirror of
https://gitlab.com/flectra-hq/flectra.git
synced 2026-08-18 17:24:42 -05:00
455 lines
18 KiB
Python
455 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Part of Odoo, Flectra. See LICENSE file for full copyright and licensing details.
|
|
|
|
import base64
|
|
import functools
|
|
import json
|
|
import logging
|
|
import math
|
|
import re
|
|
|
|
from werkzeug import urls
|
|
|
|
from flectra import fields as flectra_fields, http, tools, _, SUPERUSER_ID
|
|
from flectra.exceptions import ValidationError, AccessError, MissingError, UserError, AccessDenied
|
|
from flectra.http import content_disposition, Controller, request, route
|
|
from flectra.tools import consteq
|
|
|
|
# --------------------------------------------------
|
|
# Misc tools
|
|
# --------------------------------------------------
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
def pager(url, total, page=1, step=30, scope=5, url_args=None):
|
|
""" Generate a dict with required value to render `website.pager` template. This method compute
|
|
url, page range to display, ... in the pager.
|
|
:param url : base url of the page link
|
|
:param total : number total of item to be splitted into pages
|
|
:param page : current page
|
|
:param step : item per page
|
|
:param scope : number of page to display on pager
|
|
:param url_args : additionnal parameters to add as query params to page url
|
|
:type url_args : dict
|
|
:returns dict
|
|
"""
|
|
# Compute Pager
|
|
page_count = int(math.ceil(float(total) / step))
|
|
|
|
page = max(1, min(int(page if str(page).isdigit() else 1), page_count))
|
|
scope -= 1
|
|
|
|
pmin = max(page - int(math.floor(scope/2)), 1)
|
|
pmax = min(pmin + scope, page_count)
|
|
|
|
if pmax - pmin < scope:
|
|
pmin = pmax - scope if pmax - scope > 0 else 1
|
|
|
|
def get_url(page):
|
|
_url = "%s/page/%s" % (url, page) if page > 1 else url
|
|
if url_args:
|
|
_url = "%s?%s" % (_url, urls.url_encode(url_args))
|
|
return _url
|
|
|
|
return {
|
|
"page_count": page_count,
|
|
"offset": (page - 1) * step,
|
|
"page": {
|
|
'url': get_url(page),
|
|
'num': page
|
|
},
|
|
"page_first": {
|
|
'url': get_url(1),
|
|
'num': 1
|
|
},
|
|
"page_start": {
|
|
'url': get_url(pmin),
|
|
'num': pmin
|
|
},
|
|
"page_previous": {
|
|
'url': get_url(max(pmin, page - 1)),
|
|
'num': max(pmin, page - 1)
|
|
},
|
|
"page_next": {
|
|
'url': get_url(min(pmax, page + 1)),
|
|
'num': min(pmax, page + 1)
|
|
},
|
|
"page_end": {
|
|
'url': get_url(pmax),
|
|
'num': pmax
|
|
},
|
|
"page_last": {
|
|
'url': get_url(page_count),
|
|
'num': page_count
|
|
},
|
|
"pages": [
|
|
{'url': get_url(page_num), 'num': page_num} for page_num in range(pmin, pmax+1)
|
|
]
|
|
}
|
|
|
|
|
|
def get_records_pager(ids, current):
|
|
if current.id in ids and (hasattr(current, 'website_url') or hasattr(current, 'access_url')):
|
|
attr_name = 'access_url' if hasattr(current, 'access_url') else 'website_url'
|
|
idx = ids.index(current.id)
|
|
prev_record = idx != 0 and current.browse(ids[idx - 1])
|
|
next_record = idx < len(ids) - 1 and current.browse(ids[idx + 1])
|
|
|
|
if prev_record and prev_record[attr_name] and attr_name == "access_url":
|
|
prev_url = '%s?access_token=%s' % (prev_record[attr_name], prev_record._portal_ensure_token())
|
|
elif prev_record and prev_record[attr_name]:
|
|
prev_url = prev_record[attr_name]
|
|
else:
|
|
prev_url = prev_record
|
|
|
|
if next_record and next_record[attr_name] and attr_name == "access_url":
|
|
next_url = '%s?access_token=%s' % (next_record[attr_name], next_record._portal_ensure_token())
|
|
elif next_record and next_record[attr_name]:
|
|
next_url = next_record[attr_name]
|
|
else:
|
|
next_url = next_record
|
|
|
|
return {
|
|
'prev_record': prev_url,
|
|
'next_record': next_url,
|
|
}
|
|
return {}
|
|
|
|
|
|
def _build_url_w_params(url_string, query_params, remove_duplicates=True):
|
|
""" Rebuild a string url based on url_string and correctly compute query parameters
|
|
using those present in the url and those given by query_params. Having duplicates in
|
|
the final url is optional. For example:
|
|
|
|
* url_string = '/my?foo=bar&error=pay'
|
|
* query_params = {'foo': 'bar2', 'alice': 'bob'}
|
|
* if remove duplicates: result = '/my?foo=bar2&error=pay&alice=bob'
|
|
* else: result = '/my?foo=bar&foo=bar2&error=pay&alice=bob'
|
|
"""
|
|
url = urls.url_parse(url_string)
|
|
url_params = url.decode_query()
|
|
if remove_duplicates: # convert to standard dict instead of werkzeug multidict to remove duplicates automatically
|
|
url_params = url_params.to_dict()
|
|
url_params.update(query_params)
|
|
return url.replace(query=urls.url_encode(url_params)).to_url()
|
|
|
|
|
|
class CustomerPortal(Controller):
|
|
|
|
MANDATORY_BILLING_FIELDS = ["name", "phone", "email", "street", "city", "country_id"]
|
|
OPTIONAL_BILLING_FIELDS = ["zipcode", "state_id", "vat", "company_name"]
|
|
|
|
_items_per_page = 20
|
|
|
|
def _prepare_portal_layout_values(self):
|
|
"""Values for /my/* templates rendering.
|
|
|
|
Does not include the record counts.
|
|
"""
|
|
# get customer sales rep
|
|
sales_user = False
|
|
partner = request.env.user.partner_id
|
|
if partner.user_id and not partner.user_id._is_public():
|
|
sales_user = partner.user_id
|
|
|
|
return {
|
|
'sales_user': sales_user,
|
|
'page_name': 'home',
|
|
}
|
|
|
|
def _prepare_home_portal_values(self, counters):
|
|
"""Values for /my & /my/home routes template rendering.
|
|
|
|
Includes the record count for the displayed badges.
|
|
where 'coutners' is the list of the displayed badges
|
|
and so the list to compute.
|
|
"""
|
|
return {}
|
|
|
|
@route(['/my/counters'], type='json', auth="user", website=True)
|
|
def counters(self, counters, **kw):
|
|
return self._prepare_home_portal_values(counters)
|
|
|
|
@route(['/my', '/my/home'], type='http', auth="user", website=True)
|
|
def home(self, **kw):
|
|
values = self._prepare_portal_layout_values()
|
|
return request.render("portal.portal_my_home", values)
|
|
|
|
@route(['/my/account'], type='http', auth='user', website=True)
|
|
def account(self, redirect=None, **post):
|
|
values = self._prepare_portal_layout_values()
|
|
partner = request.env.user.partner_id
|
|
values.update({
|
|
'error': {},
|
|
'error_message': [],
|
|
})
|
|
|
|
if post and request.httprequest.method == 'POST':
|
|
error, error_message = self.details_form_validate(post)
|
|
values.update({'error': error, 'error_message': error_message})
|
|
values.update(post)
|
|
if not error:
|
|
values = {key: post[key] for key in self.MANDATORY_BILLING_FIELDS}
|
|
values.update({key: post[key] for key in self.OPTIONAL_BILLING_FIELDS if key in post})
|
|
for field in set(['country_id', 'state_id']) & set(values.keys()):
|
|
try:
|
|
values[field] = int(values[field])
|
|
except:
|
|
values[field] = False
|
|
values.update({'zip': values.pop('zipcode', '')})
|
|
partner.sudo().write(values)
|
|
if redirect:
|
|
return request.redirect(redirect)
|
|
return request.redirect('/my/home')
|
|
|
|
countries = request.env['res.country'].sudo().search([])
|
|
states = request.env['res.country.state'].sudo().search([])
|
|
|
|
values.update({
|
|
'partner': partner,
|
|
'countries': countries,
|
|
'states': states,
|
|
'has_check_vat': hasattr(request.env['res.partner'], 'check_vat'),
|
|
'redirect': redirect,
|
|
'page_name': 'my_details',
|
|
})
|
|
|
|
response = request.render("portal.portal_my_details", values)
|
|
response.headers['X-Frame-Options'] = 'DENY'
|
|
return response
|
|
|
|
@route('/my/security', type='http', auth='user', website=True, methods=['GET', 'POST'])
|
|
def security(self, **post):
|
|
values = self._prepare_portal_layout_values()
|
|
values['get_error'] = get_error
|
|
|
|
if request.httprequest.method == 'POST':
|
|
values.update(self._update_password(
|
|
post['old'].strip(),
|
|
post['new1'].strip(),
|
|
post['new2'].strip()
|
|
))
|
|
|
|
return request.render('portal.portal_my_security', values, headers={
|
|
'X-Frame-Options': 'DENY'
|
|
})
|
|
|
|
def _update_password(self, old, new1, new2):
|
|
for k, v in [('old', old), ('new1', new1), ('new2', new2)]:
|
|
if not v:
|
|
return {'errors': {'password': {k: _("You cannot leave any password empty.")}}}
|
|
|
|
if new1 != new2:
|
|
return {'errors': {'password': {'new2': _("The new password and its confirmation must be identical.")}}}
|
|
|
|
try:
|
|
request.env['res.users'].change_password(old, new1)
|
|
except UserError as e:
|
|
return {'errors': {'password': e.name}}
|
|
except AccessDenied as e:
|
|
msg = e.args[0]
|
|
if msg == AccessDenied().args[0]:
|
|
msg = _('The old password you provided is incorrect, your password was not changed.')
|
|
return {'errors': {'password': {'old': msg}}}
|
|
|
|
# update session token so the user does not get logged out (cache cleared by passwd change)
|
|
new_token = request.env.user._compute_session_token(request.session.sid)
|
|
request.session.session_token = new_token
|
|
|
|
return {'success': {'password': True}}
|
|
|
|
@http.route('/portal/attachment/add', type='http', auth='public', methods=['POST'], website=True)
|
|
def attachment_add(self, name, file, res_model, res_id, access_token=None, **kwargs):
|
|
"""Process a file uploaded from the portal chatter and create the
|
|
corresponding `ir.attachment`.
|
|
|
|
The attachment will be created "pending" until the associated message
|
|
is actually created, and it will be garbage collected otherwise.
|
|
|
|
:param name: name of the file to save.
|
|
:type name: string
|
|
|
|
:param file: the file to save
|
|
:type file: werkzeug.FileStorage
|
|
|
|
:param res_model: name of the model of the original document.
|
|
To check access rights only, it will not be saved here.
|
|
:type res_model: string
|
|
|
|
:param res_id: id of the original document.
|
|
To check access rights only, it will not be saved here.
|
|
:type res_id: int
|
|
|
|
:param access_token: access_token of the original document.
|
|
To check access rights only, it will not be saved here.
|
|
:type access_token: string
|
|
|
|
:return: attachment data {id, name, mimetype, file_size, access_token}
|
|
:rtype: dict
|
|
"""
|
|
try:
|
|
self._document_check_access(res_model, int(res_id), access_token=access_token)
|
|
except (AccessError, MissingError) as e:
|
|
raise UserError(_("The document does not exist or you do not have the rights to access it."))
|
|
|
|
IrAttachment = request.env['ir.attachment']
|
|
access_token = False
|
|
|
|
# Avoid using sudo or creating access_token when not necessary: internal
|
|
# users can create attachments, as opposed to public and portal users.
|
|
if not request.env.user.has_group('base.group_user'):
|
|
IrAttachment = IrAttachment.sudo().with_context(binary_field_real_user=IrAttachment.env.user)
|
|
access_token = IrAttachment._generate_access_token()
|
|
|
|
# At this point the related message does not exist yet, so we assign
|
|
# those specific res_model and res_is. They will be correctly set
|
|
# when the message is created: see `portal_chatter_post`,
|
|
# or garbage collected otherwise: see `_garbage_collect_attachments`.
|
|
attachment = IrAttachment.create({
|
|
'name': name,
|
|
'datas': base64.b64encode(file.read()),
|
|
'res_model': 'mail.compose.message',
|
|
'res_id': 0,
|
|
'access_token': access_token,
|
|
})
|
|
return request.make_response(
|
|
data=json.dumps(attachment.read(['id', 'name', 'mimetype', 'file_size', 'access_token'])[0]),
|
|
headers=[('Content-Type', 'application/json')]
|
|
)
|
|
|
|
@http.route('/portal/attachment/remove', type='json', auth='public')
|
|
def attachment_remove(self, attachment_id, access_token=None):
|
|
"""Remove the given `attachment_id`, only if it is in a "pending" state.
|
|
|
|
The user must have access right on the attachment or provide a valid
|
|
`access_token`.
|
|
"""
|
|
try:
|
|
attachment_sudo = self._document_check_access('ir.attachment', int(attachment_id), access_token=access_token)
|
|
except (AccessError, MissingError) as e:
|
|
raise UserError(_("The attachment does not exist or you do not have the rights to access it."))
|
|
|
|
if attachment_sudo.res_model != 'mail.compose.message' or attachment_sudo.res_id != 0:
|
|
raise UserError(_("The attachment %s cannot be removed because it is not in a pending state.", attachment_sudo.name))
|
|
|
|
if attachment_sudo.env['mail.message'].search([('attachment_ids', 'in', attachment_sudo.ids)]):
|
|
raise UserError(_("The attachment %s cannot be removed because it is linked to a message.", attachment_sudo.name))
|
|
|
|
return attachment_sudo.unlink()
|
|
|
|
def details_form_validate(self, data):
|
|
error = dict()
|
|
error_message = []
|
|
|
|
# Validation
|
|
for field_name in self.MANDATORY_BILLING_FIELDS:
|
|
if not data.get(field_name):
|
|
error[field_name] = 'missing'
|
|
|
|
# email validation
|
|
if data.get('email') and not tools.single_email_re.match(data.get('email')):
|
|
error["email"] = 'error'
|
|
error_message.append(_('Invalid Email! Please enter a valid email address.'))
|
|
|
|
# vat validation
|
|
partner = request.env.user.partner_id
|
|
if data.get("vat") and partner and partner.vat != data.get("vat"):
|
|
if partner.can_edit_vat():
|
|
if hasattr(partner, "check_vat"):
|
|
if data.get("country_id"):
|
|
data["vat"] = request.env["res.partner"].fix_eu_vat_number(int(data.get("country_id")), data.get("vat"))
|
|
partner_dummy = partner.new({
|
|
'vat': data['vat'],
|
|
'country_id': (int(data['country_id'])
|
|
if data.get('country_id') else False),
|
|
})
|
|
try:
|
|
partner_dummy.check_vat()
|
|
except ValidationError:
|
|
error["vat"] = 'error'
|
|
else:
|
|
error_message.append(_('Changing VAT number is not allowed once document(s) have been issued for your account. Please contact us directly for this operation.'))
|
|
|
|
# error message for empty required fields
|
|
if [err for err in error.values() if err == 'missing']:
|
|
error_message.append(_('Some required fields are empty.'))
|
|
|
|
unknown = [k for k in data if k not in self.MANDATORY_BILLING_FIELDS + self.OPTIONAL_BILLING_FIELDS]
|
|
if unknown:
|
|
error['common'] = 'Unknown field'
|
|
error_message.append("Unknown field '%s'" % ','.join(unknown))
|
|
|
|
return error, error_message
|
|
|
|
def _document_check_access(self, model_name, document_id, access_token=None):
|
|
document = request.env[model_name].browse([document_id])
|
|
document_sudo = document.with_user(SUPERUSER_ID).exists()
|
|
if not document_sudo:
|
|
raise MissingError(_("This document does not exist."))
|
|
try:
|
|
document.check_access_rights('read')
|
|
document.check_access_rule('read')
|
|
except AccessError:
|
|
if not access_token or not document_sudo.access_token or not consteq(document_sudo.access_token, access_token):
|
|
raise
|
|
return document_sudo
|
|
|
|
def _get_page_view_values(self, document, access_token, values, session_history, no_breadcrumbs, **kwargs):
|
|
if access_token:
|
|
# if no_breadcrumbs = False -> force breadcrumbs even if access_token to `invite` users to register if they click on it
|
|
values['no_breadcrumbs'] = no_breadcrumbs
|
|
values['access_token'] = access_token
|
|
values['token'] = access_token # for portal chatter
|
|
|
|
# Those are used notably whenever the payment form is implied in the portal.
|
|
if kwargs.get('error'):
|
|
values['error'] = kwargs['error']
|
|
if kwargs.get('warning'):
|
|
values['warning'] = kwargs['warning']
|
|
if kwargs.get('success'):
|
|
values['success'] = kwargs['success']
|
|
# Email token for posting messages in portal view with identified author
|
|
if kwargs.get('pid'):
|
|
values['pid'] = kwargs['pid']
|
|
if kwargs.get('hash'):
|
|
values['hash'] = kwargs['hash']
|
|
|
|
history = request.session.get(session_history, [])
|
|
values.update(get_records_pager(history, document))
|
|
|
|
return values
|
|
|
|
def _show_report(self, model, report_type, report_ref, download=False):
|
|
if report_type not in ('html', 'pdf', 'text'):
|
|
raise UserError(_("Invalid report type: %s", report_type))
|
|
|
|
report_sudo = request.env.ref(report_ref).with_user(SUPERUSER_ID)
|
|
|
|
if not isinstance(report_sudo, request.env.registry['ir.actions.report']):
|
|
raise UserError(_("%s is not the reference of a report", report_ref))
|
|
|
|
if hasattr(model, 'company_id'):
|
|
report_sudo = report_sudo.with_company(model.company_id)
|
|
|
|
method_name = '_render_qweb_%s' % (report_type)
|
|
report = getattr(report_sudo, method_name)([model.id], data={'report_type': report_type})[0]
|
|
reporthttpheaders = [
|
|
('Content-Type', 'application/pdf' if report_type == 'pdf' else 'text/html'),
|
|
('Content-Length', len(report)),
|
|
]
|
|
if report_type == 'pdf' and download:
|
|
filename = "%s.pdf" % (re.sub('\W+', '-', model._get_report_base_filename()))
|
|
reporthttpheaders.append(('Content-Disposition', content_disposition(filename)))
|
|
return request.make_response(report, headers=reporthttpheaders)
|
|
|
|
def get_error(e, path=''):
|
|
""" Recursively dereferences `path` (a period-separated sequence of dict
|
|
keys) in `e` (an error dict or value), returns the final resolution IIF it's
|
|
an str, otherwise returns None
|
|
"""
|
|
for k in (path.split('.') if path else []):
|
|
if not isinstance(e, dict):
|
|
return None
|
|
e = e.get(k)
|
|
|
|
return e if isinstance(e, str) else None
|