88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
"""Access数据库操作模块"""
|
|
import pyodbc
|
|
import os
|
|
|
|
|
|
class AccessDB:
|
|
def __init__(self, db_path, password):
|
|
self.db_path = db_path
|
|
self.password = password
|
|
self.connection = None
|
|
|
|
def connect(self):
|
|
"""连接Access数据库"""
|
|
try:
|
|
# 检查网络路径是否存在
|
|
if self.db_path.startswith('\\\\') and not os.path.exists(self.db_path):
|
|
raise FileNotFoundError(f"无法访问网络路径: {self.db_path}，请检查网络连接和共享设置")
|
|
|
|
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
|
|
except pyodbc.Error as e:
|
|
print(f"连接Access数据库失败: {e}")
|
|
raise
|
|
except FileNotFoundError as e:
|
|
print(f"数据库文件不可访问: {e}")
|
|
raise
|
|
|
|
def get_max_mark(self):
|
|
"""获取Access数据库中最大的Mark值"""
|
|
try:
|
|
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
|
|
except pyodbc.Error as e:
|
|
print(f"查询最大Mark值时出错: {e}")
|
|
return 0
|
|
except Exception as e:
|
|
print(f"获取最大Mark值时发生未知错误: {e}")
|
|
return 0
|
|
|
|
def query_task_status(self, mis_ids):
|
|
"""查询任务状态"""
|
|
try:
|
|
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
|
|
except pyodbc.Error as e:
|
|
print(f"查询任务状态时出错: {e}")
|
|
return {}
|
|
except Exception as e:
|
|
print(f"查询任务状态时发生未知错误: {e}")
|
|
return {}
|
|
|
|
def close(self):
|
|
"""关闭数据库连接"""
|
|
if self.connection:
|
|
self.connection.close()
|
|
self.connection = None
|