Added user-friendly CLI

classes/exceptions.py

@@ -20,3 +20,39 @@ class SubmissionDuplicatesError(Exception):
         super().__init__(
             f"Found duplicates of a photo '{file_path}': {self.duplicates}"
         )
+
+
+class UserCreationError(Exception):
+    def __init__(self, code: int, data: str) -> None:
+        self.code = code
+        self.data = data
+        super().__init__(
+            f"Could not create a new user. API returned HTTP {self.code} with content: {self.data}"
+        )
+
+
+class UserCreationDuplicateError(Exception):
+    def __init__(self, username: str) -> None:
+        self.username = username
+        super().__init__(f"User '{self.username} already exists.'")
+
+
+class AlbumCreationError(Exception):
+    def __init__(self, code: int, data: str) -> None:
+        self.code = code
+        self.data = data
+        super().__init__(
+            f"Could not create a new album. API returned HTTP {self.code} with content: {self.data}"
+        )
+
+
+class AlbumCreationDuplicateError(Exception):
+    def __init__(self, name: str) -> None:
+        self.name = name
+        super().__init__(f"Album '{self.name} already exists.'")
+
+
+class AlbumCreationNameError(Exception):
+    def __init__(self, data: dict) -> None:
+        self.data = data
+        super().__init__(data["detail"])

modules/api_client.py

@@ -11,7 +11,14 @@ import aiofiles
 from aiohttp import ClientSession, FormData
 from ujson import dumps
 
-from classes.exceptions import SubmissionUploadError
+from classes.exceptions import (
+    AlbumCreationDuplicateError,
+    AlbumCreationError,
+    AlbumCreationNameError,
+    SubmissionUploadError,
+    UserCreationDuplicateError,
+    UserCreationError,
+)
 from modules.logger import logWrite
 from modules.utils import configGet
 
@@ -169,5 +176,33 @@ async def move_pic(id: str, token: Union[str, None] = None) -> bool:
         return False
 
 
+async def create_user(username: str, email: str, password: str) -> None:
+    response = await http_session.post(
+        f'{configGet("address", "posting", "api")}/users',
+        data={"user": username, "email": email, "password": password},
+    )
+    if response.status == 409:
+        raise UserCreationDuplicateError(username)
+    elif response.status != 204:
+        raise UserCreationError(response.status, await response.text(encoding="utf-8"))
+    return None
+
+
+async def create_album(name: str, title: str) -> None:
+    token = await authorize()
+    response = await http_session.post(
+        f'{configGet("address", "posting", "api")}/albums',
+        params={"name": name, "title": title},
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    if response.status == 409:
+        raise AlbumCreationDuplicateError(name)
+    elif response.status == 406:
+        raise AlbumCreationNameError(await response.json())
+    elif response.status != 200:
+        raise AlbumCreationError(response.status, await response.text(encoding="utf-8"))
+    return None
+
+
 if __name__ == "__main__":
     print(asyncio.run(authorize()))

modules/cli.py

@@ -0,0 +1,77 @@
+import asyncio
+from sys import exit
+from traceback import print_exc
+from modules.api_client import create_album, create_user, http_session
+from argparse import ArgumentParser
+
+from modules.utils import configSet
+
+parser = ArgumentParser(
+    prog="Telegram Poster",
+    description="Bot for posting some of your stuff and also receiving submissions.",
+)
+
+parser.add_argument("--create-user", action="store_true")
+parser.add_argument("--create-album", action="store_true")
+
+args = parser.parse_args()
+
+
+async def cli_create_user() -> None:
+    print(
+        "To set up Photos API connection you need to create a new user.\nIf you have email confirmation enabled in your Photos API config - you need to use a real email that will get a confirmation code afterwards.",
+        flush=True,
+    )
+    username = input("Choose username for new Photos API user: ").strip()
+    email = input(f"Choose email for user '{username}': ").strip()
+    password = input(f"Choose password for user '{username}': ").strip()
+    try:
+        result = await create_user(username, email, password)
+        # asyncio.run(create_user(username, email, password))
+        configSet("username", username, "posting", "api")
+        configSet("password", password, "posting", "api")
+        none = input(
+            "Alright. If you have email confirmation enabled - please confirm registration by using the link in your email. After that press Enter. Otherwise just press Enter."
+        )
+    except Exception as exp:
+        print(f"Could not create a user due to {exp}", flush=True)
+        print_exc()
+        exit()
+    if not args.create_album:
+        print("You're done!", flush=True)
+        await http_session.close()
+        exit()
+
+
+async def cli_create_album() -> None:
+    print(
+        "To use Photos API your user needs to have an album to store its data.\nThis wizard will help you to create a new album with its name and title.",
+        flush=True,
+    )
+    name = input("Choose a name for your album: ").strip()
+    title = input(f"Choose a title for album '{name}': ").strip()
+    try:
+        result = await create_album(name, title)
+        # asyncio.run(create_album(name, title))
+        configSet("album", name, "posting", "api")
+    except Exception as exp:
+        print(f"Could not create an album due to {exp}", flush=True)
+        print_exc()
+        exit()
+    print("You're done!", flush=True)
+    await http_session.close()
+    exit()
+
+
+if args.create_user or args.create_album:
+    loop = asyncio.get_event_loop()
+    tasks = []
+
+    if args.create_user:
+        tasks.append(loop.create_task(cli_create_user()))
+
+    if args.create_album:
+        tasks.append(loop.create_task(cli_create_album()))
+
+    loop.run_until_complete(asyncio.wait(tasks))
+    loop.close()

poster.py

@@ -1,94 +1,16 @@
-from os import sep, remove, getpid
+from os import getpid
-from shutil import move
 from sys import exit
-from argparse import ArgumentParser
 
+from modules.cli import *
 from modules.logger import logWrite
 from modules.scheduler import scheduler
-from modules.utils import configGet, jsonLoad, jsonSave, killProc, locale
+from modules.utils import configGet, killProc, locale
-
-# Args =====================================================================================================================================
-parser = ArgumentParser(
-    prog="Telegram Poster",
-    description="Bot for posting some of your stuff and also receiving submissions.",
-)
-
-parser.add_argument("-m", "--move-sent", action="store_true")
-parser.add_argument("-c", "--cleanup", action="store_true")
-parser.add_argument("--confirm", action="store_true")
-parser.add_argument("-i", "--cleanup-index", action="store_true")
-parser.add_argument("-n", "--norun", action="store_true")
-
-args = parser.parse_args()
-
-if args.move_sent:
-    for entry in jsonLoad(configGet("index", "locations"))["sent"]:
-        try:
-            move(
-                configGet("queue", "locations") + sep + entry,
-                configGet("sent", "locations") + sep + entry,
-            )
-        except FileNotFoundError:
-            logWrite(
-                locale(
-                    "move_sent_doesnt_exist", "console", locale=configGet("locale")
-                ).format(entry)
-            )
-        except Exception as exp:
-            logWrite(
-                locale(
-                    "move_sent_doesnt_exception", "console", locale=configGet("locale")
-                ).format(entry, exp)
-            )
-    logWrite(locale("move_sent_completed", "console", locale=configGet("locale")))
-
-if args.cleanup:
-    if args.confirm:
-        index = jsonLoad(configGet("index", "locations"))
-        for entry in index["sent"]:
-            try:
-                try:
-                    remove(configGet("queue", "locations") + sep + entry)
-                except FileNotFoundError:
-                    pass
-                try:
-                    remove(configGet("sent", "locations") + sep + entry)
-                except FileNotFoundError:
-                    pass
-            except Exception as exp:
-                logWrite(
-                    locale(
-                        "cleanup_exception", "console", locale=configGet("locale")
-                    ).format(entry, exp)
-                )
-            jsonSave(index, jsonLoad(configGet("index", "locations")))
-        logWrite(locale("cleanup_completed", "console", locale=configGet("locale")))
-    else:
-        logWrite(locale("cleanup_unathorized", "console", locale=configGet("locale")))
-
-if args.cleanup_index:
-    if args.confirm:
-        index = jsonLoad(configGet("index", "locations"))
-        index["sent"] = []
-        jsonSave(index, jsonLoad(configGet("index", "locations")))
-        logWrite(
-            locale("cleanup_index_completed", "console", locale=configGet("locale"))
-        )
-    else:
-        logWrite(
-            locale("cleanup_index_unathorized", "console", locale=configGet("locale"))
-        )
-
-if args.norun:
-    logWrite(locale("passed_norun", "console", locale=configGet("locale_log")))
-    exit()
-# ===========================================================================================================================================
-
 
 # Import ===================================================================================================================================
 try:
-    from modules.app import app
     from pyrogram.sync import idle
+
+    from modules.app import app
 except ModuleNotFoundError:
     print(locale("deps_missing", "console", locale=configGet("locale")), flush=True)
     exit()