oxo/oxo.py

import io
import typing as t
from pathlib import Path

import httpx
import typer
from rich.console import Console

try:
    from importlib.metadata import version
except ImportError:
    from importlib_metadata import version

__version__ = version("oxo")

app = typer.Typer()

BASE_URL = typer.Option("https://0x0.st", envvar="OXO_BASE_URL")

err_console = Console(stderr=True, color_system=None)


def post_to(base_url: str, *, data: t.Iterable[t.Dict]) -> str:
    retval = []
    with httpx.Client() as client:
        for d in data:
            if isinstance(d.get("file"), io.BufferedReader):
                res = client.post(base_url, files=d)
                token = res.headers.get("x-token")
                if token:
                    err_console.print(
                        f"To remove post, curl -Ftoken={token} -Fdelete= {res.text.strip()}"
                    )
                    err_console.print(
                        f"To update expiration date, curl -Ftoken={token} -Fexpires=NUMHOURS {res.text.strip()}"
                    )
            else:
                res = client.post(base_url, data=d)
            res.raise_for_status()
            retval.append(res.text.strip())
    return " ".join(retval)


def post_files(base_url, files: t.List[Path], expires: t.Optional[int]):
    return post_to(base_url, data=({"file": f.open("rb")} for f in files))


def post_repost(base_url, urls: t.List[str]):
    return post_to(base_url, data=({"url": u.strip()} for u in urls))


def post_shorten(base_url, urls: t.List[str]):
    return post_to(base_url, data=({"shorten": u.strip()} for u in urls))


def version_callback(value: bool):
    if value:
        typer.echo(f"{__version__}")
        raise typer.Exit()


@app.callback()
def main(
    version: t.Optional[bool] = typer.Option(
        None,
        "--version",
        callback=version_callback,
        is_eager=True,
        help="Show the version and exit.",
    ),
):
    """A command line utility for 0x0.st compliant pastebins.

    To use a different 0x0 site, set `OXO_BASE_URL` in your environment,
    or specify it in the relevant subcommand.
    """


@app.command()
def files(
    files: t.List[Path] = typer.Argument(
        ..., min=1, exists=True, file_okay=True, dir_okay=False, resolve_path=True
    ),
    expires: t.Optional[int] = typer.Option(
        None, help="Expiration time, in hours or epoch milliseconds"
    ),
    base_url: str = BASE_URL,
):
    """Upload one or more files."""
    typer.echo(post_files(base_url, files, expires))


@app.command()
def repost(urls: t.List[str] = typer.Argument(..., min=1), base_url=BASE_URL):
    """Repost one or more urls."""
    typer.echo(post_repost(base_url, urls))


@app.command()
def shorten(urls: t.List[str] = typer.Argument(..., min=1), base_url=BASE_URL):
    """Shorten one or more urls."""
    if base_url == BASE_URL.default:
        typer.secho(
            f"Warning: shortening is often disabled "
            f"for {base_url}, command may fail",
            fg=typer.colors.RED,
            err=True,
        )
    typer.echo(post_shorten(base_url, urls))