标签：salt fun lock 操作日志 存在 文件 star 购物 架构
代码视频之一：https://www.bilibili.com/video/BV1Cc411h78m/
更多ATM代码视频请浏览：https://space.bilibili.com/516735973
在项目开发中，清晰明了的结构设计非常重要。它的重要性至少提现在三个方面：结构清晰；可维护性强；可扩展性高。
常用的项目结构设计中，三层架构设计非常实用。这种架构设计模式将整个程序分为三层：
用户视图层：用于与用户交互的，可以接受用户的输入，打印接口返回的数据。
逻辑接口层：接收视图层传递过来的参数，根据逻辑判断调用数据层加以处理并返回一个结果给用户视图层。
数据处理层：接受接口层传递过来的参数，做数据的增删改查。
项目需求
1.额度15000或自定义 --> 注册功能
2.实现购物商城，买东西加入购物车，调用信用卡接口结账 --> 购物功能、支付功能
3.可以提现，手续费5% --> 提现功能
4.支持多账户登录 --> 登录功能
5.支持账户间转账 --> 转账功能
6.记录日常消费 --> 记录流水功能
7.提供还款接口 --> 还款功能
8.ATM记录操作日志 --> 记录日志功能
9.提供管理接口，包括添加账户、用户额度，冻结账户等。。。 ---> 管理
用户视图层的功能
"用户视图层" 展示给用户选择的功能
1、注册功能
2、登录功能
3、查看余额
4、提现功能
5、还款功能
6、转账功能
7、查看流水
8、购物功能
9、结算购物车
10、查看购物车
11、管理员功能
管理员视图层的功能
1、注册功能
2、登录功能
3、锁定账户
4、解锁账户
5、更改额度
6、增加商品
一个项目如何从无到有
一
需求分析
1.
拿到项目，会先在客户那里一起讨论需求，
商量项目的功能是否能实现，周期与价格，得到一个需求文档。
2.
最后在公司内部需要开一次会议，最终得到一个开发文档，
交给不同岗位的程序员进行开发。
- Python: 后端，爬虫
- 不同的岗位：
- UI界面设计:
- 设计软件的布局，会分局软件的外观切成一张张图片。
- 前端:
- 拿到UI交给他的图片，然后去搭建网页面。
- 设计一些页面中，哪些位置需要接收数据，需要进行数据交互。
- 后端:
- 直接核心的业务逻辑，调度数据库进行数据的增删查改。
- 测试:
- 会给代码进行全面测试，比如压力测试，界面测试（CF卡箱子）。
- 运维:
- 部署项目。
二
程序的架构设计
1、程序设计的好处
1）思路清晰
2）不会出现写一半代码时推翻重写
3）方便自己或以后的同事更好维护
2、三层架构设计的好处
1）把每个功能都分层三部分，逻辑清晰
2）如果用户更换不同的用户界面或不同，
的数据储存机制都不会影响接口层的核心
逻辑代码，扩展性强。
3）可以在接口层，准确的记录日志与流水。
3、三层架构
一
用户视图层
用于与用户交互的，可以接受用户的输入，打印接口返回的数据。
二
逻辑接口层
接受
用户视图层
传递过来的参数，根据逻辑判断调用数据层加以处理，
并返回一个结果给
用户视图层。
三
数据处理层
接受接口层传递过来的参数，做数据的
- 保存数据
save()
- 查看数据
select()
- 更新数据
- 删除数据
三 分任务开发
四 测试
五 上线
执行文件
from core.user_src import run
if __name__ == ‘__main__‘:
run()
用户视图层
core
user_src.py
# 用户视图层
from db.db_handle import judge
from interface.user_interface import register_interface
from interface.user_interface import log_interface
from interface.bank_interface import withdraw_interface
from interface.bank_interface import transfer_interface
from interface.bank_interface import repay_interface
from interface.bank_interface import pay_interface
from interface.shopping_interface import add_class_interface
from interface.shopping_interface import read_interface
from interface.shopping_interface import add_car_interface
from interface.shopping_interface import shop_interface
from core.admin_src import admin_run
from lib.common import auth
# 记录用户登录状态
log = None
# 注册
def register():
print(‘注册‘.center(50, ‘?‘))
while True:
username = input(‘输入用户名>>‘).strip()
# 判断用户名是否存在
res = judge(username)
if res:
print(‘用户名已存在！‘)
continue
while True:
pwd = input(‘请输入密码‘).strip()
re_pwd = input(‘请确认密码‘).strip()
if pwd != re_pwd:
print(‘两次密码不一致‘)
continue
else:
# 保存注册信息
res = register_interface(username, pwd)
print(res)
break
break
# 登录
def login():
print(‘登录‘.center(50, ‘?‘))
while True:
username = input(‘请输入用户名>>‘).strip()
# 判断用户名是否存在
res = judge(username)
if res is None:
print(‘用户名不存在‘)
continue
count = 0
while count <= 3:
pwd = input(‘请输入密码>>‘).strip()
# 判断密码是否正确
flag, msg = log_interface(username, pwd, count)
if flag == ‘locked‘:
print(msg)
break
if flag == False:
count += 1
print(msg)
continue
else:
global log
log = username
print(msg)
break
break
# 查询余额
@auth
def check_balance():
print(‘查看余额‘.center(50, ‘?‘))
info = judge(log)
print(f‘{log}先生/女士，您的余额为{info["cash"]}元‘)
# 提现
@auth
def withdraw():
print(‘提现‘.center(50, ‘?‘))
while True:
cash = input(‘请输入提现金额‘).strip()
if not cash.isdigit():
print(‘请输入大于0的数字！‘)
# 进行提现操作
flag, msg = withdraw_interface(log, cash)
if flag:
print(msg)
break
else:
print(msg)
continue
# 转账
@auth
def transfer_account():
print(‘转账‘.center(50, ‘?‘))
while True:
payee = input(‘请输入收款人账户>>‘).strip()
payee_info = judge(payee)
if payee_info is None:
print(‘收款人账户不存在！‘)
continue
while True:
cash = input(‘请输入转账金额‘).strip()
if not cash.isdigit():
print(‘请输入大于0的数字！‘)
continue
# 进行转账操作
flag, msg = transfer_interface(log, payee, cash)
if flag:
print(msg)
break
else:
print(msg)
continue
break
# 还款
@auth
def repayment():
print(‘还款‘.center(50, ‘?‘))
while True:
cash = input(‘请输入还款金额>>‘).strip()
if not cash.isdigit():
print(‘请输入大于0的数字！‘)
continue
# 进行还款操作
res = repay_interface(log, cash)
print(res)
break
# 查看流水
@auth
def cash_flow():
print(‘银行流水‘.center(50, ‘?‘))
info = judge(log)
if info[‘flow‘]:
for i in info[‘flow‘]:
print(i)
else:
print(‘当前用户暂无资金流动‘)
# 购物
@auth
def shopping():
print(‘购物‘.center(50, ‘?‘))
# 输出商品类别新信息
res = add_class_interface()
if res:
for num, name in enumerate(res):
print(num, ‘?‘, name)
# 用户选择商品类别
while True:
choice = input(‘请输入想购买的商品类别编号>>‘).strip()
if int(choice) not in range(len(res)):
print(‘暂无此类别商品‘)
continue
# 输出商品类别中的商品
class_name = res[int(choice)]
flag, msg = read_interface(class_name)
if flag:
for num, info in enumerate(msg):
name, price = info
print(f‘[编号]?{num} [商品名称]?{name} [商品价格]?{price}‘)
else:
print(msg)
shopping()
# 用户选择商品
while True:
choice1 = input(‘请输入想购买的商品编号>>‘).strip()
if int(choice1) > len(msg) - 1:
print(‘暂无此商品...‘)
continue
while True:
# 商品名字msg[0] 价格msg[1]
choice3 = input(‘购买[y]or加入购物车[n]‘).strip()
if choice3 == ‘y‘:
name, price = msg[int(choice1)]
flag, msg = pay_interface(log, price)
if flag:
print(msg)
run()
else:
print(msg)
shopping()
elif choice3 == ‘n‘:
name, price = msg[int(choice1)]
car = {}
car[f‘{name}‘] = [price, 1]
res = add_car_interface(log, car)
print(res)
break
else:
print(‘请输入正确命令‘)
continue
break
break
else:
print(‘暂无商品类别...‘)
# shopping()
# 查看购物车
@auth
def show_shopping():
print(‘查看购物车‘.center(50, ‘?‘))
flag, msg = shop_interface(log)
if flag:
for i, v in msg.items():
price, num = v
print(f‘[商品名称]?{i} [商品价格]?{price} [商品数量]?{num}‘)
else:
print(msg)
# 结算购物车
@auth
def settle():
print(‘结算购物车‘.center(50, ‘?‘))
# 查看购物车
flag, msg = shop_interface(log)
if flag:
for i, v in msg.items():
price, num = v
print(f‘[商品名称]?{i} [商品价格]?{price} [商品数量]?{num}‘)
while True:
choice = input(‘购买所有商品[Q]or购买某个商品[F]‘).strip()
if choice == ‘Q‘:
account = 0
for i, v in msg.items():
price, num = v
sum_price = float(price) * int(num)
account += sum_price
flag, msg = pay_interface(log, account)
if flag:
print(msg)
run()
else:
print(msg)
run()
elif choice == ‘F‘:
while True:
choice1 = input(‘请输入购买的商品名称>>‘).strip()
if choice1 not in msg:
print(‘商品不存在‘)
continue
price, num = msg[choice1]
account = int(num) * float(price)
flag, msg = pay_interface(log, account)
if flag:
print(msg)
run()
else:
print(msg)
run()
else:
print(‘请输入正确指令‘)
else:
print(msg)
run()
# 管理员功能
def admin_func():
print(‘管理员功能‘.center(50, ‘?‘))
admin_run()
# 程序运行界面
def run():
dic_func = {
‘0‘: register,
‘1‘: login,
‘2‘: check_balance,
‘3‘: withdraw,
‘4‘: transfer_account,
‘5‘: repayment,
‘6‘: cash_flow,
‘7‘: shopping,
‘8‘: show_shopping,
‘9‘: settle,
‘10‘: admin_func,
‘11‘: exit
}
while True:
print(‘‘‘
?????ATM & ShoppingCar?????
[0]?注册 [1]?登录 [2]?查看余额 [3]?提现 [4]?转账
[5]?还款 [6]?银行流水 [7]?购物 [8]?查看购物车 [9]?结算购物车
[10]?管理员功能 [11]?退出
‘‘‘)
cmd = input(‘输入命令编号>>‘).strip()
if cmd not in dic_func:
print(‘更多的功能正在开发中...‘)
continue
dic_func[cmd]()
admin_src.py
# 管理员视图层
from db.db_handle import admin_judge
from interface.admin_interface import admin_register_interface
from interface.admin_interface import unlocked_interface
from interface.admin_interface import locked_interface
from interface.admin_interface import change_interface
from interface.shopping_interface import add_class_interface
from interface.shopping_interface import increase_interface
from lib.admin_common import admin_login
from lib.common import admin_auth
# 管理员注册
def admin_register():
while True:
name = input(‘请输入用户名>>‘).strip()
res = admin_judge(name)
if res:
print(‘用户名已存在！‘)
continue
while True:
pwd = input(‘请输入密码‘).strip()
re_pwd = input(‘请确认密码‘).strip()
if pwd != re_pwd:
print(‘两次密码不一致‘)
continue
else:
# 保存注册信息
res = admin_register_interface(name, pwd)
print(res)
break
break
# 管理员登录
def admin_log():
admin_login()
# 用户解锁
@admin_auth
def unlocked():
print(‘用户解锁‘.center(50, ‘?‘))
while True:
name = input(‘请输入需要解锁的账户>>‘).strip()
flag,msg = unlocked_interface(name)
if flag:
print(msg)
break
else:
print(msg)
# 用户锁定
@admin_auth
def locked():
print(‘用户锁定‘.center(50, ‘?‘))
while True:
name = input(‘请输入需要锁定的账户>>‘).strip()
flag,msg = locked_interface(name)
if flag:
print(msg)
break
else:
print(msg)
# 更改额度
@admin_auth
def change_limit():
print(‘更改额度‘.center(50, ‘?‘))
while True:
name = input(‘请输入需要更改的账户>>‘).strip()
cash = input(‘请输入新的额度>>‘)
if not cash.isdigit():
print(‘请输入大于0的数字！‘)
continue
flag, msg = change_interface(name,cash)
if flag:
print(msg)
break
else:
print(msg)
# 增加商品类别和商品信息
def add(choice):
while True:
if choice == ‘y‘:
while True:
class_name = input(‘请输入商品类别名>>‘).strip()
product_name = input(‘请输入商品名称>>‘).strip()
product_price = input(‘请输入商品价格‘).strip()
if not class_name.isalpha() or not product_name.isalpha() or not product_price.isdigit():
print(‘输入不符合标准‘)
continue
# 商品存入文件中
msg = increase_interface(class_name, product_name, product_price)
print(msg)
break
elif choice == ‘n‘:
admin_run()
else:
print(‘请输入正确命令Q‘)
continue
break
# 添加商品
@admin_auth
def increase_product():
print(‘增加商品‘.center(50, ‘?‘))
# 输出商品类别信息[] [....]
msg = add_class_interface()
if msg is None:
print(‘暂无商品类别‘)
choice1 = input(‘是否增加新的商品类别[y/n]‘).strip()
add(choice1)
else:
for num,i in enumerate(msg):
print(num,‘?‘,i)
while True:
choice = input(‘请输入添加商品所属商品类别编号‘).strip()
if not choice.isdigit():
print(‘请输入数字编号！‘)
continue
if int(choice) not in range(len(msg)):
while True:
choice1 = input(‘是否增加新的商品类别[y/n]‘).strip()
if choice1 == ‘n‘:
admin_run()
elif choice1 == ‘y‘:
add(choice1)
break
else:
print(‘请输入正确指令！‘)
continue
else:
class_name = msg[int(choice)]
while True:
product_name = input(‘请输入商品名称>>‘).strip()
product_price = input(‘请输入商品价格‘).strip()
if not product_name.isalpha() or not product_price.isdigit():
print(‘输入不符合标准‘)
continue
# 商品存入文件中
msg = increase_interface(class_name, product_name, product_price)
print(msg)
break
break
# 管理员功能界面
def admin_run():
func = {
‘0‘:exit,
‘1‘:admin_register,
‘2‘:admin_log,
‘3‘:unlocked,
‘4‘:locked,
‘5‘:change_limit,
‘6‘:increase_product
}
while True:
print(‘‘‘
??? Administrator Function ???
[0]?退出 [1]?注册 [2]?登录 [3]?用户解锁
[4]?用户锁定 [5]?更改额度 [6]?增加商品
‘‘‘)
cmd = input(‘请输入命令>>‘).strip()
if cmd not in func:
print(‘更多的功能正在开发中...‘)
continue
func[cmd]()
逻辑接口层---interface
user_interface.py
# 逻辑接口层
from lib.common import get_md5
from db.db_handle import save
from db.db_handle import judge
from lib import common
# 生成日志
user_logger = common.get_logger(‘user‘)
# 注册接口
def register_interface(username,pwd,balance=150000):
hash_pwd = get_md5(pwd)
info = {
‘username‘:username,
‘password‘:hash_pwd,
‘cash‘:balance,
‘flow‘:[],
‘shopping_car‘:{},
# True---未被冻结
‘locked‘:True
}
save(info)
res = ‘注册成功‘
user_logger.info(res)
return res
# 登录接口
def log_interface(username,password,count):
info = judge(username)
hash_pwd = get_md5(password)
if info[‘locked‘] == False:
return ‘locked‘,‘账号被锁定，请联系管理员‘
if count == 2:
info[‘locked‘] = False
res = ‘账号被锁定，请联系管理员‘
user_logger.info(res)
save(info)
return ‘locked‘, res
if hash_pwd != info[‘password‘]:
return False,‘密码错误‘
else:
res = ‘登录成功‘
user_logger.info(res)
return True,res
bank_interface.py
# 逻辑接口层
import time
from db.db_handle import judge
from db.db_handle import save
from lib import common
# 生成日志
bank_logger = common.get_logger(‘bank‘)
# 提现接口
def withdraw_interface(username,cash):
info = judge(username)
cash = float(cash)
info[‘cash‘] = float(info[‘cash‘])
# 提现需要5%手续费
if cash*1.05 > info[‘cash‘]:
return False,‘余额不足‘
else:
info[‘cash‘] -= cash*1.05
flow = f‘{time.asctime()} {username}先生/女士提现 {cash}元‘
# 添加银行流水
info[‘flow‘].append(flow)
bank_logger.info(flow)
save(info)
return True,f‘{username}先生/女士提现 {cash}元‘
# 转账接口
def transfer_interface(username,payee,cash):
user_info = judge(username)
payee_info = judge(payee)
cash = float(cash)
user_info[‘cash‘] = float(user_info[‘cash‘])
payee_info[‘cash‘] = float(payee_info[‘cash‘])
if cash > user_info[‘cash‘]:
return False,‘余额不足‘
else:
user_info[‘cash‘] -= cash
flow = f‘{time.asctime()} {username}先生/女士向 {payee}先生/女士转账 {cash}元‘
user_info[‘flow‘].append(flow)
save(user_info)
bank_logger.info(flow)
payee_info[‘cash‘] += cash
flow = f‘{time.asctime()} {payee}先生/女士收到 {username}先生/女士转账 {cash}元‘
payee_info[‘flow‘].append(flow)
save(payee_info)
return True,f‘{payee}先生/女士收到 {username}先生/女士转账 {cash}元‘
# 还款接口
def repay_interface(username,cash):
info = judge(username)
cash = float(cash)
info[‘cash‘] = float(info[‘cash‘])
info[‘cash‘] += cash
flow = f‘{time.asctime()} {username}先生/女士还款 {cash}元‘
info[‘flow‘].append(flow)
bank_logger.info(flow)
save(info)
return f‘{username}先生/女士还款 {cash}元‘
# 支付接口
def pay_interface(username,price):
info = judge(username)
price = float(price)
info[‘cash‘] = float( info[‘cash‘])
if info[‘cash‘] < float(price):
return False,‘余额不足‘
else:
info[‘cash‘] -= price
flow = f‘{time.asctime()}{username}先生/女士成功消费{price}元‘
info[‘flow‘].append(flow)
bank_logger.info(flow)
save(info)
return True,‘购买成功...‘
admin_interface.py
from lib.common import get_md5
from db.db_handle import admin_save
from db.db_handle import admin_judge
from db.db_handle import judge
from db.db_handle import save
from lib import common
# 生成日志
admin_logger = common.get_logger(‘admin‘)
# 管理员注册接口
def admin_register_interface(username,pwd):
hash_pwd = get_md5(pwd)
info = {
‘username‘:username,
‘password‘:hash_pwd,
}
admin_save(info)
res = ‘注册成功‘
admin_logger.info(res)
return res
# 管理员登录接口
def admin_log_interface(username,password):
info = admin_judge(username)
hash_pwd = get_md5(password)
if hash_pwd != info[‘password‘]:
return False,‘密码错误‘
else:
res = ‘登录成功‘
admin_logger.info(res)
return True,res
# 解锁接口
def unlocked_interface(name):
info = judge(name)
if info:
info[‘locked‘] = True
save(info)
res = ‘解锁成功‘
admin_logger.info(res)
return True,res
else:
return False,‘用户不存在‘
# 锁定接口
def locked_interface(name):
info = judge(name)
if info:
info[‘locked‘] = False
save(info)
res = ‘锁定成功‘
admin_logger.info(res)
return True,res
else:
return False,‘用户不存在‘
# 更改额度接口
def change_interface(name,cash):
info = judge(name)
if info:
info[‘cash‘] = cash
res = ‘额度更改成功‘
admin_logger.info(res)
save(info)
return True,res
else:
return False,‘用户不存在‘
shopping_interface.py
import os
from db.db_handle import judge_product
from db.db_handle import judge
from db.db_handle import save
from db.db_handle import save_product
from db.db_handle import read_info
from conf.setting import PRODUCT_PATH
from lib import common
# 生成日志
shop_logger = common.get_logger(‘shopping‘)
# 商品类别接口
def add_class_interface():
class_list = []
name_list = os.listdir(PRODUCT_PATH)
if name_list:
for i in name_list:
# i = [book.json,....]
name,_ = i.split(‘.‘)
class_list.append(name)
return class_list
else:
return None
# 添加商品信息
def increase_interface(class_name,product_name,product_price):
info = judge_product(class_name)
all_products = []
product_info = [product_name,product_price]
if info:
for i in info:
all_products.append(i)
all_products.append(product_info)
else:
all_products.append(product_info)
save_product(class_name,all_products)
res = ‘添加成功‘
shop_logger.info(res)
return res
# 查看商品类别中的具体信息
def read_interface(class_name):
res = read_info(class_name)
if res:
return True,res
else:
return False,‘此类别暂无商品‘
# 加购物车接口
def add_car_interface(username,car):
info = judge(username)
shopping_car = info[‘shopping_car‘]
for i in car:
price,num = car[i]
if i not in shopping_car:
shopping_car [i]= car[i]
else:
shopping_car[i][1] += num
info[‘shopping_car‘] = shopping_car
res = f‘成功加入购物车‘
shop_logger.info(res)
save(info)
return res
# 查看购物车
def shop_interface(username):
info = judge(username)
shopping_car = info[‘shopping_car‘]
if shopping_car:
return True,shopping_car
else:
return False,‘购物车空空如也！‘
数据处理层---db
db_handle.py
# 数据处理层
import os
import json
from conf.setting import USER_INFO_PATH
from conf.setting import ADMIN_PATH
from conf.setting import PRODUCT_PATH
from conf.setting import DB_PATH
# 判断用户是否存在
def judge(username):
# 用户信息文件路径
user_path = os.path.join(USER_INFO_PATH,f‘{username}.json‘)
if os.path.exists(user_path):
with open(user_path,‘r‘,encoding=‘utf_8‘) as f:
return json.load(f)
else:
return None
# 保存用户信息
def save(info):
username = info.get(‘username‘)
user_path = os.path.join(USER_INFO_PATH, f‘{username}.json‘)
with open(user_path, ‘w‘, encoding=‘utf_8‘) as f:
json.dump(info,f,ensure_ascii=False)
# 判断管理员是否存在
def admin_judge(username):
# 用户信息文件路径
user_path = os.path.join(ADMIN_PATH,f‘{username}.json‘)
if os.path.exists(user_path):
with open(user_path,‘r‘,encoding=‘utf_8‘) as f:
return json.load(f)
else:
return None
# 保存管理员信息
def admin_save(info):
username = info.get(‘username‘)
user_path = os.path.join(ADMIN_PATH, f‘{username}.json‘)
with open(user_path, ‘w‘, encoding=‘utf_8‘) as f:
json.dump(info,f,ensure_ascii=False)
# 判断商品类别是否存在
def judge_product(class_name):
product_path = os.path.join(PRODUCT_PATH,f‘{class_name}.json‘)
if os.path.exists(product_path):
with open(product_path,‘r‘,encoding=‘utf-8‘) as f:
try:
product_info = json.load(f)
return product_info
except json.decoder.JSONDecodeError:
return None
else:
return None
# 保存商品信息
def save_product(class_name,all_products ):
product_path = os.path.join(PRODUCT_PATH, f‘{class_name}.json‘)
with open(product_path, ‘w‘, encoding=‘utf-8‘) as f:
json.dump(all_products ,f,ensure_ascii=False)
# 查看商品信息
def read_info(class_name):
path = os.path.join(PRODUCT_PATH,f‘{class_name}.json‘)
with open(path,‘r‘,encoding=‘utf-8‘) as f:
try:
info = json.load(f)
return info
except json.decoder.JSONDecodeError:
return None
user_data文件夹
admin_data文件夹
product_data文件夹
配置信息---conf
setting.py
# 程序配置信息
import os
# 程序根文件夹路径
BASE_DIR = os.path.dirname(os.path.dirname(__file__))
# db文件夹路径
DB_PATH = os.path.join(BASE_DIR,‘db‘)
# 用户信息文件夹
USER_INFO_PATH = os.path.join(DB_PATH,‘user_data‘)
# 管理员信息文件夹
ADMIN_PATH = os.path.join(DB_PATH,‘admin_data‘)
# 商品信息文件夹
PRODUCT_PATH = os.path.join(DB_PATH,‘product_data‘)
# 日志文件
LOG_PATH = os.path.join(BASE_DIR,‘log‘)
LOG_UER_PATH = os.path.join(LOG_PATH,‘access.log‘)
# 2、强调：其中的%(name)s为getlogger时指定的名字
standard_format = ‘%(asctime)s - %(threadName)s:%(thread)d - 日志名字:%(name)s - %(filename)s:%(lineno)d -‘ ‘%(levelname)s - %(message)s‘
simple_format = ‘[%(levelname)s][%(asctime)s][%(filename)s:%(lineno)d]%(message)s‘
test_format = ‘%(asctime)s] %(message)s‘
# 3、日志配置字典
LOGGING_DIC = {
‘version‘: 1,
‘disable_existing_loggers‘: False,
‘formatters‘: {
‘standard‘: {
‘format‘: standard_format
},
‘simple‘: {
‘format‘: simple_format
},
‘test‘: {
‘format‘: test_format
},
},
‘filters‘: {},
# handlers是日志的接收者，不同的handler会将日志输出到不同的位置
‘handlers‘: {
#打印到终端的日志
‘console‘: {
‘level‘: ‘DEBUG‘,
‘class‘: ‘logging.StreamHandler‘, # 打印到屏幕
‘formatter‘: ‘simple‘
},
‘default‘: {
‘level‘: ‘DEBUG‘,
‘class‘: ‘logging.handlers.RotatingFileHandler‘, # 保存到文件
# ‘maxBytes‘: 1024*1024*5, # 日志大小 5M
# ‘maxBytes‘: 1000,
# ‘backupCount‘: 5,
‘filename‘: LOG_UER_PATH, # os.path.join(os.path.dirname(os.path.dirname(__file__)),‘log‘,‘a2.log‘)
‘encoding‘: ‘utf-8‘,
‘formatter‘: ‘standard‘,
},
#打印到文件的日志,收集info及以上的日志
‘other‘: {
‘level‘: ‘DEBUG‘,
‘class‘: ‘logging.FileHandler‘, # 保存到文件
‘filename‘: LOG_UER_PATH, # os.path.join(os.path.dirname(os.path.dirname(__file__)),‘log‘,‘a2.log‘)
‘encoding‘: ‘utf-8‘,
‘formatter‘: ‘test‘,
},
},
# loggers是日志的产生者，产生的日志会传递给handler然后控制输出
‘loggers‘: {
#logging.getLogger(__name__)拿到的logger配置
‘kkk‘: {
‘handlers‘: [‘console‘,‘other‘], # 这里把上面定义的两个handler都加上，即log数据既写入文件又打印到屏幕
‘level‘: ‘DEBUG‘, # loggers(第一层日志级别关限制)--->handlers(第二层日志级别关卡限制)
‘propagate‘: False, # 默认为True，向上（更高level的logger）传递，通常设置为False即可，否则会一份日志向上层层传递
},
‘终端提示‘: {
‘handlers‘: [‘console‘,], # 这里把上面定义的两个handler都加上，即log数据既写入文件又打印到屏幕
‘level‘: ‘DEBUG‘, # loggers(第一层日志级别关限制)--->handlers(第二层日志级别关卡限制)
‘propagate‘: False, # 默认为True，向上（更高level的logger）传递，通常设置为False即可，否则会一份日志向上层层传递
},
‘‘: {
‘handlers‘: [‘default‘, ], # 这里把上面定义的两个handler都加上，即log数据既写入文件又打印到屏幕
‘level‘: ‘DEBUG‘, # loggers(第一层日志级别关限制)--->handlers(第二层日志级别关卡限制)
‘propagate‘: False, # 默认为True，向上（更高level的logger）传递，通常设置为False即可，否则会一份日志向上层层传递
},
},
}
公共方法---lib
common.py
# 公共方法
import hashlib
import logging
import logging.config
from conf import setting
# 密码加密
def get_md5(pwd):
m = hashlib.md5()
m.update(pwd.encode(‘utf-8‘))
salt = ‘采蘑菇的小姑娘‘
m.update(salt.encode(‘utf-8‘))
return m.hexdigest()
# 用户认证装饰器
def auth(func):
from core import user_src
def wrapper(*args,**kwargs):
if user_src.log:
res = func(*args,**kwargs)
return res
else:
print(‘请登录！‘)
user_src.login()
return wrapper
# 管理员认证登录装饰器
def admin_auth(func):
from lib import admin_common
def wrapper(*args,**kwargs):
if admin_common.admin_log:
res = func(*args,**kwargs)
return res
else:
admin_common.admin_login()
return wrapper
# 日志字典的配置
def get_logger(log_type): # log_type ---> user
‘‘‘
:param log_type: 比如是 user日志，bank日志，购物商城日志
:return:
‘‘‘
# 1、加载日志配置信息
logging.config.dictConfig(
setting.LOGGING_DIC
)
# 2、获取日志对象
logger = logging.getLogger(log_type)
return logger
admin_common.py(防止循环导入问题)
from db.db_handle import admin_judge
from interface.admin_interface import admin_log_interface
# 记录登陆状态
admin_log = None
# 管理员登录
def admin_login():
print(‘登录‘.center(50, ‘?‘))
while True:
username = input(‘请输入用户名>>‘).strip()
# 判断用户名是否存在
res = admin_judge(username)
if res is None:
print(‘用户名不存在‘)
continue
while True:
pwd = input(‘请输入密码>>‘).strip()
# 判断密码是否正确
flag, msg = admin_log_interface(username, pwd)
if flag == False:
print(msg)
continue
else:
global admin_log
admin_log = username
print(msg)
break
break
log---存放操作日志
read_me----存放说明性信息
标签：salt fun lock 操作日志 存在 文件 star 购物 架构
原文地址：https://www.cnblogs.com/Kathrine/p/12650293.html
