Add support for statements (#3)

* Add prototype for document download

* Add support for statements

* Update README.md

* Bump version to 3.00

* Update EquatePlus.lua

* Add signature

Author: neatc0der

.gitignore

@@ -1,3 +1,5 @@
 .DS_Store
 credentials.txt
-equateplus_qr.png
+equateplus_qr.png
+equateplus.pckl
+documents/

EquatePlus.lua

@@ -6,7 +6,7 @@ end
 
 local baseurl=""
 local reportOnce
-local Version="2.1"
+local Version=3.00
 local CSRF_TOKEN=nil
 local CSRF2_TOKEN=nil
 local connection
@@ -378,6 +378,34 @@ function RefreshAccount (account, since)
   return {securities=securities}
 end
 
+function FetchStatements (accounts, knownIdentifiers)
+  local statements = {}
+
+  -- Load postbox page.
+  local library = JSON(connectWithCSRF("POST","https://www.equateplus.com/EquatePlusParticipant2/services/documents/library?_cId="..cId.."&_rId="..rnd(),"{\"$type\":\"Object\"}","application/json;charset=UTF-8")):dictionary()
+
+  local pattern = "(%d+)-(%d+)-(%d+)T(%d+):(%d+):(%d+)"
+  for k,document in pairs(library["documents"]) do
+    local statement = {}
+    local year, month, day, hour, minute, second = document["date"]:match(pattern)
+    statement.creationDate = os.time({year=year,month=month,day=day})
+    statement.name = document["description"]
+    statement.identifier = document["id"]
+    statement.filename = (document["description"] .. "(" .. MM.localizeDate(statement.creationDate) .. ").pdf"):gsub("/", "-")
+    if not knownIdentifiers[statement.identifier] then
+      print("Downloading statement: " .. statement.filename)
+      statement.pdf = connectWithCSRF("GET", "https://www.equateplus.com/EquatePlusParticipant2/services/statements/download?documentId="..statement.identifier.."&downloadType=inline&source=LIBRARY")
+      if startsWith(statement.pdf, "{\"$type\":\"TechnicalError\"") then
+        print("error downloading statement")
+      else
+        table.insert(statements, statement)
+      end
+    end
+  end
+
+  return {statements=statements}
+end
+
 function bugReport(status,err,v)
   if not status and reportOnce then
     reportOnce=false
@@ -389,7 +417,7 @@ end
 
 function EndSession ()
   -- Logout.
-  connectWithCSRF("GET","services/participant/logout")
+  connectWithCSRF("GET","https://www.equateplus.com/EquatePlusParticipant2/services/participant/logout")
 end
 
--- SIGNATURE: MC0CFQCAvuO8tEE66YK0ZhAm76gHE6Tw0gIUD3F98TfMYQk+RDYZYHlMFdDwSSU=
+-- SIGNATURE: MCwCFAf5dPf+zi2l1NSWRHsSzplHaJyLAhQZSM+ssqOMG1BBGvMO2OTFeWkfbA==

README.md

@@ -2,7 +2,10 @@
 
 A quick and dirty plugin for MoneyMoney for share depot. See `equateplus.py` as showcase for api calls.
 
-Plugin now supports EquateAccess App authentication.
+Plugin now supports:
+
+* EquateAccess App authentication
+* Statement download
 
 No warrenty or whatsoever!
 

_config.yml

@@ -8,19 +8,16 @@
 - equateplus_qr.png: Scan QR code with EquateAccess app
 """
 from base64 import b64decode
+from datetime import datetime
 from pathlib import Path
+from pickle import dumps, loads
 from random import randint
 from time import sleep
 
+import click
 import requests
 
 
-CREDENTIALS_PATH: Path = Path("credentials.txt")
-QR_CODE_PATH: Path = Path("equateplus_qr.png")
-
-CREDENTIALS: list[str] = CREDENTIALS_PATH.read_text().strip().split("\n")
-
-
 def debug(func):
     def wrapper(*args, **kwargs):
         print(func.__name__, end=" ", flush=True)
@@ -54,8 +51,13 @@ def get_amount(data: dict) -> float:
     return quantity["amount"]
 
 
-class Equate:
-    def __init__(self) -> None:
+class EquatePlus:
+    def __init__(self, username: str, password: str, qr_code_path: Path) -> None:
+        self.username: str = username
+        self.password: str = password
+        self.qr_code_path: Path = qr_code_path
+        self.documents_dir: Path = Path("documents")
+
         self.session: requests.Session = requests.Session()
         self.cookies: dict[str, str] = {}
         self.csrf: str | None = None
@@ -105,7 +107,7 @@ def send_user(self) -> bool:
             },
             data={
                 "csrfpId": self.csrf,
-                "isiwebuserid": CREDENTIALS[0],
+                "isiwebuserid": self.username,
                 "result": "Continue Login",
             },
             headers={
@@ -119,8 +121,8 @@ def send_credentials(self) -> bool:
             "https://www.equateplus.com/EquatePlusParticipant2/?login",
             data={
                 "csrfpId": self.csrf,
-                "isiwebuserid": CREDENTIALS[0],
-                "isiwebpasswd": CREDENTIALS[1],
+                "isiwebuserid": self.username,
+                "isiwebpasswd": self.password,
                 "result": "Continue",
             },
             headers={
@@ -134,7 +136,7 @@ def request_devices(self) -> bool:
         response: requests.Response = self.session.post(
             "https://www.equateplus.com/EquatePlusParticipant2/?login",
             data={
-                "isiwebuserid": CREDENTIALS[0],
+                "isiwebuserid": self.username,
                 "isiwebpasswd": "null",
                 "result": "null",
             },
@@ -164,7 +166,7 @@ def request_qr_code(self) -> bool:
 
         try:
             data = response.json()
-            QR_CODE_PATH.write_bytes(b64decode(data["dispatcherInformation"]["response"] + "=="))
+            self.qr_code_path.write_bytes(b64decode(data["dispatcherInformation"]["response"] + "=="))
             self.session_id = data["sessionId"]
 
             return True
@@ -253,30 +255,116 @@ def get_plan_details(self) -> bool:
         print(" ", end="", flush=True)
         return True
 
+    @debug
+    def get_documents(self) -> bool:
+        response: requests.Response = self.session.post(
+            "https://www.equateplus.com/EquatePlusParticipant2/services/documents/library",
+            params=self.ids(),
+            json={
+                "$type": "Object",
+            },
+        )
+        try:
+            for document in response.json()["documents"]:
+                date: str = datetime.fromisoformat(document["date"]).strftime("%d.%m.%Y")
+                file_name: str = document["description"] + f" ({date}).pdf"
+                file_path: Path = self.documents_dir / file_name.replace("/", "-")
+
+                if self.download_document(document["id"], file_path):
+                    print(".", end="", flush=True)
+                else:
+                    print("x", end="", flush=True)
+
+        except (KeyError, IndexError):
+            return False
+        
+        print(" ", end="", flush=True)
+        return True
+
+    def download_document(self, document_id: str, file_path: Path) -> bool:
+        response: requests.Response = self.session.get(
+            "https://www.equateplus.com/EquatePlusParticipant2/services/statements/download",
+            params={
+                "documentId": document_id,
+                "downloadType": "inline",
+                "source": "LIBRARY",
+            },
+        )
+        if response.content.startswith(b"{\"$type\":\"TechnicalError\""):
+            return False
+
+        try:
+            file_path.parent.mkdir(parents=True, exist_ok=True)
+            file_path.write_bytes(response.content)
+        
+        except:
+            return False
+
+        return True
+
     @debug
     def logout(self) -> None:
         self.session.get(
             "https://www.equateplus.com/EquatePlusParticipant2/services/participant/logout",
         )
 
 
-def main() -> None:
-    e = Equate()
+@click.command()
+@click.option("--credentials-path", type=Path, default=Path("credentials.txt"), show_default=True)
+@click.option("--qr-code-path", type=Path, default=Path("equateplus_qr.png"), show_default=True)
+@click.option("--pickle-path", type=Path, default=Path("equateplus.pckl"), show_default=True)
+@click.option("--download-documents", is_flag=True, help="download documents")
+@click.option(
+    "--store",
+    is_flag=True,
+    help="serialize EquatePlus object to file (and don't logout)",
+)
+@click.option(
+    "--restore",
+    is_flag=True,
+    help="deserialize EquatePlus object from file (and don't login)",
+)
+@click.option("--no-logout", is_flag=True, help="don't logout")
+def main(credentials_path: Path, qr_code_path: Path, pickle_path: Path, download_documents: bool, store: bool, restore: bool, no_logout: bool) -> None:
+    credentials: list[str] = credentials_path.read_text().strip().split("\n")
+    equateplus = EquatePlus(username=credentials[0], password=credentials[1], qr_code_path=qr_code_path)
+    login_successful: bool = False
     try:
-        e.initialize() and \
-        e.send_user() and \
-        e.send_credentials() and \
-        e.request_devices() and \
-        e.request_qr_code() and \
-        e.verify_qr_code() and \
-        e.complete_login() and \
-        e.get_plan_summary() and \
-        e.get_plan_details()
+        if restore:
+            equateplus = loads(pickle_path.read_bytes())
+            print("equateplus restored")
+            login_successful = True
+
+        else:
+            login_successful = equateplus.initialize() and \
+            equateplus.send_user() and \
+            equateplus.send_credentials() and \
+            equateplus.request_devices() and \
+            equateplus.request_qr_code() and \
+            equateplus.verify_qr_code() and \
+            equateplus.complete_login()
+        
+        if store:
+            if login_successful:
+                pickle_path.write_bytes(dumps(equateplus))
+                print("equateplus stored")
+            else:
+                print("equateplus not stored - login failed")
+
+        login_successful and \
+        equateplus.get_plan_summary() and \
+        equateplus.get_plan_details() and \
+        download_documents and \
+        equateplus.get_documents()
+
     finally:
-        e.logout()
+        if not store and not no_logout:
+            equateplus.logout()
+        else:
+            print("equateplus not logged out")
 
     from pprint import pprint
-    pprint(e.securities)
+    pprint(equateplus.securities)
 
 
 if __name__ == "__main__":

equateplus.py

@@ -0,0 +1,2 @@
+click
+requests