import logging from odoo.http import request _logger = logging.getLogger(__name__) def _to_bool(self, value): if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if isinstance(value, str): normalized = value.strip().lower() if normalized in {"1", "true", "t", "yes", "y", "on"}: return True if normalized in {"0", "false", "f", "no", "n", "off", ""}: return False return bool(value) def _validate_user_group_access(self, group_order, current_user): partner = current_user.partner_id if not partner or not group_order: raise ValueError("User is not a member of any consumer group in this order") user_group_ids = set(partner.group_ids.ids) for consumer_group in group_order.group_ids: if consumer_group.id in user_group_ids: return consumer_group.id _logger.warning( "_validate_user_group_access: user %s (%s) not member of any consumer group in order %s", current_user.name, current_user.id, group_order.id, ) raise ValueError("User is not a member of any consumer group in this order") def _get_consumer_group_for_user(self, group_order, current_user): partner = current_user.partner_id if not partner or not group_order: return False user_group_ids = set(partner.group_ids.ids) for consumer_group in group_order.group_ids: if consumer_group.id in user_group_ids: return consumer_group.id _logger.warning( "_get_consumer_group_for_user: User %s (%s) is not member of any consumer group in order %s", current_user.name, current_user.id, group_order.id, ) return False def _get_salesperson_for_order(self, partner): if partner.user_id and not partner.user_id._is_public(): return partner.user_id commercial_partner = partner.commercial_partner_id if commercial_partner.user_id and not commercial_partner.user_id._is_public(): return commercial_partner.user_id return False def _find_recent_draft_order(self, partner_id, group_order, request_obj=None): req = request_obj or request from datetime import datetime from datetime import timedelta today = datetime.now().date() start_of_week = today - timedelta(days=today.weekday()) end_of_week = start_of_week + timedelta(days=6) domain = [ ("partner_id", "=", partner_id), ("group_order_id", "=", group_order.id), ("state", "=", "draft"), ] if group_order.pickup_date: domain.append(("pickup_date", "=", group_order.pickup_date)) else: domain.extend( [ ("create_date", ">=", f"{start_of_week} 00:00:00"), ("create_date", "<=", f"{end_of_week} 23:59:59"), ] ) return ( req.env["sale.order"].sudo().search(domain, order="create_date desc", limit=1) ) def _validate_confirm_request(self, data, request_obj=None): req = request_obj or request order_id = data.get("order_id") if not order_id: raise ValueError("order_id is required") try: order_id = int(order_id) except (ValueError, TypeError) as err: raise ValueError(f"Invalid order_id format: {order_id}") from err group_order = req.env["group.order"].sudo().browse(order_id) if not group_order.exists(): raise ValueError(f"Order {order_id} not found") if group_order.state != "open": raise ValueError("Order is not available (not in open state)") current_user = req.env.user if not current_user.partner_id: raise ValueError("User has no associated partner") _validate_user_group_access(self, group_order, current_user) items = data.get("items", []) if not items: raise ValueError("No items in cart") _logger.info( "_validate_confirm_request: Valid request for order %d with %d items", order_id, len(items), ) return order_id, group_order, current_user, items def _validate_draft_request(self, data, request_obj=None): req = request_obj or request order_id = data.get("order_id") if not order_id: raise ValueError("order_id is required") try: order_id = int(order_id) except (ValueError, TypeError) as err: raise ValueError(f"Invalid order_id format: {order_id}") from err group_order = req.env["group.order"].sudo().browse(order_id) if not group_order.exists(): raise ValueError(f"Order {order_id} not found") current_user = req.env.user if not current_user.partner_id: raise ValueError("User has no associated partner") _validate_user_group_access(self, group_order, current_user) items = data.get("items", []) if not items: raise ValueError("No items in cart") merge_action = data.get("merge_action") existing_draft_id = data.get("existing_draft_id") _logger.info( "_validate_draft_request: Valid request for order %d with %d items (merge_action=%s)", order_id, len(items), merge_action, ) return (order_id, group_order, current_user, items, merge_action, existing_draft_id) def _validate_confirm_json(self, data, request_obj=None): req = request_obj or request order_id = data.get("order_id") if not order_id: raise ValueError("order_id is required") try: order_id = int(order_id) except (ValueError, TypeError) as err: raise ValueError(f"Invalid order_id format: {order_id}") from err group_order = req.env["group.order"].sudo().browse(order_id) if not group_order.exists(): raise ValueError(f"Order {order_id} not found") if group_order.state != "open": raise ValueError(f"Order is {group_order.state}") current_user = req.env.user if not current_user.partner_id: raise ValueError("User has no associated partner") _validate_user_group_access(self, group_order, current_user) items = data.get("items", []) if not items: raise ValueError("No items in cart") is_delivery = _to_bool(self, data.get("is_delivery", False)) _logger.info( "_validate_confirm_json: Valid request for order %d with %d items (is_delivery=%s)", order_id, len(items), is_delivery, ) return order_id, group_order, current_user, items, is_delivery def _validate_items_for_group_order(self, items, group_order, request_obj=None): req = request_obj or request if not items: return { "available_items": [], "unavailable_items": [], "unavailable_products": set(), "warning_message": "", } try: available_products = req.env["group.order"]._get_products_for_group_order( group_order.id ) available_product_ids = set(available_products.ids) except Exception as e: _logger.error( "Error getting available products for group_order %d: %s", group_order.id, e ) return { "available_items": items, "unavailable_items": [], "unavailable_products": set(), "warning_message": "", } available_items = [] unavailable_items = [] unavailable_product_ids = set() for item in items: product_id = item.get("product_id") if product_id in available_product_ids: available_items.append(item) else: unavailable_items.append(item) unavailable_product_ids.add(product_id) warning_message = "" if unavailable_items: unavailable_names = [ item.get("product_name", "Unknown") for item in unavailable_items ] warning_message = req.env._( "%(count)d product(s) from your saved order are no longer available in this group order: %(names)s. Only available products will be loaded.", count=len(unavailable_items), names=", ".join(unavailable_names), ) _logger.warning( "load_order_from_history: %d unavailable items in group_order %d (products: %s)", len(unavailable_items), group_order.id, unavailable_product_ids, ) return { "available_items": available_items, "unavailable_items": unavailable_items, "unavailable_products": unavailable_product_ids, "warning_message": warning_message, }