webshop/app/db.py

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.git import git_add_commit
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)

        # update git repo
        commit_msg = "Update product inventory"
        git_add_commit(settings.git_repo, filepath, commit_msg)


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 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_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 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 <filepath> with the customer's order

    p = Path(f"{git_repo}/{filepath}")

    with open(p, "a") as f:
        f.write(order["data"])

    git_add_commit(git_repo, filepath, order["commit_msg"])


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
    )

    if status_res:
        commit_msg = "Update order status"
        git_add_commit(git_repo, local_db_filepath, commit_msg)

    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/<order-id>.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)

    # -- git add commit newly created file
    commit_msg = f"Export order {order.order_id} to .xlsx format"
    git_add_commit(git_repo, filepath, commit_msg)