- ATM项目开发流程
- ATM+购物车逻辑分享
- 项目源代码展示
1.需求分析
ATM一系列小功能：
注册，登录，提现，转账，充值，查看流水，购物车，管理员
2.ATM架构设计
第一层
与用户直接打交道 不参与核心逻辑的编写(数据展示与获取)
cmd终端
浏览器(前端)
第二层
项目核心逻辑
py文件
后端框架(后端)
第三层
数据库
py文件 json文件
数据库(MySQL)
3.ATM目录搭建
bin文件夹——用于存储程序的启动文件
start.py
conf文件夹——用于存储程序的配置文件
settings.py
core文件夹——用于存储程序的主体逻辑
src.py
interface文件夹——用于存储程序的接口文件
user_interfface.py
shop_interface.py
back_interface.py
admin_interface.py
lib文件夹——用于存储程序的公共功能
common.py
log文件夹——用于存储程序的日志文件
log.log
db文件夹——用于存储程序的数据文件
db_handler.py
readme文件（txt文件——用于编写程序的说明
requirements.txt ——用于存储程序的第三方模块名称和 版本
ATM+购物车逻辑分享4.ATM功能搭建
空函数、功能字典、循环
1.ATM注册功能
面条版
- 获取用户数据(用户名 密码 确认密码)
- 比对两次密码是否一致
- 构建用户字典数据
- 利用json序列化到文件中
架构版
第一层：获取用户数据
第二层：校验用户是否存在；构造用户字典；完成用户注册
第三层：保存用户数据的方法（序列化功能>>>:将字典写入文本文件中）
2.ATM登录功能
第一层：获取用户数据
第二层：校验用户是否存在；获取用户字典数据；比对密码是否一致
第三层：提取用户数据的方法(反序列功能>>>:将json文件里面的数据转成字典)
3.代码封装与优化
settings.py
4.ATM密码加密功能
common.py
5.校验用户登陆装饰器
- 装饰器模板
- 全局变量
6.查看余额功能
查看的应该是当前登录用户的账户余额
也就意味着查看余额的功能需要用户登录 并且登录之后需要保存用户状态
1.如何保存用户状态
全局变量存储用户登录信息
2.如何校验用户登录
装饰器
如何区分用户是否登录成功
我们可以在第二层函数的返回值上添加一个布尔值 用于标识是否成功
为了保证统一和后期编写方便 我们将第二层所有的函数返回值全部变成两个
第一个是布尔值 第二个是具体的数据值
针对全局变量的存储位置
第一层更合适
7.提现功能
- 在第一层获取提现金额
- 调用第二层提现接口，传入当前登录用户名和金额
- 第二层获取当前用户账户余额，判断是否充足
- 如果充足的话则完成提现期间可以加上提现手续费
- 完成账户余额扣款 记录流水 并保存数据 添加日志
8.转账功能
- 在第一层获取用户想要转账的目标账户名和金额数
- 调用第二层转账接口 传入当前登录用户、转账用户、转账金额
- 获取登录用户数据、转账用户数据
- 判断转账金额是否小于登录用户的账户余额
- 账户余额大于转账金额则扣去相应的转账金额、目标账户余额增加
- 双方账户都记录流水 保存用户数据 并添加日志
9.查看账单
- 直接查询当前登录用户数据并返回账单信息即可
- 我们针对用户的提现、转账、消费 都象征性的在用户字典的flow键对于的值列表中记录一条信息即可.
10.日志功能
- 配置文件拷贝日志字典
- 公用文件编写日志函数
- 接口文件导入日志函数
11.购物车功能
直接CV即可 然后核对代码需要修改的地方(数据的获取和保存)
项目源代码展示 bin文件夹下start.py文件12.管理员功能
1.只有管理员才可以触发
2.冻结账户
3.移除账户
4.查看指定账户的余额
5.充值执行账户的余额
6.查看指定账户的账单
7.查看指定账户的购物车
ps:4567也可以直接合并 返回指定用户的字典数据
import os
import sys
# 将项目根目录添加到sys.path中
base_dir = os.path.dirname(os.path.dirname(__file__))
sys.path.append(base_dir)
# 导入run方法并启动项目
if __name__ == '__main__':
from core import src
src.run()
conf文件夹下settings.py文件
import os
BASE_DIR = os.path.dirname(os.path.dirname(__file__))
DB_PATH = os.path.join(BASE_DIR, 'db')
if not os.path.exists(DB_PATH):
os.mkdir(DB_PATH)
# 定义系统的提现利率
MONEY_RATE = 0.01
# 我偷懒 不改大写了
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'
# 自定义文件路径
LOG_DIR = os.path.join(BASE_DIR, 'log')
if not os.path.isdir(LOG_DIR):
os.mkdir(LOG_DIR)
LOGFILE_PATH = os.path.join(LOG_DIR, 'ATM.log')
# 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）传递
}, # 当键不存在的情况下 (key设为空字符串)默认都会使用该k:v配置
# '购物车记录': {
# 'handlers': ['default','console'], # 这里把上面定义的两个handler都加上，即log数据既写入文件又打印到屏幕
# 'level': 'WARNING',
# 'propagate': True, # 向上（更高level的logger）传递
# }, # 当键不存在的情况下 (key设为空字符串)默认都会使用该k:v配置
},
}
core文件夹下src.py文件
interface import user_interface, bank_interface, shop_interface, admin_interface
from lib import common
# 定义一个全局字典 记录用户登录状态
is_login = {
'username': '',
'is_admin': False
}
def register():
while True:
# 获取用户名,并且确认密码
username = input('username>>>:').strip()
password = input('password>>>:').strip()
real_pwd = input('real_pwd>>>:').strip()
# 判断用户输入两次密码是否正确
if password != real_pwd:
print('两次输入密码不匹配')
return
# 调用注册接口 完成用户注册
flag, msg = user_interface.register_interface(username, password)
print(msg)
def login():
username = input('username>>>:').strip()
password = input('password>>>:').strip()
# 调用逻辑层登录的核心逻辑
flag, msg = user_interface.login_interface(username, password)
# 用户登录成功之后修改全局字典
if flag:
is_login['username'] = username
is_login['is_admin'] = msg[1]
print(msg)
@common.outer('normal')
def check_account():
# 直接调用银行业务接口
flag, msg = bank_interface.check_account_interface(is_login.get('username'))
print(msg)
@common.outer('normal')
def withdraw():
money = input('请输入要提现的金额>>>:').strip()
if not money.isdigit():
print('金额只能是数字')
return
money = float(money)
flag, msg = bank_interface.withdraw_interface(is_login.get('username'), money)
print(msg)
@common.outer('normal')
def repay():
money = input('请输入要充值的金额>>>:').strip()
if not money.isdigit():
print('金额只能是数字')
return
money = float(money)
flag, msg = bank_interface.repay_interface(is_login.get('username'), money)
print(msg)
@common.outer('normal')
def transfer():
target_user = input('请输入你想要转给的用户名>>>:').strip()
money = input('请输入要转账的金额>>>:').strip()
if not money.isdigit():
print('金额只能是数字')
return
money = float(money)
# 调用银行转账接口
flag, msg = bank_interface.transfer_interface(is_login.get('username'), target_user, money)
print(msg)
@common.outer('normal')
def check_order():
# 直接调用银行查看账单接口
flag, msg = bank_interface.check_order_interface(is_login.get('username'))
if not flag:
print(msg)
else:
for i in msg:
print(f"""
+++++++++++++账单信息+++++++++++++
详细信息:{i}
+++++++++++++++++++++++++++++++++
""")
@common.outer('normal')
def add_shop_car():
# 直接调用第二层添加购物车功能
flag, msg = shop_interface.add_shop_car_interface(is_login.get('username'))
print(msg)
@common.outer('normal')
def check_shop_car():
# 直接调用第二层查看购物车功能
flag, msg = shop_interface.check_shop_car_interface(is_login.get('username'))
if flag:
for item in msg.items(): # ('挂壁面', [100, 3])
print(f"商品名称:{item[0]} | 商品数量:{item[1][0]} | 商品单价:{item[1][1]}")
else:
print(msg)
@common.outer('normal')
def pay_shop_car():
flag, msg = shop_interface.pay_shop_car_interface(is_login.get('username'))
print(msg)
@common.outer('admin')
def admin():
print("""
1.冻结账户
2.移除账户
3.查看指定账户的余额
4.充值执行账户的余额
5.查看指定账户的账单
6.查看指定账户的购物车
""")
choice = input('尊敬的管理员大人 请选择您的操作(q)>>>:').strip()
if choice == '1':
freeze_user = input('请输入您想要冻结的账户名>>>:').strip()
flag, msg = admin_interface.freeze_user_interface(is_login.get('username'), freeze_user)
print(msg)
elif choice == '2':
move_user = input('请输入您想要移除的账户名>>>:').strip()
flag, msg = admin_interface.move_user_interface(is_login.get('username'), move_user)
print(msg)
elif choice == '3':
pass
elif choice == '4':
pass
elif choice == '5':
pass
elif choice == '6':
pass
else:
print('您真调皮 没有该功能编号')
@common.outer('normal')
def reise_shop_car():
while True:
check_shop_car()
choice = input("""
请选择想要执行的操作
1.移除商品
2.修改数量
3.退出修改
""").strip()
if choice == '1':
print('小提示：商品编号从零开始依次往下')
user_dict = shop_interface.reise_shop_interface()
res = user_dict.get('shop_car')
shop_interface.reise1_shop_interface(res, user_dict)
elif choice == '2':
user_dict = shop_interface.reise_shop_interface()
res = user_dict.get('shop_car')
shop_interface.reise2_shop_interface(res, user_dict)
elif choice == '3':
break
else:
print('没有此功能编号')
func_dic = {
'1': register,
'2': login,
'3': check_account,
'4': withdraw,
'5': repay,
'6': transfer,
'7': check_order,
'8': add_shop_car,
'9': check_shop_car,
'10': reise_shop_car,
'11': pay_shop_car,
'12': admin
}
def run():
while True:
print("""
1.用户注册
2.用户登录
3.查看余额
4.账户提现
5.充值功能
6.转账功能
7.查看账单
8.添加购物车功能
9.查看购物车功能
10.修改购物车功能
11.结算购物车功能
12.管理员功能
""")
choice = input('请输入您想获得的编号>>>:').strip()
if choice in func_dic:
func_name = func_dic.get(choice)
func_name()
else:
print('没有该功能编号')
db文件夹下da_handler.py文件
import os
import json
from conf import settings
from lib import common
# 和数据相关
def save(user_dict):
username = user_dict.get('username')
# 获取当前路径
file_path = os.path.join(settings.DB_PATH, '%s.json' % username)
# 写入文件
with open(file_path, 'w', encoding='utf8')as f:
json.dump(user_dict, f, ensure_ascii=False)
print(f'用户{username}注册成功')
def select(username):
import os
file_path = os.path.join(settings.DB_PATH, '%s.json' % username)
if os.path.exists(file_path):
with open(file_path, 'r', encoding='utf8')as f:
return json.load(f)
def redir(username):
import os
file_path = os.path.join(settings.DB_PATH, '%s.json' % username)
os.remove(file_path)
interace文件夹下user_interface.py文件
from db import db_handlen
from lib import common
logger = common.get_logger('用户相关业务')
from core import src
def register_interface(username, password):
# 对用户输入密码进行加密处理
new_pwd = common.get_hash(password)
# 判断用户名是否存在
res = db_handlen.select(username)
if res:
return '用户名已存在'
# 创建用户临时数据
user_dict = {'username': username,
'password': new_pwd,
'balance': 15000,
'shop_car': {},
'flow': [], # 个人流水
'is_lock': False, # 判断是否被冻结
'is_admin': False # 判断是否是管理员
}
# 调入第三层封装好的数据即可
db_handlen.save(user_dict)
logger.info(f'{username}注册成功')
return True,f'{username}注册成功'
def login_interface(username, password):
# 获取当前用户名对应的用户数据（可能存在也可能不存在）
user_dict = db_handlen.select(username)
if not user_dict:
return False,'用户名不存在'
# 判断当前用户是否已经被冻结
if user_dict.get("is_lock"):
return False,'不好意思，当前账户已经被冻结'
# 将明文密码加密成密文进行比对
new_pwd = common.get_hash(password)
# 2.校验密码时是否正确
if new_pwd == user_dict.get('password'):
return False,'用户密码错误'
logger.info(f'{username}登录成功')
return True,('登录成功', user_dict.get('is_admin'))
interface文件夹下bank_interface.py文件
from db import db_handlen
from lib import common
from conf import settings
logger = common.get_logger('银行相关业务')
def check_account_interface(username):
# 根据用户获取用户数据
user_dict = db_handlen.select(username)
if user_dict:
# 生成查看余额日志
logger.info(f'{username}查看了账户余额')
# 正确则返回查看余额，错误则看不到用户数据
return True, '当前%s的账户余额:%s' % (username, user_dict.get('balance'))
return False, '查不到用户数据'
def withdraw_interface(username,money):
user_dict = db_handlen.select(username)
if not user_dict:
return False,'查不到当前用户'
# 2.获取当前用户的余额
current_balance = user_dict.get('balance')
# 3.判断账户余额是否大于或等于提现金额(注意 还有个手续费 1%)
if current_balance < money * (1 + settings.MONEY_RATE):
return False, '账户余额不足'
# 4.执行提现功能
user_dict['balance'] -= money * (1 + settings.MONEY_RATE)
# 5.添加账户流水信息
user_dict['flow'].append(
f'提现金额:{money} 手续费:{money * settings.MONEY_RATE} 账户余额:{user_dict.get("balance")}')
db_handlen.save(user_dict)
logger.info(f'当前用户:{username} 提现金额:{money} 手续费:{money * settings.MONEY_RATE}')
return True, (f"""
{'=' * 25}
用户{username}已获得查看余额资格 n
提现金额{money} n
手续费{money * (1 + settings.MONEY_RATE)}
账户内剩余{user_dict['balance']} n
{'=' * 25}
""")
def repay_interface(username,money):
user_dict = db_handlen.select(username)
if not user_dict:
return False, '查不到当前用户'
user_dict['balance'] += money
user_dict['flow'].append(f"{user_dict.get('username')}充值了{money}")
# user_dict['flow'].append(f'充值了:{money} 账户余额:{user_dict.get("balance")}')
db_handlen.save(user_dict)
logger.info(f'当前用户{username} 充值了:{money}')
return True,(f"""
{'=' * 25}
用户{username}已获得充值资格 n
账户内剩余{user_dict['balance']} n
充值金额为{money} n
{'=' * 25}
""")
def transfer_interface(current_user,target_user,money ):
target_user_dict = db_handlen.select(target_user)
if not target_user_dict:
return False, '查不到想要转账的用户'
# 2.获取登录用户数据
current_user_dict = db_handlen.select(current_user)
if not current_user_dict:
return False, '查不到当前登录的用户'
# 3.判断当前登录用户的账户余额是否满足转账的条件
current_user_balance = current_user_dict.get('balance')
if current_user_balance < money:
return False, '当前账户余额不足'
# 4.完成增减钱操作
current_user_dict['balance'] -= money
current_user_dict['flow'].append(f'给目标账户:{target_user}转了:{money}块钱')
target_user_dict['balance'] += money
target_user_dict['flow'].append(f'接收到来自:{current_user}转账的{money}块钱')
db_handlen.save(current_user_dict)
db_handlen.save(target_user_dict)
logger.info(f'当前账户:{current_user}给目标账户:{target_user}转了:{money}块钱')
return True,(f"""
{'=' * 25}
用户{current_user}已获得转账资格 n
给目标账户:{target_user}转账
转账金额为{money} n
现有余额{current_user_dict.get("balance")}
{'=' * 25}
""")
def check_order_interface(username):
user_dict = db_handlen.select(username)
if not user_dict:
return False, '查不到当前登录的用户'
flow_list = user_dict.get('flow')
logger.info(f'{username}查看了自己的账单信息')
return True, flow_list
interface文件夹下shop_interface.py文件
from lib import common
from db import db_handlen
logger = common.get_logger('购物车业务')
def add_shop_car_interface(username):
# 1.获取商品数据(直接编写)
good_list = [
['挂壁面', 3],
['印度飞饼', 22],
['极品木瓜', 666],
['土耳其土豆', 999],
['伊拉克拌面', 1000],
['董卓戏张飞公仔', 2000],
['仿真玩偶', 10000]
]
# 定义一个临时存储用户想要购买商品的字典数据 {'印度飞饼':[10, 22], '仿真玩偶':[1, 10000]}
temp_shop_dict = {}
# 2.循环打印商品信息购用户选择
while True:
# 3.打印商品数据
for i, j in enumerate(good_list): # i 数值 j 列表
print(f"商品编号:{i} | 商品名称:{j[0]} | 商品单价:{j[1]}")
# 4.获取用户想要添加的商品编号
target_good_id = input('请输入想要购买的商品编号(q)>>>:').strip()
# 添加一个结束添加购物车的标识 一旦用户输入q则将用户临时购物车字典添加到用户json数据中
if target_good_id == 'q':
user_dict = db_handlen.select(username)
# user_dict['shop_car'] = temp_shop_dict # "shop_car": {'印度飞饼':[10, 22],'土耳其土豆':[20, 999]}
"""
如果直接替换 那么可能会吧之前的购物车数据顶替 不够全面
"""
# 先获取用户原有的购物车数据
old_shop_car = user_dict.get('shop_car') # {'印度飞饼':[10, 22],'土耳其土豆':[20, 999]}
# 当前用户购物车数据 {'挂壁面': [300, 3], '土耳其土豆': [100, 999], '董卓戏张飞公仔': [200, 2000]}
for good_name in temp_shop_dict:
if good_name in old_shop_car:
old_shop_car.get('good_name')[0] += temp_shop_dict.get('good_name')[0]
else:
old_shop_car['good_name'] = temp_shop_dict.get('good_name')
# 写入数据
db_handlen.save(user_dict)
logger.info(f'{username}添加了购物车商品')
return True, f'添加购物车成功'
# 5.小的逻辑校验(是否是纯数字 是否在范围内)
if not target_good_id.isdigit():
print('商品编号只能是数字')
continue
target_good_id = int(target_good_id)
if target_good_id not in range(len(good_list)):
print('商品编号超出了范围')
continue
# 6.根据用户输入的商品编号获取商品信息
target_good_info = good_list[target_good_id] # ['印度飞饼', 22]
target_good_name = target_good_info[0]
target_good_price = target_good_info[1]
# 7.获取用户想要购买的数量
target_good_num = input('请输入想要购买的商品数量>>>:').strip()
if not target_good_num.isdigit():
print('商品数量只能是数字')
continue
target_good_num = int(target_good_num)
# 8.写入临时购物车字典
if target_good_name in temp_shop_dict:
# 获取值列表 然后自增
value_list = temp_shop_dict.get(target_good_name) # [5, 22]
value_list[0] += target_good_num # [5+5, 22]
temp_shop_dict[target_good_name] = value_list #
else:
temp_shop_dict[target_good_name] = [target_good_num, target_good_price]
def check_shop_car_interface(username):
user_dict = db_handlen.select(username)
shop_car_info = user_dict.get('shop_car') # {'飞饼':[11, 22]}
if not shop_car_info:
return False, '购物车空空如也'
logger.info(f'{username}查看了自己的购物车')
return True, shop_car_info
def pay_shop_car_interface(username):
user_dict = db_handlen.select(username)
# print(user_dict) # {'name': 'jason', 'pwd': '123', 'balance': 15000, 'shop_car': {'挂壁面': [100, 3], '土耳其土豆': [50, 999], '印度飞饼': [33, 22]}}
# 3.获取购物车商品数据
shop_car = user_dict.get('shop_car') # {'挂壁面': [100, 3], '土耳其土豆': [50, 999], '印度飞饼': [33, 22]}
if not shop_car:
print('你先去买东西 行不行')
return
# 4.计算商品的总价
total_money = 0
current_balance = user_dict.get('balance')
for values in shop_car.values(): # [100, 3] [50, 999]
values[0] = int(values[0])
total_money += values[0] * values[1]
# 5.判断余额是否充足
if current_balance >= total_money:
rest_money = current_balance - total_money
user_dict['balance'] = rest_money
user_dict['shop_car'] = {}
user_dict['flow'].append(f'{username}清空了自己的购物车 花费了{total_money}')
logger.info(f'{username}清空了自己的购物车 花费了{total_money}')
db_handlen.save(user_dict)
return True, f'今日消费:{total_money} 卡上余额:{rest_money}'
else:
return False, '你个穷鬼 买不起滚蛋!!!'
interface文件夹下admin_interface.py文件
from lib import common
from db import db_handlen
logger = common.get_logger('管理员相关业务')
def freeze_user_interface(admin_name,freeze_user):
freeze_user_dict = db_handlen.select(freeze_user)
if not freeze_user_dict:
return False, '非常抱歉 您想要冻结的账户不存在'
freeze_user_dict['is_lock'] = True
db_handlen.save(freeze_user_dict)
logger.error(f'账户{freeze_user}被管理员{admin_name}冻结')
return True, f'账户{freeze_user}冻结成功'
def move_user_interface(admin_name,move_user):
move_user_dict = db_handlen.select(move_user)
if not move_user_dict:
return False, '非常抱歉 您想要移除的账户不存在'
db_handlen.redir(move_user)
logger.error(f'账户{move_user}被管理员{admin_name}移除')
return True, f'账户{move_user}移除成功'
lib文件夹下common.py文件
from core import src
def get_hash(password):
import hashlib
md5 = hashlib.md5()
md5.update(password.encode('utf8'))
new_pwd = md5.hexdigest()
return new_pwd
import logging
import logging.config
from conf import settings
def get_logger(msg):
logging.config.dictConfig(settings.LOGGING_DIC) # 自动加载字典中的配置
logger1 = logging.getLogger(msg)
return logger1
def outer(func_type):
def login_auth(func_name):
def inner(*args, **kwargs):
# 执行被装饰函数之前可以做的额外操作
if src.is_login.get('username'):
if func_type == 'normal':
res = func_name(*args, **kwargs)
# 执行被装饰函数之后可以做的额外操作
return res
else:
if src.is_login.get('is_admin'):
res = func_name(*args, **kwargs)
# 执行被装饰函数之后可以做的额外操作
return res
else:
print('您当前不是管理员 无法执行')
else:
print('您暂未登录无法执行当前功能 请先登录')
src.login()
return inner
return login_auth
