目录：
#start.py
from core import src if __name__ == '__main__': src.run()
#admin.py
from lib import common from interface import admin_interface logger_admin=common.get_logger("admin") admin_name=None #管理员登录 def admin_login(): global admin_name count = 0 while count < 3: name = input("请输入管理员账号:>>>").strip() passwd = input("请输入密码:>>>").strip() if name == "b": break mask, msg = admin_interface.back_admin(name, passwd, logger_admin) if mask: admin_name = name break else: print(msg) else: print("输错三次，退出程序") exit() #冻结账户 def locked(): while True: lock_user=input("请输入要冻结的账户:>>>").strip() mask,msg=admin_interface.back_locked(lock_user,admin_name,logger_admin) if mask: break else: print(msg) #解冻账户 def unlocked(): while True: unlocked_user=input("请输入要解冻的账户:>>>").strip() mask,msg=admin_interface.back_unlocked(unlocked_user,admin_name,logger_admin) if mask: break else: print(msg) #添加账户 def admin_add_user(): from core import src src.register() #修改额度 def rechange_balance(): while True: name=input("请输入修改的账号:>>>").strip() balance=input("请输入修改额度:>>>").strip() if balance.isdigit(): balance=int(balance) mask,msg=admin_interface.back_rechange_balance(name,balance,admin_name,logger_admin) if mask: break else: print(msg) else: print("请输入数字(正整数)") #功能字典 func_dic={ "1":["冻结账户",locked], "2":["解冻账户",unlocked], "3":["添加账户",admin_add_user], "4":["修改额度",rechange_balance], } #主程序 def run(): admin_login() if admin_name: while True: common.print_func(func_dic) choice=input("请输入功能编号:>>>").strip() if choice == "b": break if choice in func_dic: func_dic[choice][1]() else: print("输入的功能编号不存在")
#settings.py
import os #项目根目录 BASR_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__))) #db文件目录 DB_FILE_DIR=os.path.join(BASR_DIR,"db","db_file") #日志文件路径 LOG_FILE_PATH=os.path.join(BASR_DIR,"log","access.log") #设置logging配置字典 standard_format = '[%(asctime)s][%(threadName)s:%(thread)d][日志名:%(name)s][%(filename)s:%(lineno)d]' '[%(levelname)s][%(message)s]' test_format = '%(message)s' # 3、日志配置字典 LOGGING_DIC = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'standard': { 'format': standard_format }, 'test': { 'format': test_format }, }, 'filters': {}, 'handlers': { #打印到终端的日志 'console': { 'level': 'DEBUG', 'class': 'logging.StreamHandler', # 打印到屏幕 'formatter': 'test' }, #打印到文件的日志,收集info及以上的日志 'default': { 'level': 'DEBUG', 'class': 'logging.handlers.RotatingFileHandler', # 保存到文件,日志轮转 'formatter': 'standard', # 可以定制日志文件路径 # BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # log文件的目录 # LOG_PATH = os.path.join(BASE_DIR,'a1.log') 'filename': LOG_FILE_PATH, # 日志文件 'maxBytes': 1024*1024*5, # 日志大小 5M，默认单位字节，日志大小达到5M就将其内容剪切到其他文件备份 'backupCount': 5, #最大备份5份，备份文件名access.log.1，在filename名后.1或.2、.3。。。。 'encoding': 'utf-8', # 日志文件的编码，再也不用担心中文log乱码了 }, }, 'loggers': { #logging.getLogger(__name__)拿到的logger配置 '': { 'handlers': ['default','console'], #如果k为空，代表找不到k的时候用空k,并将日志名用这个空k 'level': 'DEBUG', 'propagate': False, }, }, }
#src.py
from lib import common from conf import settings from interface import user_interface from interface import bank_interface from interface import shop_interface from core import admin logger_access=common.get_logger("access") logger_trade=common.get_logger("trade") login_user=None #用户注册 def register(): while True: add_name=input("请输入注册的用户名:>>>").strip() if add_name == "b": break add_pwd=input("请输入密码:>>>").strip() re_add_pwd=input("再次输入密码:>>>").strip() if add_pwd == re_add_pwd: mask,msg=user_interface.user_register(add_name,add_pwd,logger_access) if mask: break else: print(msg) else: print("两次密码输入不一致，请重新输入") #登录 def login(): global login_user count=0 while count < 3: inp_name=input("请输入用户名:>>>").strip() if inp_name == "b": break inp_pwd=input("请输入密码:>>>").strip() mask,msg=user_interface.user_login(inp_name,inp_pwd,logger_access) if mask: login_user=inp_name break else: print(msg) count+=1 else: print("输错三次，退出程序") #查看余额 @common.auth def check_money(): balance=user_interface.back_balance(login_user) print("用户{}余额为{}元".format(login_user,balance)) #提现 @common.auth def withdraw(): while True: withdraw_money=input("请输入提现金额:>>>").strip() if withdraw_money == "b": break if withdraw_money.isdigit(): withdraw_money=int(withdraw_money) mask,msg=bank_interface.back_withdraw(login_user,withdraw_money,logger_trade) if mask: break else: print(msg) else: print("请输入数字(正整数)") #还款 @common.auth def pay(): while True: pay_money=input("请输入还款金额:>>>").strip() if pay_money == "b": break if pay_money.isdigit(): pay_money=int(pay_money) bank_interface.back_pay(login_user,pay_money,logger_trade) break else: print("请输入数字(正整数)") #转账 @common.auth def transfer(): while True: transfer_to_user=input("请输入转入的账户:>>>").strip() if transfer_to_user == "b": break transfer_money=input("请输入转账金额:>>>").strip() if transfer_money.isdigit(): transfer_money=int(transfer_money) mask,msg=bank_interface.back_transfer(login_user,transfer_to_user,transfer_money,logger_trade) if mask: break else: print(msg) else: print("请输入数字(正整数)") #查看流水 @common.auth def check_flow(): flow=user_interface.back_flow(login_user) if not flow: print("用户{}暂时还没有流水记录") print("==========用户流水记录==========") for i in flow: print(i) print("===============end==============") #商品菜单 goods_dic={ "数码":{ "1":["iphone 11",5000], "2":["ipad",3000], "3":["dell G3",6000], }, "文具":{ "1":["钢笔",100], "2":["铅笔",3], "3":["橡皮",1], }, "生活":{ "1":["追风筝的人",45], "2":["egon传记",99], "3":["egon**三部曲",199], } } #购物功能 @common.auth def shopping(): while True: print("==========商品类型==========") for i in goods_dic: print(i) print("=============end============") choice=input("请输入商品类型:>>>").strip() if choice == "b": break if choice in goods_dic: while True: common.print_good(goods_dic[choice]) choice1=input("请输入选中商品的商品编号:(输入y结算)").strip() if choice1 == "b": break if choice1 == "y" or choice1 == "Y": #开始结算 mask,msg=shop_interface.back_car_money(login_user,logger_trade) if mask: break else: print(msg) if choice1 in goods_dic[choice]: good_name, price = goods_dic[choice][choice1] #将选择商品加入购物车 shop_interface.add_goods(login_user,good_name,price) #输出当前购物车信息 shop_car=shop_interface.back_car(login_user) print(shop_car) else: print("输入的商品编号不存在") else: print("输入的商品类型编号不存在") #查看购物车 @common.auth def check_car(): shop_car=shop_interface.back_car(login_user) print("========用户{}购物车信息========".format(login_user)) for i in shop_car: print("商品名:{},商品价格:{},商品数量:{}".format(i,shop_car[i][0],shop_car[i][1])) print("===============end===============") #管理员功能 def admin_func(): admin.run() #退出 def logout(): choice=input("是否确认退出y/n,Y/N:>>>").strip() if choice == "y" or choice == "Y": exit() elif choice == "n" or choice == "N": pass else: print("请输入y/n,Y/N") #功能字典 func_dic={ "0":["退出",logout], "1":["注册",register], "2":["登录",login], "3":["查看余额",check_money], "4":["提现",withdraw], "5":["还款",pay], "6":["转账",transfer], "7":["查看流水",check_flow], "8":["购物功能",shopping], "9":["查看购物车",check_car], "10":["管理员功能",admin_func], } #主函数 def run(): while True: common.print_func(func_dic) choice=input("请输入功能编号:>>>").strip() if choice in func_dic: func_dic[choice][1]() else: print("输入的功能编号不存在")
#db_handle.py
def load_info(user_name): #用户文件名 user_file="{}.json".format(user_name) user_file_path=os.path.join(settings.DB_FILE_DIR,user_file) if os.path.isfile(user_file_path): with open(user_file_path,"r",encoding="utf-8") as f: data=json.load(f) return data #保存或更新用户信息 def dump_info(data): # 用户文件名 user_file = "{}.json".format(data["name"]) user_file_path = os.path.join(settings.DB_FILE_DIR, user_file) with open(user_file_path,"w",encoding="utf-8") as f: json.dump(data,f,ensure_ascii=False)
#user_interface.py
from db import db_handle from lib import common #用户注册 def user_register(user_name,passwd,logger,balance=15000): #加载用户信息 data=db_handle.load_info(user_name) if data: return False,"用户已存在" #获取密码MD5 pwd_md5=common.get_md5(passwd) user_data={ "name":user_name, "passwd":pwd_md5, "balance":balance, "flow":[], "shop_car":{}, "buyed":{}, "locked":False, "admin":False } db_handle.dump_info(user_data) logger.info("用户{}注册成功".format(user_name)) return True,"注册成功" #用户登录 def user_login(user_name,passwd,logger): # 加载用户信息 data = db_handle.load_info(user_name) if not data: return False,"用户不存在" #获取密码MD5 pwd_md5=common.get_md5(passwd) if data["locked"]: return False,"用户已锁定，请联系管理员解锁" else: if user_name == data["name"] and pwd_md5 == data["passwd"]: logger.info("用户{}登录成功".format(user_name)) return True,"登录成功" else: return False,"用户名或密码错误" #查看余额 def back_balance(user_name): # 加载用户信息 data = db_handle.load_info(user_name) return data["balance"] #查看用户流水 def back_flow(user_name): # 加载用户信息 data = db_handle.load_info(user_name) return data["flow"]
#shop_interface.py
from db import db_handle #结算购物车 def back_car_money(user_name,logger): #加载用户信息 data=db_handle.load_info(user_name) shop_car=data["shop_car"] if not shop_car: return False,"当前购物车为空,先去购物吧" sum_car=0 for i in shop_car.values(): price,num=i sum_car+=(price*num) if data["balance"] >= sum_car: data["balance"] -= sum_car data["buyed"].update(shop_car) data["shop_car"].clear() data["flow"].append("用户{}购物消费{}元".format(user_name,sum_car)) logger.info("用户{}购物消费{}元".format(user_name,sum_car)) db_handle.dump_info(data) return True,"购买成功" else: return False,"余额不足,购买失败" #将商品加入购物车 def add_goods(user_name,good_name,price): # 加载用户信息 data = db_handle.load_info(user_name) if good_name in data["shop_car"]: data["shop_car"][good_name][1]+=1 else: data["shop_car"][good_name]=[price,1] db_handle.dump_info(data) #查看当前购物车 def back_car(user_name): # 加载用户信息 data = db_handle.load_info(user_name) return data["shop_car"]
#bank_interface.py
from db import db_handle #提现 def back_withdraw(user_name,money,logger): # 加载用户信息 data = db_handle.load_info(user_name) #计算提现手续费 total_money=round((money*1.005),2) if data["balance"] >= total_money: data["balance"] -= total_money data["flow"].append("用户{}成功提现{}元,手续费{}元".format(user_name,money,(total_money-money))) logger.info("用户{}成功提现{}元,手续费{}元".format(user_name,money,(total_money-money))) db_handle.dump_info(data) return True,"提现成功" else: return False,"余额不足,提现失败" #还款 def back_pay(user_name,money,logger): # 加载用户信息 data = db_handle.load_info(user_name) data["balance"] += money data["flow"].append("用户{}成功还款{}元".format(user_name,money)) logger.info("用户{}成功还款{}元".format(user_name,money)) db_handle.dump_info(data) #转账 def back_transfer(user_name,transfer_to_user,money,logger): # 加载用户信息 data = db_handle.load_info(user_name) data_to_trade = db_handle.load_info(transfer_to_user) if not data_to_trade: return False,"转入账户不存在，转账失败" if data["balance"] >= money: data["balance"]-=money data_to_trade["balance"]+=money data["flow"].append("用户{}向用户{}转账{}元".format(user_name,transfer_to_user,money)) data_to_trade["flow"].append("用户{}收到用户{}转账{}元".format(transfer_to_user,user_name,money)) logger.info("用户{}向用户{}转账{}元".format(user_name,transfer_to_user,money)) db_handle.dump_info(data) db_handle.dump_info(data_to_trade) return True,"转账成功" else: return False,"余额不足，转账失败"
#admin_interface.py
from lib import common from db import db_handle #查看是否是管理员用户 def back_admin(user_name,passwd,logger): #加载用户信息 data=db_handle.load_info(user_name) if not data: return False, "用户不存在" if not data["admin"]: return False,"此用户没有管理员权限" pwd_md5=common.get_md5(passwd) if user_name == data["name"] and pwd_md5 == data["passwd"]: logger.info("管理员用户{}登录成功".format(user_name)) return True, "登录成功" else: return False, "用户名或密码错误" #冻结账户 def back_locked(user_name,admin,logger): # 加载用户信息 data = db_handle.load_info(user_name) if not data: return False,"冻结账户不存在" data["locked"]=True logger.info("管理员用户{}冻结用户{}".format(admin,user_name)) db_handle.dump_info(data) return True,"冻结成功" #解冻账户 def back_unlocked(user_name,admin,logger): # 加载用户信息 data = db_handle.load_info(user_name) if not data: return False, "解冻账户不存在" data["locked"] = False logger.info("管理员用户{}解冻用户{}".format(admin, user_name)) db_handle.dump_info(data) return True, "解冻成功" #修改用户额度 def back_rechange_balance(user_name,balance,admin,logger): # 加载用户信息 data = db_handle.load_info(user_name) if not data: return False, "账户不存在" data["balance"]=balance logger.info("管理员用户{}修改用户{}额度为{}元".format(admin, user_name,balance)) db_handle.dump_info(data) return True,"修改成功"
#common.py
from conf import settings import logging.config import hashlib from prettytable import PrettyTable #打印功能字典菜单 def print_func(dic): x=PrettyTable(field_names=["功能编号","功能名称"]) for i in dic: x.add_row([i,dic[i][0]]) print(x) #打印商品菜单 def print_good(dic): x=PrettyTable(field_names=["商品编号","商品名称","商品价格"]) for i in dic: x.add_row([i,dic[i][0],dic[i][1]]) print(x) #获取logger def get_logger(name): logging.config.dictConfig(settings.LOGGING_DIC) logger=logging.getLogger(name) return logger #获取密码的md5值 def get_md5(pwd): m1=hashlib.md5() #加盐 salt="I love python" m1.update(pwd.encode("utf-8")) m1.update(salt.encode("utf-8")) return m1.hexdigest() #验证登录装饰器 def auth(func): from core import src def inner(*args,**kwargs): if src.login_user: res=func(*args,**kwargs) return res else: print("请先登录。。。") src.login() res = func(*args, **kwargs) return res return inner
readme.md
- ATM + 购物车 1、注册（5分） 2、登录（5分） 3、查看余额（2分） 4、提现（5分） 5、还款（5分） 6、转账（10分） 7、查看流水（2分） 8、购物功能（15分） 9、查看购物车功能（2分） 10、记录日志（5分） 11、登录认证装饰器（2分） 12、密码加密（2分） 管理员功能 - 冻结账户 - 添加账户 - 修改额度