app代码注释#马莹

my_branch
马莹 3 months ago
parent ed27107eba
commit e847acfdd0

@ -1,32 +1,48 @@
# 马莹:导入微信机器人(werobot)的会话存储基类,用于自定义会话存储方式
from werobot.session import SessionStorage
# 马莹导入微信机器人的JSON处理工具用于数据的序列化和反序列化
from werobot.utils import json_loads, json_dumps
# 马莹导入当前Django项目中djangoblog应用的缓存工具
from djangoblog.utils import cache
# 马莹定义基于Memcache的会话存储类继承自werobot的SessionStorage基类
class MemcacheStorage(SessionStorage):
# 马莹:初始化方法,设置缓存键的前缀,默认为'ws_'
def __init__(self, prefix='ws_'):
self.prefix = prefix
self.cache = cache
self.prefix = prefix # 马莹:存储缓存键的前缀
self.cache = cache # 马莹:引用导入的缓存工具实例
# 马莹:用于检查缓存存储是否可用的属性
@property
def is_available(self):
value = "1"
value = "1" # 马莹:定义测试值
# 马莹:尝试设置一个测试键值对到缓存中
self.set('checkavaliable', value=value)
# 马莹:通过比较设置的值和获取的值是否一致,判断缓存是否可用
return value == self.get('checkavaliable')
# 马莹:生成带前缀的缓存键名,避免键名冲突
def key_name(self, s):
return '{prefix}{s}'.format(prefix=self.prefix, s=s)
# 马莹从缓存中获取指定ID对应的会话数据
def get(self, id):
id = self.key_name(id)
id = self.key_name(id) # 马莹:生成带前缀的缓存键
# 马莹从缓存中获取数据若不存在则返回空字典的JSON字符串
session_json = self.cache.get(id) or '{}'
# 马莹将JSON字符串反序列化为Python字典并返回
return json_loads(session_json)
# 马莹:将会话数据存入缓存
def set(self, id, value):
id = self.key_name(id)
id = self.key_name(id) # 马莹:生成带前缀的缓存键
# 马莹将Python对象序列化为JSON字符串后存入缓存
self.cache.set(id, json_dumps(value))
# 马莹从缓存中删除指定ID的会话数据
def delete(self, id):
id = self.key_name(id)
self.cache.delete(id)
id = self.key_name(id) # 马莹:生成带前缀的缓存键
# 马莹:从缓存中删除该键对应的记录
self.cache.delete(id)

@ -1,19 +1,28 @@
# 马莹导入Django的admin模块用于后台管理功能
from django.contrib import admin
# 马莹:注册模型的地方(后续会在这里注册需要管理的模型)
# Register your models here.
# 马莹定义Commands模型的后台管理类
class CommandsAdmin(admin.ModelAdmin):
# 马莹:在后台列表页展示的字段:标题、命令、描述
list_display = ('title', 'command', 'describe')
# 马莹定义EmailSendLog模型的后台管理类
class EmailSendLogAdmin(admin.ModelAdmin):
# 马莹:在后台列表页展示的字段:标题、收件人、发送结果、创建时间
list_display = ('title', 'emailto', 'send_result', 'creation_time')
# 马莹:设置为只读的字段(无法在后台编辑)
readonly_fields = (
'title',
'emailto',
'send_result',
'creation_time',
'content')
'title', # 标题
'emailto', # 收件人
'send_result', # 发送结果
'creation_time', # 创建时间
'content' # 邮件内容
)
# 马莹:重写添加权限方法,禁止在后台手动添加记录
def has_add_permission(self, request):
return False

@ -5,24 +5,24 @@ from blog.models import Article, Category
class BlogApi:
def __init__(self):
self.searchqueryset = SearchQuerySet() # 初始化搜索查询集,用于处理文章搜索功能
self.searchqueryset.auto_query('') # 执行空查询,初始化搜索结果集(可能用于后续叠加过滤条件)
self.__max_takecount__ = 8 # 定义私有变量限制各类查询的最大返回数量为8条
self.searchqueryset = SearchQuerySet() # 马莹:初始化搜索查询集,用于处理文章搜索功能
self.searchqueryset.auto_query('') # 马莹:执行空查询,初始化搜索结果集(可能用于后续叠加过滤条件)
self.__max_takecount__ = 8 #马莹: 定义私有变量限制各类查询的最大返回数量为8条
def search_articles(self, query):
sqs = self.searchqueryset.auto_query(query) # 使用搜索查询集执行自动查询(可能包含分词、过滤等处理)
sqs = sqs.load_all() # 预加载所有关联数据,减少数据库查询次数(优化性能)
return sqs[:self.__max_takecount__] # 限制返回结果数量返回前N条匹配的文章
sqs = self.searchqueryset.auto_query(query) # 马莹:使用搜索查询集执行自动查询(可能包含分词、过滤等处理)
sqs = sqs.load_all() # 马莹:预加载所有关联数据,减少数据库查询次数(优化性能)
return sqs[:self.__max_takecount__] # 马莹:限制返回结果数量返回前N条匹配的文章
def get_category_lists(self):
return Category.objects.all() # 返回所有分类对象(未限制数量,通常分类数量较少)
return Category.objects.all() # 马莹:返回所有分类对象(未限制数量,通常分类数量较少)
def get_category_articles(self, categoryname):
articles = Article.objects.filter(category__name=categoryname) # 过滤出指定分类下的所有文章(通过外键关联查询)
articles = Article.objects.filter(category__name=categoryname) #马莹: 过滤出指定分类下的所有文章(通过外键关联查询)
if articles:
return articles[:self.__max_takecount__]
return None # 若存在符合条件的文章返回前N条否则返回None
return None # 马莹:若存在符合条件的文章返回前N条否则返回None
def get_recent_articles(self):
return Article.objects.all()[:self.__max_takecount__]
# 返回所有文章的前N条依赖于Article模型的默认排序设置
# 马莹:返回所有文章的前N条依赖于Article模型的默认排序设置

@ -1,14 +1,14 @@
import logging # 导入日志模块,用于记录程序运行过程中的日志信息
import os # 导入os模块用于与操作系统交互如获取环境变量、执行系统命令等
import logging # 马莹;导入日志模块,用于记录程序运行过程中的日志信息
import os # 马莹:导入os模块用于与操作系统交互如获取环境变量、执行系统命令等
import openai # 导入openai模块用于调用OpenAI的API服务
import openai # 马莹:导入openai模块用于调用OpenAI的API服务
from servermanager.models import commands # 从servermanager应用的models模块中导入commands模型用于操作命令相关的数据
from servermanager.models import commands # 马莹:从servermanager应用的models模块中导入commands模型用于操作命令相关的数据
logger = logging.getLogger(__name__) # 创建日志记录器,名称为当前模块名,用于记录该模块的日志
logger = logging.getLogger(__name__) # 马莹:创建日志记录器,名称为当前模块名,用于记录该模块的日志
openai.api_key = os.environ.get('OPENAI_API_KEY') # 从环境变量中获取OpenAI的API密钥并设置为openai模块的API密钥
if os.environ.get('HTTP_PROXY'): # 检查环境变量中是否设置了HTTP代理如果有则为openai模块设置代理
openai.api_key = os.environ.get('OPENAI_API_KEY') # 马莹:从环境变量中获取OpenAI的API密钥并设置为openai模块的API密钥
if os.environ.get('HTTP_PROXY'): # 马莹:检查环境变量中是否设置了HTTP代理如果有则为openai模块设置代理
openai.proxy = os.environ.get('HTTP_PROXY')
@ -25,16 +25,16 @@ class ChatGPT:
:return: GPT模型的回复内容字符串若出错则返回"服务器出错了"
"""
try:
# 调用OpenAI的ChatCompletion接口使用gpt-3.5-turbo模型
# messages参数为消息列表包含用户角色和对应的内容
# 马莹:调用OpenAI的ChatCompletion接口使用gpt-3.5-turbo模型
# 马莹:messages参数为消息列表包含用户角色和对应的内容
completion = openai.ChatCompletion.create(model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}])
# 从返回结果中提取第一个选择的消息内容并返回
# 马莹:从返回结果中提取第一个选择的消息内容并返回
return completion.choices[0].message.content
except Exception as e:
# 捕获异常并记录错误日志
# 马莹:捕获异常并记录错误日志
logger.error(e)
# 返回错误提示信息
# 马莹:返回错误提示信息
return "服务器出错了"
@ -55,18 +55,18 @@ class CommandHandler:
:param title: 命令
:return: 返回命令执行结果
"""
# 使用filter函数筛选出标题不区分大小写与输入title匹配的命令
# 将筛选结果转换为列表
# 马莹:使用filter函数筛选出标题不区分大小写与输入title匹配的命令
# 马莹:将筛选结果转换为列表
cmd = list(
filter(
lambda x: x.title.upper() == title.upper(),
self.commands))
# 如果找到匹配的命令
# 马莹:如果找到匹配的命令
if cmd:
# 调用私有方法执行命令传入命令的具体内容cmd[0].command
# 马莹:调用私有方法执行命令传入命令的具体内容cmd[0].command
return self.__run_command__(cmd[0].command)
else:
# 未找到命令时,返回提示信息
# 马莹:未找到命令时,返回提示信息
return "未找到相关命令请输入hepme获得帮助。"
def __run_command__(self, cmd):
@ -77,11 +77,11 @@ class CommandHandler:
:return: 命令执行的输出结果字符串若执行出错返回错误提示
"""
try:
# 使用os.popen执行命令并读取命令的输出结果
# 马莹:使用os.popen执行命令并读取命令的输出结果
res = os.popen(cmd).read()
return res
except BaseException:
# 捕获所有基本异常,返回命令执行出错的提示
# 马莹:捕获所有基本异常,返回命令执行出错的提示
return '命令执行出错!'
def get_help(self):
@ -91,12 +91,12 @@ class CommandHandler:
:return: 包含所有命令标题和描述的字符串每条命令占一行
"""
rsp = ''
# 遍历所有命令,拼接命令标题和描述信息
# 马莹:遍历所有命令,拼接命令标题和描述信息
for cmd in self.commands:
rsp += '{c}:{d}\n'.format(c=cmd.title, d=cmd.describe)
return rsp
# 当该模块作为主程序运行时执行以下代码
# 马莹:当该模块作为主程序运行时执行以下代码
if __name__ == '__main__':
chatbot = ChatGPT()
prompt = "写一篇1000字关于AI的论文"

@ -1,5 +1,9 @@
# 马莹导入Django的AppConfig类用于配置应用的元数据和行为
from django.apps import AppConfig
# 马莹定义名为ServermanagerConfig的应用配置类继承自AppConfig
class ServermanagerConfig(AppConfig):
# 马莹:指定当前应用的名称为'servermanager'
# 马莹这个名称会被Django用于识别应用通常与应用的目录名一致
name = 'servermanager'

@ -1,21 +1,21 @@
# Generated by Django 4.1.7 on 2023-03-02 07:14
# 说明此文件由Django 4.1.7版本自动生成生成时间为2023年3月2日7:14
# 迁移文件用于记录数据库模型的创建和修改通过Django的migrate命令同步到数据库
# 马莹:Generated by Django 4.1.7 on 2023-03-02 07:14
# 马莹:说明此文件由Django 4.1.7版本自动生成生成时间为2023年3月2日7:14
#马莹: 迁移文件用于记录数据库模型的创建和修改通过Django的migrate命令同步到数据库
from django.db import migrations, models
# 导入Django迁移模块和模型字段模块
# 马莹:导入Django迁移模块和模型字段模块
class Migration(migrations.Migration):
# 定义迁移类,所有迁移操作都在这个类中定义
initial = True # 标记为初始迁移(第一次创建模型时生成)
# 马莹:定义迁移类,所有迁移操作都在这个类中定义
initial = True # 马莹:标记为初始迁移(第一次创建模型时生成)
dependencies = [
] # 依赖的其他迁移文件列表,初始迁移无依赖,所以为空
# 若后续迁移依赖其他应用的迁移,会在此处列出,如:['appname.0001_initial']
] # 马莹:依赖的其他迁移文件列表,初始迁移无依赖,所以为空
# 马莹:若后续迁移依赖其他应用的迁移,会在此处列出,如:['appname.0001_initial']
operations = [ # 迁移操作列表,包含模型的创建、修改等操作
migrations.CreateModel( # 创建名为"commands"的模型(对应数据库表)
operations = [ # 马莹:迁移操作列表,包含模型的创建、修改等操作
migrations.CreateModel( # 马莹:创建名为"commands"的模型(对应数据库表)
name='commands',
fields=[ # 定义模型的字段(对应数据库表的列)
fields=[ # 马莹:定义模型的字段(对应数据库表的列)
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('title', models.CharField(max_length=300, verbose_name='命令标题')),
('command', models.CharField(max_length=2000, verbose_name='命令')),
@ -23,12 +23,12 @@ class Migration(migrations.Migration):
('created_time', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('last_mod_time', models.DateTimeField(auto_now=True, verbose_name='修改时间')),
],
options={ # 模型的额外配置
'verbose_name': '命令', # 模型单数显示名称(后台管理用)
'verbose_name_plural': '命令', # 模型复数显示名称(后台管理用)
}, # 若未指定ordering默认按主键id排序
options={ # 马莹:模型的额外配置
'verbose_name': '命令', # 马莹:模型单数显示名称(后台管理用)
'verbose_name_plural': '命令', # 马莹:模型复数显示名称(后台管理用)
}, # 马莹:若未指定ordering默认按主键id排序
),
migrations.CreateModel( # 创建名为"EmailSendLog"的模型(邮件发送日志)
migrations.CreateModel( # 马莹:创建名为"EmailSendLog"的模型(邮件发送日志)
name='EmailSendLog',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
@ -41,7 +41,7 @@ class Migration(migrations.Migration):
options={
'verbose_name': '邮件发送log',
'verbose_name_plural': '邮件发送log',
'ordering': ['-created_time'], # 按创建时间倒序排列(最新的日志在前)
'ordering': ['-created_time'], # 马莹:按创建时间倒序排列(最新的日志在前)
},
),
]

@ -2,35 +2,35 @@
# 说明此文件由Django 4.2.5版本自动生成生成时间为2023年9月6日13:19
# 作用:记录数据库模型的修改操作(字段重命名、配置调整等),用于同步数据库结构变更
from django.db import migrations
# 导入Django迁移模块
# 马莹:导入Django迁移模块
class Migration(migrations.Migration):
# 迁移类,所有数据库变更操作在此定义
dependencies = [ # 依赖的前置迁移文件:表示必须先执行'servermanager'应用的'0001_initial'迁移
# 才能执行当前迁移(确保修改的是已存在的模型)
# 马莹:迁移类,所有数据库变更操作在此定义
dependencies = [ # 马莹:依赖的前置迁移文件:表示必须先执行'servermanager'应用的'0001_initial'迁移
# 马莹:才能执行当前迁移(确保修改的是已存在的模型)
('servermanager', '0001_initial'),
]
operations = [ # 迁移操作列表:包含对模型的修改操作
migrations.AlterModelOptions( # 修改'EmailSendLog'模型的元配置
operations = [ # 马莹:迁移操作列表:包含对模型的修改操作
migrations.AlterModelOptions( # 马莹:修改'EmailSendLog'模型的元配置
name='emailsendlog',
options={'ordering': ['-creation_time'], 'verbose_name': '邮件发送log', 'verbose_name_plural': '邮件发送log'},
), # 1. 排序方式变更:按'creation_time'字段倒序排列(最新记录在前)
), # 马莹:1. 排序方式变更:按'creation_time'字段倒序排列(最新记录在前)
# (原配置可能是按其他字段排序,此处同步字段名变更后的排序)
# 2. 模型显示名称(单数和复数)保持不变
migrations.RenameField( # 重命名'commands'模型的字段
# 马莹:2. 模型显示名称(单数和复数)保持不变
migrations.RenameField( # 马莹:重命名'commands'模型的字段
model_name='commands',
old_name='created_time', # 原字段名:创建时间
new_name='creation_time', # 新字段名:创建时间(更简洁的命名)
old_name='created_time', # 马莹:原字段名:创建时间
new_name='creation_time', # 马莹:新字段名:创建时间(更简洁的命名)
),
migrations.RenameField( # 重命名'commands'模型的另一个字段
migrations.RenameField( # 马莹:重命名'commands'模型的另一个字段
model_name='commands',
old_name='last_mod_time', # 重命名'commands'模型的另一个字段
new_name='last_modify_time', # 重命名'commands'模型的另一个字段
old_name='last_mod_time', # 马莹:重命名'commands'模型的另一个字段
new_name='last_modify_time', # 马莹:重命名'commands'模型的另一个字段
),
migrations.RenameField( # 重命名'commands'模型的另一个字段
migrations.RenameField( # 马莹:重命名'commands'模型的另一个字段
model_name='emailsendlog',
old_name='created_time', # 原字段名:创建时间
new_name='creation_time', # 新字段名创建时间与commands模型保持命名一致
old_name='created_time', # 马莹:原字段名:创建时间
new_name='creation_time', # 马莹:新字段名创建时间与commands模型保持命名一致
),
]

@ -1,33 +1,50 @@
# 马莹导入Django的模型模块用于定义数据库模型
from django.db import models
# Create your models here.
# 马莹:定义命令模型,用于存储各类命令相关信息
class commands(models.Model):
# 马莹命令标题字段字符串类型最大长度300在admin中显示为'命令标题'
title = models.CharField('命令标题', max_length=300)
# 马莹命令内容字段字符串类型最大长度2000在admin中显示为'命令'
command = models.CharField('命令', max_length=2000)
# 马莹命令描述字段字符串类型最大长度300在admin中显示为'命令描述'
describe = models.CharField('命令描述', max_length=300)
# 马莹创建时间字段自动记录创建时间在admin中显示为'创建时间'
creation_time = models.DateTimeField('创建时间', auto_now_add=True)
# 马莹最后修改时间字段自动记录最后修改时间在admin中显示为'修改时间'
last_modify_time = models.DateTimeField('修改时间', auto_now=True)
# 马莹:定义模型实例的字符串表示,返回命令标题
def __str__(self):
return self.title
# 马莹:模型的元数据配置
class Meta:
verbose_name = '命令'
verbose_name_plural = verbose_name
verbose_name = '命令' # 马莹:模型的单数显示名称
verbose_name_plural = verbose_name # 马莹:模型的复数显示名称(与单数相同)
# 马莹:定义邮件发送日志模型,用于记录邮件发送的相关信息
class EmailSendLog(models.Model):
# 马莹收件人字段字符串类型最大长度300在admin中显示为'收件人'
emailto = models.CharField('收件人', max_length=300)
# 马莹邮件标题字段字符串类型最大长度2000在admin中显示为'邮件标题'
title = models.CharField('邮件标题', max_length=2000)
# 马莹邮件内容字段文本类型无长度限制在admin中显示为'邮件内容'
content = models.TextField('邮件内容')
# 马莹发送结果字段布尔类型默认值为False表示发送失败在admin中显示为'结果'
send_result = models.BooleanField('结果', default=False)
# 马莹创建时间字段自动记录创建时间在admin中显示为'创建时间'
creation_time = models.DateTimeField('创建时间', auto_now_add=True)
# 马莹:定义模型实例的字符串表示,返回邮件标题
def __str__(self):
return self.title
# 马莹:模型的元数据配置
class Meta:
verbose_name = '邮件发送log'
verbose_name_plural = verbose_name
ordering = ['-creation_time']
verbose_name = '邮件发送log' # 马莹:模型的单数显示名称
verbose_name_plural = verbose_name # 马莹:模型的复数显示名称(与单数相同)
ordering = ['-creation_time'] # 马莹:默认排序方式:按创建时间倒序(最新的在前)

@ -1,46 +1,62 @@
import logging
import os
import re
# 马莹:导入必要的模块
import logging # 日志模块,用于记录系统运行日志
import os # 操作系统接口模块,用于处理文件路径等
import re # 正则表达式模块,用于字符串匹配
import jsonpickle
from django.conf import settings
from werobot import WeRoBot
from werobot.replies import ArticlesReply, Article
from werobot.session.filestorage import FileStorage
import jsonpickle # 用于复杂对象的JSON序列化与反序列化
from django.conf import settings # 导入Django项目设置
from werobot import WeRoBot # 导入微信机器人框架
from werobot.replies import ArticlesReply, Article # 导入微信消息回复类型
from werobot.session.filestorage import FileStorage # 导入文件存储会话的类
from djangoblog.utils import get_sha256
from servermanager.api.blogapi import BlogApi
from servermanager.api.commonapi import ChatGPT, CommandHandler
from .MemcacheStorage import MemcacheStorage
from djangoblog.utils import get_sha256 # 导入自定义的SHA256加密工具
from servermanager.api.blogapi import BlogApi # 导入博客相关API
from servermanager.api.commonapi import ChatGPT, CommandHandler # 导入聊天机器人和命令处理工具
from .MemcacheStorage import MemcacheStorage # 导入自定义的Memcache会话存储类
# 马莹初始化微信机器人从环境变量获取token默认使用'lylinux',启用会话功能
robot = WeRoBot(token=os.environ.get('DJANGO_WEROBOT_TOKEN')
or 'lylinux', enable_session=True)
# 马莹实例化Memcache存储对象
memstorage = MemcacheStorage()
# 马莹检查Memcache是否可用可用则使用Memcache存储会话否则使用文件存储
if memstorage.is_available:
robot.config['SESSION_STORAGE'] = memstorage
else:
# 马莹:若文件存储的会话文件已存在则删除
if os.path.exists(os.path.join(settings.BASE_DIR, 'werobot_session')):
os.remove(os.path.join(settings.BASE_DIR, 'werobot_session'))
# 马莹:使用文件存储会话
robot.config['SESSION_STORAGE'] = FileStorage(filename='werobot_session')
# 马莹实例化博客API、命令处理器和日志记录器
blogapi = BlogApi()
cmd_handler = CommandHandler()
logger = logging.getLogger(__name__)
def convert_to_article_reply(articles, message):
"""
将文章列表转换为微信图文消息回复
:param articles: 文章对象列表
:param message: 接收的消息对象
:return: 图文消息回复对象
"""
reply = ArticlesReply(message=message)
# 马莹:导入自定义的模板标签用于截断文章内容
from blog.templatetags.blog_tags import truncatechars_content
for post in articles:
# 马莹:从文章内容中提取第一张图片作为图文消息的封面图
imgs = re.findall(r'(?:http\:|https\:)?\/\/.*\.(?:png|jpg)', post.body)
imgurl = ''
if imgs:
imgurl = imgs[0]
# 马莹:创建文章对象并添加到回复中
article = Article(
title=post.title,
description=truncatechars_content(post.body),
description=truncatechars_content(post.body), # 马莹:截断文章内容作为描述
img=imgurl,
url=post.get_full_url()
url=post.get_full_url() # 马莹:文章的完整链接
)
reply.add_article(article)
return reply
@ -48,10 +64,17 @@ def convert_to_article_reply(articles, message):
@robot.filter(re.compile(r"^\?.*"))
def search(message, session):
"""
处理文章搜索功能匹配以?开头的消息
:param message: 接收的消息对象
:param session: 会话对象
:return: 搜索结果的图文回复或提示信息
"""
s = message.content
searchstr = str(s).replace('?', '')
result = blogapi.search_articles(searchstr)
searchstr = str(s).replace('?', '') # 马莹:提取搜索关键词(去除开头的?
result = blogapi.search_articles(searchstr) # 马莹调用博客API搜索文章
if result:
# 马莹:将搜索结果转换为图文回复
articles = list(map(lambda x: x.object, result))
reply = convert_to_article_reply(articles, message)
return reply
@ -61,15 +84,29 @@ def search(message, session):
@robot.filter(re.compile(r'^category\s*$', re.I))
def category(message, session):
categorys = blogapi.get_category_lists()
"""
处理获取文章分类目录的功能匹配category不区分大小写
:param message: 接收的消息对象
:param session: 会话对象
:return: 分类目录字符串
"""
categorys = blogapi.get_category_lists() # 马莹调用博客API获取分类列表
# 马莹:将分类名称拼接为字符串返回
content = ','.join(map(lambda x: x.name, categorys))
return '所有文章分类目录:' + content
@robot.filter(re.compile(r'^recent\s*$', re.I))
def recents(message, session):
articles = blogapi.get_recent_articles()
"""
处理获取最新文章的功能匹配recent不区分大小写
:param message: 接收的消息对象
:param session: 会话对象
:return: 最新文章的图文回复或提示信息
"""
articles = blogapi.get_recent_articles() # 马莹调用博客API获取最新文章
if articles:
# 马莹:将最新文章转换为图文回复
reply = convert_to_article_reply(articles, message)
return reply
else:
@ -78,6 +115,12 @@ def recents(message, session):
@robot.filter(re.compile('^help$', re.I))
def help(message, session):
"""
处理帮助信息查询匹配help不区分大小写
:param message: 接收的消息对象
:param session: 会话对象
:return: 帮助信息字符串
"""
return '''欢迎关注!
默认会与图灵机器人聊天~~
你可以通过下面这些命令来获得信息
@ -100,88 +143,131 @@ def help(message, session):
@robot.filter(re.compile(r'^weather\:.*$', re.I))
def weather(message, session):
"""
处理天气查询功能建设中匹配以weather:开头的消息不区分大小写
:param message: 接收的消息对象
:param session: 会话对象
:return: 建设中提示
"""
return "建设中..."
@robot.filter(re.compile(r'^idcard\:.*$', re.I))
def idcard(message, session):
"""
处理身份证信息查询功能建设中匹配以idcard:开头的消息不区分大小写
:param message: 接收的消息对象
:param session: 会话对象
:return: 建设中提示
"""
return "建设中..."
@robot.handler
def echo(message, session):
"""
通用消息处理器处理未被上述过滤器匹配的消息
:param message: 接收的消息对象
:param session: 会话对象
:return: 消息处理结果
"""
handler = MessageHandler(message, session)
return handler.handler()
class MessageHandler:
"""消息处理类,处理用户消息并返回相应结果"""
def __init__(self, message, session):
userid = message.source
"""
初始化方法
:param message: 接收的消息对象
:param session: 会话对象
"""
userid = message.source # 获取用户ID
self.message = message
self.session = session
self.userid = userid
try:
# 马莹:从会话中获取用户信息,反序列化为对象
info = session[userid]
self.userinfo = jsonpickle.decode(info)
except Exception as e:
# 马莹:若会话中无用户信息,则初始化新的用户信息对象
userinfo = WxUserInfo()
self.userinfo = userinfo
@property
def is_admin(self):
"""判断当前用户是否为管理员"""
return self.userinfo.isAdmin
@property
def is_password_set(self):
"""判断管理员密码是否已设置"""
return self.userinfo.isPasswordSet
def save_session(self):
"""将用户信息序列化并保存到会话中"""
info = jsonpickle.encode(self.userinfo)
self.session[self.userid] = info
def handler(self):
info = self.message.content
"""处理消息的核心方法"""
info = self.message.content # 马莹:获取消息内容
# 马莹管理员退出功能若为管理员且消息为EXIT则退出管理员模式
if self.userinfo.isAdmin and info.upper() == 'EXIT':
self.userinfo = WxUserInfo()
self.save_session()
return "退出成功"
# 马莹进入管理员模式消息为ADMIN时标记用户为管理员待验证密码
if info.upper() == 'ADMIN':
self.userinfo.isAdmin = True
self.save_session()
return "输入管理员密码"
# 马莹:管理员密码验证:若为管理员且未设置密码,则验证输入的密码
if self.userinfo.isAdmin and not self.userinfo.isPasswordSet:
passwd = settings.WXADMIN
passwd = settings.WXADMIN # 马莹:从设置中获取管理员密码
if settings.TESTING:
passwd = '123'
passwd = '123' # 马莹测试环境下密码为123
# 马莹验证密码双重SHA256加密后比较
if passwd.upper() == get_sha256(get_sha256(info)).upper():
self.userinfo.isPasswordSet = True
self.userinfo.isPasswordSet = True # 标记密码已设置
self.save_session()
return "验证通过,请输入命令或者要执行的命令代码:输入helpme获得帮助"
else:
# 马莹密码错误处理超过3次则退出管理员模式
if self.userinfo.Count >= 3:
self.userinfo = WxUserInfo()
self.save_session()
return "超过验证次数"
self.userinfo.Count += 1
self.userinfo.Count += 1 # 增加错误次数
self.save_session()
return "验证失败,请重新输入管理员密码:"
# 马莹:管理员命令处理:已验证的管理员可执行命令
if self.userinfo.isAdmin and self.userinfo.isPasswordSet:
# 马莹确认执行命令若之前有命令且当前消息为Y则执行命令
if self.userinfo.Command != '' and info.upper() == 'Y':
return cmd_handler.run(self.userinfo.Command)
else:
# 马莹帮助信息输入helpme获取命令帮助
if info.upper() == 'HELPME':
return cmd_handler.get_help()
# 马莹:暂存命令并提示确认
self.userinfo.Command = info
self.save_session()
return "确认执行: " + info + " 命令?"
# 马莹普通用户消息调用ChatGPT处理
return ChatGPT.chat(info)
class WxUserInfo():
"""微信用户信息类,用于存储用户会话相关信息"""
def __init__(self):
self.isAdmin = False
self.isPasswordSet = False
self.Count = 0
self.Command = ''
self.isAdmin = False # 马莹:是否为管理员
self.isPasswordSet = False # 马莹:是否已设置管理员密码
self.Count = 0 # 马莹:密码错误次数
self.Command = '' # 马莹:待执行的命令

@ -1,7 +1,10 @@
# 马莹导入Django测试相关模块
from django.test import Client, RequestFactory, TestCase
from django.utils import timezone
# 马莹:导入微信机器人文本消息类
from werobot.messages.messages import TextMessage
# 马莹:导入项目内相关模型和工具类
from accounts.models import BlogUser
from blog.models import Category, Article
from servermanager.api.commonapi import ChatGPT
@ -11,69 +14,104 @@ from .robot import search, category, recents
# Create your tests here.
# 马莹定义服务器管理模块的测试类继承自Django的TestCase
class ServerManagerTest(TestCase):
def setUp(self):
"""测试初始化方法,在每个测试方法执行前运行"""
# 马莹创建测试客户端用于模拟HTTP请求
self.client = Client()
# 马莹:创建请求工厂,用于构造测试用的请求对象
self.factory = RequestFactory()
def test_chat_gpt(self):
"""测试ChatGPT聊天功能"""
# 马莹调用ChatGPT的chat方法发送"你好"消息
content = ChatGPT.chat("你好")
# 马莹:断言返回结果不为空
self.assertIsNotNone(content)
def test_validate_comment(self):
"""综合测试各类功能:文章搜索、分类、最新文章、命令执行、消息处理等"""
# 马莹:创建超级用户
user = BlogUser.objects.create_superuser(
email="liangliangyy1@gmail.com",
username="liangliangyy1",
password="liangliangyy1")
# 马莹:使用测试客户端登录该超级用户
self.client.login(username='liangliangyy1', password='liangliangyy1')
# 马莹:创建测试分类并保存到数据库
c = Category()
c.name = "categoryccc"
c.save()
# 马莹:创建测试文章并保存到数据库
article = Article()
article.title = "nicetitleccc"
article.body = "nicecontentccc"
article.author = user
article.category = c
article.type = 'a'
article.status = 'p'
article.author = user # 关联作者
article.category = c # 关联分类
article.type = 'a' # 文章类型
article.status = 'p' # 发布状态
article.save()
# 马莹:构造文本消息对象,内容为"nice"
s = TextMessage([])
s.content = "nice"
# 马莹:测试文章搜索功能
rsp = search(s, None)
# 马莹:测试获取分类功能
rsp = category(None, None)
# 马莹:断言分类功能返回结果不为空
self.assertIsNotNone(rsp)
# 马莹:测试获取最新文章功能
rsp = recents(None, None)
# 马莹:断言最新文章功能返回结果不是"暂时还没有文章"
self.assertTrue(rsp != '暂时还没有文章')
# 马莹:创建测试命令并保存到数据库
cmd = commands()
cmd.title = "test"
cmd.command = "ls"
cmd.describe = "test"
cmd.save()
# 马莹:实例化命令处理器
cmdhandler = CommandHandler()
# 马莹:测试执行命令功能
rsp = cmdhandler.run('test')
# 马莹:断言命令执行结果不为空
self.assertIsNotNone(rsp)
# 马莹设置消息的发送者ID
s.source = 'u'
# 马莹:设置消息内容为'test'
s.content = 'test'
# 马莹:实例化消息处理器
msghandler = MessageHandler(s, {})
# 马莹:以下注释代码为管理员权限相关配置(当前测试未启用)
# msghandler.userinfo.isPasswordSet = True
# msghandler.userinfo.isAdmin = True
# 马莹:处理消息内容'test'
msghandler.handler()
# 马莹:修改消息内容为'y'(模拟确认执行命令)
s.content = 'y'
msghandler.handler()
# 马莹:修改消息内容为身份证查询指令(建设中功能)
s.content = 'idcard:12321233'
msghandler.handler()
# 马莹:修改消息内容为天气查询指令(建设中功能)
s.content = 'weather:上海'
msghandler.handler()
# 马莹:修改消息内容为'admin'(进入管理员模式)
s.content = 'admin'
msghandler.handler()
# 马莹:修改消息内容为'123'(输入管理员密码)
s.content = '123'
msghandler.handler()
# 马莹:修改消息内容为'exit'(退出管理员模式)
s.content = 'exit'
msghandler.handler()
msghandler.handler()

@ -1,10 +1,17 @@
# 马莹导入Django的路径配置模块
from django.urls import path
# 马莹导入微信机器人框架适配Django的视图生成工具
from werobot.contrib.django import make_view
# 马莹:导入当前应用中定义的微信机器人实例
from .robot import robot
# 马莹定义应用的命名空间用于URL反向解析时区分不同应用的URL
app_name = "servermanager"
# 马莹URL路由配置列表
urlpatterns = [
# 马莹配置微信机器人的访问路径将机器人实例转换为Django可识别的视图
# 马莹:当访问 /servermanager/robot 路径时,由微信机器人处理请求
path(r'robot', make_view(robot)),
]

Loading…
Cancel
Save