odoo-tower/odoo-addons
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tower: upload laundry_management 19.0.19.0.4 (via marketplace)

Browse Source
This commit is contained in:
git_admin
2026-05-01 15:01:17 +00:00
parent 33a761413c
commit 234aee88e2
1 changed files with 765 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

765
addons/laundry_management/models/laundry_order.py Normal file
View File

@@ -0,0 +1,765 @@
import logging
from odoo import models, fields, api, _
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__)
STATES = [
('intake', 'Intake'),
('processing', 'Processing'),
('ready', 'Ready for Pickup'),
('delivered', 'Delivered'),
('cancelled', 'Cancelled'),
]
STATE_KEYS = [s[0] for s in STATES]
FINAL_STATES = {'delivered', 'cancelled'}
SOURCE_TYPES = [
('pos', 'Point of Sale'),
('manual', 'Manual / Backoffice'),
]
# Header fields the lock protects. NOT included on purpose:
# - state (workflow advance is always allowed via dedicated actions)
# - amount_settled (settlement engine writes after lock)
# - notes (managerial commentary always allowed)
# - manager_unlocked_* (the unlock wizard writes these)
# - tracking_enabled (Phase-4 prep, manager configuration)
LOCKED_HEADER_FIELDS = frozenset({
'partner_id',
'amount_total', 'amount_paid_cash', 'amount_deferred',
'order_type_id', 'attribute_ids',
'is_delivery', 'delivery_address', 'delivery_scheduled_at',
'priority_level',
'pos_order_id', 'pos_reference',
'source_type', 'name', 'company_id',
})
# Sentinel context flag set by the POS sync hook (and any other automated
# server-side path) to allow the create + write that bring a fresh
# laundry.order to life. Without this flag, locked orders refuse mutations.
POS_SYNC_CTX = 'laundry_pos_sync'
class LaundryOrder(models.Model):
"""Standalone laundry order — created from POS.
POS owns payments/sessions/accounting. This model handles operational
workflow only: intake -> processing -> ready -> delivered.
Financial model (Phase 1 fix):
amount_total = mirror of pos.order.amount_total
amount_paid_cash = real money collected at origin (cash/card)
amount_deferred = Customer Account / pay-later amount at origin
amount_settled = money collected later via settlement engine
amount_due = amount_deferred - amount_settled
`amount_due > 0` means the customer still owes real money.
"""
_name = 'laundry.order'
_description = 'Laundry Order'
_inherit = ['mail.thread', 'mail.activity.mixin']
_order = 'create_date desc, id desc'
name = fields.Char(
string='Order No.',
required=True, copy=False, readonly=True,
default='New', tracking=True,
)
# -- POS link --
# Optional: manual/backoffice orders have no POS origin. The uniqueness
# constraint below still enforces "one laundry order per pos.order" for
# POS-sourced rows.
pos_order_id = fields.Many2one(
'pos.order', string='POS Order',
index=True, readonly=True, copy=False,
ondelete='restrict',
)
pos_reference = fields.Char(
string='POS Reference',
readonly=True, copy=False, index=True,
)
# -- Customer --
partner_id = fields.Many2one(
'res.partner', string='Customer',
required=True, index=True, tracking=True,
)
partner_phone = fields.Char(
related='partner_id.phone', string='Phone', readonly=True,
)
# -- Company / Currency --
company_id = fields.Many2one(
'res.company', string='Company',
required=True, default=lambda self: self.env.company,
index=True,
)
currency_id = fields.Many2one(
related='company_id.currency_id', store=True,
)
# -- Workflow --
state = fields.Selection(
STATES,
string='Status',
default='intake',
required=True,
tracking=True,
copy=False,
index=True,
)
# -- Lines --
line_ids = fields.One2many(
'laundry.order.line', 'order_id', string='Order Lines',
)
# -- Financial snapshot --
amount_total = fields.Monetary(
string='Total',
currency_field='currency_id',
readonly=True, copy=False, store=True,
)
amount_paid_cash = fields.Monetary(
string='Paid (Cash/Card)',
currency_field='currency_id',
readonly=True, copy=False, store=True,
default=0.0,
help='Amount collected at origin via non-deferred payment methods '
'(cash, card — any pos.payment.method with split_transactions=False).',
)
amount_deferred = fields.Monetary(
string='Deferred',
currency_field='currency_id',
readonly=True, copy=False, store=True,
default=0.0,
help='Amount deferred at origin via Customer Account / pay-later '
'(any pos.payment.method with split_transactions=True).',
)
amount_settled = fields.Monetary(
string='Settled',
currency_field='currency_id',
readonly=True, copy=False, store=True,
default=0.0,
help='Amount collected later via the settlement engine '
'(account.payment + reconciliation).',
)
amount_due = fields.Monetary(
string='Due',
currency_field='currency_id',
compute='_compute_amount_due',
store=True,
help='Remaining balance the customer owes: amount_deferred - amount_settled.',
)
# Back-compat alias — views/reports still reference `amount_paid`.
# Computed, non-stored; reflects real cash collected at origin.
amount_paid = fields.Monetary(
string='Paid',
currency_field='currency_id',
compute='_compute_amount_paid_alias',
)
# -- Computed --
item_count = fields.Integer(
string='Items',
compute='_compute_item_count',
store=True,
)
# -- Order type / attributes / delivery --
order_type_id = fields.Many2one(
'laundry.order.type', string='Order Type',
index=True, tracking=True,
)
attribute_ids = fields.Many2many(
'laundry.order.attribute',
'laundry_order_attribute_rel',
'order_id', 'attribute_id',
string='Attributes',
)
is_delivery = fields.Boolean(string='Delivery', tracking=True, index=True)
delivery_address = fields.Text(string='Delivery Address')
delivery_scheduled_at = fields.Datetime(string='Scheduled At')
priority_level = fields.Selection([
('normal', 'Normal'),
('urgent', 'Urgent'),
], string='Priority', default='normal', tracking=True, index=True)
# -- Notes --
notes = fields.Text(string='Notes')
# ── Source / locking (Phase 3) ────────────────────────────────────
# source_type is the truth-bearing identity. is_from_pos is a stored
# mirror used in domains, list filters, and rule conditions where a
# selection field would be awkward.
source_type = fields.Selection(
SOURCE_TYPES,
string='Source',
required=True,
default='manual',
readonly=True,
copy=False,
index=True,
tracking=True,
help='POS-sourced orders are hard-locked: lines, prices and the '
'customer cannot be edited unless a manager grants a '
'temporary unlock window. Manual orders are editable until '
'they reach a final state (delivered / cancelled).',
)
is_from_pos = fields.Boolean(
string='From POS',
compute='_compute_is_from_pos',
store=True, index=True,
)
# Phase-4 prep — flag only, no logic wired yet.
tracking_enabled = fields.Boolean(
string='Per-Item Tracking',
default=False,
copy=False,
help='When enabled, each laundry.order.line will be advanced '
'through its own state machine. Phase 4 wires the '
'synchronization between order state and item state.',
)
# Computed: True when the order refuses mutation of LOCKED_HEADER_FIELDS
# and any line write/create/unlink. Not stored — cheap to recompute and
# depends on a Datetime that ages out without a write.
locked = fields.Boolean(
string='Locked',
compute='_compute_locked',
help='Order is read-only when True. POS-sourced orders are '
'always locked. Final-state orders (delivered, cancelled) '
'are always locked. Managers can grant a temporary unlock '
'window via the "Unlock for Editing" action.',
)
manager_unlocked_until = fields.Datetime(
string='Unlock Window Expires',
copy=False, readonly=True,
help='When set in the future, the lock guard is suspended. '
'Auto-expires; no manual re-lock required.',
)
manager_unlocked_by = fields.Many2one(
'res.users', string='Last Unlocked By',
copy=False, readonly=True,
)
manager_unlock_reason = fields.Char(
string='Last Unlock Reason',
copy=False, readonly=True,
)
# Stamped when the order moves to delivered. Powers the avg-processing
# and on-time KPIs on the Operations Dashboard. Outside
# LOCKED_HEADER_FIELDS so action_deliver can write it on POS-locked
# orders without needing the bypass context.
delivered_at = fields.Datetime(
string='Delivered At',
readonly=True, copy=False, index=True,
help='Timestamp set automatically when the order moves to '
'Delivered. Used by the analytics dashboard to compute '
'processing time and on-time delivery percentage.',
)
# -- Constraints --
_pos_order_uniq = models.Constraint(
'UNIQUE(pos_order_id)',
'A laundry order already exists for this POS order.',
)
@api.constrains('source_type', 'pos_order_id')
def _check_source_type_consistency(self):
for order in self:
if order.source_type == 'pos' and not order.pos_order_id:
raise UserError(_(
'Order "%s" is marked as POS-sourced but has no '
'linked POS order.', order.name or order.id,
))
if order.source_type == 'manual' and order.pos_order_id:
raise UserError(_(
'Order "%s" is marked as manual but has a linked '
'POS order. Set source_type="pos" to keep them '
'consistent.', order.name or order.id,
))
# -- Computed --
@api.depends('amount_deferred', 'amount_settled')
def _compute_amount_due(self):
for order in self:
order.amount_due = max(
(order.amount_deferred or 0.0) - (order.amount_settled or 0.0),
0.0,
)
@api.depends('amount_paid_cash')
def _compute_amount_paid_alias(self):
for order in self:
order.amount_paid = order.amount_paid_cash or 0.0
@api.depends('source_type')
def _compute_is_from_pos(self):
for order in self:
order.is_from_pos = order.source_type == 'pos'
@api.depends('source_type', 'state', 'manager_unlocked_until')
def _compute_locked(self):
now = fields.Datetime.now()
for order in self:
unlock_active = bool(
order.manager_unlocked_until
and order.manager_unlocked_until > now
)
base_locked = (
order.source_type == 'pos'
or order.state in FINAL_STATES
)
order.locked = base_locked and not unlock_active
# ── Lock enforcement helpers ──────────────────────────────────────
def _is_pos_sync(self):
"""True when the call originates from the POS sync hook (or any
explicit server path that opts in via the context flag).
Both create() and write() honour this so the bridge from
pos.order can build / refresh the laundry.order without fighting
its own lock guard. This is checked BEFORE anything else in
write() so indirect POS writes (stored-compute flushes, cascades)
can never raise from the guard.
"""
return bool(self.env.context.get(POS_SYNC_CTX))
def _check_lock_for_write(self, vals):
"""Raise UserError when `vals` would mutate a protected header
field on a currently-locked order. Workflow advances (state,
amount_settled, notes, manager_unlocked_*) are excluded by
whitelist (LOCKED_HEADER_FIELDS).
Note: the POS-sync bypass is already applied at the top of
`write()` — this helper is only invoked for non-bypassed paths.
"""
protected = LOCKED_HEADER_FIELDS.intersection(vals.keys())
if not protected:
return
for order in self:
if order.locked:
raise UserError(_(
'Order "%(name)s" is locked. Editable fields: state '
'transitions, internal notes, settlement amount.\n'
'To edit %(fields)s, ask a manager to use '
'"Unlock for Editing" first.',
name=order.name,
fields=', '.join(sorted(protected)),
))
def _check_lock_for_unlink(self):
"""POS-sourced and final-state orders cannot be unlinked. The
manager unlock wizard is intentionally NOT honored here — deletion
requires a stronger affordance (cancellation + audit trail), not
a temporary edit window.
"""
for order in self:
if order.source_type == 'pos':
raise UserError(_(
'Order "%(name)s" was created from POS and cannot '
'be deleted. Cancel the underlying POS order instead.',
name=order.name,
))
if order.state in FINAL_STATES:
raise UserError(_(
'Order "%(name)s" is in a final state (%(state)s) '
'and cannot be deleted.',
name=order.name,
state=dict(STATES).get(order.state, order.state),
))
@api.depends('line_ids.qty')
def _compute_item_count(self):
for order in self:
order.item_count = int(sum(order.line_ids.mapped('qty')))
# -- ORM --
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
if vals.get('name', 'New') == 'New':
vals['name'] = (
self.env['ir.sequence'].next_by_code('laundry.order')
or 'New'
)
self._apply_type_attribute_inference(vals)
return super().create(vals_list)
def write(self, vals):
# STEP 1 — POS sync bypass.
# Must be the very first thing we do. Any code path that opts
# into the context flag (POS create/sync, settlement engine,
# stored-compute flushes triggered from inside a bypassed write)
# MUST sail through unconditionally. No locking check, no
# side-effects, no iteration over self — just delegate to super.
# This guarantee is what makes POS payment/settlement flows
# immune to this model's lock guard.
if self._is_pos_sync():
if _logger.isEnabledFor(logging.DEBUG):
_logger.debug(
'laundry.order.write BYPASS ids=%s keys=%s',
self.ids, list(vals.keys()),
)
return super().write(vals)
# STEP 2 — Forensic trace (DEBUG-level; off by default in prod,
# enable with `--log-level=debug` or `--log-handler=odoo.addons
# .laundry_management.models.laundry_order:DEBUG`).
if _logger.isEnabledFor(logging.DEBUG):
keys = list(vals.keys())
for order in self:
_logger.debug(
'laundry.order.write id=%s source=%s state=%s '
'locked=%s ctx_keys=%s vals_keys=%s',
order.id, order.source_type, order.state, order.locked,
sorted(self.env.context.keys()), keys,
)
# STEP 3 — Lock guard for non-POS callers.
self._check_lock_for_write(vals)
return super().write(vals)
def unlink(self):
# Explicit: even with POS-sync context, refuse to delete a locked
# order. Deletion is not a mutation the sync path ever issues.
self._check_lock_for_unlink()
return super().unlink()
@api.model
def _apply_type_attribute_inference(self, vals):
"""Fill in priority_level / is_delivery from the selected order
type and attributes when the caller did not explicitly set them.
Rules:
- type.priority='urgent' OR any attribute with
is_priority_related=True → priority_level='urgent'
- type.is_delivery=True OR any attribute with
is_delivery_related=True → is_delivery=True
- Do NOT overwrite explicit incoming delivery_address /
delivery_scheduled_at with blank values.
"""
type_id = vals.get('order_type_id')
order_type = (
self.env['laundry.order.type'].browse(type_id) if type_id else None
)
attribute_ids = []
raw_attrs = vals.get('attribute_ids') or []
for cmd in raw_attrs:
if isinstance(cmd, (list, tuple)) and len(cmd) >= 3:
# Odoo x2m commands: (6,0,[ids]), (4,id), etc.
if cmd[0] == 6 and isinstance(cmd[2], (list, tuple)):
attribute_ids.extend(cmd[2])
elif cmd[0] == 4 and cmd[1]:
attribute_ids.append(cmd[1])
elif isinstance(cmd, int):
attribute_ids.append(cmd)
attributes = (
self.env['laundry.order.attribute'].browse(attribute_ids)
if attribute_ids else self.env['laundry.order.attribute']
)
# Priority
if 'priority_level' not in vals:
urgent = (
(order_type and order_type.priority == 'urgent')
or any(a.is_priority_related for a in attributes)
)
vals['priority_level'] = 'urgent' if urgent else 'normal'
# Delivery
if 'is_delivery' not in vals:
delivery = (
(order_type and order_type.is_delivery)
or any(a.is_delivery_related for a in attributes)
)
vals['is_delivery'] = bool(delivery)
# -- Workflow actions --
def action_process(self):
for order in self:
if order.state != 'intake':
raise UserError(_(
'Order "%(name)s" is not in Intake state.',
name=order.name,
))
order.state = 'processing'
def action_ready(self):
for order in self:
if order.state != 'processing':
raise UserError(_(
'Order "%(name)s" is not in Processing state.',
name=order.name,
))
order.state = 'ready'
def action_deliver(self):
"""Guards: must be Ready + fully paid (amount_due == 0).
Also stamps `delivered_at` so the dashboard KPIs (avg processing
time, on-time delivery %) can be computed from real data instead
of the heuristic on `write_date`.
"""
for order in self:
if order.state != 'ready':
raise UserError(_(
'Order "%(name)s" must be Ready before delivery.',
name=order.name,
))
if order.amount_due > 0:
raise UserError(_(
'Order "%(name)s" has %(due).2f outstanding. '
'Collect payment in POS before delivery.',
name=order.name,
due=order.amount_due,
))
order.write({
'state': 'delivered',
'delivered_at': fields.Datetime.now(),
})
def action_cancel(self):
"""Cancel an order. Allowed for manual orders only — POS-sourced
orders must be voided through the POS workflow to keep the sale
and the operational record in sync.
"""
for order in self:
if order.source_type == 'pos':
raise UserError(_(
'Order "%(name)s" was created from POS and cannot '
'be cancelled here. Cancel the underlying POS order '
'instead.',
name=order.name,
))
if order.state in FINAL_STATES:
raise UserError(_(
'Order "%(name)s" is already %(state)s.',
name=order.name,
state=dict(STATES).get(order.state, order.state),
))
order.state = 'cancelled'
# -- Smart button --
def action_open_pos_order(self):
self.ensure_one()
return {
'type': 'ir.actions.act_window',
'res_model': 'pos.order',
'res_id': self.pos_order_id.id,
'view_mode': 'form',
'target': 'current',
}
# ═════════════════════════════════════════════════════════════════
# POS "Laundry Orders" popup — server-side RPCs
# ---------------------------------------------------------------
# These methods are the ONLY way the POS popup interacts with the
# model. Each action delegates to the corresponding workflow method
# (`action_process` / `action_ready` / `action_deliver`) which already
# enforces state + amount_due guards server-side. Direct writes to
# LOCKED_HEADER_FIELDS are NOT exposed here — the Phase 3 lock
# remains the sole authority for business-edit protection.
# ═════════════════════════════════════════════════════════════════
def _pos_allowed_actions(self):
"""Return the list of action keys the popup may render for this
order. Pure function of state + amount_due. Final states only
allow printing.
"""
self.ensure_one()
actions = ['print_work_order']
if self.state in FINAL_STATES:
return actions
if self.state == 'intake':
actions.append('start_processing')
elif self.state == 'processing':
actions.append('mark_ready')
elif self.state == 'ready':
if self.amount_due <= 0:
actions.append('deliver')
else:
actions.append('collect_payment')
return actions
def _pos_payment_state(self):
self.ensure_one()
if self.amount_due > 0:
return 'due'
if self.amount_deferred > 0 and self.amount_settled >= self.amount_deferred:
return 'settled'
if self.amount_deferred > 0:
return 'deferred'
return 'paid'
def _pos_payload(self):
"""Compact, UI-ready dict. Single source of truth for the popup
shape — every RPC returns exactly this structure."""
self.ensure_one()
names = self.line_ids.mapped('product_id.name')
if not names:
names = self.line_ids.mapped('description')
state_selection = dict(self._fields['state'].selection)
return {
'id': self.id,
'name': self.name,
'state': self.state,
'state_label': state_selection.get(self.state) or self.state,
'pos_reference': self.pos_reference or '',
'is_from_pos': self.is_from_pos,
'create_date': fields.Datetime.to_string(self.create_date) if self.create_date else False,
'item_count': int(self.item_count or 0),
'service_summary': ', '.join(dict.fromkeys(names))[:80],
'amount_total': self.amount_total,
'amount_paid': self.amount_paid_cash,
'amount_deferred': self.amount_deferred,
'amount_settled': self.amount_settled,
'amount_due': self.amount_due,
'payment_state': self._pos_payment_state(),
'is_delivery': self.is_delivery,
'allowed_actions': self._pos_allowed_actions(),
}
@api.model
def pos_search_customer_orders(self, partner_id, search_query=False, limit=20):
"""Partner-scoped search for the POS popup.
• Always filters by partner_id (no global search in this phase).
• Optional search_query ilike-matches on name / pos_reference /
partner_phone.
• Hard-capped at 50 rows regardless of the caller's limit.
"""
if not partner_id:
return []
limit = max(1, min(int(limit or 20), 50))
q = (search_query or '').strip()
domain = [('partner_id', '=', partner_id)]
if q:
domain = [
'&',
('partner_id', '=', partner_id),
'|', '|',
('name', 'ilike', q),
('pos_reference', 'ilike', q),
('partner_phone', 'ilike', q),
]
orders = self.search(
domain, order='create_date desc, id desc', limit=limit,
)
return [o._pos_payload() for o in orders]
def pos_action_start_processing(self):
"""POS popup: advance intake → processing. Returns refreshed payload."""
self.ensure_one()
self.action_process()
return self._pos_payload()
def pos_action_mark_ready(self):
"""POS popup: advance processing → ready. Returns refreshed payload."""
self.ensure_one()
self.action_ready()
return self._pos_payload()
def pos_action_deliver(self):
"""POS popup: advance ready → delivered. Returns refreshed payload.
`action_deliver` raises UserError when amount_due > 0 — the popup
surfaces that error; no client-side duplication of the rule.
"""
self.ensure_one()
self.action_deliver()
return self._pos_payload()
@api.model
def pos_get_thermal_data(self, order_id):
"""Build a self-contained payload for the thermal Work-Order
receipt rendered by `laundry_management.LaundryWorkOrderThermal`.
Independent from `_pos_payload` — that one is for the popup list
(compact); this one carries every line + delivery meta the
cashier needs on the printed slip.
"""
order = self.browse(int(order_id))
if not order.exists():
return False
state_label = dict(order._fields['state'].selection).get(
order.state, order.state,
)
return {
'id': order.id,
'name': order.name,
'state': order.state,
'state_label': state_label,
'payment_state': order._pos_payment_state(),
'pos_reference': order.pos_reference or '',
'partner_name': order.partner_id.name or '',
# `mobile` is provided by the optional `phone` add-on; fall
# back gracefully when it isn't present in the install.
'partner_phone': (
order.partner_id.phone
or getattr(order.partner_id, 'mobile', '')
or ''
),
'company_name': order.company_id.name or '',
'create_date': fields.Datetime.to_string(order.create_date),
'lines': [{
'qty': line.qty,
'description': (
line.description
or (line.product_id.name if line.product_id else '')
),
'price_unit': line.price_unit,
'subtotal': line.subtotal,
'tracking_code': line.tracking_code or '',
} for line in order.line_ids],
'item_count': int(order.item_count or 0),
'amount_total': order.amount_total,
'amount_paid': order.amount_paid_cash,
'amount_deferred': order.amount_deferred,
'amount_settled': order.amount_settled,
'amount_due': order.amount_due,
'is_delivery': order.is_delivery,
'delivery_address': order.delivery_address or '',
'delivery_scheduled_at': (
fields.Datetime.to_string(order.delivery_scheduled_at)
if order.delivery_scheduled_at else ''
),
'currency_symbol': order.currency_id.symbol or '',
'currency_position': order.currency_id.position or 'after',
}
def action_open_unlock_wizard(self):
"""Open the manager unlock wizard pre-filled with this order.
Access is enforced inside the wizard's action method (group
check + reason required), but we also short-circuit here so
the button itself is silent for non-managers.
"""
self.ensure_one()
if not self.locked:
raise UserError(_(
'Order "%s" is already editable.', self.name,
))
return {
'type': 'ir.actions.act_window',
'res_model': 'laundry.order.unlock.wizard',
'view_mode': 'form',
'target': 'new',
'context': {
'default_order_id': self.id,
},
}