diff --git a/poetry.lock b/poetry.lock --- a/poetry.lock +++ b/poetry.lock @@ -58,6 +58,17 @@ optional = false python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*" [[package]] +name = "commonmark" +version = "0.9.1" +description = "Python parser for the CommonMark Markdown spec" +category = "main" +optional = false +python-versions = "*" + +[package.extras] +test = ["flake8 (==3.7.8)", "hypothesis (==3.55.3)"] + +[[package]] name = "idna" version = "3.3" description = "Internationalized Domain Names in Applications (IDNA)" @@ -125,6 +136,14 @@ docs = ["Sphinx (>=4)", "furo (>=2021.7. test = ["appdirs (==1.4.4)", "pytest (>=6)", "pytest-cov (>=2.7)", "pytest-mock (>=3.6)"] [[package]] +name = "pygments" +version = "2.11.2" +description = "Pygments is a syntax highlighting package written in Python." +category = "main" +optional = false +python-versions = ">=3.5" + +[[package]] name = "requests" version = "2.27.1" description = "Python HTTP for Humans." @@ -143,6 +162,22 @@ socks = ["PySocks (>=1.5.6,!=1.5.7)", "w use_chardet_on_py3 = ["chardet (>=3.0.2,<5)"] [[package]] +name = "rich" +version = "11.2.0" +description = "Render rich text, tables, progress bars, syntax highlighting, markdown and more to the terminal" +category = "main" +optional = false +python-versions = ">=3.6.2,<4.0.0" + +[package.dependencies] +colorama = ">=0.4.0,<0.5.0" +commonmark = ">=0.9.0,<0.10.0" +pygments = ">=2.6.0,<3.0.0" + +[package.extras] +jupyter = ["ipywidgets (>=7.5.1,<8.0.0)"] + +[[package]] name = "tomli" version = "2.0.0" description = "A lil' TOML parser" @@ -174,7 +209,7 @@ socks = ["PySocks (>=1.5.6,!=1.5.7,<2.0) [metadata] lock-version = "1.1" python-versions = "^3.10" -content-hash = "aa2b2339dc49cd0a2f5dbdf90b4e946bd890b422351f4d9f58bd47eb900a0edb" +content-hash = "14f945aef88b635fa5d4cc1f329572834551d8fb4fe0c7cbbbcfa419bbec84a7" [metadata.files] black = [ @@ -218,6 +253,10 @@ colorama = [ {file = "colorama-0.4.4-py2.py3-none-any.whl", hash = "sha256:9f47eda37229f68eee03b24b9748937c7dc3868f906e8ba69fbcbdd3bc5dc3e2"}, {file = "colorama-0.4.4.tar.gz", hash = "sha256:5941b2b48a20143d2267e95b1c2a7603ce057ee39fd88e7329b0c292aa16869b"}, ] +commonmark = [ + {file = "commonmark-0.9.1-py2.py3-none-any.whl", hash = "sha256:da2f38c92590f83de410ba1a3cbceafbc74fee9def35f9251ba9a971d6d66fd9"}, + {file = "commonmark-0.9.1.tar.gz", hash = "sha256:452f9dc859be7f06631ddcb328b6919c67984aca654e5fefb3914d54691aed60"}, +] idna = [ {file = "idna-3.3-py3-none-any.whl", hash = "sha256:84d9dd047ffa80596e0f246e2eab0b391788b0503584e8945f2368256d2735ff"}, {file = "idna-3.3.tar.gz", hash = "sha256:9d643ff0a55b762d5cdb124b8eaa99c66322e2157b69160bc32796e824360e6d"}, @@ -260,10 +299,18 @@ platformdirs = [ {file = "platformdirs-2.4.1-py3-none-any.whl", hash = "sha256:1d7385c7db91728b83efd0ca99a5afb296cab9d0ed8313a45ed8ba17967ecfca"}, {file = "platformdirs-2.4.1.tar.gz", hash = "sha256:440633ddfebcc36264232365d7840a970e75e1018d15b4327d11f91909045fda"}, ] +pygments = [ + {file = "Pygments-2.11.2-py3-none-any.whl", hash = "sha256:44238f1b60a76d78fc8ca0528ee429702aae011c265fe6a8dd8b63049ae41c65"}, + {file = "Pygments-2.11.2.tar.gz", hash = "sha256:4e426f72023d88d03b2fa258de560726ce890ff3b630f88c21cbb8b2503b8c6a"}, +] requests = [ {file = "requests-2.27.1-py2.py3-none-any.whl", hash = "sha256:f22fa1e554c9ddfd16e6e41ac79759e17be9e492b3587efa038054674760e72d"}, {file = "requests-2.27.1.tar.gz", hash = "sha256:68d7c56fd5a8999887728ef304a6d12edc7be74f1cfa47714fc8b414525c9a61"}, ] +rich = [ + {file = "rich-11.2.0-py3-none-any.whl", hash = "sha256:d5f49ad91fb343efcae45a2b2df04a9755e863e50413623ab8c9e74f05aee52b"}, + {file = "rich-11.2.0.tar.gz", hash = "sha256:1a6266a5738115017bb64a66c59c717e7aa047b3ae49a011ede4abdeffc6536e"}, +] tomli = [ {file = "tomli-2.0.0-py3-none-any.whl", hash = "sha256:b5bde28da1fed24b9bd1d4d2b8cba62300bfb4ec9a6187a957e8ddb9434c5224"}, {file = "tomli-2.0.0.tar.gz", hash = "sha256:c292c34f58502a1eb2bbb9f5bbc9a5ebc37bee10ffb8c2d6bbdfa8eb13cc14e1"}, diff --git a/pyproject.toml b/pyproject.toml --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ authors = ["Dennis Fink None: + + config = json.load(configfile) # type: ignore + logging.config.dictConfig(config["logging"]) + + ctx.ensure_object(dict) + ctx.obj["request_session"] = requests.Session() + ctx.obj["request_session"].headers.update( + { + "accept": "application/json", + "GROCY-API-KEY": config["grocy"]["apikey"], + } + ) + ctx.obj["base_url"] = f"{config['grocy']['url']}/api/" + + menu = Table.grid(padding=DEFAULT_PADDING) + menu.add_column(justify="left", style="green", no_wrap=True) + menu.add_column(justify="left", style="cyan", no_wrap=True) + + for task_id, task in sorted(TASK_MAP.items(), key=itemgetter(0)): + menu.add_row(task_id, task[0]) + + while True: + click.clear() + console.print(Panel(menu, title="[green bold]Menu[/green bold]")) + choice = prompt.ask( + "Enter a number to select a task", choices=list(TASK_MAP.keys()) + ) + selected_task = TASK_MAP[choice][1] + + rerun = True + while rerun: + barcode = prompt.ask("Please scan the barcode") + barcode = prepare_barcode(barcode) + + try: + selected_task(barcode) + except ( + requests.Timeout, + requests.ConnectionError, + requests.HTTPError, + requests.TooManyRedirects, + ): + rerun = False + console.input("Continue") + else: + rerun = bool( + int_prompt.ask( + "Do the same with another product", + choices=["0", "1"], + default=0, + ) + ) + + +if __name__ == "__main__": + stockcli() diff --git a/stockcli/console.py b/stockcli/console.py new file mode 100644 --- /dev/null +++ b/stockcli/console.py @@ -0,0 +1,8 @@ +from rich.console import Console +from rich.prompt import IntPrompt, Prompt + +DEFAULT_PADDING = (0, 1, 0, 0) +console = Console() +error_console = Console(stderr=True, style="bold red") +prompt = Prompt(console=console) +int_prompt = IntPrompt(console=console) diff --git a/stockcli/stock.py b/stockcli/stock.py new file mode 100644 --- /dev/null +++ b/stockcli/stock.py @@ -0,0 +1,225 @@ +from operator import itemgetter + +from rich.panel import Panel +from rich.table import Table + +from . import utils +from .console import DEFAULT_PADDING, console, int_prompt, prompt +from .style import GreenBoldText + + +def get_info_by_barcode(barcode: str) -> None: + + product = utils.get_request(f"stock/products/by-barcode/{barcode}") + + inner_product = product["product"] + product_id = inner_product["id"] + + product_group = utils.get_request( + f"objects/product_groups/{inner_product['product_group_id']}" + ) + + grid = Table.grid(padding=DEFAULT_PADDING) + grid.add_column(justify="left", no_wrap=True) + grid.add_column(justify="right", style="cyan", no_wrap=True) + grid.add_column(justify="left", no_wrap=True) + + grid.add_row(GreenBoldText("Product ID:"), product_id) + grid.add_row(GreenBoldText("Product Name:"), inner_product["name"]) + grid.add_row(GreenBoldText("Product Group:"), product_group["name"]) + + for i, barcode_data in enumerate(product["product_barcodes"], start=1): + grid.add_row( + GreenBoldText(f"Barcode N°{i}:"), + barcode_data["barcode"], + barcode_data["note"], + ) + + purchase_quantity_unit = product["default_quantity_unit_purchase"] + stock_quantity_unit = product["quantity_unit_stock"] + + purchase_to_stock_conversion = utils.get_request( + f"objects/quantity_unit_conversions?query[]=from_qu_id={purchase_quantity_unit['id']}&query[]=to_qu_id={stock_quantity_unit['id']}" + ) + + if len(purchase_to_stock_conversion) != 0: + conversion = f"{purchase_to_stock_conversion[0]['factor']} {stock_quantity_unit['name_plural']}" + else: + conversion = "" + + grid.add_row( + GreenBoldText("Purchase Quantity Unit:"), + purchase_quantity_unit["name"], + conversion, + ) + grid.add_row( + GreenBoldText("Stock Quantity Unit:"), + stock_quantity_unit["name"], + ) + + stock_amount = product["stock_amount"] + grid.add_row(GreenBoldText("Overall Stock Amount:"), str(stock_amount)) + + if stock_amount > 0: + grid.add_row(GreenBoldText("Locations:")) + locations = utils.get_request(f"stock/products/{product_id}/locations") + + for location in sorted(locations, key=itemgetter("location_name")): + grid.add_row( + f"[green bold]-[/green bold] [magenta]{location['location_name']}[/magenta]", + location["amount"], + ) + + console.print(Panel(grid, title="[green bold]Product Info[/green bold]")) + + return + + +def transfer_by_barcode(barcode: str) -> None: + + product = utils.get_request(f"stock/products/by-barcode/{barcode}") + + inner_product = product["product"] + product_id = inner_product["id"] + + locations = utils.get_request(f"stock/products/{product_id}/locations") + all_locations = utils.get_request("objects/locations") + + grid = Table.grid(padding=DEFAULT_PADDING) + grid.add_column(justify="right", style="green", no_wrap=True) + grid.add_column(justify="left", style="cyan", no_wrap=True) + grid.add_column(justify="right", style="magenta", no_wrap=True) + + grid.add_row("Product ID:", product_id, "") + grid.add_row("Product Name:", inner_product["name"], "") + grid.add_row("Locations:", "", "") + + choices = [] + + for location in sorted(locations, key=itemgetter("location_id")): + location_id = location["location_id"] + choices.append(location_id) + grid.add_row( + f"[blue]{location_id}[/blue]", + location["location_name"], + location["amount"], + ) + + choices.append("0") + console.print( + Panel( + grid, + title="[green bold]Product Info[/green bold]", + subtitle="[blue]Transfer stock[/blue]", + ) + ) + + from_id = int_prompt.ask( + "From", + choices=choices, + default=int(choices[0]), + ) + + if not from_id: + return + + grid = Table.grid(padding=DEFAULT_PADDING) + grid.add_column(justify="right", style="green", no_wrap=True) + grid.add_column(justify="left", style="cyan", no_wrap=True) + + choices = [] + for location in sorted(all_locations, key=itemgetter("id")): + location_id = location["id"] + if int(location_id) != from_id: + choices.append(location_id) + grid.add_row(f"[blue]{location_id}[/blue]", location["name"]) + + choices.append("0") + console.print( + Panel( + grid, + title="[green bold]Locations[/green bold]", + subtitle="[blue]Transfer Stock[/blue]", + ) + ) + + to_id = int_prompt.ask( + "To", + choices=choices, + default=int(choices[0]), + ) + + if not to_id: + return + + amount = int_prompt.ask("Amount") + + if not amount: + return + + data = {"amount": amount, "location_id_from": from_id, "location_id_to": to_id} + + response = utils.post_request(f"stock/products/by-barcode/{barcode}/transfer", data) + console.print("Successfully transfered!") + return + + +def add_by_barcode(barcode: str) -> None: + + product = utils.get_request(f"stock/products/by-barcode/{barcode}") + + inner_product = product["product"] + product_id = inner_product["id"] + + grid = Table.grid(padding=DEFAULT_PADDING) + grid.add_column(justify="right", style="green", no_wrap=True) + grid.add_column(justify="left", style="cyan", no_wrap=True) + grid.add_row("Product ID:", product_id) + grid.add_row("Product Name:", inner_product["name"]) + console.print( + Panel( + grid, + title="[green bold]Product Info[/green bold]", + subtitle="[blue]Add Stock[/blue]", + ) + ) + + amount = int_prompt.ask("Amount") + data = {"amount": amount, "transaction_type": "purchase"} + + response = utils.post_request(f"stock/products/by-barcode/{barcode}/add", data) + console.print("Successfully added!") + return + + +def update_by_barcode(barcode: str) -> None: + + product = utils.get_request(f"stock/products/by-barcode/{barcode}") + + inner_product = product["product"] + product_id = inner_product["id"] + + grid = Table.grid(padding=DEFAULT_PADDING) + grid.add_column(justify="right", style="green", no_wrap=True) + grid.add_column(justify="left", style="cyan", no_wrap=True) + + grid.add_row("Product ID:", product_id) + grid.add_row("Product Name:", inner_product["name"]) + grid.add_row("Overall Stock Amount:", product["stock_amount"]) + + console.print( + Panel( + grid, + title="[green bold]Product Info[/green bold]", + subtitle="[blue]Update stock[/blue]", + ) + ) + amount = prompt.ask("Amount") + + data = { + "new_amount": amount, + } + + response = utils.post_request(f"stock/products/{product_id}/inventory", data) + console.print("Successfully updated!") + return diff --git a/stockcli/style.py b/stockcli/style.py new file mode 100644 --- /dev/null +++ b/stockcli/style.py @@ -0,0 +1,8 @@ +from functools import partial + +from rich.style import Style +from rich.text import Text + +GREEN_BOLD = Style(color="green", bold=True) + +GreenBoldText = partial(Text, style=GREEN_BOLD) diff --git a/stockcli/utils.py b/stockcli/utils.py new file mode 100644 --- /dev/null +++ b/stockcli/utils.py @@ -0,0 +1,64 @@ +import logging +import string +from operator import attrgetter +from typing import Any, Optional + +import click +import requests + +from .console import error_console + +UNALLOWED_CHARACTERS = str.maketrans(dict((c, None) for c in string.whitespace)) + + +def make_request(method: str, url_path: str, data: Optional[Any] = None) -> Any: + obj = click.get_current_context().obj + session = obj["request_session"] + base_url = obj["base_url"] + requested_url = base_url + url_path + + method_function = attrgetter(method) + + try: + if data is not None: + response = method_function(session)(requested_url, json=data) + else: + response = method_function(session)(requested_url) + response.raise_for_status() + except requests.Timeout: + logging.error(f"The connection to {requested_url} timed out!") + error_console.print("Connection timed out!") + raise + except requests.ConnectionError: + logging.error(f"Couldn't establish a connection to {requested_url}!") + error_console.print("Couldn't establish a connection!") + raise + except requests.HTTPError: + logging.error(f"{requested_url} sent back an HTTPError") + error_console.print("Got the following error:") + error_message = response.json()["error_message"] + logging.error(error_message) + error_console.print(error_message) + raise + except requests.TooManyRedirects: + logging.error(f"{requested_url} had too many redirects!") + error_console.print("Too many redirects!") + raise + else: + return response.json() + + +def get_request(url_path: str) -> Any: + return make_request("get", url_path) + + +def post_request(url_path: str, data: dict[str, Any]) -> Any: + return make_request("post", url_path, data) + + +def put_request(url_path: str, data: dict[str, Any]) -> Any: + return make_request("put", url_path, data) + + +def prepare_barcode(barcode: str) -> str: + return barcode.translate(UNALLOWED_CHARACTERS)