123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778 |
- import datetime
- import psycopg2
- import json
- def get_ts_time():
- current_time = datetime.datetime.now()
- timestamp = current_time.timestamp()
- return int(timestamp)
- # 获取 Unix 时间戳
- def get_current_day_library_name():
- timestamp = get_ts_time()
- # 将时间戳转换为 datetime 对象
- date_obj = datetime.datetime.fromtimestamp(timestamp)
- # 格式化日期为 YYYYMMDD 格式
- formatted_date = date_obj.strftime("%Y%m%d")
- # 拼接字符串
- result_str = "xsinsert" + formatted_date
- return result_str
- def writeDataToCurrentDay_Library(data):
- # 连接到数据库
- with open('db_params.json', 'r') as f:
- db_params = json.load(f)
- db_params = db_params['test']
- #建立长连接
- db_name = get_current_day_library_name()
- ts_time = get_ts_time()
- conn = psycopg2.connect(**db_params)
- cursor = conn.cursor()
- # 表名
- db_name = get_current_day_library_name()
- # 遍历字典中的每个键值对
- for column_name, value in data.items():
- cursor.execute(f"""
- INSERT INTO {db_name} (nm, v,ts,createtime,factoryname,devicename,type,gatewaycode)
- VALUES (%s, %s,{ts_time},{ts_time},'lfgawqyq','plc3',4,'SN2_30101_002209_00014')
- """, (column_name, value))
- conn.commit()
- # 获取查询结果
- # 打印结果
- # 关闭游标和连接
- cursor.close()
- conn.close()
- # cursor.execute(f"""SELECT 1 FROM {db_name} """)
- # exists = cursor.fetchone()
- #
- # # 如果数据库不存在,则创建数据库
- # if not exists:
- # cursor.execute(f"CREATE DATABASE {db_name}")
- #
- # # 创建表
- # # 如果不存在,插入新行
- # cursor.execute("""
- # INSERT INTO realtimedata (nm, v,ts,createtime,factoryname,devicename,type,gatewaycode)
- # VALUES (%s, %s,get_ts_time(),get_ts_time(),'lfgawqyq','plc10','SN2_30101_002209_00014')
- # """, ('Ifgawqyq_plc3_Tag22', '3'))
- # data ={"lfgawqyq_plc11_Tag23": 40.234, "lfgawqyq_plc11_Tag124": 40.264, "lfgawqyq_plc11_Tag25": 3.0, "lfgawqyq_plc11_Tag122": 40.264, "lfgawqyq_xn_cscod": 4.689, "lfgawqyq_xn_cstn": 10.503}
- #
- # writeDataToCurrentDay_Library(data)
|