import logging from pathlib import Path from typing import NoReturn import arrow import pycountry import yaml from openpyxl import load_workbook from pydantic import ValidationError from app.parser import read_dir, read_file, parse_file from app.schema import ( CartOrderInfo, CartSession, CartSessionItem, DocumentCheckoutProductInfo, DocumentProduct, DocumentProductMeta, OrdersBlock, ShippingInfo, Settings, ProductShippingMeta, ) from app.serializer import convert_pydantic_product_to_text, format_order_items, format_order_shipping_info logger = logging.getLogger(__name__) def check_product_availability( product: DocumentProduct, product_id: str, quantity: int, size: str | None, style: str | None, settings, ) -> tuple[bool, bool, int]: """Check if selected product, eventually with specified option size and style, is available by checking the product's own inventory table. Returns a tuple of: - is_product_available: do we have at least 1 copy of it in the inventory, - is_product_selectable: can we show it in the web view or is it soldout, - inventory_amount. """ is_available = False is_selectable = False amount = 0 if style or size: product_variation_id = get_variation_id(product, size, style) else: product_variation_id = None if product_variation_id: for variant in product.meta.inventory: if variant.product_variation_id == product_variation_id: is_available = variant.amount > 0 is_selectable = (variant.amount - quantity) > 0 amount = variant.amount break elif len(product.meta.inventory) > 1: for item in product.meta.inventory: if item.amount > 0: is_available, is_selectable, amount = True, True, 1 else: is_available = product.meta.inventory[0].amount > 0 is_selectable = (product.meta.inventory[0].amount - quantity) > 0 amount = product.meta.inventory[0].amount return is_available, is_selectable, amount def update_product_inventory( filename: str, quantity: int, product_variation_id: str, settings ) -> NoReturn: """Read given Product document and update its inventory amount based on the given product's size and style. Afterwards write back document to disk. """ doc = read_file( settings.git_repo, filename, settings.document_match, settings.block_types, "products", ) inventory = doc.meta.inventory for model in inventory: if model.product_variation_id == product_variation_id: # run extra check to see if what would be decreased # would amount to at best 0 (eg. not going negative) if model.amount - quantity >= 0: model.amount = model.amount - quantity txt_doc = convert_pydantic_product_to_text(doc) filepath = f"products/{filename}/index.md" p = Path(f"{settings.git_repo}/{filepath}") if p.exists(): with open(p, "w") as f: f.write(txt_doc) def create_variation_id(product: DocumentProduct): """ Create a list of unique variation IDs mixing main product id with size and style for variation for specified product """ if len(product.meta.inventory) > 1: variation_ids = [] for variant in product.meta.inventory: if not variant.product_variation_id: unique_variation_id = f"{product.meta.product_id}_" if variant.size: unique_variation_id += f"_{variant.size}" if variant.style: unique_variation_id += f"_{variant.style}" variation_ids.append(unique_variation_id) else: variation_ids.append(None) return variation_ids return None # returns None in case there are no variations of the product or single variant only def get_products_id_db(settings) -> dict: """Returns collection of product_ids and paths associated with it""" product_data = {} filepath = f'{Path(settings.git_repo)}/products.yaml' p = Path(filepath) if p.exists(): with open(p, 'r') as f: product_data = yaml.safe_load(f) return product_data def write_product_id_db(product:DocumentProduct, settings, products_data = {}): """Adds product_id - path relation to the products.md in product folder""" filepath = f'{Path(settings.git_repo)}/products.yaml' product_file_path = f'{Path(settings.git_repo)}/{product.meta.path}/index.md' p = Path(filepath) products_data.setdefault("products", {}) if product.meta.product_id not in products_data["products"] or not p.exists(): products_data["products"][product.meta.product_id] = product_file_path with open(p, "w") as f: yaml.dump(products_data, f, default_flow_style = False) def get_variation_id(product:DocumentProduct, size: str | None, style: str | None) -> str | None: """Returns variation ID for product that has size or style""" if size or style: for variant in product.meta.inventory: if variant.size == size and variant.style == style: return variant.product_variation_id return None def write_variation_id(product:DocumentProduct, settings): """ Generate and write IDs for variation of the product into file""" filepath = f'{Path(settings.git_repo)}/{product.meta.path}/index.md' product_file = parse_file(settings.git_repo, Path(filepath), settings.block_types) ids = create_variation_id(product) if ids: for i in range(0, len(product.meta.inventory)): if ids[i]: product_file.meta.inventory[i].product_variation_id = ids[i] txt_doc = convert_pydantic_product_to_text(product_file) p = Path(filepath) if p.exists(): with open(p, "w") as f: f.write(txt_doc) # update git repo # commit_msg = "Update product inventory" # git_add_commit(settings.git_repo, filepath, commit_msg) def get_products_by_category(settings): """Helper function to product a list of products organized by their categories.""" products = read_dir( settings.git_repo, settings.document_match, settings.document_exclude, settings.block_types, exclude_doc="products", tree="products", ) categories = {} for p in products: categories[p.meta.category] = [] for p in products: block_img = [block for block in p.blocks if block.type == "image"][0] p.blocks = [] p.blocks.append(block_img) categories[p.meta.category].append(p) return categories def get_product_by_product_id( settings: Settings, product_id: str, products: dict, ) -> DocumentProduct: """Fetch a product by its product-id.""" if products: product = parse_file(settings.git_repo, Path(products["products"][product_id]), settings.block_types) return product else: return None def get_docs_by_category(category: str, settings): """Read a directory of directories and return a sublist with only the documents matching the given category. """ pages = read_dir( settings.git_repo, settings.document_match, settings.document_exclude, settings.block_types, exclude_doc=None, tree="pages", ) return [p for p in pages if p.meta.category == category] def is_order_id_open(settings: dict[str, str | list[str]], order_id: str) -> bool: """Check whether given Order ID exists already in orders/index.md and order status is not `Open`. Return appropriate boolean value. """ filename = "orders" orders = read_file( settings["git_repo"], filename, settings["document_match"], settings["block_types"], ) if orders: customer_order = [ order for order in orders.blocks if order.order_id == order_id and order.status != "Open" ] if len(customer_order) == 0: return True else: return False def prepare_order_info( order_meta: dict[str], payment_meta: dict[str], product_list: str, cart: CartSession, shipping_info: ShippingInfo, products_meta: list[DocumentProductMeta], ) -> str: """Prepare text to save into local-db that sums up the received order. """ # -- content order cart_items = [] cm = cart.meta ci = cart.items if len(ci.keys()) > 0: for k, v in ci.items(): for meta in products_meta: cart_product_id = meta.product_id if cart_product_id == v.product_id: item = CartOrderInfo( code=meta.code, title=meta.title, quantity=v.quantity, price=v.price * v.quantity, options=v.options, ) cart_items.append(item) items_table = format_order_items(cart_items) shipping_info_table = format_order_shipping_info(shipping_info) timestamp = arrow.now().format("YYYY-MM-DD HH:mm:ss ZZ") return ( f"---\n" f"type: 'order'\n" f"date: {timestamp}\n" f"order_id: '{order_meta['id']}'\n" f"product_list: '{product_list}'\n" f"payment_provider: '{payment_meta['provider']}'\n" f"payment_reference: '{payment_meta['id']}'\n" f"status: '{order_meta['status']}'\n" f"items: \n{items_table}" f"currency: '{cm.currency}'\n" f"subtotal: '{cm.subtotal}'\n" f"total_weight: '{cm.total_weight}'\n" f"shipping: '{cm.shipping}'\n" f"total: '{cm.total}'\n" f"shipping_info: \n{shipping_info_table}" ) def prepare_customer_order_data( git_repo: str, document_match: list[str], document_exclude: list[str], block_types: list[str], checkout_session_id: str, product_list: str, provider: str, order_id: str, cart: CartSession, shipping_info: ShippingInfo, ) -> dict[dict, str]: """Prepare customer order data and git commit message.""" products = read_dir( git_repo, document_match, document_exclude, block_types, exclude_doc="products", tree="products", ) products_meta = [product.meta for product in products] order_meta = {"id": order_id, "status": "Open"} payment_meta = {"id": checkout_session_id, "provider": provider} customer_order = prepare_order_info( order_meta, payment_meta, product_list, cart, shipping_info, products_meta ) commit_msg = "Add order" return {"data": customer_order, "commit_msg": commit_msg} def save_shipping_info(order: dict[str], git_repo: str, filepath: str) -> NoReturn: """Save customer's shipping info data to local git repo.""" # write new text file under with the customer's order p = Path(f"{git_repo}/{filepath}") with open(p, "a") as f: f.write(order["data"]) def update_order_info( order_id: str, order_status: str, git_repo: str, filepath: str ) -> bool: """Update customer's order info after the payment has been processed.""" p = Path(f"{git_repo}/{filepath}") try: with open(p, "r+") as f: order_content = f.read() orders = yaml.safe_load_all(order_content) updated_orders = [] for order in orders: if order: if "type" in order and order["type"] == "meta": meta_block = f"---\n" f"{yaml.safe_dump(order)}" updated_orders.append(meta_block) elif "order_id" in order: if order["order_id"] == order_id: order["status"] = order_status # prepare each order entry by hand as pyyaml format things # in a way we don't like (ref multiline values). timestamp = arrow.get(order["date"]).format( "YYYY-MM-DD HH:mm:ss ZZ" ) items_table = format_order_items(order['items']) shipping_info_table = format_order_shipping_info(ShippingInfo(**order['shipping_info'])) updated_order = ( f"---\n" f"type: 'order'\n" f"date: {timestamp}\n" f"order_id: '{order['order_id']}'\n" f"product_list: '{order['product_list']}'\n" f"payment_provider: '{order['payment_provider']}'\n" f"payment_reference: '{order['payment_reference']}'\n" f"status: '{order['status']}'\n" f"items: \n{items_table}" f"currency: '{order['currency']}'\n" f"subtotal: '{order['subtotal']}'\n" f"total_weight: '{order['total_weight']}'\n" f"shipping: '{order['shipping']}'\n" f"total: '{order['total']}'\n" f"shipping_info: \n{shipping_info_table}" ) updated_orders.append(updated_order) new_orders = "".join(updated_orders) f.seek(0) f.write(new_orders) f.truncate() return True except FileNotFoundError: return False def update_status_order( event_type: str, client_reference_id: str, git_repo: str, local_db_filepath: str ) -> bool: """Helper function to select order's status before updating it.""" status_index = { "checkout.session.completed": "Complete", # Stripe "checkout.session.async_payment_succeeded": "Complete", # Stripe "checkout.session.async_payment_failed": "Failed", # Stripe "checkout.session.expired": "Expired", # Stripe "finished": "Complete", # NOWPayments "expired": "Expired", # NOWPayments "failed": "Failed", # NOWPayments } order_status = status_index[event_type] status_res = update_order_info( client_reference_id, order_status, git_repo, local_db_filepath ) return status_res def read_order_info_field( field: str, order_id: str, git_repo: str, filepath: str ) -> bool: """Helper function to read the given field from the Order Info document as YAML.""" p = Path(f"{git_repo}/{filepath}") try: with open(p, "r+") as f: order_content = f.read() orders = yaml.safe_load_all(order_content) selected_order = [ order for order in orders if "order_id" in order and order["order_id"] == order_id ] if len(selected_order) > 0: selected_order = selected_order[0] if field in selected_order: return selected_order[field] else: None except FileNotFoundError: return None def pick_last_order_id(git_repo: str, filepath: str) -> str: """Helper function to pick the ID from the last received order from the Order Info document. Used for testing purposes. """ p = Path(f"{git_repo}/{filepath}") try: with open(p, "r+") as f: order_content = f.read() orders = yaml.safe_load_all(order_content) return list(orders)[-1]["order_id"] except FileNotFoundError: raise FileNotFoundError def get_product_values_from_checkout_string( product_info: str, ) -> list[str, int, str | None]: """Helper function to parse a string part of the checkout operation, which contains: - product path (eg. product slug) - product quantity - product's size and style """ cart_product_id, product_quantity = product_info.split("--") product_quantity = int(product_quantity) filename, options = cart_product_id.split("__") if options != "": size, style = options.split("-") else: size = None style = None return filename, product_quantity, size, style def fetch_order_product_list( client_reference_id: str, git_repo: str, local_db_filepath: str ) -> list[DocumentCheckoutProductInfo] | None: """Helper function to select order's status before updating it.""" order_product_info = read_order_info_field( "product_list", client_reference_id, git_repo, local_db_filepath ) if order_product_info is None: logger.warning("fetch_order_product_list error: no order-product-info found.") return None product_list_index = [] product_list = order_product_info.split(",") for product_info in product_list: filename, quantity, size, style = get_product_values_from_checkout_string( product_info ) try: product_info = DocumentCheckoutProductInfo( filename=filename, quantity=quantity, size=size, style=style ) product_list_index.append(product_info) except ValidationError as e: import json logger.error("pydantic validation errors =>") for e in e.errors(): logger.error(f"error => {json.dumps(e, indent=4)}") logger.error("fetch_order_product_list error") return product_list_index def export_order_to_xlsx( order: OrdersBlock, products_shipping_meta: ProductShippingMeta, order_template_filepath: str, currency: str, git_repo, email_from: str, ): """Export customer order to an .xlsx file, ready to be uploaded to the SwissPost website as part of the shipping workflow. """ # read from xslx template file (settings.order_upload.filepath) # update spreadsheet in memory # export it under `orders/upload/.xlsx` order_template_filepath = f"{git_repo}/{order_template_filepath}" workbook = load_workbook(filename=order_template_filepath) sheet = workbook.active # -- add order info order_country_iso = pycountry.countries.lookup(order.shipping_info.country).alpha_2 # invoice info sheet["A2"] = order.order_id sheet["B2"] = arrow.get(order.date).format("DD/MM/YY") sheet["C2"] = currency sheet["D2"] = order.shipping_info.first_name sheet["E2"] = order.shipping_info.last_name sheet["H2"] = order.shipping_info.address sheet["I2"] = order.shipping_info.address_no sheet["J2"] = order.shipping_info.postal_code sheet["K2"] = order.shipping_info.city sheet["M2"] = order_country_iso sheet["N2"] = order.shipping_info.email sheet["O2"] = order.shipping_info.phone_number # recipient info sheet["R2"] = order.shipping_info.first_name sheet["S2"] = order.shipping_info.last_name sheet["U2"] = order.shipping_info.address sheet["V2"] = order.shipping_info.address_no sheet["W2"] = order.shipping_info.postal_code sheet["X2"] = order.shipping_info.city sheet["Z2"] = order_country_iso sheet["AA2"] = order.shipping_info.email sheet["AB2"] = order.shipping_info.phone_number # add order product items for idx, item in enumerate(order.items): # we set the baseline to row 2 row_n = 2 + idx product = products_shipping_meta[idx] sheet[f"AC{row_n}"] = item.code sheet[f"AD{row_n}"] = item.title sheet[f"AE{row_n}"] = item.quantity sheet[f"AF{row_n}"] = item.price sheet[f"AG{row_n}"] = product.weight sheet[f"AH{row_n}"] = product.customs_tariff_code sheet[f"AI{row_n}"] = product.origin_country_iso # -- save file to disk timestamp = arrow.get(order.date).format("YYYY-MM-DD-HHmmss") filepath = f"{git_repo}/orders/upload/{timestamp}--{order.order_id}.xlsx" workbook.save(filename=filepath)