fixed a small database issue with price drop tracking and added doc strings

This commit is contained in:
tpv committed 2026-07-03 16:39:43 -07:00
1 parent a61dbdfaa9
commit 99d660ee52
6 files changed
+172 -4

No files matched your search

+18
View File
@@ -1,3 +1,5 @@
"""Config loading and mode helpers shared across monitors and the bot."""
import json import json
import logging import logging
import pathlib import pathlib
@@ -7,15 +9,19 @@ from monitor_utils.constants import DEVELOPMENT_BOT_TOKEN, LOGGERS, PRODUCTION_B
class DevelopmentMode(str, Enum): class DevelopmentMode(str, Enum):
"""Runtime mode selecting production or development behavior."""
PRODUCTION = "PRODUCTION" PRODUCTION = "PRODUCTION"
DEVELOPMENT = "DEVELOPMENT" DEVELOPMENT = "DEVELOPMENT"
def is_production(development_mode: DevelopmentMode) -> bool: def is_production(development_mode: DevelopmentMode) -> bool:
"""Return True when the mode is PRODUCTION."""
return development_mode == DevelopmentMode.PRODUCTION return development_mode == DevelopmentMode.PRODUCTION
def get_bot_token(is_production: bool) -> str: def get_bot_token(is_production: bool) -> str:
"""Return the production or development bot token for the mode."""
if is_production: if is_production:
return PRODUCTION_BOT_TOKEN return PRODUCTION_BOT_TOKEN
@@ -28,6 +34,17 @@ def get_config(
test_config_dir: str | pathlib.Path = "tests/configs", test_config_dir: str | pathlib.Path = "tests/configs",
use_production_path: bool = False, use_production_path: bool = False,
) -> tuple[dict, pathlib.Path]: ) -> tuple[dict, pathlib.Path]:
"""Load a monitor's JSON config from the production or test directory.
Args:
monitor_name: Monitor name with or without a .json suffix.
config_dir: Directory holding production configs.
test_config_dir: Directory holding test configs.
use_production_path: Load from config_dir instead of test_config_dir.
Returns:
Tuple of the parsed config dict and the resolved config path.
"""
monitor_name = monitor_name.replace(".json", "") monitor_name = monitor_name.replace(".json", "")
base = config_dir if use_production_path else test_config_dir base = config_dir if use_production_path else test_config_dir
config_path = pathlib.Path(base) / f"{monitor_name}.json" config_path = pathlib.Path(base) / f"{monitor_name}.json"
@@ -36,5 +53,6 @@ def get_config(
def set_logger_levels(loggers: list[str] = LOGGERS, level: int = logging.WARNING): def set_logger_levels(loggers: list[str] = LOGGERS, level: int = logging.WARNING):
"""Set the given loggers to the given level to quiet noisy libraries."""
for name in loggers: for name in loggers:
logging.getLogger(name).setLevel(level) logging.getLogger(name).setLevel(level)
+7
View File
@@ -1,3 +1,10 @@
"""Shared constants for monitors and the consumer bot.
Holds Mongo connection settings, bot tokens, retry and timeout tuning,
client fingerprint identifiers, webhook routing maps, product status values,
and the expected error tuple used for known-failure handling.
"""
import asyncio import asyncio
from json import JSONDecodeError from json import JSONDecodeError
+2 -2
View File
@@ -33,7 +33,7 @@ def get_mongo_client() -> AsyncIOMotorClient:
async def get_product_from_database( async def get_product_from_database(
collection: AsyncIOMotorCollection, collection: AsyncIOMotorCollection,
product_id: str, product_id: str,
) -> dict | None: ) -> dict:
return await collection.find_one({"_id": product_id}) return await collection.find_one({"_id": product_id})
@@ -41,7 +41,7 @@ async def add_products_to_database(
products: list[Product], products: list[Product],
collection: AsyncIOMotorCollection, collection: AsyncIOMotorCollection,
category_identifier: str, category_identifier: str,
) -> None: ):
if not products: if not products:
return return
+54 -1
View File
@@ -1,3 +1,5 @@
"""Dataclass models shared across monitors and the consumer bot."""
from dataclasses import field from dataclasses import field
from datetime import datetime from datetime import datetime
@@ -6,7 +8,6 @@ from pydantic.dataclasses import dataclass
@dataclass @dataclass
class ConsumerConfig: class ConsumerConfig:
# used claude to refactor the consumer building in the discord bot
"""Configuration for a single retailer's consumer. """Configuration for a single retailer's consumer.
Attributes: Attributes:
@@ -28,6 +29,15 @@ class ConsumerConfig:
@dataclass @dataclass
class PricingInfo: class PricingInfo:
"""Price fields for a product.
Attributes:
original_price: Pre-discount price.
current_price: Current listed price.
price_before_drop: Stored price prior to the last detected drop.
edu_price: Education-store price when offered.
"""
original_price: float original_price: float
current_price: float current_price: float
price_before_drop: float | None = None price_before_drop: float | None = None
@@ -36,6 +46,16 @@ class PricingInfo:
@dataclass @dataclass
class AvailabilityInfo: class AvailabilityInfo:
"""Availability and membership flags for a product.
Attributes:
available: Whether the product is purchasable.
condition: Product condition such as new or open box.
app_exclusive: Only purchasable through the retailer's app.
member_pick: Flagged as a member pick offer.
member_deal: Flagged as a member deal offer.
"""
available: bool | None = None available: bool | None = None
condition: str | None = None condition: str | None = None
app_exclusive: bool | None = None app_exclusive: bool | None = None
@@ -45,6 +65,19 @@ class AvailabilityInfo:
@dataclass @dataclass
class ProductMetadata: class ProductMetadata:
"""Presentation and routing metadata for a product.
Attributes:
image_url: Product image URL.
channel_id: Resolved Discord channel id for the ping.
customization_index: Config number when priced via customization.
member_non_member_comparison: Display string comparing member and
non-member pricing.
overstock_text: Overstock label text when present.
percentage: Computed discount percentage.
affiliate_url: Affiliate link when available.
"""
image_url: str | None = None image_url: str | None = None
channel_id: str | None = None channel_id: str | None = None
customization_index: str | None = None customization_index: str | None = None
@@ -56,6 +89,14 @@ class ProductMetadata:
@dataclass @dataclass
class ProductTiming: class ProductTiming:
"""Lifecycle timestamps for a product.
Attributes:
discovered_date: When the product was first stored.
start_date: Offer start when the retailer provides one.
end_date: Offer end when the retailer provides one.
"""
discovered_date: datetime | None = None discovered_date: datetime | None = None
start_date: datetime | None = None start_date: datetime | None = None
end_date: datetime | None = None end_date: datetime | None = None
@@ -63,6 +104,18 @@ class ProductTiming:
@dataclass @dataclass
class Product: class Product:
"""A monitored product assembled from retailer data.
Attributes:
product_id: Retailer-specific product identifier.
title: Product title.
url: Product page URL.
pricing: Price fields.
metadata: Presentation and routing metadata.
timing: Lifecycle timestamps.
availability: Availability and membership flags.
"""
product_id: str product_id: str
title: str title: str
url: str url: str
+21
View File
@@ -243,6 +243,27 @@ async def execute_pagination(
pagination_mode: str = "offset", pagination_mode: str = "offset",
**kwargs, **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.
"""
LOGGER.info("executing pagination", extra={"category": category_identifier}) LOGGER.info("executing pagination", extra={"category": category_identifier})
total_pages = (int(total_products) + products_per_page - 1) // products_per_page total_pages = (int(total_products) + products_per_page - 1) // products_per_page
semaphore = asyncio.Semaphore(max_concurrent_tasks) semaphore = asyncio.Semaphore(max_concurrent_tasks)
+70 -1
View File
@@ -1,4 +1,9 @@
# ik ua/tls doesn't match headers/header order but targeted antibots don't seem to care might fix later """Shared HTTP session helpers for all monitors.
Known gap: UA/TLS fingerprint does not always match headers/header order,
but targeted antibots do not currently seem to care.
"""
import asyncio import asyncio
import json import json
import logging import logging
@@ -42,6 +47,22 @@ async def get_base_session(
timeout: aiohttp.ClientTimeout = CLIENT_TIMEOUT, timeout: aiohttp.ClientTimeout = CLIENT_TIMEOUT,
proxies: AioProxies | None = None, proxies: AioProxies | None = None,
) -> AsyncGenerator[aiohttp.ClientSession, None]: ) -> AsyncGenerator[aiohttp.ClientSession, None]:
"""Yield an aiohttp session with a random User-Agent and attached proxy pool.
The proxy pool is stored on the session as proxy_pool for send_request to
rotate per request. The session is always closed on exit.
Args:
headers: Headers applied to the session; User-Agent is defaulted if
absent.
cookies: Cookies applied to the session.
connector: Optional TCP connector.
timeout: Client timeout applied to the session.
proxies: Proxy pool attached to the session.
Yields:
The configured aiohttp session.
"""
headers = headers if headers else {} headers = headers if headers else {}
headers.setdefault("User-Agent", UA.random) headers.setdefault("User-Agent", UA.random)
session = aiohttp.ClientSession( session = aiohttp.ClientSession(
@@ -60,6 +81,20 @@ async def get_tls_session(
timeout: int = TIMEOUT_SECONDS, timeout: int = TIMEOUT_SECONDS,
proxies: AioProxies | None = None, proxies: AioProxies | None = None,
) -> AsyncGenerator[AsyncSession, None]: ) -> AsyncGenerator[AsyncSession, None]:
"""Yield a curl_cffi session impersonating a random client fingerprint.
The proxy pool is stored on the session as proxy_pool for send_request to
rotate per request.
Args:
headers: Headers applied to the session; a Chrome User-Agent is
defaulted if absent.
timeout: Request timeout in seconds.
proxies: Proxy pool attached to the session.
Yields:
The configured curl_cffi session.
"""
headers = headers if headers else {} headers = headers if headers else {}
headers.setdefault("User-Agent", UA.chrome) headers.setdefault("User-Agent", UA.chrome)
async with AsyncSession( async with AsyncSession(
@@ -74,6 +109,12 @@ async def get_tls_session(
def build_session_details( def build_session_details(
session: aiohttp.ClientSession | AsyncSession, headers: dict = None session: aiohttp.ClientSession | AsyncSession, headers: dict = None
): ):
"""Merge headers into an existing session, defaulting a random User-Agent.
Args:
session: Session whose headers are updated in place.
headers: Headers to merge; User-Agent is defaulted if absent.
"""
headers = headers or {} headers = headers or {}
headers.setdefault("User-Agent", UA.random) headers.setdefault("User-Agent", UA.random)
session.headers.update(headers) session.headers.update(headers)
@@ -105,6 +146,34 @@ async def send_request(
reraise_json_error: bool = False, reraise_json_error: bool = False,
development_proxy: str | None = None, development_proxy: str | None = None,
) -> dict | str: ) -> dict | str:
"""Send a request through either session type with retries and proxy rotation.
Retries transport errors and timeouts up to RETRY_LIMIT with a fixed wait.
When development_proxy is set it overrides the session's proxy pool and
disables TLS verification for local traffic inspection; otherwise the next
proxy is drawn from the session's proxy_pool when present. Responses are
parsed as JSON when possible.
Args:
session: aiohttp or curl_cffi session to send through.
url: Request URL.
method: HTTP method.
params: Query parameters.
json_data: JSON body.
data: Form body.
reraise_json_error: Raise JSONDecodeError instead of returning raw
text when the response is not valid JSON.
development_proxy: Local proxy URL that overrides the pool and
disables verification.
Returns:
Parsed JSON dict, or raw response text when parsing fails and
reraise_json_error is False.
Raises:
json.JSONDecodeError: When parsing fails and reraise_json_error is
True.
"""
verify = True verify = True
if development_proxy: if development_proxy:
proxy = development_proxy proxy = development_proxy