Spaces:

albumup
/

file

Sleeping

App Files Files Community

albumup commited on Jan 17, 2025

Commit

2218121

verified ·

1 Parent(s): 991d698

Update app.py

Browse files

Files changed (1) hide show

app.py +104 -71

app.py CHANGED Viewed

@@ -1,24 +1,38 @@
 from fastapi import FastAPI, File, UploadFile, Request
-from fastapi.responses import HTMLResponse, RedirectResponse, StreamingResponse
-import requests
-import asyncio
 import os
 app = FastAPI()
 HTML_CONTENT = """
 <!DOCTYPE html>
 <html lang="en">
 <head>
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
-    <title>File Upload</title>
 </head>
 <body>
-    <h1>Upload File</h1>
-    <form action="/upload" method="post" enctype="multipart/form-data">
-        <input type="file" name="file" accept="*/*" required>
-        <button type="submit">Upload</button>
     </form>
 </body>
 </html>
@@ -28,76 +42,95 @@ HTML_CONTENT = """
 async def index():
     return HTML_CONTENT
-@app.post("/upload")
-async def handle_upload(request: Request, file: UploadFile = File(...)):
-    if not file:
-        return {"error": "No file uploaded"}, 400
-    cookies = await get_cookies()
-    if 'csrftoken' not in cookies or 'sessionid' not in cookies:
-        return {"error": "Failed to get cookies"}, 500
-    upload_result = await initiate_upload(cookies, file.filename, file.content_type)
-    if not upload_result or 'upload_url' not in upload_result:
-        return {"error": "Failed to initiate upload"}, 500
-    file_content = await file.read()
-    upload_success = await retry_upload(upload_result['upload_url'], file_content, file.content_type)
-    if not upload_success:
-        return {"error": "Upload failed"}, 500
-    base_url = str(request.base_url).rstrip('/')
-    redirect_url = f"{base_url}/upload/{upload_result['serving_url'].split('/pbxt/')[1]}"
-    return RedirectResponse(url=redirect_url, status_code=303)
-@app.get("/upload/{path:path}")
-async def handle_stream(path: str, request: Request):
-    original_url = f'https://replicate.delivery/pbxt/{path}'
-    range_header = request.headers.get('Range')
-    headers = {'Range': range_header} if range_header else {}
-    response = requests.get(original_url, headers=headers, stream=True)
-    def generate():
-        for chunk in response.iter_content(chunk_size=8192):
-            yield chunk
-    headers = dict(response.headers)
-    headers['Access-Control-Allow-Origin'] = '*'
-    return StreamingResponse(generate(), headers=headers, status_code=response.status_code)
-async def get_cookies():
-    try:
-        response = requests.get('https://replicate.com/levelsio/neon-tokyo', headers={
-            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
-        })
-        return dict(response.cookies)
-    except Exception:
-        return {}
-async def initiate_upload(cookies, filename, content_type):
-    url = f"https://replicate.com/api/upload/{filename}?content_type={content_type}"
-    headers = {
-        'X-CSRFToken': cookies['csrftoken'],
-        'Referer': 'https://replicate.com/levelsio/neon-tokyo',
-        'Origin': 'https://replicate.com'
-    }
-    response = requests.post(url, cookies=cookies, headers=headers)
-    return response.json()
-async def retry_upload(upload_url, file_content, content_type, max_retries=5):
-    delay = 1
-    for _ in range(max_retries):
-        try:
-            response = requests.put(upload_url, data=file_content, headers={'Content-Type': content_type})
-            if response.status_code == 200:
-                return True
-        except Exception:
-            pass
-        await asyncio.sleep(delay)
-        delay *= 2
-    return False
 if __name__ == "__main__":
     import uvicorn

 from fastapi import FastAPI, File, UploadFile, Request
+from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse, FileResponse
+from fastapi.staticfiles import StaticFiles
+import shutil
 import os
+import uuid
+import zipfile
+from typing import List
+import json
 app = FastAPI()
+# Create directories if they don't exist
+UPLOAD_DIR = "uploads"
+ALBUM_DIR = "albums"
+os.makedirs(UPLOAD_DIR, exist_ok=True)
+os.makedirs(ALBUM_DIR, exist_ok=True)
+# Serve static files
+app.mount("/static", StaticFiles(directory=UPLOAD_DIR), name="static")
 HTML_CONTENT = """
 <!DOCTYPE html>
 <html lang="en">
 <head>
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Album Upload</title>
 </head>
 <body>
+    <h1>Create Album</h1>
+    <form action="/create-album" method="post" enctype="multipart/form-data">
+        <input type="text" name="album_name" placeholder="Album Name" required><br><br>
+        <input type="file" name="files" accept="*/*" multiple required><br><br>
+        <button type="submit">Create Album</button>
     </form>
 </body>
 </html>
 async def index():
     return HTML_CONTENT
+@app.post("/create-album")
+async def create_album(album_name: str, files: List[UploadFile]):
+    if not files:
+        return JSONResponse({"error": "No files uploaded"}, status_code=400)
+    # Create unique album ID
+    album_id = str(uuid.uuid4())
+    album_path = os.path.join(ALBUM_DIR, album_id)
+    os.makedirs(album_path, exist_ok=True)
+    # Save album metadata
+    album_info = {
+        "album_name": album_name,
+        "files": []
+    }
+    # Save uploaded files
+    for file in files:
+        file_path = os.path.join(album_path, file.filename)
+        with open(file_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+        album_info["files"].append(file.filename)
+    # Save album metadata
+    with open(os.path.join(album_path, "album_info.json"), "w") as f:
+        json.dump(album_info, f)
+    album_url = f"/album/{album_id}"
+    return JSONResponse({
+        "message": "Album created successfully",
+        "album_url": album_url,
+        "share_url": f"/share/{album_id}"
+    })
+@app.get("/album/{album_id}")
+async def view_album(album_id: str):
+    album_path = os.path.join(ALBUM_DIR, album_id)
+    if not os.path.exists(album_path):
+        return JSONResponse({"error": "Album not found"}, status_code=404)
+    with open(os.path.join(album_path, "album_info.json")) as f:
+        album_info = json.load(f)
+    return HTMLResponse(f"""
+        <html>
+            <head><title>{album_info['album_name']}</title></head>
+            <body>
+                <h1>{album_info['album_name']}</h1>
+                <div>
+                    <h3>Files:</h3>
+                    <ul>
+                        {"".join(f'<li><a href="/download/{album_id}/{file}">{file}</a></li>' for file in album_info['files'])}
+                    </ul>
+                    <a href="/download/{album_id}/all">Download All Files (ZIP)</a>
+                </div>
+            </body>
+        </html>
+    """)
+@app.get("/share/{album_id}")
+async def share_album(album_id: str, request: Request):
+    album_path = os.path.join(ALBUM_DIR, album_id)
+    if not os.path.exists(album_path):
+        return JSONResponse({"error": "Album not found"}, status_code=404)
+    base_url = str(request.base_url).rstrip('/')
+    share_url = f"{base_url}/album/{album_id}"
+    return JSONResponse({"share_url": share_url})
+@app.get("/download/{album_id}/{filename}")
+async def download_file(album_id: str, filename: str):
+    file_path = os.path.join(ALBUM_DIR, album_id, filename)
+    if not os.path.exists(file_path):
+        return JSONResponse({"error": "File not found"}, status_code=404)
+    return FileResponse(file_path, filename=filename)
+@app.get("/download/{album_id}/all")
+async def download_all(album_id: str):
+    album_path = os.path.join(ALBUM_DIR, album_id)
+    if not os.path.exists(album_path):
+        return JSONResponse({"error": "Album not found"}, status_code=404)
+    zip_path = os.path.join(album_path, "album.zip")
+    with zipfile.ZipFile(zip_path, 'w') as zipf:
+        for file in os.listdir(album_path):
+            if file != "album.zip" and file != "album_info.json":
+                zipf.write(os.path.join(album_path, file), file)
+    return FileResponse(zip_path, filename=f"album_{album_id}.zip")
 if __name__ == "__main__":
     import uvicorn