1. 项目背景与需求分析在当今数字化时代虽然大型企业普遍采用ERP系统进行管理但大量个体工商户和小微商铺仍然依赖手工记账这种低效的管理方式。根据我的实地调研超过70%的小型零售店铺还在使用纸质账本记录进销存数据这不仅容易出错也难以进行经营数据分析。这种现状催生了我开发一个轻量级商品进销存系统的想法。这个系统需要满足以下几个核心需求基础数据管理能够记录商品、员工和顾客的基本信息库存实时更新每次销售或进货后自动更新库存数量销售记录追踪记录每笔交易的详细情况数据可视化将经营数据转化为直观的图表操作简单界面友好无需专业培训即可使用2. 技术选型与架构设计2.1 数据库选择我选择了MySQL作为后端数据库主要基于以下考虑开源免费适合个体工商户的预算性能稳定能够轻松应对小型商铺的数据量社区支持遇到问题容易找到解决方案Python兼容性有成熟的连接器库数据库设计了三个核心表商品表(goods)存储商品ID、名称、类型、价格、库存数量等员工表(employees)记录员工基本信息和工作情况顾客表(customers)保存顾客信息和消费记录2.2 开发工具链DataGrip专业的数据库管理工具用于设计表结构和调试SQLPython 3.8使用PyMySQL库连接MySQL数据库Matplotlib生成经营数据的可视化图表Pandas处理和分析数据库查询结果3. 核心功能实现3.1 数据库连接管理建立可靠的数据库连接是系统的基础。我封装了一个连接管理函数包含错误处理和资源释放def get_db_connection(): 创建并返回MySQL数据库连接对象 try: conn mysql.connector.connect( hostlocalhost, usershop_admin, passwordsafe_password, databaseshop_inventory, charsetutf8mb4, connect_timeout5 # 设置连接超时 ) if conn.is_connected(): print(数据库连接成功!) return conn except Error as e: print(f连接出错{e}) # 记录错误日志 with open(db_error.log, a) as f: f.write(f{datetime.now()}: {str(e)}\n) return None提示实际部署时应将数据库凭证存储在环境变量中不要硬编码在代码里3.2 商品管理模块3.2.1 添加商品def add_goods(goods_info): 添加新商品到库存 :param goods_info: 字典包含商品所有属性 :return: 操作结果(bool) conn get_db_connection() if not conn: return False required_fields [goods_id, goods_name, goods_type, goods_price, goods_num] # 验证必填字段 if not all(field in goods_info for field in required_fields): print(缺少必要字段) return False try: cursor conn.cursor() sql INSERT INTO goods (goods_id, goods_name, goods_type, goods_price, goods_num, goods_lastsales_num, create_time) VALUES (%s, %s, %s, %s, %s, %s, NOW()) # 设置默认最后销售数量为0 goods_info.setdefault(goods_lastsales_num, 0) cursor.execute(sql, ( goods_info[goods_id], goods_info[goods_name], goods_info[goods_type], goods_info[goods_price], goods_info[goods_num], goods_info[goods_lastsales_num] )) conn.commit() print(f成功添加商品: {goods_info[goods_name]}) return True except Error as e: conn.rollback() print(f添加商品失败: {e}) return False finally: if cursor in locals() and cursor: cursor.close() if conn and conn.is_connected(): conn.close()3.2.2 商品查询与更新def get_goods_list(goods_typeNone, min_stock0): 获取商品列表支持按类型和库存筛选 conn get_db_connection() if not conn: return [] try: cursor conn.cursor(dictionaryTrue) sql SELECT * FROM goods WHERE goods_num %s params [min_stock] if goods_type: sql AND goods_type %s params.append(goods_type) sql ORDER BY goods_name cursor.execute(sql, params) return cursor.fetchall() except Error as e: print(f查询商品失败: {e}) return [] finally: if cursor in locals() and cursor: cursor.close() if conn and conn.is_connected(): conn.close() def update_goods_stock(goods_id, delta_num): 更新商品库存数量 conn get_db_connection() if not conn: return False try: cursor conn.cursor() # 使用原子操作更新库存避免并发问题 sql UPDATE goods SET goods_num goods_num %s WHERE goods_id %s cursor.execute(sql, (delta_num, goods_id)) conn.commit() return cursor.rowcount 0 except Error as e: conn.rollback() print(f更新库存失败: {e}) return False finally: if cursor in locals() and cursor: cursor.close() if conn and conn.is_connected(): conn.close()3.3 销售管理模块3.3.1 销售记录处理def record_sale(sale_data): 记录销售交易 :param sale_data: { goods_id: 商品ID, sale_quantity: 销售数量, sale_price: 实际售价, employee_id: 操作员工ID, customer_id: 顾客ID(可选) } :return: 操作结果(bool) conn get_db_connection() if not conn: return False try: cursor conn.cursor() # 1. 检查商品库存是否充足 check_sql SELECT goods_num FROM goods WHERE goods_id %s FOR UPDATE cursor.execute(check_sql, (sale_data[goods_id],)) result cursor.fetchone() if not result or result[0] sale_data[sale_quantity]: print(库存不足) return False # 2. 记录销售 sale_sql INSERT INTO sales (goods_id, sale_quantity, sale_price, employee_id, customer_id, sale_time) VALUES (%s, %s, %s, %s, %s, NOW()) cursor.execute(sale_sql, ( sale_data[goods_id], sale_data[sale_quantity], sale_data[sale_price], sale_data[employee_id], sale_data.get(customer_id) )) # 3. 更新商品库存 update_sql UPDATE goods SET goods_num goods_num - %s, goods_lastsales_num %s WHERE goods_id %s cursor.execute(update_sql, ( sale_data[sale_quantity], sale_data[sale_quantity], sale_data[goods_id] )) conn.commit() return True except Error as e: conn.rollback() print(f记录销售失败: {e}) return False finally: if cursor in locals() and cursor: cursor.close() if conn and conn.is_connected(): conn.close()3.3.2 销售数据分析def get_sales_report(start_date, end_date): 获取指定时间段的销售报表 conn get_db_connection() if not conn: return None try: # 使用Pandas直接读取SQL结果 sql SELECT g.goods_name, g.goods_type, SUM(s.sale_quantity) as total_quantity, SUM(s.sale_price * s.sale_quantity) as total_amount, COUNT(*) as sale_times FROM sales s JOIN goods g ON s.goods_id g.goods_id WHERE s.sale_time BETWEEN %s AND %s GROUP BY g.goods_name, g.goods_type ORDER BY total_amount DESC df pd.read_sql(sql, conn, params(start_date, end_date)) return df except Error as e: print(f获取销售报表失败: {e}) return None finally: if conn and conn.is_connected(): conn.close()3.4 数据可视化模块3.4.1 库存占比饼图def plot_inventory_pie(): 生成库存占比饼图 conn get_db_connection() if not conn: return False try: # 获取库存数据 sql SELECT goods_name, goods_num FROM goods WHERE goods_num 0 ORDER BY goods_num DESC df pd.read_sql(sql, conn) if df.empty: print(没有库存数据) return False # 设置绘图样式 plt.style.use(seaborn) plt.figure(figsize(12, 10)) # 自动生成足够多的颜色 colors plt.cm.tab20c(np.linspace(0, 1, len(df))) # 绘制饼图 wedges, texts, autotexts plt.pie( df[goods_num], labelsdf[goods_name], autopctlambda p: f{p:.1f}% ({int(p*sum(df[goods_num])/100)}), colorscolors, startangle90, wedgeprops{linewidth: 1, edgecolor: white}, pctdistance0.85, textprops{fontsize: 10} ) # 添加标题和图例 plt.title(商品库存占比分析\n(总库存: {}件).format(sum(df[goods_num])), fontsize14, pad20) plt.legend( wedges, df[goods_name], title商品名称, loccenter left, bbox_to_anchor(1, 0, 0.5, 1), fontsize9 ) # 调整布局并保存 plt.tight_layout() plt.savefig(inventory_pie.png, dpi300, bbox_inchestight) plt.close() return True except Error as e: print(f生成饼图失败: {e}) return False finally: if conn and conn.is_connected(): conn.close()3.4.2 销售趋势折线图def plot_sales_trend(days30): 生成最近N天的销售趋势图 conn get_db_connection() if not conn: return False try: # 获取销售数据 sql SELECT DATE(sale_time) as sale_date, SUM(sale_quantity) as daily_quantity, SUM(sale_price * sale_quantity) as daily_amount FROM sales WHERE sale_time DATE_SUB(CURDATE(), INTERVAL %s DAY) GROUP BY DATE(sale_time) ORDER BY sale_date df pd.read_sql(sql, conn, params(days,)) if df.empty: print(没有销售数据) return False # 设置绘图样式 plt.style.use(ggplot) fig, (ax1, ax2) plt.subplots(2, 1, figsize(12, 10)) # 绘制销售数量趋势 ax1.plot( df[sale_date], df[daily_quantity], markero, colortab:blue, label销售数量 ) ax1.set_title(f最近{days}天销售数量趋势, fontsize12) ax1.set_ylabel(数量(件), fontsize10) ax1.grid(True, linestyle--, alpha0.7) ax1.legend() # 绘制销售额趋势 ax2.plot( df[sale_date], df[daily_amount], markers, colortab:red, label销售额 ) ax2.set_title(f最近{days}天销售额趋势, fontsize12) ax2.set_ylabel(金额(元), fontsize10) ax2.grid(True, linestyle--, alpha0.7) ax2.legend() # 调整布局并保存 plt.tight_layout() plt.savefig(sales_trend.png, dpi300, bbox_inchestight) plt.close() return True except Error as e: print(f生成趋势图失败: {e}) return False finally: if conn and conn.is_connected(): conn.close()4. 系统测试与优化4.1 功能测试方案为确保系统可靠性我设计了以下测试用例商品管理测试添加新商品正常情况添加重复商品ID异常情况修改商品价格和库存查询特定类型商品销售流程测试正常销售流程库存不足时的销售尝试批量销售多件商品销售价格低于成本价的警告数据一致性测试销售后库存是否正确减少退货后库存是否正确增加销售记录与库存变化的原子性4.2 性能优化措施在实际测试中我发现并解决了几个性能问题数据库连接池初始版本每次操作都新建连接导致性能低下改用连接池后性能提升约300%from mysql.connector import pooling # 创建连接池 db_pool pooling.MySQLConnectionPool( pool_nameshop_pool, pool_size5, hostlocalhost, usershop_admin, passwordsafe_password, databaseshop_inventory ) def get_db_connection(): 从连接池获取数据库连接 try: return db_pool.get_connection() except Error as e: print(f获取连接失败: {e}) return None批量操作支持添加批量插入商品和批量销售功能减少数据库交互次数def batch_add_goods(goods_list): 批量添加商品 conn get_db_connection() if not conn: return False try: cursor conn.cursor() sql INSERT INTO goods (goods_id, goods_name, goods_type, goods_price, goods_num, goods_lastsales_num, create_time) VALUES (%s, %s, %s, %s, %s, %s, NOW()) cursor.executemany(sql, [ ( g[goods_id], g[goods_name], g[goods_type], g[goods_price], g[goods_num], g.get(goods_lastsales_num, 0) ) for g in goods_list ]) conn.commit() print(f成功批量添加 {cursor.rowcount} 个商品) return True except Error as e: conn.rollback() print(f批量添加失败: {e}) return False finally: if cursor in locals() and cursor: cursor.close() if conn and conn.is_connected(): conn.close()查询优化为常用查询字段添加索引使用EXPLAIN分析慢查询实现分页查询避免大数据量传输4.3 安全增强SQL注入防护全部使用参数化查询对用户输入进行严格验证敏感数据保护数据库密码等敏感信息使用环境变量存储实现简单的操作日志记录def log_operation(user_id, operation_type, details): 记录系统操作日志 conn get_db_connection() if not conn: return try: cursor conn.cursor() sql INSERT INTO operation_logs (user_id, operation_type, operation_details, operation_time) VALUES (%s, %s, %s, NOW()) cursor.execute(sql, (user_id, operation_type, str(details))) conn.commit() except Error as e: print(f记录日志失败: {e}) finally: if cursor in locals() and cursor: cursor.close() if conn and conn.is_connected(): conn.close()5. 实际应用与扩展建议5.1 系统部署方案对于小型商铺推荐以下部署方式本地部署使用Raspberry Pi等低成本硬件安装MySQL和Python环境配置自动备份策略简易界面开发使用Tkinter开发基础GUI或使用Flask开发Web界面提供扫码枪等外设支持5.2 未来扩展方向移动端支持开发手机APP或微信小程序实现远程库存查询和销售记录智能分析功能销售预测模型自动补货建议顾客购买行为分析多店铺管理支持连锁店铺数据汇总跨店库存调拨功能5.3 使用建议数据备份策略每日自动备份数据库定期测试备份恢复流程员工培训要点基础数据录入规范日常销售操作流程简单故障排查方法季节性调整旺季前检查系统性能根据销售趋势调整库存策略这个轻量级进销存系统虽然功能简单但已经能够满足小型商铺的基本管理需求。在实际使用中建议先在小规模场景下试用根据实际经营特点逐步调整和扩展功能。对于技术能力有限的商户可以考虑使用现成的开源进销存系统或者寻找本地化服务商进行定制开发。