首页 > 编程语言 >一个练习项目，好玩的bbs-python-tornado

一个练习项目，好玩的bbs-python-tornado

时间：2024-09-02 14:05:37浏览次数：11  
标签：get python data self cursor tornado sessionId bbs id

代码：

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

相关文章

  • 一个练习项目，好玩的bbs-python-flask
    代码：fromflaskimportFlask,make_response,requestimportos.pathimportMySQLdbimportjsonimporthashlibimportrandomimportmathimportosfromdatetimeimportdatetimeapp=Flask(__name__)classDateEncoder(json.JSONEncoder):defdefault(se......
  • 一个练习项目，好玩的bbs-python-fastapi
    代码：fromfastapiimportFastAPI,Response,Cookie,Dependsfromfastapi.responsesimportJSONResponsefromfastapi.responsesimportHTMLResponseimportos.pathimportMySQLdbimportjsonimporthashlibimportrandomimportmathimportosfromdatetimeim......
  • 一个练习项目，好玩的bbs-python-bottle
    代码：frombottleimportroute,run,templatefrombottleimportBottle,request,responseimportos.pathimportMySQLdbimportjsonimporthashlibimportrandomimportmathimportosfromdatetimeimportdatetimeclassDateEncoder(json.JSONEncoder):......
  • 一个练习项目，好玩的bbs-python-pyramid
    代码：fromwsgiref.simple_serverimportmake_serverfrompyramid.configimportConfiguratorfrompyramid.viewimportview_configfrompyramid.responseimportResponseimportos.pathimportMySQLdbimportjsonimporthashlibimportrandomimportmathimport......
  • 一个练习项目，好玩的bbs-python-webpy
    代码：importwebimportos.pathimportMySQLdbimportjsonimporthashlibimportrandomimportmathimportosfromdatetimeimportdatetimeclassDateEncoder(json.JSONEncoder):defdefault(self,obj):ifisinstance(obj,datetime):......
  • [开题报告]flask框架的高校医务室管理系统（程序+论文+python）
    本系统（程序+源码+数据库+调试部署+开发环境）带论文文档1万字以上，文末可获取，系统界面在最后面。系统程序文件列表开题报告内容研究背景随着高校规模的不断扩大和学生健康意识的增强，高校医务室作为维护师生健康的重要机构，其管理效率与服务质量直接影响到师生的日常生活与学习......
  • 推荐一个Python打造的开源自动驾驶平台：Donkeycar！
    1、引言随着人工智能和自动驾驶技术的飞速发展，自动驾驶车辆的研究和开发成为了科技领域的热点。对于初学者、爱好者和学生而言，一款易于上手且功能强大的自动驾驶平台显得尤为重要。Donkeycar正是这样一款开源项目，它提供了一个轻量级、模块化的Python自驾车库，旨在促进快速实验和社区......
  • 如何使用 Python 调用 DPAPI ？
    在Windows环境下，DPAPI（DataProtectionAPI）是一种用于加密和解密数据的API，可以保护数据，使其只能由当前用户或计算机访问。在Python中，可以通过Cryptography或pywin32等库来使用DPAPI进行数据加密和解密。以下是我我做项目时使用Python调用DPAPI进行数据加密和解密的示......
  • [开题报告]flask框架的机电配件管理系统（程序+论文+python）
    本系统（程序+源码+数据库+调试部署+开发环境）带论文文档1万字以上，文末可获取，系统界面在最后面。系统程序文件列表开题报告内容研究背景在快速发展的机电行业中，配件管理作为保障设备高效运行与维护的重要环节，其重要性日益凸显。随着企业规模的扩大和生产复杂度的提升，传统的配......
  • 基于micropython的ESP8266控制触摸传感器的设计方案
       以下是一个基于MicroPython的ESP8266控制触摸传感器的设计方案：一、硬件准备1. ESP8266开发板(如NodeMCU)       2. 触摸传感器模块(如TTP223触摸传感器)   3. 杜邦线若干                    ......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99