#一個項目是怎麼從無到有的 #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('哈哈哈哈哈哈哈')