自动吐钱器——ATM机
bin
＃！_ * _ coding：utf-8 _ * _
#__author__：“lucky”
#!_*_coding:utf-8_*_
#__author__:"Alex Li"
import os
import sys
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
print(base_dir)
sys.path.append(base_dir)
from core import main
if __name__ == ‘__main__‘:
main.run()
conf
＃！_ * _ coding：utf-8 _ * _
#__author__：“lucky”
#!_*_coding:utf-8_*_
#__author__:"Alex Li"
import os
import sys
import logging
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATABASE = {
‘engine‘: ‘file_storage‘, #support mysql,postgresql in the future
‘name‘:‘accounts‘,
‘path‘: "%s/db" % BASE_DIR
}
LOG_LEVEL = logging.INFO
LOG_TYPES = {
‘transaction‘: ‘transactions.log‘,
‘access‘: ‘access.log‘,
}
TRANSACTION_TYPE = {
‘repay‘:{‘action‘:‘plus‘, ‘interest‘:0},
‘withdraw‘:{‘action‘:‘minus‘, ‘interest‘:0.05},
‘transfer‘:{‘action‘:‘minus‘, ‘interest‘:0.05},
‘consume‘:{‘action‘:‘minus‘, ‘interest‘:0},
}
core
＃！_ * _ coding：utf-8 _ * _
#__author__：“lucky”
#!_*_coding:utf-8_*_
#__author__:"lucky"
import json
import time
from core import db_handler
from conf import settings
def load_current_balance(account_id):
‘‘‘
return account balance and other basic info
:param account_id:
:return:
‘‘‘
# db_path = db_handler.db_handler(settings.DATABASE)
# account_file = "%s/%s.json" %(db_path,account_id)
#
db_api = db_handler.db_handler()
data = db_api("select * from accounts where account=%s" % account_id)
return data
# with open(account_file) as f:
# acc_data = json.load(f)
# return acc_data
def dump_account(account_data):
‘‘‘
after updated transaction or account data , dump it back to file db
:param account_data:
:return:
‘‘‘
db_api = db_handler.db_handler()
data = db_api("update accounts where account=%s" % account_data[‘id‘],account_data=account_data)
# db_path = db_handler.db_handler(settings.DATABASE)
# account_file = "%s/%s.json" %(db_path,account_data[‘id‘])
# with open(account_file, ‘w‘) as f:
# acc_data = json.dump(account_data,f)
return True
#!_*_coding:utf-8_*_
#__author__:"lucky"
import os
from core import db_handler
from conf import settings
from core import logger
import json
import time
def login_required(func):
"验证用户是否登录"
def wrapper(*args,**kwargs):
#print(‘--wrapper--->‘,args,kwargs)
if args[0].get(‘is_authenticated‘):
return func(*args,**kwargs)
else:
exit("User is not authenticated.")
return wrapper
def acc_auth(account,password):
‘‘‘
account auth func
:param account: credit account number
:param password: credit card password
:return: if passed the authentication , retun the account object, otherwise ,return None
‘‘‘
db_path = db_handler.db_handler(settings.DATABASE)
account_file = "%s/%s.json" %(db_path,account)
print(account_file)
if os.path.isfile(account_file):
with open(account_file,‘r‘) as f:
account_data = json.load(f)
if account_data[‘password‘] == password:
exp_time_stamp = time.mktime(time.strptime(account_data[‘expire_date‘], "%Y-%m-%d"))
if time.time() >exp_time_stamp:
print("\033[31;1mAccount [%s] has expired,please contact the back to get a new card!\033[0m" % account)
else: #passed the authentication
return account_data
else:
print("\033[31;1mAccount ID or password is incorrect!\033[0m")
else:
print("\033[31;1mAccount [%s] does not exist!\033[0m" % account)
def acc_auth2(account,password):
‘‘‘
优化版认证接口
:param account: credit account number
:param password: credit card password
:return: if passed the authentication , retun the account object, otherwise ,return None
‘‘‘
db_api = db_handler.db_handler()
data = db_api("select * from accounts where account=%s" % account)
if data[‘password‘] == password:
exp_time_stamp = time.mktime(time.strptime(data[‘expire_date‘], "%Y-%m-%d"))
if time.time() > exp_time_stamp:
print("\033[31;1mAccount [%s] has expired,please contact the back to get a new card!\033[0m" % account)
else: # passed the authentication
return data
else:
print("\033[31;1mAccount ID or password is incorrect!\033[0m")
def acc_login(user_data,log_obj):
‘‘‘
account login func
:user_data: user info data , only saves in memory
:return:
‘‘‘
retry_count = 0
while user_data[‘is_authenticated‘] is not True and retry_count < 3 :
account = input("\033[32;1maccount:\033[0m").strip()
password = input("\033[32;1mpassword:\033[0m").strip()
auth = acc_auth2(account, password)
if auth: #not None means passed the authentication
user_data[‘is_authenticated‘] = True
user_data[‘account_id‘] = account
#print("welcome")
return auth
retry_count +=1
else:
log_obj.error("account [%s] too many login attempts" % account)
exit()
#!_*_coding:utf-8_*_
#__author__:"lucky"
‘‘‘
handle all the database interactions
‘‘‘
import json,time ,os
from conf import settings
def file_db_handle(conn_params):
‘‘‘
parse the db file path
:param conn_params: the db connection params set in settings
:return:
‘‘‘
print(‘file db:‘,conn_params)
#db_path =‘%s/%s‘ %(conn_params[‘path‘],conn_params[‘name‘])
return file_execute
def db_handler():
‘‘‘
connect to db
:param conn_parms: the db connection params set in settings
:return:a
‘‘‘
conn_params = settings.DATABASE
if conn_params[‘engine‘] == ‘file_storage‘:
return file_db_handle(conn_params)
elif conn_params[‘engine‘] == ‘mysql‘:
pass #todo
def file_execute(sql,**kwargs):
conn_params = settings.DATABASE
db_path = ‘%s/%s‘ % (conn_params[‘path‘], conn_params[‘name‘])
print(sql,db_path)
sql_list = sql.split("where")
print(sql_list)
if sql_list[0].startswith("select") and len(sql_list)> 1:#has where clause
column,val = sql_list[1].strip().split("=")
if column == ‘account‘:
account_file = "%s/%s.json" % (db_path, val)
print(account_file)
if os.path.isfile(account_file):
with open(account_file, ‘r‘) as f:
account_data = json.load(f)
return account_data
else:
exit("\033[31;1mAccount [%s] does not exist!\033[0m" % val )
elif sql_list[0].startswith("update") and len(sql_list)> 1:#has where clause
column, val = sql_list[1].strip().split("=")
if column == ‘account‘:
account_file = "%s/%s.json" % (db_path, val)
#print(account_file)
if os.path.isfile(account_file):
account_data = kwargs.get("account_data")
with open(account_file, ‘w‘) as f:
acc_data = json.dump(account_data, f)
return True
#!_*_coding:utf-8_*_
#__author__:"lucky"
‘‘‘
handle all the logging works
‘‘‘
import logging
from conf import settings
def logger(log_type):
#create logger
logger = logging.getLogger(log_type)
logger.setLevel(settings.LOG_LEVEL)
# create console handler and set level to debug
ch = logging.StreamHandler()
ch.setLevel(settings.LOG_LEVEL)
# create file handler and set level to warning
log_file = "%s/log/%s" %(settings.BASE_DIR, settings.LOG_TYPES[log_type])
fh = logging.FileHandler(log_file)
fh.setLevel(settings.LOG_LEVEL)
# create formatter
formatter = logging.Formatter(‘%(asctime)s - %(name)s - %(levelname)s - %(message)s‘)
# add formatter to ch and fh
ch.setFormatter(formatter)
fh.setFormatter(formatter)
# add ch and fh to logger
logger.addHandler(ch)
logger.addHandler(fh)
return logger
# ‘application‘ code
‘‘‘logger.debug(‘debug message‘)
logger.info(‘info message‘)
logger.warn(‘warn message‘)
logger.error(‘error message‘)
logger.critical(‘critical message‘)‘‘‘
1 #!_*_coding:utf-8_*_
2 #__author__:"lucky"
3
4 ‘‘‘
5 main program handle module , handle all the user interaction stuff
6 ‘‘‘
7
8 from core import auth
9 from core import accounts
10 from core import logger
11 from core import accounts
12 from core import transaction
13 from core.auth import login_required
14 import time
15
16 #transaction logger
17 trans_logger = logger.logger(‘transaction‘)
18 #access logger
19 access_logger = logger.logger(‘access‘)
20
21
22 #temp account data ,only saves the data in memory
23 user_data = {
24 ‘account_id‘:None,
25 ‘is_authenticated‘:False,
26 ‘account_data‘:None
27
28 }
29
30 def account_info(acc_data):
31 print(user_data)
32
33 @login_required
34 def repay(acc_data):
35 ‘‘‘
36 print current balance and let user repay the bill
37 :return:
38 ‘‘‘
39 account_data = accounts.load_current_balance(acc_data[‘account_id‘])
40 #for k,v in account_data.items():
41 # print(k,v )
42 current_balance= ‘‘‘ --------- BALANCE INFO --------
43 Credit : %s
44 Balance: %s‘‘‘ %(account_data[‘credit‘],account_data[‘balance‘])
45 print(current_balance)
46 back_flag = False
47 while not back_flag:
48 repay_amount = input("\033[33;1mInput repay amount:\033[0m").strip()
49 if len(repay_amount) >0 and repay_amount.isdigit():
50 print(‘ddd 00‘)
51 new_balance = transaction.make_transaction(trans_logger,account_data,‘repay‘, repay_amount)
52 if new_balance:
53 print(‘‘‘\033[42;1mNew Balance:%s\033[0m‘‘‘ %(new_balance[‘balance‘]))
54
55 else:
56 print(‘\033[31;1m[%s] is not a valid amount, only accept integer!\033[0m‘ % repay_amount)
57
58 if repay_amount == ‘b‘:
59 back_flag = True
60 def withdraw(acc_data):
61 ‘‘‘
62 print current balance and let user do the withdraw action
63 :param acc_data:
64 :return:
65 ‘‘‘
66 account_data = accounts.load_current_balance(acc_data[‘account_id‘])
67 current_balance= ‘‘‘ --------- BALANCE INFO --------
68 Credit : %s
69 Balance: %s‘‘‘ %(account_data[‘credit‘],account_data[‘balance‘])
70 print(current_balance)
71 back_flag = False
72 while not back_flag:
73 withdraw_amount = input("\033[33;1mInput withdraw amount:\033[0m").strip()
74 if len(withdraw_amount) >0 and withdraw_amount.isdigit():
75 new_balance = transaction.make_transaction(trans_logger,account_data,‘withdraw‘, withdraw_amount)
76 if new_balance:
77 print(‘‘‘\033[42;1mNew Balance:%s\033[0m‘‘‘ %(new_balance[‘balance‘]))
78
79 else:
80 print(‘\033[31;1m[%s] is not a valid amount, only accept integer!\033[0m‘ % withdraw_amount)
81
82 if withdraw_amount == ‘b‘:
83 back_flag = True
84
85 def transfer(acc_data):
86 pass
87 def pay_check(acc_data):
88 pass
89 def logout(acc_data):
90 pass
91 def interactive(acc_data):
92 ‘‘‘
93 interact with user
94 :return:
95 ‘‘‘
96 menu = u‘‘‘
97 ------- Oldboy Bank ---------
98 \033[32;1m1. 账户信息
99 2. 还款(功能已实现)
100 3. 取款(功能已实现)
101 4. 转账
102 5. 账单
103 6. 退出
104 \033[0m‘‘‘
105 menu_dic = {
106 ‘1‘: account_info,
107 ‘2‘: repay,
108 ‘3‘: withdraw,
109 ‘4‘: transfer,
110 ‘5‘: pay_check,
111 ‘6‘: logout,
112 }
113 exit_flag = False
114 while not exit_flag:
115 print(menu)
116 user_option = input(">>:").strip()
117 if user_option in menu_dic:
118 print(‘accdata‘,acc_data)
119 #acc_data[‘is_authenticated‘] = False
120 menu_dic[user_option](acc_data)
121
122 else:
123 print("\033[31;1mOption does not exist!\033[0m")
124 def run():
125 ‘‘‘
126 this function will be called right a way when the program started, here handles the user interaction stuff
127 :return:
128 ‘‘‘
129 acc_data = auth.acc_login(user_data,access_logger)
130 if user_data[‘is_authenticated‘]:
131 user_data[‘account_data‘] = acc_data
132 interactive(user_data)
1 #!_*_coding:utf-8_*_
2 #__author__:"lucky"
3
4 from conf import settings
5 from core import accounts
6 from core import logger
7 #transaction logger
8
9
10
11 def make_transaction(log_obj,account_data,tran_type,amount,**others):
12 ‘‘‘
13 deal all the user transactions
14 :param account_data: user account data
15 :param tran_type: transaction type
16 :param amount: transaction amount
17 :param others: mainly for logging usage
18 :return:
19 ‘‘‘
20 amount = float(amount)
21 if tran_type in settings.TRANSACTION_TYPE:
22
23 interest = amount * settings.TRANSACTION_TYPE[tran_type][‘interest‘]
24 old_balance = account_data[‘balance‘]
25 if settings.TRANSACTION_TYPE[tran_type][‘action‘] == ‘plus‘:
26 new_balance = old_balance + amount + interest
27 elif settings.TRANSACTION_TYPE[tran_type][‘action‘] == ‘minus‘:
28 new_balance = old_balance - amount - interest
29 #check credit
30 if new_balance <0:
31 print(‘‘‘\033[31;1mYour credit [%s] is not enough for this transaction [-%s], your current balance is
32 [%s]‘‘‘ %(account_data[‘credit‘],(amount + interest), old_balance ))
33 return
34 account_data[‘balance‘] = new_balance
35 accounts.dump_account(account_data) #save the new balance back to file
36 log_obj.info("account:%s action:%s amount:%s interest:%s" %
37 (account_data[‘id‘], tran_type, amount,interest) )
38 return account_data
39 else:
40 print("\033[31;1mTransaction type [%s] is not exist!\033[0m" % tran_type)
db
accounts文件夹:
里面放一些需要存储的数据
#!_*_coding:utf-8_*_
#__author__:"lucky"
#!_*_coding:utf-8_*_
#__author__:"lucky"
import json
acc_dic = {
‘id‘: 1234,
‘password‘: ‘abc‘,
‘credit‘: 15000,
‘balance‘: 15000,
‘enroll_date‘: ‘2020-04-01‘,
‘expire_date‘: ‘2025-03-31‘,
‘pay_day‘: 22,
‘status‘: 0 # 0 = normal, 1 = locked, 2 = disabled
}
print(json.dumps(acc_dic))
log
#!_*_coding:utf-8_*_
#__author__:"lucky"
会记录用户的一些行为,比如存款或者消费
readme
程序介绍:
实现ATM常用功能
功能全部用python的基础知识实现,用到了time\os\sys\json\open\logging\函数\模块知识, 主要帮给大家一个简单的模块化编程的示例
注意:只实现了"还款"和"取现功能"
程序结构:
day5-atm/
├── README
├── atm #ATM主程目录
│ ├── __init__.py
│ ├── bin #ATM 执行文件 目录
│ │ ├── __init__.py
│ │ ├── atm.py #ATM 执行程序
│ │ └── manage.py #ATM 管理端,未实现
│ ├── conf #配置文件
│ │ ├── __init__.py
│ │ └── settings.py
│ ├── core #主要程序逻辑都 在这个目录 里
│ │ ├── __init__.py
│ │ ├── accounts.py #用于从文件里加载和存储账户数据
│ │ ├── auth.py #用户认证模块
│ │ ├── db_handler.py #数据库连接引擎
│ │ ├── logger.py #日志记录模块
│ │ ├── main.py #主逻辑交互程序
│ │ └── transaction.py #记账\还钱\取钱等所有的与账户金额相关的操作都 在这
│ ├── db #用户数据存储的地方
│ │ ├── __init__.py
│ │ ├── account_sample.py #生成一个初始的账户数据 ,把这个数据 存成一个 以这个账户id为文件名的文件,放在accounts目录 就行了,程序自己去会这里找
│ │ └── accounts #存各个用户的账户数据 ,一个用户一个文件
│ │ └── 1234.json #一个用户账户示例文件
│ └── log #日志目录
│ ├── __init__.py
│ ├── access.log #用户访问和操作的相关日志
│ └── transactions.log #所有的交易日志
└── shopping_mall #电子商城程序,需单独实现
└── __init__.py
原文：https://www.cnblogs.com/lucky-cat233/p/12616873.html