# Part of Odoo. See LICENSE file for full copyright and licensing details. import logging import os import re import time from collections import Counter from odoo import api, fields, models, tools from odoo.exceptions import UserError from odoo.fields import Domain from odoo.http import Response from odoo.tools import SQL, escape_psql from odoo.addons.base.models.ir_http import EXTENSION_TO_WEB_MIMETYPES from odoo.addons.website.tools import text_from_html logger = logging.getLogger(__name__) class PageCannotBeCached(Exception): def __init__(self, result): self.result = result class WebsitePage(models.Model): _name = 'website.page' _inherits = {'ir.ui.view': 'view_id'} _inherit = [ 'website.published.multi.mixin', 'website.searchable.mixin', 'website.page_options.mixin', ] _description = 'Page' _order = 'website_id' # for how long a cache entry is considered valid (in seconds) _CACHE_DURATION = 3600 url = fields.Char('Page URL', required=True) view_id = fields.Many2one('ir.ui.view', string='View', required=True, index=True, ondelete="cascade") view_write_uid = fields.Many2one('res.users', "Last Content Update by", related='view_id.write_uid') view_write_date = fields.Datetime("Last Content Update on", related='view_id.write_date') website_indexed = fields.Boolean('Is Indexed', default=True) menu_ids = fields.One2many('website.menu', 'page_id', 'Related Menus') is_in_menu = fields.Boolean(compute='_compute_website_menu') is_homepage = fields.Boolean(compute='_compute_is_homepage', string='Homepage') is_visible = fields.Boolean(compute='_compute_visible', string='Is Visible') is_new_page_template = fields.Boolean(string="New Page Template", help='Add this page to the "+New" page templates. It will be added to the "Custom" category.') parent_id = fields.Many2one('website.page', string="Parent Page") parent_ids = fields.Many2many('website.page', compute='_compute_parent_ids') # don't use mixin website_id but use website_id on ir.ui.view instead website_id = fields.Many2one(related='view_id.website_id', store=True, readonly=False, ondelete='cascade') arch = fields.Text(related='view_id.arch', readonly=False, depends_context=('website_id',)) @api.constrains('parent_id') def _compute_parent_ids(self): for page in self: parent_ids = [] parent = page.parent_id while parent: if parent.id == page.id: raise UserError(self.env._("This would create a circular page hierarchy.")) parent_ids.append(parent.id) parent = parent.parent_id page.parent_ids = self.browse(reversed(parent_ids)) def _compute_is_homepage(self): website = self.env['website'].get_current_website() for page in self: page.is_homepage = page.url == (website.homepage_url or page.website_id == website and '/') def _compute_visible(self): for page in self: page.is_visible = page.website_published @api.depends('menu_ids') def _compute_website_menu(self): for page in self: page.is_in_menu = bool(page.menu_ids) # This update was added to make sure the mixin calculations are correct # (page.website_url > page.url). @api.depends('url') def _compute_website_url(self): for page in self: page.website_url = page.url @api.depends_context('uid') def _compute_can_publish(self): # Note: this `if`'s purpose it to optimize the way this is computed for # multiple records. if self.env.user.has_group('website.group_website_designer'): for record in self: record.can_publish = True else: super()._compute_can_publish() def _get_most_specific_pages(self): ''' Returns the most specific pages in self. ''' ids = [] previous_page = None page_keys = self.sudo().with_context(prefetch_fields=False).search_fetch( self.env['website'].browse(self.env.context.get('website_id')).website_domain(), field_names=['key'], ).mapped('key') page_keys_counts = Counter(page_keys) # Iterate a single time on the whole list sorted on specific-website first. for page in self.sorted(key=lambda p: (p.url, not p.website_id)): if ( (not previous_page or page.url != previous_page.url) # If a generic page (niche case) has been COWed and that COWed # page received a URL change, it should not let you access the # generic page anymore, despite having a different URL. and (page.website_id or page_keys_counts[page.key] == 1) ): ids.append(page.id) previous_page = page return self.browse(ids) def copy_data(self, default=None): vals_list = super().copy_data(default=default) if not default: return vals_list for page, vals in zip(self, vals_list): if not default.get('view_id'): new_view = page.view_id.copy({'website_id': default.get('website_id')}) vals['view_id'] = new_view.id vals['key'] = new_view.key vals['url'] = default.get('url', self.env['website'].get_unique_path(page.url)) return vals_list @api.model def clone_page(self, page_id, page_name=None, clone_menu=True): """ Clone a page, given its identifier :param page_id : website.page identifier """ page = self.browse(int(page_id)) copy_param = dict(name=page_name or page.name, website_id=self.env['website'].get_current_website().id) if page_name: url = '/' + self.env['ir.http']._slugify(page_name, max_length=1024, path=True) copy_param['url'] = self.env['website'].get_unique_path(url) new_page = page.with_context(check_translations=True).copy(copy_param) # Should not clone menu if the page was cloned from one website to another # Eg: Cloning a generic page (no website) will create a page with a website, we can't clone menu (not same container) if clone_menu and new_page.website_id == page.website_id: menu = self.env['website.menu'].search([('page_id', '=', page_id)], limit=1) if menu: # If the page being cloned has a menu, clone it too menu.copy({'url': new_page.url, 'name': new_page.name, 'page_id': new_page.id}) return new_page.url def unlink(self): # When a website_page is deleted, the ORM does not delete its # ir_ui_view. So we got to delete it ourself, but only if the # ir_ui_view is not used by another website_page. views_to_delete = self.view_id.filtered( lambda v: v.page_ids <= self and not v.inherit_children_ids ) # Rebind self to avoid unlink already deleted records from `ondelete="cascade"` self = self - views_to_delete.page_ids views_to_delete.unlink() # Make sure website.is_menu_cache_disabled() will be recomputed if self: self.env.registry.clear_cache('templates') return super().unlink() def write(self, vals): for page in self: website_id = False if vals.get('website_id') or page.website_id: website_id = vals.get('website_id') or page.website_id.id # If URL has been edited, slug it if 'url' in vals: url = vals['url'] or '' url = '/' + self.env['ir.http']._slugify(url, max_length=1024, path=True) if page.url != url: url = self.env['website'].with_context(website_id=website_id).get_unique_path(url) page.menu_ids.write({'url': url}) # Sync website's homepage URL website = self.env['website'].get_current_website() page_url_normalized = {'homepage_url': page.url} website._handle_homepage_url(page_url_normalized) if website.homepage_url == page_url_normalized['homepage_url']: website.homepage_url = url vals['url'] = url # If name has changed, check for key uniqueness if 'name' in vals and page.name != vals['name']: vals['key'] = self.env['website'].with_context(website_id=website_id).get_unique_key(self.env['ir.http']._slugify(vals['name'] or '')) if 'visibility' in vals: if vals['visibility'] != 'restricted_group': vals['group_ids'] = False # write on page == write on view # the view will invalidate the 'templates' cache return super().write(vals) def get_website_meta(self): self.ensure_one() return self.view_id.get_website_meta() @api.model def _search_get_detail(self, website, order, options): with_description = options['displayDescription'] # Read access on website.page requires sudo. requires_sudo = True domain = [website.website_domain()] if not self.env.user.has_group('website.group_website_designer'): # Rule must be reinforced because of sudo. domain.append([ ('website_published', '=', True), ('website_indexed', '=', True), ]) # Prevent accessing unaccessible pages domain.append([('visibility', '!=', 'password')]) if website.is_public_user(): domain.append([('visibility', '!=', 'connected')]) domain.append(Domain.OR([ [('group_ids', '=', False)], [('group_ids', 'in', self.env.user.group_ids.ids)] ])) search_fields = ['name', 'url'] fetch_fields = ['id', 'name', 'url'] mapping = { 'name': {'name': 'name', 'type': 'text', 'match': True}, 'website_url': {'name': 'url', 'type': 'text', 'truncate': False}, } if with_description: search_fields.append('arch_db') fetch_fields.append('arch') mapping['description'] = {'name': 'arch', 'type': 'text', 'html': True, 'match': True} return { 'model': 'website.page', 'base_domain': domain, 'requires_sudo': requires_sudo, 'search_fields': search_fields, 'fetch_fields': fetch_fields, 'mapping': mapping, 'icon': 'fa-file-o', } @api.model def _search_fetch(self, search_detail, search, limit, order): with_description = 'description' in search_detail['mapping'] # Cannot rely on the super's _search_fetch because the search must be # performed among the most specific pages only. fields = search_detail['search_fields'] base_domain = Domain.AND(search_detail['base_domain']) domain = self._search_build_domain([base_domain], search, fields, search_detail.get('search_extra')) most_specific_pages = self.env['website']._get_website_pages( domain=base_domain, order=order ) if with_description and search and most_specific_pages: # Perform search in translations # TODO Remove when domains will support xml_translate fields custom_view_domain = Domain.custom( to_sql=lambda table: SQL( "%(name)s ILIKE %(search)s OR %(arch_db)s ILIKE %(search)s", name=table.name, arch_db=table.arch_db, search=f"%{escape_psql(search)}%", ) ) # most_specific_pages is already filtered and ordered pages = self.sudo().with_context(active_test=False).search( Domain('view_id', 'any', custom_view_domain) & Domain('id', 'in', most_specific_pages.ids) ) # just update the domain for filtering domain |= Domain('id', 'in', pages.ids) results = most_specific_pages.filtered_domain(domain) # already sudo def filter_page(search, page, all_pages): # Exclude pages that do not pass ACL. Rule = page.env['ir.rule'].sudo(False) if not page.filtered_domain(Rule._compute_domain('website.page', 'read')): return False if not page.view_id.filtered_domain(Rule._compute_domain('ir.ui.view', 'read')): return False if search and with_description: # Search might have matched words in the xml tags and parameters therefore we make # sure the terms actually appear inside the text. text = '%s %s %s' % (page.name, page.url, text_from_html(page.arch)) pattern = '|'.join([re.escape(search_term) for search_term in search.split()]) return re.findall('(%s)' % pattern, text, flags=re.I) if pattern else False return True results = results.filtered(lambda result: filter_page(search, result, results)) return results[:limit], len(results) def action_page_debug_view(self): return { 'type': 'ir.actions.act_window', 'res_model': 'ir.ui.view', 'res_id': self.view_id.id, 'view_mode': 'form', 'view_id': self.env.ref('website.view_view_form_extend').id, } # website cache @api.model def _allow_to_use_cache(self, request): """ Checks if the generated HTML content is eligible for caching. This is useful for preventing sensitive or dynamic content from being stored. """ return ( request.httprequest.method == "GET" and not request.params # because the parameters are not part of the cache key and request.env.user._is_public() # only cache for unlogged user and not self._get_page_info(request)['group_ids'] # do not cache elements dependent on security access ) @api.model def _allow_cache_insertion(self, layout): """ Determines whether a page is allowed to be served from the cache based on the current request, URL, or session. """ return True @api.model def _post_process_response_from_cache(self, request, response) -> None: """ A hook called after a response is retrieved from the cache. This method allows for post-processing, such as incrementing counters or modifying HTTP headers, without regenerating the entire page. """ csrf_token = request.csrf_token(None) html = response.response[0] html = re.sub(r'csrf_token: "[^"]+"', f'csrf_token: {csrf_token!r}', html) html = re.sub(r'name="csrf_token" value="[^"]+"', f'name="csrf_token" value={csrf_token!r}', html) response.response = [html] # used for _register_website_track response._cached_view_id = self._get_page_info(request)['view_id'] response._cached_page = self @api.model def _get_cache_key(self, request): """ Allows for supplementing the base cache key with custom components (e.g., from the URL or session). This is essential for ensuring that the cache serves the correct version of a page based on specific parameters like user language or currency. """ return (request.website.id, request.lang.code, request.httprequest.path, request.session.debug) def _get_response(self, request): """ Returns the response corresponding to the request. The response may or may not come from the cache. The management and logic for this cache are placed directly on the `website.page` model, as it is the source of the HTML response for these records. This approach makes it easy to control caching behavior through a few new, overridable methods: - `_allow_to_use_cache` - `_allow_cache_insertion` - `_post_process_response_from_cache` - `_get_cache_key` The cache is an ORM cache from `_get_response_cached` with an added mechanism to update its values. After a certain period, the system will fetch the true response from `_get_response_raw` and update the cached value. This approach reduces the need for frequent `clear_caches` for non-critical changes while ensuring that the cached data remains accurate over time. """ self.ensure_one() if self._allow_to_use_cache(request): try: response, cache_key = self._get_response_cached(request) except PageCannotBeCached as notCache: if notCache.result: return notCache.result[0] if time.time() < response.time + self._CACHE_DURATION: resp = Response( headers=response.headers.copy(), mimetype=response.mimetype, content_type=response.content_type, status=response.status, response=[response.response[0]], ) self._post_process_response_from_cache(request, resp) return resp # The cached response is too old and considered out-of-date. Get it # from scratch and update the cache accordingly. response = self._get_response_raw(request) self._get_response_cached.__cache__.add_value(self, request, cache_value=(response, cache_key)) return response return self._get_response_raw(request) @tools.conditional( 'xml' not in tools.config['dev_mode'], tools.ormcache('self._get_cache_key(request)', cache='templates.cached_values'), ) def _get_response_cached(self, request) -> tuple[Response, int, str]: """ Returns the response corresponding to the request. If the response exists and `_allow_cache_insertio` return True, this response is cached. """ cache_key = self._get_cache_key(request) response = self._get_response_raw(request) result = response, cache_key if not response: raise PageCannotBeCached(result) response.flatten() if not self._allow_cache_insertion(response.response[-1]): raise PageCannotBeCached(result) return result def _get_response_raw(self, request) -> Response | None: """ Returns the raw response associated with the current request. This method is called by `_get_response_cached`, which handles caching the result. It is also called directly by `_get_response` if `_allow_to_use_cache` returns False. """ req_page = request.httprequest.path # fetch all prefetchable fields to get all data at once. If we use the # default fetch(), another query is necessary to get the page fields # like 'website_meta'. fields_to_fetch = [name for name, field in self._fields.items() if field.prefetch] self.fetch(fields_to_fetch) fields_to_fetch = [name for name, field in self.view_id._fields.items() if field.prefetch] self.view_id.fetch(fields_to_fetch) if ( (self.env.user.has_group('website.group_website_designer') or self.is_visible) and ( # If a generic page (niche case) has been COWed and that COWed # page received a URL change, it should not let you access the # generic page anymore, despite having a different URL. self.website_id or self.view_id.id == self.env['ir.ui.view'].with_context(website_id=request.website.id)._get_cached_template_info(self.view_id.key)['id'] ) ): _, ext = os.path.splitext(req_page) response = request.render(self.view_id.id, { 'main_object': self, }, mimetype=EXTENSION_TO_WEB_MIMETYPES.get(ext, 'text/html')) response.time = time.time() return response return None @tools.conditional( 'xml' not in tools.config['dev_mode'], tools.ormcache('(request.httprequest.path, self.env.context.get("website_id"))', cache='templates.cached_values'), ) @api.model def _get_page_info(self, request) -> dict | None: req_page = request.httprequest.path # specific page first page_domain = Domain('url', '=', req_page) & request.website.website_domain() page = self.sudo().search_fetch(page_domain, order='website_id asc', limit=1) # case insensitive search if not page: page_domain = Domain('url', '=ilike', req_page) & request.website.website_domain() page = self.sudo().search_fetch(page_domain, order='website_id asc', limit=1) if page: return { 'id': page.id, 'url': page.url, 'view_id': page.view_id.id, 'group_ids': page.group_ids.ids, }