diff --git a/appstore/sbomp_test_ip/README.md b/appstore/sbomp_test_ip/README.md new file mode 100644 index 0000000000000000000000000000000000000000..ff2b84188a72e94caac4b839c9cf83cb1ee0fc39 --- /dev/null +++ b/appstore/sbomp_test_ip/README.md @@ -0,0 +1,10 @@ +一个测试IP是否能用的模块 + +应用场景 + 1、需要对接的终端设备很多、且终端设备IP大部分不相同的场景 + 2、对sql_lite使用的一种演示 + 3、DTCLOUD 如何调用另外一个模块的函数 + +使用方法 + self.env['sbomp.test.ip'].test_net_is_used(in_ip='127.0.0.1', in_port=503, connect_mode='socket') + self.env['sbomp.test.ip'].error_connect_ip(in_ip='127.0.0.1', in_port=503, connect_mode='socket',connect_error_info='未知错误') \ No newline at end of file diff --git a/appstore/sbomp_test_ip/__init__.py b/appstore/sbomp_test_ip/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..0650744f6bc69b9f0b865e8c7174c813a5f5995e --- /dev/null +++ b/appstore/sbomp_test_ip/__init__.py @@ -0,0 +1 @@ +from . import models diff --git a/appstore/sbomp_test_ip/__manifest__.py b/appstore/sbomp_test_ip/__manifest__.py new file mode 100644 index 0000000000000000000000000000000000000000..647da2d1e4f27a009c5363be3906e583d184e770 --- /dev/null +++ b/appstore/sbomp_test_ip/__manifest__.py @@ -0,0 +1,36 @@ +# -*- coding: utf-8 -*- +# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&& +# DTCloud360 +# QQ:35350428 +# 邮件:35350428@qq.com +# 手机：13584935775 +# 作者：'扶程星云' +# 公司网址： http://www.dtcloud360.com/ +# Copyright 中亿丰信息科技(苏州)有限公司 +# 日期：2021/10/15 +# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&& + +{ + 'name': '测试IP地址', + 'summary': """ + sbomp_test_ip + """, + 'category': 'Tools', + 'sequence': 3, + 'author': '扶程星云', + 'website': "http://www.dtcloud360.com/", + 'depends': ['base'], + 'version': '0.1', + 'data': [ + # 'data/sbomp_test_ip_reset.xml', + ], + 'installable': True, + 'application': True, + 'auto_install': True, + 'description': """ + 有时候，我们需要测试IP是否可用，然后再进行操作，特别是大量重复IP， + 我们在一定时间内是不需要重复检查一个IP是否可用的，如果一定时间内是不通的， + 我们会直接返回不通，直到再次测试时间到达。 +""", + +} diff --git a/appstore/sbomp_test_ip/data/sbomp_test_ip_reset.xml b/appstore/sbomp_test_ip/data/sbomp_test_ip_reset.xml new file mode 100644 index 0000000000000000000000000000000000000000..49d5fcbbd13332e6b9d6166397485f13c23e6a68 --- /dev/null +++ b/appstore/sbomp_test_ip/data/sbomp_test_ip_reset.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + + + diff --git a/appstore/sbomp_test_ip/models/__init__.py b/appstore/sbomp_test_ip/models/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..47f9415719e6647facc4faa733fbda8443bcdc44 --- /dev/null +++ b/appstore/sbomp_test_ip/models/__init__.py @@ -0,0 +1,2 @@ +from . import sbomp_test_ip + diff --git a/appstore/sbomp_test_ip/models/sbomp_test_ip.py b/appstore/sbomp_test_ip/models/sbomp_test_ip.py new file mode 100644 index 0000000000000000000000000000000000000000..fae24a37f523306a6b80867a4e5da1075e05d3ac --- /dev/null +++ b/appstore/sbomp_test_ip/models/sbomp_test_ip.py @@ -0,0 +1,246 @@ +# -*- coding: utf-8 -*- +from datetime import timedelta + +from dtcloud import models, fields, api + +from dtcloud.tools.sql_lite_tools import nebula_delete_sql_lite, nebula_create_sql_lite + + + +def net_is_used(in_ip='127.0.0.1', in_port=503): + ip = in_ip + table_name = 'sbomp_test_ip' + db_name = 'sbomp.db' + connect_datetime = get_location_time() + reconnect_datetime = get_location_time(add_minutes=10) + # 如果传进来的是 '192.168.3.3:503' + ip_port_list = in_ip.split(':') + if len(ip_port_list) == 2: + # print(f'ip_port_list:{ip_port_list} in_ip:{in_ip}') + ip_tmp = ip_port_list[0] + ip = ip_tmp.replace('\'', '') + port_tmp = ip_port_list[1] + port = port_tmp.replace('\'', '') + # print(f'ip_port_list ip_port_list :{ip_port_list} in_ip:{in_ip}, ip:{ip},port:{port}') + port = int(port) + else: # 如果传进来的端口是字符 + if type(in_port) == str: + port = int(in_port) + else: + port = in_port + + # 测试本地IP是否正常 + # 先查看有没有这条记录 + ip_address = str(ip) + str_port = str(port) + + where_values = 'ip_address=? and port=?' + in_parameters = (ip_address, str_port) + check_table_sbomp_test_ip() + select_sql = get_select_sql_base(table_name=table_name, select_values='id,connect_state,reconnect_datetime', + where_values=where_values) + re_state, db_list = nebula_select_sql_lite(select_sql, db_name=db_name, in_parameters=in_parameters) + if re_state: # 有返回值,正常 + for db_one in db_list: # 如果是断开状态，而且日期是小于重新检查时间，则直接返回False + s_none_format = "%Y-%m-%d %H:%M:%S" + db_one_connect_state = bool(db_one[1]) + db_one_reconnect_datetime_str = db_one[ + 2] # 2021-03-17 03:10:42.817235+08:00' does not match format '%Y-%m-%d %H:%M:%S.%f%z' + + s_tz = db_one_reconnect_datetime_str.split('+')[1] + s_format = s_none_format + '.%f+' + s_tz + db_one_reconnect_datetime = datetime.strptime(db_one_reconnect_datetime_str, s_format) + + # db_one_reconnect_datetime = datetime.strptime(db_one_reconnect_datetime_str, + # "%Y-%m-%d %H:%M:%S.%f%z") # "%Y-%m-%d %H:%M:%S.%f%z Y-%m-%dT%H:%M:%S.%f + + # print(f'db_one_reconnect_datetime:') + + db_one_reconnect_datetime = datetime.strptime(db_one_reconnect_datetime.strftime(s_none_format), + s_none_format) + + connect_datetime = datetime.strptime(connect_datetime.strftime(s_none_format), s_none_format) + + if (not db_one_connect_state) and (connect_datetime <= db_one_reconnect_datetime): + d_temp = db_one_reconnect_datetime - connect_datetime + print(f'net_is_used 离下次检查时间还有{d_temp}') + return False + + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + s.connect((ip, port)) + s.shutdown(1) + print('net_is_used %s:%d is used' % (ip, port)) + return True + except Exception as e: + print(f'net_is_used 目标地址:{ip},目标端口:{port}出错，出错信息:{e}') + connect_state = False + if re_state: + if len(db_list): # 如果有返回值,更新数据 + set_values = 'connect_datetime=?,reconnect_datetime=?,connect_state=?' + update_sql = get_update_sql_base(table_name=table_name, set_values=set_values, + where_values=where_values) + in_parameters = (connect_datetime, reconnect_datetime, connect_state, ip_address, str_port) + re_state, re_info = nebula_update_sql_lite(update_sql, db_name=db_name, in_parameters=in_parameters) + else: # 是新记录，需要插入 + set_parameters = 'ip_address,port,connect_datetime,reconnect_datetime,connect_state' + insert_sql = get_insert_sql_base(table_name=table_name, set_parameters=set_parameters) + in_parameters = (ip_address, str_port, connect_datetime, reconnect_datetime, connect_state) + re_state, re_info = nebula_update_sql_lite(update_sql=insert_sql, db_name=db_name, + in_parameters=in_parameters) + print(f'net_is_used re_state:{re_state}, db_list:{db_list}') + return False + +class sbomp_test_ip(models.Model): + _name = 'sbomp.test.ip' + _description = '测试IP地址' + _order = 'id desc' + + name = fields.Char(string='IP地址信息', copy=False, index=True, default=lambda self: u'New') + ip_address = fields.Char(string='IP地址', copy=False, index=True) + port = fields.Char(string='端口号', copy=False) + slave_number = fields.Char(string='从站地址', copy=False) + connect_datetime = fields.Datetime(string='本次测试时间', copy=False, default=lambda self: fields.Datetime.today()) + reconnect_datetime = fields.Datetime(string='重新测试时间', copy=False, default=lambda self: fields.Datetime.today()) + connect_state = fields.Boolean(string='联通状态', default=True) + connect_error_info = fields.Char(string='出错信息') + connect_mode = fields.Selection([ + ('socket', 'socket'), + ('ping', 'ping'), + ], '连接模式状态', copy=False, default='socket') + + @api.model + def delete_sql_lite_api(self): # 删除内存表 + print(f'delete_sql_lite_api -- begin--') + re, info = nebula_delete_sql_lite(db_name='sbomp.db', table_name='sbomp_test_ip_slave') + if re: + print(f'删除sbomp_test_ip_slave成功') + else: + print(f'删除sbomp_test_ip_slave失败{info}') + + re, info = nebula_delete_sql_lite(db_name='sbomp.db', table_name='sbomp_test_ip') + if re: + print(f'删除sbomp_test_ip成功') + else: + print(f'删除sbomp_test_ip失败{info}') + print(f'delete_sql_lite_api -- end--') + + def delete_sql_lite(self): # 删除内存表 + print(f'delete_sql_lite -- begin--') + re, info = nebula_delete_sql_lite(db_name='sbomp.db', table_name='sbomp_test_ip_slave') + if re: + self.env.user.notify_success(message='删除sbomp_test_ip_slave成功') + else: + self.env.user.notify_warning(message=info) + + re, info = nebula_delete_sql_lite(db_name='sbomp.db', table_name='sbomp_test_ip') + if re: + self.env.user.notify_success(message='删除sbomp_test_ip成功') + else: + self.env.user.notify_warning(message=info) + print(f'delete_sql_lite ok') + + + def create_sql_lite(self): # 创建内存数据表 + print(f'create_sql_lite -- begin--') + sql_text = '''CREATE TABLE IF NOT EXISTS sbomp_test_ip_slave + ( + id integer primary key autoincrement not null, + name varchar , + ip_address varchar , + port varchar, + slave_number varchar, + connect_datetime DATETIME NOT NULL, + reconnect_datetime DATETIME NOT NULL, + connect_state BOOLEAN NOT NULL, + connect_error_info TEXT, + connect_mode TEXT);''' + + re, info = nebula_create_sql_lite(db_name='sbomp.db', create_sql=sql_text) + if re: + self.env.user.notify_success(message='建表成功') + else: + self.env.user.notify_warning(message=info) + print(f'create_sql_lite -- end--') + + def net_db_used(self): + self.test_net_is_used(self.ip_address, self.port, self.connect_mode) + + def error_connect_ip(self, in_ip='127.0.0.1', in_port=503, slave_number='-1', connect_mode='socket', + connect_error_info='未知错误'): # 如果连接其他异常也设置为不可用 172.16.79.11 + now_datetime = fields.datetime.now() + domain = [('ip_address', '=', in_ip), ('port', '=', in_port), ('connect_mode', '=', connect_mode), + ('slave_number', '=', slave_number)] + + reconnect_datetime = now_datetime + timedelta(minutes=10) + values = { + 'connect_state': False, + 'connect_datetime': now_datetime, + 'connect_error_info': connect_error_info, + 'reconnect_datetime': reconnect_datetime, + } + db_list = self.env['sbomp.test.ip'].sudo().search(domain) + if db_list: + db_list.sudo().write(values) + else: # 新建 + my_name = 'ip:' + in_ip + ', port:' + str(in_port) + values['name'] = my_name + values['ip_address'] = in_ip + values['port'] = in_port + values['connect_mode'] = connect_mode + self.env['sbomp.test.ip'].sudo().create(values) + print( + f'error_connect_ip in_ip:{in_ip}, port:{in_port}, slave_number:{slave_number},connect_error_info:{connect_error_info}') + + def test_net_is_used(self, in_ip='null', in_port=503, slave_number='-1', connect_mode='socket'): + print(f'test_net_is_used -- in_ip:{in_ip}:{in_port}--') + if in_ip == 'null': + return False + ip = in_ip + now_datetime = fields.datetime.now() + # 如果传进来的是 '192.168.3.3:503' + ip_port_list = in_ip.split(':') + if len(ip_port_list) == 2: + ip_tmp = ip_port_list[0] + ip = ip_tmp.replace('\'', '') + port_tmp = ip_port_list[1] + port = port_tmp.replace('\'', '') + port = int(port) + else: # 如果传进来的端口是字符 + if type(in_port) == str: + port = int(in_port) + else: + port = in_port + + # 流程： 1. 传入ip、端口号、连接方式 + # 2. 检查这些参数在数据库中是不是存在，如果是 在小于重新检查时间，且上次检查是不通的，直接返回false，以避免时间浪费，后期可以用内存表的方式来遍历 + # 3. 如果不存在，或者已经到了重新检查的时间，就检查. + # 4.如果存在，还没有过重新检查时间，直接返回false.......... + # 5.检查结果，返回检查结果。 + # 6.利用finally功能，返回结果后将检查结果保存到数据中。 没有从地址，从地址=-1 + if net_is_used(in_ip=ip, in_port=port): + domain = [('ip_address', '=', ip), ('port', '=', port), ('connect_mode', '=', connect_mode), + ('slave_number', '=', slave_number), + ('reconnect_datetime', '>', now_datetime), + ('connect_state', '=', False)] # 小于重新检查时间，且上次检查是不通的，直接返回false + try: + if self.env['sbomp.test.ip'].sudo().search(domain, limit=1).id: + print(f'小于重新检查时间，且上次检查是不通的，直接返回false') + return False + except Exception as e: + print(f'目标地址:{ip},目标端口:{port},从站地址:{slave_number}出错，出错信息:{e}') + # 重新检查是否可用 + # if connect_mode == 'socket': + # return net_is_used(in_ip=ip, in_port=port) + # else: + # return True + + return True + else: + return False + +# 使用方法 +# odoo 如何调用另外一个模块的函数 +# self.env['model.name'].function_name() +# self.env['sbomp.test.ip'].test_net_is_used(in_ip='127.0.0.1', in_port=503, connect_mode='socket') +# self.env['sbomp.test.ip'].error_connect_ip(in_ip='127.0.0.1', in_port=503, connect_mode='socket',connect_error_info='未知错误') diff --git a/appstore/sbomp_test_ip/security/ir.model.access.csv b/appstore/sbomp_test_ip/security/ir.model.access.csv new file mode 100644 index 0000000000000000000000000000000000000000..bdd764dedf0b80e160ca00b1d81ba3ee6417504f --- /dev/null +++ b/appstore/sbomp_test_ip/security/ir.model.access.csv @@ -0,0 +1,5 @@ +"id","name","model_id:id","group_id:id","perm_read","perm_write","perm_create","perm_unlink" +access_sbomp_test_ip_admin,sbomp_test_ip,model_sbomp_test_ip,base.group_system,1,1,1,1 +access_sbomp_test_ip_user,sbomp_test_ip,model_sbomp_test_ip,base.group_user,1,0,0,1 + + diff --git a/appstore/sbomp_test_ip/static/description/icon.png b/appstore/sbomp_test_ip/static/description/icon.png new file mode 100644 index 0000000000000000000000000000000000000000..633a8639c2faa4b2b58e114ab7ccc3545796f5f8 Binary files /dev/null and b/appstore/sbomp_test_ip/static/description/icon.png differ diff --git a/appstore/sbomp_test_ip/views/sbomp_test_ip_views.xml b/appstore/sbomp_test_ip/views/sbomp_test_ip_views.xml new file mode 100644 index 0000000000000000000000000000000000000000..bf7931f8bd7e1e837edfb0a78a68e76a052eec48 --- /dev/null +++ b/appstore/sbomp_test_ip/views/sbomp_test_ip_views.xml @@ -0,0 +1,65 @@ + + + + + sbomp.test.ip.form + sbomp.test.ip + +
+
+
+ + +
+
+ + + + + + + + + + + + + + + + +
+ + + + + + sbomp.test.ip.tree + sbomp.test.ip + 1 + + + + + + + + + + + + + + + + IP测试 + sbomp.test.ip + tree,form + + + + + \ No newline at end of file diff --git a/dtcloud/tools/sql_lite_tools.py b/dtcloud/tools/sql_lite_tools.py new file mode 100644 index 0000000000000000000000000000000000000000..c487bf5c9d3fe051cd4032cabaf6744f16c264a3 --- /dev/null +++ b/dtcloud/tools/sql_lite_tools.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&& +# DTCloud360 +# QQ:35350428 +# 邮件:35350428@qq.com +# 手机：13584935775 +# 作者：'扶程星云' +# 公司网址： http://www.dtcloud360.com/ +# Copyright 中亿丰信息科技(苏州)有限公司 +# 日期：2021/10/15 +# 功能： 针对 sqlite3的操作 +# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&& + + +import sqlite3 + + +def nebula_delete_sql_lite(db_name='sbomp.db', table_name='sbomp_test_ip'): # 删除内存表 + # 创建与数据库的连接 + + conn = sqlite3.connect(db_name) + # 删除表 + drop_sql = "drop table IF EXISTS " + table_name + try: + conn.execute(drop_sql) + return True, 'ok' + except Exception as e: + return False, e + finally: + conn.close() + + +def nebula_create_sql_lite(create_sql, db_name='sbomp.db'): # 创建内存表 + conn = sqlite3.connect(db_name) + # 创建一个游标 cursor + cur = conn.cursor() + # 建表的sql语句 + sql_text = create_sql + # 执行sql语句 + try: + cur.execute(sql_text) + return True, 'ok' + except Exception as e: + return False, e + finally: + # 关闭游标 + cur.close() + # 关闭连接 + conn.close() + + +def nebula_select_sql_lite(select_sql, in_parameters, db_name='sbomp.db'): # 查询记录 + conn = sqlite3.connect(db_name) + # 创建一个游标 cursor + cur = conn.cursor() + # 建表的sql语句 + sql_text = select_sql + # 执行sql语句 + try: + cur.execute(sql_text, in_parameters) + db_list = cur.fetchall() + return True, db_list + except Exception as e: + return False, e + finally: + # 关闭游标 + cur.close() + # 关闭连接 + conn.close() + + +def nebula_update_sql_lite(update_sql, in_parameters, db_name='sbomp.db'): # 更新记录 + conn = sqlite3.connect(db_name) + # 创建一个游标 cursor + cur = conn.cursor() + # 更新表的sql语句 + sql_text = update_sql + # 执行sql语句 + try: + cur.execute(sql_text, in_parameters) + conn.commit() + print(f'nebula_update_sql_lite ok') + return True, 'ok' + except Exception as e: + conn.rollback() + print(e) + return False, e + finally: + # 关闭游标 + cur.close() + # 关闭连接 + conn.close() +