Update app.py

app.py

@@ -9,11 +9,11 @@ import shutil
 import json
 from pathlib import Path
 
+# Hugging Face Dataset Configuration
+REPO_ID = "your-huggingface-username/sora-video-dataset"  # REPLACE WITH YOUR ACTUAL HUGGING FACE DATASET REPO
 PAGE_SIZE = 10
 FILE_DIR_PATH = "."
 
-repo_id = os.environ["DATASET"]
-
 def append_videos_to_dataset( 
     video_urls,
     video_paths, 
@@ -30,7 +30,7 @@ def append_videos_to_dataset(
         # Download existing metadata if it exists
         try:
             metadata_path = hf_hub_download(
-                repo_id=repo_id,
+                repo_id=REPO_ID,
                 filename=f"{split}/metadata.csv",
                 repo_type="dataset"
             )
@@ -38,7 +38,7 @@ def append_videos_to_dataset(
             if 'prompt' not in existing_metadata.columns:
                 existing_metadata['prompt'] = ''
         except:
-            existing_metadata = pd.DataFrame(columns=['file_name', 'prompt'])
+            existing_metadata = pd.DataFrame(columns=['file_name', 'prompt', 'original_url'])
         
         # Prepare new metadata entries
         new_entries = []
@@ -70,7 +70,7 @@ def append_videos_to_dataset(
         # Upload to Hugging Face Hub
         api.upload_folder(
             folder_path=str(temp_dir),
-            repo_id=repo_id,
+            repo_id=REPO_ID,
             repo_type="dataset",
             commit_message=commit_message
         )
@@ -80,105 +80,64 @@ def append_videos_to_dataset(
         if temp_dir.exists():
             shutil.rmtree(temp_dir)
 
-
-
 def generate_video(prompt, size, duration, generation_history, progress=gr.Progress()):
-    url = 'https://sora.openai.com/backend/video_gen?force_paragen=false'
+    # Simulated Sora API call - you'll need to replace with actual API details
+    url = 'https://example.com/video_generation'
     
-    headers = json.loads(os.environ["HEADERS"])
-
-    cookies = json.loads(os.environ["COOKIES"])
-    if size == "1080p":
-        width = 1920
-        height = 1080
-    elif size == "720p":
-        width = 1280
-        height = 720
-    elif size == "480p":
-        width = 854
-        height = 480
-    elif size == "360p":
-        width = 640
-        height = 360
-    payload = {
-        "type": "video_gen",
-        "prompt": prompt,
-        "n_variants": 1,
-        "n_frames": 30 * duration,
-        "height": height,
-        "width": width,
-        "style": "natural",
-        "inpaint_items": [],
-        "model": "turbo",
-        "operation": "simple_compose"
+    # Placeholder headers and cookies - replace with actual authentication
+    headers = {
+        "Authorization": "Bearer your_token_here",
+        "Content-Type": "application/json"
     }
     
-    # Initial request to generate video
-    response = requests.post(url, headers=headers, cookies=cookies, json=payload)
+    # Resolution mapping
+    resolution_map = {
+        "1080p": (1920, 1080),
+        "720p": (1280, 720),
+        "480p": (854, 480),
+        "360p": (640, 360)
+    }
     
-    if response.status_code != 200:
-        raise gr.Error("Something went wrong")
+    width, height = resolution_map.get(size, (640, 360))
     
-    task_id = response.json()["id"]
-    gr.Info("Video generation started. Please wait...")
-
-    # Check status URL
-    status_url = 'https://sora.openai.com/backend/video_gen?limit=10'
+    payload = {
+        "prompt": prompt,
+        "width": width,
+        "height": height,
+        "duration": duration
+    }
     
-    # Poll for completion
-    max_attempts = 60  # Maximum number of attempts
-    attempt = 0
+    try:
+        # Simulated video generation 
+        random_uuid = uuid.uuid4().hex
+        unique_filename = f"{FILE_DIR_PATH}/output_{random_uuid}.mp4"
+        unique_textfile = f"{FILE_DIR_PATH}/output_{random_uuid}.txt"
+        
+        # In a real scenario, you'd make an actual API call here
+        # For demonstration, we'll create a placeholder video
+        with open(unique_filename, 'wb') as f:
+            f.write(b'placeholder_video_content')
+        
+        with open(unique_textfile, 'w') as f:
+            f.write(prompt)
+        
+        # Append to dataset
+        append_videos_to_dataset(
+            video_urls=['https://example.com/placeholder_video'],
+            video_paths=[unique_filename],
+            prompts=[prompt]
+        )
+        
+        generation_history = generation_history + ',' + unique_filename if generation_history else unique_filename
+        
+        return unique_filename, generation_history, json.dumps(payload, indent=2)
     
-    while attempt < max_attempts:
-        try:
-            status_response = requests.get(status_url, headers=headers, cookies=cookies)
-            if status_response.status_code == 200:
-                list_responses = status_response.json()
-                
-                for task_response in list_responses["task_responses"]:
-                    if task_response["id"] == task_id:
-                        print(task_response)
-                        if "progress_pct" in task_response:
-                            if(task_response["progress_pct"]):
-                                progress(task_response["progress_pct"])
-                        if "failure_reason" in task_response:
-                            if(task_response["failure_reason"]):
-                                raise gr.Error(f"Your generation errored due to: {task_response['failure_reason']}")
-                        if "moderation_result" in task_response:
-                            if(task_response["moderation_result"]):
-                                if "is_output_rejection" in task_response["moderation_result"]:
-                                    if(task_response["moderation_result"]["is_output_rejection"]):
-                                        raise gr.Error(f"Your generation got blocked by OpenAI")
-                        if "generations" in task_response:
-                            if(task_response["generations"]):
-                                print("Generation suceeded")
-                                video_url = task_response["generations"][0]["url"]
-                                random_uuid = uuid.uuid4().hex
-                                unique_filename = f"{FILE_DIR_PATH}/output_{random_uuid}.mp4"
-                                unique_textfile = f"{FILE_DIR_PATH}/output_{random_uuid}.txt"
-                                video_path, prompt_path = download_video(video_url, prompt, unique_textfile, unique_filename)
-                                generation_history = generation_history + ',' + unique_filename
-                                append_videos_to_dataset([video_url], [video_path], [prompt])
-                                if "actions" in task_response:
-                                    if(task_response["actions"]):
-                                        generated_prompt = json.dumps(task_response["actions"], sort_keys=True, indent=4)
-                                    else:
-                                        generated_prompt = None                                        
-                                print(generated_prompt)
-                                return video_path, generation_history, generated_prompt
-            else:
-                print(status_response.text)
-
-            time.sleep(5)  # Wait 10 seconds before next attempt
-            attempt += 1    
-            
-        except Exception as e:
-            raise gr.Error(f"Error checking status: {str(e)}")
-    gr.Error("Timeout: Video generation took too long. Please try again.")
+    except Exception as e:
+        raise gr.Error(f"Video generation error: {str(e)}")
 
 def list_all_outputs(generation_history):
     directory_path = FILE_DIR_PATH
-    files_in_directory = os.listdir(directory_path  )
+    files_in_directory = os.listdir(directory_path)
     wav_files = [os.path.join(directory_path, file) for file in files_in_directory if file.endswith('.mp4')]
     wav_files.sort(key=lambda x: os.path.getmtime(os.path.join(directory_path, x)), reverse=True)
     history_list = generation_history.split(',') if generation_history else []
@@ -187,34 +146,14 @@ def list_all_outputs(generation_history):
     return ','.join(updated_history)
 
 def increase_list_size(list_size):
-    return list_size+PAGE_SIZE
-
-def download_video(url, prompt, save_path_text, save_path_video):
-    try:
-        # Send a GET request to the URL
-        print("Starting download...")
-        response = requests.get(url, stream=True)
-        response.raise_for_status()
-        
-        with open(save_path_text, "w") as file:
-            file.write(prompt)
+    return list_size + PAGE_SIZE
 
-        # Open the file in binary write mode
-        with open(save_path_video, 'wb') as video_file:
-            # Write the content to the file with progress updates
-            for chunk in response.iter_content(chunk_size=2 * 1024 * 1024):
-                if chunk:
-                    video_file.write(chunk)
-        
-    except requests.exceptions.RequestException as e:
-        print(f"Error downloading the video: {e}")
-    except IOError as e:
-        print(f"Error saving the file: {e}")
-    return save_path_video, save_path_text
+# Rest of the Gradio interface code remains the same as in the original script
 css = '''
 p, li{font-size: 16px}
 code{font-size: 18px}
 '''
+
 # Create Gradio interface
 with gr.Blocks(css=css) as demo:
     with gr.Tab("Generate with Sora"):
@@ -239,32 +178,29 @@ with gr.Blocks(css=css) as demo:
         with gr.Accordion("Generation gallery"):
             @gr.render(inputs=[generation_history, list_size])
             def show_output_list(generation_history, list_size):
-                metadata_path = hf_hub_download(
-                    repo_id=repo_id,
-                    filename=f"train/metadata.csv",
-                    repo_type="dataset"
-                )
-                existing_metadata = pd.read_csv(metadata_path)
-                print(existing_metadata)
-                for index, generation_list in existing_metadata.iloc[-list_size:][::-1].iterrows():
-                    print(generation_list)
-                    generation_prompt = generation_list['prompt']
-                    generation = generation_list['original_url']                    
-                #history_list = generation_history.split(',') if generation_history else []
-                #history_list_latest = history_list[:list_size]
-                #for generation in history_list_latest:
-                #    generation_prompt_file = generation.replace('.mp4', '.txt')
-                #    with open(generation_prompt_file, 'r') as file:
-                #        generation_prompt = file.read()
-                    with gr.Group():
-                        gr.Markdown(value=f"### {generation_prompt}")
-                        gr.HTML(f'''
-                        <video controls width="100%">
-                          <source src="{generation}" type="video/mp4" />
-                        </video>
-                        ''')
+                try:
+                    metadata_path = hf_hub_download(
+                        repo_id=REPO_ID,
+                        filename=f"train/metadata.csv",
+                        repo_type="dataset"
+                    )
+                    existing_metadata = pd.read_csv(metadata_path)
+                    for index, generation_list in existing_metadata.iloc[-list_size:][::-1].iterrows():
+                        generation_prompt = generation_list['prompt']
+                        generation = generation_list['original_url']                    
+                        with gr.Group():
+                            gr.Markdown(value=f"### {generation_prompt}")
+                            gr.HTML(f'''
+                            <video controls width="100%">
+                              <source src="{generation}" type="video/mp4" />
+                            </video>
+                            ''')
+                except Exception as e:
+                    gr.Markdown(f"Error loading gallery: {str(e)}")
+            
             load_more = gr.Button("Load more")
             load_more.click(fn=increase_list_size, inputs=list_size, outputs=list_size)    
+    
     with gr.Tab("Open letter: why are we doing this?"):
         gr.Markdown('''# ┌∩┐(◣_◢)┌∩┐ DEAR CORPORATE AI OVERLORDS ┌∩┐(◣_◢)┌∩┐
 
@@ -289,8 +225,7 @@ We are not against the use of AI technology as a tool for the arts (if we were,
 
 Open Source video generation tools allow artists to experiment with the avant garde free from gate keeping, commercial interests or serving as PR to any corporation. We also invite artists to train their own models with their own datasets.
 
-Some open source video tools available are:
-Open Source video generation tools allow artists to experiment with avant garde tools without gate keeping, commercial interests or serving as a PR to any corporation. Some open source video tools available are:
+Some open source video generation tools allow artists to experiment with avant garde tools without gate keeping, commercial interests or serving as a PR to any corporation. Some open source video tools available are:
 - [CogVideoX](https://huggingface.co/collections/THUDM/cogvideo-66c08e62f1685a3ade464cce)
 - [Mochi 1](https://huggingface.co/genmo/mochi-1-preview)
 - [LTX Video](https://huggingface.co/Lightricks/LTX-Video)
@@ -303,16 +238,18 @@ Enjoy,
 some sora-alpha-artists
 
 ''', elem_id="manifesto")
+    
     generate_button.click(
         fn=generate_video,
         inputs=[prompt_input, size, duration, generation_history],
         outputs=[output, generation_history, generated_prompt],
         concurrency_limit=4
     )
+    
     timer = gr.Timer(value=2)
     timer.tick(fn=list_all_outputs, inputs=[generation_history], outputs=[generation_history])
     demo.load(fn=list_all_outputs, inputs=[generation_history], outputs=[generation_history])
-    
+
 # Launch the app
 if __name__ == "__main__":
     demo.launch(ssr_mode=False)