Python转换图片格式（png转换成webp）、批量上传至S3、生成json文件

刺骨的言语ヽ痛彻心扉 2022-10-10 04:48 496阅读 0赞

在工作中遇到一个需求，需要将运营给出的png格式图片转换成webp格式，并将要求的图片上传至S3，同时生成一份json文件返回给产品。  
其实一个脚本就可以搞定，但是后续每次转换上传，需要对路径等作出更改，考虑到为了让运营人员也可以轻松使用，在此使用Flask做一个页面，将所必须的信息填写好执行即可。  
**效果图：  
index页面**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70]  
**成功跳转页面：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70 1]  
成功后进入S3查看确实已经上传，生成的json文件会生成在本地。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70 2]

**失败跳转页面：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70 3]

废话少说，上代码。  
**1.首先编写一个简单的html页面，核心代码如下：**

/*页面样式引入的bootstrap*/
    <div class="row justify-content-center " style="height: 500px">
        <div class="align-self-center">
            <form action="http://localhost:8081/get/sum" target="_blank" method="POST">
                <div class="form-row">
                    <div class="form-group col-sm-6">
                        <label for="exampleInput1">AWS_ID</label>
                        <input type="password" class="form-control" name="aws_id" id="exampleInput1" aria-describedby="emailHelp" placeholder="AWS_ACCESS_KEY_ID">
                    </div>
                    <div class="form-group col-sm-6">
                        <label for="exampleInput2">AWS_KEY</label>
                        <input type="password" class="form-control" name="aws_key" id="exampleInput2" placeholder="AWS_SECRET_ACCESS_KEY">
                    </div>
                </div>
                <div class="form-row">
                    <div class="form-group col-sm-12">
                        <label for="exampleInput3">输入您的 本地上传图片文件夹路径</label>
                        <input type="text" class="form-control" name="local_path" id="exampleInput3" placeholder="eg:/Users/xm20201102/Documents/bigemoji_0519_png">
                    </div>
                    <div class="form-group col-sm-12">
                    /*输入这个是工作需求中为了生成最后的json文件*/
                        <label for="exampleInput4">输入您下载的 emojiUnicode的csv文件路径</label>
                        <input type="text" class="form-control" id="exampleInput4" name="csv_path" placeholder="eg:/Users/xm20201102/Desktop/big_emoji_list_1.csv">
                    </div>
                </div>
                <div class="form-check" align="center">
                    <input type="checkbox" class="form-check-input" id="exampleCheck1">
                    <label class="form-check-label" for="exampleCheck1">记住我的选择</label>
                </div>
                <div align="center">
                    <button type="submit" class="btn btn-primary">Go!</button>
                </div>
            </form>
        </div>
    </div>

**2.编写后台处理逻辑：**  
首先导入我们需要的包：

import requests
    from flask import Flask, request, jsonify, redirect
    import boto3
    import time
    import os
    from PIL import Image
    from flask.helpers import url_for
    from flask.templating import render_template

定义一些全局变量：

BUCKET_NAME = 'upload-bucket'  #S3上已经创建好的桶名称
    REGION_NAME = 'us-east-1'	  #S3账户时区
    #以下四个路径是S3存储桶中图片存放的路径，因为不同图片存放路径不同，一共有四个路径。
    PHOTO_DIRNAME_3d_WEBP = '3d/webp/'
    PHOTO_DIRNAME_3d_PNG = '3d/png/'
    PHOTO_DIRNAME_2d_WEBP = '2d/webp/'
    PHOTO_DIRNAME_2d_PNG = '2d/png/'
    
    S3_PATH = ''					#生成json文件时用到，根据自己需求是否保留，https://cdn域名/存放路径
    AWS_ACCESS_KEY_ID = '' 			#AWS账户Access key ID,从前端页面接收，也可写死，写死后去除前端输入框
    AWS_SECRET_ACCESS_KEY = ''		#AWS账户Secret access key,从前端页面接收，也可写死，写死后去除前端输入框
    PHOTO_FOLDER = ''				#本地图片文件夹存放路径，从前端页面接收
    BIG_EMOJI_UNICODE_LIST_CSV = '' #生成json文件时用到，根据自己需求是否保留，从前端页面接收，本次需求是从给出的csv文件中提出每行的emojiUnicode
    EMOJI_UNICODE = 'None'			#生成json文件时用到，根据自己需求是否保留，会在后续赋值
    LOCAL_PATH = ''					#生成json文件时用到，根据自己需求是否保留，会在后续赋值
    
    app = Flask(__name__)			#使用flask必须加这个
    #初始化生成一个app对象，这个对象就是Flask的当前实例对象，后面的各个方法调用都是这个实例 
    #Flask会进行一系列自己的初始化，比如web API路径初始化，web资源加载，日志模块创建等。然后返回这个创建好的对象给你

接收前端输入数据并处理：

@app.route("/get/sum", methods=["GET", "POST"])
    def get_sum():
        global AWS_ACCESS_KEY_ID
        global AWS_SECRET_ACCESS_KEY
        global PHOTO_FOLDER
        global BIG_EMOJI_UNICODE_LIST_CSV
        global LOCAL_PATH
    	#给已定义的全局变量赋值
        AWS_ACCESS_KEY_ID = request.form['aws_id']
        AWS_SECRET_ACCESS_KEY = request.form['aws_key']
        PHOTO_FOLDER = request.form['local_path']
        BIG_EMOJI_UNICODE_LIST_CSV = request.form['csv_path']
        LOCAL_PATH = PHOTO_FOLDER[:-1]
    	#若输入的信息有空，重定向到错误页面
        if AWS_ACCESS_KEY_ID == '' or AWS_SECRET_ACCESS_KEY == '' or PHOTO_FOLDER == '' or BIG_EMOJI_UNICODE_LIST_CSV == '':
            return redirect(url_for('get_error'))
    	#开始逻辑处理
        start()			#选择png格式图片并转换成webp格式
        upload_photo_512_and_150_webp()	# 批量上传512和150大小的webp图片
        upload_photo_512_png()				# 批量上传512大小的png图片
        print_json()	#输出json文件
        return redirect(url_for('get_response', path=PHOTO_FOLDER, photo='static/home3_02_05.png'))
    
    #调用成功页面
    @app.route("/get_response", methods=["GET", "POST"])
    def get_response():
        return render_template("success.html",
                               path=request.args.get('path'), photo=request.args.get('photo'))
    
    #调用失败页面
    @app.route("/get_error", methods=["GET", "POST"])
    def get_error():
        return render_template("error.html")
    
    
    
    # 去除目录中非png格式的，只选择png格式进行转换，开始转换。若没有则不转换
    def start():
        filePath = PHOTO_FOLDER
        res = os.listdir(filePath)
        for path in res:
        #mac本读取文件夹内容时，会有默认存在的.DS_Store文件，和文件排序等有关，不转换它，也可以在终端输入命令让其不生成
            if path != '.DS_Store' and path.endswith('.png'):
                to_webp(filePath + path)
    
    # 转换格式png to webp
    def to_webp(path):
        im = Image.open(path)
        save_dir = path.replace('0512', '0512_webp')
        save_dir = save_dir.replace('.png', '.webp')
        im.save(save_dir, "WEBP")
    
    # 批量上传*.webp
    def upload_photo_512_and_150_webp():
        os.chdir(PHOTO_FOLDER)
        logo_list = os.listdir(PHOTO_FOLDER)
        for logo in logo_list:
            if not os.path.isfile(os.path.abspath(logo)):
                logo_list.remove(logo)
            if logo.endswith('.webp') and logo != '.DS_Store':
                upload_photo(photo_name=logo)
    
    # 批量上传512*512.png
    def upload_photo_512_png():
        os.chdir(PHOTO_FOLDER)
        logo_list = os.listdir(PHOTO_FOLDER)
        for logo in logo_list:
            if not os.path.isfile(os.path.abspath(logo)):
                logo_list.remove(logo)
            # 只检测后缀为_512.png的上传,_150.png的不上传
            if logo.endswith('_512.png') and logo != '.DS_Store':
                upload_photo(photo_name=logo)
    
    # S3上传单个文件
    def upload_photo(bucket_name=BUCKET_NAME, photo_name=None):
        upload_photo_s3 = init_s3_photo(REGION_NAME, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
        if '3d' in photo_name:
            if photo_name.endswith('.webp'):
                with open(os.path.join(PHOTO_FOLDER, photo_name), 'rb') as f:
                    photo_stream = f.read()
                key_name = PHOTO_DIRNAME_3d_WEBP + photo_name
    
            else:
                with open(os.path.join(PHOTO_FOLDER, photo_name), 'rb') as f:
                    photo_stream = f.read()
                key_name = PHOTO_DIRNAME_3d_PNG + photo_name
        else:
            if photo_name.endswith('.webp'):
                with open(os.path.join(PHOTO_FOLDER, photo_name), 'rb') as f:
                    photo_stream = f.read()
                key_name = PHOTO_DIRNAME_2d_WEBP + photo_name
    
            else:
                with open(os.path.join(PHOTO_FOLDER, photo_name), 'rb') as f:
                    photo_stream = f.read()
                key_name = PHOTO_DIRNAME_2d_PNG + photo_name
        try:
            # upload to s3，key_name指定S3上的路径
            upload_photo_s3.Bucket(bucket_name).upload_file(Filename=photo_name, Key=key_name,
                                                          ExtraArgs={ 'ACL': 'public-read'})
    		#注意上传时必须加这个参数ExtraArgs={'ACL': 'public-read'}，否则会报没有权限错误
            print("{} upload done".format(photo_name))
        except Exception as e:
            print("upload {} error:{}".format(photo_name, e))
    
    # 生成json并写到文件里
    def print_json():
        global EMOJI_UNICODE
        data = []
        count = 0
        for root, dirs, files in os.walk(LOCAL_PATH):
            for name in files:
                if name.endswith('.webp'):
                    s3_path_512 = os.path.join(root, name).replace(LOCAL_PATH,
                                                                   S3_PATH)
                    if s3_path_512.endswith('_150.webp'):
                        webp_150 = s3_path_512
                    else:
                        webp_150 = s3_path_512.replace("_512.webp", "_150.webp")
    
                    temp_path = webp_150
                    webp_512 = temp_path.replace("_150.webp", "_512.webp")
                    id = name[:-9]
                    # 下载的csv路径
                    with open(BIG_EMOJI_UNICODE_LIST_CSV) as f:
                        for line in f:
                            if id == line.split(",")[0]:
                                EMOJI_UNICODE = line.split(",")[1]
                                break
                    if '2d' in id:
                        webp_150 = webp_150.replace("3d", "2d")
                        webp_512 = webp_512.replace("3d", "2d")
                    nameJson = { 
                        "emoji_unicode": EMOJI_UNICODE,
                        "big_emoji_id": id,
                        "big_emoji_preview": webp_150,
                        "big_emoji_url_webp": webp_512,
                        "big_emoji_url_png": webp_512.replace("webp", "png")
                    }
                    if nameJson not in data:
                        data.append(nameJson)
                    count += 1
    
        sort_data = sorted(data, key=lambda e: e.__getitem__('big_emoji_id'))
        JSON_data = str(sort_data).replace("'", "\"").replace(r"\n", "")
        now_time = int((time.time()))
        time_local = time.localtime(now_time)
        dt = time.strftime("%Y-%m-%d_%H:%M:%S", time_local)
        # 输出到当前图片目录json文件中
        with open(dt + "_upload_bigEmoji.json", 'w') as f:
            f.write(JSON_data)
    
    # 连接S3
    def init_s3_photo(region_name, aws_access_key_id,
                           aws_secret_access_key):
        s3 = boto3.resource('s3', region_name=region_name, aws_access_key_id=aws_access_key_id,
                            aws_secret_access_key=aws_secret_access_key)
        return s3
    
    #主函数入口，可以设置端口号
    if __name__ == "__main__":
        app.config["JSON_AS_ASCII"] = False
        app.run(host="127.0.0.1", port=8081)

到这里就基本完成了，还有一个成功或失败跳转的html页面没有放上来，随便做一个就行了，我们主要目的是实现需求功能，并可以持续使用。如果有问题或者指导我的小伙伴们可以评论我，看到都会回复。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70]: /images/20221005/eca8b4ce718b458c95ad320b4944a6bb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70 1]: /images/20221005/337c59b9a5384e1a81e763faada0f0a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70 2]: /images/20221005/bf17a5193f0448ef8c3cc99d98ebd7fb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDEyMzU0MA_size_16_color_FFFFFF_t_70 3]: /images/20221005/76e46f695f4449699508d32faf7d70a2.png