tumblr-cli/tcli/__main__.py

import argparse
import json
import mimetypes
import os
import sys
import textwrap
import tomllib
from pathlib import Path
from typing import Any, Sequence

import pytumblr2

from .effects import black, cyan


def xdg_config_home() -> Path:
    if xdg_path := os.environ.get("XDG_CONFIG_HOME"):
        return Path(xdg_path)
    else:
        return Path("~").expanduser() / ".config"


DEFAULT_CONFIG_PATH: Path = xdg_config_home() / "tumblrcli" / "config.toml"
type Config = dict[str, Any]


def get_tumblr(config: Config) -> pytumblr2.TumblrRestClient:
    return pytumblr2.TumblrRestClient(
        config["consumer_key"],
        config["consumer_secret"],
        config["oauth_token"],
        config["oauth_secret"],
    )


def parse_args() -> argparse.Namespace:
    class AppendOrdered(argparse.Action):
        """
        Simple Action class to append interleaved commands (image, text, etc)
        in the order that they appear.
        """

        def __call__(
            self,
            parser: argparse.ArgumentParser,
            namespace: argparse.Namespace,
            values: Any,
            option_string: str | None = None,
        ):
            seq = getattr(namespace, "sequence", None)
            if seq is None:
                seq = []
            seq.append((self.dest, values[0]))
            setattr(namespace, "sequence", seq)

    parser = argparse.ArgumentParser(
        prog="tcli",
        description="A command line interface that allows you to post to Tumblr",
    )

    parser.add_argument(
        "-c",
        "--config",
        default=DEFAULT_CONFIG_PATH,
        type=Path,
        help=f"The configuration file to use. Default: {DEFAULT_CONFIG_PATH}",
    )
    parser.add_argument(
        "-b",
        "--blog",
        type=str,
        default=None,
        help="The blog to post to. Overrides setting in configuration",
    )
    parser.add_argument(
        "-u",
        "--unattended",
        action="store_true",
        default=False,
        help="Make a post without confirmation",
    )
    parser.add_argument(
        "-d",
        "--dry-fire",
        action="store_true",
        default=False,
        help="Don't actually make the post",
    )

    subparsers = parser.add_subparsers(
        help="Post to make on Tumblr", required=True, dest="command"
    )

    post = subparsers.add_parser("post", help="Make or reblog a post on Tumblr")
    post.add_argument(
        "-g",
        "--tag",
        nargs=1,
        type=str,
        action="extend",
        help="Add a tag that will be added to the post without the leading #. Any hashes will be stripped before sending.",
    )
    post.add_argument(
        "-t",
        "--text",
        nargs=1,
        type=str,
        action=AppendOrdered,
        help="Add a paragraph of text to the post. Use '@file.txt' to load the text from file.txt.",
    )
    post.add_argument(
        "-i",
        "--image",
        nargs=1,
        type=argparse.FileType("rb"),
        action=AppendOrdered,
        help="Add an image from this path to the post.",
    )
    post.add_argument(
        "-r",
        "--reblog",
        metavar="ID",
        type=int,
        help="Reblog this post ID. Requires the -p/--parent option.",
    )
    post.add_argument(
        "-p",
        "--parent",
        type=str,
        help="Reblog from this blog. Required for the -r/--reblog option.",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    if not hasattr(args, "sequence"):
        setattr(args, "sequence", [])

    # Ensure that the config path exists but only if it's the default
    if not args.config.exists():
        print(
            f"ERROR: Configuration path {args.config} does not exist. Make sure to fill out the configuration before using the tool.",
            file=sys.stderr,
        )
        raise SystemExit(1)

    config = tomllib.loads(args.config.read_text())

    if args.blog:
        config["blog"] = args.blog

    if bool(args.parent) ^ bool(args.reblog):
        print(
            "ERROR: -r/--reblog and -p/--parent arguments are both required for reblogs.",
            file=sys.stderr,
        )
        raise SystemExit(1)

    match args.command:
        case "post":
            do_post(
                config,
                args.sequence,
                args.tag or [],
                args.reblog,
                args.parent,
                args.unattended,
                args.dry_fire,
            )


def do_post(
    config: Config,
    sequence: Sequence[tuple[str, Any]],
    tags: Sequence[str],
    reblog: int | None,
    reblog_parent: str | None,
    unattended: bool,
    dry_fire: bool,
) -> None:
    tags = [tag.replace("#", "") for tag in tags if tag.replace("#", "")]
    content: list[dict[str, Any]] = []
    media_sources: dict[str, str] = {}
    media_id = 0

    for kind, value in sequence:
        match kind:
            case "text":
                if value.startswith("@"):
                    filepath = Path(value[1:])
                    if not filepath.is_file():
                        print(
                            f"ERROR: {filepath} does not exist, exiting",
                            file=sys.stderr,
                        )
                        raise SystemExit(1)
                    value = filepath.read_text()
                for paragraph in value.split("\n\n"):
                    content += [{"type": "text", "text": paragraph.strip()}]
            case "image":
                filetype, _ = mimetypes.guess_type(value.name)
                identifier = f"media-{media_id}"
                content += [
                    {
                        "type": "image",
                        "media": {"type": filetype, "identifier": identifier},
                    }
                ]
                media_sources[identifier] = value.name
                media_id += 1

    if not content and not reblog:
        print("ERROR: no content supplied for this post.", file=sys.stderr)
        raise SystemExit(1)

    if not unattended:
        print(
            black("You are about to post this to Tumblr blog"),
            cyan(config["blog"]) + ":",
        )
        print()
        parts: list[str] = []
        for block in content:
            match block["type"]:
                case "text":
                    for chunk in block["text"].split("\n\n"):
                        parts += [
                            "\n".join(
                                black(" > ") + line for line in textwrap.wrap(chunk)
                            )
                        ]
                case "image":
                    filename = media_sources[block["media"]["identifier"]]
                    filetype = block["media"]["type"]
                    parts += [black(f" > image ({filetype} {filename})")]
        lines = f"\n{black(' > ')}\n".join(parts)
        print(lines)
        if tags:
            print()
            print("Tags: ", ", ".join([black(f"#{tag}") for tag in tags]))
        print()
        ask = True and not dry_fire
        while ask:
            try:
                response = input("Does this look okay? [y/n]: ")
            except KeyboardInterrupt:
                raise SystemExit(1)
            if response.lower() in ("y", "yes"):
                ask = False
            elif response.lower() in ("n", "no"):
                print("Confirmation failed, not posting")
                raise SystemExit(1)

    # end unattended

    tumblr = get_tumblr(config)

    if dry_fire:
        print("Dry firing, not sending text post.")
    else:
        if reblog:
            response = tumblr.reblog_post(
                config["blog"],
                parent_blogname=reblog_parent,
                id=reblog,
                content=content,
                tags=tags,
                media_sources=media_sources,
            )
        else:
            response = tumblr.create_post(
                config["blog"],
                content=content,
                tags=tags,
                media_sources=media_sources,
            )
        print("Got response:")
        print(json.dumps(response, indent=4))


try:
    main()
except (SystemExit, KeyboardInterrupt):
    print("Exiting")
    raise