作业需求
1、额度 15000或自定义
2、实现购物商城，买东西加入 购物车，调用信用卡接口结账
3、可以提现，手续费5%
4、支持多账户登录
5、支持账户间转账
6、记录每月日常消费流水
7、提供还款接口
8、ATM记录操作日志
9、提供管理接口，包括添加账户、用户额度，冻结账户等。。。
10、用户认证用装饰器
需求9、10没做
程序结构
主程序ATM
只是一个入口
知识点
获取当前文件路径：os.path.abspath(__file__)
获取当前文件的文件夹路径：
os.path.dirname(os.path.abspath(__file__))
import os import sys dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) #找到路径 sys.path.insert(0, dir) #添加路径 print(dir) #将main.py里面所有代码封装成main变量 from core import main if __name__ == "__main__": #这里我刚开始用run()就爆错了 main.run()
重要程序放在mian中：
在run中进行登陆，认证用户，然后用交互程序调用程序（扩展性更好）
""" 主逻辑交互模块 """ from core import auth from core import log from core import transaction from core import account import sys #用户数据信息 user_data = { 'account_id':None, #帐号ID 'is_authenticated':False, #是否认证 'account_data':None #帐号数据 } #调用log文件下的log方法，返回日志对象 access_logger = log.log("access") def account_info(acc_data): """ acc_data:包括ID，is_authenticaed,用户帐号信息 查看用户帐户信息 :return: """ print(acc_data) def repay(acc_data): """ acc_data:包括ID，is_authenticaed,用户帐号信息 还款功能 :return: """ print(acc_data) #调用account模块的load_account方法，从数据库从load出用户信息 account_data = account.load_account(acc_data["id"]) print(account_data) current_balance = """ -------------BALANCE INFO-------------- Credit:%s Balance:%s """ % (account_data["credit"], account_data["balance"]) print(current_balance) back_flag = False while not back_flag: repay_amount = input(" 33[31;1mInput repay amount(b=back): 33[0m").strip() #如果用户输入整型数字 if len(repay_amount) > 0 and repay_amount.isdigit(): #调用transaction模块的方法,参数分别是用户帐户信息，交易类型，交易金额 new_account_data = transaction.make_transaction(account_data, "repay", repay_amount) if new_account_data: print(" 33[42;1mNew Balance:%s 33[0m" % new_account_data["balance"]) else: print(" 33[31;1m%s is not valid amount,Only accept interger! 33[0m" % repay_amount) if repay_amount =="b" or repay_amount == "back": back_flag = True def withdraw(acc_data): """ 取款功能 :return: """ account_data = account.load_account(acc_data["id"]) print(account_data) current_balance = """ -------------BALANCE INFO-------------- Credit:%s Balance:%s """ % (account_data["credit"], account_data["balance"]) print(current_balance) back_flag = False while not back_flag: withdraw_amount = input(" 33[31;1mInput withdraw amount(b=back): 33[0m").strip() #如果用户输入整型数字 if len(withdraw_amount) > 0 and withdraw_amount.isdigit(): #调用transaction模块的方法,参数分别是用户帐户信息，交易类型，交易金额 new_account_data = transaction.make_transaction(account_data, "withdraw", withdraw_amount) if new_account_data: print(" 33[42;1mNew Balance:%s 33[0m" % new_account_data["balance"]) else: print(" 33[31;1m%s is not valid amount,Only accept interger! 33[0m" % withdraw_amount) if withdraw_amount =="b" or withdraw_amount == "back": back_flag = True def transfer(acc_data): """ 转帐 :return: """ account_data = account.load_account(acc_data["id"]) current_balance = """ -------------BALANCE INFO-------------- Credit:%s Balance:%s """ % (account_data["credit"], account_data["balance"]) print(current_balance) back_flag = False while not back_flag: #转帐给的帐户 transfer_name = input(" 33[31;1mInput the account you transfer to: 33[0m").strip() transfer_amount = input(" 33[31;1mInput transfer amount(b=back): 33[0m").strip() #如果用户输入整型数字 if len(transfer_amount) > 0 and transfer_amount.isdigit(): #若用户存在，将用户信息读出 transfer_data = auth.auth_exist(transfer_name) if transfer_data: #调用transaction模块的方法,参数分别是用户帐户信息，交易类型，交易金额 new_account_data = transaction.make_transaction(account_data, "transfer_out", transfer_amount) new_transfer_data = transaction.make_transaction(transfer_data, "repay", transfer_amount) if new_account_data: print(" 33[31;1mTransfer to [%s]success 33[0m" % new_transfer_data["id"]) print(" 33[42;1mNew Balance:%s 33[0m" % new_account_data["balance"]) else: print(" 33[31;1m%s is not valid amount,Only accept interger! 33[0m" % transfer_amount) if transfer_amount =="b" or transfer_amount == "back": back_flag = True def paycheck(): """ 转帐检查 :return: """ pass def logout(acc_data): """ 退出登陆 :return: """ sys.exit() def interactive(acc_data): """ 用户交互 :return: """ msg = ( """ -------------ZhangChengLiang Bank--------------- 33[31;1m 1. 账户信息 2. 存款 3. 取款 4. 转账 5. 账单 6. 退出 33[0m""" ) menu_dic = { "1":account_info, "2":repay, "3":withdraw, "4":transfer, "5":paycheck, "6":logout } flag = False while not flag: print(msg) choice = input("<<<:").strip() if choice in menu_dic: #很重要！！省了很多代码，不用像之前一个一个判断！ menu_dic[choice](acc_data) else: print(" 33[31;1mYou choice doesn't exist! 33[0m") def run(): """ 当程序启动时调用，用于实现主要交互逻辑 :return: """ # 调用认证模块,返回用户文件json.load后的字典,传入access_logger日志对象 access_data = auth.access_login(user_data, access_logger) print("AA") if user_data["is_authenticated"]: #如果用户认证成功 print("has authenticated") #将用户文件的字典赋给user_data["account_data"] user_data["account_data"] = access_data interactive(user_data) #用户交互开始
配置文件
""" 初始化的配置 """ import logging import os #到ATM目录，方便后面创建帐户文件 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) LOGIN_LEVEL = logging.INFO #初始化日志级别 LOGIN_TYPE = { "access":"access.log" } DATABASE = { "db_tool":"file_storage", #文件存储，这里可拓展成数据库形式的 "name":"accounts", #db下的文件名 "path":"%s/db" % BASE_DIR } #用户交易类型，每个类型对应一个字典，包括帐户金额变动方式，利息 TRANSACTION_TYPE = { "repay":{"action":"plus", "interest":0}, #存款 "withdraw":{"action":"minus", "interest": 0.05}, #取款(提现) "transfer_out":{"action":"minus", "interest": 0.05} #转帐(出) }
用户认证
""" 认证模块 """ print("BB") import os import json import time from core import db_handle from conf import settings from core import account def access_auth(account, password, log_obj): """ 下面的access_login调用access_auth方法，用于登陆 :param acount: 用户名 :param password: 密码 :return:如果未超期，返回字典，超期则打印相应提示 """ db_path = db_handle.handle(settings.DATABASE) #调用db_handle下的handle方法,返回路径/db/accounts print(db_path) account_file = "%s/%s.json" % (db_path, account) #用户文件 print(account_file) if os.path.isfile(account_file): #如果用户文件存在(即用户存在) with open(account_file, "r", encoding="utf-8") as f: #打开文件 account_data = json.load(f) #file_data为字典形式 print(account_data) if account_data["password"] == password: expire_time = time.mktime(time.strptime(account_data["expire_date"], "%Y-%m-%d")) print(expire_time) print(time.strptime(account_data["expire_date"], "%Y-%m-%d")) if time.time() > expire_time: #如果信用卡已超期 log_obj.error("Account [%s] had expired,Please contract the bank" % account) print(" 33[31;1mAccount [%s] had expired,Please contract the bank" % account) else: #信用卡未超期，返回用户数据的字典 print("return") log_obj.info("Account [%s] logging success" % account) return account_data else: log_obj.error("Account or Passworddoes not correct!") print(" 33[31;1mAccount or Passworddoes not correct! 33[0m") else: #用户不存在 log_obj.error("Account [%s] does not exist!" % account) print(" 33[31;1mAccount [%s] does not exist! 33[0m" % account) def access_login(user_data, log_obj): """ 用记登陆，当登陆失败超过三次则退出 :param user_data: main.py里面的字典 :return:若用户帐号密码正确且信用卡未超期，返回用户数据的字典 """ retry = 0 while not user_data["is_authenticated"] and retry < 3: account = input("Account:").strip() password = input("Password:").strip() #用户帐号密码正确且信用卡未超期，返回用户数据的字典 user_auth_data = access_auth(account, password, log_obj) if user_auth_data: user_data["is_authenticated"] = True #用户认证为True user_data["id"] = account #用户帐号ID为帐号名 print("welcome") return user_auth_data retry += 1 #登陆和信用卡认证出错，则次数加1 else: #若次数超过三次，打印相关信息并退出 print("Account [%s] try logging too many times..." % account) log_obj.error("Account [%s] try logging too many times..." % account) exit() def auth_exist(account_name): """ 用来判断用户是否存在(转帐时可用),存在则将用户信息读出 :param account_name: :return: """ db_path = db_handle.handle(settings.DATABASE) #调用db_handle下的handle方法,返回路径/db/accounts print(db_path) account_file = "%s/%s.json" % (db_path, account_name) #用户文件 print(account_file) if os.path.isfile(account_file): #如果用户文件存在(即用户存在) account_data = account.load_account(account_name) print(account_data) return account_data else: print(" 33[31;1mError:[%s] is not exist! 33[0m" % account_name) return False
数据的使用，包含文件和数据库两种
""" 处理与数据库的交互，若是file_db_storage,返回路径 """ def file_db_handle(database): """ 数据存在文件 :param database: :return: 返回路径 ATM/db/accounts """ db_path = "%s/%s" % (database["path"], database["name"]) print(db_path) return db_path def mysql_db_handle(database): """ 处理mysql数据库，这里用文件来存数据， 保留这个方法主要为了程序拓展性 :return: """ pass def handle(database): """ 对某种数据库形式处于是 本程序用的是文件处理file_storage :param database: settings里面的DATABASE :return: 返回路径 """ if database["db_tool"] == "file_storage": return file_db_handle(database) if database["db_tool"] == "mysql": return mysql_db_handle(database)
为保证用户信息的实时性，用户数据需多次获取存入，因此单独写读写程序
""" 用于处理用户信息的load or dump 每进行一个操作就将信息更新到数据库 """ from core import db_handle from conf import settings import json def load_account(account_id): """ 将用户信息从文件中load出来 :return: 用户信息的字典 """ #返回路径 ATM/db/accounts db_path = db_handle.handle(settings.DATABASE) account_file = "%s/%s.json" % (db_path, account_id) with open(account_file, "r", encoding="utf-8") as f: account_data = json.load(f) return account_data def dump_account(account_data): """ 将已更改的用户信息更新到用户文件 :param account_data: 每操作后用户的信息 :return: """ db_path = db_handle.handle(settings.DATABASE) account_file = "%s/%s.json" % (db_path, account_data["id"]) with open(account_file, "w", encoding="utf-8") as f: json.dump(account_data, f) print("dump success")
还有log
import logging from conf import settings def log(logging_type): """ main模块调用access_logger = log.log("access") :param logging_type: "access" :return: 返回logger日志对象 """ logger = logging.getLogger(logging_type) #传日志用例，生成日志对象 logger.setLevel(settings.LOGIN_LEVEL) #设置日志级别 ch = logging.StreamHandler() #日志打印到屏幕，获取对象 ch.setLevel(settings.LOGIN_LEVEL) # 获取文件日志对象及日志文件 log_file = "%s/log/%s" % (settings.BASE_DIR, settings.LOGIN_TYPE[logging_type]) fh = logging.FileHandler(log_file) fh.setLevel(settings.LOGIN_LEVEL) #日志格式 formatter = logging.Formatter("%(asctime)s-%(name)s-%(levelname)s-%(message)s") #输出格式 ch.setFormatter(formatter) fh.setFormatter(formatter) #把日志打印到指定的handler logger.addHandler(ch) logger.addHandler(fh) return logger #log方法返回logger对象 # logger.debug('debugmessage') # logger.info('infomessage') # logger.warn('warnmessage') # logger.error('errormessage') # logger.critical('criticalmessage')
万事儿开头难，实践出真知