feat: kwai kolors (#262)

* feat: kwai kolors
* docs: update docs
* chore: format

Author: timerring

README.md

@@ -18,7 +18,8 @@
   <img src="assets/qwen-color.svg" alt="Qwen-2.5-72B-Instruct" width="60" height="60" />
   <img src="assets/minimax-color.svg" alt="Minimax" width="20" height="60" />
   <img src="assets/minimax-text.svg" alt="Minimax" width="60" height="60" />
-
+  <img src="assets/siliconcloud-color.svg" alt="SiliconFlow" width="15" height="60" />
+  <img src="assets/siliconcloud-text.svg" alt="SiliconFlow" width="100" height="60" />
 </div>
 
 ##  1. Introduction
@@ -45,6 +46,7 @@
 - **( :tada: NEW)自动多平台循环直播推流**：该工具已经开源 [looplive](https://github.com/timerring/looplive) 是一个 7 x 24 小时全自动**循环多平台同时推流**直播工具。
 - **( :tada: NEW)自动生成风格变换的视频封面**：采用图生图多模态模型，自动获取视频截图并上传风格变换后的视频封面。
   - `Minimax image-01`
+  - `Kwai Kolors`
 
 项目架构流程如下：
 
@@ -195,6 +197,12 @@ MLLM 模型主要用于自动切片后的切片标题生成，此功能默认关
 
 在项目的自动切片功能需要使用到 Minimax 模型，请自行[注册账号](https://www.minimax.chat/)并申请 API Key，填写到 `bilive.toml` 文件中对应的 `MINIMAX_API_KEY` 中。
 
+##### 3.2.5 Kwai Kolors 模型
+
+> 如需使用 Kwai Kolors 模型，请将 `bilive.toml` 文件中 `generate_cover` 参数设置为 `true`，并将 `IMAGE_GEN_MODEL` 参数设置为 `siliconflow`，采用 siliconflow 部署的 Kolors 模型。
+
+请自行[注册账号](https://cloud.siliconflow.cn/i/3Szr5BVg)并申请 API Key，填写到 `bilive.toml` 文件中对应的 `SILICONFLOW_API_KEY` 中。
+
 #### 4. bilitool 登录
 
 > 由于一般日志打印不出二维码效果（docker 的日志不确定是否能打印，等发布新image时再修改，docker 版本请先参考文档 [bilive](https://bilive.timerring.com)，本 README 只针对源码部署），所以这步需要提前在机器上安装 [bilitool](https://github.com/timerring/bilitool):

assets/siliconcloud-color.svg

@@ -37,5 +37,6 @@ qwen_api_key = "" # Apply for your own Qwen API key at https://bailian.console.a
 
 [cover]
 generate_cover = false # whether to generate cover
-image_gen_model = "minimax" # the image generation model, can be "minimax"
+image_gen_model = "minimax" # the image generation model, can be "minimax" or "siliconflow"
 minimax_api_key = "" # Apply for your own Minimax API key at https://platform.minimaxi.com/user-center/basic-information/interface-key
+siliconflow_api_key = "" # Apply for your own SiliconFlow API key at https://cloud.siliconflow.cn/i/3Szr5BVg

assets/siliconcloud-text.svg

@@ -75,3 +75,4 @@ def get_interface_config():
 GENERATE_COVER = config.get('cover', {}).get('generate_cover')
 IMAGE_GEN_MODEL = config.get('cover', {}).get('image_gen_model')
 MINIMAX_API_KEY = config.get('cover', {}).get('minimax_api_key')
+SILICONFLOW_API_KEY = config.get('cover', {}).get('siliconflow_api_key')

bilive.toml

@@ -3,6 +3,7 @@
 from src.config import IMAGE_GEN_MODEL
 import subprocess
 
+
 def cut_cover_use_ffmpeg(video_path):
     """Cut cover use ffmpeg
     Args:
@@ -13,10 +14,20 @@ def cut_cover_use_ffmpeg(video_path):
     upload_log.info("begin to generate cover")
     cover_path = video_path[:-4] + ".jpg"
     ffmpeg_command = [
-        'ffmpeg', '-y', '-i', video_path, '-t', '1', '-r', '1', cover_path
+        "ffmpeg",
+        "-y",
+        "-i",
+        video_path,
+        "-t",
+        "1",
+        "-r",
+        "1",
+        cover_path,
     ]
     try:
-        result = subprocess.run(ffmpeg_command, check=True, capture_output=True, text=True)
+        result = subprocess.run(
+            ffmpeg_command, check=True, capture_output=True, text=True
+        )
         upload_log.debug(f"FFmpeg output: {result.stdout}")
         if result.stderr:
             upload_log.debug(f"FFmpeg debug: {result.stderr}")
@@ -33,6 +44,7 @@ def cover_generator(model_type):
     Returns:
         function: wrapped title generation function
     """
+
     def decorator(func):
         def wrapper(video_path):
             cover_path = cut_cover_use_ffmpeg(video_path)
@@ -41,13 +53,21 @@ def wrapper(video_path):
                 return None
             if model_type == "minimax":
                 from .image_model_sdk.minimax_sdk import minimax_generate_cover
+
                 return minimax_generate_cover(cover_path)
+            elif model_type == "siliconflow":
+                from .image_model_sdk.kolors_sdk import kolors_generate_cover
+
+                return kolors_generate_cover(cover_path)
             else:
                 upload_log.error(f"Unsupported model type: {model_type}")
                 return None
+
         return wrapper
+
     return decorator
 
+
 @cover_generator(IMAGE_GEN_MODEL)
 def generate_cover(video_path):
     """Generate cover for video
@@ -56,4 +76,4 @@ def generate_cover(video_path):
     Returns:
         str: generated cover
     """
-    pass  # The actual implementation is handled by the decorator
+    pass  # The actual implementation is handled by the decorator

src/config.py

@@ -0,0 +1,49 @@
+import requests
+import base64
+import time
+import os
+from src.config import SILICONFLOW_API_KEY
+
+
+def kolors_generate_cover(your_file_path):
+    """Generater cover image using SiliconFlow api of Kolors(Kwai)
+    Args:
+        your_file_path: str, path to the image file
+    Returns:
+        str, local download path of the generated cover image file
+    """
+    with open(your_file_path, "rb") as image_file:
+        data = base64.b64encode(image_file.read()).decode("utf-8")
+
+    payload = {
+        "model": "Kwai-Kolors/Kolors",
+        "prompt": "这是一个视频截图，请尝试生成对应的日本动漫类型的封面",
+        "image_size": "1024x1024",
+        "batch_size": 1,
+        "num_inference_steps": 20,
+        "guidance_scale": 7.5,
+        "image": f"data:image/webp;base64,{data}",
+    }
+    headers = {
+        "Authorization": f"Bearer {SILICONFLOW_API_KEY}",
+        "Content-Type": "application/json",
+    }
+    url = "https://api.siliconflow.cn/v1/images/generations"
+    response = requests.request("POST", url, json=payload, headers=headers)
+    if response.status_code == 200:
+        image_url = response.json()["images"][0]["url"]
+        img_data = requests.get(image_url).content
+        cover_name = time.strftime("%Y%m%d%H%M%S") + ".png"
+        temp_cover_path = os.path.join(os.path.dirname(your_file_path), cover_name)
+        with open(temp_cover_path, "wb") as handler:
+            handler.write(img_data)
+        os.remove(your_file_path)
+        return temp_cover_path
+    else:
+        print(response.text, flush=True)
+        return None
+
+
+if __name__ == "__main__":
+    your_file_path = ""
+    print(kolors_generate_cover(your_file_path))

src/cover/cover_generator.py

@@ -13,41 +13,41 @@ def minimax_generate_cover(your_file_path):
     Returns:
         str, local download path of the generated cover image file
     """
-    cover_name = time.strftime("%Y%m%d%H%M%S") + ".png"
-    temp_cover_path = os.path.join(os.path.dirname(your_file_path), cover_name)
 
     with open(your_file_path, "rb") as image_file:
-        data = base64.b64encode(image_file.read()).decode('utf-8')
+        data = base64.b64encode(image_file.read()).decode("utf-8")
 
-    payload = json.dumps({
-        "model": "image-01",
-        "prompt": "这是一个视频截图，请生成其对应的吉普力风格的图片",
-        "subject_reference": [
-            {
-                "type": "character",
-                "image_file": f"data:image/jpeg;base64,{data}"
-            }
-        ],
-        "n": 2
-    })
+    payload = json.dumps(
+        {
+            "model": "image-01",
+            "prompt": "这是一个视频截图，请生成其对应的吉普力风格的图片",
+            "subject_reference": [
+                {"type": "character", "image_file": f"data:image/jpeg;base64,{data}"}
+            ],
+            "n": 2,
+        }
+    )
     headers = {
-        'Authorization': f'Bearer {MINIMAX_API_KEY}',
-        'Content-Type': 'application/json'
+        "Authorization": f"Bearer {MINIMAX_API_KEY}",
+        "Content-Type": "application/json",
     }
 
     url = "https://api.minimax.chat/v1/image_generation"
     response = requests.request("POST", url, headers=headers, data=payload).json()
-    if response['base_resp']['status_code'] == 0:
-        image_url = response['data']['image_urls'][0]
+    if response["base_resp"]["status_code"] == 0:
+        image_url = response["data"]["image_urls"][0]
         img_data = requests.get(image_url).content
-        with open(temp_cover_path, 'wb') as handler:
+        cover_name = time.strftime("%Y%m%d%H%M%S") + ".png"
+        temp_cover_path = os.path.join(os.path.dirname(your_file_path), cover_name)
+        with open(temp_cover_path, "wb") as handler:
             handler.write(img_data)
         os.remove(your_file_path)
         return temp_cover_path
     else:
-        print(response['base_resp']['error_msg'])
+        print(response["base_resp"]["error_msg"], flush=True)
         return None
 
+
 if __name__ == "__main__":
     your_file_path = ""
-    print(minimax_generate_cover(your_file_path))
+    print(minimax_generate_cover(your_file_path))