大致的层级结构
全部代码
bin目录下的start.py
我将这部分作为程序的启动入口，代码如下
from core.src import run
if __name__ == '__main__':
run()
conf下的seetings.py
设置相关
import sys
from pathlib import Path
path = Path(__file__)
PATH_DIR = path.parent.parent
sys.path.insert(0, str(PATH_DIR)) # 根目录
DB_DIR = PATH_DIR / 'DB'
PATH_DICT = {} # 存放全部路径数据的字典
db_path_list = ['user_data', "bank_data", "user_flow", "user_log"]
for p in db_path_list:
item = DB_DIR / p
if not item.exists():
Path(item).mkdir()
PATH_DICT.setdefault(p, item)
core下的src.py
核心文件，除了功能菜单无多余的代码
active函数只有一个功能，那就是程序一开始启动就将数据读取出来，就算读取不到也不会报错
from core.bank import *
from core.users import *
from lib.common import status
from DB.db_hander import active
func_dict = {
"1": ("注册功能", register),
"2": ("登录功能", login),
"3": ("激活银行卡", active_bankcard),
"4": ("取款", withdraw_money),
"5": ("转账", transfer_account),
"6": ("充值金额", recharge),
"7": ("查看流水", show_flow),
"8": ("个人信息", show_self_data),
"9": ("切换用户", logout),
"0": ("退出系统", None)
}
def run():
while True:
active()
print("功能菜单".center(40, "="))
for index, function in func_dict.items():
print(f"{index}.{function[0]:10}", end="\t")
if int(index) % 3 == 0:
print("\n", end="")
else:
print("欢迎使用".center(40, "="))
status()
choice = input("输入序号进入功能：")
if choice == "0":
print("已退出，期待再次使用！")
break
if choice not in func_dict:
print("选择错误，请重试。")
continue
func_dict[choice][1]()
core下的users.py
登录注册功能
import json
from textwrap import dedent
from DB.db_hander import *
from conf.settings import PATH_DICT
from lib.common import status_dict
from lib.common import admin_dict
from lib.common import symbol
from lib.common import md5_encryption
from lib.common import login_auth
from lib.common import activa_bank_auth
def get_user_pwd():
username = input("请输入用户名：")
password = input("请输入6位密码：")
return username, password
def register():
"""注册"""
try:
read_name_pwd("user_pwd.json")
username, password = get_user_pwd()
if username in all_user_data:
print(f"用户名[{username}]已存在，请勿重复注册 ^_^")
return
assert password.isdigit() and len(password) == 6, "错误，密码必须是6位数字"
except Exception as e:
print(e)
else:
if username == admin_dict['username'] and password == admin_dict['password']:
role = "admin"
else:
role = "normal"
flag, code_symbol = symbol()
if flag:
rel_password = md5_encryption(password, code_symbol)
data = creat_user_format(username=username, password=rel_password, role=role, salt=code_symbol)
save_name_pwd("user_pwd.json", data)
save_user_log(f"{username}_log.txt", username, "注册成功")
print(f"恭喜，用户[{username}]注册成功")
def login():
"""登录"""
read_name_pwd("user_pwd.json")
username, password = get_user_pwd()
if username in all_user_data:
salt = all_user_data.get(username).get("salt")
rel_password = all_user_data.get(username).get("password")
flag, code_symbol = symbol()
if flag:
user_password = md5_encryption(password, salt)
if user_password == rel_password:
save_user_log(f"{username}_log.txt", username, "登录成功")
status_dict['username'] = username
status_dict['identity'] = all_user_data.get(username).get("role")
status_dict['is_login'] = True
print(f"用户[{username}]登录成功")
read_bank_data(f"user_bank.json") # 登录成功之后加载银行卡信息，就算没有也不会报错
else:
save_user_log(f"{username}_log.txt", username, "登录失败")
print("密码错误")
else:
print(f"用户[{username}]不存在")
@login_auth
@activa_bank_auth
def show_self_data():
path = PATH_DICT.get("user_data") / "user_pwd.json"
with open(path, encoding="utf-8") as file:
user_data = json.load(file)
path = PATH_DICT.get("bank_data") / "user_bank.json"
with open(path, encoding="utf-8") as file:
bank_data = json.load(file)
LOGIN_NAME = fetch_login_name()
login_pwd = user_data[LOGIN_NAME].get("password")
role = user_data[LOGIN_NAME].get("role")
bank_number = bank_data[LOGIN_NAME].get("bankcard_number")
bankcard_password = bank_data[LOGIN_NAME].get("bankcard_password")
balance = bank_data[LOGIN_NAME].get("balance")
print(f"用户{LOGIN_NAME}的信息如下".center(36, "-"))
print(dedent(f"""
姓名:{LOGIN_NAME}
登录密码:{login_pwd}
身份:{role}
银行账号:{bank_number}
取款密码:{bankcard_password}
余额:{balance}
"""))
print("^_^敏感信息已加密^_^".rjust(36))
save_user_log(f"{LOGIN_NAME}_log.txt", LOGIN_NAME, "查看个人信息成功")
def logout():
if status_dict['is_login'] is False:
print("当前还没有用户登录哦~")
else:
username = fetch_login_name()
print(f"用户[{username}]已退出系统")
status_dict['username'] = ''
status_dict['is_login'] = False
core下的bank.py
和银行打交道的代码一般我会放在这个目录里面
from lib.common import login_auth
from lib.common import status_dict
from lib.common import activa_bank_auth
from lib.common import bank_data_dict
from lib.common import symbol
from lib.common import fetch_login_name
from lib.common import md5_encryption
from lib.common import bankcard_username
from DB.db_hander import save_bank_data
from DB.db_hander import creat_user_format
from DB.db_hander import save_user_log
from DB.db_hander import save_user_flow
from DB.db_hander import read_user_flow
@login_auth
def active_bankcard():
"""激活银行卡功能"""
LOGIN_NAME = fetch_login_name()
if bank_data_dict.get(LOGIN_NAME) is not None and bank_data_dict.get(LOGIN_NAME)['active_card'] == "YES":
print(f"用户[{LOGIN_NAME}], 您已激活过银行卡，请勿重复激活。")
save_user_log(f"{LOGIN_NAME}_log.txt", LOGIN_NAME, "尝试重复执行银行卡")
return
try:
bankcard_number = input("请输入6位银行卡卡号：")
assert bankcard_number.isdigit() and len(bankcard_number) == 6, "错误，银行卡号必须是6位纯数字"
bankcard_password = input("请输入3位取款密码：")
assert bankcard_password.isdigit() and len(bankcard_password) == 3, "错误，取款密码必须是3位纯数字"
except Exception as e:
print(e)
else:
balance = 1000
flag, code_symbol = symbol(4)
if flag:
rel_bankcard_password = md5_encryption(bankcard_password, code_symbol)
data = creat_user_format(username=LOGIN_NAME, bankcard_number=bankcard_number,
bankcard_password=rel_bankcard_password, balance=balance, active_card="YES",
salt=code_symbol)
save_bank_data("user_bank.json", data=data)
save_user_log(f"{LOGIN_NAME}_log.txt", LOGIN_NAME, "成功激活银行卡")
save_user_flow(f"{LOGIN_NAME}_bank_log.txt", LOGIN_NAME, f"初始化银行卡信息成功 余额为 {balance}元")
print("恭喜！已成功激活银行卡 ^_^")
status_dict['is_active_bankcard'] = "YES"
# status_dict = {"username": "", "identity": "", "is_login": False, "is_active_bankcard": False}
# 会复用的代码
# 校验金额
def get_bank_input(key="取款"):
try:
LOGIN_NAME = fetch_login_name()
out_balance = bank_data_dict[LOGIN_NAME].get("balance") # int
balance = input(f"当前余额为{out_balance},请输入{key}金额：")
if key == "取款":
assert balance.isdigit() and float(balance) <= out_balance, f"错误，{key}金额应小于等于{out_balance}！"
bankcard_password = input("请输入3位取款密码：")
assert bankcard_password.isdigit() or len(bankcard_password) == 3, "错误，密码必须是3位纯数字"
except Exception as e:
print(e)
else:
balance = float(balance)
return out_balance, balance, bankcard_password
@login_auth
@activa_bank_auth
def show_flow():
"""查看流水"""
LOGIN_NAME = fetch_login_name()
read_user_flow(f"{LOGIN_NAME}_bank_log.txt")
save_user_log(f"{LOGIN_NAME}_log.txt", LOGIN_NAME, "查看余额成功`")
# 会复用的代码
# 获取加盐的取款密码
def fetchBankPasswordWithSalt(bankcard_password):
LOGIN_NAME = fetch_login_name()
rel_bankcard_password = bank_data_dict.get(LOGIN_NAME).get("bankcard_password")
salt = bank_data_dict.get(LOGIN_NAME).get("salt")
user_bankcard_password = md5_encryption(bankcard_password, salt)
return rel_bankcard_password, user_bankcard_password
@login_auth
@activa_bank_auth
def withdraw_money():
"""取款功能"""
LOGIN_NAME = fetch_login_name()
flag = fetchBankcardNumber(LOGIN_NAME)
if flag:
try:
fetchBankData_andUpdateData(LOGIN_NAME=LOGIN_NAME, key="取款")
except Exception:
pass
@login_auth
@activa_bank_auth
def transfer_account():
"""转账功能"""
if len(bankcard_username) < 2:
print("错误！当前系统里激活过银行卡的用户太少")
else:
print(f"当前系统里面有{len(bankcard_username)}个已激活银行卡的用户")
try:
LOGIN_NAME = fetch_login_name()
for name, value in bank_data_dict.items():
print(f"用户名:{name}\t银行账号:{value['bankcard_number']}")
transfer_number = input("请输入对方的六位账号：")
assert bankcard_username[transfer_number] != LOGIN_NAME, "错误！无法转账给自己"
assert transfer_number in bankcard_username, "错误，账号不存在。"
assert transfer_number.isdigit() or len(transfer_number) == 6, "错误，银行卡必须是6位纯数字"
to_name = bankcard_username[transfer_number]
out_balance, balance, bankcard_password = get_bank_input("转账")
except Exception as e:
print(e)
else:
flag = symbol(4) # 验证码
if flag:
rel_bankcard_password, user_bankcard_password = fetchBankPasswordWithSalt(bankcard_password)
if user_bankcard_password == rel_bankcard_password:
balance = float(balance)
# 余额 交易金额
last_balance = out_balance - balance
print(f"取款成功，当前剩余金额为：{last_balance}元")
# 更新数据到字典里面
bank_data_dict[LOGIN_NAME]['balance'] = last_balance
bank_data_dict[to_name]['balance'] += balance
# 这里应该写入流水
save_user_flow(f"{LOGIN_NAME}_bank_log.txt", LOGIN_NAME,
f"向 {to_name} 转账 {balance}元 余额为 {last_balance}元")
save_user_flow(f"{to_name}_bank_log.txt", to_name,
f"收到 {LOGIN_NAME} 转账 {balance}元 余额为 {bank_data_dict[to_name]['balance']}元")
# 更新本地文件
save_bank_data("user_bank.json", {LOGIN_NAME: bank_data_dict[LOGIN_NAME]})
save_bank_data("user_bank.json", {to_name: bank_data_dict[to_name]})
else:
# 验证码输入错误的情况下应该记录日志
save_user_flow(f"{LOGIN_NAME}_bank_log.txt", LOGIN_NAME, "验证码输入错误")
# 取款和充值金额会复用
def fetchBankcardNumber(login_name):
try:
bankcard_number = input("请输入6位银行卡卡号：")
assert bankcard_number.isdigit() or len(bankcard_number) == 6, "错误，银行卡号必须是6位纯数字"
if bankcard_number != bank_data_dict[login_name]['bankcard_number']:
print("错误，银行卡账号不匹配。")
return
except Exception as e:
print(e)
return
else:
return True
@login_auth
@activa_bank_auth
def recharge():
"""充值功能"""
LOGIN_NAME = fetch_login_name()
flag = fetchBankcardNumber(LOGIN_NAME)
if flag:
try:
fetchBankData_andUpdateData(LOGIN_NAME=LOGIN_NAME, key="充值")
except Exception:
pass
# 会复用的代码
# 获取用户输入并写入日志里面
def fetchBankData_andUpdateData(LOGIN_NAME, key="取款"):
out_balance, balance, bankcard_password = get_bank_input(key)
flag, code_symbol = symbol(4) # 验证码
if flag:
rel_bankcard_password, user_bankcard_password = fetchBankPasswordWithSalt(bankcard_password)
if user_bankcard_password == rel_bankcard_password:
if key == "取款":
last_balance = out_balance - float(balance)
else:
last_balance = out_balance + float(balance)
print(f" {key}成功，当前剩余金额为：{last_balance}元")
# 这里应该写入流水
save_user_flow(f"{LOGIN_NAME}_bank_log.txt", LOGIN_NAME,
f"充值金钱 {balance}元 余额为 {last_balance}元")
# 更新数据到字典里面
bank_data_dict[LOGIN_NAME]['balance'] = last_balance
# 更新本地文件
save_bank_data('user_bank.json', {LOGIN_NAME: bank_data_dict[LOGIN_NAME]})
else:
# 验证码输入错误的情况下应该记录日志
save_user_flow(f"{LOGIN_NAME}_bank_log.txt", LOGIN_NAME, "验证码输入错误")
DB下的db_hander.py
文件操作相关的代码
import json
from pathlib import Path
from conf.settings import PATH_DICT
from lib.common import now
from lib.common import status_dict
from lib.common import bank_data_dict
from lib.common import all_user_data
from lib.common import fetch_login_name
from lib.common import bankcard_username
# 创建格式化的数据
def creat_user_format(**kwargs):
username = kwargs.get("username")
data = {username: kwargs}
return data
# 保存用户名和密码的函数
def save_name_pwd(path, data):
path = PATH_DICT.get("user_data") / path
all_user_data.update(data)
with open(path, "w", encoding="utf-8") as file:
json.dump(all_user_data, file, ensure_ascii=False)
# db_path_list = ['user_data', "bank_data", "user_flow", "user_log"]
# 保存银行卡信息的函数
def save_bank_data(path, data):
# print(f"即将被写入的银行数据 --->", data)
path = PATH_DICT.get("bank_data") / path
bank_data_dict.update(data)
with open(path, "w", encoding="utf-8") as file:
json.dump(bank_data_dict, file, ensure_ascii=False)
# 保存用户日志的函数
def save_user_log(path, username, data):
# 当前用户 username 于 xx年xx月xx日 xx时xx分xx秒 注册成功
path = PATH_DICT.get("user_log") / path
with open(path, "at", encoding="utf-8") as file:
file.write(f"前用户 {username} 于 {now()} {data}" + "\n")
# 查看流水的函数
def read_user_flow(path):
path = PATH_DICT.get("user_flow") / path
with open(path, encoding="utf-8") as file:
for line in file:
print(line.strip())
# 保存用户流水的日志
def save_user_flow(path, username, data):
# # 当前用户 username 于 xx年xx月xx日 xx时xx分xx秒 初始化银行卡信息成功 余额为 last_balance
path = PATH_DICT.get("user_flow") / path
with open(path, "at", encoding="utf-8") as file:
file.write(f"当前用户 {username} 于 {now()} {data}" + "\n")
# 读取用户名和密码的函数
def read_name_pwd(path):
path = PATH_DICT.get("user_data") / path
if not Path(path).exists():
return
with open(path, encoding="utf-8") as file:
data = json.load(file)
all_user_data.update(data)
# 读取银行信息的函数
# bank_data_dict = {}
def read_bank_data(path):
path = PATH_DICT.get("bank_data") / path
if not Path(path).exists():
return
with open(path, encoding="utf-8") as file:
data = json.load(file)
bank_data_dict.update(data)
for username, v in bank_data_dict.items():
bankcard_number = v.get("bankcard_number")
bankcard_username.setdefault(bankcard_number, username)
LOGIN_NAME = fetch_login_name()
if LOGIN_NAME in bank_data_dict:
active_card = bank_data_dict.get(LOGIN_NAME).get("active_card")
status_dict['is_active_bankcard'] = active_card # 如果用户已经注册过，下次使用就不需要重复激活银行卡了
# 加载本地读取数据的函数，一开始就加载
def active():
read_name_pwd("user_pwd.json")
read_bank_data("user_bank.json")
lib下的common.py
我在这里存放了一些公共的功能，比如登录认证装饰器，全局字典等。
import time
import string
import random
from hashlib import md5
from functools import wraps
def now():
return time.strftime("%Y年%m月%d日 %H点%M分%S秒")
# 登录成功后修改状态的字典
status_dict = {"username": "", "identity": "", "is_login": False, "is_active_bankcard": "NO"}
# 管理员字典信息
admin_dict = {"username": "eva", "password": "112233"}
# 银行操作字典
bank_data_dict = {}
# 方便转账功能
bankcard_username = {}
# 使用json去创建数据
all_user_data = {}
# 登录验证语法糖，很多功能如果不登录是无法使用的
def login_auth(func):
@wraps(func)
def inner(*args, **kwargs):
if status_dict['is_login'] is False:
print("尚未登录，请登录后使用此功能。")
else:
return func(*args, **kwargs)
return inner
# 激活银行卡语法糖验证
def activa_bank_auth(func):
@wraps(func)
def inner(*args, **kwargs):
if status_dict['is_active_bankcard'] == "NO":
print("尚未激活银行卡，请激活银行卡后使用此功能。")
else:
return func(*args, **kwargs)
return inner
# 单纯获取全局登录的字典
def fetch_login_name():
username = status_dict['username']
return username
# 就一个功能，显示当前登录的用户
def status():
username, identity, is_login, is_active_bankcard = status_dict.values()
if is_login:
print(f"当前登录用户: {username}")
print()
# 生成验证码的函数
def symbol(number=4):
text = string.ascii_letters + string.digits
# 生成的验证码
code_symbol = "".join(random.choices(text, k=number))
user_input = input(f"请输入看到的{number}位验证码（不区分大小写）: --> {code_symbol} <-- ").strip()
if user_input.casefold() == code_symbol.casefold():
return True, code_symbol
else:
print("对不起，验证码错误！")
return False, code_symbol
# md5加密
def md5_encryption(text: str, salt):
md5_obj = md5()
content = text + salt
md5_obj.update(content.encode("utf-8"))
return md5_obj.hexdigest()
效果演示
本地文件效果
DB\bank_data\user_data.json
{
"eva": {
"username": "eva",
"bankcard_number": "888888",
"bankcard_password": "c3a88567e5d5a669f0e5304cbe87d4a9",
"balance": 890.0,
"active_card": "YES",
"salt": "ghQc"
},
"jack": {
"username": "jack",
"bankcard_number": "666666",
"bankcard_password": "533921648684c184c20382ba9a8b1296",
"balance": 1618.0,
"active_card": "YES",
"salt": "9nmh"
}
}
DB\user_data\user_pwd.json
{
"eva": {
"username": "eva",
"password": "619be042e858c9d73c6beb7d289662f0",
"role": "admin",
"salt": "7TaZ"
},
"jack": {
"username": "jack",
"password": "4d1c1c363dc0386bce9581e5e8968cc0",
"role": "normal",
"salt": "kX2r"
}
}
DB\user_flow\eva_bank_log.txt
当前用户 eva 于 2023年12月18日 16点50分32秒 初始化银行卡信息成功 余额为 1000元
当前用户 eva 于 2023年12月18日 16点50分44秒 充值金钱 192.0元 余额为 808.0元
当前用户 eva 于 2023年12月18日 16点52分22秒 收到 jack 转账 82.0元 余额为 890.0元
DB\user_flow\jack_bank_log.txt
当前用户 jack 于 2023年12月18日 16点51分34秒 初始化银行卡信息成功 余额为 1000元
当前用户 jack 于 2023年12月18日 16点51分55秒 充值金钱 123.0元 余额为 877.0元
当前用户 jack 于 2023年12月18日 16点52分22秒 向 eva 转账 82.0元 余额为 795.0元
当前用户 jack 于 2023年12月18日 16点52分36秒 充值金钱 823.0元 余额为 1618.0元
DB\user_log\eva_log.txt
前用户 eva 于 2023年12月18日 16点50分16秒 注册成功
前用户 eva 于 2023年12月18日 16点50分22秒 登录成功
前用户 eva 于 2023年12月18日 16点50分32秒 成功激活银行卡
DB\user_log\jack_log.txt
前用户 jack 于 2023年12月18日 16点51分02秒 注册成功
前用户 jack 于 2023年12月18日 16点51分18秒 登录成功
前用户 jack 于 2023年12月18日 16点51分34秒 成功激活银行卡
前用户 jack 于 2023年12月18日 16点52分53秒 查看余额成功`
前用户 jack 于 2023年12月18日 16点52分57秒 查看个人信息成功
使用到的技术&模块
1. 装饰器，登录验证装饰器以及激活银行卡两个装饰器
2. 随机生成验证码(random模块以及string模块)
3. 格式化打印字符串(textwrap模块)
4. MD5加密（hashlib模块）
5. 记录时间（time模块）
6. 路径加载（sys模块）
7. 操作路径（pathlib模块）
8. 写入本地json文件（json模块）
写在后面
需要优化改进的地方还有很多很多，看到这篇文章的你我一起努力。
（山腰太挤了，我们山顶见）