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

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

时间：2024-09-02 14:05:54浏览次数：10  
标签：cherrypy python bbs self cursor result data id

代码：

import cherrypy
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 Root(object):
    conn = None
    cursor = None
    secretKey = 'saacac3423@21212'
    pagesize = 20
    default_dir = ''
    
    def __init__(self):
        self.conn = MySQLdb.Connection('127.0.0.1', 'root', '123456', 'my_bbs')
        self.cursor = self.conn.cursor(cursorclass = MySQLdb.cursors.DictCursor)
        self.default_dir = 'static'
        
    def __del__(self):
        self.cursor.close()
        self.conn.close()
    
    """ 提供静态文件的访问 """
    @cherrypy.expose
    def static(self, *args, **kwargs):
        path = os.path.dirname(__file__)+'/'+self.default_dir +'/'+ os.path.join(*args)
        if os.path.isfile(path):
            return cherrypy.lib.static.serve_file(path)
        else:
            raise cherrypy.HTTPError(404, 'File not found')
 
        
    def readTemplate(self, tpName, data = {}):
        file = open("templates/" + tpName, "r", encoding="utf-8")
        content = file.read()
        file.close()
        
        for k,v in data.items():
            content = content.replace("${"+k+"}", v)
        
        return content
        
    def getloginuserinfo(self, sessionId):
        try:
            sessionIdHead = cherrypy.request.headers['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}
            
        return result
    
    def error(self, code, msg):
        return self.response(code, msg, None)
    
    def success(self, data = {}):
        return self.response(0, '', data)
    
class User(Root):
    @cherrypy.expose
    def register(self, username, password, nickname):
        sql = "select id,username,nickname,addTime from user where username='%s'" % username
        self.cursor.execute(sql)
        data = self.cursor.fetchone()
        if data != None:
            result = self.error(1, '用户名已经存在')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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
            result = self.success(insertId)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '注册失败')
            
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def login(self, username, 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:
            result = self.error(1, '用户名或者密码错误')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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
            result = self.success(data)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '保存会话id失败')
            
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def logout(self, sessionId):
        data = super().getloginuserinfo(sessionId)
        
        if data == None:
            result = self.success(None)
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
        
        if data['sessionId'] == '':
            result = self.success(data)
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        try:
            sql = "update user set sessionId='' where sessionId='%s'" % sessionId
            self.cursor.execute(sql)
            self.conn.commit()
            data['sessionId'] = ''
            result = self.success(data)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '删除会话id失败')
            
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def getuserinfo(self, sessionId):
        data = super().getloginuserinfo(sessionId.lower())
        if data == None:
            result = self.error(1, '获取用户信息失败')
        elif data['id'] <= 0:
            result = self.error(1, '请先登录')
        else:
            result = self.success(data)
            
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
class Post(Root):
    @cherrypy.expose
    def list(self, keyword = '', page = 1):
        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()
        
        result = self.success({'totalpage' : totalpage, 'data' : data})
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def detail(self, 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()
        
        result = self.success(data)
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def add(self, title, content, sessionId):
        userinfo = super().getloginuserinfo(sessionId)
        userId = userinfo['id']
        userNickename = userinfo['nickname']
        
        if userId <= 0:
            result = self.error(1, '请先登录')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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
            result = self.success(insertId)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '发帖失败')
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def edit(self, id, title, content, sessionId):
        userinfo = super().getloginuserinfo(sessionId)
        userId = userinfo['id']
        userNickename = userinfo['nickname']
        
        if userId <= 0:
            result = self.error(1, '请先登录')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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()
            result = self.success(None)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '编辑帖子失败')
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def delete(self, id, sessionId):
        userinfo = super().getloginuserinfo(sessionId)
        userId = userinfo['id']
        userNickename = userinfo['nickname']
        
        if userId <= 0:
            result = self.error(1, '请先登录')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        try:
            sql = "update content set isDel=1 where id=%s and userId=%s" % (id, userId)
            self.cursor.execute(sql)
            self.conn.commit()
            result = self.success(None)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '删除帖子失败')
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
class Reply(Root):
    @cherrypy.expose
    def list(self, contentId = 0, page = 1):
        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()
        
        result = self.success({'totalpage' : totalpage, 'data' : data})
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def detail(self, 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()
        
        result = self.success(data)
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def add(self, contentId, content, sessionId):
        userinfo = super().getloginuserinfo(sessionId)
        userId = userinfo['id']
        userNickename = userinfo['nickname']
        
        if userId <= 0:
            result = self.error(1, '请先登录')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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
            result = self.success(insertId)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '回复失败')
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def edit(self, id, content, sessionId):
        userinfo = super().getloginuserinfo(sessionId)
        userId = userinfo['id']
        userNickename = userinfo['nickname']
        
        if userId <= 0:
            result = self.error(1, '请先登录')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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()
            result = self.success(None)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '编辑回复失败')
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
    @cherrypy.expose
    def delete(self, id, sessionId):
        userinfo = super().getloginuserinfo(sessionId)
        userId = userinfo['id']
        userNickename = userinfo['nickname']
        
        if userId <= 0:
            result = self.error(1, '请先登录')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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:
            result = self.error(1, '回复不存在')
            return json.dumps(result, cls = DateEncoder, ensure_ascii = False)

        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()
            result = self.success(None)
        except MySQLdb.Error as e:
            self.conn.rollback()
            result = self.error(1, '删除回复失败')
        
        return json.dumps(result, cls = DateEncoder, ensure_ascii = False)
    
class Index(Root):
    @cherrypy.expose
    def index(self):
        return "此站接口使用python.cherrypy实现，<a href='api.html' target='_blank'>接口列表</a>";

if __name__ == '__main__':
    server_config = {
        'server.socket_host': '0.0.0.0',
        'server.socket_port': 1083
    }
    cherrypy.config.update(server_config)
    
    cherrypy.tree.mount(Index(), '/', {})
    cherrypy.tree.mount(User(), '/user', {})
    cherrypy.tree.mount(Post(), '/post', {})
    cherrypy.tree.mount(Reply(), '/reply', {})
    
    cherrypy.engine.start()
    cherrypy.engine.block()

 

输出：

C:\Users\jsb>D:\software\Python310\python.exe D:\workspace\studys\study_bbs\start_web_cherrypy.py
[02/Sep/2024:14:00:15] ENGINE Bus STARTING
CherryPy Checker:
The Application mounted at '' has an empty config.

[02/Sep/2024:14:00:15] ENGINE Started monitor thread 'Autoreloader'.
[02/Sep/2024:14:00:15] ENGINE Serving on http://0.0.0.0:1083
[02/Sep/2024:14:00:15] ENGINE Bus STARTED

 

标签：cherrypy,python,bbs,self,cursor,result,data,id
From： https://www.cnblogs.com/xuxiaobo/p/18392582

相关文章

  • 一个练习项目，好玩的bbs-python-tornado
    代码：importos.pathimporttornado.httpserverimporttornado.webimporttornado.optionsimporttornado.ioloopfromtornado.optionsimportdefine,optionsimportMySQLdbimportjsonimporthashlibimportrandomimportmathimportosfromdatetimeimportda......
  • 一个练习项目，好玩的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万字以上，文末可获取，系统界面在最后面。系统程序文件列表开题报告内容研究背景在快速发展的机电行业中，配件管理作为保障设备高效运行与维护的重要环节，其重要性日益凸显。随着企业规模的扩大和生产复杂度的提升，传统的配......

赞助商

阅读排行

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

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

Copyright © 2020-2023 IPS99 版权所有 IPS99