jikimo/test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增 Redis 工具与控制器，更新同步逻辑相关文件

Browse Source
This commit is contained in:
陈赓
2025-07-15 11:28:12 +08:00
parent 7fca59322e
commit cdbc277a94
3 changed files with 238 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
requirements.txt
View File

@@ -4,4 +4,4 @@ wechatpy==1.8.18
pycryptodome==3.22.0
openupgradelib==3.10.0
opcua==0.98.13
openpyxl
openpyxl

1
sf_mrs_connect/controllers/__init__.py
View File

@@ -1 +1,2 @@
from . import controllers
from . import sync_controller

372
sf_mrs_connect/models/sync_common.py
View File

@@ -5,9 +5,11 @@ import base64
import traceback
import requests
from odoo import models
from odoo import models,api,fields
from odoo.exceptions import ValidationError
from odoo.addons.sf_base.commons.common import Common
from .redis_utils import RedisClient
from .common import Common
from odoo.addons.jikimo_sf.sf_base.commons.common import Common
_logger = logging.getLogger(__name__)
@@ -1506,149 +1508,157 @@ class SyncMulti_Mounting_Type(models.Model):
raise ValidationError("联装类型认证未通过")
class SyncFixtureModel(models.Model):
_inherit = 'sf.fixture.model'
_description = '同步夹具型号列表'
# class SyncFixtureModel(models.Model):
# _inherit = 'sf.fixture.model'
# _description = '同步夹具型号列表'
url = '/api/fixture_model/list'
# url = '/api/fixture_model/list'
# 定时同步夹具型号列表
def sync_fixture_model_yesterday(self):
config = self.env['res.config.settings'].get_values()
headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
strUrl = config['sf_url'] + self.url
r = requests.post(strUrl, json={}, data=None, headers=headers)
r = r.json()
result = json.loads(r['result'])
if result['status'] == 1:
if result.get('fixture_model_yesterday_list'):
for item in result['fixture_model_yesterday_list']:
if item:
fixture_model = self.search([("code", '=', item['code']), ('active', 'in', [True, False])])
val = {
"name": item['name'],
"code": item['code'],
"fixture_material_id": self.env['sf.fixture.material'].search(
[('code', '=', item['fixture_material_code'])]).id,
"multi_mounting_type_id": self.env['sf.multi_mounting.type'].search(
[('code', '=', item['multi_mounting_type_code'])]).id,
"brand_id": self.env['sf.machine.brand'].search([('code', '=', item['brand_code'])]).id,
"model_file": '' if not item['model_file'] else base64.b64decode(item['model_file']),
"status": item['status'],
"active": item['active'],
}
if not fixture_model:
self.create(val)
else:
fixture_model.write(val)
# # 定时同步夹具型号列表
# def sync_fixture_model_yesterday(self):
# config = self.env['res.config.settings'].get_values()
# headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
# strUrl = config['sf_url'] + self.url
# r = requests.post(strUrl, json={}, data=None, headers=headers)
# r = r.json()
# result = json.loads(r['result'])
# if result['status'] == 1:
# if result.get('fixture_model_yesterday_list'):
# for item in result['fixture_model_yesterday_list']:
# if item:
# fixture_model = self.search([("code", '=', item['code']), ('active', 'in', [True, False])])
# val = {
# "name": item['name'],
# "code": item['code'],
# "fixture_material_id": self.env['sf.fixture.material'].search(
# [('code', '=', item['fixture_material_code'])]).id,
# "multi_mounting_type_id": self.env['sf.multi_mounting.type'].search(
# [('code', '=', item['multi_mounting_type_code'])]).id,
# "brand_id": self.env['sf.machine.brand'].search([('code', '=', item['brand_code'])]).id,
# "model_file": '' if not item['model_file'] else base64.b64decode(item['model_file']),
# "status": item['status'],
# "active": item['active'],
# }
# if not fixture_model:
# self.create(val)
# else:
# fixture_model.write(val)
else:
raise ValidationError("夹具型号认证未通过")
# else:
# raise ValidationError("夹具型号认证未通过")
# 定时同步所有夹具型号列表
def sync_all_fixture_model(self):
config = self.env['res.config.settings'].get_values()
headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
strUrl = config['sf_url'] + self.url
r = requests.post(strUrl, json={}, data=None, headers=headers)
r = r.json()
result = json.loads(r['result'])
# print('result:%s' % result)
if result['status'] == 1:
if result.get('fixture_model_all_list'):
for item in result['fixture_model_all_list']:
if item:
fixture_model = self.search([('code', '=', item['code']), ('active', 'in', [True, False])])
val = {
"name": item['name'],
"code": item['code'],
"fixture_material_id": self.env['sf.fixture.material'].search(
[('code', '=', item['fixture_material_code'])]).id,
"multi_mounting_type_id": self.env['sf.multi_mounting.type'].search(
[('code', '=', item['multi_mounting_type_code'])]).id,
"brand_id": self.env['sf.machine.brand'].search([('code', '=', item['brand_code'])]).id,
"model_file": '' if not item['model_file'] else base64.b64decode(item['model_file']),
"status": item['status'],
"active": item['active'],
}
if not fixture_model:
self.create(val)
else:
fixture_model.write(val)
else:
raise ValidationError("夹具型号认证未通过")
# # 定时同步所有夹具型号列表
# def sync_all_fixture_model(self):
# config = self.env['res.config.settings'].get_values()
# headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
# strUrl = config['sf_url'] + self.url
# r = requests.post(strUrl, json={}, data=None, headers=headers)
# r = r.json()
# result = json.loads(r['result'])
# # print('result:%s' % result)
# if result['status'] == 1:
# if result.get('fixture_model_all_list'):
# for item in result['fixture_model_all_list']:
# if item:
# fixture_model = self.search([('code', '=', item['code']), ('active', 'in', [True, False])])
# val = {
# "name": item['name'],
# "code": item['code'],
# "fixture_material_id": self.env['sf.fixture.material'].search(
# [('code', '=', item['fixture_material_code'])]).id,
# "multi_mounting_type_id": self.env['sf.multi_mounting.type'].search(
# [('code', '=', item['multi_mounting_type_code'])]).id,
# "brand_id": self.env['sf.machine.brand'].search([('code', '=', item['brand_code'])]).id,
# "model_file": '' if not item['model_file'] else base64.b64decode(item['model_file']),
# "status": item['status'],
# "active": item['active'],
# }
# if not fixture_model:
# self.create(val)
# else:
# fixture_model.write(val)
# else:
# raise ValidationError("夹具型号认证未通过")
class SyncfixtureMaterialsBasicParameters(models.Model):
_inherit = 'sf.fixture.materials.basic.parameters'
_description = '同步夹具型号基本参数列表'
# class SyncfixtureMaterialsBasicParameters(models.Model):
# _inherit = 'sf.fixture.materials.basic.parameters'
# _description = '同步夹具型号基本参数列表'
url = '/api/fixture_parameters/list'
# url = '/api/fixture_parameters/list'
# 定时同步夹具型号基本信息
def sync_fixture_materials_basic_parameters_yesterday(self):
config = self.env['res.config.settings'].get_values()
headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
strUrl = config['sf_url'] + self.url
r = requests.post(strUrl, json={}, data=None, headers=headers)
r = r.json()
result = json.loads(r['result'])
if result['status'] == 1:
if result.get('fixture_parameters_yesterday_list'):
all_list = result.get('fixture_parameters_yesterday_list')
if all_list.get('zero_chuck_all_list'):
self._write_or_create(all_list.get('zero_chuck_yesterday_list'), '零点卡盘')
if all_list.get('zero_tray_all_list'):
self._write_or_create(all_list.get('zero_tray_yesterday_list'), '零点托盘')
if all_list.get('pneumatic_fixture_all_list'):
self._write_or_create(all_list.get('pneumatic_fixture_yesterday_list'), '气动夹具')
if all_list.get('jaw_vice_all_list'):
self._write_or_create(all_list.get('jaw_vice_yesterday_list'), '虎钳夹具')
if all_list.get('magnet_fixture_all_list'):
self._write_or_create(all_list.get('magnet_fixture_yesterday_list'), '磁吸夹具')
if all_list.get('adapter_board_all_list'):
self._write_or_create(all_list.get('adapter_board_yesterday_list'), '转接板(锁板)夹具')
if all_list.get('scroll_chuck_all_list'):
self._write_or_create(all_list.get('scroll_chuck_yesterday_list'), '三爪卡盘')
else:
raise ValidationError("夹具型号基本参数认证未通过")
# # 定时同步夹具型号基本信息
# def sync_fixture_materials_basic_parameters_yesterday(self):
# config = self.env['res.config.settings'].get_values()
# headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
# strUrl = config['sf_url'] + self.url
# r = requests.post(strUrl, json={}, data=None, headers=headers)
# r = r.json()
# result = json.loads(r['result'])
# if result['status'] == 1:
# if result.get('fixture_parameters_yesterday_list'):
# all_list = result.get('fixture_parameters_yesterday_list')
# if all_list.get('zero_chuck_all_list'):
# self._write_or_create(all_list.get('zero_chuck_yesterday_list'), '零点卡盘')
# if all_list.get('zero_tray_all_list'):
# self._write_or_create(all_list.get('zero_tray_yesterday_list'), '零点托盘')
# if all_list.get('pneumatic_fixture_all_list'):
# self._write_or_create(all_list.get('pneumatic_fixture_yesterday_list'), '气动夹具')
# if all_list.get('jaw_vice_all_list'):
# self._write_or_create(all_list.get('jaw_vice_yesterday_list'), '虎钳夹具')
# if all_list.get('magnet_fixture_all_list'):
# self._write_or_create(all_list.get('magnet_fixture_yesterday_list'), '磁吸夹具')
# if all_list.get('adapter_board_all_list'):
# self._write_or_create(all_list.get('adapter_board_yesterday_list'), '转接板(锁板)夹具')
# if all_list.get('scroll_chuck_all_list'):
# self._write_or_create(all_list.get('scroll_chuck_yesterday_list'), '三爪卡盘')
# if all_list.get('air_tray_all_list'):
# self._write_or_create(all_list.get('air_tray_all_list'),'气吹托盘')
# if all_list.get('magnet_tray_all_list'):
# self._write_or_create(all_list.get('magnet_tray_all_list'),'磁吸托盘')
# else:
# raise ValidationError("夹具型号基本参数认证未通过")
# 定时同步所有夹具型号基本信息
def sync_all_fixture_materials_basic_parameters(self):
config = self.env['res.config.settings'].get_values()
headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
strUrl = config['sf_url'] + self.url
r = requests.post(strUrl, json={}, data=None, headers=headers)
r = r.json()
result = json.loads(r['result'])
if result['status'] == 1:
if result.get('fixture_parameters_all_list'):
all_list = result.get('fixture_parameters_all_list')
if all_list.get('zero_chuck_all_list'):
self._write_or_create(all_list.get('zero_chuck_all_list'), '零点卡盘')
if all_list.get('zero_tray_all_list'):
self._write_or_create(all_list.get('zero_tray_all_list'), '零点托盘')
if all_list.get('pneumatic_fixture_all_list'):
self._write_or_create(all_list.get('pneumatic_fixture_all_list'), '气动夹具')
if all_list.get('jaw_vice_all_list'):
self._write_or_create(all_list.get('jaw_vice_all_list'), '虎钳夹具')
if all_list.get('magnet_fixture_all_list'):
self._write_or_create(all_list.get('magnet_fixture_all_list'), '磁吸夹具')
if all_list.get('adapter_board_all_list'):
self._write_or_create(all_list.get('adapter_board_all_list'), '转接板(锁板)夹具')
if all_list.get('scroll_chuck_all_list'):
self._write_or_create(all_list.get('scroll_chuck_all_list'), '三爪卡盘')
else:
raise ValidationError("夹具型号基本参数认证未通过")
# # 定时同步所有夹具型号基本信息
# def sync_all_fixture_materials_basic_parameters(self):
# config = self.env['res.config.settings'].get_values()
# headers = Common.get_headers(self, config['token'], config['sf_secret_key'])
# strUrl = config['sf_url'] + self.url
# r = requests.post(strUrl, json={}, data=None, headers=headers)
# r = r.json()
# result = json.loads(r['result'])
# if result['status'] == 1:
# if result.get('fixture_parameters_all_list'):
# all_list = result.get('fixture_parameters_all_list')
# if all_list.get('zero_chuck_all_list'):
# self._write_or_create(all_list.get('zero_chuck_all_list'), '零点卡盘')
# if all_list.get('zero_tray_all_list'):
# self._write_or_create(all_list.get('zero_tray_all_list'), '零点托盘')
# if all_list.get('pneumatic_fixture_all_list'):
# self._write_or_create(all_list.get('pneumatic_fixture_all_list'), '气动夹具')
# if all_list.get('jaw_vice_all_list'):
# self._write_or_create(all_list.get('jaw_vice_all_list'), '虎钳夹具')
# if all_list.get('magnet_fixture_all_list'):
# self._write_or_create(all_list.get('magnet_fixture_all_list'), '磁吸夹具')
# if all_list.get('adapter_board_all_list'):
# self._write_or_create(all_list.get('adapter_board_all_list'), '转接板(锁板)夹具')
# if all_list.get('scroll_chuck_all_list'):
# self._write_or_create(all_list.get('scroll_chuck_all_list'), '三爪卡盘')
# if all_list.get('air_tray_all_list'):
# self._write_or_create(all_list.get('air_tray_all_list'),'气吹托盘')
# if all_list.get('magnet_tray_all_list'):
# self._write_or_create(all_list.get('magnet_tray_all_list'),'磁吸托盘')
# else:
# raise ValidationError("夹具型号基本参数认证未通过")
def _write_or_create(self, fixture_parameters_list, material_name):
for item in fixture_parameters_list:
if item:
basic_parameters = self.search([('code', '=', item.get('code')), ('active', 'in', [True, False])])
if not basic_parameters:
self.create(self._get_basic_parameters_list(item, material_name))
else:
basic_parameters.write(self._get_basic_parameters_list(item, material_name))
# def _write_or_create(self, fixture_parameters_list, material_name):
# for item in fixture_parameters_list:
# if item:
# basic_parameters = self.search([('code', '=', item.get('code')), ('active', 'in', [True, False])])
# if not basic_parameters:
# self.create(self._get_basic_parameters_list(item, material_name))
# else:
# basic_parameters.write(self._get_basic_parameters_list(item, material_name))
class SyncFunctionalFixtureType(models.Model):
@@ -3230,4 +3240,94 @@ class EmbryoRedundancySync(models.Model):
"height": item['height'],
"active": item['active'],
"remark": item['remark'],
})
})
class SyncFixtureModel(models.Model):
_inherit = 'sf.fixture.model'
_description = 'Redis 优先同步夹具型号'
def sync_all_fixture_model(self):
rc = RedisClient()
key = 'mrs:fixture_model_all_list'
all_list = rc.get_json(key)
if not all_list:
raise ValidationError(f"Redis 中未找到 key={key}")
for item in all_list:
if not item or not item.get('code'):
continue
record = self.search([('code', '=', item['code'])], limit=1)
vals = {
'name': item['name'],
'code': item['code'],
'fixture_material_id': self.env['sf.fixture.material']
.search([('code', '=', item['fixture_material_code'])], limit=1).id,
'multi_mounting_type_id': self.env['sf.multi_mounting.type']
.search([('code', '=', item['multi_mounting_type_code'])], limit=1).id,
'brand_id': self.env['sf.machine.brand']
.search([('code', '=', item['brand_code'])], limit=1).id,
'model_file': base64.b64decode(item['model_file']) if item.get('model_file') else False,
'status': item['status'],
'active': item['active'],
}
if record:
record.write(vals)
else:
self.create(vals)
class SyncfixtureMaterialsBasicParameters(models.Model):
_inherit = 'sf.fixture.materials.basic.parameters'
_description = 'Redis 优先同步夹具基本参数'
def sync_all_fixture_materials_basic_parameters(self):
rc = RedisClient()
key = 'mrs:fixture_param_all_list'
all_list = rc.get_json(key)
if not all_list:
raise ValidationError(f"Redis 中未找到 key={key}")
def _sync_list(param_list, material_name):
for item in param_list or []:
if not item or not item.get('code'):
continue
record = self.search([('code', '=', item['code'])], limit=1)
vals = self._get_basic_parameters_list(item, material_name)
if record:
record.write(vals)
else:
self.create(vals)
_sync_list(all_list.get('zero_chuck_all_list'), '零点卡盘')
_sync_list(all_list.get('zero_tray_all_list'), '零点托盘')
_sync_list(all_list.get('pneumatic_fixture_all_list'), '气动夹具')
_sync_list(all_list.get('jaw_vice_all_list'), '虎钳夹具')
_sync_list(all_list.get('magnet_fixture_all_list'), '磁吸夹具')
_sync_list(all_list.get('adapter_board_all_list'), '转接板(锁板)夹具')
_sync_list(all_list.get('scroll_chuck_all_list'), '三爪卡盘')
_sync_list(all_list.get('air_tray_all_list'), '气吹托盘')
_sync_list(all_list.get('magnet_tray_all_list'), '磁吸托盘')
def _get_basic_parameters_list(self, item, material_name):
"""
统一结构化 item 数据，供写入模型字段使用（你应当根据 material_name 自定义字段映射）
"""
return {
'name': item.get('name'),
'code': item.get('code'),
'length': item.get('length'),
'width': item.get('width'),
'height': item.get('height'),
'diameter': item.get('diameter'),
'weight': item.get('weight'),
'fixture_model_id': self.env['sf.fixture.model'].search([('code', '=', item.get('fixture_model_code'))], limit=1).id,
'materials_model_id': self.env['sf.materials.model'].search([('code', '=', item.get('material_code'))], limit=1).id,
'active': item.get('active', True),
# 你可以根据 material_name 判断类型并补充字段
}