forked from kalanakt/All-Url-Uploader
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathhelp_Nekmo_ffmpeg.py
159 lines (149 loc) · 5.1 KB
/
help_Nekmo_ffmpeg.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# (c) Shrimadhav U K
# the logging things
from hachoir.parser import createParser
from hachoir.metadata import extractMetadata
import time
import os
import asyncio
import logging
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
async def place_water_mark(input_file, output_file, water_mark_file):
watermarked_file = output_file + ".watermark.png"
metadata = extractMetadata(createParser(input_file))
width = metadata.get("width")
# https://stackoverflow.com/a/34547184/4723940
shrink_watermark_file_genertor_command = [
"ffmpeg",
"-i", water_mark_file,
"-y -v quiet",
"-vf",
"scale={}*0.5:-1".format(width),
watermarked_file
]
# print(shrink_watermark_file_genertor_command)
process = await asyncio.create_subprocess_exec(
*shrink_watermark_file_genertor_command,
# stdout must a pipe to be accessible as process.stdout
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
# Wait for the subprocess to finish
stdout, stderr = await process.communicate()
e_response = stderr.decode().strip()
t_response = stdout.decode().strip()
commands_to_execute = [
"ffmpeg",
"-i", input_file,
"-i", watermarked_file,
"-filter_complex",
# https://stackoverflow.com/a/16235519
# "\"[0:0] scale=400:225 [wm]; [wm][1:0] overlay=305:0 [out]\"",
# "-map \"[out]\" -b:v 896k -r 20 -an ",
"\"overlay=(main_w-overlay_w):(main_h-overlay_h)\"",
# "-vf \"drawtext=text='@FFMovingPictureExpertGroupBOT':x=W-(W/2):y=H-(H/2):fontfile=" + Config.FONT_FILE + ":fontsize=12:fontcolor=white:shadowcolor=black:shadowx=5:shadowy=5\"",
output_file
]
# print(commands_to_execute)
process = await asyncio.create_subprocess_exec(
*commands_to_execute,
# stdout must a pipe to be accessible as process.stdout
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
# Wait for the subprocess to finish
stdout, stderr = await process.communicate()
e_response = stderr.decode().strip()
t_response = stdout.decode().strip()
return output_file
async def take_screen_shot(video_file, output_directory, ttl):
# https://stackoverflow.com/a/13891070/4723940
out_put_file_name = output_directory + \
"/" + str(time.time()) + ".jpg"
file_genertor_command = [
"ffmpeg",
"-ss",
str(ttl),
"-i",
video_file,
"-vframes",
"1",
out_put_file_name
]
# width = "90"
process = await asyncio.create_subprocess_exec(
*file_genertor_command,
# stdout must a pipe to be accessible as process.stdout
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
# Wait for the subprocess to finish
stdout, stderr = await process.communicate()
e_response = stderr.decode().strip()
t_response = stdout.decode().strip()
if os.path.lexists(out_put_file_name):
return out_put_file_name
else:
return None
# https://github.com/Nekmo/telegram-upload/blob/master/telegram_upload/video.py#L26
async def cult_small_video(video_file, output_directory, start_time, end_time):
# https://stackoverflow.com/a/13891070/4723940
out_put_file_name = output_directory + \
"/" + str(round(time.time())) + ".mp4"
file_genertor_command = [
"ffmpeg",
"-i",
video_file,
"-ss",
start_time,
"-to",
end_time,
"-async",
"1",
"-strict",
"-2",
out_put_file_name
]
process = await asyncio.create_subprocess_exec(
*file_genertor_command,
# stdout must a pipe to be accessible as process.stdout
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
# Wait for the subprocess to finish
stdout, stderr = await process.communicate()
e_response = stderr.decode().strip()
t_response = stdout.decode().strip()
if os.path.lexists(out_put_file_name):
return out_put_file_name
else:
return None
async def generate_screen_shots(
video_file,
output_directory,
is_watermarkable,
wf,
min_duration,
no_of_photos
):
metadata = extractMetadata(createParser(video_file))
duration = 0
if metadata is not None:
if metadata.has("duration"):
duration = metadata.get('duration').seconds
if duration > min_duration:
images = []
ttl_step = duration // no_of_photos
current_ttl = ttl_step
for looper in range(0, no_of_photos):
ss_img = await take_screen_shot(video_file, output_directory, current_ttl)
current_ttl = current_ttl + ttl_step
if is_watermarkable:
ss_img = await place_water_mark(ss_img, output_directory + "/" + str(time.time()) + ".jpg", wf)
images.append(ss_img)
return images
else:
return None