1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
import requests
import json
from pathlib import Path
from datetime import datetime
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 bookmark 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, bookmark_dir: str):
"""Initializes a JForm instance.
name: Internal identifier. Must be unique.
url: URL of the API endpoint jForm exposed to creators of questionnaire.
bookmark_dir: A directory to store bookmark files, which are named under the format
"<name>.bookmark".
"""
self._name = name
self.url = url
self._bookmark_fp = Path(bookmark_dir) / (f"{name}.bookmark")
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_bookmark(self) -> int:
"""Read bookmark file and returns contents. No safeguards."""
try:
with open(self._bookmark_fp) as f:
bookmark = f.read()
f.close()
return int(bookmark)
except FileNotFoundError:
# create file
with open(self._bookmark_fp, "x") as f:
f.close()
return 0
def _write_bookmark(self, bookmark: int) -> None:
"""Write into bookmark file."""
try:
with open(self._bookmark_fp, "x") as f:
f.write(str(bookmark))
f.close()
except FileExistsError:
with open(self._bookmark_fp, "w") as f:
f.write(str(bookmark))
f.close()
def get_unread(self) -> list:
"""Get unread answers to required fields as a list of dicts, most recent last.
Keys of a dict in the list:
- name: (string) Tenant's name.
- id: (uint) Tenant's student/faculty ID.
- phone: (string) Tenant's phone number.
- key: (uint) Number of key to umbrella.
- date: (datetime.datetime) When the jForm answer sheet was submitted.
"""
bookmark = self._read_bookmark()
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 bookmark. next time, stop before this id.
latest_id = sheets[0]["id"]
self._write_bookmark(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"] <= bookmark:
# is bookmark or earlier than bookmark
found_read = True
break
ans = sheet["answers"]
unread.append(
{
"jform_name": self._name,
"name": ans[0]["answer"],
"id": ans[1]["answer"],
"phone": ans[2]["answer"],
"key": int(ans[3]["answer"]),
"date": datetime.fromisoformat(sheet["submitted_at"]),
}
)
page += 1
return unread
|
