初始化技术博客项目
功能特性: - 前端文章展示 (Markdown渲染、代码高亮) - 后台管理系统 (文章、分类、标签、作者管理) - 图片上传 (点击、拖拽、粘贴) - 网站设置 (Logo、底部宣传图片上传) - 访问统计 (IP记录、趋势图表) - 参考来源、附件支持 - 单端口部署 (16012) 技术栈: - Flask 3.0 + SQLAlchemy - SQLite数据库 - marked.js + highlight.js (Markdown渲染) - Chart.js (统计图表)
This commit is contained in:
10
app/services/__init__.py
Normal file
10
app/services/__init__.py
Normal file
@@ -0,0 +1,10 @@
|
||||
"""服务层模块"""
|
||||
from app.services.article_service import ArticleService
|
||||
from app.services.category_service import CategoryService
|
||||
from app.services.tag_service import TagService
|
||||
from app.services.admin_service import AdminService
|
||||
from app.services.site_config_service import SiteConfigService
|
||||
from app.services.visit_service import VisitService
|
||||
from app.services.upload_service import UploadService, AuthorService, ReferenceService, AttachmentService
|
||||
|
||||
__all__ = ['ArticleService', 'CategoryService', 'TagService', 'AdminService', 'SiteConfigService', 'VisitService', 'UploadService', 'AuthorService', 'ReferenceService', 'AttachmentService']
|
||||
48
app/services/admin_service.py
Normal file
48
app/services/admin_service.py
Normal file
@@ -0,0 +1,48 @@
|
||||
"""管理员服务"""
|
||||
from datetime import datetime
|
||||
from app import db
|
||||
from app.models import Admin
|
||||
|
||||
|
||||
class AdminService:
|
||||
"""管理员业务逻辑"""
|
||||
|
||||
@staticmethod
|
||||
def authenticate(username, password):
|
||||
"""验证管理员登录"""
|
||||
admin = Admin.query.filter_by(username=username).first()
|
||||
if admin and admin.is_active and admin.check_password(password):
|
||||
# 更新最后登录时间
|
||||
admin.last_login = datetime.utcnow()
|
||||
db.session.commit()
|
||||
return admin
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(admin_id):
|
||||
"""通过ID获取管理员"""
|
||||
return Admin.query.get(admin_id)
|
||||
|
||||
@staticmethod
|
||||
def get_by_username(username):
|
||||
"""通过用户名获取管理员"""
|
||||
return Admin.query.filter_by(username=username).first()
|
||||
|
||||
@staticmethod
|
||||
def create(username, password, email=None):
|
||||
"""创建管理员"""
|
||||
admin = Admin(username=username, email=email)
|
||||
admin.set_password(password)
|
||||
db.session.add(admin)
|
||||
db.session.commit()
|
||||
return admin
|
||||
|
||||
@staticmethod
|
||||
def update_password(admin_id, old_password, new_password):
|
||||
"""更新密码"""
|
||||
admin = Admin.query.get(admin_id)
|
||||
if admin and admin.check_password(old_password):
|
||||
admin.set_password(new_password)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
128
app/services/article_service.py
Normal file
128
app/services/article_service.py
Normal file
@@ -0,0 +1,128 @@
|
||||
"""文章服务"""
|
||||
from datetime import datetime
|
||||
from slugify import slugify
|
||||
from app import db
|
||||
from app.models import Article, Category, Tag
|
||||
|
||||
|
||||
class ArticleService:
|
||||
"""文章业务逻辑"""
|
||||
|
||||
@staticmethod
|
||||
def get_published_list(page=1, per_page=10, category_id=None, tag_id=None):
|
||||
"""获取已发布文章列表"""
|
||||
query = Article.query.filter_by(is_published=True)
|
||||
|
||||
if category_id:
|
||||
query = query.filter_by(category_id=category_id)
|
||||
if tag_id:
|
||||
query = query.join(Article.tags).filter(Tag.id == tag_id)
|
||||
|
||||
return query.order_by(Article.published_at.desc()).paginate(
|
||||
page=page, per_page=per_page, error_out=False
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_by_slug(slug):
|
||||
"""通过slug获取文章"""
|
||||
article = Article.query.filter_by(slug=slug, is_published=True).first()
|
||||
if article:
|
||||
# 增加浏览次数
|
||||
article.view_count += 1
|
||||
db.session.commit()
|
||||
return article
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(article_id):
|
||||
"""通过ID获取文章"""
|
||||
return Article.query.get(article_id)
|
||||
|
||||
@staticmethod
|
||||
def get_all_list(page=1, per_page=20, is_published=None):
|
||||
"""获取所有文章列表（后台用）"""
|
||||
query = Article.query
|
||||
if is_published is not None:
|
||||
query = query.filter_by(is_published=is_published)
|
||||
return query.order_by(Article.updated_at.desc()).paginate(
|
||||
page=page, per_page=per_page, error_out=False
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create(title, content, summary=None, author_id=None, category_id=None, tag_ids=None,
|
||||
cover_image=None, is_published=False):
|
||||
"""创建文章"""
|
||||
# 生成唯一slug
|
||||
slug = slugify(title)
|
||||
count = Article.query.filter(Article.slug.like(f'{slug}%')).count()
|
||||
if count > 0:
|
||||
slug = f'{slug}-{count + 1}'
|
||||
|
||||
article = Article(
|
||||
title=title,
|
||||
slug=slug,
|
||||
content=content,
|
||||
summary=summary or content[:200] + '...' if len(content) > 200 else content,
|
||||
author_id=author_id,
|
||||
category_id=category_id,
|
||||
cover_image=cover_image,
|
||||
is_published=is_published,
|
||||
published_at=datetime.utcnow() if is_published else None
|
||||
)
|
||||
|
||||
# 添加标签
|
||||
if tag_ids:
|
||||
tags = Tag.query.filter(Tag.id.in_(tag_ids)).all()
|
||||
article.tags = tags
|
||||
|
||||
db.session.add(article)
|
||||
db.session.commit()
|
||||
return article
|
||||
|
||||
@staticmethod
|
||||
def update(article_id, **kwargs):
|
||||
"""更新文章"""
|
||||
article = Article.query.get(article_id)
|
||||
if not article:
|
||||
return None
|
||||
|
||||
# 处理发布状态变化
|
||||
if 'is_published' in kwargs and kwargs['is_published'] and not article.is_published:
|
||||
kwargs['published_at'] = datetime.utcnow()
|
||||
|
||||
# 处理标签
|
||||
tag_ids = kwargs.pop('tag_ids', None)
|
||||
if tag_ids is not None:
|
||||
tags = Tag.query.filter(Tag.id.in_(tag_ids)).all()
|
||||
article.tags = tags
|
||||
|
||||
# 更新其他字段
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(article, key):
|
||||
setattr(article, key, value)
|
||||
|
||||
db.session.commit()
|
||||
return article
|
||||
|
||||
@staticmethod
|
||||
def delete(article_id):
|
||||
"""删除文章"""
|
||||
article = Article.query.get(article_id)
|
||||
if article:
|
||||
db.session.delete(article)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def get_hot_articles(limit=5):
|
||||
"""获取热门文章"""
|
||||
return Article.query.filter_by(is_published=True)\
|
||||
.order_by(Article.view_count.desc())\
|
||||
.limit(limit).all()
|
||||
|
||||
@staticmethod
|
||||
def get_recent_articles(limit=5):
|
||||
"""获取最新文章"""
|
||||
return Article.query.filter_by(is_published=True)\
|
||||
.order_by(Article.published_at.desc())\
|
||||
.limit(limit).all()
|
||||
62
app/services/category_service.py
Normal file
62
app/services/category_service.py
Normal file
@@ -0,0 +1,62 @@
|
||||
"""分类服务"""
|
||||
from app import db
|
||||
from app.models import Category
|
||||
|
||||
|
||||
class CategoryService:
|
||||
"""分类业务逻辑"""
|
||||
|
||||
@staticmethod
|
||||
def get_all():
|
||||
"""获取所有分类"""
|
||||
return Category.query.order_by(Category.sort_order).all()
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(category_id):
|
||||
"""通过ID获取分类"""
|
||||
return Category.query.get(category_id)
|
||||
|
||||
@staticmethod
|
||||
def get_by_slug(slug):
|
||||
"""通过slug获取分类"""
|
||||
return Category.query.filter_by(slug=slug).first()
|
||||
|
||||
@staticmethod
|
||||
def create(name, slug=None, description=None, sort_order=0):
|
||||
"""创建分类"""
|
||||
from slugify import slugify
|
||||
slug = slug or slugify(name)
|
||||
|
||||
category = Category(
|
||||
name=name,
|
||||
slug=slug,
|
||||
description=description,
|
||||
sort_order=sort_order
|
||||
)
|
||||
db.session.add(category)
|
||||
db.session.commit()
|
||||
return category
|
||||
|
||||
@staticmethod
|
||||
def update(category_id, **kwargs):
|
||||
"""更新分类"""
|
||||
category = Category.query.get(category_id)
|
||||
if not category:
|
||||
return None
|
||||
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(category, key):
|
||||
setattr(category, key, value)
|
||||
|
||||
db.session.commit()
|
||||
return category
|
||||
|
||||
@staticmethod
|
||||
def delete(category_id):
|
||||
"""删除分类"""
|
||||
category = Category.query.get(category_id)
|
||||
if category:
|
||||
db.session.delete(category)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
50
app/services/site_config_service.py
Normal file
50
app/services/site_config_service.py
Normal file
@@ -0,0 +1,50 @@
|
||||
"""网站配置服务"""
|
||||
from app import db
|
||||
from app.models import SiteConfig
|
||||
|
||||
|
||||
class SiteConfigService:
|
||||
"""网站配置业务逻辑"""
|
||||
|
||||
@staticmethod
|
||||
def get_config():
|
||||
"""获取网站配置"""
|
||||
return SiteConfig.get_config()
|
||||
|
||||
@staticmethod
|
||||
def update(**kwargs):
|
||||
"""更新网站配置"""
|
||||
config = SiteConfig.get_config()
|
||||
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(config, key):
|
||||
setattr(config, key, value)
|
||||
|
||||
db.session.commit()
|
||||
return config
|
||||
|
||||
@staticmethod
|
||||
def get_dict():
|
||||
"""获取配置字典（用于模板全局变量）"""
|
||||
config = SiteConfig.get_config()
|
||||
return {
|
||||
'site_name': config.site_name,
|
||||
'site_subtitle': config.site_subtitle,
|
||||
'site_logo': config.site_logo,
|
||||
'site_keywords': config.site_keywords,
|
||||
'site_description': config.site_description,
|
||||
'footer_text': config.footer_text,
|
||||
'footer_copyright': config.footer_copyright,
|
||||
'footer_technology': config.footer_technology,
|
||||
'promo_image_1': config.promo_image_1,
|
||||
'promo_image_1_title': config.promo_image_1_title,
|
||||
'promo_image_1_link': config.promo_image_1_link,
|
||||
'promo_image_2': config.promo_image_2,
|
||||
'promo_image_2_title': config.promo_image_2_title,
|
||||
'promo_image_2_link': config.promo_image_2_link,
|
||||
'promo_image_3': config.promo_image_3,
|
||||
'promo_image_3_title': config.promo_image_3_title,
|
||||
'contact_email': config.contact_email,
|
||||
'contact_github': config.contact_github,
|
||||
'articles_per_page': config.articles_per_page,
|
||||
}
|
||||
57
app/services/tag_service.py
Normal file
57
app/services/tag_service.py
Normal file
@@ -0,0 +1,57 @@
|
||||
"""标签服务"""
|
||||
from app import db
|
||||
from app.models import Tag
|
||||
|
||||
|
||||
class TagService:
|
||||
"""标签业务逻辑"""
|
||||
|
||||
@staticmethod
|
||||
def get_all():
|
||||
"""获取所有标签"""
|
||||
return Tag.query.order_by(Tag.name).all()
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(tag_id):
|
||||
"""通过ID获取标签"""
|
||||
return Tag.query.get(tag_id)
|
||||
|
||||
@staticmethod
|
||||
def get_by_slug(slug):
|
||||
"""通过slug获取标签"""
|
||||
return Tag.query.filter_by(slug=slug).first()
|
||||
|
||||
@staticmethod
|
||||
def create(name, slug=None, color='#007bff'):
|
||||
"""创建标签"""
|
||||
from slugify import slugify
|
||||
slug = slug or slugify(name)
|
||||
|
||||
tag = Tag(name=name, slug=slug, color=color)
|
||||
db.session.add(tag)
|
||||
db.session.commit()
|
||||
return tag
|
||||
|
||||
@staticmethod
|
||||
def update(tag_id, **kwargs):
|
||||
"""更新标签"""
|
||||
tag = Tag.query.get(tag_id)
|
||||
if not tag:
|
||||
return None
|
||||
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(tag, key):
|
||||
setattr(tag, key, value)
|
||||
|
||||
db.session.commit()
|
||||
return tag
|
||||
|
||||
@staticmethod
|
||||
def delete(tag_id):
|
||||
"""删除标签"""
|
||||
tag = Tag.query.get(tag_id)
|
||||
if tag:
|
||||
db.session.delete(tag)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
255
app/services/upload_service.py
Normal file
255
app/services/upload_service.py
Normal file
@@ -0,0 +1,255 @@
|
||||
"""文章扩展服务"""
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from werkzeug.utils import secure_filename
|
||||
from app import db
|
||||
from app.models import ArticleAuthor, Reference, Attachment, UploadedImage, Article
|
||||
|
||||
|
||||
class UploadService:
|
||||
"""文件上传服务"""
|
||||
|
||||
UPLOAD_FOLDER = 'uploads'
|
||||
ALLOWED_IMAGE_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'webpo', 'svg'}
|
||||
ALLOWED_FILE_EXTENSIONS = {'pdf', 'doc', 'docx', 'xls', 'xlsx', 'zip', 'rar', 'tar', 'gz', 'txt', 'md'}
|
||||
MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB
|
||||
|
||||
@staticmethod
|
||||
def get_upload_folder():
|
||||
"""获取上传目录路径"""
|
||||
from flask import current_app
|
||||
upload_folder = os.path.join(current_app.root_path, '..', UploadService.UPLOAD_FOLDER)
|
||||
os.makedirs(upload_folder, exist_ok=True)
|
||||
return upload_folder
|
||||
|
||||
@staticmethod
|
||||
def allowed_image(filename):
|
||||
"""检查是否是允许的图片格式"""
|
||||
return '.' in filename and \
|
||||
filename.rsplit('.', 1)[1].lower() in UploadService.ALLOWED_IMAGE_EXTENSIONS
|
||||
|
||||
@staticmethod
|
||||
def allowed_file(filename):
|
||||
"""检查是否是允许的文件格式"""
|
||||
ext = filename.rsplit('.', 1)[1].lower() if '.' in filename else ''
|
||||
return ext in UploadService.ALLOWED_IMAGE_EXTENSIONS or \
|
||||
ext in UploadService.ALLOWED_FILE_EXTENSIONS
|
||||
|
||||
@staticmethod
|
||||
def upload_image(file, article_id=None, uploader='admin'):
|
||||
"""上传图片"""
|
||||
if not file or not UploadService.allowed_image(file.filename):
|
||||
return None, '不支持的图片格式'
|
||||
|
||||
# 检查文件大小
|
||||
file.seek(0, os.SEEK_END)
|
||||
size = file.tell()
|
||||
file.seek(0)
|
||||
if size > UploadService.MAX_FILE_SIZE:
|
||||
return None, '文件大小超过限制（最大10MB）'
|
||||
|
||||
# 生成唯一文件名
|
||||
ext = file.filename.rsplit('.', 1)[1].lower()
|
||||
filename = f'{datetime.now().strftime("%Y%m%d")}_{uuid.uuid4().hex[:8]}.{ext}'
|
||||
|
||||
# 保存文件
|
||||
filepath = os.path.join(UploadService.get_upload_folder(), filename)
|
||||
file.save(filepath)
|
||||
|
||||
# 记录到数据库
|
||||
image = UploadedImage(
|
||||
filename=filename,
|
||||
filepath=filepath,
|
||||
article_id=article_id,
|
||||
uploader=uploader
|
||||
)
|
||||
db.session.add(image)
|
||||
db.session.commit()
|
||||
|
||||
return image, None
|
||||
|
||||
@staticmethod
|
||||
def upload_attachment(file, article_id, description=None):
|
||||
"""上传附件"""
|
||||
if not file or not UploadService.allowed_file(file.filename):
|
||||
return None, '不支持的文件格式'
|
||||
|
||||
# 检查文件大小
|
||||
file.seek(0, os.SEEK_END)
|
||||
size = file.tell()
|
||||
file.seek(0)
|
||||
if size > UploadService.MAX_FILE_SIZE:
|
||||
return None, '文件大小超过限制（最大10MB）'
|
||||
|
||||
# 生成唯一文件名
|
||||
original_name = secure_filename(file.filename)
|
||||
ext = original_name.rsplit('.', 1)[1].lower() if '.' in original_name else ''
|
||||
filename = f'{datetime.now().strftime("%Y%m%d")}_{uuid.uuid4().hex[:8]}_{original_name}'
|
||||
|
||||
# 保存文件
|
||||
filepath = os.path.join(UploadService.get_upload_folder(), filename)
|
||||
file.save(filepath)
|
||||
|
||||
# 记录到数据库
|
||||
attachment = Attachment(
|
||||
article_id=article_id,
|
||||
filename=original_name,
|
||||
filepath=filepath,
|
||||
filesize=size,
|
||||
filetype=ext,
|
||||
description=description
|
||||
)
|
||||
db.session.add(attachment)
|
||||
db.session.commit()
|
||||
|
||||
return attachment, None
|
||||
|
||||
@staticmethod
|
||||
def delete_image(image_id):
|
||||
"""删除图片"""
|
||||
image = UploadedImage.query.get(image_id)
|
||||
if image:
|
||||
# 删除文件
|
||||
if os.path.exists(image.filepath):
|
||||
os.remove(image.filepath)
|
||||
# 删除记录
|
||||
db.session.delete(image)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def delete_attachment(attachment_id):
|
||||
"""删除附件"""
|
||||
attachment = Attachment.query.get(attachment_id)
|
||||
if attachment:
|
||||
# 删除文件
|
||||
if os.path.exists(attachment.filepath):
|
||||
os.remove(attachment.filepath)
|
||||
# 删除记录
|
||||
db.session.delete(attachment)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def get_image_url(filename):
|
||||
"""获取图片URL（相对路径）"""
|
||||
return f'/uploads/{filename}'
|
||||
|
||||
@staticmethod
|
||||
def get_image_url(filename):
|
||||
"""获取图片URL"""
|
||||
return f'/uploads/{filename}'
|
||||
|
||||
|
||||
class AuthorService:
|
||||
"""作者服务"""
|
||||
|
||||
@staticmethod
|
||||
def get_all():
|
||||
"""获取所有作者"""
|
||||
return ArticleAuthor.query.order_by(ArticleAuthor.name).all()
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(author_id):
|
||||
"""通过ID获取作者"""
|
||||
return ArticleAuthor.query.get(author_id)
|
||||
|
||||
@staticmethod
|
||||
def create(name, avatar=None, bio=None, website=None):
|
||||
"""创建作者"""
|
||||
author = ArticleAuthor(name=name, avatar=avatar, bio=bio, website=website)
|
||||
db.session.add(author)
|
||||
db.session.commit()
|
||||
return author
|
||||
|
||||
@staticmethod
|
||||
def update(author_id, **kwargs):
|
||||
"""更新作者"""
|
||||
author = ArticleAuthor.query.get(author_id)
|
||||
if author:
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(author, key):
|
||||
setattr(author, key, value)
|
||||
db.session.commit()
|
||||
return author
|
||||
|
||||
@staticmethod
|
||||
def delete(author_id):
|
||||
"""删除作者"""
|
||||
author = ArticleAuthor.query.get(author_id)
|
||||
if author:
|
||||
db.session.delete(author)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class ReferenceService:
|
||||
"""参考来源服务"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_article(article_id):
|
||||
"""获取文章的参考来源"""
|
||||
return Reference.query.filter_by(article_id=article_id).order_by(Reference.created_at).all()
|
||||
|
||||
@staticmethod
|
||||
def add(article_id, title, url=None, description=None):
|
||||
"""添加参考来源"""
|
||||
ref = Reference(article_id=article_id, title=title, url=url, description=description)
|
||||
db.session.add(ref)
|
||||
db.session.commit()
|
||||
return ref
|
||||
|
||||
@staticmethod
|
||||
def update(reference_id, **kwargs):
|
||||
"""更新参考来源"""
|
||||
ref = Reference.query.get(reference_id)
|
||||
if ref:
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(ref, key):
|
||||
setattr(ref, key, value)
|
||||
db.session.commit()
|
||||
return ref
|
||||
|
||||
@staticmethod
|
||||
def delete(reference_id):
|
||||
"""删除参考来源"""
|
||||
ref = Reference.query.get(reference_id)
|
||||
if ref:
|
||||
db.session.delete(ref)
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def update_article_references(article_id, references_data):
|
||||
"""批量更新文章参考来源"""
|
||||
# 删除旧的参考来源
|
||||
Reference.query.filter_by(article_id=article_id).delete()
|
||||
|
||||
# 添加新的参考来源
|
||||
for ref_data in references_data:
|
||||
if ref_data.get('title'): # 只添加有标题的
|
||||
ReferenceService.add(
|
||||
article_id=article_id,
|
||||
title=ref_data.get('title'),
|
||||
url=ref_data.get('url'),
|
||||
description=ref_data.get('description')
|
||||
)
|
||||
|
||||
|
||||
class AttachmentService:
|
||||
"""附件服务"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_article(article_id):
|
||||
"""获取文章附件"""
|
||||
return Attachment.query.filter_by(article_id=article_id).order_by(Attachment.upload_time).all()
|
||||
|
||||
@staticmethod
|
||||
def delete_attachment(attachment_id):
|
||||
"""删除附件"""
|
||||
return UploadService.delete_attachment(attachment_id)
|
||||
162
app/services/visit_service.py
Normal file
162
app/services/visit_service.py
Normal file
@@ -0,0 +1,162 @@
|
||||
"""访问统计服务"""
|
||||
from datetime import datetime, date, timedelta
|
||||
from app import db
|
||||
from app.models import VisitLog, DailyStats, Article
|
||||
|
||||
|
||||
class VisitService:
|
||||
"""访问统计业务逻辑"""
|
||||
|
||||
@staticmethod
|
||||
def log_visit(request, article_id=None):
|
||||
"""记录访问日志"""
|
||||
from flask import request as flask_request
|
||||
|
||||
visit = VisitLog(
|
||||
ip_address=VisitService.get_client_ip(flask_request),
|
||||
user_agent=flask_request.user_agent.string if flask_request.user_agent else None,
|
||||
referer=flask_request.referrer,
|
||||
path=flask_request.path,
|
||||
method=flask_request.method,
|
||||
article_id=article_id
|
||||
)
|
||||
|
||||
db.session.add(visit)
|
||||
db.session.commit()
|
||||
|
||||
# 更新每日统计
|
||||
VisitService.update_daily_stats()
|
||||
|
||||
return visit
|
||||
|
||||
@staticmethod
|
||||
def get_client_ip(request):
|
||||
"""获取客户端真实IP"""
|
||||
# 检查代理头
|
||||
if request.headers.get('X-Forwarded-For'):
|
||||
return request.headers.get('X-Forwarded-For').split(',')[0].strip()
|
||||
elif request.headers.get('X-Real-IP'):
|
||||
return request.headers.get('X-Real-IP')
|
||||
else:
|
||||
return request.remote_addr or '0.0.0.0'
|
||||
|
||||
@staticmethod
|
||||
def update_daily_stats():
|
||||
"""更新每日统计"""
|
||||
today = date.today()
|
||||
stats = DailyStats.query.filter_by(date=today).first()
|
||||
|
||||
if not stats:
|
||||
stats = DailyStats(date=today)
|
||||
db.session.add(stats)
|
||||
|
||||
# 计算今日数据
|
||||
today_start = datetime.combine(today, datetime.min.time())
|
||||
today_end = datetime.combine(today, datetime.max.time())
|
||||
|
||||
stats.total_visits = VisitLog.query.filter(
|
||||
VisitLog.visited_at >= today_start,
|
||||
VisitLog.visited_at <= today_end
|
||||
).count()
|
||||
|
||||
stats.unique_ips = db.session.query(
|
||||
db.func.count(db.func.distinct(VisitLog.ip_address))
|
||||
).filter(
|
||||
VisitLog.visited_at >= today_start,
|
||||
VisitLog.visited_at <= today_end
|
||||
).scalar()
|
||||
|
||||
stats.article_views = VisitLog.query.filter(
|
||||
VisitLog.visited_at >= today_start,
|
||||
VisitLog.visited_at <= today_end,
|
||||
VisitLog.article_id.isnot(None)
|
||||
).count()
|
||||
|
||||
# 找出今日最热门文章
|
||||
top_article_result = db.session.query(
|
||||
VisitLog.article_id,
|
||||
db.func.count(VisitLog.id).label('views')
|
||||
).filter(
|
||||
VisitLog.visited_at >= today_start,
|
||||
VisitLog.visited_at <= today_end,
|
||||
VisitLog.article_id.isnot(None)
|
||||
).group_by(VisitLog.article_id).order_by(db.desc('views')).first()
|
||||
|
||||
if top_article_result:
|
||||
stats.top_article_id = top_article_result[0]
|
||||
stats.top_article_views = top_article_result[1]
|
||||
|
||||
db.session.commit()
|
||||
return stats
|
||||
|
||||
@staticmethod
|
||||
def get_stats_summary():
|
||||
"""获取统计摘要"""
|
||||
return {
|
||||
'total_visits': VisitLog.get_total_count(),
|
||||
'today_visits': VisitLog.get_today_count(),
|
||||
'unique_ips': VisitLog.get_unique_ip_count(),
|
||||
'daily_stats': DailyStats.get_stats_list(30)
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def get_visit_list(page=1, per_page=50, path_filter=None):
|
||||
"""获取访问日志列表"""
|
||||
query = VisitLog.query
|
||||
|
||||
if path_filter:
|
||||
query = query.filter(VisitLog.path.like(f'%{path_filter}%'))
|
||||
|
||||
return query.order_by(VisitLog.visited_at.desc()).paginate(
|
||||
page=page, per_page=per_page, error_out=False
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_top_ips(limit=10):
|
||||
"""获取访问最多的IP"""
|
||||
from sqlalchemy import func
|
||||
|
||||
result = db.session.query(
|
||||
VisitLog.ip_address,
|
||||
func.count(VisitLog.id).label('visit_count'),
|
||||
func.min(VisitLog.visited_at).label('first_visit'),
|
||||
func.max(VisitLog.visited_at).label('last_visit')
|
||||
).group_by(VisitLog.ip_address).order_by(db.desc('visit_count')).limit(limit).all()
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def get_top_articles(days=7, limit=10):
|
||||
"""获取热门文章"""
|
||||
start_date = date.today() - timedelta(days=days)
|
||||
start_datetime = datetime.combine(start_date, datetime.min.time())
|
||||
|
||||
result = db.session.query(
|
||||
Article,
|
||||
db.func.count(VisitLog.id).label('views')
|
||||
).join(VisitLog).filter(
|
||||
VisitLog.visited_at >= start_datetime,
|
||||
VisitLog.article_id.isnot(None)
|
||||
).group_by(Article.id).order_by(db.desc('views')).limit(limit).all()
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def get_top_paths(limit=20):
|
||||
"""获取访问最多的路径"""
|
||||
from sqlalchemy import func
|
||||
|
||||
result = db.session.query(
|
||||
VisitLog.path,
|
||||
func.count(VisitLog.id).label('visit_count')
|
||||
).group_by(VisitLog.path).order_by(db.desc('visit_count')).limit(limit).all()
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def clear_old_logs(days=90):
|
||||
"""清理旧日志"""
|
||||
cutoff_date = datetime.now() - timedelta(days=days)
|
||||
deleted = VisitLog.query.filter(VisitLog.visited_at < cutoff_date).delete()
|
||||
db.session.commit()
|
||||
return deleted
|
||||
Reference in New Issue
Block a user