Merge branch 'wjl_branch'

pull/7/head
ni-jie404 4 months ago
commit 6a585ee2ab

@ -5,6 +5,7 @@ from djangoblog.utils import cache # 项目中封装的缓存对象(如 Redis
class MemcacheStorage(SessionStorage):
# wjl
"""
自定义的会话存储类基于缓存系统 Redis/Memcached实现 WeRoBot 微信机器人的会话管理
用于在微信消息处理过程中持久化用户会话数据如上下文状态等
@ -13,6 +14,7 @@ class MemcacheStorage(SessionStorage):
"""
def __init__(self, prefix='ws_'):
# wjl
"""
初始化 MemcacheStorage 实例
@ -23,8 +25,8 @@ class MemcacheStorage(SessionStorage):
- 设置键名前缀
- 绑定项目全局的缓存实例cache
"""
self.prefix = prefix # 设置键名前缀
self.cache = cache # 使用项目中已配置的缓存后端(如 Redis
self.prefix = prefix #wjl# 设置键名前缀
self.cache = cache #wjl# 使用项目中已配置的缓存后端(如 Redis
@property
def is_available(self):
@ -40,10 +42,11 @@ class MemcacheStorage(SessionStorage):
bool: True 表示缓存可用False 表示不可用如缓存服务宕机
"""
value = "1"
self.set('checkavaliable', value=value) # 写入测试数据
return value == self.get('checkavaliable') # 验证能否正确读取
self.set('checkavaliable', value=value) #wjl# 写入测试数据
return value == self.get('checkavaliable') #wjl# 验证能否正确读取
def key_name(self, s):
# wjl
"""
为给定的会话 ID 生成带前缀的完整缓存键名
@ -59,6 +62,7 @@ class MemcacheStorage(SessionStorage):
return '{prefix}{s}'.format(prefix=self.prefix, s=s)
def get(self, id):
# wjl
"""
根据会话 ID 从缓存中获取会话数据
@ -74,10 +78,11 @@ class MemcacheStorage(SessionStorage):
3. 使用 json_loads JSON 字符串反序列化为 Python 字典
"""
id = self.key_name(id)
session_json = self.cache.get(id) or '{}' # 如果未找到,返回空 JSON 字符串
session_json = self.cache.get(id) or '{}' #wjl # 如果未找到,返回空 JSON 字符串
return json_loads(session_json)
def set(self, id, value):
# wjl
"""
将会话数据存储到缓存中
@ -91,9 +96,10 @@ class MemcacheStorage(SessionStorage):
3. 存入缓存使用默认过期时间
"""
id = self.key_name(id)
self.cache.set(id, json_dumps(value)) # 序列化并存储
self.cache.set(id, json_dumps(value)) #wjl# 序列化并存储
def delete(self, id):
# wjl
"""
根据会话 ID 从缓存中删除会话数据

@ -1,33 +1,35 @@
from django.contrib import admin
# Register your models here.
# 本文件用于将数据模型注册到 Django 管理后台Admin并自定义其展示和操作方式。
# 当前未立即注册模型,需在下方通过 admin.site.register() 显式注册。
#wjl# Register your models here.
#wjl# 本文件用于将数据模型注册到 Django 管理后台Admin并自定义其展示和操作方式。
#wjl# 当前未立即注册模型,需在下方通过 admin.site.register() 显式注册。
class CommandsAdmin(admin.ModelAdmin):
# wjl
"""
Django Admin 配置类用于自定义 'commands' 模型在管理后台中的列表展示
控制在模型列表页面显示哪些字段
"""
# 定义在管理后台的模型列表页面中要显示的字段列
# 用户将看到三列:'title'(命令标题)、'command'(命令内容)、'describe'(描述)
#wjl# 定义在管理后台的模型列表页面中要显示的字段列
#wjl# 用户将看到三列:'title'(命令标题)、'command'(命令内容)、'describe'(描述)
list_display = ('title', 'command', 'describe')
class EmailSendLogAdmin(admin.ModelAdmin):
# wjl
"""
Django Admin 配置类用于自定义 'EmailSendLog' 模型在管理后台中的行为
该日志模型主要用于查看禁止添加新记录并将所有字段设为只读
"""
# 定义在管理后台的模型列表页面中要显示的字段列
# 显示:邮件标题、收件人、发送结果、创建时间
#wjl# 定义在管理后台的模型列表页面中要显示的字段列
#wjl# 显示:邮件标题、收件人、发送结果、创建时间
list_display = ('title', 'emailto', 'send_result', 'creation_time')
# 定义在编辑或查看单条记录时,哪些字段为只读(不可编辑)
# 所有字段均设为只读,防止日志被意外修改
#wjl # 定义在编辑或查看单条记录时,哪些字段为只读(不可编辑)
#wjl# 所有字段均设为只读,防止日志被意外修改
readonly_fields = (
'title',
'emailto',
@ -36,13 +38,13 @@ class EmailSendLogAdmin(admin.ModelAdmin):
'content'
)
# 重写权限方法,禁止用户在管理后台添加新的 EmailSendLog 记录
# 因为邮件日志应由程序自动创建,不允许手动添加
#
# 参数:
# request: 当前的 HTTP 请求对象
#
# 返回:
# bool: 始终返回 False表示禁止添加操作
#wjl# 重写权限方法,禁止用户在管理后台添加新的 EmailSendLog 记录
#wjl# 因为邮件日志应由程序自动创建,不允许手动添加
#wjl#
#wjl# 参数:
#wjl # request: 当前的 HTTP 请求对象
#wjl#
#wjl# 返回:
#wjl# bool: 始终返回 False表示禁止添加操作
def has_add_permission(self, request):
return False

@ -4,11 +4,13 @@ from blog.models import Article, Category
class BlogApi:
#wjl
"""
博客 API 封装了与博客相关的数据查询功能包括全文搜索分类获取分类文章获取和最新文章获取
"""
def __init__(self):
#wjl
"""
初始化 BlogApi 实例
创建一个 SearchQuerySet 对象用于全文搜索并执行一个空查询以准备搜索环境
@ -19,6 +21,7 @@ class BlogApi:
self.__max_takecount__ = 8 # 设置最大返回文章数量为 8 篇
def search_articles(self, query):
#wjl
"""
根据用户输入的关键词查询文章
@ -37,6 +40,7 @@ class BlogApi:
return sqs[:self.__max_takecount__] # 返回最多 __max_takecount__ 条搜索结果
def get_category_lists(self):
#wjl
"""
获取所有文章分类的列表
@ -49,6 +53,7 @@ class BlogApi:
return Category.objects.all()
def get_category_articles(self, categoryname):
#wjl
"""
根据分类名称获取该分类下的文章列表
@ -68,6 +73,7 @@ class BlogApi:
return None # 无文章则返回 None
def get_recent_articles(self):
#wjl
"""
获取最新的文章列表按模型默认排序通常是按创建时间倒序

@ -5,18 +5,19 @@ import openai
from servermanager.models import commands
# 获取当前模块的 logger用于记录日志
#wjl# 获取当前模块的 logger用于记录日志
logger = logging.getLogger(__name__)
# 从环境变量中读取 OpenAI API 密钥
#wjl# 从环境变量中读取 OpenAI API 密钥
openai.api_key = os.environ.get('OPENAI_API_KEY')
# 如果设置了 HTTP 代理,则配置 openai 库使用该代理
#wjl# 如果设置了 HTTP 代理,则配置 openai 库使用该代理
if os.environ.get('HTTP_PROXY'):
openai.proxy = os.environ.get('HTTP_PROXY')
class ChatGPT:
#wjl
"""
封装与 OpenAI ChatGPT 模型进行交互的功能
提供静态方法 `chat` 用于发送用户提示并获取模型回复
@ -24,6 +25,7 @@ class ChatGPT:
@staticmethod
def chat(prompt):
# wjl
"""
调用 OpenAI GPT-3.5 Turbo 模型生成回复
@ -41,29 +43,32 @@ class ChatGPT:
"""
try:
completion = openai.ChatCompletion.create(
model="gpt-3.5-turbo", # 使用 GPT-3.5 Turbo 模型
messages=[{"role": "user", "content": prompt}] # 构造对话消息
model="gpt-3.5-turbo", #wjl# 使用 GPT-3.5 Turbo 模型
messages=[{"role": "user", "content": prompt}] #wjl# 构造对话消息
)
return completion.choices[0].message.content # 返回模型生成的文本
return completion.choices[0].message.content #wjl# 返回模型生成的文本
except Exception as e:
logger.error(e) # 记录异常信息到日志
return "服务器出错了" # 返回用户友好的错误提示
logger.error(e) #wjl# 记录异常信息到日志
return "服务器出错了" #wjl # 返回用户友好的错误提示
class CommandHandler:
# wjl
"""
命令处理器类用于管理查找和执行预定义的系统命令
从数据库加载命令列表支持通过名称查找并执行命令以及获取帮助信息
"""
def __init__(self):
# wjl
"""
初始化 CommandHandler 实例
从数据库中加载所有预定义的命令对象存储在实例变量 self.commands
"""
self.commands = commands.objects.all() # 查询数据库中所有命令记录
self.commands = commands.objects.all() #wjl# 查询数据库中所有命令记录
def run(self, title):
# wjl
"""
根据命令标题查找并执行对应的系统命令
@ -80,14 +85,15 @@ class CommandHandler:
"""
cmd = list(
filter(
lambda x: x.title.upper() == title.upper(), # 忽略大小写比较
lambda x: x.title.upper() == title.upper(), #wjl# 忽略大小写比较
self.commands))
if cmd:
return self.__run_command__(cmd[0].command) # 执行找到的命令
return self.__run_command__(cmd[0].command) #wjl# 执行找到的命令
else:
return "未找到相关命令请输入hepme获得帮助。"
def __run_command__(self, cmd):
# wjl
"""
执行给定的系统命令shell 命令
@ -102,12 +108,13 @@ class CommandHandler:
捕获所有异常包括执行失败权限问题等防止程序崩溃
"""
try:
res = os.popen(cmd).read() # 执行命令并读取输出
res = os.popen(cmd).read() #wjl# 执行命令并读取输出
return res
except BaseException:
return '命令执行出错!'
def get_help(self):
# wjl
"""
生成帮助信息列出所有可用的命令及其描述
@ -119,11 +126,12 @@ class CommandHandler:
"""
rsp = ''
for cmd in self.commands:
rsp += '{c}:{d}\n'.format(c=cmd.title, d=cmd.describe) # 拼接命令帮助信息
rsp += '{c}:{d}\n'.format(c=cmd.title, d=cmd.describe) #wjl# 拼接命令帮助信息
return rsp
if __name__ == '__main__':
# wjl
"""
主程序入口用于测试 ChatGPT 类的功能
创建 ChatGPT 实例发送一个测试提示并打印模型回复

@ -1,7 +1,7 @@
# 导入 AppConfig它是 Django 中用于配置应用程序的基类
#wjl# 导入 AppConfig它是 Django 中用于配置应用程序的基类
from django.apps import AppConfig
#wjl
class ServermanagerConfig(AppConfig):
"""
这是 'servermanager' 应用的配置类
@ -11,13 +11,13 @@ class ServermanagerConfig(AppConfig):
此类的作用是告诉 Django 这个应用的基本信息例如它的 Python 路径name
"""
# 指定该应用的完整 Python 导入路径
# Django 使用此属性来唯一标识和导入 'servermanager' 应用
# 通常与应用所在文件夹的名称一致
#wjl# 指定该应用的完整 Python 导入路径
#wjl# Django 使用此属性来唯一标识和导入 'servermanager' 应用
#wjl# 通常与应用所在文件夹的名称一致
name = 'servermanager'
# 注意:
# - 该类目前只设置了最基本的 name 属性。
# - 如果需要在应用启动时执行某些初始化代码,可以重写 ready() 方法。
# 例如:注册信号处理器、启动后台任务等。
# - 由于当前没有复杂的初始化需求,因此未定义其他方法或属性。
#wjl# 注意:
#wjl# - 该类目前只设置了最基本的 name 属性。
#wjl# - 如果需要在应用启动时执行某些初始化代码,可以重写 ready() 方法。
#wjl# 例如:注册信号处理器、启动后台任务等。
#wjl# - 由于当前没有复杂的初始化需求,因此未定义其他方法或属性。

@ -1,7 +1,9 @@
#wjl
from django.db import migrations, models
class Migration(migrations.Migration):
# wjl
"""
Django 迁移类定义了数据库模式的变更
此迁移是应用的初始迁移initial=True用于创建两个新的数据模型表
@ -9,60 +11,60 @@ class Migration(migrations.Migration):
- EmailSendLog记录邮件发送的日志信息
"""
# 标记此迁移为“初始迁移”,表示这是应用的第一个迁移文件
#wjl # 标记此迁移为“初始迁移”,表示这是应用的第一个迁移文件
initial = True
# 定义此迁移所依赖的其他迁移
# 当前为空列表,表示该迁移不依赖于任何其他迁移(即它是最初的)
#wjl# 定义此迁移所依赖的其他迁移
#wjl # 当前为空列表,表示该迁移不依赖于任何其他迁移(即它是最初的)
dependencies = [
]
# 定义此迁移要执行的数据库操作列表
#wjl# 定义此迁移要执行的数据库操作列表
operations = [
# 第一个操作:创建名为 'commands' 的数据模型表
migrations.CreateModel(
name='commands', # 数据库表对应的模型名称
fields=[ # 定义该模型包含的字段(即数据库表的列)
# 主键 ID 字段:自增的 BigAutoField
#wjl# 主键 ID 字段:自增的 BigAutoField
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
# 命令标题字段:最大长度 300 的字符串
#wjl# 命令标题字段:最大长度 300 的字符串
('title', models.CharField(max_length=300, verbose_name='命令标题')),
# 实际执行的命令字段:最大长度 2000 的字符串
#wjl# 实际执行的命令字段:最大长度 2000 的字符串
('command', models.CharField(max_length=2000, verbose_name='命令')),
# 命令描述字段:最大长度 300 的字符串
#wjl# 命令描述字段:最大长度 300 的字符串
('describe', models.CharField(max_length=300, verbose_name='命令描述')),
# 创建时间字段:自动在对象创建时设置当前时间
#wjl# 创建时间字段:自动在对象创建时设置当前时间
('created_time', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
# 最后修改时间字段:每次对象保存时自动更新为当前时间
#wjl# 最后修改时间字段:每次对象保存时自动更新为当前时间
('last_mod_time', models.DateTimeField(auto_now=True, verbose_name='修改时间')),
],
options={ # 模型的元选项配置
'verbose_name': '命令', # 单数形式的可读名称
'verbose_name_plural': '命令', # 复数形式的可读名称(中文通常与单数相同)
options={ #wjl# 模型的元选项配置
'verbose_name': '命令', #wjl# 单数形式的可读名称
'verbose_name_plural': '命令', #wjl# 复数形式的可读名称(中文通常与单数相同)
},
),
# 第二个操作:创建名为 'EmailSendLog' 的数据模型表
#wjl# 第二个操作:创建名为 'EmailSendLog' 的数据模型表
migrations.CreateModel(
name='EmailSendLog', # 数据库表对应的模型名称
fields=[ # 定义该模型包含的字段(即数据库表的列)
# 主键 ID 字段:自增的 BigAutoField
name='EmailSendLog', #wjl# 数据库表对应的模型名称
fields=[ #wjl# 定义该模型包含的字段(即数据库表的列)
#wjl# 主键 ID 字段:自增的 BigAutoField
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
# 收件人邮箱字段:最大长度 300 的字符串
#wjl# 收件人邮箱字段:最大长度 300 的字符串
('emailto', models.CharField(max_length=300, verbose_name='收件人')),
# 邮件标题字段:最大长度 2000 的字符串
#wjl# 邮件标题字段:最大长度 2000 的字符串
('title', models.CharField(max_length=2000, verbose_name='邮件标题')),
# 邮件内容字段:长文本字段
#wjl# 邮件内容字段:长文本字段
('content', models.TextField(verbose_name='邮件内容')),
# 发送结果字段:布尔值,默认为 False表示失败
#wjl# 发送结果字段:布尔值,默认为 False表示失败
('send_result', models.BooleanField(default=False, verbose_name='结果')),
# 创建时间字段:自动在记录创建时设置当前时间
#wjl# 创建时间字段:自动在记录创建时设置当前时间
('created_time', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
],
options={ # 模型的元选项配置
'verbose_name': '邮件发送log', # 单数形式的可读名称
'verbose_name_plural': '邮件发送log', # 复数形式的可读名称
'ordering': ['-created_time'], # 默认排序规则:按创建时间倒序排列(最新的在前)
'verbose_name': '邮件发送log', #wjl# 单数形式的可读名称
'verbose_name_plural': '邮件发送log', #wjl# 复数形式的可读名称
'ordering': ['-created_time'], #wjl# 默认排序规则:按创建时间倒序排列(最新的在前)
},
),
]

@ -2,50 +2,51 @@ from django.db import migrations
class Migration(migrations.Migration):
# wjl
"""
Django 迁移类定义了对数据库模式的变更
此迁移文件基于 'servermanager' 应用的初始迁移0001_initial进行修改
主要目的是统一字段命名规范如将 created_time 改为 creation_time并更新模型排序规则
"""
# 定义此迁移所依赖的其他迁移
# 依赖于 servermanager 应用的 0001_initial 迁移,确保基础表已存在
#wjl# 定义此迁移所依赖的其他迁移
#wjl# 依赖于 servermanager 应用的 0001_initial 迁移,确保基础表已存在
dependencies = [
('servermanager', '0001_initial'), # 格式:(应用名, 迁移文件名)
('servermanager', '0001_initial'), #wjl# 格式:(应用名, 迁移文件名)
]
# 定义此迁移要执行的数据库操作列表
#wjl# 定义此迁移要执行的数据库操作列表
operations = [
# 操作 1修改模型 EmailSendLog 的元选项Meta options
#wjl# 操作 1修改模型 EmailSendLog 的元选项Meta options
migrations.AlterModelOptions(
name='emailsendlog', # 要修改的模型名称
name='emailsendlog', #wjl# 要修改的模型名称
options={
# 更新排序字段为 'creation_time'(倒序),确保最新记录排在前面
#wjl# 更新排序字段为 'creation_time'(倒序),确保最新记录排在前面
'ordering': ['-creation_time'],
# 保持模型的可读名称不变
#wjl# 保持模型的可读名称不变
'verbose_name': '邮件发送log',
'verbose_name_plural': '邮件发送log'
},
),
# 操作 2将 commands 模型中的字段 created_time 重命名为 creation_time
#wjl # 操作 2将 commands 模型中的字段 created_time 重命名为 creation_time
migrations.RenameField(
model_name='commands', # 要操作的模型名称
old_name='created_time', # 原字段名
new_name='creation_time', # 新字段名
model_name='commands', #wjl# 要操作的模型名称
old_name='created_time', #wjl # 原字段名
new_name='creation_time', #wjl# 新字段名
),
# 操作 3将 commands 模型中的字段 last_mod_time 重命名为 last_modify_time
#wjl# 操作 3将 commands 模型中的字段 last_mod_time 重命名为 last_modify_time
migrations.RenameField(
model_name='commands', # 要操作的模型名称
old_name='last_mod_time', # 原字段名
new_name='last_modify_time', # 新字段名
model_name='commands', #wjl# 要操作的模型名称
old_name='last_mod_time', #wjl # 原字段名
new_name='last_modify_time', #wjl# 新字段名
),
# 操作 4将 emailsendlog 模型中的字段 created_time 重命名为 creation_time
#wjl# 操作 4将 emailsendlog 模型中的字段 created_time 重命名为 creation_time
migrations.RenameField(
model_name='emailsendlog', # 要操作的模型名称
old_name='created_time', # 原字段名
new_name='creation_time', # 新字段名
model_name='emailsendlog', #wjl# 要操作的模型名称
old_name='created_time', #wjl# 原字段名
new_name='creation_time', #wjl# 新字段名
),
]

@ -1,35 +1,37 @@
from django.db import models
# Create your models here.
# 定义了两个数据模型:
# 1. commands用于存储可执行的系统命令
# 2. EmailSendLog用于记录邮件发送的历史日志
#wjl# Create your models here.
#wjl# 定义了两个数据模型:
#wjl# 1. commands用于存储可执行的系统命令
#wjl# 2. EmailSendLog用于记录邮件发送的历史日志
class commands(models.Model):
# wjl
"""
数据模型commands
用于存储系统中预定义的可执行命令例如服务器管理命令脚本路径等
可通过 Django Admin 管理后台进行增删改查
"""
# 命令的标题/名称,用于标识该命令(如 "重启服务"
#wjl# 命令的标题/名称,用于标识该命令(如 "重启服务"
title = models.CharField('命令标题', max_length=300)
# 实际要执行的命令字符串(如 "systemctl restart nginx"
#wjl# 实际要执行的命令字符串(如 "systemctl restart nginx"
command = models.CharField('命令', max_length=2000)
# 对该命令的简要描述,说明其用途
#wjl# 对该命令的简要描述,说明其用途
describe = models.CharField('命令描述', max_length=300)
# 记录该命令首次创建的时间,自动在创建时设置为当前时间
#wjl# 记录该命令首次创建的时间,自动在创建时设置为当前时间
creation_time = models.DateTimeField('创建时间', auto_now_add=True)
# 记录该命令最后一次修改的时间,每次保存时自动更新为当前时间
#wjl# 记录该命令最后一次修改的时间,每次保存时自动更新为当前时间
last_modify_time = models.DateTimeField('修改时间', auto_now=True)
def __str__(self):
# wjl
"""
返回该模型实例的字符串表示
Django Admin 或外键显示时将显示命令的标题
@ -40,37 +42,40 @@ class commands(models.Model):
return self.title
class Meta:
# wjl
"""
模型元数据配置
定义模型的可读名称用于在 Django Admin 后台显示
"""
verbose_name = '命令' # 单数形式的名称
verbose_name_plural = verbose_name # 复数形式的名称(中文通常与单数相同)
verbose_name = '命令' #wjl# 单数形式的名称
verbose_name_plural = verbose_name #wjl #wjl# 复数形式的名称(中文通常与单数相同)
class EmailSendLog(models.Model):
# wjl
"""
数据模型EmailSendLog
用于记录系统发送邮件的日志信息包括收件人内容结果和时间
所有字段均为只读仅用于查看历史记录不可手动修改
"""
# 邮件收件人地址(支持多个,以逗号分隔)
#wjl# 邮件收件人地址(支持多个,以逗号分隔)
emailto = models.CharField('收件人', max_length=300)
# 邮件标题
#wjl# 邮件标题
title = models.CharField('邮件标题', max_length=2000)
# 邮件正文内容,支持长文本
#wjl # 邮件正文内容,支持长文本
content = models.TextField('邮件内容')
# 邮件是否发送成功True 表示成功False 表示失败
#wjl# 邮件是否发送成功True 表示成功False 表示失败
send_result = models.BooleanField('结果', default=False)
# 邮件记录创建时间,自动在插入时设置为当前时间
#wjl# 邮件记录创建时间,自动在插入时设置为当前时间
creation_time = models.DateTimeField('创建时间', auto_now_add=True)
def __str__(self):
# wjl
"""
返回该模型实例的字符串表示
Django Admin 或外键显示时将显示邮件的标题
@ -81,10 +86,11 @@ class EmailSendLog(models.Model):
return self.title
class Meta:
# wjl
"""
模型元数据配置
定义模型的可读名称和默认排序规则
"""
verbose_name = '邮件发送log' # 单数形式的名称
verbose_name_plural = verbose_name # 复数形式的名称
ordering = ['-creation_time'] # 默认排序:按创建时间倒序排列(最新的在前)
verbose_name = '邮件发送log' #wjl# 单数形式的名称
verbose_name_plural = verbose_name #wjl # 复数形式的名称
ordering = ['-creation_time'] #wjl# 默认排序:按创建时间倒序排列(最新的在前)

@ -15,35 +15,36 @@ from servermanager.api.commonapi import ChatGPT, CommandHandler # 聊天与命
from .MemcacheStorage import MemcacheStorage # 自定义基于缓存的会话存储
# 初始化微信机器人实例
#wjl# 初始化微信机器人实例
robot = WeRoBot(
token=os.environ.get('DJANGO_WEROBOT_TOKEN') or 'lylinux', # 微信公众号 Token
enable_session=True # 启用会话功能,用于维护用户状态
token=os.environ.get('DJANGO_WEROBOT_TOKEN') or 'lylinux', #wjl# 微信公众号 Token
enable_session=True #wjl# 启用会话功能,用于维护用户状态
)
# 创建自定义的缓存会话存储实例(基于 Redis/Memcached
#wjl# 创建自定义的缓存会话存储实例(基于 Redis/Memcached
memstorage = MemcacheStorage()
# 检查缓存存储是否可用
#wjl# 检查缓存存储是否可用
if memstorage.is_available:
# 如果可用,使用 MemcacheStorage 作为会话后端
#wjl# 如果可用,使用 MemcacheStorage 作为会话后端
robot.config['SESSION_STORAGE'] = memstorage
else:
# 如果不可用,回退到文件存储
# 删除旧的会话文件(避免冲突或损坏)
#wjl# 如果不可用,回退到文件存储
#wjl # 删除旧的会话文件(避免冲突或损坏)
session_file = os.path.join(settings.BASE_DIR, 'werobot_session')
if os.path.exists(session_file):
os.remove(session_file)
# 使用本地文件存储会话数据
#wjl # 使用本地文件存储会话数据
robot.config['SESSION_STORAGE'] = FileStorage(filename='werobot_session')
# 实例化博客 API 和命令处理器
blogapi = BlogApi() # 提供博客文章查询服务
cmd_handler = CommandHandler() # 提供服务器命令执行功能
logger = logging.getLogger(__name__) # 获取当前模块的日志记录器
#wjl# 实例化博客 API 和命令处理器
blogapi = BlogApi() #wjl # 提供博客文章查询服务
cmd_handler = CommandHandler()#wjl # 提供服务器命令执行功能
logger = logging.getLogger(__name__) #wjl# 获取当前模块的日志记录器
def convert_to_article_reply(articles, message):
# wjl
"""
将博客文章列表转换为微信图文消息回复格式
@ -60,18 +61,18 @@ def convert_to_article_reply(articles, message):
3. 构造 Article 对象并添加到回复中
"""
reply = ArticlesReply(message=message)
from blog.templatetags.blog_tags import truncatechars_content # 导入截断内容的模板标签
from blog.templatetags.blog_tags import truncatechars_content #wjl# 导入截断内容的模板标签
for post in articles:
# 使用正则从文章内容中提取第一张图片 URLpng/jpg
#wjl# 使用正则从文章内容中提取第一张图片 URLpng/jpg
imgs = re.findall(r'(?:http\:|https\:)?\/\/.*\.(?:png|jpg)', post.body)
imgurl = imgs[0] if imgs else '' # 如果有图取第一张,否则为空
imgurl = imgs[0] if imgs else '' #wjl# 如果有图取第一张,否则为空
article = Article(
title=post.title,
description=truncatechars_content(post.body), # 截断内容作为描述
description=truncatechars_content(post.body), #wjl# 截断内容作为描述
img=imgurl,
url=post.get_full_url() # 文章完整 URL
url=post.get_full_url() #wjl# 文章完整 URL
)
reply.add_article(article)
return reply
@ -79,6 +80,7 @@ def convert_to_article_reply(articles, message):
@robot.filter(re.compile(r"^\?.*"))
def search(message, session):
# wjl
"""
处理以 '?' 开头的消息用于搜索博客文章
@ -104,6 +106,7 @@ def search(message, session):
@robot.filter(re.compile(r'^category\s*$', re.I))
def category(message, session):
# wjl
"""
处理 "category" 消息返回所有文章分类目录
@ -115,12 +118,13 @@ def category(message, session):
文本消息列出所有分类名称
"""
categorys = blogapi.get_category_lists()
content = ','.join(map(lambda x: x.name, categorys)) # 拼接分类名
content = ','.join(map(lambda x: x.name, categorys)) #wjl# 拼接分类名
return '所有文章分类目录:' + content
@robot.filter(re.compile(r'^recent\s*$', re.I))
def recents(message, session):
# wjl
"""
处理 "recent" 消息返回最新发布的文章
@ -141,6 +145,7 @@ def recents(message, session):
@robot.filter(re.compile('^help$', re.I))
def help(message, session):
# wjl
"""
处理 "help" 消息返回帮助文档
@ -173,6 +178,7 @@ def help(message, session):
@robot.filter(re.compile(r'^weather\:.*$', re.I))
def weather(message, session):
# wjl
"""
处理 "weather:" 开头的消息天气查询功能
当前为占位符功能正在建设中
@ -189,6 +195,7 @@ def weather(message, session):
@robot.filter(re.compile(r'^idcard\:.*$', re.I))
def idcard(message, session):
# wjl
"""
处理 "idcard:" 开头的消息身份证信息查询功能
当前为占位符功能正在建设中
@ -205,6 +212,7 @@ def idcard(message, session):
@robot.handler
def echo(message, session):
# wjl
"""
默认消息处理器当没有其他 filter 匹配时调用
创建 MessageHandler 实例处理消息
@ -221,6 +229,7 @@ def echo(message, session):
class MessageHandler:
# wjl
"""
消息处理器类负责处理用户消息尤其是管理员命令和认证流程
使用会话session维护用户状态是否管理员是否已认证等
@ -229,28 +238,29 @@ class MessageHandler:
def __init__(self, message, session):
self.message = message
self.session = session
self.userid = message.source # 用户唯一标识OpenID
self.userid = message.source #wjl# 用户唯一标识OpenID
# 尝试从会话中加载用户信息
try:
info = session[self.userid]
self.userinfo = jsonpickle.decode(info) # 反序列化为 WxUserInfo 对象
self.userinfo = jsonpickle.decode(info) #wjl# 反序列化为 WxUserInfo 对象
except Exception as e:
# 如果出错(首次访问或会话丢失),创建默认用户信息
#wjl# 如果出错(首次访问或会话丢失),创建默认用户信息
userinfo = WxUserInfo()
self.userinfo = userinfo
@property
def is_admin(self):
"""判断当前用户是否为管理员"""
#wjl"""判断当前用户是否为管理员"""
return self.userinfo.isAdmin
@property
def is_password_set(self):
"""判断当前管理员是否已完成密码验证"""
#wjl"""判断当前管理员是否已完成密码验证"""
return self.userinfo.isPasswordSet
def save_session(self):
# wjl
"""
将当前用户信息保存回会话
使用 jsonpickle 序列化对象并存入 session
@ -259,6 +269,7 @@ class MessageHandler:
self.session[self.userid] = info
def handler(self):
# wjl
"""
核心消息处理逻辑根据用户状态和输入内容返回相应响应
@ -274,65 +285,66 @@ class MessageHandler:
"""
info = self.message.content
# 退出管理员模式
#wjl# 退出管理员模式
if self.userinfo.isAdmin and info.upper() == 'EXIT':
self.userinfo = WxUserInfo()
self.save_session()
return "退出成功"
# 请求进入管理员模式
#wjl# 请求进入管理员模式
if info.upper() == 'ADMIN':
self.userinfo.isAdmin = True
self.save_session()
return "输入管理员密码"
# 管理员密码验证阶段
#wjl# 管理员密码验证阶段
if self.userinfo.isAdmin and not self.userinfo.isPasswordSet:
# 获取配置中的管理员密码SHA256 加密后)
#wjl# 获取配置中的管理员密码SHA256 加密后)
passwd = settings.WXADMIN
if settings.TESTING:
passwd = '123' # 测试环境下使用简单密码
# 验证用户输入的密码(双重 SHA256
passwd = '123' #wjl# 测试环境下使用简单密码
#wjl# 验证用户输入的密码(双重 SHA256
if passwd.upper() == get_sha256(get_sha256(info)).upper():
self.userinfo.isPasswordSet = True
self.save_session()
return "验证通过,请输入命令或者要执行的命令代码:输入helpme获得帮助"
return #wjl"验证通过,请输入命令或者要执行的命令代码:输入helpme获得帮助"
else:
# 密码错误次数限制
#wjl# 密码错误次数限制
if self.userinfo.Count >= 3:
self.userinfo = WxUserInfo()
self.save_session()
return "超过验证次数"
return #wjl"超过验证次数"
self.userinfo.Count += 1
self.save_session()
return "验证失败,请重新输入管理员密码:"
return #wjl"验证失败,请重新输入管理员密码:"
# 管理员已认证,可执行命令
#wjl # 管理员已认证,可执行命令
if self.userinfo.isAdmin and self.userinfo.isPasswordSet:
if self.userinfo.Command != '' and info.upper() == 'Y':
# 用户确认执行命令
#wjl# 用户确认执行命令
return cmd_handler.run(self.userinfo.Command)
else:
if info.upper() == 'HELPME':
# 显示命令帮助
#wjl# 显示命令帮助
return cmd_handler.get_help()
# 记录待执行的命令,等待用户确认
#wjl# 记录待执行的命令,等待用户确认
self.userinfo.Command = info
self.save_session()
return "确认执行: " + info + " 命令?"
# 默认行为:调用 ChatGPT 进行普通聊天
#wjl # 默认行为:调用 ChatGPT 进行普通聊天
return ChatGPT.chat(info)
class WxUserInfo():
# wjl
"""
微信用户信息类用于在会话中存储用户状态
包括是否为管理员是否已通过密码验证尝试次数待执行命令等
"""
def __init__(self):
self.isAdmin = False # 是否请求成为管理员
self.isPasswordSet = False # 是否已通过密码验证
self.Count = 0 # 密码尝试次数
self.Command = '' # 待执行的命令
self.isAdmin = False #wjl # 是否请求成为管理员
self.isPasswordSet = False #wjl# 是否已通过密码验证
self.Count = 0 #wjl # 密码尝试次数
self.Command = '' #wjl# 待执行的命令

@ -1,19 +1,20 @@
from django.test import Client, RequestFactory, TestCase
from django.utils import timezone # 用于处理时间相关的测试
from django.utils import timezone #wjl# 用于处理时间相关的测试
# 导入 WeRoBot 消息类,用于模拟微信消息
#wjl# 导入 WeRoBot 消息类,用于模拟微信消息
from werobot.messages.messages import TextMessage
# 导入项目中的模型和 API
from accounts.models import BlogUser # 用户模型(用于创建管理员)
from blog.models import Category, Article # 博客分类和文章模型
from servermanager.api.commonapi import ChatGPT # 聊天功能接口
from .models import commands # 命令模型
from .robot import MessageHandler, CommandHandler # 核心机器人处理器
from .robot import search, category, recents # 微信机器人命令函数
#wjl# 导入项目中的模型和 API
from accounts.models import BlogUser #wjl# 用户模型(用于创建管理员)
from blog.models import Category, Article #wjl # 博客分类和文章模型
from servermanager.api.commonapi import ChatGPT #wjl# 聊天功能接口
from .models import commands #wjl# 命令模型
from .robot import MessageHandler, CommandHandler #wjl# 核心机器人处理器
from .robot import search, category, recents #wjl# 微信机器人命令函数
class ServerManagerTest(TestCase):
# wjl
"""
Django 测试用例类用于对 servermanager 应用的核心功能进行单元测试
测试内容包括
@ -24,6 +25,7 @@ class ServerManagerTest(TestCase):
"""
def setUp(self):
# wjl
"""
在每个测试方法执行前自动运行的初始化方法
设置测试所需的公共环境
@ -32,10 +34,11 @@ class ServerManagerTest(TestCase):
注意虽然此处创建了 factory但在当前测试中并未实际使用
"""
self.client = Client() # Django 测试客户端
self.factory = RequestFactory() # 请求工厂,用于创建模拟请求
self.client = Client() #wjl# Django 测试客户端
self.factory = RequestFactory() #wjl# 请求工厂,用于创建模拟请求
def test_chat_gpt(self):
# wjl
"""
测试 ChatGPT 聊天功能是否正常工作
@ -49,6 +52,7 @@ class ServerManagerTest(TestCase):
self.assertIsNotNone(content)
def test_validate_comment(self):
# wjl
"""
综合测试方法覆盖多个功能点
名称 'validate_comment' 不准确实际测试的是 servermanager 的核心功能
@ -61,88 +65,88 @@ class ServerManagerTest(TestCase):
5. 模拟完整管理员会话流程登录执行命令退出等
"""
# 1. 创建超级用户
#wjl # 1. 创建超级用户
user = BlogUser.objects.create_superuser(
email="liangliangyy1@gmail.com",
username="liangliangyy1",
password="liangliangyy1")
# 使用测试客户端登录该用户
#wjl# 使用测试客户端登录该用户
self.client.login(username='liangliangyy1', password='liangliangyy1')
# 2. 创建博客分类
#wjl# 2. 创建博客分类
c = Category()
c.name = "categoryccc"
c.save()
# 3. 创建一篇已发布的文章
#wjl# 3. 创建一篇已发布的文章
article = Article()
article.title = "nicetitleccc"
article.body = "nicecontentccc"
article.author = user
article.category = c
article.type = 'a' # 文章类型
article.status = 'p' # 发布状态
article.type = 'a' #wjl# 文章类型
article.status = 'p' #wjl # 发布状态
article.save()
# 4. 模拟微信文本消息
#wjl# 4. 模拟微信文本消息
s = TextMessage([])
s.content = "nice" # 搜索关键词
s.content = "nice" #wjl# 搜索关键词
# 5. 测试 search 命令
rsp = search(s, None) # 调用 search 过滤器
# 断言有响应(即使未找到文章也应返回提示)
#wjl # 5. 测试 search 命令
#wjlrsp = search(s, None) # 调用 search 过滤器
#wjl # 断言有响应(即使未找到文章也应返回提示)
# 6. 测试 category 命令
#wjl # 6. 测试 category 命令
rsp = category(None, None)
self.assertIsNotNone(rsp) # 确保返回了分类列表
# 7. 测试 recents 命令
#wjl# 7. 测试 recents 命令
rsp = recents(None, None)
# 断言返回结果不是“暂时还没有文章”,说明能获取到文章
#wjl# 断言返回结果不是“暂时还没有文章”,说明能获取到文章
self.assertTrue(rsp != '暂时还没有文章')
# 8. 测试命令执行功能
#wjl # 8. 测试命令执行功能
cmd = commands()
cmd.title = "test"
cmd.command = "ls" # 测试命令(列出目录)
cmd.command = "ls" #wjl # 测试命令(列出目录)
cmd.describe = "test"
cmd.save() # 保存到数据库
cmd.save() #wjl # 保存到数据库
# 实例化命令处理器
#wjl# 实例化命令处理器
cmdhandler = CommandHandler()
# 执行名为 'test' 的命令
#wjl # 执行名为 'test' 的命令
rsp = cmdhandler.run('test')
# 断言命令执行有返回结果
#wjl# 断言命令执行有返回结果
self.assertIsNotNone(rsp)
# 9. 模拟管理员会话流程
s.source = 'u' # 设置用户标识OpenID
s.content = 'test' # 用户输入内容
#wjl# 9. 模拟管理员会话流程
s.source = 'u' #wjl# 设置用户标识OpenID
s.content = 'test' #wjl # 用户输入内容
# 创建消息处理器实例,传入模拟消息和空会话
#wjl# 创建消息处理器实例,传入模拟消息和空会话
msghandler = MessageHandler(s, {})
# 模拟用户行为序列:
msghandler.handler() # 处理 "test" 消息(应提示确认)
#wjl# 模拟用户行为序列:
msghandler.handler() #wjl# 处理 "test" 消息(应提示确认)
s.content = 'y' # 用户确认执行
msghandler.handler() # 执行命令
s.content = 'y' #wjl# 用户确认执行
msghandler.handler() #wjl# 执行命令
s.content = 'idcard:12321233' # 尝试身份证查询(占位功能)
s.content = 'idcard:12321233' #wjl# 尝试身份证查询(占位功能)
msghandler.handler()
s.content = 'weather:上海' # 尝试天气查询(占位功能)
s.content = 'weather:上海' #wjl# 尝试天气查询(占位功能)
msghandler.handler()
s.content = 'admin' # 请求进入管理员模式
s.content = 'admin' #wjl# 请求进入管理员模式
msghandler.handler()
s.content = '123' # 输入密码(测试环境下有效)
s.content = '123' #wjl# 输入密码(测试环境下有效)
msghandler.handler()
s.content = 'exit' # 退出管理员模式
s.content = 'exit' #wjl # 退出管理员模式
msghandler.handler()
# 注意:此测试未包含 assert 断言来验证每个步骤的结果,
# 主要是通过调用来检查是否抛出异常,确保代码路径可执行。
#wjl# 注意:此测试未包含 assert 断言来验证每个步骤的结果,
#wjl# 主要是通过调用来检查是否抛出异常,确保代码路径可执行。

@ -1,4 +1,4 @@
# 导入 Django URL 路由模块
#wjl# 导入 Django URL 路由模块
from django.urls import path
# 导入 WeRoBot 与 Django 集成的工具函数
from werobot.contrib.django import make_view
@ -6,17 +6,18 @@ from werobot.contrib.django import make_view
# 从当前应用的 robot 模块导入已配置的 WeRoBot 实例
from .robot import robot
# 定义应用命名空间
#wjl# 定义应用命名空间
app_name = "servermanager"
"""
应用命名空间用于在 Django 项目中唯一标识此应用的 URL
在模板或 reverse() 函数中可通过 'servermanager:xxx' 引用此应用的 URL
"""
# 定义 URL 路由列表
#wjl# 定义 URL 路由列表
urlpatterns = [
# 将微信机器人接入点绑定到特定 URL 路径
#wjl# 将微信机器人接入点绑定到特定 URL 路径
path(r'robot', make_view(robot)),
#wjl
"""
URL 路由配置

@ -1,4 +1,4 @@
# Create your views here.
#wjl# Create your views here.
"""
视图模块views.py
"""
Loading…
Cancel
Save