AI 生成: 一个基于浏览器网址的跨设备文件快传工具,支持拖拽上传、二维码扫描、实时传输进度显示。
This commit is contained in:
@@ -0,0 +1,137 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
QuickShare - 浏览器跨设备文件快传服务
|
||||
"""
|
||||
import os
|
||||
import socket
|
||||
import time
|
||||
from flask import Flask, render_template, request, jsonify, send_from_directory, abort
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
# 上传目录
|
||||
UPLOAD_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'uploads')
|
||||
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
||||
|
||||
# 允许所有文件类型,最大 2GB
|
||||
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
|
||||
app.config['MAX_CONTENT_LENGTH'] = 2 * 1024 * 1024 * 1024
|
||||
|
||||
|
||||
def get_local_ip():
|
||||
"""获取本机局域网 IP 地址"""
|
||||
try:
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
s.connect(('8.8.8.8', 80))
|
||||
ip = s.getsockname()[0]
|
||||
s.close()
|
||||
return ip
|
||||
except Exception:
|
||||
return '127.0.0.1'
|
||||
|
||||
|
||||
def format_size(size):
|
||||
"""格式化文件大小"""
|
||||
for unit in ['B', 'KB', 'MB', 'GB']:
|
||||
if size < 1024:
|
||||
return f"{size:.1f} {unit}"
|
||||
size /= 1024
|
||||
return f"{size:.1f} TB"
|
||||
|
||||
|
||||
@app.route('/')
|
||||
def index():
|
||||
"""主页面"""
|
||||
ip = get_local_ip()
|
||||
port = int(os.environ.get('PORT', 5000))
|
||||
url = f"http://{ip}:{port}"
|
||||
return render_template('index.html', url=url)
|
||||
|
||||
|
||||
@app.route('/upload', methods=['POST'])
|
||||
def upload_file():
|
||||
"""接收上传的文件"""
|
||||
if 'file' not in request.files:
|
||||
return jsonify({'success': False, 'error': '没有文件'}), 400
|
||||
|
||||
file = request.files['file']
|
||||
if file.filename == '':
|
||||
return jsonify({'success': False, 'error': '文件名为空'}), 400
|
||||
|
||||
# 安全文件名,避免路径穿越
|
||||
filename = secure_filename(file.filename)
|
||||
if not filename:
|
||||
filename = f"file_{int(time.time())}"
|
||||
|
||||
# 避免覆盖同名文件
|
||||
save_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
|
||||
base, ext = os.path.splitext(filename)
|
||||
counter = 1
|
||||
while os.path.exists(save_path):
|
||||
filename = f"{base}_{counter}{ext}"
|
||||
save_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
|
||||
counter += 1
|
||||
|
||||
file.save(save_path)
|
||||
size = os.path.getsize(save_path)
|
||||
|
||||
return jsonify({
|
||||
'success': True,
|
||||
'filename': filename,
|
||||
'size': format_size(size),
|
||||
'size_bytes': size
|
||||
})
|
||||
|
||||
|
||||
@app.route('/files')
|
||||
def list_files():
|
||||
"""列出已上传的文件"""
|
||||
files = []
|
||||
if os.path.exists(UPLOAD_FOLDER):
|
||||
for name in os.listdir(UPLOAD_FOLDER):
|
||||
path = os.path.join(UPLOAD_FOLDER, name)
|
||||
if os.path.isfile(path):
|
||||
stat = os.stat(path)
|
||||
files.append({
|
||||
'name': name,
|
||||
'size': format_size(stat.st_size),
|
||||
'size_bytes': stat.st_size,
|
||||
'mtime': stat.st_mtime
|
||||
})
|
||||
# 按修改时间倒序
|
||||
files.sort(key=lambda x: x['mtime'], reverse=True)
|
||||
return jsonify(files)
|
||||
|
||||
|
||||
@app.route('/download/<path:filename>')
|
||||
def download_file(filename):
|
||||
"""下载文件"""
|
||||
# 安全检查
|
||||
if '..' in filename or filename.startswith('/'):
|
||||
abort(400)
|
||||
return send_from_directory(app.config['UPLOAD_FOLDER'], filename, as_attachment=True)
|
||||
|
||||
|
||||
@app.route('/delete/<path:filename>', methods=['DELETE'])
|
||||
def delete_file(filename):
|
||||
"""删除文件"""
|
||||
if '..' in filename or filename.startswith('/'):
|
||||
abort(400)
|
||||
path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
|
||||
if os.path.isfile(path):
|
||||
os.remove(path)
|
||||
return jsonify({'success': True})
|
||||
return jsonify({'success': False, 'error': '文件不存在'}), 404
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
port = int(os.environ.get('PORT', 5000))
|
||||
ip = get_local_ip()
|
||||
print("\n" + "=" * 50)
|
||||
print(" QuickShare 已启动")
|
||||
print("=" * 50)
|
||||
print(f" 本机访问: http://localhost:{port}")
|
||||
print(f" 局域网访问: http://{ip}:{port}")
|
||||
print("=" * 50 + "\n")
|
||||
app.run(host='0.0.0.0', port=port, debug=False, threaded=True)
|
||||
Reference in New Issue
Block a user