Basic interfacing with jForm API

Retrieve unread answer sheets from jForm.

2 files changed, 113 insertions, 0 deletions

diff --git a/jimbrella/config.py b/jimbrella/config.py
index 26111f1..1bde65b 100644
--- a/jimbrella/config.py
+++ b/jimbrella/config.py
@@ -1,3 +1,12 @@
+# jForm
+JFORM_TAKEAWAY_URL = (
+    "https://wj.sjtu.edu.cn/api/v1/public/export/900e25bf6a039da04da7c5165a7cd41a/json"
+)
+# JFORM_GIVEBACK_URL= (
+#     "https://wj.sjtu.edu.cn/api/v1/public/export/900e25bf6a039da04da7c5165a7cd41a/json"
+# )
+JFORM_CHECKPOINTS_PATH = "/home/fkfd/p/jimbrella/jform"
+
 # database
 DATABASE_PATH = "/home/fkfd/p/jimbrella/jimbrella.db.csv"
 
diff --git a/jimbrella/jform.py b/jimbrella/jform.py
new file mode 100644
index 0000000..dff6654
--- /dev/null
+++ b/jimbrella/jform.py
@@ -0,0 +1,104 @@
+import requests
+import json
+
+
+class JForm:
+    """Retrieves results from one jForm questionnaire.
+
+    In this context, "JForm" refers to this class, and "jForm", the online service also known as
+    SJTU Questionnaires, hosted on https://wj.sjtu.edu.cn/.
+
+    JImbrella's database relies on JForm to sync its database against jForm's.
+
+    Each JForm object requires a checkpoint file in which to store the id of the latest answer
+    sheet it has read. Each time the API detects the presence of new answer sheets, the file will
+    be overwritten.
+    """
+
+    def __init__(self, name: str, url: str, checkpoint_fp: str):
+        self._name = name  # internal identifier
+        self.url = url
+        self._checkpoint_fp = checkpoint_fp
+
+    def _get(self, page=1) -> dict:
+        """Internal method to HTTP GET the API in JSON format."""
+        resp = requests.get(
+            self.url,
+            params={
+                "params": json.dumps(
+                    {
+                        "current": page,
+                        "pageSize": 100,
+                    }
+                ),
+                "sort": json.dumps({"id": "desc"}),
+            },
+        )
+        return resp
+
+    def _read_checkpoint(self) -> int:
+        """Read checkpoint file and returns contents. No safeguards."""
+        try:
+            with open(self._checkpoint_fp) as f:
+                checkpoint = f.read()
+                f.close()
+                return int(checkpoint)
+        except FileNotFoundError:
+            return 0
+
+    def _write_checkpoint(self, checkpoint: int) -> None:
+        """Write into checkpoint file."""
+        try:
+            with open(self._checkpoint_fp, "x") as f:
+                f.write(str(checkpoint))
+                f.close()
+        except FileExistsError:
+            with open(self._checkpoint_fp, "w") as f:
+                f.write(str(checkpoint))
+                f.close()
+
+    def get_unread(self) -> list:
+        """Get unread answers to required fields as a list of dicts, most recent last."""
+        checkpoint = self._read_checkpoint()
+        unread = []
+        latest_id = 0
+        page = 1
+        found_read = False
+        while not found_read:
+            try:
+                resp = self._get(page=page)
+            except:
+                break  # quietly abort
+            if resp.status_code != 200:
+                break
+
+            sheets = resp.json()["data"]["rows"]
+            if not latest_id:
+                # the first page of sheets we have retrieved this run.
+                # on this page, the first answer sheet is the latest.
+                # update checkpoint. next time, stop before this id.
+                latest_id = sheets[0]["id"]
+                self._write_checkpoint(latest_id)
+
+            if not sheets:
+                # somehow jForm doesn't respond with a 404
+                # when we exceed the total pages of a questionnaire
+                # instead it just returns 200 along with an empty data field
+                break
+            for sheet in sheets:
+                if sheet["id"] <= checkpoint:
+                    # is checkpoint or earlier than checkpoint
+                    found_read = True
+                    break
+                ans = sheet["answers"]
+                unread.append(
+                    {
+                        "name": ans[0]["answer"],
+                        "id": ans[1]["answer"],
+                        "phone": ans[2]["answer"],
+                        "key": ans[3]["answer"],
+                    }
+                )
+            page += 1
+
+        return unread