Ubuntu
155 lines
5.6 KiB
Python
155 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
图片存储方案迁移脚本
|
|
将图片从旧的层级目录结构迁移到新的扁平结构
|
|
|
|
旧结构: /{module}/{year}/{month}/{filename}
|
|
/{module}/{year}/{month}/thumbnails/thumb_{filename}.jpg
|
|
新结构: /{filename}
|
|
/thumbnails/{filename_stem}.jpg
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import shutil
|
|
from pathlib import Path
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from app.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
# 数据库路径
|
|
DB_PATH = "./cms.db"
|
|
UPLOAD_DIR = Path(settings.UPLOAD_DIR)
|
|
|
|
def migrate_images():
|
|
"""迁移图片路径"""
|
|
engine = create_engine(f"sqlite:///{DB_PATH}")
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
db = SessionLocal()
|
|
|
|
try:
|
|
# 获取所有图片记录
|
|
result = db.execute(text("SELECT id, file_path, thumb_path, filename FROM images"))
|
|
images = result.fetchall()
|
|
|
|
print(f"找到 {len(images)} 条图片记录需要迁移")
|
|
|
|
migrated_count = 0
|
|
skipped_count = 0
|
|
error_count = 0
|
|
|
|
for img in images:
|
|
try:
|
|
img_id = img[0]
|
|
old_file_path = img[1] # e.g., /news/2026/04/20260420_121931_b91c5d3a.jpg
|
|
old_thumb_path = img[2] # e.g., /news/2026/04/thumbnails/thumb_20260420_121931_b91c5d3a.jpg.jpg
|
|
filename = img[3] # e.g., 20260420_121931_b91c5d3a.jpg
|
|
|
|
# 如果已经是新格式，跳过
|
|
if old_file_path.startswith("/") and "/" not in old_file_path.lstrip("/"):
|
|
print(f"跳过 (已是新格式): {old_file_path}")
|
|
skipped_count += 1
|
|
continue
|
|
|
|
# 新的路径格式
|
|
new_file_path = f"/{filename}"
|
|
new_thumb_path = f"/thumbnails/{Path(filename).stem}.jpg" if old_thumb_path else None
|
|
|
|
# 物理文件路径
|
|
old_file_full = UPLOAD_DIR / old_file_path.lstrip("/")
|
|
new_file_full = UPLOAD_DIR / filename
|
|
|
|
# 移动原图文件
|
|
if old_file_full.exists() and old_file_full != new_file_full:
|
|
try:
|
|
shutil.move(str(old_file_full), str(new_file_full))
|
|
print(f"移动文件: {old_file_full} -> {new_file_full}")
|
|
except Exception as e:
|
|
print(f"移动文件失败 {old_file_full}: {e}")
|
|
error_count += 1
|
|
continue
|
|
|
|
# 处理缩略图
|
|
if old_thumb_path:
|
|
old_thumb_full = UPLOAD_DIR / old_thumb_path.lstrip("/")
|
|
new_thumb_full = UPLOAD_DIR / "thumbnails" / f"{Path(filename).stem}.jpg"
|
|
|
|
if old_thumb_full.exists() and old_thumb_full != new_thumb_full:
|
|
try:
|
|
shutil.move(str(old_thumb_full), str(new_thumb_full))
|
|
print(f"移动缩略图: {old_thumb_full} -> {new_thumb_full}")
|
|
except Exception as e:
|
|
print(f"移动缩略图失败 {old_thumb_full}: {e}")
|
|
|
|
# 更新数据库
|
|
db.execute(
|
|
text("UPDATE images SET file_path = :new_path, thumb_path = :new_thumb WHERE id = :id"),
|
|
{"new_path": new_file_path, "new_thumb": new_thumb_path, "id": img_id}
|
|
)
|
|
|
|
migrated_count += 1
|
|
print(f"迁移成功 ID={img_id}: {old_file_path} -> {new_file_path}")
|
|
|
|
except Exception as e:
|
|
print(f"迁移失败 ID={img_id}: {e}")
|
|
error_count += 1
|
|
|
|
db.commit()
|
|
|
|
print("\n" + "="*60)
|
|
print(f"迁移完成!")
|
|
print(f"成功: {migrated_count}")
|
|
print(f"跳过 (已是新格式): {skipped_count}")
|
|
print(f"失败: {error_count}")
|
|
print("="*60)
|
|
|
|
# 清理空目录
|
|
cleanup_empty_dirs(UPLOAD_DIR)
|
|
|
|
except Exception as e:
|
|
print(f"迁移过程出错: {e}")
|
|
db.rollback()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def cleanup_empty_dirs(root_dir: Path):
|
|
"""清理空目录"""
|
|
removed_count = 0
|
|
for dirpath, dirnames, filenames in os.walk(str(root_dir), topdown=False):
|
|
dir_path = Path(dirpath)
|
|
# 跳过根目录和 thumbnails 目录
|
|
if dir_path == root_dir or dir_path.name == "thumbnails":
|
|
continue
|
|
|
|
# 检查是否为空
|
|
try:
|
|
if dir_path.exists() and not any(dir_path.iterdir()):
|
|
dir_path.rmdir()
|
|
print(f"删除空目录: {dir_path}")
|
|
removed_count += 1
|
|
except Exception as e:
|
|
print(f"删除目录失败 {dir_path}: {e}")
|
|
|
|
print(f"清理完成，共删除 {removed_count} 个空目录")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("图片存储方案迁移工具")
|
|
print("="*60)
|
|
print(f"数据库: {DB_PATH}")
|
|
print(f"上传目录: {UPLOAD_DIR}")
|
|
print("="*60)
|
|
|
|
confirm = input("\n确定要开始迁移吗? (yes/no): ")
|
|
if confirm.lower() == "yes":
|
|
migrate_images()
|
|
else:
|
|
print("已取消迁移")
|