import os import tempfile from flask import Flask, request, jsonify, send_file from wsgidav.wsgidav_app import WsgiDAVApp from hf_bucket_provider import HfBucketProvider HF_TOKEN = os.environ.get("HF_TOKEN") if not HF_TOKEN: raise ValueError("HF_TOKEN environment variable not set. Please add it to Space Secrets.") BUCKET_ID = "nagose/filebed" app = Flask(__name__) # WebDAV 配置：禁用目录浏览器，确保响应为标准的 WebDAV XML 格式 dav_config = { "provider_mapping": { "/": HfBucketProvider(BUCKET_ID, HF_TOKEN) }, "http_authenticator": { "accept_basic": True, "accept_digest": False, "default_to_digest": False, }, "simple_dc": { "user_mapping": {"*": True} }, "dir_browser": {"enable": False}, # 禁用内置的 HTML 目录列表，避免干扰 WebDAV 响应 "verbose": 1, } dav_app = WsgiDAVApp(dav_config) @app.route('/webdav', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'PROPFIND', 'PROPPATCH', 'MKCOL', 'COPY', 'MOVE', 'LOCK', 'UNLOCK']) @app.route('/webdav/', methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'PROPFIND', 'PROPPATCH', 'MKCOL', 'COPY', 'MOVE', 'LOCK', 'UNLOCK']) def webdav_handler(path): environ = request.environ environ['PATH_INFO'] = '/' + path environ['SCRIPT_NAME'] = '/webdav' return dav_app(environ, lambda status, headers, exc_info=None: None) # 调试路由：测试 WebDAV 提供者的根目录成员 @app.route('/debug/webdav-root') def debug_webdav_root(): try: provider = HfBucketProvider(BUCKET_ID, HF_TOKEN) members = provider._list_directory('/') return jsonify({'members': members}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/debug/list') def debug_list(): from huggingface_hub import list_bucket_tree try: items = list_bucket_tree( bucket_id=BUCKET_ID, recursive=True, token=HF_TOKEN ) items_list = list(items) result = [] for item in items_list: entry = { 'path': item.path, 'type': item.type, 'size': item.size, } if hasattr(item, 'last_modified') and item.last_modified: entry['last_modified'] = item.last_modified.isoformat() else: entry['last_modified'] = None result.append(entry) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 # 美化后的前端界面（包含调试按钮） HTML_CONTENT = """ HF Bucket 管理器

HF Bucket 管理器

WebDAV 服务已挂载在 /webdav 路径，您可以使用任何 WebDAV 客户端连接。

例如：https://你的-space地址/webdav

上传测试

文件列表

""" @app.route('/') def index(): return HTML_CONTENT @app.route('/upload', methods=['POST']) def upload(): from huggingface_hub import batch_bucket_files if 'file' not in request.files: return jsonify({'error': 'No file part'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': 'No selected file'}), 400 filename = file.filename with tempfile.NamedTemporaryFile(delete=False) as tmp: file.save(tmp.name) try: batch_bucket_files( bucket_id=BUCKET_ID, add=[(tmp.name, filename)], token=HF_TOKEN ) print(f"[UPLOAD] Success: {filename}") except Exception as e: print(f"[UPLOAD] Error: {e}") return jsonify({'error': str(e)}), 500 finally: os.unlink(tmp.name) return jsonify({'success': True, 'filename': filename}) @app.route('/list') def list_files(): from huggingface_hub import list_bucket_tree try: items = list_bucket_tree( bucket_id=BUCKET_ID, recursive=True, token=HF_TOKEN ) items_list = list(items) files = [item.path for item in items_list if item.type == 'file'] return jsonify(files) except Exception as e: print(f"[LIST] Error: {e}") return jsonify({'error': str(e)}), 500 @app.route('/file/') def get_file(filename): from huggingface_hub import download_bucket_files try: with tempfile.TemporaryDirectory() as tmpdir: local_path = os.path.join(tmpdir, filename) download_bucket_files( bucket_id=BUCKET_ID, files=[(filename, local_path)], token=HF_TOKEN ) return send_file(local_path, as_attachment=True, download_name=filename) except Exception as e: print(f"[DOWNLOAD] Error: {e}") return jsonify({'error': str(e)}), 500 @app.route('/delete/', methods=['DELETE']) def delete_file(filename): from huggingface_hub import batch_bucket_files try: batch_bucket_files( bucket_id=BUCKET_ID, delete=[filename], token=HF_TOKEN ) print(f"[DELETE] Success: {filename}") return jsonify({'success': True}) except Exception as e: print(f"[DELETE] Error: {e}") return jsonify({'error': str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)