ai动态报表

This commit is contained in:
李鹏宇
2026-05-29 14:16:01 +08:00
parent 805e26300c
commit 0cfa751878
4 changed files with 140 additions and 1 deletions
+70
View File
@@ -5,6 +5,30 @@ from odoo import fields, models, api
_logger = logging.getLogger(__name__)
def _strip_face_template(message_data_str):
"""移除 message_data JSON 中的 face_template 字段(图片base64太长)"""
if not message_data_str:
return message_data_str
try:
data = json.loads(message_data_str)
if isinstance(data, dict) and 'face_template' in data:
data.pop('face_template')
_logger.info("已移除 face_template 字段")
return json.dumps(data, ensure_ascii=False)
if isinstance(data, list):
changed = False
for item in data:
if isinstance(item, dict) and 'face_template' in item:
item.pop('face_template')
changed = True
if changed:
_logger.info("已移除列表中的 face_template 字段")
return json.dumps(data, ensure_ascii=False)
return message_data_str
except (json.JSONDecodeError, TypeError):
return message_data_str
class IotMessage(models.Model):
_name = 'iot.message'
_description = 'mqtt 消息,发送和接收'
@@ -21,6 +45,52 @@ class IotMessage(models.Model):
# 响应信息
res_data = fields.Text(string='发送后的响应信息')
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
if vals.get('message_data'):
vals['message_data'] = _strip_face_template(vals['message_data'])
return super().create(vals_list)
def write(self, vals):
if vals.get('message_data'):
vals['message_data'] = _strip_face_template(vals['message_data'])
return super().write(vals)
def action_clear_face_template(self):
"""批量清除已有记录的 face_template 参数"""
count = 0
for record in self:
if not record.message_data:
continue
try:
data = json.loads(record.message_data)
changed = False
if isinstance(data, dict) and 'face_template' in data:
data.pop('face_template')
changed = True
elif isinstance(data, list):
for item in data:
if isinstance(item, dict) and 'face_template' in item:
item.pop('face_template')
changed = True
if changed:
record.write({'message_data': json.dumps(data, ensure_ascii=False)})
count += 1
except (json.JSONDecodeError, TypeError):
continue
_logger.info(f"已清理 {count} 条记录的 face_template")
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': '完成',
'message': f'已清理 {count} 条记录的 face_template',
'sticky': False,
'type': 'success',
}
}
@api.model
def to_reply(self, data):
_logger.info(f'返回消息:{data}')
+3
View File
@@ -8,6 +8,9 @@
<field name="model">iot.message</field>
<field name="arch" type="xml">
<tree string="Iot 消息">
<header>
<button name="action_clear_face_template" string="清除 face_template" type="object"/>
</header>
<field name="name"/>
<field name="topic"/>
<field name="device_sn"/>
+2 -1
View File
@@ -9,4 +9,5 @@ from . import res_partner
from . import base_month_year
from . import system_logs
from . import res_users
from . import mail_thread
from . import mail_thread
from . import ir_ui_menu
+65
View File
@@ -0,0 +1,65 @@
# -*- coding: utf-8 -*-
from odoo import api, fields, models, _
import pypinyin
import re
class IrUiMenu(models.Model):
_inherit = 'ir.ui.menu'
name_pinyin = fields.Char(
string='菜单名拼音',
compute='_compute_name_pinyin',
store=True,
index=True,
)
@api.depends('name')
def _compute_name_pinyin(self):
for menu in self:
if menu.name:
# 全拼:如 "通知公告" → "tongzhigonggao"
full_pinyin = ''.join(pypinyin.lazy_pinyin(menu.name)).lower()
# 首字母:如 "通知公告" → "tzgg"
first_letters = ''.join([
p[0] for p in pypinyin.pinyin(menu.name)
]).lower()
menu.name_pinyin = f"{full_pinyin} {first_letters}"
else:
menu.name_pinyin = False
def name_search(self, name='', args=None, operator='ilike', limit=100):
"""
重写 name_search 支持全拼/首字母模糊搜索菜单
"""
if args is None:
args = []
if name and operator in ('ilike', 'like', '=', '=like'):
# 清理输入:去掉空格、转小写
search_term = name.strip().lower().replace(' ', '')
if not search_term:
return super().name_search(name=name, args=args, operator=operator, limit=limit)
# 生成搜索词的拼音和首字母
search_pinyin = ''.join(pypinyin.lazy_pinyin(search_term)).lower()
search_first = ''.join([p[0] for p in pypinyin.pinyin(search_term)]).lower()
# 构建拼音搜索域:匹配 name_pinyin 字段
pinyin_domain = [
'|',
('name_pinyin', 'ilike', search_term), # 直接搜索词
('name_pinyin', 'ilike', search_pinyin), # 全拼搜索
]
# 如果首字母不同,也加上
if search_first != search_pinyin:
pinyin_domain = ['|'] + pinyin_domain + [('name_pinyin', 'ilike', search_first)]
# 组合:原名搜索 OR 拼音搜索
full_domain = ['|'] + expression.AND([args, [('name', operator, name)]]) + expression.AND([args, pinyin_domain])
# 去重:用 browse 把结果合并(优先按原名匹配排序)
name_results = self.search(expression.AND([args, [('name', operator, name)]]), limit=limit)
pinyin_results = self.search(expression.AND([args, pinyin_domain]), limit=limit)
all_ids = list(dict.fromkeys(name_results.ids + pinyin_results.ids))[:limit]
return self.browse(all_ids).name_get()
return super().name_search(name=name, args=args, operator=operator, limit=limit)