Python架构及部分ATM项目

一、一个项目是如何从无到有

1、需求分析

1.拿到项目，会先在客户那里一起讨论需求，
商量项目的功能是否能实现，周期与价格，得到一个需求文档。

2.最后在公司内部需要开一次会议，最终得到一个开发文档，
交给不同岗位的程序员进行开发。
    - Python: 后端，爬虫
    
    - 不同的岗位：
        - UI界面设计:
            - 设计软件的布局，会分局软件的外观切成一张张图片。
        
        - 前端:
            - 拿到UI交给他的图片，然后去搭建网页面。
            - 设计一些页面中，哪些位置需要接收数据，需要进行数据交互。
        
        - 后端:
            - 直接核心的业务逻辑，调度数据库进行数据的增删查改。
        
        - 测试:
            - 会给代码进行全面测试，比如压力测试，界面测试（CF卡箱子）。
        
        - 运维:
            - 部署项目。

2、程序的架构设计

1、程序设计的好处

1）思路清晰
2）不会出现写一半代码时推翻重写
3）方便自己或以后的同事更好维护

2、三层架构设计的好处

1）把每个功能都分层三部分，逻辑清晰
2）如果用户更换不同的用户界面或不同，的数据储存机制都不会影响接口层的核心
逻辑代码，扩展性强。
3）可以在接口层，准确的记录日志与流水。

3、三层架构

　　1 用户视图层

　　　用户与用户交互的，可以接受用户的输入，打印接口返回的数据

　　2 逻辑接口层

　　　接收用户视图层传过来的参数，构建逻辑判断调用的数据层加以处理，并返回一个结果给用户视图层

　　3 数据处理层

　　　接收接口传递过来的参数，做数据的增删改查

三 分任务开发

四 测试

五 上线

---

二、项目的说明书

项目：ATM + 购物车

项目需求:

1.额度15000或自定义     -->  注册功能
2.实现购物商城，买东西加入购物车，调用信用卡接口结账  --> 购物功能、支付功能
3.可以提现，手续费5%   --> 提现功能
4.支持多账户登录  --> 登录功能
5.支持账户间转账  --> 转账功能
6.记录日常消费 -->  记录流水功能
7.提供还款接口 -->  还款功能
8.ATM记录操作日志 --> 记录日志功能
9.提供管理接口，包括添加账户、用户额度，冻结账户等... ---> 管理员功能
10.用户认证用装饰器  --> 登录认证装饰器

项目思路流程框架图：

"用户视图层" 展示给用户选择的功能：

1、注册功能
2、登录功能
3、查看余额
4、提现功能
5、还款功能
6、转账功能
7、查看流水
8、购物功能
9、查看购物车
10、管理员功能

开始创建目录：

conf
	--settings 存放配置信息
core
	--src 用户视图层
interface  -- 逻辑接口层
	--user_interface 用户相关接口（注册、登录）
	--bank_interface 银行相关接口（查看余额，提现、还款、转账）
	--shop_interface 购物车相关接口（购物功能、查看购物车）
db
	--user_data
		--deimos.json
		--aaa.json
	--db_handler 数据处理层，专门用于处理数据
lib
	--common
log
	--access.log
start.py  -- 程序的入口

用户视图层相关代码：

'''
第一层: 用户视图层
'''

from interface import user_interface
from interface import bank_interface
from interface import shop_interface
from lib import common

global_name = None
# 1、注册功能
# 面条版
'''
# 面条版
def register():
    while True:
        username = input('请输入您的账号：>>').strip()
        password = input('请输入您的密码：>>').strip()
        re_password = input('请重复输入您的账号：>>').strip()
        # 小逻辑处理
        if password == re_password:
            import json
            import os
            from conf import settings
            user_path = os.path.join(
                settings.USER_DATA_PATH, f'{username}.json'
            )
            # 查看用户是否存在
            if os.path.exists(user_path):
                print('请重新输入')
                continue
            # 若用户存在重新输入
                with open(user_path,'r',encoding='utf-8') as rf:
                    data = json.load(rf)
                if data:
                    print('账号存在，请重新输入')
                    continue
            # 用户不存在，则保存用户数据
            # 阻止用户的数据字典信息
            user_dic = {
                'username': username,
                'password': password,
                'balance': 15000,
                'flow': [],  # 用于记录用户流水的列表
                'shop_car': {},  # 用户记录用户购物车
                'locked': False  # 用于记录账户是否被冻结
            }
            # 用户数据，tank.json 存不是目的，目的是为了取数据
            with open(user_path, 'w', encoding='utf-8')as f:
                # ensure_ascii=False 显示文件当中的中文数据，使得美观
                json.dump(user_dic, f,ensure_ascii=False)
'''



# 注册功能
def register():
    while True:
        user_name = input('请输入您的账号:>>').strip()
        user_pwd = input('请输入您的密码:>>').strip()
        re_user_pwd = input('请再次输入您的密码:>>').strip()
        if user_pwd == re_user_pwd:
            flag, msg = user_interface.register_interface(user_name, user_pwd)
        if flag:
            print(msg)
            break
        else:
            print(msg)


# 登录功能
def login():
    while True:
        user_name = input('请输入您的账号:>>').strip()
        user_pwd = input('请输入您的密码:>>').strip()
        flag, msg = user_interface.login_interface(user_name, user_pwd)
        if flag:
            print(msg)
            global global_name
            global_name = user_name
            break
        else:
            print(msg)


# 查看余额
@common.login_auth
def check_balance():
    balance = user_interface.check_balance_interface(global_name)
    print(f'用户{global_name}的账户余额为{balance}元')


# 提现功能
@common.login_auth
def withdraw():
    while True:
        input_money = input('请输入您要提现的金额：>>').strip()
        if not input_money.isdigit():
            print('格式不规范，请重新输入')
            continue
        flag, msg = bank_interface.withdraw_interface(global_name, input_money)
        if flag:
            print(msg)
            break
        else:
            print(msg)


# 还款功能
@common.login_auth
def repay():
    while True:
        input_money = input('请输入您要还的金额：>>').strip()
        if not input_money.isdigit():
            print('格式不规范，请重新输入')
            continue
        input_money = float(input_money)
        if input_money > 0:
            flag, msg = bank_interface.repay_interface(global_name, input_money)
            if flag:
                print(msg)
                break
        else:
            print('输入金额不能少于0')


# 转账功能
@common.login_auth
def transfer():
    while True:
        transfer_name = input('请输入转入的账号:>>').strip()
        money = input('请输入转账金额:>>').strip()
        if global_name == transfer_name:
            print('不能给自己转账！！')
            continue
        if not money.isdigit():
            print('格式错误，请重新输入正确金额！')
            continue
        money = float(money)
        if money > 0:
            flag, msg = bank_interface.transfer_interface(global_name, transfer_name, money)
            if flag:
                print(msg)
                break
            else:
                print(msg)
        else:
            print('格式错误，请重新输入正确金额！')


# 查看流水
@common.login_auth
def check_flow():
    flow = bank_interface.check_flow_interface(global_name)
    if flow:
        for line in flow:
            print(line)

    else:
        print('当前用户没有流水')


# 购物功能
@common.login_auth
def shopping():
    shop_list = [
        ['包子', 11],
        ['啤机', 12],
        ['鸡爪', 13],
        ['水杯', 14],
        ['酸奶', 16],
    ]
    shop_car = {}
    while True:
        print('========购物商城===========')
        for index,shop in enumerate(shop_list):
            shop_name,shop_price = shop
            print(f'商品编号为{index}	',f'商品名称为{shop_name}	',f'商品价格为{shop_price}	')
        print('=======欢迎选购============')
        choice = input('请输入你要购买的商品，(是否结账输入n or y):>>').strip()
        if choice == 'n':
            if not shop_car:
                print('购物车是空的，不能添加')
                continue
            flag, msg = shop_interface.add_shop_car_interface(global_name, shop_car)
            if flag:
                print(msg)
                break
            else:
                print(msg)
        elif choice == 'y':
            if not shop_car:
                print('购物车是空的，不能支付')
                continue
            flag, msg = shop_interface.shopping_interface(global_name, shop_car)
            if flag:
                print(msg)
                break
            else:
                print(msg)
        if not choice.isdigit():
            print('请输入正确编号！')
            continue
        choice = int(choice)
        if choice not in range(len(shop_list)):
            print('请输入正确的编号！')
            continue
        shop_name,shop_price = shop_list[choice]
        if shop_name in shop_car:
            shop_car[shop_name][1]+=1
        else:
            shop_car[shop_name] = [shop_price,1]
        print('当前购物车',shop_car)


# 查看购物车
@common.login_auth
def check_shop_car():
    shop_car = shop_interface.check_shop_car_interface(global_name)
    print(shop_car)


# 管理员功能
@common.login_auth
def admin():
    from core import admin
    admin.admin_run()


# 创建函数字典
func_dict = {
    '0': exit,
    '1': register,
    '2': login,
    '3': check_balance,
    '4': withdraw,
    '5': repay,
    '6': transfer,
    '7': check_flow,
    '8': shopping,
    '9': check_shop_car,
    '10': admin,
}


# 视图层主程序
def run():
    while True:
        print(
            """
            0、退出
            1、注册功能
            2、登录功能
            3、查看余额
            4、提现功能
            5、还款功能
            6、转账功能
            7、查看流水
            8、购物功能
            9、查看购物车
            10、管理员功能
            """
        )
        num = input('请输入功能编号：>>').strip()
        if num not in func_dict:
            print('请输入正确的编号！！')
            continue

        func_dict.get(num)()

逻辑接口层相关代码：

"""
银行相关接口
"""
from db import db_handler
from lib import common


bank_logger = common.get_logger(log_type='bank')


def withdraw_interface(username, money):
    user_dic = db_handler.select(username)
    balance = float(user_dic.get('balance'))

    money1 = float(money) * 1.05
    money2 = float(money) * 0.05
    if money1 <= balance:
        balance1 = balance - money1
        user_dic['balance'] = balance1
        flow = f'用户{username}提现成功，提现金额为{money}元,手续费为{round(money2, 2)}元'
        user_dic['flow'].append(flow)
        db_handler.save(user_dic)
        bank_logger.info(flow)
        return True, flow
    return False, '提现金额不足，请重新输入'


def repay_interface(username, money):
    user_dic = db_handler.select(username)
    user_dic['balance'] += float(money)
    flow = f'账户{username}还款成功，还款金额为{money}元'
    user_dic['flow'].append(flow)
    bank_logger.info(flow)
    db_handler.save(user_dic)
    return True, flow


def transfer_interface(global_name, transfer_name, money):
    login_dic = db_handler.select(global_name)
    transfer_dic = db_handler.select(transfer_name)
    if not transfer_dic:
        return False, f'目标用户不存在'
    if login_dic['balance'] >= money:
        login_dic['balance'] -= money
        transfer_dic['balance'] += money
        flow_login = f'用户{global_name}给用户{transfer_name}转账{money}元'
        flow_transfer = f'用户{transfer_name}收到用户{global_name}的转账{money}元'
        login_dic['flow'].append(flow_login)
        transfer_dic['flow'].append(flow_transfer)
        bank_logger.info(flow_login)
        bank_logger.info(flow_transfer)
        db_handler.save(login_dic)
        db_handler.save(transfer_dic)
        return True, flow_login
    else:
        return False, f'当前用户转账资金不足'


def check_flow_interface(name):
    login_dic = db_handler.select(name)
    return login_dic['flow']


def pay_interface(global_name,cost):
    user_dic = db_handler.select(global_name)
    if user_dic.get('balance') >=cost:
        user_dic['balance'] -= cost
        flow = f'用户{global_name}消费金额{cost}元'
        bank_logger.info(flow)
        user_dic['flow'].append(flow)
        db_handler.save(user_dic)
        return True
    else:
        return False



'''
第二层: 用户接口
'''
from db import db_handler
from lib import common

user_logger = common.get_logger(log_type='bank')


# 注册接口
def register_interface(user_name,user_pwd,balance = 15000):
    # 查看用户是否存在
    user_dic = db_handler.select(user_name)
    # 不存在则注册，并且做密码加密
    if user_dic:
        return False,f'账户{user_name}已存在'

    user_pwd = common.pwd_md5(user_pwd)
    # 用户字典信息
    user_dic = {
        'username':user_name,
        'password':user_pwd,
        'balance':balance,
        'flow':[],
        'shop_car':{},
        # 账户状态
        'locked':False
    }
    # 保存数据
    db_handler.save(user_dic)
    msg = f'{user_name}注册成功'
    user_logger.info(msg)
    return True,msg


# 登录接口
def login_interface(user_name,user_pwd):
    user_dic = db_handler.select(user_name)
    if user_dic:
        if user_dic.get('locked'):
            return False,f'账户{user_name}已经被锁定'
        user_pwd = common.pwd_md5(user_pwd)
        if user_pwd == user_dic.get('password'):
            msg = f'用户{user_name}登陆成功'
            user_logger.info(msg)
            return True,msg
        else:
            msg = f'用户{user_name}密码错误'
            user_logger.info(msg)
            return False,msg
    return False,f'用户{user_name}不存在，请重新输入'


# 查看账户余额
def check_balance_interface(username):
    user_dic = db_handler.select(username)
    return user_dic.get('balance')


'''
第三层: 购物车接口
'''
from db import db_handler
from lib import common


shop_logger = common.get_logger(log_type='shop')


def add_shop_car_interface(global_name, shopping_car):
    user_dic = db_handler.select(global_name)
    shop_car = user_dic.get('shop_car')
    for shop_name, price_number in shopping_car.items():
        number = price_number[1]
        if shop_name in shop_car:
            user_dic['shop_car'][shop_name][1] += number
        else:
            user_dic['shop_car'].update(
                {shop_name: price_number}
            )
    db_handler.save(user_dic)
    return True, '添加购物车成功'


def shopping_interface(global_name, shop_car):
    cost = 0
    for price_number in shop_car.values():
        price,number = price_number
        cost += float(price*number)
        from interface import bank_interface
        flag = bank_interface.pay_interface(global_name,cost)
        if flag:
            msg = f'用户{global_name}支付成功，支付金额为{cost}元'
            shop_logger.info(msg)
            return True,msg
        return False,'支付失败，余额不足'


def check_shop_car_interface(global_name):
    user_dic = db_handler.select(global_name)
    return user_dic['shop_car']

数据处理层相关代码：

'''
第三层: 数据处理层
用于专门处理数据的
'''
import os
from conf import settings
import json


def select(user_name):
    user_path = os.path.join(settings.USER_DATA_PATH,f'{user_name}.json')
    if os.path.exists(user_path):
        with open(user_path,'r',encoding='utf-8')as f:
            user_dic = json.load(f)
            return user_dic


def save(user_dic):
    username = user_dic.get('username')
    user_path = os.path.join(settings.USER_DATA_PATH,f'{username}.json')
    with open(user_path,'w',encoding='utf-8')as f:
        json.dump(user_dic,f,ensure_ascii=False)

管理员相关代码：

"""
admin.py
"""
from core import src
from interface import admin_interface

def add_user():
    src.register()


def changer_balance():
    while True:
        change_name = input('请输入需要修改额度的账户：>>').strip()
        money = input('请输入需要修改额度：>>').strip()
        if not money.isdigit():
            print('格式错误')
            continue
        flag,msg = admin_interface.changer_balance_interface(change_name,money)
        if flag:
            print(msg)
            break
        else:
            print(msg)


def lock_user():
    while True:
        lock_name = input('请输入需要冻结的账户：>>').strip()
        flag, msg = admin_interface.lock_user_interface(lock_name)
        if flag:
            print(msg)
            break
        else:
            print(msg)


admin_dic = {
    '1':add_user,
    '2':changer_balance,
    '3':lock_user,

}


def admin_run():
    while True:
        print("""
        1、添加账户
        2、修改额度
        3、冻结账户
        """)
        num = input('请输入功能编号：>>').strip()
        if num not in admin_dic:
            print('请输入正确的编号！！')
            continue

        admin_dic.get(num)()

"""
admin_interface.py
"""        
from db import db_handler
from lib import common

admin_logger = common.get_logger(log_type='admin')


def changer_balance_interface(change_name,money):
    user_dic = db_handler.select(change_name)
    if user_dic:
        user_dic['balance'] = float(money)
        db_handler.save(user_dic)
        msg = f'管理员修改用户{change_name}额度修改成功！'
        admin_logger.info(msg)
        return True,msg
    return False,'账户不存在'


def lock_user_interface(lock_name):
    user_dic = db_handler.select(lock_name)
    if user_dic:
        user_dic['locked'] = True
        db_handler.save(user_dic)
        msg = f'{lock_name}账户冻结成功'
        admin_logger.info(msg)
        return True,msg
    return False,'冻结用户不存在'

项目配置：

import os

BASE_PATH = os.path.dirname(os.path.dirname(__file__))

USER_DATA_PATH = os.path.join(BASE_PATH,'db','user_data')

# print(USER_DATA_PATH)


"""
logging配置
"""

# 定义三种日志输出格式 开始
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.log'

# 如果不存在定义的日志目录就创建一个
if not os.path.isdir(logfile_dir):
    os.mkdir(logfile_dir)

# log文件的全路径
logfile_path = os.path.join(logfile_dir, logfile_name)

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）传递
        },
    },
}

公共方法:

import hashlib
from conf import settings
import logging.config


def pwd_md5(user_pwd):
    md5_obj = hashlib.md5()
    md5_obj.update(user_pwd.encode('utf-8'))
    salt = '不要想着破解密码了'
    md5_obj.update(salt.encode('utf-8'))
    return md5_obj.hexdigest()


# 登录认证装饰器
def login_auth(func):
    from core import src
    def wrapper(*args,**kwargs):
        if src.global_name:
            res = func(*args,**kwargs)
            return res
        else:
            print('您还没有登录，请登录后再使用该功能')
            src.login()
    return wrapper


def get_logger(log_type):
    logging.config.dictConfig(
        settings.LOGGING_DIC
    )
    logger = logging.getLogger(log_type)
    return logger