105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2026 Nexa Systems Inc.
|
|
# License OPL-1 (Odoo Proprietary License v1.0)
|
|
|
|
import logging
|
|
|
|
from . import models
|
|
from . import wizard
|
|
from . import controllers
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _post_init_assign_groups(env):
|
|
"""Give all internal users the Fusion Inventory / User group,
|
|
and all admins the Manager group on install.
|
|
Also force-recompute margin and auto-detect brands."""
|
|
fi_user = env.ref('fusion_inventory.group_fusion_inventory_user', raise_if_not_found=False)
|
|
fi_manager = env.ref('fusion_inventory.group_fusion_inventory_manager', raise_if_not_found=False)
|
|
|
|
if fi_user:
|
|
internal_users = env['res.users'].sudo().search([
|
|
('active', '=', True),
|
|
('share', '=', False),
|
|
])
|
|
fi_user.sudo().write({'user_ids': [(4, u.id) for u in internal_users]})
|
|
|
|
if fi_manager:
|
|
group_admin = env.ref('base.group_system', raise_if_not_found=False)
|
|
if group_admin:
|
|
admin_users = group_admin.sudo().user_ids.filtered('active')
|
|
fi_manager.sudo().write({'user_ids': [(4, u.id) for u in admin_users]})
|
|
|
|
_recompute_margins(env)
|
|
_auto_detect_brands(env)
|
|
_sync_costs_from_bills(env)
|
|
|
|
|
|
def _recompute_margins(env):
|
|
"""Force-recompute x_fi_margin_pct for all products with a sale price and cost."""
|
|
try:
|
|
products = env['product.template'].sudo().search([
|
|
('list_price', '>', 0),
|
|
('standard_price', '>', 0),
|
|
])
|
|
if products:
|
|
products._compute_margin_pct()
|
|
_logger.info('Recomputed margin for %d products', len(products))
|
|
except Exception:
|
|
_logger.warning('Failed to recompute margins', exc_info=True)
|
|
|
|
|
|
def _sync_costs_from_bills(env):
|
|
"""Update every product's cost from its latest posted vendor bill line."""
|
|
try:
|
|
products = env['product.template'].sudo().search([])
|
|
products._sync_cost_from_latest_bill()
|
|
_logger.info('Post-init cost sync complete for %d products', len(products))
|
|
except Exception:
|
|
_logger.warning('Failed to sync costs from vendor bills', exc_info=True)
|
|
|
|
|
|
def _auto_detect_brands(env):
|
|
"""Create product.brand records from existing PO vendors
|
|
and link products to their respective brands."""
|
|
try:
|
|
Brand = env['product.brand'].sudo()
|
|
PO = env['purchase.order'].sudo()
|
|
|
|
orders = PO.search([('state', 'in', ('purchase', 'done'))])
|
|
if not orders:
|
|
_logger.info('No confirmed POs found, skipping brand auto-detection')
|
|
return
|
|
|
|
vendor_ids = list(set(orders.mapped('partner_id').ids))
|
|
partners = env['res.partner'].sudo().browse(vendor_ids)
|
|
|
|
created = 0
|
|
linked = 0
|
|
for vendor in partners:
|
|
if not vendor.supplier_rank:
|
|
continue
|
|
|
|
brand = Brand.search([('partner_id', '=', vendor.id)], limit=1)
|
|
if not brand:
|
|
brand = Brand.create({
|
|
'name': vendor.name,
|
|
'partner_id': vendor.id,
|
|
'logo': vendor.image_128 or False,
|
|
})
|
|
created += 1
|
|
|
|
vendor_orders = orders.filtered(lambda o: o.partner_id.id == vendor.id)
|
|
product_tmpls = vendor_orders.mapped('order_line.product_id.product_tmpl_id')
|
|
for tmpl in product_tmpls:
|
|
if brand.id not in tmpl.x_fi_brand_ids.ids:
|
|
tmpl.write({'x_fi_brand_ids': [(4, brand.id)]})
|
|
linked += 1
|
|
|
|
_logger.info(
|
|
'Brand auto-detection: %d brands created, %d product-brand links added',
|
|
created, linked)
|
|
except Exception:
|
|
_logger.warning('Failed to auto-detect brands from POs', exc_info=True)
|