代码注释 #6

Open
pbytuefo4 wants to merge 0 commits from yxy_branch into develop

@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="FacetManager">
<facet type="django" name="Django">
<configuration>
<option name="rootFolder" value="$MODULE_DIR$/djangoblog/src/DjangoBlog-master/DjangoBlog-master" />
<option name="settingsModule" value="settings.py" />
<option name="manageScript" value="$MODULE_DIR$/djangoblog/src/DjangoBlog-master/DjangoBlog-master/manage.py" />
<option name="environment" value="&lt;map/&gt;" />
<option name="doNotUseTestRunner" value="false" />
<option name="trackFilePattern" value="migrations" />
</configuration>
</facet>
</component>
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$/djangoblog/src/DjangoBlog-master/DjangoBlog-master" isTestSource="false" />
</content>
<orderEntry type="jdk" jdkName="Python 3.12 (PythonProject1)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
<component name="PyDocumentationSettings">
<option name="format" value="PLAIN" />
<option name="myDocStringFormat" value="Plain" />
</component>
</module>

@ -1,7 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="Python 3.12" />
<option name="sdkName" value="Python 3.12 (PythonProject1)" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12" project-jdk-type="Python SDK" />
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12 (PythonProject1)" project-jdk-type="Python SDK" />
</project>

@ -2,7 +2,7 @@
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/zyl_django.iml" filepath="$PROJECT_DIR$/.idea/zyl_django.iml" />
<module fileurl="file://$PROJECT_DIR$/.idea/DjangoBlog-yxy_branch.iml" filepath="$PROJECT_DIR$/.idea/DjangoBlog-yxy_branch.iml" />
</modules>
</component>
</project>

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>

@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="Python 3.12" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12" project-jdk-type="Python SDK" />
</project>

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/zyl_django.iml" filepath="$PROJECT_DIR$/.idea/zyl_django.iml" />
</modules>
</component>
</project>

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>

@ -0,0 +1,8 @@
# 默认忽略的文件
/shelf/
/workspace.xml
# 基于编辑器的 HTTP 客户端请求
/httpRequests/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml

@ -0,0 +1,6 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>
</component>

@ -0,0 +1,189 @@
# 导入Django表单模块用于创建自定义表单
from django import forms
# 导入Django admin模块用于注册模型到后台管理系统
from django.contrib import admin
# 导入获取用户模型的函数,用于处理作者关联
from django.contrib.auth import get_user_model
# 导入reverse函数用于生成URL
from django.urls import reverse
# 导入format_html用于在admin中生成HTML代码
from django.utils.html import format_html
# 导入国际化工具,用于翻译后台显示文本
from django.utils.translation import gettext_lazy as _
# 导入当前应用的Article模型
from .models import Article
class ArticleForm(forms.ModelForm):
"""
自定义文章表单用于在admin中自定义文章的编辑界面
可以在这里添加自定义字段验证 widgets 或修改表单行为
目前注释掉了pagedown编辑器的配置如需使用可取消注释
"""
# body = forms.CharField(widget=AdminPagedownWidget()) # 富文本编辑器配置
class Meta:
model = Article # 关联的模型
fields = '__all__' # 包含模型的所有字段
# 自定义批量操作:发布选中的文章
def makr_article_publish(modeladmin, request, queryset):
# 将选中文章的状态更新为'p'(published)
queryset.update(status='p')
# 自定义批量操作:将选中的文章设为草稿
def draft_article(modeladmin, request, queryset):
# 将选中文章的状态更新为'd'(draft)
queryset.update(status='d')
# 自定义批量操作:关闭选中文章的评论
def close_article_commentstatus(modeladmin, request, queryset):
# 将选中文章的评论状态更新为'c'(closed)
queryset.update(comment_status='c')
# 自定义批量操作:开启选中文章的评论
def open_article_commentstatus(modeladmin, request, queryset):
# 将选中文章的评论状态更新为'o'(open)
queryset.update(comment_status='o')
# 为批量操作设置显示名称(支持国际化)
makr_article_publish.short_description = _('Publish selected articles')
draft_article.short_description = _('Draft selected articles')
close_article_commentstatus.short_description = _('Close article comments')
open_article_commentstatus.short_description = _('Open article comments')
class ArticlelAdmin(admin.ModelAdmin):
"""
文章模型的Admin配置类自定义文章在后台的显示和操作方式
"""
list_per_page = 20 # 每页显示20条记录
search_fields = ('body', 'title') # 可搜索的字段
form = ArticleForm # 使用自定义的表单
# 列表页显示的字段
list_display = (
'id', # 文章ID
'title', # 标题
'author', # 作者
'link_to_category', # 分类(带链接)
'creation_time', # 创建时间
'views', # 浏览量
'status', # 状态
'type', # 类型(文章/页面)
'article_order' # 排序序号
)
# 列表页可点击跳转编辑的字段
list_display_links = ('id', 'title')
# 可筛选的字段(右侧过滤器)
list_filter = ('status', 'type', 'category')
# 多对多字段的水平选择器
filter_horizontal = ('tags',)
# 编辑页排除的字段(这些字段通常自动生成,不需要手动编辑)
exclude = ('creation_time', 'last_modify_time')
# 启用"在站点上查看"功能
view_on_site = True
# 注册批量操作
actions = [
makr_article_publish,
draft_article,
close_article_commentstatus,
open_article_commentstatus
]
def link_to_category(self, obj):
"""
自定义列表字段显示分类并添加跳转链接到分类编辑页
Args:
obj: 当前文章对象
Returns:
HTML代码带链接的分类名称
"""
# 获取分类模型的元数据用于生成URL
info = (obj.category._meta.app_label, obj.category._meta.model_name)
# 生成分类编辑页的URL
link = reverse('admin:%s_%s_change' % info, args=(obj.category.id,))
# 返回带链接的HTML
return format_html(u'<a href="%s">%s</a>' % (link, obj.category.name))
# 自定义字段的显示名称
link_to_category.short_description = _('category')
def get_form(self, request, obj=None, **kwargs):
"""
重写表单获取方法自定义表单字段
这里限制了作者只能选择超级用户
"""
form = super(ArticlelAdmin, self).get_form(request, obj, **kwargs)
# 作者字段只显示超级用户
form.base_fields['author'].queryset = get_user_model().objects.filter(is_superuser=True)
return form
def save_model(self, request, obj, form, change):
"""
重写保存模型的方法
可以在这里添加额外的保存逻辑如自动填充某些字段
目前使用默认实现
"""
super(ArticlelAdmin, self).save_model(request, obj, form, change)
def get_view_on_site_url(self, obj=None):
"""
自定义"在站点上查看"的链接
Args:
obj: 文章对象
Returns:
文章的前台访问URL或网站首页
"""
if obj:
# 如果有文章对象返回文章的完整URL
url = obj.get_full_url()
return url
else:
# 如果没有对象(如在列表页),返回网站首页
from djangoblog.utils import get_current_site
site = get_current_site().domain
return site
class TagAdmin(admin.ModelAdmin):
"""标签模型的Admin配置"""
# 编辑页排除的字段(自动生成)
exclude = ('slug', 'last_mod_time', 'creation_time')
class CategoryAdmin(admin.ModelAdmin):
"""分类模型的Admin配置"""
# 列表页显示的字段
list_display = ('name', 'parent_category', 'index')
# 编辑页排除的字段
exclude = ('slug', 'last_mod_time', 'creation_time')
class LinksAdmin(admin.ModelAdmin):
"""链接模型的Admin配置"""
exclude = ('last_mod_time', 'creation_time')
class SideBarAdmin(admin.ModelAdmin):
"""侧边栏模型的Admin配置"""
list_display = ('name', 'content', 'is_enable', 'sequence')
exclude = ('last_mod_time', 'creation_time')
class BlogSettingsAdmin(admin.ModelAdmin):
"""博客设置模型的Admin配置"""
pass # 使用默认配置

@ -0,0 +1,15 @@
# 从Django的apps模块导入AppConfig类用于定义应用的配置
from django.apps import AppConfig
class BlogConfig(AppConfig):
"""
博客应用blog的配置类
Django通过此类识别和配置应用的基本信息
包括应用名称默认自动生成的主键类型等
当项目启动时Django会加载每个应用的AppConfig子类
"""
# 定义应用的名称,必须与应用的实际目录名一致
# 这个名称用于Django内部识别应用例如在INSTALLED_APPS中注册时使用
name = 'blog'

@ -0,0 +1,73 @@
# 导入日志模块,用于记录系统运行时的信息和错误
import logging
# 从django.utils导入timezone用于获取当前时间
from django.utils import timezone
# 导入自定义的缓存工具和获取博客设置的工具函数
from djangoblog.utils import cache, get_blog_setting
# 导入当前应用下的Category分类和Article文章模型
from .models import Category, Article
# 创建日志记录器,用于记录当前模块的日志信息
logger = logging.getLogger(__name__)
def seo_processor(requests):
"""
自定义上下文处理器用于在所有模板中全局共享SEO相关的配置和数据
上下文处理器是Django的一个功能允许你在所有模板中自动添加变量
无需在每个视图函数中单独传递特别适合网站全局配置信息的共享
Args:
requests: Django请求对象包含当前请求的相关信息如域名协议等
Returns:
dict: 包含网站配置分类页面等信息的字典将被注入到所有模板中
"""
# 定义缓存键,用于标识当前处理器的缓存数据
key = 'seo_processor'
# 尝试从缓存中获取数据,减少数据库查询和计算开销
value = cache.get(key)
# 如果缓存中存在数据,直接返回缓存内容
if value:
return value
else:
# 缓存未命中时,记录日志并重新计算数据
logger.info('set processor cache.')
# 获取博客的全局设置(从数据库或其他配置源)
setting = get_blog_setting()
# 构建需要传递给模板的全局变量字典
value = {
'SITE_NAME': setting.site_name, # 网站名称
'SHOW_GOOGLE_ADSENSE': setting.show_google_adsense, # 是否显示谷歌广告
'GOOGLE_ADSENSE_CODES': setting.google_adsense_codes, # 谷歌广告代码
'SITE_SEO_DESCRIPTION': setting.site_seo_description, # 网站SEO描述用于搜索引擎
'SITE_DESCRIPTION': setting.site_description, # 网站描述
'SITE_KEYWORDS': setting.site_keywords, # 网站关键词用于SEO
# 网站基础URL如https://example.com/
'SITE_BASE_URL': requests.scheme + '://' + requests.get_host() + '/',
'ARTICLE_SUB_LENGTH': setting.article_sub_length, # 文章摘要长度
'nav_category_list': Category.objects.all(), # 导航栏显示的所有分类
# 导航栏显示的页面(类型为'p'即page状态为'p'即published
'nav_pages': Article.objects.filter(
type='p',
status='p'),
'OPEN_SITE_COMMENT': setting.open_site_comment, # 是否开启网站评论功能
'BEIAN_CODE': setting.beian_code, # 网站备案号
'ANALYTICS_CODE': setting.analytics_code, # 网站统计代码如Google Analytics
"BEIAN_CODE_GONGAN": setting.gongan_beiancode, # 公安备案号
"SHOW_GONGAN_CODE": setting.show_gongan_code, # 是否显示公安备案号
"CURRENT_YEAR": timezone.now().year, # 当前年份(用于页脚版权信息等)
"GLOBAL_HEADER": setting.global_header, # 全局页眉代码如额外的CSS/JS
"GLOBAL_FOOTER": setting.global_footer, # 全局页脚代码
"COMMENT_NEED_REVIEW": setting.comment_need_review, # 评论是否需要审核
}
# 将数据存入缓存有效期为10小时60秒*60分*10小时
cache.set(key, value, 60 * 60 * 10)
# 返回构建的全局变量字典
return value

@ -0,0 +1,267 @@
# 导入时间处理模块
import time
# 导入Elasticsearch客户端相关模块
import elasticsearch.client
# 导入Django配置模块
from django.conf import settings
# 导入Elasticsearch DSL相关组件用于定义文档结构
from elasticsearch_dsl import Document, InnerDoc, Date, Integer, Long, Text, Object, GeoPoint, Keyword, Boolean
from elasticsearch_dsl.connections import connections
# 导入博客文章模型,用于数据同步
from blog.models import Article
# 检查是否启用Elasticsearch通过判断配置中是否存在ELASTICSEARCH_DSL
ELASTICSEARCH_ENABLED = hasattr(settings, 'ELASTICSEARCH_DSL')
# 如果启用了Elasticsearch则进行初始化配置
if ELASTICSEARCH_ENABLED:
# 创建Elasticsearch连接从Django配置中获取主机地址
connections.create_connection(
hosts=[settings.ELASTICSEARCH_DSL['default']['hosts']])
# 导入Elasticsearch客户端并初始化
from elasticsearch import Elasticsearch
es = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
# 初始化IngestClient用于处理数据预处理管道
from elasticsearch.client import IngestClient
c = IngestClient(es)
# 尝试获取名为'geoip'的管道,如果不存在则创建
try:
c.get_pipeline('geoip')
except elasticsearch.exceptions.NotFoundError:
# 创建geoip管道通过ip地址解析地理位置信息
c.put_pipeline('geoip', body='''{
"description" : "Add geoip info", # 管道描述:添加地理信息
"processors" : [
{
"geoip" : {
"field" : "ip" # 基于ip字段解析地理信息
}
}
]
}''')
# 定义地理位置信息内部文档(嵌套在主文档中)
class GeoIp(InnerDoc):
continent_name = Keyword() # 大洲名称( Keyword类型不分词适合精确查询
country_iso_code = Keyword() # 国家ISO代码如CN、US
country_name = Keyword() # 国家名称
location = GeoPoint() # 经纬度坐标Elasticsearch地理点类型
# 定义用户代理浏览器信息内部文档
class UserAgentBrowser(InnerDoc):
Family = Keyword() # 浏览器家族如Chrome、Firefox
Version = Keyword() # 浏览器版本
# 定义用户代理操作系统信息内部文档(继承浏览器结构,字段相同)
class UserAgentOS(UserAgentBrowser):
pass
# 定义用户代理设备信息内部文档
class UserAgentDevice(InnerDoc):
Family = Keyword() # 设备家族如iPhone、Windows
Brand = Keyword() # 设备品牌如Apple、Samsung
Model = Keyword() # 设备型号如iPhone 13
# 定义用户代理整体信息内部文档(整合浏览器、系统、设备信息)
class UserAgent(InnerDoc):
browser = Object(UserAgentBrowser, required=False) # 浏览器信息(可选)
os = Object(UserAgentOS, required=False) # 操作系统信息(可选)
device = Object(UserAgentDevice, required=False) # 设备信息(可选)
string = Text() # 原始用户代理字符串(如"Mozilla/5.0..."
is_bot = Boolean() # 是否为爬虫机器人
# 定义性能日志文档(记录访问性能数据)
class ElapsedTimeDocument(Document):
url = Keyword() # 访问的URL精确匹配
time_taken = Long() # 页面加载耗时(毫秒)
log_datetime = Date() # 日志记录时间
ip = Keyword() # 访问者IP地址
geoip = Object(GeoIp, required=False) # 地理位置信息由geoip管道生成
useragent = Object(UserAgent, required=False) # 用户代理信息
# 索引配置
class Index:
name = 'performance' # 索引名称performance性能日志
settings = {
"number_of_shards": 1, # 主分片数量
"number_of_replicas": 0 # 副本分片数量单节点环境设为0
}
# 文档类型配置Elasticsearch 7+后逐渐废弃但DSL仍保留兼容
class Meta:
doc_type = 'ElapsedTime'
# 性能日志文档管理器(处理索引创建、删除、数据写入)
class ElaspedTimeDocumentManager:
@staticmethod
def build_index():
"""创建performance索引如果不存在"""
from elasticsearch import Elasticsearch
# 连接Elasticsearch
client = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
# 检查索引是否存在
res = client.indices.exists(index="performance")
if not res:
# 初始化索引根据ElapsedTimeDocument的定义创建映射
ElapsedTimeDocument.init()
@staticmethod
def delete_index():
"""删除performance索引"""
from elasticsearch import Elasticsearch
es = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
# 忽略400索引不存在和404请求错误
es.indices.delete(index='performance', ignore=[400, 404])
@staticmethod
def create(url, time_taken, log_datetime, useragent, ip):
"""创建一条性能日志记录并写入Elasticsearch"""
# 确保索引存在
ElaspedTimeDocumentManager.build_index()
# 构建用户代理信息对象
ua = UserAgent()
ua.browser = UserAgentBrowser()
ua.browser.Family = useragent.browser.family # 浏览器家族
ua.browser.Version = useragent.browser.version_string # 浏览器版本
ua.os = UserAgentOS()
ua.os.Family = useragent.os.family # 操作系统家族
ua.os.Version = useragent.os.version_string # 操作系统版本
ua.device = UserAgentDevice()
ua.device.Family = useragent.device.family # 设备家族
ua.device.Brand = useragent.device.brand # 设备品牌
ua.device.Model = useragent.device.model # 设备型号
ua.string = useragent.ua_string # 原始用户代理字符串
ua.is_bot = useragent.is_bot # 是否为爬虫
# 构建性能日志文档
doc = ElapsedTimeDocument(
meta={
# 用当前时间戳毫秒作为文档ID
'id': int(round(time.time() * 1000))
},
url=url, # 访问URL
time_taken=time_taken, # 耗时
log_datetime=log_datetime, # 日志时间
useragent=ua, # 用户代理信息
ip=ip # IP地址
)
# 保存文档时应用geoip管道自动解析IP对应的地理位置
doc.save(pipeline="geoip")
# 定义文章文档(用于博客文章的搜索索引)
class ArticleDocument(Document):
# 文章内容使用ik分词器max_word最大化分词smart智能分词
body = Text(analyzer='ik_max_word', search_analyzer='ik_smart')
# 文章标题(同上分词配置)
title = Text(analyzer='ik_max_word', search_analyzer='ik_smart')
# 作者信息(嵌套对象)
author = Object(properties={
'nickname': Text(analyzer='ik_max_word', search_analyzer='ik_smart'), # 作者昵称
'id': Integer() # 作者ID
})
# 分类信息(嵌套对象)
category = Object(properties={
'name': Text(analyzer='ik_max_word', search_analyzer='ik_smart'), # 分类名称
'id': Integer() # 分类ID
})
# 标签信息(嵌套对象列表)
tags = Object(properties={
'name': Text(analyzer='ik_max_word', search_analyzer='ik_smart'), # 标签名称
'id': Integer() # 标签ID
})
pub_time = Date() # 发布时间
status = Text() # 文章状态(如发布、草稿)
comment_status = Text() # 评论状态(如开启、关闭)
type = Text() # 文章类型(如原创、转载)
views = Integer() # 浏览量
article_order = Integer() # 文章排序权重
# 索引配置
class Index:
name = 'blog' # 索引名称blog博客文章
settings = {
"number_of_shards": 1,
"number_of_replicas": 0
}
# 文档类型配置
class Meta:
doc_type = 'Article'
# 文章文档管理器(处理文章索引的创建、更新、重建)
class ArticleDocumentManager():
def __init__(self):
"""初始化时创建索引(如果不存在)"""
self.create_index()
def create_index(self):
"""创建blog索引根据ArticleDocument定义初始化映射"""
ArticleDocument.init()
def delete_index(self):
"""删除blog索引"""
from elasticsearch import Elasticsearch
es = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
es.indices.delete(index='blog', ignore=[400, 404])
def convert_to_doc(self, articles):
"""将Django模型对象列表转换为ArticleDocument列表"""
return [
ArticleDocument(
meta={'id': article.id}, # 用文章ID作为文档ID
body=article.body, # 文章内容
title=article.title, # 文章标题
author={
'nickname': article.author.username, # 作者用户名
'id': article.author.id # 作者ID
},
category={
'name': article.category.name, # 分类名称
'id': article.category.id # 分类ID
},
# 转换标签列表(多对多关系)
tags=[{'name': t.name, 'id': t.id} for t in article.tags.all()],
pub_time=article.pub_time, # 发布时间
status=article.status, # 文章状态
comment_status=article.comment_status, # 评论状态
type=article.type, # 文章类型
views=article.views, # 浏览量
article_order=article.article_order # 排序权重
) for article in articles
]
def rebuild(self, articles=None):
"""重建索引(默认同步所有文章,可指定文章列表)"""
# 初始化索引结构
ArticleDocument.init()
# 如果未指定文章,则同步所有文章
articles = articles if articles else Article.objects.all()
# 转换模型为文档对象
docs = self.convert_to_doc(articles)
# 批量保存文档
for doc in docs:
doc.save()
def update_docs(self, docs):
"""更新文档列表(批量保存)"""
for doc in docs:
doc.save()

@ -0,0 +1,41 @@
# 导入日志模块,用于记录搜索相关日志
import logging
# 导入Django的表单模块用于构建自定义表单
from django import forms
# 导入Haystack的搜索表单基类用于扩展搜索功能
from haystack.forms import SearchForm
# 创建日志记录器,使用当前模块名作为日志器名称
logger = logging.getLogger(__name__)
class BlogSearchForm(SearchForm):
"""
博客搜索表单类继承自Haystack的SearchForm
用于自定义博客搜索的表单验证和搜索逻辑
"""
# 定义搜索查询字段required=True表示该字段为必填项
# 用户输入的搜索关键词将通过该字段传递
querydata = forms.CharField(required=True)
def search(self):
"""
重写父类的search方法实现自定义搜索逻辑
该方法会处理搜索请求并返回搜索结果
"""
# 调用父类的search方法获取初始搜索结果集
# 父类方法会处理Haystack的核心搜索逻辑
datas = super(BlogSearchForm, self).search()
# 检查表单数据是否有效,若无效则返回无查询结果的默认响应
if not self.is_valid():
return self.no_query_found()
# 如果表单验证通过且存在查询数据querydata
if self.cleaned_data['querydata']:
# 记录搜索关键词到日志,方便后续分析用户搜索行为
logger.info(self.cleaned_data['querydata'])
# 返回处理后的搜索结果集
return datas

@ -0,0 +1,45 @@
# 导入Django命令基类用于创建自定义管理命令
from django.core.management.base import BaseCommand
# 导入博客相关的Elasticsearch文档管理器和配置
from blog.documents import (
ElapsedTimeDocument, # 耗时统计文档模型
ArticleDocumentManager, # 文章文档管理器
ElaspedTimeDocumentManager, # 耗时统计文档管理器原拼写可能存在笔误应为Elapsed
ELASTICSEARCH_ENABLED # Elasticsearch启用状态标记
)
# TODO: 后续可优化为支持参数化(如指定重建的索引类型等)
class Command(BaseCommand):
"""
Django自定义管理命令构建Elasticsearch搜索索引
用于初始化或重建文章和耗时统计相关的搜索索引
"""
# 命令的帮助信息使用python manage.py help build_index时显示
help = 'build search index'
def handle(self, *args, **options):
"""
命令核心执行方法
当运行python manage.py build_index时调用
"""
# 仅在Elasticsearch启用时执行索引构建
if ELASTICSEARCH_ENABLED:
# 构建耗时统计文档的索引
ElaspedTimeDocumentManager.build_index()
# 初始化耗时统计文档的索引结构
elapsed_manager = ElapsedTimeDocument()
elapsed_manager.init() # 创建索引映射
# 处理文章文档索引:先删除旧索引,再重建
article_manager = ArticleDocumentManager()
article_manager.delete_index() # 删除现有文章索引
article_manager.rebuild() # 重新创建索引并同步数据
# 输出成功信息到控制台
self.stdout.write(self.style.SUCCESS('Successfully built search indexes'))
else:
# 当Elasticsearch未启用时提示用户
self.stdout.write(self.style.WARNING('Elasticsearch is not enabled, skipping index build'))

@ -0,0 +1,32 @@
# 导入Django命令基类用于创建自定义管理命令
from django.core.management.base import BaseCommand
# 导入博客应用中的标签和分类模型
from blog.models import Tag, Category
# TODO: 后续可优化为支持参数化(如指定输出格式、过滤条件等)
class Command(BaseCommand):
"""
Django自定义管理命令生成搜索关键词列表
提取所有标签和分类的名称用于构建搜索提示词或关键词库
"""
# 命令的帮助信息执行python manage.py help build_search_words时显示
help = 'build search words'
def handle(self, *args, **options):
"""
命令核心执行逻辑
当运行python manage.py build_search_words时调用
"""
# 1. 提取所有标签Tag的名称并转换为列表
# 2. 提取所有分类Category的名称并转换为列表
# 3. 合并两个列表并通过set去重确保关键词唯一
datas = set(
[tag.name for tag in Tag.objects.all()] + # 标签名称列表
[category.name for category in Category.objects.all()] # 分类名称列表
)
# 将去重后的关键词按行打印输出
# 格式为每个关键词单独一行,便于后续处理(如写入文件或导入搜索提示库)
print('\n'.join(datas))

@ -0,0 +1,26 @@
# 导入Django命令基类用于创建自定义管理命令
from django.core.management.base import BaseCommand
# 导入项目自定义的缓存工具封装自djangoblog.utils
from djangoblog.utils import cache
class Command(BaseCommand):
"""
Django自定义管理命令清除系统所有缓存
用于手动触发缓存清理确保缓存数据与数据库同步
"""
# 命令的帮助信息执行python manage.py help clear_cache时显示
help = 'clear the whole cache'
def handle(self, *args, **options):
"""
命令核心执行逻辑
当运行python manage.py clear_cache时调用
"""
# 调用缓存工具的clear()方法,清除所有缓存数据
# 这里的cache是项目自定义的缓存实例可能封装了Django原生缓存或其他缓存后端
cache.clear()
# 向控制台输出成功信息使用Django命令的样式工具显示绿色成功提示
self.stdout.write(self.style.SUCCESS('Cleared cache\n'))

@ -0,0 +1,76 @@
# 导入Django用户模型、密码加密、命令基类
from django.contrib.auth import get_user_model
from django.contrib.auth.hashers import make_password
from django.core.management.base import BaseCommand
# 导入博客应用的核心模型
from blog.models import Article, Tag, Category
class Command(BaseCommand):
"""
Django自定义管理命令创建测试数据
用于快速生成用户分类标签和文章等测试数据方便开发和测试
"""
# 命令的帮助信息执行python manage.py help create_testdata时显示
help = 'create test datas'
def handle(self, *args, **options):
"""
命令核心执行逻辑
当运行python manage.py create_testdata时调用生成测试数据
"""
# 1. 创建或获取测试用户
# get_or_create存在则获取不存在则创建避免重复生成
# make_password加密密码安全存储
user = get_user_model().objects.get_or_create(
email='test@test.com', # 测试邮箱
username='测试用户', # 用户名
password=make_password('test!q@w#eTYU') # 加密后的密码
)[0] # [0]取返回元组中的用户对象
# 2. 创建分类(含层级关系)
# 创建父分类(无上级分类)
pcategory = Category.objects.get_or_create(
name='我是父类目',
parent_category=None # 顶级分类
)[0]
# 创建子分类(关联父分类)
category = Category.objects.get_or_create(
name='子类目',
parent_category=pcategory # 关联到父分类
)[0]
category.save() # 保存子分类
# 3. 创建基础标签(供所有测试文章共用)
basetag = Tag()
basetag.name = "标签" # 标签名称
basetag.save()
# 4. 批量创建测试文章1-19共19篇
for i in range(1, 20):
# 创建或获取文章
article = Article.objects.get_or_create(
category=category, # 关联到子分类
title=f'nice title {i}', # 文章标题(带序号)
body=f'nice content {i}', # 文章内容(带序号)
author=user # 关联到测试用户
)[0]
# 为每篇文章创建专属标签
tag = Tag()
tag.name = f"标签{i}" # 标签名称(带序号)
tag.save()
# 给文章添加标签(专属标签 + 基础标签)
article.tags.add(tag)
article.tags.add(basetag)
article.save() # 保存文章(更新标签关联)
# 5. 清除缓存(确保新生成的测试数据能立即生效)
from djangoblog.utils import cache
cache.clear()
# 输出成功信息到控制台
self.stdout.write(self.style.SUCCESS('created test datas \n'))

@ -0,0 +1,88 @@
# 导入Django命令基类用于创建自定义管理命令
from django.core.management.base import BaseCommand
# 导入搜索引擎推送工具、站点配置和博客模型
from djangoblog.spider_notify import SpiderNotify # 搜索引擎推送工具
from djangoblog.utils import get_current_site # 获取当前站点信息
from blog.models import Article, Tag, Category # 博客核心模型
# 获取当前站点的域名用于生成完整URL
site = get_current_site().domain
class Command(BaseCommand):
"""
Django自定义管理命令向百度搜索引擎推送URL
用于主动告知百度爬虫网站的更新内容加速收录
"""
# 命令的帮助信息执行python manage.py help ping_baidu时显示
help = 'notify baidu url'
def add_arguments(self, parser):
"""
定义命令参数指定需要推送的URL类型
通过parser添加命令行参数限制可选值
"""
parser.add_argument(
'data_type', # 参数名称
type=str,
choices=[ # 可选参数值
'all', # 推送所有类型(文章、标签、分类)
'article', # 仅推送文章
'tag', # 仅推送标签页
'category' # 仅推送分类页
],
help='指定推送类型article(所有文章)、tag(所有标签)、category(所有分类)、all(全部)'
)
def get_full_url(self, path):
"""
生成完整的URL域名+相对路径
:param path: 模型实例的相对路径/article/1.html
:return: 完整的URL字符串如https://example.com/article/1.html
"""
return f"https://{site}{path}"
def handle(self, *args, **options):
"""
命令核心执行逻辑
根据参数类型收集URL推送给百度搜索引擎
"""
# 获取用户指定的推送类型
data_type = options['data_type']
self.stdout.write(f'开始收集{data_type}类型的URL...')
# 存储待推送的URL列表
urls = []
# 1. 收集文章URL已发布状态
if data_type == 'article' or data_type == 'all':
# 筛选所有已发布的文章
for article in Article.objects.filter(status='p'):
# 调用文章模型的get_full_url方法获取完整URL
urls.append(article.get_full_url())
# 2. 收集标签页URL
if data_type == 'tag' or data_type == 'all':
for tag in Tag.objects.all():
# 获取标签页的相对路径再生成完整URL
relative_url = tag.get_absolute_url()
urls.append(self.get_full_url(relative_url))
# 3. 收集分类页URL
if data_type == 'category' or data_type == 'all':
for category in Category.objects.all():
# 获取分类页的相对路径再生成完整URL
relative_url = category.get_absolute_url()
urls.append(self.get_full_url(relative_url))
# 输出待推送的URL数量
self.stdout.write(
self.style.SUCCESS(f'准备推送{len(urls)}条URL...')
)
# 调用工具类向百度推送URL
SpiderNotify.baidu_notify(urls)
# 推送完成,输出成功信息
self.stdout.write(self.style.SUCCESS('URL推送完成'))

@ -0,0 +1,86 @@
import requests # 用于发送HTTP请求验证图片URL有效性
from django.core.management.base import BaseCommand
from django.templatetags.static import static # 生成静态文件URL
# 导入项目工具和模型用户头像保存、OAuth用户模型、OAuth管理工具
from djangoblog.utils import save_user_avatar # 保存用户头像到本地的工具函数
from oauth.models import OAuthUser # OAuth关联用户模型存储第三方登录用户信息
from oauth.oauthmanager import get_manager_by_type # 根据 OAuth 类型获取对应管理器
class Command(BaseCommand):
"""
Django自定义管理命令同步用户头像
用于检查并更新OAuth用户的头像URL确保头像可访问无效则重新获取或使用默认头像
"""
# 命令的帮助信息执行python manage.py help sync_user_avatar时显示
help = 'sync user avatar'
def test_picture(self, url):
"""
验证图片URL是否有效可访问且返回200状态码
:param url: 头像图片的URL
:return: 有效则返回True否则返回False
"""
try:
# 发送GET请求超时2秒检查状态码是否为200
if requests.get(url, timeout=2).status_code == 200:
return True
except:
# 任何异常(超时、连接错误等)均视为无效
pass
return False
def handle(self, *args, **options):
"""
命令核心执行逻辑
遍历所有OAuth用户检查并同步头像URL
"""
# 获取项目静态文件的基础URL用于判断头像是否为本地静态文件
static_url = static("../")
# 获取所有OAuth用户
users = OAuthUser.objects.all()
self.stdout.write(f'开始同步{len(users)}个用户头像')
# 遍历每个用户处理头像
for u in users:
self.stdout.write(f'开始同步:{u.nickname}') # 输出当前处理的用户名
url = u.picture # 获取用户当前的头像URL
if url: # 如果用户已有头像URL
# 情况1头像URL是本地静态文件以static_url开头
if url.startswith(static_url):
# 验证本地头像是否有效
if self.test_picture(url):
self.stdout.write(f' 头像有效,跳过:{url}')
continue # 有效则跳过处理
else:
# 本地头像无效,尝试重新获取
self.stdout.write(f' 本地头像无效,尝试重新获取')
if u.metadata: # 如果存在第三方平台返回的元数据(可能包含头像信息)
# 根据OAuth类型如qq、weibo获取对应的管理器
manage = get_manager_by_type(u.type)
# 从元数据中提取最新头像URL
url = manage.get_picture(u.metadata)
# 保存头像到本地并返回新的URL
url = save_user_avatar(url)
else:
# 无元数据,使用默认头像
url = static('blog/img/avatar.png')
else:
# 情况2头像URL是第三方链接非本地文件保存到本地
self.stdout.write(f' 第三方头像,保存到本地')
url = save_user_avatar(url)
else:
# 情况3用户无头像URL使用默认头像
self.stdout.write(f' 无头像,使用默认头像')
url = static('blog/img/avatar.png')
# 更新用户头像并保存
if url:
self.stdout.write(f' 结束同步:{u.nickname}.url:{url}')
u.picture = url
u.save() # 保存更新后的头像URL
self.stdout.write('所有用户头像同步完成')

@ -0,0 +1,90 @@
# 导入日志模块,用于记录中间件运行过程中的日志信息
import logging
# 导入时间模块,用于计算页面渲染耗时
import time
# 从ipware工具导入获取客户端IP的函数
from ipware import get_client_ip
# 从user_agents工具导入解析用户代理的函数
from user_agents import parse
# 导入博客相关的ES配置和文档管理器用于记录页面加载时间
from blog.documents import ELASTICSEARCH_ENABLED, ElaspedTimeDocumentManager
# 创建当前模块的日志记录器
logger = logging.getLogger(__name__)
class OnlineMiddleware(object):
"""
自定义Django中间件用于
1. 计算页面渲染耗时
2. 收集客户端信息IP用户代理
3. 在启用Elasticsearch时记录访问性能数据
4. 替换响应中的特定标记为实际加载时间
"""
def __init__(self, get_response=None):
"""
中间件初始化方法
:param get_response: Django框架传入的处理响应的函数用于链式调用中间件
"""
self.get_response = get_response
# 调用父类初始化方法Python 2兼容写法在Python 3中可省略
super().__init__()
def __call__(self, request):
"""
中间件核心处理方法在请求到达视图前和响应返回客户端前执行
:param request: Django的请求对象包含客户端请求信息
:return: 处理后的响应对象
"""
# 记录请求处理开始时间(用于计算耗时)
start_time = time.time()
# 调用下一个中间件或视图函数,获取响应对象
response = self.get_response(request)
# 从请求头中获取用户代理字符串(如浏览器型号、系统等信息)
http_user_agent = request.META.get('HTTP_USER_AGENT', '')
# 获取客户端IP地址第二个返回值为是否是公开IP此处暂不使用
ip, _ = get_client_ip(request)
# 解析用户代理字符串,转换为可操作的对象(方便提取浏览器、系统等信息)
user_agent = parse(http_user_agent)
# 判断响应是否为非流式响应(流式响应无法修改内容,如文件下载)
if not response.streaming:
try:
# 计算页面渲染总耗时(当前时间 - 开始时间)
cast_time = time.time() - start_time
# 如果启用了Elasticsearch记录性能数据
if ELASTICSEARCH_ENABLED:
# 将耗时转换为毫秒并保留2位小数
time_taken = round((cast_time) * 1000, 2)
# 获取当前请求的URL路径
url = request.path
# 导入Django的时区工具用于记录当前时间
from django.utils import timezone
# 通过文档管理器向Elasticsearch插入一条性能记录
ElaspedTimeDocumentManager.create(
url=url, # 访问的URL
time_taken=time_taken, # 页面加载耗时(毫秒)
log_datetime=timezone.now(), # 记录时间(当前时区)
useragent=user_agent, # 解析后的用户代理信息
ip=ip # 客户端IP地址
)
# 将响应内容中的<!!LOAD_TIMES!!>标记替换为实际耗时保留前5位字符
# 注需确保响应内容为bytes类型因此使用str.encode转换
response.content = response.content.replace(
b'<!!LOAD_TIMES!!>', str.encode(str(cast_time)[:5])
)
# 捕获所有异常,避免中间件错误导致请求失败
except Exception as e:
# 记录异常信息到日志
logger.error("Error in OnlineMiddleware: %s" % e)
# 返回处理后的响应对象
return response

@ -0,0 +1,202 @@
# 生成信息由Django 4.1.7在2023-03-02 07:14自动生成的迁移文件
# 迁移文件用于定义数据库表结构通过Django的迁移系统创建或修改数据库表
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion # 用于定义外键删除时的行为
import django.utils.timezone # 用于处理时间字段的默认值
import mdeditor.fields # 导入Markdown编辑器字段用于文章正文
class Migration(migrations.Migration):
"""
数据库迁移类定义博客系统初始表结构的迁移操作
所有模型的首次迁移会创建对应的数据库表
"""
# 标记为初始迁移(首次创建表结构)
initial = True
# 依赖关系当前迁移依赖于Django用户模型的迁移
# 因为Article模型关联了用户表作者需确保用户表先创建
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
# 迁移操作:创建所有模型对应的数据库表
operations = [
# 创建"网站配置"表BlogSettings
migrations.CreateModel(
name='BlogSettings',
fields=[
# 自增主键BigAutoField支持更大的数值范围适合大数据量
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('sitename', models.CharField(default='', max_length=200, verbose_name='网站名称')), # 网站名称
('site_description', models.TextField(default='', max_length=1000, verbose_name='网站描述')), # 网站描述(用于前端展示)
('site_seo_description', models.TextField(default='', max_length=1000, verbose_name='网站SEO描述')), # 用于搜索引擎优化的描述
('site_keywords', models.TextField(default='', max_length=1000, verbose_name='网站关键字')), # SEO关键字多个用逗号分隔
('article_sub_length', models.IntegerField(default=300, verbose_name='文章摘要长度')), # 文章列表页显示的摘要长度
('sidebar_article_count', models.IntegerField(default=10, verbose_name='侧边栏文章数目')), # 侧边栏显示的文章数量
('sidebar_comment_count', models.IntegerField(default=5, verbose_name='侧边栏评论数目')), # 侧边栏显示的评论数量
('article_comment_count', models.IntegerField(default=5, verbose_name='文章页面默认显示评论数目')), # 文章详情页默认显示的评论数量
('show_google_adsense', models.BooleanField(default=False, verbose_name='是否显示谷歌广告')), # 开关:是否显示谷歌广告
('google_adsense_codes', models.TextField(blank=True, default='', max_length=2000, null=True, verbose_name='广告内容')), # 谷歌广告代码
('open_site_comment', models.BooleanField(default=True, verbose_name='是否打开网站评论功能')), # 开关:是否允许评论
('beiancode', models.CharField(blank=True, default='', max_length=2000, null=True, verbose_name='备案号')), # 网站备案号ICP备案
('analyticscode', models.TextField(default='', max_length=1000, verbose_name='网站统计代码')), # 第三方统计代码(如百度统计)
('show_gongan_code', models.BooleanField(default=False, verbose_name='是否显示公安备案号')), # 开关:是否显示公安备案号
('gongan_beiancode', models.TextField(blank=True, default='', max_length=2000, null=True, verbose_name='公安备案号')), # 公安备案号
],
options={
'verbose_name': '网站配置', # 模型的单数显示名称
'verbose_name_plural': '网站配置', # 模型的复数显示名称(因配置通常只有一条记录,复数同单数)
},
),
# 创建"友情链接"表Links
migrations.CreateModel(
name='Links',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=30, unique=True, verbose_name='链接名称')), # 友情链接名称(唯一)
('link', models.URLField(verbose_name='链接地址')), # 链接的URL地址
('sequence', models.IntegerField(unique=True, verbose_name='排序')), # 排序序号(唯一,控制显示顺序)
('is_enable', models.BooleanField(default=True, verbose_name='是否显示')), # 开关:是否在页面显示
# 显示类型:指定链接在哪些页面显示
('show_type', models.CharField(
choices=[('i', '首页'), ('l', '列表页'), ('p', '文章页面'), ('a', '全站'), ('s', '友情链接页面')],
default='i',
max_length=1,
verbose_name='显示类型'
)),
('created_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='创建时间')), # 创建时间
('last_mod_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='修改时间')), # 最后修改时间
],
options={
'verbose_name': '友情链接',
'verbose_name_plural': '友情链接',
'ordering': ['sequence'], # 默认按排序序号升序排列
},
),
# 创建"侧边栏"表SideBar
migrations.CreateModel(
name='SideBar',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100, verbose_name='标题')), # 侧边栏模块标题
('content', models.TextField(verbose_name='内容')), # 侧边栏内容支持HTML
('sequence', models.IntegerField(unique=True, verbose_name='排序')), # 排序序号(控制多个侧边栏的显示顺序)
('is_enable', models.BooleanField(default=True, verbose_name='是否启用')), # 开关:是否显示该侧边栏
('created_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='创建时间')),
('last_mod_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='修改时间')),
],
options={
'verbose_name': '侧边栏',
'verbose_name_plural': '侧边栏',
'ordering': ['sequence'], # 按排序序号升序排列
},
),
# 创建"标签"表Tag
migrations.CreateModel(
name='Tag',
fields=[
('id', models.AutoField(primary_key=True, serialize=False)), # 自增主键
('created_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='创建时间')),
('last_mod_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='修改时间')),
('name', models.CharField(max_length=30, unique=True, verbose_name='标签名')), # 标签名称(唯一)
('slug', models.SlugField(blank=True, default='no-slug', max_length=60)), # URL友好的标识符用于生成标签页URL
],
options={
'verbose_name': '标签',
'verbose_name_plural': '标签',
'ordering': ['name'], # 按标签名升序排列
},
),
# 创建"分类"表Category
migrations.CreateModel(
name='Category',
fields=[
('id', models.AutoField(primary_key=True, serialize=False)),
('created_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='创建时间')),
('last_mod_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='修改时间')),
('name', models.CharField(max_length=30, unique=True, verbose_name='分类名')), # 分类名称(唯一)
('slug', models.SlugField(blank=True, default='no-slug', max_length=60)), # 用于生成分类页URL的标识符
('index', models.IntegerField(default=0, verbose_name='权重排序-越大越靠前')), # 权重值,控制分类在页面的显示优先级
# 自关联外键:支持分类层级(父分类->子分类)
# on_delete=models.CASCADE表示若父分类删除子分类也会被删除
('parent_category', models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
to='blog.category',
verbose_name='父级分类'
)),
],
options={
'verbose_name': '分类',
'verbose_name_plural': '分类',
'ordering': ['-index'], # 按权重降序排列(权重越大越靠前)
},
),
# 创建"文章"表Article
migrations.CreateModel(
name='Article',
fields=[
('id', models.AutoField(primary_key=True, serialize=False)),
('created_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='创建时间')),
('last_mod_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='修改时间')),
('title', models.CharField(max_length=200, unique=True, verbose_name='标题')), # 文章标题(唯一)
('body', mdeditor.fields.MDTextField(verbose_name='正文')), # 文章正文使用Markdown编辑器
('pub_time', models.DateTimeField(default=django.utils.timezone.now, verbose_name='发布时间')), # 发布时间
# 文章状态草稿d/发表p
('status', models.CharField(
choices=[('d', '草稿'), ('p', '发表')],
default='p',
max_length=1,
verbose_name='文章状态'
)),
# 评论状态打开o/关闭c
('comment_status', models.CharField(
choices=[('o', '打开'), ('c', '关闭')],
default='o',
max_length=1,
verbose_name='评论状态'
)),
# 内容类型文章a/页面p如关于页、联系页
('type', models.CharField(
choices=[('a', '文章'), ('p', '页面')],
default='a',
max_length=1,
verbose_name='类型'
)),
('views', models.PositiveIntegerField(default=0, verbose_name='浏览量')), # 浏览量(非负整数)
('article_order', models.IntegerField(default=0, verbose_name='排序,数字越大越靠前')), # 排序值,控制文章在列表中的位置
('show_toc', models.BooleanField(default=False, verbose_name='是否显示toc目录')), # 开关:是否显示文章目录
# 外键关联作者Django用户模型
# on_delete=models.CASCADE表示若作者账号删除其文章也会被删除
('author', models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to=settings.AUTH_USER_MODEL,
verbose_name='作者'
)),
# 外键:关联分类
# on_delete=models.CASCADE表示若分类删除该分类下的文章也会被删除
('category', models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to='blog.category',
verbose_name='分类'
)),
# 多对多关系:文章与标签(一篇文章可关联多个标签,一个标签可关联多篇文章)
('tags', models.ManyToManyField(blank=True, to='blog.tag', verbose_name='标签集合')),
],
options={
'verbose_name': '文章',
'verbose_name_plural': '文章',
# 默认排序规则:先按排序值降序(值越大越靠前),再按发布时间降序(最新的在前)
'ordering': ['-article_order', '-pub_time'],
'get_latest_by': 'id', # 按ID获取最新记录ID自增越大越新
},
),
]

@ -0,0 +1,45 @@
# 生成信息由Django 4.1.7在2023-03-29 06:08自动生成的迁移文件
from django.db import migrations, models
class Migration(migrations.Migration):
"""
数据库迁移类为网站配置表添加新字段
用于扩展网站配置功能支持全局头部和尾部内容的设置
"""
# 依赖关系当前迁移依赖于博客应用的初始迁移0001_initial
# 确保在初始表结构创建之后再执行此迁移
dependencies = [
('blog', '0001_initial'), # 依赖blog应用的第一个迁移文件
]
# 迁移操作为BlogSettings模型添加两个新字段
operations = [
# 为BlogSettings添加"公共尾部"字段
migrations.AddField(
model_name='blogsettings', # 目标模型:网站配置表
name='global_footer', # 新字段名称
field=models.TextField(
blank=True, # 允许表单提交为空
default='', # 默认值为空字符串
null=True, # 数据库中允许为NULL
verbose_name='公共尾部' # 管理界面显示的字段名称
),
# 字段作用存储网站全局共用的尾部HTML内容如版权信息、备案号等
# 可在所有页面底部统一显示,避免重复开发
),
# 为BlogSettings添加"公共头部"字段
migrations.AddField(
model_name='blogsettings', # 目标模型:网站配置表
name='global_header', # 新字段名称
field=models.TextField(
blank=True,
default='',
null=True,
verbose_name='公共头部'
),
# 字段作用存储网站全局共用的头部HTML内容如公共导航、统计代码等
# 可在所有页面顶部统一显示,方便全局修改
),
]

@ -0,0 +1,31 @@
# 生成信息由Django 4.2.1版本在2023-05-09 07:45自动生成的迁移文件
from django.db import migrations, models
class Migration(migrations.Migration):
"""
数据库迁移类为网站配置表添加评论审核开关字段
用于控制用户评论是否需要管理员审核后才显示增强内容管理能力
"""
# 依赖关系当前迁移依赖于博客应用的上一个迁移文件0002_...
# 确保在之前的表结构变更完成后再执行本次迁移
dependencies = [
('blog', '0002_blogsettings_global_footer_and_more'),
]
# 迁移操作为BlogSettings模型添加评论审核开关字段
operations = [
migrations.AddField(
model_name='blogsettings', # 目标模型网站配置表BlogSettings
name='comment_need_review', # 新字段名称:评论是否需要审核
field=models.BooleanField(
default=False, # 默认值为False评论无需审核提交后直接显示
verbose_name='评论是否需要审核' # 管理后台显示的字段名称
),
# 字段作用:
# - 当值为True时用户提交的评论需管理员在后台审核通过后才会在前端显示
# - 当值为False时评论提交后立即显示无需审核
# 用于防止垃圾评论或违规内容直接展示,提升网站内容安全性
),
]

@ -0,0 +1,40 @@
# 生成信息由Django 4.2.1版本在2023-05-09 07:51自动生成的迁移文件
from django.db import migrations
class Migration(migrations.Migration):
"""
数据库迁移类重命名BlogSettings模型中的多个字段
目的是统一字段命名规范采用下划线命名法提升代码可读性和一致性
"""
# 依赖关系当前迁移依赖于上一个迁移文件0003_...
# 确保在添加评论审核字段之后执行字段重命名操作
dependencies = [
('blog', '0003_blogsettings_comment_need_review'),
]
# 迁移操作批量重命名BlogSettings模型的字段
operations = [
# 重命名"analyticscode"字段为"analytics_code"
migrations.RenameField(
model_name='blogsettings', # 目标模型:网站配置表
old_name='analyticscode', # 旧字段名(驼峰式命名,不规范)
new_name='analytics_code', # 新字段名下划线命名符合Python规范
# 字段含义存储网站统计代码如百度统计、Google Analytics
),
# 重命名"beiancode"字段为"beian_code"
migrations.RenameField(
model_name='blogsettings',
old_name='beiancode', # 旧字段名(连写,不规范)
new_name='beian_code', # 新字段名(下划线分隔,更清晰)
# 字段含义存储网站ICP备案号
),
# 重命名"sitename"字段为"site_name"
migrations.RenameField(
model_name='blogsettings',
old_name='sitename', # 旧字段名(连写,不规范)
new_name='site_name', # 新字段名(下划线分隔,符合命名习惯)
# 字段含义:存储网站名称
),
]

@ -0,0 +1,107 @@
# 生成信息由Django 4.2.5版本在2023-09-06 13:13自动生成的迁移文件
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion # 外键删除行为处理
import django.utils.timezone # 时间字段默认值
import mdeditor.fields # Markdown编辑器字段
class Migration(migrations.Migration):
"""
数据库迁移类统一模型的字段命名和 verbose_name 为英文
可能是为了国际化适配或代码规范统一将中文标识改为英文
"""
# 依赖关系:依赖用户模型和上一个迁移文件
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('blog', '0004_rename_analyticscode_blogsettings_analytics_code_and_more'),
]
operations = [
# 1. 修改模型的元数据选项verbose_name 改为英文)
migrations.AlterModelOptions(
name='article',
options={'get_latest_by': 'id', 'ordering': ['-article_order', '-pub_time'],
'verbose_name': 'article', 'verbose_name_plural': 'article'},
),
migrations.AlterModelOptions(
name='category',
options={'ordering': ['-index'], 'verbose_name': 'category', 'verbose_name_plural': 'category'},
),
migrations.AlterModelOptions(
name='links',
options={'ordering': ['sequence'], 'verbose_name': 'link', 'verbose_name_plural': 'link'},
),
migrations.AlterModelOptions(
name='sidebar',
options={'ordering': ['sequence'], 'verbose_name': 'sidebar', 'verbose_name_plural': 'sidebar'},
),
migrations.AlterModelOptions(
name='tag',
options={'ordering': ['name'], 'verbose_name': 'tag', 'verbose_name_plural': 'tag'},
),
# 2. 删除旧的时间字段(中文命名相关)
migrations.RemoveField(model_name='article', name='created_time'),
migrations.RemoveField(model_name='article', name='last_mod_time'),
migrations.RemoveField(model_name='category', name='created_time'),
migrations.RemoveField(model_name='category', name='last_mod_time'),
migrations.RemoveField(model_name='links', name='created_time'),
migrations.RemoveField(model_name='sidebar', name='created_time'),
migrations.RemoveField(model_name='tag', name='created_time'),
migrations.RemoveField(model_name='tag', name='last_mod_time'),
# 3. 添加新的时间字段(英文命名)
migrations.AddField(
model_name='article',
name='creation_time', # 新创建时间字段(英文命名)
field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='creation time'),
),
migrations.AddField(
model_name='article',
name='last_modify_time', # 新最后修改时间字段
field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='modify time'),
),
migrations.AddField(model_name='category', name='creation_time', field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='creation time')),
migrations.AddField(model_name='category', name='last_modify_time', field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='modify time')),
migrations.AddField(model_name='links', name='creation_time', field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='creation time')),
migrations.AddField(model_name='sidebar', name='creation_time', field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='creation time')),
migrations.AddField(model_name='tag', name='creation_time', field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='creation time')),
migrations.AddField(model_name='tag', name='last_modify_time', field=models.DateTimeField(default=django.utils.timezone.now, verbose_name='modify time')),
# 4. 修改所有字段的 verbose_name 为英文(仅列举部分代表性字段)
migrations.AlterField(
model_name='article',
name='article_order',
field=models.IntegerField(default=0, verbose_name='order'), # 原"排序"改为"order"
),
migrations.AlterField(
model_name='article',
name='author',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL, verbose_name='author'), # 原"作者"改为"author"
),
migrations.AlterField(
model_name='article',
name='body',
field=mdeditor.fields.MDTextField(verbose_name='body'), # 原"正文"改为"body"
),
migrations.AlterField(
model_name='article',
name='comment_status',
field=models.CharField(choices=[('o', 'Open'), ('c', 'Close')], default='o', max_length=1, verbose_name='comment status'), # 状态选项和描述均改为英文
),
# ... 省略其他字段的AlterField均为verbose_name改为英文
# 友情链接模型的显示类型选项修改(中文场景改为英文场景)
migrations.AlterField(
model_name='links',
name='show_type',
field=models.CharField(
choices=[('i', 'index'), ('l', 'list'), ('p', 'post'), ('a', 'all'), ('s', 'slide')],
default='i',
max_length=1,
verbose_name='show type'
),
),
]

@ -0,0 +1,30 @@
# 生成信息由Django 4.2.7版本在2024-01-26 02:41自动生成的迁移文件
from django.db import migrations
class Migration(migrations.Migration):
"""
数据库迁移类修改BlogSettings模型的元数据选项
将模型的显示名称从之前的命名可能为中文或其他语言统一改为英文适配国际化需求
"""
# 依赖关系当前迁移依赖于博客应用的上一个迁移文件0005_...
# 确保在之前的模型结构调整完成后再执行本次元数据修改
dependencies = [
('blog', '0005_alter_article_options_alter_category_options_and_more'),
]
# 迁移操作修改BlogSettings模型的元数据选项
operations = [
migrations.AlterModelOptions(
name='blogsettings', # 目标模型网站配置表BlogSettings
options={
'verbose_name': 'Website configuration', # 模型单数显示名称(改为英文)
'verbose_name_plural': 'Website configuration' # 模型复数显示名称(改为英文,因配置通常为单条记录,复数同单数)
},
# 修改目的:
# 1. 统一模型显示名称为英文,适配国际化场景(如多语言网站后台)
# 2. 使模型名称更符合英文开发环境的命名习惯,提升代码一致性
# 3. 之前的版本可能使用中文(如"网站配置")或其他命名,此处统一规范化
),
]

@ -0,0 +1,415 @@
# 导入日志模块,用于记录系统运行日志
import logging
# 导入正则表达式模块用于处理文本中的匹配如提取图片URL
import re
# 导入抽象基类相关工具,用于定义抽象方法
from abc import abstractmethod
# 导入Django配置、异常、模型、URL反转等核心功能
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import models
from django.urls import reverse
from django.utils.timezone import now
from django.utils.translation import gettext_lazy as _ # 国际化翻译工具
# 导入markdown编辑器字段用于文章内容编辑
from mdeditor.fields import MDTextField
# 导入slug生成工具用于生成URL友好的标识符
from uuslug import slugify
# 导入自定义工具:缓存装饰器、缓存操作、获取当前站点信息
from djangoblog.utils import cache_decorator, cache
from djangoblog.utils import get_current_site
# 创建当前模块的日志记录器
logger = logging.getLogger(__name__)
class LinkShowType(models.TextChoices):
"""
链接展示位置的枚举类
定义友情链接在网站中的展示位置选项
"""
I = ('i', _('index')) # 首页展示
L = ('l', _('list')) # 列表页展示
P = ('p', _('post')) # 文章详情页展示
A = ('a', _('all')) # 所有页面展示
S = ('s', _('slide')) # 幻灯片展示
class BaseModel(models.Model):
"""
模型基类所有其他模型的父类
封装通用字段和方法减少代码重复
"""
# 自增主键ID
id = models.AutoField(primary_key=True)
# 创建时间字段,默认为当前时间
creation_time = models.DateTimeField(_('creation time'), default=now)
# 最后修改时间字段,默认为当前时间
last_modify_time = models.DateTimeField(_('modify time'), default=now)
def save(self, *args, **kwargs):
"""
重写保存方法扩展功能
1. 单独处理文章阅读量更新避免更新其他字段
2. 自动生成slugURL友好标识符
"""
# 判断是否是更新文章阅读量的操作
is_update_views = isinstance(
self,
Article) and 'update_fields' in kwargs and kwargs['update_fields'] == ['views']
if is_update_views:
# 仅更新阅读量字段(优化性能)
Article.objects.filter(pk=self.pk).update(views=self.views)
else:
# 若模型包含slug字段则自动生成slug基于title或name字段
if 'slug' in self.__dict__:
# 优先使用title字段否则使用name字段作为slug源
slug_source = getattr(self, 'title') if 'title' in self.__dict__ else getattr(self, 'name')
setattr(self, 'slug', slugify(slug_source)) # 生成URL友好的slug
# 调用父类保存方法
super().save(*args, **kwargs)
def get_full_url(self):
"""生成包含域名的完整URL用于外部链接或分享"""
site_domain = get_current_site().domain # 获取当前站点域名
full_url = f"https://{site_domain}{self.get_absolute_url()}"
return full_url
class Meta:
abstract = True # 声明为抽象基类,不生成数据库表
@abstractmethod
def get_absolute_url(self):
"""
抽象方法获取模型实例的相对URL
子类必须实现用于生成详情页链接
"""
pass
class Article(BaseModel):
"""
文章模型
存储博客文章的核心信息
"""
# 文章状态选项:草稿/已发布
STATUS_CHOICES = (
('d', _('Draft')), # 草稿
('p', _('Published')), # 已发布
)
# 评论状态选项:开启/关闭
COMMENT_STATUS = (
('o', _('Open')), # 允许评论
('c', _('Close')), # 关闭评论
)
# 内容类型选项:文章/页面(如关于页、联系页)
TYPE = (
('a', _('Article')), # 普通文章
('p', _('Page')), # 独立页面
)
title = models.CharField(_('title'), max_length=200, unique=True) # 文章标题(唯一)
body = MDTextField(_('body')) # 文章内容markdown格式
pub_time = models.DateTimeField(
_('publish time'), blank=False, null=False, default=now) # 发布时间
status = models.CharField(
_('status'), max_length=1, choices=STATUS_CHOICES, default='p') # 文章状态
comment_status = models.CharField(
_('comment status'), max_length=1, choices=COMMENT_STATUS, default='o') # 评论状态
type = models.CharField(_('type'), max_length=1, choices=TYPE, default='a') # 内容类型
views = models.PositiveIntegerField(_('views'), default=0) # 阅读量
author = models.ForeignKey(
settings.AUTH_USER_MODEL,
verbose_name=_('author'),
blank=False,
null=False,
on_delete=models.CASCADE # 关联用户,用户删除时文章也删除
)
article_order = models.IntegerField(
_('order'), blank=False, null=False, default=0) # 文章排序权重
show_toc = models.BooleanField(_('show toc'), blank=False, null=False, default=False) # 是否显示目录
category = models.ForeignKey(
'Category',
verbose_name=_('category'),
on_delete=models.CASCADE,
blank=False,
null=False # 关联分类,分类删除时文章也删除
)
tags = models.ManyToManyField('Tag', verbose_name=_('tag'), blank=True) # 多对多关联标签
def body_to_string(self):
"""将文章内容转换为字符串(用于搜索等场景)"""
return self.body
def __str__(self):
"""模型实例的字符串表示(用于后台显示)"""
return self.title
class Meta:
ordering = ['-article_order', '-pub_time'] # 排序规则:先按排序权重降序,再按发布时间降序
verbose_name = _('article') # 模型显示名称(单数)
verbose_name_plural = verbose_name # 模型显示名称(复数)
get_latest_by = 'id' # 按id获取最新记录
def get_absolute_url(self):
"""生成文章详情页的相对URL"""
return reverse('blog:detailbyid', kwargs={
'article_id': self.id,
'year': self.creation_time.year,
'month': self.creation_time.month,
'day': self.creation_time.day
})
@cache_decorator(60 * 60 * 10) # 缓存10小时
def get_category_tree(self):
"""获取当前文章所属分类的层级树(含父级分类)"""
tree = self.category.get_category_tree()
# 转换为 (分类名称, 分类URL) 的列表
return [(c.name, c.get_absolute_url()) for c in tree]
def save(self, *args, **kwargs):
"""重写保存方法(可扩展,此处直接调用父类方法)"""
super().save(*args, **kwargs)
def viewed(self):
"""增加阅读量并保存仅更新views字段"""
self.views += 1
self.save(update_fields=['views']) # 只更新views字段优化性能
def comment_list(self):
"""获取当前文章的有效评论列表(带缓存)"""
cache_key = f'article_comments_{self.id}'
# 尝试从缓存获取
cached_comments = cache.get(cache_key)
if cached_comments:
logger.info(f'从缓存获取文章评论: {self.id}')
return cached_comments
# 缓存未命中,从数据库查询
comments = self.comment_set.filter(is_enable=True).order_by('-id') # 按ID降序最新在前
cache.set(cache_key, comments, 60 * 100) # 缓存100分钟
logger.info(f'缓存文章评论: {self.id}')
return comments
def get_admin_url(self):
"""获取后台管理编辑页面的URL"""
# 自动获取模型的app标签和模型名称
info = (self._meta.app_label, self._meta.model_name)
return reverse(f'admin:{info[0]}_{info[1]}_change', args=(self.pk,))
@cache_decorator(expiration=60 * 100) # 缓存100分钟
def next_article(self):
"""获取下一篇文章ID更大、已发布的第一篇"""
return Article.objects.filter(id__gt=self.id, status='p').order_by('id').first()
@cache_decorator(expiration=60 * 100) # 缓存100分钟
def prev_article(self):
"""获取上一篇文章ID更小、已发布的最后一篇"""
return Article.objects.filter(id__lt=self.id, status='p').first()
def get_first_image_url(self):
"""从文章内容中提取第一张图片的URL用于封面图等场景"""
# 正则匹配markdown图片格式: ![描述](URL)
match = re.search(r'!\[.*?\]\((.+?)\)', self.body)
if match:
return match.group(1) # 返回匹配到的URL
return "" # 无图片时返回空
class Category(BaseModel):
"""
分类模型
用于文章的分类管理支持层级结构父分类
"""
name = models.CharField(_('category name'), max_length=30, unique=True) # 分类名称(唯一)
parent_category = models.ForeignKey(
'self',
verbose_name=_('parent category'),
blank=True,
null=True,
on_delete=models.CASCADE # 父分类删除时,子分类也删除
)
slug = models.SlugField(default='no-slug', max_length=60, blank=True) # 分类的URL标识符
index = models.IntegerField(default=0, verbose_name=_('index')) # 排序权重(值越大越靠前)
class Meta:
ordering = ['-index'] # 按排序权重降序排列
verbose_name = _('category')
verbose_name_plural = verbose_name
def get_absolute_url(self):
"""生成分类详情页的相对URL"""
return reverse('blog:category_detail', kwargs={'category_name': self.slug})
def __str__(self):
return self.name
@cache_decorator(60 * 60 * 10) # 缓存10小时
def get_category_tree(self):
"""
递归获取当前分类的层级树含所有父级分类
例如子分类 -> 父分类 -> 顶级分类
"""
categorys = []
def parse(category):
categorys.append(category)
if category.parent_category: # 若存在父分类,继续递归
parse(category.parent_category)
parse(self)
return categorys
@cache_decorator(60 * 60 * 10) # 缓存10小时
def get_sub_categorys(self):
"""
递归获取当前分类的所有子分类含多级子分类
"""
categorys = []
all_categorys = Category.objects.all() # 获取所有分类
def parse(category):
if category not in categorys:
categorys.append(category)
# 获取当前分类的直接子分类
childs = all_categorys.filter(parent_category=category)
for child in childs:
if child not in categorys:
categorys.append(child)
parse(child) # 递归处理子分类
parse(self)
return categorys
class Tag(BaseModel):
"""
标签模型
用于文章的标签管理多对多关系
"""
name = models.CharField(_('tag name'), max_length=30, unique=True) # 标签名称(唯一)
slug = models.SlugField(default='no-slug', max_length=60, blank=True) # 标签的URL标识符
def __str__(self):
return self.name
def get_absolute_url(self):
"""生成标签详情页的相对URL"""
return reverse('blog:tag_detail', kwargs={'tag_name': self.slug})
@cache_decorator(60 * 60 * 10) # 缓存10小时
def get_article_count(self):
"""获取该标签关联的文章数量(去重)"""
return Article.objects.filter(tags__name=self.name).distinct().count()
class Meta:
ordering = ['name'] # 按名称升序排列
verbose_name = _('tag')
verbose_name_plural = verbose_name
class Links(models.Model):
"""
友情链接模型
存储网站的友情链接信息
"""
name = models.CharField(_('link name'), max_length=30, unique=True) # 链接名称(唯一)
link = models.URLField(_('link')) # 链接URL
sequence = models.IntegerField(_('order'), unique=True) # 排序序号(唯一,用于控制显示顺序)
is_enable = models.BooleanField(
_('is show'), default=True, blank=False, null=False) # 是否启用(显示)
show_type = models.CharField(
_('show type'),
max_length=1,
choices=LinkShowType.choices,
default=LinkShowType.I # 默认为首页展示
)
creation_time = models.DateTimeField(_('creation time'), default=now) # 创建时间
last_mod_time = models.DateTimeField(_('modify time'), default=now) # 最后修改时间
class Meta:
ordering = ['sequence'] # 按排序序号升序排列
verbose_name = _('link')
verbose_name_plural = verbose_name
def __str__(self):
return self.name
class SideBar(models.Model):
"""
侧边栏模型
用于展示网站侧边栏内容支持HTML
"""
name = models.CharField(_('title'), max_length=100) # 侧边栏标题
content = models.TextField(_('content')) # 侧边栏内容HTML格式
sequence = models.IntegerField(_('order'), unique=True) # 排序序号(控制显示顺序)
is_enable = models.BooleanField(_('is enable'), default=True) # 是否启用
creation_time = models.DateTimeField(_('creation time'), default=now)
last_mod_time = models.DateTimeField(_('modify time'), default=now)
class Meta:
ordering = ['sequence'] # 按排序序号升序排列
verbose_name = _('sidebar')
verbose_name_plural = verbose_name
def __str__(self):
return self.name
class BlogSettings(models.Model):
"""
博客配置模型
存储网站的全局设置单例模式仅允许一条记录
"""
site_name = models.CharField(
_('site name'), max_length=200, null=False, blank=False, default='') # 网站名称
site_description = models.TextField(
_('site description'), max_length=1000, null=False, blank=False, default='') # 网站描述
site_seo_description = models.TextField(
_('site seo description'), max_length=1000, null=False, blank=False, default='') # SEO描述
site_keywords = models.TextField(
_('site keywords'), max_length=1000, null=False, blank=False, default='') # 网站关键词SEO
article_sub_length = models.IntegerField(_('article sub length'), default=300) # 文章摘要长度
sidebar_article_count = models.IntegerField(_('sidebar article count'), default=10) # 侧边栏文章数量
sidebar_comment_count = models.IntegerField(_('sidebar comment count'), default=5) # 侧边栏评论数量
article_comment_count = models.IntegerField(_('article comment count'), default=5) # 文章页显示评论数量
show_google_adsense = models.BooleanField(_('show adsense'), default=False) # 是否显示谷歌广告
google_adsense_codes = models.TextField(
_('adsense code'), max_length=2000, null=True, blank=True, default='') # 谷歌广告代码
open_site_comment = models.BooleanField(_('open site comment'), default=True) # 是否开启全站评论
global_header = models.TextField("公共头部", null=True, blank=True, default='') # 全局头部HTML代码
global_footer = models.TextField("公共尾部", null=True, blank=True, default='') # 全局尾部HTML代码
beian_code = models.CharField(
'备案号', max_length=2000, null=True, blank=True, default='') # 网站备案号
analytics_code = models.TextField(
"网站统计代码", max_length=1000, null=False, blank=False, default='') # 统计代码(如百度统计)
show_gongan_code = models.BooleanField(
'是否显示公安备案号', default=False, null=False) # 是否显示公安备案号
gongan_beiancode = models.TextField(
'公安备案号', max_length=2000, null=True, blank=True, default='') # 公安备案号
comment_need_review = models.BooleanField(
'评论是否需要审核', default=False, null=False) # 评论是否需要审核后显示
class Meta:
verbose_name = _('Website configuration')
verbose_name_plural = verbose_name
def __str__(self):
return self.site_name
def clean(self):
"""
数据验证确保仅存在一条配置记录
在保存前调用用于防止创建多条配置
"""
if BlogSettings.objects.exclude(id=self.id).count():
raise ValidationError(_('There can only be one configuration')) # 仅允许一个配置记录
def save(self, *args, **kwargs):
"""保存配置后清除缓存(确保配置实时生效)"""
super().save(*args, **kwargs)
from djangoblog.utils import cache
cache.clear() # 清除所有缓存

@ -0,0 +1,32 @@
# 导入Haystack的索引模块用于定义搜索索引
from haystack import indexes
# 导入博客文章模型,作为搜索索引的数据源
from blog.models import Article
class ArticleIndex(indexes.SearchIndex, indexes.Indexable):
"""
文章搜索索引类用于配置Haystack搜索的索引规则
继承自Haystack的SearchIndex搜索索引基类和Indexable可索引接口
"""
# 定义主搜索字段:
# - document=True标记为主要搜索字段Haystack默认以此字段作为全文检索的基础
# - use_template=True指定使用模板来构建索引内容模板通常存放于templates/search/indexes/[app名]/[模型名]_text.txt
text = indexes.CharField(document=True, use_template=True)
def get_model(self):
"""
必须实现的方法指定该索引对应的模型
返回值为需要被索引的Django模型类
"""
return Article
def index_queryset(self, using=None):
"""
定义需要被索引的数据集
筛选出状态为"已发布"status='p'的文章仅对这些文章建立搜索索引
:param using: 可选参数指定搜索引擎多引擎场景下使用
:return: 查询集QuerySet包含需要被索引的模型实例
"""
return self.get_model().objects.filter(status='p')

@ -0,0 +1,408 @@
import hashlib # 用于Gravatar头像的MD5哈希计算
import logging # 日志记录
import random # 随机选择样式
import urllib # URL编码处理
from django import template # 模板标签核心模块
from django.conf import settings # 项目配置
from django.db.models import Q # 数据库查询条件
from django.shortcuts import get_object_or_404 # 获取对象或返回404
from django.template.defaultfilters import stringfilter # 字符串过滤器装饰器
from django.templatetags.static import static # 静态文件URL生成
from django.urls import reverse # URL反向解析
from django.utils.safestring import mark_safe # 标记安全HTML字符串
# 导入项目模型
from blog.models import Article, Category, Tag, Links, SideBar, LinkShowType
from comments.models import Comment
# 导入工具类和插件
from djangoblog.utils import CommonMarkdown, sanitize_html # Markdown处理和HTML净化
from djangoblog.utils import cache # 缓存工具
from djangoblog.utils import get_current_site # 获取当前站点信息
from oauth.models import OAuthUser # OAuth用户模型
from djangoblog.plugin_manage import hooks # 插件钩子
# 日志配置
logger = logging.getLogger(__name__)
# 注册模板标签库
register = template.Library()
@register.simple_tag(takes_context=True)
def head_meta(context):
"""
页面头部元信息标签通过插件钩子扩展
用于动态生成SEO相关的meta标签如titlekeywords等
:param context: 模板上下文
:return: 经过插件处理的安全HTML字符串
"""
return mark_safe(hooks.apply_filters('head_meta', '', context))
@register.simple_tag
def timeformat(data):
"""
时间格式化标签
datetime 对象格式化为 settings.TIME_FORMAT 定义的样式
:param data: datetime对象
:return: 格式化后的时间字符串失败返回空
"""
try:
return data.strftime(settings.TIME_FORMAT)
except Exception as e:
logger.error(e)
return ""
@register.simple_tag
def datetimeformat(data):
"""
日期时间格式化标签
datetime 对象格式化为 settings.DATE_TIME_FORMAT 定义的样式
:param data: datetime对象
:return: 格式化后的日期时间字符串失败返回空
"""
try:
return data.strftime(settings.DATE_TIME_FORMAT)
except Exception as e:
logger.error(e)
return ""
@register.filter()
@stringfilter
def custom_markdown(content):
"""
Markdown渲染过滤器
将Markdown格式的文本转换为HTML并标记为安全
:param content: Markdown文本
:return: 安全的HTML字符串
"""
return mark_safe(CommonMarkdown.get_markdown(content))
@register.simple_tag
def get_markdown_toc(content):
"""
获取Markdown内容的目录TOC
用于生成文章目录导航
:param content: Markdown文本
:return: 目录的HTML字符串
"""
from djangoblog.utils import CommonMarkdown
body, toc = CommonMarkdown.get_markdown_with_toc(content)
return mark_safe(toc)
@register.filter()
@stringfilter
def comment_markdown(content):
"""
评论内容的Markdown渲染过滤器
先转换为HTML再通过sanitize_html净化过滤危险标签
:param content: 评论的Markdown文本
:return: 安全的HTML字符串
"""
content = CommonMarkdown.get_markdown(content)
return mark_safe(sanitize_html(content))
@register.filter(is_safe=True)
@stringfilter
def truncatechars_content(content):
"""
文章内容摘要过滤器
根据网站配置的摘要长度截断HTML内容保留标签结构
:param content: 文章HTML内容
:return: 截断后的安全HTML字符串
"""
from django.template.defaultfilters import truncatechars_html
from djangoblog.utils import get_blog_setting
blogsetting = get_blog_setting() # 获取网站配置
return truncatechars_html(content, blogsetting.article_sub_length)
@register.filter(is_safe=True)
@stringfilter
def truncate(content):
"""
简单截断过滤器纯文本
去除HTML标签后截断前150个字符
:param content: 带HTML的文本
:return: 截断后的纯文本
"""
from django.utils.html import strip_tags
return strip_tags(content)[:150]
@register.inclusion_tag('blog/tags/breadcrumb.html')
def load_breadcrumb(article):
"""
面包屑导航标签
生成文章的分类层级导航首页 > 技术 > Python > 文章标题
:param article: 文章对象
:return: 包含导航层级和标题的上下文
"""
names = article.get_category_tree() # 获取分类层级列表
from djangoblog.utils import get_blog_setting
blogsetting = get_blog_setting()
site = get_current_site().domain
names.append((blogsetting.site_name, '/')) # 添加首页
names = names[::-1] # 反转层级顺序(从顶级到当前)
return {
'names': names,
'title': article.title,
'count': len(names) + 1
}
@register.inclusion_tag('blog/tags/article_tag_list.html')
def load_articletags(article):
"""
文章标签列表标签
生成文章关联的标签列表包含标签URL文章数量和随机样式
:param article: 文章对象
:return: 包含标签信息的上下文
"""
tags = article.tags.all()
tags_list = []
for tag in tags:
url = tag.get_absolute_url() # 标签页URL
count = tag.get_article_count() # 标签关联的文章数
# 随机选择Bootstrap样式如primary、success等
tags_list.append((
url, count, tag, random.choice(settings.BOOTSTRAP_COLOR_TYPES)
))
return {'article_tags_list': tags_list}
@register.inclusion_tag('blog/tags/sidebar.html')
def load_sidebar(user, linktype):
"""
侧边栏内容标签
加载侧边栏所需数据热门文章分类标签云等并使用缓存优化性能
:param user: 当前用户
:param linktype: 链接显示类型控制友情链接显示场景
:return: 侧边栏数据上下文
"""
# 缓存键:区分不同链接类型的侧边栏
cachekey = "sidebar" + linktype
value = cache.get(cachekey)
if value: # 命中缓存直接返回
value['user'] = user
return value
else: # 未命中缓存,重新计算并缓存
logger.info('load sidebar')
from djangoblog.utils import get_blog_setting
blogsetting = get_blog_setting() # 网站配置
# 侧边栏数据查询
recent_articles = Article.objects.filter(status='p')[:blogsetting.sidebar_article_count] # 最新文章
sidebar_categorys = Category.objects.all() # 所有分类
extra_sidebars = SideBar.objects.filter(is_enable=True).order_by('sequence') # 自定义侧边栏
most_read_articles = Article.objects.filter(status='p').order_by('-views')[
:blogsetting.sidebar_article_count] # 热门文章
dates = Article.objects.datetimes('creation_time', 'month', order='DESC') # 文章归档日期
# 符合显示类型的友情链接
links = Links.objects.filter(is_enable=True).filter(
Q(show_type=str(linktype)) | Q(show_type=LinkShowType.A)
)
# 最新评论
commment_list = Comment.objects.filter(is_enable=True).order_by('-id')[:blogsetting.sidebar_comment_count]
# 标签云(根据文章数量计算字体大小)
sidebar_tags = None
tags = Tag.objects.all()
if tags and len(tags) > 0:
# 过滤有文章的标签
tag_with_count = [(t, t.get_article_count()) for t in tags if t.get_article_count()]
if tag_with_count:
total = sum([t[1] for t in tag_with_count])
avg = total / len(tag_with_count) # 平均文章数
increment = 5 # 字体大小增量
# 计算每个标签的字体大小(与平均数量成正比)
sidebar_tags = [
(t[0], t[1], (t[1] / avg) * increment + 10)
for t in tag_with_count
]
random.shuffle(sidebar_tags) # 随机排序
# 组装侧边栏数据
value = {
'recent_articles': recent_articles,
'sidebar_categorys': sidebar_categorys,
'most_read_articles': most_read_articles,
'article_dates': dates,
'sidebar_comments': commment_list,
'sidabar_links': links,
'show_google_adsense': blogsetting.show_google_adsense,
'google_adsense_codes': blogsetting.google_adsense_codes,
'open_site_comment': blogsetting.open_site_comment,
'show_gongan_code': blogsetting.show_gongan_code,
'sidebar_tags': sidebar_tags,
'extra_sidebars': extra_sidebars
}
# 缓存3小时
cache.set(cachekey, value, 60 * 60 * 3)
logger.info(f'set sidebar cache.key:{cachekey}')
value['user'] = user
return value
@register.inclusion_tag('blog/tags/article_meta_info.html')
def load_article_metas(article, user):
"""
文章元信息标签
加载文章的元数据作者发布时间分类等
:param article: 文章对象
:param user: 当前用户
:return: 包含文章和用户的上下文
"""
return {'article': article, 'user': user}
@register.inclusion_tag('blog/tags/article_pagination.html')
def load_pagination_info(page_obj, page_type, tag_name):
"""
分页导航标签
根据不同页面类型首页标签页分类页等生成上一页/下一页链接
:param page_obj: Django分页对象
:param page_type: 页面类型如分类标签归档作者文章归档等
:param tag_name: 标签/分类/作者名称用于URL参数
:return: 包含分页链接的上下文
"""
previous_url = ''
next_url = ''
# 首页分页
if page_type == '':
if page_obj.has_next():
next_url = reverse('blog:index_page', kwargs={'page': page_obj.next_page_number()})
if page_obj.has_previous():
previous_url = reverse('blog:index_page', kwargs={'page': page_obj.previous_page_number()})
# 标签页分页
elif page_type == '分类标签归档':
tag = get_object_or_404(Tag, name=tag_name)
if page_obj.has_next():
next_url = reverse('blog:tag_detail_page',
kwargs={'page': page_obj.next_page_number(), 'tag_name': tag.slug})
if page_obj.has_previous():
previous_url = reverse('blog:tag_detail_page',
kwargs={'page': page_obj.previous_page_number(), 'tag_name': tag.slug})
# 作者文章分页
elif page_type == '作者文章归档':
if page_obj.has_next():
next_url = reverse('blog:author_detail_page',
kwargs={'page': page_obj.next_page_number(), 'author_name': tag_name})
if page_obj.has_previous():
previous_url = reverse('blog:author_detail_page',
kwargs={'page': page_obj.previous_page_number(), 'author_name': tag_name})
# 分类页分页
elif page_type == '分类目录归档':
category = get_object_or_404(Category, name=tag_name)
if page_obj.has_next():
next_url = reverse('blog:category_detail_page',
kwargs={'page': page_obj.next_page_number(), 'category_name': category.slug})
if page_obj.has_previous():
previous_url = reverse('blog:category_detail_page',
kwargs={'page': page_obj.previous_page_number(), 'category_name': category.slug})
return {
'previous_url': previous_url,
'next_url': next_url,
'page_obj': page_obj
}
@register.inclusion_tag('blog/tags/article_info.html')
def load_article_detail(article, isindex, user):
"""
文章详情标签
加载文章详情页或列表页的展示内容列表页显示摘要详情页显示完整内容
:param article: 文章对象
:param isindex: 是否为列表页True/False
:param user: 当前用户
:return: 包含文章展示信息的上下文
"""
from djangoblog.utils import get_blog_setting
blogsetting = get_blog_setting()
return {
'article': article,
'isindex': isindex,
'user': user,
'open_site_comment': blogsetting.open_site_comment, # 是否允许评论
}
@register.filter
def gravatar_url(email, size=40):
"""
Gravatar头像URL过滤器
生成用户的Gravatar头像URL优先使用OAuth用户的头像
:param email: 用户邮箱
:param size: 头像尺寸
:return: 头像URL字符串
"""
cachekey = f'gravatat/{email}'
url = cache.get(cachekey)
if url: # 缓存命中
return url
else: # 缓存未命中
# 优先使用OAuth用户的头像
oauth_users = OAuthUser.objects.filter(email=email)
if oauth_users:
valid_avatars = [user for user in oauth_users if user.picture]
if valid_avatars:
return valid_avatars[0].picture
# 生成Gravatar URL邮箱MD5哈希 + 尺寸 + 默认头像)
email = email.encode('utf-8')
default_avatar = static('blog/img/avatar.png') # 本地默认头像
url = f"https://www.gravatar.com/avatar/{hashlib.md5(email.lower()).hexdigest()}?{urllib.parse.urlencode({'d': default_avatar, 's': str(size)})}"
# 缓存10小时
cache.set(cachekey, url, 60 * 60 * 10)
logger.info(f'set gravatar cache.key:{cachekey}')
return url
@register.filter
def gravatar(email, size=40):
"""
Gravatar头像标签
生成包含头像图片的HTML标签
:param email: 用户邮箱
:param size: 头像尺寸
:return: 安全的img标签HTML字符串
"""
url = gravatar_url(email, size)
return mark_safe(f'<img src="{url}" height="{size}" width="{size}">')
@register.simple_tag
def query(qs, **kwargs):
"""
查询集过滤标签
在模板中对查询集进行过滤{% query books author=author as mybooks %}
:param qs: Django查询集
:param kwargs: 过滤条件键值对
:return: 过滤后的查询集
"""
return qs.filter(**kwargs)
@register.filter
def addstr(arg1, arg2):
"""
字符串拼接过滤器
将两个参数转换为字符串并拼接
:param arg1: 第一个参数
:param arg2: 第二个参数
:return: 拼接后的字符串
"""
return str(arg1) + str(arg2)

@ -0,0 +1,331 @@
import os
import requests
# 导入Django核心模块配置、文件上传、命令调用、分页、静态文件、测试工具、URL反转、时区
from django.conf import settings
from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.management import call_command
from django.core.paginator import Paginator
from django.templatetags.static import static
from django.test import Client, RequestFactory, TestCase
from django.urls import reverse
from django.utils import timezone
# 导入项目相关模型和工具用户、博客模型、表单、模板标签、工具函数、OAuth相关
from accounts.models import BlogUser
from blog.forms import BlogSearchForm
from blog.models import Article, Category, Tag, SideBar, Links
from blog.templatetags.blog_tags import load_pagination_info, load_articletags
from djangoblog.utils import get_current_site, get_sha256
from oauth.models import OAuthUser, OAuthConfig
class ArticleTest(TestCase):
"""
博客核心功能测试类
测试文章分类标签搜索权限等核心业务逻辑
"""
def setUp(self):
"""
测试前的初始化方法
创建测试客户端和请求工厂用于模拟HTTP请求
"""
self.client = Client() # 模拟用户浏览器的客户端
self.factory = RequestFactory() # 用于构造请求对象的工厂
def test_validate_article(self):
"""
测试文章相关核心功能
- 用户模型操作
- 分类标签侧边栏链接等模型CRUD
- 文章发布分页搜索评论等流程
- 页面访问状态码验证
"""
# 获取当前站点域名
site = get_current_site().domain
# 创建或获取测试用户(管理员)
user = BlogUser.objects.get_or_create(
email="liangliangyy@gmail.com",
username="liangliangyy")[0]
user.set_password("liangliangyy") # 设置密码
user.is_staff = True # 允许登录admin
user.is_superuser = True # 超级管理员权限
user.save()
# 测试用户个人页面访问
response = self.client.get(user.get_absolute_url())
self.assertEqual(response.status_code, 200) # 验证页面正常访问
# 测试admin后台页面访问未登录状态实际会跳转登录页
self.client.get('/admin/servermanager/emailsendlog/')
self.client.get('admin/admin/logentry/')
# 创建测试侧边栏
s = SideBar()
s.sequence = 1 # 排序序号
s.name = 'test' # 名称
s.content = 'test content' # 内容
s.is_enable = True # 启用
s.save()
# 创建测试分类
category = Category()
category.name = "category" # 分类名称
category.creation_time = timezone.now()
category.last_mod_time = timezone.now()
category.save()
# 创建测试标签
tag = Tag()
tag.name = "nicetag" # 标签名称
tag.save()
# 创建测试文章
article = Article()
article.title = "nicetitle" # 标题
article.body = "nicecontent" # 内容
article.author = user # 作者
article.category = category # 所属分类
article.type = 'a' # 类型为文章
article.status = 'p' # 状态为已发布
article.save()
# 验证标签关联(初始无标签)
self.assertEqual(0, article.tags.count())
# 关联标签并验证
article.tags.add(tag)
article.save()
self.assertEqual(1, article.tags.count())
# 批量创建20篇测试文章用于测试分页
for i in range(20):
article = Article()
article.title = f"nicetitle{i}"
article.body = f"nicetitle{i}"
article.author = user
article.category = category
article.type = 'a'
article.status = 'p'
article.save()
article.tags.add(tag)
article.save()
# 测试Elasticsearch搜索如果启用
from blog.documents import ELASTICSEARCH_ENABLED
if ELASTICSEARCH_ENABLED:
call_command("build_index") # 构建搜索索引
response = self.client.get('/search', {'q': 'nicetitle'}) # 执行搜索
self.assertEqual(response.status_code, 200) # 验证搜索页正常
# 测试文章详情页访问
response = self.client.get(article.get_absolute_url())
self.assertEqual(response.status_code, 200)
# 测试搜索引擎推送功能
from djangoblog.spider_notify import SpiderNotify
SpiderNotify.notify(article.get_absolute_url()) # 推送文章URL到搜索引擎
# 测试标签页访问
response = self.client.get(tag.get_absolute_url())
self.assertEqual(response.status_code, 200)
# 测试分类页访问
response = self.client.get(category.get_absolute_url())
self.assertEqual(response.status_code, 200)
# 测试搜索功能(无结果场景)
response = self.client.get('/search', {'q': 'django'})
self.assertEqual(response.status_code, 200)
# 测试文章标签模板标签
s = load_articletags(article)
self.assertIsNotNone(s) # 验证模板标签返回结果
# 登录测试用户
self.client.login(username='liangliangyy', password='liangliangyy')
# 测试归档页访问
response = self.client.get(reverse('blog:archives'))
self.assertEqual(response.status_code, 200)
# 测试各种场景下的分页功能
# 1. 所有文章分页
p = Paginator(Article.objects.all(), settings.PAGINATE_BY)
self.check_pagination(p, '', '')
# 2. 标签筛选分页
p = Paginator(Article.objects.filter(tags=tag), settings.PAGINATE_BY)
self.check_pagination(p, '分类标签归档', tag.slug)
# 3. 作者筛选分页
p = Paginator(
Article.objects.filter(author__username='liangliangyy'),
settings.PAGINATE_BY
)
self.check_pagination(p, '作者文章归档', 'liangliangyy')
# 4. 分类筛选分页
p = Paginator(Article.objects.filter(category=category), settings.PAGINATE_BY)
self.check_pagination(p, '分类目录归档', category.slug)
# 测试搜索表单
f = BlogSearchForm()
f.search() # 调用搜索方法
# 测试百度搜索引擎推送
SpiderNotify.baidu_notify([article.get_full_url()])
# 测试头像相关模板标签
from blog.templatetags.blog_tags import gravatar_url, gravatar
u = gravatar_url('liangliangyy@gmail.com') # 获取头像URL
u = gravatar('liangliangyy@gmail.com') # 生成头像HTML
# 测试友情链接
link = Links(
sequence=1,
name="lylinux",
link='https://wwww.lylinux.net'
)
link.save()
response = self.client.get('/links.html') # 访问友情链接页
self.assertEqual(response.status_code, 200)
# 测试RSS订阅
response = self.client.get('/feed/')
self.assertEqual(response.status_code, 200)
# 测试站点地图
response = self.client.get('/sitemap.xml')
self.assertEqual(response.status_code, 200)
# 测试admin后台操作删除文章、访问日志
self.client.get("/admin/blog/article/1/delete/")
self.client.get('/admin/servermanager/emailsendlog/')
self.client.get('/admin/admin/logentry/')
self.client.get('/admin/admin/logentry/1/change/')
def check_pagination(self, p, type, value):
"""
测试分页功能的辅助方法
验证分页控件生成的URL是否可正常访问
"""
# 遍历所有分页页面
for page in range(1, p.num_pages + 1):
# 获取分页信息(通过模板标签)
s = load_pagination_info(p.page(page), type, value)
self.assertIsNotNone(s) # 验证分页信息生成正常
# 测试上一页链接
if s['previous_url']:
response = self.client.get(s['previous_url'])
self.assertEqual(response.status_code, 200)
# 测试下一页链接
if s['next_url']:
response = self.client.get(s['next_url'])
self.assertEqual(response.status_code, 200)
def test_image(self):
"""
测试图片上传功能
- 未授权上传
- 授权上传
- 头像保存工具函数
- 邮件发送工具函数
"""
# 下载测试图片Python官方logo
rsp = requests.get('https://www.python.org/static/img/python-logo.png')
imagepath = os.path.join(settings.BASE_DIR, 'python.png') # 保存路径
with open(imagepath, 'wb') as file:
file.write(rsp.content)
# 测试未授权上传预期403禁止访问
rsp = self.client.post('/upload')
self.assertEqual(rsp.status_code, 403)
# 生成上传签名基于SECRET_KEY的双重SHA256加密
sign = get_sha256(get_sha256(settings.SECRET_KEY))
# 测试授权上传
with open(imagepath, 'rb') as file:
# 构造上传文件对象
imgfile = SimpleUploadedFile(
'python.png', file.read(), content_type='image/jpg'
)
form_data = {'python.png': imgfile}
# 带签名上传
rsp = self.client.post(
f'/upload?sign={sign}', form_data, follow=True
)
self.assertEqual(rsp.status_code, 200) # 验证上传成功
# 清理测试文件
os.remove(imagepath)
# 测试用户头像保存和邮件发送工具函数
from djangoblog.utils import save_user_avatar, send_email
send_email(['qq@qq.com'], 'testTitle', 'testContent') # 测试发送邮件
save_user_avatar('https://www.python.org/static/img/python-logo.png') # 测试保存头像
def test_errorpage(self):
"""测试错误页面404页面"""
rsp = self.client.get('/eee') # 访问不存在的URL
self.assertEqual(rsp.status_code, 404) # 验证返回404
def test_commands(self):
"""
测试Django自定义命令
- 索引构建缓存清理数据同步等
"""
# 创建测试用户
user = BlogUser.objects.get_or_create(
email="liangliangyy@gmail.com",
username="liangliangyy")[0]
user.set_password("liangliangyy")
user.is_staff = True
user.is_superuser = True
user.save()
# 创建OAuth配置
c = OAuthConfig()
c.type = 'qq' # QQ登录
c.appkey = 'appkey'
c.appsecret = 'appsecret'
c.save()
# 创建关联用户的OAuth账号
u = OAuthUser()
u.type = 'qq'
u.openid = 'openid'
u.user = user # 关联本地用户
u.picture = static("/blog/img/avatar.png") # 头像
u.metadata = '''
{
"figureurl": "https://qzapp.qlogo.cn/qzapp/101513904/C740E30B4113EAA80E0D9918ABC78E82/30"
}''' # 第三方平台返回的元数据
u.save()
# 创建未关联本地用户的OAuth账号
u = OAuthUser()
u.type = 'qq'
u.openid = 'openid1'
u.picture = 'https://qzapp.qlogo.cn/qzapp/101513904/C740E30B4113EAA80E0D9918ABC78E82/30'
u.metadata = '''
{
"figureurl": "https://qzapp.qlogo.cn/qzapp/101513904/C740E30B4113EAA80E0D9918ABC78E82/30"
}'''
u.save()
# 测试Elasticsearch索引构建命令
from blog.documents import ELASTICSEARCH_ENABLED
if ELASTICSEARCH_ENABLED:
call_command("build_index")
# 测试其他自定义命令
call_command("ping_baidu", "all") # 百度链接推送
call_command("create_testdata") # 创建测试数据
call_command("clear_cache") # 清理缓存
call_command("sync_user_avatar") # 同步用户头像
call_command("build_search_words") # 构建搜索关键词

@ -0,0 +1,98 @@
# 导入Django URL路径处理和缓存装饰器
from django.urls import path
from django.views.decorators.cache import cache_page
# 导入当前应用的视图模块
from . import views
# 定义应用命名空间用于模板中URL反向解析如{% url 'blog:index' %}
app_name = "blog"
# URL路由配置列表映射URL路径到对应的视图
urlpatterns = [
# 首页路由
path(
r'', # 匹配根路径(如域名/
views.IndexView.as_view(), # 关联首页视图(基于类的视图)
name='index' # 路由名称,用于反向解析
),
# 首页分页路由(带页码参数)
path(
r'page/<int:page>/', # 匹配带页码的路径(如/page/2/
views.IndexView.as_view(), # 复用首页视图处理分页
name='index_page' # 路由名称
),
# 文章详情页路由按日期和ID
path(
r'article/<int:year>/<int:month>/<int:day>/<int:article_id>.html',
# 匹配路径格式article/年/月/日/文章ID.html如article/2023/10/01/1.html
views.ArticleDetailView.as_view(), # 关联文章详情视图
name='detailbyid' # 路由名称
),
# 分类详情页路由
path(
r'category/<slug:category_name>.html',
# 匹配路径category/分类别名.html如category/tech.htmlslug表示URL友好的字符串
views.CategoryDetailView.as_view(), # 关联分类详情视图
name='category_detail' # 路由名称
),
# 分类详情页分页路由
path(
r'category/<slug:category_name>/<int:page>.html',
# 匹配带页码的分类路径如category/tech/2.html
views.CategoryDetailView.as_view(), # 复用分类视图处理分页
name='category_detail_page' # 路由名称
),
# 作者文章列表路由
path(
r'author/<author_name>.html',
# 匹配路径author/用户名.html如author/admin.html
views.AuthorDetailView.as_view(), # 关联作者文章列表视图
name='author_detail' # 路由名称
),
# 作者文章列表分页路由
path(
r'author/<author_name>/<int:page>.html',
# 匹配带页码的作者路径如author/admin/2.html
views.AuthorDetailView.as_view(), # 复用作者视图处理分页
name='author_detail_page' # 路由名称
),
# 标签详情页路由
path(
r'tag/<slug:tag_name>.html',
# 匹配路径tag/标签别名.html如tag/python.html
views.TagDetailView.as_view(), # 关联标签详情视图
name='tag_detail' # 路由名称
),
# 标签详情页分页路由
path(
r'tag/<slug:tag_name>/<int:page>.html',
# 匹配带页码的标签路径如tag/python/2.html
views.TagDetailView.as_view(), # 复用标签视图处理分页
name='tag_detail_page' # 路由名称
),
# 文章归档页路由(带缓存)
path(
'archives.html', # 匹配路径archives.html
cache_page(60 * 60)(views.ArchivesView.as_view()), # 缓存60分钟60秒*60
name='archives' # 路由名称
),
# 友情链接页路由
path(
'links.html', # 匹配路径links.html
views.LinkListView.as_view(), # 关联友情链接视图
name='links' # 路由名称
),
# 文件上传接口路由
path(
r'upload', # 匹配路径upload
views.fileupload, # 关联文件上传视图函数(基于函数的视图)
name='upload' # 路由名称
),
# 清理缓存接口路由
path(
r'clean', # 匹配路径clean
views.clean_cache_view, # 关联清理缓存视图函数
name='clean' # 路由名称
),
]

@ -0,0 +1,498 @@
import logging
import os
import uuid # 用于生成唯一文件名
# 导入Django核心模块配置、分页、HTTP响应、视图工具、翻译等
from django.conf import settings
from django.core.paginator import Paginator
from django.http import HttpResponse, HttpResponseForbidden
from django.shortcuts import get_object_or_404, render
from django.templatetags.static import static # 生成静态文件URL
from django.utils import timezone # 处理时间
from django.utils.translation import gettext_lazy as _ # 国际化翻译
from django.views.decorators.csrf import csrf_exempt # 豁免CSRF验证用于文件上传
from django.views.generic.detail import DetailView # 详情页通用视图
from django.views.generic.list import ListView # 列表页通用视图
from haystack.views import SearchView # 搜索视图
# 导入项目模型、表单、工具和插件
from blog.models import Article, Category, LinkShowType, Links, Tag
from comments.forms import CommentForm # 评论表单
from djangoblog.plugin_manage import hooks # 插件钩子
from djangoblog.plugin_manage.hook_constants import ARTICLE_CONTENT_HOOK_NAME # 文章内容钩子常量
from djangoblog.utils import cache, get_blog_setting, get_sha256 # 缓存、配置和加密工具
# 创建当前模块的日志记录器
logger = logging.getLogger(__name__)
class ArticleListView(ListView):
"""
文章列表基类视图
封装文章列表页的通用逻辑分页缓存上下文处理
被首页分类标签作者等列表页继承
"""
# 模板路径:所有文章列表页共用此模板
template_name = 'blog/article_index.html'
# 上下文变量名:模板中用{{ article_list }}访问列表数据
context_object_name = 'article_list'
# 页面类型描述(如"分类目录归档"),子类需重写
page_type = ''
# 分页大小:从配置中获取
paginate_by = settings.PAGINATE_BY
# 分页参数名URL中页码的参数名如?page=2
page_kwarg = 'page'
# 友情链接显示类型默认为列表页L
link_type = LinkShowType.L
def get_view_cache_key(self):
"""获取视图缓存的key未实际使用预留扩展"""
return self.request.get['pages']
@property
def page_number(self):
"""获取当前页码从URL参数或kwargs中提取"""
page_kwarg = self.page_kwarg
# 优先从URL路径参数获取再从GET参数获取默认1
page = self.kwargs.get(page_kwarg) or self.request.GET.get(page_kwarg) or 1
return page
def get_queryset_cache_key(self):
"""
抽象方法获取查询集的缓存key
子类必须实现用于区分不同页面的缓存
"""
raise NotImplementedError()
def get_queryset_data(self):
"""
抽象方法获取查询集数据
子类必须实现定义具体的文章筛选逻辑
"""
raise NotImplementedError()
def get_queryset_from_cache(self, cache_key):
"""
从缓存获取或生成查询集数据
:param cache_key: 缓存唯一标识
:return: 文章查询集
"""
# 尝试从缓存获取
value = cache.get(cache_key)
if value:
logger.info(f'从缓存获取数据key: {cache_key}')
return value
else:
# 缓存未命中,执行查询并缓存
article_list = self.get_queryset_data()
cache.set(cache_key, article_list)
logger.info(f'设置缓存key: {cache_key}')
return article_list
def get_queryset(self):
"""
重写父类方法从缓存获取查询集
优化性能减少数据库查询
"""
cache_key = self.get_queryset_cache_key()
return self.get_queryset_from_cache(cache_key)
def get_context_data(self, **kwargs):
"""
扩展上下文数据添加友情链接显示类型
"""
kwargs['linktype'] = self.link_type
return super().get_context_data(** kwargs)
class IndexView(ArticleListView):
"""
首页视图
继承文章列表基类展示所有已发布的文章
"""
# 友情链接显示类型首页I
link_type = LinkShowType.I
def get_queryset_data(self):
"""获取首页文章列表已发布的普通文章type='a'"""
return Article.objects.filter(type='a', status='p')
def get_queryset_cache_key(self):
"""生成首页缓存key包含页码"""
return f'index_{self.page_number}'
class ArticleDetailView(DetailView):
"""
文章详情页视图
展示单篇文章的详细内容评论等
"""
template_name = 'blog/article_detail.html' # 详情页模板
model = Article # 关联的模型
pk_url_kwarg = 'article_id' # URL中主键的参数名
context_object_name = "article" # 模板中文章对象的变量名
def get_context_data(self, **kwargs):
"""
构建详情页上下文数据
- 评论表单
- 评论分页
- 上下篇文章
- 插件钩子处理
"""
# 初始化评论表单
comment_form = CommentForm()
# 获取当前文章的所有有效评论
article_comments = self.object.comment_list()
# 筛选顶级评论(无父评论)
parent_comments = article_comments.filter(parent_comment=None)
# 获取博客配置(评论分页大小)
blog_setting = get_blog_setting()
# 初始化评论分页器
paginator = Paginator(parent_comments, blog_setting.article_comment_count)
# 处理评论页码参数
page = self.request.GET.get('comment_page', '1')
if not page.isnumeric():
page = 1
else:
page = int(page)
page = max(1, min(page, paginator.num_pages)) # 限制页码范围
# 获取当前页的评论
p_comments = paginator.page(page)
# 生成上下页评论的URL
next_page = p_comments.next_page_number() if p_comments.has_next() else None
prev_page = p_comments.previous_page_number() if p_comments.has_previous() else None
if next_page:
kwargs['comment_next_page_url'] = f'{self.object.get_absolute_url()}?comment_page={next_page}#commentlist-container'
if prev_page:
kwargs['comment_prev_page_url'] = f'{self.object.get_absolute_url()}?comment_page={prev_page}#commentlist-container'
# 添加上下文数据
kwargs['form'] = comment_form # 评论表单
kwargs['article_comments'] = article_comments # 所有评论
kwargs['p_comments'] = p_comments # 当前页评论
kwargs['comment_count'] = len(article_comments) if article_comments else 0 # 评论总数
# 上下篇文章
kwargs['next_article'] = self.object.next_article
kwargs['prev_article'] = self.object.prev_article
# 调用父类方法获取基础上下文
context = super().get_context_data(**kwargs)
# 获取当前文章对象
article = self.object
# 执行插件动作钩子:通知插件"文章详情已获取"
hooks.run_action('after_article_body_get', article=article, request=self.request)
# 执行插件过滤钩子:允许插件修改文章正文(如添加水印、解析特殊标签等)
article.body = hooks.apply_filters(
ARTICLE_CONTENT_HOOK_NAME,
article.body,
article=article,
request=self.request
)
return context
class CategoryDetailView(ArticleListView):
"""
分类详情页视图
展示指定分类及子分类下的所有文章
"""
page_type = "分类目录归档" # 页面类型描述
def get_queryset_data(self):
"""
获取分类下的文章列表
1. 根据URL中的分类slug获取分类对象
2. 包含所有子分类的文章
3. 仅展示已发布状态
"""
slug = self.kwargs['category_name']
category = get_object_or_404(Category, slug=slug) # 获取分类不存在则404
# 记录分类名称(用于上下文)
self.categoryname = category.name
# 获取当前分类及所有子分类的名称列表
categorynames = [c.name for c in category.get_sub_categorys()]
# 筛选属于这些分类且已发布的文章
return Article.objects.filter(category__name__in=categorynames, status='p')
def get_queryset_cache_key(self):
"""生成分类页面的缓存key包含分类名和页码"""
slug = self.kwargs['category_name']
category = get_object_or_404(Category, slug=slug)
self.categoryname = category.name
return f'category_list_{self.categoryname}_{self.page_number}'
def get_context_data(self, **kwargs):
"""扩展上下文:添加页面类型和分类名称"""
# 处理分类名称(去除路径前缀,仅保留最后一级)
try:
categoryname = self.categoryname.split('/')[-1]
except:
categoryname = self.categoryname
kwargs['page_type'] = self.page_type
kwargs['tag_name'] = categoryname # 模板中统一用tag_name显示当前分类/标签/作者名
return super().get_context_data(** kwargs)
class AuthorDetailView(ArticleListView):
"""
作者详情页视图
展示指定作者发布的所有文章
"""
page_type = '作者文章归档' # 页面类型描述
def get_queryset_cache_key(self):
"""生成作者页面的缓存key包含作者名和页码"""
from uuslug import slugify # 确保作者名URL友好
author_name = slugify(self.kwargs['author_name'])
return f'author_{author_name}_{self.page_number}'
def get_queryset_data(self):
"""获取指定作者的已发布文章"""
author_name = self.kwargs['author_name']
return Article.objects.filter(author__username=author_name, type='a', status='p')
def get_context_data(self, **kwargs):
"""扩展上下文:添加页面类型和作者名"""
kwargs['page_type'] = self.page_type
kwargs['tag_name'] = self.kwargs['author_name']
return super().get_context_data(** kwargs)
class TagDetailView(ArticleListView):
"""
标签详情页视图
展示指定标签关联的所有文章
"""
page_type = '分类标签归档' # 页面类型描述
def get_queryset_data(self):
"""获取指定标签的已发布文章"""
slug = self.kwargs['tag_name']
tag = get_object_or_404(Tag, slug=slug) # 获取标签不存在则404
self.name = tag.name # 记录标签名
return Article.objects.filter(tags__name=self.name, type='a', status='p')
def get_queryset_cache_key(self):
"""生成标签页面的缓存key包含标签名和页码"""
slug = self.kwargs['tag_name']
tag = get_object_or_404(Tag, slug=slug)
self.name = tag.name
return f'tag_{self.name}_{self.page_number}'
def get_context_data(self, **kwargs):
"""扩展上下文:添加页面类型和标签名"""
kwargs['page_type'] = self.page_type
kwargs['tag_name'] = self.name
return super().get_context_data(** kwargs)
class ArchivesView(ArticleListView):
"""
文章归档页面视图
展示所有已发布文章的归档列表按时间分组
"""
page_type = '文章归档'
paginate_by = None # 归档页不分页
page_kwarg = None # 无需页码参数
template_name = 'blog/article_archives.html' # 归档页专用模板
def get_queryset_data(self):
"""获取所有已发布文章(用于归档)"""
return Article.objects.filter(status='p').all()
def get_queryset_cache_key(self):
"""归档页缓存key固定值因不分页"""
return 'archives'
class LinkListView(ListView):
"""
友情链接页面视图
展示所有启用的友情链接
"""
model = Links # 关联链接模型
template_name = 'blog/links_list.html' # 链接页模板
def get_queryset(self):
"""仅获取启用的友情链接"""
return Links.objects.filter(is_enable=True)
class EsSearchView(SearchView):
"""
搜索视图基于Haystack
处理全文搜索请求并返回结果
"""
def get_context(self):
"""构建搜索结果页面的上下文数据"""
# 构建分页器和当前页数据
paginator, page = self.build_page()
context = {
"query": self.query, # 搜索关键词
"form": self.form, # 搜索表单
"page": page, # 当前页结果
"paginator": paginator, # 分页器
"suggestion": None, # 搜索建议(默认无)
}
# 如果搜索引擎支持拼写建议,添加建议内容
if hasattr(self.results, "query") and self.results.query.backend.include_spelling:
context["suggestion"] = self.results.query.get_spelling_suggestion()
# 添加额外上下文
context.update(self.extra_context())
return context
@csrf_exempt # 豁免CSRF验证用于外部调用上传
def fileupload(request):
"""
文件上传接口图床功能
仅允许POST请求且需验证签名
"""
if request.method == 'POST':
# 获取签名参数
sign = request.GET.get('sign', None)
if not sign:
return HttpResponseForbidden() # 无签名则禁止
# 验证签名双重SHA256加密基于SECRET_KEY
if sign != get_sha256(get_sha256(settings.SECRET_KEY)):
return HttpResponseForbidden() # 签名无效则禁止
# 存储上传文件的URL
response = []
# 处理每个上传的文件
for filename in request.FILES:
# 生成时间目录(按年/月/日)
timestr = timezone.now().strftime('%Y/%m/%d')
# 图片文件扩展名
imgextensions = ['jpg', 'png', 'jpeg', 'bmp']
# 检查是否为图片
fname = str(filename)
isimage = any(ext in fname.lower() for ext in imgextensions)
# 确定存储目录(图片和普通文件分开)
base_dir = os.path.join(
settings.STATICFILES,
"image" if isimage else "files",
timestr
)
# 确保目录存在
if not os.path.exists(base_dir):
os.makedirs(base_dir)
# 生成唯一文件名UUID+原扩展名)
file_ext = os.path.splitext(filename)[-1]
savepath = os.path.normpath(
os.path.join(base_dir, f"{uuid.uuid4().hex}{file_ext}")
)
# 安全检查:防止路径穿越
if not savepath.startswith(base_dir):
return HttpResponse("Invalid path")
# 保存文件
with open(savepath, 'wb+') as wfile:
for chunk in request.FILES[filename].chunks():
wfile.write(chunk)
# 压缩图片(如果是图片文件)
if isimage:
from PIL import Image
try:
with Image.open(savepath) as image:
# 优化图片质量20%质量,启用优化)
image.save(savepath, quality=20, optimize=True)
except Exception as e:
logger.error(f"图片压缩失败: {e}")
# 生成文件的访问URL
url = static(savepath)
response.append(url)
# 返回所有上传文件的URL
return HttpResponse(response)
else:
# 仅允许POST请求
return HttpResponse("only for post")
def page_not_found_view(request, exception, template_name='blog/error_page.html'):
"""
404错误页面视图
处理页面未找到的情况
"""
if exception:
logger.error(exception) # 记录错误详情
url = request.get_full_path() # 获取请求的URL
return render(
request,
template_name,
{
'message': _('Sorry, the page you requested is not found. Please click the home page to see others.'),
'statuscode': '404'
},
status=404
)
def server_error_view(request, template_name='blog/error_page.html'):
"""
500错误页面视图
处理服务器内部错误
"""
return render(
request,
template_name,
{
'message': _('Sorry, the server is busy. Please click the home page to see others.'),
'statuscode': '500'
},
status=500
)
def permission_denied_view(request, exception, template_name='blog/error_page.html'):
"""
403错误页面视图
处理权限不足的情况
"""
if exception:
logger.error(exception) # 记录错误详情
return render(
request,
template_name,
{
'message': _('Sorry, you do not have permission to access this page.'),
'statuscode': '403'
},
status=403
)
def clean_cache_view(request):
"""
清理缓存接口
调用后清除所有缓存数据
"""
cache.clear()
return HttpResponse('ok')

Some files were not shown because too many files have changed in this diff Show More

Loading…
Cancel
Save