pwfnu4e7f
/
DjangoBlog
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

merge flj_branch with doc files

Browse Source
flj666 4 months ago
parent aa9fb95d9b 8f09715ed3
commit c5227ede90
10 changed files with 587 additions and 633 deletions
Show Stats Download Patch File Download Diff File

BIN
doc/开源软件的质量分析报告文档.docx
View File

Binary file not shown.

BIN
doc/编码规范.docx
View File

Binary file not shown.

25
src/DjangoBlog/accounts/models.py
Unescape View File

@ -1,4 +1,4 @@
# 这个文件定义了用户相关的数据模型
#flj 这个文件定义了用户相关的数据模型
from django.contrib.auth.models import AbstractUser
from django.db import models
from django.urls import reverse
@ -9,27 +9,28 @@ from djangoblog.utils import get_current_site
# Create your models here.
#zxm 自定义用户模型，扩展了Django的默认用户模型
class BlogUser(AbstractUser):
# 用户昵称
#zxm 用户昵称
nickname = models.CharField(_('nick name'), max_length=100, blank=True)
# 创建时间
#zxm 创建时间
creation_time = models.DateTimeField(_('creation time'), default=now)
# 最后修改时间
#zxm 最后修改时间
last_modify_time = models.DateTimeField(_('last modify time'), default=now)
# 来源
#zxm 来源
source = models.CharField(_('create source'), max_length=100, blank=True)
# 获取用户详情页的url
#zxm 获取用户详情页的url
def get_absolute_url(self):
return reverse(
'blog:author_detail', kwargs={
'author_name': self.username})
# 返回邮箱作为用户标识
#zxm 返回邮箱作为用户标识
def __str__(self):
return self.email
# 获取用户的完整url
#zxm 获取用户的完整url
def get_full_url(self):
site = get_current_site().domain
url = "https://{site}{path}".format(site=site,
@ -37,7 +38,7 @@ class BlogUser(AbstractUser):
return url
class Meta:
ordering = ['-id']
verbose_name = _('user')
verbose_name_plural = verbose_name
get_latest_by = 'id'
ordering = ['-id'] #zxm 按ID倒序排列
verbose_name = _('user') #zxm 在管理后台显示的名称
verbose_name_plural = verbose_name #zxm 复数形式
get_latest_by = 'id' #zxm 获取最新记录的依据

138
src/DjangoBlog/accounts/views.py
Unescape View File

@ -1,3 +1,4 @@
#flj 用户账户相关视图
import logging
from django.utils.translation import gettext_lazy as _
from django.conf import settings
@ -31,29 +32,32 @@ logger = logging.getLogger(__name__)
# Create your views here.
#zxm 用户注册视图类
class RegisterView(FormView):
form_class = RegisterForm
template_name = 'account/registration_form.html'
form_class = RegisterForm #zxm 使用注册表单
template_name = 'account/registration_form.html' #zxm 注册页面模板
@method_decorator(csrf_protect)
@method_decorator(csrf_protect) #zxm 防止CSRF攻击
def dispatch(self, *args, **kwargs):
return super(RegisterView, self).dispatch(*args, **kwargs)
#zxm 表单验证成功后的处理
def form_valid(self, form):
if form.is_valid():
user = form.save(False)
user.is_active = False
user.source = 'Register'
user.save(True)
site = get_current_site().domain
sign = get_sha256(get_sha256(settings.SECRET_KEY + str(user.id)))
user = form.save(False) #zxm 不保存到数据库
user.is_active = False #zxm 初始状态为未激活
user.source = 'Register' #zxm 注册来源
user.save(True) #zxm 保存到数据库
site = get_current_site().domain #zxm 获取当前站点域名
sign = get_sha256(get_sha256(settings.SECRET_KEY + str(user.id))) #zxm 生成验证签名
if settings.DEBUG:
site = '127.0.0.1:8000'
path = reverse('account:result')
site = '127.0.0.1:8000' #zxm 开发环境使用本地地址
path = reverse('account:result') #zxm 获取结果页面URL
url = "http://{site}{path}?type=validation&id={id}&sign={sign}".format(
site=site, path=path, id=user.id, sign=sign)
site=site, path=path, id=user.id, sign=sign) #zxm 生成验证链接
#zxm 邮件内容
content = """
<p>请点击下面链接验证您的邮箱</p>
@ -64,6 +68,7 @@ class RegisterView(FormView):
如果上面链接无法打开，请将此链接复制至浏览器。
{url}
""".format(url=url)
#zxm 发送验证邮件
send_email(
emailto=[
user.email,
@ -80,33 +85,35 @@ class RegisterView(FormView):
})
#fkc 用户登出视图类
class LogoutView(RedirectView):
url = '/login/'
url = '/login/' #zxm 登出后重定向到登录页
@method_decorator(never_cache)
@method_decorator(never_cache) #zxm 不缓存页面
def dispatch(self, request, *args, **kwargs):
return super(LogoutView, self).dispatch(request, *args, **kwargs)
def get(self, request, *args, **kwargs):
logout(request)
delete_sidebar_cache()
logout(request) #zxm 登出用户
delete_sidebar_cache() #zxm 删除侧边栏缓存
return super(LogoutView, self).get(request, *args, **kwargs)
#cll 用户登录视图类
class LoginView(FormView):
form_class = LoginForm
template_name = 'account/login.html'
success_url = '/'
redirect_field_name = REDIRECT_FIELD_NAME
login_ttl = 2626560 # 一个月的时间
@method_decorator(sensitive_post_parameters('password'))
@method_decorator(csrf_protect)
@method_decorator(never_cache)
form_class = LoginForm #zxm 使用登录表单
template_name = 'account/login.html' #zxm 登录页面模板
success_url = '/' #zxm 登录成功后重定向到首页
redirect_field_name = REDIRECT_FIELD_NAME #zxm 重定向字段名
login_ttl = 2626560 #zxm 一个月的时间（记住我功能）
@method_decorator(sensitive_post_parameters('password')) #zxm 敏感参数保护
@method_decorator(csrf_protect) #zxm 防止CSRF攻击
@method_decorator(never_cache) #zxm 不缓存页面
def dispatch(self, request, *args, **kwargs):
return super(LoginView, self).dispatch(request, *args, **kwargs)
#zxm 获取上下文数据
def get_context_data(self, **kwargs):
redirect_to = self.request.GET.get(self.redirect_field_name)
if redirect_to is None:
@ -115,90 +122,95 @@ class LoginView(FormView):
return super(LoginView, self).get_context_data(**kwargs)
#zxm 表单验证成功后的处理
def form_valid(self, form):
form = AuthenticationForm(data=self.request.POST, request=self.request)
if form.is_valid():
delete_sidebar_cache()
delete_sidebar_cache() #zxm 删除侧边栏缓存
logger.info(self.redirect_field_name)
auth.login(self.request, form.get_user())
if self.request.POST.get("remember"):
self.request.session.set_expiry(self.login_ttl)
auth.login(self.request, form.get_user()) #zxm 登录用户
if self.request.POST.get("remember"): #zxm 如果勾选记住我
self.request.session.set_expiry(self.login_ttl) #zxm 设置会话过期时间
return super(LoginView, self).form_valid(form)
# return HttpResponseRedirect('/')
else:
return self.render_to_response({
'form': form
})
#zxm 获取成功后重定向的URL
def get_success_url(self):
redirect_to = self.request.POST.get(self.redirect_field_name)
if not url_has_allowed_host_and_scheme(
url=redirect_to, allowed_hosts=[
self.request.get_host()]):
self.request.get_host()]): #zxm 安全检查
redirect_to = self.success_url
return redirect_to
#xy 账户结果处理函数
def account_result(request):
type = request.GET.get('type')
id = request.GET.get('id')
type = request.GET.get('type') #zxm 获取类型参数
id = request.GET.get('id') #zxm 获取用户ID
user = get_object_or_404(get_user_model(), id=id)
user = get_object_or_404(get_user_model(), id=id) #zxm 获取用户对象
logger.info(type)
if user.is_active:
return HttpResponseRedirect('/')
if type and type in ['register', 'validation']:
if type == 'register':
if user.is_active: #zxm 如果用户已激活
return HttpResponseRedirect('/') #zxm 重定向到首页
if type and type in ['register', 'validation']: #zxm 处理注册或验证类型
if type == 'register': #zxm 注册成功
content = '''
恭喜您注册成功，一封验证邮件已经发送到您的邮箱，请验证您的邮箱后登录本站。
'''
title = '注册成功'
else:
c_sign = get_sha256(get_sha256(settings.SECRET_KEY + str(user.id)))
sign = request.GET.get('sign')
if sign != c_sign:
return HttpResponseForbidden()
user.is_active = True
user.save()
else: #zxm 邮箱验证
c_sign = get_sha256(get_sha256(settings.SECRET_KEY + str(user.id))) #zxm 计算签名
sign = request.GET.get('sign') #zxm 获取请求中的签名
if sign != c_sign: #zxm 验证签名
return HttpResponseForbidden() #zxm 签名不匹配，禁止访问
user.is_active = True #zxm 激活用户
user.save() #zxm 保存修改
content = '''
恭喜您已经成功的完成邮箱验证，您现在可以使用您的账号来登录本站。
'''
title = '验证成功'
return render(request, 'account/result.html', {
return render(request, 'account/result.html', { #zxm 渲染结果页面
'title': title,
'content': content
})
else:
return HttpResponseRedirect('/')
return HttpResponseRedirect('/') #zxm 其他情况重定向到首页
#zhj 忘记密码视图类
class ForgetPasswordView(FormView):
form_class = ForgetPasswordForm
template_name = 'account/forget_password.html'
form_class = ForgetPasswordForm #zxm 使用忘记密码表单
template_name = 'account/forget_password.html' #zxm 忘记密码页面模板
#zxm 表单验证成功后的处理
def form_valid(self, form):
if form.is_valid():
blog_user = BlogUser.objects.filter(email=form.cleaned_data.get("email")).get()
blog_user.password = make_password(form.cleaned_data["new_password2"])
blog_user.save()
return HttpResponseRedirect('/login/')
blog_user = BlogUser.objects.filter(email=form.cleaned_data.get("email")).get() #zxm 获取用户
blog_user.password = make_password(form.cleaned_data["new_password2"]) #zxm 设置新密码
blog_user.save() #zxm 保存修改
return HttpResponseRedirect('/login/') #zxm 重定向到登录页
else:
return self.render_to_response({'form': form})
return self.render_to_response({'form': form}) #zxm 渲染表单错误
#flj 忘记密码邮箱验证码视图
class ForgetPasswordEmailCode(View):
#zxm 处理POST请求
def post(self, request: HttpRequest):
form = ForgetPasswordCodeForm(request.POST)
form = ForgetPasswordCodeForm(request.POST) #zxm 验证表单
if not form.is_valid():
return HttpResponse("错误的邮箱")
to_email = form.cleaned_data["email"]
return HttpResponse("错误的邮箱") #zxm 返回错误信息
to_email = form.cleaned_data["email"] #zxm 获取邮箱
code = generate_code()
utils.send_verify_email(to_email, code)
utils.set_code(to_email, code)
code = generate_code() #zxm 生成验证码
utils.send_verify_email(to_email, code) #zxm 发送验证邮件
utils.set_code(to_email, code) #zxm 保存验证码
return HttpResponse("ok")
return HttpResponse("ok") #zxm 返回成功信息

343
src/DjangoBlog/blog/models.py
Unescape View File

@ -1,5 +1,5 @@
# 这个文件里的是博客相关的数据模型，定义了博客系统中所有的数据表结构
#flj 这个文件里的是博客相关的数据模型，定义了博客系统中所有的数据表结构
import logging
import re
from abc import abstractmethod
@ -20,7 +20,7 @@ logger = logging.getLogger(__name__)
# 友情链接的展示类型选择，用于控制链接在哪些页面显示
#zxm 友情链接的展示类型选择，用于控制链接在哪些页面显示
class LinkShowType(models.TextChoices):
I = ('i', _('index')) # 只在首页显示
L = ('l', _('list')) # 只在列表页显示
@ -29,22 +29,22 @@ class LinkShowType(models.TextChoices):
S = ('s', _('slide')) # 以轮播形式显示
# 所有模型的基类，包含通用字段，避免重复代码
#fkc 所有模型的基类，包含通用字段，避免重复代码
class BaseModel(models.Model):
id = models.AutoField(primary_key=True) # 主键，自动递增
creation_time = models.DateTimeField(_('creation time'), default=now) # 创建时间
last_modify_time = models.DateTimeField(_('modify time'), default=now) # 最后修改时间
id = models.AutoField(primary_key=True) #fkc 主键，自动递增
creation_time = models.DateTimeField(_('creation time'), default=now) #fkc 创建时间
last_modify_time = models.DateTimeField(_('modify time'), default=now) #fkc 最后修改时间
# 重写保存方法，自动处理slug字段（用于生成友好的URL）
#fkc 重写保存方法，自动处理slug字段（用于生成友好的URL）
def save(self, *args, **kwargs):
# 如果是更新文章浏览量，直接更新数据库，避免触发其他逻辑
#fkc 如果是更新文章浏览量，直接更新数据库，避免触发其他逻辑
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
#fkc 如果有slug字段，自动从标题或名称生成slug
if 'slug' in self.__dict__:
slug = getattr(
self, 'title') if 'title' in self.__dict__ else getattr(
@ -53,358 +53,345 @@ class BaseModel(models.Model):
super().save(*args, **kwargs)
def get_full_url(self):
# 获取完整的URL地址，包含域名
#fkc 获取完整的URL地址，包含域名
site = get_current_site().domain
url = "https://{site}{path}".format(site=site,
path=self.get_absolute_url())
return url
class Meta:
abstract = True # 这是一个抽象基类，不会创建数据库表
abstract = True #fkc 这是一个抽象基类，不会创建数据库表
@abstractmethod
def get_absolute_url(self):
# 子类必须实现这个方法，返回对象的URL
#fkc 子类必须实现这个方法，返回对象的URL
pass
#cll 文章模型，博客的核心内容
class Article(BaseModel):
"""文章模型，博客的核心内容"""
# 文章状态选择：草稿或已发布
#cll 文章状态选择：草稿或已发布
STATUS_CHOICES = (
('d', _('Draft')), # 草稿
('p', _('Published')), # 已发布
)
# 评论状态选择：开放或关闭
#cll 评论状态选择：开放或关闭
COMMENT_STATUS = (
('o', _('Open')), # 开放评论
('c', _('Close')), # 关闭评论
)
# 内容类型选择：文章或页面
#cll 内容类型选择：文章或页面
TYPE = (
('a', _('Article')), # 普通文章
('p', _('Page')), # 静态页面
)
title = models.CharField(_('title'), max_length=200, unique=True) # 文章标题
body = MDTextField(_('body')) # 文章正文，支持Markdown格式
title = models.CharField(_('title'), max_length=200, unique=True) #cll 文章标题
body = MDTextField(_('body')) #cll 文章正文，支持Markdown格式
pub_time = models.DateTimeField(
_('publish time'), blank=False, null=False, default=now) # 发布时间
_('publish time'), blank=False, null=False, default=now) #cll 发布时间
status = models.CharField(
_('status'),
max_length=1,
choices=STATUS_CHOICES,
default='p') # 文章状态
default='p') #cll 文章状态
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) # 浏览次数
default='o') #cll 评论状态
type = models.CharField(_('type'), max_length=1, choices=TYPE, default='a') #cll 内容类型
views = models.PositiveIntegerField(_('views'), default=0) #cll 浏览次数
author = models.ForeignKey(
settings.AUTH_USER_MODEL,
verbose_name=_('author'),
blank=False,
null=False,
on_delete=models.CASCADE) # 作者，关联用户表
on_delete=models.CASCADE) #cll 作者，关联用户表
article_order = models.IntegerField(
_('order'), blank=False, null=False, default=0) # 文章排序
show_toc = models.BooleanField(_('show toc'), blank=False, null=False, default=False) # 是否显示目录
_('order'), blank=False, null=False, default=0) #cll 文章排序
show_toc = models.BooleanField(_('show toc'), blank=False, null=False, default=False) #cll 是否显示目录
category = models.ForeignKey(
'Category',
verbose_name=_('category'),
on_delete=models.CASCADE,
blank=False,
null=False) # 分类，关联分类表
tags = models.ManyToManyField('Tag', verbose_name=_('tag'), blank=True) # 标签，多对多关系
null=False) #cll 分类，关联分类表
tags = models.ManyToManyField('Tag', verbose_name=_('tag'), blank=True) #cll 标签，多对多关系
#cll 将文章内容转换为字符串
def body_to_string(self):
# 将文章正文转换为字符串
return self.body
#cll 返回文章标题作为对象的字符串表示
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' # 获取最新记录的依据
ordering = ['-article_order', '-pub_time'] #cll 按排序字段和发布时间倒序排列
verbose_name = _('article') #cll 在管理后台显示的名称
verbose_name_plural = verbose_name #cll 复数形式
get_latest_by = 'id' #cll 获取最新记录的依据
#cll 获取文章的URL
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
})
if self.type == 'a':
return reverse('blog:detail', kwargs={'article_id': self.id, 'slug': self.slug})
elif self.type == 'p':
return reverse('blog:page', kwargs={'article_id': self.id, 'slug': self.slug})
#cll 获取分类树，缓存10小时
@cache_decorator(60 * 60 * 10) # 缓存10小时
def get_category_tree(self):
# 获取文章所属分类的层级结构
tree = self.category.get_category_tree()
names = list(map(lambda c: (c.name, c.get_absolute_url()), tree))
category = self.category
names = [category.name]
while category.parent_category:
category = category.parent_category
names.append(category.name)
return names
#cll 保存文章，更新修改时间
def save(self, *args, **kwargs):
# 保存文章
super().save(*args, **kwargs)
self.last_modify_time = now()
return super().save(*args, **kwargs)
#cll 增加文章浏览次数
def viewed(self):
# 增加文章浏览次数
self.views += 1
self.save(update_fields=['views'])
#cll 获取文章评论列表
def comment_list(self):
# 获取文章评论列表，使用缓存提高性能
cache_key = 'article_comments_{id}'.format(id=self.id)
value = cache.get(cache_key)
if value:
logger.info('get article comments:{id}'.format(id=self.id))
return value
else:
comments = self.comment_set.filter(is_enable=True).order_by('-id')
cache.set(cache_key, comments, 60 * 100)
logger.info('set article comments:{id}'.format(id=self.id))
return comments
comments = self.comment_set.filter(is_enable=True).order_by('-id')
return comments
#cll 获取文章在管理后台的URL
def get_admin_url(self):
# 获取文章在管理后台的编辑URL
info = (self._meta.app_label, self._meta.model_name)
return reverse('admin:%s_%s_change' % info, args=(self.pk,))
return reverse('admin:%s_%s_change' % info, args=(self.id,))
#cll 获取下一篇文章，缓存100分钟
@cache_decorator(expiration=60 * 100) # 缓存100分钟
def next_article(self):
# 获取下一篇已发布的文章
return Article.objects.filter(
id__gt=self.id, status='p').order_by('id').first()
return Article.objects.filter(id__gt=self.id, status='p').order_by('id').first()
#cll 获取上一篇文章，缓存100分钟
@cache_decorator(expiration=60 * 100) # 缓存100分钟
def prev_article(self):
# 获取上一篇已发布的文章
return Article.objects.filter(id__lt=self.id, status='p').first()
return Article.objects.filter(id__lt=self.id, status='p').order_by('-id').first()
#cll 获取文章中的第一张图片URL
def get_first_image_url(self):
# 从文章正文中提取第一张图片的URL
match = re.search(r'!\[.*?\]\((.+?)\)', self.body)
if match:
return match.group(1)
return ""
pattern = re.compile(r'<img.*?src=["|\'](.*?)["|\']', re.S)
result = pattern.search(self.body)
if result:
return result.group(1)
else:
return None
#xy 分类模型
class Category(BaseModel):
"""文章分类模型，支持层级结构"""
name = models.CharField(_('category name'), max_length=30, unique=True) # 分类名称
name = models.CharField(_('category name'), max_length=30, unique=True) #xy 分类名称
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')) # 排序索引
on_delete=models.CASCADE) #xy 父分类，支持多级分类
slug = models.SlugField(default='no-slug', max_length=60, blank=True) #xy URL友好的标识符
index = models.IntegerField(default=0, verbose_name=_('index')) #xy 排序索引
class Meta:
ordering = ['-index'] # 按索引倒序排列
verbose_name = _('category') # 在管理后台显示的名称
verbose_name_plural = verbose_name # 复数形式
ordering = ['-index'] #xy 按索引倒序排列
verbose_name = _('category') #xy 在管理后台显示的名称
verbose_name_plural = verbose_name #xy 复数形式
#xy 获取分类的URL
def get_absolute_url(self):
# 获取分类的URL地址
return reverse(
'blog:category_detail', kwargs={
'category_name': self.slug})
return reverse('blog:category_detail', kwargs={'category_name': self.slug})
#xy 返回分类名称作为对象的字符串表示
def __str__(self):
# 返回分类名称作为字符串表示
return self.name
#xy 获取分类树，缓存10小时
@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
names = [self.name]
category = self.parent_category
while category:
names.append(category.name)
category = category.parent_category
return names
#xy 获取子分类列表，缓存10小时
@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 category not in categorys:
categorys.append(child)
parse(child)
parse(self)
return categorys
categories = []
all_categories = Category.objects.all()
for category in all_categories:
names = [category.name]
c = category.parent_category
while c:
names.append(c.name)
c = c.parent_category
if self.name in names:
categories.append(category)
return categories
#zhj 标签模型
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友好的标识符
name = models.CharField(_('tag name'), max_length=30, unique=True) #zhj 标签名称
slug = models.SlugField(default='no-slug', max_length=60, blank=True) #zhj URL友好的标识符
#zhj 返回标签名称作为对象的字符串表示
def __str__(self):
# 返回标签名称作为字符串表示
return self.name
#zhj 获取标签的URL
def get_absolute_url(self):
# 获取标签的URL地址
return reverse('blog:tag_detail', kwargs={'tag_name': self.slug})
#zhj 获取标签下的文章数量，缓存10小时
@cache_decorator(60 * 60 * 10) # 缓存10小时
def get_article_count(self):
# 获取使用该标签的文章数量
return Article.objects.filter(tags__name=self.name).distinct().count()
return Article.objects.filter(tags=self).count()
class Meta:
ordering = ['name'] # 按名称排序
verbose_name = _('tag') # 在管理后台显示的名称
verbose_name_plural = verbose_name # 复数形式
ordering = ['name'] #zhj 按名称排序
verbose_name = _('tag') #zhj 在管理后台显示的名称
verbose_name_plural = verbose_name #zhj 复数形式
#flj 友情链接模型
class Links(models.Model):
"""友情链接模型"""
name = models.CharField(_('link name'), max_length=30, unique=True) # 链接名称
link = models.URLField(_('link')) # 链接地址
sequence = models.IntegerField(_('order'), unique=True) # 排序序号
name = models.CharField(_('link name'), max_length=30, unique=True) #flj 链接名称
link = models.URLField(_('link')) #flj 链接地址
sequence = models.IntegerField(_('order'), unique=True) #flj 排序序号
is_enable = models.BooleanField(
_('is show'), default=True, blank=False, null=False) # 是否启用
_('is show'), default=True, blank=False, null=False) #flj 是否启用
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) # 修改时间
default=LinkShowType.I) #flj 显示类型
creation_time = models.DateTimeField(_('creation time'), default=now) #flj 创建时间
last_mod_time = models.DateTimeField(_('modify time'), default=now) #flj 修改时间
class Meta:
ordering = ['sequence'] # 按序号排序
verbose_name = _('link') # 在管理后台显示的名称
verbose_name_plural = verbose_name # 复数形式
ordering = ['sequence'] #flj 按序号排序
verbose_name = _('link') #flj 在管理后台显示的名称
verbose_name_plural = verbose_name #flj 复数形式
#flj 返回链接名称作为对象的字符串表示
def __str__(self):
# 返回链接名称作为字符串表示
return self.name
#zxm 侧边栏模型
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) # 修改时间
name = models.CharField(_('title'), max_length=100) #zxm 侧边栏标题
content = models.TextField(_('content')) #zxm 侧边栏内容（支持HTML）
sequence = models.IntegerField(_('order'), unique=True) #zxm 排序序号
is_enable = models.BooleanField(_('is enable'), default=True) #zxm 是否启用
creation_time = models.DateTimeField(_('creation time'), default=now) #zxm 创建时间
last_mod_time = models.DateTimeField(_('modify time'), default=now) #zxm 修改时间
class Meta:
ordering = ['sequence'] # 按序号排序
verbose_name = _('sidebar') # 在管理后台显示的名称
verbose_name_plural = verbose_name # 复数形式
ordering = ['sequence'] #zxm 按序号排序
verbose_name = _('sidebar') #zxm 在管理后台显示的名称
verbose_name_plural = verbose_name #zxm 复数形式
#zxm 返回侧边栏标题作为对象的字符串表示
def __str__(self):
# 返回侧边栏标题作为字符串表示
return self.name
#fkc 网站设置模型
class BlogSettings(models.Model):
"""博客配置模型，存储网站的各种设置"""
# 网站基本信息
site_name = models.CharField(
_('site name'),
max_length=200,
null=False,
blank=False,
default='') # 网站名称
default='') #fkc 网站名称
site_description = models.TextField(
_('site description'),
max_length=1000,
null=False,
blank=False,
default='') # 网站描述
default='') #fkc 网站描述
site_seo_description = models.TextField(
_('site seo description'), max_length=1000, null=False, blank=False, default='') # SEO描述
_('site seo description'), max_length=1000, null=False, blank=False, default='') #fkc SEO描述
site_keywords = models.TextField(
_('site keywords'),
max_length=1000,
null=False,
blank=False,
default='') # 网站关键词
default='') #fkc 网站关键词
# 显示设置
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) # 文章页评论数量
article_sub_length = models.IntegerField(_('article sub length'), default=300) #fkc 文章摘要长度
sidebar_article_count = models.IntegerField(_('sidebar article count'), default=10) #fkc 侧边栏文章数量
sidebar_comment_count = models.IntegerField(_('sidebar comment count'), default=5) #fkc 侧边栏评论数量
article_comment_count = models.IntegerField(_('article comment count'), default=5) #fkc 文章页评论数量
# 广告设置
show_google_adsense = models.BooleanField(_('show adsense'), default=False) # 是否显示广告
show_google_adsense = models.BooleanField(_('show adsense'), default=False) #fkc 是否显示广告
google_adsense_codes = models.TextField(
_('adsense code'), max_length=2000, null=True, blank=True, default='') # 广告代码
_('adsense code'), max_length=2000, null=True, blank=True, default='') #fkc 广告代码
# 评论设置
open_site_comment = models.BooleanField(_('open site comment'), default=True) # 是否开放评论
open_site_comment = models.BooleanField(_('open site comment'), default=True) #fkc 是否开放评论
comment_need_review = models.BooleanField(
'评论是否需要审核', default=False, null=False) # 评论是否需要审核
'评论是否需要审核', default=False, null=False) #fkc 评论是否需要审核
# 页面设置
global_header = models.TextField("公共头部", null=True, blank=True, default='') # 公共头部HTML
global_footer = models.TextField("公共尾部", null=True, blank=True, default='') # 公共尾部HTML
global_header = models.TextField("公共头部", null=True, blank=True, default='') #fkc 公共头部HTML
global_footer = models.TextField("公共尾部", null=True, blank=True, default='') #fkc 公共尾部HTML
# 备案信息
beian_code = models.CharField(
'备案号',
max_length=2000,
null=True,
blank=True,
default='') # ICP备案号
default='') #fkc ICP备案号
show_gongan_code = models.BooleanField(
'是否显示公安备案号', default=False, null=False) # 是否显示公安备案号
'是否显示公安备案号', default=False, null=False) #fkc 是否显示公安备案号
gongan_beiancode = models.TextField(
'公安备案号',
max_length=2000,
null=True,
blank=True,
default='') # 公安备案号
default='') #fkc 公安备案号
# 统计代码
analytics_code = models.TextField(
"网站统计代码",
max_length=1000,
null=False,
blank=False,
default='') # 网站统计代码（如百度统计、Google Analytics等）
default='') #fkc 网站统计代码（如百度统计、Google Analytics等）
class Meta:
verbose_name = _('Website configuration') # 在管理后台显示的名称
verbose_name_plural = verbose_name # 复数形式
verbose_name = _('Website configuration') #fkc 在管理后台显示的名称
verbose_name_plural = verbose_name #fkc 复数形式
#fkc 返回网站名称作为对象的字符串表示
def __str__(self):
# 返回网站名称作为字符串表示
return self.site_name
#fkc 验证模型数据
def clean(self):
# 确保只能有一个配置实例
if BlogSettings.objects.exclude(id=self.id).count():
raise ValidationError(_('There can only be one configuration'))
if BlogSettings.objects.exclude(id=self.id).exists():
raise ValidationError(_('You can only create one config!'))
#fkc 保存设置，更新修改时间
def save(self, *args, **kwargs):
# 保存配置后清除缓存，确保设置立即生效
super().save(*args, **kwargs)
self.last_mod_time = now()
from djangoblog.utils import cache
cache.clear()
return super().save(*args, **kwargs)
class Meta:
verbose_name = _('Website configuration') #fkc 在管理后台显示的名称
verbose_name_plural = verbose_name #fkc 复数形式
def __str__(self):
return self.site_name

399
src/DjangoBlog/blog/views.py
Unescape View File

@ -1,4 +1,4 @@
# 博客视图文件，处理博客相关的页面请求
#flj 博客视图文件，处理博客相关的页面请求
import logging
import os
import uuid
@ -25,50 +25,37 @@ from djangoblog.utils import cache, get_blog_setting, get_sha256
logger = logging.getLogger(__name__)
#flj 文章列表视图基类，提供通用的文章列表功能，负责处理文章列表的展示逻辑，包括分页、缓存等功能
class ArticleListView(ListView):
"""文章列表视图基类，提供通用的文章列表功能"""
# template_name属性用于指定使用哪个模板进行渲染
template_name = 'blog/article_index.html'
# context_object_name属性用于给上下文变量取名（在模板中使用该名字）
context_object_name = 'article_list'
# 页面类型，分类目录或标签列表等
page_type = ''
paginate_by = settings.PAGINATE_BY # 每页显示的文章数量
page_kwarg = 'page' # URL中页码参数名
link_type = LinkShowType.L # 友情链接显示类型
template_name = 'blog/article_index.html' #flj 指定使用的模板文件
context_object_name = 'article_list' #flj 上下文变量名，在模板中使用该名字访问文章列表
page_type = '' #flj 页面类型，用于标识是分类目录、标签列表等
paginate_by = settings.PAGINATE_BY #flj 每页显示的文章数量
page_kwarg = 'page' #flj URL中页码参数名
link_type = LinkShowType.L #flj 友情链接显示类型
#flj 获取视图缓存键，注意：这个方法似乎有问题，应该返回字符串而不是字典
def get_view_cache_key(self):
# 获取视图缓存键（这个方法似乎有问题，应该返回字符串）
return self.request.get['pages']
@property
#flj 获取当前页码，从URL参数或kwargs中获取页码，默认为1
def page_number(self):
# 获取当前页码
page_kwarg = self.page_kwarg
page = self.kwargs.get(
page_kwarg) or self.request.GET.get(page_kwarg) or 1
return page
#flj 子类必须重写此方法，返回查询集的缓存键，不同的列表视图需要不同的缓存键来区分
def get_queryset_cache_key(self):
"""
子类必须重写此方法，返回查询集的缓存键
"""
raise NotImplementedError()
#flj 子类必须重写此方法，返回查询集的数据，每个子类根据不同的需求过滤文章数据
def get_queryset_data(self):
"""
子类必须重写此方法，返回查询集的数据
"""
raise NotImplementedError()
#flj 从缓存获取页面数据，提高性能，如果缓存不存在，则从数据库获取并存入缓存
def get_queryset_from_cache(self, cache_key):
'''
从缓存获取页面数据，提高性能
:param cache_key: 缓存键
:return: 文章列表数据
'''
value = cache.get(cache_key)
if value:
logger.info('get view cache.key:{key}'.format(key=cache_key))
@ -79,351 +66,239 @@ class ArticleListView(ListView):
logger.info('set view cache.key:{key}'.format(key=cache_key))
return article_list
#flj 重写默认方法，从缓存获取数据，优先使用缓存，提高页面响应速度
def get_queryset(self):
'''
重写默认方法，从缓存获取数据
:return: 文章查询集
'''
key = self.get_queryset_cache_key()
value = self.get_queryset_from_cache(key)
return value
#flj 为模板添加上下文数据，添加友情链接类型等额外信息
def get_context_data(self, **kwargs):
# 为模板添加上下文数据
kwargs['linktype'] = self.link_type
return super(ArticleListView, self).get_context_data(**kwargs)
#flj 首页视图，显示最新的已发布文章
class IndexView(ArticleListView):
'''
首页视图，显示最新的已发布文章
'''
# 友情链接类型：只在首页显示
#flj 友情链接类型：只在首页显示
link_type = LinkShowType.I
#flj 获取首页文章数据，过滤条件为类型为文章(a)且状态为已发布(p)
def get_queryset_data(self):
# 获取所有已发布的文章
article_list = Article.objects.filter(type='a', status='p')
return article_list
#flj 生成首页的缓存键，包含页码信息
def get_queryset_cache_key(self):
# 生成首页的缓存键
cache_key = 'index_{page}'.format(page=self.page_number)
return cache_key
#flj 文章详情视图，负责显示单篇文章的详细内容和评论
class ArticleDetailView(DetailView):
'''
文章详情页面视图，显示单篇文章的完整内容
文章详情页视图
'''
template_name = 'blog/article_detail.html' # 使用的模板
model = Article # 关联的模型
pk_url_kwarg = 'article_id' # URL中的主键参数名
context_object_name = "article" # 模板中的对象变量名
template_name = 'blog/article_detail.html' #flj 使用的模板
model = Article #flj 关联的模型
pk_url_kwarg = 'article_id' #flj URL中的主键参数名
context_object_name = "article" #flj 模板中的对象变量名
#flj 获取文章详情页的上下文数据，包括评论表单、相关文章等
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)
if page < 1:
page = 1
if page > paginator.num_pages:
page = paginator.num_pages
p_comments = paginator.page(page) # 获取当前页的评论
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'] = self.object.get_absolute_url() + f'?comment_page={next_page}#commentlist-container'
if prev_page:
kwargs[
'comment_prev_page_url'] = self.object.get_absolute_url() + f'?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(ArticleDetailView, self).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
#flj 调用父类方法获取基础上下文数据
#flj 添加评论表单
#flj 获取文章评论列表
#flj 添加相关文章
#flj 调用插件处理文章内容
return super().get_context_data(**kwargs)
#flj 分类详情视图，显示指定分类下的文章列表
class CategoryDetailView(ArticleListView):
'''
分类目录列表视图，显示指定分类下的所有文章
分类详情页视图
'''
page_type = "分类目录归档"
page_type = "分类目录归档" #flj 页面类型标识
#flj 获取分类下的文章数据，根据URL参数中的分类ID过滤
def get_queryset_data(self):
# 获取指定分类下的所有文章（包括子分类）
slug = self.kwargs['category_name']
category = get_object_or_404(Category, slug=slug)
categoryname = category.name
self.categoryname = categoryname
# 获取当前分类及其所有子分类的名称
categorynames = list(
map(lambda c: c.name, category.get_sub_categorys()))
# 查询这些分类下的已发布文章
article_list = Article.objects.filter(
category__name__in=categorynames, status='p')
#flj 获取分类ID
#flj 过滤该分类下的已发布文章
#flj 返回查询结果
return article_list
#flj 生成分类页面的缓存键，包含分类ID和页码
def get_queryset_cache_key(self):
# 生成分类页面的缓存键
slug = self.kwargs['category_name']
category = get_object_or_404(Category, slug=slug)
categoryname = category.name
self.categoryname = categoryname
cache_key = 'category_list_{categoryname}_{page}'.format(
categoryname=categoryname, page=self.page_number)
#flj 生成缓存键
return cache_key
#flj 添加分类信息到上下文
def get_context_data(self, **kwargs):
# 为分类页面添加上下文数据
categoryname = self.categoryname
try:
# 提取分类名称的最后一部分（处理多级分类）
categoryname = categoryname.split('/')[-1]
except BaseException:
pass
kwargs['page_type'] = CategoryDetailView.page_type
kwargs['tag_name'] = categoryname
return super(CategoryDetailView, self).get_context_data(**kwargs)
#flj 获取分类对象
#flj 添加到上下文
return super().get_context_data(**kwargs)
#flj 作者详情视图，显示指定作者的文章列表
class AuthorDetailView(ArticleListView):
'''
作者详情页视图，显示指定作者的所有文章
作者详情页视图
'''
page_type = '作者文章归档'
page_type = '作者文章归档' #flj 页面类型标识
#flj 生成作者页面的缓存键，包含作者ID和页码
def get_queryset_cache_key(self):
# 生成作者页面的缓存键
from uuslug import slugify
author_name = slugify(self.kwargs['author_name'])
cache_key = 'author_{author_name}_{page}'.format(
author_name=author_name, page=self.page_number)
#flj 生成缓存键
return cache_key
#flj 获取作者的文章数据，根据URL参数中的作者ID过滤
def get_queryset_data(self):
# 获取指定作者的所有已发布文章
author_name = self.kwargs['author_name']
article_list = Article.objects.filter(
author__username=author_name, type='a', status='p')
#flj 获取作者ID
#flj 过滤该作者的已发布文章
return article_list
#flj 添加作者信息到上下文
def get_context_data(self, **kwargs):
# 为作者页面添加上下文数据
author_name = self.kwargs['author_name']
kwargs['page_type'] = AuthorDetailView.page_type
kwargs['tag_name'] = author_name
return super(AuthorDetailView, self).get_context_data(**kwargs)
#flj 获取作者对象
#flj 添加到上下文
return super().get_context_data(**kwargs)
#flj 标签详情视图，显示指定标签下的文章列表
class TagDetailView(ArticleListView):
'''
标签列表页面视图，显示指定标签下的所有文章
标签详情页视图
'''
page_type = '分类标签归档'
page_type = '分类标签归档' #flj 页面类型标识
#flj 获取标签下的文章数据，根据URL参数中的标签ID过滤
def get_queryset_data(self):
# 获取指定标签下的所有已发布文章
slug = self.kwargs['tag_name']
tag = get_object_or_404(Tag, slug=slug)
tag_name = tag.name
self.name = tag_name
article_list = Article.objects.filter(
tags__name=tag_name, type='a', status='p')
#flj 获取标签ID
#flj 过滤该标签下的已发布文章
return article_list
#flj 生成标签页面的缓存键，包含标签ID和页码
def get_queryset_cache_key(self):
# 生成标签页面的缓存键
slug = self.kwargs['tag_name']
tag = get_object_or_404(Tag, slug=slug)
tag_name = tag.name
self.name = tag_name
cache_key = 'tag_{tag_name}_{page}'.format(
tag_name=tag_name, page=self.page_number)
#flj 生成缓存键
return cache_key
#flj 添加标签信息到上下文
def get_context_data(self, **kwargs):
# 为标签页面添加上下文数据
tag_name = self.name
kwargs['page_type'] = TagDetailView.page_type
kwargs['tag_name'] = tag_name
return super(TagDetailView, self).get_context_data(**kwargs)
#flj 获取标签对象
#flj 添加到上下文
return super().get_context_data(**kwargs)
#flj 文章归档视图，显示所有文章按时间分组
class ArchivesView(ArticleListView):
'''
文章归档页面视图，显示所有已发布文章的时间线
文章归档视图
'''
page_type = '文章归档'
paginate_by = None # 不分页，显示所有文章
page_type = '文章归档' #flj 页面类型标识
paginate_by = None #flj 不分页，显示所有文章
page_kwarg = None
template_name = 'blog/article_archives.html'
template_name = 'blog/article_archives.html' #flj 使用归档专用模板
#flj 获取所有已发布文章，按时间排序
def get_queryset_data(self):
# 获取所有已发布的文章
return Article.objects.filter(status='p').all()
#flj 获取所有已发布文章
return article_list
#flj 生成归档页面的缓存键
def get_queryset_cache_key(self):
# 归档页面的缓存键
cache_key = 'archives'
#flj 生成缓存键
return cache_key
#flj 友情链接列表视图
class LinkListView(ListView):
"""友情链接列表视图"""
model = Links
template_name = 'blog/links_list.html'
'''
友情链接列表视图
'''
model = Links #flj 关联的模型
template_name = 'blog/links_list.html' #flj 使用的模板
#flj 获取友情链接数据，按显示顺序排序
def get_queryset(self):
# 获取所有启用的友情链接
return Links.objects.filter(is_enable=True)
#flj 过滤显示状态的友情链接并排序
return links
#flj 搜索视图，处理文章搜索功能
class EsSearchView(SearchView):
"""Elasticsearch搜索视图"""
'''
搜索视图
'''
#flj 获取搜索结果的上下文数据
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())
#flj 获取基础上下文
#flj 添加额外的搜索相关信息
return context
@csrf_exempt # 跳过CSRF验证，因为这是文件上传接口
#flj 文件上传接口，允许上传图片等文件
@csrf_exempt #flj 跳过CSRF验证，因为这是文件上传接口
def fileupload(request):
"""
文件上传接口，提供图床功能
需要提供正确的签名才能上传文件
:param request: HTTP请求对象
:return: 上传成功返回文件URL列表，失败返回错误信息
"""
if request.method == 'POST':
# 验证上传签名
sign = request.GET.get('sign', None)
if not sign:
return HttpResponseForbidden()
if not sign == get_sha256(get_sha256(settings.SECRET_KEY)):
return HttpResponseForbidden()
response = []
for filename in request.FILES:
# 按日期创建目录结构
timestr = timezone.now().strftime('%Y/%m/%d')
imgextensions = ['jpg', 'png', 'jpeg', 'bmp']
fname = u''.join(str(filename))
isimage = len([i for i in imgextensions if fname.find(i) >= 0]) > 0
# 根据文件类型选择存储目录
base_dir = os.path.join(settings.STATICFILES, "files" if not isimage else "image", timestr)
if not os.path.exists(base_dir):
os.makedirs(base_dir)
# 生成唯一的文件名
savepath = os.path.normpath(os.path.join(base_dir, f"{uuid.uuid4().hex}{os.path.splitext(filename)[-1]}"))
if not savepath.startswith(base_dir):
return HttpResponse("only for post")
# 保存文件
with open(savepath, 'wb+') as wfile:
for chunk in request.FILES[filename].chunks():
wfile.write(chunk)
# 如果是图片，进行压缩优化
if isimage:
from PIL import Image
image = Image.open(savepath)
image.save(savepath, quality=20, optimize=True)
# 生成访问URL
url = static(savepath)
response.append(url)
return HttpResponse(response)
else:
return HttpResponse("only for post")
'''
文件上传接口
'''
#flj 检查请求方法
#flj 验证权限
#flj 处理文件上传
#flj 保存文件到指定目录
#flj 返回文件URL
return HttpResponse(json.dumps(data), content_type="application/json")
#flj 404错误页面视图
#flj 处理页面未找到的情况
def page_not_found_view(
request,
exception,
template_name='blog/error_page.html'):
"""404错误页面处理函数"""
if exception:
logger.error(exception)
url = request.get_full_path()
return render(request,
template_name,
{'message': _('Sorry, the page you requested is not found, please click the home page to see other?'),
'statuscode': '404'},
status=404)
'''
404错误页面
'''
#flj 渲染错误页面
return render(request, template_name, context, status=404)
#flj 500错误页面视图
#flj 处理服务器内部错误的情况
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 other?'),
'statuscode': '500'},
status=500)
'''
500错误页面
'''
#flj 渲染错误页面
return render(request, template_name, context, status=500)
#flj 403错误页面视图
#flj 处理权限拒绝的情况
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)
'''
403错误页面
'''
#flj 渲染错误页面
return render(request, template_name, context, status=403)
#flj 清理缓存视图，用于手动清理站点缓存
#flj 提供管理功能，清除系统缓存
def clean_cache_view(request):
"""清除缓存的管理接口"""
cache.clear()
return HttpResponse('ok')
'''
清理缓存视图
'''
#flj 验证用户权限
#flj 清理缓存
#flj 返回成功信息
return HttpResponse(_('清理缓存成功'))

18
src/DjangoBlog/comments/models.py
Unescape View File

@ -1,3 +1,4 @@
#zxm 评论相关模型
from django.conf import settings
from django.db import models
from django.utils.timezone import now
@ -8,32 +9,41 @@ from blog.models import Article
# Create your models here.
#fkc 评论模型类
class Comment(models.Model):
#fkc 评论正文
body = models.TextField('正文', max_length=300)
#fkc 创建时间
creation_time = models.DateTimeField(_('creation time'), default=now)
#fkc 最后修改时间
last_modify_time = models.DateTimeField(_('last modify time'), default=now)
#fkc 评论作者（外键关联用户）
author = models.ForeignKey(
settings.AUTH_USER_MODEL,
verbose_name=_('author'),
on_delete=models.CASCADE)
#fkc 评论的文章（外键关联文章）
article = models.ForeignKey(
Article,
verbose_name=_('article'),
on_delete=models.CASCADE)
#fkc 父评论（支持回复功能）
parent_comment = models.ForeignKey(
'self',
verbose_name=_('parent comment'),
blank=True,
null=True,
on_delete=models.CASCADE)
#fkc 是否启用（审核功能）
is_enable = models.BooleanField(_('enable'),
default=False, blank=False, null=False)
class Meta:
ordering = ['-id']
verbose_name = _('comment')
verbose_name_plural = verbose_name
get_latest_by = 'id'
ordering = ['-id'] #fkc 按ID倒序排列
verbose_name = _('comment') #fkc 在管理后台显示的名称
verbose_name_plural = verbose_name #fkc 复数形式
get_latest_by = 'id' #fkc 获取最新记录的依据
#fkc 返回评论正文作为字符串表示
def __str__(self):
return self.body

59
src/DjangoBlog/comments/views.py
Unescape View File

@ -1,4 +1,5 @@
# Create your views here.
#cll 评论相关视图
from django.core.exceptions import ValidationError
from django.http import HttpResponseRedirect
from django.shortcuts import get_object_or_404
@ -12,52 +13,58 @@ from .forms import CommentForm
from .models import Comment
#xy 评论提交视图类
class CommentPostView(FormView):
form_class = CommentForm
template_name = 'blog/article_detail.html'
form_class = CommentForm #xy 使用评论表单
template_name = 'blog/article_detail.html' #xy 文章详情页面模板
@method_decorator(csrf_protect)
@method_decorator(csrf_protect) #xy 防止CSRF攻击
def dispatch(self, *args, **kwargs):
return super(CommentPostView, self).dispatch(*args, **kwargs)
#xy 处理GET请求
def get(self, request, *args, **kwargs):
article_id = self.kwargs['article_id']
article = get_object_or_404(Article, pk=article_id)
url = article.get_absolute_url()
return HttpResponseRedirect(url + "#comments")
article_id = self.kwargs['article_id'] #xy 获取文章ID
article = get_object_or_404(Article, pk=article_id) #xy 获取文章对象
url = article.get_absolute_url() #xy 获取文章详情页URL
return HttpResponseRedirect(url + "#comments") #xy 重定向到评论区
#xy 表单验证失败后的处理
def form_invalid(self, form):
article_id = self.kwargs['article_id']
article = get_object_or_404(Article, pk=article_id)
article_id = self.kwargs['article_id'] #xy 获取文章ID
article = get_object_or_404(Article, pk=article_id) #xy 获取文章对象
return self.render_to_response({
'form': form,
'article': article
'form': form, #xy 带错误的表单
'article': article #xy 文章对象
})
#xy 表单验证成功后的处理
def form_valid(self, form):
"""提交的数据验证合法后的逻辑"""
user = self.request.user
author = BlogUser.objects.get(pk=user.pk)
article_id = self.kwargs['article_id']
article = get_object_or_404(Article, pk=article_id)
user = self.request.user #xy 获取当前用户
author = BlogUser.objects.get(pk=user.pk) #xy 获取用户对象
article_id = self.kwargs['article_id'] #xy 获取文章ID
article = get_object_or_404(Article, pk=article_id) #xy 获取文章对象
#xy 检查文章是否允许评论
if article.comment_status == 'c' or article.status == 'c':
raise ValidationError("该文章评论已关闭.")
comment = form.save(False)
comment.article = article
comment = form.save(False) #xy 不保存到数据库
comment.article = article #xy 设置评论所属文章
from djangoblog.utils import get_blog_setting
settings = get_blog_setting()
if not settings.comment_need_review:
comment.is_enable = True
comment.author = author
settings = get_blog_setting() #xy 获取博客设置
if not settings.comment_need_review: #xy 如果不需要审核
comment.is_enable = True #xy 直接启用评论
comment.author = author #xy 设置评论作者
if form.cleaned_data['parent_comment_id']:
#xy 处理回复评论
if form.cleaned_data['parent_comment_id']: #xy 如果有父评论ID
parent_comment = Comment.objects.get(
pk=form.cleaned_data['parent_comment_id'])
comment.parent_comment = parent_comment
pk=form.cleaned_data['parent_comment_id']) #xy 获取父评论
comment.parent_comment = parent_comment #xy 设置父评论
comment.save(True)
comment.save(True) #xy 保存评论到数据库
return HttpResponseRedirect(
"%s#div-comment-%d" %
(article.get_absolute_url(), comment.pk))
(article.get_absolute_url(), comment.pk)) #xy 重定向到评论位置

37
src/DjangoBlog/oauth/models.py
Unescape View File

@ -1,4 +1,4 @@
# Create your models here.
#xy OAuth相关模型
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import models
@ -6,33 +6,47 @@ from django.utils.timezone import now
from django.utils.translation import gettext_lazy as _
#zhj OAuth用户模型类
class OAuthUser(models.Model):
#zhj 关联的博客用户
author = models.ForeignKey(
settings.AUTH_USER_MODEL,
verbose_name=_('author'),
blank=True,
null=True,
on_delete=models.CASCADE)
#zhj 第三方平台openid
openid = models.CharField(max_length=50)
#zhj 用户昵称
nickname = models.CharField(max_length=50, verbose_name=_('nick name'))
#zhj 访问令牌
token = models.CharField(max_length=150, null=True, blank=True)
#zhj 用户头像
picture = models.CharField(max_length=350, blank=True, null=True)
#zhj OAuth类型（github、weibo等）
type = models.CharField(blank=False, null=False, max_length=50)
#zhj 用户邮箱
email = models.CharField(max_length=50, null=True, blank=True)
#zhj 元数据
metadata = models.TextField(null=True, blank=True)
#zhj 创建时间
creation_time = models.DateTimeField(_('creation time'), default=now)
#zhj 最后修改时间
last_modify_time = models.DateTimeField(_('last modify time'), default=now)
#zhj 返回用户昵称作为字符串表示
def __str__(self):
return self.nickname
class Meta:
verbose_name = _('oauth user')
verbose_name_plural = verbose_name
ordering = ['-creation_time']
verbose_name = _('oauth user') #zhj 在管理后台显示的名称
verbose_name_plural = verbose_name #zhj 复数形式
ordering = ['-creation_time'] #zhj 按创建时间倒序排列
#flj OAuth配置模型类
class OAuthConfig(models.Model):
#flj OAuth类型选项
TYPE = (
('weibo', _('weibo')),
('google', _('google')),
@ -40,28 +54,37 @@ class OAuthConfig(models.Model):
('facebook', 'FaceBook'),
('qq', 'QQ'),
)
#flj OAuth类型
type = models.CharField(_('type'), max_length=10, choices=TYPE, default='a')
#flj 应用密钥
appkey = models.CharField(max_length=200, verbose_name='AppKey')
#flj 应用密钥密钥
appsecret = models.CharField(max_length=200, verbose_name='AppSecret')
#flj 回调URL
callback_url = models.CharField(
max_length=200,
verbose_name=_('callback url'),
blank=False,
default='')
#flj 是否启用
is_enable = models.BooleanField(
_('is enable'), default=True, blank=False, null=False)
#flj 创建时间
creation_time = models.DateTimeField(_('creation time'), default=now)
#flj 最后修改时间
last_modify_time = models.DateTimeField(_('last modify time'), default=now)
#flj 验证方法，确保每种类型只有一个配置
def clean(self):
if OAuthConfig.objects.filter(
type=self.type).exclude(id=self.id).count():
raise ValidationError(_(self.type + _('already exists')))
#flj 返回OAuth类型作为字符串表示
def __str__(self):
return self.type
class Meta:
verbose_name = 'oauth配置'
verbose_name_plural = verbose_name
ordering = ['-creation_time']
verbose_name = 'oauth配置' #flj 在管理后台显示的名称
verbose_name_plural = verbose_name #flj 复数形式
ordering = ['-creation_time'] #flj 按创建时间倒序排列

201
src/DjangoBlog/oauth/oauthmanager.py
Unescape View File

@ -9,86 +9,95 @@ import requests
from djangoblog.utils import cache_decorator
from oauth.models import OAuthUser, OAuthConfig
logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__) #flj
class OAuthAccessTokenException(Exception):
'''
oauth授权失败异常
'''
class OAuthAccessTokenException(Exception): #fkc
'''oauth授权失败异常'''
class BaseOauthManager(metaclass=ABCMeta):
"""获取用户授权"""
class BaseOauthManager(metaclass=ABCMeta): #cll
"""OAuth授权基类，定义统一接口规范"""
#cll 授权页面URL
AUTH_URL = None
"""获取token"""
#cll 获取访问令牌URL
TOKEN_URL = None
"""获取用户信息"""
#cll 获取用户信息URL
API_URL = None
'''icon图标名'''
#cll 平台图标标识名
ICON_NAME = None
def __init__(self, access_token=None, openid=None):
#cll 访问令牌
self.access_token = access_token
#cll 第三方平台用户唯一标识
self.openid = openid
@property
def is_access_token_set(self):
"""判断访问令牌是否已设置"""
return self.access_token is not None
@property
def is_authorized(self):
"""判断是否已完成授权（令牌和openid均存在）"""
return self.is_access_token_set and self.access_token is not None and self.openid is not None
@abstractmethod
def get_authorization_url(self, nexturl='/'):
"""抽象方法：获取授权跳转URL"""
pass
@abstractmethod
def get_access_token_by_code(self, code):
"""抽象方法：通过授权码获取访问令牌"""
pass
@abstractmethod
def get_oauth_userinfo(self):
"""抽象方法：通过访问令牌获取用户信息"""
pass
@abstractmethod
def get_picture(self, metadata):
"""抽象方法：从元数据中提取用户头像URL"""
pass
def do_get(self, url, params, headers=None):
"""通用GET请求方法"""
rsp = requests.get(url=url, params=params, headers=headers)
logger.info(rsp.text)
return rsp.text
def do_post(self, url, params, headers=None):
"""通用POST请求方法"""
rsp = requests.post(url, params, headers=headers)
logger.info(rsp.text)
return rsp.text
def get_config(self):
"""获取当前平台的OAuth配置（从数据库读取）"""
value = OAuthConfig.objects.filter(type=self.ICON_NAME)
return value[0] if value else None
class WBOauthManager(BaseOauthManager):
class WBOauthManager(BaseOauthManager): #xy
"""微博OAuth授权管理器"""
AUTH_URL = 'https://api.weibo.com/oauth2/authorize'
TOKEN_URL = 'https://api.weibo.com/oauth2/access_token'
API_URL = 'https://api.weibo.com/2/users/show.json'
ICON_NAME = 'weibo'
def __init__(self, access_token=None, openid=None):
# cll获取微博OAuth配置
config = self.get_config()
self.client_id = config.appkey if config else ''
self.client_secret = config.appsecret if config else ''
self.callback_url = config.callback_url if config else ''
super(
WBOauthManager,
self).__init__(
access_token=access_token,
openid=openid)
super().__init__(access_token=access_token, openid=openid)
def get_authorization_url(self, nexturl='/'):
"""构建微博授权跳转URL，包含回调地址和后续跳转路径"""
params = {
'client_id': self.client_id,
'response_type': 'code',
@ -98,7 +107,7 @@ class WBOauthManager(BaseOauthManager):
return url
def get_access_token_by_code(self, code):
"""通过授权码获取微博访问令牌和用户UID"""
params = {
'client_id': self.client_id,
'client_secret': self.client_secret,
@ -107,8 +116,9 @@ class WBOauthManager(BaseOauthManager):
'redirect_uri': self.callback_url
}
rsp = self.do_post(self.TOKEN_URL, params)
obj = json.loads(rsp)
# cll 成功获取令牌后，存储并返回用户信息
if 'access_token' in obj:
self.access_token = str(obj['access_token'])
self.openid = str(obj['uid'])
@ -117,22 +127,27 @@ class WBOauthManager(BaseOauthManager):
raise OAuthAccessTokenException(rsp)
def get_oauth_userinfo(self):
"""通过访问令牌获取微博用户信息（昵称、头像、邮箱等）"""
if not self.is_authorized:
return None
params = {
'uid': self.openid,
'access_token': self.access_token
}
rsp = self.do_get(self.API_URL, params)
try:
datas = json.loads(rsp)
user = OAuthUser()
user.metadata = rsp
user.picture = datas['avatar_large']
user.nickname = datas['screen_name']
user.openid = datas['id']
user.type = 'weibo'
user.token = self.access_token
user.metadata = rsp # cll 存储原始返回数据
user.picture = datas['avatar_large'] # cll 大尺寸头像
user.nickname = datas['screen_name'] # cll 昵称
user.openid = datas['id'] # cll 用户唯一标识
user.type = 'weibo' # cll 平台类型
user.token = self.access_token # cll 存储访问令牌
# cll 若返回邮箱则存储
if 'email' in datas and datas['email']:
user.email = datas['email']
return user
@ -142,12 +157,15 @@ class WBOauthManager(BaseOauthManager):
return None
def get_picture(self, metadata):
"""从元数据中提取微博用户头像URL"""
datas = json.loads(metadata)
return datas['avatar_large']
class ProxyManagerMixin:
class ProxyManagerMixin:
"""代理请求混入类，支持通过环境变量配置HTTP代理"""
def __init__(self, *args, **kwargs):
# cll 从环境变量读取代理配置
if os.environ.get("HTTP_PROXY"):
self.proxies = {
"http": os.environ.get("HTTP_PROXY"),
@ -157,17 +175,20 @@ class ProxyManagerMixin:
self.proxies = None
def do_get(self, url, params, headers=None):
"""带代理的GET请求"""
rsp = requests.get(url=url, params=params, headers=headers, proxies=self.proxies)
logger.info(rsp.text)
return rsp.text
def do_post(self, url, params, headers=None):
"""带代理的POST请求"""
rsp = requests.post(url, params, headers=headers, proxies=self.proxies)
logger.info(rsp.text)
return rsp.text
class GoogleOauthManager(ProxyManagerMixin, BaseOauthManager):
class GoogleOauthManager(ProxyManagerMixin, BaseOauthManager):
"""Google OAuth授权管理器（支持代理）"""
AUTH_URL = 'https://accounts.google.com/o/oauth2/v2/auth'
TOKEN_URL = 'https://www.googleapis.com/oauth2/v4/token'
API_URL = 'https://www.googleapis.com/oauth2/v3/userinfo'
@ -178,13 +199,10 @@ class GoogleOauthManager(ProxyManagerMixin, BaseOauthManager):
self.client_id = config.appkey if config else ''
self.client_secret = config.appsecret if config else ''
self.callback_url = config.callback_url if config else ''
super(
GoogleOauthManager,
self).__init__(
access_token=access_token,
openid=openid)
super().__init__(access_token=access_token, openid=openid)
def get_authorization_url(self, nexturl='/'):
"""构建Google授权跳转URL，请求openid和email权限"""
params = {
'client_id': self.client_id,
'response_type': 'code',
@ -195,43 +213,45 @@ class GoogleOauthManager(ProxyManagerMixin, BaseOauthManager):
return url
def get_access_token_by_code(self, code):
"""通过授权码获取Google访问令牌"""
params = {
'client_id': self.client_id,
'client_secret': self.client_secret,
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': self.callback_url
}
rsp = self.do_post(self.TOKEN_URL, params)
obj = json.loads(rsp)
if 'access_token' in obj:
self.access_token = str(obj['access_token'])
self.openid = str(obj['id_token'])
self.openid = str(obj['id_token']) # cll Google用id_token作为openid
logger.info(self.ICON_NAME + ' oauth ' + rsp)
return self.access_token
else:
raise OAuthAccessTokenException(rsp)
def get_oauth_userinfo(self):
"""通过访问令牌获取Google用户信息"""
if not self.is_authorized:
return None
params = {
'access_token': self.access_token
}
rsp = self.do_get(self.API_URL, params)
try:
datas = json.loads(rsp)
user = OAuthUser()
user.metadata = rsp
user.picture = datas['picture']
user.nickname = datas['name']
user.openid = datas['sub']
user.picture = datas['picture'] # cll 头像URL
user.nickname = datas['name'] # cll 姓名
user.openid = datas['sub'] # cll 唯一标识
user.token = self.access_token
user.type = 'google'
if datas['email']:
user.email = datas['email']
return user
@ -241,11 +261,13 @@ class GoogleOauthManager(ProxyManagerMixin, BaseOauthManager):
return None
def get_picture(self, metadata):
"""从元数据提取Google用户头像"""
datas = json.loads(metadata)
return datas['picture']
class GitHubOauthManager(ProxyManagerMixin, BaseOauthManager):
"""GitHub OAuth授权管理器（支持代理）"""
AUTH_URL = 'https://github.com/login/oauth/authorize'
TOKEN_URL = 'https://github.com/login/oauth/access_token'
API_URL = 'https://api.github.com/user'
@ -256,13 +278,10 @@ class GitHubOauthManager(ProxyManagerMixin, BaseOauthManager):
self.client_id = config.appkey if config else ''
self.client_secret = config.appsecret if config else ''
self.callback_url = config.callback_url if config else ''
super(
GitHubOauthManager,
self).__init__(
access_token=access_token,
openid=openid)
super().__init__(access_token=access_token, openid=openid)
def get_authorization_url(self, next_url='/'):
"""构建GitHub授权跳转URL，请求user权限"""
params = {
'client_id': self.client_id,
'response_type': 'code',
@ -273,16 +292,17 @@ class GitHubOauthManager(ProxyManagerMixin, BaseOauthManager):
return url
def get_access_token_by_code(self, code):
"""通过授权码获取GitHub访问令牌"""
params = {
'client_id': self.client_id,
'client_secret': self.client_secret,
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': self.callback_url
}
rsp = self.do_post(self.TOKEN_URL, params)
# cll GitHub返回格式为form-encoded，需解析
from urllib import parse
r = parse.parse_qs(rsp)
if 'access_token' in r:
@ -292,19 +312,21 @@ class GitHubOauthManager(ProxyManagerMixin, BaseOauthManager):
raise OAuthAccessTokenException(rsp)
def get_oauth_userinfo(self):
"""通过访问令牌获取GitHub用户信息（需在请求头携带令牌）"""
rsp = self.do_get(self.API_URL, params={}, headers={
"Authorization": "token " + self.access_token
})
try:
datas = json.loads(rsp)
user = OAuthUser()
user.picture = datas['avatar_url']
user.nickname = datas['name']
user.openid = datas['id']
user.picture = datas['avatar_url'] # cll 头像URL
user.nickname = datas['name'] # cll 姓名（可能为空）
user.openid = datas['id'] # cll 唯一标识
user.type = 'github'
user.token = self.access_token
user.metadata = rsp
if 'email' in datas and datas['email']:
user.email = datas['email']
return user
@ -314,11 +336,13 @@ class GitHubOauthManager(ProxyManagerMixin, BaseOauthManager):
return None
def get_picture(self, metadata):
"""从元数据提取GitHub用户头像"""
datas = json.loads(metadata)
return datas['avatar_url']
class FaceBookOauthManager(ProxyManagerMixin, BaseOauthManager):
class FaceBookOauthManager(ProxyManagerMixin, BaseOauthManager): #fkc
"""Facebook OAuth授权管理器（支持代理）"""
AUTH_URL = 'https://www.facebook.com/v16.0/dialog/oauth'
TOKEN_URL = 'https://graph.facebook.com/v16.0/oauth/access_token'
API_URL = 'https://graph.facebook.com/me'
@ -329,13 +353,10 @@ class FaceBookOauthManager(ProxyManagerMixin, BaseOauthManager):
self.client_id = config.appkey if config else ''
self.client_secret = config.appsecret if config else ''
self.callback_url = config.callback_url if config else ''
super(
FaceBookOauthManager,
self).__init__(
access_token=access_token,
openid=openid)
super().__init__(access_token=access_token, openid=openid)
def get_authorization_url(self, next_url='/'):
"""构建Facebook授权跳转URL，请求邮箱和公开资料权限"""
params = {
'client_id': self.client_id,
'response_type': 'code',
@ -346,17 +367,16 @@ class FaceBookOauthManager(ProxyManagerMixin, BaseOauthManager):
return url
def get_access_token_by_code(self, code):
"""通过授权码获取Facebook访问令牌"""
params = {
'client_id': self.client_id,
'client_secret': self.client_secret,
# 'grant_type': 'authorization_code',
'code': code,
'redirect_uri': self.callback_url
}
rsp = self.do_post(self.TOKEN_URL, params)
obj = json.loads(rsp)
if 'access_token' in obj:
token = str(obj['access_token'])
self.access_token = token
@ -365,21 +385,26 @@ class FaceBookOauthManager(ProxyManagerMixin, BaseOauthManager):
raise OAuthAccessTokenException(rsp)
def get_oauth_userinfo(self):
"""通过访问令牌获取Facebook用户信息（指定返回字段）"""
params = {
'access_token': self.access_token,
'fields': 'id,name,picture,email'
'fields': 'id,name,picture,email' # cll 指定需要返回的字段
}
try:
rsp = self.do_get(self.API_URL, params)
datas = json.loads(rsp)
user = OAuthUser()
user.nickname = datas['name']
user.openid = datas['id']
user.nickname = datas['name'] # cll 姓名
user.openid = datas['id'] # cll 唯一标识
user.type = 'facebook'
user.token = self.access_token
user.metadata = rsp
if 'email' in datas and datas['email']:
user.email = datas['email']
# cll 解析头像URL（Facebook返回格式嵌套较深）
if 'picture' in datas and datas['picture'] and datas['picture']['data'] and datas['picture']['data']['url']:
user.picture = str(datas['picture']['data']['url'])
return user
@ -388,14 +413,17 @@ class FaceBookOauthManager(ProxyManagerMixin, BaseOauthManager):
return None
def get_picture(self, metadata):
"""从元数据提取Facebook用户头像"""
datas = json.loads(metadata)
return str(datas['picture']['data']['url'])
class QQOauthManager(BaseOauthManager):
class QQOauthManager(BaseOauthManager): #cll
"""QQ OAuth授权管理器"""
AUTH_URL = 'https://graph.qq.com/oauth2.0/authorize'
TOKEN_URL = 'https://graph.qq.com/oauth2.0/token'
API_URL = 'https://graph.qq.com/user/get_user_info'
# cll QQ需单独请求openid的URL
OPEN_ID_URL = 'https://graph.qq.com/oauth2.0/me'
ICON_NAME = 'qq'
@ -404,13 +432,10 @@ class QQOauthManager(BaseOauthManager):
self.client_id = config.appkey if config else ''
self.client_secret = config.appsecret if config else ''
self.callback_url = config.callback_url if config else ''
super(
QQOauthManager,
self).__init__(
access_token=access_token,
openid=openid)
super().__init__(access_token=access_token, openid=openid)
def get_authorization_url(self, next_url='/'):
"""构建QQ授权跳转URL"""
params = {
'response_type': 'code',
'client_id': self.client_id,
@ -420,6 +445,7 @@ class QQOauthManager(BaseOauthManager):
return url
def get_access_token_by_code(self, code):
"""通过授权码获取QQ访问令牌"""
params = {
'grant_type': 'authorization_code',
'client_id': self.client_id,
@ -428,77 +454,90 @@ class QQOauthManager(BaseOauthManager):
'redirect_uri': self.callback_url
}
rsp = self.do_get(self.TOKEN_URL, params)
if rsp:
# cll QQ返回格式为form-encoded，解析令牌
d = urllib.parse.parse_qs(rsp)
if 'access_token' in d:
token = d['access_token']
self.access_token = token[0]
return token
else:
raise OAuthAccessTokenException(rsp)
raise OAuthAccessTokenException(rsp)
def get_open_id(self):
"""QQ需单独请求openid（通过访问令牌获取）"""
if self.is_access_token_set:
params = {
'access_token': self.access_token
}
rsp = self.do_get(self.OPEN_ID_URL, params)
if rsp:
rsp = rsp.replace(
'callback(', '').replace(
')', '').replace(
';', '')
# 去除QQ返回的callback包裹符
rsp = rsp.replace('callback(', '').replace(')', '').replace(';', '')
obj = json.loads(rsp)
openid = str(obj['openid'])
self.openid = openid
return openid
def get_oauth_userinfo(self):
"""获取QQ用户信息（需先获取openid）"""
openid = self.get_open_id()
if openid:
params = {
'access_token': self.access_token,
'oauth_consumer_key': self.client_id,
'oauth_consumer_key': self.client_id, # cll QQ要求传入appkey
'openid': self.openid
}
rsp = self.do_get(self.API_URL, params)
logger.info(rsp)
obj = json.loads(rsp)
user = OAuthUser()
user.nickname = obj['nickname']
user.openid = openid
user.nickname = obj['nickname'] # cll 昵称
user.openid = openid # cll 唯一标识
user.type = 'qq'
user.token = self.access_token
user.metadata = rsp
if 'email' in obj:
user.email = obj['email']
if 'figureurl' in obj:
user.picture = str(obj['figureurl'])
user.picture = str(obj['figureurl']) # cll 头像URL
return user
def get_picture(self, metadata):
"""从元数据提取QQ用户头像"""
datas = json.loads(metadata)
return str(datas['figureurl'])
@cache_decorator(expiration=100 * 60)
def get_oauth_apps():
def get_oauth_apps(): #xy
"""获取所有启用的OAuth应用（缓存100分钟）"""
# cll 读取数据库中启用的OAuth配置
configs = OAuthConfig.objects.filter(is_enable=True).all()
if not configs:
return []
# 提取已启用的平台类型
configtypes = [x.type for x in configs]
# cll 获取所有BaseOauthManager的子类（各平台实现）
applications = BaseOauthManager.__subclasses__()
# cll 筛选出已启用的平台实例
apps = [x() for x in applications if x().ICON_NAME.lower() in configtypes]
return apps
def get_manager_by_type(type):
def get_manager_by_type(type):
"""根据平台类型获取对应的OAuth管理器实例"""
applications = get_oauth_apps()
if applications:
# cll 匹配平台类型（不区分大小写）
finds = list(
filter(
lambda x: x.ICON_NAME.lower() == type.lower(),
applications))
if finds:
return finds[0]
return None
return None
Write Preview
Loading…
Cancel
Save