Create app.py

app.py

@@ -0,0 +1,167 @@
+import gradio as gr
+import requests
+from PIL import Image
+import io
+import os
+import subprocess
+import sys
+import atexit
+import time
+
+API_KEY = os.environ.get("SILICONFLOW_API_KEY")
+if not API_KEY:
+    raise ValueError("请设置SILICONFLOW_API_KEY环境变量")
+
+API_URL = "https://api.siliconflow.cn/v1/image/generations"
+TEXT_API_URL = "https://api.siliconflow.cn/v1/chat/completions"
+HEADERS = {
+    "accept": "application/json",
+    "content-type": "application/json",
+    "Authorization": f"Bearer {API_KEY}"
+}
+
+# Cloudflared 安装和设置
+def install_cloudflared():
+    if sys.platform.startswith('linux'):
+        subprocess.run(["curl", "-L", "--output", "cloudflared", "https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-linux-amd64"])
+        subprocess.run(["chmod", "+x", "cloudflared"])
+    elif sys.platform == 'darwin':
+        subprocess.run(["brew", "install", "cloudflare/cloudflare/cloudflared"])
+    elif sys.platform == 'win32':
+        subprocess.run(["powershell", "-Command", "Invoke-WebRequest -Uri https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-windows-amd64.exe -OutFile cloudflared.exe"])
+    else:
+        raise OSError("Unsupported operating system")
+
+cloudflared_process = None
+
+def start_cloudflared(port):
+    global cloudflared_process
+    if sys.platform.startswith('linux') or sys.platform == 'darwin':
+        cloudflared_process = subprocess.Popen(["./cloudflared", "tunnel", "--url", f"http://localhost:{port}"])
+    elif sys.platform == 'win32':
+        cloudflared_process = subprocess.Popen(["cloudflared.exe", "tunnel", "--url", f"http://localhost:{port}"])
+    atexit.register(stop_cloudflared)
+
+def stop_cloudflared():
+    global cloudflared_process
+    if cloudflared_process:
+        cloudflared_process.terminate()
+        cloudflared_process.wait()
+
+# 其余函数保持不变
+def generate_image(model, prompt, image_size, batch_size=1, num_inference_steps=20):
+    data = {
+        "model": model,
+        "prompt": prompt,
+        "image_size": image_size,
+        "batch_size": batch_size,
+        "num_inference_steps": num_inference_steps
+    }
+
+    print(f"Sending request to API with data: {data}")
+    response = requests.post(API_URL, headers=HEADERS, json=data)
+    print(f"API response status code: {response.status_code}")
+
+    if response.status_code == 200:
+        response_json = response.json()
+        print(f"API response: {response_json}")
+        if "images" in response_json:
+            images = []
+            for img_data in response_json["images"]:
+                image_url = img_data["url"]
+                try:
+                    image_response = requests.get(image_url)
+                    image = Image.open(io.BytesIO(image_response.content))
+                    images.append(image)
+                except Exception as e:
+                    print(f"Error fetching image: {e}")
+            return images if images else "Error: No images could be fetched"
+        else:
+            return "No image data in response"
+    else:
+        return f"Error: {response.status_code}, {response.text}"
+
+def use_gemma_model(prompt, task):
+    data = {
+        "model": "google/gemma-2-9b-it",
+        "messages": [
+            {"role": "system", "content": f"You are an AI assistant that helps with {task}. Respond concisely."},
+            {"role": "user", "content": prompt}
+        ],
+        "temperature": 0.7,
+        "max_tokens": 150
+    }
+    response = requests.post(TEXT_API_URL, headers=HEADERS, json=data)
+    if response.status_code == 200:
+        return response.json()['choices'][0]['message']['content'].strip()
+    else:
+        return f"Error: {response.status_code}, {response.text}"
+
+def enhance_prompt(prompt):
+    return use_gemma_model(f"Enhance this image prompt for better results: {prompt}", "prompt enhancement")
+
+def translate_prompt(prompt):
+    return use_gemma_model(f"Translate this text to English: {prompt}", "translation")
+
+def generate_with_options(model, prompt, image_width, image_height, batch_size, num_inference_steps, enhance, translate):
+    try:
+        if enhance:
+            prompt = enhance_prompt(prompt)
+        if translate:
+            prompt = translate_prompt(prompt)
+        image_size = f"{image_width}x{image_height}"
+        result = generate_image(model, prompt, image_size, batch_size, num_inference_steps)
+        if isinstance(result, str):  # 如果返回的是错误消息
+            return None, result  # 返回 None 作为图像，错误消息作为文本
+        if isinstance(result, list):  # 如果返回的是图像列表
+            return result, None
+        return result, None  # 返回图像和 None 作为错误消息
+    except Exception as e:
+        return None, f"An error occurred: {str(e)}"
+
+def create_model_interface(model_name, default_steps=20):
+    with gr.Tab(model_name):
+        with gr.Row():
+            with gr.Column():
+                prompt = gr.Textbox(label="Prompt")
+                enhance = gr.Checkbox(label="Enhance Prompt")
+                translate = gr.Checkbox(label="Translate Prompt")
+                image_width = gr.Number(label="Image Width", value=1024, step=64)
+                image_height = gr.Number(label="Image Height", value=1024, step=64)
+                batch_size = gr.Slider(minimum=1, maximum=4, step=1, label="Batch Size", value=1)
+                num_inference_steps = gr.Slider(minimum=1, maximum=50, step=1, label="Inference Steps", value=default_steps)
+                generate_button = gr.Button("Generate")
+            with gr.Column():
+                output = gr.Gallery(label="Generated Images")
+                error_output = gr.Textbox(label="Error Message", visible=False)
+        
+        generate_button.click(
+            fn=generate_with_options,
+            inputs=[
+                gr.Textbox(value=model_name, visible=False),
+                prompt,
+                image_width,
+                image_height,
+                batch_size,
+                num_inference_steps,
+                enhance,
+                translate
+            ],
+            outputs=[output, error_output]
+        )
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Image Generation with FLUX Models")
+    
+    create_model_interface("black-forest-labs/FLUX.1-dev", default_steps=20)
+    create_model_interface("black-forest-labs/FLUX.1-schnell")
+    create_model_interface("Pro/black-forest-labs/FLUX.1-schnell")
+
+if __name__ == "__main__":
+    port = 7860  # 或者您想使用的其他端口
+    print("Installing Cloudflared...")
+    install_cloudflared()
+    print("Starting Cloudflared...")
+    start_cloudflared(port)
+    print("Cloudflared started. Launching Gradio app...")
+    demo.launch(server_name="0.0.0.0", server_port=port)