fix(knesset_committee_decisions): paginate OData responses

datapackage_pipelines_budgetkey/pipelines/knesset/knesset_committee_decisions.py

@@ -6,41 +6,70 @@
 OUTPUT_PATH = '/var/datapackages/knesset/knesset_committee_decisions'
 os.makedirs(OUTPUT_PATH, exist_ok=True)
 
+
+def _odata_paged(url):
+    """Iterate every record from an OData v4 endpoint.
+
+    Knesset's OData v4 server returns at most 100 rows per response and
+    advertises the next page in `@odata.nextLink`. The previous version of
+    this pipeline ignored that link and consumed only the first 100 rows
+    of the global `KNS_DocumentCommitteeSession` query — which, ordered by
+    `Id`, are all from Knesset 18/20 (2016). The result was an `index.csv`
+    that froze in 2016 even though the live OData store now has 12.5M+
+    matching documents.
+    """
+    while url:
+        resp = requests.get(url).json()
+        for item in resp.get('value', []):
+            yield item
+        url = resp.get('@odata.nextLink')
+
+
 def flow(*_):
     out = []
     downloaded = 0
-    committees = requests.get('https://knesset.gov.il/OdataV4/ParliamentInfo/KNS_Committee?$filter=CommitteeTypeID%20eq%2070&$orderby=Name').json()
     committees = [
-        dict(
-            id=x['Id'],
-            knesset_num=x['KnessetNum'],
+        dict(id=x['Id'], knesset_num=x['KnessetNum'])
+        for x in _odata_paged(
+            'https://knesset.gov.il/OdataV4/ParliamentInfo/KNS_Committee'
+            '?$filter=CommitteeTypeID%20eq%2070&$orderby=Name'
         )
-        for x in committees['value']
     ]
     print(f'GOT {len(committees)} committees')
-    documents = requests.get(f'https://knesset.gov.il/OdataV4/ParliamentInfo/KNS_DocumentCommitteeSession?$filter=GroupTypeID%20eq%20106&$orderby=Id').json()
-    print(f'GOT {len(documents["value"])} documents')
+
     for committee in committees:
-        sessions = requests.get(f'https://knesset.gov.il/OdataV4/ParliamentInfo/KNS_CommitteeSession?$filter=CommitteeID%20eq%20{committee["id"]}&$orderby=Id').json()
-        print(f'GOT {len(sessions["value"])} sessions for committee {committee["id"]}')
-        for session in sessions['value']:
-            sessionID = session['Id']
-            # documents = requests.get(f'https://knesset.gov.il/OdataV4/ParliamentInfo/KNS_DocumentCommitteeSession?$filter=GroupTypeID%20eq%20106&$orderby=Id').json()
-            for document in documents['value']:
-                if document['ApplicationDesc'].lower() == 'pdf' and document['CommitteeSessionID'] == sessionID:
-                    doc = dict(
-                        url=document['FilePath'],
-                        filename=f'{document["Id"]}.pdf',
-                        date=document['LastUpdatedDate'],
-                        knesset_num=committee['knesset_num']
-                    )
-                    out.append(doc)
-                    outpath = os.path.join(OUTPUT_PATH, doc['filename'])
-                    if not os.path.exists(outpath):
-                        document = requests.get(doc['url'], stream=True)
-                        with open(outpath, 'wb') as o:
-                            shutil.copyfileobj(document.raw, o)
-                        downloaded += 1
+        sessions = list(_odata_paged(
+            'https://knesset.gov.il/OdataV4/ParliamentInfo/KNS_CommitteeSession'
+            f'?$filter=CommitteeID%20eq%20{committee["id"]}&$orderby=Id'
+        ))
+        print(f'GOT {len(sessions)} sessions for committee {committee["id"]}')
+        for session in sessions:
+            session_id = session['Id']
+            # Per-session, paginated document fetch. Pushing the
+            # `CommitteeSessionID eq …` predicate into OData keeps each
+            # response small (typically 0–3 PDFs per session) and lets us
+            # walk the entire history correctly.
+            documents = _odata_paged(
+                'https://knesset.gov.il/OdataV4/ParliamentInfo/KNS_DocumentCommitteeSession'
+                f'?$filter=CommitteeSessionID%20eq%20{session_id}'
+                '%20and%20GroupTypeID%20eq%20106&$orderby=Id'
+            )
+            for document in documents:
+                if document.get('ApplicationDesc', '').lower() != 'pdf':
+                    continue
+                doc = dict(
+                    url=document['FilePath'],
+                    filename=f'{document["Id"]}.pdf',
+                    date=document['LastUpdatedDate'],
+                    knesset_num=committee['knesset_num'],
+                )
+                out.append(doc)
+                outpath = os.path.join(OUTPUT_PATH, doc['filename'])
+                if not os.path.exists(outpath):
+                    pdf_resp = requests.get(doc['url'], stream=True)
+                    with open(outpath, 'wb') as o:
+                        shutil.copyfileobj(pdf_resp.raw, o)
+                    downloaded += 1
 
         print(f'DOWNLOADED {downloaded} out of {len(out)} total documents (now in committee {committee["id"]})')
 
@@ -52,4 +81,4 @@ def flow(*_):
 
 
 if __name__ == "__main__":
-    flow()
+    flow()