init code

.gitignore

@@ -0,0 +1 @@
+*.pyc

app.py

@@ -1,154 +1,81 @@
+import logging
+import os
+import sys
+
 import gradio as gr
-import numpy as np
-import random
-
-# import spaces #[uncomment to use ZeroGPU]
-from diffusers import DiffusionPipeline
-import torch
-
-device = "cuda" if torch.cuda.is_available() else "cpu"
-model_repo_id = "stabilityai/sdxl-turbo"  # Replace to the model you would like to use
-
-if torch.cuda.is_available():
-    torch_dtype = torch.float16
-else:
-    torch_dtype = torch.float32
-
-pipe = DiffusionPipeline.from_pretrained(model_repo_id, torch_dtype=torch_dtype)
-pipe = pipe.to(device)
-
-MAX_SEED = np.iinfo(np.int32).max
-MAX_IMAGE_SIZE = 1024
-
-
-# @spaces.GPU #[uncomment to use ZeroGPU]
-def infer(
-    prompt,
-    negative_prompt,
-    seed,
-    randomize_seed,
-    width,
-    height,
-    guidance_scale,
-    num_inference_steps,
-    progress=gr.Progress(track_tqdm=True),
-):
-    if randomize_seed:
-        seed = random.randint(0, MAX_SEED)
-
-    generator = torch.Generator().manual_seed(seed)
-
-    image = pipe(
-        prompt=prompt,
-        negative_prompt=negative_prompt,
-        guidance_scale=guidance_scale,
-        num_inference_steps=num_inference_steps,
-        width=width,
-        height=height,
-        generator=generator,
-    ).images[0]
-
-    return image, seed
-
-
-examples = [
-    "Astronaut in a jungle, cold color palette, muted colors, detailed, 8k",
-    "An astronaut riding a green horse",
-    "A delicious ceviche cheesecake slice",
-]
-
-css = """
-#col-container {
-    margin: 0 auto;
-    max-width: 640px;
-}
-"""
-
-with gr.Blocks(css=css) as demo:
-    with gr.Column(elem_id="col-container"):
-        gr.Markdown(" # Text-to-Image Gradio Template")
 
-        with gr.Row():
-            prompt = gr.Text(
-                label="Prompt",
-                show_label=False,
-                max_lines=1,
-                placeholder="Enter your prompt",
-                container=False,
-            )
-
-            run_button = gr.Button("Run", scale=0, variant="primary")
-
-        result = gr.Image(label="Result", show_label=False)
-
-        with gr.Accordion("Advanced Settings", open=False):
-            negative_prompt = gr.Text(
-                label="Negative prompt",
-                max_lines=1,
-                placeholder="Enter a negative prompt",
-                visible=False,
-            )
-
-            seed = gr.Slider(
-                label="Seed",
-                minimum=0,
-                maximum=MAX_SEED,
-                step=1,
-                value=0,
-            )
-
-            randomize_seed = gr.Checkbox(label="Randomize seed", value=True)
-
-            with gr.Row():
-                width = gr.Slider(
-                    label="Width",
-                    minimum=256,
-                    maximum=MAX_IMAGE_SIZE,
-                    step=32,
-                    value=1024,  # Replace with defaults that work for your model
-                )
-
-                height = gr.Slider(
-                    label="Height",
-                    minimum=256,
-                    maximum=MAX_IMAGE_SIZE,
-                    step=32,
-                    value=1024,  # Replace with defaults that work for your model
-                )
-
-            with gr.Row():
-                guidance_scale = gr.Slider(
-                    label="Guidance scale",
-                    minimum=0.0,
-                    maximum=10.0,
-                    step=0.1,
-                    value=0.0,  # Replace with defaults that work for your model
-                )
-
-                num_inference_steps = gr.Slider(
-                    label="Number of inference steps",
-                    minimum=1,
-                    maximum=50,
-                    step=1,
-                    value=2,  # Replace with defaults that work for your model
-                )
-
-        gr.Examples(examples=examples, inputs=[prompt])
-    gr.on(
-        triggers=[run_button.click, prompt.submit],
-        fn=infer,
-        inputs=[
-            prompt,
-            negative_prompt,
-            seed,
-            randomize_seed,
-            width,
-            height,
-            guidance_scale,
-            num_inference_steps,
-        ],
-        outputs=[result, seed],
+from gradio.processing_utils import PUBLIC_HOSTNAME_WHITELIST
+
+PUBLIC_HOSTNAME_WHITELIST.append("aigame-skyreels.oss-accelerate.aliyuncs.com")
+PUBLIC_HOSTNAME_WHITELIST.append("aigame-skyreels.oss-cn-shanghai.aliyuncs.com")
+PUBLIC_HOSTNAME_WHITELIST.append("skyreels-infer-dev.oss-cn-shenzhen.aliyuncs.com")
+
+
+def _setup_logging() -> None:
+    """Configure logging to output to console with detailed format."""
+    log_format = "%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s"
+    logging.basicConfig(
+        level=logging.INFO,
+        format=log_format,
+        datefmt="%Y-%m-%d %H:%M:%S",
+        handlers=[logging.StreamHandler()],
     )
 
+
+# Initialize logging at module load
+_setup_logging()
+
+
+def _ensure_import_paths() -> None:
+    # Make sure the current directory is on sys.path so `pages` can be imported
+    current_dir = os.path.dirname(os.path.abspath(__file__))
+    if current_dir not in sys.path:
+        sys.path.append(current_dir)
+
+
+def create_app() -> gr.Blocks:
+    _ensure_import_paths()
+    # Delayed import to ensure sys.path is prepared
+    from pages import page_map
+
+    with gr.Blocks(title="SkyReels") as demo:
+        # State to store current logged in username
+        current_user = gr.State(value="default_user")
+
+        with gr.Row():
+            gr.Markdown("## SkyReels")
+        gr.Markdown("Use the tabs below to select a feature page.")
+        gr.Markdown("For API access, please refer to https://platform.skyreels.ai/")
+
+        with gr.Tabs():
+            video_extension_pages = ["video_shot_extend", "video_shot_switching"]
+            video_extension_pages_set = set(video_extension_pages)
+            categorized_pages = video_extension_pages_set
+            activate_pages = [
+                "video_shot_extend",
+                "video_shot_switching",
+                "audio2video_single",
+                "multimodel",
+            ]
+            other_pages = [p for p in activate_pages if p not in categorized_pages]
+            # Render other pages
+            for page_name in other_pages:
+                if page_name in page_map:
+                    page_map[page_name]().render(current_user)
+            with gr.Tab(label="Video Extension"):
+                with gr.Tabs():
+                    for page_name in video_extension_pages:
+                        if page_name in activate_pages and page_name in page_map:
+                            page_map[page_name]().render(current_user)
+
+    return demo
+
+
 if __name__ == "__main__":
-    demo.launch()
+    app = create_app()
+    # ssr_mode=False disables Server-Side Rendering which can cause connection issues in Spaces
+    app.queue(max_size=64).launch(
+        server_name="0.0.0.0", 
+        server_port=7860,
+        ssr_mode=False
+    )

pages/__init__.py

@@ -0,0 +1,11 @@
+from .audio2video_single import Audio2VideoSinglePage
+from .video_shot_extend import VideoShotExtendPage
+from .video_shot_switching import VideoShotSwitchingPage
+from .multimodel import MultiModelPage
+
+page_map = {
+    "audio2video_single": Audio2VideoSinglePage,
+    "video_shot_extend": VideoShotExtendPage,
+    "video_shot_switching": VideoShotSwitchingPage,
+    "multimodel": MultiModelPage,
+}

pages/audio2video_single.py

@@ -0,0 +1,105 @@
+import os
+from typing import List, Optional
+
+import gradio as gr
+
+from pages.base_page import BasePage
+from utils.ossutils import upload_to_oss
+
+
+class Audio2VideoSinglePage(BasePage):
+    @property
+    def page_name(self) -> str:
+        return "Talking Avatar(Single Actor)"
+
+    @property
+    def page_description(self) -> str:
+        return """
+            ## Single-Actor Avatar
+            1. Provide a video description prompt.
+            2. Upload a starting image.
+            3. Upload an audio file.
+            4. Click "Start Generation."
+            """
+
+    @property
+    def submit_endpoint(self) -> str:
+        return "/api/v1/video/audio2video/single/submit"
+
+    @property
+    def query_endpoint(self) -> str:
+        return "/api/v1/video/audio2video/single/task"
+
+    def render_input_components(self) -> List[gr.Component]:
+        with gr.Row():
+            prompt = gr.Textbox(
+                label="Prompt",
+                placeholder="Enter video description prompt",
+                lines=3,
+                value="a person is talking",
+            )
+        with gr.Row():
+            image_input = gr.Image(
+                label="Upload First Frame Image",
+                sources=["upload"],
+                type="filepath",
+            )
+        with gr.Row():
+            audio_input = gr.Audio(
+                label="Upload Audio",
+                sources=["upload"],
+                type="filepath",
+            )
+
+        return [
+            prompt,
+            image_input,
+            audio_input,
+        ]
+
+    def _build_payload(
+        self,
+        prompt: str,
+        image_path: Optional[str],
+        audio_path: Optional[str],
+    ):
+        if not prompt:
+            raise ValueError("Prompt is required.")
+        if not image_path:
+            raise ValueError("Please upload an image file.")
+        if not audio_path:
+            raise ValueError("Please upload an audio file.")
+
+        # Upload image to OSS
+        if isinstance(image_path, str) and image_path.startswith("http"):
+            image_url = image_path
+        else:
+            if not (isinstance(image_path, str) and os.path.isfile(image_path)):
+                raise ValueError("Invalid image file.")
+            image_url = upload_to_oss(image_path)
+
+        # Upload audio to OSS
+        if isinstance(audio_path, str) and audio_path.startswith("http"):
+            audio_url = audio_path
+        else:
+            if not (isinstance(audio_path, str) and os.path.isfile(audio_path)):
+                raise ValueError("Invalid audio file.")
+            audio_url = upload_to_oss(audio_path)
+
+        payload = {
+            "prompt": prompt,
+            "mode": "std",
+            "first_frame_image": image_url,
+            "audios": [audio_url],
+        }
+
+        return payload
+
+    def render_example(self, inputs: List[gr.Component]):
+        gr.Examples(
+            label="Examples (click to autofill)",
+            examples=[],
+            inputs=inputs,
+            examples_per_page=6,
+            cache_examples=False,
+        )

pages/base_page.py

@@ -0,0 +1,242 @@
+import os
+import json
+import logging
+import time
+from abc import ABC, abstractmethod
+from typing import Dict, List, Optional, Tuple
+
+import gradio as gr
+import requests
+from retry import retry
+
+logger = logging.getLogger(__name__)
+
+
+class AbstractPage(ABC):
+    @abstractmethod
+    def render(self, current_user=None):
+        pass
+
+
+class BasePage(AbstractPage):
+    base_url = "https://apis.skyreels.ai/"
+    token = ""
+    poll_interval = 5
+    loop_limit = 2000
+
+    @property
+    @abstractmethod
+    def page_name(self) -> str:
+        pass
+
+    @property
+    @abstractmethod
+    def page_description(self) -> str:
+        pass
+
+    @property
+    @abstractmethod
+    def submit_endpoint(self) -> str:
+        pass
+
+    @property
+    @abstractmethod
+    def query_endpoint(self) -> str:
+        pass
+
+    @abstractmethod
+    def render_input_components(self) -> List[gr.Component]:
+        pass
+
+    @abstractmethod
+    def _build_payload(self, *args, **kwargs):
+        pass
+
+    @abstractmethod
+    def render_example(self, inputs: List[gr.Component]):
+        pass
+
+    @retry(tries=3, delay=10, backoff=2)
+    def _submit_task(self, payload: dict) -> Tuple[str, dict]:
+        submit_url = f"{self.base_url}{self.submit_endpoint}"
+        headers = {
+            "accept": "application/json",
+            "Content-Type": "application/json",
+        }
+        if self.token:
+            headers["Authorization"] = self.token
+        
+        #payload["api_key"] = os.environ.get("API_KEY")
+        real_payload = payload.copy()
+        real_payload["api_key"] = os.environ.get("API_KEY")
+
+        response = requests.post(
+            submit_url, headers=headers, data=json.dumps(real_payload), timeout=30
+        )
+        if response.status_code != 200:
+            raise ValueError(
+                f"Failed to submit task: {response.status_code} {response.text}"
+            )
+        body = response.json()
+        if "task_id" not in body:
+            raise ValueError(f"Response missing task_id: {body}")
+        return body["task_id"], body
+
+    @retry(tries=100, delay=5)
+    def _query_task(self, task_id: str) -> Dict:
+        query_url = f"{self.base_url}{self.query_endpoint}/{task_id}"
+        headers = {
+            "accept": "application/json",
+        }
+        if self.token:
+            headers["Authorization"] = self.token
+
+        response = requests.get(query_url, headers=headers, timeout=30)
+        response.raise_for_status()
+        return response.json()
+
+    def _extract_video_url_from_data(self, data: Dict) -> Optional[str]:
+        if not isinstance(data, dict):
+            return None
+        value = data.get("video_url")
+        return value if isinstance(value, str) and value != "" else None
+
+    def _run_generation(self, *args, current_user=None, **kwargs):
+        try:
+            log_text = ""
+            payload_obj = None
+            yield (
+                "Submitting task…",
+                log_text,
+                None,  # payload_json (none yet)
+                None,  # video
+                gr.update(value="Generating", interactive=False),
+            )
+
+            payload = self._build_payload(*args, **kwargs)
+            log_text += f"[{time.strftime('%H:%M:%S')}] payload: {json.dumps(payload, ensure_ascii=False)}\n"
+            payload_obj = payload
+
+            yield ("Payload ready", log_text, payload_obj, None, gr.update())
+
+            task_id, submit_body = self._submit_task(payload)
+            log_text += f"[{time.strftime('%H:%M:%S')}] submit resp: {json.dumps(submit_body, ensure_ascii=False)}\n"
+
+            # Log user submission info
+            logger.info(
+                f"Task submitted - user: {current_user}, task_id: {task_id}, payload: {json.dumps(payload, ensure_ascii=False)}"
+            )
+
+            yield (
+                f"Task submitted. task_id: `{task_id}`",
+                log_text,
+                payload_obj,
+                None,
+                gr.update(),
+            )
+
+            for _ in range(self.loop_limit):
+                rsp = self._query_task(task_id)
+                status = rsp.get("status")
+                data = rsp.get("data")
+                msg = rsp.get("msg")
+                log_text += f"[{time.strftime('%H:%M:%S')}] poll resp: {json.dumps(rsp, ensure_ascii=False)}\n"
+                yield (f"Status: `{status}`", log_text, payload_obj, None, gr.update())
+
+                if status == "success":
+                    video_url = self._extract_video_url_from_data(
+                        data if isinstance(data, dict) else {}
+                    )
+                    if video_url:
+                        yield (
+                            "Completed ✅",
+                            log_text,
+                            payload_obj,
+                            video_url,
+                            gr.update(value="Start Generating", interactive=True),
+                        )
+                    else:
+                        yield (
+                            "Completed (no video URL found)",
+                            log_text,
+                            payload_obj,
+                            None,
+                            gr.update(value="Start Generating", interactive=True),
+                        )
+                    return
+
+                elif status == "failed":
+                    yield (
+                        f"Failed ❌: {msg}",
+                        log_text,
+                        payload_obj,
+                        None,
+                        gr.update(value="Start Generating", interactive=True),
+                    )
+                    return
+
+                time.sleep(self.poll_interval)
+
+            yield (
+                f"Failed ❌: Task Timeout: {task_id}",
+                log_text,
+                payload_obj,
+                None,
+                gr.update(value="Start Generating", interactive=True),
+            )
+            return
+
+        except Exception as e:
+            yield (
+                f"Error: {e}",
+                "",
+                payload_obj,
+                None,
+                gr.update(value="Start Generating", interactive=True),
+            )
+
+    def render(self, current_user=None):
+        with gr.TabItem(self.page_name):
+            # Center content to ~80% width using 1-8-1 column scaling
+            with gr.Row():
+                gr.Column(scale=1)
+                with gr.Column(scale=8):
+                    gr.Markdown(self.page_description)
+
+                    input_components = self.render_input_components()
+
+                    run_btn = gr.Button("Start Generating", variant="primary")
+
+                    with gr.Row():
+                        status_md = gr.Markdown()
+                    with gr.Row():
+                        payload_json = gr.JSON(label="Payload")
+                    with gr.Row():
+                        rsp_logs = gr.Textbox(label="Logs", lines=14, interactive=False)
+                    with gr.Row():
+                        video = gr.Video(label="Result Video")
+
+                    self.render_example(input_components)
+
+                    if current_user is not None:
+
+                        def run_with_user(*args):
+                            # Last arg is current_user from State
+                            *input_args, user = args
+                            yield from self._run_generation(
+                                *input_args, current_user=user
+                            )
+
+                        # Use the passed current_user State directly in inputs
+                        run_btn.click(
+                            fn=run_with_user,
+                            inputs=input_components + [current_user],
+                            outputs=[status_md, rsp_logs, payload_json, video, run_btn],
+                        )
+                    else:
+                        run_btn.click(
+                            fn=self._run_generation,
+                            inputs=input_components,
+                            outputs=[status_md, rsp_logs, payload_json, video, run_btn],
+                        )
+                gr.Column(scale=1)

pages/multimodel.py

@@ -0,0 +1,146 @@
+from typing import List, Optional
+
+import gradio as gr
+
+from pages.base_page import BasePage
+from utils.ossutils import upload_to_oss
+
+
+class MultiModelPage(BasePage):
+    @property
+    def page_name(self) -> str:
+        # return "多模型生成（视频延长/ 切镜/多主体）"
+        return "Reference to Video"
+
+    @property
+    def page_description(self) -> str:
+        return """
+        ### Reference to Video
+        - Supports uploading 1-4 reference images locally,
+        - After clicking "Start Generation", it will automatically poll task status and display video link when completed.
+        """
+
+    @property
+    def submit_endpoint(self) -> str:
+        return "/api/v1/video/multiobject/submit"
+
+    @property
+    def query_endpoint(self) -> str:
+        return "/api/v1/video/multiobject/task"
+
+    def render_input_components(self) -> List[gr.Component]:
+        with gr.Row():
+            prompt = gr.Textbox(
+                label="Prompt",
+                placeholder="Describe the scene…",
+                lines=4,
+            )
+
+        with gr.Row():
+            image_file_0 = gr.Image(
+                label="Reference image 0 (local)",
+                type="filepath",
+                sources=["upload", "clipboard"],
+            )
+            image_file_1 = gr.Image(
+                label="Reference image 1 (local)",
+                type="filepath",
+                sources=["upload", "clipboard"],
+            )
+            image_file_2 = gr.Image(
+                label="Reference image 2 (local)",
+                type="filepath",
+                sources=["upload", "clipboard"],
+            )
+            image_file_3 = gr.Image(
+                label="Reference image 3 (local)",
+                type="filepath",
+                sources=["upload", "clipboard"],
+            )
+
+            aspect_ratio = gr.Dropdown(
+                label="Aspect ratio",
+                choices=["16:9", "9:16"],
+                value="16:9",
+            )
+        return [
+            prompt,
+            image_file_0,
+            image_file_1,
+            image_file_2,
+            image_file_3,
+            aspect_ratio,
+        ]
+
+    def _build_payload(
+        self,
+        prompt: str,
+        image_file_0: Optional[str],
+        image_file_1: Optional[str],
+        image_file_2: Optional[str],
+        image_file_3: Optional[str],
+        aspect_ratio: str,
+    ):
+        if all(
+            image_file is None
+            for image_file in [image_file_0, image_file_1, image_file_2, image_file_3]
+        ):
+            raise ValueError("Please upload at least one reference image")
+        reference_list = []
+        if image_file_0 is not None:
+            if isinstance(image_file_0, str) and image_file_0.startswith("http"):
+                image_file_0_url = image_file_0
+            else:
+                image_file_0_url = upload_to_oss(image_file_0)
+            reference_list.append(image_file_0_url)
+
+        if image_file_1 is not None:
+            if isinstance(image_file_1, str) and image_file_1.startswith("http"):
+                image_file_1_url = image_file_1
+            else:
+                image_file_1_url = upload_to_oss(image_file_1)
+            reference_list.append(image_file_1_url)
+
+        if image_file_2 is not None:
+            if isinstance(image_file_2, str) and image_file_2.startswith("http"):
+                image_file_2_url = image_file_2
+            else:
+                image_file_2_url = upload_to_oss(image_file_2)
+            reference_list.append(image_file_2_url)
+
+        if image_file_3 is not None:
+            if isinstance(image_file_3, str) and image_file_3.startswith("http"):
+                image_file_3_url = image_file_3
+            else:
+                image_file_3_url = upload_to_oss(image_file_3)
+            reference_list.append(image_file_3_url)
+
+        assert (
+            len(reference_list) <= 4 and len(reference_list) >= 1
+        ), "Number of reference images must be between 1 and 4"
+
+        payload = {
+            # "task_type": "mutil_object",
+            "prompt": prompt or "",
+            "ref_images": reference_list,
+            "aspect_ratio": aspect_ratio,
+        }
+        return payload
+
+    def render_example(self, inputs: List[gr.Component]):
+        gr.Examples(
+            label="Examples (Click to auto-fill)",
+            examples=[
+                [
+                    "The man pours the milk from the cup into a plate for the puppy to drink.",
+                    "https://aigame-html.oss-cn-shanghai.aliyuncs.com/skyreels_evaluation_samples_control/multi_concept_benchmark/thing/bowl/1.png",
+                    "https://aigame-html.oss-cn-shanghai.aliyuncs.com/skyreels_evaluation_samples_control/multi_concept_benchmark/thing/animal/dog/4.jpg",
+                    "https://aigame-html.oss-cn-shanghai.aliyuncs.com/skyreels_evaluation_samples_control/multi_concept_benchmark/thing/cup/1.jpg",
+                    "https://aigame-html.oss-cn-shanghai.aliyuncs.com/skyreels_evaluation_samples_control/multi_concept_benchmark/human/man/synthesis/2.png",
+                    "16:9",
+                ],
+            ],
+            inputs=inputs,
+            examples_per_page=6,
+            cache_examples=False,
+        )

pages/video_shot_extend.py

@@ -0,0 +1,98 @@
+from typing import List
+
+import gradio as gr
+
+from pages.base_page import BasePage
+from utils.ossutils import upload_to_oss
+
+
+class VideoShotExtendPage(BasePage):
+    @property
+    def page_name(self) -> str:
+        return "Single-Shot Video Extension"
+
+    @property
+    def page_description(self) -> str:
+        return """
+        ## Single-Shot Video Extension
+        1. Upload a local video file.
+        2. Set the extension duration, choosing from 5 to 30 seconds.
+        3. Click "Start Generation." A loading status will appear below to indicate that the process has started successfully.
+        """
+
+    @property
+    def submit_endpoint(self) -> str:
+        return "/api/v1/video/extension/submit"
+
+    @property
+    def query_endpoint(self) -> str:
+        return "/api/v1/video/extension/task"
+
+    def render_input_components(self) -> List[gr.Component]:
+        with gr.Row():
+            prompt = gr.Textbox(
+                label="Prompt",
+                placeholder="Describe the scene…",
+                lines=4,
+            )
+
+        with gr.Row():
+            reference_video = gr.Video(
+                label="Reference video (local)",
+                sources=["upload"],
+            )
+
+        with gr.Row():
+            duration = gr.Dropdown(
+                label="duration",
+                choices=["5", "10", "15", "20", "25", "30"],
+                value="5",
+            )
+        return [prompt, reference_video, duration]
+
+    def _build_payload(
+        self,
+        prompt: str,
+        reference_video: str,
+        duration: str,
+    ):
+        if reference_video is None:
+            raise ValueError("Please upload a reference video.")
+        elif isinstance(reference_video, str) and reference_video.startswith("http"):
+            reference_video_url = reference_video
+        else:
+            reference_video_url = upload_to_oss(reference_video)
+
+        if prompt is None:
+            raise ValueError("Please enter the prompt.")
+
+        if duration is None:
+            raise ValueError("Please select the duration.")
+
+        payload = {
+            "prompt": prompt or "",
+            "prefix_video": reference_video_url,
+            "duration": int(duration),
+        }
+
+        return payload
+
+    def render_example(self, inputs: List[gr.Component]):
+        gr.Examples(
+            label="Examples (click to autofill)",
+            examples=[
+                [
+                    "A man is running along the beach, with the camera zooming out.",
+                    "https://aigame-html.oss-cn-shanghai.aliyuncs.com/skyreels_evaluation_samples_control/video_extend_benchmark_v1_5/0.mp4",
+                    "5",
+                ],
+                [
+                    "The camera gradually zooms in and focuses on the white building in the center of the video",
+                    "https://aigame-skyreels-pro-sv.oss-us-west-1.aliyuncs.com/skyreel_assets/20250623/2ac2f142-30dc-490d-97a4-5a68b3f3ff4f.mp4",
+                    "5",
+                ],
+            ],
+            inputs=inputs,
+            examples_per_page=6,
+            cache_examples=False,
+        )

pages/video_shot_switching.py

@@ -0,0 +1,148 @@
+from typing import List
+
+import gradio as gr
+
+from pages.base_page import BasePage
+from utils.ossutils import upload_to_oss
+
+
+class VideoShotSwitchingPage(BasePage):
+    @property
+    def page_name(self) -> str:
+        return "Shot Switching Video Extension"
+
+    @property
+    def page_description(self) -> str:
+        return """
+        ## Shot Switching Video Extension
+        Extend an existing clip while **switching shots** with professional transition patterns (e.g., Cut-In / Cut-Out / Reverse Shot / Multi-Angle / Cut Away).
+
+        1. Upload a reference video (or use an example URL).
+        2. Choose a cut type (or Auto).
+        3. Select duration (2–5 seconds, per API).
+        4. Click "Start Generating."
+
+        ### Cut Type Parameter Details
+        - **Auto**: The model automatically determines the appropriate cut type based on the scene context (default).
+        - **Cut-In**: Transitions from a wide shot to a close-up within the current scene.
+        - **Cut-Out**: Transitions from a close-up to a wide shot within the current scene.
+        - **Shot/Reverse Shot**: In dialogue scenes, transitions from a shot facing one person to a shot facing the other person.
+        - **Multi-Angle**: Switches to a different angle to show the current scene.
+        - **Cut Away**: Transitions to a new area within the current scene.
+        """
+
+    @property
+    def submit_endpoint(self) -> str:
+        return "/api/v1/video/extension/cutshot/submit"
+
+    @property
+    def query_endpoint(self) -> str:
+        return "/api/v1/video/extension/cutshot/task"
+
+    def render_input_components(self) -> List[gr.Component]:
+        with gr.Row():
+            prompt = gr.Textbox(
+                label="prompt",
+                placeholder="Describe the scene…",
+                lines=4,
+            )
+
+        with gr.Row():
+            prefix_video = gr.Video(
+                label="prefix_video (local)",
+                sources=["upload"],
+            )
+
+        with gr.Row():
+            duration = gr.Dropdown(
+                label="duration",
+                choices=["2", "3", "4", "5"],
+                value="5",
+            )
+
+            cut_type = gr.Dropdown(
+                label="cut_type",
+                choices=[
+                    "Auto",
+                    "Cut-In",
+                    "Cut-Out",
+                    "Shot/Reverse Shot",
+                    "Multi-Angle",
+                    "Cut Away",
+                ],
+                value="Auto",
+            )
+
+        return [prompt, prefix_video, duration, cut_type]
+
+    def _build_payload(
+        self,
+        prompt: str,
+        prefix_video: str,
+        duration: str,
+        cut_type: str,
+    ):
+        if not prompt:
+            raise ValueError("Please enter the prompt.")
+
+        if prefix_video is None:
+            raise ValueError("Please upload a prefix_video.")
+        elif isinstance(prefix_video, str) and prefix_video.startswith("http"):
+            prefix_video_url = prefix_video
+        else:
+            prefix_video_url = upload_to_oss(prefix_video)
+
+        if duration is None:
+            raise ValueError("Please select the duration.")
+        duration_int = int(duration)
+        if duration_int < 2 or duration_int > 5:
+            raise ValueError("duration must be between 2 and 5 seconds (per API).")
+
+        if not cut_type:
+            raise ValueError("Please select the cut_type.")
+
+        payload = {
+            "prompt": prompt,
+            "prefix_video": prefix_video_url,
+            "duration": duration_int,
+            "cut_type": cut_type,
+        }
+        return payload
+
+    def render_example(self, inputs: List[gr.Component]):
+        gr.Examples(
+            label="Examples (click to autofill)",
+            examples=[
+                [
+                    # self.API_KEY_PLACEHOLDER,
+                    "The camera focus on the man.",
+                    "https://aigame-html.oss-cn-shanghai.aliyuncs.com/skyreels_api/examples/video_extension/test.mp4",
+                    "5",
+                    "Auto",
+                ],
+                [
+                    #self.API_KEY_PLACEHOLDER,
+                    "Close-up on the girl's face as she aims, sweating",
+                    "https://skyreels-infer-dev.oss-cn-shenzhen.aliyuncs.com/user/yuzhe.jin/data/shot_extention/2.mp4",
+                    "5",
+                    "Auto",
+                ],
+                [
+                    # self.API_KEY_PLACEHOLDER,
+                    "The camera pulls back to reveal the girl lying in a meadow of flowers.",
+                    "https://skyreels-infer-dev.oss-cn-shenzhen.aliyuncs.com/user/yuzhe.jin/data/shot_extention/3.mp4",
+                    "5",
+                    "Auto",
+                ],
+                [
+                    # self.API_KEY_PLACEHOLDER,
+                    "Create a top side angle view of the robot playing the guitar",
+                    "https://skyreels-infer-dev.oss-cn-shenzhen.aliyuncs.com/user/yuzhe.jin/data/shot_extention/6.mp4",
+                    "5",
+                    "Auto",
+                ],
+            ],
+            inputs=inputs,
+            examples_per_page=6,
+            cache_examples=False,
+        )

requirements.txt

@@ -1,6 +1,3 @@
-accelerate
-diffusers
-invisible_watermark
-torch
-transformers
-xformers
+oss2
+retry
+gradio==6.4.0

utils/ossutils.py

@@ -0,0 +1,42 @@
+import os
+from datetime import datetime
+import uuid
+
+import oss2
+from retry import retry
+
+oss_ak = os.environ.get("ALIYUN_ACCESS_KEY_ID")
+oss_sk = os.environ.get("ALIYUN_ACCESS_KEY_SECRET")
+oss_endpoint = os.environ.get("ALIYUN_ENDPOINT")
+oss_bucket_name = os.environ.get("ALIYUN_BUCKET_NAME")
+
+# 配置 OSS 连接信息 - 请替换为你的实际信息
+
+auth = oss2.Auth(oss_ak, oss_sk)
+# Configure timeouts via config; use both connect and read timeout
+# service = oss2.Service(auth, oss_endpoint, connect_timeout=60)
+bucket = oss2.Bucket(auth, oss_endpoint, oss_bucket_name, connect_timeout=60)
+
+
+@retry(tries=2)
+def upload_to_oss(file_path):
+    if not os.path.isfile(file_path):
+        raise FileNotFoundError(f"文件不存在: {file_path}")
+    file_name = os.path.basename(file_path)
+    file_ext = file_name.split(".")[-1] if "." in file_name else ""
+    rand4 = uuid.uuid4().hex[:4]
+    unique_key = (
+        f"gradio_upload/{datetime.now().strftime('%Y%m%d%H%M%S')}_{rand4}.{file_ext}"
+    )
+    try:
+        with open(file_path, "rb") as f:
+            result = bucket.put_object(unique_key, f)
+        if result.status == 200:
+            oss_url = f"https://{bucket.bucket_name}.{bucket.endpoint.split('//')[1]}/{unique_key}"
+            return oss_url
+        else:
+            print(f"上传失败，错误码：{result.status}")
+            raise RuntimeError(f"上传失败，错误码：{result.status}")
+    except Exception as e:
+        print(f"上传出错：{e}")
+        raise RuntimeError(f"上传出错：{e}") from e