268 lines
8.3 KiB
Python
268 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
文章API服务器
|
||
提供文章列表获取和文章详情获取的接口
|
||
"""
|
||
|
||
from flask import Blueprint, request, jsonify
|
||
from datetime import datetime, timedelta
|
||
import logging
|
||
from database import db
|
||
from bson import ObjectId
|
||
|
||
# 创建蓝图
|
||
article_bp = Blueprint('article', __name__, url_prefix='/api/article')
|
||
|
||
# 获取数据库集合
|
||
articles_collection = db['articles']
|
||
|
||
def format_time(time_obj):
|
||
"""格式化时间"""
|
||
if not time_obj:
|
||
return ""
|
||
|
||
if isinstance(time_obj, datetime):
|
||
return time_obj.strftime("%Y-%m-%d %H:%M:%S")
|
||
else:
|
||
return str(time_obj)
|
||
|
||
def format_article_item(doc):
|
||
"""格式化文章数据项"""
|
||
return {
|
||
"_id": str(doc.get("_id", "")),
|
||
"title": doc.get("title", ""),
|
||
"author_id": doc.get("author_id", ""),
|
||
"cover_image": doc.get("cover_image", ""),
|
||
"status": doc.get("status", ""),
|
||
"summary": doc.get("summary", ""),
|
||
"created_at": format_time(doc.get("created_at")),
|
||
"likes": doc.get("likes", []),
|
||
"likes_count": len(doc.get("likes", []))
|
||
}
|
||
|
||
def get_article_list(page=1, limit=20, sort_by="created_at", status=None):
|
||
"""获取文章列表（分页）"""
|
||
try:
|
||
# 计算跳过的数量
|
||
skip = (page - 1) * limit
|
||
|
||
# 构建查询条件
|
||
query_condition = {}
|
||
if status:
|
||
query_condition["status"] = status
|
||
|
||
# 设置排序字段
|
||
sort_field = sort_by if sort_by in ["created_at", "title"] else "created_at"
|
||
sort_order = -1 # 降序
|
||
|
||
# 查询数据
|
||
cursor = articles_collection.find(query_condition).sort(sort_field, sort_order).skip(skip).limit(limit)
|
||
docs = list(cursor)
|
||
|
||
# 获取总数
|
||
total = articles_collection.count_documents(query_condition)
|
||
|
||
# 格式化数据
|
||
article_list = []
|
||
for doc in docs:
|
||
item = format_article_item(doc)
|
||
article_list.append(item)
|
||
|
||
return {
|
||
"success": True,
|
||
"data": article_list,
|
||
"pagination": {
|
||
"page": page,
|
||
"limit": limit,
|
||
"total": total,
|
||
"pages": (total + limit - 1) // limit,
|
||
"has_next": page * limit < total,
|
||
"has_prev": page > 1
|
||
},
|
||
"sort_by": sort_by,
|
||
"status_filter": status,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
|
||
except Exception as e:
|
||
logging.error(f"获取文章列表失败: {e}")
|
||
return {"success": False, "message": f"获取数据失败: {str(e)}"}
|
||
|
||
def search_articles(keyword, page=1, limit=10):
|
||
"""搜索文章"""
|
||
try:
|
||
if not keyword:
|
||
return {"success": False, "message": "请提供搜索关键词"}
|
||
|
||
# 计算跳过的数量
|
||
skip = (page - 1) * limit
|
||
|
||
# 构建搜索条件（模糊匹配标题和内容）
|
||
search_condition = {
|
||
"$or": [
|
||
{"title": {"$regex": keyword, "$options": "i"}},
|
||
{"content": {"$regex": keyword, "$options": "i"}},
|
||
{"summary": {"$regex": keyword, "$options": "i"}}
|
||
]
|
||
}
|
||
|
||
# 查询数据
|
||
cursor = articles_collection.find(search_condition).sort("created_at", -1).skip(skip).limit(limit)
|
||
docs = list(cursor)
|
||
|
||
# 获取搜索结果总数
|
||
total = articles_collection.count_documents(search_condition)
|
||
|
||
# 格式化数据
|
||
search_results = []
|
||
for doc in docs:
|
||
item = format_article_item(doc)
|
||
search_results.append(item)
|
||
|
||
return {
|
||
"success": True,
|
||
"data": search_results,
|
||
"keyword": keyword,
|
||
"pagination": {
|
||
"page": page,
|
||
"limit": limit,
|
||
"total": total,
|
||
"pages": (total + limit - 1) // limit,
|
||
"has_next": page * limit < total,
|
||
"has_prev": page > 1
|
||
},
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
|
||
except Exception as e:
|
||
logging.error(f"搜索文章失败: {e}")
|
||
return {"success": False, "message": f"搜索失败: {str(e)}"}
|
||
|
||
def get_article_detail(article_id):
|
||
"""获取文章详情"""
|
||
try:
|
||
# 尝试通过ObjectId查找
|
||
try:
|
||
doc = articles_collection.find_one({"_id": ObjectId(article_id)})
|
||
except:
|
||
# 如果ObjectId无效，尝试其他字段
|
||
doc = articles_collection.find_one({
|
||
"$or": [
|
||
{"title": article_id},
|
||
{"author_id": article_id}
|
||
]
|
||
})
|
||
|
||
if not doc:
|
||
return {"success": False, "message": "未找到文章信息"}
|
||
|
||
# 格式化详细信息
|
||
detail = format_article_item(doc)
|
||
|
||
return {
|
||
"success": True,
|
||
"data": detail,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
|
||
except Exception as e:
|
||
logging.error(f"获取文章详情失败: {e}")
|
||
return {"success": False, "message": f"获取详情失败: {str(e)}"}
|
||
|
||
def get_statistics():
|
||
"""获取统计信息"""
|
||
try:
|
||
# 基本统计
|
||
total_articles = articles_collection.count_documents({})
|
||
|
||
if total_articles == 0:
|
||
return {"success": False, "message": "暂无数据"}
|
||
|
||
# 按状态统计
|
||
status_stats = []
|
||
for status in ["draft", "published", "archived"]:
|
||
count = articles_collection.count_documents({"status": status})
|
||
status_stats.append({"status": status, "count": count})
|
||
|
||
# 获取最新更新时间
|
||
latest_doc = articles_collection.find().sort("created_at", -1).limit(1)
|
||
latest_time = ""
|
||
if latest_doc:
|
||
latest_list = list(latest_doc)
|
||
if latest_list:
|
||
latest_time = format_time(latest_list[0].get("created_at"))
|
||
|
||
return {
|
||
"success": True,
|
||
"data": {
|
||
"total_articles": total_articles,
|
||
"status_stats": status_stats,
|
||
"latest_update": latest_time
|
||
},
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
|
||
except Exception as e:
|
||
logging.error(f"获取统计信息失败: {e}")
|
||
return {"success": False, "message": f"获取统计失败: {str(e)}"}
|
||
|
||
# 路由定义
|
||
@article_bp.route('/list')
|
||
def get_articles():
|
||
"""获取文章列表"""
|
||
page = int(request.args.get('page', 1))
|
||
limit = int(request.args.get('limit', 20))
|
||
sort_by = request.args.get('sort', 'created_at')
|
||
status = request.args.get('status')
|
||
|
||
result = get_article_list(page, limit, sort_by, status)
|
||
return jsonify(result)
|
||
|
||
@article_bp.route('/search')
|
||
def search():
|
||
"""搜索文章"""
|
||
keyword = request.args.get('q', '')
|
||
page = int(request.args.get('page', 1))
|
||
limit = int(request.args.get('limit', 10))
|
||
result = search_articles(keyword, page, limit)
|
||
return jsonify(result)
|
||
|
||
@article_bp.route('/detail')
|
||
def get_detail():
|
||
"""获取文章详情"""
|
||
article_id = request.args.get('id', '')
|
||
result = get_article_detail(article_id)
|
||
return jsonify(result)
|
||
|
||
@article_bp.route('/stats')
|
||
def get_stats():
|
||
"""获取统计信息"""
|
||
result = get_statistics()
|
||
return jsonify(result)
|
||
|
||
@article_bp.route('/health')
|
||
def health_check():
|
||
"""健康检查"""
|
||
try:
|
||
# 检查数据库连接
|
||
total_records = articles_collection.count_documents({})
|
||
|
||
return jsonify({
|
||
"success": True,
|
||
"message": "服务正常",
|
||
"data": {
|
||
"database": "连接正常",
|
||
"total_records": total_records,
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
})
|
||
except Exception as e:
|
||
return jsonify({
|
||
"success": False,
|
||
"message": f"服务异常: {str(e)}",
|
||
"data": {
|
||
"database": "连接失败",
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
}) |