diff --git a/auth/__init__.py b/auth/__init__.py new file mode 100644 index 0000000..2f85cc9 --- /dev/null +++ b/auth/__init__.py @@ -0,0 +1,6 @@ +""" +Auth package for user authentication and session management +""" +from .session_manager import SessionManager + +__all__ = ['SessionManager'] diff --git a/auth/session_manager.py b/auth/session_manager.py new file mode 100644 index 0000000..c7275de --- /dev/null +++ b/auth/session_manager.py @@ -0,0 +1,130 @@ +""" +Session Manager - Singleton pattern for managing authenticated user session +""" +from typing import Optional + + +class SessionManager: + """ + Singleton session manager to maintain authenticated user state + throughout the application lifecycle. + """ + + _instance = None + + def __new__(cls): + """Implement singleton pattern""" + if cls._instance is None: + cls._instance = super(SessionManager, cls).__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + """Initialize the session manager (only once)""" + if self._initialized: + return + + self._current_user = None # {username, user_type} + self._initialized = True + + @classmethod + def get_instance(cls) -> 'SessionManager': + """ + Get the singleton instance of SessionManager + + Returns: + The singleton SessionManager instance + """ + if cls._instance is None: + cls._instance = cls() + return cls._instance + + def login(self, username: str, password: str) -> bool: + """ + Authenticate and login a user + + Args: + username: The username to authenticate + password: The password to verify + + Returns: + True if login successful, False otherwise + """ + from db.bip_users_dao import BIPUsersDAO + + dao = BIPUsersDAO() + user_info = dao.authenticate(username, password) + + if user_info: + self._current_user = { + 'username': user_info['username'], + 'user_type': user_info['user_type'] + } + return True + return False + + def logout(self): + """Logout the current user and clear session""" + self._current_user = None + + def is_authenticated(self) -> bool: + """ + Check if a user is currently authenticated + + Returns: + True if user is logged in, False otherwise + """ + return self._current_user is not None + + def is_admin(self) -> bool: + """ + Check if the current user is an admin + + Returns: + True if current user is admin, False otherwise + """ + if not self._current_user: + return False + return self._current_user.get('user_type') == 'Admin' + + def is_guest(self) -> bool: + """ + Check if the current user is a guest + + Returns: + True if current user is guest, False otherwise + """ + if not self._current_user: + return False + return self._current_user.get('user_type') == 'Guest' + + def get_username(self) -> Optional[str]: + """ + Get the current username + + Returns: + Current username if authenticated, None otherwise + """ + if not self._current_user: + return None + return self._current_user.get('username') + + def get_user_type(self) -> Optional[str]: + """ + Get the current user type + + Returns: + Current user type if authenticated, None otherwise + """ + if not self._current_user: + return None + return self._current_user.get('user_type') + + def get_user_info(self) -> Optional[dict]: + """ + Get all current user information + + Returns: + Dict with username and user_type if authenticated, None otherwise + """ + return self._current_user diff --git a/db/bip_users_dao.py b/db/bip_users_dao.py new file mode 100644 index 0000000..5562107 --- /dev/null +++ b/db/bip_users_dao.py @@ -0,0 +1,172 @@ +""" +BIPUsers DAO - Data access object for user authentication and management +""" +from typing import Optional, Dict, Any, List +from db.connection import get_connection + + +class BIPUsersDAO: + """Data access object for BIPUsers table""" + + def authenticate(self, username: str, password: str) -> Optional[Dict[str, Any]]: + """ + Authenticate a user with username and password + + Args: + username: The username to authenticate + password: The password to verify + + Returns: + Dict with user info if authentication successful, None otherwise + Returns: {id, username, user_type} + """ + sql = """ + SELECT [ID], [UserName], [UserType] + FROM [dbo].[BIPUsers] + WHERE [UserName] = ? AND [Password] = ? + """ + with get_connection() as db: + results = db.execute_query(sql, (username, password)) + if results: + return { + 'id': results[0]['ID'], + 'username': results[0]['UserName'], + 'user_type': results[0]['UserType'] + } + return None + + def get_all_users(self) -> List[Dict[str, Any]]: + """ + Get all users from the database + + Returns: + List of user dictionaries: [{id, username, user_type, create_time}] + """ + sql = """ + SELECT [ID], [UserName], [UserType], [CreateTime] + FROM [dbo].[BIPUsers] + ORDER BY [UserName] + """ + with get_connection() as db: + results = db.execute_query(sql) + return [ + { + 'id': row['ID'], + 'username': row['UserName'], + 'user_type': row['UserType'], + 'create_time': row['CreateTime'] + } + for row in results + ] + + def create_user(self, username: str, password: str, user_type: str) -> bool: + """ + Create a new user + + Args: + username: The username (must be unique) + password: The password (plain text for internal tool) + user_type: User type ('Admin', 'User', or 'Guest') + + Returns: + True if successful, False otherwise + """ + sql = """ + INSERT INTO [dbo].[BIPUsers] ([UserName], [Password], [UserType]) + VALUES (?, ?, ?) + """ + try: + with get_connection() as db: + db.execute_update(sql, (username, password, user_type)) + return True + except Exception as e: + print(f"Error creating user: {e}") + return False + + def update_user_type(self, username: str, user_type: str) -> bool: + """ + Update a user's type + + Args: + username: The username to update + user_type: New user type ('Admin', 'User', or 'Guest') + + Returns: + True if successful, False otherwise + """ + sql = """ + UPDATE [dbo].[BIPUsers] + SET [UserType] = ? + WHERE [UserName] = ? + """ + try: + with get_connection() as db: + db.execute_update(sql, (user_type, username)) + return True + except Exception as e: + print(f"Error updating user type: {e}") + return False + + def update_password(self, username: str, new_password: str) -> bool: + """ + Update a user's password + + Args: + username: The username to update + new_password: The new password (plain text for internal tool) + + Returns: + True if successful, False otherwise + """ + sql = """ + UPDATE [dbo].[BIPUsers] + SET [Password] = ? + WHERE [UserName] = ? + """ + try: + with get_connection() as db: + db.execute_update(sql, (new_password, username)) + return True + except Exception as e: + print(f"Error updating password: {e}") + return False + + def delete_user(self, username: str) -> bool: + """ + Delete a user + + Args: + username: The username to delete + + Returns: + True if successful, False otherwise + """ + sql = """ + DELETE FROM [dbo].[BIPUsers] + WHERE [UserName] = ? + """ + try: + with get_connection() as db: + db.execute_update(sql, (username,)) + return True + except Exception as e: + print(f"Error deleting user: {e}") + return False + + def user_exists(self, username: str) -> bool: + """ + Check if a username already exists + + Args: + username: The username to check + + Returns: + True if username exists, False otherwise + """ + sql = """ + SELECT COUNT(*) as count FROM [dbo].[BIPUsers] + WHERE [UserName] = ? + """ + with get_connection() as db: + results = db.execute_query(sql, (username,)) + return results[0]['count'] > 0 if results else False diff --git a/db/materials_to_be_deleted_records_dao.py b/db/materials_to_be_deleted_records_dao.py index 6c29018..d169564 100644 --- a/db/materials_to_be_deleted_records_dao.py +++ b/db/materials_to_be_deleted_records_dao.py @@ -194,6 +194,20 @@ class MaterialsToBeDeletedDAO: results = db.execute_query(sql) return [r['ManagerName'] for r in results if r.get('ManagerName')] + def get_records_by_manager(self, manager_name: str) -> List[Dict[str, Any]]: + """ + Get all material records for a specific manager. + + This is an alias for get_materials_by_manager for consistency. + + Args: + manager_name: Manager name + + Returns: + List of material records for the specified manager + """ + return self.get_materials_by_manager(manager_name) + def get_record_by_material_code(self, material_code: str) -> Optional[Dict[str, Any]]: """ Get a specific record by material code. diff --git a/gui/login_dialog.py b/gui/login_dialog.py new file mode 100644 index 0000000..a90ad8e --- /dev/null +++ b/gui/login_dialog.py @@ -0,0 +1,162 @@ +""" +Login Dialog - Modal dialog for user authentication +""" +import tkinter as tk +from tkinter import ttk, messagebox +from typing import Optional, Tuple + + +class LoginDialog: + """ + Modal login dialog that captures user credentials + + Usage: + login_dialog = LoginDialog(parent) + credentials = login_dialog.get_credentials() + if credentials: + username, password = credentials + # Proceed with authentication + """ + + def __init__(self, parent): + """ + Initialize the login dialog + + Args: + parent: The parent Tkinter window + """ + self.parent = parent + self.result = None # Will hold (username, password) or None + self.dialog = None + + # Create dialog as modal + self._create_dialog() + self.parent.wait_window(self.dialog) + + def _create_dialog(self): + """Create the login dialog UI""" + self.dialog = tk.Toplevel(self.parent) + self.dialog.title("ERP 自动化工具 - 登录") + self.dialog.geometry("400x250") + self.dialog.resizable(False, False) + + # Center the dialog on parent + self.dialog.transient(self.parent) + self.dialog.grab_set() + + # Calculate position to center on parent + self.parent.update_idletasks() + parent_x = self.parent.winfo_x() + parent_y = self.parent.winfo_y() + parent_width = self.parent.winfo_width() + parent_height = self.parent.winfo_height() + + dialog_width = 400 + dialog_height = 250 + x = parent_x + (parent_width - dialog_width) // 2 + y = parent_y + (parent_height - dialog_height) // 2 + self.dialog.geometry(f"{dialog_width}x{dialog_height}+{x}+{y}") + + # Create UI elements + self._create_widgets() + + # Bind Enter key to login button + self.dialog.bind('', lambda e: self._on_login()) + + # Focus on username entry + self.username_entry.focus_set() + + def _create_widgets(self): + """Create the dialog widgets""" + # Main frame with padding + main_frame = ttk.Frame(self.dialog, padding="20") + main_frame.pack(fill=tk.BOTH, expand=True) + + # Title + title_label = ttk.Label( + main_frame, + text="请登录", + font=('', 16, 'bold') + ) + title_label.pack(pady=(0, 20)) + + # Username field + username_frame = ttk.Frame(main_frame) + username_frame.pack(fill=tk.X, pady=5) + + ttk.Label(username_frame, text="用户名:", width=10).pack(side=tk.LEFT) + self.username_entry = ttk.Entry(username_frame) + self.username_entry.pack(side=tk.LEFT, fill=tk.X, expand=True) + + # Password field + password_frame = ttk.Frame(main_frame) + password_frame.pack(fill=tk.X, pady=5) + + ttk.Label(password_frame, text="密码:", width=10).pack(side=tk.LEFT) + self.password_entry = ttk.Entry(password_frame, show="*") + self.password_entry.pack(side=tk.LEFT, fill=tk.X, expand=True) + + # Buttons frame + button_frame = ttk.Frame(main_frame) + button_frame.pack(pady=(20, 0)) + + # Login button + login_btn = ttk.Button( + button_frame, + text="登录", + command=self._on_login, + width=10 + ) + login_btn.pack(side=tk.LEFT, padx=5) + + # Cancel button + cancel_btn = ttk.Button( + button_frame, + text="取消", + command=self._on_cancel, + width=10 + ) + cancel_btn.pack(side=tk.LEFT, padx=5) + + # Version info + version_label = ttk.Label( + main_frame, + text="v1.0", + font=('', 8), + foreground='gray' + ) + version_label.pack(side=tk.BOTTOM, pady=10) + + def _on_login(self): + """Handle login button click""" + username = self.username_entry.get().strip() + password = self.password_entry.get().strip() + + if not username: + messagebox.showwarning("输入错误", "请输入用户名") + self.username_entry.focus_set() + return + + if not password: + messagebox.showwarning("输入错误", "请输入密码") + self.password_entry.focus_set() + return + + # Return credentials for validation + self.result = (username, password) + self.dialog.destroy() + + def _on_cancel(self): + """Handle cancel button click""" + self.result = None + self.dialog.destroy() + + def get_credentials(self) -> Optional[Tuple[str, str]]: + """ + Get the entered credentials + + Returns: + Tuple of (username, password) if user clicked Login, + None if user clicked Cancel or closed the dialog + """ + return self.result diff --git a/gui/main_window.py b/gui/main_window.py index a578542..d03da2b 100644 --- a/gui/main_window.py +++ b/gui/main_window.py @@ -18,18 +18,21 @@ from gui.settings_tab import SettingsTab class MainWindow: """主窗口类""" - def __init__(self, root: tk.Tk): + def __init__(self, root: tk.Tk, session_manager): """ 初始化主窗口 Args: root: Tk 根窗口 + session_manager: SessionManager 实例,用于用户认证和权限管理 """ self.root = root self.config = ConfigManager() + self.session_manager = session_manager - # 设置窗口属性 - self.root.title("ERP 自动化工具 v1.0") + # 设置窗口属性(包含用户信息) + user_type_display = "管理员" if session_manager.is_admin() else "用户" + self.root.title(f"ERP 自动化工具 v1.0 - {session_manager.get_username()} ({user_type_display})") self.root.geometry("1000x700") # 设置最小窗口大小 @@ -72,8 +75,8 @@ class MainWindow: self.extraction_tab = DataExtractionTab(self.notebook, self.config) self.notebook.add(self.extraction_tab, text="数据提取") - # 物料校验标签页 - self.validation_tab = MaterialValidationTab(self.notebook, self.config) + # 物料校验标签页(传入 session_manager) + self.validation_tab = MaterialValidationTab(self.notebook, self.config, self.session_manager) self.notebook.add(self.validation_tab, text="物料校验") # 数据查询标签页 @@ -97,6 +100,15 @@ class MainWindow: ) status_label.pack(side=tk.LEFT, padx=5) + # 用户信息显示 + user_type_display = "管理员" if self.session_manager.is_admin() else "用户" + self.user_info_var = tk.StringVar() + self.user_info_var.set(f"当前用户: {self.session_manager.get_username()} ({user_type_display})") + user_label = ttk.Label( + self.status_bar, textvariable=self.user_info_var, anchor=tk.E + ) + user_label.pack(side=tk.RIGHT, padx=5) + # 配置状态指示 self.config_status = tk.StringVar() self.config_status.set("配置已加载") diff --git a/gui/material_type_management_dialog.py b/gui/material_type_management_dialog.py index 63325f9..6ea9b8e 100644 --- a/gui/material_type_management_dialog.py +++ b/gui/material_type_management_dialog.py @@ -332,14 +332,16 @@ class MaterialTypeManagementDialog(tk.Toplevel): ROW_STATUS_MODIFIED = "modified" ROW_STATUS_DELETED = "deleted" - def __init__(self, parent, title: str = "类型管理"): + def __init__(self, parent, session_manager, title: str = "类型管理"): """初始化对话框 Args: parent: 父窗口 + session_manager: SessionManager 实例,用于权限控制 title: 窗口标题 """ super().__init__(parent) + self.session_manager = session_manager self.title(title) self.geometry("900x650") @@ -515,6 +517,16 @@ class MaterialTypeManagementDialog(tk.Toplevel): widget.destroy() self.manager_checkboxes.clear() + # PERMISSION CHECK: 非管理员用户隐藏筛选 UI + if not self.session_manager.is_admin(): + self.managers = [self.session_manager.get_username()] + ttk.Label( + self.filter_frame, + text=f"仅显示您的数据(负责人:{self.session_manager.get_username()})" + ).pack(anchor="w") + return + + # 管理员:获取所有负责人 from db.materials_to_be_deleted_dao import MaterialsTypeToBeDeletedDAO dao = MaterialsTypeToBeDeletedDAO() self.managers = dao.get_managers() @@ -591,7 +603,13 @@ class MaterialTypeManagementDialog(tk.Toplevel): from db.materials_to_be_deleted_dao import MaterialsTypeToBeDeletedDAO dao = MaterialsTypeToBeDeletedDAO() - self.original_data = dao.get_all_materials() + + # PERMISSION CHECK: 非管理员用户只加载自己的数据 + if not self.session_manager.is_admin(): + self.original_data = dao.get_materials_by_manager(self.session_manager.get_username()) + else: + self.original_data = dao.get_all_materials() + self.current_data = list(self.original_data) self.row_status = {} self.original_values = {} @@ -627,7 +645,11 @@ class MaterialTypeManagementDialog(tk.Toplevel): def _apply_filter(self): """应用筛选条件""" - selected_managers = self._get_selected_managers() + # PERMISSION CHECK: 非管理员用户跳过筛选(已在数据库层面筛选) + if not self.session_manager.is_admin(): + selected_managers = [self.session_manager.get_username()] + else: + selected_managers = self._get_selected_managers() if not selected_managers: self.current_data = [] @@ -639,12 +661,7 @@ class MaterialTypeManagementDialog(tk.Toplevel): ] # 添加新增的记录 - new_records = [ - r for r, status in self.row_status.items() - if status == self.ROW_STATUS_NEW - ] - - # 从行状态中恢复新增记录的数据 + new_records = [] for item_id in list(self.row_status.keys()): if self.row_status[item_id] == self.ROW_STATUS_NEW: # 尝试从 tree 中获取数据 diff --git a/gui/material_validation_tab.py b/gui/material_validation_tab.py index 8ee961e..8e458c9 100644 --- a/gui/material_validation_tab.py +++ b/gui/material_validation_tab.py @@ -131,16 +131,18 @@ class CheckboxTreeview(ttk.Treeview): class MaterialValidationTab(ttk.Frame): """物料校验标签页""" - def __init__(self, parent, config: ConfigManager): + def __init__(self, parent, config: ConfigManager, session_manager): """ 初始化物料校验标签页 Args: parent: 父容器 config: 配置管理器 + session_manager: SessionManager 实例,用于权限控制 """ super().__init__(parent) self.config = config + self.session_manager = session_manager self.validating = False self.validation_results = None self.material_records_cache = None # 缓存完整的物料记录 @@ -421,7 +423,17 @@ class MaterialValidationTab(ttk.Frame): widget.destroy() self.manager_checkboxes.clear() - # 从两个表获取负责人列表并合并去重 + # PERMISSION CHECK: 非管理员用户只能看到自己的数据 + if not self.session_manager.is_admin(): + self.managers = [self.session_manager.get_username()] + # 不显示复选框,直接显示提示信息 + ttk.Label( + self.filter_frame, + text=f"仅显示您的数据(负责人:{self.session_manager.get_username()})" + ).pack(anchor="w") + return + + # 管理员:从两个表获取负责人列表并合并去重 from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO record_dao = MaterialsToBeDeletedDAO() @@ -501,7 +513,11 @@ class MaterialValidationTab(ttk.Frame): if not self.material_records_cache: return - selected_managers = self._get_selected_managers() + # PERMISSION CHECK: 非管理员用户强制筛选到当前用户 + if not self.session_manager.is_admin(): + selected_managers = [self.session_manager.get_username()] + else: + selected_managers = self._get_selected_managers() if not selected_managers: # 没有选中任何负责人,清空表格 @@ -541,7 +557,12 @@ class MaterialValidationTab(ttk.Frame): # 从数据库获取已标记删除的记录 from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO record_dao = MaterialsToBeDeletedDAO() - marked_records = record_dao.get_all_records() + + # PERMISSION CHECK: 非管理员用户只获取自己的记录 + if not self.session_manager.is_admin(): + marked_records = record_dao.get_materials_by_manager(self.session_manager.get_username()) + else: + marked_records = record_dao.get_all_records() # Build dictionary: MaterialCode -> ManagerName marked_codes_dict = { @@ -795,7 +816,12 @@ class MaterialValidationTab(ttk.Frame): # 从数据库获取已标记删除的 MaterialCode -> ManagerName 映射 from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO record_dao = MaterialsToBeDeletedDAO() - marked_records = record_dao.get_all_records() + + # PERMISSION CHECK: 非管理员用户只获取自己的记录 + if not self.session_manager.is_admin(): + marked_records = record_dao.get_materials_by_manager(self.session_manager.get_username()) + else: + marked_records = record_dao.get_all_records() # Build dictionary: MaterialCode -> ManagerName marked_codes_dict = { @@ -1067,4 +1093,4 @@ class MaterialValidationTab(ttk.Frame): def open_type_management(self): """打开类型管理窗口""" - dialog = MaterialTypeManagementDialog(self, title="物料类型管理") + dialog = MaterialTypeManagementDialog(self, self.session_manager, title="物料类型管理") diff --git a/init_users.sql b/init_users.sql new file mode 100644 index 0000000..85e5f76 --- /dev/null +++ b/init_users.sql @@ -0,0 +1,23 @@ +-- ============================================= +-- Initial User Setup for [dbo].[BIPUsers] Table +-- ============================================= +-- This script creates initial users for the ERP Automation Tool +-- +-- IMPORTANT: Change these passwords before deploying to production! +-- ============================================= + +-- Create admin user (can see all data) +INSERT INTO [dbo].[BIPUsers] (UserName, Password, UserType) VALUES ('admin', 'admin123', 'Admin'); + +-- Create regular users (replace with actual usernames) +-- These users will only see records where ManagerName matches their username +INSERT INTO [dbo].[BIPUsers] (UserName, Password, UserType) VALUES ('user1', 'password1', 'User'); +INSERT INTO [dbo].[BIPUsers] (UserName, Password, UserType) VALUES ('user2', 'password2', 'User'); + +-- Example: Create users for specific managers +-- Replace 'ManagerName1', 'ManagerName2' with actual manager names from your data +-- INSERT INTO [dbo].[BIPUsers] (UserName, Password, UserType) VALUES ('张三', 'password123', 'User'); +-- INSERT INTO [dbo].[BIPUsers] (UserName, Password, UserType) VALUES ('李四', 'password456', 'User'); + +-- Verify created users +SELECT * FROM [dbo].[BIPUsers]; diff --git a/main_ui.py b/main_ui.py index 563e8c1..a883168 100644 --- a/main_ui.py +++ b/main_ui.py @@ -14,10 +14,14 @@ ERP 自动化工具 - GUI 版本 - 物料校验:校验物料状态并匹配待删除物料 - 数据查询:查询生产订单号等信息 - 设置管理:管理系统配置 + - 用户认证:基于用户权限的数据访问控制 """ import tkinter as tk +from tkinter import messagebox from gui.main_window import MainWindow +from gui.login_dialog import LoginDialog +from auth.session_manager import SessionManager def main(): @@ -25,8 +29,27 @@ def main(): # 创建根窗口 root = tk.Tk() - # 创建主窗口 - app = MainWindow(root) + # 显示登录对话框 + login_dialog = LoginDialog(root) + credentials = login_dialog.get_credentials() + + if not credentials: + # 用户取消登录或关闭对话框 + root.destroy() + return + + username, password = credentials + + # 认证并初始化会话 + session_manager = SessionManager.get_instance() + + if not session_manager.login(username, password): + messagebox.showerror("登录失败", "用户名或密码错误") + root.destroy() + return + + # 创建主窗口(传入 session_manager) + app = MainWindow(root, session_manager) # 启动主事件循环 root.mainloop()