140 lines
6.6 KiB
Python
140 lines
6.6 KiB
Python
import logging
|
|
from odoo import api, fields, models, _
|
|
from odoo.osv import expression
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
class StockLocation(models.Model):
|
|
_inherit = 'stock.location'
|
|
|
|
def _get_allowed_locations(self):
|
|
"""
|
|
Deep Data Audit: Log the names and types of every location found.
|
|
"""
|
|
ctx = self.env.context
|
|
|
|
# Identification variables
|
|
mo_id = (ctx.get('active_mo_id') or
|
|
ctx.get('default_production_id') or
|
|
ctx.get('production_id') or
|
|
(ctx.get('active_id') if ctx.get('active_model') == 'mrp.production' else None))
|
|
|
|
allowed_ids = []
|
|
source_name = "None"
|
|
|
|
# 1. MO Lookup
|
|
if mo_id:
|
|
mo = self.env['mrp.production'].sudo().browse(mo_id)
|
|
if mo.exists() and mo.allowed_source_location_ids:
|
|
allowed_ids = mo.allowed_source_location_ids.ids
|
|
source_name = f"MO {mo.name}"
|
|
|
|
# 2. Context ID Fallback
|
|
if not allowed_ids:
|
|
target_keys = ['allowed_source_location_ids', 'default_allowed_source_location_ids']
|
|
for key in target_keys:
|
|
val = ctx.get(key)
|
|
if val:
|
|
if isinstance(val, list):
|
|
if all(isinstance(x, int) for x in val):
|
|
allowed_ids = val
|
|
else:
|
|
for entry in val:
|
|
if isinstance(entry, (list, tuple)):
|
|
if entry[0] == 6: allowed_ids = entry[2]; break
|
|
if entry[0] == 4: allowed_ids.append(entry[1])
|
|
elif isinstance(val, int):
|
|
allowed_ids = [val]
|
|
|
|
if allowed_ids:
|
|
source_name = f"Context {key}"
|
|
break
|
|
|
|
# 3. Final IDs check & Logging
|
|
if allowed_ids:
|
|
# Audit the location names
|
|
locations = self.env['stock.location'].sudo().browse(allowed_ids)
|
|
loc_names = [f"{l.display_name} (ID: {l.id})" for l in locations]
|
|
_logger.error(f"DEBUG_RESTRICT: Identified {len(loc_names)} Allowed Locations for {source_name}: {loc_names}")
|
|
return allowed_ids
|
|
|
|
# 4. Fallback to default
|
|
if ctx.get('default_location_id'):
|
|
return [ctx.get('default_location_id')]
|
|
|
|
return False
|
|
|
|
class StockQuant(models.Model):
|
|
_inherit = 'stock.quant'
|
|
|
|
@api.model
|
|
def web_search_read(self, domain, specification, offset=0, limit=None, order=None, count_limit=None):
|
|
"""
|
|
UI Override with GLOBAL AUDIT to find hiding stock.
|
|
"""
|
|
ctx = self.env.context
|
|
if not ctx.get('skip_location_restriction') and ctx.get('uid'):
|
|
allowed_ids = self.env['stock.location']._get_allowed_locations()
|
|
if allowed_ids:
|
|
# Strip native filters
|
|
clean_leaves = []
|
|
product_id = None
|
|
for leaf in domain:
|
|
if isinstance(leaf, (list, tuple)):
|
|
if len(leaf) == 3:
|
|
if leaf[0] == 'location_id': continue
|
|
if leaf[0] == 'product_id': product_id = leaf[2]
|
|
clean_leaves.append(leaf)
|
|
|
|
# Apply multi-location filter
|
|
domain = expression.AND([clean_leaves, [('location_id', 'child_of', allowed_ids)]])
|
|
|
|
# EXECUTE SEARCH once internally to see if it's empty
|
|
# If it's empty, we do a global warehouse search to tell the user WHERE his stock is.
|
|
res = super().web_search_read(domain, specification, offset=offset, limit=limit, order=order, count_limit=count_limit)
|
|
|
|
if res.get('length') == 0 and product_id:
|
|
# Global Audit!
|
|
_logger.error(f"DEBUG_RESTRICT: Catalog is EMPTY for product {product_id} in {allowed_ids}.")
|
|
all_quants = self.sudo().search([('product_id', '=', product_id), ('quantity', '>', 0)])
|
|
if all_quants:
|
|
loc_summary = [f"{q.location_id.display_name} (Qty: {q.quantity})" for q in all_quants]
|
|
_logger.error(f"DEBUG_RESTRICT: AUDIT: Product was found in {len(all_quants)} OTHER locations: {loc_summary}")
|
|
else:
|
|
_logger.error(f"DEBUG_RESTRICT: AUDIT: Product has ZERO quantity currently in the entire database (sudo).")
|
|
|
|
return res
|
|
|
|
return super().web_search_read(domain, specification, offset=offset, limit=limit, order=order, count_limit=count_limit)
|
|
|
|
class StockLot(models.Model):
|
|
_inherit = 'stock.lot'
|
|
|
|
@api.model
|
|
def name_search(self, name='', args=None, operator='ilike', limit=100):
|
|
ctx = self.env.context
|
|
if not ctx.get('skip_location_restriction') and ctx.get('uid'):
|
|
allowed_ids = self.env['stock.location']._get_allowed_locations()
|
|
if allowed_ids:
|
|
quant_domain = [('location_id', 'child_of', allowed_ids), ('quantity', '>', 0), ('lot_id', '!=', False)]
|
|
if ctx.get('default_product_id'):
|
|
quant_domain.append(('product_id', '=', ctx.get('default_product_id')))
|
|
|
|
quants = self.env['stock.quant'].with_context(skip_location_restriction=True).sudo().search(quant_domain)
|
|
args = expression.AND([args or [], [('id', 'in', quants.mapped('lot_id').ids)]])
|
|
return super().name_search(name, args=args, operator=operator, limit=limit)
|
|
|
|
@api.model
|
|
def web_search_read(self, domain, specification, offset=0, limit=None, order=None, count_limit=None):
|
|
ctx = self.env.context
|
|
if not ctx.get('skip_location_restriction') and ctx.get('uid'):
|
|
allowed_ids = self.env['stock.location']._get_allowed_locations()
|
|
if allowed_ids:
|
|
quant_domain = [('location_id', 'child_of', allowed_ids), ('quantity', '>', 0), ('lot_id', '!=', False)]
|
|
if ctx.get('default_product_id'):
|
|
quant_domain.append(('product_id', '=', ctx.get('default_product_id')))
|
|
|
|
quants = self.env['stock.quant'].with_context(skip_location_restriction=True).sudo().search(quant_domain)
|
|
domain = expression.AND([domain, [('id', 'in', quants.mapped('lot_id').ids)]])
|
|
return super().web_search_read(domain, specification, offset=offset, limit=limit, order=order, count_limit=count_limit)
|