Python PySide + SQLite3 开发的 《️ POS点销管理系统》可用初型

图：

目录：

开发说明书：

POS点销管理系统开发说明

1. 系统概述

本系统是一个基于 Python PySide6 开发的现代化 POS 点销管理系统，集成了商品管理、库存管理、会员管理、订单管理等核心功能。

2. 技术栈

• 开发语言: Python 3.8+

• GUI 框架: PySide6

• 数据库: SQLite3

• 样式框架: Qss

3. 系统功能模块

3.1 基础功能

• 用户登录

  • 默认管理员账号: admin

  • 默认管理员密码: admin123

  • 多角色权限控制(管理员/收银员/经理)

3.2 核心模块

1. 商品管理

   • 商品增删改查

   • 商品分类管理

   • 商品图片管理

   • 商品导入导出

   • 条码扫描支持

2. 库存管理

   • 库存状态查看

   • 库存调整记录

   • 库存预警

   • 出入库管理

3. 会员管理

   • 会员信息管理

   • 会员积分系统

   • 会员等级管理

   • 消费记录查询

4. 订单管理

   • 订单创建与结算

   • 订单状态管理

   • 订单历史查询

   • 退款处理

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)

2. 商品类别表 (categories)

3. 订单表 (orders)

4. 订单明细表 (order_details)

5. 员工表 (employees)

6. 客户表 (customers)

7. 库存交易记录表 (inventory_transactions)

8. 积分历史表 (points_history)

约束说明

• 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 0

if __name__ == '__main__':
    sys.exit(main())

db_manager.py

import sqlite3
from datetime import datetime
import os
import random
import hashlib

class DatabaseManager:
    def __init__(self, db_path="db/pos_system.db"):
        # 确保数据库目录存在
        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        self.db_path = db_path
        self.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_timestamp
                AFTER UPDATE ON products
                BEGIN
                    UPDATE products SET updated_at = CURRENT_TIMESTAMP
                    WHERE product_id = NEW.product_id;
                END
            ''')

            # 添加默认管理员账号
            cursor.execute('SELECT COUNT(*) FROM employees')
            if cursor.fetchone()[0] == 0:
                # 默认密码: admin123
                default_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.lastrowid

    def 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 False
            
        query = '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 > 0
            except Exception as e:
                conn.rollback()
                raise e

    def 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 > 0

    def 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.lastrowid

    def 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 False
                
            params.append(category_id)
            query = f'UPDATE categories SET {", ".join(update_fields)} WHERE category_id = ?'
            cursor.execute(query, params)
            return cursor.rowcount > 0

    def 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 False
            cursor.execute('DELETE FROM categories WHERE category_id = ?', (category_id,))
            return cursor.rowcount > 0

    def 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.lastrowid

    def 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', &