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
| import base64
import logging
import os
import hashlib
import subprocess
from pathlib import Path
from uuid import uuid4
from colorama import Fore
from signalbot import Command, Context, SignalBot
logger = logging.getLogger("signalbot")
logger.setLevel(logging.INFO)
logging.basicConfig(format="%(asctime)s [%(name)s:%(levelname)s]> %(message)s")
TAG = f"[{Fore.LIGHTYELLOW_EX}SignalBot{Fore.RESET}]"
class SBot(Command):
async def handle(self, c: Context):
logger.info(f'{TAG} Received "{c.message.text}" from {c.message.source}')
if c.message.source in ["+49160000000"]:
await c.send("I'll try to download the video...")
video_name = Path(f"./videos/{uuid4()}")
if not video_name.parent.exists():
os.makedirs(video_name.parent)
self.download_video(c.message.text, video_name)
video_file = None
for path, _, files in os.walk(video_name.parent):
for file in [Path(path, file) for file in files]:
if file.stem == video_name.stem:
video_file = file
break
await c.send('Video will be converted...')
converted_video_file = Path(f"./converted/{uuid4()}.mp4")
if not converted_video_file.parent.exists():
os.makedirs(converted_video_file.parent)
if self.convert_video(video_file, converted_video_file) and converted_video_file.exists():
os.unlink(video_file)
with open(converted_video_file, "rb") as file:
video_b64 = base64.b64encode(file.read()).decode()
await c.send("Successfully downloaded video", base64_attachments=[video_b64])
os.unlink(converted_video_file)
def download_video(self, url, output_path):
try:
subprocess.run(["yt-dlp", url, "-o", output_path.absolute()])
return True
except Exception as e:
return False
def convert_video(self, input_path, output_path):
try:
subprocess.run(
[
"ffmpeg",
"-hwaccel",
"vaapi",
"-vaapi_device",
"/dev/dri/renderD128",
"-i",
f'{input_path.absolute()}',
"-vf",
"format=nv12,hwupload",
"-map",
"0:0",
"-map",
"0:1",
"-c:v",
"hevc_vaapi",
"-qp",
"32",
f'{output_path.absolute()}',
]
)
return True
except Exception as e:
return False
if __name__ == "__main__":
bot = SignalBot(
{
"signal_service": "127.0.0.1:8080",
"phone_number": "+491760000000",
}
)
bot.register(SBot())
bot.start()
|