图:
目录:
开发说明书:
POS点销管理系统开发说明
1. 系统概述
本系统是一个基于 Python PySide6 开发的现代化 POS 点销管理系统,集成了商品管理、库存管理、会员管理、订单管理等核心功能。
2. 技术栈
-
开发语言: Python 3.8+
-
GUI 框架: PySide6
-
数据库: SQLite3
-
样式框架: Qss
3. 系统功能模块
3.1 基础功能
-
用户登录
-
默认管理员账号: admin
-
默认管理员密码: admin123
-
多角色权限控制(管理员/收银员/经理)
-
3.2 核心模块
-
商品管理
-
商品增删改查
-
商品分类管理
-
商品图片管理
-
商品导入导出
-
条码扫描支持
-
-
库存管理
-
库存状态查看
-
库存调整记录
-
库存预警
-
出入库管理
-
-
会员管理
-
会员信息管理
-
会员积分系统
-
会员等级管理
-
消费记录查询
-
-
订单管理
-
订单创建与结算
-
订单状态管理
-
订单历史查询
-
退款处理
-
3.3 界面设计
-
现代化 Design 风格
-
响应式布局
-
直观的操作界面
-
统一的视觉风格
4. 目录结构
├── main.py # 程序入口
├── database/
│ └── db_manager.py # 数据库管理
├── style/
│ └── style.qss # 界面样式
└── views/
├── login_window.py # 登录窗口
├── main_window.py # 主窗口
├── product_window.py # 商品管理
├── category_window.py # 分类管理
├── inventory_window.py # 库存管理
├── customer_window.py # 会员管理
├── employee_window.py # 员工管理
└── order_window.py # 订单管理
数据库结构:
1. 商品表 (products)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
product_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 商品ID(主键,自增) |
product_code | VARCHAR(50) | UNIQUE, NOT NULL | 商品编码(唯一,非空) |
product_name | VARCHAR(100) | NOT NULL | 商品名称(非空) |
category_id | INTEGER | 类别ID | |
unit_price | DECIMAL(10,2) | NOT NULL | 单价(非空) |
stock_quantity | INTEGER | NOT NULL, DEFAULT 0 | 库存数量(非空,默认0) |
image_path | TEXT | 图片路径 | |
created_at | TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 创建时间(默认当前时间) |
updated_at | TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 更新时间(默认当前时间) |
2. 商品类别表 (categories)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
category_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 类别ID(主键,自增) |
category_name | VARCHAR(50) | NOT NULL | 类别名称(非空) |
description | TEXT | 类别描述 |
3. 订单表 (orders)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
order_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 订单ID(主键,自增) |
order_number | VARCHAR(50) | UNIQUE, NOT NULL | 订单编号(唯一,非空) |
customer_id | INTEGER | 客户ID | |
employee_id | INTEGER | NOT NULL | 员工ID(非空) |
order_date | TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 订单日期(默认当前时间) |
total_amount | DECIMAL(10,2) | NOT NULL | 总金额(非空) |
payment_method | TEXT | CHECK: IN ('cash', 'card', 'mobile_payment'), NOT NULL | 支付方式(现金/卡/移动支付,非空) |
status | TEXT | CHECK: IN ('completed', 'cancelled', 'refunded'), DEFAULT 'completed' | 订单状态(完成/取消/退款,默认完成) |
4. 订单明细表 (order_details)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
order_detail_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 订单明细ID(主键,自增) |
order_id | INTEGER | NOT NULL, FOREIGN KEY REFERENCES orders(order_id) | 订单ID(外键,非空) |
product_id | INTEGER | NOT NULL, FOREIGN KEY REFERENCES products(product_id) | 商品ID(外键,非空) |
quantity | INTEGER | NOT NULL | 数量(非空) |
unit_price | DECIMAL(10,2) | NOT NULL | 单价(非空) |
subtotal | DECIMAL(10,2) | NOT NULL | 小计(非空) |
5. 员工表 (employees)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
employee_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 员工ID(主键,自增) |
username | VARCHAR(50) | UNIQUE, NOT NULL | 用户名(唯一,非空) |
password_hash | VARCHAR(255) | NOT NULL | 密码哈希值(非空) |
full_name | VARCHAR(100) | NOT NULL | 姓名(非空) |
role | TEXT | CHECK: IN ('admin', 'cashier', 'manager'), NOT NULL | 角色(管理员/收银员/经理,非空) |
contact_number | VARCHAR(20) | 联系电话 | |
| VARCHAR(100) | 邮箱 | |
is_active | BOOLEAN | DEFAULT 1 | 是否激活(默认是) |
6. 客户表 (customers)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
customer_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 客户ID(主键,自增) |
customer_name | VARCHAR(100) | NOT NULL | 客户姓名(非空) |
phone_number | VARCHAR(20) | 电话号码 | |
| VARCHAR(100) | 邮箱 | |
membership_level | TEXT | CHECK: IN ('regular', 'silver', 'gold'), DEFAULT 'regular' | 会员等级(普通/银卡/金卡,默认普通) |
points | INTEGER | DEFAULT 0 | 积分(默认0) |
created_at | TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 创建时间(默认当前时间) |
7. 库存交易记录表 (inventory_transactions)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
transaction_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 库存记录ID(主键,自增) |
product_id | INTEGER | NOT NULL, FOREIGN KEY REFERENCES products(product_id) | 商品ID(外键,非空) |
transaction_type | TEXT | CHECK: IN ('in', 'out'), NOT NULL | 交易类型(入库/出库,非空) |
quantity | INTEGER | NOT NULL | 数量(非空) |
transaction_date | TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 交易日期(默认当前时间) |
notes | TEXT | 备注 | |
employee_id | INTEGER | NOT NULL, FOREIGN KEY REFERENCES employees(employee_id) | 员工ID(外键,非空) |
8. 积分历史表 (points_history)
字段名 | 数据类型 | 约束/说明 | 中文说明 |
history_id | INTEGER | PRIMARY KEY, AUTOINCREMENT | 积分历史ID(主键,自增) |
customer_id | INTEGER | NOT NULL, FOREIGN KEY REFERENCES customers(customer_id) | 客户ID(外键,非空) |
points_change | INTEGER | NOT NULL | 积分变动(非空) |
notes | TEXT | 备注 | |
created_at | TIMESTAMP | DEFAULT CURRENT_TIMESTAMP | 创建时间(默认当前时间) |
约束说明
- PRIMARY KEY: 主键,唯一标识每一行
- AUTOINCREMENT: 自动递增,通常用于主键
- UNIQUE: 唯一约束,确保字段值不重复
- NOT NULL: 非空约束,确保字段值不能为空
- FOREIGN KEY: 外键,用于关联其他表
- CHECK: 检查约束,确保字段值符合指定条件
- DEFAULT: 默认值,当未提供字段值时使用默认值
代码:
main.py
import sys
import warnings
from PySide6.QtWidgets import QApplication, QDialog
from views.main_window import MainWindow
from views.login_window import LoginDialog
from database.db_manager import DatabaseManager# 添加警告过滤
warnings.filterwarnings("ignore", message="zbar.*")def load_stylesheet(filename):with open(filename, "r", encoding='utf-8') as f:return f.read()def main():# 创建应用程序实例app = QApplication(sys.argv)# 应用 Fluent UI 风格app.setStyle("Fusion")# 加载 QSS 文件stylesheet = load_stylesheet("style/style.qss")app.setStyleSheet(stylesheet)# 初始化数据库管理器db_manager = DatabaseManager()# 创建并显示登录窗口login_dialog = LoginDialog(db_manager)if login_dialog.exec() == QDialog.Accepted:# 登录成功,创建并显示主窗口window = MainWindow(db_manager)window.handle_login_success(login_dialog.user_info) # 传递用户信息window.show()return app.exec()else:# 登录失败或取消,直接退出程序return 0if __name__ == '__main__':sys.exit(main())
db_manager.py
import sqlite3
from datetime import datetime
import os
import random
import hashlibclass DatabaseManager:def __init__(self, db_path="db/pos_system.db"):# 确保数据库目录存在os.makedirs(os.path.dirname(db_path), exist_ok=True)self.db_path = db_pathself.create_tables()def get_connection(self):return sqlite3.connect(self.db_path)def create_tables(self):with self.get_connection() as conn:cursor = conn.cursor()# 创建商品表 (products)cursor.execute('''CREATE TABLE IF NOT EXISTS products (product_id INTEGER PRIMARY KEY AUTOINCREMENT,product_code VARCHAR(50) UNIQUE NOT NULL,product_name VARCHAR(100) NOT NULL,category_id INTEGER,unit_price DECIMAL(10,2) NOT NULL,stock_quantity INTEGER NOT NULL DEFAULT 0,image_path TEXT,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,FOREIGN KEY (category_id) REFERENCES categories(category_id))''')# 创建商品类别表 (categories)cursor.execute('''CREATE TABLE IF NOT EXISTS categories (category_id INTEGER PRIMARY KEY AUTOINCREMENT,category_name VARCHAR(50) NOT NULL,description TEXT)''')# 创建订单表 (orders)cursor.execute('''CREATE TABLE IF NOT EXISTS orders (order_id INTEGER PRIMARY KEY AUTOINCREMENT,order_number VARCHAR(50) UNIQUE NOT NULL,customer_id INTEGER,employee_id INTEGER NOT NULL,order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,total_amount DECIMAL(10,2) NOT NULL,payment_method TEXT CHECK(payment_method IN ('现金', '微信', '支付宝')) NOT NULL,status TEXT CHECK(status IN ('完成', '取消', '已退款')) DEFAULT '完成',FOREIGN KEY (customer_id) REFERENCES customers(customer_id),FOREIGN KEY (employee_id) REFERENCES employees(employee_id))''')# 创建订单明细表 (order_details)cursor.execute('''CREATE TABLE IF NOT EXISTS order_details (order_detail_id INTEGER PRIMARY KEY AUTOINCREMENT,order_id INTEGER NOT NULL,product_id INTEGER NOT NULL,quantity INTEGER NOT NULL,unit_price DECIMAL(10,2) NOT NULL,subtotal DECIMAL(10,2) NOT NULL,FOREIGN KEY (order_id) REFERENCES orders(order_id),FOREIGN KEY (product_id) REFERENCES products(product_id))''')# 创建员工表 (employees)cursor.execute('''CREATE TABLE IF NOT EXISTS employees (employee_id INTEGER PRIMARY KEY AUTOINCREMENT,username VARCHAR(50) UNIQUE NOT NULL,password_hash VARCHAR(255) NOT NULL,full_name VARCHAR(100) NOT NULL,role TEXT CHECK(role IN ('admin', 'cashier', 'manager')) NOT NULL,contact_number VARCHAR(20),email VARCHAR(100),is_active BOOLEAN DEFAULT 1)''')# 创建客户表 (customers)cursor.execute('''CREATE TABLE IF NOT EXISTS customers (customer_id INTEGER PRIMARY KEY AUTOINCREMENT,customer_name VARCHAR(100) NOT NULL,phone_number VARCHAR(20),email VARCHAR(100),membership_level TEXT CHECK(membership_level IN ('普通', 'Vip', '黄金Vip')) DEFAULT '普通',points INTEGER DEFAULT 0,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')# 创建库存交易记录表 (inventory_transactions)cursor.execute('''CREATE TABLE IF NOT EXISTS inventory_transactions (transaction_id INTEGER PRIMARY KEY AUTOINCREMENT,product_id INTEGER NOT NULL,transaction_type TEXT CHECK(transaction_type IN ('in', 'out')) NOT NULL,quantity INTEGER NOT NULL,transaction_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,notes TEXT,employee_id INTEGER NOT NULL,FOREIGN KEY (product_id) REFERENCES products(product_id),FOREIGN KEY (employee_id) REFERENCES employees(employee_id))''')# 创建积分历史表 (points_history)cursor.execute('''CREATE TABLE IF NOT EXISTS points_history (history_id INTEGER PRIMARY KEY AUTOINCREMENT,customer_id INTEGER NOT NULL,points_change INTEGER NOT NULL,notes TEXT,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,FOREIGN KEY (customer_id) REFERENCES customers(customer_id))''')# 创建触发器:更新商品的更新时间cursor.execute('''CREATE TRIGGER IF NOT EXISTS update_product_timestampAFTER UPDATE ON productsBEGINUPDATE products SET updated_at = CURRENT_TIMESTAMPWHERE product_id = NEW.product_id;END''')# 添加默认管理员账号cursor.execute('SELECT COUNT(*) FROM employees')if cursor.fetchone()[0] == 0:# 默认密码: admin123default_password = hashlib.sha256('admin123'.encode()).hexdigest()cursor.execute('''INSERT INTO employees (username, password_hash, full_name, role)VALUES (?, ?, ?, ?)''', ('admin', default_password, '系统管理员', 'admin'))# 添加默认收银员账号cashier_password = hashlib.sha256('cashier123'.encode()).hexdigest()cursor.execute('''INSERT INTO employees (username, password_hash, full_name, role)VALUES (?, ?, ?, ?)''', ('cashier', cashier_password, '收银员', 'cashier'))conn.commit()def add_product(self, product_code, product_name, category_id, unit_price, stock_quantity=0, image_path=None):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('''INSERT INTO products (product_code, product_name, category_id, unit_price, stock_quantity, image_path)VALUES (?, ?, ?, ?, ?, ?)''', (product_code, product_name, category_id, unit_price, stock_quantity, image_path))return cursor.lastrowiddef get_all_products(self):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('''SELECT p.*, c.category_name FROM products p LEFT JOIN categories c ON p.category_id = c.category_id''')return cursor.fetchall()def get_product_by_code(self, product_code):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('SELECT * FROM products WHERE product_code = ?', (product_code,))return cursor.fetchone()def update_product(self, product_id, **kwargs):"""更新商品信息"""allowed_fields = {'product_code', 'product_name', 'category_id','unit_price', 'stock_quantity', 'image_path'}update_fields = {k: v for k, v in kwargs.items() if k in allowed_fields}if not update_fields:return Falsequery = 'UPDATE products SET ' + ', '.join(f'{k} = ?' for k in update_fields.keys())query += ' WHERE product_id = ?'with self.get_connection() as conn:cursor = conn.cursor()try:cursor.execute(query, list(update_fields.values()) + [product_id])return cursor.rowcount > 0except Exception as e:conn.rollback()raise edef delete_product(self, product_id):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('DELETE FROM products WHERE product_id = ?', (product_id,))return cursor.rowcount > 0def add_category(self, category_name, description=None):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('''INSERT INTO categories (category_name, description)VALUES (?, ?)''', (category_name, description))return cursor.lastrowiddef get_all_categories(self):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('SELECT * FROM categories')return cursor.fetchall()def update_category(self, category_id, category_name=None, description=None):with self.get_connection() as conn:cursor = conn.cursor()update_fields = []params = []if category_name is not None:update_fields.append('category_name = ?')params.append(category_name)if description is not None:update_fields.append('description = ?')params.append(description)if not update_fields:return Falseparams.append(category_id)query = f'UPDATE categories SET {", ".join(update_fields)} WHERE category_id = ?'cursor.execute(query, params)return cursor.rowcount > 0def delete_category(self, category_id):with self.get_connection() as conn:cursor = conn.cursor()# 检查是否有商品使用此类别cursor.execute('SELECT COUNT(*) FROM products WHERE category_id = ?', (category_id,))if cursor.fetchone()[0] > 0:return Falsecursor.execute('DELETE FROM categories WHERE category_id = ?', (category_id,))return cursor.rowcount > 0def add_employee(self, username, password_hash, full_name, role, contact_number=None, email=None):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('''INSERT INTO employees (username, password_hash, full_name, role, contact_number, email)VALUES (?, ?, ?, ?, ?, ?)''', (username, password_hash, full_name, role, contact_number, email))return cursor.lastrowiddef get_all_employees(self):with self.get_connection() as conn:cursor = conn.cursor()cursor.execute('SELECT * FROM employees')return cursor.fetchall()def update_employee(self, employee_id, **kwargs):allowed_fields = {'username', 'password_hash', &