代码:
import os.path import tornado.httpserver import tornado.web import tornado.options import tornado.ioloop from tornado.options import define, options import MySQLdb import json import hashlib import random import math import os from datetime import datetime class DateEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.strftime("%Y-%m-%d %H:%M:%S") else: return json.JSONEncoder.default(self, obj) class BaseHandler: conn = None cursor = None secretKey = 'saacac3423@21212' pagesize = 20 def __init__(self): self.conn = MySQLdb.Connection('127.0.0.1', 'root', '123456', 'my_bbs') self.cursor = self.conn.cursor(cursorclass = MySQLdb.cursors.DictCursor) def __del__(self): self.cursor.close() self.conn.close() def getloginuserinfo(self, sessionId): try: sessionIdHead = self.get_secure_cookie("sessionId") except: sessionIdHead = '' if sessionIdHead is not None and sessionIdHead != '': sessionId = sessionIdHead sql = "select id,username,nickname,addTime,sessionId from user where sessionId='%s'" % sessionId self.cursor.execute(sql) data = self.cursor.fetchone() if data is None: data = {'id' : 0, 'username' : '', 'nickname' : '', 'addTime' : '', 'sessionId' : ''} return data def response(self, code, msg, data): if code != 0: result = {'code' : code, 'msg' : msg, 'data' : None} else: result = {'code' : 0, 'msg' : '', 'data' : data} result = json.dumps(result, cls = DateEncoder, ensure_ascii = False) return result def error(self, code, msg): self.write(self.response(code, msg, None)) def success(self, data = {}): self.write(self.response(0, '', data)) class IndexHandler(tornado.web.RequestHandler, BaseHandler): def get(self): self.write("此站接口使用python.tornado实现,<a href='api.html' target='_blank'>接口列表</a>") class RegisterHandler(tornado.web.RequestHandler, BaseHandler): def get(self): username = self.get_argument("username", "") password = self.get_argument("password", "") nickname = self.get_argument("nickname", "") sql = "select id,username,nickname,addTime from user where username='%s'" % username self.cursor.execute(sql) data = self.cursor.fetchone() if data != None: self.error(1, '用户名已经存在') return True try: passwordMd5 = hashlib.md5(password.encode(encoding='utf-8')).hexdigest() sql = "insert into user(username, password, nickname) value('%s', '%s', '%s')" % (username, passwordMd5, nickname) self.cursor.execute(sql) self.conn.commit() insertId = self.cursor.lastrowid self.success(insertId) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '注册失败') class LoginHandler(tornado.web.RequestHandler, BaseHandler): def get(self): username = self.get_argument("username", "") password = self.get_argument("password", "") passwordMd5 = hashlib.md5(password.encode(encoding='utf-8')).hexdigest() sql = "select id,username,nickname,addTime from user where username='%s' and password='%s'" % (username, passwordMd5) self.cursor.execute(sql) data = self.cursor.fetchone() if data == None: self.error(1, '用户名或者密码错误') return True tmpSessionId = self.secretKey + str(data['id']) + str(data['addTime']) tmpSessionId = hashlib.md5(tmpSessionId.encode(encoding='utf-8')).hexdigest() try: sql = "update user set sessionId='%s' where id=%s" % (tmpSessionId, data['id']) self.cursor.execute(sql) self.conn.commit() data['sessionId'] = tmpSessionId self.success(data) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '保存会话id失败') class LogoutHandler(tornado.web.RequestHandler, BaseHandler): def get(self): sessionId = self.get_argument("sessionId", "") data = super().getloginuserinfo(sessionId) if data == None: self.success(None) return True if data['sessionId'] == '': self.success(data) return True try: sql = "update user set sessionId='' where sessionId='%s'" % sessionId self.cursor.execute(sql) self.conn.commit() data['sessionId'] = '' self.success(data) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '删除会话id失败') class GetuserinfoHandler(tornado.web.RequestHandler, BaseHandler): def get(self): sessionId = self.get_argument("sessionId", "") userinfo = super().getloginuserinfo(sessionId) self.success(userinfo) class PostlistHandler(tornado.web.RequestHandler, BaseHandler): def get(self): page = self.get_argument("page", "1") keyword = self.get_argument("keyword", "") page = int(page) if page <= 0: page = 1 addsql = " isDel=0 " if keyword is not None and keyword != '': addsql = " isDel=0 and title like '%"+keyword+"%' " start = (page - 1) * self.pagesize sql1 = "select count(1) as count from content where %s" % addsql self.cursor.execute(sql1) countdata = self.cursor.fetchone() totalpage = math.ceil(countdata['count'] / float(self.pagesize)) data = [] if totalpage > 0: sql2 = "select id,title,userId,userNickename,replyNum,updateTime from content where %s order by updateTime desc limit %s,%s" % (addsql, start, self.pagesize) self.cursor.execute(sql2) data = self.cursor.fetchall() self.success({'totalpage' : totalpage, 'data' : data}) class PostdetailHandler(tornado.web.RequestHandler, BaseHandler): def get(self): id = self.get_argument("id", "0") sql = "select id,title,content,userId,userNickename,replyNum,updateTime from content where isDel=0 and id=%s" % id self.cursor.execute(sql) data = self.cursor.fetchone() self.success(data) class PostaddHandler(tornado.web.RequestHandler, BaseHandler): def get(self): title = self.get_argument("title", "") content = self.get_argument("content", "") sessionId = self.get_argument("sessionId", "") userinfo = super().getloginuserinfo(sessionId) userId = userinfo['id'] userNickename = userinfo['nickname'] if userId <= 0: self.error(1, '请先登录') return True try: sql = "insert into content(title, content, userId, userNickename) value('%s', '%s', %s, '%s')" % (title, content, userId, userNickename) self.cursor.execute(sql) self.conn.commit() insertId = self.cursor.lastrowid self.success(insertId) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '发帖失败') class PosteditHandler(tornado.web.RequestHandler, BaseHandler): def get(self): id = self.get_argument("id", "0") title = self.get_argument("title", "") content = self.get_argument("content", "") sessionId = self.get_argument("sessionId", "") userinfo = super().getloginuserinfo(sessionId) userId = userinfo['id'] userNickename = userinfo['nickname'] if userId <= 0: self.error(1, '请先登录') return True try: sql = "update content set title='%s',content='%s',userId=%s,userNickename='%s' where id=%s and userId=%s" % (title, content, userId, userNickename, id, userId) self.cursor.execute(sql) self.conn.commit() self.success(None) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '编辑帖子失败') class PostdeleteHandler(tornado.web.RequestHandler, BaseHandler): def get(self): id = self.get_argument("id", "0") sessionId = self.get_argument("sessionId", "") userinfo = super().getloginuserinfo(sessionId) userId = userinfo['id'] userNickename = userinfo['nickname'] if userId <= 0: self.error(1, '请先登录') return True try: sql = "update content set isDel=1 where id=%s and userId=%s" % (id, userId) self.cursor.execute(sql) self.conn.commit() self.success(None) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '删除帖子失败') class ReplylistHandler(tornado.web.RequestHandler, BaseHandler): def get(self): page = self.get_argument("page", "1") contentId = self.get_argument("contentId", "0") page = int(page) if page <= 0: page = 1 start = (page - 1) * self.pagesize sql1 = "select count(1) as count from reply where isDel=0 and contentId=%s" % contentId self.cursor.execute(sql1) countdata = self.cursor.fetchone() totalpage = math.ceil(countdata['count'] / float(self.pagesize)) data = [] if totalpage > 0: sql2 = "select id,content,replyUserId,replyUserNickename,addTime from reply where isDel=0 and contentId=%s order by id asc limit %s,%s" % (contentId, start, self.pagesize) self.cursor.execute(sql2) data = self.cursor.fetchall() self.success({'totalpage' : totalpage, 'data' : data}) class ReplydetailHandler(tornado.web.RequestHandler, BaseHandler): def get(self): id = self.get_argument("id", "0") sql = "select id,content,replyUserId,replyUserNickename,addTime from reply where isDel=0 and id=%s" % id self.cursor.execute(sql) data = self.cursor.fetchone() self.success(data) class ReplyaddHandler(tornado.web.RequestHandler, BaseHandler): def get(self): contentId = self.get_argument("contentId", "0") content = self.get_argument("content", "") sessionId = self.get_argument("sessionId", "") userinfo = super().getloginuserinfo(sessionId) userId = userinfo['id'] userNickename = userinfo['nickname'] if userId <= 0: self.error(1, '请先登录') return True try: sql2 = "update content set replyNum=replyNum+1 where id=%s" % contentId self.cursor.execute(sql2) sql1 = "insert into reply(contentId, content, replyUserId, replyUserNickename) value(%s, '%s', %s, '%s')" % (contentId, content, userId, userNickename) self.cursor.execute(sql1) self.conn.commit() insertId = self.cursor.lastrowid self.success(insertId) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '回复失败') class ReplyeditHandler(tornado.web.RequestHandler, BaseHandler): def get(self): id = self.get_argument("id", "0") content = self.get_argument("content", "") sessionId = self.get_argument("sessionId", "") userinfo = super().getloginuserinfo(sessionId) userId = userinfo['id'] userNickename = userinfo['nickname'] if userId <= 0: self.error(1, '请先登录') return True try: sql = "update reply set content='%s',replyUserId=%s,replyUserNickename='%s' where id=%s and replyUserId=%s" % (content, userId, userNickename, id, userId) self.cursor.execute(sql) self.conn.commit() self.success(None) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '编辑回复失败') class ReplydeleteHandler(tornado.web.RequestHandler, BaseHandler): def get(self): id = self.get_argument("id", "0") sessionId = self.get_argument("sessionId", "") userinfo = super().getloginuserinfo(sessionId) userId = userinfo['id'] userNickename = userinfo['nickname'] if userId <= 0: self.error(1, '请先登录') return True sql = "select id,content,replyUserId,replyUserNickename,addTime,contentId from reply where isDel=0 and id=%s" % id self.cursor.execute(sql) contentdata = self.cursor.fetchone() if contentdata is None: self.error(1, '回复不存在') return True try: sql2 = "update content set replyNum=replyNum-1 where id=%s" % contentdata['contentId'] self.cursor.execute(sql2) sql1 = "update reply set isDel=1 where id=%s and replyUserId=%s" % (id, userId) self.cursor.execute(sql1) self.conn.commit() self.success(None) except MySQLdb.Error as e: self.conn.rollback() self.error(1, '删除回复失败') if __name__ == "__main__": port = 1088 define("port", default=port, help="run on the given port", type = int) print("python.tornado Server is running on port %d\n" % port) tornado.options.parse_command_line() app = tornado.web.Application( handlers=[ (r"/", IndexHandler), (r"/user/register", RegisterHandler), (r"/user/login", LoginHandler), (r"/user/logout", LogoutHandler), (r"/user/getuserinfo", GetuserinfoHandler), (r"/post/list", PostlistHandler), (r"/post/detail", PostdetailHandler), (r"/post/add", PostaddHandler), (r"/post/edit", PosteditHandler), (r"/post/delete", PostdeleteHandler), (r"/reply/list", ReplylistHandler), (r"/reply/detail", ReplydetailHandler), (r"/reply/add", ReplyaddHandler), (r"/reply/edit", ReplyeditHandler), (r"/reply/delete", ReplydeleteHandler), ], cookie_secret="bZJc2sWbQLKos6GkHn/VB9oXwQt8S0R0kRvJ5/xJ89E=", xsrf_cookies=True, login_url="/xsrf", template_path=os.path.join(os.path.dirname(__file__), "templates"), static_path=os.path.join(os.path.dirname(__file__), "static"), debug=True ) http_server = tornado.httpserver.HTTPServer(app) http_server.listen(options.port) tornado.ioloop.IOLoop.instance().start()
输出:
D:\workspace\studys\study_pys\pc_app\dist>D:\software\Python310\python.exe D:\workspace\studys\study_bbs\start_web_tornado.py python.tornado Server is running on port 1088
标签:get,python,data,self,cursor,tornado,sessionId,bbs,id From: https://www.cnblogs.com/xuxiaobo/p/18392587