app.py

import os
import gradio as gr
from modelscope.pipelines import pipeline
from modelscope.outputs import OutputKeys
import tempfile
import uplink_python
from uplink_python.uplink import Uplink
from uplink_python.module_classes import Permission, SharePrefix, Config
import datetime

# Define the text to video synthesis pipeline
p = pipeline("text-to-video-synthesis", "damo/text-to-video-synthesis")

# Storj configuration information
MY_API_KEY = "1H9SGajiNtFMJtjoseMKuhiLpuPHq3Sy7of6eNUjs8yeRE1DkmRF4XJsTm1ExXLbsxqJ4t5fRZobh4tyFmEdocNCFsJhukakve5C4V59TEEzqTaLVdjFc7ovCkmuRGotR9c2aqDdt5RThRTtKcXaVE5uCtuFQ9tZvnsX"
MY_SATELLITE = "12EayRS2V1kEsWESU9QMRseFhdxYxKicsiFmxrsLZHeLUtdps3S@us1.storj.io:7777"
MY_BUCKET = "huggingface-demo"
MY_ENCRYPTION_PASSPHRASE = "."

# Initialize Storj Uplink
uplink = Uplink()

# Implement the generate_video function
def generate_video(text_input: str, duration: int) -> str:
    test_text = {'text': text_input, 'duration': duration}
    output_video_path = p(test_text,)[OutputKeys.OUTPUT_VIDEO]

    access = uplink.request_access_with_passphrase(MY_SATELLITE, MY_API_KEY, MY_ENCRYPTION_PASSPHRASE)
    project = access.open_project()
    project.ensure_bucket(MY_BUCKET)
    video_key = os.path.basename(output_video_path)
    with open(output_video_path, "rb") as f:
        upload = project.upload_object(MY_BUCKET, video_key)
        upload.write_file(f)
        upload.commit()

    permissions = Permission(allow_download=True)
    shared_prefix = [SharePrefix(bucket=MY_BUCKET, prefix=video_key)]
    shared_access = access.share(permissions, shared_prefix)
    serialized_shared_access = shared_access.serialize()
    deserialized_shared_access = uplink.parse_access(serialized_shared_access)
    project_with_shared_access = deserialized_shared_access.open_project()

    expiration = datetime.timedelta(hours=1)
    expires_at = (datetime.datetime.utcnow() + expiration).timestamp()
    storj_video_url = f"{MY_SATELLITE}/{MY_BUCKET}/{video_key}?access={serialized_shared_access}&expiresAt={expires_at}"
    download = project_with_shared_access.download_object(MY_BUCKET, video_key)

    chunk_size = 1024 * 1024
    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
    try:
        with open(temp_file.name, "wb") as dst:
            while True:
                try:
                    chunk = download.read(chunk_size)  # Read a chunk from the download
                except Exception as e:
                    print(f"Error while reading from download: {e}")
                    break

                if not chunk:
                    break

                dst.write(chunk)  # Write the chunk to the destination file
    except Exception as e:
        print(f"Error while writing to file: {e}")

    temp_file.seek(0)
    project.close()

    return temp_file.name, storj_video_url

# Define the tweet_video function
def tweet_video(text_input: str, duration: int, button_click: bool) -> str:
    if button_click:
        tweet_text = f"Check out this amazing video generated by Text-to-Video Synthesis! {text_input}"
        tweet_url = f"https://twitter.com/intent/tweet?text={tweet_text}"
        return tweet_url
    else:
        return "Hello " + text_input + "!"

# Set theme
my_theme = gr.Theme.from_hub("bethecloud/storj_theme")

# Define the input components
text_input = gr.inputs.Textbox(lines=3, placeholder="Enter a short text to generate video")
duration = gr.inputs.Slider(minimum=1, maximum=10, default=5, label="Duration (seconds)")
button = gr.inputs.Button(label="Share on Twitter")

# Define the output components
video_output = gr.outputs.Video(label="Generated Video")
url_output = gr.outputs.Textbox(label="Storj Video URL")

# Create the Gradio interface
iface = gr.Interface(
    fn=tweet_video,
    inputs=[text_input, duration],
    outputs=[video_output, url_output],
    title="Text-to-Video Synthesis",
    description="Generate a video based on the input text and share it on Twitter.",
    theme=my_theme
)

# Define the action for the "Share on Twitter" button
def on_button_click():
    storj_video_url = iface.get_component("Storj Video URL").data
    text_input = iface.get_component("Enter a short text to generate video").data
    duration = iface.get_component("Duration (seconds)").data
    tweet_url = tweet_video(text_input, duration, True)
    gr.Interface.open_url(tweet_url)

# Add the "Share on Twitter" button and its action to the interface
iface.set_component("Share on Twitter", gr.Button(on_button_click, label="Share on Twitter"))

# Launch the interface
iface.launch(share=True, debug=True)