分层架构购物车

大致的层级结构

全部代码

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模块）

写在后面

需要优化改进的地方还有很多很多，看到这篇文章的你我一起努力。
（山腰太挤了，我们山顶见）