Ubuntu
327 lines
8.8 KiB
Python
327 lines
8.8 KiB
Python
"""
|
|
CMS 数据库操作工具 — 草稿文章查找、内容更新
|
|
|
|
用法:
|
|
from cms_ops import find_draft_by_title, update_article_content
|
|
"""
|
|
|
|
import sqlite3
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict, Any
|
|
|
|
# 数据库路径（相对脚本所在目录）
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
PROJECT_ROOT = os.path.dirname(SCRIPT_DIR) # website-new 根目录
|
|
DB_PATH = os.path.join(PROJECT_ROOT, "backend", "cms.db")
|
|
|
|
|
|
def _get_conn():
|
|
"""获取数据库连接"""
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def find_draft_by_title(keyword: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
按标题关键词查找最新一篇草稿文章。
|
|
|
|
Args:
|
|
keyword: 标题关键词
|
|
|
|
Returns:
|
|
文章字典，未找到返回 None
|
|
"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""SELECT * FROM articles
|
|
WHERE status = 'draft' AND title LIKE ?
|
|
ORDER BY created_at DESC
|
|
LIMIT 1""",
|
|
(f"%{keyword}%",),
|
|
)
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
if row:
|
|
return dict(row)
|
|
return None
|
|
|
|
|
|
def find_draft_by_column(column_name: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
查找某栏目下最新一篇草稿文章。
|
|
|
|
Args:
|
|
column_name: 栏目名称（模糊匹配）
|
|
|
|
Returns:
|
|
文章字典，未找到返回 None
|
|
"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""SELECT articles.* FROM articles
|
|
JOIN columns ON articles.column_id = columns.id
|
|
WHERE articles.status = 'draft' AND columns.name LIKE ?
|
|
ORDER BY articles.created_at DESC
|
|
LIMIT 1""",
|
|
(f"%{column_name}%",),
|
|
)
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
if row:
|
|
return dict(row)
|
|
return None
|
|
|
|
|
|
def find_draft_by_id(article_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
按 ID 查找草稿文章。
|
|
|
|
Args:
|
|
article_id: 文章 ID
|
|
|
|
Returns:
|
|
文章字典，未找到返回 None
|
|
"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT * FROM articles WHERE id = ? AND status = 'draft'",
|
|
(article_id,),
|
|
)
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
if row:
|
|
return dict(row)
|
|
return None
|
|
|
|
|
|
def list_latest_drafts(limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""
|
|
列出最新的 N 篇草稿文章。
|
|
|
|
Args:
|
|
limit: 返回数量
|
|
|
|
Returns:
|
|
文章字典列表
|
|
"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""SELECT id, title, column_id, content_type,
|
|
length(content) as content_len, created_at
|
|
FROM articles
|
|
WHERE status = 'draft'
|
|
ORDER BY created_at DESC
|
|
LIMIT ?""",
|
|
(limit,),
|
|
)
|
|
rows = cursor.fetchall()
|
|
conn.close()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def get_article_full(article_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
获取文章完整内容（不限制状态）。
|
|
|
|
Args:
|
|
article_id: 文章 ID
|
|
|
|
Returns:
|
|
文章字典，包含 content 字段
|
|
"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM articles WHERE id = ?", (article_id,))
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
if row:
|
|
return dict(row)
|
|
return None
|
|
|
|
|
|
def update_article_content(
|
|
article_id: int,
|
|
new_content: str,
|
|
status: str = "published",
|
|
new_title: Optional[str] = None,
|
|
new_author: Optional[str] = None,
|
|
new_cover_image: Optional[str] = None,
|
|
) -> bool:
|
|
"""
|
|
更新文章内容、状态和可选的标题/作者/封面图。
|
|
|
|
Args:
|
|
article_id: 文章 ID
|
|
new_content: 改写后的正文
|
|
status: 新状态，默认 'published'
|
|
new_title: 可选，改写后的标题
|
|
new_author: 可选，改写后的作者
|
|
new_cover_image: 可选，新的封面图路径
|
|
|
|
Returns:
|
|
是否更新成功
|
|
"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
sets = ["content = ?", "status = ?", "updated_at = ?"]
|
|
params = [new_content, status, now]
|
|
|
|
if new_title:
|
|
sets.append("title = ?")
|
|
params.append(new_title)
|
|
|
|
if new_author:
|
|
sets.append("author = ?")
|
|
params.append(new_author)
|
|
|
|
if new_cover_image:
|
|
sets.append("cover_image = ?")
|
|
params.append(new_cover_image)
|
|
|
|
params.append(article_id)
|
|
cursor.execute(
|
|
f"UPDATE articles SET {', '.join(sets)} WHERE id = ?",
|
|
params,
|
|
)
|
|
|
|
conn.commit()
|
|
affected = cursor.rowcount
|
|
conn.close()
|
|
return affected > 0
|
|
|
|
|
|
def _cleanup_cover_file(cover_image: str):
|
|
"""
|
|
删除封面图文件及对应缩略图（与后台 file_utils.cleanup_file_by_path 逻辑对齐）。
|
|
|
|
Args:
|
|
cover_image: 封面图路径（/storage/admin/images/... 或相对路径）
|
|
"""
|
|
if not cover_image or cover_image.startswith(("http://", "https://")):
|
|
return
|
|
|
|
upload_dir = os.path.join(PROJECT_ROOT, "storage", "admin", "images")
|
|
thumbnail_dir = os.path.join(upload_dir, "thumbnails")
|
|
|
|
# 兼容两种路径格式
|
|
if cover_image.startswith("/storage/admin/images/"):
|
|
rel_path = cover_image[len("/storage/admin/images/"):]
|
|
else:
|
|
rel_path = cover_image.lstrip("/")
|
|
|
|
full_path = os.path.join(upload_dir, rel_path)
|
|
|
|
# 删除原文件
|
|
try:
|
|
if os.path.isfile(full_path):
|
|
os.remove(full_path)
|
|
except OSError:
|
|
pass
|
|
|
|
# 同步删除缩略图（与原图同名，扩展名 .jpg）
|
|
try:
|
|
stem = os.path.splitext(os.path.basename(rel_path))[0]
|
|
thumb_path = os.path.join(thumbnail_dir, f"{stem}.jpg")
|
|
if os.path.isfile(thumb_path):
|
|
os.remove(thumb_path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def delete_article(article_id: int) -> bool:
|
|
"""
|
|
删除文章（物理删除），同步清理封面图片文件及缩略图。
|
|
与后台 routers/articles.py 的 delete_article 行为对齐。
|
|
|
|
Args:
|
|
article_id: 文章 ID
|
|
|
|
Returns:
|
|
是否删除成功
|
|
"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
|
|
# 先查 cover_image 以便删除后清理文件
|
|
cursor.execute("SELECT cover_image FROM articles WHERE id = ?", (article_id,))
|
|
row = cursor.fetchone()
|
|
cover_image = row["cover_image"] if row else None
|
|
|
|
cursor.execute("DELETE FROM articles WHERE id = ?", (article_id,))
|
|
conn.commit()
|
|
affected = cursor.rowcount
|
|
conn.close()
|
|
|
|
if affected > 0 and cover_image:
|
|
_cleanup_cover_file(cover_image)
|
|
|
|
return affected > 0
|
|
|
|
|
|
def count_drafts() -> int:
|
|
"""统计草稿文章数量"""
|
|
conn = _get_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT COUNT(*) FROM articles WHERE status = 'draft'")
|
|
count = cursor.fetchone()[0]
|
|
conn.close()
|
|
return count
|
|
|
|
|
|
# ---- 命令行模式 ----
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
if len(sys.argv) < 2:
|
|
print("用法:")
|
|
print(" python cms_ops.py list [limit] 列表最新草稿")
|
|
print(" python cms_ops.py search <keyword> 按标题搜索")
|
|
print(" python cms_ops.py get <id> 获取文章完整内容")
|
|
print(" python cms_ops.py count 统计草稿数量")
|
|
sys.exit(0)
|
|
|
|
cmd = sys.argv[1]
|
|
|
|
if cmd == "list":
|
|
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 10
|
|
drafts = list_latest_drafts(limit)
|
|
print(f"最新 {len(drafts)} 篇草稿:\n")
|
|
for d in drafts:
|
|
title = d["title"][:50]
|
|
print(f" ID={d['id']:<5} | {title:<50s} | {d['content_len']} chars | {d['created_at']}")
|
|
|
|
elif cmd == "search":
|
|
keyword = sys.argv[2]
|
|
article = find_draft_by_title(keyword)
|
|
if article:
|
|
print(f"找到: ID={article['id']} | {article['title']}")
|
|
print(f"内容类型: {article['content_type']}")
|
|
print(f"内容长度: {len(article.get('content', ''))} chars")
|
|
print(f"摘要: {article.get('summary', '')}")
|
|
print(f"\n正文:\n{article.get('content', '')[:500]}...")
|
|
else:
|
|
print(f"未找到匹配 '{keyword}' 的草稿")
|
|
|
|
elif cmd == "get":
|
|
article_id = int(sys.argv[2])
|
|
article = get_article_full(article_id)
|
|
if article:
|
|
print(f"ID={article['id']} | {article['title']}")
|
|
print(f"状态: {article['status']}")
|
|
print(f"内容类型: {article['content_type']}")
|
|
print(f"封面: {article.get('cover_image', '')}")
|
|
print(f"\n正文:\n{article.get('content', '')}")
|
|
else:
|
|
print(f"找不到 ID={article_id} 的文章")
|
|
|
|
elif cmd == "count":
|
|
print(f"草稿文章: {count_drafts()} 篇")
|