76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
import psycopg2
|
|
from datetime import datetime
|
|
from miniohelp import *
|
|
|
|
def insert_data(table_name, data, table = 'danger'):
|
|
"""
|
|
向 PostgreSQL 数据库的指定表中插入数据
|
|
:param table_name: 表名
|
|
:param data: 字典格式的数据（列名: 值）
|
|
"""
|
|
# 验证表名是否合法
|
|
if table_name not in [table]: # 在这里加入你的表名
|
|
print(f"错误: 无效的表名 {table_name}")
|
|
return
|
|
|
|
try:
|
|
sql_yaml = data.pop('sql_yaml')
|
|
# 数据库配置
|
|
sql_config = read_sql_config(sql_yaml)
|
|
db_host = sql_config['host']
|
|
db_port = sql_config['port']
|
|
db_database = sql_config['database']
|
|
db_user = sql_config['user']
|
|
db_password = sql_config['password']
|
|
# 连接数据库
|
|
with psycopg2.connect(
|
|
host=db_host,
|
|
port=db_port,
|
|
database=db_database,
|
|
user=db_user,
|
|
password=db_password
|
|
) as conn:
|
|
with conn.cursor() as cursor:
|
|
# 生成 SQL 语句
|
|
columns = ', '.join(data.keys())
|
|
values = ', '.join(['%s'] * len(data))
|
|
sql = f"INSERT INTO {table_name} ({columns}) VALUES ({values})"
|
|
|
|
# 执行 SQL 语句
|
|
cursor.execute(sql, tuple(data.values()))
|
|
|
|
# 提交事务
|
|
conn.commit()
|
|
print("数据插入成功")
|
|
|
|
except psycopg2.Error as e:
|
|
print("数据库错误:", e.pgcode, e.pgerror) # 打印详细的错误码和错误信息
|
|
|
|
# 示例数据
|
|
data = {
|
|
'orgcode': 'bdzl',
|
|
'sn': '123456',
|
|
'snname': '测试机',
|
|
'spaceid': '97123d28-40a3-4dd1-9021-616a7d60dce8',
|
|
'spacename': '目标检测',
|
|
'guid': '97123d28-40a3-4dd1-9021-616a7d60dce8',
|
|
'level': '一般',
|
|
'title': 'pedestrian',
|
|
'dangertype': '目标',
|
|
'content': '目标检测',
|
|
'gps1': "",
|
|
'dealresult': "",
|
|
'imgobjectname': 'AIResults/20250320/97123d28-40a3-4dd1-9021-616a7d60dce8.jpg',
|
|
'state': 0,
|
|
'createtm': '2025-03-20 16:12:04',
|
|
'dealtm': '2025-03-20 16:12:04',
|
|
'createuser': "",
|
|
'dealuser': "",
|
|
'checkuser': "",
|
|
'remarks': "",
|
|
'sql_yaml': "", # 数据库配置文件路径
|
|
}
|
|
|
|
# 调用插入函数
|
|
insert_data('danger', data)
|