scripts/cronjob/media/compress.py

from pymongo import UpdateOne
from lib.all import *
from PIL import Image
import oss2

service_name = 'compress'
log_dir = '/app/log'
logger = Logger(service_name, log_dir=log_dir)

access_key_id = "LTAI5tAdu5LRvZwm4LJa21Fo"
access_key_secret = "WGvSQsDralTfFAAxhEqLBOgbXqflHo"
endpoint_internal = "https://oss-cn-hangzhou-internal.aliyuncs.com"
bucket_name = "wishpal-ironfan-media"

HorizontalView = 1  # 横图
VerticalView = 2  # 竖图

Resize720P = 720
Resize1080P = 1080
Resize1440P = 1440


def calc_new_file_size(img_size: tuple, resize_p: int = Resize720P):
    w = int(img_size[0])
    h = int(img_size[1])

    view_type = HorizontalView if w > h else VerticalView

    if resize_p == Resize720P:
        resize_format = 1280
    elif resize_p == Resize1080P:
        resize_format = 1920
    elif resize_p == Resize1440P:
        resize_format = 2560
    else:
        resize_format = 1280

    if view_type == HorizontalView:
        reduce_rate = safe_div(resize_format, w)
        new_w = int(float(w) * reduce_rate)
        new_h = int(float(h) * reduce_rate)
        return new_w, new_h
    else:
        reduce_rate = safe_div(resize_format, h)
        new_w = int(float(w) * reduce_rate)
        new_h = int(float(h) * reduce_rate)
        return new_w, new_h


def compress_and_save_image(img_path: str, dst_path: str, resize_p: int):
    try:
        img = Image.open(img_path)
        resize_img = img.resize((calc_new_file_size(img.size, resize_p)))
        resize_img.save(dst_path)
        return "success"
    except Exception as e:
        return str(e)


class S:
    def __init__(self):
        self.bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint_internal, bucket_name)
        self.col_image = MongoDB(
            host="mongodb://root:Wishpal2024@dds-bp1a72ede37c55d4242490.mongodb.rds.aliyuncs.com:3717/admin",
            port=3717,
            db="media",
            collection="image"
        )
        print(self.col_image)

    def __del__(self):
        self.col_image.close()

    def save_img_from_oss(self, oss_src_id: str, local_path: str):
        return self.bucket.get_object_to_file(oss_src_id, local_path)

    def upload_img_to_oss(self, local_path: str, oss_src_id: str):
        return self.bucket.put_object_from_file(oss_src_id, local_path)

    def deal_one(self, oss_src_id: str, oss_src_download_path: str, fmt: str, resize_p: int):
        oss_src_id_python_type = oss_src_id.replace("/", "_")
        # 当前目录
        cur_dir = os.getcwd() + "/"
        # 源文件转resize_p
        oss_resize_dir = "imgprod{}".format(resize_p)
        oss_resize_src_id = oss_src_id.replace("imgprod", oss_resize_dir)
        local_resize_path = cur_dir + oss_src_id_python_type.replace("imgprod", oss_resize_dir) + ".{}".format(fmt)
        ret = compress_and_save_image(oss_src_download_path, local_resize_path, resize_p)
        if ret != "success":
            logger.Error("compress_and_save_image {} fail, err: {}, src: {}, local: {}".format(resize_p, ret, oss_src_download_path, local_resize_path))
            return None

        try:
            ret = self.upload_img_to_oss(local_resize_path, oss_resize_src_id)
            if ret.status != 200:
                logger.Error("upload_img_to_oss {} fail, err: {}, src: {}, local: {}".format(resize_p, ret, oss_src_download_path, local_resize_path))
                return None

        except Exception as e:
            logger.Error("upload_img_to_oss {} fail, err: {}, src: {}, local: {}".format(resize_p, str(e), oss_src_download_path, local_resize_path))
            return None

        file_size = os.path.getsize(local_resize_path)
        q = {
            "src_id": oss_src_id
        }
        up = {
            "$set": {
                "size_{}".format(resize_p): file_size,
                "src_id_{}".format(resize_p): oss_resize_src_id
            }
        }
        uo = UpdateOne(q, up)
        os.remove(local_resize_path)
        return uo

    def get_wait_deal_images(self):
        q = {
            "status": 0
        }
        images = list()
        images_tmp = self.col_image.find(q)
        logger.Info("len(images_tmp): {}".format(len(images_tmp)))
        for img in images_tmp:
            fmt = safe_get_str(img, "fmt")
            if fmt.find("heic") > 0:
                continue
            images.append(img)
        return images

    def proc(self):
        images = self.get_wait_deal_images()
        logger.Info("====== {} ====== len(images): {}".format(get_time_str_by_ts(int(time.time())), len(images)))
        idx = 0
        for image in images:
            idx += 1
            # 当前目录
            cur_dir = os.getcwd() + "/"

            oss_src_id = safe_get_str(image, "src_id")
            src_id_python_type = oss_src_id.replace("/", "_")
            fmt_ori = safe_get_str(image, "fmt")
            fmt = fmt_ori.replace("image/", "")

            bulk_reqs = list()

            status = 100

            # 先下载到本地，源文件 imgprod/a4/98/22b4-391d-445d-834c-5048fb77653c
            oss_src_download_path = cur_dir + src_id_python_type + ".{}".format(fmt)
            try:
                self.save_img_from_oss(oss_src_id, oss_src_download_path)
            except Exception as e:
                logger.Error("save_img_from_oss fail, err: {}, src_id: {}".format(str(e), oss_src_id))
                continue

            # 720P
            uo720 = self.deal_one(oss_src_id, oss_src_download_path, fmt, Resize720P)
            if not uo720:
                status = 101
                logger.Error("deal_one 720 error, oss_src_id: {}".format(oss_src_id))
            else:
                bulk_reqs.append(uo720)

            # 1080P
            uo1080 = self.deal_one(oss_src_id, oss_src_download_path, fmt, Resize1080P)
            if not uo1080:
                status = 101
                logger.Error("deal_one 1080 error, oss_src_id: {}".format(oss_src_id))
            else:
                bulk_reqs.append(uo1080)

            # 1440P
            uo1440 = self.deal_one(oss_src_id, oss_src_download_path, fmt, Resize1440P)
            if not uo1440:
                status = 101
                logger.Error("deal_one 1440 error, oss_src_id: {}".format(oss_src_id))
            else:
                bulk_reqs.append(uo1440)

            src_size = os.path.getsize(oss_src_download_path)
            q1 = {
                "src_id": oss_src_id,
            }
            up1 = {
                "$set": {
                    "size_src": src_size,
                    "resize_t": int(time.time()),
                    "status": status
                }
            }
            bulk_reqs.append(UpdateOne(q1, up1))

            os.remove(oss_src_download_path)
            mongo_ret = self.col_image.bulk_write(bulk_reqs)
            if not mongo_ret:
                logger.Error("bulk_write fail, oss_src_id: {}".format(oss_src_id))

            logger.Info("{}/{}, success, oss_src_id: {}, id: {}".format(idx, len(images), oss_src_id, safe_get_int(image, "_id")))


s = S()
while True:
    s.proc()
    time.sleep(5)