练习题:
1.logging模块有几个日志级别？
#INFO,WARNING,DEBUG,CRITICAL,ERROR
2.请配置logging模块，使其在屏幕和文件里同时打印以下格式的日志
2017-10-18 15:56:26,613 - access - ERROR - account [1234] too many login attempts
代码
import logging
from logging import handlers
logger = logging.getLogger('access')
logger.setLevel(logging.ERROR)
ch = logging.StreamHandler()
fh = handlers.TimedRotatingFileHandler(interval=1,when='D',filename='web.log', backupCount=10)
logger.addHandler(fh)
logger.addHandler(ch)
file_format = logging.Formatter(fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(file_format)
fh.setFormatter(file_format) logger.error('account [1234] too many login attempts')
3.json、pickle、shelve三个区别是什么？
首先，这三个模块都是序列化工具。
1. json是所有语言的序列化工具,优点跨语言、体积小.只能序列化一些基本的数据类型。int\str\list\tuple\dict
pickle是python语言特有序列化工具，所有数据都能序列化。只能在python中使用，存储数据占空间大.
shelve模块是一个简单的k,v将内存数据通过文件持久化的模块，可以持久化任何pickle可支持的python数据格式。
4.json的作用是什么？
序列化，把dict，tuple，list等内存对象转换为字符串，持久化数据，存储到硬盘中或网络传输，因为网络和硬盘只接受bytes
5.subprocess执行命令方法有几种？
run()方法
call()方法
Popen()方法
6.为什么要设计好目录结构？
1.可读性高: 不熟悉这个项目的代码的人，一眼就能看懂目录结构，知道程序启动脚本是哪个，测试目录在哪儿，配置文件在哪儿等等。从而非常快速的了解这个项目。
2.可维护性高: 定义好组织规则后，维护者就能很明确地知道，新增的哪个文件和代码应该放在什么目录之下。这个好处是，随着时间的推移，代码/配置的规模增加，项目结构不会混乱，仍然能够组织良好。
7.打印出命令行的第一个参数。例如：
python argument.py luffy
打印出 luffy
import sys
print(sys.argv[1])
8.以下代码
'''
Linux当前目录/usr/local/nginx/html/
文件名：index.html
'''
import os
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath('index.html')))
print(BASE_DIR) 打印
/usr/local/nginx/
9.os.path.dirname和os.path.abspath含义是什么？
os.path.dirname：指定文件的目录
os.path.abspath：指定文件的绝对路径
10.通过configparser模块完成以下功能
文件名my.cnf
[DEFAULT] [client]
port = 3306
socket = /data/mysql_3306/mysql.sock [mysqld]
explicit_defaults_for_timestamp = true
port = 3306
socket = /data/mysql_3306/mysql.sock
back_log = 80
basedir = /usr/local/mysql
tmpdir = /tmp
datadir = /data/mysql_3306
default-time-zone = '+8:00'
修改时区 default-time-zone = '+8:00' 为 校准的全球时间 +00:00
删除 explicit_defaults_for_timestamp = true
为DEFAULT增加一条 character-set-server = utf8
import configparser
import os
config = configparser.ConfigParser()
config.read('my.cnf')
new_file = 'my.cnf%s'%('new')
with open(new_file,'w') as new_f: print(config.sections())
print(config)
# 为DEFAULT增加一条 character-set-server = utf8
config['DEFAULT']={'character-set-server':'utf8'}
# 修改时区 default-time-zone = '+8:00' 为 校准的全球时间 +00:00
config.set('mysqld','default-time-zone','+00:00')
# 删除 explicit_defaults_for_timestamp = true
config.remove_option('mysqld','explicit_defaults_for_timestamp')
config.write(new_f)
os.remove('my.cnf')
os.rename(new_file,'my.cnf')
11.写一个6位随机验证码程序（使用random模块),要求验证码中至少包含一个数字、一个小写字母、一个大写字母.
import random
import string code_li = [] code_li.append(random.choice(string.ascii_lowercase))
code_li.append(random.choice(string.digits))
code_li.append(random.choice(string.ascii_uppercase))
while len(code_li) < 6:
code_li.append(random.choice(string.digits+string.ascii_lowercase+string.ascii_uppercase))
print(code_li)
q_code=''.join(code_li)
print(q_code)
12.利用正则表达式提取到 luffycity.com ，内容如下
s=''' <!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>luffycity.com</title>
</head>
<body>
</body>
</html>
''' import re
match_s = re.findall('<title>(.*?)</title>',s)
print(match_s[0])
13.写一个用户登录验证程序，文件如下 1234.json
1234.json
{"expire_date": "2021-01-01", "id": 1234, "status": 0, "pay_day": 22, "password": "abc"}
'''
写一个用户登录验证程序，文件如下
1234.json 用户名为json文件名，密码为 password。
判断是否过期，与expire_date进行对比。
登陆成功后，打印“登陆成功”，三次登陆失败，status值改为1，并且锁定账号。
'''
import json
import os
import time
import hashlib count = 0
exit_flag =False while count < 3:
user = input('输入用户名： ')
f = user.strip()+'.json'
if os.path.exists(f):
fp = open(f,'r+',encoding='utf-8')
j_user = json.load(fp)
if j_user["status"] == 1:
print('账号已经锁定')
break
else:
expire_dt = j_user["expire_date"] current_st = time.time()
expire_st = time.mktime(time.strptime(expire_dt,'%Y-%m-%d'))
# print(expire_st,current_st)
if current_st > expire_st:
print('用户已经过期')
break
else:
while count < 3:
pwd = input('输入密码： ')
if pwd.strip() == j_user["password"]:
print('用户[%s]登录成功'%user)
exit_flag = True
break
else:
print('密码不对')
if count == 2:
print('用户登录已超过3次，锁定账号')
j_user["status"] = 1
fp.seek(0)
fp.truncate() # 清空文件内容
json.dump(j_user,fp) # 写入锁定信息
count += 1
if exit_flag:
break
else:
print('用户不存在')
count += 1
14. 把第13题三次验证的密码进行hashlib加密处理。即：json文件保存为md5的值，然后用md5的值进行验证。
加密密码
f ='1234.json'
fp = open(f,'r+',encoding='utf-8')
j_user = json.load(fp)
md = hashlib.md5()
md.update('abc'.encode('utf-8'))
md_pwd = md.hexdigest()
print(md_pwd)
j_user["password"] = md_pwd
fp.seek(0)
fp.truncate() # 清空文件内容
json.dump(j_user,fp) # 写入md5密码信息
fp.close()
用户登录验证
count = 0
exit_flag = False
md = hashlib.md5() while count < 3:
user = input('输入用户名： ')
f = user.strip()+'.json'
if os.path.exists(f):
fp = open(f,'r+',encoding='utf-8')
j_user = json.load(fp)
if j_user["status"] == 1:
print('账号已经锁定')
break
else:
expire_dt = j_user["expire_date"] current_st = time.time()
expire_st = time.mktime(time.strptime(expire_dt,'%Y-%m-%d'))
# print(expire_st,current_st)
if current_st > expire_st:
print('用户已经过期')
break
else:
while count < 3:
pwd = input('输入密码： ')
md.update(pwd.strip().encode('utf-8'))
md5_pwd = md.hexdigest()
if md5_pwd == j_user["password"]:
print('用户[%s]登录成功'%user)
exit_flag = True
break
else:
print('密码不对')
if count == 2:
print('用户登录已超过3次，锁定账号')
j_user["status"] = 1
fp.seek(0)
fp.truncate() # 清空文件内容
json.dump(j_user,fp) # 写入锁定信息
fp.close() count += 1
if exit_flag:
break
else:
print('用户不存在')
count += 1
15.
最近luffy买了个tesla，通过转账的形式，并且支付了5%的手续费，tesla价格为75万。文件为json，请用程序实现该转账行为。 需求如下：
-
目录结构为
.
├── account
│ ├── luffy.json
│ └── tesla.json
└── bin
└── start.py
当执行start.py时，出现交互窗口
------- Luffy Bank ---------
1. 账户信息
2. 转账
选择1 账户信息 显示luffy的当前账户余额。
选择2 转账 直接扣掉75万和利息费用并且tesla账户增加75万
-
对上题增加一个需求：提现。 目录结构如下
.
├── account
│ └── luffy.json
├── bin
│ └── start.py
└── core
└── withdraw.py
当执行start.py时，出现交互窗口
------- Luffy Bank ---------
1. 账户信息
2. 提现
选择1 账户信息 显示luffy的当前账户余额和信用额度。
选择2 提现 提现金额应小于等于信用额度，利息为5%，提现金额为用户自定义。
尝试把上一章的验证用户登陆的装饰器添加到提现和转账的功能上。
-
对第15题的用户转账、登录、提现操作均通过logging模块记录日志,日志文件位置如下
.
├── account
│ └── luffy.json
├── bin
│ └── start.py
└── core
| └── withdraw.py
└── logs
└── bank.log
目录结构
代码 settings.py
import os '''
日志文件设置
'''
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LOG_LEVEL='INFO'
LOG_FILE='bank.log' '''
交易类型
'''
TRANS_TYPE={
'withdraw':{'interest':0.05,'action':'minus'},
'transfer': {'interest': 0.05, 'action': 'minus'}
} print(BASE_DIR)
代码 my_logset.py
#!/usr/bin/python
# -*- coding: utf-8 -*-
# @Time : 2018/4/23 8:55
# @Author : hyang
# @File : my_logset.py
# @Software: PyCharm import logging
import os
from conf import settings # 日志格式
log_format = '[%(asctime)s - %(levelname)s - %(name)s - %(filename)s - %(lineno)d ] %(message)s ' def get_mylogger(name):
"""
get log
:param name:
:return:
"""
logger = logging.getLogger(name)
logger.setLevel(settings.LOG_LEVEL) console_handler = logging.StreamHandler()
# 文件绝对路径
logfile_path = os.path.join(settings.BASE_DIR, 'log',settings.LOG_FILE)
file_handler = logging.FileHandler(logfile_path) logger.addHandler(console_handler)
logger.addHandler(file_handler) file_format = logging.Formatter(fmt=log_format)
console_format = logging.Formatter(fmt=log_format, datefmt='%Y-%m-%d %H:%M:%S ') console_handler.setFormatter(console_format)
file_handler.setFormatter(file_format) return logger if __name__ == '__main__':
log = get_mylogger('access')
log.info('access')
log.error('Error') log1 = get_mylogger('trans')
log1.info('trans')
代码 auth.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018/5/14 22:57
# @Author : hyang
# @Site :
# @File : auth.py
# @Software: PyCharm import json
import time
import os
from conf import settings
from log import my_logset
from functools import wraps # logger = my_logset.get_mylogger('access') def login_required(func):
"""
登录认证装饰器
:param func:
:return:
"""
@wraps(func)
def wrapper(*args,**kwargs):
# print(args[0].get('is_authenticated'))
if args[0].get('is_authenticated'):
print('execute %s'%func.__name__)
res = func(*args, **kwargs)
else:
exit('user is not authenticated')
return res return wrapper def acc_login(user, pwd,logger):
"""
用户登录
:param log_obj:
:return:
"""
# 账号文件
account_file = os.path.join(settings.BASE_DIR,'account','%s.json'%user)
if os.path.isfile(account_file):
user_fp = open(account_file,'r',encoding='utf-8')
account_data = json.load(fp=user_fp)
if account_data["password"] == pwd:
exp_time_stamp = time.mktime(time.strptime(account_data["expire_date"],'%Y-%m-%d'))
status = account_data['status']
if time.time() > exp_time_stamp:
msg = 'Account [%s] has expired,please contact the back to get a new card!' % user
logger.error(msg) elif status != 0:
msg = 'Account [%s] has frozen' % user
logger.error(msg) else:
logger.info('***********欢迎{}登录***********'.format(user))
return account_data
else:
logger.error("Account ID or password is incorrect!")
else:
msg = "Account [%s] does not exist!" % user
logger.error(msg) def auth_acc(user_data,logger):
"""
用户登录
:return:
"""
retry_count = 0
if user_data['is_authenticated'] is not True:
while retry_count < 3:
user = input('input username: ').strip()
pwd = input('input password: ').strip() acc_data = acc_login(user, pwd,logger)
if acc_data:
user_data['is_authenticated'] = True
return acc_data retry_count +=1
else:
exit("account too many login attempts" ) if __name__ == '__main__':
logger = my_logset.get_mylogger('access')
user_data = {'is_authenticated':False}
auth_acc(user_data,logger)
代码 main.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018/5/14 23:55
# @Author : hyang
# @Site :
# @File : main.py
# @Software: PyCharm import os
import json
from core import auth
from core.auth import login_required
from conf import settings
from log import my_logset user_data = {
'user_id':None,
'is_authenticated':False,
'user':None
} # transaction logger
trans_logger = my_logset.get_mylogger('transaction')
# access logger
access_logger = my_logset.get_mylogger('access') @login_required
def account_info(user_data):
"""
print acoount_info
:param acc_data:
:return:
"""
curr_data = user_data['user']
current_user = '''
--------- USER INFO --------
user_id: {}
Credit : {}
Balance: {}
expire_date: {}
----------------------------------
'''.format(curr_data['id'], curr_data['credit'], curr_data['balance'],curr_data['expire_date'])
access_logger.info(current_user) @login_required
def withdraw(user_data):
""" print current balance and let user do the withdraw action
:param user_data:
:return:
"""
curr_data = user_data['user']
back_flag = False
while not back_flag:
withdraw_amount = input("Input withdraw amount:").strip()
if withdraw_amount.isdigit():
old_bal = curr_data['balance']
curr_data['balance'] = old_bal - float(withdraw_amount)*settings.TRANS_TYPE['withdraw']['interest']
if curr_data['balance']:
msg = "New Balance:%s" % (curr_data['balance'])
trans_logger.info(msg)
save_data(curr_data)
elif withdraw_amount == 'b':
back_flag = True
else:
msg = "[%s] is not a valid amount, only accept integer!" % withdraw_amount
trans_logger.error(msg) @login_required
def transfer(user_data):
"""
print current balance and let user do the transfer action
:param user_data:
:return:
"""
curr_data = user_data['user']
back_flag = False
while not back_flag:
transfer_amount = input("Input transfer amount:").strip()
tesla_data = get_data('tesla') # 得到转账人的信息
if transfer_amount.isdigit():
transfer_amount = float(transfer_amount) # 转账金额转化float
old_bal = curr_data['balance']
new_tesla_bal = tesla_data['balance'] + transfer_amount # 得到转账人的余额
curr_data['balance'] = old_bal - float(transfer_amount) * settings.TRANS_TYPE['withdraw']['interest']
tesla_data['balance'] = new_tesla_bal
if curr_data['balance']:
msg = "New Balance: %s new_tesla_bal: %s " % (curr_data['balance'], new_tesla_bal)
trans_logger.info(msg)
# 保存数据
save_data(curr_data)
save_data(tesla_data)
elif transfer_amount == 'b':
back_flag = True
else:
msg = "[%s] is not a valid amount, only accept integer!" % transfer_amount
trans_logger.error(msg) def logout(user_data):
"""
user logout
:param acc_data:
:return:
"""
msg = "%s logout" % user_data['user_id']
user_data['is_authenticated']= False def save_data(acc_data):
"""
保存acc_data
:param acc_data:
:return:
"""
file = os.path.join(settings.BASE_DIR,'account',acc_data['id']+'.json')
user_fp = open(file,'w',encoding='utf-8')
print('save_data: ',file, acc_data)
json.dump(acc_data,user_fp,ensure_ascii=False)
user_fp.close() def get_data(user_id):
"""
得到acc_data
:param user_id:
:return:
"""
file = os.path.join(settings.BASE_DIR, 'account', user_id+'.json')
user_fp = open(file, 'r', encoding='utf-8')
acc_data = json.load(user_fp)
user_fp.close()
return acc_data def interactive(acc_data): '''
interact with user
:return:
'''
menu = """
------- Bank ---------
1. 账户信息(功能已实现)
2. 取款(功能已实现)
3. 转账(功能已实现)
4. 用户退出(功能已实现)
"""
menu_dic = {
'': account_info,
'': withdraw,
'': transfer,
'': logout,
}
exit_flag = False
while not exit_flag:
print(menu)
user_option = input(">>:").strip()
if user_option in menu_dic:
# print('accdata', acc_data)
# print(menu_dic[user_option], acc_data)
menu_dic[user_option](acc_data)
else:
print("Option does not exist!", "error")
exit_flag = True def run():
'''
this function will be called right a way when the program started, here handles the user interaction stuff
:return:
'''
acc_data = auth.auth_acc(user_data,access_logger)
# print(acc_data)
# 如果用户认证成功
if user_data['is_authenticated']:
user_data['user'] = acc_data
user_data['user_id'] = acc_data.get('id')
interactive(user_data) if __name__ == '__main__':
run()
# print(a[0])
代码 start.py
import sys,os BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASE_DIR) # 加入环境变量 from core import main as m if __name__ == '__main__':
m.run()