78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
from dataclasses import dataclass
|
|
import datetime
|
|
import os
|
|
import re
|
|
|
|
from amazonorders.conf import AmazonOrdersConfig
|
|
from amazonorders.orders import AmazonOrders
|
|
from amazonorders.transactions import AmazonTransactions
|
|
from amazonorders.session import AmazonSession
|
|
from amazonorders.entity.order import Order
|
|
from amazonorders.entity.transaction import Transaction
|
|
import xdg_base_dirs
|
|
|
|
|
|
@dataclass
|
|
class AmazonAccount:
|
|
email: str
|
|
password: str
|
|
|
|
|
|
def get_accounts(env: os._Environ | dict[str, str] = os.environ) -> list[AmazonAccount]:
|
|
accts = []
|
|
for k, v in env.items():
|
|
if k == 'AMAZON_EMAIL':
|
|
pwd = env['AMAZON_PASSWORD']
|
|
acct = AmazonAccount(email=v, password=pwd)
|
|
accts.append(acct)
|
|
elif m := re.match(r'AMAZON_EMAIL_(\d+)', k):
|
|
idx = m.group(1)
|
|
pwd = env[f'AMAZON_PASSWORD_{idx}']
|
|
acct = AmazonAccount(email=v, password=pwd)
|
|
accts.append(acct)
|
|
|
|
return accts
|
|
|
|
|
|
class Session:
|
|
orders: AmazonOrders
|
|
transactions: AmazonTransactions
|
|
|
|
def __init__(self, acct: AmazonAccount):
|
|
self.acct = acct
|
|
self._cookie_jar_path = xdg_base_dirs.xdg_data_home() / f'catnab/cookies/{self.acct.email}.json'
|
|
|
|
def __enter__(self):
|
|
config = AmazonOrdersConfig(data={'cookie_jar_path': self._cookie_jar_path})
|
|
self.amazon_session = AmazonSession(
|
|
username=self.acct.email,
|
|
password=self.acct.password,
|
|
config=config
|
|
)
|
|
self.amazon_session.login()
|
|
|
|
self.orders = AmazonOrders(self.amazon_session)
|
|
self.transactions = AmazonTransactions(self.amazon_session)
|
|
|
|
return self
|
|
|
|
def __exit__(self, *exc_info):
|
|
pass
|
|
|
|
def list_transactions(self, days: int) -> list[Transaction]:
|
|
return self.transactions.get_transactions(days)
|
|
|
|
def get_order(self, order_number: str) -> Order:
|
|
return self.orders.get_order(order_number)
|
|
|
|
def get_orders(self, days: int) -> list[Order]:
|
|
today = datetime.datetime.today()
|
|
since_date = today - datetime.timedelta(days=days)
|
|
years = range(since_date.year, today.year + 1)
|
|
|
|
orders = []
|
|
for year in years:
|
|
orders += self.orders.get_order_history(year=year)
|
|
|
|
return orders
|