Asset-Compression-Manager/acm/config.py

import importlib.metadata
import json
import logging
import typing

from pydantic import BaseModel, BaseSettings, validator

from acm.utility import get_string_sha256sum, get_string_xor
from acm.version import VERSION

LOG = logging.getLogger("acm.config")


class ACMProfileProcessorOptions(BaseModel):
    force_preserve_smaller_input: bool = False


class ACMProfileProcessor(BaseModel):
    name: str
    version: typing.Optional[str]
    processors: typing.List[str]
    extensions: typing.List[str]
    output_extension: str
    options: ACMProfileProcessorOptions
    command: str
    signature: typing.Optional[str]

    @validator('version', always=True)
    def version_validator(cls, v, values) -> str:
        # TODO Set the version to the app version if not provided
        if v is None:
            return VERSION

    @validator('signature', always=True)
    def signature_validator(cls, v, values) -> str:
        signature_keys = ["name", "version", "processors", "extensions", "output_extension", "command"]
        signature_values = [value for key, value in values.items() if key in signature_keys]
        return get_string_sha256sum(json.dumps(signature_values))


class ACMProfile(BaseModel):
    name: str
    version: typing.Optional[str]
    processors: typing.List[ACMProfileProcessor]
    signature: typing.Optional[str]

    @validator('version', always=True)
    def version_validator(cls, v, values) -> str:
        if v is None:
            return VERSION

    @validator('signature', always=True)
    def hash_signature_validator(cls, v, values) -> str:
        signature_keys = ["name", "version"]
        signature_values = [value for key, value in values.items() if key in signature_keys]
        signature = get_string_sha256sum(json.dumps(signature_values))
        
        processor_signatures = [processor.signature for processor in values["processors"]]
        if len(processor_signatures) > 1:
            combined_processor_signature = get_string_xor(*processor_signatures)
        else:
            combined_processor_signature = processor_signatures[0]

        return get_string_sha256sum(signature + combined_processor_signature)

    def get_processor_names(self) -> typing.List[str]:
        return [processor.name for processor in self.processors]

    def get_processor(self, name: str) -> typing.Optional[ACMProfileProcessor]:
        for processor in self.processors:
            if name == processor.name:
                return processor
        return None


class ACMS3(BaseModel):
    secure: bool = False,
    host: str = "127.0.0.1:9000"
    access_key: typing.Optional[str]
    secret_key: typing.Optional[str]


class ACMConfig(BaseSettings):
    concurrency: int = 0
    debug: bool = False
    s3: typing.Optional[ACMS3]
    version: typing.Optional[str]
    profiles: typing.List[ACMProfile]
    signature: typing.Optional[str]

    @validator('version', always=True)
    def version_validator(cls, v, values) -> str:
        if v is None:
            return VERSION

    @validator('signature', always=True)
    def signature_validator(cls, v, values) -> str:
        signature_keys = ["version"]
        signature_values = [value for key, value in values.items() if key in signature_keys]
        signature = get_string_sha256sum(json.dumps(signature_values))
        
        profiles_signatures = [profiles.signature for profiles in values["profiles"]]
        if len(profiles_signatures) > 1:
            combined_profiles_signature = get_string_xor(*profiles_signatures)
        else:
            combined_profiles_signature = profiles_signatures[0]

        return get_string_sha256sum(signature + combined_profiles_signature)

    class Config:
        env_prefix = 'ACM_'
        env_nested_delimiter = '__'

    def get_profile_names(self) -> typing.List[str]:
        return [profile.name for profile in self.profiles]

    def get_profile(self, name: str) -> typing.Optional[ACMProfile]:
        for profile in self.profiles:
            if name == profile.name:
                return profile
        return None


def get_default_config():
    """
    Returns the default ACM config
    """
    acm_profiles = []

    # default #
    acm_default_processors = []
    acm_default_processors.append(ACMProfileProcessor(
        name = "jpeg",
        processors = ["cjpeg"],
        extensions = ["jpg", "jpeg"],
        output_extension = "jpg",
        options = ACMProfileProcessorOptions(force_preserve_smaller_input=True),
        command = "cjpeg -optimize -quality 90 -progressive -outfile {output_file} {input_file}"
    ))
    acm_default_processors.append(ACMProfileProcessor(
        name = "png",
        processors = ["optipng"],
        extensions = ["png"],
        output_extension = "png",
        options = ACMProfileProcessorOptions(force_preserve_smaller_input=True),
        command = "optipng -o2 -strip all -out {output_file} {input_file}"
    ))
    acm_default_processors.append(ACMProfileProcessor(
        name = "video",
        processors = ["ffmpeg"],
        extensions = ["mp4","webm"],
        output_extension = "webm",
        options = ACMProfileProcessorOptions(),
        command = "optipng -o2 -strip all -out {output_file} {input_file}"
    ))
    acm_default_processors.append(ACMProfileProcessor(
        name = "audio",
        processors = ["ffmpeg","opusenc"],
        extensions = ["wav","mp3"],
        output_extension = "ogg",
        options = ACMProfileProcessorOptions(),
        command = "optipng -o2 -strip all -out {output_file} {input_file}"
    ))
    acm_profiles.append(ACMProfile(
        name = "default",
        processors = acm_default_processors
    ))

    # placebo #
    acm_placebo_processors = []
    acm_placebo_processors.append(ACMProfileProcessor(
        name = "jpeg",
        processors = ["cjpeg"],
        extensions = ["jpg", "jpeg"],
        output_extension = "jpg",
        options = ACMProfileProcessorOptions(),
        command = "cp {input_file} {output_file}"
    ))
    acm_placebo_processors.append(ACMProfileProcessor(
        name = "png",
        processors = ["optipng"],
        extensions = ["png"],
        output_extension = "png",
        options = ACMProfileProcessorOptions(),
        command = "cp {input_file} {output_file}"
    ))
    acm_placebo_processors.append(ACMProfileProcessor(
        name = "video",
        processors = ["ffmpeg"],
        extensions = ["mp4","webm"],
        output_extension = "webm",
        options = ACMProfileProcessorOptions(),
        command = "cp {input_file} {output_file}"
    ))
    acm_placebo_processors.append(ACMProfileProcessor(
        name = "audio",
        processors = ["ffmpeg","opusenc"],
        extensions = ["wav","mp3"],
        output_extension = "ogg",
        options = ACMProfileProcessorOptions(),
        command = "cp {input_file} {output_file}"
    ))
    acm_profiles.append(ACMProfile(
        name = "placebo",
        processors = acm_placebo_processors
    ))

    # webp #
    acm_webp_processors = []
    acm_webp_processors.append(ACMProfileProcessor(
        name = "jpeg",
        processors = ["cwebp"],
        extensions = ["jpg", "jpeg"],
        output_extension = "jpg",
        options = ACMProfileProcessorOptions(),
        command = "cwebp -jpeg_like -q 90 -o {output_file} {input_file}"
    ))
    acm_webp_processors.append(ACMProfileProcessor(
        name = "png",
        processors = ["cwebp"],
        extensions = ["png"],
        output_extension = "png",
        options = ACMProfileProcessorOptions(),
        command = "cwebp -lossless -o {output_file} {input_file}"
    ))
    acm_profiles.append(ACMProfile(
        name = "webp",
        processors = acm_webp_processors
    ))

    # aggressive #
    acm_aggressive_processors = []
    acm_aggressive_processors.append(ACMProfileProcessor(
        name = "jpeg",
        processors = ["cjpeg"],
        extensions = ["jpg", "jpeg"],
        output_extension = "jpg",
        options = ACMProfileProcessorOptions(force_preserve_smaller_input=True),
        command = "export FILE={output_file} && export TEMP_FILE=${FILE}_tmp.jpg && ffmpeg -i {input_file} -vf scale=-1:720 ${TEMP_FILE} && cjpeg -optimize -quality 75 -progressive -outfile {output_file} ${TEMP_FILE} && rm ${TEMP_FILE}"
    ))
    acm_aggressive_processors.append(ACMProfileProcessor(
        name = "png",
        processors = ["optipng"],
        extensions = ["png"],
        output_extension = "png",
        options = ACMProfileProcessorOptions(force_preserve_smaller_input=True),
        command = "optipng -o2 -strip all -out {output_file} {input_file}"
    ))
    acm_aggressive_processors.append(ACMProfileProcessor(
        name = "video",
        processors = ["ffmpeg"],
        extensions = ["mp4","webm"],
        output_extension = "webm",
        options = ACMProfileProcessorOptions(),
        command = "ffmpeg -hide_banner -loglevel panic -i {input_file} -vf scale=-1:720 -c:v libvpx-vp9 -b:v 0 -crf 38 -c:a libopus {output_file}"
    ))
    acm_aggressive_processors.append(ACMProfileProcessor(
        name = "audio",
        processors = ["ffmpeg","opusenc"],
        extensions = ["wav","mp3"],
        output_extension = "ogg",
        options = ACMProfileProcessorOptions(),
        command = "ffmpeg -hide_banner -loglevel panic -i {input_file} -f wav -| opusenc --bitrate 64 --vbr --downmix-stereo --discard-comments --discard-pictures - {output_file}"
    ))
    acm_profiles.append(ACMProfile(
        name = "aggressive",
        processors = acm_aggressive_processors
    ))

    # aggressive-webp #
    acm_aggressive_webp_processors = []
    acm_aggressive_webp_processors.append(ACMProfileProcessor(
        name = "jpeg",
        processors = ["cwebp"],
        extensions = ["jpg", "jpeg"],
        output_extension = "jpg",
        options = ACMProfileProcessorOptions(),
        command = "export FILE={output_file} && export TEMP_FILE=${FILE}_tmp.jpg && ffmpeg -i {input_file} -vf scale=-1:720 ${TEMP_FILE} && cwebp -jpeg_like -q 75 -o {output_file} ${TEMP_FILE} && rm ${TEMP_FILE}"
    ))
    acm_aggressive_webp_processors.append(ACMProfileProcessor(
        name = "png",
        processors = ["optipng"],
        extensions = ["png"],
        output_extension = "png",
        options = ACMProfileProcessorOptions(),
        command = "cwebp -o {output_file} ${input_file}"
    ))
    acm_aggressive_webp_processors.append(ACMProfileProcessor(
        name = "video",
        processors = ["ffmpeg"],
        extensions = ["mp4","webm"],
        output_extension = "webm",
        options = ACMProfileProcessorOptions(),
        command = "ffmpeg -hide_banner -loglevel panic -i {input_file} -vf scale=-1:720 -c:v libvpx-vp9 -b:v 0 -crf 38 -c:a libopus {output_file}"
    ))
    acm_aggressive_webp_processors.append(ACMProfileProcessor(
        name = "audio",
        processors = ["ffmpeg","opusenc"],
        extensions = ["wav","mp3"],
        output_extension = "ogg",
        options = ACMProfileProcessorOptions(),
        command = "ffmpeg -hide_banner -loglevel panic -i {input_file} -f wav -| opusenc --bitrate 64 --vbr --downmix-stereo --discard-comments --discard-pictures - {output_file}"
    ))
    acm_profiles.append(ACMProfile(
        name = "aggressive-webp",
        processors = acm_aggressive_webp_processors
    ))

    return ACMConfig(
        profiles=acm_profiles
    )