1b-agent/enterprise-addons
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
19.0
enterprise-addons/documents/models/documents_document.py
T

2844 lines
140 KiB
Python
Raw Permalink Blame History

import base64
import contextlib
import io
import logging
import re
import string
import uuid
from ast import literal_eval
from collections import OrderedDict, defaultdict
import requests
from dateutil.relativedelta import relativedelta
from urllib.parse import quote
from werkzeug.urls import url_encode
import odoo
from odoo import _, api, Command, fields, models, modules, SUPERUSER_ID
from odoo.exceptions import AccessError, MissingError, UserError, ValidationError
from odoo.fields import Domain
from odoo.tools import groupby, SQL
from odoo.tools.image import image_process
from odoo.tools.mimetypes import get_extension
from odoo.tools.misc import clean_context, consteq
from odoo.tools.pdf import PdfFileReader
from odoo.addons.mail.tools import link_preview
_logger = logging.getLogger(__name__)
def _sanitize_file_extension(extension):
""" Remove leading and trailing spacing + Remove leading "." """
return re.sub(r'^[\s.]+|\s+$', '', extension)
class DocumentsDocument(models.Model):
_name = 'documents.document'
_description = 'Document'
_inherit = ['mail.thread', 'mail.activity.mixin', 'mail.alias.mixin.optional']
_mail_post_access = 'read'
_order = 'sequence, id desc'
_parent_name = 'folder_id'
_parent_store = True
_systray_view = 'activity'
# Attachment
attachment_id = fields.Many2one('ir.attachment', ondelete='cascade', bypass_search_access=True, copy=False)
attachment_name = fields.Char('Attachment Name', related='attachment_id.name', readonly=False)
attachment_type = fields.Selection(string='Attachment Type', related='attachment_id.type', readonly=False)
is_editable_attachment = fields.Boolean(default=False, help='True if we can edit the link attachment.')
is_multipage = fields.Boolean(
'Is considered multipage', compute='_compute_is_multipage', store=True, readonly=False)
datas = fields.Binary(related='attachment_id.raw', related_sudo=True, readonly=False, prefetch=False)
raw = fields.Binary(related='attachment_id.raw', related_sudo=True, readonly=False, prefetch=False)
file_extension = fields.Char('File Extension', copy=True, store=True, readonly=False,
compute='_compute_file_extension', inverse='_inverse_file_extension')
file_size = fields.Integer(compute='_compute_file_size', store=True)
checksum = fields.Char(related='attachment_id.checksum')
mimetype = fields.Char(related='attachment_id.mimetype')
res_model = fields.Char('Resource Model', compute="_compute_res_record", recursive=True,
inverse="_inverse_res_record", store=True)
res_id = fields.Many2oneReference('Resource ID', compute="_compute_res_record", recursive=True,
inverse="_inverse_res_record", store=True, model_field="res_model")
res_name = fields.Char('Resource Name', compute="_compute_res_name", compute_sudo=True)
index_content = fields.Text(related='attachment_id.index_content')
description = fields.Text('Attachment Description', related='attachment_id.description', readonly=False)
# Versioning
previous_attachment_ids = fields.Many2many('ir.attachment', string="History", bypass_search_access=True)
# Document
name = fields.Char('Name', copy=True, store=True, compute='_compute_name_and_preview', readonly=False,
translate=True, tracking=True)
active = fields.Boolean(default=True, string="Active")
thumbnail = fields.Binary(
readonly=False, store=True, attachment=True, compute='_compute_thumbnail', recursive=True)
thumbnail_status = fields.Selection([
('present', 'Present'), # Document has a thumbnail
('error', 'Error'), # Error when generating the thumbnail
('client_generated', 'Client Generated'), # The PDF thumbnail is generated by the user browser
('restricted', 'Inaccessible'), # Shortcut to no-permission source
], compute="_compute_thumbnail", store=True, readonly=False, recursive=True,
)
url = fields.Char('Link URL', index=True, size=1024, tracking=True)
url_preview_image = fields.Char(
'URL Preview Image', store=True, compute='_compute_name_and_preview', readonly=False)
res_model_name = fields.Char(compute='_compute_res_model_name')
type = fields.Selection([('url', 'URL'), ('binary', 'File'), ('folder', 'Folder')],
default='binary', string='Type', required=True, readonly=True, index=True)
shortcut_document_id = fields.Many2one('documents.document', 'Source Document', ondelete='cascade',
index='btree_not_null')
shortcut_document_owner_id = fields.Many2one(
'res.users', 'Source Document Owner', related="shortcut_document_id.owner_id", store=True)
shortcut_ids = fields.One2many('documents.document', 'shortcut_document_id')
favorited_ids = fields.Many2many('res.users', string="Favorite of")
is_favorited = fields.Boolean(compute='_compute_is_favorited', inverse='_inverse_is_favorited')
tag_ids = fields.Many2many('documents.tag', 'document_tag_rel', string="Tags")
partner_id = fields.Many2one('res.partner', string="Contact", tracking=True, index='btree_not_null')
owner_id = fields.Many2one(
'res.users', tracking=True, index=True, string="Owner", copy=False,
default=lambda self: self.env.user.id if self.env.user.active else False)
lock_uid = fields.Many2one('res.users', string="Locked by", tracking=True)
request_activity_id = fields.Many2one('mail.activity')
requestee_partner_id = fields.Many2one('res.partner')
sequence = fields.Integer('Sequence', default=10)
# Access
document_token = fields.Char(
required=True, copy=False, index='btree',
default=lambda __: base64.urlsafe_b64encode(uuid.uuid4().bytes).decode().removesuffix('=='))
access_token = fields.Char(compute='_compute_access_token')
access_url = fields.Char(string='Access url', compute='_compute_access_url')
is_access_via_link_hidden = fields.Boolean(
'Link Access Hidden', index=True,
help='If "True", only people given direct access to this document will be able to view it. '
'If "False", access with the link also given to all who can access the parent folder.'
)
access_via_link = fields.Selection(
[('view', 'Viewer'), ('edit', 'Editor'), ('none', 'None')],
string='Link Access Rights', default='none', required=True, index=True)
access_internal = fields.Selection(
[('view', 'Viewer'), ('edit', 'Editor'), ('none', 'None')],
string='Internal Users Rights', default='none', required=True, index=True)
access_ids = fields.One2many('documents.access', 'document_id', string="Allowed Access")
user_permission = fields.Selection(
[('edit', 'Editor'), ('view', 'Viewer'), ('none', 'None')], string='User permission',
compute='_compute_user_permission', search='_search_user_permission', compute_sudo=True)
user_can_move = fields.Boolean(string='Can move it', compute='_compute_user_can_move')
# Folder = parent document
parent_path = fields.Char(index=True) # see '_parent_store' implementation in the ORM for details
folder_id = fields.Many2one('documents.document', string='Folder', ondelete='set null', tracking=True,
domain="[('type', '=', 'folder'), ('shortcut_document_id', '=', False)]",
required=False, index=True, search='_search_folder_id')
user_folder_id = fields.Char(string='Parent', compute='_compute_user_folder_id', search='_search_user_folder_id')
children_ids = fields.One2many('documents.document', 'folder_id')
deletion_delay = fields.Integer("Deletion delay", compute="_compute_deletion_delay",
help="Delay after permanent deletion of the document in the trash (days)")
company_id = fields.Many2one('res.company', string='Company', store=True, readonly=False, index=True)
# Activity
create_activity_option = fields.Boolean(string='Create a new activity', compute='_compute_create_activity_option',
store=True, readonly=False)
create_activity_type_id = fields.Many2one('mail.activity.type', string="Activity type")
create_activity_summary = fields.Char('Summary')
create_activity_date_deadline_range = fields.Integer(string='Due Date In')
create_activity_date_deadline_range_type = fields.Selection([
('days', 'Days'),
('weeks', 'Weeks'),
('months', 'Months'),
], string='Due type', default='days')
create_activity_note = fields.Html(string="Note")
create_activity_user_id = fields.Many2one('res.users', string='Responsible')
# Actions that we can do on the document
available_embedded_actions_ids = fields.Many2many(
'ir.embedded.actions', 'Available Actions', compute='_compute_available_embedded_actions_ids',
groups='base.group_user')
# Alias
alias_tag_ids = fields.Many2many('documents.tag', 'document_alias_tag_rel', string="Alias Tags")
mail_alias_domain_count = fields.Integer("Mail Alias Domain Count", compute='_compute_mail_alias_domain_count')
# UI fields
last_access_date_group = fields.Selection(selection=[
('0_older', 'Older'),
('1_month', 'This Month'),
('2_week', 'This Week'),
('3_day', 'Today'),
], string="Last Accessed On", compute='_compute_last_access_date_group', search='_search_last_access_date_group')
_attachment_unique = models.Constraint(
'unique (attachment_id)',
"This attachment is already a document",
)
_document_token_unique = models.Constraint(
'unique (document_token)',
"Access tokens already used.",
)
_folder_id_not_id = models.Constraint(
'check(folder_id <> id)',
"A folder cannot be included in itself",
)
_shortcut_document_id_not_id = models.Constraint(
'check(shortcut_document_id <> id)',
"A shortcut cannot point to itself",
)
_res_model_res_id_idx = models.Index("(res_model, res_id)")
@api.depends('document_token')
def _compute_access_token(self):
for document in self:
document.access_token = f"{document.document_token}o{document.id or 0:x}"
@api.depends('access_token')
def _compute_access_url(self):
for document in self:
document.access_url = f'{document.sudo().get_base_url()}/odoo/documents/{quote(document.access_token, safe="")}'
@api.depends('create_activity_type_id', 'create_activity_user_id')
def _compute_create_activity_option(self):
to_activate = self.filtered(lambda d: d.create_activity_type_id and d.create_activity_user_id)
to_activate.create_activity_option = True
(self - to_activate).create_activity_option = False
@api.depends("folder_id", "company_id")
@api.depends_context("uid", "allowed_company_ids", "documents_show_parent_name")
def _compute_display_name(self):
accessible_records = self._filtered_access('read')
not_accessible_records = self - accessible_records
not_accessible_records.display_name = _("Restricted")
folders = accessible_records.filtered(lambda d: d.type == 'folder')
for record in folders:
if record.user_permission != 'none':
record.display_name = (
record.name
if not self.env.context.get('documents_show_parent_name') or not record.folder_id
else _("%(record)s (in %(parent)s)", record=record.name, parent=record.folder_id.name)
)
else:
record.display_name = _("Restricted Folder")
for record in accessible_records - folders:
record.display_name = record.name
@api.depends('name', 'type', 'shortcut_document_id.name')
def _compute_file_extension(self):
for record in self:
if record.type != 'binary':
record.file_extension = False
elif record.shortcut_document_id.name:
file_extension = _sanitize_file_extension(get_extension(record.shortcut_document_id.name.strip()))
record.file_extension = file_extension or False
elif record.name:
record.file_extension = _sanitize_file_extension(get_extension(record.name.strip())) or False
@api.depends('attachment_id', 'shortcut_document_id.attachment_id')
def _compute_file_size(self):
shortcuts = self.filtered('shortcut_document_id')
for document in self - shortcuts:
document.file_size = document.attachment_id.file_size
for document in shortcuts:
document.file_size = document.shortcut_document_id.file_size
def _inverse_file_extension(self):
for record in self:
file_extension = _sanitize_file_extension(record.file_extension) if record.file_extension else False
(record | record.shortcut_ids).file_extension = file_extension
@api.constrains('document_token')
def _check_document_token(self):
charset = set(string.ascii_letters + string.digits + '-_')
for document in self:
if len(document.document_token or '') != 22 or set(document.document_token) - charset:
raise ValidationError(_('Invalid document token'))
@api.constrains('shortcut_document_id', 'shortcut_ids', 'type', 'folder_id', 'children_ids', 'company_id')
def _check_shortcut_fields(self):
errors = []
wrong_types, wrong_companies = self.browse(), self.browse()
# Access rights can allow for a document to be edited without having access to its parent folder
wrong_parents_sudo = self.folder_id.sudo().filtered('shortcut_document_id')
for target in self.filtered('shortcut_ids'):
for shortcut in target.shortcut_ids:
if shortcut.type != target.type:
wrong_types |= shortcut
for shortcut in self.filtered('shortcut_document_id'):
if shortcut.type != shortcut.shortcut_document_id.type:
wrong_types |= shortcut
if shortcut.children_ids:
wrong_parents_sudo |= shortcut
if (shortcut.shortcut_document_id.company_id
and shortcut.shortcut_document_id.company_id != shortcut.company_id):
wrong_companies |= shortcut
if wrong_types:
message = _("The following documents/shortcuts have a type mismatch: \n")
documents_list = "\n- ".join(wrong_types.mapped('name'))
errors.append(f'{message}\n- {documents_list}')
if wrong_parents_sudo:
message = _("The following shortcuts cannot be set as documents parents: \n")
shortcuts_list = "\n- ".join(wrong_parents_sudo.mapped('name'))
errors.append(f'{message}\n- {shortcuts_list}')
if wrong_companies:
message = _("The following documents/shortcuts have a company mismatch: \n")
shortcuts_list = "\n- ".join(wrong_companies.mapped('name'))
errors.append(f'{message}\n- {shortcuts_list}')
if errors:
raise ValidationError('\n\n'.join(errors))
@api.constrains('owner_id', 'folder_id')
def _check_root_documents_owner_id(self):
root_documents = self.filtered(lambda d: not d.folder_id)
unauthorized_owners_sudo = root_documents._get_unauthorized_root_document_owners_sudo()
if unauthorized_owners_sudo:
users_documents_list = [
(document.owner_id.name, document.name)
for document in root_documents
if document.owner_id in unauthorized_owners_sudo
]
raise ValidationError(
_("The following user(s) cannot own root documents/folders: \n- %(lines)s",
lines="\n-".join(f'{user_name}: {doc_name}' for user_name, doc_name in users_documents_list))
)
def _get_unauthorized_root_document_owners_sudo(self):
""" Return sudo'ed documents records as only used by system process."""
return self.mapped('owner_id').sudo().filtered('share')
@api.constrains('type', 'alias_name')
def _check_alias(self):
wrong_records = self.filtered(lambda d: (d.type != 'folder' or d.shortcut_document_id) and d.alias_name)
if wrong_records:
raise ValidationError(_(
"The following documents can't have alias: \n- %(records)s",
records="\n-".join(wrong_records.mapped('name'))))
@api.constrains('res_model')
def _check_res_model(self):
if self.filtered(lambda d: d.res_model == 'documents.document'):
raise ValidationError(_('A document can not be linked to itself or another document.'))
def _is_company_root_folder(self):
self.ensure_one()
return self.type == 'folder' and not self.folder_id and not self.owner_id
@api.depends_context('uid')
@api.depends('folder_id', 'folder_id.user_permission', 'owner_id', 'active')
def _compute_user_folder_id(self):
SHARED = 'SHARED' if not self.env.user.share else False
self.user_folder_id = False # Inaccessible
active_documents = self.filtered('active')
(self - active_documents).user_folder_id = "TRASH"
for document in active_documents.filtered(lambda d: d.user_permission != 'none'):
if document.folder_id:
if document.folder_id.user_permission != 'none':
document.user_folder_id = str(document.folder_id.id)
else:
document.user_folder_id = SHARED
elif self.env.user.share:
document.user_folder_id = False
elif not document.owner_id:
document.user_folder_id = 'COMPANY'
elif document.owner_id == self.env.user:
document.user_folder_id = 'MY' # Root of user's drive
else:
document.user_folder_id = SHARED # Root of another user's drive
def _search_user_folder_id(self, operator, operand):
"""Search domain for user_folder_id virtual folder_id.
Note that searching in "RECENT" is allowed for practicality w.r.t. webclient
even though no record will have "RECENT" as computed `user_folder_id`
"""
if operator not in ('in', 'child_of'):
return NotImplemented
values = {operand} if isinstance(operand, str) else set(operand)
if 'TRASH' in values:
# Would need `active_test=False` in context
raise UserError(_("Searching on TRASH is not supported."))
domain_parts = []
folder_ids = []
for value in values:
if isinstance(value, int):
value = str(value)
elif not isinstance(value, str):
raise UserError(_("Invalid search operand."))
if not value and self.env.user.share:
domain_parts.append(Domain("folder_id", "=", False) | Domain('folder_id', 'not any', []))
elif not value:
domain_parts.append(Domain.FALSE)
elif value == "COMPANY":
domain_parts.append(Domain('folder_id', '=', False) & Domain('owner_id', '=', False))
elif value == "MY":
domain_parts.append(Domain('folder_id', '=', False) & Domain('owner_id', '=', self.env.user.id))
elif value == "RECENT":
domain_parts.append(Domain(
'access_ids', 'any',
Domain('partner_id', '=', self.env.user.partner_id.id) & Domain('last_access_date', '!=', False)))
elif value == "SHARED":
# Find records without permission on folder_id as directly searching on user_permission = 'none' is not allowed.
domain_parts.append(
Domain('folder_id', '!=', False) & Domain('folder_id', 'not any', [])
| Domain("folder_id", "=", False) & Domain("owner_id", "not in", [self.env.user.id, False])
)
elif value.isnumeric():
folder_ids.append(int(value))
else:
raise UserError(_("Unknown searched value %s", value))
if folder_ids:
domain_parts.append(Domain('folder_id', 'in', folder_ids))
domain = Domain.OR(domain_parts)
if operator == 'child_of':
if len(values) > 1:
raise UserError(_('Only one value can be searched for children of `user_folder_id`'))
return self._get_child_of_domain(domain, values.pop())
return domain
@api.model
def _get_child_of_domain(self, roots_domain, value: str | int):
"""Make sure that all intermediate folders are also part of the result."""
if not isinstance(value, str | int):
raise UserError(_('Only one string or number value can be searched for documents `child_of`.'))
if value == 'SHARED':
# Can't use sudo speedup here
shared_roots = self.with_context(active_test=False).search_fetch(roots_domain, ['id'])
return Domain('id', 'child_of', shared_roots.ids)
candidates, top_level_folders = (
query.select(*(self._field_to_sql(query.table, fname, query) for fname in ('id', 'folder_id')))
for query in (
self.with_context(active_test=False)._search([('type', '=', 'folder')]),
self.with_context(active_test=False)._search(roots_domain & Domain('type', '=', 'folder'))
)
)
children = SQL(
"""
WITH RECURSIVE
candidates as (%(candidates)s),
top_level as (%(top_level_folders)s),
children AS (
SELECT id
FROM top_level
UNION ALL
SELECT c.id
FROM candidates c
JOIN children f
ON c.folder_id = f.id
)
SELECT id FROM children
""",
candidates=candidates,
top_level_folders=top_level_folders,
)
return roots_domain | Domain('folder_id', 'any', children)
@api.model
def _search_folder_id(self, operator, operand):
if operator != 'child_of':
return Domain(Domain('folder_id', operator, operand), internal=True)
values = {operand} if isinstance(operand, int) else set(operand)
if len(values) > 1:
raise UserError(_("Only one value can be searched for child of `folder_id`."))
value = values.pop()
return self._get_child_of_domain(Domain('folder_id', '=', value) | Domain('id', '=', value), value)
@api.model
def _clean_vals_for_user_folder_id(self, vals):
"""Update vals to integrate `user_folder_id`.
This allows to
* Override context-provided values if `user_folder_id` is defined
* Handle constraints on moving only on `folder_id` and `owner_id` instead
of duplicating them for `user_folder_id`
* Centralize logic about vals vs context defaults
Note that passing any values for `folder_id` and `owner_id` in vals or context
will discard default_user_folder_id.
:param dict vals: Values for record
:raises UserError: on invalid new `user_folder_id` or conflict with `folder_id`
or `owner_id` in `vals`
"""
user_folder_id = vals.get('user_folder_id')
if not user_folder_id:
if (
(default_user_folder_id := self.env.context.get('default_user_folder_id'))
and 'folder_id' not in vals
and 'owner_id' not in vals
and 'default_folder_id' not in self.env.context
and 'default_owner_id' not in self.env.context
):
user_folder_id = default_user_folder_id
else:
return
if user_folder_id == "COMPANY":
new_vals = {'owner_id': False, 'folder_id': False}
elif user_folder_id == "MY":
if not self.env.user.active:
raise UserError(_("Inactive user cannot create/move in 'My Drive'."))
new_vals = {'owner_id': self.env.user.id, 'folder_id': False}
elif user_folder_id == "RECENT":
raise UserError(_("Documents cannot be created or moved in 'Recent'."))
elif user_folder_id == "SHARED":
raise UserError(_("Documents cannot be created or moved in 'Shared With Me'."))
elif user_folder_id == "TRASH":
raise UserError(_("Documents cannot be created or moved in the trash."))
elif user_folder_id.isnumeric():
new_vals = {"folder_id": int(user_folder_id)}
else:
raise UserError(_("Unexpected user_folder_id value %s", user_folder_id))
message = _("Conflicting values passed with user_folder_id.")
if (folder_id := vals.get('folder_id')) and folder_id != new_vals['folder_id']:
raise UserError(message)
if (owner_id := vals.get('owner_id')) and 'owner_id' in new_vals and owner_id != new_vals['owner_id']:
raise UserError(message)
vals.update(new_vals)
@api.depends('attachment_id', 'url', 'shortcut_document_id')
def _compute_name_and_preview(self):
request_session = requests.Session()
shortcuts = self.filtered('shortcut_document_id')
for record in self - shortcuts:
if record.attachment_id:
record.name = record.attachment_id.name
record.url_preview_image = False
elif record.url:
preview = link_preview.get_link_preview_from_url(record.url, request_session)
if not preview:
continue
if preview.get('og_title'):
record.name = preview['og_title']
if preview.get('og_image'):
record.url_preview_image = preview['og_image']
for shortcut in shortcuts:
shortcut.name = shortcut.name or shortcut.shortcut_document_id.name
shortcut.url_preview_image = shortcut.url_preview_image or shortcut.shortcut_document_id.url_preview_image
@api.depends_context('uid', 'allowed_company_ids')
@api.depends('access_ids', 'access_internal', 'access_via_link', 'owner_id', 'is_access_via_link_hidden',
'company_id', 'folder_id.access_ids', 'folder_id.access_internal', 'folder_id.access_via_link',
'folder_id.owner_id', 'folder_id.company_id', 'shortcut_document_id', 'shortcut_document_owner_id')
def _compute_user_permission(self):
if self.env.user.has_group('documents.group_documents_system'):
for document in self:
if (
not (company := document.company_id)
or company in self.env.companies
or company not in self.env.user.with_context(active_test=True).company_ids
):
document.user_permission = 'edit'
else:
document.user_permission = 'none'
return
permission_by_document = self._get_permission_without_token_multi()
for document in self:
if document.company_id and not document.company_id.active:
document.user_permission = 'none'
continue
document.user_permission = permission_by_document[document]
if document.user_permission == 'view' and document.access_via_link == 'edit':
document.user_permission = 'edit'
elif (
document.shortcut_document_id
and document.owner_id == self.env.user
and (document.user_permission == 'view' or document.shortcut_document_owner_id == self.env.user)
):
# Extend shortcut owner permission when target is accessible
document.user_permission = 'edit'
if document.user_permission == 'none' and document.folder_id and document.access_via_link != 'none' \
and not document.is_access_via_link_hidden \
and (document.company_id in self.env.companies or document.company_id not in self.env.user.with_context(active_test=True).company_ids):
# If the user can access the parent, they have the link.
# This only works one level up, as it mimics accessing through the interface.
with contextlib.suppress(AccessError):
if document.folder_id._get_permission_without_token() != 'none':
document.user_permission = document.access_via_link
def _get_permission_without_token(self):
self.ensure_one()
return self._get_permission_without_token_multi()[self]
def _get_permission_without_token_multi(self):
permission_by_document = {}
documents_to_process = self
for document in self:
exclude_ownership = bool(document.shortcut_document_id)
is_user_company = document.company_id and document.company_id in self.env.user.with_context(active_test=False).company_ids
is_disabled_company = is_user_company and document.company_id not in self.env.companies
if is_disabled_company:
permission_by_document[document] = 'none'
documents_to_process -= document
continue
if document.owner_id == self.env.user and not exclude_ownership:
permission_by_document[document] = 'edit'
documents_to_process -= document
continue
permission_by_document[document] = 'none'
if not documents_to_process:
return permission_by_document
# access with <documents.access>
access_by_document = self.env['documents.access']._read_group(
domain=[
('partner_id', '=', self.env.user.partner_id.id),
('document_id', 'in', documents_to_process.ids),
'|',
('expiration_date', '=', False),
('expiration_date', '>', fields.Datetime.now()),
],
groupby=['document_id'],
aggregates=['id:recordset'],
)
# `access` is a singleton, since there can be only 1 access per (document_id, partner_id)
for document, access in access_by_document:
if access:
permission_by_document[document] = access.role or document.access_via_link
# access as internal
for document in documents_to_process:
if (
not self.env.user.share
and permission_by_document[document] != "edit"
and document.access_internal != 'none'
and (not document.company_id or document.company_id in self.env.companies)
):
permission_by_document[document] = (
'edit'
if self.env.user.has_group('documents.group_documents_manager')
else document.access_internal
)
return permission_by_document
def _search_user_permission(self, operator, value, exclude_ownership=False):
if self.env.user._is_public():
return Domain.FALSE
searched_roles = {'view', 'edit', 'none'}
if operator == 'in':
searched_roles.intersection_update(value)
elif operator == 'not in':
searched_roles.difference_update(value)
else:
return NotImplemented
searched_roles.discard('none')
if not searched_roles:
return Domain.FALSE
searched_roles = list(searched_roles)
other_company = Domain('company_id', '!=', False) & Domain('company_id', 'not in', self.env.user.with_context(active_test=False).company_ids.ids)
allowed_or_no_company = Domain('company_id', 'in', [False] + self.env.companies.ids)
any_except_disabled_company = Domain.OR([
Domain('company_id', 'in', self.env.companies.ids),
Domain('company_id', 'not in', self.env.user.company_ids.ids),
Domain('company_id.active', '=', False),
])
any_except_disabled_and_archived_company = (
Domain('company_id', 'in', self.env.companies.ids)
| Domain('company_id', 'not in', self.env.user.with_context(active_test=False).company_ids.ids)
)
if self.env.user.has_group('documents.group_documents_system'):
if searched_roles == ['view']:
return Domain.FALSE # System Administrator has "edit" on all documents, so finds none with "view" only.
# System Administrator should always be able to edit documents from archived companies (even with active_test=False)
return any_except_disabled_company
# Access from membership
if searched_roles == ['view']:
access_level_domain = (
(Domain('role', '=', 'view') & Domain('document_id.access_via_link', 'in', ('none', 'view')))
| (Domain('role', '=', False) & Domain('document_id.access_via_link', '=', 'view'))
)
elif searched_roles == ['edit']:
access_level_domain = Domain('role', '=', 'edit') | Domain('document_id.access_via_link', '=', 'edit')
else:
access_level_domain = Domain('role', 'in', ('view', 'edit')) | Domain('document_id.access_via_link', '!=', 'none')
access_domain = Domain('access_ids', 'any', Domain.AND((
access_level_domain,
Domain('partner_id', '=', self.env.user.partner_id.id),
Domain('expiration_date', '=', False) | Domain('expiration_date', '>', fields.Datetime.now()),
)))
# Access from ownership
if exclude_ownership:
owner_domain = Domain.FALSE
else:
owner_domain = Domain('owner_id', '=', self.env.user.id) & Domain.OR([
[('shortcut_document_id', '=', False)],
[('shortcut_document_owner_id', '=', self.env.user.id)],
# extend permission to edit on shortcuts when otherwise viewer (synced with target)
# optimized to avoid recursive call if owner_domain is not going to be used (see below)
# or if everything we need is already in `access_domain`
self._search_user_permission('in', ['view'], exclude_ownership=True)
if set(searched_roles) == {'edit'}
else Domain.FALSE,
])
direct_domain = any_except_disabled_and_archived_company & (
access_domain if 'edit' not in searched_roles else access_domain | owner_domain
)
# Access form access_internal
if self.env.user.has_group('documents.group_documents_manager'):
if searched_roles == ['view']:
direct_domain &= Domain('access_internal', '=', 'none') | other_company
else:
direct_domain |= Domain('access_internal', 'in', ('view', 'edit')) & allowed_or_no_company
elif not self.env.user.share:
if searched_roles == ['view']:
internal_domain = Domain('access_internal', '=', 'view') & Domain('access_via_link', 'in', ('none', 'view'))
elif searched_roles == ['edit']:
internal_domain = Domain('access_internal', '=', 'edit') | (
Domain('access_internal', '=', 'view') & Domain('access_via_link', '=', 'edit')
)
else:
internal_domain = Domain('access_internal', 'in', ('view', 'edit'))
direct_domain |= internal_domain & allowed_or_no_company
if exclude_ownership:
return direct_domain
# Look one level up for links unless hidden
link_via_parent_domain = Domain.AND([
any_except_disabled_and_archived_company,
[('access_via_link', 'in', searched_roles)],
[('is_access_via_link_hidden', '=', False)],
[('folder_id', 'any', direct_domain)],
])
return direct_domain | link_via_parent_domain
@api.depends('datas', 'mimetype')
def _compute_is_multipage(self):
for document in self:
# external computation to be extended
document.is_multipage = bool(document._get_is_multipage()) # None => False
@api.depends('attachment_id', 'attachment_id.res_model', 'attachment_id.res_id',
'shortcut_document_id.res_model', 'shortcut_document_id.res_id')
def _compute_res_record(self):
for record in self:
attachment = record.attachment_id
if attachment:
record.res_model = (attachment.res_model != 'documents.document' and attachment.res_model) or False
record.res_id = (attachment.res_model != 'documents.document' and attachment.res_id) or False
if record.shortcut_document_id:
record.res_model = record.shortcut_document_id.res_model
record.res_id = record.shortcut_document_id.res_id
@api.depends('attachment_id', 'res_model', 'res_id')
def _compute_res_name(self):
for record in self:
if record.attachment_id:
record.res_name = record.attachment_id.res_name
elif record.res_id and record.res_model:
try:
record.res_name = self.env[record.res_model].browse(record.res_id).display_name
except MissingError:
record.res_name = False
else:
record.res_name = False
def _inverse_res_record(self):
for record in self:
attachment = record.attachment_id.with_context(no_document=True)
# If no linked record, link the attachment to the document
# (so users see the attachment in the technical view if they have access to the document)
res_model, res_id = record.res_model, record.res_id
if not res_model:
res_model = 'documents.document'
res_id = record.id
if attachment and (attachment.res_model, attachment.res_id) != (res_model, res_id):
# Avoid inconsistency in the data, write both at the same time.
# In case a check_access is done between res_id and res_model modification,
# an access error can be received. (Mail causes this check_access)
attachment.sudo().write({'res_model': res_model, 'res_id': res_id})
@api.depends('checksum', 'shortcut_document_id.thumbnail', 'shortcut_document_id.thumbnail_status',
'shortcut_document_id.user_permission')
def _compute_thumbnail(self):
for document in self:
if document.shortcut_document_id:
if document.shortcut_document_id:
document.thumbnail = document.shortcut_document_id.thumbnail
document.thumbnail_status = document.shortcut_document_id.thumbnail_status
else:
document.thumbnail = False
document.thumbnail_status = 'restricted'
elif document.mimetype and (
document.mimetype.startswith('application/pdf') or document.mimetype.startswith('image/webp')):
# These thumbnails are generated by the client. To force the generation, we invalidate the thumbnail.
document.thumbnail = False
document.thumbnail_status = 'client_generated'
elif document.mimetype and document.mimetype.startswith('image/'):
try:
document.thumbnail = base64.b64encode(image_process(document.raw, size=(200, 140), crop='center'))
document.thumbnail_status = 'present'
except (UserError, TypeError):
document.thumbnail = False
document.thumbnail_status = 'error'
else:
document.thumbnail = False
document.thumbnail_status = False
@api.depends('type')
def _compute_deletion_delay(self):
folders = self.filtered(lambda d: d.type == 'folder')
folders.deletion_delay = self.get_deletion_delay()
(self - folders).deletion_delay = False
def _get_folder_embedded_actions(self, folder_ids):
"""Return the enabled actions for the given folder."""
folders_sudo = self.env['documents.document'].sudo().search([
('id', 'in', folder_ids),
'|', ('user_permission', '!=', 'none'), ('children_ids', 'any', [('user_permission', '!=', 'none')])
])
if not folders_sudo:
return {}
all_embedded_actions_sudo = self.env['ir.embedded.actions'].sudo().search(
domain=Domain.AND([
self.env['ir.embedded.actions'].sudo()._get_documents_embed_base_domain(),
[('parent_res_id', 'in', (folders_sudo + folders_sudo.shortcut_document_id).ids)],
]),
order='sequence',
)
# Filtering on action_id.groups_id above is not possible because the orm "considers" action_id
# to be of the ir.actions.action model, that does not have a groups_id field.
accessible_server_actions_ids = self.env['ir.actions.server'].sudo().search(
Domain.AND([
[('id', 'in', all_embedded_actions_sudo.action_id.ids)],
self._get_embeddable_server_action_domain(),
])).ids
embedded_actions = all_embedded_actions_sudo.filtered(
lambda e: e.action_id.id in accessible_server_actions_ids).sudo(False)
# group after ordering by `ir.embedded.actions` sequence
actions_per_folder = embedded_actions.grouped('parent_res_id')
targets_to_shortcuts_sudo = folders_sudo.grouped('shortcut_document_id')
actions_per_shortcut_folder = {
shortcut_sudo.id: actions
for target_sudo, shortcuts_sudo in targets_to_shortcuts_sudo.items()
for shortcut_sudo in shortcuts_sudo
if (actions := actions_per_folder.get(target_sudo.id))
}
return actions_per_folder | actions_per_shortcut_folder
@api.depends_context('uid', 'allowed_company_ids')
@api.depends('folder_id')
def _compute_available_embedded_actions_ids(self):
embedded_actions = self._get_folder_embedded_actions(self.folder_id.ids)
embedded_actions_per_folder = {
folder_id: actions.ids
for folder_id, actions in embedded_actions.items()
}
self.available_embedded_actions_ids = False
for document in self.filtered(lambda d: d.type != 'folder' and not d.shortcut_document_id):
document.available_embedded_actions_ids = embedded_actions_per_folder.get(document.folder_id.id, False)
def _get_last_access_date_group_cte(self):
return SQL("""
WITH last_access_date AS (
SELECT (CASE
WHEN last_access_date > NOW() - INTERVAL '1 days' THEN '3_day'
WHEN last_access_date > NOW() - INTERVAL '7 days' THEN '2_week'
WHEN last_access_date > NOW() - INTERVAL '1 months' THEN '1_month'
ELSE '0_older'
END) AS date,
document_id
FROM documents_access
WHERE partner_id = %s
)
""", self.env.user.partner_id.id)
def _compute_mail_alias_domain_count(self):
self.mail_alias_domain_count = self.env['mail.alias.domain'].sudo().search_count([])
@api.depends('access_ids')
def _compute_last_access_date_group(self):
self.env.cr.execute(SQL(
"""(%s SELECT document_id, date FROM last_access_date WHERE document_id = ANY(%s))""",
self._get_last_access_date_group_cte(),
self.ids,
))
values = {line['document_id']: line['date'] for line in self.env.cr.dictfetchall()}
for document in self:
document.last_access_date_group = values.get(document.id)
def _search_last_access_date_group(self, operator, operand):
if operator != 'in':
return NotImplemented
values = set(operand)
if False in values:
query = SQL("(%s SELECT document_id FROM last_access_date)", self._get_last_access_date_group_cte())
domain = [('id', 'not in', query)]
if len(values) > 1:
values.remove(False)
domain += self._search_last_access_date_group(operator, values)
return domain
query = SQL(
"""(%s SELECT document_id FROM last_access_date WHERE date IN %s)""",
self._get_last_access_date_group_cte(), tuple(values))
return [('id', 'in', query)]
def _field_to_sql(self, alias, fname, query=None) -> SQL:
if fname == 'last_access_date_group':
# Allow to group on the field
return SQL("""(%s SELECT date FROM last_access_date WHERE document_id = %s)""",
self._get_last_access_date_group_cte(), SQL.identifier(alias, 'id'))
return super()._field_to_sql(alias, fname, query)
def _order_field_to_sql(self, alias, field_name, direction, nulls, query):
if field_name == 'last_access_date_group':
sql_field = SQL(
"SELECT last_access_date FROM documents_access WHERE partner_id = %s AND document_id = %s",
self.env.user.partner_id.id,
SQL.identifier(alias, 'id')
)
return SQL("(%s) %s %s", sql_field, direction, nulls)
if field_name == 'is_folder':
sql_field = SQL("%s != 'folder'", SQL.identifier(alias, 'type'))
return SQL("(%s) %s %s", sql_field, direction, nulls)
return super()._order_field_to_sql(alias, field_name, direction, nulls, query)
@api.model
def get_previewable_file_extensions(self):
return {'bmp', 'png', 'jpg', 'jpeg', 'pdf', 'gif', 'txt'}
def action_move_folder(self, target, before_folder_id=False):
"""Move one folder to the given position and update its sequence.
If no parent_folder is given, check whether the
parent is 'COMPANY' or 'MY'. If no before_folder is given, place it as
last child of its parent (last root if no parent is given)
:param str target: user_folder_id of the new parent folder
:param int|bool before_folder_id: id of the folder before which to move
"""
self.ensure_one()
if self.type != 'folder' or not self.active:
return
values = {'user_folder_id': target}
sibling_folders_domain = (
Domain('type', '=', 'folder')
& Domain('id', '!=', self.id)
& Domain('user_folder_id', '=', target)
)
# If before_folder is indeed a sibling given the passed target (as it could have been moved by someone else),
# assign its current sequence value to the current record and shift the following folders to keep ordering.
if before_folder := self.browse(before_folder_id):
located_after_domain = (
Domain('sequence', '>', before_folder.sequence)
| (Domain('sequence', '=', before_folder.sequence) & Domain('id', '<=', before_folder_id))
)
folders_to_resequence_domain = sibling_folders_domain & located_after_domain
folders_to_resequence_sudo = self.sudo().search(folders_to_resequence_domain)
if before_folder == folders_to_resequence_sudo[0]:
values['sequence'] = before_folder.sequence
new_sequence = before_folder.sequence + 1
for folder_sudo in folders_to_resequence_sudo:
if folder_sudo.sequence >= new_sequence:
break
folder_sudo.sequence = new_sequence
new_sequence += 1
return self.write(values)
# Otherwise, move the folder as last child of its parent
if result := self.env['documents.document'].sudo().search_read(
sibling_folders_domain, fields=['sequence'], order="sequence DESC", limit=1):
values['sequence'] = result[0]["sequence"] + 1
return self.write(values)
def action_change_owner(self, new_user_id):
if not self.env.user._is_admin() and not self.env.user.has_group('documents.group_documents_system'):
if any(document.owner_id != self.env.user for document in self):
raise AccessError(_("You are not allowed to change ownerships of documents you do not own."))
self.owner_id = new_user_id
@api.model
def _ensure_user_role_without_propagation(self, role, documents_per_user):
"""Set role membership without propagating to children."""
existing_access = self.env['documents.access'].sudo().search(Domain.OR(
[('partner_id', '=', owner.partner_id.id), ('document_id', 'in', documents.ids)]
for owner, documents in documents_per_user.items()
))
existing_access.role = role
existing_access_values = {(a.partner_id, a.document_id) for a in existing_access}
self.env['documents.access'].sudo().create([
{'partner_id': owner.partner_id.id, 'document_id': document.id, 'role': role}
for owner, documents in documents_per_user.items()
for document in documents
if (owner.partner_id, document) not in existing_access_values
])
def action_create_shortcut(self, location_user_folder_id=None):
"""Create a shortcut to self in a specific user_folder or as a sibling.
:param str | None location_user_folder_id: Optional: where to create the shortcut.
"""
if not self.ids:
return self.browse()
if len(self.folder_id.ids) > 1 and location_user_folder_id is None:
raise UserError(_("A destination is required when creating multiple shortcuts at once."))
if location_user_folder_id is False:
raise UserError(_('Ambiguous shortcut target location.'))
if location_user_folder_id is not None:
try:
location_folder_id = int(location_user_folder_id)
except ValueError: # Company, My => False
location_folder_id = False
else:
location_folder_id = None
location = self.browse(location_folder_id) if location_folder_id is not None else self.folder_id
if self.shortcut_document_id:
targets = self.filtered(lambda d: not d.shortcut_document_id) | self.shortcut_document_id
return targets.action_create_shortcut(str(location.id) if location else location_user_folder_id)
if location_folder_id and location.shortcut_document_id:
return self.action_create_shortcut(str(location.shortcut_document_id.id))
if location and location.user_permission != 'edit':
raise AccessError(_("You are not allowed to write in this folder."))
self.check_access('read')
return self.sudo().create([{
"user_folder_id": str(location.id) if location else location_user_folder_id,
"shortcut_document_id": document.id,
"access_internal": document.access_internal or 'view',
"access_via_link": document.access_via_link or 'none',
"access_ids": [
Command.create({
"partner_id": access.partner_id.id,
"role": access.role,
})
for access in document.access_ids if access.role
],
**{
field_name: (value.id if isinstance((value := document[field_name]), models.Model) else value)
for field_name in self._get_shortcuts_copy_fields()
}
} for document in self]).sudo(False)
def action_delete_from_history(self, attachment_id):
"""Delete an attachment from the document's history."""
self.ensure_one()
attachment = self.env['ir.attachment'].browse(attachment_id)
if (
attachment not in self.previous_attachment_ids
and (attachment != self.attachment_id or not self.previous_attachment_ids)
):
raise UserError(_('You cannot delete this attachment.'))
if attachment == self.attachment_id:
self.attachment_id = max(self.previous_attachment_ids, key=lambda a: (a.create_date, a.id))
attachment.unlink()
@api.model
def _get_shortcuts_copy_fields(self):
# Note that current simple usage in action_create_shortcut supports scalar and m2o fields.
return {'company_id', 'file_size', 'file_extension', 'is_access_via_link_hidden', 'is_multipage',
'name', 'partner_id', 'type', 'url', 'url_preview_image'}
def action_update_access_rights(self, access_internal=None, access_via_link=None, is_access_via_link_hidden=None,
partners=None, no_propagation=False):
"""Update access to a document and propagate if applicable.
This method can be called to update the access of internal users, with
the link, as well as a set of partners and roles to the records in self
and their children (except shortcuts), and shortcuts pointing to them
as they are kept synchronized.
Modifications to internal users and link access are propagated down to
children until the new value is already present. Note that changes to
the discoverability(`is_access_via_link_hidden`) are never propagated.
For partners, all changes are applied to all children regardless of the
existing rights structure.
:param str | None access_internal: optional new permission level for internal users
:param str | None access_via_link: optional new permission level for partners with the link
:param bool|None is_access_via_link_hidden: optional new value for discoverability
:param dict[str | int | res.partner(), tuple[str | bool | None, str | datetime | bool | None] partners:
Mapping of partner(_id) to the tuple:
role: 'edit', 'view', False (=>delete),
expiration: datetime string, False (removed/None)
:param bool no_propagation: whether to propagate rights to sub-folders
"""
if len(self.ids) == 0:
return
try:
self.check_access('write')
except UserError:
raise AccessError(self.env._("You are not allowed to update these access rights."))
if self.shortcut_document_id:
raise UserError(_("You can not update the access of a shortcut, update its target instead."))
# Check inputs as we are going to bypass the ORM in the private method(s)
access_options = {'view', 'edit', 'none', None}
hidden_options = {None, True, False}
role_options = {'edit', 'view', False, None}
incorrect_fields_to_options = {
**({'is_access_via_link_hidden': hidden_options} if is_access_via_link_hidden not in hidden_options else {}),
**({'access_via_link': access_options} if access_via_link not in access_options else {}),
**({'access_internal': access_options} if access_internal not in access_options else {}),
**({'partners.role': role_options}
if any(role not in role_options for (__, (role, __)) in (partners or {}).items()) else {})
}
if incorrect_fields_to_options:
hints = "\n- " + "\n- ".join(f'{name}: {options}' for name, options in incorrect_fields_to_options.items())
raise UserError(_(
"Incorrect values. Use one of the following for the following fields: %(hints)s.)", hints=hints
))
changes_by_document_dict = self._action_update_access(
access_internal, access_via_link, is_access_via_link_hidden, no_propagation=no_propagation)
if partners:
partners = {
self.env['res.partner'].browse(int(partner)) if isinstance(partner, str | int) else partner:
(role, fields.Datetime.to_datetime(exp) if exp and isinstance(exp, str) else exp)
for partner, (role, exp) in (partners or {}).items()
}
created_or_updated_access, removed_access = self._action_update_members(partners, no_propagation=no_propagation)
self._update_changes_by_document_dict(created_or_updated_access, removed_access, changes_by_document_dict)
self.env['documents.access.tracking']._create_access_tracking(changes_by_document_dict)
return self.mapped('user_permission')
def _action_update_access(self, access_internal, access_via_link, is_access_via_link_hidden,
no_propagation=False):
"""Update the access on self and children.
Stop the propagation when the value is already the right one.
:param str | None access_internal: change the `access_internal` if not None
:param str | None access_via_link: change the `access_via_link` if not None
:param bool | None is_access_via_link_hidden: change the `is_access_via_link_hidden` if not None
:param bool no_propagation: whether to propagate access update to sub-folders
"""
self.flush_model()
changes_by_document_dict = defaultdict(dict)
for field, value in (
('access_internal', access_internal),
('access_via_link', access_via_link),
('is_access_via_link_hidden', is_access_via_link_hidden),
):
if value is None:
continue
# never propagate discoverability
skip_propagation = no_propagation or field == 'is_access_via_link_hidden'
# records that we might need to update
candidates_domain = Domain([
(field, '!=', value),
# the update is done only "target -> shortcut",
# but not "shortcut -> target"
('shortcut_document_id', '=', False),
('id', 'in' if skip_propagation else 'child_of', self.ids),
])
candidates_domain &= self._get_access_update_domain()
candidates_query = self.with_context(active_test=False)._search(candidates_domain)
candidates = candidates_query.select(*(
self._field_to_sql(candidates_query.table, fname, candidates_query)
for fname in ('id', 'folder_id', 'shortcut_document_id', field)
))
self.env.cr.execute(SQL("""
WITH RECURSIVE candidates AS (%(candidates)s),
-- explore the folders
documents_to_update AS (
SELECT id, %(field)s
FROM candidates
WHERE id = ANY(%(root_ids)s)
UNION
SELECT child.id, child.%(field)s
FROM candidates AS child
JOIN documents_to_update AS parent
ON child.folder_id = parent.id
),
documents_and_shortcuts AS (
SELECT id, %(field)s FROM documents_to_update
UNION
-- document.shortcut_ids
-- update in "SUDO" to keep them synchronized
SELECT shortcut.id, shortcut.%(field)s
FROM documents_document AS shortcut
JOIN documents_to_update
ON documents_to_update.id = shortcut.shortcut_document_id
)
UPDATE documents_document
SET %(field)s = %(value)s
FROM documents_and_shortcuts AS doc
-- document | document.children_ids | document.shortcut_ids
WHERE documents_document.id = doc.id
RETURNING doc.id, doc.%(field)s
""", field=SQL(field), value=value, root_ids=self.ids, candidates=candidates))
for id, old_value in self.env.cr.fetchall():
changes_by_document_dict[id][field] = old_value
self.invalidate_model([
'access_internal',
'access_via_link',
'is_access_via_link_hidden',
'user_permission',
])
return changes_by_document_dict
def _action_update_members(self, partners, no_propagation=False):
"""Update the members access on all files bellow the current folder.
:param partners: Partners to add as members / change access
:param bool no_propagation: whether to propagate members update to sub-folders
"""
self.env['documents.access'].flush_model()
partners_to_remove = self.env['res.partner']
# {(role, expiration_date): partners}
values_to_update = defaultdict(lambda: self.env['res.partner'])
for partner, (role, expiration_date) in partners.items():
if role is False:
# remove the members
partners_to_remove |= partner
elif role is not None or expiration_date is not None:
values_to_update[role, expiration_date] |= partner
# use `_search` to respect access rules and to use `_search_user_permission`
to_update_domain = Domain([
('shortcut_document_id', '=', False), # update "target -> shortcuts" but not "shortcut -> target"
('id', 'in' if no_propagation else 'child_of', self.ids),
])
to_update_domain &= self._get_access_update_domain()
documents = self.with_context(active_test=False)._search(to_update_domain).select()
created_or_updated_access = []
for (role, expiration_date), partners in values_to_update.items():
if role not in ('edit', 'view'):
raise UserError(_("Invalid role.")) # The public method would have returned a more insightful message
update_fields = [SQL('role = %(role)s', role=role)]
if expiration_date is not None:
update_fields.append(SQL(
'expiration_date = %(expiration_date)s',
expiration_date=expiration_date or None,
))
update_fields = SQL(',').join(update_fields)
self.env.cr.execute(SQL(
"""
WITH documents AS (%(documents)s),
documents_and_shortcuts AS (
SELECT * FROM documents
UNION
-- document.shortcut_ids
SELECT shortcut.id
FROM documents_document AS shortcut
JOIN documents AS document
ON document.id = shortcut.shortcut_document_id
),
existing AS (
SELECT document_id, partner_id, role, expiration_date
FROM documents_access
JOIN documents_and_shortcuts
ON document_id = documents_and_shortcuts.id
AND partner_id = any(%(partner_ids)s)
),
updated_or_created AS (
INSERT INTO documents_access (
document_id,
partner_id,
role,
expiration_date
) (
SELECT DISTINCT ON (doc.id, partner_id) doc.id,
partner_id,
%(role)s,
%(expiration_date)s
FROM documents_and_shortcuts AS doc
JOIN LATERAL UNNEST(%(partner_ids)s) AS partner_id ON TRUE
)
ON CONFLICT (document_id, partner_id) DO UPDATE SET %(update_fields)s
RETURNING document_id, partner_id, role, expiration_date
)
SELECT 'existing' as action, * FROM existing
UNION ALL
SELECT 'upsert' as action, * FROM updated_or_created
ORDER BY action ASC
""",
documents=documents,
partner_ids=partners.ids,
expiration_date=expiration_date or None,
role=role,
update_fields=update_fields,
))
created_or_updated_access += self.env.cr.fetchall()
removed_access = []
if partners_to_remove:
self.env.cr.execute(SQL("""
WITH documents AS (%(documents)s),
docs_and_shortcuts AS (
SELECT id
FROM documents
UNION
SELECT shortcut.id
FROM documents AS doc
JOIN documents_document AS shortcut
ON doc.id = shortcut.shortcut_document_id
)
DELETE FROM documents_access AS access
USING docs_and_shortcuts AS doc
WHERE access.document_id = doc.id
AND access.partner_id = ANY(%(partner_ids)s)
RETURNING access.document_id, access.partner_id
""", documents=documents, partner_ids=partners_to_remove.ids))
removed_access = self.env.cr.fetchall()
self.env['documents.document'].invalidate_model([
'access_ids',
'user_permission',
])
self.env['documents.access'].invalidate_model()
return created_or_updated_access, removed_access
@api.model
def _update_changes_by_document_dict(
self, created_or_updated_access, removed_access, changes_by_document_dict):
old_values = defaultdict(dict)
for action, doc, partner, role, exp in created_or_updated_access:
exp = fields.Date.to_string(exp) or 'None'
partner_dict = (changes_by_document_dict.setdefault(doc, {})
.setdefault('members', {'added': {}, 'updated': {}, 'removed': []}))
if action == 'upsert':
if old := old_values[doc].get(partner):
partner_dict['updated'][partner] = {
'role': (old['role'], role),
'expiration_date': (old['expiration_date'], exp),
}
else:
partner_dict['added'][partner] = {
'role': role,
'expiration_date': exp,
}
elif action == 'existing':
old_values[doc][partner] = {
'role': role,
'expiration_date': exp,
}
for doc, partner in removed_access:
(changes_by_document_dict.setdefault(doc, {})
.setdefault('members', {'added': {}, 'updated': {}, 'removed': []})['removed'].append(partner))
def _update_company(self, company_id):
"""Apply company to documents and children, without stopping (see _action_update_members).
:param int|bool company_id: Id to set or False
"""
self.flush_model()
to_update_domain = Domain.AND((
Domain('id', 'in', self.ids) | Domain('company_id', '!=', company_id),
# the update is done only "target -> shortcut",
# but not "shortcut -> target"
[('shortcut_document_id', '=', False)],
[('id', 'child_of', self.ids)],
[] if self.env.su else [('user_permission', '=', 'edit')],
))
to_update = self.with_context(active_test=False)._search(to_update_domain).select()
# update shortcuts in sudo to keep them synchronized
shortcuts_union = SQL("""
UNION
SELECT shortcut.id
FROM documents_document AS shortcut
JOIN documents_to_update
ON documents_to_update.id = shortcut.shortcut_document_id
""" if company_id else "")
self.env.cr.execute(SQL("""
WITH documents_to_update AS (%(to_update)s),
documents_and_shortcuts AS (
SELECT id FROM documents_to_update
%(shortcuts_union)s
)
UPDATE documents_document
SET %(field)s = %(value)s
FROM documents_and_shortcuts AS doc
WHERE documents_document.id = doc.id
""", field=SQL('company_id'), value=company_id or None, to_update=to_update, shortcuts_union=shortcuts_union))
self.invalidate_model(['company_id', 'user_permission'])
def _get_access_update_domain(self):
return Domain.TRUE if self.env.su else Domain('user_permission', '=', 'edit')
def get_formview_action(self, access_uid=None):
"""Return the `document_action_preference` action as the default view for M2O fields."""
if not self.has_access("read"):
return {
"type": "ir.actions.client",
"tag": "display_notification",
"params": {
"message": self.env._("Document not found or inaccessible."),
"type": "danger",
},
}
default_user_folder_id = str(self.id) if self.type == "folder" and self.active else self.user_folder_id
action = self.env["ir.actions.actions"]._for_xml_id("documents.document_action_preference")
action["context"] = {
**self.env.context,
"documents_show_default_breadcrumb": True,
"no_documents_unique_folder_id": True,
"searchpanel_default_user_folder_id": default_user_folder_id,
"documents_init_document_id": self.id if self.type != "folder" or not self.active else False,
}
return action
@api.model
def get_documents_actions(self, folder_id):
"""Return the available actions and a key to know if the action is embedded on the folder."""
if not isinstance(folder_id, int):
raise ValueError("Invalid folder_id")
folder = self.env['documents.document'].search([('id', '=', folder_id)])
if not folder:
raise UserError(_('This folder does not exist or is not accessible.'))
embedded_actions = self._get_folder_embedded_actions(folder.ids)
embedded_actions = embedded_actions[folder.id].action_id.ids if embedded_actions else []
actions = self.env['ir.actions.server'].sudo().search(self._get_embeddable_server_action_domain())
return [{
"id": action.id,
"name": action.display_name,
"is_embedded": action.id in embedded_actions
} for action in actions]
@api.model
def _get_embeddable_server_action_domain(self):
"""Wrap `_get_base_server_actions_domain`'s domain to exclude children and actions with invalid children."""
candidate_actions_sudo = self.env["ir.actions.server"].sudo()._search(
Domain.AND([
self._get_base_server_actions_domain(),
Domain.OR([[('group_ids', 'any', [('id', 'in', self.env.user.all_group_ids.ids)])],
[('group_ids', '=', False)]]),
]),
)
return Domain.AND([
[('id', 'in', candidate_actions_sudo)],
[('parent_id', '=', False)], # no child action
[('child_ids', 'not any', [('id', 'not in', candidate_actions_sudo)])], # no invalid child
])
@api.model
def _get_base_server_actions_domain(self):
"""Return the base domain for actions applicable to documents in the current context.
!Meant to be wrapped by _get_embeddable_server_action_domain. Override to add validity conditions.
"""
return Domain.AND([
[('model_id', '=', self.env['ir.model']._get_id('documents.document'))],
[('usage', 'in', ('ir_actions_server', 'documents_embedded'))],
])
@api.model
def action_folder_embed_action(self, folder_id, action_id):
"""Enable / disable the action for the given folder
:param int folder_id: The folder on which we pin the actions
:param int action_id: The id of the action to enable
"""
if not self.env.user.has_group('documents.group_documents_user') and not self.env.su:
raise AccessError(_("You are not allowed to pin/unpin embedded Actions."))
server_actions_groups_domain = [
'|', ('group_ids', 'any', [('id', 'in', self.env.user.all_group_ids.ids)]),
('group_ids', '=', False),
]
action = self.env['ir.actions.server'].sudo().search([('id', '=', action_id), *server_actions_groups_domain])
if not action:
raise UserError(_('This action does not exist.'))
if action.type != 'ir.actions.server':
raise UserError(_('You cannot pin that type of action.'))
folder = self.env['documents.document'].browse(folder_id).sudo().exists()
if not folder or folder.type != 'folder':
raise UserError(_('You cannot pin an action on that document.'))
if folder.shortcut_document_id:
return self.action_folder_embed_action(folder.shortcut_document_id.id, action_id)
all_embedded_actions_sudo = self.env['ir.embedded.actions'].sudo().search(
Domain.AND([
self.env['ir.embedded.actions'].sudo()._get_documents_embed_base_domain(),
[('action_id', '=', action_id), ('parent_res_id', '=', folder_id)],
])
)
# See _get_folder_embedded_actions
accessible_server_action_ids = self.env['ir.actions.server'].sudo().search([
('id', 'in', all_embedded_actions_sudo.action_id.ids),
*server_actions_groups_domain
]).ids
embedded_actions_sudo = all_embedded_actions_sudo.filtered(
lambda e: e.action_id.id in accessible_server_action_ids)
if embedded_actions_sudo:
embedded_actions_sudo.unlink()
else:
# first pinned action should be displayed first
last_action = self.env['ir.embedded.actions'].search(
[], order='sequence DESC', limit=1)
self.env['ir.embedded.actions'].create({
'name': action.name,
'parent_action_id': self.env.ref("documents.document_action").id,
'action_id': action.id,
'parent_res_model': 'documents.document',
'parent_res_id': folder_id,
'groups_ids': self.env.ref('base.group_user').ids,
'sequence': last_action.sequence + 1 if last_action else 1,
})
return self.get_documents_actions(folder_id)
@api.model
def action_execute_embedded_action(self, action_id):
"""Execute an embedded action on context records.
:param int action_id: id of embedded action to be run on context provided records.
"""
if self.env.user.share:
raise AccessError(_("You are not allowed to execute embedded actions."))
if self.env.context.get('active_model') != 'documents.document':
raise UserError(_("Unavailable action."))
ids = self.env.context.get('active_ids', [self.env.context['active_id']] if self.env.context.get('active_id') else [])
if not ids:
raise UserError(_("Missing documents reference."))
embedded_action = self.env['ir.embedded.actions'].browse([action_id])
if all(action_id in document.available_embedded_actions_ids.ids for document in self.browse(ids)):
return self.env['ir.actions.server'].with_context(documents_active_ids=ids).browse(embedded_action.action_id.id).run()
raise UserError(_("Unavailable action."))
def _embed_action(self, action_id):
"""Embed a server action on the current folder(s) if not already done."""
IrEmbeddedActions = self.env['ir.embedded.actions']
embedded_actions = self._get_folder_embedded_actions(self.ids)
new_embedding_folders = self.env['documents.document']
for folder in self:
if action_id not in embedded_actions.get(folder.id, IrEmbeddedActions).action_id.ids:
folder.action_folder_embed_action(folder.id, action_id)
new_embedding_folders |= folder
return new_embedding_folders
def action_link_to_record(self, model=False):
"""Open the `link_to_record_wizard` to choose a record to link to the current documents.
This method can be used inside server actions.
"""
context = {
'default_document_ids': self.ids,
'default_resource_ref': False,
'default_is_readonly_model': False,
'default_model_ref': False,
}
if documents_link_record := self.filtered('res_model'):
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'type': 'warning',
'message': _(
"Already linked Documents: %s",
", ".join(documents_link_record.mapped('name'))
),
}
}
if model:
self.env[model].check_access('write')
context['default_is_readonly_model'] = True
context['default_model_id'] = self.env['ir.model']._get_id(model)
first_valid_id = self.env[model].search([], limit=1).id
if not first_valid_id:
raise UserError(_("There are no records to link this document. Create one first."))
context['default_resource_ref'] = f'{model},{first_valid_id}'
return {
'name': _('Choose a record to link'),
'type': 'ir.actions.act_window',
'res_model': 'documents.link_to_record_wizard',
'view_mode': 'form',
'target': 'new',
'views': [(False, "form")],
'context': context,
}
def _notify_get_recipients_groups(self, message, model_description, msg_vals=False):
groups = super()._notify_get_recipients_groups(
message, model_description, msg_vals=msg_vals
)
if len(self.ids) != 1:
return groups
group_values = {
'active': True,
'button_access': {'url': self.access_url},
'has_button_access': True,
}
return [
('group_documents_document_people_with_access',
lambda pdata:
(pdata['uid'] and self.with_user(pdata['uid']).user_permission != 'none') or
(pdata['id'] and self.access_via_link != 'none'
and self.access_ids.filtered(lambda a: a.partner_id.id == pdata['id'] and a.role)),
group_values)
] + groups
def get_deletion_delay(self):
return int(self.env['ir.config_parameter'].sudo().get_param('documents.deletion_delay', '30'))
def _get_is_multipage(self):
"""Whether the document can be considered multipage, if able to determine.
:return: `None` if mimetype not handled or url-type attachment, `False` if single page or error occurred, `True` otherwise.
:rtype: bool | None
"""
if self.mimetype not in ('application/pdf', 'application/pdf;base64') or self.attachment_type == 'url':
return None
decoded = base64.b64decode(self.datas)
# Avoid warning in tests due to IrActionsReport._pre_render_qweb_pdf rendering pdf as html
# It is done before even reading the PDF as PdfFileReader emit warning in that case
if modules.module.current_test and b'<!DOCTYPE html>' in decoded[:32]:
_logger.info("Skip _get_is_multipage of %r: html content detected in pdf document while in testing mode",
self.name)
return False
stream = io.BytesIO(decoded)
try:
return PdfFileReader(stream, strict=False).numPages > 1
except AttributeError:
raise # If PyPDF's API changes and the `numPages` property isn't there anymore, not if its computation fails.
except Exception: # noqa: BLE001
message = ('Impossible to count pages in %r. It could be due to a malformed document or a '
'(possibly known) issue within PyPDF2.')
_logger.warning(message, self.name, exc_info=True)
return False
@api.depends('active', 'user_permission', 'folder_id.user_permission', 'owner_id', 'user_folder_id')
@api.depends_context('uid')
def _compute_user_can_move(self):
"""Compute whether the env user can move the document.
A document can be moved if it is not in the trash and can be deleted, i.e.,
* if it is owned, or
* if the user can edit it AND its containing folder AND it is not protected (see _is_company_root_folder)
"""
active_documents = self.filtered('active')
(self - active_documents).user_can_move = False
user_can_unlink_map = active_documents._get_user_can_unlink()
for document in active_documents:
document.user_can_move = user_can_unlink_map[document]
def _get_user_can_unlink(self):
if self.env.is_admin() or self.env.user.has_group('documents.group_documents_system'):
return {document: True for document in self}
is_manager = self.env.user.has_group('documents.group_documents_manager')
return {
document: (
document.owner_id == self.env.user
or (
document.user_permission == 'edit'
and (not document.folder_id or document.folder_id.user_permission == 'edit')
and (is_manager or document.user_folder_id != 'COMPANY' or document.type != 'folder')
)
) for document in self
}
@api.depends('favorited_ids')
@api.depends_context('uid')
def _compute_is_favorited(self):
favorited = self._filtered_access('read').filtered(lambda d: self.env.user in d.favorited_ids)
favorited.is_favorited = True
(self - favorited).is_favorited = False
def _inverse_is_favorited(self):
unfavorited_documents = favorited_documents = self.env['documents.document'].sudo()
for document in self:
if self.env.user in document.favorited_ids:
unfavorited_documents |= document
else:
favorited_documents |= document
favorited_documents.write({'favorited_ids': [(4, self.env.uid)]})
unfavorited_documents.write({'favorited_ids': [(3, self.env.uid)]})
@api.depends('res_model')
def _compute_res_model_name(self):
for record in self:
if record.res_model:
record.res_model_name = self.env['ir.model']._get(record.res_model).display_name
else:
record.res_model_name = False
@api.constrains('url')
def _check_url(self):
for document in self.filtered("url"):
if not document.url.startswith(('https://', 'http://', 'ftp://')):
raise ValidationError(_('URL %s does not seem complete, as it does not begin with http(s):// or ftp://', document.url))
@api.model
def message_new(self, msg_dict, custom_values=None):
# When an email comes, create a document with the default values,
# then let `_message_post_after_hook` create one document per attachment.
custom_values = custom_values or {}
folder = self.env['documents.document'].browse(custom_values.get('folder_id'))
custom_values['name'] = _('Mail: %s', msg_dict.get('subject'))
if 'company_id' not in custom_values:
custom_values['company_id'] = folder.company_id.id
if 'tag_ids' not in custom_values:
custom_values['tag_ids'] = folder.alias_tag_ids.ids
else:
tags = custom_values['tag_ids']
if tags and isinstance(tags[0], list | tuple):
# we have a list of m2m commands
if all(len(t) >= 2 and t[0] == Command.LINK for t in tags):
tags = [t[1] for t in tags]
elif len(tags) == 1 and len(tags[0]) == 3 and tags[0][0] == Command.SET:
tags = tags[0][2]
else: # do not support other commands
tags = []
custom_values['tag_ids'] = self.env['documents.tag'].browse(tags).exists().ids
custom_values['active'] = False
return super().message_new(msg_dict, custom_values).with_context(document_message_new=True)
def _alias_get_creation_values(self):
values = super()._alias_get_creation_values()
values['alias_model_id'] = self.env['ir.model']._get('documents.document').id
if self.id:
values['alias_defaults'] = literal_eval(self.alias_defaults or "{}")
values['alias_defaults'] |= {'folder_id': self.id}
return values
def message_post(self, *, message_type='notification', **kwargs):
""" Prevent document creation when posting message with attachment on a document (_create_attachments_for_post).
If new documents must be created (ex.: alias on document folder), it will be handled by the
_message_post_after_hook based on the context variable "document_message_new" (ignoring no_document)
That variable is set to True by the message_new method (and not set by message_update method).
"""
return super(DocumentsDocument, self.with_context(no_document=True)).message_post(
message_type=message_type, **kwargs)
def _message_post_after_hook(self, message, msg_vals):
# If the res model was an attachment and a mail, adds all the custom values of the linked
# document settings to the attachments of the mail. If it was only a new email converts
# its body to an attachment for the given document (use case: invoice/receipt sent as an email)
if message.message_type != 'email' or not self.env.context.get("document_message_new"):
return super()._message_post_after_hook(message, msg_vals)
m2m_commands = msg_vals['attachment_ids']
attachments = self.env['ir.attachment'].browse([x[1] for x in m2m_commands])
disable_mail_to_document = literal_eval(self.env['ir.config_parameter'].get_param('documents.disable_mail_to_document', default="0"))
documents = None
if attachments:
self.attachment_id = False
documents = self.env['documents.document'].create([{
**self._message_post_after_hook_template_values(),
'name': attachment.name,
'attachment_id': attachment.id,
'company_id': self.folder_id.company_id.id,
} for attachment in attachments])
for attachment, document in zip(attachments, documents):
attachment.write({
'res_model': 'documents.document',
'res_id': document.id,
})
sub_message_values = {
'author_id': msg_vals.get('author_id'),
'body': msg_vals.get('body', ''),
'email_from': msg_vals.get('email_from'),
'message_type': 'email',
'subject': msg_vals.get('subject') or self.name,
'subtype_id': msg_vals.get('subtype_id'),
'subtype_xmlid': msg_vals.get('subtype_xmlid'),
}
sub_message_values.pop('model', None)
sub_message_values.pop('res_id', None)
sub_message_values.pop('attachment_ids', None)
document.message_post(**sub_message_values)
elif not self.attachment_id and not disable_mail_to_document:
attachment = self.env['ir.attachment'].create({
'name': msg_vals.get('subject') or msg_vals.get('email_from', _('email')),
'type': 'binary',
'raw': message.body,
'mimetype': 'application/documents-email', # Custom mimetype. Only for preview in Documents
'res_model': 'documents.document',
})
document = self.env['documents.document'].create({
**self._message_post_after_hook_template_values(),
'attachment_id': attachment.id,
})
message.res_id = document.id
attachment.res_id = document.id
documents = document
# Activity settings set through alias_defaults values has precedence over the activity folder settings
if documents:
for document in documents:
if self.create_activity_option:
document.documents_set_activity(settings_record=self)
elif self.folder_id.create_activity_option:
document.documents_set_activity(settings_record=self.folder_id)
return super()._message_post_after_hook(message, msg_vals)
def _message_post_after_hook_template_values(self):
"""Values that will be taken from the document template."""
return {
'folder_id': self.folder_id.id,
'owner_id': self.folder_id.owner_id.id,
'partner_id': self.partner_id.id,
'tag_ids': self.tag_ids.ids,
}
def documents_set_activity(self, settings_record=None):
"""
Generate an activity based on the fields of settings_record.
:param settings_record: the record that contains the activity fields.
settings_record.create_activity_type_id (required)
settings_record.create_activity_summary
settings_record.create_activity_note
settings_record.create_activity_date_deadline_range
settings_record.create_activity_date_deadline_range_type
settings_record.create_activity_user_id
"""
if settings_record and settings_record.create_activity_type_id:
for record in self:
activity_vals = {
'activity_type_id': settings_record.create_activity_type_id.id,
'summary': settings_record.create_activity_summary or '',
'note': settings_record.create_activity_note or '',
}
if settings_record.create_activity_date_deadline_range > 0:
activity_vals['date_deadline'] = fields.Date.context_today(settings_record) + relativedelta(
**{
settings_record.create_activity_date_deadline_range_type: settings_record.create_activity_date_deadline_range})
if settings_record._fields.get(
'create_has_owner_activity') and settings_record.create_has_owner_activity and record.owner_id:
user = record.owner_id
elif settings_record._fields.get('create_activity_user_id') and settings_record.create_activity_user_id:
user = settings_record.create_activity_user_id
elif settings_record._fields.get('user_id') and settings_record.user_id:
user = settings_record.user_id
else:
user = self.env.user
if user:
activity_vals['user_id'] = user.id
record.activity_schedule(**activity_vals)
def copy_data(self, default=None):
default = dict(default or {})
if 'user_folder_id' in default:
self._clean_vals_for_user_folder_id(default)
vals_list = super().copy_data(default=default)
if 'name' not in default:
for document, vals in zip(self, vals_list):
vals['name'] = document.name if self.env.context.get('documents_copy_skip_rename') else _("%s (copy)", document.name)
for vals in vals_list:
# Avoid to propagate folder access as we want to copy the document accesses alone
vals['access_ids'] = default.get('access_ids', False)
if 'owner_id' not in vals:
vals['owner_id'] = self.env.user.id
return vals_list
def copy(self, default=None):
if not self:
return self
if not all(self.mapped('active')):
raise UserError(_('You cannot duplicate document(s) in the Trash.'))
if default and default.get('user_folder_id') == 'MY':
default['owner_id'] = self.env.user.id
# As we avoid to propagate the folder permission by setting access_ids to False (see copy_data), user has no
# right to create the document. So after checking permission, we execute the copy in sudo.
self.env['documents.document'].check_access('create')
self.check_access('read')
documents_order = {doc.id: idx for idx, doc in enumerate(self)}
new_documents = [self.browse()] * len(self)
is_manager = self.env.is_admin() or self.env.user.has_group('documents.group_documents_manager')
skip_documents = self.env.context.get('documents_copy_folders_only')
shortcuts = self.filtered('shortcut_document_id')
if shortcuts and not skip_documents:
for destination, targets in self._get_copy_shortcuts_destinations(shortcuts, default):
new_shortcuts = targets.action_create_shortcut(location_user_folder_id=destination)
for new_shortcut, target in zip(new_shortcuts, targets):
new_shortcut.name = _("%s (copy)", target.name)
new_documents[documents_order[target.id]] = new_shortcut
folders = (self - shortcuts).filtered(lambda d: d.type == 'folder')
if folders:
if not is_manager and default and default.get('user_folder_id') == 'COMPANY':
raise AccessError(_('Only Documents Managers can create in company folder.'))
embedded_actions = self._get_folder_embedded_actions(folders.ids)
new_folders = folders.sudo()._copy_with_access(default=default).sudo(False)
# move in "My Drive" if needed
self.browse([
new_folder.id
for old_folder, new_folder in zip(folders, new_folders)
if old_folder._cannot_create_sibling()
]).sudo().write({'folder_id': False})
for old_folder, new_folder in zip(folders, new_folders):
if folder_embedded_actions := embedded_actions.get(old_folder.id):
embedded_actions_copies = folder_embedded_actions.copy()
embedded_actions_copies.parent_res_id = new_folder.id
# no need to check for permission as all the checks have been done
children_default = {'folder_id': new_folder.id}
owner_id_in_default = (default or {}).get('owner_id') is not None
if owner_id_in_default:
children_default.update(owner_id=default['owner_id'])
# check if we are not copying a folder into itself or one of its descendants
if new_folder.parent_path.startswith(old_folder.parent_path):
raise UserError(_("You cannot copy a folder into itself or into one of its own descendants."))
old_folder.children_ids.with_context(documents_copy_skip_rename=True).copy(children_default)
new_documents[documents_order[old_folder.id]] = new_folder
if is_manager and old_folder._is_company_root_folder() and not owner_id_in_default:
new_folder.owner_id = old_folder.owner_id
if not skip_documents and (documents_sudo := (self - shortcuts - folders).sudo()):
new_binaries_sudo = documents_sudo._copy_with_access(default=default)
for old_document_sudo, new_binary_sudo in zip(documents_sudo, new_binaries_sudo):
new_documents[documents_order[old_document_sudo.id]] = new_binary_sudo.sudo(False)
if (
is_manager
and 'owner_id' not in (default or {})
and (not old_document_sudo.owner_id and not old_document_sudo.folder_id) # company root
):
new_binary_sudo.owner_id = False
# move in "My Drive" if needed
self.browse([
new_binary_sudo.id for new_binary_sudo in new_binaries_sudo
if new_binary_sudo.sudo(self.env.su)._cannot_create_sibling()
]).sudo().write({'folder_id': False})
if to_copy_attachment_sudo := documents_sudo._copy_attachment_filter(default):
new_attachments_iterator = iter(to_copy_attachment_sudo.attachment_id.with_context(no_document=True).copy())
# Avoid recompute based on attachment_id
with self.env.protecting(self._get_fields_to_recompute(depends=['attachment_id']), new_binaries_sudo):
for old_document_sudo, new_binary_sudo in zip(documents_sudo, new_binaries_sudo):
if old_document_sudo._copy_attachment_filter(default):
new_attachment = next(new_attachments_iterator)
new_binary_sudo.write({
'attachment_id': new_attachment.id,
'res_id': False,
'res_model': False,
})
return self.browse([new_document.id for new_document in new_documents])
def _cannot_create_sibling(self):
"""Return whether the user is not allowed to create in the same folder, used for copy."""
self.ensure_one()
if self.env.su:
return False
if self.folder_id:
# do not check edit access rule, to allow copying in root company folders
return self.folder_id.user_permission != 'edit'
return (
# allow the manager to copy root folder without moving them to his drive
not self.env.user.has_group('documents.group_documents_manager')
# anyone can copy in one's drive
and self.owner_id != self.env.user
)
def _copy_attachment_filter(self, default):
if default and 'attachment_id' in default:
return self.env['documents.document']
return self.filtered('attachment_id')
def _get_copy_shortcuts_destinations(self, shortcuts, default):
"""Integrate copy `default` and access rights to return valid destinations for shortcuts to copy."""
default = default or {}
folder_id = default.get('folder_id')
user_folder_id = default.get('user_folder_id')
prefetch_ids = None
candidates = {}
if user_folder_id:
if user_folder_id.isnumeric():
candidates[self.browse(int(user_folder_id))] = shortcuts
else:
return ((user_folder_id, shortcuts),)
elif folder_id is not None:
candidates[self.browse(folder_id)] = shortcuts
else:
candidates = shortcuts.grouped('folder_id')
prefetch_ids = shortcuts.folder_id.ids
targets_per_destination = defaultdict(self.browse)
for destination, destination_shortcuts in candidates.items():
if isinstance(destination, str):
pass
elif not self.env.su and destination and destination.with_prefetch(prefetch_ids).user_permission != 'edit':
destination = 'MY'
else:
destination = str(destination.id)
targets_per_destination[destination] |= destination_shortcuts
return targets_per_destination.items()
@api.model
def _get_fields_to_recompute(self, depends):
"""
Get copyable `compute stored` fields that need recomputation
based on the provided dependencies.
"""
if not depends:
return []
fields_to_recompute = set()
fields_compute_stored = {
field
for field in self._fields.values()
if field.copy and field.store and field.compute
}
for field_dependence in (self._fields[depend] for depend in depends):
fields_dependent = set(self.pool.get_dependent_fields(field_dependence))
fields_to_recompute |= fields_compute_stored & fields_dependent
return fields_to_recompute
def _copy_with_access(self, default):
"""Copy documents with their access. !Assumes that access rights were checked before! """
if not self:
return self
res = super().copy(default=default)
if default and 'access_ids' in default:
return res
access_vals_list = []
for doc, doc_copied in zip(self, res):
owner_partner = doc_copied.owner_id.partner_id # already done at doc_copied creation
doc_access_to_have = doc.access_ids.filtered('role')
doc_access_to_create = doc_access_to_have.filtered(
lambda a: a.partner_id not in doc_copied.access_ids.partner_id | owner_partner)
access_vals_list += doc_access_to_create.copy_data(default={'document_id': doc_copied.id})
self.env['documents.access'].sudo().create(access_vals_list)
return res
def toggle_favorited(self):
self.ensure_one()
self.sudo().is_favorited = not self.is_favorited
return self.is_favorited
def toggle_favorited_multi(self): # TODO remove in master and directly modify toggle_favorited
for record in self:
record.toggle_favorited()
def access_content(self):
self.ensure_one()
action = {
'type': "ir.actions.act_url",
'target': "new",
}
if self.url:
action['url'] = self.url
elif self.type == 'binary':
action['url'] = f'/documents/content/{quote(self.access_token, safe="")}'
return action
def open_resource(self):
self.ensure_one()
if self.res_model and self.res_id:
view_id = self.env[self.res_model].get_formview_id(self.res_id)
return {
'res_id': self.res_id,
'res_model': self.res_model,
'type': 'ir.actions.act_window',
'views': [[view_id, 'form']],
}
def toggle_lock(self):
"""
sets a lock user, the lock user is the user who locks a file for themselves, preventing data replacement
and archive (therefore deletion) for any user but himself.
Any user with the edit permission can unlock the file.
"""
self.ensure_one()
if self.lock_uid:
self.lock_uid = False
else:
self.lock_uid = self.env.uid
def action_archive(self):
if not self:
return
to_archive_sudo = self.sudo().with_context(active_test=False).search([('id', 'child_of', self.ids)])
active_documents = to_archive_sudo.filtered(self._active_name).sudo(False)
if not active_documents:
return
active_documents._raise_if_unauthorized_archive()
active_documents._raise_if_used_folder()
deletion_date = fields.Date.to_string(fields.Date.today() + relativedelta(days=self.get_deletion_delay()))
log_message = _("This file has been sent to the trash and will be deleted forever on the %s", deletion_date)
active_documents._message_log_batch(bodies={doc.id: log_message for doc in active_documents})
return super(DocumentsDocument, active_documents).action_archive()
def action_unarchive(self):
self_archived = self.filtered(lambda d: not d.active)
if not self_archived:
return
archived_top_parent_documents = self.env["documents.document"].sudo().search(
Domain.AND((
Domain('id', 'parent_of', self_archived.ids),
Domain('id', 'not in', self_archived.ids),
Domain('active', '=', False),
Domain('folder_id', '=', False) | Domain('folder_id.active', '=', True),
))
).sudo(False)
if archived_top_parent_documents:
raise UserError(_(
"Item(s) you wish to restore are included in archived folders. "
"To restore these items, you must restore the following including folders instead:\n"
"- %(folders_list)s",
folders_list="\n-".join(archived_top_parent_documents.mapped('name')))) # "Restricted" if not allowed
# Leave archived children (and descendants) the current user doesn't have access to.
to_unarchive_candidate_documents = self.env['documents.document'].with_context(active_test=False).search(
[('id', 'child_of', self_archived.ids)])
seen_documents, to_unarchive_ids = set(), set()
def add_if_can_be_restored(doc):
if doc in seen_documents or seen_documents.add(doc):
return doc.id in to_unarchive_ids
if not doc.folder_id or doc.folder_id.sudo().active or add_if_can_be_restored(doc.folder_id):
to_unarchive_ids.add(doc.id)
return True
return False
for document in to_unarchive_candidate_documents:
add_if_can_be_restored(document)
to_unarchive_documents = to_unarchive_candidate_documents.filtered(lambda d: d.id in to_unarchive_ids)
log_message = _("This document has been restored.")
to_unarchive_documents._message_log_batch(bodies={doc.id: log_message for doc in to_unarchive_documents})
return super(DocumentsDocument, to_unarchive_documents).action_unarchive()
@api.model_create_multi
def create(self, vals_list):
"""Access rights fields (access_ids), access_internal, and access_via_link are inherited from containing folder
unless specified in vals or context defaults.
"""
attachments = []
for vals in vals_list:
keys = [key for key in vals if
self._fields[key].related and self._fields[key].related.split('.')[0] == 'attachment_id']
attachment_dict = {key: vals.pop(key) for key in keys if key in vals}
attachment = self.env['ir.attachment'].browse(vals.get('attachment_id'))
self._clean_vals_for_user_folder_id(vals)
if attachment and attachment_dict:
attachment.write(attachment_dict)
elif attachment_dict:
attachment_dict.setdefault('name', vals.get('name', 'unnamed'))
# default_res_model and default_res_id will cause unique constraints to trigger.
attachment = self.env['ir.attachment'].with_context(clean_context(self.env.context)).create(attachment_dict)
vals['attachment_id'] = attachment.id
vals['name'] = vals.get('name', attachment.name)
attachments.append(attachment)
# don't allow using default_access_ids
documents = super(DocumentsDocument, self.with_context(default_access_ids=None)).create(vals_list)
is_manager = self.env.is_admin() or self.env.user.has_group('documents.group_documents_manager')
if not is_manager:
if any(d.alias_name for d in documents):
raise AccessError(_('Only Documents Managers can set aliases.'))
if any(d._is_company_root_folder() for d in documents):
raise AccessError(_('Only Documents Managers can create in company folder.'))
for document, attachment in zip(documents, attachments):
if attachment and not attachment.res_id and (
not attachment.res_model or attachment.res_model == 'documents.document'):
attachment.with_context(no_document=True).write({
'res_model': 'documents.document',
'res_id': document.id,
})
return documents
def _prepare_create_values(self, vals_list):
old_vals_list = [vals.copy() for vals in vals_list]
vals_list = super()._prepare_create_values(vals_list)
folders = self.env['documents.document'].browse(v['folder_id'] for v in vals_list if v.get('folder_id'))
users = self.env['res.users'].browse(v['owner_id'] for v in vals_list if v.get('owner_id'))
folders.fetch(('access_internal', 'access_via_link', 'access_ids', 'active', 'company_id', 'owner_id'))
(users | folders.owner_id).fetch(['partner_id'])
vals_list_to_update_linked_record = []
for vals, old_vals in zip(vals_list, old_vals_list):
owner = self.env['res.users'].browse(vals.get('owner_id', self.env.user.active and self.env.user.id))
if owner and not owner.active:
_logger.warning(
"Documents: Creating document(s) as %s" % (
"superuser" if owner.id == SUPERUSER_ID
else f"archived user (id={owner.id})"),
)
owner = self.env['res.users']
vals['owner_id'] = False
vals_values = {'owner_id': owner.id}
if vals.get('shortcut_document_id'):
self.browse(vals.get('shortcut_document_id')).check_access('read')
folder = self.env['documents.document'].browse(vals.get('folder_id', False))
if folder:
if not folder.active:
raise UserError(self.env._('It is not possible to create documents in an archived folder.'))
vals_values.update({
'access_via_link': folder.access_via_link,
'access_internal': folder.access_internal,
})
if folder.company_id:
vals_values['company_id'] = folder.company_id.id
vals.update((k, v) for k, v in vals_values.items() if k not in old_vals)
# Add folder-inherited members without overriding provided values
if ('shortcut_document_id' not in old_vals and old_vals.get('access_ids') not in (False, Command.set([]), [])
and folder and (inherited_access_ids := folder._get_inherited_access_ids_vals())):
vals_access_ids_to_check = vals['access_ids'] if old_vals.get('access_ids') else []
partner_ids = [val[2]['partner_id'] for val in vals_access_ids_to_check]
access_vals_to_add = [v for v in inherited_access_ids if v['partner_id'] not in partner_ids]
vals['access_ids'] += [Command.create(access_vals) for access_vals in access_vals_to_add]
# Ensure owner logged access
if owner:
vals['access_ids'] = vals['access_ids'] or []
for values in vals['access_ids']:
if values and values[2] and values[2]['partner_id'] == owner.partner_id.id:
values[2]['last_access_date'] = fields.Datetime.now()
break
else:
vals['access_ids'] += [
Command.create({'partner_id': owner.partner_id.id, 'last_access_date': fields.Datetime.now()})
]
# If res_model and res_id are not set, we must get it from the related attachment if set (prepare list)
if 'res_model' not in vals and 'res_id' not in vals and isinstance(vals.get('attachment_id'), int):
vals_list_to_update_linked_record.append(vals)
# For the next step, we need to ensure the related ref is present by getting it from attachment if needed
if vals_list_to_update_linked_record:
attachment_by_id = self.env['ir.attachment'].browse(
[vals['attachment_id'] for vals in vals_list_to_update_linked_record]).grouped('id')
for vals in vals_list_to_update_linked_record:
attachment = attachment_by_id[vals['attachment_id']]
vals['res_model'] = False if attachment.res_model == 'documents.document' else attachment.res_model
vals['res_id'] = False if attachment.res_model == 'documents.document' else attachment.res_id
# Delegate vals_list update to _prepare_create_values_for_model to add values depending on related record
updated_vals_list = []
for res_model, model_vals_tuple_list in groupby(zip(vals_list, old_vals_list), lambda v: v[0].get('res_model')):
updated_vals_list += self._prepare_create_values_for_model(
res_model,
[vals_tuple[0] for vals_tuple in model_vals_tuple_list],
[vals_tuple[1] for vals_tuple in model_vals_tuple_list],
)
return updated_vals_list
def _get_inherited_access_ids_vals(self):
"""Get access values to create when creating a document inside a folder (self).
:rtype: list[dict]
:return: vals_list for folder child `access_ids`.
"""
self.ensure_one()
vals = [
{'partner_id': access.partner_id.id, 'role': access.role, 'expiration_date': access.expiration_date}
for access in self.access_ids.filtered('role')
if access.partner_id != self.owner_id.partner_id
]
if self.owner_id:
vals += [{'partner_id': self.owner_id.partner_id.id, 'role': 'edit'}]
return vals
def _prepare_create_values_for_model(self, res_model, vals_list, pre_vals_list):
"""Override to add values depending on related model/record"""
if (
res_model
and issubclass(self.pool[res_model], self.pool['documents.mixin'])
and not self.env.context.get('no_document')
):
return self.env[res_model]._prepare_document_create_values_for_linked_records(
res_model, vals_list, pre_vals_list)
return vals_list
def write(self, vals):
if 'shortcut_document_id' in vals:
raise UserError(_("Shortcuts cannot change target document."))
self._clean_vals_for_user_folder_id(vals)
is_manager = self.env.is_admin() or self.env.user.has_group('documents.group_documents_manager')
pinned_folders_start = self.filtered(lambda d: d._is_company_root_folder())
previous_owner_access_to_keep = {}
documents_per_initial_active = {}
if (owner_id := vals.get('owner_id')) is not None:
can_change_owner = is_manager or all(
(not d.owner_id and d.user_can_move) or d.owner_id == self.env.user for d in self)
if not can_change_owner:
raise AccessError(_("You cannot change the owner of documents you do not own."))
if not isinstance(owner_id, int | bool | None):
owner_id = owner_id.id
documents_changing_owner = self.filtered(lambda d: d.owner_id and d.owner_id.id != owner_id)
previous_owner_access_to_keep.update(documents_changing_owner.grouped('owner_id'))
new_parent_folder = self.browse()
documents_to_move, documents_to_move_per_initial_folder = self.browse(), self.browse()
if folder_id := vals.get('folder_id'):
new_parent_folder = self.browse(folder_id)
if new_parent_folder.type != 'folder':
raise UserError(_("Invalid folder id"))
documents_to_move = self.filtered(lambda d: d.folder_id != new_parent_folder)
if documents_to_move and not new_parent_folder.active:
raise UserError(_("It is not possible to move documents into archived folders."))
if documents_to_move and not self.env.su:
if new_parent_folder.user_permission != 'edit':
raise AccessError(_("You can't access that folder_id."))
for doc in documents_to_move:
if doc.user_permission != 'edit':
raise AccessError(_("You are not allowed to move (some of) these documents."))
if not doc.user_can_move:
raise AccessError(_("You can't move documents you do not own out of folders you cannot edit."))
if new_parent_folder.shortcut_document_id:
return self.write(vals | {'folder_id': new_parent_folder.shortcut_document_id.id})
for doc in documents_to_move:
to_active = vals.get('active')
if (
not doc.active and not to_active
or doc.folder_id and not doc.folder_id.active and (not to_active or doc.folder_id not in self)
):
raise UserError(_("It is not possible to move archived documents."))
documents_to_move_per_initial_folder = documents_to_move.grouped('folder_id')
if (to_active := vals.get('active')) is not None:
if self.env.user.share and not self.env.su:
raise UserError(_("You are not allowed to (un)archive documents."))
if not to_active:
self.check_access('unlink') # As archived gc leads to unlink after `deletion_delay` days.
documents_per_initial_active = self.grouped('active')
attachment_id = vals.get('attachment_id')
if attachment_id:
self.ensure_one()
attachments_was_present = []
for record in self:
attachments_was_present.append(bool(record.attachment_id))
if record.type == 'binary' and ('datas' in vals or 'url' in vals) and not record.datas:
body = _("Document Request: %(name)s Uploaded by: %(user)s", name=record.name, user=self.env.user.name)
record.with_context(no_document=True).message_post(body=body)
if record.attachment_id and record._has_versioning():
# versioning
if attachment_id and attachment_id != record.attachment_id.id:
# Link the new attachment to the related record and link the previous one
# to the document.
attachment = self.env["ir.attachment"].browse(attachment_id)
if (attachment.res_model, attachment.res_id) != (record.res_model, record.res_id):
attachment.with_context(no_document=True).write({
"res_model": record.res_model or "documents.document",
"res_id": record.res_id if record.res_model else record.id,
})
related_record = record.res_model and self.env[record.res_model].browse(record.res_id)
if (
not hasattr(related_record, "message_main_attachment_id")
or related_record.message_main_attachment_id
!= record.attachment_id
):
record.attachment_id.with_context(no_document=True).write(
{"res_model": "documents.document", "res_id": record.id}
)
if attachment_id in record.previous_attachment_ids.ids:
record.previous_attachment_ids = [(3, attachment_id, False)]
record.previous_attachment_ids = [(4, record.attachment_id.id, False)]
if 'datas' in vals:
old_attachment = record.attachment_id.with_context(no_document=True).copy()
# removes the link between the old attachment and the record.
old_attachment.write({
'res_model': 'documents.document',
'res_id': record.id,
})
record.previous_attachment_ids = [(4, old_attachment.id, False)]
elif not record.attachment_id and vals.get('datas') and not vals.get('attachment_id'):
res_model = vals.get('res_model', record.res_model)
res_id = vals.get('res_id', record.res_id)
if res_model and not self.env[res_model].browse(res_id).exists():
record.res_model = False
record.res_id = False
attachment = self.env['ir.attachment'].with_context(no_document=True).create({
'name': vals.get('name', record.name),
'res_model': record.res_model or 'documents.document',
'res_id': record.res_id if record.res_model else record.id,
})
record.attachment_id = attachment.id
# pops the datas and/or the mimetype key(s) to explicitly write them in batch on the ir.attachment
# so the mimetype is properly set. The reason was because the related keys are not written in batch
# and because mimetype is readonly on `ir.attachment` (which prevents writing through the related).
attachment_dict = {key: vals.pop(key) for key in ['datas', 'mimetype'] if key in vals}
if not is_manager and set(vals) & set(self.env['mail.alias.mixin']._fields):
raise AccessError(_('Only Documents managers can set an alias.'))
write_result = super().write(vals)
if attachment_dict:
self.attachment_id.write(attachment_dict)
if 'attachment_id' in vals:
self.attachment_id.check_access('read')
if (new_active := vals.get('active')) is not None:
if not new_active:
if self.sudo().search([('id', 'child_of', self.ids), ('active', '=', True)]):
raise UserError(_('Operation not supported. Please use "Move to Trash" / `action_archive` instead.'))
if archived_documents := documents_per_initial_active.get(True): # Log moved to trash instead of "archived"
for folder, children in archived_documents.filtered('folder_id').grouped('folder_id').items():
folder.sudo(self.env.user in children.owner_id).message_post(
body=_('The following documents have been sent to trash: %(documents)s.',
documents=', '.join(children.mapped('display_name')))
)
elif new_active:
if self.sudo().search([('id', 'parent_of', self.ids), ('active', '=', False)]):
raise UserError(_('Operation not supported. Please use "Restore" / `action_unarchive` instead.'))
if restored_documents := documents_per_initial_active.get(False): # Log restored instead of "unarchived"
for folder, children in restored_documents.filtered('folder_id').grouped('folder_id').items():
folder.sudo(self.env.user in children.owner_id).message_post(
body=_('The following documents have been restored from the trash: %(documents)s.',
documents=', '.join(children.mapped('display_name')))
)
if not is_manager and self.filtered(lambda d: d._is_company_root_folder()) != pinned_folders_start:
raise AccessError(_("Only Documents Managers can create in company folder."))
for document, attachment_was_present in zip(self, attachments_was_present):
if document.request_activity_id and document.attachment_id and not attachment_was_present:
feedback = _("Document Request: %(name)s Uploaded by: %(user)s",
name=self.name, user=self.env.user.name)
document.with_context(no_document=True).request_activity_id.action_feedback(
feedback=feedback, attachment_ids=[document.attachment_id.id])
if ((company_id := vals.get('company_id')) is not None) and self.shortcut_ids | self.children_ids:
self._update_company(company_id)
# Ensure edit role for previous owners
self._ensure_user_role_without_propagation('edit', previous_owner_access_to_keep)
if new_parent_folder and (documents_to_sync := documents_to_move.filtered(lambda d: not d.shortcut_document_id)):
documents_to_sync.action_update_access_rights(
access_internal=new_parent_folder.access_internal,
access_via_link=new_parent_folder.access_via_link,
# Simply add partners of destination
partners={access.partner_id: (access.role, access.expiration_date)
for access in new_parent_folder.access_ids if access.role},
)
# Propagate folder company unless passed as well (already done)
if 'company_id' not in vals:
documents_to_sync._update_company(new_parent_folder.company_id.id)
if new_parent_folder and documents_to_move:
for folder, documents in documents_to_move.grouped('folder_id').items():
folder.message_post_with_source(
source_ref='documents.folder_notification_move_in',
render_values={"documents": documents},
)
for folder, documents in documents_to_move_per_initial_folder.items():
folder.message_post_with_source(
source_ref='documents.folder_notification_move_out',
render_values={"documents": documents},
)
return write_result
@api.onchange("attachment_id")
def _onchange_attachment_id(self):
self.attachment_id.check_access('read')
def _has_versioning(self):
self.ensure_one()
return True
@api.model
def _pdf_split(self, new_files=None, open_files=None, vals=None):
vals = vals or {}
new_attachments = self.env['ir.attachment']._pdf_split(new_files=new_files, open_files=open_files)
new_documents = self.create([
dict(vals, attachment_id=attachment.id) for attachment in new_attachments
])
# Prevent concurrent update error on accessing these documents for the first time on exiting the split tool
env_partner = self.env.user.partner_id
documents_not_member = new_documents.filtered(lambda d: env_partner not in d.access_ids.partner_id)
self.env['documents.access'].sudo().create([
{'document_id': doc.id, 'partner_id': env_partner.id, 'last_access_date': fields.Datetime.now()}
for doc in documents_not_member
])
return new_documents
@api.model
def search_panel_select_range(self, field_name, **kwargs):
def convert_user_folder_ids_to_int(vals):
"""Convert user_folder_id to int where applicable to construct categoryTree matching on id of parent."""
if (user_folder_id := vals['user_folder_id']) and user_folder_id.isnumeric():
vals['user_folder_id'] = int(user_folder_id)
if field_name == 'user_folder_id':
enable_counters = kwargs.get('enable_counters', False)
search_panel_fields = self._get_search_panel_fields()
domain = Domain('type', '=', 'folder')
if unique_folder_id := self.env.context.get('documents_unique_folder_id'):
if access_token := self.env.context.get('documents_shared_access_token'):
# Enable non-public access via link through a knowledge embedded view without extra rpc call.
self._search_panel_access_via_link(unique_folder_id, access_token)
values = self.env['documents.document'].search_read(
domain & Domain('folder_id', 'child_of', unique_folder_id),
search_panel_fields,
)
for record in values:
convert_user_folder_ids_to_int(record)
if record['id'] == unique_folder_id:
record['user_folder_id'] = False # Set as root
return {
'parent_field': 'user_folder_id',
'values': values,
}
records = self.env['documents.document'].search_read(domain, search_panel_fields)
alias_tag_data = {}
if not self.env.user.share:
alias_tag_ids = {alias_tag_id for rec in records for alias_tag_id in rec['alias_tag_ids']}
alias_tag_data = {
alias_tag['id']: {
'id': alias_tag.id,
'color': alias_tag.color,
'display_name': alias_tag.display_name
} for alias_tag in self.env['documents.tag'].browse(alias_tag_ids)
}
domain_image = {}
if enable_counters:
model_domain = Domain.AND([
kwargs.get('search_domain', []),
kwargs.get('category_domain', []),
kwargs.get('filter_domain', []),
Domain(field_name, '!=', False),
])
domain_image = self._search_panel_domain_image(field_name, model_domain, enable_counters)
# Read the targets in batch
targets = self.browse(r['shortcut_document_id'][0] for r in records if r['shortcut_document_id'])
targets_data = targets.grouped('id')
values_range = OrderedDict()
for record in records:
record_id = record['id']
convert_user_folder_ids_to_int(record)
if not self.env.user.share:
record['alias_tag_ids'] = [alias_tag_data[tag_id] for tag_id in record['alias_tag_ids']]
if enable_counters:
image_element = domain_image.get(record_id)
record['__count'] = image_element['__count'] if image_element else 0
if record['shortcut_document_id']:
target = targets_data[record['shortcut_document_id'][0]]
record['target_user_permission'] = target.user_permission
record['target_access_token'] = target.access_token
values_range[record_id] = record
if enable_counters:
self._search_panel_global_counters(values_range, 'user_folder_id')
special_roots = []
if not self.env.user.share:
special_roots = [
{'bold': True, 'childrenIds': [], 'parentId': False, 'user_permission': 'edit'} | values
for values in [
{
'display_name': _("Company"),
'id': 'COMPANY',
'description': _("Common roots for all company users."),
'user_permission': 'edit' if self.env.user.has_group('documents.group_documents_manager') else 'view',
}, {
'display_name': _("My Drive"),
'id': 'MY',
'user_permission': 'edit',
'description': _("Your individual space."),
}, {
'display_name': _("Shared with me"),
'id': 'SHARED',
'description': _("Additional documents you have access to."),
}, {
'display_name': _("Recent"),
'id': 'RECENT',
'description': _("Recently accessed documents."),
}]
]
if not self.env.context.get('documents_search_panel_no_trash'):
special_roots.append({
'display_name': _("Trash"),
'id': 'TRASH',
'description': _("Items in trash will be deleted forever after %s days.",
self.get_deletion_delay()),
})
return {
'parent_field': 'user_folder_id',
'values': list(values_range.values()) + special_roots,
}
return super().search_panel_select_range(field_name)
@api.model
def _search_panel_access_via_link(self, folder_id, access_token):
"""Log access if necessary to access the record and it is accessible via link.
In this case, both folder_id and access_token are necessary and must agree,
and access is only logged if access_via_link is not "none".
Notably useful for knowledge embedded views.
"""
folder_sudo = self.browse(folder_id).sudo()
if (
folder_sudo.user_permission == "none"
and folder_sudo.access_via_link != "none"
and consteq(folder_sudo.access_token, access_token)
):
self.env['documents.access'].sudo().create({
'document_id': folder_sudo.id,
'partner_id': self.env.user.partner_id.id,
'last_access_date': fields.Datetime.now(),
})
@api.readonly
@api.model
def get_document_max_upload_limit(self):
ICP = self.env['ir.config_parameter'].sudo()
for key in ('document.max_fileupload_size', 'web.max_file_upload_size'):
value = ICP.get_param(key, default=None)
if value is None:
continue
try:
return int(value) or None
except ValueError:
_logger.error("invalid %s: %r", key, value)
return odoo.http.DEFAULT_MAX_CONTENT_LENGTH
@api.readonly
@api.model
def get_details_panel_res_models(self):
"""Return the list of models that a document can be linked to via the details panel.
:rtype: list[str]
"""
functional_models = [
"account.move", "fleet.vehicle", "hr.expense", "hr.leave", "product.product", "project.project",
"project.task", "purchase.order", "sale.order",
]
return [
model for model in functional_models
if (res_model := self.env.get(model)) is not None and res_model.has_access('read')
]
@api.model
def _get_traceback_folder_sudo(self):
folder_id = self.env["ir.config_parameter"].sudo().get_param('documents.support_folder', False)
folder_sudo = self.env["documents.document"].sudo().browse(int(folder_id))
if not folder_sudo or not folder_sudo.exists():
folder_sudo = self.env["documents.document"].sudo().create({
'name': self.env._('Support'),
'type': 'folder',
'access_internal': 'none',
'access_via_link': 'none'
})
self.env["ir.config_parameter"].sudo().set_param('documents.support_folder', folder_sudo.id)
return folder_sudo
def unlink(self):
"""Clean unused linked records too.
This applies to:
* Children documents when deleting the parent folder
* Parent folder if it is archived and has no other children and user is allowed to
* Attachment if document-related record is deleted
"""
to_delete = self.sudo().with_context(active_test=False).search([('id', 'child_of', self.ids)]).sudo(False)
removable_parent_folders = self.with_context(active_test=False).folder_id.filtered(
lambda folder: len(folder.children_ids - self) == 0 and not folder.active and folder.id not in self.ids)
removable_attachments = self.attachment_id.filtered(lambda a: a.res_model != 'documents.document')
res = super(DocumentsDocument, to_delete).unlink()
if removable_attachments:
removable_attachments.unlink()
if removable_parent_folders:
with contextlib.suppress(AccessError):
removable_parent_folders.unlink()
return res
def _check_access(self, operation: str):
"""Prevent unlink when not the owner and no edit permission on the containing folder."""
res = super()._check_access(operation)
if self.env.su or operation != 'unlink':
return res
forbidden_documents = res[0] if res else self.browse()
can_unlink_map = self._get_user_can_unlink()
forbidden_documents |= self.filtered(lambda d: not can_unlink_map[d])
if forbidden_documents:
# Hide potentially unknown inaccessible content's name.
message = _("You do not have sufficient access rights to delete these documents.")
return forbidden_documents, lambda: AccessError(message)
return res
@api.ondelete(at_uninstall=False)
def _unlink_except_unauthorized(self):
"""deprecated: legacy hook as the logic is now included in `_check_access`. todo: remove in master."""
self._raise_if_unauthorized_archive()
@api.ondelete(at_uninstall=False)
def _unlink_except_company_folders(self):
self._raise_if_used_folder()
def _raise_if_used_folder(self):
if folder_ids := self.filtered(lambda d: d.type == 'folder').ids:
company_used_folders_domain = self.env['res.company']._get_used_folder_ids_domain(folder_ids)
if self.env['res.company'].sudo().search_count(company_used_folders_domain, limit=1):
raise ValidationError(_("Impossible to delete folders used by other applications."))
def _raise_if_unauthorized_archive(self):
"""deprecated: legacy hook as the logic is now included in `_check_access`. todo: remove in master."""
@api.autovacuum
def _gc_clear_bin(self):
"""Files are deleted automatically from the trash bin after the configured remaining days."""
self.search(self._get_gc_clear_bin_domain(), limit=1000).unlink()
@api.model
def _get_gc_clear_bin_domain(self):
deletion_delay = self.get_deletion_delay()
return [
('active', '=', False),
('write_date', '<=', fields.Datetime.now() - relativedelta(days=deletion_delay)),
]
@api.model
def _get_search_panel_fields(self):
"""Return the list of fields used by the search panel."""
search_panel_fields = ['access_internal', 'access_token', 'access_via_link', 'active', 'company_id',
'description', 'display_name', 'user_folder_id', 'is_access_via_link_hidden',
'is_favorited', 'mail_alias_domain_count',
'owner_id', 'shortcut_document_id', 'user_permission']
if not self.env.user.share:
search_panel_fields += ['alias_domain_id', 'alias_name', 'alias_tag_ids', 'create_activity_type_id',
'create_activity_user_id', 'partner_id']
return search_panel_fields
def _get_access_action(self, access_uid=None, force_website=False):
self.ensure_one()
if access_uid and not force_website and self.active and self.env.user.has_group("documents.group_documents_user"):
url_params = url_encode({
'documents_init_document_id': self.id,
'view_id': self.env.ref("documents.document_view_kanban").id,
'menu_id': self.env.ref("documents.menu_root").id,
'folder_id': self.folder_id.id,
})
return {
"type": "ir.actions.act_url",
"url": f"/odoo/action-documents.document_action?{url_params}"
}
return super()._get_access_action(access_uid=access_uid, force_website=force_website)
@api.model
def _data_embed_if_records_exist(self, folder_xmlid, server_action_xmlid):
if (
(action := self.env.ref(server_action_xmlid, raise_if_not_found=False))
and (folder := self.env.ref(folder_xmlid, raise_if_not_found=False))
):
self.action_folder_embed_action(folder.id, action.id)
def add_documents_attachment(self, res_model, res_id, is_public=False):
"""Unified method to create document attachments with optional plugin handling
:param str res_model: model name to attach the document to
:param int res_id: record ID to attach the document to
:param bool is_public: specify attachment can publicly accessible
:return: Data of newly created attachments
:rtype: list[dict]
"""
# Build attachment data with conditional plugin parameters
new_attachments = self.env['ir.attachment']
for attachment in self.attachment_id:
copied = attachment.copy({
"res_model": res_model,
"res_id": res_id,
"res_field": False,
"public": is_public,
"original_id": attachment.id,
})
new_attachments |= copied
# Generate access tokens if needed
if is_public:
for attachment in new_attachments:
attachment.generate_access_token()
return [attachment._get_media_info() for attachment in new_attachments]
Reference in new issue View Git Blame Copy Permalink