diff --git a/spreadsheet_oca/README.rst b/spreadsheet_oca/README.rst index 20fb5ca4..3f2808b9 100644 --- a/spreadsheet_oca/README.rst +++ b/spreadsheet_oca/README.rst @@ -11,7 +11,7 @@ Spreadsheet Oca !! This file is generated by oca-gen-addon-readme !! !! changes will be overwritten. !! !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! - !! source digest: sha256:1310916cafc940d7dcb638553bb2072e73f47cc887916d5c384b1cec841b4223 + !! source digest: sha256:afa8556b48ef425ac88f6fa7b69a9f5d76aaef94958f9ce0e14e0efd97024dc0 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! .. |badge1| image:: https://img.shields.io/badge/maturity-Beta-yellow.png diff --git a/spreadsheet_oca/__manifest__.py b/spreadsheet_oca/__manifest__.py index 4b6779cc..e401a367 100644 --- a/spreadsheet_oca/__manifest__.py +++ b/spreadsheet_oca/__manifest__.py @@ -5,7 +5,7 @@ "name": "Spreadsheet Oca", "summary": """ Allow to edit spreadsheets""", - "version": "18.0.1.3.1", + "version": "18.0.1.3.0", "license": "AGPL-3", "author": "CreuBlanca,Odoo Community Association (OCA)", "website": "https://github.com/OCA/spreadsheet", @@ -14,6 +14,7 @@ "security/security.xml", "security/ir.model.access.csv", "views/spreadsheet_spreadsheet.xml", + "views/spreadsheet_writeback_views.xml", "data/spreadsheet_spreadsheet_import_mode.xml", "wizards/spreadsheet_select_row_number.xml", "wizards/spreadsheet_spreadsheet_import.xml", diff --git a/spreadsheet_oca/controllers/__init__.py b/spreadsheet_oca/controllers/__init__.py index 12a7e529..ab2a1c30 100644 --- a/spreadsheet_oca/controllers/__init__.py +++ b/spreadsheet_oca/controllers/__init__.py @@ -1 +1 @@ -from . import main +from . import spreadsheet_writeback diff --git a/spreadsheet_oca/controllers/spreadsheet_writeback.py b/spreadsheet_oca/controllers/spreadsheet_writeback.py new file mode 100644 index 00000000..402fa4ce --- /dev/null +++ b/spreadsheet_oca/controllers/spreadsheet_writeback.py @@ -0,0 +1,197 @@ +# Copyright 2025 Ledo Enterprises LLC +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +""" +Controller for cell writeback: edit list cells to update Odoo records. + +The JS list cell-edit handler POSTs here with: + spreadsheet_id — int + model — str (e.g. "sale.order") + record_id — int + field_name — str (e.g. "name") + new_value — any (JSON-decoded by Odoo's JSON-RPC dispatcher) + +Returns a JSON-serialisable dict: + {'success': True, 'old_value': str, 'new_value': str, 'log_id': int} + or + {'error': ''} + +All exceptions are caught so a writeback failure never results in a +500 error reaching the browser. +""" + +import logging + +from odoo import _ +from odoo.exceptions import AccessError +from odoo.http import Controller, request, route + +from ..models.spreadsheet_writeback import WRITEBACK_FIELD_TYPES + +_logger = logging.getLogger(__name__) + + +class SpreadsheetWriteback(Controller): + @route( + "/spreadsheet/writeback", + type="json", + auth="user", + methods=["POST"], + ) + def writeback(self, spreadsheet_id, model, record_id, field_name, new_value): + """ + Write a single field value to an Odoo record on behalf of the + spreadsheet's List cell-edit handler. + + Security checks (in order): + 1. spreadsheet.writeback_enabled must be True. + 2. Current user must have read access on the spreadsheet record. + 3. model must be a registered model in this environment. + 4. The target record must exist. + 5. Current user must have write access on the target record. + + The old value is captured before the write and stored in the audit + log. For relational fields str() is used which may not be + directly re-writable; see the model docstring for details. + """ + log_vals_base = { + "spreadsheet_id": spreadsheet_id, + "res_model": model, + "record_id": record_id, + "field_name": field_name, + "new_value": str(new_value), + } + + try: + # 1. Load spreadsheet and check writeback_enabled + spreadsheet = request.env["spreadsheet.spreadsheet"].browse(spreadsheet_id) + if not spreadsheet.exists(): + return {"error": "Spreadsheet not found."} + + if not spreadsheet.writeback_enabled: + return {"error": "Writeback not enabled for this spreadsheet."} + + # 2. Check spreadsheet read access + try: + spreadsheet.check_access("read") + except AccessError: + return {"error": "Access denied to spreadsheet."} + + # 3. Validate model + if model not in request.env: + return {"error": f"Model {model!r} is not available."} + + # 4. Load and check record existence + record = request.env[model].browse(record_id) + if not record.exists(): + return {"error": f"Record {model}({record_id}) not found."} + + # 5. Check write access on the target record + try: + record.check_access("write") + except AccessError: + _logger.warning( + "Writeback: user %d denied write on %s(%d)", + request.env.uid, + model, + record_id, + ) + return {"error": "Access denied: no write access on record."} + + # 6. Validate field_name exists and is writable + model_fields = request.env[model]._fields + if field_name not in model_fields: + return { + "error": _( + "Field %(field)s does not exist on model %(model)s.", + field=field_name, + model=model, + ) + } + field_obj = model_fields[field_name] + if field_obj.type not in WRITEBACK_FIELD_TYPES: + return { + "error": _( + "Field %(field)s on %(model)s is a %(type)s field." + " Only simple value fields can be written back, because" + " the previous value is stored as text so it can be" + " rolled back.", + field=field_name, + model=model, + type=field_obj.type, + ) + } + if field_obj.readonly or field_obj.compute: + return { + "error": _( + "Field %(field)s on %(model)s is computed or readonly" + " and cannot be written to.", + field=field_name, + model=model, + ) + } + + # Capture old value before writing + old_value = record[field_name] + old_value_str = str(old_value) + + # Perform the write + record.write({field_name: new_value}) + + # Create audit log (sudo so the log can always be written + # regardless of the user's access on spreadsheet.writeback.log) + log = ( + request.env["spreadsheet.writeback.log"] + .sudo() + .create( + dict( + log_vals_base, + old_value=old_value_str, + status="ok", + ) + ) + ) + + # Post a brief chatter note on the spreadsheet + spreadsheet.sudo().message_post( + body=_( + "Writeback: field %(field)s on " + "%(model)s #%(record_id)d changed " + "from %(old)s to %(new)s.", + field=field_name, + model=model, + record_id=record_id, + old=old_value_str, + new=str(new_value), + ), + subtype_xmlid="mail.mt_note", + ) + + return { + "success": True, + "old_value": old_value_str, + "new_value": str(new_value), + "log_id": log.id, + } + + except Exception as exc: + _logger.exception( + "Writeback error: spreadsheet=%d model=%s record=%d field=%s", + spreadsheet_id, + model, + record_id, + field_name, + ) + # Attempt to write an error log (best effort — use sudo and + # ignore any secondary failure so the route always returns JSON) + try: + request.env["spreadsheet.writeback.log"].sudo().create( + dict( + log_vals_base, + status="error", + error_message=str(exc)[:255], + ) + ) + except Exception: + _logger.exception("Failed to create writeback error log") + + return {"error": str(exc)} diff --git a/spreadsheet_oca/demo/demo_kpi_dashboard.json b/spreadsheet_oca/demo/demo_kpi_dashboard.json new file mode 100644 index 00000000..ba59dfac --- /dev/null +++ b/spreadsheet_oca/demo/demo_kpi_dashboard.json @@ -0,0 +1,85 @@ +{ + "version": 21, + "sheets": [ + { + "id": "dashboard", + "name": "Dashboard", + "colNumber": 26, + "rowNumber": 100, + "rows": {}, + "cols": {}, + "merges": [], + "cells": { + "A1": {"content": "KPI", "style": 1, "border": 1}, + "B1": {"content": "Target", "style": 1, "border": 1}, + "C1": {"content": "Actual", "style": 1, "border": 1}, + "D1": {"content": "Variance", "style": 1, "border": 1}, + "E1": {"content": "Status", "style": 1, "border": 1}, + "A2": {"content": "Cost Per Lead"}, + "B2": {"content": "50", "format": 1}, + "C2": {"content": "42", "format": 1}, + "D2": {"content": "=C2-B2", "format": 1}, + "E2": {"content": "=IF(C2<=B2,\"On Track\",\"Over\")"}, + "A3": {"content": "Revenue Growth"}, + "B3": {"content": "0.15", "format": 2}, + "C3": {"content": "0.12", "format": 2}, + "D3": {"content": "=C3-B3", "format": 2}, + "E3": {"content": "=IF(C3>=B3,\"On Track\",\"Below\")"}, + "A4": {"content": "Customer Churn"}, + "B4": {"content": "0.05", "format": 2}, + "C4": {"content": "0.032", "format": 2}, + "D4": {"content": "=C4-B4", "format": 2}, + "E4": {"content": "=IF(C4<=B4,\"On Track\",\"High\")"}, + "A5": {"content": "Avg Deal Size"}, + "B5": {"content": "25000", "format": 1}, + "C5": {"content": "28500", "format": 1}, + "D5": {"content": "=C5-B5", "format": 1}, + "E5": {"content": "=IF(C5>=B5,\"On Track\",\"Low\")"}, + "A6": {"content": "NPS Score"}, + "B6": {"content": "70"}, + "C6": {"content": "78"}, + "D6": {"content": "=C6-B6"}, + "E6": {"content": "=IF(C6>=B6,\"On Track\",\"Low\")"}, + "A8": {"content": "Last Updated"}, + "B8": {"content": "2026-03-01"}, + "A9": {"content": "Updated By"}, + "B9": {"content": "Admin"} + }, + "conditionalFormats": [], + "figures": [], + "filterTables": [], + "tables": [], + "dataValidationRules": [], + "comments": {}, + "headerGroups": {"ROW": [], "COL": []}, + "areGridLinesVisible": true, + "isVisible": true + } + ], + "settings": {}, + "customTableStyles": {}, + "styles": { + "1": {"bold": true, "align": "center"} + }, + "formats": { + "1": "$#,##0", + "2": "0.00%" + }, + "borders": { + "1": { + "top": ["thin", "#000"], + "bottom": ["thin", "#000"], + "left": ["thin", "#000"], + "right": ["thin", "#000"] + } + }, + "revisionId": "START_REVISION", + "uniqueFigureIds": true, + "odooVersion": 12, + "globalFilters": [], + "pivots": {}, + "pivotNextId": 1, + "lists": {}, + "listNextId": 1, + "chartOdooMenusReferences": {} +} diff --git a/spreadsheet_oca/demo/demo_pivot_dashboard.json b/spreadsheet_oca/demo/demo_pivot_dashboard.json new file mode 100644 index 00000000..2ed2f20e --- /dev/null +++ b/spreadsheet_oca/demo/demo_pivot_dashboard.json @@ -0,0 +1,98 @@ +{ + "version": 21, + "sheets": [ + { + "id": "sheet_partners", + "name": "Partners by Country", + "colNumber": 26, + "rowNumber": 100, + "rows": {}, + "cols": { + "0": {"size": 220}, + "1": {"size": 140}, + "2": {"size": 140}, + "3": {"size": 140} + }, + "merges": [], + "cells": { + "A1": {"content": "=PIVOT(1)"} + }, + "conditionalFormats": [], + "figures": [], + "filterTables": [], + "tables": [], + "dataValidationRules": [], + "comments": {}, + "headerGroups": {"ROW": [], "COL": []}, + "areGridLinesVisible": true, + "isVisible": true + }, + { + "id": "sheet_regions", + "name": "Regions per Country", + "colNumber": 26, + "rowNumber": 100, + "rows": {}, + "cols": { + "0": {"size": 220}, + "1": {"size": 140} + }, + "merges": [], + "cells": { + "A1": {"content": "=PIVOT(2)"} + }, + "conditionalFormats": [], + "figures": [], + "filterTables": [], + "tables": [], + "dataValidationRules": [], + "comments": {}, + "headerGroups": {"ROW": [], "COL": []}, + "areGridLinesVisible": true, + "isVisible": true + } + ], + "settings": {}, + "customTableStyles": {}, + "styles": {}, + "formats": {}, + "borders": {}, + "revisionId": "START_REVISION", + "uniqueFigureIds": true, + "odooVersion": 12, + "globalFilters": [], + "pivots": { + "1": { + "type": "ODOO", + "id": "1", + "formulaId": "1", + "name": "Partners by Country & Type", + "model": "res.partner", + "domain": [["active", "=", true]], + "context": {}, + "measures": [{"id": "__count", "fieldName": "__count"}], + "rows": [{"fieldName": "country_id", "order": "desc"}], + "columns": [{"fieldName": "is_company"}], + "sortedColumn": null, + "fieldMatching": {} + }, + "2": { + "type": "ODOO", + "id": "2", + "formulaId": "2", + "name": "Regions per Country", + "model": "res.country.state", + "domain": [], + "context": {}, + "measures": [{"id": "__count", "fieldName": "__count"}], + "rows": [{"fieldName": "country_id", "order": "desc"}], + "columns": [], + "sortedColumn": null, + "fieldMatching": {} + } + }, + "pivotNextId": 3, + "lists": {}, + "listNextId": 1, + "chartOdooMenusReferences": {} +} diff --git a/spreadsheet_oca/demo/spreadsheet_spreadsheet.xml b/spreadsheet_oca/demo/spreadsheet_spreadsheet.xml index 11222ed5..ca966eb5 100644 --- a/spreadsheet_oca/demo/spreadsheet_spreadsheet.xml +++ b/spreadsheet_oca/demo/spreadsheet_spreadsheet.xml @@ -1,11 +1,125 @@ + + + + Müller GmbH + + + + + Hans Weber + + + + + Dupont SA + + + + + Marie Leclerc + + + + + British Solutions Ltd + + + + + James Clarke + + + + + Tanaka Industries + + + + + Silva Comércio Ltda + + + + + Ana Costa + + + + Patel Technologies Pvt Ltd + + + + + Priya Sharma + + + + + Outback Systems Pty Ltd + + + + + + - Demo spreadsheet + Sales Pipeline Summary + + + + KPI Dashboard + + + + + + + Partner Pivot Dashboard + + + + + + + + + res.partner + + phone + +1 555-0101 + +1 555-0199 + ok + + + + + res.partner + + name + Dupont SA + Dupont SA International + rolled_back + diff --git a/spreadsheet_oca/models/__init__.py b/spreadsheet_oca/models/__init__.py index 3e07c12a..6c0b2e1d 100644 --- a/spreadsheet_oca/models/__init__.py +++ b/spreadsheet_oca/models/__init__.py @@ -1,7 +1,9 @@ from . import ir_model +from . import cell_ref # noqa: F401 — helpers shared by the pivot features from . import spreadsheet_abstract from . import spreadsheet_spreadsheet_tag from . import spreadsheet_spreadsheet from . import spreadsheet_oca_revision from . import ir_websocket from . import spreadsheet_spreadsheet_import_mode +from . import spreadsheet_writeback diff --git a/spreadsheet_oca/models/cell_ref.py b/spreadsheet_oca/models/cell_ref.py new file mode 100644 index 00000000..4e41851b --- /dev/null +++ b/spreadsheet_oca/models/cell_ref.py @@ -0,0 +1,140 @@ +# Copyright 2026 Ledo Enterprises LLC +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +""" +Shared cell-reference helpers for spreadsheet_oca. + +Used by spreadsheet_alert, spreadsheet_scenario, and spreadsheet_input_param +to avoid duplicating cell-address parsing and raw-JSON access logic. +""" + +import re + +# Pre-compiled pattern: column letters + row number (1-based, no zero row). +_CELL_REF_RE = re.compile(r"^([A-Za-z]+)([1-9][0-9]*)$") + + +def _idx_to_cell_address(col_idx, row_idx): + """Convert 0-based (col, row) to cell address like 'A1', 'B3', 'AA12'.""" + col_str = "" + c = col_idx + while True: + col_str = chr(ord("A") + c % 26) + col_str + c = c // 26 - 1 + if c < 0: + break + return f"{col_str}{row_idx + 1}" + + +def parse_cell_ref(ref): + """ + Parse a bare cell reference like 'B3' or 'AA12' into (col_index, row_index). + + Both indices are 0-based to match the o-spreadsheet JSON cell-map format. + Returns (None, None) on invalid input (empty string, zero row, etc.). + """ + m = _CELL_REF_RE.match(ref.strip()) + if not m: + return None, None + col_str, row_str = m.group(1).upper(), m.group(2) + col_idx = 0 + for ch in col_str: + col_idx = col_idx * 26 + (ord(ch) - ord("A") + 1) + col_idx -= 1 # convert to 0-based + row_idx = int(row_str) - 1 # convert to 0-based + return col_idx, row_idx + + +def parse_cell_key(key): + """ + Parse a possibly-qualified cell key into (sheet_name_or_None, col_idx, row_idx). + + Supported formats: + - ``"B3"`` — no sheet qualifier; sheet_name = None + - ``"Sheet1!B3"`` — explicit sheet qualifier + """ + key = key.strip() + if "!" in key: + sheet_part, addr_part = key.split("!", 1) + sheet_name = sheet_part.strip() + else: + sheet_name = None + addr_part = key + col_idx, row_idx = parse_cell_ref(addr_part) + return sheet_name, col_idx, row_idx + + +def _resolve_sheet(sheets, sheet_name=None): + """Return the target sheet dict from a list of sheets. + + If *sheet_name* is given, searches case-insensitively; falls back to the + first sheet if not found. Returns None when *sheets* is empty. + """ + if not sheets: + return None + if sheet_name: + for s in sheets: + if s.get("name", "").lower() == sheet_name.lower(): + return s + return sheets[0] + + +def read_cell_value(spreadsheet_raw, cell_ref, sheet_name=None): + """ + Read the value of a cell from a spreadsheet_raw JSON dict. + + *cell_ref* may be bare (``"B3"``) or sheet-qualified (``"Sheet1!B3"``). + *sheet_name*, when provided, overrides any sheet qualifier embedded in + *cell_ref* and forces lookup in the named sheet (falling back to sheet 0). + + Return value priority: + 1. The cell's evaluated ``"value"`` key (set by o-spreadsheet when the + workbook is saved after formula evaluation in the browser). + 2. The cell's ``"content"`` string (for static / hand-typed cells). + 3. ``None`` when the cell, sheet, or raw JSON is absent. + """ + sheets = (spreadsheet_raw or {}).get("sheets", []) + ref_sheet, col_idx, row_idx = parse_cell_key(cell_ref) + if col_idx is None: + return None + + target_name = sheet_name or ref_sheet + target_sheet = _resolve_sheet(sheets, target_name) + if target_sheet is None: + return None + + cells = target_sheet.get("cells", {}) + cell_addr = _idx_to_cell_address(col_idx, row_idx) + cell_data = cells.get(cell_addr, {}) + if not cell_data: + return None + + value = cell_data.get("value") + if value is None: + value = cell_data.get("content") + return value if value != "" else None + + +def write_cell_content(spreadsheet_raw, cell_ref, value, sheet_name=None): + """ + Write a value into ``cells[row][col]["content"]`` of *spreadsheet_raw* in-place. + + Creates nested dicts as needed. *cell_ref* and *sheet_name* follow the + same conventions as :func:`read_cell_value`. + + Returns the (mutated) *spreadsheet_raw* dict. + """ + sheets = (spreadsheet_raw or {}).get("sheets", []) + ref_sheet, col_idx, row_idx = parse_cell_key(cell_ref) + if col_idx is None: + return spreadsheet_raw + + target_name = sheet_name or ref_sheet + target_sheet = _resolve_sheet(sheets, target_name) + if target_sheet is None: + return spreadsheet_raw + + cells = target_sheet.setdefault("cells", {}) + cell_addr = _idx_to_cell_address(col_idx, row_idx) + cell_data = cells.setdefault(cell_addr, {}) + cell_data["content"] = str(value) if value is not None else "" + return spreadsheet_raw diff --git a/spreadsheet_oca/models/pivot_data.py b/spreadsheet_oca/models/pivot_data.py new file mode 100644 index 00000000..916158cd --- /dev/null +++ b/spreadsheet_oca/models/pivot_data.py @@ -0,0 +1,422 @@ +# Copyright 2026 Ledo Enterprises LLC +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +""" +Server-side pivot data helper. + +Replicates the read_group strategy used by the Odoo web PivotModel +(addons/web/static/src/views/pivot/pivot_model.js) to produce pivot table +data server-side, without executing any JavaScript. + +The JS pivot loads data by: + 1. Computing all row-groupby prefixes ("sections"): + rows=["partner_id","date:month"] → [[], ["partner_id"], + ["partner_id","date:month"]] + 2. Computing all col-groupby prefixes ("sections"): + cols=["stage_id"] → [[], ["stage_id"]] + 3. Taking the cartesian product (row_prefix × col_prefix) for "divisors". + 4. For each divisor [rowPrefix, colPrefix], calling: + read_group(domain, fields=measureSpecs, + groupby=rowPrefix+colPrefix, lazy=False) + +This module replicates that strategy in Python and exposes: + - ``get_pivot_data(model, domain, context, rows, columns, measures)`` + +Rows / columns are lists of dimension dicts: + {"fieldName": "date_order", "granularity": "month"} + {"fieldName": "partner_id"} (no granularity) + +Measures are lists of measure dicts: + {"fieldName": "amount_total", "aggregator": "sum"} + {"fieldName": "__count"} +""" + +import itertools +import logging + +from odoo.exceptions import AccessError + +_logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Helpers mirroring the JS helpers in pivot_model.js +# --------------------------------------------------------------------------- + +DATE_GRANULARITIES = {"day", "week", "month", "quarter", "year"} + +# Shown when a group key is empty (read_group returns False for "not set"). +_NONE_LABEL = "None" + + +def _dimension_to_groupby(dim): + """Convert a dimension dict to an Odoo read_group groupby string. + + {"fieldName": "date_order", "granularity": "month"} → "date_order:month" + {"fieldName": "partner_id"} → "partner_id" + """ + name = dim["fieldName"] + gran = dim.get("granularity") + return f"{name}:{gran}" if gran else name + + +def _sections(lst): + """Return all prefixes of lst including the empty prefix. + + sections(["a", "b", "c"]) → [[], ["a"], ["a", "b"], ["a", "b", "c"]] + + Mirrors the JS ``sections()`` helper. + """ + return [lst[:i] for i in range(len(lst) + 1)] + + +def _measure_to_field_spec(measure): + """Convert a measure dict to a read_group ``fields`` element. + + {"fieldName": "amount_total", "aggregator": "sum"} → "amount_total:sum" + {"fieldName": "__count"} → "__count" + """ + if measure["fieldName"] == "__count": + return "__count" + agg = measure.get("aggregator") or "sum" + return f"{measure['fieldName']}:{agg}" + + +# --------------------------------------------------------------------------- +# Main computation +# --------------------------------------------------------------------------- + + +def get_pivot_data(env, model_name, domain, context, row_dims, col_dims, measures): + """Compute pivot table data using the same read_group strategy as the JS. + + Returns a dict: + { + "fields": {fieldName: {type, string, ...}}, + "groups": [ + { + "rowValues": ["2026-01", ...], # normalised group key values + "colValues": ["Confirmed", ...], + "rowLabels": ["January 2026", ...], # human-readable + "colLabels": ["Confirmed", ...], + "rowGroupBy": ["date_order:month"], + "colGroupBy": ["stage_id"], + "count": 12, + "measures": {"amount_total:sum": 9800.0, ...}, + }, + ... + ], + "rowDimensions": [{"fieldName": ..., "granularity": ...}, ...], + "colDimensions": [{"fieldName": ..., "granularity": ...}, ...], + "measureSpecs": ["amount_total:sum", ...], + "partial": False, # True if any subtotal query failed + "failedGroupBys": [], # the groupbys that failed, when partial + } + + An :class:`~odoo.exceptions.AccessError` propagates rather than degrading + to an empty subtotal, so a permission problem is never silently rendered + as a zero. + """ + Model = env[model_name].with_context(**(context or {})) + + # ── 1. Fields metadata (needed for label resolution) ───────────────── + all_field_names = [d["fieldName"] for d in row_dims + col_dims] + [ + m["fieldName"] for m in measures if m["fieldName"] != "__count" + ] + # fields_get returns {fieldName: {type, string, selection, ...}} + fields_meta = Model.fields_get( + all_field_names, attributes=["type", "string", "selection"] + ) + + # ── 2. Build groupby strings ────────────────────────────────────────── + row_groupbys = [_dimension_to_groupby(d) for d in row_dims] + col_groupbys = [_dimension_to_groupby(d) for d in col_dims] + measure_specs = [_measure_to_field_spec(m) for m in measures] + + # Ensure count is always fetched (JS always adds __count implicitly) + field_specs_with_count = measure_specs + ( + [] if "__count" in measure_specs else ["__count"] + ) + + # ── 3. Compute divisors (cartesian product of all prefixes) ────────── + row_sections = _sections(row_groupbys) + col_sections = _sections(col_groupbys) + divisors = list(itertools.product(row_sections, col_sections)) + + # ── 4. Fire read_group for each divisor ────────────────────────────── + groups = [] + failed_groupbys = [] + for row_prefix, col_prefix in divisors: + groupby = row_prefix + col_prefix + try: + # read_group is still the public API in 18.0 and is what the JS + # pivot uses. It is kept here (rather than _read_group) because + # this code needs the formatted values and __domain that only + # read_group returns. 19.0 deprecates it in favour of + # formatted_read_group — this is the single call site to port. + results = Model.read_group( + domain=domain or [], + fields=field_specs_with_count, + groupby=groupby, + lazy=False, + ) + except AccessError: + # Never mask a permission problem as an empty subtotal. + raise + except Exception: + # One malformed divisor must not lose the whole pivot, but the + # caller has to be able to tell a partial result from a complete + # one — hence ``partial``/``failedGroupBys`` in the return value. + _logger.exception( + "read_group failed for model=%s groupby=%s", model_name, groupby + ) + failed_groupbys.append(groupby) + continue + + for rg in results: + group_entry = { + "rowGroupBy": row_prefix, + "colGroupBy": col_prefix, + "rowValues": _extract_group_values(rg, row_prefix, fields_meta), + "colValues": _extract_group_values(rg, col_prefix, fields_meta), + "rowLabels": _extract_group_labels(rg, row_prefix, fields_meta), + "colLabels": _extract_group_labels(rg, col_prefix, fields_meta), + "count": rg.get("__count", 0), + "measures": _extract_measures(rg, measures, fields_meta), + "domain": rg.get("__domain", []), + } + groups.append(group_entry) + + return { + "fields": fields_meta, + "groups": groups, + "rowDimensions": row_dims, + "colDimensions": col_dims, + "measureSpecs": measure_specs, + "partial": bool(failed_groupbys), + "failedGroupBys": failed_groupbys, + } + + +def _extract_group_values(rg_row, groupby_list, fields_meta): + """Extract normalised group values from a read_group result row. + + Many2one fields return (id, display_name) — we normalise to the id (int). + Date/datetime fields return a formatted string (Odoo already handles + granularity in the groupby key). + """ + values = [] + for gb_spec in groupby_list: + field_name = gb_spec.split(":")[0] + raw = rg_row.get(gb_spec) or rg_row.get(field_name) + if raw is False or raw is None: + values.append(False) + elif isinstance(raw, list | tuple) and len(raw) == 2: + # Many2one: (id, display_name) — store id; JS uses id for grouping + values.append(raw[0]) + else: + values.append(raw) + return values + + +def _extract_group_labels(rg_row, groupby_list, fields_meta): + """Human-readable counterpart of :func:`_extract_group_values`. + + The values are ids because that is what grouping needs, but anything that + renders a group to a human — Chatter summaries, digests, XLSX — needs the + label instead, or a many2one group shows up as a bare database id. + """ + labels = [] + for gb_spec in groupby_list: + field_name = gb_spec.split(":")[0] + raw = rg_row.get(gb_spec) or rg_row.get(field_name) + meta = fields_meta.get(field_name) or {} + if raw is False or raw is None: + labels.append(_NONE_LABEL) + elif isinstance(raw, list | tuple) and len(raw) == 2: + labels.append(raw[1]) # many2one display_name + elif meta.get("type") == "selection": + selection = dict(meta.get("selection") or []) + labels.append(selection.get(raw, raw)) + else: + labels.append(raw) + return [str(label) for label in labels] + + +def _extract_measures(rg_row, measures, fields_meta): + """Extract measure values from a read_group result row.""" + result = {} + for measure in measures: + fname = measure["fieldName"] + agg = measure.get("aggregator") + if fname == "__count": + result["__count"] = rg_row.get("__count", 0) + continue + # read_group key: field_name (no aggregator suffix in result keys) + raw = rg_row.get(fname, 0) + if isinstance(raw, list | tuple): + # Many2one used as measure — count distinct occurrences + raw = 1 if raw else 0 + if raw is False: + raw = 0 + spec_key = f"{fname}:{agg}" if agg else fname + result[spec_key] = raw + return result + + +# --------------------------------------------------------------------------- +# Shared helpers for pivot iteration and HTML rendering +# --------------------------------------------------------------------------- + + +def collect_pivot_summaries(env, spreadsheet_raw, domain_transform=None): + """Iterate over ODOO-type pivots and return fresh data for each. + + Args: + env: Odoo environment. + spreadsheet_raw: dict — the spreadsheet's raw JSON data. + domain_transform: optional callable(domain) -> domain, applied to each + pivot's domain before querying (e.g. parameter substitution). + + Returns: + A tuple ``(summaries, failed_names)`` where *summaries* is a list of + ``{"name": ..., "model": ..., "result": ...}`` dicts, and + *failed_names* is a list of pivot display names that could not be loaded. + """ + pivots = spreadsheet_raw.get("pivots", {}) + summaries = [] + failed_names = [] + for pivot_id, pivot_def in pivots.items(): + if pivot_def.get("type") != "ODOO": + continue + model_name = pivot_def.get("model") + pivot_name = pivot_def.get("name") or f"Pivot #{pivot_id}" + if not model_name or model_name not in env: + _logger.warning( + "collect_pivot_summaries: unknown model %r — skipping pivot %s", + model_name, + pivot_id, + ) + failed_names.append(pivot_name) + continue + try: + domain = pivot_def.get("domain", []) + if domain_transform: + domain = domain_transform(domain) + result = get_pivot_data( + env, + model_name, + domain, + pivot_def.get("context", {}), + pivot_def.get("rows", []), + pivot_def.get("columns", []), + pivot_def.get("measures", []), + ) + summaries.append( + { + "name": pivot_name, + "model": model_name, + "result": result, + } + ) + except Exception: + _logger.exception( + "collect_pivot_summaries: failed to compute pivot %s", + pivot_id, + ) + failed_names.append(pivot_name) + return summaries, failed_names + + +def render_pivot_table_html(summary, max_rows=10): + """Render a single pivot summary as an HTML table string. + + Args: + summary: dict with keys ``"name"``, ``"model"``, ``"result"`` + (as returned by ``collect_pivot_summaries``). + max_rows: maximum number of detail rows to include before truncating. + + Returns: + str — HTML fragment for the pivot table. + """ + result = summary["result"] + name = summary["name"] + model = summary["model"] + parts = [] + + parts.append( + f'

{name}' + f' ({model})

' + ) + + row_dims = result.get("rowDimensions", []) + groups = result.get("groups", []) + + # Grand total row + grand_totals = [ + g for g in groups if g["rowGroupBy"] == [] and g["colGroupBy"] == [] + ] + if grand_totals: + gt = grand_totals[0] + count = gt.get("count", 0) + parts.append( + f'

Total records: {count}

' + ) + for key, val in gt.get("measures", {}).items(): + if key != "__count" and val is not None: + parts.append( + f'

{key}: {val}

' + ) + + # Row breakdown table + if row_dims: + row_gb = [d["fieldName"] for d in row_dims] + row_groups = [ + g for g in groups if g["rowGroupBy"] == row_gb and g["colGroupBy"] == [] + ] + if row_groups: + measure_keys = [ + k for k in (row_groups[0].get("measures") or {}) if k != "__count" + ] + headers = ["Group"] + measure_keys + ["Count"] + parts.append( + '' + ) + parts.append("") + for h in headers: + parts.append( + '' + ) + parts.append("") + for g in row_groups[:max_rows]: + label = ", ".join( + g.get("rowLabels") or [str(v) for v in g["rowValues"]] + ) + parts.append("") + parts.append( + f'' + ) + for mk in measure_keys: + val = g.get("measures", {}).get(mk, "") + parts.append( + '' + ) + parts.append( + ''.format(g.get("count", "")) + ) + parts.append("") + if len(row_groups) > max_rows: + colspan = len(headers) + extra = len(row_groups) - max_rows + more_text = f"and {extra} more rows" + parts.append( + f'" + ) + parts.append("
{h}
{label}{val}{}
' + f"… {more_text}
") + + return "".join(parts) diff --git a/spreadsheet_oca/models/spreadsheet_spreadsheet.py b/spreadsheet_oca/models/spreadsheet_spreadsheet.py index 55a9ae9f..bc8b949a 100644 --- a/spreadsheet_oca/models/spreadsheet_spreadsheet.py +++ b/spreadsheet_oca/models/spreadsheet_spreadsheet.py @@ -6,6 +6,9 @@ from io import BytesIO from odoo import _, api, fields, models +from odoo.exceptions import UserError + +from .spreadsheet_writeback import coerce_old_value class SpreadsheetSpreadsheet(models.Model): @@ -56,10 +59,134 @@ class SpreadsheetSpreadsheet(models.Model): string="Tags", comodel_name="spreadsheet.spreadsheet.tag" ) + # ── DRY helper for read_group-based count fields ───────────────────────── + + def _compute_related_count(self, comodel, field_name, extra_domain=None): + """Compute a count field by grouping *comodel* on ``spreadsheet_id``. + + By default the domain filters on ``active=True``; pass *extra_domain* + to override (e.g. ``[("status", "!=", "error")]`` for writeback logs). + """ + domain = [("spreadsheet_id", "in", self.ids)] + if extra_domain is not None: + domain += extra_domain + else: + domain.append(("active", "=", True)) + counts = self.env[comodel].read_group( + domain, ["spreadsheet_id"], ["spreadsheet_id"] + ) + count_map = {c["spreadsheet_id"][0]: c["spreadsheet_id_count"] for c in counts} + for rec in self: + rec[field_name] = count_map.get(rec.id, 0) + @api.depends("name") def _compute_filename(self): for record in self: - record.filename = "%s.json" % (self.name or _("Unnamed")) + record.filename = f"{record.name or _('Unnamed')}.json" + + # ── Writeback ────────────────────────────────────────────────────────── + writeback_enabled = fields.Boolean( + default=False, + tracking=True, + help=( + "Allow users to write Odoo record values directly from this " + "spreadsheet's List views. Each change is logged and reversible." + ), + ) + writeback_log_count = fields.Integer( + compute="_compute_writeback_log_count", + ) + + def _compute_writeback_log_count(self): + self._compute_related_count( + "spreadsheet.writeback.log", + "writeback_log_count", + extra_domain=[("status", "!=", "error")], + ) + + def action_open_writeback_log(self): + self.ensure_one() + return { + "type": "ir.actions.act_window", + "name": _("Writeback Log"), + "res_model": "spreadsheet.writeback.log", + "view_mode": "list,form", + "domain": [("spreadsheet_id", "=", self.id)], + "context": {"default_spreadsheet_id": self.id}, + } + + @api.model + def action_rollback_writeback(self, log_id): + log = self.env["spreadsheet.writeback.log"].sudo().browse(log_id) + if not log.exists(): + raise UserError( + _("Writeback log entry %(log_id)d not found.", log_id=log_id) + ) + + if log.old_value is False or log.old_value is None: + raise UserError( + _( + "Cannot roll back log %(log_id)d: previous value is unknown.", + log_id=log_id, + ) + ) + + if log.res_model not in self.env: + raise UserError( + _( + "Cannot roll back log %(log_id)d:" + " model %(model)r is not available.", + log_id=log_id, + model=log.res_model, + ) + ) + + # Access check: ensure the calling user has write permission on the + # target model/record before we escalate to sudo(). + self.env[log.res_model].check_access("write") + record = self.env[log.res_model].browse(log.record_id) + if not record.exists(): + raise UserError( + _( + "Cannot roll back log %(log_id)d: record " + "%(model)s(%(record_id)d) no longer exists.", + log_id=log_id, + model=log.res_model, + record_id=log.record_id, + ) + ) + record.check_access("write") + + field = record._fields.get(log.field_name) + if field is None: + raise UserError( + _( + "Cannot roll back log %(log_id)d: field %(field)s no longer" + " exists on %(model)s.", + log_id=log_id, + field=log.field_name, + model=log.res_model, + ) + ) + record.write({log.field_name: coerce_old_value(field, log.old_value)}) + log.write({"status": "rolled_back"}) + + spreadsheet = log.spreadsheet_id.sudo() + spreadsheet.message_post( + body=_( + "Writeback rolled back: field %(field)s on " + "%(model)s #%(record_id)d restored to " + "%(old_value)s (was %(new_value)s).", + field=log.field_name, + model=log.res_model, + record_id=log.record_id, + old_value=log.old_value, + new_value=log.new_value, + ), + subtype_xmlid="mail.mt_note", + ) + + return True def create_document_from_attachment(self, attachment_ids): attachments = self.env["ir.attachment"].browse(attachment_ids) diff --git a/spreadsheet_oca/models/spreadsheet_writeback.py b/spreadsheet_oca/models/spreadsheet_writeback.py new file mode 100644 index 00000000..9192318e --- /dev/null +++ b/spreadsheet_oca/models/spreadsheet_writeback.py @@ -0,0 +1,111 @@ +# Copyright 2025 Ledo Enterprises LLC +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +""" +Cell writeback: edit list cells to update Odoo records. + +Allows users to write Odoo record field values directly from a +spreadsheet's List view. Each cell edit in the browser posts to +/spreadsheet/writeback (controllers/spreadsheet_writeback.py) which +calls the target model's write() and records an audit log entry here. + +JavaScript integration (not yet implemented): the JS list cell-edit +handler will POST to /spreadsheet/writeback with the spreadsheet_id, +model, record_id, field_name and new_value. The controller returns a +JSON dict with {success, old_value, new_value, log_id} or {error}. + +old_value is stored as text, so writeback is restricted to the field +types whose value survives a str() round-trip — see WRITEBACK_FIELD_TYPES. +Relational and binary fields are rejected at the controller rather than +stored in a form that cannot be rolled back. + +Rolling a value back goes through coerce_old_value(), which converts the +stored text back to the field's own type. Booleans matter most here: +str(False) is "False", and writing that string straight back would set the +field to True, because a non-empty string is truthy. +""" + +import logging + +from odoo import fields, models + +_logger = logging.getLogger(__name__) + +# Field types whose value survives being stored as text and written back. +# Relational and binary types are deliberately absent: str(recordset) yields +# e.g. "res.partner(7,)", which cannot be written back to the field. +WRITEBACK_FIELD_TYPES = { + "char", + "text", + "integer", + "float", + "monetary", + "boolean", + "date", + "datetime", + "selection", +} + + +def coerce_old_value(field, stored): + """Convert a stored old_value string back to *field*'s own type. + + Odoo coerces text to date, datetime, selection and the numeric types on + write, but not to boolean — ``bool("False")`` is True — so booleans are + converted explicitly. + """ + if field.type == "boolean": + return str(stored).strip().lower() in ("true", "1") + if field.type == "integer": + return int(float(stored)) + if field.type in ("float", "monetary"): + return float(stored) + return stored + + +class SpreadsheetWritebackLog(models.Model): + _name = "spreadsheet.writeback.log" + _description = "Spreadsheet Writeback Audit Log" + _order = "writeback_at desc" + + spreadsheet_id = fields.Many2one( + "spreadsheet.spreadsheet", + required=True, + ondelete="cascade", + index=True, + string="Spreadsheet", + ) + res_model = fields.Char(required=True, string="Model") + record_id = fields.Integer(required=True) + field_name = fields.Char(required=True, string="Field") + old_value = fields.Char(string="Previous Value") + new_value = fields.Char(required=True) + user_id = fields.Many2one( + "res.users", + default=lambda self: self.env.user, + readonly=True, + ) + writeback_at = fields.Datetime( + default=fields.Datetime.now, + readonly=True, + string="Written At", + ) + status = fields.Selection( + [ + ("ok", "Success"), + ("error", "Error"), + ("rolled_back", "Rolled Back"), + ], + default="ok", + ) + error_message = fields.Char(string="Error") + + def action_rollback(self): + """ + Roll back this log entry by restoring old_value to the target record. + + Called from the form view "Roll Back" button (type="object"). + Delegates to SpreadsheetSpreadsheet.action_rollback_writeback so the + rollback logic lives in one place. + """ + self.ensure_one() + self.spreadsheet_id.action_rollback_writeback(self.id) diff --git a/spreadsheet_oca/readme/DEVELOP.md b/spreadsheet_oca/readme/DEVELOP.md index 0afb4a08..4c82d718 100644 --- a/spreadsheet_oca/readme/DEVELOP.md +++ b/spreadsheet_oca/readme/DEVELOP.md @@ -1,3 +1,30 @@ If you want to develop custom business functions, you can add others, based on the file + +## Server-side pivot computation + +`models/pivot_data.py` reproduces the strategy of the JavaScript +`PivotModel` (`web/static/src/views/pivot/pivot_model.js`) in Python, so +pivot data can be produced without a browser — for scheduled refreshes, +emails or exports. + +For row groupbys `R` and column groupbys `C` it queries every *prefix* +pair of `R` and `C` (the "divisors"), which is what produces the subtotal +rows a pivot needs. That means **(len(R)+1) x (len(C)+1) queries** per +pivot, matching what the JS does. + +The useful entry points are: + +- `get_pivot_data(env, model, domain, context, rows, cols, measures)` — + one pivot. Groups come back with `rowValues`/`colValues` (ids, for + grouping) and `rowLabels`/`colLabels` (display names, for rendering). + If a subtotal query fails the result is still returned, but flagged + with `partial` and `failedGroupBys`; an `AccessError` is re-raised + rather than degraded into an empty group. +- `collect_pivot_summaries(env, spreadsheet_raw, domain_transform=None)` — + every Odoo pivot in a workbook. +- `render_pivot_table_html(summary, max_rows=10)` — one summary as HTML. + +`models/cell_ref.py` holds the A1-notation helpers used to read and write +individual cells of the `spreadsheet_raw` JSON. diff --git a/spreadsheet_oca/readme/USAGE.md b/spreadsheet_oca/readme/USAGE.md index 935707dc..f6aad5cc 100644 --- a/spreadsheet_oca/readme/USAGE.md +++ b/spreadsheet_oca/readme/USAGE.md @@ -37,3 +37,18 @@ > - `ODOO.ACCOUNT.GROUP(type)`: Returns the account ids of a given > group where type should be a value of the `account_type` field of > `account.account` model. (`income`, `asset_receivable`, etc.) + +## **Write spreadsheet edits back to Odoo records** + +Tick **Writeback Enabled** on a spreadsheet to let edits made in a List cell +update the underlying Odoo record. + +Every write is checked against the editing user's own access rights — the +spreadsheet must be readable by them, and the target record writable — and each +change is recorded in **Writeback Logs** with the previous value, so it can be +rolled back from the log's **Roll Back** button. + +Only simple value fields can be written back: text, numbers, dates, booleans +and selections. Relational fields (Many2one, tags, attachments) are rejected, +because the previous value is stored as text and a related record cannot be +restored from its text form. diff --git a/spreadsheet_oca/security/ir.model.access.csv b/spreadsheet_oca/security/ir.model.access.csv index 1898b166..d619fca7 100644 --- a/spreadsheet_oca/security/ir.model.access.csv +++ b/spreadsheet_oca/security/ir.model.access.csv @@ -6,3 +6,5 @@ access_spreadsheet_import_mode,access_spreadsheet_oca_revision,model_spreadsheet access_spreadsheet_select_row_number,access_spreadsheet_select_row_number,model_spreadsheet_select_row_number,base.group_user,1,1,1,1 access_spreadsheet_spreadsheet_tag,access_spreadsheet_spreadsheet_tag,model_spreadsheet_spreadsheet_tag,spreadsheet_oca.group_user,1,0,0,0 access_spreadsheet_spreadsheet_manager_tag,access_spreadsheet_spreadsheet_manager_tag,model_spreadsheet_spreadsheet_tag,spreadsheet_oca.group_manager,1,1,1,1 +access_spreadsheet_writeback_log_user,access_spreadsheet_writeback_log_user,model_spreadsheet_writeback_log,base.group_user,1,0,0,0 +access_spreadsheet_writeback_log_manager,access_spreadsheet_writeback_log_manager,model_spreadsheet_writeback_log,spreadsheet_oca.group_manager,1,1,1,1 diff --git a/spreadsheet_oca/security/security.xml b/spreadsheet_oca/security/security.xml index aa94100a..783d0edf 100644 --- a/spreadsheet_oca/security/security.xml +++ b/spreadsheet_oca/security/security.xml @@ -62,4 +62,25 @@ [('group_ids','in', user.groups_id.ids)] + + + + + Writeback Log: follow spreadsheet access + + + [ + '|', '|', '|', + ('spreadsheet_id.owner_id', '=', user.id), + ('spreadsheet_id.contributor_ids', '=', user.id), + ('spreadsheet_id.contributor_group_ids', 'in', user.groups_id.ids), + ('spreadsheet_id.reader_ids', '=', user.id), + ] + + + Writeback Log: manager full access + + + [(1, '=', 1)] + diff --git a/spreadsheet_oca/static/description/index.html b/spreadsheet_oca/static/description/index.html index 739577db..46936a35 100644 --- a/spreadsheet_oca/static/description/index.html +++ b/spreadsheet_oca/static/description/index.html @@ -3,7 +3,7 @@ -README.rst +Spreadsheet Oca -
+
+

Spreadsheet Oca

- - -Odoo Community Association - -
-

Spreadsheet Oca

-

Beta License: AGPL-3 OCA/spreadsheet Translate me on Weblate Try me on Runboat

+

Beta License: AGPL-3 OCA/spreadsheet Translate me on Weblate Try me on Runboat

This module adds a functionality for adding and editing Spreadsheets using Odoo CE.

It is an alternative to the proprietary module spreadsheet_edition @@ -397,9 +392,9 @@

Spreadsheet Oca

-

Usage

+

Usage

-

Create a new spreadsheet

+

Create a new spreadsheet

-

Development

+

Development

If you want to develop custom business functions, you can add others, based on the file https://github.com/odoo/odoo/blob/16.0/addons/spreadsheet_account/static/src/accounting_functions.js

-

Bug Tracker

+

Bug Tracker

Bugs are tracked on GitHub Issues. In case of trouble, please check there if your issue has already been reported. If you spotted it first, help us to smash it by providing a detailed and welcomed @@ -461,15 +456,15 @@

Bug Tracker

Do not contact contributors directly about support or help with technical issues.

-

Credits

+

Credits

-

Authors

+

Authors

  • CreuBlanca
-

Contributors

+

Contributors

-

Maintainers

+

Maintainers

This module is maintained by the OCA.

Odoo Community Association @@ -500,6 +495,5 @@

Maintainers

-
diff --git a/spreadsheet_oca/tests/__init__.py b/spreadsheet_oca/tests/__init__.py new file mode 100644 index 00000000..2e0432c8 --- /dev/null +++ b/spreadsheet_oca/tests/__init__.py @@ -0,0 +1,3 @@ +from . import test_spreadsheet +from . import test_pivot_data +from . import test_writeback diff --git a/spreadsheet_oca/tests/test_pivot_data.py b/spreadsheet_oca/tests/test_pivot_data.py new file mode 100644 index 00000000..b1341934 --- /dev/null +++ b/spreadsheet_oca/tests/test_pivot_data.py @@ -0,0 +1,198 @@ +# Copyright 2026 Ledo Enterprises LLC +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +""" +Tests for server-side pivot data computation. + +These tests verify that get_pivot_data() produces the same grouping +structure that the Odoo JS PivotModel would produce via read_group. + +Run against odoo_test (which has sale, account installed): + docker exec -i odoo-prod odoo test -d odoo_test \ + --test-tags spreadsheet_oca.TestPivotData --stop-after-init +""" + +from odoo.tests import TransactionCase + +from ..models.pivot_data import _dimension_to_groupby, _sections, get_pivot_data + + +class TestPivotDataHelpers(TransactionCase): + """Unit tests for the pure-Python helpers (no DB needed).""" + + def test_sections_empty(self): + self.assertEqual(_sections([]), [[]]) + + def test_sections_one(self): + self.assertEqual(_sections(["a"]), [[], ["a"]]) + + def test_sections_two(self): + self.assertEqual(_sections(["a", "b"]), [[], ["a"], ["a", "b"]]) + + def test_dimension_no_granularity(self): + self.assertEqual( + _dimension_to_groupby({"fieldName": "partner_id"}), "partner_id" + ) + + def test_dimension_with_granularity(self): + self.assertEqual( + _dimension_to_groupby({"fieldName": "date_order", "granularity": "month"}), + "date_order:month", + ) + + +class TestPivotData(TransactionCase): + """Integration tests using res.partner (always available, no demo needed).""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + # Create a handful of partners in different countries to group by + cls.country_be = cls.env.ref("base.be") + cls.country_us = cls.env.ref("base.us") + cls.partners = cls.env["res.partner"].create( + [ + {"name": "Alpha", "country_id": cls.country_be.id, "is_company": True}, + {"name": "Beta", "country_id": cls.country_be.id, "is_company": True}, + {"name": "Gamma", "country_id": cls.country_us.id, "is_company": True}, + {"name": "Delta", "country_id": cls.country_us.id, "is_company": False}, + ] + ) + cls.domain = [("id", "in", cls.partners.ids)] + + # ── Helpers ───────────────────────────────────────────────────────────── + + def _run(self, row_dims, col_dims, measures): + return get_pivot_data( + self.env, + "res.partner", + self.domain, + {}, + row_dims, + col_dims, + measures, + ) + + def _groups_for(self, result, row_prefix, col_prefix): + """Return groups matching the given row/col groupby prefix.""" + return [ + g + for g in result["groups"] + if g["rowGroupBy"] == row_prefix and g["colGroupBy"] == col_prefix + ] + + # ── Grand-total (no groupby) ──────────────────────────────────────────── + + def test_grand_total_count(self): + """With no dims, one group with count = number of partners.""" + result = self._run([], [], [{"fieldName": "__count"}]) + groups = self._groups_for(result, [], []) + self.assertEqual(len(groups), 1) + self.assertEqual(groups[0]["count"], 4) + + # ── Single row groupby ────────────────────────────────────────────────── + + def test_row_groupby_country(self): + """Row groupby country_id → one group per country + grand total.""" + row_dims = [{"fieldName": "country_id"}] + result = self._run(row_dims, [], [{"fieldName": "__count"}]) + + # Grand total (rowGroupBy=[], colGroupBy=[]) + totals = self._groups_for(result, [], []) + self.assertEqual(len(totals), 1) + self.assertEqual(totals[0]["count"], 4) + + # Per-country groups (rowGroupBy=["country_id"], colGroupBy=[]) + country_groups = self._groups_for(result, ["country_id"], []) + self.assertEqual(len(country_groups), 2) + counts_by_country = {g["rowValues"][0]: g["count"] for g in country_groups} + self.assertEqual(counts_by_country[self.country_be.id], 2) + self.assertEqual(counts_by_country[self.country_us.id], 2) + + def test_many2one_groups_carry_display_labels(self): + """A many2one group must expose its name, not just its database id. + + rowValues stays an id because grouping needs it, but anything that + renders a group for a human reads rowLabels — without it a Chatter + summary shows "13" where it should show "Belgium". + """ + result = self._run( + [{"fieldName": "country_id"}], [], [{"fieldName": "__count"}] + ) + country_groups = self._groups_for(result, ["country_id"], []) + + labels = {g["rowLabels"][0] for g in country_groups} + self.assertEqual( + labels, {self.country_be.display_name, self.country_us.display_name} + ) + for group in country_groups: + self.assertIsInstance(group["rowValues"][0], int) + + # ── Row + col groupby ─────────────────────────────────────────────────── + + def test_row_and_col_groupby(self): + """Row=country_id, Col=is_company → 2×2 cell values.""" + row_dims = [{"fieldName": "country_id"}] + col_dims = [{"fieldName": "is_company"}] + result = self._run(row_dims, col_dims, [{"fieldName": "__count"}]) + + # Divisors: ([], []) ([], [is_company]) + # ([country_id], []) ([country_id], [is_company]) + # → 4 divisors, each producing N read_group rows + divisor_keys = { + (tuple(g["rowGroupBy"]), tuple(g["colGroupBy"])) for g in result["groups"] + } + self.assertIn(((), ()), divisor_keys) + self.assertIn(((), ("is_company",)), divisor_keys) + self.assertIn((("country_id",), ()), divisor_keys) + self.assertIn((("country_id",), ("is_company",)), divisor_keys) + + # BE / is_company=True → Alpha + Beta = 2 + cell_groups = self._groups_for(result, ["country_id"], ["is_company"]) + be_company = [ + g + for g in cell_groups + if g["rowValues"] == [self.country_be.id] and g["colValues"] == [True] + ] + self.assertEqual(len(be_company), 1) + self.assertEqual(be_company[0]["count"], 2) + + # US / is_company=False → Delta = 1 + us_individual = [ + g + for g in cell_groups + if g["rowValues"] == [self.country_us.id] and g["colValues"] == [False] + ] + self.assertEqual(len(us_individual), 1) + self.assertEqual(us_individual[0]["count"], 1) + + # ── Return structure ──────────────────────────────────────────────────── + + def test_return_fields_metadata(self): + """Result includes fields metadata for all used fields.""" + row_dims = [{"fieldName": "country_id"}] + result = self._run(row_dims, [], [{"fieldName": "__count"}]) + self.assertIn("country_id", result["fields"]) + self.assertEqual(result["fields"]["country_id"]["type"], "many2one") + + def test_return_dimensions_and_specs(self): + """Result echoes back row/col dims and measure specs.""" + row_dims = [{"fieldName": "country_id"}] + measures = [{"fieldName": "__count"}] + result = self._run(row_dims, [], measures) + self.assertEqual(result["rowDimensions"], row_dims) + self.assertEqual(result["colDimensions"], []) + self.assertEqual(result["measureSpecs"], ["__count"]) + + # ── Domain filtering ──────────────────────────────────────────────────── + + def test_domain_filters_correctly(self): + """Domain restricts records — only BE partners.""" + be_domain = [ + ("id", "in", self.partners.ids), + ("country_id", "=", self.country_be.id), + ] + result = get_pivot_data( + self.env, "res.partner", be_domain, {}, [], [], [{"fieldName": "__count"}] + ) + totals = self._groups_for(result, [], []) + self.assertEqual(totals[0]["count"], 2) diff --git a/spreadsheet_oca/tests/test_spreadsheet.py b/spreadsheet_oca/tests/test_spreadsheet.py new file mode 100644 index 00000000..f27e7196 --- /dev/null +++ b/spreadsheet_oca/tests/test_spreadsheet.py @@ -0,0 +1,28 @@ +# Copyright 2026 Ledo Enterprises LLC +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo.tests.common import TransactionCase + + +class TestSpreadsheetFilename(TransactionCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.Spreadsheet = cls.env["spreadsheet.spreadsheet"] + + def test_filename_single_record(self): + spreadsheet = self.Spreadsheet.create({"name": "Budget"}) + self.assertEqual(spreadsheet.filename, "Budget.json") + + def test_filename_is_computed_per_record(self): + """Each record must get its own name, not the first record's. + + _compute_filename used to read ``self.name`` while iterating over + ``self``, which raises a singleton error on a multi-record set. + """ + alpha = self.Spreadsheet.create({"name": "Alpha"}) + beta = self.Spreadsheet.create({"name": "Beta"}) + + both = alpha | beta + + self.assertEqual(both.mapped("filename"), ["Alpha.json", "Beta.json"]) diff --git a/spreadsheet_oca/tests/test_writeback.py b/spreadsheet_oca/tests/test_writeback.py new file mode 100644 index 00000000..7db5f283 --- /dev/null +++ b/spreadsheet_oca/tests/test_writeback.py @@ -0,0 +1,445 @@ +# Copyright 2025 Ledo Enterprises LLC +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +""" +Tests for cell writeback: edit list cells to update Odoo records. + +Controller tests exercise the model-level logic directly rather than +going through the HTTP stack (no test HTTP client is needed for unit +tests in Odoo's TransactionCase framework). + +The controller's writeback() method is tested indirectly by calling +the underlying model operations; the controller integration is +implicitly covered by the fact that the controller delegates entirely +to env[model].write() and env['spreadsheet.writeback.log'].create(), +both of which are exercised here. +""" + +import logging +from unittest.mock import patch + +from odoo.exceptions import AccessError +from odoo.tests import TransactionCase + +_logger = logging.getLogger(__name__) + + +class TestWriteback(TransactionCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + + # Create a spreadsheet with writeback enabled + cls.spreadsheet = cls.env["spreadsheet.spreadsheet"].create( + { + "name": "Writeback Test Spreadsheet", + "writeback_enabled": True, + } + ) + + # Create a spreadsheet with writeback disabled for negative tests + cls.spreadsheet_off = cls.env["spreadsheet.spreadsheet"].create( + { + "name": "Writeback Disabled Spreadsheet", + "writeback_enabled": False, + } + ) + + # Use res.partner as the target model — always available in CE + cls.partner = cls.env["res.partner"].create( + {"name": "Writeback Test Partner", "email": "wb@test.example"} + ) + + # ── Helper ──────────────────────────────────────────────────────────────── + + def _simulate_writeback( + self, + spreadsheet, + model, + record_id, + field_name, + new_value, + env=None, + ): + """ + Simulate the controller's writeback logic using the ORM directly, + mirroring what SpreadsheetWriteback.writeback() does. + + Returns the same dict the controller would return: either + {'success': True, 'old_value': ..., 'new_value': ..., 'log_id': ...} + or {'error': ''}. + """ + if env is None: + env = self.env + + log_vals_base = { + "spreadsheet_id": spreadsheet.id, + "res_model": model, + "record_id": record_id, + "field_name": field_name, + "new_value": str(new_value), + } + + try: + if not spreadsheet.exists(): + return {"error": "Spreadsheet not found."} + + if not spreadsheet.writeback_enabled: + return {"error": "Writeback not enabled for this spreadsheet."} + + try: + spreadsheet.check_access("read") + except AccessError: + return {"error": "Access denied to spreadsheet."} + + if model not in env: + return {"error": f"Model {model!r} is not available."} + + record = env[model].browse(record_id) + if not record.exists(): + return {"error": f"Record {model}({record_id}) not found."} + + try: + record.check_access("write") + except AccessError: + return { + "error": "Access denied: you do not have" + " write access on this record." + } + + old_value = record[field_name] + old_value_str = str(old_value) + + record.write({field_name: new_value}) + + log = ( + env["spreadsheet.writeback.log"] + .sudo() + .create( + dict( + log_vals_base, + old_value=old_value_str, + status="ok", + ) + ) + ) + + spreadsheet.sudo().message_post( + body=( + f"Writeback: field {field_name} on " + f"{model} #{record_id} changed " + f"from {old_value_str}" + f" to {new_value}." + ), + subtype_xmlid="mail.mt_note", + ) + + return { + "success": True, + "old_value": old_value_str, + "new_value": str(new_value), + "log_id": log.id, + } + + except Exception as exc: + try: + env["spreadsheet.writeback.log"].sudo().create( + dict( + log_vals_base, + status="error", + error_message=str(exc)[:255], + ) + ) + except Exception: + _logger.debug("Failed to create writeback error log entry") + return {"error": str(exc)} + + # ── test_writeback_disabled_returns_error ────────────────────────────────── + + def test_writeback_disabled_returns_error(self): + """Controller returns an error dict when writeback_enabled is False.""" + result = self._simulate_writeback( + self.spreadsheet_off, + "res.partner", + self.partner.id, + "name", + "Should Not Happen", + ) + self.assertIn("error", result) + self.assertNotIn("success", result) + self.assertIn("not enabled", result["error"].lower()) + + # ── test_writeback_creates_log ───────────────────────────────────────────── + + def test_writeback_creates_log(self): + """A successful writeback creates a spreadsheet.writeback.log record.""" + log_count_before = self.env["spreadsheet.writeback.log"].search_count( + [("spreadsheet_id", "=", self.spreadsheet.id)] + ) + result = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + "Writeback Log Test", + ) + self.assertTrue(result.get("success"), result) + log_count_after = self.env["spreadsheet.writeback.log"].search_count( + [("spreadsheet_id", "=", self.spreadsheet.id)] + ) + self.assertGreater(log_count_after, log_count_before) + + # ── test_writeback_updates_record ────────────────────────────────────────── + + def test_writeback_updates_record(self): + """The target record's field is actually changed after a writeback.""" + new_name = "Updated By Writeback" + result = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + new_name, + ) + self.assertTrue(result.get("success"), result) + self.partner.invalidate_recordset() + self.assertEqual(self.partner.name, new_name) + + # ── test_writeback_log_contains_old_value ───────────────────────────────── + + def test_writeback_log_contains_old_value(self): + """The log entry captures the old value before the write.""" + # Set a known starting name + self.partner.write({"name": "Known Old Name"}) + result = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + "New Name After Writeback", + ) + self.assertTrue(result.get("success"), result) + self.assertEqual(result["old_value"], "Known Old Name") + + log = self.env["spreadsheet.writeback.log"].browse(result["log_id"]) + self.assertEqual(log.old_value, "Known Old Name") + self.assertEqual(log.field_name, "name") + self.assertEqual(log.status, "ok") + + # ── test_writeback_access_denied ────────────────────────────────────────── + + def test_writeback_access_denied(self): + """ + A user without write access on the target record gets an error dict. + + We patch check_access on the record to raise AccessError, simulating + a restricted user without going through the full ir.rule machinery. + """ + # Create the record as admin + protected_partner = ( + self.env["res.partner"].sudo().create({"name": "Protected Partner"}) + ) + + # Patch check_access to always raise AccessError for this test + with patch.object( + type(protected_partner), + "check_access", + side_effect=AccessError("Access denied"), + ): + result = self._simulate_writeback( + self.spreadsheet, + "res.partner", + protected_partner.id, + "name", + "Should Fail", + ) + + self.assertIn("error", result) + self.assertIn("access denied", result["error"].lower()) + + # ── test_rollback_restores_value ─────────────────────────────────────────── + + def test_rollback_restores_value(self): + """action_rollback_writeback restores the old field value.""" + original_name = "Pre-Rollback Name" + self.partner.write({"name": original_name}) + + result = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + "Changed Name", + ) + self.assertTrue(result.get("success"), result) + self.partner.invalidate_recordset() + self.assertEqual(self.partner.name, "Changed Name") + + # Roll back via the spreadsheet model method + self.spreadsheet.action_rollback_writeback(result["log_id"]) + + self.partner.invalidate_recordset() + self.assertEqual(self.partner.name, original_name) + + # ── test_rollback_marks_log_rolled_back ─────────────────────────────────── + + def test_rollback_marks_log_rolled_back(self): + """The log entry status changes to 'rolled_back' after a rollback.""" + self.partner.write({"name": "Before Rollback Mark"}) + result = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + "After Write", + ) + self.assertTrue(result.get("success"), result) + + log = self.env["spreadsheet.writeback.log"].browse(result["log_id"]) + self.assertEqual(log.status, "ok") + + self.spreadsheet.action_rollback_writeback(result["log_id"]) + log.invalidate_recordset() + self.assertEqual(log.status, "rolled_back") + + # ── test_smart_button_count ──────────────────────────────────────────────── + + def test_smart_button_count(self): + """writeback_log_count reflects successful log entries.""" + self.spreadsheet.invalidate_recordset() + count_before = self.spreadsheet.writeback_log_count + + # Write once + self.partner.write({"name": "Count Test 1"}) + r1 = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + "Count Test 2", + ) + self.assertTrue(r1.get("success"), r1) + + # Write again + r2 = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + "Count Test 3", + ) + self.assertTrue(r2.get("success"), r2) + + self.spreadsheet.invalidate_recordset() + self.assertEqual(self.spreadsheet.writeback_log_count, count_before + 2) + + # ── test_writeback_posts_chatter_message ────────────────────────────────── + + def test_writeback_posts_chatter_message(self): + """A chatter note appears on the spreadsheet after a writeback.""" + self.partner.write({"name": "Chatter Test Start"}) + msg_count_before = self.env["mail.message"].search_count( + [ + ("res_id", "=", self.spreadsheet.id), + ("model", "=", "spreadsheet.spreadsheet"), + ] + ) + result = self._simulate_writeback( + self.spreadsheet, + "res.partner", + self.partner.id, + "name", + "Chatter Test End", + ) + self.assertTrue(result.get("success"), result) + + msg_count_after = self.env["mail.message"].search_count( + [ + ("res_id", "=", self.spreadsheet.id), + ("model", "=", "spreadsheet.spreadsheet"), + ] + ) + self.assertGreater(msg_count_after, msg_count_before) + + +# TODO: Add HttpCase tests for the /spreadsheet/writeback JSON-RPC endpoint. +# +# An HttpCase would POST to the controller route directly, verifying the full +# HTTP stack (routing, CSRF, JSON-RPC serialisation, auth). However, this +# requires: +# 1. A logged-in browser session (HttpCase.authenticate + url_open), or +# manually crafting a JSON-RPC request with session cookies. +# 2. The Odoo test HTTP server running (HttpCase spins one up, but it uses +# a separate transaction — test data created in setUpClass is not visible +# unless using ``@tagged("post_install", "-at_install")``) . +# 3. Careful handling of the JSON-RPC envelope ({"jsonrpc": "2.0", "method": +# "call", "params": {...}}) which Odoo's ``type="json"`` routes expect. +# +# For now the model-level tests above cover the business logic; the controller +# is a thin wrapper. A proper HttpCase should be added when the module gets +# integration / end-to-end test infrastructure. + + +class TestWritebackValueRoundTrip(TransactionCase): + """old_value is stored as text, so rolling back must restore the type.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.spreadsheet = cls.env["spreadsheet.spreadsheet"].create( + {"name": "Round Trip Sheet", "writeback_enabled": True} + ) + cls.partner = cls.env["res.partner"].create( + {"name": "Round Trip Partner", "is_company": False} + ) + + def _log(self, field_name, old_value): + return self.env["spreadsheet.writeback.log"].create( + { + "spreadsheet_id": self.spreadsheet.id, + "res_model": "res.partner", + "record_id": self.partner.id, + "field_name": field_name, + "old_value": old_value, + "new_value": "irrelevant", + } + ) + + def test_rollback_restores_false_boolean(self): + """str(False) is "False", which is truthy — it must not roll back True.""" + self.partner.is_company = True + log = self._log("is_company", str(False)) + + self.spreadsheet.action_rollback_writeback(log.id) + + self.assertFalse( + self.partner.is_company, + "a boolean that was False must roll back to False, not True", + ) + + def test_rollback_restores_true_boolean(self): + self.partner.is_company = False + log = self._log("is_company", str(True)) + self.spreadsheet.action_rollback_writeback(log.id) + self.assertTrue(self.partner.is_company) + + def test_rollback_restores_numeric_types(self): + original = self.partner.color + self.partner.color = original + 3 + log = self._log("color", str(original)) + self.spreadsheet.action_rollback_writeback(log.id) + self.assertEqual(self.partner.color, original) + + def test_relational_fields_are_not_writeback_candidates(self): + """A many2one cannot survive a str() round-trip, so it is excluded.""" + from ..models.spreadsheet_writeback import WRITEBACK_FIELD_TYPES + + for fname in ("parent_id", "category_id", "image_1920"): + field = self.env["res.partner"]._fields[fname] + self.assertNotIn( + field.type, + WRITEBACK_FIELD_TYPES, + f"{fname} ({field.type}) must not be writeback-eligible", + ) + self.assertIn( + self.env["res.partner"]._fields["name"].type, WRITEBACK_FIELD_TYPES + ) diff --git a/spreadsheet_oca/views/spreadsheet_writeback_views.xml b/spreadsheet_oca/views/spreadsheet_writeback_views.xml new file mode 100644 index 00000000..6529a7c9 --- /dev/null +++ b/spreadsheet_oca/views/spreadsheet_writeback_views.xml @@ -0,0 +1,127 @@ + + + + + + spreadsheet.writeback.log.list + spreadsheet.writeback.log + + + + + + + + + + + + + + + + + spreadsheet.writeback.log.form + spreadsheet.writeback.log + +
+
+
+ + + + + + + + + + + + + + + + + + + +
+ + + + + + Writeback Log + spreadsheet.writeback.log + list,form + {} + + + + + spreadsheet.spreadsheet.form.writeback + spreadsheet.spreadsheet + + + + + + + + + + + spreadsheet.spreadsheet.form.writeback_enabled + spreadsheet.spreadsheet + + + + + + + + + +