pycharm代码
"""
商品进销存管理系统 - 命令行界面
《Python数据分析》课程期末项目
"""
# 扩展包区域
import csv
from datetime import datetime
import numpy as np
#常量定义区域
#文件路径
GOODS_FILE = "goods.csv"
PURCHASE_FILE = "purchase.csv"
SALE_FILE = "sales.csv"
# CSV文件表头
GOODS_HEADER = ['goods_id', 'name', 'spec', 'sale_price', 'latest_cost', 'stock_qty', 'min_stock']
PURCHASE_HEADER = ['purchase_id', 'date', 'goods_id', 'quantity', 'cost_price']
SALES_HEADER = ['sale_id', 'date', 'goods_id', 'quantity', 'unit_price', 'cost_price']
# 通用方法区
'''获取goods.csv文件内容'''
def get_goods_lists():
goods_List = []
file = open(GOODS_FILE,"r",encoding="utf-8")
csv_reader = csv.reader(file)
for index, row in enumerate(csv_reader):
if index == 0:
continue
goods_List.append(row)
file.close()
return goods_List
'''获取purchase.csv文件内容'''
def get_purchase_lists():
purchase_List = []
file = open(PURCHASE_FILE,"r",encoding="utf-8")
csv_reader = csv.reader(file)
for index, row in enumerate(csv_reader):
if index == 0:
continue
purchase_List.append(row)
file.close()
return purchase_List
'''获取sales.csv文件内容'''
def get_sales_lists():
sales_List = []
file = open(SALE_FILE, "r", encoding="utf-8")
csv_reader = csv.reader(file)
for index, row in enumerate(csv_reader):
if index == 0:
continue
sales_List.append(row)
file.close()
return sales_List
'''写入CSV文件'''
def write_csv(file_name, header, data_list):
file = open(file_name, "w", encoding="utf-8", newline="")
csv_writer = csv.writer(file)
csv_writer.writerow(header)
csv_writer.writerows(data_list)
file.close()
# 方法区
'''显示主菜单'''
def show_main_menu():
print("\n" + "=" * 35)
print(" 商品进销存管理系统 v1.0")
print("=" * 35)
print("1. 商品管理")
print("2. 进货管理")
print("3. 销售管理")
print("4. 库存查询与统计")
print("5. 财务统计")
print("6. 退出系统")
print("=" * 35)
'''程序主入口'''
def main():
while True:
show_main_menu()
choice = input("请选择操作（1-6）： ").strip()
if choice=="1":
good_menu()
elif choice == "2":
purchase_menu()
elif choice == "3":
sales_menu()
elif choice == "4":
inventory_menu()
elif choice == "5":
finance_menu()
elif choice == "6":
print("\n感谢使用，再见！")
break
else:
print("无效选项，请输入 1-6 之间的数字！")
'''商品管理子菜单'''
def good_menu():
while True:
print("\n=== 商品管理 ===")
print("1. 添加商品")
print("2. 修改商品")
print("3. 删除商品")
print("4. 返回主菜单")
choice = input("请选择： ").strip()
if choice == "1":
add_goods()
elif choice == "2":
update_goods()
elif choice == "3":
del_goods()
elif choice == "4":
break
else:
print("无效选项，请重新选择！")
'''进货管理子菜单'''
def purchase_menu():
while True:
print("\n=== 进货管理 ===")
print("1. 进货登记")
print("2. 查询进货记录")
print("3. 返回主菜单")
choice = input("请选择： ").strip()
if choice == "1":
purchase_register()
elif choice == "2":
query_purchase()
elif choice == "3":
break
else:
print("无效选项，请重新选择！")
'''销售管理子菜单'''
def sales_menu():
while True:
print("\n=== 销售管理 ===")
print("1. 销售登记")
print("2. 查询销售记录")
print("3. 返回主菜单")
choice = input("请选择： ").strip()
if choice == "1":
sales_register()
elif choice == "2":
query_sales()
elif choice == "3":
break
else:
print("无效选项，请重新选择！")
'''库存查询与统计子菜单'''
def inventory_menu():
while True:
print("\n=== 库存查询与统计 ===")
print("1. 查看所有库存列表（含预警）")
print("2. 库存总金额")
print("3. 库存预警商品列表")
print("4. 返回主菜单")
choice = input("请选择： ").strip()
if choice == "1":
query_inventory()
elif choice == "2":
query_inventory_total()
elif choice == "3":
query_inventory_warning()
elif choice == "4":
break
else:
print("无效选项，请重新选择！")
'''财务统计子菜单'''
def finance_menu():
while True:
print("\n=== 财务统计 ===")
print("1. 时间段销售统计")
print("2. 单品利润排行榜")
print("3. 返回主菜单")
choice = input("请选择： ").strip()
if choice == "1":
query_sales_time()
elif choice == "2":
query_sales_profit()
elif choice == "3":
break
else:
print("无效选项，请重新选择！")
'''添加商品'''
def add_goods():
print("\n=== 添加商品 ===")
goods_id = input("请输入商品编号： ").strip()
# 非空验证
if not goods_id:
print("错误：商品编号不能为空！")
return
# 获取商品列表
goods_list = get_goods_lists()
# 判断商品编号是否重复
for goods in goods_list:
if goods[0] == goods_id:
print("错误：商品编号已存在！")
return
# 获取商品名称
name = input("请输入商品名称： ").strip()
spec = input("请输入规格（如 550ml）： ").strip()
try:
sale_price = float(input("请输入销售单价（元）： "))
latest_cost = float(input("请输入最新进价（元）： "))
stock_qty = int(input("请输入初始库存数量： "))
min_stock = int(input("请输入最低库存预警值： "))
if sale_price <= 0 or latest_cost <= 0 or stock_qty < 0 or min_stock < 0:
print("错误：价格和数量必须为正数！")
return
except ValueError:
print("错误：请输入有效的数字！")
return
# CSV文件写入
new_goods = [goods_id, name, spec, sale_price, latest_cost, stock_qty, min_stock]
goods_list.append(new_goods)
file = open(GOODS_FILE, "w", encoding="utf-8", newline="")
csv_writer = csv.writer(file)
csv_writer.writerow(GOODS_HEADER)
csv_writer.writerows(goods_list)
file.close()
print("商品添加成功！")
'''修改商品'''
def update_goods():
print("\n--- 修改商品 ---")
goods_id = input("请输入要修改的商品编号： ").strip()
# 非空验证
if not goods_id:
print("错误：商品编号不能为空！")
return
# 获取商品列表
goods_list = get_goods_lists()
# 获取需要修改的商品
target_goods = None
index_goods = -1
for index, goods in enumerate(goods_list):
if goods[0] == goods_id:
target_goods = goods
index_goods = index
break
if target_goods is None:
print(f"错误：商品编号 {goods_id} 不存在！请先添加新商品")
return
print(f"\n当前商品信息：")
print(f"商品编号： {target_goods[0]}、商品名称：{target_goods[1]}、规格：{target_goods[2]}")
print(f"销售单价： {target_goods[3]}、最新进价：{target_goods[4]}")
print(f"库存数量： {target_goods[5]}、库存预警值： {target_goods[6]}")
print("\n请选择要修改的字段： ")
print("1. 商品名称 2. 规格 3. 销售单价 4. 最新进价 5. 最低库存预警")
choice = input("请输入选项(1-5)").strip()
try:
if choice == "1":
new_name = input("请输入新的商品名称： ").strip()
if not new_name:
print("错误：商品名称不能为空！")
return
target_goods[1] = new_name
elif choice == "2":
new_spec = input("请输入新的规格： ").strip()
if not new_spec:
print("错误：规格不能为空！")
return
target_goods[2] = new_spec
elif choice == "3":
new_sale_price = float(input("请输入新的销售单价："))
if new_sale_price <= 0:
print(f"错误：销售单价必须为正数！")
return
target_goods[3] = new_sale_price
elif choice == "4":
new_latest_cost = float(input("请输入新的最新进价："))
if new_latest_cost <= 0:
print(f"错误：最新进价必须为正数！")
return
target_goods[4] = new_latest_cost
elif choice == "5":
new_min_stock = int(input("请输入新的最低库存预警值："))
if new_min_stock < 0:
print(f"错误：最低库存预警值必须为正数！")
return
target_goods[6] = new_min_stock
except ValueError:
print("无效选项！")
return
# 修改的结果写入csv文件
goods_list[index_goods] = target_goods
file = open(GOODS_FILE, "w", encoding="utf-8", newline="")
csv_writer = csv.writer(file)
csv_writer.writerow(GOODS_HEADER)
csv_writer.writerows(goods_list)
file.close()
print(f"√ 商品 {goods_id} 修改成功！")
'''删除商品'''
def del_goods():
print("\n--- 删除商品 ---")
goods_id = input("请输入要删除的商品编号： ").strip()
goods_list = get_goods_lists()
# 查找商品
target_goods = None
for goods in goods_list:
if goods[0] == goods_id:
target_goods = goods
break
if target_goods is None:
print(f"错误：商品编号 {goods_id} 不存在！")
return
print(f"商品： {target_goods[1]}({goods_id})")
choice = input(f"同时删除该商品的所有进货/销售记录（Y/N）： ").strip().upper()
#删除商品记录
goods_list.remove(target_goods)
write_csv(GOODS_FILE, GOODS_HEADER, goods_list)
# 级联删除进货和销售记录
if choice == "Y":
# 删除进货记录
# 获取进货所有进货记录
purchase_list = get_purchase_lists()
for purchase in purchase_list:
if purchase[2] == goods_id:
#删除操作
purchase_list.remove(purchase)
# 写入csv文件
write_csv(PURCHASE_FILE, PURCHASE_HEADER, purchase_list)
#删除销售记录
sales_list = get_sales_lists()
for sale in sales_list:
if sale[2] == goods_id:
sales_list.remove(sale)
# 写入csv文件
write_csv(SALE_FILE, SALES_HEADER, sales_list)
print(f"√ 商品 {goods_id} 及其相关记录已删除！")
elif choice == "N":
print(f"√ 商品 {goods_id} 已删除（历史记录保留）！")
else:
print("无效选项！")
'''进货登记'''
def purchase_register():
print("\n=== 进货登记 ===")
goods_id = input("请输入商品编号： ").strip()
# 非空验证
if not goods_id:
print("错误：商品编号不能为空！")
return
# 获取商品列表
goods_list = get_goods_lists()
# 查找对应商品
target_goods = None
for goods in goods_list:
if goods[0] == goods_id:
target_goods = goods
break
if target_goods is None:
print(f"错误：商品编号 {goods_id} 不存在！")
return
print(f"商品名称:{target_goods[1]},当前库存: {target_goods[5]},最新进价: {target_goods[4]}")
try:
quantity = int(input("请输入进货数量： "))
cost_price = float(input("请输入进价： "))
if quantity <= 0 or cost_price <= 0:
print("错误：数量和价格必须为正数！")
return
except ValueError:
print(f"错误：请输入有效的数字！ ")
return
# 获取当前时间
now = datetime.now()
date = now.strftime("%Y%m%d%H%M")
# 生成进货编号
purchase_id = "PO"+date
# 进货时间
purchase_date = now.strftime("%Y-%m-%d")
# 生成新的进货记录
new_purchase = [purchase_id, purchase_date, goods_id, quantity, cost_price]
# 获取进货记录
purchase_list = get_purchase_lists()
purchase_list.append(new_purchase)
# csv文件写入
write_csv(PURCHASE_FILE, PURCHASE_HEADER, purchase_list)
# 更新商品库存
target_goods[5] = int(target_goods[5])+quantity
target_goods[4] = cost_price
write_csv(GOODS_FILE, GOODS_HEADER, goods_list)
print("√ 进货记录已保存！！")
'''查询进货记录'''
def query_purchase():
print("\n=== 查询进货记录 ===")
print("1. 按商品编号查询")
print("2. 按时间段查询")
print("3. 返回上一级菜单")
choice = input("请选择： ").strip()
#获取进货记录
purchase_list = get_purchase_lists()
if purchase_list is None:
print("错误：没有进货记录！")
return
if choice == "1":
goods_id = input("请输入商品编号： ").strip()
# 非空验证
if not goods_id:
print("错误：商品编号不能为空！")
return
# 查询结果
result = []
for purchase in purchase_list:
if purchase[2] == goods_id:
result.append(purchase)
if result is None:
print(f"未找到商品 {goods_id} 的进货记录。")
return
print(f"\n进货记录（商品 {goods_id}）：")
print("进货单号\t\t\t\t进货日期\t\t\t数量\t\t单价")
for purchase in result:
print(f"{purchase[0]}\t\t\t{purchase[1]}\t\t\t{purchase[3]}\t\t{purchase[4]}")
elif choice == "2":
start_date_str = input("请输入开始日期（格式：2020-01-01）： ").strip()
end_date_str = input("请输入结束日期（格式：2020-01-01）： ").strip()
# 非空验证
if not start_date_str or not end_date_str:
print("错误：日期不能为空！")
return
# 日期格式验证
try:
start_date = datetime.strptime(start_date_str, "%Y-%m-%d")
end_date = datetime.strptime(end_date_str, "%Y-%m-%d")
except ValueError:
print("错误：日期格式不正确！")
return
# 查询结果
result = []
for purchase in purchase_list:
if start_date <= datetime.strptime(purchase[1], "%Y-%m-%d") <= end_date:
result.append(purchase)
if result is None:
print(f"未找到 {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}的进货记录。")
return
print(f"\n进货记录（{start_date} 至 {end_date}）：")
print("进货单号\t\t\t\t进货日期\t\t\t数量\t\t单价")
for purchase in result:
print(f"{purchase[0]}\t\t\t{purchase[1]}\t\t\t{purchase[3]}\t\t{purchase[4]}")
elif choice == "3":
return
else:
print("无效选项，请重新选择！")
'''销售登记'''
def sales_register():
print("\n=== 销售登记 ===")
goods_id = input("请输入商品编号： ").strip()
# 非空验证
if not goods_id:
print("错误：商品编号不能为空！")
return
# 获取商品列表
goods_list = get_goods_lists()
# 查找对应商品
target_goods = None
for goods in goods_list:
if goods[0] == goods_id:
target_goods = goods
break
if target_goods is None:
print(f"错误：商品编号 {goods_id} 不存在！")
return
print(f"商品名称:{target_goods[1]},当前库存: {target_goods[5]},"
f"售价价格: {target_goods[3]},成本价格: {target_goods[4]}")
try:
quantity = int(input("请输入销售数量： "))
if quantity <= 0:
print("错误：销售数量必须为正整数！")
return
except ValueError:
print("错误：请输入有效的数字！")
return
if quantity > int(target_goods[5]):
print(f"错误：库存不足！当前库存仅 {target_goods[5]} 件，无法销售 {quantity} 件。")
return
# 获取当前时间
now = datetime.now()
date = now.strftime("%Y%m%d%H%M")
# 生成销售编号
sale_id = "SO"+date
# 销售时间
sale_date = now.strftime("%Y-%m-%d")
# 生成新的销售记录
new_sale = [sale_id, sale_date, goods_id, quantity, target_goods[3], target_goods[4]]
# 获取销售记录
sales_list = get_sales_lists()
sales_list.append(new_sale)
# csv文件写入
write_csv(SALE_FILE, SALES_HEADER, sales_list)
# 更新商品库存
target_goods[5] = int(target_goods[5]) - quantity
write_csv(GOODS_FILE, GOODS_HEADER, goods_list)
# 输出销售单号
print(f"销售单号：{sale_id}")
print(f"销售数量:{quantity}")
print(f"销售金额:{round(int(quantity)*float(target_goods[3]),2)}")
print(f"销售成本:{round(int(quantity)*float(target_goods[4]),2)}")
print(f"库存数量:{target_goods[5]}")
print("√ 销售记录已保存！！")
'''查询销售记录'''
def query_sales():
print("\n=== 查询销售记录 ===")
print("1. 按商品编号查询")
print("2. 按时间段查询")
print("3. 返回上一级菜单")
choice = input("请选择： ").strip()
# 获取销售记录
sales_list = get_sales_lists()
if sales_list is None:
print("错误：没有销售记录！")
return
if choice == "1":
goods_id = input("请输入商品编号： ").strip()
# 非空验证
if not goods_id:
print("错误：商品编号不能为空！")
return
# 查询结果
result = []
for sale in sales_list:
if sale[2] == goods_id:
result.append(sale)
if result is None:
print(f"未找到商品 {goods_id} 的销售记录。")
return
print(f"\n销售记录（商品 {goods_id}）：")
print("销售单号\t\t\t\t销售日期\t\t\t数量\t\t单价\t\t成本")
for sale in result:
print(f"{sale[0]}\t\t\t{sale[1]}\t\t\t{sale[3]}\t\t{sale[4]}\t\t{sale[5]}")
elif choice == "2":
start_date_str = input("请输入开始日期（格式：2020-01-01）： ").strip()
end_date_str = input("请输入结束日期（格式：2020-01-01）： ").strip()
# 非空验证
if not start_date_str or not end_date_str:
print("错误：日期不能为空！")
return
# 日期格式验证
try:
start_date = datetime.strptime(start_date_str, "%Y-%m-%d")
end_date = datetime.strptime(end_date_str, "%Y-%m-%d")
except ValueError:
print("错误：日期格式不正确！")
return
# 查询结果
result = []
for sale in sales_list:
if start_date <= datetime.strptime(sale[1], "%Y-%m-%d") <= end_date:
result.append(sale)
if result is None:
print(f"未找到 {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}的销售记录。")
return
print(f"\n销售记录（{start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}）： ")
print("销售单号\t\t\t\t销售日期\t\t\t数量\t\t单价\t\t成本")
for sale in result:
print(f"{sale[0]}\t\t\t{sale[1]}\t\t\t{sale[3]}\t\t{sale[4]}\t\t{sale[5]}")
elif choice == "3":
return
else:
print("无效选项，请重新选择！")
'''查看所有库存列表（含预警）'''
def query_inventory():
print("\n=== 查看所有库存列表（含预警） ===")
goods_list = get_goods_lists()
if goods_list is None:
print("错误：没有商品记录！")
return
print(f"\n商品列表： ")
print(f"{'商品编号':<8} {'商品名称':<12} {'规格':<10} {'售价':<6} {'成本':<6} {'库存':<6} {'预警':<6}")
# 打印数据
for goods in goods_list:
print(f"{goods[0]:<8} {goods[1]:<12} {goods[2]:<10} {goods[3]:<6} {goods[4]:<6} {goods[5]:<6} "
f"{'!!!' if int(goods[5]) < int(goods[6]) else '正常':<6}")
'''库存总金额'''
def query_inventory_total():
print("\n=== 库存总金额 ===")
goods_list = get_goods_lists()
if goods_list is None:
print("错误：没有商品记录！")
return
# NumPy 实现：向量化计算库存总金额
# 读取numpy数组
goods_arr = np.loadtxt(GOODS_FILE,
delimiter=",",
dtype=np.float32,
skiprows=1,
encoding="utf-8",
usecols=(3,5))
# 计算库存总金额
total = np.sum(goods_arr[:,0]*goods_arr[:,1])
print(f"正在使用 NumPy 计算库存总金额...")
print(f"库存总金额（按最新进价计算）= {round(total,2)} 元")
'''库存预警商品列表'''
def query_inventory_warning():
print("\n=== 库存预警商品列表 ===")
goods_list = get_goods_lists()
if goods_list is None:
print("错误：没有商品记录！")
return
print(f"\n库存预警商品列表： ")
print(f"{'商品编号':<8} {'商品名称':<12} {'规格':<10} {'售价':<6} {'成本':<6} {'库存':<6} {'预警':<6}")
goods_list_waring = []
for goods in goods_list:
if int(goods[5]) < int(goods[6]):
goods_list_waring.append(goods)
if goods_list_waring is None:
print("没有库存预警商品！")
return
# 打印数据
for goods in goods_list_waring:
print(f"{goods[0]:<8} {goods[1]:<12} {goods[2]:<10} {goods[3]:<6} {goods[4]:<6} {goods[5]:<6} "
f"{'!!!':<6}")
'''时间段销售统计'''
def query_sales_time():
print("\n=== 时间段销售统计 ===")
start_date_str = input("请输入开始日期（格式：2020-01-01）： ").strip()
end_date_str = input("请输入结束日期（格式：2020-01-01）： ").strip()
# 非空验证
if not start_date_str or not end_date_str:
print("错误：日期不能为空！")
return
# 日期格式验证
try:
start_date = datetime.strptime(start_date_str, "%Y-%m-%d")
end_date = datetime.strptime(end_date_str, "%Y-%m-%d")
except ValueError:
print("错误：日期格式不正确！")
return
# 获取销售记录
sales_list = get_sales_lists()
if sales_list is None:
print("错误：没有销售记录！")
return
# 查询结果
result = []
for sale in sales_list:
if start_date <= datetime.strptime(sale[1], "%Y-%m-%d") <= end_date:
result.append(sale)
if result is None:
print(f"未找到 {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}的销售记录。")
return
quantity_list = []
unit_price_list = []
cost_price_list = []
for sale in result:
quantity_list.append(int(sale[3]))
unit_price_list.append(float(sale[4]))
cost_price_list.append(float(sale[5]))
quantity_arr = np.array(quantity_list)
unit_price_arr = np.array(unit_price_list)
cost_price_arr = np.array(cost_price_list)
total_amount = np.sum(unit_price_arr*quantity_arr)
total_cost = np.sum(cost_price_arr*quantity_arr)
total_profit = total_amount - total_cost
profit_rate = (total_profit / total_amount * 100) if total_amount > 0 else 0
print(f"\n统计结果（{start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}）： ")
print(f"数量量：共{len(result)}笔销售记录")
print(f"销售总金额：{round(total_amount,2)}元")
print(f"销售总成本：{round(total_cost,2)}元")
print(f"销售利润：{round(total_profit,2)}元")
print(f"利润率：{round(profit_rate,2)}%")
'''单品利润排行榜'''
def query_sales_profit():
print("\n=== 单品利润排行榜 ===")
#获取商品列表
goods_list = get_goods_lists()
if goods_list is None:
print("错误：没有商品记录！")
return
# 获取销售记录
sales_list = get_sales_lists()
if sales_list is None:
print("错误：没有销售记录！")
return
goods_profit = 0.0
goods_profit_list =[]
for goods in goods_list:
for sale in sales_list:
if sale[2] == goods[0]:
goods_profit += (float(sale[4]) - float(sale[5])) * int(sale[3])
goods_profit_list.append([goods[0],goods[1],round(goods_profit,2)])
goods_profit = 0.0
# 按利润降序排序
goods_profit_list.sort(key=lambda x:x[2],reverse=True)
print(f"\n单品利润排行榜：")
for goods in goods_profit_list:
print(f"商品编号:{goods[0]} 商品名称:{goods[1]} 利润:{goods[2]}")
# 程序运行入口
if __name__ == "__main__":
main()
以上代码均为复制老师的完整代码可运行