479 lines
17 KiB
Python
479 lines
17 KiB
Python
import asyncio
|
|
import logging
|
|
from typing import Awaitable, Callable, Iterable
|
|
|
|
import aiohttp
|
|
from aioproxies import AioProxies
|
|
from curl_cffi.requests import AsyncSession
|
|
from log_setup import setup_logging
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
|
|
from monitor_utils.config import (
|
|
DevelopmentMode,
|
|
get_config,
|
|
is_production,
|
|
set_logger_levels,
|
|
)
|
|
from monitor_utils.constants import (
|
|
EXPECTED_TASK_ERRORS,
|
|
LOGGING_LEVEL,
|
|
)
|
|
from monitor_utils.database import get_collection
|
|
from monitor_utils.models import (
|
|
AvailabilityInfo,
|
|
PricingInfo,
|
|
Product,
|
|
ProductMetadata,
|
|
ProductTiming,
|
|
)
|
|
from monitor_utils.sessions import get_base_session, get_tls_session
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def init_monitor(
|
|
monitor_name: str,
|
|
development_mode: DevelopmentMode,
|
|
production_proxy_path: str = "proxies.txt",
|
|
testing_proxy_path: str = "configs/proxies.txt",
|
|
logging_level=LOGGING_LEVEL,
|
|
) -> tuple[list[str], AsyncIOMotorCollection, AioProxies]:
|
|
"""Run shared monitor startup and return categories, collection, and proxies.
|
|
|
|
Configures JSON logging under the monitor's name, applies logger levels,
|
|
loads the monitor config and Mongo collection, and loads proxies from the
|
|
production or testing path based on development_mode.
|
|
|
|
Args:
|
|
monitor_name: Monitor identifier used for logging and config lookup.
|
|
development_mode: Selects production or testing config, collection,
|
|
and proxy path.
|
|
production_proxy_path: Proxy file path used in production mode.
|
|
testing_proxy_path: Proxy file path used in testing mode.
|
|
logging_level: Level passed to setup_logging.
|
|
|
|
Returns:
|
|
Tuple of category names, Mongo collection, and loaded proxies.
|
|
"""
|
|
|
|
use_production = is_production(development_mode)
|
|
setup_logging(
|
|
name="logs/latest.log",
|
|
history_name=monitor_name,
|
|
level=logging_level,
|
|
output="json",
|
|
)
|
|
set_logger_levels()
|
|
category_names, _, collection = setup_config(
|
|
monitor_name=monitor_name, development_mode=development_mode
|
|
)
|
|
proxy_path = production_proxy_path if use_production else testing_proxy_path
|
|
proxies = AioProxies.from_file(path=proxy_path)
|
|
return category_names, collection, proxies
|
|
|
|
|
|
def setup_config(
|
|
monitor_name: str, development_mode: DevelopmentMode
|
|
) -> tuple[list[str], dict[int], AsyncIOMotorCollection]:
|
|
"""Load monitor config and resolve the Mongo collection for the mode.
|
|
|
|
Testing mode appends a _testing suffix to the collection name so
|
|
production data is never touched during development.
|
|
|
|
Args:
|
|
monitor_name: Monitor identifier used for config and collection lookup.
|
|
development_mode: Selects the production or testing config path and
|
|
collection.
|
|
|
|
Returns:
|
|
Tuple of category names, channel id mapping, and Mongo collection.
|
|
"""
|
|
|
|
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 build_product_details(
|
|
product: dict,
|
|
min_percentage: int,
|
|
min_price_drop: float | None = None,
|
|
) -> Product | None:
|
|
"""Build a Product from a Mongo document if it passes discount thresholds.
|
|
|
|
Skips documents with a missing or zero original price. Passing requires
|
|
meeting min_percentage, or min_price_drop when provided. Channel routing
|
|
is owned by the consumer bot; no channel id is attached here.
|
|
|
|
Args:
|
|
product: Mongo document for the product.
|
|
min_percentage: Minimum discount percentage to pass.
|
|
min_price_drop: Optional minimum absolute price drop to pass.
|
|
|
|
Returns:
|
|
The built Product, or None when skipped or below thresholds.
|
|
"""
|
|
|
|
product_id = product["_id"]
|
|
current_price = product["currentPrice"]
|
|
original_price = product["originalPrice"]
|
|
if not original_price:
|
|
LOGGER.warning(
|
|
"missing or zero original price, skipping",
|
|
extra={"product_id": product_id},
|
|
)
|
|
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"],
|
|
),
|
|
)
|
|
return built
|
|
|
|
|
|
async def execute_monitor_logic(
|
|
category_identifiers: list[str] | dict[str, str],
|
|
task_func: Callable[..., Awaitable[None]],
|
|
*,
|
|
proxies: AioProxies,
|
|
headers: dict | None = None,
|
|
cookies: dict | None = None,
|
|
pre_task_func: Callable[..., Awaitable[None]] | None = None,
|
|
use_tls: bool = False,
|
|
**kwargs,
|
|
):
|
|
"""Run the monitor's category tasks in a loop with a fresh session per cycle.
|
|
|
|
Each cycle opens a TLS or base session, optionally runs pre_task_func,
|
|
dispatches task_func per category concurrently, logs expected failures,
|
|
raises unexpected ones, then sleeps briefly before the next cycle.
|
|
|
|
Args:
|
|
category_identifiers: List of category ids, or a mapping of category
|
|
id to query id; mapping entries pass query_category_id to
|
|
task_func.
|
|
task_func: Awaitable invoked per category with the identifier,
|
|
session, and **kwargs.
|
|
proxies: Proxies applied to the session.
|
|
headers: Headers applied to the session.
|
|
cookies: Cookies applied to the base session; ignored for TLS.
|
|
pre_task_func: Optional awaitable run once per cycle before category
|
|
tasks, receiving session and **kwargs.
|
|
use_tls: Use the TLS fingerprint session instead of the base session.
|
|
**kwargs: Forwarded verbatim to pre_task_func and every task_func call.
|
|
"""
|
|
|
|
while True:
|
|
session_ctx = (
|
|
get_tls_session(
|
|
headers=headers,
|
|
proxies=proxies,
|
|
)
|
|
if use_tls
|
|
else get_base_session(headers=headers, cookies=cookies, proxies=proxies)
|
|
)
|
|
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 exc:
|
|
LOGGER.warning("failed to get products", exc_info=exc)
|
|
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(
|
|
"failed to get products",
|
|
extra={"category": category_identifier},
|
|
exc_info=result,
|
|
)
|
|
elif isinstance(result, Exception):
|
|
LOGGER.error(
|
|
"unexpected error",
|
|
extra={"category": category_identifier},
|
|
exc_info=result,
|
|
)
|
|
raise result
|
|
|
|
LOGGER.info("sleeping before next tasks", extra={"delay": 0.30})
|
|
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,
|
|
):
|
|
"""Paginate a category by dispatching concurrent page or offset tasks to func.
|
|
|
|
Runs pages 2..N under a semaphore, retries failed pages with exponential
|
|
backoff, and logs any pages that still fail after retries.
|
|
|
|
Args:
|
|
category_identifier: Label used for logging only; not passed to func.
|
|
total_products: Total product count used to derive page count.
|
|
products_per_page: Page size used to derive page count and offsets.
|
|
func: Awaitable task invoked per page; receives session plus either
|
|
offset or page_number depending on pagination_mode.
|
|
session: HTTP session forwarded to each func call.
|
|
max_concurrent_tasks: Semaphore limit for concurrent page fetches.
|
|
start_offset: Base offset added when pagination_mode is offset.
|
|
pagination_mode: Either offset or page, selecting the kwarg sent to func.
|
|
**kwargs: Forwarded verbatim to every func call. Any identifier func
|
|
needs (category_id, category_name, category_path, or bestbuy's
|
|
query_category_id) must be supplied here, since naming varies
|
|
per monitor.
|
|
"""
|
|
|
|
total_pages = (int(total_products) + products_per_page - 1) // products_per_page
|
|
LOGGER.info(
|
|
"executing pagination",
|
|
extra={"category": category_identifier, "total": total_pages},
|
|
)
|
|
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):
|
|
"""Fetch a single page or offset under the shared semaphore."""
|
|
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(
|
|
"failed to get products after retries",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"failed": failed_pages,
|
|
},
|
|
)
|
|
|
|
if stop_event.is_set():
|
|
LOGGER.info("pagination stopped early", extra={"category": category_identifier})
|
|
|
|
|
|
async def fetch_all_pages(
|
|
page_range: Iterable[int],
|
|
fetch_func: Callable[[int], Awaitable[None]],
|
|
category_identifier: str,
|
|
pagination_label: str,
|
|
) -> list[int]:
|
|
"""Dispatch all page fetches concurrently and return failed page numbers.
|
|
|
|
Args:
|
|
page_range: Page numbers to fetch.
|
|
fetch_func: Awaitable invoked per page number.
|
|
category_identifier: Label used for logging.
|
|
pagination_label: Either offset or page, used for logging.
|
|
|
|
Returns:
|
|
Page numbers whose fetch raised an expected error.
|
|
"""
|
|
|
|
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]:
|
|
"""Log page results and collect failures, raising on unexpected errors.
|
|
|
|
Expected errors are logged as warnings and their pages collected for
|
|
retry. Unexpected exceptions are logged and re-raised. Successful pages
|
|
are logged only when part of a retry attempt.
|
|
|
|
Args:
|
|
page_numbers: Page numbers matching results by position.
|
|
results: Gathered results or exceptions per page.
|
|
category_identifier: Label used for logging.
|
|
pagination_label: Either offset or page, used for logging.
|
|
retry_attempt: Retry attempt number when handling retry results.
|
|
|
|
Returns:
|
|
Page numbers whose fetch raised an expected error.
|
|
"""
|
|
|
|
failed_pages = []
|
|
for page_number, result in zip(page_numbers, results):
|
|
if isinstance(result, EXPECTED_TASK_ERRORS):
|
|
LOGGER.warning(
|
|
"failed",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"page": page_number,
|
|
"retry": retry_attempt,
|
|
},
|
|
exc_info=result,
|
|
)
|
|
failed_pages.append(page_number)
|
|
elif isinstance(result, Exception):
|
|
LOGGER.error(
|
|
"unexpected error",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"page": page_number,
|
|
"retry": retry_attempt,
|
|
},
|
|
exc_info=result,
|
|
)
|
|
raise result
|
|
elif retry_attempt:
|
|
LOGGER.info(
|
|
"retry succeeded",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"page": page_number,
|
|
"retry": retry_attempt,
|
|
},
|
|
)
|
|
|
|
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 failed pages with exponential backoff until exhausted or clear.
|
|
|
|
Sleeps 2**attempt seconds before each round and re-dispatches only the
|
|
pages that failed the previous round.
|
|
|
|
Args:
|
|
failed_pages: Page numbers to retry.
|
|
fetch_func: Awaitable invoked per page number.
|
|
category_identifier: Label used for logging.
|
|
pagination_label: Either offset or page, used for logging.
|
|
max_retries: Maximum retry rounds.
|
|
|
|
Returns:
|
|
Page numbers still failing after all retries.
|
|
"""
|
|
|
|
retry_attempt = 1
|
|
while failed_pages and retry_attempt <= max_retries:
|
|
await asyncio.sleep(2**retry_attempt)
|
|
LOGGER.info(
|
|
"retrying failed pages",
|
|
extra={
|
|
"category": category_identifier,
|
|
"label": pagination_label,
|
|
"count": len(failed_pages),
|
|
"attempt": retry_attempt,
|
|
"max_retries": 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
|