jsasg/orico-official-website
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Merge branch 'dev'

Browse Source
This commit is contained in:
jsasg
2025-12-22 10:15:48 +08:00
parent c767c20641 33929b8284
commit 24e5554bc4
9 changed files with 1587 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
app/index/view/pc/topic_laptop/index.html
View File

@@ -62,11 +62,7 @@
<div class="swiper-container-texts-p">{$tfi.desc||raw|htmlspecialchars_decode}</div>
<div class="swiper-container-texts-img">
<a href="{$tfi.link}">
{eq name=":cookie('think_lang')" value="en-us"}
<img src="__IMAGES__/topic_laptop/eljgd.png" alt="" >
{else/}
<img src="__IMAGES__/topic_laptop/ljgd.png" alt="" >
{/eq}
<img src="{$tfi.extra_image}" alt="" >
</a>
</div>
</div>

186
scripts/QUICK_START.md Normal file
View File

@@ -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/*

415
scripts/README.md Normal file
View File

@@ -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. 确认服务器配置正确
---
**提示**：首次使用前，建议在测试服务器上进行小规模测试。

24
scripts/config.json Normal file
View File

@@ -0,0 +1,24 @@
{
"source": {
"type": "local",
"base_dir": "public/storage",
"ssh": {
"host": "source-server.example.com",
"port": 22,
"username": "username",
"password": "your_password_here",
"key_file": ""
}
},
"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": ""
}
}
}

24
scripts/examples/config.example.json Normal file
View File

@@ -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": ""
}
}
}

12
scripts/examples/images.example.txt Normal file
View File

@@ -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

912
scripts/image_migrate.py Executable file
View File

@@ -0,0 +1,912 @@
#!/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 Exception as listdir_e:
# 目录确实不存在且创建失败
print(f"错误: 无法创建目录 {current_path}: {listdir_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("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("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("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:
with open("fails.txt", "a", encoding="utf-8") as file:
file.write(f"{source_path}\n")
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("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()

2
scripts/images.txt Normal file
View File

@@ -0,0 +1,2 @@
images/banner/20251217/c16f15d090170b9cecaf6b88e02384f3_thumb.png
images/banner/20251217/c16f15d090170b9cecaf6b88e02384f3.png

8
scripts/requirements.txt Normal file
View File

@@ -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