在免费吧(free8.net)看到 WispByte 为注册用户提供的1G空间,512M内存的小容器,其目的可能是为了让用户免费体验一下产品,然后将免费用户转化为付费用户吧。
当然,wispbyte是为了让用户测试TG机器人方面的,但我另辟蹊径,使用python做了一个在线FTP管理器,凑巧可以连接free.fr的免费空间。代码功能下:
代码如下:
import os
import time
import shutil
import zipfile
import tarfile
import json
from flask import Flask, render_template_string, request, redirect, url_for, session, send_file
from ftplib import FTP
from werkzeug.utils import secure_filename
app = Flask(__name__)
#自定义安全码
app.secret_key = 'your_secret_key_here'
# HTML 模板
HTML_TEMPLATE = '''
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Web FTP 文件管理器</title>
<link href="https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/[email protected]/font/bootstrap-icons.css">
<style>
html, body { height: 100%; }
body { background-color: #f8f9fa; font-size: 14px; color: #333; display: flex; flex-direction: column; }
.main-content { flex: 1; }
.navbar { box-shadow: 0 2px 4px rgba(0,0,0,.05); }
.card { border: none; box-shadow: 0 0.125rem 0.25rem rgba(0, 0, 0, 0.075); }
.table th { font-weight: 600; background-color: #f1f3f5; color: #495057; border-top: none; }
.table td { vertical-align: middle; }
.btn-action { padding: 0.25rem 0.5rem; font-size: 0.825rem; }
.breadcrumb-container { background: #e9ecef; padding: 10px 15px; border-radius: 0.375rem; margin-bottom: 20px; }
textarea.code-editor { font-family: SFMono-Regular, Menlo, Monaco, Consolas, monospace; font-size: 13px; height: 500px; }
.batch-toolbar { background: #cfe2ff; border: 1px solid #b6d4fe; color: #084298; padding: 10px 15px; border-radius: 0.375rem; margin-bottom: 15px; display: none; align-items: center; justify-content: space-between; flex-wrap: wrap; gap: 10px; }
.drop-zone { border: 2px dashed #0d6efd; border-radius: 0.5rem; padding: 30px; text-align: center; background: #f8f9fa; transition: all 0.3s; cursor: pointer; }
.drop-zone.dragover { background: #e2f0d9; border-color: #198754; }
.footer { background-color: #ffffff; border-top: 1px solid #dee2e6; color: #6c757d; font-size: 13px; }
</style>
</head>
<body>
<div class="main-content">
<nav class="navbar navbar-expand-lg navbar-dark bg-dark mb-4">
<div class="container-fluid">
<a class="navbar-brand" href="/"><i class="bi bi-folder2-open"></i> Web FTP Manager</a>
{% if session.get('logged_in') %}
<div class="d-flex align-items-center">
<span class="text-light me-3 small"><i class="bi bi-hdd-network"></i> {{ session.host }}</span>
<a href="/logout" class="btn btn-outline-light btn-sm"><i class="bi bi-box-arrow-right"></i> 断开连接</a>
</div>
{% endif %}
</div>
</nav>
<div class="container pb-5">
{% if error %}
<div class="alert alert-danger alert-dismissible fade show" role="alert">
<i class="bi bi-exclamation-triangle-fill"></i> {{ error }}
<button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button>
</div>
{% endif %}
{% if not session.get('logged_in') %}
<div class="row justify-content-center mt-5">
<div class="col-md-5">
<div class="card shadow-sm">
<div class="card-body p-4">
<h3 class="card-title mb-4 text-center text-secondary"><i class="bi bi-shield-lock"></i> FTP 身份认证</h3>
<form method="post" action="/login">
<div class="mb-3">
<label class="form-label">服务器地址 (Host)</label>
<input type="text" class="form-control" name="host" placeholder="例如: reishi.cc" required>
</div>
<div class="mb-3">
<label class="form-label">端口 (Port)</label>
<input type="number" class="form-control" name="port" value="21" required>
</div>
<div class="mb-3">
<label class="form-label">用户名 (Username)</label>
<input type="text" class="form-control" name="username" required>
</div>
<div class="mb-3">
<label class="form-label">密码 (Password)</label>
<input type="password" class="form-control" name="password" required>
</div>
<button type="submit" class="btn btn-primary w-100 py-2"><i class="bi bi-plug"></i> 立即连接</button>
</form>
</div>
</div>
</div>
</div>
{% else %}
{% if mode == 'edit' %}
<div class="card">
<div class="card-header bg-secondary text-white d-flex justify-content-between align-items-center">
<span><i class="bi bi-file-earmark-code"></i> 正在编辑: <strong>{{ filename }}</strong></span>
<a href="/" class="btn btn-light btn-sm">返回列表</a>
</div>
<div class="card-body">
<form method="post" action="/save">
<input type="hidden" name="filename" value="{{ filename }}">
<textarea name="content" class="form-control code-editor mb-3">{{ content }}</textarea>
<button type="submit" class="btn btn-success"><i class="bi bi-save"></i> 保存修改</button>
<a href="/" class="btn btn-secondary">取消</a>
</form>
</div>
</div>
{% else %}
<div class="breadcrumb-container d-flex justify-content-between align-items-center flex-wrap">
<ol class="breadcrumb mb-0">
<li class="breadcrumb-item"><a href="/navigate?path=/" class="text-decoration-none"><i class="bi bi-house-door"></i> 根目录</a></li>
{% set path_accum = "" %}
{% for part in session.cwd.strip('/').split('/') %}
{% if part %}
{% set path_accum = path_accum + '/' + part %}
<li class="breadcrumb-item"><a href="/navigate?path={{ path_accum }}" class="text-decoration-none">{{ part }}</a></li>
{% endif %}
{% endfor %}
</ol>
<div class="btn-group">
<button type="button" class="btn btn-sm btn-outline-primary" data-bs-toggle="modal" data-bs-target="#uploadModal"><i class="bi bi-cloud-upload"></i> 上传文件/文件夹</button>
<button type="button" class="btn btn-sm btn-outline-success" data-bs-toggle="modal" data-bs-target="#mkdirModal"><i class="bi bi-folder-plus"></i> 新建目录</button>
<button type="button" class="btn btn-sm btn-outline-secondary" data-bs-toggle="modal" data-bs-target="#mkfileModal"><i class="bi bi-file-earmark-plus"></i> 新建文件</button>
</div>
</div>
<div class="batch-toolbar" id="batchBar">
<div><i class="bi bi-check2-square"></i> 已选中 <strong id="selectedCount">0</strong> 项</div>
<div class="btn-group">
<button class="btn btn-danger btn-sm" onclick="batchAction('delete')"><i class="bi bi-trash"></i> 批量删除</button>
<button class="btn btn-warning btn-sm" onclick="batchAction('move')"><i class="bi bi-arrow-right-square"></i> 批量移动</button>
<button class="btn btn-success btn-sm" onclick="batchAction('copy')"><i class="bi bi-files"></i> 批量复制</button>
<button class="btn btn-info btn-sm text-white" data-bs-toggle="modal" data-bs-target="#compressModal"><i class="bi bi-file-earmark-zip"></i> 打包选中项</button>
</div>
</div>
<div class="card">
<div class="card-body p-0">
<div class="table-responsive">
<table class="table table-hover mb-0 align-middle">
<thead>
<tr>
<th style="width: 40px;"><input type="checkbox" class="form-check-input" id="selectAll" onclick="toggleSelectAll(this)"></th>
<th>名称</th>
<th style="width: 100px;">类型</th>
<th style="width: 120px;">大小</th>
<th style="width: 160px;">修改时间</th>
<th style="width: 260px; text-align: right;">操作</th>
</tr>
</thead>
<tbody>
{% if session.cwd != '/' %}
<tr>
<td></td>
<td colspan="5">
<a href="/cd?dir=.." class="text-decoration-none text-dark fw-bold"><i class="bi bi-arrow-up-circle"></i> .. (返回上级目录)</a>
</td>
</tr>
{% endif %}
{% for item in files %}
<tr>
<td><input type="checkbox" class="form-check-input item-checkbox" value="{{ item.name }}" onchange="updateBatchBar()"></td>
<td>
{% if item.is_dir %}
<a href="/cd?dir={{ item.name }}" class="text-decoration-none fw-bold text-dark"><i class="bi bi-folder-fill text-warning"></i> {{ item.name }}</a>
{% else %}
<i class="bi bi-file-earmark-text text-secondary"></i> {{ item.name }}
{% endif %}
</td>
<td><span class="badge bg-light text-dark border">{{ '目录' if item.is_dir else '文件' }}</span></td>
<td class="text-muted small">{{ item.size }}</td>
<td class="text-muted small">{{ item.time }}</td>
<td class="text-end">
<div class="btn-group">
{% if item.is_dir %}
<a href="/cd?dir={{ item.name }}" class="btn btn-outline-primary btn-action"><i class="bi bi-folder2-open"></i> 进入</a>
{% else %}
<a href="/download/{{ item.name }}" class="btn btn-outline-primary btn-action"><i class="bi bi-download"></i></a>
{% if item.name.lower().endswith(('.zip', '.rar', '.tar', '.gz', '.tgz', '.7z')) %}
<button class="btn btn-outline-warning btn-action" onclick="openUnzipModal('{{ item.name }}')" title="解压缩"><i class="bi bi-file-earmark-zip"></i></button>
{% else %}
<a href="/edit?name={{ item.name }}" class="btn btn-outline-warning btn-action" title="编辑"><i class="bi bi-pencil-square"></i></a>
{% endif %}
{% endif %}
<button class="btn btn-outline-secondary btn-action" onclick="renameItem('{{ item.name }}')" title="改名"><i class="bi bi-cursor-text"></i></button>
<a href="/delete?name={{ item.name }}" class="btn btn-outline-danger btn-action" onclick="return confirm('确定要递归删除 {{ item.name }} 及其所有内容吗?');" title="删除"><i class="bi bi-trash"></i></a>
</div>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
<!-- 弹窗:解压设置(支持选择同名策略) -->
<div class="modal fade" id="unzipModal" tabindex="-1">
<div class="modal-dialog">
<div class="modal-content">
<form method="get" action="/unzip">
<div class="modal-header">
<h5 class="modal-title"><i class="bi bi-file-earmark-zip"></i> 解压选项</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<input type="hidden" name="name" id="unzipFileName">
<p class="text-muted">正在解压文件:<strong id="unzipFileLabel"></strong></p>
<div class="mb-3">
<label class="form-label">遇到同名文件时:</label>
<select class="form-select" name="conflict">
<option value="rename">自动重命名 (例如 index(1).php)</option>
<option value="skip">跳过不解压</option>
<option value="overwrite">直接覆盖</option>
</select>
</div>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary btn-sm" data-bs-dismiss="modal">取消</button>
<button type="submit" class="btn btn-warning btn-sm">开始解压</button>
</div>
</form>
</div>
</div>
</div>
<!-- 弹窗:支持拖拽上传文件和文件夹 -->
<div class="modal fade" id="uploadModal" tabindex="-1">
<div class="modal-dialog modal-lg">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title"><i class="bi bi-cloud-upload"></i> 上传文件或整个文件夹</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<div id="dropZone" class="drop-zone mb-3">
<i class="bi bi-cloud-arrow-up display-4 text-primary"></i>
<p class="mt-2 mb-1 fw-bold">将文件或文件夹**混合拖拽到此处**上传</p>
<p class="text-muted small mb-0">完美支持多文件、整文件夹及混合同时拖入</p>
<input type="file" id="fileInput" multiple webkitdirectory style="display:none;" onchange="handleFiles(this.files)">
</div>
<div class="d-flex justify-content-between align-items-center">
<button type="button" class="btn btn-outline-primary btn-sm" onclick="document.getElementById('fileInput').click()"><i class="bi bi-folder2-open"></i> 选择本地文件/文件夹</button>
<span id="uploadStatus" class="text-muted small"></span>
</div>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary btn-sm" data-bs-dismiss="modal">关闭</button>
</div>
</div>
</div>
</div>
<!-- 弹窗:新建文件夹 -->
<div class="modal fade" id="mkdirModal" tabindex="-1">
<div class="modal-dialog">
<div class="modal-content">
<form method="post" action="/mkdir">
<div class="modal-header">
<h5 class="modal-title"><i class="bi bi-folder-plus"></i> 创建新文件夹</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<input type="text" class="form-control" name="dirname" placeholder="请输入文件夹名称" required>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary btn-sm" data-bs-dismiss="modal">取消</button>
<button type="submit" class="btn btn-success btn-sm">创建</button>
</div>
</form>
</div>
</div>
</div>
<!-- 弹窗:新建文本文件 -->
<div class="modal fade" id="mkfileModal" tabindex="-1">
<div class="modal-dialog">
<div class="modal-content">
<form method="post" action="/mkfile">
<div class="modal-header">
<h5 class="modal-title"><i class="bi bi-file-earmark-plus"></i> 创建空白文件</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<input type="text" class="form-control" name="filename" placeholder="例如: index.html 或 config.json" required>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary btn-sm" data-bs-dismiss="modal">取消</button>
<button type="submit" class="btn btn-secondary btn-sm">创建并编辑</button>
</div>
</form>
</div>
</div>
</div>
<!-- 弹窗:打包选中项 -->
<div class="modal fade" id="compressModal" tabindex="-1">
<div class="modal-dialog">
<div class="modal-content">
<form id="compressForm" method="post" action="/batch_compress">
<div class="modal-header">
<h5 class="modal-title"><i class="bi bi-file-earmark-zip"></i> 打包选中项</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<input type="hidden" name="items" id="compressItemsInput">
<div class="mb-3">
<label class="form-label">压缩包文件名</label>
<input type="text" class="form-control" name="archive_name" value="archive" required>
</div>
<div class="mb-3">
<label class="form-label">压缩格式</label>
<select class="form-select" name="archive_format">
<option value="zip">zip (.zip)</option>
<option value="gztar">tar.gz (.tar.gz)</option>
<option value="tar">tar (.tar)</option>
</select>
</div>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary btn-sm" data-bs-dismiss="modal">取消</button>
<button type="submit" class="btn btn-primary btn-sm" onclick="prepareCompress()">开始打包</button>
</div>
</form>
</div>
</div>
</div>
<script src="https://cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.bundle.min.js"></script>
<script>
function openUnzipModal(filename) {
document.getElementById('unzipFileName').value = filename;
document.getElementById('unzipFileLabel').innerText = filename;
new bootstrap.Modal(document.getElementById('unzipModal')).show();
}
function toggleSelectAll(source) {
let checkboxes = document.getElementsByClassName('item-checkbox');
for(let i=0; i<checkboxes.length; i++) {
checkboxes[i].checked = source.checked;
}
updateBatchBar();
}
function updateBatchBar() {
let checkboxes = document.getElementsByClassName('item-checkbox');
let count = 0;
for(let i=0; i<checkboxes.length; i++) {
if(checkboxes[i].checked) count++;
}
document.getElementById('selectedCount').innerText = count;
document.getElementById('batchBar').style.display = count > 0 ? 'flex' : 'none';
}
function renameItem(oldName) {
let newName = prompt("请输入新的名称:", oldName);
if (newName && newName.trim() !== "" && newName !== oldName) {
let form = document.createElement('form');
form.method = 'POST'; form.action = '/rename';
let inputOld = document.createElement('input'); inputOld.type = 'hidden'; inputOld.name = 'old_name'; inputOld.value = oldName;
let inputNew = document.createElement('input'); inputNew.type = 'hidden'; inputNew.name = 'new_name'; inputNew.value = newName.trim();
form.appendChild(inputOld); form.appendChild(inputNew);
document.body.appendChild(form); form.submit();
}
}
function batchAction(actionType) {
let checkboxes = document.getElementsByClassName('item-checkbox');
let selected = [];
for(let i=0; i<checkboxes.length; i++) {
if(checkboxes[i].checked) selected.push(checkboxes[i].value);
}
if(selected.length === 0) return;
if(actionType === 'delete') {
if(!confirm("确定要递归删除选中的 " + selected.length + " 个项目及其所有内容吗?")) return;
submitBatchForm('/batch_delete', { items: JSON.stringify(selected) });
} else if(actionType === 'move' || actionType === 'copy') {
let targetDir = prompt("请输入目标文件夹路径(例如 /public_html):");
if(!targetDir) return;
submitBatchForm('/batch_' + actionType, { items: JSON.stringify(selected), target: targetDir });
}
}
function prepareCompress() {
let checkboxes = document.getElementsByClassName('item-checkbox');
let selected = [];
for(let i=0; i<checkboxes.length; i++) {
if(checkboxes[i].checked) selected.push(checkboxes[i].value);
}
document.getElementById('compressItemsInput').value = JSON.stringify(selected);
}
function submitBatchForm(actionUrl, params) {
let form = document.createElement('form');
form.method = 'POST'; form.action = actionUrl;
for(let key in params) {
let input = document.createElement('input');
input.type = 'hidden'; input.name = key; input.value = params[key];
form.appendChild(input);
}
document.body.appendChild(form); form.submit();
}
const dropZone = document.getElementById('dropZone');
dropZone.addEventListener('dragover', (e) => { e.preventDefault(); dropZone.classList.add('dragover'); });
dropZone.addEventListener('dragleave', () => { dropZone.classList.remove('dragover'); });
dropZone.addEventListener('drop', async (e) => {
e.preventDefault();
dropZone.classList.remove('dragover');
let items = e.dataTransfer.items;
let fileQueue = [];
let promises = [];
for (let i = 0; i < items.length; i++) {
let item = items[i].webkitGetAsEntry();
if (item) promises.push(traverseFileTree(item, "", fileQueue));
}
await Promise.all(promises);
uploadQueue(fileQueue);
});
function traverseFileTree(item, path, queue) {
return new Promise((resolve) => {
if (item.isFile) {
item.file((file) => {
queue.push({ file: file, relativePath: path + file.name });
resolve();
});
} else if (item.isDirectory) {
let dirReader = item.createReader();
dirReader.readEntries(async (entries) => {
let subPromises = [];
for (let i = 0; i < entries.length; i++) {
subPromises.push(traverseFileTree(entries[i], path + item.name + "/", queue));
}
await Promise.all(subPromises);
resolve();
});
} else {
resolve();
}
});
}
function handleFiles(files) {
let queue = [];
for (let i = 0; i < files.length; i++) {
let f = files[i];
queue.push({ file: f, relativePath: f.webkitRelativePath || f.name });
}
uploadQueue(queue);
}
async function uploadQueue(queue) {
let statusEl = document.getElementById('uploadStatus');
let total = queue.length, current = 0;
for (let item of queue) {
current++;
statusEl.innerText = `正在上传 (${current}/${total}): ${item.relativePath}`;
let formData = new FormData();
formData.append('file', item.file);
formData.append('relative_path', item.relativePath);
try {
await fetch('/upload', { method: 'POST', body: formData });
} catch (err) { console.error(err); }
}
statusEl.innerText = "所有文件上传完毕!正在刷新...";
setTimeout(() => { window.location.reload(); }, 1000);
}
</script>
{% endif %}
{% endif %}
</div>
</div>
<!-- 版权与版本信息 Footer -->
<footer class="footer py-3 text-center">
<div class="container">
<span>© 2026 <a href="https://reishi.cc?ref=webftp" target="_blank">reishi.cc</a>. All Rights Reserved.</span>
<span class="ms-3 text-muted">Version <strong>v0.0.16</strong></span>
</div>
</footer>
</body>
</html>
'''
def get_ftp_connection():
ftp = FTP()
ftp.connect(session['host'], int(session['port']), timeout=15)
ftp.login(session['username'], session['password'])
if 'cwd' in session and session['cwd']:
try:
ftp.cwd(session['cwd'])
except:
pass
return ftp
def recursive_delete(ftp, name):
try:
ftp.delete(name)
return
except Exception:
pass
try:
ftp.cwd(name)
except Exception:
return
try:
try:
entries = [item_name for item_name, _ in ftp.mlsd() if item_name not in ['.', '..']]
except Exception:
entries = [e for e in ftp.nlst() if e not in ['.', '..']]
for entry in entries:
recursive_delete(ftp, entry)
ftp.cwd('..')
ftp.rmd(name)
except Exception as e:
try:
ftp.cwd('..')
except:
pass
raise e
@app.route('/')
def index():
files = []
error = request.args.get('error')
if session.get('logged_in'):
try:
ftp = get_ftp_connection()
file_list = []
try:
for name, facts in ftp.mlsd():
if name in ['.', '..']: continue
is_dir = facts.get('type') == 'dir'
size = facts.get('size', '-')
if not is_dir and size != '-':
size = f"{int(size) / 1024:.1f} KB" if int(size) > 1024 else f"{size} B"
else:
size = '-'
raw_time = facts.get('modify', '')
m_time = f"{raw_time[:4]}-{raw_time[4:6]}-{raw_time[6:8]}" if len(raw_time) >= 8 else '-'
file_list.append({'name': name, 'is_dir': is_dir, 'size': size, 'time': m_time})
except Exception:
for name in ftp.nlst():
if name in ['.', '..']: continue
file_list.append({'name': name, 'is_dir': False, 'size': '-', 'time': '-'})
files = sorted(file_list, key=lambda x: (not x['is_dir'], x['name']))
ftp.quit()
except Exception as e:
error = f"获取目录失败: {str(e)}"
return render_template_string(HTML_TEMPLATE, files=files, error=error, mode='normal')
@app.route('/login', methods=['POST'])
def login():
session['host'] = request.form['host']
session['port'] = request.form['port']
session['username'] = request.form['username']
session['password'] = request.form['password']
session['cwd'] = '/'
try:
ftp = get_ftp_connection()
ftp.quit()
session['logged_in'] = True
except Exception as e:
session.clear()
return redirect(url_for('index', error=f"连接失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/logout')
def logout():
session.clear()
return redirect(url_for('index'))
@app.route('/cd')
def cd():
target = request.args.get('dir')
if 'logged_in' in session and target:
try:
ftp = get_ftp_connection()
if target == '..': ftp.cwd('..')
else: ftp.cwd(target)
session['cwd'] = ftp.pwd()
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"无法进入目录: {str(e)}"))
return redirect(url_for('index'))
@app.route('/navigate')
def navigate():
target_path = request.args.get('path', '/')
if 'logged_in' in session:
try:
ftp = get_ftp_connection()
ftp.cwd(target_path)
session['cwd'] = ftp.pwd()
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"无法跳转目录: {str(e)}"))
return redirect(url_for('index'))
@app.route('/mkdir', methods=['POST'])
def mkdir():
dirname = request.form['dirname']
if 'logged_in' in session and dirname:
try:
ftp = get_ftp_connection()
ftp.mkd(dirname)
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"创建文件夹失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/mkfile', methods=['POST'])
def mkfile():
filename = request.form['filename']
if 'logged_in' in session and filename:
local_path = os.path.join('/tmp', secure_filename(filename))
try:
with open(local_path, 'w', encoding='utf-8') as f: f.write('')
ftp = get_ftp_connection()
with open(local_path, 'rb') as f: ftp.storbinary(f'STOR {filename}', f)
ftp.quit()
if os.path.exists(local_path): os.remove(local_path)
return redirect(url_for('edit_file', name=filename))
except Exception as e:
return redirect(url_for('index', error=f"创建文件失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/upload', methods=['POST'])
def upload():
if 'logged_in' not in session: return "Unauthorized", 401
file = request.files.get('file')
relative_path = request.form.get('relative_path', file.filename if file else '')
if file:
local_path = os.path.join('/tmp', f"upload_{time.time()}_{secure_filename(file.filename)}")
file.save(local_path)
try:
ftp = get_ftp_connection()
parts = relative_path.replace('\\', '/').split('/')
if len(parts) > 1:
current_cwd = ftp.pwd()
for folder in parts[:-1]:
if folder:
try: ftp.mkd(folder)
except: pass
try: ftp.cwd(folder)
except: pass
with open(local_path, 'rb') as f: ftp.storbinary(f'STOR {parts[-1]}', f)
ftp.cwd(current_cwd)
else:
with open(local_path, 'rb') as f: ftp.storbinary(f'STOR {parts[-1]}', f)
ftp.quit()
except Exception as e:
return str(e), 500
if os.path.exists(local_path): os.remove(local_path)
return "OK", 200
@app.route('/download/<path:filename>')
def download_file(filename):
if 'logged_in' not in session: return redirect(url_for('index'))
local_path = os.path.join('/tmp', secure_filename(filename))
try:
ftp = get_ftp_connection()
with open(local_path, 'wb') as f: ftp.retrbinary(f'RETR {filename}', f.write)
ftp.quit()
return send_file(local_path, as_attachment=True)
except Exception as e:
return f"下载失败: {e}"
@app.route('/unzip')
def unzip_file():
filename = request.args.get('name')
conflict_mode = request.args.get('conflict', 'rename') # rename, skip, overwrite
if 'logged_in' not in session or not filename:
return redirect(url_for('index'))
local_zip_path = os.path.join('/tmp', f"unzip_{time.time()}_{secure_filename(filename)}")
extract_dir = local_zip_path + "_extracted"
try:
ftp = get_ftp_connection()
with open(local_zip_path, 'wb') as f:
ftp.retrbinary(f'RETR {filename}', f.write)
os.makedirs(extract_dir, exist_ok=True)
if zipfile.is_zipfile(local_zip_path):
with zipfile.ZipFile(local_zip_path, 'r') as zip_ref:
zip_ref.extractall(extract_dir)
elif tarfile.is_tarfile(local_zip_path):
with tarfile.open(local_zip_path, 'r:*') as tar_ref:
tar_ref.extractall(extract_dir)
else:
raise Exception("暂不支持该压缩格式或文件损坏")
def upload_extracted_dir_with_conflict(local_base, remote_ftp, mode):
for root, dirs, files in os.walk(local_base):
rel_path = os.path.relpath(root, local_base).replace('\\', '/')
current_target_dir = remote_ftp.pwd()
if rel_path != '.':
for part in rel_path.split('/'):
try: remote_ftp.mkd(part)
except: pass
remote_ftp.cwd(part)
# 实时获取当前服务器目录下的文件列表,用以精准比对服务器上是否有同名文件
try:
server_existing_items = [name for name, _ in remote_ftp.mlsd() if name not in ['.', '..']]
except Exception:
try:
server_existing_items = [e for e in remote_ftp.nlst() if e not in ['.', '..']]
except Exception:
server_existing_items = []
for file in files:
target_filename = file
# 只有当服务器同目录下真的存在该文件名时,才触发策略
if target_filename in server_existing_items:
if mode == 'skip':
continue
elif mode == 'rename':
name_part, ext_part = os.path.splitext(file)
counter = 1
while True:
new_name = f"{name_part}({counter}){ext_part}"
if new_name not in server_existing_items:
target_filename = new_name
break
counter += 1
elif mode == 'overwrite':
pass
l_file = os.path.join(root, file)
with open(l_file, 'rb') as lf:
remote_ftp.storbinary(f'STOR {target_filename}', lf)
# 将新上传或重命名的文件名加入列表,防止压缩包自身内部出现多重同名文件时发生冲突
server_existing_items.append(target_filename)
remote_ftp.cwd(current_target_dir)
upload_extracted_dir_with_conflict(extract_dir, ftp, conflict_mode)
ftp.quit()
if os.path.exists(local_zip_path): os.remove(local_zip_path)
if os.path.exists(extract_dir): shutil.rmtree(extract_dir)
except Exception as e:
return redirect(url_for('index', error=f"解压失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/batch_compress', methods=['POST'])
def batch_compress():
items = json.loads(request.form.get('items', '[]'))
archive_name = request.form.get('archive_name', 'archive').strip()
archive_format = request.form.get('archive_format', 'zip')
if 'logged_in' not in session or not items:
return redirect(url_for('index'))
temp_dir = os.path.join('/tmp', f"compress_{time.time()}")
os.makedirs(temp_dir, exist_ok=True)
try:
ftp = get_ftp_connection()
def download_item_recursive(ftp_conn, name, local_dest_dir):
try:
l_path = os.path.join(local_dest_dir, name)
with open(l_path, 'wb') as f: ftp_conn.retrbinary(f'RETR {name}', f.write)
except Exception:
sub_dir = os.path.join(local_dest_dir, name)
os.makedirs(sub_dir, exist_ok=True)
try:
ftp_conn.cwd(name)
try: entries = [e for e, _ in ftp_conn.mlsd() if e not in ['.', '..']]
except: entries = [e for e in ftp_conn.nlst() if e not in ['.', '..']]
for entry in entries:
download_item_recursive(ftp_conn, entry, sub_dir)
ftp_conn.cwd('..')
except: pass
for name in items:
download_item_recursive(ftp, name, temp_dir)
archive_base = os.path.join('/tmp', f"{archive_name}")
shutil.make_archive(archive_base, archive_format, root_dir=temp_dir)
ext = '.tar.gz' if archive_format == 'gztar' else f'.{archive_format}'
final_archive_filename = archive_name + ext if not archive_name.endswith(ext) else archive_name
local_final_archive = archive_base + ext
with open(local_final_archive, 'rb') as f:
ftp.storbinary(f'STOR {final_archive_filename}', f)
ftp.quit()
if os.path.exists(temp_dir): shutil.rmtree(temp_dir)
if os.path.exists(local_final_archive): os.remove(local_final_archive)
except Exception as e:
return redirect(url_for('index', error=f"打包失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/edit')
def edit_file():
filename = request.args.get('name')
if 'logged_in' not in session or not filename: return redirect(url_for('index'))
local_path = os.path.join('/tmp', secure_filename(filename))
content = ""
try:
ftp = get_ftp_connection()
with open(local_path, 'wb') as f: ftp.retrbinary(f'RETR {filename}', f.write)
ftp.quit()
with open(local_path, 'r', encoding='utf-8', errors='ignore') as f: content = f.read()
if os.path.exists(local_path): os.remove(local_path)
except Exception as e:
return redirect(url_for('index', error=f"无法读取文件进行编辑: {str(e)}"))
return render_template_string(HTML_TEMPLATE, mode='edit', filename=filename, content=content)
@app.route('/save', methods=['POST'])
def save_file():
filename = request.form['filename']
content = request.form['content']
if 'logged_in' not in session or not filename: return redirect(url_for('index'))
local_path = os.path.join('/tmp', secure_filename(filename))
try:
with open(local_path, 'w', encoding='utf-8') as f: f.write(content)
ftp = get_ftp_connection()
with open(local_path, 'rb') as f: ftp.storbinary(f'STOR {filename}', f)
ftp.quit()
if os.path.exists(local_path): os.remove(local_path)
except Exception as e:
return redirect(url_for('index', error=f"保存文件失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/rename', methods=['POST'])
def rename_file():
old_name = request.form['old_name']
new_name = request.form['new_name']
if 'logged_in' in session and old_name and new_name:
try:
ftp = get_ftp_connection()
ftp.rename(old_name, new_name)
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"重命名失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/delete')
def delete_file():
name = request.args.get('name')
if 'logged_in' in session and name:
try:
ftp = get_ftp_connection()
recursive_delete(ftp, name)
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"删除失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/batch_delete', methods=['POST'])
def batch_delete():
items = json.loads(request.form.get('items', '[]'))
if 'logged_in' in session and items:
try:
ftp = get_ftp_connection()
for name in items:
try: recursive_delete(ftp, name)
except: pass
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"批量删除发生错误: {str(e)}"))
return redirect(url_for('index'))
@app.route('/batch_move', methods=['POST'])
def batch_move():
items = json.loads(request.form.get('items', '[]'))
target = request.form.get('target')
if 'logged_in' in session and items and target:
try:
ftp = get_ftp_connection()
for name in items:
new_path = os.path.join(target, name).replace('\\', '/')
ftp.rename(name, new_path)
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"批量移动失败: {str(e)}"))
return redirect(url_for('index'))
@app.route('/batch_copy', methods=['POST'])
def batch_copy():
items = json.loads(request.form.get('items', '[]'))
target = request.form.get('target')
if 'logged_in' in session and items and target:
try:
ftp = get_ftp_connection()
for name in items:
local_path = os.path.join('/tmp', secure_filename(name))
with open(local_path, 'wb') as f: ftp.retrbinary(f'RETR {name}', f.write)
current_cwd = ftp.pwd()
ftp.cwd(target)
with open(local_path, 'rb') as f: ftp.storbinary(f'STOR {name}', f)
ftp.cwd(current_cwd)
if os.path.exists(local_path): os.remove(local_path)
ftp.quit()
except Exception as e:
return redirect(url_for('index', error=f"批量复制失败: {str(e)}"))
return redirect(url_for('index'))
if __name__ == '__main__':
port = int(os.environ.get('SERVER_PORT', os.environ.get('PORT', 3424)))
app.run(host='0.0.0.0', port=port, debug=True)
在wispbyte的机器上用,需要在机器管理中的startup里,Package Manager中添加flask并点击add package,代码的最后第二行中的3424改为自己的端口,保存代码为app.py ,然后再执行开机。
此代码同样适用于 adkynet 的python机器上使用,需要在startup的Additional Python packages中,添加flask paramiko后,再开机。
不定时关闭的在线演示:
1、wispbyte:http://78.154.103.29:13078/
2、adkynet:http://83.150.217.93:3375/