feat: 图片迁移脚本

scripts/QUICK_START.md

@@ -0,0 +1,186 @@
+# 图片迁移脚本快速开始指南
+
+## 概述
+这是一个通过SSH将本地图片迁移到远程服务器的Python脚本。支持保持相对路径结构，自动创建目录，批量处理图片文件。
+
+## 环境要求
+- Python 3.6+
+- paramiko库 (SSH客户端库)
+
+## 快速开始
+
+### 1. 安装依赖
+```bash
+cd /var/www/html/orico-official-website/scripts
+pip install -r requirements.txt
+```
+
+### 2. 创建配置文件
+```bash
+# 创建示例配置文件
+python image_migrate.py --create-config config.json
+```
+
+编辑生成的 `config.json` 文件，填写目标服务器信息：
+- 修改 `target.ssh` 部分的目标服务器地址、用户名、密码/密钥
+- 设置 `source.base_dir` 为本地图片目录路径
+
+### 3. 准备图片路径列表
+创建 `images.txt` 文件，每行一个相对路径：
+```txt
+uploads/products/2024/01/product1.jpg
+uploads/products/2024/01/product2.jpg
+assets/images/logo.png
+uploads/banners/main-banner.jpg
+```
+
+### 4. 执行迁移
+```bash
+python image_migrate.py --config config.json --input images.txt --verbose
+```
+
+## 配置文件示例（本地源）
+```json
+{
+    "source": {
+        "type": "local",
+        "base_dir": "/var/www/html/orico-official-website/public/uploads"
+    },
+    "target": {
+        "type": "ssh",
+        "base_dir": "/var/www/html/images",
+        "ssh": {
+            "host": "your-server.com",
+            "port": 22,
+            "username": "deploy",
+            "password": "your_password",
+            "key_file": "/path/to/private/key"
+        }
+    }
+}
+```
+
+## 常用命令
+
+### 基本迁移
+```bash
+python image_migrate.py --config config.json --input images.txt
+```
+
+### 详细输出模式
+```bash
+python image_migrate.py --config config.json --input images.txt --verbose
+```
+
+### 覆盖已存在的文件
+```bash
+python image_migrate.py --config config.json --input images.txt --overwrite
+```
+
+### 直接从命令行指定文件
+```bash
+python image_migrate.py --config config.json \
+    "uploads/test1.jpg" \
+    "uploads/test2.jpg" \
+    "assets/logo.png"
+```
+
+### 不使用配置文件，直接指定参数
+```bash
+python image_migrate.py \
+    --source-type local \
+    --source-dir /path/to/local/images \
+    --target-host server.example.com \
+    --target-user username \
+    --target-dir /remote/path/images \
+    --input images.txt \
+    --verbose
+```
+
+## 常见场景
+
+### 场景1：迁移网站上传目录
+```bash
+# 配置文件中的源目录设置为：
+"source": {
+    "type": "local",
+    "base_dir": "/var/www/html/orico-official-website/public/uploads"
+}
+
+# 执行迁移
+python image_migrate.py --config config.json --input uploads_list.txt
+```
+
+### 场景2：从数据库导出路径并迁移
+```bash
+# 从数据库导出图片路径
+mysql -u username -p database -e "SELECT image_path FROM products" --skip-column-names > products.txt
+
+# 清理路径（如果需要）
+sed -i 's/^\.\///' products.txt
+
+# 迁移图片
+python image_migrate.py --config config.json --input products.txt --verbose
+```
+
+### 场景3：迁移整个目录结构
+```bash
+# 使用find命令生成所有图片文件列表
+find /path/to/local/images -type f \( -name "*.jpg" -o -name "*.png" -o -name "*.gif" -o -name "*.webp" \) | sed 's|^/path/to/local/images/||' > all_images.txt
+
+# 迁移所有图片
+python image_migrate.py --config config.json --input all_images.txt
+```
+
+## 路径说明
+- **相对路径**：相对于配置文件中指定的 `base_dir`
+- **示例**：如果 `base_dir` 是 `/var/www/html/images`，相对路径 `uploads/test.jpg` 对应完整路径 `/var/www/html/images/uploads/test.jpg`
+- **目标路径**：在目标服务器上保持相同的相对路径结构
+
+## 故障排查
+
+### 1. 连接失败
+```bash
+# 测试SSH连接
+ssh -p 22 username@server.example.com
+
+# 检查配置文件中的主机、端口、用户名是否正确
+```
+
+### 2. 文件不存在
+```bash
+# 检查本地文件是否存在
+ls -la /var/www/html/orico-official-website/public/uploads/uploads/test.jpg
+
+# 使用详细模式查看完整路径
+python image_migrate.py --config config.json --input images.txt --verbose
+```
+
+### 3. 权限不足
+```bash
+# 检查本地文件读取权限
+ls -la /path/to/local/images
+
+# 检查目标目录写入权限（通过SSH）
+ssh username@server.example.com "ls -la /remote/path/images"
+```
+
+## 注意事项
+1. 建议使用SSH密钥认证而非密码（更安全）
+2. 首次使用前，建议用少量文件测试
+3. 大文件传输可能需要较长时间，建议分批处理
+4. 使用 `--verbose` 参数可查看详细传输信息
+5. 脚本会自动创建目标服务器上的目录结构
+
+## 获取帮助
+```bash
+# 查看完整帮助
+python image_migrate.py --help
+
+# 查看详细使用说明
+cat README.md
+```
+
+---
+*快速开始指南更新日期：2024年*
+*脚本位置：/var/www/html/orico-official-website/scripts/*

scripts/README.md

@@ -0,0 +1,415 @@
+# 图片迁移脚本 (Image Migration Script)
+
+从本地目录或SSH服务器迁移图片到远程SSH服务器，保持相对路径结构一致。
+
+## 功能特性
+
+- ✅ **灵活的源支持**：支持本地目录(`local`)和SSH服务器(`ssh`)作为源
+- ✅ **远程目标**：通过SSH迁移到远程服务器
+- ✅ **路径保持**：在目标服务器上保持相同的相对路径和文件名
+- ✅ **配置灵活**：支持配置文件和命令行参数
+- ✅ **目录自动创建**：递归创建目标目录结构
+- ✅ **重复文件处理**：默认跳过已存在文件，支持覆盖选项
+- ✅ **进度统计**：显示详细的传输进度和统计信息
+- ✅ **错误处理**：完善的连接管理和错误处理
+
+## 环境要求
+
+- Python 3.6+
+- paramiko 库（用于SSH连接）
+
+## 快速开始
+
+### 1. 安装依赖
+
+```bash
+cd /var/www/html/orico-official-website/scripts
+pip install -r requirements.txt
+```
+
+### 2. 创建配置文件
+
+```bash
+# 创建示例配置文件
+python image_migrate.py --create-config config.json
+
+# 编辑配置文件，填写真实的服务器信息
+vim config.json
+```
+
+### 3. 准备图片路径列表
+
+创建一个文本文件（如 `images.txt`），每行一个相对路径：
+
+```
+uploads/products/2024/01/product1.jpg
+uploads/products/2024/01/product2.jpg
+assets/images/logo.png
+```
+
+### 4. 执行迁移
+
+#### 从本地目录迁移到远程服务器（最常用）
+
+```bash
+# 使用配置文件迁移（源为本地目录）
+python image_migrate.py --config config.json --input images.txt
+
+# 或直接指定参数迁移（源为本地目录）
+python image_migrate.py \
+  --source-type local --source-dir /path/to/local/images \
+  --target-host dst.example.com --target-user user2 --target-dir /home/user/images \
+  --input images.txt --verbose
+```
+
+#### 从SSH服务器迁移到另一个SSH服务器
+
+```bash
+# 使用配置文件迁移（源为SSH服务器）
+python image_migrate.py --config config.json --input images.txt
+
+# 或直接指定参数迁移（源为SSH服务器）
+python image_migrate.py \
+  --source-type ssh --source-host src.example.com --source-user user1 --source-dir /var/www/images \
+  --target-host dst.example.com --target-user user2 --target-dir /home/user/images \
+  --input images.txt --verbose
+```
+
+## 配置文件格式
+
+配置文件为JSON格式，支持两种源类型配置：
+
+### 本地目录作为源（推荐）
+
+```json
+{
+    "source": {
+        "type": "local",
+        "base_dir": "/path/to/local/images"
+    },
+    "target": {
+        "type": "ssh",
+        "base_dir": "/var/www/html/images",
+        "ssh": {
+            "host": "target-server.example.com",
+            "port": 22,
+            "username": "username",
+            "password": "your_password_here",
+            "key_file": "/path/to/private/key"
+        }
+    }
+}
+```
+
+### SSH服务器作为源
+
+```json
+{
+    "source": {
+        "type": "ssh",
+        "base_dir": "/var/www/html/images",
+        "ssh": {
+            "host": "source-server.example.com",
+            "port": 22,
+            "username": "username",
+            "password": "your_password_here",
+            "key_file": "/path/to/private/key"
+        }
+    },
+    "target": {
+        "type": "ssh",
+        "base_dir": "/var/www/html/images",
+        "ssh": {
+            "host": "target-server.example.com",
+            "port": 22,
+            "username": "username",
+            "password": "your_password_here",
+            "key_file": "/path/to/private/key"
+        }
+    }
+}
+```
+
+**注意：**
+
+- `type`：可以是 `"local"`（本地目录）或 `"ssh"`（SSH服务器）
+- `base_dir`：图片的基础目录，图片的相对路径将相对于这个目录
+- 对于SSH类型，需要提供`ssh`配置部分
+- `password` 和 `key_file` 二选一即可
+- 目标服务器**必须**是SSH类型
+
+## 命令行参数
+
+### 主要选项
+
+| 参数                   | 说明                   |
+| ---------------------- | ---------------------- |
+| `--config FILE`        | 使用配置文件           |
+| `--create-config FILE` | 创建示例配置文件       |
+| `--input FILE`         | 包含图片路径列表的文件 |
+| `--verbose`, `-v`      | 显示详细输出           |
+| `--overwrite`          | 覆盖已存在的文件       |
+
+### 源服务器参数（可覆盖配置文件）
+
+| 参数                     | 说明                                                         |
+| ------------------------ | ------------------------------------------------------------ |
+| `--source-type TYPE`     | 源类型: `local`(本地目录) 或 `ssh`(SSH服务器)，默认: `local` |
+| `--source-dir DIR`       | **必需** 源服务器图片基础目录                                |
+| `--source-host HOST`     | 源服务器地址（仅SSH类型需要）                                |
+| `--source-port PORT`     | 源服务器端口（仅SSH类型，默认: 22）                          |
+| `--source-user USER`     | 源服务器用户名（仅SSH类型需要）                              |
+| `--source-password PASS` | 源服务器密码（仅SSH类型需要）                                |
+| `--source-key FILE`      | 源服务器私钥文件路径（仅SSH类型需要）                        |
+
+### 目标服务器参数（可覆盖配置文件）
+
+| 参数                     | 说明                                 |
+| ------------------------ | ------------------------------------ |
+| `--target-host HOST`     | **必需** 目标服务器地址（必须是SSH） |
+| `--target-port PORT`     | 目标服务器端口（默认: 22）           |
+| `--target-user USER`     | **必需** 目标服务器用户名            |
+| `--target-password PASS` | 目标服务器密码                       |
+| `--target-key FILE`      | 目标服务器私钥文件路径               |
+| `--target-dir DIR`       | **必需** 目标服务器图片基础目录      |
+
+## 使用示例
+
+### 示例1：从本地目录迁移到远程服务器（推荐）
+
+```bash
+# 1. 创建配置文件
+python image_migrate.py --create-config myconfig.json
+
+# 2. 编辑配置文件
+# 将 source.type 设置为 "local"
+# 设置 source.base_dir 为本地目录路径
+# 填写目标服务器的SSH信息
+
+# 3. 创建图片路径文件
+echo "uploads/test.jpg" > images.txt
+echo "assets/logo.png" >> images.txt
+
+# 4. 执行迁移
+python image_migrate.py --config myconfig.json --input images.txt
+```
+
+### 示例2：直接从命令行迁移本地图片
+
+```bash
+python image_migrate.py \
+  --source-type local \
+  --source-dir /home/user/my_website/images \
+  --target-host 192.168.1.200 \
+  --target-user deploy \
+  --target-key ~/.ssh/id_rsa \
+  --target-dir /var/www/html/images \
+  --input images.txt \
+  --verbose
+```
+
+### 示例3：从SSH服务器迁移到另一个SSH服务器
+
+```bash
+python image_migrate.py \
+  --source-type ssh \
+  --source-host old-server.example.com \
+  --source-user admin \
+  --source-key ~/.ssh/id_rsa \
+  --source-dir /var/www/images \
+  --target-host new-server.example.com \
+  --target-user deploy \
+  --target-dir /home/deploy/images \
+  --input images.txt \
+  --overwrite
+```
+
+### 示例4：直接从命令行指定图片路径
+
+```bash
+python image_migrate.py --config config.json \
+  "uploads/product1.jpg" \
+  "uploads/product2.jpg" \
+  "assets/logo.png"
+```
+
+### 示例5：批量迁移数据库导出的图片路径
+
+```bash
+# 从数据库导出图片路径（假设每行一个路径）
+mysql -e "SELECT image_path FROM products" --skip-column-names > products.txt
+
+# 清理路径（如果需要）
+sed -i 's/^\.\///' products.txt
+
+# 迁移图片
+python image_migrate.py --config config.json --input products.txt --verbose
+```
+
+## 图片路径格式
+
+图片路径应为**相对路径**，相对于配置文件中指定的 `base_dir`：
+
+```
+# 正确示例（相对路径）
+uploads/products/2024/01/product1.jpg
+assets/images/logo.png
+
+# 错误示例（绝对路径）
+/var/www/html/images/uploads/product1.jpg  # 错误！应该使用相对路径
+
+# 在目标服务器上，文件将保存为：
+# 目标 base_dir + 相对路径
+```
+
+## 工作流程
+
+1. **连接建立**：连接到目标SSH服务器（源为本地时不需要连接）
+2. **路径处理**：为每个图片构建完整源路径和目标路径
+3. **目录创建**：在目标服务器上创建必要的目录结构
+4. **文件检查**：检查源文件是否存在，目标文件是否已存在（跳过）
+5. **文件传输**：通过本地临时文件传输（源→本地临时文件→目标）
+6. **统计报告**：显示传输结果和统计信息
+
+## 故障排除
+
+### 常见问题
+
+#### 1. 连接失败
+
+```
+错误: 连接失败: [Errno 111] Connection refused
+```
+
+- 检查目标服务器地址和端口
+- 确认SSH服务正在运行
+- 检查防火墙设置
+
+#### 2. 认证失败
+
+```
+错误: Authentication failed.
+```
+
+- 确认用户名和密码正确
+- 检查SSH密钥权限（chmod 600 ~/.ssh/id_rsa）
+- 确认密钥已添加到目标服务器的 authorized_keys
+
+#### 3. 源文件不存在
+
+```
+错误: 源文件不存在: /path/to/local/images/uploads/test.jpg
+```
+
+- 确认相对路径正确
+- 确认源基础目录正确
+- 使用 `--verbose` 查看完整路径
+
+#### 4. 权限不足
+
+```
+错误: Permission denied
+```
+
+- 确认用户有读取源文件的权限（本地文件）
+- 确认用户有写入目标目录的权限
+
+#### 5. 目录创建失败
+
+```
+错误: 无法创建目标目录: /var/www/html/images/uploads
+```
+
+- 检查目标目录的写入权限
+- 确认目标服务器磁盘空间充足
+
+### 调试技巧
+
+1. **使用详细模式**：添加 `--verbose` 参数查看详细信息
+2. **测试连接**：手动SSH连接测试目标服务器是否可达
+3. **检查路径**：在源目录上确认文件存在
+4. **查看日志**：脚本会显示详细的错误信息
+
+## 性能优化
+
+### 大文件传输
+
+- 脚本使用流式传输，适用于大文件
+- 临时文件存储在系统临时目录，传输完成后自动清理
+
+### 批量传输
+
+- 对于大量文件，建议分批处理
+- 可以使用多个图片路径文件，分别执行迁移
+
+### 网络优化
+
+- 确保到目标服务器的网络连接稳定
+- 考虑使用内网传输以提高速度
+
+## 安全注意事项
+
+1. **认证方式**：建议使用SSH密钥认证而非密码
+2. **权限控制**：配置文件建议设置权限 `chmod 600 config.json`
+3. **最小权限**：使用具有最小必要权限的账户执行迁移
+4. **本地文件安全**：确保本地目录不包含敏感信息
+5. **临时文件清理**：脚本会自动清理临时传输文件
+
+## 扩展和定制
+
+### 修改脚本
+
+- 可以在 `ImageMigrator` 类中添加额外的功能
+- 支持断点续传、并行传输等高级功能
+
+### 集成到工作流
+
+- 可以与其他脚本结合使用
+- 可以作为CI/CD流程的一部分
+
+### 添加进度条
+
+```python
+# 可选：安装 tqdm 库
+pip install tqdm
+
+# 在脚本中添加进度条显示
+```
+
+## 文件结构
+
+```
+scripts/
+├── image_migrate.py          # 主脚本
+├── image_migrate_backup.py   # 原始版本备份
+├── requirements.txt          # Python依赖
+├── README.md                # 本文档
+└── examples/                # 示例文件
+    ├── config.example.json  # 配置文件示例
+    └── images.example.txt   # 图片路径示例
+```
+
+## 版本更新
+
+### v2.0 更新
+
+- ✅ 新增本地目录(`local`)作为源的支持
+- ✅ 默认源类型改为`local`
+- ✅ 简化配置文件结构
+- ✅ 改进错误处理和临时文件管理
+
+## 许可证
+
+本脚本根据项目需要自由使用和修改。
+
+## 支持
+
+如有问题，请：
+
+1. 检查本文档的故障排除部分
+2. 使用 `--verbose` 参数获取详细信息
+3. 确认服务器配置正确
+
+---
+
+**提示**：首次使用前，建议在测试服务器上进行小规模测试。

scripts/examples/config.example.json

@@ -0,0 +1,24 @@
+{
+    "source": {
+        "type": "local",
+        "base_dir": "/var/www/html/orico-official-website/public/storage",
+        "ssh": {
+            "host": "source-server.example.com",
+            "port": 22,
+            "username": "username",
+            "password": "your_password_here",
+            "key_file": "/path/to/private/key"
+        }
+    },
+    "target": {
+        "type": "ssh",
+        "base_dir": "/www/wwwroot/orico-official-website/public/storage",
+        "ssh": {
+            "host": "47.91.149.172",
+            "port": 22,
+            "username": "root",
+            "password": "Orico666tx5d",
+            "key_file": ""
+        }
+    }
+}

scripts/examples/images.example.txt

@@ -0,0 +1,12 @@
+# 图片路径示例文件
+# 每行一个相对路径（相对于配置文件中指定的基础目录）
+# 空行和以 # 开头的行会被忽略
+# 路径使用正斜杠 /，即使在Windows系统上
+
+# 注意：这些路径都是相对路径
+# 实际文件在源服务器上的完整路径是：源基础目录 + 相对路径
+# 例如：如果源基础目录是 /var/www/html/images
+# 那么 uploads/products/2024/01/product1.jpg 对应的完整路径是：
+# /var/www/html/images/uploads/products/2024/01/product1.jpg
+
+images/article/logo.png

scripts/image_migrate.py

@@ -0,0 +1,910 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+图片迁移脚本 - 通过SSH或本地目录迁移图片到远程服务器
+保持相对路径结构一致
+"""
+
+import argparse
+import json
+import os
+import shutil
+import sys
+import tempfile
+import time
+from typing import Any, Dict, List, Optional, Tuple
+
+# paramiko 导入被延迟到实际需要时
+paramiko: Any = None
+
+
+def ensure_paramiko() -> None:
+    """确保paramiko已导入，如果未安装则报错"""
+    global paramiko
+    if paramiko is None:
+        try:
+            import paramiko as paramiko_module
+
+            paramiko = paramiko_module
+        except ImportError:
+            print("错误: 需要安装paramiko库")
+            print("请运行: pip install paramiko")
+            sys.exit(1)
+
+
+class ServerConfig:
+    """服务器配置（支持SSH或本地目录）"""
+
+    def __init__(
+        self,
+        config_type: str = "ssh",  # "ssh" 或 "local"
+        host: str = "",
+        port: int = 22,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        key_file: Optional[str] = None,
+        base_dir: Optional[str] = None,
+    ):
+        """
+        初始化服务器配置
+
+        Args:
+            config_type: 配置类型，'ssh' 表示SSH服务器，'local' 表示本地目录
+            host: 服务器地址（对于local类型可为None）
+            port: 服务器端口（仅SSH类型）
+            username: 用户名（仅SSH类型）
+            password: 密码（仅SSH类型）
+            key_file: 私钥文件路径（仅SSH类型）
+            base_dir: 基础目录路径
+        """
+        self.config_type = config_type
+        self.host = host
+        self.port = port
+        self.username = username
+        self.password = password
+        self.key_file = key_file
+        self.base_dir = base_dir.rstrip("/") if base_dir else ""
+
+    def to_dict(self) -> Dict:
+        """转换为字典"""
+        return {
+            "type": self.config_type,
+            "host": self.host,
+            "port": self.port,
+            "username": self.username,
+            "password": self.password,
+            "key_file": self.key_file,
+            "base_dir": self.base_dir,
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> "ServerConfig":
+        """从字典创建"""
+        ssh_data = data.get("ssh", {})
+        return cls(
+            config_type=data.get("type", "ssh"),
+            host=ssh_data.get("host", "localhost"),
+            port=ssh_data.get("port", 22),
+            username=ssh_data.get("username", ""),
+            password=ssh_data.get("password", ""),
+            key_file=ssh_data.get("key_file", ""),
+            base_dir=data.get("base_dir", ""),
+        )
+
+    def is_local(self) -> bool:
+        """是否是本地目录配置"""
+        return self.config_type == "local"
+
+    def __str__(self) -> str:
+        if self.is_local():
+            return f"LocalDirectory(base_dir={self.base_dir})"
+        else:
+            return f"SSHServer(host={self.host}:{self.port}, user={self.username}, base_dir={self.base_dir})"
+
+
+class ImageMigrator:
+    """图片迁移器"""
+
+    def __init__(
+        self,
+        source_config: ServerConfig,
+        target_config: ServerConfig,
+        verbose: bool = False,
+        overwrite: bool = False,
+    ) -> None:
+        """
+        初始化迁移器
+
+        Args:
+            source_config: 源服务器配置（可以是本地目录或SSH服务器）
+            target_config: 目标服务器配置（必须为SSH服务器）
+            verbose: 是否显示详细输出
+            overwrite: 是否覆盖已存在的文件
+        """
+        self.source_config = source_config
+        self.target_config = target_config
+        self.verbose = verbose
+        self.overwrite = overwrite
+
+        # SSH客户端连接（仅用于SSH类型的服务器）
+        self.source_client: Any = None
+        self.target_client: Any = None
+        self.source_sftp: Any = None
+        self.target_sftp: Any = None
+
+        # 统计信息
+        self.stats = {
+            "total": 0,
+            "success": 0,
+            "failed": 0,
+            "skipped": 0,
+            "bytes_transferred": 0,
+        }
+
+    def connect(self) -> bool:
+        """连接到服务器"""
+        try:
+            # 确保paramiko已导入
+            ensure_paramiko()
+
+            # 连接目标服务器（必须为SSH）
+            if self.target_config.is_local():
+                print("错误: 目标服务器必须为SSH服务器，不能是本地目录")
+                return False
+
+            self.target_client = paramiko.SSHClient()
+            self.target_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+            if self.verbose:
+                print(
+                    f"连接到目标服务器: {self.target_config.host}:{self.target_config.port}"
+                )
+
+            if self.target_config.key_file:
+                key = paramiko.RSAKey.from_private_key_file(self.target_config.key_file)
+
+                self.target_client.connect(
+                    hostname=self.target_config.host,
+                    port=self.target_config.port,
+                    username=self.target_config.username,
+                    pkey=key,
+                )
+            else:
+                self.target_client.connect(
+                    hostname=self.target_config.host,
+                    port=self.target_config.port,
+                    username=self.target_config.username,
+                    password=self.target_config.password,
+                )
+
+            self.target_sftp = self.target_client.open_sftp()
+
+            # 检查SFTP服务器的工作目录
+            if self.verbose:
+                try:
+                    cwd = self.target_sftp.getcwd()
+                    print(f"DEBUG: 目标SFTP服务器当前工作目录: {cwd}")
+                except Exception as e:
+                    print(f"DEBUG: 无法获取目标SFTP服务器工作目录: {e}")
+
+            # 连接源服务器（如果是SSH类型）
+            if not self.source_config.is_local():
+                self.source_client = paramiko.SSHClient()
+                self.source_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+                if self.verbose:
+                    print(
+                        f"连接到源服务器: {self.source_config.host}:{self.source_config.port}"
+                    )
+
+                if self.source_config.key_file:
+                    key = paramiko.RSAKey.from_private_key_file(
+                        self.source_config.key_file
+                    )
+
+                    self.source_client.connect(
+                        hostname=self.source_config.host,
+                        port=self.source_config.port,
+                        username=self.source_config.username,
+                        pkey=key,
+                    )
+                else:
+                    self.source_client.connect(
+                        hostname=self.source_config.host,
+                        port=self.source_config.port,
+                        username=self.source_config.username,
+                        password=self.source_config.password,
+                    )
+
+                self.source_sftp = self.source_client.open_sftp()
+
+            if self.verbose:
+                source_type = (
+                    "本地目录" if self.source_config.is_local() else "SSH服务器"
+                )
+                print(f"连接成功! 源: {source_type}, 目标: SSH服务器")
+            return True
+
+        except Exception as e:
+            print(f"连接失败: {e}")
+            self.close()
+            return False
+
+    def close(self) -> None:
+        """关闭所有连接"""
+        if self.source_sftp:
+            self.source_sftp.close()
+        if self.source_client:
+            self.source_client.close()
+        if self.target_sftp:
+            self.target_sftp.close()
+        if self.target_client:
+            self.target_client.close()
+
+        if self.verbose:
+            print("连接已关闭")
+
+    def ensure_target_directory(self, remote_dir: str) -> bool:
+        """确保目标目录存在（递归创建）"""
+        try:
+            if self.verbose:
+                print(f"DEBUG ensure_target_directory: 检查/创建目录: {remote_dir}")
+
+            # 确保路径以斜杠开头（绝对路径）
+            if not remote_dir.startswith("/"):
+                remote_dir = "/" + remote_dir
+                if self.verbose:
+                    print(
+                        f"DEBUG ensure_target_directory: 转换为绝对路径: {remote_dir}"
+                    )
+
+            # 首先检查目录是否已经存在
+            try:
+                # 尝试列出目录内容
+                self.target_sftp.listdir(remote_dir)
+                if self.verbose:
+                    print(f"DEBUG ensure_target_directory: 目录已存在: {remote_dir}")
+                return True
+            except (FileNotFoundError, IOError):
+                # 目录不存在，需要递归创建
+                if self.verbose:
+                    print(
+                        f"DEBUG ensure_target_directory: 目录不存在，开始递归创建: {remote_dir}"
+                    )
+
+                # 分割路径为各个组件，保留空字符串以处理根目录
+                parts = [p for p in remote_dir.split("/") if p]
+
+                # 从根目录开始逐级创建
+                current_path = ""
+                for i, part in enumerate(parts):
+                    # 构建当前路径（确保以斜杠开头）
+                    if current_path:
+                        current_path = f"{current_path}/{part}"
+                    else:
+                        current_path = f"/{part}"
+
+                    # 检查当前路径是否存在
+                    try:
+                        self.target_sftp.listdir(current_path)
+                        if self.verbose:
+                            print(
+                                f"DEBUG ensure_target_directory: 路径已存在: {current_path}"
+                            )
+                    except (FileNotFoundError, IOError):
+                        # 当前路径不存在，尝试创建
+                        if self.verbose:
+                            print(
+                                f"DEBUG ensure_target_directory: 创建目录: {current_path}"
+                            )
+                        try:
+                            self.target_sftp.mkdir(current_path)
+                            if self.verbose:
+                                print(
+                                    f"DEBUG ensure_target_directory: 目录创建成功: {current_path}"
+                                )
+                        except Exception as mkdir_e:
+                            # 如果创建失败，可能是权限问题或目录已存在
+                            if self.verbose:
+                                print(
+                                    f"DEBUG ensure_target_directory: 创建目录时出错 {current_path}: {mkdir_e}"
+                                )
+                            # 尝试检查目录是否真的不存在
+                            try:
+                                self.target_sftp.listdir(current_path)
+                                if self.verbose:
+                                    print(
+                                        f"DEBUG ensure_target_directory: 目录实际上已存在: {current_path}"
+                                    )
+                            except:
+                                # 目录确实不存在且创建失败
+                                print(f"错误: 无法创建目录 {current_path}: {mkdir_e}")
+                                return False
+
+                # 最终确认目录是否创建成功
+                try:
+                    self.target_sftp.listdir(remote_dir)
+                    if self.verbose:
+                        print(
+                            f"DEBUG ensure_target_directory: 目录创建完成: {remote_dir}"
+                        )
+                    return True
+                except Exception as final_check_e:
+                    if self.verbose:
+                        print(
+                            f"DEBUG ensure_target_directory: 最终检查失败 {remote_dir}: {final_check_e}"
+                        )
+                    return False
+        except Exception as e:
+            print(f"创建目录失败 {remote_dir}: {e}")
+            if self.verbose:
+                import traceback
+
+                traceback.print_exc()
+            return False
+
+    def get_source_file_info(self, source_path: str) -> Tuple[bool, Optional[int]]:
+        """
+        获取源文件信息
+
+        Returns:
+            Tuple[exists: bool, size: Optional[int]]
+        """
+        try:
+            if self.source_config.is_local():
+                # 本地文件
+                if self.verbose:
+                    print(f"DEBUG: 检查本地文件路径: {source_path}")
+                    print(f"DEBUG: 当前工作目录: {os.getcwd()}")
+                    print(f"DEBUG: 绝对路径: {os.path.abspath(source_path)}")
+                    print(f"DEBUG: 文件是否存在: {os.path.exists(source_path)}")
+
+                if os.path.exists(source_path):
+                    if self.verbose:
+                        print(f"DEBUG: 文件大小: {os.path.getsize(source_path)} bytes")
+                    return True, os.path.getsize(source_path)
+                else:
+                    if self.verbose:
+                        print(f"DEBUG: 文件不存在!")
+                    return False, 0
+            else:
+                # SSH远程文件
+                source_attr = self.source_sftp.stat(source_path)
+                return True, source_attr.st_size
+
+        except Exception as e:
+            if self.verbose:
+                print(f"DEBUG: 获取文件信息时出错: {e}")
+            return False, 0
+
+    def read_source_file(self, source_path: str, temp_path: str) -> bool:
+        """
+        读取源文件到临时文件
+
+        Returns:
+            bool: 是否成功
+        """
+        try:
+            if self.source_config.is_local():
+                # 从本地文件复制
+                if self.verbose:
+                    print(f"DEBUG read_source_file: 开始复制文件")
+                    print(f"DEBUG read_source_file: 源路径: {source_path}")
+                    print(f"DEBUG read_source_file: 临时路径: {temp_path}")
+                    print(
+                        f"DEBUG read_source_file: 源文件是否存在: {os.path.exists(source_path)}"
+                    )
+                    print(
+                        f"DEBUG read_source_file: 源文件绝对路径: {os.path.abspath(source_path)}"
+                    )
+
+                shutil.copy2(source_path, temp_path)
+
+                if self.verbose:
+                    print(f"DEBUG read_source_file: 复制完成")
+                    print(
+                        f"DEBUG read_source_file: 临时文件是否存在: {os.path.exists(temp_path)}"
+                    )
+                    if os.path.exists(temp_path):
+                        print(
+                            f"DEBUG read_source_file: 临时文件大小: {os.path.getsize(temp_path)} bytes"
+                        )
+
+                return True
+            else:
+                # 从SSH服务器下载
+                self.source_sftp.get(source_path, temp_path)
+
+                return True
+        except Exception as e:
+            print(f"读取源文件失败 {source_path}: {e}")
+            import traceback
+
+            traceback.print_exc()
+            return False
+
+    def transfer_file(self, relative_path: str) -> bool:
+        """
+        传输单个文件
+
+        Args:
+            relative_path: 相对路径（相对于基础目录）
+
+        Returns:
+            bool: 是否成功
+        """
+        try:
+            # 构建完整路径
+            source_path = f"{self.source_config.base_dir}/{relative_path}"
+            target_path = f"{self.target_config.base_dir}/{relative_path}"
+
+            if self.verbose:
+                source_type = "本地" if self.source_config.is_local() else "远程"
+                print(f"传输: {relative_path}")
+                print(f"  源({source_type}): {source_path}")
+                print(f"  目标(远程): {target_path}")
+
+            # 检查源文件是否存在并获取大小
+            source_exists, file_size = self.get_source_file_info(source_path)
+            if not source_exists:
+                print(f"错误: 源文件不存在: {source_path}")
+                self.stats["failed"] += 1
+                return False
+
+            # 检查目标文件是否已存在（根据overwrite选项处理）
+            try:
+                self.target_sftp.stat(target_path)
+
+                if not self.overwrite:
+                    if self.verbose:
+                        print(f"跳过: 目标文件已存在: {target_path}")
+                    self.stats["skipped"] += 1
+                    return True
+                else:
+                    if self.verbose:
+                        print(f"覆盖: 目标文件已存在: {target_path}")
+            except FileNotFoundError:
+                if self.verbose:
+                    print(f"DEBUG: 目标文件不存在，将创建新文件: {target_path}")
+            except Exception as e:
+                if self.verbose:
+                    print(f"DEBUG: 检查目标文件时出错: {e}")
+
+            # 确保目标目录存在
+            target_dir = os.path.dirname(target_path)
+            if self.verbose:
+                print(f"DEBUG: 目标目录: {target_dir}")
+                print(f"DEBUG: 目标路径: {target_path}")
+            if target_dir and not self.ensure_target_directory(target_dir):
+                print(f"错误: 无法创建目标目录: {target_dir}")
+                self.stats["failed"] += 1
+                return False
+
+            # 传输文件
+            start_time = time.time()
+            temp_path = None
+
+            try:
+                # 创建临时文件
+                with tempfile.NamedTemporaryFile(delete=False) as temp_file:
+                    temp_path = temp_file.name
+
+                if self.verbose:
+                    print(f"DEBUG: 创建临时文件: {temp_path}")
+
+                # 读取源文件到临时文件
+                if not self.read_source_file(source_path, temp_path):
+                    self.stats["failed"] += 1
+                    return False
+
+                # 上传到目标服务器
+                if self.verbose:
+                    print(f"DEBUG: 开始上传到目标服务器: {temp_path} -> {target_path}")
+                    print(f"DEBUG: 临时文件大小: {os.path.getsize(temp_path)} bytes")
+                    print(f"DEBUG: 临时文件是否存在: {os.path.exists(temp_path)}")
+                    # 尝试列出目标目录内容
+                    try:
+                        target_dir = os.path.dirname(target_path)
+                        dir_list = self.target_sftp.listdir(target_dir)
+                        print(f"DEBUG: 目标目录内容: {dir_list}")
+                    except Exception as e:
+                        print(f"DEBUG: 无法列出目标目录内容 {target_dir}: {e}")
+                        # 尝试检查目录是否存在
+                        try:
+                            self.target_sftp.stat(target_dir)
+                            print(f"DEBUG: 但目录stat成功: {target_dir}")
+                        except Exception as stat_e:
+                            print(f"DEBUG: 目录stat也失败: {stat_e}")
+
+                self.target_sftp.put(temp_path, target_path)
+
+                if self.verbose:
+                    print(f"DEBUG: 上传完成")
+
+            except Exception as e:
+                if self.verbose:
+                    print(f"DEBUG: 上传过程中出错: {e}")
+                    import traceback
+
+                    traceback.print_exc()
+                raise
+            finally:
+                # 清理临时文件
+                if temp_path and os.path.exists(temp_path):
+                    try:
+                        if self.verbose:
+                            print(f"DEBUG: 清理临时文件: {temp_path}")
+                        os.remove(temp_path)
+                    except Exception as e:
+                        if self.verbose:
+                            print(f"DEBUG: 清理临时文件时出错: {e}")
+
+            # 记录统计
+            transfer_time = time.time() - start_time
+            speed = (file_size or 0) / transfer_time / 1024 if transfer_time > 0 else 0
+
+            if self.verbose:
+                print(
+                    f"  完成: {file_size} bytes, 耗时: {transfer_time:.2f}s, 速度: {speed:.2f} KB/s"
+                )
+
+            self.stats["success"] += 1
+            self.stats["bytes_transferred"] += file_size or 0
+            return True
+
+        except Exception as e:
+            print(f"传输失败 {relative_path}: {e}")
+            if self.verbose:
+                import traceback
+
+                traceback.print_exc()
+            self.stats["failed"] += 1
+            return False
+
+    def migrate_images(self, image_paths: List[str]) -> Dict[str, int]:
+        """
+        迁移图片列表
+
+        Args:
+            image_paths: 相对路径列表
+
+        Returns:
+            Dict: 统计信息
+        """
+        if not self.connect():
+            return self.stats
+
+        try:
+            self.stats["total"] = len(image_paths)
+
+            source_type = "本地目录" if self.source_config.is_local() else "SSH服务器"
+            print(f"开始从{source_type}迁移 {len(image_paths)} 个图片到SSH服务器...")
+
+            for i, relative_path in enumerate(image_paths, 1):
+                # 清理路径
+                relative_path = relative_path.strip()
+                if not relative_path:
+                    continue
+
+                # 显示进度
+                print(f"[{i}/{len(image_paths)}] {relative_path}", end="")
+
+                # 传输文件
+                success = self.transfer_file(relative_path)
+
+                if success:
+                    print(" ✓")
+                else:
+                    print(" ✗")
+
+            # 打印统计信息
+            print("\n" + "=" * 50)
+            print("迁移完成!")
+            print(f"总计: {self.stats['total']}")
+            print(f"成功: {self.stats['success']}")
+            print(f"失败: {self.stats['failed']}")
+            print(f"跳过: {self.stats['skipped']}")
+            print(f"传输字节: {self.stats['bytes_transferred']:,}")
+
+            return self.stats
+
+        finally:
+            self.close()
+
+
+def load_config(config_file: str) -> Dict:
+    """加载配置文件"""
+    try:
+        with open(config_file, "r", encoding="utf-8") as f:
+            config = json.load(f)
+        return config
+    except Exception as e:
+        print(f"加载配置文件失败 {config_file}: {e}")
+        return {}
+
+
+def save_config(config_file: str, config: Dict):
+    """保存配置文件"""
+    try:
+        with open(config_file, "w", encoding="utf-8") as f:
+            json.dump(config, f, indent=2, ensure_ascii=False)
+        print(f"配置文件已保存: {config_file}")
+    except Exception as e:
+        print(f"保存配置文件失败 {config_file}: {e}")
+
+
+def create_example_config(config_file: str):
+    """创建示例配置文件"""
+    example_config = {
+        "source": {
+            "type": "local",  # 可以是 "local" 或 "ssh"
+            "base_dir": "/path/to/local/images",
+            # 如果是SSH类型，需要以下配置
+            "ssh": {
+                "host": "source-server.example.com",
+                "port": 22,
+                "username": "username",
+                "password": "your_password_here",
+                "key_file": "/path/to/private/key",
+            },
+        },
+        "target": {
+            "type": "ssh",  # 目标必须是SSH类型
+            "base_dir": "/var/www/html/images",
+            "ssh": {
+                "host": "target-server.example.com",
+                "port": 22,
+                "username": "username",
+                "password": "your_password_here",
+                "key_file": "/path/to/private/key",
+            },
+        },
+    }
+
+    save_config(config_file, example_config)
+    print("已创建示例配置文件")
+    print("注意: 源服务器可以是本地目录(local)或SSH服务器(ssh)")
+    print("      目标服务器必须是SSH服务器(ssh)")
+
+
+def read_image_paths(image_list_file: str) -> List[str]:
+    """从文件读取图片路径列表"""
+    paths = []
+    try:
+        with open(image_list_file, "r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if line and not line.startswith("#"):
+                    paths.append(line)
+        print(f"从文件读取 {len(paths)} 个图片路径")
+    except Exception as e:
+        print(f"读取图片路径文件失败 {image_list_file}: {e}")
+    return paths
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="从本地目录或SSH服务器迁移图片到远程SSH服务器，保持相对路径结构",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+示例:
+  # 使用配置文件迁移（源为本地目录）
+  python image_migrate.py --config config.json --input images.txt
+
+  # 使用配置文件迁移（源为SSH服务器）
+  # 配置文件中 source.type="ssh"
+
+  # 直接指定参数迁移（源为本地目录）
+  python image_migrate.py \\
+    --source-type local --source-dir /path/to/local/images \\
+    --target-host dst.example.com --target-user user2 --target-dir /home/user/images \\
+    --input images.txt
+
+  # 直接指定参数迁移（源为SSH服务器）
+  python image_migrate.py \\
+    --source-type ssh --source-host src.example.com --source-user user1 --source-dir /var/www/images \\
+    --target-host dst.example.com --target-user user2 --target-dir /home/user/images \\
+    --input images.txt
+
+  # 创建示例配置文件
+  python image_migrate.py --create-config config.json
+
+  # 从命令行直接指定图片路径
+  python image_migrate.py --config config.json images/product1.jpg images/product2.jpg
+        """,
+    )
+
+    # 配置文件选项
+    parser.add_argument("--config", help="配置文件路径")
+    parser.add_argument("--create-config", metavar="FILE", help="创建示例配置文件")
+
+    # 源服务器选项
+    parser.add_argument(
+        "--source-type",
+        choices=["local", "ssh"],
+        default="local",
+        help="源服务器类型: local(本地目录) 或 ssh(SSH服务器)，默认: local",
+    )
+    parser.add_argument("--source-host", help="源服务器地址（仅SSH类型需要）")
+    parser.add_argument(
+        "--source-port",
+        type=int,
+        default=22,
+        help="源服务器端口（仅SSH类型，默认: 22）",
+    )
+    parser.add_argument("--source-user", help="源服务器用户名（仅SSH类型需要）")
+    parser.add_argument("--source-password", help="源服务器密码（仅SSH类型需要）")
+    parser.add_argument("--source-key", help="源服务器私钥文件路径（仅SSH类型需要）")
+    parser.add_argument("--source-dir", help="源服务器图片基础目录")
+
+    # 目标服务器选项
+    parser.add_argument("--target-host", help="目标服务器地址")
+    parser.add_argument(
+        "--target-port", type=int, default=22, help="目标服务器端口（默认: 22）"
+    )
+    parser.add_argument("--target-user", help="目标服务器用户名")
+    parser.add_argument("--target-password", help="目标服务器密码")
+    parser.add_argument("--target-key", help="目标服务器私钥文件路径")
+    parser.add_argument("--target-dir", help="目标服务器图片基础目录")
+
+    # 输入选项
+    parser.add_argument(
+        "--input", "-i", help="包含图片路径列表的文件（每行一个相对路径）"
+    )
+    parser.add_argument("--verbose", "-v", action="store_true", help="显示详细输出")
+    parser.add_argument("--overwrite", action="store_true", help="覆盖已存在的文件")
+
+    # 直接传递图片路径
+    parser.add_argument("image_paths", nargs="*", help="直接指定图片相对路径")
+
+    args = parser.parse_args()
+
+    # 创建示例配置文件
+    if args.create_config:
+        create_example_config(args.create_config)
+        return
+
+    # 检查必要参数（如果不是创建配置文件）
+    if not args.config and not (
+        args.source_dir and args.target_host and args.target_user and args.target_dir
+    ):
+        print("错误: 必须提供配置文件或指定必要的服务器参数")
+        print("必要参数: --source-dir, --target-host, --target-user, --target-dir")
+        parser.print_help()
+        sys.exit(1)
+
+    # 加载配置
+    config = {}
+    if args.config:
+        config = load_config(args.config)
+        if not config:
+            print("错误: 无法加载配置文件")
+            sys.exit(1)
+
+    # 构建源服务器配置
+    source_config_data = config.get("source", {})
+
+    # 命令行参数覆盖配置文件
+    if args.source_type:
+        source_config_data["type"] = args.source_type
+
+    # 如果是SSH类型，需要SSH配置
+    if args.source_type == "ssh" or source_config_data.get("type") == "ssh":
+        ssh_config = source_config_data.get("ssh", {})
+
+        if args.source_host:
+            ssh_config["host"] = args.source_host
+        if args.source_port:
+            ssh_config["port"] = args.source_port
+        if args.source_user:
+            ssh_config["username"] = args.source_user
+        if args.source_password:
+            ssh_config["password"] = args.source_password
+        if args.source_key:
+            ssh_config["key_file"] = args.source_key
+
+        source_config_data["ssh"] = ssh_config
+
+        # 检查必要的SSH参数
+        if not ssh_config.get("host"):
+            print("错误: 源服务器为SSH类型，必须指定 --source-host")
+            sys.exit(1)
+        if not ssh_config.get("username"):
+            print("错误: 源服务器为SSH类型，必须指定 --source-user")
+            sys.exit(1)
+    else:
+        # 本地类型，不需要SSH配置
+        source_config_data.pop("ssh", None)
+
+    # 设置基础目录
+    if args.source_dir:
+        source_config_data["base_dir"] = args.source_dir
+    elif "base_dir" not in source_config_data:
+        print("错误: 必须指定源服务器图片基础目录 (--source-dir)")
+        sys.exit(1)
+
+    # 构建目标服务器配置
+    target_config_data = config.get("target", {})
+    target_config_data["type"] = "ssh"  # 目标必须是SSH
+
+    ssh_config = target_config_data.get("ssh", {})
+
+    # 命令行参数覆盖配置文件
+    if args.target_host:
+        ssh_config["host"] = args.target_host
+    if args.target_port:
+        ssh_config["port"] = args.target_port
+    if args.target_user:
+        ssh_config["username"] = args.target_user
+    if args.target_password:
+        ssh_config["password"] = args.target_password
+    if args.target_key:
+        ssh_config["key_file"] = args.target_key
+
+    target_config_data["ssh"] = ssh_config
+
+    # 设置基础目录
+    if args.target_dir:
+        target_config_data["base_dir"] = args.target_dir
+    elif "base_dir" not in target_config_data:
+        print("错误: 必须指定目标服务器图片基础目录 (--target-dir)")
+        sys.exit(1)
+
+    # 检查必要的目标SSH参数
+    if not ssh_config.get("host"):
+        print("错误: 必须指定目标服务器地址 (--target-host)")
+        sys.exit(1)
+    if not ssh_config.get("username"):
+        print("错误: 必须指定目标服务器用户名 (--target-user)")
+        sys.exit(1)
+
+    # 创建服务器配置对象
+    try:
+        source_config = ServerConfig.from_dict(source_config_data)
+        target_config = ServerConfig.from_dict(target_config_data)
+
+        if args.verbose:
+            print(f"源服务器配置: {source_config}")
+            print(f"目标服务器配置: {target_config}")
+    except Exception as e:
+        print(f"创建服务器配置失败: {e}")
+        sys.exit(1)
+
+    # 获取图片路径列表
+    image_paths = []
+
+    # 从文件读取
+    if args.input:
+        file_paths = read_image_paths(args.input)
+        image_paths.extend(file_paths)
+
+    # 从命令行参数添加
+    if args.image_paths:
+        image_paths.extend(args.image_paths)
+
+    if not image_paths:
+        print("错误: 没有指定要迁移的图片路径")
+        print("请使用 --input 指定文件或直接在命令行提供路径")
+        sys.exit(1)
+
+    # 去重
+    image_paths = list(set(image_paths))
+
+    # 创建并运行迁移器
+    migrator = ImageMigrator(
+        source_config=source_config,
+        target_config=target_config,
+        verbose=args.verbose,
+        overwrite=args.overwrite,
+    )
+
+    # 执行迁移
+    stats = migrator.migrate_images(image_paths)
+
+    # 如果有失败，返回错误代码
+    if stats["failed"] > 0:
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

scripts/requirements.txt

@@ -0,0 +1,8 @@
+# Requirements for image migration script
+paramiko>=2.12.0
+
+# Optional: for better performance with large files
+# Optional: for progress bars
+# tqdm>=4.66.0
+
+# Install with: pip install -r requirements.txt