app.py

import requests
from PIL import Image
import oss2
import os
import gradio as gr
import uuid
import json
import base64
from io import BytesIO
from datetime import datetime
from oss2.credentials import EnvironmentVariableCredentialsProvider,StaticCredentialsProvider

# 下载网络图片
def download_image(url):
    try:
        response = requests.get(url, timeout=10)  # 添加超时处理，避免长时间等待
        response.raise_for_status()  # 如果状态码不是200，抛出HTTPError
        return Image.open(BytesIO(response.content))
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"图片下载失败: {e}")



def formatNow():
    now = datetime.now()
    return now.strftime("%Y-%m-%d %H:%M:%S")



def remove_metadata(image, output_path=None):
    # 获取图片的格式
    img_format = image.format.lower()

    # 如果没有指定输出路径，自动生成路径
    if output_path is None:
        output_path = f"downloaded_image_no_metadata.{img_format}"

    # 根据图片的格式保存图片，不包含元数据
    image.save(output_path, img_format)

    return output_path

# 上传到阿里云 OSS
def upload_to_oss(file_path, object_name, bucket_name, access_key_id, access_key_secret, security_token, endpoint):
    try:
        auth = oss2.ProviderAuthV4(StaticCredentialsProvider(access_key_id,access_key_secret,security_token))
        # auth = oss2.ProviderAuthV4(access_key_id, access_key_secret, security_token)
        # 填写Bucket名称。
        bucket = oss2.Bucket(auth, f'https://{endpoint}', bucket_name, region=endpoint.split('.')[0].split('-', 1)[1])
        result = bucket.put_object_from_file(object_name, file_path)
        # 返回上传的文件 URL
        return f"https://{bucket_name}.{endpoint}/{object_name}"
    except oss2.exceptions.OssError as e:
        print(f"OSS上传失败: {e}")
        return None

def compress_image(image, output_path, quality=85):
    img_format = image.format.lower()

    if img_format in ['jpeg', 'webp']:
        image.save(output_path, img_format.upper(), quality=quality)
    elif img_format == 'png':
        image.save(output_path, 'PNG', optimize=True)
    elif img_format == 'gif':
        image = image.convert("P", palette=Image.ADAPTIVE)
        image.save(output_path, optimize=True)
    elif img_format == 'tiff':
        image.save(output_path, 'TIFF', compression='tiff_deflate')
    else:
        raise Exception(f"不支持的图片格式: {img_format}")

    return output_path
    
# 删除本地文件
def delete_local_file(file_path):
    if os.path.exists(file_path):
        os.remove(file_path)
        print(f"{file_path} 已删除。")
    else:
        print(f"{file_path} 文件不存在。")


def run(params):
    params = json.loads(params)
    url = params['url']
    version_id = params['id']
    access_key_id = params['access_key_id']
    access_key_secret = params['access_key_secret']
    securityToken = params['securityToken']
    endpoint = params['endpoint']
    bucket_name = params['bucket_name']
    upload_filename = params['upload_filename']
    callback_url = params['callback_url']
    model_type = params['model_type']
    compress = params.get('compress',False)
    quality = params.get('quality',85)
    env_text = params.get('env','-')
    logPrefix = f"{formatNow()}-{env_text}-"
    # url,version_id,access_key_id,access_key_secret,securityToken,endpoint,bucket_name,upload_filename,callback_url,callback_header_key,callback_header_secret
    # 网络图片 URL 和本地保存路径
    tmp_file_name = uuid.uuid4()
    result_image = f'{tmp_file_name}_compressed_image.jpg'
    print(f"开始运行图片 {url}")
    # 步骤 1：下载图片到本地
    downloaded_image = download_image(url)
    
    if downloaded_image:
        if compress:
            output_image_path = compress_image(downloaded_image,result_image)
        else:
            output_image_path = remove_metadata(downloaded_image,result_image)
            print(f"{logPrefix}元数据移除完成")
        if access_key_id and endpoint and bucket_name:
            print(f'{logPrefix}上传OSS')
            # 步骤 3：上传到阿里云 OSS
            oss_url = upload_to_oss(output_image_path, upload_filename, bucket_name, access_key_id, access_key_secret,securityToken, endpoint)
        
            # 步骤 4：删除本地的临时文件
            delete_local_file(result_image)  # 删除压缩后的图片
            if oss_url:
                print(f"{logPrefix}图片上传成功，URL：{oss_url}")
                res = requests.post(callback_url,json={
                    'version_id': version_id,
                    'model_type': model_type,
                    'filename': upload_filename
                })
                print(f"回调结果 {res.status_code}")
                return f"{logPrefix}图片上传成功，URL：{oss_url}"
            else:
                return f"{logPrefix}上传到 OSS 失败"
        else:
            print(f'{logPrefix}不上传OSS')
            with open(output_image_path,'rb') as f:
                image_base64 = base64.b64encode(f.read())
            print(f"{logPrefix}运行完成")
            delete_local_file(output_image_path)  # 删除压缩后的图片
            return image_base64
    else:
        return '图片下载失败'
    return '运行失败'

with gr.Blocks() as demo:
    
    with gr.Column(elem_id="col-container"):
        with gr.Row():
            params = gr.Text(
                label="图片地址",
                show_label=False,
                max_lines=1,
                placeholder="input image url",
                container=False,
            ) 


        with gr.Row():
            output = gr.Textbox(
                label="Output",
                placeholder="Result will be displayed here",
                lines=10,
                interactive=False
            )
        
        with gr.Row():
            run_button = gr.Button("Run", scale=0)

    gr.on(
        triggers=[run_button.click],
        fn = run,
        inputs = [params],
        outputs = [output],
        concurrency_limit=20
    )

demo.launch()