355 lines
15 KiB
Python
355 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
import logging
|
||
|
||
from odoo import api, fields, models
|
||
from odoo.exceptions import UserError
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
|
||
class YuthonInventoryTask(models.Model):
|
||
"""盘点任务：定义一次盘点的范围、人员和时间
|
||
每次点击「确认并生成盘点表」会创建一条盘点表头记录（yuthon.inventory.sheet），
|
||
头记录下包含多条明细行（yuthon.inventory.sheet.line），每条对应一个产品/设备
|
||
"""
|
||
_name = 'yuthon.inventory.task'
|
||
_description = '盘点任务'
|
||
_order = 'create_date desc, id desc'
|
||
|
||
name = fields.Char(string='盘点名称', required=True, index=True)
|
||
code = fields.Char(string='编号', readonly=True, copy=False, default='New')
|
||
|
||
planned_date = fields.Datetime(string='计划盘点时间')
|
||
employee_ids = fields.Many2many(
|
||
'hr.employee',
|
||
'inventory_task_employee_rel',
|
||
'task_id',
|
||
'employee_id',
|
||
string='盘点人员',
|
||
required=True,
|
||
)
|
||
company_ids = fields.Many2many(
|
||
'res.company',
|
||
string='公司范围',
|
||
required=True,
|
||
default=lambda self: self.env.company,
|
||
)
|
||
|
||
include_fixed_assets = fields.Boolean(string='盘点固定资产', default=True)
|
||
include_office_supplies = fields.Boolean(string='盘点办公用品', default=True)
|
||
|
||
state = fields.Selection(
|
||
[('draft', '草稿'), ('confirmed', '已确认'), ('done', '已完成')],
|
||
string='状态',
|
||
default='draft',
|
||
index=True,
|
||
readonly=True,
|
||
copy=False,
|
||
)
|
||
|
||
# 盘点表头记录（每次确认生成一条）
|
||
sheet_ids = fields.One2many(
|
||
'yuthon.inventory.sheet',
|
||
'task_id',
|
||
string='盘点表',
|
||
readonly=True,
|
||
)
|
||
sheet_count = fields.Integer(string='盘点表数', compute='_compute_sheet_count')
|
||
|
||
note = fields.Text(string='备注')
|
||
reminded = fields.Boolean(string='已提醒', default=False, copy=False)
|
||
|
||
create_uid = fields.Many2one('res.users', string='创建人', readonly=True)
|
||
create_date = fields.Datetime(string='创建时间', readonly=True)
|
||
|
||
@api.depends('sheet_ids')
|
||
def _compute_sheet_count(self):
|
||
"""计算盘点表头数量"""
|
||
for task in self:
|
||
task.sheet_count = len(task.sheet_ids)
|
||
|
||
@api.model_create_multi
|
||
def create(self, vals_list):
|
||
"""创建时自动生成编号，格式：INV-202606-001"""
|
||
for vals in vals_list:
|
||
if vals.get('code', 'New') == 'New':
|
||
vals['code'] = self.env['ir.sequence'].next_by_code('yuthon.inventory.task') or 'New'
|
||
return super().create(vals_list)
|
||
|
||
def action_confirm(self):
|
||
"""确认任务：生成一条盘点表头 + 所有明细行，状态改为已确认"""
|
||
self.ensure_one()
|
||
if self.state != 'draft':
|
||
raise UserError('只有草稿状态的任务才能确认！')
|
||
if not self.include_fixed_assets and not self.include_office_supplies:
|
||
raise UserError('请至少选择一种盘点内容（固定资产或办公用品）！')
|
||
if not self.sheet_ids:
|
||
self._generate_sheets()
|
||
self.state = 'confirmed'
|
||
|
||
def action_done(self):
|
||
"""完成任务：批量提交所有待盘点的明细行
|
||
逐行执行提交逻辑（与点击每行「提交修改」按钮效果相同）
|
||
"""
|
||
self.ensure_one()
|
||
if self.state != 'confirmed':
|
||
raise UserError('只有已确认状态的任务才能提交！')
|
||
|
||
# 收集所有待盘点的明细行
|
||
pending_lines = self.env['yuthon.inventory.sheet.line']
|
||
for sheet in self.sheet_ids:
|
||
pending_lines |= sheet.line_ids.filtered(lambda l: l.state == 'pending')
|
||
|
||
if not pending_lines:
|
||
raise UserError('没有需要提交的盘点明细！')
|
||
|
||
# 逐行提交
|
||
for line in pending_lines:
|
||
try:
|
||
line.action_submit_line()
|
||
except UserError as e:
|
||
# 单行失败不中断，记录错误继续处理其他行
|
||
_logger.warning('明细行[%s]提交失败: %s', line.product_name, str(e))
|
||
|
||
# 标记任务完成
|
||
self.state = 'done'
|
||
|
||
def action_draft(self):
|
||
"""回到草稿：清空盘点表，允许重新生成"""
|
||
self.ensure_one()
|
||
if self.state == 'done':
|
||
raise UserError('已完成的任务不能回到草稿！')
|
||
self.state = 'draft'
|
||
# 清空盘点表（删除头记录，级联删除明细行）
|
||
self.sheet_ids.unlink()
|
||
|
||
def _generate_sheets(self):
|
||
"""根据任务条件生成 1 条盘点表头 + N 条明细行
|
||
- 办公用品从 product.template（"办公用品"标签）+ stock.quant 取数
|
||
- 固定资产从 maintenance.equipment 取数
|
||
"""
|
||
self.ensure_one()
|
||
Sheet = self.env['yuthon.inventory.sheet']
|
||
SheetLine = self.env['yuthon.inventory.sheet.line']
|
||
|
||
_logger.info('=== _generate_sheets 开始 ===')
|
||
_logger.info('任务ID=%s, name=%s, code=%s', self.id, self.name, self.code)
|
||
|
||
# 创建一条盘点表头
|
||
sheet = Sheet.create({
|
||
'task_id': self.id,
|
||
})
|
||
_logger.info('创建盘点表头 ID=%s', sheet.id)
|
||
|
||
line_count = 0
|
||
|
||
# ===== 办公用品明细行 =====
|
||
if self.include_office_supplies:
|
||
# 只取任务选定公司范围内的产品
|
||
product_tmpl_domain = [
|
||
('type', 'in', ('product', 'consu')),
|
||
'|',
|
||
('company_id', 'in', self.company_ids.ids),
|
||
('company_id', '=', False),
|
||
]
|
||
office_tags = self.env['product.tag'].search([('name', '=', '办公用品')])
|
||
_logger.info('办公用品标签查询 → 找到 %d 个标签', len(office_tags))
|
||
|
||
if office_tags:
|
||
office_domain = product_tmpl_domain + [('product_tag_ids', 'in', office_tags.ids)]
|
||
office_tmpls = self.env['product.template'].search(office_domain)
|
||
_logger.info('办公用品模板查询 → 找到 %d 个模板', len(office_tmpls))
|
||
|
||
if office_tmpls:
|
||
products = self.env['product.product'].search([
|
||
('product_tmpl_id', 'in', office_tmpls.ids),
|
||
])
|
||
_logger.info('办公用品变体查询 → 找到 %d 个产品', len(products))
|
||
|
||
if products:
|
||
# 只查询有实际库存的 quant（quantity > 0），
|
||
# 确保盘点明细只带出仓库里真正有货的产品
|
||
quant_domain = [
|
||
('product_id', 'in', products.ids),
|
||
('location_id.usage', '=', 'internal'),
|
||
('quantity', '>', 0),
|
||
]
|
||
quants = self.env['stock.quant'].search(quant_domain)
|
||
_logger.info('办公用品 stock.quant 查询（quantity>0）→ %d 条记录', len(quants))
|
||
|
||
# 按产品汇总库存数量
|
||
product_summary = {}
|
||
for quant in quants:
|
||
pid = quant.product_id.id
|
||
if pid not in product_summary:
|
||
product_summary[pid] = {'qty': 0.0}
|
||
product_summary[pid]['qty'] += quant.quantity
|
||
|
||
# 只创建有实际库存的产品的明细行（跳过库存为 0 的产品）
|
||
for pid, summary in product_summary.items():
|
||
product = self.env['product.product'].browse(pid)
|
||
if not product.exists():
|
||
continue
|
||
# 使用产品自身的公司ID，确保后续提交时公司匹配
|
||
line_company = product.company_id.id if product.company_id else (
|
||
self.company_ids[0].id if self.company_ids else False
|
||
)
|
||
SheetLine.create({
|
||
'sheet_id': sheet.id,
|
||
'sheet_type': 'office',
|
||
'product_id': product.id,
|
||
'on_hand_qty': summary['qty'],
|
||
'company_id': line_company,
|
||
})
|
||
line_count += 1
|
||
_logger.info('办公用品明细行创建完成（共 %d 行）', line_count)
|
||
|
||
# ===== 固定资产明细行 =====
|
||
if self.include_fixed_assets:
|
||
# 只取任务选定公司范围内的设备
|
||
eq_domain = [
|
||
'|',
|
||
('company_id', 'in', self.company_ids.ids),
|
||
('company_id', '=', False),
|
||
]
|
||
equipments = self.env['maintenance.equipment'].search(eq_domain)
|
||
_logger.info('固定资产设备查询 → 找到 %d 台设备', len(equipments))
|
||
|
||
for eq in equipments:
|
||
try:
|
||
on_hand = float(eq.equipment_number or 0)
|
||
except (ValueError, TypeError):
|
||
on_hand = 0.0
|
||
|
||
eq_company = False
|
||
if eq.company_id:
|
||
eq_company = eq.company_id.id
|
||
elif self.company_ids:
|
||
eq_company = self.company_ids[0].id
|
||
|
||
SheetLine.create({
|
||
'sheet_id': sheet.id,
|
||
'sheet_type': 'fixed_asset',
|
||
'equipment_id': eq.id,
|
||
'on_hand_qty': on_hand,
|
||
'user_name': eq.employee_id.name if eq.employee_id else '',
|
||
'department_name': eq.department_id1.name if eq.department_id1 else '',
|
||
'company_id': eq_company,
|
||
})
|
||
line_count += 1
|
||
_logger.info('固定资产明细行创建完成')
|
||
|
||
if line_count == 0:
|
||
raise UserError('根据所选条件未找到任何可盘点的数据，请检查盘点内容设置！')
|
||
_logger.info('=== _generate_sheets 完成：头记录=1, 明细行=%d ===', line_count)
|
||
|
||
def _adjust_inventory(self, lines=None):
|
||
"""提交盘点结果时，按类型分别处理明细行：
|
||
- 办公用品：通过 stock.quant 调整库存（标准 Odoo 方式）
|
||
- 固定资产：更新 maintenance.equipment 的 equipment_number 字段
|
||
"""
|
||
self.ensure_one()
|
||
if lines is None:
|
||
# 收集所有头记录下的所有明细行
|
||
lines = self.env['yuthon.inventory.sheet.line']
|
||
for sheet in self.sheet_ids:
|
||
lines |= sheet.line_ids
|
||
|
||
diff_lines = lines.filtered(lambda l: l.difference_qty != 0)
|
||
if not diff_lines:
|
||
return
|
||
|
||
office_lines = diff_lines.filtered(lambda l: l.sheet_type == 'office')
|
||
asset_lines = diff_lines.filtered(lambda l: l.sheet_type == 'fixed_asset')
|
||
|
||
# ===== 办公用品：通过 stock.quant.inventory_quantity 调整库存 =====
|
||
for line in office_lines:
|
||
product = line.product_id
|
||
actual_qty = line.actual_qty
|
||
|
||
quants = self.env['stock.quant'].search([
|
||
('product_id', '=', product.id),
|
||
('location_id.usage', '=', 'internal'),
|
||
'|',
|
||
('company_id', 'in', self.company_ids.ids),
|
||
('company_id', '=', False),
|
||
])
|
||
if not quants:
|
||
default_location = self.env.ref('stock.stock_location_stock', raise_if_not_found=False)
|
||
if not default_location:
|
||
continue
|
||
quant = self.env['stock.quant'].create({
|
||
'product_id': product.id,
|
||
'location_id': default_location.id,
|
||
'quantity': 0.0,
|
||
'company_id': self.company_ids[0].id if self.company_ids else False,
|
||
})
|
||
else:
|
||
quant = quants.filtered(lambda q: q.quantity > 0.0)[:1] or quants[:1]
|
||
|
||
quant.inventory_quantity = actual_qty
|
||
quant.action_apply_inventory()
|
||
|
||
self.env['yuthon.inventory.adjustment'].create({
|
||
'task_id': self.id,
|
||
'line_id': line.id,
|
||
'product_id': product.id,
|
||
'adjust_type': 'increase' if line.difference_qty > 0 else 'decrease',
|
||
'qty_before': line.on_hand_qty,
|
||
'qty_after': actual_qty,
|
||
'adjust_qty': abs(line.difference_qty),
|
||
'reason': line.difference_reason,
|
||
'company_id': line.company_id.id if line.company_id else False,
|
||
})
|
||
|
||
# ===== 固定资产：更新 maintenance.equipment 的 equipment_number 字段 =====
|
||
for line in asset_lines:
|
||
equipment = line.equipment_id
|
||
actual_qty = line.actual_qty
|
||
|
||
if actual_qty == int(actual_qty):
|
||
equipment.equipment_number = str(int(actual_qty))
|
||
else:
|
||
equipment.equipment_number = str(round(actual_qty, 2))
|
||
|
||
_logger.info('固定资产盘点: 设备[%s](id=%s) 数量从 %s 更新为 %s',
|
||
equipment.name, equipment.id, line.on_hand_qty, equipment.equipment_number)
|
||
|
||
self.env['yuthon.inventory.adjustment'].create({
|
||
'task_id': self.id,
|
||
'line_id': line.id,
|
||
'equipment_id': equipment.id,
|
||
'product_id': False,
|
||
'adjust_type': 'increase' if line.difference_qty > 0 else 'decrease',
|
||
'qty_before': line.on_hand_qty,
|
||
'qty_after': actual_qty,
|
||
'adjust_qty': abs(line.difference_qty),
|
||
'reason': line.difference_reason,
|
||
'company_id': line.company_id.id if line.company_id else False,
|
||
})
|
||
|
||
def action_open_sheets(self):
|
||
"""跳转到当前任务的盘点表头列表"""
|
||
self.ensure_one()
|
||
return {
|
||
'type': 'ir.actions.act_window',
|
||
'name': '盘点表 - %s' % self.name,
|
||
'res_model': 'yuthon.inventory.sheet',
|
||
'view_mode': 'tree,form',
|
||
'domain': [('task_id', '=', self.id)],
|
||
'target': 'current',
|
||
}
|
||
|
||
def action_send_reminder(self):
|
||
"""手动触发提醒（供 cron 调用）"""
|
||
now = fields.Datetime.now()
|
||
tasks = self.search([
|
||
('state', '=', 'confirmed'),
|
||
('planned_date', '!=', False),
|
||
('planned_date', '<=', now),
|
||
('reminded', '=', False),
|
||
])
|
||
for task in tasks:
|
||
# TODO: 接入企业微信后，此处发送消息给 task.employee_ids
|
||
task.reminded = True
|