ptp_slowpoke.py

from tokenize import String
import requests
from requests import Session
from requests_toolbelt import MultipartEncoder
import traceback
from typing import Any, List, NoReturn, Optional, Type, NamedTuple, Dict, Iterable, Union
from os import PathLike
import colorama 
import sys
from pathlib import Path
from typing import TypeVar
import inspect

fix = TypeVar('fix', bound='VPath')
AnyPath = Union[PathLike[str], str]

class Colours:
    """Colour constants"""
    FAIL_DIM: str = colorama.Back.RED + colorama.Fore.BLACK + colorama.Style.NORMAL
    FAIL_BRIGHT: str = colorama.Back.RED + colorama.Fore.WHITE + colorama.Style.NORMAL
    WARN: str = colorama.Back.YELLOW + colorama.Fore.BLACK + colorama.Style.NORMAL
    INFO: str = colorama.Back.BLUE + colorama.Fore.WHITE + colorama.Style.BRIGHT
    RESET: str = colorama.Style.RESET_ALL
    FAILS: List[str] = [FAIL_DIM, FAIL_BRIGHT]
    
class VPath(Path):
    """Modified version of pathlib.Path"""
    # pylint: disable=no-member
    _flavour = type(Path())._flavour  # type: ignore

    def format(self, *args: Any, **kwargs: Any) -> fix:
        """
        :return:        Formatted version of `vpath`,
                        using substitutions from args and kwargs.
                        The substitutions are identified by braces ('{' and '}')
        """
        return VPath(self.to_str().format(*args, **kwargs))

    def set_track(self, track_number: int, /) -> fix:
        """
        Set the track number by replacing the substitution "{track_number}"
        by the track_number specified

        :param track_number:        Track number
        :return:                    Formatted VPath
        """
        return self.format(track_number=track_number)

    def to_str(self) -> str:
        """
        :return:        String representation of the path, suitable for
                        passing to system calls.
        """
        return str(self)
    
class SlowPicsConf(NamedTuple):
    collection_name: str = VPath(inspect.stack()[-1].filename).stem
    """
    Slowpics's collection name.\n
    Default is the name of the current script
    """

    public: bool = True
    """Make the comparison public"""

    optimise: bool = True
    """If 'true", images will be losslessly optimised"""

    nsfw: bool = False
    """If images not suitable for minors (nudity, gore, etc.)"""

    remove_after: Optional[int] = None
    """Remove after N days"""

class VPath(Path):
    """Modified version of pathlib.Path"""
    # pylint: disable=no-member
    _flavour = type(Path())._flavour  # type: ignore

    def format(self, *args: Any, **kwargs: Any) -> VPath:
        """
        :return:        Formatted version of `vpath`,
                        using substitutions from args and kwargs.
                        The substitutions are identified by braces ('{' and '}')
        """
        return VPath(self.to_str().format(*args, **kwargs))

    def set_track(self, track_number: int, /) -> VPath:
        """
        Set the track number by replacing the substitution "{track_number}"
        by the track_number specified

        :param track_number:        Track number
        :return:                    Formatted VPath
        """
        return self.format(track_number=track_number)

    def to_str(self) -> str:
        """
        :return:        String representation of the path, suitable for
                        passing to system calls.
        """
        return str(self)
    
class Status:
    __slots__ = ()

    @staticmethod
    def fail(string: str, /, *, exception: Type[BaseException] = Exception, chain_err: Optional[BaseException] = None) -> NoReturn:
        curr_split: List[str] = []

        # All that stuff is just for alternating colours lmao
        if chain_err:
            class _Exception(BaseException):
                __cause__ = chain_err

            curr = _Exception()

            for p in traceback.format_exception(None, curr, None)[:-1]:
                curr_split.extend(p.splitlines(keepends=True))

        for p in traceback.format_stack()[:-2]:
            curr_split.extend(p.splitlines(keepends=True))

        curr_split.append(f'{exception.__name__}: {string}')

        curr_split = [Colours.FAILS[i % 2] + line + Colours.RESET for i, line in enumerate(curr_split[::-1])][::-1]
        sys.exit(''.join(curr_split) + Colours.RESET)

    @staticmethod
    def warn(string: str, /) -> None:
        print(f'{Colours.WARN}{string}{Colours.RESET}')

    @staticmethod
    def info(string: str, /, **kwargs: Any) -> None:
        print(f'{Colours.INFO}{string}{Colours.RESET}', **kwargs)

    @staticmethod
    def logo() -> None:
        print("")
            
def _get_slowpics_header(content_length: str, content_type: str, sess: Session) -> Dict[str, str]:
    return {
        "Accept": "*/*",
        "Accept-Encoding": "gzip, deflate, br",
        "Accept-Language": "en-US,en;q=0.5",
        "Content-Length": content_length,
        "Content-Type": content_type,
        "Origin": "https://slow.pics/",
        "Referer": "https://slow.pics/comparison",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Site": "same-origin",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        "X-XSRF-TOKEN": sess.cookies.get_dict()["XSRF-TOKEN"]
    }
    
def _make_api_compatible(config: SlowPicsConf) -> Dict[str, str]:
    conf = {
        'collectionName': config.collection_name,
        'public': str(config.public).lower(),
        'optimizeImages': str(config.optimise).lower(),
        'hentai': str(config.nsfw).lower(),
    }
    if config.remove_after is not None:
        conf.update({'removeAfter': str(config.remove_after)})
    return conf

def upload_to_slowpics(path: VPath, 
                        clips: Dict[str, str], 
                        config: SlowPicsConf, 
                        num_links: int,
                        frames: Optional[Iterable[int]] = None                      
                    ) -> None:
    """
    Upload to slow.pics with given configuration

    :param config:              NamedTuple which contains the uploading configuration
    """
    
    frames = sorted(range(int(abs(num_links))))
    # Upload to slow.pics
    all_images = [sorted((path / name).glob('*.png')) for name in clips.keys()]
    # print(all_images)
    fields: Dict[str, Any] = {}

    for i, (name, images) in enumerate(zip(list(clips.keys()), all_images)):
        for j, (image, frame) in enumerate(zip(images, frames)):
            fields[f'comparisons[{j}].name'] = str(frame)
            fields[f'comparisons[{j}].images[{i}].name'] = name
            fields[f'comparisons[{j}].images[{i}].file'] = (image.name, image.read_bytes(), 'image/png')

    with Session() as sess:
        sess.get('https://slow.pics/api/comparison')
        # TODO: yeet this
        files = MultipartEncoder(_make_api_compatible(config) | fields)

        Status.info('Uploading images...')
        print()
        url = sess.post(
            'https://slow.pics/api/comparison', data=files.to_string(),
            headers=_get_slowpics_header(str(files.len), files.content_type, sess)
        )

    slowpics_url = f'https://slow.pics/c/{url.text}'
    Status.info(f'Slowpics url: {slowpics_url}')

    url_file = path / 'slow.pics.url'
    url_file.write_text(f'[InternetShortcut]\nURL={slowpics_url}', encoding='utf-8')
    Status.info(f'url file copied to "{url_file.resolve().to_str()}"')
    
def links() -> List[str]:
    return [
            ["https://ptpimg.me/45y772.png",
            "https://ptpimg.me/9qy70i.png",
            "https://ptpimg.me/2u23a1.png",
            "https://ptpimg.me/90om13.png",
            "https://ptpimg.me/28w2k0.png",
            "https://ptpimg.me/lr573k.png",
            "https://ptpimg.me/r780l5.png"
            ], 
            [
            "https://ptpimg.me/m70cd7.png",
            "https://ptpimg.me/17fe16.png",
            "https://ptpimg.me/yz6fe4.png",
            "https://ptpimg.me/b7ja3b.png",
            "https://ptpimg.me/q8a921.png",
            "https://ptpimg.me/0df9g3.png",
            "https://ptpimg.me/z4golv.png"
            ]
        ]

def downloader(num_clips: int, path: AnyPath = "comps"):
    path = VPath(path)
    try:
        path.mkdir(parents=True)
    except FileExistsError:
        Status.warn(f'Folder "{path}" already exists, skipping creation')
    
    headers = {
        'Accept': '*/*',
        'Accept-Language': 'en-US,en;q=0.9',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'Pragma': 'no-cache',
        'Referer': 'https://ptp.me/',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.67 Safari/537.36',
        'X-Requested-With': 'XMLHttpRequest',
    }
    
    for i in range(1, num_clips+1):
        try:
            pathf = path / f"src{i}"
            pathf.mkdir(parents=True)
        except FileExistsError:
            pass
        
        link_list = links()
        
        for comp_no, collection in enumerate(link_list):
            for j, link in zip(list(range(len(collection))), collection):
                if j+1 == i:
                    r = requests.get(link, headers=headers)
                    with open(f"{path}/src{j+1}/{comp_no+1}_{j+1}.png", "wb") as f:
                        f.write(r.content)
        
        
def main():
    link_list = links()
    num_clips = len(link_list[0])
    path = VPath("_comps")
    downloader(num_clips, path)
    clips = {f"src{key}": value for (key, value) in zip(range(1, num_clips+1), range(1, num_clips+1))}
    config = SlowPicsConf(
                    collection_name="Test",
                    public=False,
                    optimise=True,
                    nsfw=False,
                )
    num_links = len(links())
    upload_to_slowpics(path, clips, config, num_links)
    
if __name__ == "__main__":
    main()