diff --git a/amazonorders/cli.py b/amazonorders/cli.py index a75d0d0..5bb8384 100644 --- a/amazonorders/cli.py +++ b/amazonorders/cli.py @@ -18,6 +18,7 @@ from amazonorders.exception import AmazonOrdersError, AmazonOrdersAuthError from amazonorders.orders import AmazonOrders from amazonorders.session import AmazonSession, IODefault +from amazonorders.transactions import AmazonTransactions logger = logging.getLogger("amazonorders") @@ -205,7 +206,7 @@ def transactions(ctx: Context, **kwargs: Any): ) click.echo("Info: Fetching transaction history, this might take a minute ...") - amazon_transactions = AmazonOrders(amazon_session, config=ctx.obj["conf"]) + amazon_transactions = AmazonTransactions(amazon_session, config=ctx.obj["conf"]) transactions = amazon_transactions.get_transactions(days=days) diff --git a/amazonorders/entity/transaction.py b/amazonorders/entity/transaction.py index 6bf6fc5..b32f212 100644 --- a/amazonorders/entity/transaction.py +++ b/amazonorders/entity/transaction.py @@ -18,7 +18,9 @@ class Transaction(Parsable): An Amazon Transaction """ - def __init__(self, parsed: Tag, config: AmazonOrdersConfig, completed_date: date) -> None: + def __init__( + self, parsed: Tag, config: AmazonOrdersConfig, completed_date: date + ) -> None: super().__init__(parsed, config) #: The Transaction completed date. @@ -33,6 +35,8 @@ def __init__(self, parsed: Tag, config: AmazonOrdersConfig, completed_date: date self.is_refund: bool = self.grand_total > 0 #: The Transaction order number. self.order_number: str = self.safe_parse(self._parse_order_number) + #: The Transaction order details link. + self.order_details_link: str = self.safe_parse(self._parse_order_details_link) #: The Transaction seller name. self.seller: str = self.safe_simple_parse( selector=self.config.selectors.FIELD_TRANSACTION_SELLER_NAME_SELECTOR @@ -45,14 +49,30 @@ def __str__(self) -> str: # pragma: no cover return f"Transaction {self.completed_date}: Order #{self.order_number}, Grand Total {self.grand_total}" def _parse_grand_total(self) -> float: - value = self.simple_parse(self.config.selectors.FIELD_TRANSACTION_GRAND_TOTAL_SELECTOR) + value = self.simple_parse( + self.config.selectors.FIELD_TRANSACTION_GRAND_TOTAL_SELECTOR + ) value = self.to_currency(value) return value def _parse_order_number(self) -> str: - value = self.simple_parse(self.config.selectors.FIELD_TRANSACTION_ORDER_NUMBER_SELECTOR) + value = self.simple_parse( + self.config.selectors.FIELD_TRANSACTION_ORDER_NUMBER_SELECTOR + ) match = re.match(".*#([0-9-]+)$", value) value = match.group(1) if match else "" return value + + def _parse_order_details_link(self) -> str: + value = self.simple_parse( + self.config.selectors.FIELD_TRANSACTION_ORDER_LINK_SELECTOR, link=True + ) + + if not value and self.order_number: + value = ( + f"{self.config.constants.ORDER_DETAILS_URL}?orderID={self.order_number}" + ) + + return value diff --git a/amazonorders/lib/__init__.py b/amazonorders/lib/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/amazonorders/lib/transactions.py b/amazonorders/lib/transactions.py deleted file mode 100644 index a65077b..0000000 --- a/amazonorders/lib/transactions.py +++ /dev/null @@ -1,62 +0,0 @@ -__copyright__ = "Copyright (c) 2024 Jeff Sawatzky" -__license__ = "MIT" - -import datetime -from typing import Dict, List, Optional, Tuple - -from bs4 import Tag - -from amazonorders.conf import AmazonOrdersConfig -from amazonorders.entity.transaction import Transaction - - -def parse_transaction_form_tag( - form_tag: Tag, config: AmazonOrdersConfig -) -> Tuple[List[Transaction], Optional[str], Optional[Dict[str, str]]]: - transactions = [] - date_container_tags = form_tag.select( - config.selectors.TRANSACTION_DATE_CONTAINERS_SELECTOR - ) - for date_container_tag in date_container_tags: - date_tag = date_container_tag.select_one( - config.selectors.FIELD_TRANSACTION_COMPLETED_DATE_SELECTOR - ) - assert date_tag is not None - - date_str = date_tag.text - date = datetime.datetime.strptime( - date_str, config.constants.TRANSACTION_DATE_FORMAT - ).date() - - transactions_container_tag = date_container_tag.find_next_sibling( - config.selectors.TRANSACTIONS_CONTAINER_SELECTOR - ) - assert isinstance(transactions_container_tag, Tag) - - transaction_tags = transactions_container_tag.select( - config.selectors.TRANSACTIONS_SELECTOR - ) - for transaction_tag in transaction_tags: - transaction = Transaction(transaction_tag, config, date) - transactions.append(transaction) - - form_state_input = form_tag.select_one( - config.selectors.TRANSACTIONS_NEXT_PAGE_INPUT_STATE_SELECTOR - ) - form_ie_input = form_tag.select_one( - config.selectors.TRANSACTIONS_NEXT_PAGE_INPUT_IE_SELECTOR - ) - next_page_input = form_tag.select_one( - config.selectors.TRANSACTIONS_NEXT_PAGE_INPUT_SELECTOR - ) - if not next_page_input or not form_state_input or not form_ie_input: - return (transactions, None, None) - - next_page_post_url = str(form_tag["action"]) - next_page_post_data = { - "ppw-widgetState": str(form_state_input["value"]), - "ie": str(form_ie_input["value"]), - str(next_page_input["name"]): "", - } - - return (transactions, next_page_post_url, next_page_post_data) diff --git a/amazonorders/orders.py b/amazonorders/orders.py index b52adb1..8e24012 100644 --- a/amazonorders/orders.py +++ b/amazonorders/orders.py @@ -8,9 +8,7 @@ from amazonorders import util from amazonorders.conf import AmazonOrdersConfig from amazonorders.entity.order import Order -from amazonorders.entity.transaction import Transaction from amazonorders.exception import AmazonOrdersError, AmazonOrdersNotFoundError -from amazonorders.lib.transactions import parse_transaction_form_tag from amazonorders.session import AmazonSession logger = logging.getLogger(__name__) @@ -125,48 +123,3 @@ def get_order(self, order: Order = self.config.order_cls(order_details_tag, self.config, full_details=True) return order - - def get_transactions( - self, - days: int = 365, - ) -> List[Transaction]: - """ - Get the Amazon transactions for the given number of days. - - :param days: The number of days worth of transactions to get. - :return: A list of the requested Transactions. - """ - if not self.amazon_session.is_authenticated: - raise AmazonOrdersError("Call AmazonSession.login() to authenticate first.") - - min_date = datetime.date.today() - datetime.timedelta(days=days) - - self.amazon_session.get(self.config.constants.TRANSACTION_HISTORY_LANDING_URL) - assert self.amazon_session.last_response_parsed - - form_tag = self.amazon_session.last_response_parsed.select_one( - self.config.selectors.TRANSACTION_HISTORY_FORM_SELECTOR - ) - - transactions: List[Transaction] = [] - while form_tag: - loaded_transactions, next_page_post_url, next_page_post_data = ( - parse_transaction_form_tag(form_tag, self.config) - ) - for transaction in loaded_transactions: - if transaction.completed_date >= min_date: - transactions.append(transaction) - else: - return transactions - - if next_page_post_url is None: - return transactions - - self.amazon_session.post(next_page_post_url, data=next_page_post_data) - assert self.amazon_session.last_response_parsed - - form_tag = self.amazon_session.last_response_parsed.select_one( - self.config.selectors.TRANSACTION_HISTORY_FORM_SELECTOR - ) - - return transactions diff --git a/amazonorders/selectors.py b/amazonorders/selectors.py index 28475a1..33d641e 100644 --- a/amazonorders/selectors.py +++ b/amazonorders/selectors.py @@ -1,8 +1,6 @@ __copyright__ = "Copyright (c) 2024 Alex Laird" __license__ = "MIT" -from amazonorders.constants import Constants - class Selectors: """ @@ -117,7 +115,7 @@ class Selectors: # CSS selectors for Transaction fields ##################################### - TRANSACTION_HISTORY_FORM_SELECTOR = f"form[method='post'][action$='{Constants.TRANSACTION_HISTORY_LANDING_ROUTE}']" + TRANSACTION_HISTORY_FORM_SELECTOR = "form:has(input[name='ppw-widgetState'])" TRANSACTION_DATE_CONTAINERS_SELECTOR = "div.apx-transaction-date-container" TRANSACTIONS_CONTAINER_SELECTOR = "div" TRANSACTIONS_SELECTOR = "div.apx-transactions-line-item-component-container" @@ -138,6 +136,9 @@ class Selectors: FIELD_TRANSACTION_ORDER_NUMBER_SELECTOR = ( "div.apx-transactions-line-item-component-container > div:nth-child(2) a.a-link-normal" ) + FIELD_TRANSACTION_ORDER_LINK_SELECTOR = ( + "div.apx-transactions-line-item-component-container > div:nth-child(2) a.a-link-normal" + ) FIELD_TRANSACTION_SELLER_NAME_SELECTOR = ( "div.apx-transactions-line-item-component-container > div:nth-child(3) span.a-size-base" ) diff --git a/amazonorders/transactions.py b/amazonorders/transactions.py new file mode 100644 index 0000000..1139a78 --- /dev/null +++ b/amazonorders/transactions.py @@ -0,0 +1,148 @@ +__copyright__ = "Copyright (c) 2024 Jeff Sawatzky" +__license__ = "MIT" + +import datetime +import logging +from typing import Dict, List, Optional, Tuple + +from bs4 import Tag + +from amazonorders.conf import AmazonOrdersConfig +from amazonorders.entity.transaction import Transaction +from amazonorders.exception import AmazonOrdersError +from amazonorders.session import AmazonSession + +logger = logging.getLogger(__name__) + + +def _parse_transaction_form_tag( + form_tag: Tag, config: AmazonOrdersConfig +) -> Tuple[List[Transaction], Optional[str], Optional[Dict[str, str]]]: + transactions = [] + date_container_tags = form_tag.select( + config.selectors.TRANSACTION_DATE_CONTAINERS_SELECTOR + ) + for date_container_tag in date_container_tags: + date_tag = date_container_tag.select_one( + config.selectors.FIELD_TRANSACTION_COMPLETED_DATE_SELECTOR + ) + if not date_tag: + logger.warning("Could not find date tag in transaction form.") + continue + + date_str = date_tag.text + date = datetime.datetime.strptime( + date_str, config.constants.TRANSACTION_DATE_FORMAT + ).date() + + transactions_container_tag = date_container_tag.find_next_sibling( + config.selectors.TRANSACTIONS_CONTAINER_SELECTOR + ) + if not isinstance(transactions_container_tag, Tag): + logger.warning( + "Could not find transactions container tag in transaction form." + ) + continue + + transaction_tags = transactions_container_tag.select( + config.selectors.TRANSACTIONS_SELECTOR + ) + for transaction_tag in transaction_tags: + transaction = Transaction(transaction_tag, config, date) + transactions.append(transaction) + + form_state_input = form_tag.select_one( + config.selectors.TRANSACTIONS_NEXT_PAGE_INPUT_STATE_SELECTOR + ) + form_ie_input = form_tag.select_one( + config.selectors.TRANSACTIONS_NEXT_PAGE_INPUT_IE_SELECTOR + ) + next_page_input = form_tag.select_one( + config.selectors.TRANSACTIONS_NEXT_PAGE_INPUT_SELECTOR + ) + if not next_page_input or not form_state_input or not form_ie_input: + return (transactions, None, None) + + next_page_post_url = str(form_tag["action"]) + next_page_post_data = { + "ppw-widgetState": str(form_state_input["value"]), + "ie": str(form_ie_input["value"]), + str(next_page_input["name"]): "", + } + + return (transactions, next_page_post_url, next_page_post_data) + + +class AmazonTransactions: + """ + Using an authenticated :class:`~amazonorders.session.AmazonSession`, can be used to query Amazon + for Transaction details and history. + """ + + def __init__( + self, + amazon_session: AmazonSession, + debug: Optional[bool] = None, + config: Optional[AmazonOrdersConfig] = None, + ) -> None: + if not debug: + debug = amazon_session.debug + if not config: + config = amazon_session.config + + #: The AmazonSession to use for requests. + self.amazon_session: AmazonSession = amazon_session + #: The AmazonOrdersConfig to use. + self.config: AmazonOrdersConfig = config + + #: Set logger ``DEBUG`` and send output to ``stderr``. + self.debug: bool = debug + if self.debug: + logger.setLevel(logging.DEBUG) + + def get_transactions( + self, + days: int = 365, + ) -> List[Transaction]: + """ + Get the Amazon transactions for the given number of days. + + :param days: The number of days worth of transactions to get. + :return: A list of the requested Transactions. + """ + if not self.amazon_session.is_authenticated: + raise AmazonOrdersError("Call AmazonSession.login() to authenticate first.") + + min_date = datetime.date.today() - datetime.timedelta(days=days) + + self.amazon_session.get(self.config.constants.TRANSACTION_HISTORY_LANDING_URL) + if not self.amazon_session.last_response_parsed: + raise AmazonOrdersError("Could not get transaction history landing page.") + + form_tag = self.amazon_session.last_response_parsed.select_one( + self.config.selectors.TRANSACTION_HISTORY_FORM_SELECTOR + ) + + transactions: List[Transaction] = [] + while form_tag: + loaded_transactions, next_page_post_url, next_page_post_data = ( + _parse_transaction_form_tag(form_tag, self.config) + ) + for transaction in loaded_transactions: + if transaction.completed_date >= min_date: + transactions.append(transaction) + else: + return transactions + + if next_page_post_url is None: + return transactions + + self.amazon_session.post(next_page_post_url, data=next_page_post_data) + if not self.amazon_session.last_response_parsed: + raise AmazonOrdersError("Could not get next transaction history page.") + + form_tag = self.amazon_session.last_response_parsed.select_one( + self.config.selectors.TRANSACTION_HISTORY_FORM_SELECTOR + ) + + return transactions diff --git a/tests/entity/test_transaction.py b/tests/entity/test_transaction.py index 57a74e3..3f8b2ec 100644 --- a/tests/entity/test_transaction.py +++ b/tests/entity/test_transaction.py @@ -48,6 +48,7 @@ def test_parse(self): self.assertEqual(transaction.completed_date, date(2024, 1, 1)) self.assertEqual(transaction.payment_method, "My Payment Method") self.assertEqual(transaction.order_number, "123-4567890-1234567") + self.assertEqual(transaction.order_details_link, "https://www.amazon.com/gp/css/summary/edit.html?orderID=123-4567890-1234567") # noqa self.assertEqual(transaction.seller, "AMZN Mktp COM") self.assertEqual(transaction.grand_total, -12.34) self.assertEqual(transaction.is_refund, False) @@ -94,6 +95,7 @@ def test_parse_refund(self): self.assertEqual(transaction.completed_date, date(2024, 1, 1)) self.assertEqual(transaction.payment_method, "My Payment Method") self.assertEqual(transaction.order_number, "123-4567890-1234567") + self.assertEqual(transaction.order_details_link, "https://www.amazon.com/gp/css/summary/edit.html?orderID=123-4567890-1234567") # noqa self.assertEqual(transaction.seller, "AMZN Mktp COM") self.assertEqual(transaction.grand_total, 12.34) self.assertEqual(transaction.is_refund, True) diff --git a/tests/lib/__init__.py b/tests/lib/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/test_orders.py b/tests/test_orders.py index 700f348..ee81a3f 100644 --- a/tests/test_orders.py +++ b/tests/test_orders.py @@ -5,7 +5,6 @@ import unittest import responses -from freezegun import freeze_time from amazonorders.exception import AmazonOrdersError from amazonorders.orders import AmazonOrders @@ -379,23 +378,3 @@ def test_temp_order_details_file(self): self.assertIsNotNone(order.items[0].link) self.assertIsNotNone(order.items[0].price) self.assertTrue(len(order.shipments) > 0) - - @freeze_time("2024-10-11") - @responses.activate - def test_transactions_command(self): - # GIVEN - days = 1 - self.amazon_session.is_authenticated = True - with open(os.path.join(self.RESOURCES_DIR, "get-transactions.html"), "r", encoding="utf-8") as f: - responses.add( - responses.GET, - f"{self.test_config.constants.TRANSACTION_HISTORY_LANDING_URL}", - body=f.read(), - status=200, - ) - - # WHEN - transactions = self.amazon_orders.get_transactions(days=days) - - # THEN - self.assertEqual(1, len(transactions)) diff --git a/tests/lib/test_transactions.py b/tests/test_transactions.py similarity index 79% rename from tests/lib/test_transactions.py rename to tests/test_transactions.py index d40ec14..21cdde5 100644 --- a/tests/lib/test_transactions.py +++ b/tests/test_transactions.py @@ -1,14 +1,57 @@ __copyright__ = "Copyright (c) 2024 Jeff Sawatzky" __license__ = "MIT" +import os +import responses from bs4 import BeautifulSoup +from freezegun import freeze_time -from amazonorders.lib.transactions import parse_transaction_form_tag +from amazonorders.session import AmazonSession +from amazonorders.transactions import AmazonTransactions, _parse_transaction_form_tag from tests.unittestcase import UnitTestCase -class TestTransactions(UnitTestCase): +class TestOrders(UnitTestCase): + temp_order_history_file_path = os.path.join( + os.path.abspath(os.path.dirname(__file__)), "output", "temp-order-history.html" + ) + temp_order_details_file_path = os.path.join( + os.path.abspath(os.path.dirname(__file__)), "output", "temp-order-details.html" + ) + + def setUp(self): + super().setUp() + + self.amazon_session = AmazonSession( + "some-username", "some-password", config=self.test_config + ) + + self.amazon_transactions = AmazonTransactions(self.amazon_session) + + @freeze_time("2024-10-11") + @responses.activate + def test_transactions_command(self): + # GIVEN + days = 1 + self.amazon_session.is_authenticated = True + with open( + os.path.join(self.RESOURCES_DIR, "get-transactions.html"), + "r", + encoding="utf-8", + ) as f: + responses.add( + responses.GET, + f"{self.test_config.constants.TRANSACTION_HISTORY_LANDING_URL}", + body=f.read(), + status=200, + ) + + # WHEN + transactions = self.amazon_transactions.get_transactions(days=days) + + # THEN + self.assertEqual(1, len(transactions)) def test_parse_transaction_form_tag(self): # GIVEN @@ -16,11 +59,15 @@ def test_parse_transaction_form_tag(self): form_tag = parsed.select_one("form") # WHEN - transactions, next_page_url, next_page_data = parse_transaction_form_tag(form_tag, self.test_config) + transactions, next_page_url, next_page_data = _parse_transaction_form_tag( + form_tag, self.test_config + ) # THEN self.assertEqual(len(transactions), 2) - self.assertEqual(next_page_url, "https://www.amazon.com:443/cpe/yourpayments/transactions") + self.assertEqual( + next_page_url, "https://www.amazon.com:443/cpe/yourpayments/transactions" + ) self.assertEqual( next_page_data, {