ai_marketplace_monitor package

Submodules

ai_marketplace_monitor.ai module

class ai_marketplace_monitor.ai.AIBackend(config: AIConfig, logger: Logger | None = None)

Bases: Generic[TAIConfig]

connect() → None

evaluate(listing: Listing, item_config: TItemConfig, marketplace_config: TMarketplaceConfig) → AIResponse

classmethod get_config(**kwargs: Any) → TAIConfig

get_prompt(listing: Listing, item_config: TItemConfig, marketplace_config: TMarketplaceConfig) → str

class ai_marketplace_monitor.ai.AIConfig(name: str, enabled: bool | None = None, api_key: str | None = None, provider: str | None = None, model: str | None = None, base_url: str | None = None, max_retries: int = 10, timeout: int | None = None)

Bases: BaseConfig

api_key: str | None = None

base_url: str | None = None

handle_api_key() → None

handle_max_retries() → None

handle_provider() → None

handle_timeout() → None

max_retries: int = 10

model: str | None = None

provider: str | None = None

timeout: int | None = None

class ai_marketplace_monitor.ai.AIResponse(score: int, comment: str, name: str = '')

Bases: object

NOT_EVALUATED: ClassVar = 'Not evaluated by AI'

comment: str

property conclusion: str

classmethod from_cache(listing: Listing, item_config: TItemConfig, marketplace_config: TMarketplaceConfig, local_cache: Cache | None = None) → AIResponse | None

name: str = ''

score: int

property stars: str

property style: str

to_cache(listing: Listing, item_config: TItemConfig, marketplace_config: TMarketplaceConfig, local_cache: Cache | None = None) → None

class ai_marketplace_monitor.ai.AIServiceProvider(*values)

Bases: Enum

ANTHROPIC = 'Anthropic'

DEEPSEEK = 'DeepSeek'

OLLAMA = 'Ollama'

OPENAI = 'OpenAI'

class ai_marketplace_monitor.ai.AnthropicBackend(config: AIConfig, logger: Logger | None = None)

Bases: AIBackend

connect() → None

default_model = 'claude-sonnet-4-20250514'

evaluate(listing: Listing, item_config: TItemConfig, marketplace_config: TMarketplaceConfig) → AIResponse

classmethod get_config(**kwargs: Any) → AnthropicConfig

class ai_marketplace_monitor.ai.AnthropicConfig(name: str, enabled: bool | None = None, api_key: str | None = None, provider: str | None = None, model: str | None = None, base_url: str | None = None, max_retries: int = 10, timeout: int | None = None)

Bases: AIConfig

handle_api_key() → None

class ai_marketplace_monitor.ai.DeekSeekConfig(name: str, enabled: bool | None = None, api_key: str | None = None, provider: str | None = None, model: str | None = None, base_url: str | None = None, max_retries: int = 10, timeout: int | None = None)

Bases: OpenAIConfig

class ai_marketplace_monitor.ai.DeepSeekBackend(config: AIConfig, logger: Logger | None = None)

Bases: OpenAIBackend

base_url: str | None = 'https://api.deepseek.com'

default_model = 'deepseek-chat'

classmethod get_config(**kwargs: Any) → DeekSeekConfig

class ai_marketplace_monitor.ai.OllamaBackend(config: AIConfig, logger: Logger | None = None)

Bases: OpenAIBackend

default_model = 'deepseek-r1:14b'

classmethod get_config(**kwargs: Any) → OllamaConfig

class ai_marketplace_monitor.ai.OllamaConfig(name: str, enabled: bool | None = None, api_key: str | None = 'ollama', provider: str | None = None, model: str | None = None, base_url: str | None = None, max_retries: int = 10, timeout: int | None = None)

Bases: OpenAIConfig

api_key: str | None = 'ollama'

handle_base_url() → None

handle_model() → None

class ai_marketplace_monitor.ai.OpenAIBackend(config: AIConfig, logger: Logger | None = None)

Bases: AIBackend

base_url: str | None = None

connect() → None

default_model = 'gpt-4o'

evaluate(listing: Listing, item_config: TItemConfig, marketplace_config: TMarketplaceConfig) → AIResponse

classmethod get_config(**kwargs: Any) → OpenAIConfig

class ai_marketplace_monitor.ai.OpenAIConfig(name: str, enabled: bool | None = None, api_key: str | None = None, provider: str | None = None, model: str | None = None, base_url: str | None = None, max_retries: int = 10, timeout: int | None = None)

Bases: AIConfig

handle_api_key() → None

ai_marketplace_monitor.cli module

Console script for ai-marketplace-monitor.

ai_marketplace_monitor.cli.main(config_files: Path] | None, <typer.models.OptionInfo object at 0x7d8f0ac39b20>]=None, headless: Annotated[bool | None, <typer.models.OptionInfo object at 0x7d8f0ac39bb0>]=False, clear_cache: Annotated[str | None, <typer.models.OptionInfo object at 0x7d8f0ac39c40>]=None, verbose: Annotated[bool | None, <typer.models.OptionInfo object at 0x7d8f0ac39cd0>]=False, items: List[str] | None, <typer.models.OptionInfo object at 0x7d8f0ac39d60>]=None, for_item: Annotated[str | None, <typer.models.OptionInfo object at 0x7d8f0ac39df0>]=None, webui: Annotated[bool, <typer.models.OptionInfo object at 0x7d8f0ac39e80>]=True, webui_host: Annotated[str, <typer.models.OptionInfo object at 0x7d8f0ac39f10>]='127.0.0.1', webui_port: Annotated[int, <typer.models.OptionInfo object at 0x7d8f0ac39fa0>]=8467, webui_log_retention: Annotated[int, <typer.models.OptionInfo object at 0x7d8f0ac3a030>]=2000, version: Annotated[bool | None, <typer.models.OptionInfo object at 0x7d8f0ac3a0c0>]=None) → None

Console script for AI Marketplace Monitor.

ai_marketplace_monitor.cli.version_callback(value: bool) → None

Callback function for the –version option.

Parameters:

value (-) – The value provided for the –version option.

Raises:

• - typer.Exit – Raises an Exit exception if the –version option is provided,

• printing the Awesome CLI version and exiting the program. –

ai_marketplace_monitor.config module

class ai_marketplace_monitor.config.Config(config_files: List[pathlib.Path], logger: logging.Logger | None = None)

Bases: Generic[TAIConfig, TItemConfig, TMarketplaceConfig]

ai: Dict[str, TAIConfig]

expand_notifications(logger: Logger | None = None) → None

expand_regions() → None

get_ai_config(config: Dict[str, Any]) → None

get_item_config(config: Dict[str, Any]) → None

get_marketplace_config(config: Dict[str, Any]) → None

get_monitor_config(config: Dict[str, Any]) → None

get_notification_config(config: Dict[str, Any]) → None

get_region_config(config: Dict[str, Any]) → None

get_translator_config(config: Dict[str, Any]) → None

get_user_config(config: Dict[str, Any]) → None

item: Dict[str, TItemConfig]

marketplace: Dict[str, TMarketplaceConfig]

monitor: MonitorConfig

notification: Dict[str, NotificationConfig]

region: Dict[str, RegionConfig]

translator: Dict[str, Translator]

user: Dict[str, UserConfig]

validate_ais() → None

validate_items() → None

validate_sections(config: Dict[str, Any]) → None

validate_users() → None

Check if notified users exists

class ai_marketplace_monitor.config.ConfigItem(*values)

Bases: Enum

AI = 'ai'

ITEM = 'item'

MARKETPLACE = 'marketplace'

MONITOR = 'monitor'

NOTIFICATION = 'notification'

REGION = 'region'

TRANSLATION = 'translation'

USER = 'user'

ai_marketplace_monitor.email_notify module

class ai_marketplace_monitor.email_notify.EmailNotificationConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = False, instance_rate_limit: float = 1.0, global_rate_limit: int = 10, _handles_own_rate_limiting: bool = False, _last_send_time: float | None = None, email: List[str] | None = None, smtp_server: str | None = None, smtp_port: int | None = None, smtp_username: str | None = None, smtp_password: str | None = None, smtp_from: str | None = None)

Bases: NotificationConfig

email: List[str] | None = None

get_html_message(listings: List[Listing], ratings: List[AIResponse], notification_status: List[NotificationStatus], force: bool = False, logger: Logger | None = None) → Tuple[str, list[Tuple[bytes, str, str]]]

get_text_message(listings: List[Listing], ratings: List[AIResponse], notification_status: List[NotificationStatus], force: bool = False, logger: Logger | None = None) → str

get_title(listings: List[Listing], notification_status: List[NotificationStatus], force: bool = False) → str

handle_email() → None

handle_smtp_from() → None

handle_smtp_password() → None

handle_smtp_port() → None

handle_smtp_server() → None

handle_smtp_username() → None

notify(listings: List[Listing], ratings: List[AIResponse], notification_status: List[NotificationStatus], force: bool = False, logger: Logger | None = None) → bool

notify_method = 'email'

required_fields: ClassVar[List[str]] = ['email', 'smtp_password']

send_email_message(title: str, message: str, html: str, images: List[Tuple[bytes, str, str]], logger: Logger | None = None) → bool

smtp_from: str | None = None

smtp_password: str | None = None

smtp_port: int | None = None

smtp_server: str | None = None

smtp_username: str | None = None

ai_marketplace_monitor.facebook module

class ai_marketplace_monitor.facebook.Availability(*values)

Bases: Enum

ALL = 'all'

INSTOCK = 'in'

OUTSTOCK = 'out'

class ai_marketplace_monitor.facebook.Category(*values)

Bases: Enum

APPAREL = 'apparel'

ELECTRONICS = 'electronics'

ENTERTAINMENT = 'entertainment'

FAMILY = 'family'

FREE = 'free'

FREE_STUFF = 'freestuff'

GARDEN = 'garden'

HOBBIES = 'hobbies'

HOME_GOODS = 'homegoods'

HOME_IMPROVEMENT = 'homeimprovement'

HOME_SALES = 'homesales'

MUSICAL_INSTRUMENTS = 'musicalinstruments'

OFFICE_SUPPLIES = 'officesupplies'

PET_SUPPLIES = 'petsupplies'

PROPERTY_RENTALS = 'propertyrentals'

SPORTING_GOODS = 'sportinggoods'

TICKETS = 'tickets'

TOYS = 'toys'

VEHICLES = 'vehicles'

VIDEO_GAMES = 'videogames'

class ai_marketplace_monitor.facebook.Condition(*values)

Bases: Enum

NEW = 'new'

USED_FAIR = 'used_fair'

USED_GOOD = 'used_good'

USED_LIKE_NEW = 'used_like_new'

class ai_marketplace_monitor.facebook.DateListed(*values)

Bases: Enum

ANYTIME = 0

PAST_24_HOURS = 1

PAST_MONTH = 30

PAST_WEEK = 7

class ai_marketplace_monitor.facebook.DeliveryMethod(*values)

Bases: Enum

ALL = 'all'

LOCAL_PICK_UP = 'local_pick_up'

SHIPPING = 'shipping'

class ai_marketplace_monitor.facebook.FacebookAutoItemWithAboutAndDescriptionPage(page: Page, translator: Translator | None = None, logger: Logger | None = None)

Bases: FacebookRegularItemPage

get_condition() → str

get_description() → str

get_price() → str

verify_layout() → bool

class ai_marketplace_monitor.facebook.FacebookAutoItemWithDescriptionPage(page: Page, translator: Translator | None = None, logger: Logger | None = None)

Bases: FacebookAutoItemWithAboutAndDescriptionPage

get_condition() → str

get_description() → str

get_price() → str

verify_layout() → bool

class ai_marketplace_monitor.facebook.FacebookItemConfig(name: str, enabled: bool | None = None, seller_locations: List[str] | None = None, availability: List[str] | None = None, condition: List[str] | None = None, date_listed: List[int] | None = None, delivery_method: List[str] | None = None, category: str | None = None, ai: List[str] | None = None, exclude_sellers: List[str] | None = None, notify: List[str] | None = None, search_city: List[str] | None = None, city_name: List[str] | None = None, radius: List[int] | None = None, currency: List[str] | None = None, search_interval: int | None = None, max_search_interval: int | None = None, start_at: List[str] | None = None, search_region: List[str] | None = None, max_price: str | None = None, min_price: str | None = None, rating: List[int] | None = None, prompt: str | None = None, extra_prompt: str | None = None, rating_prompt: str | None = None, searched_count: int = 0, search_phrases: List[str] = <factory>, keywords: List[str] | None = None, antikeywords: List[str] | None = None, description: str | None = None, marketplace: str | None = None)

Bases: ItemConfig, FacebookMarketItemCommonConfig

class ai_marketplace_monitor.facebook.FacebookItemPage(page: Page, translator: Translator | None = None, logger: Logger | None = None)

Bases: WebPage

get_condition() → str

get_description() → str

get_image_url() → str

get_location() → str

get_price() → str

get_seller() → str

get_title() → str

parse(post_url: str) → Listing

verify_layout() → bool

class ai_marketplace_monitor.facebook.FacebookMarketItemCommonConfig(name: str, enabled: bool | None = None, seller_locations: List[str] | None = None, availability: List[str] | None = None, condition: List[str] | None = None, date_listed: List[int] | None = None, delivery_method: List[str] | None = None, category: str | None = None)

Bases: BaseConfig

Item options that can be defined in marketplace

This class defines and processes options that can be specified in both marketplace and item sections, specific to facebook marketplace

availability: List[str] | None = None

category: str | None = None

condition: List[str] | None = None

date_listed: List[int] | None = None

delivery_method: List[str] | None = None

handle_availability() → None

handle_category() → None

handle_condition() → None

handle_date_listed() → None

handle_delivery_method() → None

handle_seller_locations() → None

seller_locations: List[str] | None = None

class ai_marketplace_monitor.facebook.FacebookMarketplace(name: str, browser: Browser | None, keyboard_monitor: KeyboardMonitor | None = None, logger: Logger | None = None)

Bases: Marketplace

check_listing(item: Listing, item_config: FacebookItemConfig, description_available: bool = True) → bool

classmethod get_config(**kwargs: Any) → FacebookMarketplaceConfig

classmethod get_item_config(**kwargs: Any) → FacebookItemConfig

get_listing_details(post_url: str, item_config: ItemConfig, price: str | None = None, title: str | None = None) → Tuple[Listing, bool]

initial_url = 'https://www.facebook.com/login/device-based/regular/login/'

login() → None

name = 'facebook'

page: Page | None

search(item_config: FacebookItemConfig) → Generator[Listing, None, None]

class ai_marketplace_monitor.facebook.FacebookMarketplaceConfig(name: str, enabled: bool | None = None, seller_locations: List[str] | None = None, availability: List[str] | None = None, condition: List[str] | None = None, date_listed: List[int] | None = None, delivery_method: List[str] | None = None, category: str | None = None, ai: List[str] | None = None, exclude_sellers: List[str] | None = None, notify: List[str] | None = None, search_city: List[str] | None = None, city_name: List[str] | None = None, radius: List[int] | None = None, currency: List[str] | None = None, search_interval: int | None = None, max_search_interval: int | None = None, start_at: List[str] | None = None, search_region: List[str] | None = None, max_price: str | None = None, min_price: str | None = None, rating: List[int] | None = None, prompt: str | None = None, extra_prompt: str | None = None, rating_prompt: str | None = None, market_type: str | None = 'facebook', language: str | None = None, monitor_config: MonitorConfig | None = None, login_wait_time: int | None = None, password: str | None = None, username: str | None = None)

Bases: MarketplaceConfig, FacebookMarketItemCommonConfig

Options specific to facebook marketplace

This class defines and processes options that can be specified in the marketplace.facebook section only. None of the options are required.

handle_login_wait_time() → None

handle_password() → None

handle_username() → None

login_wait_time: int | None = None

password: str | None = None

username: str | None = None

class ai_marketplace_monitor.facebook.FacebookRegularItemPage(page: Page, translator: Translator | None = None, logger: Logger | None = None)

Bases: FacebookItemPage

get_condition() → str

get_description() → str

get_image_url() → str

get_location() → str

get_price() → str

get_seller() → str

get_title() → str

verify_layout() → bool

class ai_marketplace_monitor.facebook.FacebookRentalItemPage(page: Page, translator: Translator | None = None, logger: Logger | None = None)

Bases: FacebookRegularItemPage

get_condition() → str

get_description() → str

verify_layout() → bool

class ai_marketplace_monitor.facebook.FacebookSearchResultPage(page: Page, translator: Translator | None = None, logger: Logger | None = None)

Bases: WebPage

get_listings() → List[Listing]

ai_marketplace_monitor.facebook.parse_listing(page: Page, post_url: str, translator: Translator | None = None, logger: Logger | None = None) → Listing | None

ai_marketplace_monitor.listing module

class ai_marketplace_monitor.listing.Listing(marketplace: str, name: str, id: str, title: str, image: str, price: str, post_url: str, location: str, seller: str, condition: str, description: str)

Bases: object

condition: str

property content: Tuple[str, str, str]

description: str

classmethod from_cache(post_url: str, local_cache: Cache | None = None) → Listing | None

property hash: str

id: str

image: str

location: str

marketplace: str

name: str

post_url: str

price: str

seller: str

title: str

to_cache(post_url: str, local_cache: Cache | None = None) → None

ai_marketplace_monitor.marketplace module

class ai_marketplace_monitor.marketplace.ItemConfig(name: str, enabled: bool | None = None, ai: List[str] | None = None, exclude_sellers: List[str] | None = None, notify: List[str] | None = None, search_city: List[str] | None = None, city_name: List[str] | None = None, radius: List[int] | None = None, currency: List[str] | None = None, search_interval: int | None = None, max_search_interval: int | None = None, start_at: List[str] | None = None, search_region: List[str] | None = None, max_price: str | None = None, min_price: str | None = None, rating: List[int] | None = None, prompt: str | None = None, extra_prompt: str | None = None, rating_prompt: str | None = None, searched_count: int = 0, search_phrases: List[str] = <factory>, keywords: List[str] | None = None, antikeywords: List[str] | None = None, description: str | None = None, marketplace: str | None = None)

Bases: MarketItemCommonConfig

This class defined options that can only be specified for items.

antikeywords: List[str] | None = None

description: str | None = None

handle_antikeywords() → None

handle_description() → None

handle_keywords() → None

handle_search_phrases() → None

keywords: List[str] | None = None

marketplace: str | None = None

search_phrases: List[str]

searched_count: int = 0

class ai_marketplace_monitor.marketplace.MarketItemCommonConfig(name: str, enabled: bool | None = None, ai: List[str] | None = None, exclude_sellers: List[str] | None = None, notify: List[str] | None = None, search_city: List[str] | None = None, city_name: List[str] | None = None, radius: List[int] | None = None, currency: List[str] | None = None, search_interval: int | None = None, max_search_interval: int | None = None, start_at: List[str] | None = None, search_region: List[str] | None = None, max_price: str | None = None, min_price: str | None = None, rating: List[int] | None = None, prompt: str | None = None, extra_prompt: str | None = None, rating_prompt: str | None = None)

Bases: BaseConfig

Item options that can be specified in market (non-marketplace specifc)

This class defines and processes options that can be specified in both marketplace and item sections, generic to all marketplaces

ai: List[str] | None = None

city_name: List[str] | None = None

currency: List[str] | None = None

exclude_sellers: List[str] | None = None

extra_prompt: str | None = None

handle_ai() → None

handle_city_name() → None

handle_currency() → None

handle_exclude_sellers() → None

handle_extra_prompt() → None

handle_max_price() → None

handle_max_search_interval() → None

handle_min_price() → None

handle_notify() → None

handle_prompt() → None

handle_radius() → None

handle_rating() → None

handle_rating_prompt() → None

handle_search_city() → None

handle_search_interval() → None

handle_search_region() → None

handle_start_at() → None

max_price: str | None = None

max_search_interval: int | None = None

min_price: str | None = None

notify: List[str] | None = None

prompt: str | None = None

radius: List[int] | None = None

rating: List[int] | None = None

rating_prompt: str | None = None

search_city: List[str] | None = None

search_interval: int | None = None

search_region: List[str] | None = None

start_at: List[str] | None = None

class ai_marketplace_monitor.marketplace.MarketPlace(*values)

Bases: Enum

FACEBOOK = 'facebook'

class ai_marketplace_monitor.marketplace.Marketplace(name: str, browser: Browser | None, keyboard_monitor: KeyboardMonitor | None = None, logger: Logger | None = None)

Bases: Generic[TMarketplaceConfig, TItemConfig]

configure(config: TMarketplaceConfig, translator: Translator | None = None) → None

create_page(swap_proxy: bool = False) → Page

classmethod get_config(**kwargs: Any) → TMarketplaceConfig

classmethod get_item_config(**kwargs: Any) → TItemConfig

goto_url(url: str, attempt: int = 0) → None

page: Page | None

search(item: TItemConfig) → Generator[Listing, None, None]

set_browser(browser: Browser | None = None) → None

stop() → None

class ai_marketplace_monitor.marketplace.MarketplaceConfig(name: str, enabled: bool | None = None, ai: List[str] | None = None, exclude_sellers: List[str] | None = None, notify: List[str] | None = None, search_city: List[str] | None = None, city_name: List[str] | None = None, radius: List[int] | None = None, currency: List[str] | None = None, search_interval: int | None = None, max_search_interval: int | None = None, start_at: List[str] | None = None, search_region: List[str] | None = None, max_price: str | None = None, min_price: str | None = None, rating: List[int] | None = None, prompt: str | None = None, extra_prompt: str | None = None, rating_prompt: str | None = None, market_type: str | None = 'facebook', language: str | None = None, monitor_config: MonitorConfig | None = None)

Bases: MarketItemCommonConfig

Generic marketplace config

handle_language() → None

handle_market_type() → None

language: str | None = None

market_type: str | None = 'facebook'

monitor_config: MonitorConfig | None = None

class ai_marketplace_monitor.marketplace.WebPage(page: Page, translator: Translator | None = None, logger: Logger | None = None)

Bases: object

translator: Translator

ai_marketplace_monitor.monitor module

class ai_marketplace_monitor.monitor.MarketplaceMonitor(config_files: List[Path] | None, headless: bool | None, logger: Logger | None)

Bases: object

active_marketplaces: ClassVar = {}

check_items(items: List[str] | None = None, for_item: str | None = None) → None

Main function to monitor the marketplace.

evaluate_by_ai(item: Listing, item_config: TItemConfig, marketplace_config: TMarketplaceConfig) → AIResponse

handle_pause() → None

Handle interruption signal.

load_ai_agents() → None

Load the AI agent.

load_config_file() → Config

Load the configuration file.

schedule_jobs() → None

Schedule jobs to run periodically.

search_item(marketplace_config: TMarketplaceConfig, marketplace: Marketplace, item_config: TItemConfig) → None

Search for an item on the marketplace.

start_monitor() → None

Main function to monitor the marketplace.

stop_monitor() → None

Stop the monitor.

ai_marketplace_monitor.notification module

class ai_marketplace_monitor.notification.NotificationConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = False, instance_rate_limit: float = 1.0, global_rate_limit: int = 10, _handles_own_rate_limiting: bool = False, _last_send_time: float | None = None)

Bases: BaseConfig

classmethod get_config(**kwargs: Any) → NotificationConfig | None

Get the specific subclass name from the specified keys, for validation purposes

global_rate_limit: int = 10

handle_max_retries() → None

handle_retry_delay() → None

instance_rate_limit: float = 1.0

max_retries: int = 5

classmethod notify_all(config: NotificationConfig, *args, **kwargs: Any) → bool

Call the notify method of all subclasses

rate_limit_enabled: bool = False

required_fields: ClassVar[List[str]] = []

retry_delay: int = 60

send_message(title: str, message: str, logger: Logger | None = None) → bool

send_message_with_retry(title: str, message: str, logger: Logger | None = None) → bool

Enhanced retry method with rate limiting support.

Subclasses that set _handles_own_rate_limiting = True (e.g. Telegram, which applies async rate limiting inside its own send_message) will NOT get sync rate limiting here — avoiding a double-wait.

class ai_marketplace_monitor.notification.NotificationStatus(*values)

Bases: Enum

EXPIRED = 1

LISTING_CHANGED = 3

LISTING_DISCOUNTED = 4

NOTIFIED = 2

NOT_NOTIFIED = 0

class ai_marketplace_monitor.notification.PushNotificationConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = False, instance_rate_limit: float = 1.0, global_rate_limit: int = 10, _handles_own_rate_limiting: bool = False, _last_send_time: float | None = None, message_format: str | None = None, with_description: int | None = None)

Bases: NotificationConfig

handle_message_format() → None

handle_with_description() → None

message_format: str | None = None

notify(listings: List[Listing], ratings: List[AIResponse], notification_status: List[NotificationStatus], force: bool = False, logger: Logger | None = None) → bool

notify_method = 'push_notification'

with_description: int | None = None

ai_marketplace_monitor.ntfy module

class ai_marketplace_monitor.ntfy.NtfyNotificationConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = False, instance_rate_limit: float = 1.0, global_rate_limit: int = 10, _handles_own_rate_limiting: bool = False, _last_send_time: float | None = None, message_format: str | None = None, with_description: int | None = None, ntfy_server: str | None = None, ntfy_topic: str | None = None)

Bases: PushNotificationConfig

handle_ntfy_server() → None

handle_ntfy_topic() → None

message_format: str | None = None

notify_method = 'ntfy'

ntfy_server: str | None = None

ntfy_topic: str | None = None

required_fields: ClassVar[List[str]] = ['ntfy_server', 'ntfy_topic']

send_message(title: str, message: str, logger: Logger | None = None) → bool

ai_marketplace_monitor.pushbullet module

class ai_marketplace_monitor.pushbullet.PushbulletNotificationConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = False, instance_rate_limit: float = 1.0, global_rate_limit: int = 10, _handles_own_rate_limiting: bool = False, _last_send_time: float | None = None, message_format: str | None = None, with_description: int | None = None, pushbullet_token: str | None = None, pushbullet_proxy_type: str | None = None, pushbullet_proxy_server: str | None = None)

Bases: PushNotificationConfig

handle_message_format() → None

handle_pushbullet_proxy_server() → None

handle_pushbullet_proxy_type() → None

handle_pushbullet_token() → None

notify_method = 'pushbullet'

pushbullet_proxy_server: str | None = None

pushbullet_proxy_type: str | None = None

pushbullet_token: str | None = None

required_fields: ClassVar[List[str]] = ['pushbullet_token']

send_message(title: str, message: str, logger: Logger | None = None) → bool

ai_marketplace_monitor.pushover module

class ai_marketplace_monitor.pushover.PushoverNotificationConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = False, instance_rate_limit: float = 1.0, global_rate_limit: int = 10, _handles_own_rate_limiting: bool = False, _last_send_time: float | None = None, message_format: str | None = None, with_description: int | None = None, pushover_user_key: str | None = None, pushover_api_token: str | None = None)

Bases: PushNotificationConfig

handle_message_format() → None

handle_pushover_api_token() → None

handle_pushover_user_key() → None

notify_method = 'pushover'

pushover_api_token: str | None = None

pushover_user_key: str | None = None

required_fields: ClassVar[List[str]] = ['pushover_user_key', 'pushover_api_token']

send_message(title: str, message: str, logger: Logger | None = None) → bool

ai_marketplace_monitor.region module

class ai_marketplace_monitor.region.RegionConfig(name: str, enabled: bool | None = None, search_city: List[str] = <factory>, full_name: str = '', radius: List[int] = <factory>, city_name: List[str] = <factory>, currency: List[str] = <factory>)

Bases: BaseConfig

city_name: List[str]

currency: List[str]

full_name: str = ''

handle_city_name() → None

handle_currency() → None

handle_radius() → None

handle_search_city() → None

radius: List[int]

search_city: List[str]

ai_marketplace_monitor.telegram module

class ai_marketplace_monitor.telegram.TelegramNotificationConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = True, instance_rate_limit: float = 1.0, global_rate_limit: int = 30, _handles_own_rate_limiting: bool = True, _last_send_time: float | None = None, message_format: str | None = None, with_description: int | None = None, telegram_token: str | None = None, telegram_chat_id: str | None = None)

Bases: PushNotificationConfig

global_rate_limit: int = 30

handle_telegram_chat_id() → None

handle_telegram_token() → None

notify_method = 'telegram'

rate_limit_enabled: bool = True

required_fields: ClassVar[List[str]] = ['telegram_token', 'telegram_chat_id']

send_message(title: str, message: str, logger: Logger | None = None) → bool

Send message using asyncio.run to call async Telegram operations.

telegram_chat_id: str | None = None

telegram_token: str | None = None

ai_marketplace_monitor.user module

class ai_marketplace_monitor.user.User(config: UserConfig, logger: Logger | None = None)

Bases: object

classmethod get_config(**kwargs: Any) → UserConfig

notification_status(listing: Listing, local_cache: Cache | None = None) → NotificationStatus

notified_key(listing: Listing) → Tuple[str, str, str, str]

notify(listings: List[Listing], ratings: List[AIResponse], item_config: TItemConfig, local_cache: Cache | None = None, force: bool = False) → None

time_since_notification(listing: Listing, local_cache: Cache | None = None) → int

to_cache(listing: Listing, local_cache: Cache | None = None) → None

class ai_marketplace_monitor.user.UserConfig(name: str, enabled: bool | None = None, max_retries: int = 5, retry_delay: int = 60, rate_limit_enabled: bool = False, instance_rate_limit: float = 1.0, global_rate_limit: int = 10, _handles_own_rate_limiting: bool = False, _last_send_time: float | None = None, message_format: str | None = None, with_description: int | None = None, telegram_token: str | None = None, telegram_chat_id: str | None = None, ntfy_server: str | None = None, ntfy_topic: str | None = None, pushover_user_key: str | None = None, pushover_api_token: str | None = None, pushbullet_token: str | None = None, pushbullet_proxy_type: str | None = None, pushbullet_proxy_server: str | None = None, email: List[str] | None = None, smtp_server: str | None = None, smtp_port: int | None = None, smtp_username: str | None = None, smtp_password: str | None = None, smtp_from: str | None = None, notify_with: List[str] | None = None, remind: int | None = None)

Bases: EmailNotificationConfig, PushbulletNotificationConfig, PushoverNotificationConfig, NtfyNotificationConfig, TelegramNotificationConfig

UserConfiguration

Derive from EmailNotificationConfig, PushbulletNotificationConfig allows the user config class to use settings from both classes.

It is possible to dynamically added these classes as parent class of UserConfig, but it is troublesome to make sure that these classes are imported.

handle_notify_with() → None

handle_remind() → None

notify_with: List[str] | None = None

remind: int | None = None

ai_marketplace_monitor.utils module

class ai_marketplace_monitor.utils.BaseConfig(name: str, enabled: bool | None = None)

Bases: object

enabled: bool | None = None

handle_enabled() → None

property hash: str

name: str

class ai_marketplace_monitor.utils.CacheType(*values)

Bases: Enum

AI_INQUIRY = 'ai-inquiries'

COUNTERS = 'counters'

LISTING_DETAILS = 'listing-details'

USER_NOTIFIED = 'user-notifications'

class ai_marketplace_monitor.utils.ChangeHandler(files: List[str])

Bases: FileSystemEventHandler

on_created(event: FileSystemEvent) → None

Called when a file or directory is created.

Parameters:

event (DirCreatedEvent or FileCreatedEvent) – Event representing file/directory creation.

on_deleted(event: FileSystemEvent) → None

Called when a file or directory is deleted.

Parameters:

event (DirDeletedEvent or FileDeletedEvent) – Event representing file/directory deletion.

on_modified(event: FileSystemEvent) → None

Called when a file or directory is modified.

Parameters:

event (DirModifiedEvent or FileModifiedEvent) – Event representing file/directory modification.

on_moved(event: FileSystemEvent) → None

Called when a file or a directory is moved or renamed.

Parameters:

event (DirMovedEvent or FileMovedEvent) – Event representing file/directory movement.

class ai_marketplace_monitor.utils.Counter

Bases: object

increment(counter_key: CounterItem, item_name: str, by: int = 1) → None

class ai_marketplace_monitor.utils.CounterItem(*values)

Bases: Enum

AI_QUERY = 'Total AI Queries'

EXCLUDED_LISTING = 'Listing excluded'

FAILED_AI_QUERY = 'Failed AI Queries)'

LISTING_EXAMINED = 'Total listing examined'

LISTING_QUERY = 'New listing fetched'

NEW_AI_QUERY = 'New AI Queries'

NEW_VALIDATED_LISTING = 'New validated listing'

NOTIFICATIONS_SENT = 'Notifications sent'

REMINDERS_SENT = 'Reminders sent'

SEARCH_PERFORMED = 'Search performed'

class ai_marketplace_monitor.utils.Currency(*values)

Bases: Enum

ARS_unsupported = 'ARS'

AUD = 'AUD'

BGN = 'BGN'

BRL = 'BRL'

CAD = 'CAD'

CHF = 'CHF'

CNY = 'CNY'

CYP = 'CYP'

CZK = 'CZK'

DKK = 'DKK'

EEK = 'EEK'

EUR = 'EUR'

GBP = 'GBP'

HKD = 'HKD'

HRK = 'HRK'

HUF = 'HUF'

IDR = 'IDR'

ILS = 'ILS'

INR = 'INR'

ISK = 'ISK'

JPY = 'JPY'

KRW = 'KRW'

LTL = 'LTL'

LVL = 'LVL'

MTL = 'MTL'

MXN = 'MXN'

MYR = 'MYR'

NOK = 'NOK'

NZD = 'NZD'

PHP = 'PHP'

PLN = 'PLN'

ROL = 'ROL'

RON = 'RON'

RUB = 'RUB'

SEK = 'SEK'

SGD = 'SGD'

SIT = 'SIT'

SKK = 'SKK'

THB = 'THB'

TRL = 'TRL'

TRY = 'TRY'

USD = 'USD'

ZAR = 'ZAR'

class ai_marketplace_monitor.utils.KeyboardMonitor

Bases: object

confirm(msg: str | None = None) → bool

confirm_character = 'c'

is_confirmed() → bool

is_paused() → bool

is_sleeping() → bool

set_paused(paused: bool = True) → None

start() → None

start_sleeping() → None

stop() → None

class ai_marketplace_monitor.utils.MonitorConfig(name: str, enabled: bool | None = None, proxy_server: List[str] | None = None, proxy_bypass: str | None = None, proxy_username: str | None = None, proxy_password: str | None = None)

Bases: BaseConfig

get_proxy_options() → ProxySettings | None

handle_proxy_bypass() → None

handle_proxy_password() → None

handle_proxy_server() → None

handle_proxy_username() → None

proxy_bypass: str | None = None

proxy_password: str | None = None

proxy_server: List[str] | None = None

proxy_username: str | None = None

class ai_marketplace_monitor.utils.SleepStatus(*values)

Bases: Enum

BY_FILE_CHANGE = 2

BY_KEYBOARD = 1

NOT_DISRUPTED = 0

class ai_marketplace_monitor.utils.Translator(locale: str | None = None, dictionary: Dict[str, str] | None = None)

Bases: object

ai_marketplace_monitor.utils.aimm_event(kind: str, **fields: Any) → Dict[str, Any]

Build a structured-event payload for a log call.

Usage:

logger.info(message, extra=aimm_event(“ai_eval”, score=5, …))

The web UI surfaces these structured fields in its filter dropdowns (kind / item / score) and in the expand-row detail pane.

ai_marketplace_monitor.utils.calculate_file_hash(file_paths: List[Path]) → str

Calculate the SHA-256 hash of the file content.

ai_marketplace_monitor.utils.convert_to_seconds(time_str: str) → int

ai_marketplace_monitor.utils.doze(duration: int, files: List[Path] | None = None, keyboard_monitor: KeyboardMonitor | None = None) → SleepStatus

Sleep for a specified duration while monitoring the change of files.

Returns:

if doze was done naturally. 1: if doze was disrupted by keyboard 2: if doze was disrupted by file change

Return type:

0

ai_marketplace_monitor.utils.extract_price(price: str) → str

ai_marketplace_monitor.utils.fetch_with_retry(url: str, timeout: int = 10, max_retries: int = 3, backoff_factor: float = 1.5, logger: Logger | None = None) → Tuple[bytes, str] | None

Fetch URL content with retry logic

Parameters:

• url – URL to fetch

• timeout – Timeout in seconds

• max_retries – Maximum number of retry attempts

• backoff_factor – Multiplier for exponential backoff

• logger – logger object

Returns:

Tuple of (content, content_type) if successful, None if failed

ai_marketplace_monitor.utils.hash_dict(obj: Dict[str, Any]) → str

Hash a dictionary to a string.

ai_marketplace_monitor.utils.hilight(text: str, style: str = 'name') → str

Highlight the keywords in the text with the specified color.

ai_marketplace_monitor.utils.is_substring(var1: str | List[str], var2: str | List[str], logger: Logger | None = None) → bool

Check if var1 is a substring of var2, after normalizing both strings. One of them can be a list of strings.

var1: can be a single string, or a list of string, for which a condition of OR is assumed.

this program will parse var11 for “AND”, “OR” and “NOT”, and return the results of the logical expression.

var2: one or more strings for testing if strings in “var1” is a substring.

ai_marketplace_monitor.utils.merge_dicts(dicts: list) → dict

Merge a list of dictionaries into a single dictionary, including nested dictionaries.

Parameters:

dicts – A list of dictionaries to merge.

Returns:

A single merged dictionary.

ai_marketplace_monitor.utils.normalize_string(string: str) → str

Normalize a string by replacing multiple spaces (including space, tab, and newline) with a single space.

ai_marketplace_monitor.utils.resize_image_data(image_data: bytes, max_width: int = 800, max_height: int = 600) → bytes

Module contents

Top-level package for ai-marketplace-monitor.