Compact post fix, deps bump, tiktok and yt shorts fix backport, galle…

…ry support

Pipfile

@@ -5,9 +5,9 @@ name = "pypi"
 
 [packages]
 tiktokapipy = "==0.2.4.post2"
-playwright = "==1.43.0"
+playwright = "==1.46.0"
 "discord.py" = "==2.4.0"
-instaloader = "==4.12"
+instaloader = "==4.13"
 facebook-scraper = "==0.2.59"
 python-magic = "0.4.27"
 redvid = "==2.0.5"
@@ -17,6 +17,7 @@ asyncpraw = "==7.7.1"
 twscrape = "==0.13"
 pytube2 = "==15.0.6"
 lxml-html-clean = "==0.1.1"
+RedDownloader = "==4.3.0"
 
 [dev-packages]
 black = "==24.4.2"

downloader/__init__.py

@@ -1,6 +1,12 @@
+# pylint: skip-file
+import re
+
 import pydantic_core
+from pytube import cipher
+from pytube.innertube import _default_clients
 
 
+# Monkey patch tiktok lib
 old_validate_python = pydantic_core.SchemaValidator.validate_python
 
 
@@ -15,3 +21,43 @@ def validate(*args, **kwargs):
 
 
 pydantic_core.SchemaValidator.validate_python = validate
+
+
+# Monkeypatch youtube lib
+# Age restriction bypass - https://stackoverflow.com/a/78267693/10428848
+_default_clients["ANDROID_MUSIC"] = _default_clients["ANDROID_CREATOR"]
+
+
+# https://github.com/pytube/pytube/issues/1954#issuecomment-2218287594
+def patched_get_throttling_function_name(js: str) -> str:
+    function_patterns = [
+        # https://github.com/ytdl-org/youtube-dl/issues/29326#issuecomment-865985377
+        # https://github.com/yt-dlp/yt-dlp/commit/48416bc4a8f1d5ff07d5977659cb8ece7640dcd8
+        # var Bpa = [iha];
+        # ...
+        # a.C && (b = a.get("n")) && (b = Bpa[0](b), a.set("n", b),
+        # Bpa.length || iha("")) }};
+        # In the above case, `iha` is the relevant function name
+        r'a\.[a-zA-Z]\s*&&\s*\([a-z]\s*=\s*a\.get\("n"\)\)\s*&&.*?\|\|\s*([a-z]+)',
+        r'\([a-z]\s*=\s*([a-zA-Z0-9$]+)(\[\d+\])?\([a-z]\)',
+        r'\([a-z]\s*=\s*([a-zA-Z0-9$]+)(\[\d+\])\([a-z]\)',
+    ]
+    for pattern in function_patterns:
+        regex = re.compile(pattern)
+        function_match = regex.search(js)
+        if function_match:
+            if len(function_match.groups()) == 1:
+                return function_match.group(1)
+            idx = function_match.group(2)
+            if idx:
+                idx = idx.strip("[]")
+                array = re.search(r'var {nfunc}\s*=\s*(\[.+?\]);'.format(nfunc=re.escape(function_match.group(1))), js)
+                if array:
+                    array = array.group(1).strip("[]").split(",")
+                    array = [x.strip() for x in array]
+                    return array[int(idx)]
+
+    raise Exception("get_throttling_function_name")
+
+
+cipher.get_throttling_function_name = patched_get_throttling_function_name

downloader/reddit.py

@@ -1,14 +1,19 @@
 import datetime
+import glob
 import io
 import os
+import shutil
 import typing
+import uuid
 
 import asyncpraw
 import redvid
 import requests
+from RedDownloader import RedDownloader as reddit_downloader
 from asyncpraw import exceptions as praw_exceptions
 
 import models
+import utils
 from downloader import base
 
 
@@ -39,6 +44,7 @@ class RedditClient(base.BaseClient):
     def __init__(self, url: str):
         super(RedditClient, self).__init__(url=url)
         self.client = RedditClientSingleton.get_instance()
+        self.downloader = reddit_downloader
 
     async def get_post(self) -> models.Post:
         post = models.Post(url=self.url)
@@ -79,9 +85,64 @@ async def _hydrate_post(self, post: models.Post) -> bool:
             with open(f'/tmp/{submission.id}.mp4', 'rb') as f:
                 post.buffer = io.BytesIO(f.read())
             os.remove(f'/tmp/{submission.id}.mp4')
+        elif submission.url.startswith('https://www.reddit.com/gallery/'):
+            post.buffer = self._download_and_merge_gallery(url=submission.url)
 
         return True
 
+    def _hydrate_post_no_login(self, post: models.Post) -> bool:
+        try:
+            media = self.downloader.Download(url=post.url, quality=720, destination='/tmp/', output=str(uuid.uuid4()))
+        except Exception:
+            return False
+
+        post.description = media.GetPostTitle().Get()
+        post.author = media.GetPostAuthor().Get()
+        post.spoiler = self._is_nsfw()
+
+        files = glob.glob(os.path.join(media.destination, f'{media.output}*'))
+        if not files:
+            return True
+
+        if os.path.isfile(files[0]):
+            with open(files[0], 'rb') as f:
+                post.buffer = io.BytesIO(f.read())
+        else:
+            photos = [os.path.join(files[0], photo) for photo in os.listdir(files[0])]
+            post.buffer = utils.combine_images(photos)
+            for photo in photos:
+                os.remove(photo)
+
+        for file in files:
+            if os.path.isfile(file):
+                os.remove(file)
+            else:
+                os.rmdir(file)
+
+        return True
+
+    def _download_and_merge_gallery(self, url: str) -> typing.Optional[io.BytesIO]:
+        path = f'/tmp/{uuid.uuid4()}'
+        os.mkdir(path)
+
+        self.downloader.Download(url=url, destination=path, verbose=False)
+        files = os.listdir(path)
+        if len(files) == 0:
+            return None
+        if len(files) == 1:
+            fp = f'{path}/{files[0]}'
+            if not os.path.isdir(fp):
+                with utils.temp_open(fp, 'rb') as f:
+                    return io.BytesIO(f.read())
+
+        downloaded_path = f'{path}/downloaded'
+        files = sorted(os.listdir(downloaded_path))
+
+        image_buffer = utils.combine_images([f'{downloaded_path}/{f}' for f in files])
+
+        shutil.rmtree(path)
+        return image_buffer
+
     def _is_nsfw(self) -> bool:
         content = str(requests.get(self.url).content)
         return 'nsfw":true' in content or 'isNsfw":true' in content

downloader/tiktok.py

@@ -37,7 +37,7 @@ async def get_post(self) -> models.Post:
                 )
             else:
                 buffer = await self._download(
-                    url=video.video.download_addr,
+                    url=video.video.play_addr or video.video.download_addr,
                     cookies=cookies,
                     headers=headers,
                 )

models/post.py

@@ -15,13 +15,13 @@ class Post:
     buffer: typing.Optional[io.BytesIO] = None
     spoiler: bool = False
     created: typing.Optional[datetime.datetime] = None
-    compact_post = os.environ.get('COMPACT_POST') or 'false'
+    compact_post = os.environ.get('COMPACT_POST', 'False').lower() in ('true', '1', 't')
 
     def __str__(self) -> str:
         description = self.description or '❌'
 
-        if self.compact_post:
-            return ('🔗 URL: {url}\n').format(url=self.url)
+        if self.compact_post is True:
+            return f'🔗 URL: {self.url}\n'
 
         return (
             '🔗 URL: {url}\n'

utils.py

@@ -1,12 +1,15 @@
 import asyncio
 import io
+import os
 import mimetypes
 import random
 import re
 import tempfile
 import typing
+from contextlib import contextmanager
 
 import magic
+from PIL import Image as pil_image
 
 emoji = ['😼', '😺', '😸', '😹', '😻', '🙀', '😿', '😾', '😩', '🙈', '🙉', '🙊', '😳']
 
@@ -53,3 +56,47 @@ async def resize(buffer: io.BytesIO, extension: str = 'mp4') -> io.BytesIO:
 
 def random_emoji() -> str:
     return random.choice(emoji)
+
+
+def combine_images(image_fps: typing.List[str | io.BytesIO], gap: int = 10, quality: int = 85) -> io.BytesIO:
+    images = [pil_image.open(path) for path in image_fps]
+    widths, heights = zip(*(im.size for im in images))
+
+    new_image = pil_image.new('RGBA', (sum(widths), max(heights)))
+    offset = 0
+    for image in images:
+        new_image.paste(image, (offset, 0))
+        offset += image.size[0] + gap
+
+    image_bytes = io.BytesIO()
+    new_image.save(image_bytes, format='PNG', quality=quality, optimize=True)
+    image_bytes.seek(0)
+
+    return image_bytes
+
+
+def resize_image(buffer: io.BytesIO, factor: float = 0.75) -> io.BytesIO:
+    if factor == 1.0:
+        return buffer
+
+    image = pil_image.open(buffer)
+    width, height = image.size
+
+    new_image = image.resize((int(width * factor), int(height * factor)), pil_image.Resampling.NEAREST)
+
+    image_bytes = io.BytesIO()
+    new_image.save(image_bytes, format='PNG', optimize=True)
+    image_bytes.seek(0)
+
+    return image_bytes
+
+
+@contextmanager
+def temp_open(path: str, mode: str = 'rb'):
+    f = open(path, mode)  # pylint: disable=unspecified-encoding
+
+    try:
+        yield f
+    finally:
+        f.close()
+        os.remove(path)