Files

609 lines
24 KiB
Python

# Part of Odoo. See LICENSE file for full copyright and licensing details.
import base64
import contextlib
import io
import json
import zipfile
from lxml import etree
from odoo import _, fields, models, api
from odoo.exceptions import UserError, AccessError, ValidationError
from odoo.fields import Command, Domain
from odoo.tools import consteq
from odoo.tools.image import image_process
from odoo.tools.misc import DEFAULT_SERVER_DATE_FORMAT
SUPPORTED_PATHS = (
"[Content_Types].xml",
"xl/sharedStrings.xml",
"xl/styles.xml",
"xl/workbook.xml",
"_rels/",
"xl/_rels",
"xl/charts/",
"xl/drawings/",
"xl/externalLinks/",
"xl/pivotTables/",
"xl/tables/",
"xl/theme/",
"xl/worksheets/",
"xl/media",
)
XLSX_MIME_TYPES = [
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"application/wps-office.xlsx",
]
class DocumentsDocument(models.Model):
_name = 'documents.document'
_inherit = ["documents.document", "spreadsheet.mixin"]
spreadsheet_binary_data = fields.Binary(compute='_compute_spreadsheet_binary_data', inverse='_inverse_spreadsheet_binary_data', default=None)
spreadsheet_thumbnail_checksum = fields.Char(
compute='_compute_spreadsheet_thumbnail_checksum',
export_string_translation=False
)
excel_export = fields.Binary()
handler = fields.Selection([
("spreadsheet", "Spreadsheet"),
("frozen_folder", "Frozen Folder"),
("frozen_spreadsheet", "Frozen Spreadsheet"),
], ondelete={"spreadsheet": "cascade", "frozen_folder": "cascade", "frozen_spreadsheet": "cascade"})
_spreadsheet_access_via_link = models.Constraint(
"CHECK((handler != 'spreadsheet') OR access_via_link != 'edit')",
"To share a spreadsheet in edit mode, add the user in the accesses",
)
_frozen_spreadsheet_access_via_link_access_internal = models.Constraint(
"CHECK((handler != 'frozen_spreadsheet') OR (access_via_link != 'edit' AND access_internal != 'edit'))",
"A frozen spreadsheet can not be editable",
)
def action_freeze_and_copy(self, spreadsheet_data, excel_files):
"""Render the spreadsheet in JS, and then make a copy to share it.
:param spreadsheet_data: The spreadsheet data to save
:param excel_files: The files to download
"""
self.ensure_one()
# we will copy in SUDO to skip check on frozen spreadsheets
# (see @_check_spreadsheet_share)
self.env['documents.document'].check_access('create')
self.check_access('read')
if not self.env.su and self.folder_id and self.folder_id.user_permission != 'edit':
raise AccessError(_('You are not allowed to freeze spreadsheets in Company'))
folder_sudo = self.env['documents.document'].sudo().search([
('folder_id', '=', self.folder_id.id),
('type', '=', 'folder'),
('handler', '=', 'frozen_folder'),
], limit=1)
if not folder_sudo:
folder_sudo = self.env['documents.document'].sudo().create({
'name': _('Frozen spreadsheets'),
'type': 'folder',
'handler': 'frozen_folder',
'folder_id': self.folder_id.id,
'access_via_link': 'none',
'access_internal': 'none',
'access_ids': False,
'owner_id': False,
})
if isinstance(spreadsheet_data, dict):
spreadsheet_data = json.dumps(spreadsheet_data)
document = self.sudo().copy({
'name': _('Frozen at %(date)s: %(name)s',
date=fields.Date.today().strftime(DEFAULT_SERVER_DATE_FORMAT), name=self.name),
'access_internal': 'none' if self.access_internal == 'none' else 'view',
'access_via_link': 'view',
'spreadsheet_data': spreadsheet_data,
'folder_id': folder_sudo.id,
'excel_export': base64.b64encode(self.env['spreadsheet.mixin']._zip_xslx_files(excel_files)),
'handler': 'frozen_spreadsheet',
'is_access_via_link_hidden': True,
'access_ids': [Command.create({
'partner_id': access.partner_id.id,
'role': 'view' if access.role == 'edit' else access.role,
}) for access in self.access_ids if access.role],
})
return {'id': document.id, 'shortcut_document_id': document.shortcut_document_id.id}
def _get_access_update_domain(self):
"""Allow to change the access of the frozen folders / spreadsheets only if we open their share panel."""
return super()._get_access_update_domain() & (
Domain('id', 'in', self.ids)
| Domain('handler', 'not in', ('frozen_folder', 'frozen_spreadsheet'))
)
@api.model_create_multi
def create(self, vals_list):
vals_list = self._resize_spreadsheet_thumbnails(vals_list)
documents = super().create(vals_list)
documents._update_spreadsheet_contributors()
return documents
def write(self, vals):
if 'handler' not in vals and 'mimetype' in vals and vals['mimetype'] != 'application/o-spreadsheet':
vals['handler'] = False
if 'spreadsheet_data' in vals:
self._update_spreadsheet_contributors()
if all(document.handler in ("spreadsheet", "frozen_spreadsheet") for document in self):
vals = self._resize_thumbnail_value(vals)
return super().write(vals)
def dispatch_spreadsheet_message(self, message, access_token=None):
if self.sudo().handler == "frozen_spreadsheet":
return False
return super().dispatch_spreadsheet_message(message, access_token)
def _get_serialized_spreadsheet_data_body(self, access_token=None):
body = super()._get_serialized_spreadsheet_data_body(access_token)
self._update_spreadsheet_contributors()
return body
def _get_spreadsheet_metadata(self, access_token=None):
if self.sudo().handler not in ("spreadsheet", "frozen_spreadsheet"):
raise ValidationError(_("The spreadsheet you are trying to access does not exist."))
data = super()._get_spreadsheet_metadata(access_token)
self._update_spreadsheet_contributors()
sudo_self = self.sudo()
return {
**data,
'handler': sudo_self.handler,
'access_url': sudo_self.access_url,
'is_favorited': sudo_self.is_favorited,
'is_archived': sudo_self.active is False,
'folder_id': sudo_self.folder_id.id,
'copy_in_my_drive': self._cannot_create_sibling(),
}
@api.constrains('owner_id')
def _check_owner_is_internal_user(self):
spreadsheet_docs = self.filtered(lambda rec: rec.handler in ("spreadsheet", "frozen_spreadsheet"))
if any(user.share for user in spreadsheet_docs.owner_id):
raise AccessError(_("Portal users cannot be the owner of a spreadsheet."))
def _check_spreadsheet_share(self, operation, access_token):
if not self.env.su and operation == 'write' and self.sudo().handler == 'frozen_spreadsheet':
raise AccessError(_("You can not edit a frozen spreadsheet"))
with contextlib.suppress(AccessError):
super()._check_spreadsheet_share(operation, access_token)
return
if (
not access_token
or not consteq(access_token, self.sudo().access_token)
or (operation == 'write' and self.sudo().access_via_link != 'edit')
or (operation == 'read' and self.sudo().access_via_link == 'none')
):
raise AccessError(_("You don't have access to this document"))
def _compute_file_extension(self):
""" Spreadsheet documents do not have file extension. """
spreadsheet_docs = self.filtered(lambda rec: rec.handler in ("spreadsheet", "frozen_spreadsheet"))
spreadsheet_docs.file_extension = False
super(DocumentsDocument, self - spreadsheet_docs)._compute_file_extension()
def _has_versioning(self):
return super()._has_versioning() and self.handler not in ("spreadsheet", "frozen_spreadsheet")
@api.depends("attachment_id", "handler")
def _compute_spreadsheet_data(self):
for document in self.with_context(bin_size=False):
if document.handler in ("spreadsheet", "frozen_spreadsheet"):
document.spreadsheet_data = document.attachment_id.raw
else:
document.spreadsheet_data = False
@api.depends("datas", "handler")
def _compute_spreadsheet_binary_data(self):
for document in self:
if document.handler in ("spreadsheet", "frozen_spreadsheet"):
document.spreadsheet_binary_data = document.datas
else:
document.spreadsheet_binary_data = False
def _inverse_spreadsheet_binary_data(self):
for document in self:
if document.handler in ("spreadsheet", "frozen_spreadsheet"):
document.write({
"datas": document.spreadsheet_binary_data,
"mimetype": "application/o-spreadsheet"
})
@api.depends_context('uid')
@api.depends("display_thumbnail")
def _compute_spreadsheet_thumbnail_checksum(self):
spreadsheets = self.filtered(lambda doc: doc.handler == "spreadsheet")
thumbnails = self.env['ir.attachment'].sudo().search([
('res_model', '=', self._name),
('res_field', '=', 'display_thumbnail'),
('res_id', 'in', spreadsheets.ids),
('create_uid', '=', self.env.uid)
])
thumbnails_by_documents = thumbnails.grouped("res_id")
for document in self:
thumbnail = thumbnails_by_documents.get(document.id)
if thumbnail:
document.spreadsheet_thumbnail_checksum = thumbnail.checksum
else:
document.spreadsheet_thumbnail_checksum = False
@api.depends("checksum", "handler")
def _compute_thumbnail(self):
# Spreadsheet thumbnails cannot be computed from their binary data.
# They should be saved independently.
spreadsheets = self.filtered(lambda d: d.handler in ("spreadsheet", "frozen_spreadsheet"))
super(DocumentsDocument, self - spreadsheets)._compute_thumbnail()
def _copy_spreadsheet_image_attachments(self):
spreadsheets = self.filtered(lambda d: d.handler in ("spreadsheet", "frozen_spreadsheet"))
super(DocumentsDocument, spreadsheets)._copy_spreadsheet_image_attachments()
def _copy_attachment_filter(self, default):
return super()._copy_attachment_filter(default).filtered(
lambda d: d.handler not in ("spreadsheet", "frozen_spreadsheet"))
def _resize_thumbnail_value(self, vals):
if 'display_thumbnail' in vals:
return dict(
vals,
display_thumbnail=base64.b64encode(image_process(base64.b64decode(vals['display_thumbnail'] or ''), size=(150, 150), crop='center')),
)
return vals
def _resize_spreadsheet_thumbnails(self, vals_list):
return [
(
self._resize_thumbnail_value(vals)
if vals.get('handler') in ('spreadsheet', 'frozen_spreadsheet')
else vals
)
for vals in vals_list
]
def _update_spreadsheet_contributors(self):
"""Add the current user to the spreadsheet contributors.
"""
can_write = self.env['spreadsheet.contributor'].has_access('write')
can_create = self.env['spreadsheet.contributor'].has_access('create')
if not can_write or not can_create:
return
for document in self:
if document.handler == 'spreadsheet':
self.env['spreadsheet.contributor']._update(self.env.user, document)
@api.model
def action_open_new_spreadsheet(self, vals=None):
if vals is None:
vals = {}
spreadsheet = self.create({
"name": _("Untitled spreadsheet"),
"mimetype": "application/o-spreadsheet",
"datas": self._empty_spreadsheet_data_base64(),
"handler": "spreadsheet",
**vals,
})
action_open = spreadsheet.action_open_spreadsheet()
action_open['params']['is_new_spreadsheet'] = True
return action_open
@api.model
def _get_spreadsheets_to_display(self, domain, offset=0, limit=None):
"""
Get all the spreadsheets, with the spreadsheet that the user has recently
opened at first.
"""
# TODO: improve with _read_group to not fetch all records
Contrib = self.env["spreadsheet.contributor"]
visible_docs = self.search(Domain.AND([domain, [("handler", "in", ("spreadsheet", "frozen_spreadsheet"))]]))
contribs = Contrib.search(
[
("document_id", "in", visible_docs.ids),
("user_id", "=", self.env.user.id),
],
order="last_update_date DESC, id DESC",
)
user_docs = contribs.document_id
# Intersection is used to apply the `domain` to `user_doc`, the union is
# here to keep only the visible docs, but with the order of contribs.
docs = ((user_docs & visible_docs) | visible_docs)
if limit:
docs = docs[offset:offset + limit]
else:
docs = docs[offset:]
return docs.read(["display_name", "display_thumbnail"])
@api.model
def _get_shortcuts_copy_fields(self):
return super()._get_shortcuts_copy_fields() | {'handler'}
def _clone_xlsx_into_spreadsheet(self, archive_source=False):
"""Clone an XLSX document into a new document with its content unzipped, and return the new document id"""
unzipped, attachments = self._unzip_xlsx()
def adjust_role(partner, role):
"""Ensure non-internal users do not have 'edit' access."""
user_ids = partner.with_context(active_test=False).user_ids
return 'view' if not user_ids or all(user_ids.mapped("share")) else role
access_ids = [
Command.create({
'partner_id': acc.partner_id.id,
'role': adjust_role(acc.partner_id, acc.role)
})
for acc in self.access_ids.filtered("role")
]
# Keep the linked record: copy() does not include these computed fields.
linked_record_values = {}
if self.res_model and self.res_id:
linked_record_values = {
"res_model": self.res_model,
"res_id": self.res_id,
}
doc = self.copy({
"access_ids": access_ids,
"attachment_id": False,
"handler": "spreadsheet",
"mimetype": "application/o-spreadsheet",
"name": self.name.removesuffix(".xlsx"),
"spreadsheet_data": json.dumps(unzipped),
**linked_record_values,
})
for attachment in attachments:
attachment.write({'res_id': doc.id})
return doc.id
def _clone_csv_into_spreadsheet(self):
csv_data = self._read_csv()
spreadsheet_data = self._convert_csv_to_spreadsheet_data(csv_data)
# Keep the linked record: copy() does not include these computed fields.
linked_record_values = {}
if self.res_model and self.res_id:
linked_record_values = {
"res_model": self.res_model,
"res_id": self.res_id,
}
doc = self.copy({
'attachment_id': False,
'handler': 'spreadsheet',
'mimetype': 'application/o-spreadsheet',
'name': self.name.removesuffix('.csv'),
'spreadsheet_data': json.dumps(spreadsheet_data),
**linked_record_values,
})
return doc.id
def import_to_spreadsheet(self, archive_source=False):
self.ensure_one()
if self.mimetype == 'text/csv':
document_id = self._clone_csv_into_spreadsheet()
else:
document_id = self._clone_xlsx_into_spreadsheet()
if archive_source:
self.action_archive()
return document_id
def _read_csv(self):
base_import_record = self.env['base_import.import'].create({
'file': self.attachment_id.raw,
'file_type': 'text/csv',
})
_, csv_content = base_import_record._read_csv({'encoding': 'utf-8', 'quoting': '"'})
return csv_content
def _convert_csv_to_spreadsheet_data(self, csv_data):
def get_column_name(n):
"""Helper function to get the spreadsheet column
name in letters (A -> Z) from the column index.
Example: 1 -> A, 2 -> B, 27 -> AA, 28 -> AB, etc.
"""
res = []
while n > 0:
n, remainder = divmod(n - 1, 26)
res.append(chr(65 + remainder))
return ''.join(res[::-1])
data = {'sheets': [{'cells': {}}]}
for row_index, row in enumerate(csv_data):
for col_index, cell_content in enumerate(row):
if cell_content:
col_name = get_column_name(col_index + 1)
cell_ref = f'{col_name}{row_index + 1}'
data['sheets'][0]['cells'][cell_ref] = cell_content
return data
def _get_is_multipage(self):
"""Override for spreadsheets and xlsx."""
is_multipage = super()._get_is_multipage()
if is_multipage is not None:
return is_multipage
if self.mimetype in XLSX_MIME_TYPES and self.attachment_id.raw:
file = io.BytesIO(self.attachment_id.raw)
if not zipfile.is_zipfile(file):
return None
with zipfile.ZipFile(file) as archive:
if '[Content_Types].xml' not in archive.namelist():
# the xlsx file is invalid
return None
with archive.open("[Content_Types].xml") as myfile:
content = myfile.read()
tree = etree.fromstring(content)
nodes = tree.xpath(
"//ns:Override[@ContentType='application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml']",
namespaces={"ns": "http://schemas.openxmlformats.org/package/2006/content-types"}
)
return len(nodes) > 1
if self.handler in ("spreadsheet", "frozen_spreadsheet"):
spreadsheet_data = json.loads(self.spreadsheet_data or '{}')
if spreadsheet_data.get("sheets"):
return len(spreadsheet_data["sheets"]) > 1
return self._is_xlsx_data_multipage(spreadsheet_data)
@api.model
def _is_xlsx_data_multipage(self, spreadsheet_data):
for filename, content in spreadsheet_data.items():
if filename.endswith("workbook.xml.rels"):
tree = etree.fromstring(content.encode())
nodes = tree.findall(
'.//rels:Relationship',
{'rels': 'http://schemas.openxmlformats.org/package/2006/relationships'}
)
found_first_sheet = False
for node in nodes:
if node.attrib["Type"].endswith('/relationships/worksheet'):
if found_first_sheet:
return True
found_first_sheet = True
break
return False
def _unzip_xlsx(self):
file = io.BytesIO(self.attachment_id.raw)
if not zipfile.is_zipfile(file) or self.mimetype not in XLSX_MIME_TYPES:
raise UserError(_("The file is not a xlsx file"))
unzipped_size = 0
with zipfile.ZipFile(file) as input_zip:
if len(input_zip.infolist()) > 1000:
raise UserError(_("The xlsx file is too big"))
if "[Content_Types].xml" not in input_zip.namelist() or \
not any(name.startswith("xl/") for name in input_zip.namelist()):
raise UserError(_("The xlsx file is corrupted"))
unzipped = {}
attachments = []
for info in input_zip.infolist():
if not (info.filename.endswith((".xml", ".xml.rels")) or "media/image" in info.filename) or\
not info.filename.startswith(SUPPORTED_PATHS):
# Don't extract files others than xmls or unsupported xmls
continue
unzipped_size += info.file_size
if unzipped_size > 50 * 1000 * 1000: # 50MB
raise UserError(_("The xlsx file is too big"))
if info.filename.endswith((".xml", ".xml.rels")):
unzipped[info.filename] = input_zip.read(info.filename).decode()
elif "media/image" in info.filename:
image_file = input_zip.read(info.filename)
attachment = self._upload_image_file(image_file, info.filename)
attachments.append(attachment)
unzipped[info.filename] = {
"imageSrc": "/web/image/" + str(attachment.id),
}
return unzipped, attachments
def _upload_image_file(self, image_file, filename):
attachment_model = self.env['ir.attachment']
attachment = attachment_model.create({
'name': filename,
'datas': base64.encodebytes(image_file),
'res_model': "documents.document",
})
attachment._post_add_create()
return attachment
def action_open_spreadsheet(self):
self.ensure_one()
return {
'type': 'ir.actions.client',
'tag': 'action_open_spreadsheet',
'params': {
'spreadsheet_id': self.id,
}
}
@api.readonly
@api.model
def get_spreadsheets(self, domain=(), offset=0, limit=None):
domain = Domain.AND([domain, [("handler", "=", "spreadsheet")]])
return {
"records": self._get_spreadsheets_to_display(domain, offset, limit),
"total": self.search_count(domain),
}
@api.model
def _get_spreadsheet_selector(self):
return {
"model": self._name,
"display_name": _("Spreadsheets"),
"sequence": 0,
"allow_create": True,
}
def _contains_live_data(self):
"""Return true if the spreadsheet contains live data, like Odoo pivots, chart, etc."""
self.ensure_one()
if self.handler != 'spreadsheet':
return False
snapshot = json.loads(self._get_spreadsheet_serialized_snapshot())
if snapshot.get("lists") or snapshot.get("pivots") or snapshot.get("chartOdooMenusReferences"):
return True
for sheet in snapshot.get("sheets", []):
for figure in sheet.get("figures", []):
if figure.get("data", {}).get("type", "").startswith("odoo_"):
return True
return any(
command.get("type") in ("INSERT_ODOO_LIST", "ADD_PIVOT", "LINK_ODOO_MENU_TO_CHART") or (
command.get("type") == "CREATE_CHART" and
command.get("definition", {}).get("type", "").startswith("odoo_")
)
for revision in self.spreadsheet_revision_ids
for command in json.loads(revision.commands).get("commands", [])
)
def _get_writable_record_name_field(self):
return 'name'
@api.model
def create_xlsx_attachment_from_spreadsheet(self, spreadsheet_files):
"""Creates XLSX attachments from spreadsheet JSON exports
:param spreadsheet_files: list of dicts with 'name', 'files', 'res_model' and 'res_id'
:return: list of media info dicts
:rtype: list[dict]
"""
attachments = []
for sheet in spreadsheet_files:
content = self.env['spreadsheet.mixin']._zip_xslx_files(sheet["files"])
attachment = self.env["ir.attachment"].create({
"name": sheet["name"],
"datas": base64.b64encode(content).decode(),
"res_model": sheet["res_model"],
"res_id": sheet["res_id"],
"type": "binary",
"mimetype": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
})
attachments.append(attachment)
return [a._get_media_info() for a in attachments]