# -*- coding: utf-8 -*- from odoo import api, fields, models, tools, _ from odoo.exceptions import ValidationError, UserError import logging from datetime import datetime, timedelta, date _logger = logging.getLogger(__name__) class YuthonNotice(models.Model): _name = "yuthon.notice" _description = '通知公告' _inherit = ['mail.thread.main.attachment', 'mail.activity.mixin'] _rec_name = 'name' _order = "sequence desc" name = fields.Char(string='标题') code = fields.Char(string="编号", default=lambda self: self.env['ir.sequence'].next_by_code('notice_code')) users_id = fields.Many2one('res.users', string="发布人", default=lambda self: self.env.user) company_id = fields.Many2one(related="users_id.employee_id.company_id", string="公司", store=True) department_id = fields.Many2one(related='users_id.employee_id.department_id', string="部门") notice_type_id = fields.Many2one('yuthon.notice.type', string="类型") create_date1 = fields.Date(string="发布日期", default=fields.Date.today()) start_date = fields.Date(string="生效日期", default=fields.Date.today()) end_date = fields.Date(string="终止日期") sequence = fields.Integer(string="顺序", default=1, compute='_compute_sequence') employee_ids = fields.Many2many('hr.employee', string="人员范围") department_ids = fields.Many2many('hr.department', string='部门范围') notice_state = fields.Selection([('no', '未发布'), ('yes', '已发布'), ('stop', '已终止')], default='no', string="公告状态") is_notice_roger = fields.Selection([('yes', '已读'), ('no', '未读')], default='no', string="确认状态") is_change_notice = fields.Boolean(string="是否发送通知", default=True) attn_notice = fields.Html(string="经办人意见") notice_text = fields.Html(string="正文") notice_text1 = fields.Html(string="正文") read_number = fields.Integer(string='阅读量', compute="_compute_read_number", store=True) is_this_company = fields.Boolean(string='本公司', default=True) approval_user_id = fields.Many2one('res.users', string="审核人") approval_date = fields.Datetime(string="审核日期") is_sticky = fields.Boolean(string="是否置顶") urgency_level = fields.Selection([('normal', '普件'), ('urgent', '急件'), ('emergency', '特急件')], string='紧急度', default='normal') active = fields.Boolean('Active', default=True, tracking=True) work_end = fields.Boolean(string="结束") users_ids = fields.Many2many( 'res.users', relation='yuthon_notice_users_rel', column1='notice_id', column2='user_id', string='用户' ) self_users_ids = fields.Many2many( 'res.users', relation='yuthon_notice_current_users_rel', column1='notice_id', column2='user_id', string="当前经办人", default=lambda self: [(6, 0, [self.env.user.id])] ) # @api.onchange('users_ids') # def _onchange_users_ids(self): # today = date.today() # self.start_date = today # self.create_date1 = today is_company = fields.Boolean(string="本公司", default=True, compute='_compute_is_company', search='_search_part_of_company') publish_date = fields.Date(string='发布日期') @api.depends_context('uid', 'company') @api.depends('company_id') def _compute_is_company(self): active_company_ids = self.env.companies for employee in self: employee.is_company = employee.company_id in active_company_ids def _search_part_of_company(self, operator, value): company_ids = self.env.companies.ids if not value: operator = '!=' if operator == '=' else '=' if operator == '=': return [('company_id', 'in', company_ids)] else: return [('company_id', 'not in', company_ids)] @api.model_create_multi def create(self, vals_list): for i in vals_list: if i['department_ids'] == i['employee_ids'] == i['users_role_ids'] == i['users_role_group_ids']: raise UserError('请填写发布的范围') return super().create(vals_list) def _get_default_name_title(self): return self._description if self._description else self._name name_title = fields.Char(default=_get_default_name_title) @api.constrains('employee_ids', 'department_ids', 'users_role_ids', 'users_role_group_ids') def _constrains_employee_related_fields(self): """ 约束验证：当关联字段变化时，更新users_ids """ def get_department_employees(department_id): """递归获取部门及其子部门的员工对应的user_id""" valid_user_ids = set() # 查找当前部门的员工（使用id替代id.origin） employee_ids = self.env['hr.employee'].search([('department_id', '=', department_id.id)]) # 收集有效的user_id valid_user_ids.update([emp.user_id.id for emp in employee_ids if emp.user_id]) # 递归处理子部门 child_ids = self.env['hr.department'].search([('parent_id', '=', department_id.id)]) for child in child_ids: valid_user_ids.update(get_department_employees(child)) return valid_user_ids for notice in self: # 避免重复触发约束（更新users_ids时跳过本次约束检查） if notice.env.context.get('skip_constrains'): continue # 获取部门关联的用户ID department_user_ids = set() for department_id in notice.department_ids: department_user_ids.update(get_department_employees(department_id)) group_role_ids = notice.users_role_group_ids.user_role_ids all_users = set( notice.employee_ids.mapped('user_id').ids + group_role_ids.mapped('personnel_ids.user_id').ids + notice.users_role_ids.mapped('personnel_ids.user_id').ids + list(department_user_ids) ) notice.with_context(skip_constrains=True).users_ids = [(6, 0, list(all_users))] if notice.users_ids and notice.is_change_notice: conf_old = self.env['yuthon.confirm.users'].sudo().search([('notice_id', '=', notice.id)]) old_employee_ids = conf_old.mapped('employee_id.id') new_employee_ids = self.env['hr.employee'].sudo().search([ ('user_id', 'in', notice.users_ids.ids) ]).ids add_employee_ids = list(set(new_employee_ids) - set(old_employee_ids)) del_employee_ids = list(set(old_employee_ids) - set(new_employee_ids)) # 3. 处理删除： if del_employee_ids: del_conf = conf_old.filtered(lambda c: c.employee_id.id in del_employee_ids) del_conf.unlink() if add_employee_ids: for emp_id in add_employee_ids: employee = self.env['hr.employee'].sudo().browse(emp_id) if not employee: continue # 防重复：如果已经存在该员工的确认记录，跳过 exist = self.env['yuthon.confirm.users'].sudo().search([ ('notice_id', '=', notice.id), ('employee_id', '=', emp_id), ], limit=1) if exist: continue self.env['yuthon.confirm.users'].sudo().create({ 'notice_id': notice.id, 'employee_id': emp_id, 'notice_code': notice.code, 'notice_name': notice.name, 'release_date': notice.create_date1, 'state': 'to_sent', }) def get_no_read_count(self): # 只统计当前用户发布范围内的未读公告，与列表 domain 保持一致 return self.search_count([ ('users_ids', 'in', self.env.uid), ('is_notice_roger', '=', 'no'), ('notice_state', '=', 'yes'), ]) def read(self, fields=None, load='_classic_read'): # 标记是否真的有未读变已读，避免没变化也触发菜单刷新 changed = False for con in self: # 只有全局状态仍是未读时才处理 if con.is_notice_roger == 'no': # 查找当前用户对应的公告确认记录 con_id = self.env['yuthon.confirm.users'].search([ ('notice_id', '=', con.id), ('employee_id', '=', con.env.user.employee_id.id) ]) # 确认是当前员工本人，再标记为已读 if con_id.employee_id == con.env.user.employee_id: con.is_notice_roger = 'yes' con_id.write({'state': 'yes'}) changed = True res = super(YuthonNotice, self).read(fields=fields, load=load) # 如果有记录从 未读 变为 已读，清除菜单缓存并通知前端刷新角标 if changed: self.env.registry.clear_cache() self.env['sun.badge.menu.mixin'].sudo().trigger_menu_reload() return res @api.depends('read_number') def _compute_read_number(self): notice_confirm_ids = self.env['yuthon.confirm.users'].search( [('notice_id.id', '=', self.id), ('state', '=', 'yes')]) for read in self: if notice_confirm_ids: read.read_number = len(notice_confirm_ids) else: read.read_number = 0 def on_sequence(self): self.sequence = 1 @api.onchange('is_sticky') def _compute_sequence(self): for i in self: if i.is_sticky: i.sequence = 1 else: i.sequence = 0 def no_sequence(self): self.sequence = 0 def notice_publish(self): """发布通知公告：给 yuthon.confirm.users 中尚未发送的记录发企微消息。 - 每人有且仅发一条：msg_id 为空才发，已发过的不重发 - 发送成功才写回 msg_id（企微返回的msgid）并更新状态为"待确认" - 单条失败不影响其他记录，失败记录保持"待发送"状态 """ self.ensure_one() if self.notice_state == 'yes': return self.notice_state = 'yes' if not self.is_change_notice: return # 重新发布时，已发送过消息的记录重置为"待确认"，等待用户重新确认（不重发消息） self.env['yuthon.confirm.users'].sudo().search([ ('notice_id', '=', self.id), ('msg_id', '!=', False), ]).write({'state': 'no'}) # 只发送尚未收到消息的记录（msg_id 为空），保证每人有且仅一条 confirm_records = self.env['yuthon.confirm.users'].sudo().search([ ('notice_id', '=', self.id), ('msg_id', '=', False), ]) sent_count = 0 fail_count = 0 for rec in confirm_records: employee = rec.employee_id if not employee or not employee.wecom_userid: _logger.warning("公告[%s]确认记录%s跳过：员工或企微userid为空", self.name, rec.id) continue url = f'https://phone.thtzjt.com/phone/Notice/{self.id}?employee_id={employee.id}' try: result_data = self.sudo().env["wecom.apps"].sync_send_message_textcard( category='notice', touser=employee.wecom_userid, title="通知公告", description=f"{self.name} 发布人: {self.users_id.name or ''}", url=url, btntxt="详细信息" ) except Exception: _logger.exception("公告[%s]给员工%s发送企微消息异常", self.name, employee.name) fail_count += 1 continue # 发送成功：返回元组第一项为 True 且携带 msgid，才写回 msg_id 并更新状态 if result_data and result_data[0] and isinstance(result_data[1], dict) and result_data[1].get('msgid'): rec.sudo().write({ 'msg_id': result_data[1].get('msgid'), 'state': 'no', }) sent_count += 1 else: _logger.warning("公告[%s]给员工%s发送失败，返回：%s", self.name, employee.name, result_data) fail_count += 1 _logger.info("公告[%s]发布完成，成功发送%s条，失败%s条", self.name, sent_count, fail_count) # 发布完毕弹窗提醒 return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': '发布完毕', 'message': f'已成功发送 {sent_count} 条消息' + (f'，{fail_count} 条失败' if fail_count else ''), 'type': 'success', 'sticky': False, }, } def look_up_notice_users(self): return { "type": "ir.actions.act_window", "name": "通知公告确认信息", "domain": [("notice_id", "=", self.id)], "res_model": "yuthon.confirm.users", "view_mode": "tree", "context": { "search_default_group_state": True, }, } def write(self, vals): """重写write方法，检测work_end字段变化，触发AI主动感知""" result = super().write(vals) # 检测work_end从False变为True if 'work_end' in vals and vals['work_end']: for record in self: if record.work_end: # 确保当前值确实是True self._trigger_ai_guide(record) return result def _trigger_ai_guide(self, record): """触发AI主动感知：查找匹配的规则，推送消息到聊天框""" # 查找匹配的规则 rules = self.env['ai.rule'].sudo().search([ ('rule_type', '=', 'active_guide'), ('guide_model', '=', self._name), ('guide_field', '=', 'work_end'), ('guide_field_value', '=', 'True'), ('active', '=', True), ]) for rule in rules: # 生成提醒内容（替换变量） message = rule.guide_message or f'{record.name} 流程已结束' message = message.replace('{record_name}', record.name or '') message = message.replace('{user_name}', self.env.user.name or '') # 自动组装按钮动作（后台自动注入 record_id） action_config = {} if rule.guide_method_name: action_config = { 'type': 'execute_tool', 'tool': rule.guide_method_name, 'params': { 'record_id': record.id, 'model': self._name, } } # 查找或创建AI对话（用当前用户的默认对话） conversation = self.env['ai.conversation'].sudo().search([ ('create_uid', '=', self.env.user.id), ], limit=1, order='write_date desc') if not conversation: conversation = self.env['ai.conversation'].sudo().create({ 'name': 'AI 对话', }) # 创建AI消息（带action） self.env['ai.message'].sudo().create({ 'conversation_id': conversation.id, 'role': 'assistant', 'content': message, 'action': json.dumps({ 'type': 'guide_button', 'label': rule.guide_button_label or '确认执行', 'action': action_config, }, ensure_ascii=False) if action_config else None, }) # TODO: 通过bus.bus推送到前端，让聊天框自动显示新消息 # self.env['bus.bus'].sudo()._sendone(self.env.user.partner_id, 'ai_guide_notification', { # 'conversation_id': conversation.id, # 'message': message, # }) def stop_notice(self): self.write({ 'notice_state': 'stop', 'end_date': fields.Date.today(), }) def action_print_notice(self): """打印通知公告""" self.ensure_one() report = self.env.ref('yuthon_notice.action_notice_print_report') return report.report_action(self) class YuthonCsTree(models.Model): _name = 'yuthon.cs.tree' _description = 'Yuthon Customer Service Tree' name = fields.Char(string='Name', required=True)