working listing and agent feature

Author: offish

src/tf2_utils/backpack_tf.py

@@ -1,14 +1,17 @@
 import requests
 
-from dataclasses import dataclass, asdict
+from dataclasses import dataclass, field
+
+from .schema import SchemaItemsUtils
+from .sku import sku_to_quality, sku_is_craftable
+from . import __title__
 
 
 __all__ = [
     "Currencies",
     "Enity",
-    "ItemResolvable",
+    # "ItemResolvable",
     "ItemDocument",
-    "ListingResolvable",
     "Listing",
     "BackpackTF",
 ]
@@ -22,55 +25,58 @@ class Currencies:
 
 @dataclass
 class Enity:
-    name: str
-    id: int
-    color: str
-
-
-@dataclass
-class ItemResolvable:
-    item: str
-    quality: str | int
-    tradable: bool
-    craftable: str
-    priceindex: str
+    name: str = ""
+    id: int = 0
+    color: str = ""
 
 
 @dataclass
 class ItemDocument:
+    appid: int
     baseName: str
-    name: str
+    defindex: int
+    id: str
     imageUrl: str
-    quantity: int
+    marketName: str
+    name: str
+    # origin:None
+    originalId: str
+    price: dict
     quality: Enity
-    rarity: Enity
-    paint: Enity
-    particle: Enity
-    elevatedQuality: Enity
+    summary: str
+    # class:list
+    slot: str
+    tradable: bool
+    craftable: bool
 
 
 @dataclass
-class ListingResolvable:
-    id: int  # asset_id if this is set, its a sell order
-    item: dict
-    details: str
-    currencies: Currencies
+class ItemResolvable:
+    baseName: str
+    craftable: bool
+    quality: Enity
+    tradable: bool = True
 
 
 @dataclass
 class Listing:
     id: str
-    appid: int
-    bumpedAt: str
-    listedAt: str
-    details: str
-    intent: str
     steamid: str
+    appid: int
     currencies: Currencies
-    promoted: bool
+    value: dict
     tradeOffersPreferred: bool
+    buyoutOnly: bool
+    details: str
+    listedAt: int
+    bumpedAt: int
+    intent: str
     count: int
+    status: str
+    source: str
     item: ItemDocument
+    user: dict
+    userAgent: dict = field(default_factory=dict)
 
 
 class BackpackTFException(Exception):
@@ -82,63 +88,60 @@ class NeedsAPIKey(BackpackTFException):
 
 
 class BackpackTF:
-    URL = "https://backpack.tf/api"
+    URL = "https://api.backpack.tf/api"
 
-    def __init__(self, api_key: str) -> None:
+    def __init__(
+        self, token: str, api_key: str = "", user_agent="listed with <3"
+    ) -> None:
+        self.token = token
         self.api_key = api_key
+        self.user_agent = user_agent
         self.user_token = None
+        self.schema = SchemaItemsUtils()
 
-    def _get_request(self, endpoint: str, params: dict = {}) -> dict:
-        if self.api_key:
-            params["apiKey"] = self.api_key
+        self.__headers = {"User-Agent": f"{__title__} | {self.user_agent}"}
 
-        response = requests.get(self.URL + endpoint, params=params)
-        return response.json()
-
-    def _post_request(self, endpoint: str, json: dict = {}) -> dict:
-        if self.api_key:
-            json["apiKey"] = self.api_key
-
-        response = requests.post(self.URL + endpoint, json=json)
-        return response.json()
-
-    def _delete_request(self, endpoint: str, params: dict = {}) -> dict:
-        if self.api_key:
-            params["apiKey"] = self.api_key
-
-        response = requests.delete(self.URL + endpoint, params=params)
-        return response.json()
-
-    def _patch_request(self, endpoint: str, params: dict = {}) -> dict:
-        if self.api_key:
-            params["apiKey"] = self.api_key
-
-        response = requests.patch(self.URL + endpoint, params=params)
+    def request(self, method: str, endpoint: str, params: dict = {}, **kwargs) -> dict:
+        params["token"] = self.token
+        response = requests.request(
+            method, self.URL + endpoint, params=params, headers=self.__headers, **kwargs
+        )
         return response.json()
 
     def get_listings(self, skip: int = 0, limit: int = 100) -> dict:
         return self._get_request(
             "/v2/classifieds/listings", {"skip": skip, "limit": limit}
         )
 
-    def create_listing(self, listing: ListingResolvable) -> Listing:
-        return self._post_request("/v2/classifieds/listings", asdict(listing))
-
-    def make_listing(
-        self, id: int, item: dict, details: str, currencies: dict
+    def _construct_listing_item(self, sku: str) -> dict:
+        return {
+            "baseName": self.schema.sku_to_base_name(sku),
+            "craftable": sku_is_craftable(sku),
+            "tradable": True,
+            "quality": {"id": sku_to_quality(sku)},
+        }
+
+    def _construct_listing(
+        self, sku: str, intent: str, currencies: dict, details: str, asset_id: int = 0
+    ) -> dict:
+        listing = {
+            "item": self._construct_listing_item(sku),
+            "details": details,
+            "currencies": Currencies(**currencies).__dict__,
+        }
+
+        if intent == "sell":
+            listing["id"] = asset_id
+
+        return listing
+
+    def create_listing(
+        self, sku: str, intent: str, currencies: dict, details: str, asset_id: int = 0
     ) -> Listing:
-        listing = ListingResolvable(id, item, details, Currencies(**currencies))
-        return self.create_listing(listing)
-
+        listing = self._construct_listing(sku, intent, currencies, details, asset_id)
+        response = self.request("POST", "/v2/classifieds/listings", json=listing)
 
-if __name__ == "__main__":
-    # BackpackTF.create_listing(ListingResolvable(1, 1, 1, 1, 1))
+        return Listing(**response)
 
-    # listin = ListingResolvable("1", {}, "my details", Currencies(1, 1.5))
-    # listin = ListingResolvable(
-    #     "1", {}, "my details", Currencies(**{"keys": 1, "metal": 1.5})
-    # )
-    listin = ListingResolvable("1", {}, "my details", Currencies(**{"keys": 2}))
-    print(asdict(listin))
-    # curren = Currencies(1, 1.5)
-    # print(curren.__dict__)
+    def register_user_agent(self) -> dict:
+        return self.request("POST", "/agent/pulse")

src/tf2_utils/schema.py

@@ -26,6 +26,9 @@ def defindex_to_name(self, defindex: int) -> str:
 
         return self.defindex_names.get(str(defindex), "")
 
+    def defindex_to_full_name(self, defindex: int) -> str:
+        return self.defindex_full_names.get(str(defindex), "")
+
     def name_to_defindex(self, name: str) -> int:
         if name == "Random Craft Weapon":
             return -50
@@ -142,6 +145,10 @@ def sku_to_base_name(self, sku: str) -> str:
         defindex = sku_to_defindex(sku)
         return self.defindex_to_name(defindex)
 
+    def sku_to_full_name(self, sku: str) -> str:
+        defindex = sku_to_defindex(sku)
+        return self.defindex_to_full_name(defindex)
+
     def sku_to_name(self, sku: str, use_uncraftable: bool = True) -> str:
         name = self.sku_to_base_name(sku)
         craftable = ""

src/tf2_utils/sockets.py

@@ -1,6 +1,7 @@
 from .prices_tf import PricesTF
 
 import json
+from typing import Callable, Any
 
 from websockets.sync.client import ClientConnection, connect
 
@@ -10,7 +11,7 @@ class BackpackTFSocket:
 
     def __init__(
         self,
-        callback,
+        callback: Callable[[dict | list[dict]], None],
         solo_entries: bool = True,
         headers: dict = {"batch-test": True},
         max_size: int | None = None,
@@ -58,7 +59,7 @@ class PricesTFSocket:
 
     def __init__(
         self,
-        callback,
+        callback: Callable[[dict], Any],
         settings: dict = {},
     ) -> None:
         """

tests/test_backpack_tf.py

@@ -1,8 +1,64 @@
 from unittest import TestCase
 
-from src.tf2_utils import BackpackTF
+from .config import BACKPACK_TF_API_KEY
+from src.tf2_utils import BackpackTF, Currencies
 
 
 class TestBackpackTF(TestCase):
     def setUp(cls) -> None:
-        cls.bptf = BackpackTF("test")
+        cls.bptf = BackpackTF(BACKPACK_TF_API_KEY)
+
+    def test_currencies(self):
+        self.assertEqual(Currencies(1, 1.5).__dict__, {"keys": 1, "metal": 1.5})
+        self.assertEqual(Currencies().__dict__, {"keys": 0, "metal": 0.0})
+        self.assertEqual(
+            Currencies(**{"metal": 10.55}).__dict__, {"keys": 0, "metal": 10.55}
+        )
+
+    def test_construct_listing_item(self):
+        self.assertDictEqual(
+            self.bptf._construct_listing_item("263;6"),
+            {
+                "baseName": "Ellis' Cap",
+                "craftable": True,
+                "quality": {"id": 6},
+                "tradable": True,
+            },
+        )
+
+    def test_construct_listing(self):
+        self.assertDictEqual(
+            self.bptf._construct_listing(
+                "263;6",
+                "sell",
+                {"keys": 1, "metal": 1.55},
+                "my description",
+                13201231975,
+            ),
+            {
+                "item": {
+                    "baseName": "Ellis' Cap",
+                    "craftable": True,
+                    "quality": {"id": 6},
+                    "tradable": True,
+                },
+                "currencies": {"keys": 1, "metal": 1.55},
+                "details": "my description",
+                "id": 13201231975,
+            },
+        )
+        self.assertDictEqual(
+            self.bptf._construct_listing(
+                "263;6", "buy", {"keys": 1, "metal": 1.55}, "my description"
+            ),
+            {
+                "item": {
+                    "baseName": "Ellis' Cap",
+                    "craftable": True,
+                    "quality": {"id": 6},
+                    "tradable": True,
+                },
+                "currencies": {"keys": 1, "metal": 1.55},
+                "details": "my description",
+            },
+        )