ATM + 购物车
https://www.cnblogs.com/kermitjam/articles/10687180.html
readme 内容
前戏: 一个项目是如何从无到有的.
一 需求分析
对项目需求进行分析,并提取出相应的功能.
- 额度15000或自定义 ---> 注册功能
- 实现购物商城,买东西加入购物车,调用信用卡接口结账 ---> 购物车, 支付功能
- 可以提现,手续费5% ---> 提现
- 支持多账户登录 ---> 登陆
- 支持账户间转账 ---> 转账
- 记录每月日常消费流水 ---> 记录流水
- 提供还款接口 ---> 还款
- ATM记录操作日志 ---> 日志功能
- 提供管理接口,包括添加账户、用户额度,冻结账户等... ---> 管理员功能
- 用户认证功能 ---> 登陆认证,使用装饰器
二 程序的架构设计
用户功能层:
负责接收用户输入的内容,并返回结果给用户.
做一些小的逻辑判断.
接口层:
处理业务逻辑.
数据处理层:
对文件进行处理.
增,删,查,改操作.
三层架构的好处:
1.代码结构清晰
2.可扩展性强.
3.易维护,管理.
三 分任务开发
- CTO
- 技术总监
- 架构师
- 项目经理
- 普通开发
UI: 用户界面设计师
前端: 网页的开发
后端: 写业务逻辑,写接口
测试: 测试软件
运维: 部署项目.
四 测试
- 手动测试
传统人工去手动测试.
- 自动化测试
通过脚本模拟人的行为,自动化执行测试.
- 黑盒测试:
对用户界面进行测试.
- 白盒测试:
对软件的性能进行测试,例如每分钟能接收多少并发量.
五 上线运行
把测试好的代码交给运维人员, 部署上线.
conf/settings
import os BASE_PATH = os.path.dirname(os.path.dirname(__file__)) DB_PATH = os.path.join(BASE_PATH,\'db\') # 定义三种日志输出格式 开始 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\' # 定义日志输出格式 结束 # ****************注意1: log文件的目录 BASE_PATH = os.path.dirname(os.path.dirname(__file__)) logfile_dir = os.path.join(BASE_PATH, \'log\') # print(logfile_dir) # ****************注意2: log文件名 logfile_name = \'atm_shop_log.log\' # 如果不存在定义的日志目录就创建一个 if not os.path.isdir(logfile_dir): os.mkdir(logfile_dir) # log文件的全路径 logfile_path = os.path.join(logfile_dir, logfile_name) # ****************注意3: log配置字典 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\' }, # 打印到文件的日志,收集info及以上的日志 \'default\': { \'level\': \'DEBUG\', \'class\': \'logging.handlers.RotatingFileHandler\', # 保存到文件 \'formatter\': \'standard\', \'filename\': logfile_path, # 日志文件 \'maxBytes\': 1024*1024*5, # 日志大小 5M \'backupCount\': 5, \'encoding\': \'utf-8\', # 日志文件的编码,再也不用担心中文log乱码了 }, }, \'loggers\': { #logging.getLogger(__name__)拿到的logger配置 \'\': { \'handlers\': [\'default\', \'console\'], # 这里把上面定义的两个handler都加上,即log数据既写入文件又打印到屏幕 \'level\': \'DEBUG\', \'propagate\': True, # 向上(更高level的logger)传递 }, }, }
core/src
from interface import user_interface from lib import common from interface import bank_interface from interface import shop_interface from interface import admin_interface user_online={\'user\':None} def register(): print(\'注册功能\') while True: inp_username=input(\'请输入用户名:\').strip() flag=user_interface.check_user_interface(inp_username) if flag: print(\'用户已存在,请重新输入\') continue inp_pwd=input(\'请输入密码:\').strip() re_pwd=input(\'请再次输入密码:\').strip() if inp_pwd==re_pwd: msg=user_interface.register_interface(inp_username,inp_pwd) if msg: print(msg) break else: print(\'注册失败\') else: print(\'两次密码不一致!\') def login(): while True: inp_username = input(\'请输入用户名\').strip() flag=user_interface.check_user_interface(inp_username) if not flag: print(\'用户不存在\') continue inp_pwd = input(\'请输入密码\').strip() flag,msg = user_interface.login_interface(inp_username,inp_pwd) if flag: print(msg) user_online[\'user\']=inp_username break else: print(msg) #查看余额 @common.login_auth def check_bal(): print(\'查看余额\') bal = user_interface.check_bal_interface(user_online[\'user\']) print(\'余额为: \',bal) #提现 @common.login_auth def withdraw(): while True: money = input(\'请输入提现金额(按q退出):\').strip() if money == \'q\': break if not money.isdigit(): print(\'请输入数字!\') continue money = int(money) flag,msg = bank_interface.withdraw_interface(user_online[\'user\'],money) if flag: print(msg) break else: print(msg) @common.login_auth def repay(): while True: money = input(\'请输入还款金额(按q退出):\').strip() if money == \'q\': break if not money.isdigit(): print(\'请输入数字!\') continue money = int(money) flag, msg = bank_interface.repay_interface(user_online[\'user\'], money) if flag: print(msg) break # 转账 @common.login_auth def transfer(): while True: to_user = input(\'请输入转账目标用户(输入q退出):\').strip() if to_user == \'q\': break if to_user == user_online[\'user\']: print(\'自己不可以给自己转账!\') continue flag = user_interface.check_user_interface(to_user) if not flag: print(\'用户不存在!\') continue money = input(\'请输入转账金额:\').strip() if not money.isdigit(): print(\'请输入数字!\') continue money = int(money) flag,msg=bank_interface.transfer_interface(to_user,user_online[\'user\'],money) if flag: print(msg) break else: print(msg) # 查看流水 @common.login_auth def check_flow(): flow_list = bank_interface.check_flow_interface(user_online[\'user\']) for flow in flow_list: print(flow) # 购物 @common.login_auth def shopping(): good_list = [ [\'Tesla\',500], [\'宝马\',350], [\'奔驰\',400], [\'奥迪\',300], [\'mazda\',200] ] user_bal=user_interface.check_bal_interface(user_online[\'user\']) shop_cart = shop_interface.check_shop_cart_interface(user_online[\'user\']) cost = 0 for i in range(len(good_list)): if shop_cart.get(good_list[i][0]): # 若存在,计算购物车价格 cost += shop_cart[good_list[i][0]] * good_list[i][1] while True: for index,price in enumerate(good_list): print(index,price) choice = input(\'请输入商品编号(输入q退出):\').strip() if choice == \'q\': print(f\'购物车为{shop_cart},余额为{user_bal},总费用为{cost}\') cmd = input(\'是否结账,请输入y/n:\').strip() if cmd ==\'y\': flag,msg = shop_interface.buy_interface(user_online[\'user\'],cost) if flag: print(msg) print(f\'已购买{shop_cart},余额为{user_bal-cost},总费用为{cost}\') break else: print(msg) elif cmd ==\'n\': flag,msg=shop_interface.add_shop_cart_interface(user_online[\'user\'],shop_cart) if flag: print(msg) break elif not choice.isdigit(): print(\'输入必须为数字!\') continue elif choice in [f\'{i}\' for i in range(len(good_list))]: good_name,good_price = good_list[int(choice)] if user_bal > cost + good_price: if good_name in shop_cart: shop_cart[good_name] += 1 else: shop_cart[good_name] = 1 cost += good_price else: print(\'金额不足\') print(f\'购物车为{shop_cart},余额为{user_bal},总费用为{cost}\') else: print(\'输入不在范围内\') continue # 参看购物车 @common.login_auth def check_shop_cart(): shop_cart = shop_interface.check_shop_cart_interface(user_online[\'user\']) if not shop_cart: print(\'购物车是空的,请选择购物车功能\') print(shop_cart) while shop_cart != {}: # 若购物车不为空 cmd = input(\'是否清空购物车,输入y/n\') if cmd == \'y\': flag,msg=shop_interface.clear_shop_cart_interface(user_online[\'user\']) if flag: print(msg) break elif cmd == \'n\': break else: print(\'输入格式不正确\') # 注销 @common.login_auth def logout(): flag,msg = user_interface.logout_interface() if flag: print(msg) #冻结账户 def lock_user(): print(\'冻结用户\') user= input(\'请输入要冻结的用户:\').strip() flag=user_interface.check_user_interface(user) if flag: msg = admin_interface.lock_interface(user) print(msg) else: print(\'目标用户不存在\') #解冻用户 def unlock_user(): print(\'解冻用户\') user = input(\'请输入要解结的用户:\').strip() flag = user_interface.check_user_interface(user) if flag: msg = admin_interface.unlock_interface(user) print(msg) else: print(\'目标用户不存在\') #修改用户额度 def change_limit(): print(\'修改用户额度\') user = input(\'请输入需要修改额度的用户:\').strip() flag = user_interface.check_user_interface(user) if not flag: print(\'用户不存在\') return limit = input(\'请输入要修改的额度\').strip() if limit.isdigit(): limit = int(limit) msg=admin_interface.change_limit_interface(user,limit) print(msg) else: print(\'请输入数字\') admin_func_dic={ \'1\':lock_user, \'2\':unlock_user, \'3\':change_limit } @common.login_auth def admin_sys(): while True: print(\'\'\' 1.冻结用户 2.解冻用户 3.用户额度 \'\'\') choice = input(\'请输入管理员功能(按q退出):\').strip() if choice == \'q\': break elif choice not in admin_func_dic: print(\'请重新输入\') continue else: admin_func_dic[choice]() func_dic = { \'1\':register, \'2\':login, \'3\':check_bal, \'4\':withdraw, \'5\':repay, \'6\':transfer, \'7\':check_flow, \'8\':shopping, \'9\':check_shop_cart, \'10\':logout, \'11\':admin_sys } def run(): while True: print(\'\'\' 1.注册 2.登陆 3.查看余额 4.提现 5.还款 6.转账 7.查看流水 8.购物车功能 9.查看购物车(可清空) 10.注销 11.管理员功能 \'\'\') choice = input(\'请输入功能编号(按q退出):\').strip() if choice ==\'q\': break elif choice in func_dic: func_dic[choice]() else: print(\'请输入正确编号\')
db/db_handler
import json import os from conf import settings def save(user_dic): user_path = f\'{settings.DB_PATH}/{user_dic["user"]}.json\' with open(user_path,\'w\',encoding=\'utf8\')as f: json.dump(user_dic,f,ensure_ascii=False) def select(user): user_path = f\'{settings.DB_PATH}/{user}.json\' if os.path.exists(user_path): with open(user_path,\'r\',encoding=\'utf8\')as f: user_dic=json.load(f) return user_dic
自动生成的ldc.json文件内容
[2019-11-22 19:15:15,294][MainThread:19088][task_id:user][user_interface.py:37][INFO][ldc用户登陆成功!] [2019-11-22 19:16:02,424][MainThread:19088][task_id:bank][bank_interface.py:53][INFO][ldc用户给zxc转账120元] [2019-11-22 19:16:12,150][MainThread:19088][task_id:bank][bank_interface.py:34][INFO][[ldc]用户参看流水] [2019-11-22 19:16:19,305][MainThread:19088][task_id:user][user_interface.py:45][INFO][ldc参看余额!] [2019-11-22 19:16:19,307][MainThread:19088][task_id:shop][shop_interface.py:10][INFO][ldc查看了购物车] [2019-11-22 19:17:02,286][MainThread:19088][task_id:bank][bank_interface.py:67][INFO][ldc用户支付400成功] [2019-11-22 19:17:02,290][MainThread:19088][task_id:shop][shop_interface.py:19][INFO][ldc购买成功] [2019-11-22 19:17:20,777][MainThread:19088][task_id:user][user_interface.py:45][INFO][ldc参看余额!] [2019-11-22 19:17:20,778][MainThread:19088][task_id:shop][shop_interface.py:10][INFO][ldc查看了购物车] [2019-11-22 19:19:32,075][MainThread:19088][task_id:bank][bank_interface.py:67][INFO][ldc用户支付0成功] [2019-11-22 19:19:32,079][MainThread:19088][task_id:shop][shop_interface.py:19][INFO][ldc购买成功] [2019-11-22 19:19:57,101][MainThread:19088][task_id:admin][admin_interface.py:24][INFO][修改用户ldc额度成功] [2019-11-22 23:31:16,539][MainThread:2176][task_id:user][user_interface.py:24][INFO][qwe用户注册成功!] [2019-11-22 23:31:26,546][MainThread:2176][task_id:user][user_interface.py:37][INFO][zxc用户登路成功!] [2019-11-22 23:31:39,420][MainThread:2176][task_id:bank][bank_interface.py:53][INFO][zxc用户给qwe转账100元] [2019-11-22 23:31:53,696][MainThread:2176][task_id:bank][bank_interface.py:34][INFO][[zxc]用户参看流水] [2019-11-22 23:32:39,981][MainThread:2176][task_id:user][user_interface.py:45][INFO][zxc参看余额!] [2019-11-22 23:32:39,985][MainThread:2176][task_id:shop][shop_interface.py:10][INFO][zxc查看了购物车] [2019-11-22 23:33:01,157][MainThread:2176][task_id:bank][bank_interface.py:67][INFO][zxc用户支付1500成功] [2019-11-22 23:33:01,160][MainThread:2176][task_id:shop][shop_interface.py:19][INFO][zxc购买成功] [2019-11-22 23:33:16,300][MainThread:2176][task_id:user][user_interface.py:53][INFO][zxc注销!]
interface/admin_interface
from db import db_handler from lib import common user_logger=common.get_logger(\'admin\') def lock_interface(user): user_dic = db_handler.select(user) user_dic[\'lock\'] = True db_handler.save(user_dic) user_logger.info(f\'{user}用户冻结成功\') return f\'{user}用户冻结成功\' def unlock_interface(user): user_dic = db_handler.select(user) user_dic[\'lock\'] = False db_handler.save(user_dic) user_logger.info(f\'{user}用户解冻成功\') return f\'{user}用户解冻成功\' def change_limit_interface(user,limit): user_dic = db_handler.select(user) user_dic[\'balance\']=limit db_handler.save(user_dic) user_logger.info(f\'修改用户{user}额度成功\') return f\'修改用户{user}额度成功\'
interface/bank_interface
from db import db_handler from lib import common user_loggor = common.get_logger(\'bank\') # 提现接口 def withdraw_interface(user,money): user_dic = db_handler.select(user) if user_dic[\'balance\'] >= money*1.05: user_dic[\'balance\'] -= money*1.05 msg = f\'[{user}]用户提现[{money}元,手续费[{money * 0.05}]元]\' user_dic[\'flow\'].append(msg) db_handler.save(user_dic) user_loggor.info(f\'[{user}]用户提现[{money}元,手续费[{money * 0.05}]元]\') return True,msg else: user_loggor.info(f\'[{user}]用户提现余额不足\') return False,\'余额不足\' # 还款接口 def repay_interface(user, money): user_dic = db_handler.select(user) user_dic[\'balance\'] += money msg = f\'[{user}]用户还款[{money}元]\' user_dic[\'flow\'].append(msg) db_handler.save(user_dic) user_loggor.info(f\'[{user}]用户还款[{money}元]\') return True,msg # 参看流水接口 def check_flow_interface(user): user_dic=db_handler.select(user) user_loggor.info(f\'[{user}]用户参看流水\') return user_dic[\'flow\'] def transfer_interface(to_user,from_user,money): to_user_dic = db_handler.select(to_user) from_user_dic = db_handler.select(from_user) if from_user_dic[\'balance\'] >= money: from_user_dic[\'balance\'] -= money to_user_dic[\'balance\'] += money # 流水信息 from_user_flow = f\'{from_user}用户给{to_user}转账{money}元\' to_user_flow = f\'{to_user}用户收到{from_user}转账{money}元\' # 记录流水 from_user_dic[\'flow\'].append(from_user_flow) to_user_dic[\'flow\'].append(to_user_flow) db_handler.save(from_user_dic) db_handler.save(to_user_dic) user_loggor.info(f\'{from_user}用户给{to_user}转账{money}元\') return True,from_user_flow else: user_loggor.info(f\'{from_user}用户余额不足\') return False,\'余额不足\' # 银行支付接口 def pay_interface(user,cost): user_dic=db_handler.select(user) if user_dic[\'balance\'] >= cost: user_dic[\'balance\'] -= cost user_dic[\'flow\'].append(f\'{user}用户支付{cost}成功\') db_handler.save(user_dic) user_loggor.info(f\'{user}用户支付{cost}成功\') return True else: user_loggor.info(f\'{user}用户支付{cost}失败\') return False
interface/shop_interface
from db import db_handler from interface import bank_interface from lib import common user_logger=common.get_logger(\'shop\') # 查看购物车 def check_shop_cart_interface(user): user_dic = db_handler.select(user) user_logger.info(f\'{user}查看了购物车\') return user_dic[\'shop_cart\'] def buy_interface(user,cost): flag=bank_interface.pay_interface(user,cost) if flag: user_dic = db_handler.select(user) user_dic[\'shop_cart\'] = {} # 清空购物车 db_handler.save(user_dic) user_logger.info(f\'{user}购买成功\') return True,\'购物成功\' else: user_logger.info(f\'{user}余额不足,购物失败\') return False,\'余额不足,购物失败\' def add_shop_cart_interface(user,shop_cart): user_dic = db_handler.select(user) user_dic[\'shop_cart\'] = shop_cart db_handler.save(user_dic) user_logger.info(f\'{user}添加购物车成功\') return True,\'添加购物车成功\' def clear_shop_cart_interface(user): user_dic = db_handler.select(user) user_dic[\'shop_cart\'] = {} db_handler.save(user_dic) user_logger.info(f\'{user}购物车清空成功\') return True,\'购物车清空成功\'
interface/user_interface
from db import db_handler from lib import common user_logger=common.get_logger(\'user\') # 参看用户接口 def check_user_interface(user): user_dic=db_handler.select(user) if user_dic: return True # 注册接口 def register_interface(user,pwd,balance = 15000): pwd = common.get_md5(pwd) user_dic={ \'user\':user, \'pwd\':pwd, \'balance\':balance, \'flow\':[], \'shop_cart\':{}, \'lock\':False } db_handler.save(user_dic) user_logger.info(f\'{user_dic["user"]}用户注册成功!\') return f\'{user}用户注册成功!\' def login_interface(user,pwd): user_dic = db_handler.select(user) pwd = common.get_md5(pwd) if user_dic[\'lock\']: user_logger.info(f\'{user}用户已被锁定,请联系管理员\') return False,\'用户已被锁定,请联系管理员\' if user_dic[\'pwd\'] == pwd: user_logger.info(f\'{user}用户登路成功!\') return True,f\'{user}用户登路成功!\' user_logger.info(f\'{user}输入密码错误!\') return False,f\'{user}密码错误!\' def check_bal_interface(user): user_dic=db_handler.select(user) user_logger.info(f\'{user}参看余额!\') return user_dic["balance"] # 注销 def logout_interface(): from core import src user = src.user_online[\'user\'] src.user_online[\'user\']=None user_logger.info(f\'{user}注销!\') return True,f\'{user}用户注销成功\'
lib/common
import hashlib import logging.config from conf import settings def get_md5(pwd): md5_obj=hashlib.md5() md5_obj.update(pwd.encode(\'utf8\')) sal=\'LDC专用\' md5_obj.update(sal.encode(\'utf8\')) res = md5_obj.hexdigest() return res # 登陆认证功能 def login_auth(func): from core import src def inner(*args,**kwargs): if src.user_online[\'user\']: res = func(*args,**kwargs) return res else: print(\'请先登录\') src.login() return inner def get_logger(type_name): logging.config.dictConfig(settings.LOGGING_DIC) logger = logging.getLogger(type_name) return logger
log文件夹内自动生成的文件
[2019-11-22 19:15:15,294][MainThread:19088][task_id:user][user_interface.py:37][INFO][ldc用户登陆成功!] [2019-11-22 19:16:02,424][MainThread:19088][task_id:bank][bank_interface.py:53][INFO][ldc用户给zxc转账120元] [2019-11-22 19:16:12,150][MainThread:19088][task_id:bank][bank_interface.py:34][INFO][[ldc]用户参看流水] [2019-11-22 19:16:19,305][MainThread:19088][task_id:user][user_interface.py:45][INFO][ldc参看余额!] [2019-11-22 19:16:19,307][MainThread:19088][task_id:shop][shop_interface.py:10][INFO][ldc查看了购物车] [2019-11-22 19:17:02,286][MainThread:19088][task_id:bank][bank_interface.py:67][INFO][ldc用户支付400成功] [2019-11-22 19:17:02,290][MainThread:19088][task_id:shop][shop_interface.py:19][INFO][ldc购买成功] [2019-11-22 19:17:20,777][MainThread:19088][task_id:user][user_interface.py:45][INFO][ldc参看余额!] [2019-11-22 19:17:20,778][MainThread:19088][task_id:shop][shop_interface.py:10][INFO][ldc查看了购物车] [2019-11-22 19:19:32,075][MainThread:19088][task_id:bank][bank_interface.py:67][INFO][ldc用户支付0成功] [2019-11-22 19:19:32,079][MainThread:19088][task_id:shop][shop_interface.py:19][INFO][ldc购买成功] [2019-11-22 19:19:57,101][MainThread:19088][task_id:admin][admin_interface.py:24][INFO][修改用户ldc额度成功] [2019-11-22 23:31:16,539][MainThread:2176][task_id:user][user_interface.py:24][INFO][qwe用户注册成功!] [2019-11-22 23:31:26,546][MainThread:2176][task_id:user][user_interface.py:37][INFO][zxc用户登路成功!] [2019-11-22 23:31:39,420][MainThread:2176][task_id:bank][bank_interface.py:53][INFO][zxc用户给qwe转账100元] [2019-11-22 23:31:53,696][MainThread:2176][task_id:bank][bank_interface.py:34][INFO][[zxc]用户参看流水] [2019-11-22 23:32:39,981][MainThread:2176][task_id:user][user_interface.py:45][INFO][zxc参看余额!] [2019-11-22 23:32:39,985][MainThread:2176][task_id:shop][shop_interface.py:10][INFO][zxc查看了购物车] [2019-11-22 23:33:01,157][MainThread:2176][task_id:bank][bank_interface.py:67][INFO][zxc用户支付1500成功] [2019-11-22 23:33:01,160][MainThread:2176][task_id:shop][shop_interface.py:19][INFO][zxc购买成功] [2019-11-22 23:33:16,300][MainThread:2176][task_id:user][user_interface.py:53][INFO][zxc注销!]
start.py 启动文件
注意:此处添加环境变量可以省略因为start在项目路径下。当start.py在bin目录下,必须加如环境变量,否则无法引用项目其他文件夹中的模块
import os import sys from core import src BASE_PATH = os.path.dirname(__file__) sys.path.append(BASE_PATH) if __name__ == \'__main__\': src.run()