epicenter/get_orders.py

import time
import requests
import json
import os

from dotenv import load_dotenv
from datetime import datetime
from telegram_bot import order_epicenter_error
from models import db, EpicenterOrder, Product, EpicenterOrderProduct


load_dotenv()

LOGIN = os.getenv("LOGIN")
PASSWORD = os.getenv("EPIC_PASSWORD")


def get_auth_token(login: str, password: str) -> str:

    headers = {
        "authority": "core-api.epicentrm.cloud",
        "accept": "application/json",
        "accept-language": "ru",
        "content-type": "application/json",
        "origin": "https://admin.epicentrm.com.ua",
        "referer": "https://admin.epicentrm.com.ua/",
        "sec-ch-ua": '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Linux"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    }

    json_data = {
        "login": login,
        "password": password,
    }

    response = requests.post(
        "https://core-api.epicentrm.cloud/v1/users/login",
        headers=headers,
        json=json_data,
    )

    auth_token = response.json()["token"]["auth"]

    return auth_token


def is_new_order(order_id: str, auth_token: str) -> bool:
    headers = {
        "authority": "core-api.epicentrm.cloud",
        "accept": "application/json",
        "accept-language": "en-US,en;q=0.9,ru;q=0.8,ru-RU;q=0.7",
        "authorization": f"Bearer {auth_token}",
        "origin": "https://admin.epicentrm.com.ua",
        "referer": "https://admin.epicentrm.com.ua/",
        "sec-ch-ua": '"Not A(Brand";v="99", "Google Chrome";v="121", "Chromium";v="121"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    }

    response = requests.get(
        f"https://core-api.epicentrm.cloud/v2/oms/orders/{order_id}",
        headers=headers,
    )

    data = json.loads(response.text)

    if (
        data["statusCode"] == "new"
        and data["statusCode"] != "canceled"
        and data["statusCode"] != "sent"
    ) and data["statusCode"] != "confirmed":
        return True
    return False


def activate_order(order_id: str, auth_token: str):

    first_headers = {
        "authority": "core-api.epicentrm.cloud",
        "accept": "*/*",
        "accept-language": "en-US,en;q=0.9,ru;q=0.8,ru-RU;q=0.7",
        "access-control-request-headers": "authorization,content-type",
        "access-control-request-method": "POST",
        "origin": "https://admin.epicentrm.com.ua",
        "referer": "https://admin.epicentrm.com.ua/",
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    }

    requests.options(
        f"https://core-api.epicentrm.cloud/v2/oms/orders/{order_id}/change-status/to/confirmed_by_merchant",
        headers=first_headers,
    )

    second_headers = {
        "authority": "core-api.epicentrm.cloud",
        "accept": "*/*",
        "accept-language": "en-US,en;q=0.9,ru;q=0.8,ru-RU;q=0.7",
        "authorization": f"Bearer {auth_token}",
        # 'content-length': '0',
        "content-type": "application/json",
        "origin": "https://admin.epicentrm.com.ua",
        "referer": "https://admin.epicentrm.com.ua/",
        "sec-ch-ua": '"Not A(Brand";v="99", "Google Chrome";v="121", "Chromium";v="121"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    }

    requests.post(
        f"https://core-api.epicentrm.cloud/v2/oms/orders/{order_id}"
        f"/change-status/to/confirmed_by_merchant",
        headers=second_headers,
    )


def get_delivery_info(
    shipment_settlementId: str, shipment_officeId: str, auth_token: str
):

    headers = {
        "authority": "core-api.epicentrm.cloud",
        "accept": "application/json",
        "accept-language": "ru",
        "authorization": f"Bearer " f"{auth_token}",
        "if-none-match": 'W/"196-wqlAXM70iFf1sVnvW/hGnongIL0"',
        "origin": "https://admin.epicentrm.com.ua",
        "referer": "https://admin.epicentrm.com.ua/",
        "sec-ch-ua": '"Not A(Brand";v="99", "Google Chrome";v="121", "Chromium";v="121"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    }

    response = requests.get(
        f"https://core-api.epicentrm.cloud/v2/deliveries/nova_poshta/settlements/{shipment_settlementId}"
        f"/offices/{shipment_officeId}",
        headers=headers,
    )

    shipment_data = json.loads(response.text)

    return shipment_data["title"]


def get_first_page_order_ids(auth_token: str):
    headers = {
        "authority": "core-api.epicentrm.cloud",
        "accept": "application/json",
        "accept-language": "ru",
        "authorization": f"Bearer {auth_token}",
        "if-none-match": 'W/"6e25-Lp/88DlhwX6wEIzxPDq/6WiYQf8"',
        "origin": "https://admin.epicentrm.com.ua",
        "referer": "https://admin.epicentrm.com.ua/",
        "sec-ch-ua": '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Linux"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    }

    params = {
        "offset": "0",
        "limit": "10",
    }

    response = requests.get(
        "https://core-api.epicentrm.cloud/v2/oms/orders", params=params, headers=headers
    )

    data = json.loads(response.text)

    order_ids = [order["id"] for order in data["items"]]

    return order_ids


def get_order_info(order_id: str, auth_token: str):
    headers = {
        "authority": "core-api.epicentrm.cloud",
        "accept": "application/json",
        "accept-language": "en-US,en;q=0.9,ru;q=0.8,ru-RU;q=0.7",
        "authorization": f"Bearer {auth_token}",
        "if-none-match": 'W/"4ced-XZFKegDeJ8gOKv2D+e5bORHnOoM"',
        "origin": "https://admin.epicentrm.com.ua",
        "referer": "https://admin.epicentrm.com.ua/",
        "sec-ch-ua": '"Not A(Brand";v="99", "Google Chrome";v="121", "Chromium";v="121"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    }

    response = requests.get(
        f"https://core-api.epicentrm.cloud/v2/oms/orders/{order_id}",
        headers=headers,
    )

    order = json.loads(response.text)
    products = []

    if is_new_order(order_id, auth_token):
        activate_order(order_id, auth_token)
        time.sleep(10)
        return get_order_info(order_id, auth_token)
    else:
        # get product info
        for item in range(len(order["items"])):
            title = order["items"][item]["title"]
            sku = order["items"][item]["sku"]
            product_price = order["items"][item]["price"]
            quantity = order["items"][item]["quantity"]
            products.append(
                {
                    "name": title,
                    "sku": sku,
                    "price": product_price,
                    "quantity": quantity,
                }
            )

        # User info
        first_name = order["address"]["firstName"]
        last_name = order["address"]["lastName"]
        patronymic = order["address"]["patronymic"]
        email = order["address"]["email"]
        phone_number = order["address"]["phone"]

        # Delivery info
        delivery_info = get_delivery_info(
            order["address"]["shipment"]["settlementId"],
            order["address"]["shipment"]["officeId"],
            auth_token,
        )
        delivery_ref = order["address"]["shipment"]["officeId"]
        settlement = order["settlement"]["title"]

        # Order info
        order_number = order["number"]
        order_status = order["statusCode"]
        created_at = order["createdAt"]
        subtotal = order["subtotal"]

        order_data = {
            "first_name": first_name,
            "last_name": last_name,
            "patronymic": patronymic,
            "email": email,
            "phone_number": phone_number,
            "delivery_info": delivery_info,
            "delivery_ref": delivery_ref,
            "settlement": settlement,
            "order_number": order_number,
            "order_id": order_id,
            "products": products,
            "order_status": order_status,
            "created_at": created_at,
            "subtotal": subtotal,
        }

    return order_data


def insert_order_info(order_info):
    with db.atomic():
        # Insert EpicenterOrder
        epicenter_model = EpicenterOrder.create(
            order_number=order_info["order_number"],
            order_id=order_info["order_id"],
            order_status=order_info["order_status"],
            order_creation_date=datetime.strptime(
                order_info["created_at"], "%Y-%m-%dT%H:%M:%S%z"
            ).strftime("%Y-%m-%d %H:%M:%S"),
            full_name=(
                f"{order_info['first_name']} {order_info['last_name']} "
                f"{order_info['patronymic'] if order_info['patronymic'] else ''}"
            ),
            email=order_info["email"],
            phone_number=order_info["phone_number"],
            delivery_address=", ".join(order_info["delivery_info"].split(",")[1:]),
            delivery_post_number=f"Склад {order_info['delivery_info'].split(',')[0]}".replace(
                "№", "#"
            ).rstrip(),
            delivery_ref=order_info["delivery_ref"],
            settlement=order_info["settlement"],
            amount=order_info["subtotal"],
        )

        # Insert or update products and associate them with the order
        product_records = []
        for product_data in order_info["products"]:
            sku = product_data["sku"]
            existing_product = Product.select().where(Product.sku == sku).first()
            if existing_product:
                # Update existing product
                existing_product.price = product_data["price"]
                existing_product.save()
            else:
                # Create new product
                existing_product = Product.create(
                    name=product_data["name"], sku=sku, price=product_data["price"]
                )
            product_records.append(existing_product)
            # Associate product with the order and specify quantity
            EpicenterOrderProduct.create(
                epicenter_order=epicenter_model,
                product=existing_product,
                quantity=product_data["quantity"],
            )


def write_orders(auth_token: str):

    # get order_ids from first 10 orders
    order_ids = get_first_page_order_ids(auth_token)

    for order_id in order_ids:
        print(order_id)

        existing_order = (
            EpicenterOrder.select().where(EpicenterOrder.order_id == order_id).first()
        )

        if existing_order:
            continue

        order_info = get_order_info(order_id=order_id, auth_token=auth_token)
        insert_order_info(order_info=order_info)

        db.close()
        # request.get()url Alex  send order_number

    print("DONE")


def run_all():
    try:
        while True:
            auth_token = get_auth_token(LOGIN, PASSWORD)
            write_orders(auth_token=auth_token)
            time.sleep(1800)
    except Exception as e:
        # print(e.with_traceback())
        exception_text = f"🆘An error in epicenter orders script. Please change something!🆘\n Exception: {e}"
        order_epicenter_error(exception_text)


if __name__ == "__main__":
    run_all()