317 lines
11 KiB
Python
317 lines
11 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Awaitable, Callable, Iterable
|
|
|
|
import aiohttp
|
|
from curl_cffi.requests import AsyncSession
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
|
|
from monitor_utils.constants import (
|
|
EXPECTED_TASK_ERRORS,
|
|
FLAG_WEBHOOK_MAP,
|
|
BUCKET_WEBHOOK_MAP,
|
|
LOGGING_LEVEL,
|
|
)
|
|
from monitor_utils.config import get_config, is_production, DevelopmentMode
|
|
from monitor_utils.models import (
|
|
Product,
|
|
ProductMetadata,
|
|
ProductTiming,
|
|
PricingInfo,
|
|
AvailabilityInfo,
|
|
)
|
|
from monitor_utils.sessions import get_base_session, get_tls_session
|
|
from monitor_utils.database import get_collection
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def setup_config(
|
|
monitor_name: str, development_mode: DevelopmentMode
|
|
) -> tuple[list[str], dict, AsyncIOMotorCollection]:
|
|
use_production = is_production(development_mode)
|
|
config, _ = get_config(
|
|
monitor_name=monitor_name,
|
|
use_production_path=use_production,
|
|
)
|
|
|
|
category_names = config["categories"]
|
|
collection = get_collection(
|
|
monitor_name if use_production else f"{monitor_name}_testing"
|
|
)
|
|
return category_names, config["channels"], collection
|
|
|
|
|
|
def get_channel_id(
|
|
product: Product, channel_ids: dict, use_price_drop_bucket: bool = False
|
|
) -> str:
|
|
# KeyError is not a issue here bc the percentage filtering per monitor handles it
|
|
if product.availability.app_exclusive:
|
|
return get_single_channel_id(channel_ids["appExclusive"])
|
|
|
|
if product.availability.available is False:
|
|
return get_single_channel_id(channel_ids["soldOut"])
|
|
|
|
if product.timing.end_date:
|
|
now = datetime.now(timezone.utc)
|
|
if (product.timing.end_date - now) <= timedelta(days=1):
|
|
return get_single_channel_id(channel_ids["event"])
|
|
|
|
for attr, key in FLAG_WEBHOOK_MAP:
|
|
if getattr(product.availability, attr, False):
|
|
return get_single_channel_id(channel_ids[key])
|
|
|
|
percentage = product.metadata.percentage or 0
|
|
percentage_bucket = min(90, (percentage // 10) * 10)
|
|
if use_price_drop_bucket:
|
|
return get_single_channel_id(channel_ids[BUCKET_WEBHOOK_MAP["price_drop"]])
|
|
|
|
return get_single_channel_id(channel_ids[BUCKET_WEBHOOK_MAP[percentage_bucket]])
|
|
|
|
|
|
def get_single_channel_id(value: str | list[str]) -> str:
|
|
if isinstance(value, list):
|
|
return value[0]
|
|
|
|
return value
|
|
|
|
|
|
def build_product_details(
|
|
product: dict,
|
|
channel_ids: dict,
|
|
min_percentage: int,
|
|
min_price_drop: float | None = None,
|
|
) -> Product | None:
|
|
product_id = product["_id"]
|
|
current_price = product["currentPrice"]
|
|
original_price = product["originalPrice"]
|
|
if not original_price:
|
|
LOGGER.warning(f"{product_id} - Missing or zero original price, skipping")
|
|
return
|
|
|
|
percentage = int(((original_price - current_price) / original_price) * 100)
|
|
price_drop = original_price - current_price
|
|
passes_percentage = percentage >= min_percentage
|
|
passes_price_drop = min_price_drop is not None and price_drop >= min_price_drop
|
|
if not passes_percentage and not passes_price_drop:
|
|
return
|
|
|
|
built = Product(
|
|
product_id=product_id,
|
|
title=product["title"],
|
|
url=product["url"],
|
|
metadata=ProductMetadata(
|
|
image_url=product["imageUrl"],
|
|
customization_index=product["customizationIndex"],
|
|
member_non_member_comparison=product["memberNonMemberComparison"],
|
|
overstock_text=product["overstockText"],
|
|
percentage=percentage,
|
|
affiliate_url=product.get("affiliateUrl"),
|
|
),
|
|
timing=ProductTiming(
|
|
discovered_date=product["discoveredDate"],
|
|
start_date=product["startDate"],
|
|
end_date=product["endDate"],
|
|
),
|
|
pricing=PricingInfo(
|
|
current_price=current_price,
|
|
original_price=original_price,
|
|
price_before_drop=product["priceBeforeDrop"],
|
|
edu_price=product["eduPrice"],
|
|
),
|
|
availability=AvailabilityInfo(
|
|
available=product["available"],
|
|
condition=product["condition"],
|
|
app_exclusive=product["appExclusive"],
|
|
member_pick=product["memberPick"],
|
|
member_deal=product["memberDeal"],
|
|
),
|
|
)
|
|
built.metadata.channel_id = get_channel_id(
|
|
built, channel_ids, passes_price_drop and not passes_percentage
|
|
)
|
|
return built
|
|
|
|
|
|
async def execute_monitor_logic(
|
|
category_identifiers: list[str] | dict[str, str],
|
|
task_func: Callable[..., Awaitable[None]],
|
|
headers: dict | None = None,
|
|
cookies: dict | None = None,
|
|
pre_task_func: Callable[..., Awaitable[None]] | None = None,
|
|
use_tls: bool = False,
|
|
**kwargs,
|
|
):
|
|
while True:
|
|
session_ctx = (
|
|
get_tls_session(headers=headers)
|
|
if use_tls
|
|
else get_base_session(headers=headers, cookies=cookies)
|
|
)
|
|
async with session_ctx as session:
|
|
LOGGER.info("Starting tasks...")
|
|
if pre_task_func:
|
|
try:
|
|
await pre_task_func(session=session, **kwargs)
|
|
except EXPECTED_TASK_ERRORS as error:
|
|
LOGGER.warning("Failed to get products", exc_info=error)
|
|
except Exception:
|
|
LOGGER.exception("Unexpected error")
|
|
raise
|
|
|
|
if isinstance(category_identifiers, dict):
|
|
tasks = [
|
|
task_func(category_id, session, query_category_id=query, **kwargs)
|
|
for category_id, query in category_identifiers.items()
|
|
]
|
|
category_keys = list(category_identifiers.keys())
|
|
else:
|
|
tasks = [
|
|
task_func(category_identifier, session, **kwargs)
|
|
for category_identifier in category_identifiers
|
|
]
|
|
category_keys = category_identifiers
|
|
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
for category_identifier, result in zip(category_keys, results):
|
|
if isinstance(result, EXPECTED_TASK_ERRORS):
|
|
LOGGER.warning(
|
|
f"{category_identifier} - Failed to get products",
|
|
exc_info=result,
|
|
)
|
|
elif isinstance(result, Exception):
|
|
LOGGER.error(
|
|
f"{category_identifier} - Unexpected error",
|
|
exc_info=result,
|
|
)
|
|
raise result
|
|
|
|
LOGGER.info(
|
|
"Sleeping 0.30 seconds before next tasks"
|
|
) # Might add option to increase/decrease this delay
|
|
await asyncio.sleep(0.30)
|
|
|
|
|
|
async def execute_pagination(
|
|
category_identifier: str,
|
|
total_products: int,
|
|
products_per_page: int,
|
|
func: Callable[..., Awaitable[None]],
|
|
session: aiohttp.ClientSession | AsyncSession,
|
|
max_concurrent_tasks: int = 5,
|
|
start_offset: int = 24,
|
|
pagination_mode: str = "offset",
|
|
**kwargs,
|
|
):
|
|
LOGGER.info(f"{category_identifier} - executing pagination")
|
|
total_pages = (int(total_products) + products_per_page - 1) // products_per_page
|
|
semaphore = asyncio.Semaphore(max_concurrent_tasks)
|
|
stop_event = asyncio.Event()
|
|
pagination_label = "offset" if pagination_mode == "offset" else "page"
|
|
page_range = range(2, total_pages + 1)
|
|
|
|
async def fetch_page(page_index: int):
|
|
async with semaphore:
|
|
if stop_event.is_set():
|
|
return
|
|
|
|
if pagination_mode == "offset":
|
|
current_offset = start_offset + page_index * products_per_page
|
|
await func(
|
|
session=session,
|
|
offset=current_offset,
|
|
**kwargs,
|
|
)
|
|
else:
|
|
await func(
|
|
session=session,
|
|
page_number=page_index,
|
|
**kwargs,
|
|
)
|
|
|
|
failed_pages = await fetch_all_pages(
|
|
page_range, fetch_page, category_identifier, pagination_label
|
|
)
|
|
if failed_pages and not stop_event.is_set():
|
|
failed_pages = await retry_failed_pages(
|
|
failed_pages, fetch_page, category_identifier, pagination_label
|
|
)
|
|
|
|
if failed_pages and not stop_event.is_set():
|
|
LOGGER.error(
|
|
f"{category_identifier} - failed to get products after retries: "
|
|
f"{pagination_label}s {failed_pages}"
|
|
)
|
|
|
|
if stop_event.is_set():
|
|
LOGGER.info(f"{category_identifier} - pagination stopped early")
|
|
|
|
|
|
async def fetch_all_pages(
|
|
page_range: Iterable[int],
|
|
fetch_func: Callable[[int], Awaitable[None]],
|
|
category_identifier: str,
|
|
pagination_label: str,
|
|
) -> list[int]:
|
|
tasks = [fetch_func(page) for page in page_range]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
return handle_page_results(
|
|
list(page_range), results, category_identifier, pagination_label
|
|
)
|
|
|
|
|
|
def handle_page_results(
|
|
page_numbers: list[int],
|
|
results: list,
|
|
category_identifier: str,
|
|
pagination_label: str,
|
|
retry_attempt: int | None = None,
|
|
) -> list[int]:
|
|
failed_pages = []
|
|
retry_suffix = f" Retry {retry_attempt} " if retry_attempt else " "
|
|
|
|
for page_number, result in zip(page_numbers, results):
|
|
if isinstance(result, EXPECTED_TASK_ERRORS):
|
|
LOGGER.warning(
|
|
f"{category_identifier} - {pagination_label} {page_number} -{retry_suffix} failed",
|
|
exc_info=result,
|
|
)
|
|
failed_pages.append(page_number)
|
|
elif isinstance(result, Exception):
|
|
LOGGER.error(
|
|
f"{category_identifier} - {pagination_label} {page_number} -{retry_suffix} unexpected error",
|
|
exc_info=result,
|
|
)
|
|
raise result
|
|
elif retry_attempt:
|
|
LOGGER.info(
|
|
f"{category_identifier} - {pagination_label} {page_number} - retry {retry_attempt} succeeded"
|
|
)
|
|
|
|
return failed_pages
|
|
|
|
|
|
async def retry_failed_pages(
|
|
failed_pages: list[int],
|
|
fetch_func: Callable[[int], Awaitable[None]],
|
|
category_identifier: str,
|
|
pagination_label: str,
|
|
max_retries: int = 3,
|
|
) -> list[int]:
|
|
retry_attempt = 1
|
|
while failed_pages and retry_attempt <= max_retries:
|
|
await asyncio.sleep(2**retry_attempt)
|
|
LOGGER.info(
|
|
f"{category_identifier} - retrying {len(failed_pages)} failed {pagination_label}s "
|
|
f"(attempt {retry_attempt}/{max_retries})"
|
|
)
|
|
tasks = [fetch_func(page_number) for page_number in failed_pages]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
failed_pages = handle_page_results(
|
|
failed_pages, results, category_identifier, pagination_label, retry_attempt
|
|
)
|
|
retry_attempt += 1
|
|
|
|
return failed_pages
|