99 lines
2.5 KiB
Python
Executable File
99 lines
2.5 KiB
Python
Executable File
#!/usr/bin/env python3.9
|
|
import argparse
|
|
import concurrent.futures
|
|
import dataclasses
|
|
import re
|
|
import sys
|
|
import time
|
|
import typing
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import logging
|
|
from functools import cache
|
|
|
|
|
|
import httpx
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class UploadResult:
|
|
image_url: str
|
|
thumbnail_url: str
|
|
|
|
session = httpx.Client(timeout=20)
|
|
|
|
@cache
|
|
def get_auth_token():
|
|
logging.info('extracting auth token')
|
|
res = session.get("https://jerking.empornium.ph/?agree-consent", follow_redirects=True)
|
|
try:
|
|
auth_token = re.search(r'name="auth_token" value="([^"]+)"', res.text).group(1)
|
|
except AttributeError:
|
|
auth_token = re.search(r'auth_token = "([^"]+)"', res.text).group(1)
|
|
|
|
logging.info(f'got {auth_token=}')
|
|
|
|
return auth_token
|
|
|
|
def upload_image(image_path: Path) -> UploadResult:
|
|
auth_token = get_auth_token()
|
|
|
|
logging.info(f'uploading {image_path.name}')
|
|
|
|
with image_path.open("rb") as f:
|
|
res = session.post(
|
|
"https://jerking.empornium.ph/json",
|
|
headers={"accept": "application/json"},
|
|
data={
|
|
"thumb_width": "400",
|
|
"thumb_height": "400",
|
|
"thumb_crop": "false",
|
|
"medium_width": "600",
|
|
"medium_crop": "false",
|
|
"type": "file",
|
|
"action": "upload",
|
|
"timestamp": str(round(time.time() * 1000)),
|
|
"auth_token": auth_token,
|
|
"nsfw": "1",
|
|
},
|
|
files={
|
|
"source": f,
|
|
},
|
|
)
|
|
|
|
data = res.json()
|
|
logging.info(f'got response={data}')
|
|
image_url = data["image"]["url"]
|
|
thumbnail_url = data["image"]["display_url"]
|
|
|
|
logging.info(f'uploaded {image_path.name}. url={image_url}')
|
|
|
|
return UploadResult(image_url, thumbnail_url)
|
|
|
|
|
|
def parse_args(argv: Optional[typing.Sequence[str]] = None) -> argparse.Namespace:
|
|
arger = argparse.ArgumentParser()
|
|
arger.add_argument('image_paths', nargs='+', action='store', type=Path, help='Path to image to upload')
|
|
if not argv or len(argv) == 0:
|
|
arger.print_help()
|
|
sys.exit(1)
|
|
|
|
return arger.parse_args(argv)
|
|
|
|
|
|
def main():
|
|
args = parse_args(sys.argv[1:])
|
|
image_paths = args.image_paths
|
|
|
|
get_auth_token()
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(pool.map(upload_image, image_paths))
|
|
for it in results:
|
|
print(it.image_url)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
logging.basicConfig(level=logging.INFO)
|
|
main()
|