Flux 图片编辑

from openai import OpenAI

# 初始化客户端
client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="https://api.yelinai.com/v1"
)

def edit_local_image(image_path, prompt, aspect_ratio="1:1", mask_path=None, 
                    seed=None, safety_tolerance=1, output_format="png"):
    """编辑本地图片"""
    try:
        # 准备基础参数
        edit_params = {
            "model": "flux-kontext-max",
            "image": open(image_path, "rb"),
            "prompt": prompt,
            "extra_body": {
                "aspect_ratio": aspect_ratio,
                "safety_tolerance": safety_tolerance,
                "output_format": output_format
            }
        }
        
        # 可选参数
        if seed is not None:
            edit_params["extra_body"]["seed"] = seed
        
        # 如果有蒙版，添加到参数中
        if mask_path:
            edit_params["mask"] = open(mask_path, "rb")
        
        # 调用编辑 API
        response = client.images.edit(**edit_params)
        
        # 返回编辑后的图片 URL（注意：URL 有效期 10 分钟）
        return response.data[0].url
    
    except Exception as e:
        print(f"编辑失败: {e}")
        return None
    
    finally:
        # 确保文件句柄关闭
        if 'edit_params' in locals():
            if 'image' in edit_params:
                edit_params['image'].close()
            if 'mask' in edit_params:
                edit_params['mask'].close()

# 使用示例
edited_url = edit_local_image(
    image_path="portrait.jpg",
    prompt="Change the background to a beautiful sunset sky",
    aspect_ratio="2:3",
    seed=42,  # 可重现的结果
    safety_tolerance=1,  # 安全级别 (0-2)
    output_format="jpeg"  # 或 "png"
)

if edited_url:
    print(f"编辑完成: {edited_url}")
    print("注意：URL 将在 10 分钟后失效，请及时下载！")

import requests
import tempfile
import os
from urllib.parse import urlparse

def download_image_from_url(url):
    """从 URL 下载图片到临时文件"""
    try:
        response = requests.get(url, timeout=30)
        response.raise_for_status()
        
        # 获取文件扩展名
        parsed_url = urlparse(url)
        extension = parsed_url.path.split('.')[-1] if '.' in parsed_url.path else 'jpg'
        
        # 创建临时文件
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=f'.{extension}')
        temp_file.write(response.content)
        temp_file.close()
        
        return temp_file.name
    
    except Exception as e:
        print(f"下载图片失败: {e}")
        return None

def edit_online_image(image_url, prompt, aspect_ratio="1:1", mask_url=None):
    """编辑在线图片"""
    image_path = None
    mask_path = None
    
    try:
        # 下载原图
        image_path = download_image_from_url(image_url)
        if not image_path:
            return None
        
        # 下载蒙版（如果有）
        if mask_url:
            mask_path = download_image_from_url(mask_url)
        
        # 编辑图片
        result = edit_local_image(image_path, prompt, aspect_ratio, mask_path)
        
        return result
    
    finally:
        # 清理临时文件
        for path in [image_path, mask_path]:
            if path and os.path.exists(path):
                os.unlink(path)

# 使用示例
edited_url = edit_online_image(
    image_url="https://example.com/photo.jpg",
    prompt="Remove the person in the background, keep the foreground intact",
    aspect_ratio="16:9"
)

def batch_edit_images(edit_tasks):
    """批量编辑图片"""
    results = []
    
    for task in edit_tasks:
        try:
            print(f"正在编辑: {task['image_path']}")
            print(f"编辑指令: {task['prompt']}")
            
            edited_url = edit_local_image(
                image_path=task['image_path'],
                prompt=task['prompt'],
                aspect_ratio=task.get('aspect_ratio', '1:1'),
                mask_path=task.get('mask_path')
            )
            
            if edited_url:
                # 下载编辑后的图片
                filename = save_edited_image(edited_url, task['image_path'])
                
                results.append({
                    'original': task['image_path'],
                    'prompt': task['prompt'],
                    'edited_url': edited_url,
                    'saved_file': filename,
                    'success': True
                })
            else:
                results.append({
                    'original': task['image_path'],
                    'prompt': task['prompt'],
                    'success': False,
                    'error': 'API 调用失败'
                })
        
        except Exception as e:
            results.append({
                'original': task['image_path'],
                'prompt': task['prompt'],
                'success': False,
                'error': str(e)
            })
    
    return results

def save_edited_image(url, original_path):
    """保存编辑后的图片"""
    import time
    
    response = requests.get(url)
    if response.status_code == 200:
        # 生成新文件名
        base_name = os.path.splitext(os.path.basename(original_path))[0]
        timestamp = int(time.time())
        filename = f"{base_name}_edited_{timestamp}.png"
        
        with open(filename, 'wb') as f:
            f.write(response.content)
        
        print(f"已保存: {filename}")
        return filename
    
    return None

# 批量编辑示例
edit_tasks = [
    {
        'image_path': 'photo1.jpg',
        'prompt': 'Change the sky to a dramatic stormy sky',
        'aspect_ratio': '3:2'
    },
    {
        'image_path': 'photo2.jpg', 
        'prompt': 'Remove all people from the image',
        'aspect_ratio': '16:9',
        'mask_path': 'mask2.png'
    },
    {
        'image_path': 'photo3.jpg',
        'prompt': 'Make the image look like a vintage photograph',
        'aspect_ratio': '1:1'
    }
]

results = batch_edit_images(edit_tasks)

# 输出结果
for result in results:
    if result['success']:
        print(f"✅ {result['original']} -> {result['saved_file']}")
    else:
        print(f"❌ {result['original']} -> {result['error']}")

# 添加文字到图像
add_text = edit_local_image(
    image_path="poster.jpg",
    prompt='Add the text "SALE 50% OFF" in bold red letters at the top center',
    aspect_ratio="3:2"
)

# 修改现有文字
modify_text = edit_local_image(
    image_path="sign.jpg", 
    prompt='Change the text from "Open" to "Closed"',
    aspect_ratio="1:1"
)

# 使用引号确保文字准确性
precise_text = edit_local_image(
    image_path="banner.jpg",
    prompt='Replace the existing text with exactly "Welcome to Our Store"',
    aspect_ratio="7:3"
)

# 保持光照一致性
lighting_edit = edit_local_image(
    image_path="interior.jpg",
    prompt="Add a lamp in the corner, matching the existing warm lighting",
    aspect_ratio="16:9"
)

# 保持透视和比例
perspective_edit = edit_local_image(
    image_path="street.jpg", 
    prompt="Add a bicycle parked next to the building, matching the street perspective",
    aspect_ratio="3:2"
)

# 维持角色特征
character_edit = edit_local_image(
    image_path="person.jpg",
    prompt="Change the person's shirt to blue while keeping all facial features identical",
    aspect_ratio="2:3",
    seed=42  # 使用固定种子确保一致性
)

# 更换背景
background_edit = edit_local_image(
    image_path="portrait.jpg",
    prompt="Replace the background with a modern office environment",
    aspect_ratio="2:3"
)

# 去除背景
remove_bg = edit_local_image(
    image_path="product.jpg", 
    prompt="Remove the background, make it transparent",
    aspect_ratio="1:1"
)

# 添加元素
add_element = edit_local_image(
    image_path="room.jpg",
    prompt="Add a beautiful plant in the corner of the room",
    aspect_ratio="16:9"
)

# 删除物体
remove_object = edit_local_image(
    image_path="street.jpg",
    prompt="Remove the car from the street, replace with empty road",
    aspect_ratio="3:2"
)

# 替换物体
replace_object = edit_local_image(
    image_path="table.jpg",
    prompt="Replace the apple on the table with an orange", 
    aspect_ratio="1:1"
)

# 艺术风格
artistic_style = edit_local_image(
    image_path="landscape.jpg",
    prompt="Transform into an impressionist painting style",
    aspect_ratio="3:2"
)

# 季节变换
season_change = edit_local_image(
    image_path="park.jpg",
    prompt="Change from summer to autumn, add colorful fall leaves",
    aspect_ratio="16:9"
)

# 使用蒙版精确编辑特定区域
masked_edit = edit_local_image(
    image_path="group_photo.jpg",
    prompt="Blur the faces of the people in the masked area",
    aspect_ratio="3:2",
    mask_path="face_mask.png"  # 白色区域会被编辑
)

from PIL import Image, ImageDraw

def create_circular_mask(image_path, center, radius):
    """创建圆形蒙版"""
    # 打开原图获取尺寸
    with Image.open(image_path) as img:
        width, height = img.size
    
    # 创建黑色背景蒙版
    mask = Image.new('L', (width, height), 0)
    draw = ImageDraw.Draw(mask)
    
    # 画白色圆形（编辑区域）
    left = center[0] - radius
    top = center[1] - radius
    right = center[0] + radius
    bottom = center[1] + radius
    
    draw.ellipse([left, top, right, bottom], fill=255)
    
    # 保存蒙版
    mask_path = f"circular_mask_{int(time.time())}.png"
    mask.save(mask_path)
    return mask_path

def create_rectangular_mask(image_path, bbox):
    """创建矩形蒙版"""
    with Image.open(image_path) as img:
        width, height = img.size
    
    mask = Image.new('L', (width, height), 0)
    draw = ImageDraw.Draw(mask)
    
    # 画白色矩形（编辑区域）
    draw.rectangle(bbox, fill=255)
    
    mask_path = f"rect_mask_{int(time.time())}.png"
    mask.save(mask_path)
    return mask_path

# 使用示例
mask_path = create_circular_mask("photo.jpg", center=(500, 300), radius=100)
edited = edit_local_image("photo.jpg", "Add sunglasses", mask_path=mask_path)

# ❌ 模糊描述
prompt = "make it better"

# ✅ 具体且详细的指令
prompt = """
Remove the person standing in the background on the left side,
fill the area with matching landscape elements,
maintain the original lighting and perspective
"""

# ✅ 文字编辑使用引号
text_prompt = 'Change the store sign text to "OPEN 24/7" in red letters'

# ✅ 保持一致性的指令
consistency_prompt = """
Add a red sports car parked in the driveway,
matching the existing lighting conditions and perspective,
ensure it fits naturally with the house architecture
"""

# ✅ 精确的局部编辑
precise_prompt = """
Change only the woman's dress color from blue to emerald green,
keep all other elements including lighting, background, and pose unchanged
"""

def create_professional_prompt(action, target, context_preservation):
    """创建专业级编辑提示词"""
    return f"""
    {action} {target}.
    Maintain the original {context_preservation}.
    Ensure the edit looks natural and seamlessly integrated.
    """

# 示例使用
prompt = create_professional_prompt(
    action="Replace",
    target="the cloudy sky with a clear sunset sky",
    context_preservation="lighting, shadows, and overall atmosphere"
)

def recommend_aspect_ratio(edit_type, original_ratio=None):
    """根据编辑类型推荐宽高比"""
    recommendations = {
        "portrait_edit": "2:3",
        "landscape_edit": "3:2", 
        "social_media": "1:1",
        "banner_creation": "7:3",
        "mobile_content": "9:16",
        "desktop_wallpaper": "16:9"
    }
    
    # 如果有原图比例，优先保持
    if original_ratio:
        return original_ratio
    
    return recommendations.get(edit_type, "1:1")

import time

def edit_with_retry(image_path, prompt, max_retries=3, **kwargs):
    """带重试机制的图片编辑"""
    for attempt in range(max_retries):
        try:
            result = edit_local_image(image_path, prompt, **kwargs)
            if result:
                return result
        except Exception as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 指数退避
                print(f"编辑失败，{wait_time}秒后重试... (尝试 {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                print(f"编辑最终失败: {e}")
                raise e
    
    return None

import time
import requests
from concurrent.futures import ThreadPoolExecutor

def download_with_urgency(url, filename, max_attempts=3):
    """紧急下载，考虑 10 分钟失效期"""
    start_time = time.time()
    
    for attempt in range(max_attempts):
        try:
            response = requests.get(url, timeout=30)
            response.raise_for_status()
            
            with open(filename, 'wb') as f:
                f.write(response.content)
            
            elapsed = time.time() - start_time
            print(f"✅ 下载成功: {filename} (用时 {elapsed:.1f}s)")
            return True
            
        except Exception as e:
            elapsed = time.time() - start_time
            if elapsed > 540:  # 9 分钟，留 1 分钟缓冲
                print(f"⚠️ 警告：接近 URL 失效时间！")
            
            if attempt < max_attempts - 1:
                print(f"下载失败，重试... (尝试 {attempt + 1}/{max_attempts})")
                time.sleep(1)
            else:
                print(f"❌ 下载最终失败: {e}")
                return False
    
    return False

def batch_edit_with_immediate_download(tasks):
    """批量编辑并立即下载"""
    def edit_and_download(task):
        # 编辑
        edited_url = edit_local_image(**task['edit_params'])
        if not edited_url:
            return {'success': False, 'error': 'Edit failed'}
        
        # 立即下载
        filename = f"{task['name']}_edited_{int(time.time())}.png"
        success = download_with_urgency(edited_url, filename)
        
        return {
            'success': success,
            'filename': filename if success else None,
            'url': edited_url,
            'task_name': task['name']
        }
    
    # 并行处理以节省时间
    with ThreadPoolExecutor(max_workers=3) as executor:
        results = list(executor.map(edit_and_download, tasks))
    
    return results

def validate_edit_result(original_path, edited_url, prompt):
    """验证编辑结果质量"""
    try:
        # 下载编辑后的图片进行分析
        response = requests.get(edited_url)
        edited_image = Image.open(BytesIO(response.content))
        original_image = Image.open(original_path)
        
        # 基础检查
        checks = {
            "size_reasonable": edited_image.size[0] * edited_image.size[1] > 50000,
            "format_valid": edited_image.format in ['PNG', 'JPEG', 'WEBP'],
            "not_corrupted": edited_image.mode in ['RGB', 'RGBA', 'L']
        }
        
        return all(checks.values()), checks
    
    except Exception as e:
        return False, {"error": str(e)}

class FluxEditTracker:
    """Flux 编辑使用统计"""
    
    def __init__(self):
        self.edit_history = []
        self.total_cost = 0
    
    def track_edit(self, image_path, prompt, aspect_ratio, success, cost=0.07):
        """记录编辑操作"""
        record = {
            "timestamp": time.time(),
            "image_path": image_path,
            "prompt": prompt,
            "aspect_ratio": aspect_ratio,
            "success": success,
            "cost": cost if success else 0
        }
        
        self.edit_history.append(record)
        if success:
            self.total_cost += cost
    
    def get_stats(self):
        """获取使用统计"""
        if not self.edit_history:
            return {}
        
        total_edits = len(self.edit_history)
        successful_edits = sum(1 for r in self.edit_history if r["success"])
        success_rate = successful_edits / total_edits * 100
        
        # 最常用的宽高比
        ratios = [r["aspect_ratio"] for r in self.edit_history]
        most_common_ratio = max(set(ratios), key=ratios.count)
        
        return {
            "total_edits": total_edits,
            "successful_edits": successful_edits,
            "success_rate": f"{success_rate:.1f}%",
            "total_cost": f"\`$\{self.total_cost:.2f}",
            "avg_cost_per_edit": f"\`$\{self.total_cost/successful_edits:.3f}" if successful_edits > 0 else "$0",
            "most_common_ratio": most_common_ratio
        }

# 使用示例
tracker = FluxEditTracker()

# 编辑时记录
result = edit_local_image("photo.jpg", "Add rainbow in sky", "16:9")
tracker.track_edit("photo.jpg", "Add rainbow in sky", "16:9", bool(result))

# 查看统计
print(tracker.get_stats())

# 下载多图处理脚本
curl -O https://raw.githubusercontent.com/laozhang-api/ai-api-code-samples/main/flux-Image-API/flux-simple-batch.sh
chmod +x flux-simple-batch.sh

# 查看使用说明
./flux-simple-batch.sh --help

# 编辑配置并运行
# 1. 设置 API_KEY
# 2. 配置 IMAGE_PAIRS 数组
# 3. 自定义 BATCH_PROMPT
./flux-simple-batch.sh

BATCH_PROMPT="Transfer the pattern/design from the left image to the clothes of the model in the right image, with the size being 2/3 of the width of the chest, located in the middle. Make it look natural and well-integrated."

BATCH_PROMPT="Combine the artistic style from the left image with the subject matter from the right image, creating a harmonious artistic fusion."

BATCH_PROMPT="Apply the decorative elements from the left image to the building in the right image, maintaining architectural harmony."