import os
import io
import shutil
import tempfile
from flask import Flask, render_template_string, request, redirect, send_file, jsonify
from huggingface_hub import HfApi, HfFileSystem, login, upload_file, delete_file, move_file
# Set env vars
REPO_ID = os.getenv("REPO_ID")
HF_TOKEN = os.getenv("HF_TOKEN")
app = Flask(__name__)
api = HfApi()
fs = HfFileSystem(token=HF_TOKEN)
login(token=HF_TOKEN)
TEMPLATE = """
HuggingFace Drive
HuggingFace Drive - {{ path or 'Root' }}
{% if parent %}
{% endif %}
{% for item in items %}
{% if item['type'] == 'dir' %}
📁 {{ item['name'] }}
{% else %}
📄 {{ item['name'] }}
{% endif %}
{% endfor %}
"""
def list_dataset_files(path=""):
files = fs.ls(f"datasets/{REPO_ID}/{path}", detail=True)
folders = []
items = []
seen = set()
for file in files:
relative_path = file["name"].replace(f"datasets/{REPO_ID}/", "")
parts = relative_path.split("/")
if len(parts) > 1:
folder_path = "/".join(parts[:1])
full_folder = f"{path}/{folder_path}".strip("/")
if full_folder not in seen:
folders.append({"type": "dir", "name": folder_path, "path": full_folder})
seen.add(full_folder)
elif len(parts) == 1:
items.append({"type": "file", "name": parts[0], "path": f"{path}/{parts[0]}".strip("/")})
return sorted(folders + items, key=lambda x: x["name"].lower())
@app.route("/", methods=["GET"])
def index():
path = request.args.get("path", "").strip("/")
parent = "/".join(path.split("/")[:-1]) if path else None
items = list_dataset_files(path)
return render_template_string(TEMPLATE, items=items, path=path, parent=parent)
@app.route("/download")
def download():
path = request.args.get("path")
with tempfile.NamedTemporaryFile(delete=False) as tmp:
fs.download(f"datasets/{REPO_ID}/{path}", tmp.name)
return send_file(tmp.name, as_attachment=True, download_name=os.path.basename(path))
@app.route("/upload", methods=["POST"])
def upload():
file = request.files["file"]
path = request.form.get("path", "").strip("/")
dest = f"{path}/{file.filename}".strip("/")
with tempfile.NamedTemporaryFile() as tmp:
file.save(tmp.name)
upload_file(path_or_fileobj=tmp.name, path_in_repo=dest, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN)
return redirect(f"/?path={path}")
@app.route("/delete", methods=["POST"])
def delete():
data = request.get_json()
delete_file(repo_id=REPO_ID, path_in_repo=data["path"], repo_type="dataset", token=HF_TOKEN)
return jsonify({"status": "deleted"})
@app.route("/create_folder", methods=["POST"])
def create_folder():
data = request.get_json()
path = data["path"].strip("/")
dummy_path = f"{path}/.keep"
with tempfile.NamedTemporaryFile() as tmp:
tmp.write(b"")
tmp.flush()
upload_file(path_or_fileobj=tmp.name, path_in_repo=dummy_path, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN)
return jsonify({"status": "folder_created"})
@app.route("/rename", methods=["POST"])
def rename():
data = request.get_json()
move_file(repo_id=REPO_ID, src_path=data["old_path"], dst_path=data["new_path"], repo_type="dataset", token=HF_TOKEN)
return jsonify({"status": "renamed"})
if __name__ == "__main__":
app.run(debug=True, port=7860)