一、首先优化相关的处理请求的类（mian.py）
import tornado.web
from PIL import Image
from pycket.session import SessionMixin
from utiles.orm import OrmHandler
from models.db import Session
from models.auth import User, Post
class BaseHandler(tornado.web.RequestHandler, SessionMixin):
def get_current_user(self):
return self.session.get('username_session')
def initialize(self):
self.db_session = Session()
self.orm = OrmHandler(self.db_session)
def on_finish(self):
self.db_session.close()
class ExploreHandler(BaseHandler):
"""
发现或最近上传的图片页面 /explore ExploreHandler
"""
@tornado.web.authenticated
def get(self):
posts = self.orm.get_all_posts()
self.render('explore.html', posts=posts, username=self.current_user)
def post(self):
pass
class IndexHandler(BaseHandler):
"""
所关注的用户图片流 / IndexHandler
"""
@tornado.web.authenticated
def get(self):
users = self.orm.get_all_users()
username = self.current_user
user = self.db_session.query(User).filter_by(username=username).first()
post_id = user.id
self.render('user/information.html', users=users, username=username, post_id=post_id)
def post(self):
pass
class PostHandler(BaseHandler):
"""
单个图片详情页面 /post/id PostHandler
"""
@tornado.web.authenticated
def get(self, post_id):
session = Session()
user = session.query(User).filter_by(username=self.current_user).first()
if post_id == str(user.id):
posts = session.query(Post).filter_by(user_id=post_id).all()
self.render('post.html', post_id=post_id, posts=posts, username=self.current_user)
else:
self.redirect('/upload')
session.close()
class One_pictureHandler(BaseHandler):
def get(self, post_id):
session = Session()
post_one = session.query(Post).filter_by(id=post_id).first()
self.render('one_picture.html', post=post_one)
class UploadHandler(BaseHandler):
"""
用户上传图片信息
"""
@tornado.web.authenticated
def get(self):
self.render('user/upload.html', username=self.current_user)
@tornado.web.authenticated
def post(self):
try:
files = self.request.files['picture']#list类型中包含一个字典
if files[0]:
dict_img = files[0]
filename = dict_img['filename']
print(filename)
print(dict_img['content_type'])
save_path = 'static/upload/{}'.format(filename)
with open(save_path, 'wb') as f:
f.write(dict_img['body'])
im = Image.open(save_path)
im.thumbnail((200, 200))
im.save('static/thumbs/thumb_{}'.format(filename), 'JPEG')
post_id = self.orm.add_post(self.current_user, 'upload/{}'.format(filename), 'thumbs/thumb_{}'.format(filename))
print(post_id)
if post_id:
self.redirect('/post/{}'.format(str(post_id)))
else:
self.write("上传失败，系统不听话")
except Exception as e:
print(e)
self.redirect('/upload')
class RegisterHandler(BaseHandler):
"""
用户注册
"""
def get(self):
self.render('user/register.html')
def post(self):
username = self.get_argument('username', '')
password = self.get_argument('password', '')
email = self.get_argument('email', '')
gender = self.get_argument('gender', '')
age = self.get_argument('age', ' ')
if not self.db_session.query(User.username).filter(User.username == username).first():
value = self.orm.add_user(username, password, email, age, gender)
if value == 'ok':
self.session.set("username_session", username)
self.redirect('/')
else:
self.write("不好意思哦，注册失败！！！！")
else:
self.write('用户名已经存在！！！')
class LoginHandler(BaseHandler):
def get(self):
next_url = self.get_argument('next', '/')
print(next_url)
if self.current_user:
print("此时的session对应的值为{}".format(self.current_user))
self.redirect(next_url)
else:
self.render('user/login.html', next_url=next_url,msg="暂时无问题！！")
def post(self):
next_url = self.get_argument('next_url', '/')
username = self.get_argument("username", '')
password = self.get_argument("password", '')
ret = self.orm.user_login(username, password)
if ret is not None:
if not self.session.get('username_session'):
if ret['result'] == True:
self.session.set("username_session", username)
self.redirect(next_url)
else:
self.render('user/login.html', next_url=next_url, msg=ret['msg'])
else:
self.render('user/login.html', next_url=next_url, msg='用户名不存在')
class LogoutHandler(BaseHandler):
@tornado.web.authenticated
def get(self):
self.session.delete('username_session')
next_url = self.get_argument('next', '/')
self.render('user/login.html', msg="退出成功！！", next_url=next_url)
class CheckHandler(BaseHandler):
@tornado.web.authenticated
def get(self):
user = self.session.query(User).filter_by(username=self.current_user).first()
post_id = user.id
self.redirect('/post/{}'.format(str(post_id)))
二、相关辅助函数的代码优化（写成一个类，orm.py文件中）
from models.auth import User, Post
from hashlib import md5
def hash(text):
return md5('{}+{}'.format(text, 'jiayan').encode()).hexdigest()
class OrmHandler:
"""
配合RequestHandler实例化的session一起使用
"""
def __init__(self, session):
self.session = session
def get_all_posts(self):
"""
获取所有图片
"""
data_all = self.session.query(Post).all()
return data_all
def get_all_users(self):
"""
获取所有用户的信息
:return:
"""
users = self.session.query(User.id, User.username, User.gender, User.age, User.creatime, User.emial).all()
return users
def add_user(self,username, password, email, age, gender):
"""
进行用户注册
:param username:
:param password:
:param email:
:return:
"""
if username and password:
user = User(username=username, password=password, emial=email, age=age, gender=gender)
self.session.add(user)
self.session.commit()
return 'ok'
else:
return 'false'
def add_post(self, username, image_url, thumb_url):
user = self.session.query(User).filter_by(username=username).first()
post = Post(user_id=user.id, image_url=image_url, thumb_url=thumb_url)
self.session.add(post)
self.session.commit()
post_id = post.user_id
return post_id
def user_login(self, username, password):
"""
用户登录模块
:param username:
:param password:
:return:
"""
try:
ret = {'result': False}
if username and password:
user = self.session.query(User).filter_by(username=username).first()
# if hash(session.query(User.password).filter(User.username == username).first()[0]) == hash(password):
if hash(user.password) == hash(password):
ret['result'] = True
else:
ret['msg'] = '密码错误'
else:
ret['msg'] = '用户名和密码不能为空'
return ret
except:
pass
