Апдейт
This commit is contained in:
93
main.py
93
main.py
@ -1,9 +1,9 @@
|
|||||||
from logging import getLogger
|
from logging import getLogger
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from time import time
|
from time import time
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
import ffmpeg
|
import ffmpeg
|
||||||
from ffmpeg.nodes import FilterableStream
|
|
||||||
from rich.prompt import Prompt
|
from rich.prompt import Prompt
|
||||||
|
|
||||||
from utils.log import configure_logging
|
from utils.log import configure_logging
|
||||||
@ -12,71 +12,72 @@ PATH = Path(__file__).parent
|
|||||||
INPUT_DIR = PATH / 'input'
|
INPUT_DIR = PATH / 'input'
|
||||||
OUTPUT_DIR = PATH / 'output'
|
OUTPUT_DIR = PATH / 'output'
|
||||||
|
|
||||||
OUTPUT_ARGS = {
|
OUTPUT_ARGS: dict[str, Any] = {
|
||||||
'vcodec': 'libvpx-vp9',
|
|
||||||
'pix_fmt': 'yuva420p',
|
|
||||||
'row-mt': 1,
|
'row-mt': 1,
|
||||||
'format': 'webm',
|
|
||||||
'loglevel': 'error',
|
'loglevel': 'error',
|
||||||
'an': None,
|
|
||||||
'y': None,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
logger = getLogger(__name__)
|
logger = getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def compress(input_file: Path, output_file: Path, sticker_width: int, bitrate: int | None = None):
|
def compress(input_file: Path, output_file: Path, sticker_width: int, max_size: int):
|
||||||
file_input = ffmpeg.input(input_file)
|
file_input = ffmpeg.input(input_file)
|
||||||
file_output = OUTPUT_DIR / input_file.name.replace(input_file.suffix, '.webm')
|
file_output = OUTPUT_DIR / input_file.name.replace(input_file.suffix, '.webm')
|
||||||
|
|
||||||
if not isinstance(file_input, FilterableStream):
|
stream = file_input.fps(fps=25, round='up')
|
||||||
logger.error(f'Failed to process file: {input_file}')
|
stream = stream.scale(w=sticker_width, h=-1, force_original_aspect_ratio='decrease')
|
||||||
raise Exception('Failed to process file')
|
stream = stream.loop(loop=0)
|
||||||
|
stream = stream.crop(
|
||||||
stream = ffmpeg.filter(file_input, 'fps', fps=25, round='up')
|
out_w=f'min(min(iw,ih),{sticker_width})', out_h=f'min(min(iw,ih),{sticker_width})'
|
||||||
stream = ffmpeg.filter(
|
|
||||||
stream, 'scale', sticker_width, -1, force_original_aspect_ratio='decrease'
|
|
||||||
)
|
)
|
||||||
stream = ffmpeg.filter(
|
stream = stream.pad(color='0x00000000')
|
||||||
stream, 'pad', sticker_width, sticker_width, '(ow-iw)/2', '(oh-ih)/2', color='0x00000000'
|
|
||||||
)
|
|
||||||
stream = ffmpeg.filter(stream, 'loop', 0, 1)
|
|
||||||
|
|
||||||
if input_file.suffix == '.gif':
|
if input_file.suffix in ('.gif', '.mp4'):
|
||||||
duration = float(ffmpeg.probe(input_file)['streams'][0]['duration'])
|
duration = float(ffmpeg.probe(input_file)['streams'][0]['duration'])
|
||||||
speed_factor = 1
|
speed_factor = 1
|
||||||
|
|
||||||
if duration > 3:
|
if duration > 3:
|
||||||
speed_factor = duration / 3
|
speed_factor = duration / 3
|
||||||
|
|
||||||
stream = ffmpeg.filter(stream, 'setpts', f'PTS/{speed_factor}')
|
stream = stream.setpts(expr=f'PTS*{speed_factor}')
|
||||||
|
|
||||||
args = OUTPUT_ARGS
|
output = ffmpeg.output(
|
||||||
|
stream,
|
||||||
if bitrate:
|
filename=file_output,
|
||||||
args['video_bitrate'] = f'{bitrate}k'
|
vcodec='libvpx-vp9',
|
||||||
|
pix_fmt='yuva420p',
|
||||||
output = ffmpeg.output(stream, filename=file_output, **args)
|
extra_options=OUTPUT_ARGS,
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
ffmpeg.run(output)
|
output.run(overwrite_output=True)
|
||||||
except Exception:
|
except Exception as e:
|
||||||
|
logger.error(f"Error during first encoding: {e}")
|
||||||
exit(1)
|
exit(1)
|
||||||
|
|
||||||
file_size = file_output.stat().st_size
|
file_size = file_output.stat().st_size
|
||||||
|
|
||||||
if file_size / 1024 > 256:
|
if file_size / 1024 > max_size:
|
||||||
if not bitrate:
|
initial_duration = float(ffmpeg.probe(input_file)['format']['duration'])
|
||||||
bitrate = int(int(ffmpeg.probe(output_file)['format']['bit_rate']) / 1000)
|
bitrate = int(max_size * 8 / initial_duration * .65)
|
||||||
|
|
||||||
bitrate = int((256 * 1024) / (file_size / bitrate))
|
temp_output = file_output.with_suffix('.tmp.webm')
|
||||||
|
stream = ffmpeg.input(output_file, c='libvpx-vp9')
|
||||||
|
|
||||||
logger.info(
|
output = ffmpeg.output(
|
||||||
f'File size is too high [{int(file_size / 1024)} KB], reducing to {bitrate}k...'
|
stream,
|
||||||
|
filename=temp_output,
|
||||||
|
vcodec='libvpx-vp9',
|
||||||
|
pix_fmt='yuva420p',
|
||||||
|
b=f'{bitrate}k',
|
||||||
|
extra_options=OUTPUT_ARGS,
|
||||||
)
|
)
|
||||||
|
|
||||||
return compress(input_file, output_file, sticker_width, bitrate)
|
output.run(overwrite_output=True)
|
||||||
|
|
||||||
|
file_output.unlink()
|
||||||
|
temp_output.rename(output_file)
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
@ -89,29 +90,37 @@ def main():
|
|||||||
for file in OUTPUT_DIR.iterdir():
|
for file in OUTPUT_DIR.iterdir():
|
||||||
file.unlink()
|
file.unlink()
|
||||||
|
|
||||||
sticker_type = Prompt.ask('Do you want to create stickers or emoji?', choices=['s', 'e'])
|
sticker_type = Prompt.ask(
|
||||||
|
'Do you want to create stickers or emoji or icon?', choices=['s', 'e', 'i']
|
||||||
|
)
|
||||||
|
|
||||||
match sticker_type:
|
match sticker_type:
|
||||||
case 's':
|
case 's':
|
||||||
STICKER_WIDTH = 512
|
width = 512
|
||||||
|
size = 256
|
||||||
case 'e':
|
case 'e':
|
||||||
STICKER_WIDTH = 100
|
width = 100
|
||||||
|
size = 256
|
||||||
|
case 'i':
|
||||||
|
width = 100
|
||||||
|
size = 32
|
||||||
case _:
|
case _:
|
||||||
STICKER_WIDTH = 512
|
width = 512
|
||||||
|
size = 256
|
||||||
|
|
||||||
started_at = time()
|
started_at = time()
|
||||||
|
|
||||||
for file in INPUT_DIR.iterdir():
|
for file in INPUT_DIR.iterdir():
|
||||||
logger.info(f'Processing {file.name}...')
|
logger.info(f'Processing {file.name}...')
|
||||||
|
|
||||||
if file.suffix not in ('.png', '.jpg', '.webp', '.gif'):
|
if file.suffix not in ('.png', '.jpg', '.webp', '.gif', '.mp4'):
|
||||||
logger.warning(f'Unsupported file type: {file.suffix}')
|
logger.warning(f'Unsupported file type: {file.suffix}')
|
||||||
continue
|
continue
|
||||||
|
|
||||||
file_path = INPUT_DIR / file.name
|
file_path = INPUT_DIR / file.name
|
||||||
file_output = OUTPUT_DIR / file.name.replace(file.suffix, '.webm')
|
file_output = OUTPUT_DIR / file.name.replace(file.suffix, '.webm')
|
||||||
|
|
||||||
compress(file_path, file_output, STICKER_WIDTH)
|
compress(file_path, file_output, width, size)
|
||||||
|
|
||||||
logger.info(f'Finished in {round(time() - started_at, 2)} seconds')
|
logger.info(f'Finished in {round(time() - started_at, 2)} seconds')
|
||||||
|
|
||||||
|
|||||||
@ -5,8 +5,8 @@ description = "A simple tool to convert any images and/or gifs to webm extension
|
|||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.13"
|
requires-python = ">=3.13"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"ffmpeg-python==0.2.0",
|
|
||||||
"rich==13.9.4",
|
"rich==13.9.4",
|
||||||
|
"typed-ffmpeg==2.6.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
[dependency-groups]
|
[dependency-groups]
|
||||||
|
|||||||
@ -13,3 +13,5 @@ def configure_logging():
|
|||||||
datefmt=DATE_FORMAT,
|
datefmt=DATE_FORMAT,
|
||||||
handlers=[RichHandler(show_time=False, rich_tracebacks=True)],
|
handlers=[RichHandler(show_time=False, rich_tracebacks=True)],
|
||||||
)
|
)
|
||||||
|
|
||||||
|
logging.getLogger('ffmpeg').setLevel(logging.ERROR)
|
||||||
|
|||||||
Reference in New Issue
Block a user