62 lines
1.6 KiB
Python
62 lines
1.6 KiB
Python
|
"""Access数据库操作模块"""
|
||
|
|
import pyodbc
|
||
|
|
|
||
|
|
|
||
|
|
class AccessDB:
|
||
|
|
def __init__(self, db_path, password):
|
||
|
|
self.db_path = db_path
|
||
|
|
self.password = password
|
||
|
|
self.connection = None
|
||
|
|
|
||
|
|
def connect(self):
|
||
|
|
"""连接Access数据库"""
|
||
|
|
conn_str = (
|
||
|
|
r'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};'
|
||
|
|
f'DBQ={self.db_path};'
|
||
|
|
f'PWD={self.password};'
|
||
|
|
)
|
||
|
|
self.connection = pyodbc.connect(conn_str)
|
||
|
|
return self.connection
|
||
|
|
|
||
|
|
def get_max_mark(self):
|
||
|
|
"""获取Access数据库中最大的Mark值"""
|
||
|
|
if not self.connection:
|
||
|
|
self.connect()
|
||
|
|
|
||
|
|
cursor = self.connection.cursor()
|
||
|
|
cursor.execute("SELECT MAX(Mark) FROM Produce")
|
||
|
|
max_mark = cursor.fetchone()[0]
|
||
|
|
|
||
|
|
if max_mark is None:
|
||
|
|
max_mark = 0
|
||
|
|
|
||
|
|
return max_mark
|
||
|
|
|
||
|
|
def query_task_status(self, mis_ids):
|
||
|
|
"""查询任务状态"""
|
||
|
|
if not self.connection:
|
||
|
|
self.connect()
|
||
|
|
|
||
|
|
if not mis_ids:
|
||
|
|
return {}
|
||
|
|
|
||
|
|
cursor = self.connection.cursor()
|
||
|
|
placeholders = ','.join('?' * len(mis_ids))
|
||
|
|
query = f"SELECT MISID, Flag FROM Produce WHERE MISID IN ({placeholders})"
|
||
|
|
cursor.execute(query, mis_ids)
|
||
|
|
results = cursor.fetchall()
|
||
|
|
|
||
|
|
current_tasks = {}
|
||
|
|
for row in results:
|
||
|
|
mark = row[0]
|
||
|
|
flag = row[1] if row[1] is not None else ""
|
||
|
|
current_tasks[mark] = flag
|
||
|
|
|
||
|
|
return current_tasks
|
||
|
|
|
||
|
|
def close(self):
|
||
|
|
"""关闭数据库连接"""
|
||
|
|
if self.connection:
|
||
|
|
self.connection.close()
|
||
|
|
self.connection = None
|