ATM項目

項目

#一個項目是怎麼從無到有的

#1.需求分析
	額度15000或者自定義		--> 註冊功能
    實現購物商城,買東西自動加入購物車,調用信用卡接口結帳		--> 購物車、支付功能
    能夠提現,支付手續費5%	--> 提現
    支持多帳戶登陸		--> 登陸
    支持帳戶間轉帳		--> 轉帳,記錄流水
    提供還款接口		--> 還款
    ATM記錄操做日誌	--> 日誌功能
    提供管理接口,包括添加帳戶、用戶額度、凍結帳戶...		--> 管理員功能
    用戶認證功能		--> 登陸認證,使用裝飾器

#2.程序的架構設計
	用戶功能層,接收用戶輸入,輸出執行結果,還能夠作小的邏輯判斷
    接口層,業務邏輯的處理、日誌、流水
    數據處理層,數據的處理(增刪改查)
    
    #三層架構的好處
    	1.代碼的結構更加清晰
        2.可擴展性強
        3.易於維護、管理

#3.分任務開發
	- CTO
    	- 技術總監
        	- 架構師
            	- 項目經理
                	- 普通開發
                    - UI:用戶界面設計師
                    - 前端:網頁的開發
                    - 後端:寫業務邏輯、寫接口
                    - 測試:測試軟件
                    - 運維:部署項目
                    
#4.測試
	- 1.手動測試
    	傳統人工去手動測試
    - 2.自動化測試
    	經過腳本模擬人的行爲,自動化執行測試
    - 3.黑盒測試
    	對用戶界面進行測試 
    - 4.白盒測試
    	對軟件的性能進行測試,例如每分鐘能接收多少併發量
        
#5.上線運行

實例

參考網站html

啓動文件

#建立項目目錄ATM
#ATM/bin/start.py

import sys
import os
BASE_dir = os.path.dirname(os.path.dirname(__file__))
sys.path.append(BASE_dir)
from core import src
if __name__ == '__main__':
    src.run()

配置文件

#ATM/conf/settings.py

import os
base_path = os.path.dirname(os.path.dirname(__file__))
db_path = os.path.join(base_path, 'db')
#不要使用%s拼接路徑,要使用os模塊
LOG_path = os.path.join(base_path,'log')

"""
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'

# 定義日誌輸出格式 結束
logfile_dir = os.path.join(base_path, 'log')  # log文件的目錄
logfile_name = 'atm_shop_log.log'  # log文件名

# 若是不存在定義的日誌目錄就建立一個
if not os.path.isdir(logfile_dir):
    os.mkdir(logfile_dir)

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

# 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)傳遞
        },
    },
}

核心文件

#ATM/core/src.py

import time
from ATM.interface import user_interface
from ATM.lib import comman
from ATM.interface import bank_interface
from ATM.interface import shop_interface
from ATM.interface import admin_interface

user_info = {
    'user':None,
}
#註冊
def register():
    while True:
        user = input('請輸入用戶名>>>: ').strip()
        # 調用接口,判斷用戶是否存在
        flag = user_interface.check_user_interface(user)
        if flag:
            print('用戶已存在,請從新輸入!')
            continue
        pwd = input('請輸入密碼>>>: ').strip()
        re_pwd = input('請確認密碼>>>: ').strip()
        if pwd == re_pwd:
            #調用接口,保存用戶信息
            msg = user_interface.register_interface(user,pwd)
            if msg:
                print(msg)
                break
            else:
                print('註冊失敗')
        else:
            print('兩次輸入密碼不一致')
#登陸
def login():
    while True:
        user = input('請輸入用戶名>>>: ').strip()
        flag = user_interface.check_user_interface(user)
        if not flag:
            print('用戶不存在,請從新輸入!')
            continue
        pwd = input('請輸入用戶密碼>>>: ').strip()
        flag,msg = user_interface.login_interface(user,pwd)
        if flag:
            print(msg)
            #登陸成功後,作一個記錄
            user_info['user'] = user
            break
        else:
            print(msg)
            break
#查看餘額
@comman.login_auth
def check_balance():
    balabce = user_interface.check_balance_interface(user_info['user'])
    msg = f"{user_info['user']}餘額爲: "
    print(msg,balabce)
#提現
@comman.login_auth
def withdraw():
    while True:
        money = input('請輸入提現金額>>>: ').strip()
        if money.isdigit():
            money = int(money)
            flag,msg = bank_interface.withdraw_interface(user_info['user'],money)
            if flag:
                print(msg)
                break
            else:
                print(msg)
        else:
            print('請輸入數字')
#還款
@comman.login_auth
def pay_bak():
    while True:
        money = input('請輸入還款金額>>>: ').strip()
        if not money.isdigit():
            print('請輸入數字!')
            continue
        money = int(money)
        flag,msg = bank_interface.repay_interface(user_info['user'],money)
        if flag:
            print(msg)
            break
#轉帳
@comman.login_auth
def transfer():
    while True:
        to_user = input('請輸入轉帳用戶>>>: ').strip()
        flag = user_interface.check_user_interface(to_user)
        if not flag:
            print('目標用戶不存在!')
            continue
        #輸入轉帳金額
        money = input('請輸入轉帳金額>>>: ').strip()
        if not money.isdigit():
            print('請輸入數字')
            continue
        money = int(money)
        flag,msg = bank_interface.transfer_interface(to_user,user_info['user'],money)
        if flag:
            print(msg)
            break
        else:
            print(msg)
#查看流水
@comman.login_auth
def check_record():
    flow_list = bank_interface.check_flow_interface(user_info['user'])
    for flow in flow_list:
        print(flow)
#購物車功能
@comman.login_auth
def shopping_cart():
    good_list = [
        ['襯衫', 500],
        ['褲子', 300],
        ['鞋子', 150],
        ['外套', 1000]
    ]
    #用戶餘額
    user_balance = user_interface.check_balance_interface(user_info['user'])
    #購物車
    shop_cart = {}
    #清空購物車,用戶須要支付的金額
    cost = 0
    while True:
        # 打印商品信息
        for index, good_price in enumerate(good_list, 1):
            print(index, good_price)
        # 選擇商品編號或者輸入q退出
        choice = input('請輸入商品編號或輸入q退出>>>: ').strip()
        if choice.isdigit():
            choice = int(choice)
            if choice >= 0 and choice <= len(good_list):
                good_name, good_price = good_list[choice - 1]
                if user_balance >= good_price:
                    if good_name in shop_cart:
                        shop_cart[good_name] += 1
                    else:
                        shop_cart[good_name] = 1
                    cost += good_price
                else:
                    print('餘額不足,沒法添加購物車!')
            else:
                print('商品編號不存在!')
        elif choice == 'q':
            commit = input('是否確認結帳,請輸入(y/n)').strip()
            if commit == 'y':
                # 調用商城支付功能,並調用銀行支付接口
                flag, msg = shop_interface.buy_shop_interface(user_info['user'], cost)
                if flag:
                    print(msg)
                    break
                else:
                    print(msg)
            elif commit == 'n':
                # 調用添加購物車接口
                flag, msg = shop_interface.add_shop_cart_interface(user_info['user'], shop_cart)
                if flag:
                    print(msg)
                    break
                else:
                    print(msg)
        else:
            print('請輸入商品編號或輸入q')
#查看購物車
@comman.login_auth
def check_shop_cart():
    shop_cart = shop_interface.check_shop_cart_interface(user_info['user'])
    if not shop_cart:
        print('購物車爲空')
    print(shop_cart)
#註銷
@comman.login_auth
def logout():
    while True:
        flag,msg = user_interface.logout_interface(user_info['user'])
        if flag:
            print(msg)
            break
#管理員功能
def lock_user():
    while True:
        user = input('請輸入須要凍結的帳戶>>>: ').strip()
        flag = user_interface.check_user_interface(user)
        if flag:
            msg = admin_interface.lock_interface(user)
            print(msg)
            break
        else:
            print('用戶不存在,請從新輸入!')
def unlock_user():
    while True:
        user = input('請輸入須要解凍的帳戶>>>: ').strip()
        flag = user_interface.check_user_interface(user)
        if flag:
            msg = admin_interface.unlock_interface(user)
            print(msg)
            break
        else:
            print('用戶不存在,請從新輸入!')
def change_limit():
    while True:
        user = input('請輸入須要修改額度的帳戶>>>: ').strip()
        flag = user_interface.check_user_interface(user)
        limit = input('請輸入修改額度>>>: ').strip()
        if flag:
            if limit.isdigit():
                limit = input(limit)
                #調用修改額度接口
                msg = admin_interface.change_limit_interface(user,limit)
                print(msg)
                break
            else:
                print('輸入額度無效,請從新輸入')
                continue
        else:
            print('用戶不存在,請從新輸入!')
admin_func_dic = {
    '1':lock_user,
    '2':unlock_user,
    '3':change_limit,
}
@comman.login_auth
def admin_sys():
    while True:
        print('''
            1.凍結帳戶
            2.解凍帳戶
            3.用戶額度
        ''')
        choice = input('請選擇管理員功能>>>: ').strip()
        if choice.isdigit():
            if choice in admin_func_dic:
                admin_func_dic[choice]()
                break
            else:
                print('相關功能還在開發中...')
                continue
        else:
            print('輸入無效,請從新輸入!')
my_dict = {
    '1':register,
    '2':login,
    '3':check_balance,
    '4':withdraw,
    '5':pay_bak,
    '6':transfer,
    '7':check_record,
    '8':shopping_cart,
    '9':check_shop_cart,
    '10':logout,
    '11':admin_sys
}
def run():
    while True:
        print("""
        1  register(註冊)
        2  login(登陸)
        3  check_balance(查看餘額)
        4  withdraw(提現)
        5  pay_bak(還款)
        6  transfer(轉帳)
        7  check_record(查看流水)
        8  shopping_cart(購物車功能)
        9  check_shop_cart(查看購物車)
        10 logout(註銷)
        11 admin_sys(管理員功能)
        """)
        choice = input('please choice func to operate>>>: ').strip()
        if choice == 'q':
            break
        if choice not in my_dict:
            print('相關功能還在開發中')
            time.sleep(1)
            continue
        my_dict.get(choice)()

文件數據操做

# ATM/db/db_handler.py 

import json
import os
from ATM.conf import settings
#保存用戶數據
def save(user_dic):
    user_path = f'{settings.db_path}/{user_dic["user"]}.json'
    with open(user_path, 'w', encoding='utf-8') as f:
        #中文不轉碼
        json.dump(user_dic, f,ensure_ascii=False)
        f.flush()
#查看用戶數據
def select(user):
    user_path = f'{settings.db_path}/{user}.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

接口文件

用戶相關接口

# ATM/interface/user_interface

from ATM.db import db_handler
from ATM.lib import comman
import os

user_logger = comman.get_logger('user')
#註冊接口
def register_interface(user,pwd,balance=15000):
    pwd = comman.get_md5(pwd)
    user_dic = {
        'user': user,
        'pwd': pwd,
        'balance': balance,
        'flow':[],
        'shop_cart':{},
        'lock':False
    }
    db_handler.save(user_dic)
    user_logger.info(f'{user_dic["user"]}用戶註冊成功!')
    return f'{user_dic["user"]}用戶註冊成功!'
#查看用戶接口
def check_user_interface(user):
    #調用數據處理層的select函數,查看用戶信息
    user_dic = db_handler.select(user)
    if user_dic:
        return True
#登陸接口
def login_interface(user,pwd):
    pwd = comman.get_md5(pwd)
    user_dic = db_handler.select(user)
    if user_dic['lock']:
        user_logger.info(f'用戶{user}已經被凍結!')
        return False,f'用戶{user}已經被凍結!'
    if user_dic['pwd'] == pwd:
        user_logger.info(f'用戶{user}登陸成功')
        return True,f'用戶{user}登陸成功'
    user_logger.info(f'用戶{user}登陸失敗,密碼錯誤')
    return False,f'用戶{user}登陸失敗,密碼錯誤'
#查看餘額接口
def check_balance_interface(user):
    user_dic = db_handler.select(user)
    return user_dic['balance']
#註銷接口
def logout_interface(user):
    from ATM.core import src
    src.user_info['user'] = None
    user_logger.info(f'用戶{user}註銷成功!歡迎下次使用')
    return True,f'用戶{user}註銷成功!歡迎下次使用'

商店相關接口

from ATM.db import db_handler
from ATM.interface import bank_interface
from ATM.lib import comman

user_logger = comman.get_logger('shop')
#商城結帳接口
def buy_shop_interface(user,cost):
    flag,msg = bank_interface.pay_interface(user,cost)
    if flag:
        user_logger.info(f'用戶{user}購物成功,餘額爲{msg}')
        return True,f'購物成功,餘額爲{msg}'
    else:
        user_logger.info(f'用戶{user}餘額爲{msg},餘額不足,購物失敗!')
        return False,f'餘額爲{msg},餘額不足,購物失敗!'
#添加購物車接口
def add_shop_cart_interface(user,shop_cart):
    user_dic = db_handler.select(user)
    old_cart = user_dic['shop_cart']
    #循環當前購物車
    for shop in shop_cart:
        if shop in old_cart:
            #把回購的商品信息添加到原來的購物車
            old_cart[shop] += shop_cart[shop]
        else:
            #把新商品添加到購物車
            old_cart[shop] = shop_cart[shop]
    user_dic['shop_cart'].update(old_cart)
    db_handler.save(user_dic)
    user_logger.info(f'用戶{user}添加購物車成功')
    return True,'添加購物車成功'
#查看購物車接口
def check_shop_cart_interface(user):
    user_dic = db_handler.select(user)
    user_logger.info(f'用戶{user}查看購物車')
    return user_dic['shop_cart']

銀行相關接口

from ATM.db import db_handler
from ATM.lib import comman

user_logger = comman.get_logger('bank')
#提現接口
def withdraw_interface(user,money):
    user_dic = db_handler.select(user)
    if user_dic['balance'] >= money*1.05:
        user_dic['balance']-=money*1.05
        msg = f'用戶{user}提現[{money}]元成功!'
        user_dic['flow'].append(msg)
        db_handler.save(user_dic)
        user_logger.info(f'用戶{user}提現[{money}]元成功!')
        return True,msg
    user_logger.info(f'用戶{user}提現失敗,餘額不足...')
    return False,'餘額不足...'
#還款接口
def repay_interface(user,money):
    user_dic = db_handler.select(user)
    user_dic['balance'] +=money
    msg = f'用戶{user},還款{money}成功'
    user_dic['flow'].append(msg)
    db_handler.save(user_dic)
    user_logger.info(f'用戶{user},還款{money}成功')
    return True,msg
#轉帳接口
def transfer_interface(to_user,from_user,money):
    to_user_dic = db_handler.select(to_user)
    from_user_dic = db_handler.select(from_user)
    if from_user_dic['balance'] >=money:
        #金額操做
        from_user_dic['balance'] -=money
        to_user_dic['balance'] +=money
        #拼接流水
        to_user_flow = f'{to_user}用戶接受{from_user}用戶轉帳{money}元'
        from_user_flow = f'{from_user}用戶給{to_user}用戶轉帳{money}元'
        #記錄流水
        from_user_dic['flow'].append(from_user_flow)
        to_user_dic['flow'].append(to_user_flow)
        #保存用戶餘額信息
        db_handler.save(from_user_dic)
        db_handler.save(to_user_dic)
        return True,from_user_flow
    user_logger.info(f'用戶{from_user}轉帳失敗,餘額不足,請充值!')
    return False,'餘額不足,請充值!'
#查看流水接口
def check_flow_interface(user):
    user_dic = db_handler.select(user)
    user_logger.info(f'用戶{user}查看了銀行流水')
    return user_dic['flow']
#銀行支付接口
def pay_interface(user,cost):
    user_dic = db_handler.select(user)
    if user_dic['balance'] >= cost:
        user_dic['balance'] -=cost
        user_dic['flow'].append(f'{user}用戶支付{cost}成功')
        db_handler.save(user_dic)
        user_logger.info(f'用戶{user}支付{cost}成功')
        return True,user_dic['balance']
    else:
        user_logger.info(f'用戶{user}支付{cost}失敗')
        return False,user_dic['balance']
if __name__ == '__main__':
    res = pay_interface('syy',1)
    print(res)

管理員相關接口

from ATM.db import db_handler
from ATM.lib import comman

user_logger = comman.get_logger('admin')
#凍結帳戶接口
def lock_interface(user):
    user_dic = db_handler.select(user)
    user_dic['lock'] = True
    db_handler.save(user_dic)
    user_logger.info(f'用戶{user}凍結成功')
    return f'用戶{user}凍結成功'
#解凍帳戶接口
def unlock_interface(user):
    user_dic = db_handler.select(user)
    user_dic['lock'] = False
    db_handler.save(user_dic)
    user_logger.info(f'用戶{user}解凍成功')
    return f'用戶{user}解凍成功'
#修改餘額接口
def change_limit_interface(user,limit):
    user_dic = db_handler.select(user)
    user_dic['balance'] = limit
    db_handler.save(user_dic)
    user_logger.info(f'用戶{user}修改餘額成功')
    return f'用戶{user}修改額度成功'

通用文件

#ATM/lib/comman.py

import hashlib
import logging.config
from ATM.conf import settings

#登陸認證
def login_auth(func):
    from core import src
    def inner(*args,**kwargs):
        #判斷用戶是否登陸,是則執行被裝飾函數,不然要求登陸
        if src.user_info['user']:
            res = func(*args,**kwargs)
            return res
        else:
            print('請先登陸...')
            src.login()
    return inner

#md5加密
def get_md5(pwd):
    val = '加鹽'
    md5 = hashlib.md5()
    md5.update(val.encode('utf-8'))
    md5.update(pwd.encode('utf-8'))
    res = md5.hexdigest()
    return res

#日誌功能
def get_logger(type_name):
    logging.config.dictConfig(settings.LOGGING_DIC)
    logger = logging.getLogger(type_name)
    return logger

if __name__ == '__main__':
    logger = get_logger('syy')
    logger.info('哈哈哈哈哈哈哈')
相關文章
相關標籤/搜索