AI Image Generation API: Xay Pipeline Tao Anh Tu Dong

mkdir image-pipeline && cd image-pipeline
python -m venv venv
source venv/bin/activate
pip install openai requests pillow python-dotenv

OPENAI_API_KEY=sk-your-openai-key-here
STABILITY_API_KEY=sk-your-stability-key-here
OUTPUT_DIR=./generated_images

# pipeline.py
import os
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv
 
load_dotenv()
 
class ImagePipeline:
    def __init__(self):
        self.output_dir = Path(os.getenv("OUTPUT_DIR", "./generated_images"))
        self.output_dir.mkdir(parents=True, exist_ok=True)
 
    def save_image(self, image_bytes: bytes, prefix: str, prompt: str) -> Path:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        slug = prompt[:30].replace(" ", "_").lower()
        filename = f"{prefix}_{slug}_{timestamp}.png"
        filepath = self.output_dir / filename
        filepath.write_bytes(image_bytes)
        return filepath

# openai_generator.py
import base64
from openai import OpenAI
from pipeline import ImagePipeline
 
class DallEGenerator(ImagePipeline):
    def __init__(self):
        super().__init__()
        self.client = OpenAI()
 
    def generate(self, prompt: str, size: str = "1024x1024",
                 quality: str = "standard") -> dict:
        response = self.client.images.generate(
            model="dall-e-3",
            prompt=prompt,
            size=size,
            quality=quality,
            response_format="b64_json",
            n=1
        )
 
        image_data = base64.b64decode(response.data[0].b64_json)
        filepath = self.save_image(image_data, "dalle", prompt)
        revised_prompt = response.data[0].revised_prompt
 
        return {
            "filepath": str(filepath),
            "revised_prompt": revised_prompt,
            "model": "dall-e-3",
            "size": size
        }

# stability_generator.py
import requests
import os
from pipeline import ImagePipeline
 
class StabilityGenerator(ImagePipeline):
    def __init__(self):
        super().__init__()
        self.api_key = os.getenv("STABILITY_API_KEY")
        self.base_url = "https://api.stability.ai/v2beta"
 
    def generate(self, prompt: str, negative_prompt: str = "",
                 aspect_ratio: str = "1:1",
                 style_preset: str = None) -> dict:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Accept": "image/*"
        }
 
        payload = {
            "prompt": prompt,
            "negative_prompt": negative_prompt,
            "aspect_ratio": aspect_ratio,
            "output_format": "png"
        }
 
        if style_preset:
            payload["style_preset"] = style_preset
 
        response = requests.post(
            f"{self.base_url}/stable-image/generate/core",
            headers=headers,
            files={"none": ""},
            data=payload
        )
 
        if response.status_code != 200:
            raise Exception(f"Stability API error: {response.text}")
 
        filepath = self.save_image(response.content, "stability", prompt)
 
        return {
            "filepath": str(filepath),
            "model": "stable-diffusion",
            "aspect_ratio": aspect_ratio
        }

# orchestrator.py
import json
from datetime import datetime
from dalle_generator import DallEGenerator
from stability_generator import StabilityGenerator
 
class ContentOrchestrator:
    def __init__(self):
        self.dalle = DallEGenerator()
        self.stability = StabilityGenerator()
 
    def generate_batch(self, prompts: list[str],
                       providers: list[str] = None) -> list[dict]:
        if providers is None:
            providers = ["dalle", "stability"]
 
        results = []
 
        for prompt in prompts:
            result = {"prompt": prompt, "outputs": [], "timestamp": datetime.now().isoformat()}
 
            if "dalle" in providers:
                try:
                    dalle_result = self.dalle.generate(prompt)
                    result["outputs"].append(dalle_result)
                except Exception as e:
                    result["outputs"].append({"error": str(e), "model": "dall-e-3"})
 
            if "stability" in providers:
                try:
                    stability_result = self.stability.generate(prompt)
                    result["outputs"].append(stability_result)
                except Exception as e:
                    result["outputs"].append({"error": str(e), "model": "stable-diffusion"})
 
            results.append(result)
 
        # Luu manifest de theo doi
        manifest_path = self.dalle.output_dir / "manifest.json"
        with open(manifest_path, "w") as f:
            json.dump(results, f, indent=2)
 
        return results
 
# Su dung
if __name__ == "__main__":
    orchestrator = ContentOrchestrator()
    prompts = [
        "A minimalist blog header showing abstract data visualization",
        "Professional product photo of a wireless headphone on marble surface",
        "Isometric illustration of a modern home office setup"
    ]
    results = orchestrator.generate_batch(prompts)
    print(f"Da tao {len(results)} bo anh")

# Them vao orchestrator.py
COST_MAP = {
    "dall-e-3": {"standard": 0.040, "hd": 0.080},
    "stable-diffusion": {"core": 0.03}
}
 
def estimate_batch_cost(prompts: list, providers: list, quality: str = "standard") -> float:
    total = 0.0
    for _ in prompts:
        if "dalle" in providers:
            total += COST_MAP["dall-e-3"][quality]
        if "stability" in providers:
            total += COST_MAP["stable-diffusion"]["core"]
    return total