feat: add transaction export to YNAB import format

pytr/api.py

@@ -387,6 +387,13 @@ async def portfolio(self):
     async def portfolio_status(self):
         return await self.subscribe({'type': 'portfolioStatus'})
 
+    async def timeline_transactions(self, after=None):
+        sub = {"type":"timelineTransactions"}
+        if after is not None:
+            sub.update({"after": after})
+
+        return await self.subscribe(sub)
+
     async def compact_portfolio(self):
         return await self.subscribe({'type': 'compactPortfolio'})
 

pytr/main.py

@@ -16,6 +16,7 @@
 from pytr.portfolio import Portfolio
 from pytr.alarms import Alarms
 from pytr.details import Details
+from pytr.transactions import Transactions
 
 
 def get_main_parser():
@@ -155,6 +156,20 @@ def formatter(prog):
     parser_completion = parser_cmd.add_parser(
         'completion', formatter_class=argparse.ArgumentDefaultsHelpFormatter, help=info, description=info
     )
+    # transactions
+    info = 'Export all transactions to csv'
+    transactions = parser_cmd.add_parser(
+        'transactions',
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+        parents=[parser_login_args],
+        help=info,
+        description=info,
+    )
+    transactions.add_argument(
+        'output', help='Output path of CSV file', metavar='OUTPUT', type=Path)
+    transactions.add_argument(
+        '--last_days', help='Number of last days to include (use 0 get all days, defaults to 14)', metavar='DAYS', default=14, type=int
+    )
     shtab.add_argument_to(parser_completion, "shell", parent=parser)
     return parser
 
@@ -221,6 +236,12 @@ def main():
         p.get()
         if args.output is not None:
             p.portfolio_to_csv()
+    elif args.command == 'transactions':
+        Transactions(
+            login(phone_no=args.phone_no, pin=args.pin, web=not args.applogin),
+            output_path=args.output,
+            last_days=args.last_days
+        ).get()
     elif args.command == 'export_transactions':
         export_transactions(args.input, args.output, args.lang)
     elif args.version:

pytr/transactions.py

@@ -0,0 +1,48 @@
+import asyncio
+from pytr.utils import preview
+import json
+from datetime import datetime, timedelta
+
+class Transactions:
+    def __init__(self, tr,output_path, last_days):
+        self.tr = tr
+        self.output_path = output_path
+        self.last_days = last_days
+        self.transactions = []
+
+    async def loop(self):
+        recv = 0
+        await self.tr.timeline_transactions()
+        while True:
+            _subscription_id, subscription, response = await self.tr.recv()
+
+            if subscription['type'] == 'timelineTransactions':
+                self.transactions.extend(response["items"])
+
+                # Transactions in the response are ordered from newest to oldest
+                # If the oldest (= last) transaction is older than what we want, exit the loop
+                t = self.transactions[-1]
+                if datetime.fromisoformat(t['timestamp']) < datetime.now().astimezone() - timedelta(days=self.last_days):
+                    return
+
+                await self.tr.timeline_transactions(response["cursors"]["after"])
+
+            else:
+                print(f"unmatched subscription of type '{subscription['type']}':\n{preview(response)}")
+
+            if recv == 1:
+                return
+
+    def output(self):
+        transactions = [
+            t
+            for t in self.transactions
+            if datetime.fromisoformat(t['timestamp']) > datetime.now().astimezone() - timedelta(days=self.last_days)
+        ]
+
+        with open(self.output_path, mode='w', encoding='utf-8') as output_file:
+            json.dump(transactions, output_file)
+
+    def get(self):
+        asyncio.get_event_loop().run_until_complete(self.loop())
+        self.output()