|
|
import json import logging import typing
from pydantic import BaseModel, BaseSettings, validator
from acm.utility import get_string_sha256sum, get_string_xor
LOG = logging.getLogger("acm.config")
# Application Version VERSION = "2.0.0"
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('processors', always=True) # def processors_validator(cls, v, values) -> str: # # Collapse the same named processors into a single processor at the correct index
@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)
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 )
|