# 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]