Spaces:

WACATW
/

pan

Running

pan / app.py
Update app.py
647c883 verified about 1 month ago
25.8 kB
import os
import io
import json
import mimetypes
from datetime import datetime
from flask import Flask, render_template, request, jsonify, send_file, abort, session, redirect, url_for
from werkzeug.security import generate_password_hash, check_password_hash
import re
import uuid

#GRAPHICAL_API_TOKEN = "676767"

def safe_filename(filename):
    """安全的文件名处理，支持中文"""
    if not filename:
        return 'unnamed'
    
    # 移除路径分隔符和危险字符
    filename = re.sub(r'[\\/:*?"<>|\x00-\x1f]', '_', filename)
    # 移除开头的点和空格
    filename = filename.lstrip('. ')
    # 确保文件名不为空
    if not filename:
        filename = 'unnamed'
    return filename

def safe_username(username):
    """安全的用户名处理"""
    if not username:
        return None
    # 只允许字母、数字、下划线、中文
    if not re.match(r'^[a-zA-Z0-9_\u4e00-\u9fa5]{3,20}$', username):
        return None
    return username

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 500 * 1024 * 1024 * 1024
app.secret_key = os.environ.get('SECRET_KEY', 'cloud-disk-secret-key-2024')

# Session cookie 配置
app.config['SESSION_COOKIE_SAMESITE'] = 'None'
app.config['SESSION_COOKIE_SECURE'] = True
app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['PERMANENT_SESSION_LIFETIME'] = 30 * 24 * 60 * 60

# 如果是 HTTPS 环境，取消下面两行的注释
# app.config['SESSION_COOKIE_SAMESITE'] = 'None'
# app.config['SESSION_COOKIE_SECURE'] = True

# Configuration
STORAGE_BASE_PATH = os.environ.get('STORAGE_PATH', '/data')
USERS_FILE = os.path.join(STORAGE_BASE_PATH, 'users.json')
MAX_STORAGE_PER_USER = 1 * 1024 * 1024 * 1024  # 1GB per user

# Ensure storage directory exists
os.makedirs(STORAGE_BASE_PATH, exist_ok=True)

# Initialize users file
if not os.path.exists(USERS_FILE):
    with open(USERS_FILE, 'w', encoding='utf-8') as f:
        json.dump({}, f)

def load_users():
    """加载用户数据"""
    try:
        with open(USERS_FILE, 'r', encoding='utf-8') as f:
            return json.load(f)
    except:
        return {}

def save_users(users):
    """保存用户数据"""
    with open(USERS_FILE, 'w', encoding='utf-8') as f:
        json.dump(users, f, ensure_ascii=False, indent=2)

def get_user_storage_path(username):
    """获取用户的存储目录"""
    return os.path.join(STORAGE_BASE_PATH, 'users', username)

def ensure_user_storage(username):
    """确保用户存储目录存在"""
    path = get_user_storage_path(username)
    os.makedirs(path, exist_ok=True)
    return path

def calculate_user_storage(username):
    """计算用户已使用的存储空间"""
    user_path = get_user_storage_path(username)
    if not os.path.exists(user_path):
        return 0
    
    total_size = 0
    for dirpath, dirnames, filenames in os.walk(user_path):
        for filename in filenames:
            filepath = os.path.join(dirpath, filename)
            try:
                total_size += os.path.getsize(filepath)
            except:
                pass
    return total_size

def get_current_user():
    """获取当前登录用户"""
    return session.get('username')

def login_required_api():
    """API 登录检查装饰器辅助函数"""
    username = get_current_user()
    if not username:
        return False, jsonify({'success': False, 'error': '请先登录', 'need_login': True})
    return True, username

def get_full_path(username, path=''):
    """获取用户目录下的完整路径"""
    user_path = get_user_storage_path(username)
    if path and path != '/':
        # 规范化路径，防止路径遍历
        path = path.strip('/')
        full_path = os.path.normpath(os.path.join(user_path, path))
        # 确保路径在用户目录内
        if not full_path.startswith(os.path.normpath(user_path)):
            return None
        return full_path
    return user_path

def list_files(username, path=''):
    """列出用户目录下的文件"""
    full_path = get_full_path(username, path)
    if not full_path or not os.path.exists(full_path):
        return []
    
    result = []
    try:
        for item in os.listdir(full_path):
            item_path = os.path.join(full_path, item)
            is_dir = os.path.isdir(item_path)
            stat = os.stat(item_path)
            
            result.append({
                'name': item,
                'type': 'directory' if is_dir else 'file',
                'size': stat.st_size if not is_dir else 0,
                'modified': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S')
            })
        
        # Sort: directories first, then files
        result.sort(key=lambda x: (x['type'] != 'directory', x['name'].lower()))
        return result
    except Exception as e:
        print(f"Error listing files: {e}")
        return []

def upload_file(username, file_storage, path=''):
    """上传文件到用户目录"""
    # 检查配额
    used_storage = calculate_user_storage(username)
    file_content = file_storage.read()
    file_size = len(file_content)
    
    if used_storage + file_size > MAX_STORAGE_PER_USER:
        return False, f"存储空间不足，您已使用 {format_size(used_storage)}，还剩 {format_size(MAX_STORAGE_PER_USER - used_storage)}"
    
    full_path = get_full_path(username, path)
    if not full_path:
        return False, "路径无效"
    
    try:
        filename = safe_filename(file_storage.filename)
        file_path = os.path.join(full_path, filename)
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        
        with open(file_path, 'wb') as f:
            f.write(file_content)
        
        return True, "上传成功"
    except Exception as e:
        print(f"Error uploading file: {e}")
        return False, str(e)

def download_file(username, filename, path=''):
    """下载用户文件"""
    full_path = get_full_path(username, path)
    if not full_path:
        return None
    
    file_path = os.path.join(full_path, filename)
    if os.path.exists(file_path) and os.path.isfile(file_path):
        return file_path
    return None

def create_directory(username, dirname, path=''):
    """创建文件夹"""
    full_path = get_full_path(username, path)
    if not full_path:
        return False, "路径无效"
    
    try:
        dirname = safe_filename(dirname)
        dir_path = os.path.join(full_path, dirname)
        os.makedirs(dir_path, exist_ok=True)
        return True, "文件夹创建成功"
    except Exception as e:
        print(f"Error creating directory: {e}")
        return False, str(e)

def delete_item(username, filename, path=''):
    """删除文件或文件夹"""
    full_path = get_full_path(username, path)
    if not full_path:
        return False, "路径无效"
    
    try:
        item_path = os.path.join(full_path, filename)
        if os.path.isdir(item_path):
            import shutil
            shutil.rmtree(item_path)
        else:
            os.remove(item_path)
        return True, "删除成功"
    except Exception as e:
        print(f"Error deleting item: {e}")
        return False, str(e)

def format_size(size_bytes):
    """格式化文件大小"""
    if size_bytes == 0 or size_bytes == '-':
        return '0 B'
    try:
        size_bytes = int(size_bytes)
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if size_bytes < 1024.0:
                return f"{size_bytes:.2f} {unit}"
            size_bytes /= 1024.0
        return f"{size_bytes:.2f} PB"
    except:
        return '-'

# Template filter
app.jinja_env.filters['format_size'] = format_size

@app.route('/')
def index():
    """主页面"""
    username = get_current_user()
    if not username:
        return redirect(url_for('login_page'))
    
    current_path = request.args.get('path', '')
    files = list_files(username, current_path)
    
    # Build breadcrumb
    breadcrumb = []
    if current_path:
        parts = current_path.strip('/').split('/')
        acc = ''
        for part in parts:
            acc += '/' + part if acc else part
            breadcrumb.append({'name': part, 'path': acc})
    
    # Calculate storage
    used_storage = calculate_user_storage(username)
    storage_percent = (used_storage / MAX_STORAGE_PER_USER) * 100
    
    return render_template('index.html', 
                           username=username,
                           files=files, 
                           current_path=current_path,
                           breadcrumb=breadcrumb,
                           used_storage=used_storage,
                           max_storage=MAX_STORAGE_PER_USER,
                           storage_percent=storage_percent,
                           used_storage_formatted=format_size(used_storage),
                           max_storage_formatted=format_size(MAX_STORAGE_PER_USER))

@app.route('/login', methods=['GET', 'POST'])
def login_page():
    """登录页面（支持表单提交）"""
    if get_current_user():
        return redirect(url_for('index'))
    
    error = None
    
    if request.method == 'POST':
        # 表单提交方式
        username = request.form.get('username', '').strip()
        password = request.form.get('password', '')
        
        if not username or not password:
            error = '请输入用户名和密码'
        else:
            users = load_users()
            user = users.get(username)
            
            if not user or not check_password_hash(user['password_hash'], password):
                error = '用户名或密码错误'
            else:
                # 登录成功
                ensure_user_storage(username)
                session['username'] = username
                session.permanent = True
                # 打印日志验证
                print("写入session成功，username=", username, "session内容=", session.get("username"))
                return redirect(url_for('index'))
    
    return render_template('login.html', error=error)

@app.route('/register', methods=['GET', 'POST'])
def register_page():
    """注册页面（支持表单提交）"""
    if get_current_user():
        return redirect(url_for('index'))
    
    error = None
    
    if request.method == 'POST':
        # 表单提交方式
        username = request.form.get('username', '').strip()
        password = request.form.get('password', '')
        confirm_password = request.form.get('confirm_password', '')
        
        # 验证用户名
        if not safe_username(username):
            error = '用户名格式不正确，3-20位字母、数字、下划线或中文'
        # 验证密码
        elif len(password) < 6:
            error = '密码至少6位'
        elif password != confirm_password:
            error = '两次密码不一致'
        else:
            # 检查用户是否存在
            users = load_users()
            if username in users:
                error = '用户名已存在'
            else:
                # 创建用户
                users[username] = {
                    'password_hash': generate_password_hash(password),
                    'created_at': datetime.now().isoformat(),
                    'max_storage': MAX_STORAGE_PER_USER
                }
                save_users(users)
                
                # 创建用户目录
                ensure_user_storage(username)
                
                # 自动登录
                session['username'] = username
                session.permanent = True
                return redirect(url_for('index'))
    
    return render_template('register.html', error=error)

# ========== Auth APIs ==========

@app.route('/api/register', methods=['POST'])
def api_register():
    """注册 API"""
    data = request.get_json()
    username = data.get('username', '').strip()
    password = data.get('password', '')
    confirm_password = data.get('confirm_password', '')
    
    # 验证用户名
    if not safe_username(username):
        return jsonify({'success': False, 'error': '用户名格式不正确，3-20位字母、数字、下划线或中文'})
    
    # 验证密码
    if len(password) < 6:
        return jsonify({'success': False, 'error': '密码至少6位'})
    
    if password != confirm_password:
        return jsonify({'success': False, 'error': '两次密码不一致'})
    
    # 检查用户是否存在
    users = load_users()
    if username in users:
        return jsonify({'success': False, 'error': '用户名已存在'})
    
    # 创建用户
    users[username] = {
        'password_hash': generate_password_hash(password),
        'created_at': datetime.now().isoformat(),
        'max_storage': MAX_STORAGE_PER_USER
    }
    save_users(users)
    
    # 创建用户目录
    ensure_user_storage(username)
    
    # 自动登录
    session['username'] = username
    
    return jsonify({'success': True, 'message': '注册成功', 'username': username})

@app.route('/api/login', methods=['POST'])
def api_login():
    """登录 API"""
    data = request.get_json()
    username = data.get('username', '').strip()
    password = data.get('password', '')
    
    if not username or not password:
        return jsonify({'success': False, 'error': '请输入用户名和密码'})
    
    users = load_users()
    user = users.get(username)
    
    if not user or not check_password_hash(user['password_hash'], password):
        return jsonify({'success': False, 'error': '用户名或密码错误'})
    
    # 确保用户目录存在
    ensure_user_storage(username)
    
    session['username'] = username
    return jsonify({'success': True, 'message': '登录成功', 'username': username})

@app.route('/api/logout', methods=['POST'])
def api_logout():
    """退出登录 API"""
    session.pop('username', None)
    return jsonify({'success': True, 'message': '已退出登录'})

@app.route('/api/user-info')
def api_user_info():
    """获取用户信息 API"""
    is_logged_in, result = login_required_api()
    if not is_logged_in:
        return result
    
    username = result
    used_storage = calculate_user_storage(username)
    
    return jsonify({
        'success': True,
        'username': username,
        'used_storage': used_storage,
        'max_storage': MAX_STORAGE_PER_USER,
        'used_storage_formatted': format_size(used_storage),
        'max_storage_formatted': format_size(MAX_STORAGE_PER_USER),
        'storage_percent': (used_storage / MAX_STORAGE_PER_USER) * 100
    })

# ========== File APIs ==========

@app.route('/api/list')
def api_list():
    """API: 列出文件"""
    is_logged_in, result = login_required_api()
    if not is_logged_in:
        return result
    
    username = result
    path = request.args.get('path', '')
    files = list_files(username, path)
    return jsonify({'success': True, 'files': files, 'path': path})

@app.route('/api/upload', methods=['POST'])
def api_upload():
    """API: 上传文件"""
    is_logged_in, result = login_required_api()
    if not is_logged_in:
        return result
    
    username = result
    
    if 'file' not in request.files:
        return jsonify({'success': False, 'error': '没有选择文件'})
    
    file = request.files['file']
    if file.filename == '':
        return jsonify({'success': False, 'error': '没有选择文件'})
    
    path = request.form.get('path', '')
    success, message = upload_file(username, file, path)
    
    return jsonify({'success': success, 'message': message})

@app.route('/api/download')
def api_download():
    """API: 下载文件"""
    is_logged_in, result = login_required_api()
    if not is_logged_in:
        return redirect(url_for('login_page'))
    
    username = result
    filename = request.args.get('filename', '')
    path = request.args.get('path', '')
    
    if not filename:
        abort(400)
    
    file_path = download_file(username, filename, path)
    if not file_path:
        abort(404)
    
    # Determine MIME type
    mime_type, _ = mimetypes.guess_type(filename)
    if not mime_type:
        mime_type = 'application/octet-stream'
    
    return send_file(
        file_path,
        mimetype=mime_type,
        as_attachment=True,
        download_name=filename
    )

@app.route('/api/mkdir', methods=['POST'])
def api_mkdir():
    """API: 创建文件夹"""
    is_logged_in, result = login_required_api()
    if not is_logged_in:
        return result
    
    username = result
    data = request.get_json()
    dirname = data.get('dirname', '')
    path = data.get('path', '')
    
    if not dirname:
        return jsonify({'success': False, 'error': '请输入文件夹名称'})
    
    success, message = create_directory(username, dirname, path)
    return jsonify({'success': success, 'message': message})

@app.route('/api/delete', methods=['POST'])
def api_delete():
    """API: 删除文件/文件夹"""
    is_logged_in, result = login_required_api()
    if not is_logged_in:
        return result
    
    username = result
    data = request.get_json()
    filename = data.get('filename', '')
    path = data.get('path', '')
    
    if not filename:
        return jsonify({'success': False, 'error': '请选择要删除的文件'})
    
    success, message = delete_item(username, filename, path)
    return jsonify({'success': success, 'message': message})

@app.route('/api/storage-info')
def api_storage_info():
    """API: 获取存储信息"""
    is_logged_in, result = login_required_api()
    if not is_logged_in:
        return result
    
    username = result
    used_storage = calculate_user_storage(username)
    
    return jsonify({
        'success': True,
        'storage_type': 'local',
        'used_storage': used_storage,
        'max_storage': MAX_STORAGE_PER_USER,
        'used_storage_formatted': format_size(used_storage),
        'max_storage_formatted': format_size(MAX_STORAGE_PER_USER),
        'storage_percent': (used_storage / MAX_STORAGE_PER_USER) * 100
    })
# 1. 查看文本文件接口
@app.route('/api/view', methods=['GET'])
def api_view():
    if 'username' not in session:
        return jsonify({'success': False, 'error': '未登录'}), 401
    
    filename = request.args.get('filename')
    current_path = request.args.get('path', '')
    
    if not filename or filename == 'undefined':
        return jsonify({'success': False, 'error': '文件名无效'}), 400
    
    user_dir = get_user_storage_path(session['username'])
    file_path = os.path.join(user_dir, current_path, filename)
    
    # 安全检查：防止路径穿越
    if not os.path.abspath(file_path).startswith(os.path.abspath(user_dir)):
        return jsonify({'success': False, 'error': '非法路径'}), 403
    
    if not os.path.isfile(file_path):
        return jsonify({'success': False, 'error': '文件不存在'}), 404
    
    # 限制文件大小，比如只允许查看 2MB 以内的文本
    if os.path.getsize(file_path) > 2 * 1024 * 1024:
        return jsonify({'success': False, 'error': '文件过大，不支持在线预览'}), 400
    
    # 判断是否为文本文件
    text_extensions = ['.txt', '.md', '.py', '.json', '.html', '.css', '.js', '.log', '.csv', '.xml']
    ext = os.path.splitext(file_path)[1].lower()
    if ext not in text_extensions:
        return jsonify({'success': False, 'error': '该文件类型不支持在线查看'}), 400
    
    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
        return jsonify({'success': True, 'content': content, 'filename': filename})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)}), 500


# 2. 移动文件接口
@app.route('/api/move', methods=['POST'])
def api_move():
    if 'username' not in session:
        return jsonify({'success': False, 'error': '未登录'}), 401
    
    data = request.get_json()
    filename = data.get('filename')
    source_path = data.get('source_path', '')
    target_dir = data.get('target_dir', '')
    
    if not filename:
        return jsonify({'success': False, 'error': '文件名无效'}), 400
    
    user_dir = get_user_storage_path(session['username'])
    abs_source = os.path.join(user_dir, source_path, filename)
    abs_target_dir = os.path.join(user_dir, target_dir)
    
    # 安全检查
    if not os.path.abspath(abs_source).startswith(os.path.abspath(user_dir)) or \
       not os.path.abspath(abs_target_dir).startswith(os.path.abspath(user_dir)):
        return jsonify({'success': False, 'error': '非法路径'}), 403
    
    if not os.path.exists(abs_source):
        return jsonify({'success': False, 'error': '源文件不存在'}), 404
    
    try:
        os.makedirs(abs_target_dir, exist_ok=True)
        shutil.move(abs_source, os.path.join(abs_target_dir, filename))
        return jsonify({'success': True, 'message': '移动成功'})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)}), 500

# 配置区域：设置你的固定 API Token（请务必修改为一个复杂的字符串）
GRAPHICAL_API_TOKEN = "676767"

def check_graphical_auth():
    """检查图形化编程 API 的认证"""
    token = request.headers.get('X-API-Token') or request.args.get('token')
    print(f"！！！后端收到 token: '{token}' | 后端预期密码: '{GRAPHICAL_API_TOKEN}'")
    if token != GRAPHICAL_API_TOKEN:
        return False, jsonify({'success': False, 'error': 'Token 无效'}), 401
    return True, None

@app.route('/api/graphical/upload', methods=['POST'])
def graphical_upload():
    """图形化编程专用：通过 JSON 上传文件"""
    auth_result = check_graphical_auth()
    if not auth_result[0]: return auth_result[1], auth_result[2]
    
    data = request.get_json()
    username = data.get('username')  # 需要在请求中指定用户名
    filename = data.get('filename')
    content_base64 = data.get('content')  # Base64 编码的文件内容
    
    if not username or not filename or not content_base64:
        return jsonify({'success': False, 'error': '缺少必要参数'}), 400
    
    import base64
    try:
        file_content = base64.b64decode(content_base64)
    except:
        return jsonify({'success': False, 'error': '内容解码失败'}), 400
    
    # 检查用户是否存在（可选）
    users = load_users()
    if username not in users:
        return jsonify({'success': False, 'error': '用户不存在'}), 404
    
    ensure_user_storage(username)
    # 调用原有的上传逻辑
    class FileStorage:
        def __init__(self, filename, content):
            self.filename = filename
            self.content = content
        def read(self):
            return self.content
    
    success, message = upload_file(username, FileStorage(filename, file_content))
    return jsonify({'success': success, 'message': message})

@app.route('/api/graphical/download', methods=['GET'])
def graphical_download():
    """图形化编程专用：通过 GET 下载文件"""
    auth_result = check_graphical_auth()
    if not auth_result[0]: return auth_result[1], auth_result[2]
    
    username = request.args.get('username')
    filename = request.args.get('filename')
    path = request.args.get('path', '')
    
    if not username or not filename:
        return jsonify({'success': False, 'error': '缺少用户名或文件名'}), 400
    
    users = load_users()
    if username not in users:
        return jsonify({'success': False, 'error': '用户不存在'}), 404
    
    file_path = download_file(username, filename, path)
    if not file_path:
        return jsonify({'success': False, 'error': '文件不存在'}), 404
    
    # 返回 Base64 编码的文件内容
    import base64
    try:
        with open(file_path, 'rb') as f:
            file_content = f.read()
        content_base64 = base64.b64encode(file_content).decode('utf-8')
        return jsonify({
            'success': True, 
            'filename': filename,
            'content': content_base64
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)}), 500

@app.route('/api/graphical/list', methods=['GET'])
def graphical_list():
    """图形化编程专用：列出文件"""
    auth_result = check_graphical_auth()
    if not auth_result[0]: return auth_result[1], auth_result[2]
    
    username = request.args.get('username')
    path = request.args.get('path', '')
    
    if not username:
        return jsonify({'success': False, 'error': '缺少用户名'}), 400
    
    users = load_users()
    if username not in users:
        return jsonify({'success': False, 'error': '用户不存在'}), 404
    
    files = list_files(username, path)
    return jsonify({'success': True, 'files': files, 'path': path})

@app.route('/api/graphical/delete', methods=['POST'])
def graphical_delete():
    """图形化编程专用：删除文件"""
    auth_result = check_graphical_auth()
    if not auth_result[0]: return auth_result[1], auth_result[2]
    
    data = request.get_json()
    username = data.get('username')
    filename = data.get('filename')
    path = data.get('path', '')
    
    if not username or not filename:
        return jsonify({'success': False, 'error': '缺少用户名或文件名'}), 400
    
    users = load_users()
    if username not in users:
        return jsonify({'success': False, 'error': '用户不存在'}), 404
    
    success, message = delete_item(username, filename, path)
    return jsonify({'success': success, 'message': message})
    
if __name__ == '__main__':
    port = int(os.environ.get('PORT', 7860))
    app.run(host='0.0.0.0', port=port, debug=False)