Code improvement

pyproject.toml

@@ -37,6 +37,7 @@ diskcache = "^5.6.3"
 watchdog = "^6.0.0"
 openai = "^1.60.1"
 parsedatetime = "^2.6"
+humanize = "^4.11.0"
 schedule = "^1.2.2"
 tomli = { version = "2.2.1", markers = "python_version < '3.11'" }
 

src/ai_marketplace_monitor/ai.py

@@ -49,10 +49,8 @@ def get_config(cls: Type["AIBackend"], **kwargs: Any) -> TAIConfig:
     def connect(self: "AIBackend") -> None:
         raise NotImplementedError("Connect method must be implemented by subclasses.")
 
-    def get_prompt(
-        self: "AIBackend", listing: SearchedItem, item_name: str, item_config: TItemConfig
-    ) -> str:
-        prompt = f"""A user would like to buy a {item_name} from facebook marketplace.
+    def get_prompt(self: "AIBackend", listing: SearchedItem, item_config: TItemConfig) -> str:
+        prompt = f"""A user would like to buy a {item_config.name} from facebook marketplace.
             He used keywords "{'" and "'.join(item_config.keywords)}" to perform the search."""
         if item_config.description:
             prompt += f""" He also added description "{item_config.description}" to describe the item he is interested in."""
@@ -83,9 +81,7 @@ def get_prompt(
         self.logger.debug(f"Prompt: {prompt}")
         return prompt
 
-    def confirm(
-        self: "AIBackend", listing: SearchedItem, item_name: str, item_config: TItemConfig
-    ) -> bool:
+    def confirm(self: "AIBackend", listing: SearchedItem, item_config: TItemConfig) -> bool:
         raise NotImplementedError("Confirm method must be implemented by subclasses.")
 
 
@@ -106,11 +102,9 @@ def connect(self: "OpenAIBackend") -> None:
                 timeout=10,
             )
 
-    def confirm(
-        self: "OpenAIBackend", listing: SearchedItem, item_name: str, item_config: TItemConfig
-    ) -> bool:
+    def confirm(self: "OpenAIBackend", listing: SearchedItem, item_config: TItemConfig) -> bool:
         # ask openai to confirm the item is correct
-        prompt = self.get_prompt(listing, item_name, item_config)
+        prompt = self.get_prompt(listing, item_config)
 
         assert self.client is not None
 

src/ai_marketplace_monitor/config.py

@@ -1,6 +1,7 @@
 import os
 import sys
 from dataclasses import dataclass, field
+from itertools import chain
 from logging import Logger
 from typing import Any, Dict, Generic, List
 
@@ -145,37 +146,20 @@ def validate_users(self: "Config") -> None:
 
     def expand_regions(self: "Config") -> None:
         # if region is specified in other section, they must exist
-        for marketplace_config in self.marketplace.values():
-            if marketplace_config.search_region is None:
+        for config in chain(self.marketplace.values(), self.item.values()):
+            if config.search_region is None:
                 continue
+            config.search_city = []
+            config.radius = []
 
-            marketplace_config.search_city = []
-            marketplace_config.radius = []
-
-            for region in marketplace_config.search_region:
+            for region in config.search_region:
                 region_config: RegionConfig = self.region[region]
                 if region not in self.region:
                     raise ValueError(
-                        f"Region [magenta]{region}[/magenta] specified in [magenta]{marketplace_config.name}[/magenta] does not exist."
-                    )
-                # if region is specified, expand it into search_city
-                marketplace_config.search_city.extend(region_config.search_city)
-                # set radius, if market_config already has radius, they should be the same
-                marketplace_config.radius.extend(region_config.radius)
-
-        # if region is specified in any of the items, do the same
-        for item_config in self.item.values():
-            # expand region into item_config's search_city
-            if item_config.search_region is None:
-                continue
-            item_config.search_city = []
-            item_config.radius = []
-            for region in item_config.search_region:
-                region_config = self.region[region]
-                if region not in self.region:
-                    raise ValueError(
-                        f"Region [magenta]{region}[/magenta] specified in [magenta]{item_config.name}[/magenta] does not exist."
+                        f"Region [magenta]{region}[/magenta] specified in [magenta]{config.name}[/magenta] does not exist."
                     )
-                # if region is specified, expand it into search_city
-                item_config.search_city.extend(region_config.search_city)
-                item_config.radius.extend(region_config.radius)
+                # avoid duplicated addition of search_city
+                for search_city, radius in zip(region_config.search_city, region_config.radius):
+                    if search_city not in config.search_city:
+                        config.search_city.append(search_city)
+                        config.radius.append(radius)

src/ai_marketplace_monitor/config.toml

@@ -1,3 +1,14 @@
+#
+# Region definitions.
+# - full_name and city_name are for readability/booktracking purpose only
+# - different radius can be used for different search_city. In this case
+#   radius should an array with the same length as search_city
+#
+# Usage：
+# - ·search_city` will be ignored if `search_region` is specified.
+# - under the hood search_city is replaced by `search_city` of the regions
+#
+
 [region.usa]
 full_name = "USA (without AK or HI)"
 radius = 500

src/ai_marketplace_monitor/facebook.py

@@ -14,6 +14,7 @@
 from .item import SearchedItem
 from .marketplace import ItemConfig, Marketplace, MarketplaceConfig
 from .utils import (
+    CacheType,
     DataClassWithHandleFunc,
     cache,
     convert_to_seconds,
@@ -49,14 +50,17 @@ class Availability(Enum):
 
 @dataclass
 class FacebookMarketItemCommonConfig(DataClassWithHandleFunc):
+    """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
+    """
+
     acceptable_locations: List[str] | None = None
     availability: str | None = None
     condition: List[str] | None = None
     date_listed: str | None = None
     delivery_method: str | None = None
-    exclude_sellers: List[str] | None = None
-    max_price: int | None = None
-    min_price: int | None = None
 
     def handle_acceptable_locations(self: "FacebookMarketItemCommonConfig") -> None:
         if self.acceptable_locations is None:
@@ -113,35 +117,15 @@ def handle_delivery_method(self: "FacebookMarketItemCommonConfig") -> None:
                 f"Item [magenta]{self.name}[/magenta] delivery_method must be one of 'local_pick_up' and 'shipping'."
             )
 
-    def handle_exclude_sellers(self: "FacebookMarketItemCommonConfig") -> None:
-        if self.exclude_sellers is None:
-            return
-
-        if isinstance(self.exclude_sellers, str):
-            self.exclude_sellers = [self.exclude_sellers]
-        if not isinstance(self.exclude_sellers, list) or not all(
-            isinstance(x, str) for x in self.exclude_sellers
-        ):
-            raise ValueError(
-                f"Item [magenta]{self.name}[/magenta] exclude_sellers must be a list."
-            )
-
-    def handle_max_price(self: "FacebookMarketItemCommonConfig") -> None:
-        if self.max_price is None:
-            return
-        if not isinstance(self.max_price, int):
-            raise ValueError(f"Item [magenta]{self.name}[/magenta] max_price must be an integer.")
-
-    def handle_min_price(self: "FacebookMarketItemCommonConfig") -> None:
-        if self.min_price is None:
-            return
-
-        if not isinstance(self.min_price, int):
-            raise ValueError(f"Item [magenta]{self.name}[/magenta] min_price must be an integer.")
-
 
 @dataclass
 class FacebookMarketplaceConfig(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.
+    """
+
     login_wait_time: int | None = None
     password: str | None = None
     username: str | None = None
@@ -305,14 +289,15 @@ def search(
                     # so we do not copy title, description etc from the detailed result
                     item.description = details.description
                     item.seller = details.seller
+                    item.name = item_config.name
                     self.logger.debug(
                         f"""New item "{item.title}" from https://www.facebook.com{item.post_url} is sold by "{item.seller}" and with description "{item.description[:100]}..." """
                     )
                     if self.filter_item(item, item_config):
                         yield item
 
     def get_item_details(self: "FacebookMarketplace", post_url: str) -> SearchedItem:
-        details = cache.get(("get_item_details", post_url))
+        details = cache.get((CacheType.ITEM_DETAILS.value, post_url))
         if details is not None:
             return details
 
@@ -322,7 +307,7 @@ def get_item_details(self: "FacebookMarketplace", post_url: str) -> SearchedItem
         assert self.page is not None
         self.goto_url(f"https://www.facebook.com{post_url}")
         details = FacebookItemPage(self.page.content(), self.logger).parse(post_url)
-        cache.set(("get_item_details", post_url), details, tag="item_details")
+        cache.set((CacheType.ITEM_DETAILS.value, post_url), details, tag="item_details")
         return details
 
     def filter_item(
@@ -441,6 +426,7 @@ def parse_listing(
         # Append the parsed data to the list.
         return SearchedItem(
             marketplace="facebook",
+            name="",
             id=post_url.split("?")[0].rstrip("/").split("/")[-1],
             title=title,
             image=image,
@@ -546,6 +532,7 @@ def parse(self: "FacebookItemPage", post_url: str) -> SearchedItem:
         self.logger.info(f"Parsing item [magenta]{title}[/magenta]")
         res = SearchedItem(
             marketplace="facebook",
+            name="",
             id=item_id,
             title=title,
             image=self.get_image_url(),

src/ai_marketplace_monitor/item.py

@@ -1,12 +1,13 @@
-from dataclasses import dataclass, field
-from typing import List
+from dataclasses import dataclass
+from typing import Tuple
 
-from .utils import DataClassWithHandleFunc
+from .utils import CacheType
 
 
 @dataclass
 class SearchedItem:
     marketplace: str
+    name: str
     # unique identification
     id: str
     title: str
@@ -17,11 +18,6 @@ class SearchedItem:
     seller: str
     description: str
 
-
-@dataclass
-class ItemConfig(DataClassWithHandleFunc):
-    """Generic item config"""
-
-    notify: List[str] = field(default_factory=list)
-    search_interval: int = 30
-    max_search_interval: int = 60
+    @property
+    def user_notified_key(self: "SearchedItem") -> Tuple[str, str, str]:
+        return (CacheType.USER_NOTIFIED.value, self.marketplace, self.id)

src/ai_marketplace_monitor/marketplace.py

@@ -11,7 +11,13 @@
 
 @dataclass
 class MarketItemCommonConfig(DataClassWithHandleFunc):
+    """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
+    """
+
+    exclude_sellers: List[str] | None = None
     max_search_interval: int | None = None
     notify: List[str] | None = None
     search_city: List[str] | None = None
@@ -22,6 +28,19 @@ class MarketItemCommonConfig(DataClassWithHandleFunc):
     max_price: int | None = None
     min_price: int | None = None
 
+    def handle_exclude_sellers(self: "MarketItemCommonConfig") -> None:
+        if self.exclude_sellers is None:
+            return
+
+        if isinstance(self.exclude_sellers, str):
+            self.exclude_sellers = [self.exclude_sellers]
+        if not isinstance(self.exclude_sellers, list) or not all(
+            isinstance(x, str) for x in self.exclude_sellers
+        ):
+            raise ValueError(
+                f"Item [magenta]{self.name}[/magenta] exclude_sellers must be a list."
+            )
+
     def handle_max_search_interval(self: "MarketItemCommonConfig") -> None:
         if self.max_search_interval is None:
             return
@@ -137,14 +156,15 @@ class MarketplaceConfig(MarketItemCommonConfig):
 
 @dataclass
 class ItemConfig(MarketItemCommonConfig):
-    """Generic item config"""
+    """This class defined options that can only be specified for items."""
 
+    # keywords is required, all others are optional
     keywords: List[str] = field(default_factory=list)
     exclude_keywords: List[str] | None = None
     exclude_by_description: List[str] | None = None
     description: str | None = None
     enabled: bool | None = None
-    marketplace: str = "facebook"
+    marketplace: str | None = None
 
     def handle_keywords(self: "ItemConfig") -> None:
         if isinstance(self.keywords, str):
@@ -202,7 +222,7 @@ def get_config(cls: Type["Marketplace"], **kwargs: Any) -> TMarketplaceConfig:
     def get_item_config(cls: Type["Marketplace"], **kwargs: Any) -> TItemConfig:
         raise NotImplementedError("get_config method must be implemented by subclasses.")
 
-    def configure(self: "Marketplace", config: MarketplaceConfig) -> None:
+    def configure(self: "Marketplace", config: TMarketplaceConfig) -> None:
         self.config = config
 
     def set_browser(self: "Marketplace", browser: Browser) -> None: