conf===>settings.py
''' 存放项目相关配置信息（相关路径的配置） ''' import os #获取文件跟目录路径 BASE_PATH=os.path.dirname(os.path.dirname(__file__)) #转到db目录下 DB_PATH=os.path.join(BASE_PATH,'db') #转到user_data目录下 USER_DATA=os.path.join(DB_PATH,'user_data') #转到shop_data目录下 SHOP_DATA=os.path.join(DB_PATH,'shop_data') #日志的配置 # 定义三种日志输出格式 standard_format = '[%(asctime)s][%(threadName)s:%(thread)d][task_id:%(name)s][%(filename)s:%(lineno)d]' \ '[%(levelname)s][%(message)s]' # 其中name为getlogger指定的名字 simple_format = '[%(levelname)s][%(asctime)s][%(filename)s:%(lineno)d]%(message)s' id_simple_format = '[%(levelname)s][%(asctime)s] %(message)s' # 定义日志输出格式 logfile_dir = os.path.join(BASE_PATH, 'log') # print(logfile_dir) logfile_name = 'ATM.log' if not os.path.isdir(logfile_dir): os.mkdir(logfile_dir) # log文件的全路径 logfile_path = os.path.join(logfile_dir, logfile_name) LOGGING_DIC = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'standard': { 'format': standard_format }, 'simple': { 'format': simple_format }, }, 'filters': {}, 'handlers': { # 打印到终端的日志 'console': { 'level': 'DEBUG', 'class': 'logging.StreamHandler', # 打印到屏幕 'formatter': 'simple' }, # 打印到文件的日志 'default': { 'level': 'DEBUG', 'class': 'logging.handlers.RotatingFileHandler', # 保存到文件 'formatter': 'standard', 'filename': logfile_path, # 日志文件 'maxBytes': 1024 * 1024 * 5, # 日志大小 5M 'backupCount': 5, 'encoding': 'utf-8', }, }, 'loggers': { # logging.getLogger(__name__)拿到的logger配置 '': { 'handlers': ['default', 'console'], 'level': 'DEBUG', 'propagate': True, }, }, }
core====>admin.py
from interface import admin_interface def change_locked(): #给用户账户解锁 while True: username=input('请输入解锁账户名：').strip() flag,msg=admin_interface.unlockuser(username) if flag: print(msg) break else: print(msg) def change_money(): while True: username=input('请输入修改信用额度的账户名：').strip() creditsmoney = input('请输入用户额度：').strip() flag,msg=admin_interface.change_creditsmoney(username,creditsmoney) if flag: print(msg) break else: print(msg) def add_user(): while True: username=input('请输入需要设置管理员权限的账户名：').strip() flag,msg=admin_interface.add_admin(username) if flag: print(msg) break else: print(msg) def add_shop(): while True: print("=======温情提示：输入q/Q则退出添加商品=====》") shopname=input('请输入要添加的商品名称：').strip() if shopname=='q'or shopname=='Q': break unitprice= input('请输入商品单价：').strip() total = input('请输入添加商品总数：').strip() if total.isdigit(): flag,msg=admin_interface.add_shop(shopname,unitprice,total) if flag: print(msg) continue else: print(msg) continue else: print('输入的数量不合法，添加失败，请重新输入') choice_dic = { '1': change_locked, '2': change_money, '3': add_user, '4': add_shop, '0': exit } #管理员功能启动入口 def admin_run(): while True: print(""" ========= ~欢迎进入管理员界面~ ======== 1 用户账户解锁 2 修改额度 3 设置管理员账号 4 添加商品库存 0 退出 ================ end ================= """) choice = input('请输入功能编号：').strip() if choice in choice_dic: choice_dic[choice]() else: print('输入的编号不存在或者不合法，请重新输入~~') continue
core====>src.py
''' 用户视图层，存放项目核心逻辑。。。 ''' #调用不同的接口模块 from interface import user_interface from interface import bank_interface from interface import shop_interface from lib import common # 全局变量，记录用户是否已登录 Login_user=None def register(): while True: username=input("请输入注册用户姓名：").strip() password = input("请输入密码：").strip() password1 = input("请再次输入密码：").strip() if password==password1: #把用户视图层得到的数据传给接口层进行处理 flag,msg=user_interface.register_interface(username,password1) if flag: print(msg) break else: print(msg) def login(): count=0 while count<=3: username = input("请输入用户名：").strip() password = input("请输入密码：").strip() flag,msg=user_interface.login_interface(username, password) if flag=='M': #返回值为M，密码错误 count+=1 if count==3: print('密码输错三次，已被锁定！') #把锁定值传去数据处理层进行更新保存 user_interface.login_locked(username,1) break print('%s还有%s次机会哦~！' %(msg,(3-count))) continue elif flag=='L': #返回为L账户被锁定 print(msg) break elif flag:#返回为1，登陆成功 print(msg) global Login_user Login_user=username break else: print(msg)#返回为0，用户名不存在 break @common.login_auth def check_mongey(): user_money=bank_interface.checkmongey_interface(Login_user) print("您的账户余额为：%s" %user_money) @common.login_auth def payback_money(): while True: money=input('请输入还款金额：').strip() if money.isdigit(): flag,msg=bank_interface.paybackmoney_interface(Login_user,money) if flag: print(msg) break elif flag=='N': print(msg) break else: print(msg) continue else: print('请输入合法数字！') @common.login_auth def withdraw_money(): while True: money=input('请输入您要提现的金额：').strip() if money.isdigit(): flag,msg=bank_interface.withdrawmoney_interface(Login_user,money) if flag: print(msg) break else: print(msg) continue else: print('请输入合法数字！') @common.login_auth def transfer_money(): while True: collectuser = input('请输入您要转账的账户名：').strip() transfermoney = input('请输入您要转账金额：').strip() if transfermoney.isdigit(): flag, msg = bank_interface.transfermoney_interface(Login_user,collectuser,transfermoney) if flag: print(msg) break else: print(msg) continue else: print('请输入合法数字！') @common.login_auth def shoppingcart(): from db import db_handler shop_dic=db_handler.shopdata() #商品库存数据 print(shop_dic) user_dic=db_handler.userdata(Login_user) shopcar_dic=user_dic['shoppingcart'] #用户购物车数据 print(shopcar_dic) print('商城商品展示如下：') for shopneme,shop_list in shop_dic.items(): unit_price,total=shop_list print(f'商品名称：{shopneme}，单价：{unit_price}元，剩余库存：{total}个') print('温情提示：输入q/Q则结束商品添加！') while True: shop_name=input('请输入您要购买的商品名称：').strip() if shop_name=='q'or shop_name=='Q': print('已退出！') break if shop_name in shop_dic: count=input('请输入商品个数：').strip() if count.isdigit(): flag,msg=shop_interface.add_shopcar(Login_user,shop_name,count) if flag: print(msg) continue else: print(msg) continue else: print("商品个数大于库存或不合法，请重新输入") else: print("商品不存在，请重新输入~！") @common.login_auth def check_shoppingcart(): usershopcar_dic=shop_interface.checkshop_interface(Login_user) all_total=0 if usershopcar_dic: for shopname,shop_list in usershopcar_dic.items(): price,count=shop_list price=float(price) count=float(count) total=float(price*count) all_total+=total print(f'商品名称：{shopname}，单价：{price}元，数量：{count}个，总价：{total}') while True: choice=input("是否对购物车商品进行结算？（Y：结算/N：放弃）：").strip() if choice=='Y': #调用银行结算接口进行结算 flag,msg=bank_interface.paymoney_interface(Login_user,all_total) if flag: print(msg) break else: print(msg) break if choice=='N': break else: print('您的购物车空空如也~~！') @common.login_auth def check_paylist(): user_flow = bank_interface.paylist_interface(Login_user) if user_flow: for flow in user_flow: print(flow) else: print('当前用户没有流水记录！') @common.login_auth def admin(): from core import admin #首先要判断该账号是否是管理账号 flag=user_interface.isadmin(Login_user) if flag: #返回值为1，则是管理员账号，否则不是管理员 admin.admin_run() else: print('非管理员账号，无法使用该功能！') #用户choice第一次选择，使用以下功能： choice_dic={ '1': register, '2': login, '3': check_mongey, '4': payback_money, '5': withdraw_money, '6': transfer_money, '7': shoppingcart, '8': check_shoppingcart, '9': check_paylist, '10': admin, '0': exit } #项目启动程序 def run(): while True: print(""" ========= ~ATM + 购物车~ ======== 1 注册功能 2 登录功能 3 查看余额 4 还款功能 5 提现功能 6 转账功能 7 购物功能 8 查看购物车 9 查看流水 10 管理员功能 0 退出 ============ end ============== """) choice=input('请输入功能编号：').strip() if choice in choice_dic: choice_dic[choice]() else: print('输入的编号不存在或者不合法，请重新输入~~') continue
db====>db_handler.py
''' 数据处理层，用于处理相关数据（数据的存取，更新等） ''' import json from conf import settings import os #查看用户信息数据，调用json模块 def userdata(username): #获取用户名的文件，需要先进行路径拼接 user_path=os.path.join(settings.USER_DATA,f'{username}.json') #进行校验 if os.path.exists(user_path): with open(user_path,'r',encoding='utf-8')as f: user_dic=json.load(f) return user_dic def userdata_save(user_dic): #首先需要用户名来拼接用户的文件名，所以先从字典里取出用户名 username=user_dic.get('username') user_path=os.path.join(settings.USER_DATA,f'{username}.json') #保存数据 with open(user_path,'w',encoding='utf-8')as f: json.dump(user_dic,f,ensure_ascii=False) #查看商品库存信息数据，调用json模块 def shopdata(): #获取商品库存的文件路径，需要先进行路径拼接 shopstocks_path=os.path.join(settings.SHOP_DATA,'shopstocks.json') #进行校验 if os.path.exists(shopstocks_path): with open(shopstocks_path,'r',encoding='utf-8')as f: shopstocks_dic=json.load(f) return shopstocks_dic def shopdata_save(shopstocks_dic): #保存数据 shopstocks_path=os.path.join(settings.SHOP_DATA,'shopstocks.json') with open(shopstocks_path,'w',encoding='utf-8')as f: json.dump(shopstocks_dic,f,ensure_ascii=False)
interface====>admin_interface.py
from db import db_handler #用户账户解锁接口 def unlockuser(username): user_dic=db_handler.userdata(username) if user_dic: user_dic['locked']=0 msg=f'用户{username}解锁成功！' db_handler.userdata_save(user_dic) return True,msg else: msg=f'解锁用户不存在，请核实！' return False,msg #用户信用额度修改接口 def change_creditsmoney(username,creditsmoney): user_dic=db_handler.userdata(username) if user_dic: user_dic['creditsmoney']=int(creditsmoney) msg=f'{username}账户额度修改成功，额度为：{creditsmoney}元！' db_handler.userdata_save(user_dic) return True,msg else: msg=f'修改账户不存在，请核实！' return False,msg #设置管理账号接口 def add_admin(username): user_dic=db_handler.userdata(username) if user_dic: user_dic['is_admin']=True msg=f'修改{username}账号为管理员~！' db_handler.userdata_save(user_dic) return True,msg else: msg=f'修改账户不存在，请核实！' return False,msg #管理员添加商品库存接口 def add_shop(shopname,price,total): shopstocks_dic=db_handler.shopdata() price=int(price) total=int(total) if shopname in shopstocks_dic: shopstocks_dic[shopname][0]=price shopstocks_dic[shopname][1]+=total msg=f'{shopname}商品数量更新成功~~~' db_handler.shopdata_save(shopstocks_dic) return True, msg else: shopstocks_dic[shopname]=[price,total] msg=f'{shopname}商品添加成功~~~' db_handler.shopdata_save(shopstocks_dic) return True,msg
interface====>bank_interface.py
''' 银行相关业务的接口 ''' #查询账户余额接口 from db import db_handler from lib import common bank_logger=common.logger(log_type='bank') #用户查询余额接口 def checkmongey_interface(username): user_dic=db_handler.userdata(username) return user_dic['money'] #用户还款接口 def paybackmoney_interface(username,paymoney): #paymoney=300 user_dic=db_handler.userdata(username) money=float(user_dic['money']) #money=15000 creditsmoney=float(user_dic['creditsmoney']) #获取账号信用额度creditsmoney 20000 re_money=creditsmoney-money #re_money=500 paymoney=float(paymoney) #paymoney=300 leftmoney=creditsmoney-(money+paymoney) #leftmoney=20000-(15000+300) if money<=creditsmoney: #判断账户金额是否超过信用额度，超过则不需要还款 if paymoney<=re_money: msg=f'用户{username}还款{paymoney}元成功，还剩{leftmoney}元未还款！' bank_logger.info(msg) user_dic['user_flow'].append(msg) user_dic['money']=money+paymoney db_handler.userdata_save(user_dic) return True,msg else: msg=f'当前用户全部还款金额为：{re_money}元，请重新输入' return False,msg else: msg=f'您的账户余额大于信用额度，不需要进行还款~' return 'N',msg #提现接口 def withdrawmoney_interface(username,withdraw_money,hand=0.005): #用户：tank提现金额：10000：手续费：0.5%（10000*0.5%） 得到10000-（10000*0.5%） user_dic=db_handler.userdata(username) withdraw_money=float(withdraw_money) #withdraw_money=10000 money=float(user_dic['money']) #用户账户余额 money=10000 handmoney=withdraw_money*hand #手续费10000*0.5% actualmoney=withdraw_money-handmoney if withdraw_money<=money: msg=f'用户{username}提现金额：{withdraw_money}元，扣除手续费：{handmoney}元，实际取款：{actualmoney}元！' bank_logger.info(msg) user_dic['user_flow'].append(msg) user_dic['money']=money-withdraw_money db_handler.userdata_save(user_dic) return True,msg msg=f'您的账户余额不足，请查询好充值！' return False,msg #转账接口 def transfermoney_interface(transfer_user,collect_user,transfermoney): transf_user_dic=db_handler.userdata(transfer_user) collect_user_dic=db_handler.userdata(collect_user) if collect_user_dic: money=float(transf_user_dic['money']) transfermoney=float(transfermoney) if transfermoney<=money: msg=f'用户{transfer_user}成功向{collect_user}的账户，转账{transfermoney}元！' bank_logger.info(msg) msg1=f'用户{collect_user}收到{transfer_user}账户的转账{transfermoney}元！' bank_logger.info(msg1) transf_user_dic['user_flow'].append(msg) collect_user_dic['user_flow'].append(msg1) transf_user_dic['money']=money-transfermoney money=float(collect_user_dic['money']) collect_user_dic['money']=money+transfermoney db_handler.userdata_save(transf_user_dic) db_handler.userdata_save(collect_user_dic) return True,msg else: msg=f'转账失败，账户余额不足，请先查询！' return False,msg else: msg=f'对方账户不存在，请先核实！' return False,msg #付款接口 def paymoney_interface(username,paymoney): user_dic=db_handler.userdata(username) money=float(user_dic['money']) paymoney=float(paymoney) if user_dic: if paymoney<=money: user_dic['money']=money-paymoney msg=f'用户{username}支付{paymoney}元成功，准备发货啦~' bank_logger.info(msg) user_dic['shoppingcart']={} #付款成功后，清空购物车 user_dic['user_flow'].append(msg) db_handler.userdata_save(user_dic) return True,msg else: msg=f'您的账户余额不足，请先查询！' return False,msg #查看消费流水接口 def paylist_interface(username): user_dic=db_handler.userdata(username) return user_dic['user_flow']
interface====>shop_interface.py
''' 购物商城相关业务的接口 ''' from db import db_handler from lib import common shop_logger=common.logger(log_type='shop') #用户添加购物车商品接口 def add_shopcar(username,shopname,count): shop_dic=db_handler.shopdata() #获取商品库存信息所有商品 """ shop_dic={"suger": [20, 50], "mac": [3000, 50], "phone": [2000, 50], "ice": [400, 500]} """ """ "shoppingcart": {"phone": [2000,1]} shopcar_dic={"phone": [2000,1]} """ user_dic=db_handler.userdata(username) #获取用户信息 shopcar_dic=user_dic['shoppingcart'] #用户信息中的购物车信息 if shopname in shop_dic: total=int(shop_dic[shopname][1]) #库存数量 unit_price=float(shop_dic[shopname][0]) #商品单价 count=int(count) if count<=total: shop_dic[shopname][1]=total-count if shopname in shopcar_dic: shopcar_dic[shopname][1]+=count msg=f'已将{shopname}添加到购物车！' shop_logger.info(msg) db_handler.userdata_save(user_dic) db_handler.shopdata_save(shop_dic) return True, msg else: shopcar_dic[shopname]=[unit_price,count] user_dic['shoppingcart']=shopcar_dic msg=f'已将{shopname}添加到购物车！' shop_logger.info(msg) db_handler.userdata_save(user_dic) db_handler.shopdata_save(shop_dic) return True,msg else: msg=f'商品库存不足，还剩{total}个，请重新输入' return False,msg #查看购物车接口 def checkshop_interface(username): user_dic=db_handler.userdata(username) usershopcar_dic=user_dic['shoppingcart'] return usershopcar_dic
interface====>user_interface.py
''' 逻辑接口层，用户接口 ''' from db import db_handler from lib import common user_logger=common.logger(log_type='user') #注册接口，接收注册数据进行处理 def register_interface(username,password,creditsmoney=20000): #调用用户数据处理层，进行数据的校验判断 user_dic=db_handler.userdata(username) if user_dic: return False,'您注册的用户名已存在！' #如果不存在，则将密码进行加密后，传到数据处理层进行保存 password=common.md5password(password) money=creditsmoney #用户字典信息 user_dic={ 'username': username, 'password': password, 'money': money, 'creditsmoney': creditsmoney, 'user_flow': [], 'shoppingcart':{}, 'locked':0, 'is_admin': False, } #传到数据处理层进行保存 msg=f'您已注册成功，您的信用默认额度为：{creditsmoney}，如需更改请联系管理员！====》 请先登陆！' user_logger.info(msg) db_handler.userdata_save(user_dic) return True,msg #登陆接口 def login_interface(username,password): #调用数据处理层处理数据，查看用户名是否存在 user_dic=db_handler.userdata(username) if user_dic: #先判断用户是否被锁定 locked=user_dic.get('locked') if locked==1: #1 被锁，0 未被锁 msg=f'您的账户已被锁定，请联系管理员解锁！' return 'L',msg if locked==0: #把密码进行加密 password=common.md5password(password) md_password=user_dic.get('password') if password==md_password: msg=f'你已登陆成功！' user_logger.info(msg) return True,msg else: msg=f'您的密码输入有误，请重新输入,' return 'M',msg else: msg=f'您输入的用户名不存在或错误，请先核实！' return False,msg #用户被锁定接口 def login_locked(username,lock): user_dic=db_handler.userdata(username) user_dic['locked']=lock db_handler.userdata_save(user_dic) #判断用户账号是否是管理员账号 def isadmin(username): user_dic=db_handler.userdata(username) flag=user_dic['is_admin'] return flag
lib====>common.py
''' 存放公共的使用方法 ''' import hashlib from conf import settings #对密码进行处理，不加盐 def md5password(password): md_password=(hashlib.md5(password.encode('utf-8'))).hexdigest() return md_password #登陆功能添加装饰器 def login_auth(func): from core import src def inner(*args, **kwargs): if src.Login_user: res = func(*args, **kwargs) return res else: print('请先登录后使用~!') src.login() return inner #添加日志功能，导入日志模块 import logging.config def logger(log_type): #获取日志配置信息获取日志路径 logging.config.dictConfig(settings.LOGGING_DIC) logger=logging.getLogger(log_type) return logger
start.py
""" 程序的启动入口 """ import os import sys import core #添加环境解释器的环境变量 sys.path.append(os.path.dirname(__file__)) #引用逻辑层里面的启动程序入口，开始启动程序 from core import src if __name__=='__main__': src.run()