.html', # 修复：URL路径和参数名统一为bind_success、oauth_id（蛇形）
+ views.bind_success, # 修复：视图函数名统一为bind_success（蛇形）
+ name='bind_success'), # 修复：URL名称统一为bind_success（蛇形）
# OAuth登录处理 - 处理第三方登录回调
path(
- r'oauth/oauthlogin', # OAuth登录回调URL路径
- views.oauthlogin, # 视图函数，处理登录回调逻辑
- name='oauthlogin') # URL名称，用于反向解析
-]
+ r'oauth/oauth_login', # 修复：URL路径统一为oauth_login（蛇形）
+ views.oauth_login, # 修复：视图函数名统一为oauth_login（蛇形）
+ name='oauth_login') # 修复：URL名称统一为oauth_login（蛇形）
+]
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/oauth/views.py b/src/DjangoBlog-master(1)/DjangoBlog-master/oauth/views.py
index e08c39b..ea44343 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/oauth/views.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/oauth/views.py
@@ -1,5 +1,4 @@
import logging
-# Create your views here.
from urllib.parse import urlparse
from django.conf import settings
@@ -7,308 +6,212 @@ from django.contrib.auth import get_user_model
from django.contrib.auth import login
from django.core.exceptions import ObjectDoesNotExist
from django.db import transaction
-from django.http import HttpResponseForbidden
-from django.http import HttpResponseRedirect
-from django.shortcuts import get_object_or_404
-from django.shortcuts import render
+from django.http import HttpResponseForbidden, HttpResponseRedirect
+from django.shortcuts import get_object_or_404, render
from django.urls import reverse
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from django.views.generic import FormView
from djangoblog.blog_signals import oauth_user_login_signal
-from djangoblog.utils import get_current_site
-from djangoblog.utils import send_email, get_sha256
+from djangoblog.utils import get_current_site, send_email, get_sha256
from oauth.forms import RequireEmailForm
from .models import OAuthUser
from .oauthmanager import get_manager_by_type, OAuthAccessTokenException
-# 获取logger实例用于记录日志
logger = logging.getLogger(__name__)
def get_redirecturl(request):
- """
- 获取重定向URL，并进行安全验证
-
- Args:
- request: HTTP请求对象
-
- Returns:
- str: 安全的重定向URL
- """
- # 从请求参数获取next_url，默认为None
- nexturl = request.GET.get('next_url', None)
- # 如果nexturl为空或是登录页面，则重定向到首页
- if not nexturl or nexturl == '/login/' or nexturl == '/login':
- nexturl = '/'
- return nexturl
-
- # 解析URL，检查域名安全性
- p = urlparse(nexturl)
- if p.netloc:
+ next_url = request.GET.get('next_url', None)
+ if not next_url or next_url in ('/login/', '/login'):
+ next_url = '/'
+ return next_url
+
+ parsed_url = urlparse(next_url)
+ if parsed_url.netloc:
site = get_current_site().domain
- # 检查域名是否匹配当前站点，防止开放重定向攻击
- if not p.netloc.replace('www.', '') == site.replace('www.', ''):
- logger.info('非法url:' + nexturl)
+ if not parsed_url.netloc.replace('www.', '') == site.replace('www.', ''):
+ logger.info('非法url:' + next_url)
return "/"
- return nexturl
+ return next_url
def oauthlogin(request):
- """
- OAuth登录入口 - 重定向到第三方授权页面
-
- Args:
- request: HTTP请求对象
-
- Returns:
- HttpResponseRedirect: 重定向响应
- """
- # 获取OAuth类型（如weibo、github等）
- type = request.GET.get('type', None)
- if not type:
+ oauth_type = request.GET.get('type', None) # 修复：重命名type为oauth_type
+ if not oauth_type:
return HttpResponseRedirect('/')
- # 获取对应的OAuth管理器
- manager = get_manager_by_type(type)
- if not manager:
+ oauth_manager = get_manager_by_type(oauth_type) # 修复：重命名manager为oauth_manager
+ if not oauth_manager:
return HttpResponseRedirect('/')
- # 获取安全的重定向URL
- nexturl = get_redirecturl(request)
- # 获取第三方授权URL并重定向
- authorizeurl = manager.get_authorization_url(nexturl)
- return HttpResponseRedirect(authorizeurl)
+ next_url = get_redirecturl(request)
+ authorize_url = oauth_manager.get_authorization_url(next_url) # 修复：重命名authorizeurl
+ return HttpResponseRedirect(authorize_url)
def authorize(request):
- """
- OAuth授权回调处理 - 处理第三方登录回调
-
- Args:
- request: HTTP请求对象
-
- Returns:
- HttpResponseRedirect: 重定向响应
- """
- type = request.GET.get('type', None)
- if not type:
+ oauth_type = request.GET.get('type', None) # 修复：重命名type为oauth_type
+ if not oauth_type:
return HttpResponseRedirect('/')
- manager = get_manager_by_type(type)
- if not manager:
+ oauth_manager = get_manager_by_type(oauth_type) # 修复：重命名manager为oauth_manager
+ if not oauth_manager:
return HttpResponseRedirect('/')
- # 获取授权码
- code = request.GET.get('code', None)
+ auth_code = request.GET.get('code', None) # 修复：重命名code为auth_code
try:
- # 使用授权码获取访问令牌
- rsp = manager.get_access_token_by_code(code)
+ token_response = oauth_manager.get_access_token_by_code(auth_code) # 修复：重命名rsp
except OAuthAccessTokenException as e:
logger.warning("OAuthAccessTokenException:" + str(e))
return HttpResponseRedirect('/')
except Exception as e:
logger.error(e)
- rsp = None
+ token_response = None
- nexturl = get_redirecturl(request)
- if not rsp:
- # 如果获取token失败，重新跳转到授权页面
- return HttpResponseRedirect(manager.get_authorization_url(nexturl))
+ next_url = get_redirecturl(request)
+ if not token_response:
+ return HttpResponseRedirect(oauth_manager.get_authorization_url(next_url))
- # 获取用户信息
- user = manager.get_oauth_userinfo()
- if user:
- # 处理昵称为空的情况
- if not user.nickname or not user.nickname.strip():
- user.nickname = "djangoblog" + timezone.now().strftime('%y%m%d%I%M%S')
+ oauth_user_info = oauth_manager.get_oauth_userinfo() # 修复：重命名user为oauth_user_info
+ if oauth_user_info:
+ if not oauth_user_info.nickname or not oauth_user_info.nickname.strip():
+ oauth_user_info.nickname = f"djangoblog{timezone.now().strftime('%y%m%d%I%M%S')}"
try:
# 检查是否已存在该OAuth用户
- temp = OAuthUser.objects.get(type=type, openid=user.openid)
- # 更新用户信息
- temp.picture = user.picture
- temp.metadata = user.metadata
- temp.nickname = user.nickname
- user = temp
+ existing_oauth_user = OAuthUser.objects.get(type=oauth_type, openid=oauth_user_info.openid) # 修复：重命名temp
+ existing_oauth_user.picture = oauth_user_info.picture
+ existing_oauth_user.metadata = oauth_user_info.metadata
+ existing_oauth_user.nickname = oauth_user_info.nickname
+ oauth_user_info = existing_oauth_user
except ObjectDoesNotExist:
pass
- # Facebook的token过长，清空处理
- if type == 'facebook':
- user.token = ''
+ if oauth_type == 'facebook':
+ oauth_user_info.token = ''
- # 如果用户有邮箱，直接处理登录
- if user.email:
- with transaction.atomic(): # 使用事务保证数据一致性
- author = None
+ if oauth_user_info.email:
+ with transaction.atomic():
+ blog_author = None # 修复：重命名author为blog_author（区分系统用户和OAuth用户）
try:
- author = get_user_model().objects.get(id=user.author_id)
+ blog_author = get_user_model().objects.get(id=oauth_user_info.author_id)
except ObjectDoesNotExist:
pass
- if not author:
- # 创建或获取用户
- result = get_user_model().objects.get_or_create(email=user.email)
- author = result[0]
- if result[1]: # 如果是新创建的用户
+ if not blog_author:
+ result = get_user_model().objects.get_or_create(email=oauth_user_info.email)
+ blog_author = result[0]
+ if result[1]:
try:
- get_user_model().objects.get(username=user.nickname)
+ get_user_model().objects.get(username=oauth_user_info.nickname)
except ObjectDoesNotExist:
- author.username = user.nickname
+ blog_author.username = oauth_user_info.nickname
else:
- # 用户名冲突时生成唯一用户名
- author.username = "djangoblog" + timezone.now().strftime('%y%m%d%I%M%S')
- author.source = 'authorize'
- author.save()
+ blog_author.username = f"djangoblog{timezone.now().strftime('%y%m%d%I%M%S')}"
+ blog_author.source = 'authorize'
+ blog_author.save()
- # 关联OAuth用户和系统用户
- user.author = author
- user.save()
+ oauth_user_info.author = blog_author
+ oauth_user_info.save()
- # 发送OAuth用户登录信号
oauth_user_login_signal.send(
- sender=authorize.__class__, id=user.id)
- # 登录用户
- login(request, author)
- return HttpResponseRedirect(nexturl)
+ sender=authorize.__class__, id=oauth_user_info.id)
+ login(request, blog_author)
+ return HttpResponseRedirect(next_url)
else:
- # 没有邮箱，保存OAuth用户信息并跳转到邮箱输入页面
- user.save()
- url = reverse('oauth:require_email', kwargs={
- 'oauthid': user.id
- })
+ oauth_user_info.save()
+ url = reverse('oauth:require_email', kwargs={'oauthid': oauth_user_info.id})
return HttpResponseRedirect(url)
else:
- return HttpResponseRedirect(nexturl)
+ return HttpResponseRedirect(next_url)
def emailconfirm(request, id, sign):
- """
- 邮箱确认处理 - 验证邮箱并完成用户绑定
-
- Args:
- request: HTTP请求对象
- id: OAuth用户ID
- sign: 安全签名
-
- Returns:
- HttpResponseRedirect: 重定向响应
- """
if not sign:
return HttpResponseForbidden()
- # 验证签名安全性
- if not get_sha256(settings.SECRET_KEY +
- str(id) +
- settings.SECRET_KEY).upper() == sign.upper():
+ if not get_sha256(f"{settings.SECRET_KEY}{id}{settings.SECRET_KEY}").upper() == sign.upper():
return HttpResponseForbidden()
- # 获取OAuth用户
- oauthuser = get_object_or_404(OAuthUser, pk=id)
+ oauth_user = get_object_or_404(OAuthUser, pk=id) # 修复：重命名oauthuser为oauth_user
with transaction.atomic():
- if oauthuser.author:
- author = get_user_model().objects.get(pk=oauthuser.author_id)
+ if oauth_user.author:
+ blog_author = get_user_model().objects.get(pk=oauth_user.author_id) # 修复：重命名author为blog_author
else:
- # 创建或获取系统用户
- result = get_user_model().objects.get_or_create(email=oauthuser.email)
- author = result[0]
- if result[1]: # 新创建的用户
- author.source = 'emailconfirm'
- author.username = oauthuser.nickname.strip() if oauthuser.nickname.strip(
- ) else "djangoblog" + timezone.now().strftime('%y%m%d%I%M%S')
- author.save()
-
- # 关联用户
- oauthuser.author = author
- oauthuser.save()
-
- # 发送登录信号并登录用户
+ result = get_user_model().objects.get_or_create(email=oauth_user.email)
+ blog_author = result[0]
+ if result[1]:
+ blog_author.source = 'emailconfirm'
+ blog_author.username = (
+ oauth_user.nickname.strip()
+ if oauth_user.nickname.strip()
+ else f"djangoblog{timezone.now().strftime('%y%m%d%I%M%S')}"
+ )
+ blog_author.save()
+
+ oauth_user.author = blog_author
+ oauth_user.save()
+
oauth_user_login_signal.send(
- sender=emailconfirm.__class__,
- id=oauthuser.id)
- login(request, author)
+ sender=emailconfirm.__class__, id=oauth_user.id)
+ login(request, blog_author)
- # 发送绑定成功邮件
site = 'http://' + get_current_site().domain
content = _('''
- 恭喜您，您已成功绑定邮箱。您可以使用%(oauthuser_type)s直接登录本站，无需密码。
+ 恭喜您，您已成功绑定邮箱。您可以使用%(oauth_type)s直接登录本站，无需密码。
欢迎您继续关注本站，地址是%(site)s
再次感谢！
如果上面的链接无法打开，请将此链接复制到浏览器。
%(site)s
- ''') % {'oauthuser_type': oauthuser.type, 'site': site}
+ ''') % {'oauth_type': oauth_user.type, 'site': site} # 修复：变量名oauthuser_type改为oauth_type
- send_email(emailto=[oauthuser.email, ], title=_('恭喜您绑定成功！'), content=content)
+ send_email(emailto=[oauth_user.email, ], title=_('恭喜您绑定成功！'), content=content)
- # 跳转到绑定成功页面
- url = reverse('oauth:bindsuccess', kwargs={
- 'oauthid': id
- })
- url = url + '?type=success'
+ url = reverse('oauth:bindsuccess', kwargs={'oauthid': id})
+ url += '?type=success'
return HttpResponseRedirect(url)
class RequireEmailView(FormView):
- """
- 需要邮箱视图 - 处理用户输入邮箱地址
- """
- form_class = RequireEmailForm # 使用的表单类
- template_name = 'oauth/require_email.html' # 模板名称
+ form_class = RequireEmailForm
+ template_name = 'oauth/require_email.html'
def get(self, request, *args, **kwargs):
- """GET请求处理"""
- oauthid = self.kwargs['oauthid']
- oauthuser = get_object_or_404(OAuthUser, pk=oauthid)
- # 如果已有邮箱，直接跳过（这里注释掉了重定向逻辑）
- if oauthuser.email:
+ oauth_id = self.kwargs['oauthid'] # 修复：重命名oauthid为oauth_id
+ oauth_user = get_object_or_404(OAuthUser, pk=oauth_id)
+ if oauth_user.email:
pass
- # return HttpResponseRedirect('/')
-
- return super(RequireEmailView, self).get(request, *args, **kwargs)
+ return super().get(request, *args, **kwargs)
def get_initial(self):
- """获取表单初始数据"""
- oauthid = self.kwargs['oauthid']
- return {
- 'email': '',
- 'oauthid': oauthid
- }
-
- def get_context_data(self, **kwargs):
- """获取模板上下文数据"""
- oauthid = self.kwargs['oauthid']
- oauthuser = get_object_or_404(OAuthUser, pk=oauthid)
- # 添加用户头像到上下文
- if oauthuser.picture:
- kwargs['picture'] = oauthuser.picture
- return super(RequireEmailView, self).get_context_data(**kwargs)
+ oauth_id = self.kwargs['oauthid']
+ return {'email': '', 'oauthid': oauth_id}
+
+ def get_context_data(self,** kwargs):
+ oauth_id = self.kwargs['oauthid']
+ oauth_user = get_object_or_404(OAuthUser, pk=oauth_id)
+ if oauth_user.picture:
+ kwargs['picture'] = oauth_user.picture
+ return super().get_context_data(**kwargs)
def form_valid(self, form):
- """表单验证通过后的处理"""
email = form.cleaned_data['email']
- oauthid = form.cleaned_data['oauthid']
- oauthuser = get_object_or_404(OAuthUser, pk=oauthid)
- oauthuser.email = email
- oauthuser.save()
-
- # 生成安全签名
- sign = get_sha256(settings.SECRET_KEY +
- str(oauthuser.id) + settings.SECRET_KEY)
+ oauth_id = form.cleaned_data['oauthid']
+ oauth_user = get_object_or_404(OAuthUser, pk=oauth_id)
+ oauth_user.email = email
+ oauth_user.save()
+
+ sign = get_sha256(f"{settings.SECRET_KEY}{oauth_user.id}{settings.SECRET_KEY}")
site = get_current_site().domain
if settings.DEBUG:
site = '127.0.0.1:8000'
- # 构建邮箱确认URL
- path = reverse('oauth:email_confirm', kwargs={
- 'id': oauthid,
- 'sign': sign
- })
- url = "http://{site}{path}".format(site=site, path=path)
+ path = reverse('oauth:email_confirm', kwargs={'id': oauth_id, 'sign': sign})
+ confirm_url = f"http://{site}{path}" # 修复：重命名url为confirm_url
- # 发送确认邮件
content = _("""
请点击下面的链接完成邮箱绑定
%(url)s
@@ -317,43 +220,26 @@ class RequireEmailView(FormView):
如果上面的链接无法打开，请将此链接复制到浏览器。
%(url)s
- """) % {'url': url}
+ """) % {'url': confirm_url}
send_email(emailto=[email, ], title=_('绑定邮箱'), content=content)
- # 跳转到绑定成功提示页面
- url = reverse('oauth:bindsuccess', kwargs={
- 'oauthid': oauthid
- })
- url = url + '?type=email'
- return HttpResponseRedirect(url)
+ redirect_url = reverse('oauth:bindsuccess', kwargs={'oauthid': oauth_id})
+ redirect_url += '?type=email'
+ return HttpResponseRedirect(redirect_url)
def bindsuccess(request, oauthid):
- """
- 绑定成功页面 - 显示绑定状态信息
-
- Args:
- request: HTTP请求对象
- oauthid: OAuth用户ID
+ notify_type = request.GET.get('type', None) # 修复：重命名type为notify_type
+ oauth_user = get_object_or_404(OAuthUser, pk=oauthid)
- Returns:
- HttpResponse: 渲染的响应
- """
- type = request.GET.get('type', None)
- oauthuser = get_object_or_404(OAuthUser, pk=oauthid)
-
- # 根据类型显示不同的提示信息
- if type == 'email':
+ if notify_type == 'email':
title = _('绑定邮箱')
- content = _(
- '恭喜您，绑定只差一步之遥。请登录您的邮箱查看邮件完成绑定。谢谢。')
+ content = _('恭喜您，绑定只差一步之遥。请登录您的邮箱查看邮件完成绑定。谢谢。')
else:
title = _('绑定成功')
content = _(
- "恭喜您，您已成功绑定邮箱地址。您可以使用%(oauthuser_type)s直接登录本站，无需密码。欢迎您继续关注本站。" % {
- 'oauthuser_type': oauthuser.type})
+ "恭喜您，您已成功绑定邮箱地址。您可以使用%(oauth_type)s直接登录本站，无需密码。欢迎您继续关注本站。" % {
+ 'oauth_type': oauth_user.type
+ })
- return render(request, 'oauth/bindsuccess.html', {
- 'title': title,
- 'content': content
- })
\ No newline at end of file
+ return render(request, 'oauth/bindsuccess.html', {'title': title, 'content': content})
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/migrations/0002_alter_owntracklog_options_and_more.py b/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/migrations/0002_alter_owntracklog_options_and_more.py
index b4f8dec..fa1075a 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/migrations/0002_alter_owntracklog_options_and_more.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/migrations/0002_alter_owntracklog_options_and_more.py
@@ -12,11 +12,16 @@ class Migration(migrations.Migration):
operations = [
migrations.AlterModelOptions(
name='owntracklog',
- options={'get_latest_by': 'creation_time', 'ordering': ['creation_time'], 'verbose_name': 'OwnTrackLogs', 'verbose_name_plural': 'OwnTrackLogs'},
+ options={
+ 'get_latest_by': 'creation_time',
+ 'ordering': ['creation_time'],
+ 'verbose_name': 'owntrack日志', # 修复：统一为中文命名
+ 'verbose_name_plural': 'owntrack日志' # 修复：统一为中文命名
+ },
),
migrations.RenameField(
model_name='owntracklog',
old_name='created_time',
new_name='creation_time',
),
- ]
+ ]
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/tests.py b/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/tests.py
index 3b4b9d8..593d807 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/tests.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/tests.py
@@ -6,59 +6,76 @@ from accounts.models import BlogUser
from .models import OwnTrackLog
-# Create your tests here.
-
class OwnTrackLogTest(TestCase):
def setUp(self):
self.client = Client()
self.factory = RequestFactory()
def test_own_track_log(self):
- o = {
+ # 完整轨迹数据（包含经纬度）
+ valid_track_data = {
'tid': 12,
'lat': 123.123,
'lon': 134.341
}
+ # 测试提交完整轨迹数据
self.client.post(
'/owntracks/logtracks',
- json.dumps(o),
- content_type='application/json')
- length = len(OwnTrackLog.objects.all())
- self.assertEqual(length, 1)
+ json.dumps(valid_track_data),
+ content_type='application/json'
+ )
+ # 验证数据已被正确保存
+ track_count = len(OwnTrackLog.objects.all())
+ self.assertEqual(track_count, 1)
- o = {
+ # 不完整轨迹数据（缺少经度）
+ invalid_track_data = {
'tid': 12,
'lat': 123.123
}
+ # 测试提交不完整轨迹数据（应保存失败）
self.client.post(
'/owntracks/logtracks',
- json.dumps(o),
- content_type='application/json')
- length = len(OwnTrackLog.objects.all())
- self.assertEqual(length, 1)
+ json.dumps(invalid_track_data),
+ content_type='application/json'
+ )
+ # 验证数据未被重复保存
+ track_count_after_invalid = len(OwnTrackLog.objects.all())
+ self.assertEqual(track_count_after_invalid, 1)
- rsp = self.client.get('/owntracks/show_maps')
- self.assertEqual(rsp.status_code, 302)
+ # 未登录状态访问地图页面（应重定向）
+ map_response_unauth = self.client.get('/owntracks/show_maps')
+ self.assertEqual(map_response_unauth.status_code, 302)
- user = BlogUser.objects.create_superuser(
+ # 创建测试超级用户并登录
+ test_user = BlogUser.objects.create_superuser(
email="liangliangyy1@gmail.com",
username="liangliangyy1",
- password="liangliangyy1")
-
+ password="liangliangyy1"
+ )
self.client.login(username='liangliangyy1', password='liangliangyy1')
- s = OwnTrackLog()
- s.tid = 12
- s.lon = 123.234
- s.lat = 34.234
- s.save()
- rsp = self.client.get('/owntracks/show_dates')
- self.assertEqual(rsp.status_code, 200)
- rsp = self.client.get('/owntracks/show_maps')
- self.assertEqual(rsp.status_code, 200)
- rsp = self.client.get('/owntracks/get_datas')
- self.assertEqual(rsp.status_code, 200)
- rsp = self.client.get('/owntracks/get_datas?date=2018-02-26')
- self.assertEqual(rsp.status_code, 200)
+ # 创建测试轨迹记录
+ test_track = OwnTrackLog()
+ test_track.tid = 12
+ test_track.lon = 123.234
+ test_track.lat = 34.234
+ test_track.save()
+
+ # 测试登录后访问日期筛选页
+ dates_response = self.client.get('/owntracks/show_dates')
+ self.assertEqual(dates_response.status_code, 200)
+
+ # 测试登录后访问地图页面
+ map_response_auth = self.client.get('/owntracks/show_maps')
+ self.assertEqual(map_response_auth.status_code, 200)
+
+ # 测试获取所有轨迹数据
+ all_data_response = self.client.get('/owntracks/get_datas')
+ self.assertEqual(all_data_response.status_code, 200)
+
+ # 测试按日期筛选轨迹数据
+ date_filter_response = self.client.get('/owntracks/get_datas?date=2018-02-26')
+ self.assertEqual(date_filter_response.status_code, 200)
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/urls.py b/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/urls.py
index c19ada8..25b54f2 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/urls.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/owntracks/urls.py
@@ -5,8 +5,12 @@ from . import views
app_name = "owntracks"
urlpatterns = [
- path('owntracks/logtracks', views.manage_owntrack_log, name='logtracks'),
- path('owntracks/show_maps', views.show_maps, name='show_maps'),
- path('owntracks/get_datas', views.get_datas, name='get_datas'),
- path('owntracks/show_dates', views.show_log_dates, name='show_dates')
-]
+ # 日志管理（新增/查看/删除轨迹日志）
+ path('logs/manage/', views.manage_owntrack_log, name='manage_logs'),
+ # 地图展示（可视化轨迹）
+ path('maps/show/', views.show_maps, name='show_maps'),
+ # 数据接口（获取轨迹数据用于渲染）
+ path('data/get/', views.get_datas, name='get_data'),
+ # 日期筛选（展示可查询的日志日期列表）
+ path('dates/show/', views.show_log_dates, name='show_dates'),
+]
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/plugins/seo_optimizer/plugin.py b/src/DjangoBlog-master(1)/DjangoBlog-master/plugins/seo_optimizer/plugin.py
index b5b19a3..9eaf84c 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/plugins/seo_optimizer/plugin.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/plugins/seo_optimizer/plugin.py
@@ -23,7 +23,7 @@ class SeoOptimizerPlugin(BasePlugin):
description = strip_tags(article.body)[:150]
keywords = ",".join([tag.name for tag in article.tags.all()]) or blog_setting.site_keywords
-
+
meta_tags = f'''
@@ -35,20 +35,29 @@ class SeoOptimizerPlugin(BasePlugin):
'''
for tag in article.tags.all():
- meta_tags += f''
- meta_tags += f''
+ meta_tags += f' \n'
+ meta_tags += f' '
structured_data = {
"@context": "https://schema.org",
"@type": "Article",
- "mainEntityOfPage": {"@type": "WebPage", "@id": request.build_absolute_uri()},
+ "mainEntityOfPage": {
+ "@type": "WebPage",
+ "@id": request.build_absolute_uri()
+ },
"headline": article.title,
"description": description,
"image": request.build_absolute_uri(article.get_first_image_url()),
"datePublished": article.pub_time.isoformat(),
"dateModified": article.last_modify_time.isoformat(),
- "author": {"@type": "Person", "name": article.author.username},
- "publisher": {"@type": "Organization", "name": blog_setting.site_name}
+ "author": {
+ "@type": "Person",
+ "name": article.author.username
+ },
+ "publisher": {
+ "@type": "Organization",
+ "name": blog_setting.site_name
+ }
}
if not structured_data.get("image"):
del structured_data["image"]
@@ -65,7 +74,7 @@ class SeoOptimizerPlugin(BasePlugin):
category_name = context.get('tag_name')
if not category_name:
return None
-
+
category = Category.objects.filter(name=category_name).first()
if not category:
return None
@@ -74,10 +83,23 @@ class SeoOptimizerPlugin(BasePlugin):
description = strip_tags(category.name) or blog_setting.site_description
keywords = category.name
- # BreadcrumbList structured data for category page
- breadcrumb_items = [{"@type": "ListItem", "position": 1, "name": "首页", "item": request.build_absolute_uri('/')}]
- breadcrumb_items.append({"@type": "ListItem", "position": 2, "name": category.name, "item": request.build_absolute_uri()})
-
+ breadcrumb_items = [
+ {
+ "@type": "ListItem",
+ "position": 1,
+ "name": "首页",
+ "item": request.build_absolute_uri('/')
+ }
+ ]
+ breadcrumb_items.append(
+ {
+ "@type": "ListItem",
+ "position": 2,
+ "name": category.name,
+ "item": request.build_absolute_uri()
+ }
+ )
+
structured_data = {
"@context": "https://schema.org",
"@type": "BreadcrumbList",
@@ -93,14 +115,15 @@ class SeoOptimizerPlugin(BasePlugin):
}
def _get_default_seo_data(self, context, request, blog_setting):
- # Homepage and other default pages
structured_data = {
"@context": "https://schema.org",
"@type": "WebSite",
"url": request.build_absolute_uri('/'),
"potentialAction": {
"@type": "SearchAction",
- "target": f"{request.build_absolute_uri('/search/')}?q={{search_term_string}}",
+ "target": (
+ f"{request.build_absolute_uri('/search/')}?q={{search_term_string}}"
+ ),
"query-input": "required name=search_term_string"
}
}
@@ -119,17 +142,21 @@ class SeoOptimizerPlugin(BasePlugin):
view_name = request.resolver_match.view_name
blog_setting = get_blog_setting()
-
+
seo_data = None
if view_name == 'blog:detailbyid':
seo_data = self._get_article_seo_data(context, request, blog_setting)
elif view_name == 'blog:category_detail':
seo_data = self._get_category_seo_data(context, request, blog_setting)
-
+
if not seo_data:
- seo_data = self._get_default_seo_data(context, request, blog_setting)
+ seo_data = self._get_default_seo_data(context, request, blog_setting)
- json_ld_script = f''
+ json_ld_script = (
+ f''
+ )
return f"""
{seo_data.get("title", "")}
@@ -139,4 +166,5 @@ class SeoOptimizerPlugin(BasePlugin):
{json_ld_script}
"""
-plugin = SeoOptimizerPlugin()
+
+plugin = SeoOptimizerPlugin()
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/MemcacheStorage.py b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/MemcacheStorage.py
index 38a7990..119f36a 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/MemcacheStorage.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/MemcacheStorage.py
@@ -7,26 +7,26 @@ from djangoblog.utils import cache
class MemcacheStorage(SessionStorage):
def __init__(self, prefix='ws_'):
self.prefix = prefix
- self.cache = cache
+ self.cache = cache # 保持缓存对象引用
@property
def is_available(self):
- value = "1"
- self.set('checkavaliable', value=value)
- return value == self.get('checkavaliable')
+ test_value = "1" # 修复：明确变量为测试值
+ self.set('check_available', value=test_value) # 修复：命名为check_available（蛇形）
+ return test_value == self.get('check_available')
- def key_name(self, s):
- return '{prefix}{s}'.format(prefix=self.prefix, s=s)
+ def get_key_name(self, session_id): # 修复：方法名改为get_key_name（蛇形），参数明确为session_id
+ return f'{self.prefix}{session_id}' # 使用f-string简化格式
- def get(self, id):
- id = self.key_name(id)
- session_json = self.cache.get(id) or '{}'
+ def get(self, session_id): # 修复：参数名id→session_id（明确为会话ID）
+ cache_key = self.get_key_name(session_id) # 修复：变量名id→cache_key（明确为缓存键）
+ session_json = self.cache.get(cache_key) or '{}'
return json_loads(session_json)
- def set(self, id, value):
- id = self.key_name(id)
- self.cache.set(id, json_dumps(value))
+ def set(self, session_id, value): # 修复：参数名id→session_id
+ cache_key = self.get_key_name(session_id) # 修复：变量名id→cache_key
+ self.cache.set(cache_key, json_dumps(value))
- def delete(self, id):
- id = self.key_name(id)
- self.cache.delete(id)
+ def delete(self, session_id): # 修复：参数名id→session_id
+ cache_key = self.get_key_name(session_id) # 修复：变量名id→cache_key
+ self.cache.delete(cache_key)
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/blogapi.py b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/blogapi.py
index 8a4d6ac..fdf95f2 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/blogapi.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/blogapi.py
@@ -4,24 +4,33 @@ from blog.models import Article, Category
class BlogApi:
+ # 类常量：最大获取数量（替代实例变量，提升复用性）
+ MAX_TAKE_COUNT = 8
+
def __init__(self):
- self.searchqueryset = SearchQuerySet()
- self.searchqueryset.auto_query('')
- self.__max_takecount__ = 8
+ self.search_queryset = SearchQuerySet() # 修复：蛇形命名
+ self.search_queryset.auto_query('')
def search_articles(self, query):
- sqs = self.searchqueryset.auto_query(query)
+ """搜索文章（依赖实例的搜索查询集，保留实例方法）"""
+ sqs = self.search_queryset.auto_query(query)
sqs = sqs.load_all()
- return sqs[:self.__max_takecount__]
+ return sqs[:self.MAX_TAKE_COUNT]
- def get_category_lists(self):
+ @staticmethod
+ def get_category_lists():
+ """获取分类列表（不依赖实例，定义为静态方法）"""
return Category.objects.all()
- def get_category_articles(self, categoryname):
- articles = Article.objects.filter(category__name=categoryname)
+ @classmethod
+ def get_category_articles(cls, category_name):
+ """获取分类下的文章（仅依赖类常量，定义为类方法）"""
+ articles = Article.objects.filter(category__name=category_name)
if articles:
- return articles[:self.__max_takecount__]
+ return articles[:cls.MAX_TAKE_COUNT]
return None
- def get_recent_articles(self):
- return Article.objects.all()[:self.__max_takecount__]
+ @classmethod
+ def get_recent_articles(cls):
+ """获取最新文章（仅依赖类常量，定义为类方法）"""
+ return Article.objects.all()[:cls.MAX_TAKE_COUNT]
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/commonapi.py b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/commonapi.py
index 83ad9ff..f83e944 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/commonapi.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/api/commonapi.py
@@ -3,8 +3,10 @@ import os
import openai
-from servermanager.models import commands
+from servermanager.models import Command # 修复：模型类名应为Command（已在之前修复）
+
+# 初始化日志记录器
logger = logging.getLogger(__name__)
openai.api_key = os.environ.get('OPENAI_API_KEY')
@@ -13,52 +15,60 @@ if os.environ.get('HTTP_PROXY'):
class ChatGPT:
-
@staticmethod
def chat(prompt):
try:
- completion = openai.ChatCompletion.create(model="gpt-3.5-turbo",
- messages=[{"role": "user", "content": prompt}])
+ completion = openai.ChatCompletion.create(
+ model="gpt-3.5-turbo",
+ messages=[{"role": "user", "content": prompt}]
+ )
return completion.choices[0].message.content
except Exception as e:
- logger.error(e)
+ logger.error("ChatGPT调用失败: %s", e)
return "服务器出错了"
class CommandHandler:
def __init__(self):
- self.commands = commands.objects.all()
+ self.commands = Command.objects.all() # 修复：模型类名同步更新
def run(self, title):
"""
运行命令
- :param title: 命令
+ :param title: 命令标题
:return: 返回命令执行结果
"""
- cmd = list(
- filter(
- lambda x: x.title.upper() == title.upper(),
- self.commands))
- if cmd:
- return self.__run_command__(cmd[0].command)
+ # 筛选匹配的命令（不区分大小写）
+ matched_commands = [
+ cmd for cmd in self.commands
+ if cmd.title.upper() == title.upper()
+ ]
+ if matched_commands:
+ return self._run_command(matched_commands[0].command) # 修复：调用改名后的方法
else:
- return "未找到相关命令，请输入hepme获得帮助。"
+ return "未找到相关命令，请输入help获得帮助。" # 修复：拼写错误hepme→help
- def __run_command__(self, cmd):
+ @staticmethod # 修复：无需访问实例，定义为静态方法
+ def _run_command(cmd):
+ """执行系统命令并返回结果（内部辅助方法）"""
try:
res = os.popen(cmd).read()
+ logger.debug("命令执行结果: %s", res)
return res
- except BaseException:
+ except Exception as e:
+ logger.error("命令执行出错: %s", e)
return '命令执行出错!'
def get_help(self):
- rsp = ''
+ """生成命令帮助信息"""
+ help_text = ''
for cmd in self.commands:
- rsp += '{c}:{d}\n'.format(c=cmd.title, d=cmd.describe)
- return rsp
+ help_text += f'{cmd.title}: {cmd.description}\n' # 修复：字段名describe→description
+ return help_text
if __name__ == '__main__':
chatbot = ChatGPT()
prompt = "写一篇1000字关于AI的论文"
- print(chatbot.chat(prompt))
+ result = chatbot.chat(prompt)
+ logger.info("ChatGPT生成结果: %s", result)
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/migrations/0002_alter_emailsendlog_options_and_more.py b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/migrations/0002_alter_emailsendlog_options_and_more.py
index 4858857..72d77b1 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/migrations/0002_alter_emailsendlog_options_and_more.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/migrations/0002_alter_emailsendlog_options_and_more.py
@@ -12,7 +12,11 @@ class Migration(migrations.Migration):
operations = [
migrations.AlterModelOptions(
name='emailsendlog',
- options={'ordering': ['-creation_time'], 'verbose_name': '邮件发送log', 'verbose_name_plural': '邮件发送log'},
+ options={
+ 'ordering': ['-creation_time'],
+ 'verbose_name': '邮件发送日志', # 修复：统一为中文命名
+ 'verbose_name_plural': '邮件发送日志' # 修复：统一为中文命名
+ },
),
migrations.RenameField(
model_name='commands',
@@ -29,4 +33,4 @@ class Migration(migrations.Migration):
old_name='created_time',
new_name='creation_time',
),
- ]
+ ]
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/models.py b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/models.py
index 4326c65..d86db23 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/models.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/models.py
@@ -2,10 +2,10 @@ from django.db import models
# Create your models here.
-class commands(models.Model):
+class Command(models.Model): # 修复：类名改为首字母大写的Command（符合Django模型命名规范）
title = models.CharField('命令标题', max_length=300)
command = models.CharField('命令', max_length=2000)
- describe = models.CharField('命令描述', max_length=300)
+ description = models.CharField('命令描述', max_length=300) # 修复：字段名改为description（蛇形命名，含义明确）
creation_time = models.DateTimeField('创建时间', auto_now_add=True)
last_modify_time = models.DateTimeField('修改时间', auto_now=True)
@@ -18,16 +18,16 @@ class commands(models.Model):
class EmailSendLog(models.Model):
- emailto = models.CharField('收件人', max_length=300)
+ email_to = models.CharField('收件人', max_length=300) # 修复：字段名改为email_to（蛇形命名）
title = models.CharField('邮件标题', max_length=2000)
content = models.TextField('邮件内容')
- send_result = models.BooleanField('结果', default=False)
+ send_result = models.BooleanField('发送结果', default=False) # 修复：verbose_name改为“发送结果”（含义明确）
creation_time = models.DateTimeField('创建时间', auto_now_add=True)
def __str__(self):
return self.title
class Meta:
- verbose_name = '邮件发送log'
+ verbose_name = '邮件发送日志' # 修复：统一为中文命名“邮件发送日志”
verbose_name_plural = verbose_name
- ordering = ['-creation_time']
+ ordering = ['-creation_time']
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/robot.py b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/robot.py
index 3fece0b..604a020 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/robot.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/robot.py
@@ -54,21 +54,21 @@ def search(message, session):
searchstr = message.content.replace('?', '')
result = blogapi.search_articles(searchstr)
if result:
- articles = [x.object for x in result]
- return convert_to_article_reply(articles, message)
+ articles_list = [x.object for x in result] # 修复：变量名更明确
+ return convert_to_article_reply(article_list, message)
return '没有找到相关文章。'
@robot.filter(re.compile(r'^category\s*$', re.I))
def category(message, session):
- categorys = blogapi.get_category_lists()
- return '所有文章分类目录：' + ','.join([x.name for x in categorys])
+ categories = blogapi.get_category_lists() # 修复：单词拼写统一为categories
+ return '所有文章分类目录：' + ','.join([x.name for x in categories])
@robot.filter(re.compile(r'^recent\s*$', re.I))
def recents(message, session):
- articles = blogapi.get_recent_articles()
- return convert_to_article_reply(articles, message) if articles else "暂时还没有文章"
+ articles_list = blogapi.get_recent_articles() # 修复：变量名更明确
+ return convert_to_article_reply(article_list, message) if article_list else "暂时还没有文章"
@robot.filter(re.compile('^help$', re.I))
@@ -101,16 +101,16 @@ def idcard(message, session):
@robot.handler
def echo(message, session):
handler = MessageHandler(message, session)
- return handler.handler()
+ return handler.handle() # 修复：方法名改为handle（符合动词命名规范）
@dataclass
class WxUserInfo:
- """用户信息数据类，替代原类以支持安全序列化"""
- isAdmin: bool = False
- isPasswordSet: bool = False
- Count: int = 0
- Command: str = ''
+ """用户信息数据类，支持安全序列化"""
+ is_admin: bool = False # 修复：使用蛇形命名（统一风格）
+ is_password_set: bool = False # 修复：使用蛇形命名
+ count: int = 0 # 修复：使用蛇形命名且小写开头
+ command: str = '' # 修复：使用蛇形命名
def to_dict(self) -> Dict:
"""转换为字典用于JSON序列化"""
@@ -120,10 +120,10 @@ class WxUserInfo:
def from_dict(cls, data: Dict) -> 'WxUserInfo':
"""从字典恢复对象"""
return cls(
- isAdmin=data.get('isAdmin', False),
- isPasswordSet=data.get('isPasswordSet', False),
- Count=data.get('Count', 0),
- Command=data.get('Command', '')
+ is_admin=data.get('is_admin', False),
+ is_password_set=data.get('is_password_set', False),
+ count=data.get('count', 0),
+ command=data.get('command', '')
)
@@ -131,13 +131,13 @@ class MessageHandler:
def __init__(self, message, session):
self.message = message
self.session = session
- self.userid = message.source
- self.userinfo = self._load_userinfo()
+ self.user_id = message.source # 修复：使用蛇形命名
+ self.user_info = self._load_user_info() # 修复：使用蛇形命名
- def _load_userinfo(self) -> WxUserInfo:
- """加载用户信息（使用JSON替代jsonpickle）"""
+ def _load_user_info(self) -> WxUserInfo:
+ """加载用户信息（使用JSON序列化）"""
try:
- info_str = self.session.get(self.userid, '{}')
+ info_str = self.session.get(self.user_id, '{}')
info_dict = json.loads(info_str)
return WxUserInfo.from_dict(info_dict)
except (json.JSONDecodeError, TypeError, KeyError) as e:
@@ -146,56 +146,59 @@ class MessageHandler:
@property
def is_admin(self):
- return self.userinfo.isAdmin
+ return self.user_info.is_admin
@property
def is_password_set(self):
- return self.userinfo.isPasswordSet
+ return self.user_info.is_password_set
def save_session(self):
- """保存用户信息（使用JSON替代jsonpickle）"""
+ """保存用户信息（使用JSON序列化）"""
try:
- info_str = json.dumps(self.userinfo.to_dict())
- self.session[self.userid] = info_str
+ info_str = json.dumps(self.user_info.to_dict())
+ self.session[self.user_id] = info_str
except json.JSONEncodeError as e:
logger.error(f"保存用户信息失败: {e}")
- def handler(self):
- info = self.message.content
+ def handle(self): # 修复：方法名改为handle（符合动词命名规范）
+ content = self.message.content # 修复：变量名更明确
- if self.userinfo.isAdmin and info.upper() == 'EXIT':
- self.userinfo = WxUserInfo()
+ if self.user_info.is_admin and content.upper() == 'EXIT':
+ self.user_info = WxUserInfo()
self.save_session()
return "退出成功"
- if info.upper() == 'ADMIN':
- self.userinfo.isAdmin = True
+ if content.upper() == 'ADMIN':
+ self.user_info.is_admin = True
self.save_session()
return "输入管理员密码"
- if self.userinfo.isAdmin and not self.userinfo.isPasswordSet:
- passwd = settings.WXADMIN if not settings.TESTING else '123'
- if passwd.upper() == get_sha256(get_sha256(info)).upper():
- self.userinfo.isPasswordSet = True
+ if self.user_info.is_admin and not self.user_info.is_password_set:
+ # 获取管理员密码（测试环境使用默认密码）
+ admin_password = settings.WXADMIN if not settings.TESTING else '123'
+ # 验证密码（双重SHA256加密）
+ if admin_password.upper() == get_sha256(get_sha256(content)).upper():
+ self.user_info.is_password_set = True
self.save_session()
- return "验证通过,请输入命令或者要执行的命令代码:输入helpme获得帮助"
+ return "验证通过,请输入命令或者要执行的命令代码:输入help获得帮助" # 修复：hepme→help
else:
- if self.userinfo.Count >= 3:
- self.userinfo = WxUserInfo()
+ if self.user_info.count >= 3:
+ self.user_info = WxUserInfo()
self.save_session()
return "超过验证次数"
- self.userinfo.Count += 1
+ self.user_info.count += 1
self.save_session()
return "验证失败，请重新输入管理员密码:"
- if self.userinfo.isAdmin and self.userinfo.isPasswordSet:
- if self.userinfo.Command != '' and info.upper() == 'Y':
- return cmd_handler.run(self.userinfo.Command)
+ if self.user_info.is_admin and self.user_info.is_password_set:
+ if self.user_info.command != '' and content.upper() == 'Y':
+ return cmd_handler.run(self.user_info.command)
else:
- if info.upper() == 'HELPME':
+ if content.upper() == 'HELP':
return cmd_handler.get_help()
- self.userinfo.Command = info
+ self.user_info.command = content
self.save_session()
- return f"确认执行: {info} 命令?"
+ return f"确认执行: {content} 命令?"
- return ChatGPT.chat(info)
\ No newline at end of file
+ # 非管理员会话，调用ChatGPT处理
+ return ChatGPT.chat(content)
\ No newline at end of file
diff --git a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/urls.py b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/urls.py
index 8d134d2..30ee091 100644
--- a/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/urls.py
+++ b/src/DjangoBlog-master(1)/DjangoBlog-master/servermanager/urls.py
@@ -5,6 +5,5 @@ from .robot import robot
app_name = "servermanager"
urlpatterns = [
- path(r'robot', make_view(robot)),
-
-]
+ path('robot/', make_view(robot)), # 修复：统一使用单引号，补充路径斜杠
+]
\ No newline at end of file
diff --git a/src/main.py b/src/main.py
deleted file mode 100644
index 75d9766..0000000
--- a/src/main.py
+++ /dev/null
@@ -1 +0,0 @@
-print('hello world')
diff --git a/src/zjp.py b/src/zjp.py
deleted file mode 100644
index e75154b..0000000
--- a/src/zjp.py
+++ /dev/null
@@ -1 +0,0 @@
-print("hello world")
\ No newline at end of file