From 6c47eb1ddeb03ed2034fa298e2b673b9379924b7 Mon Sep 17 00:00:00 2001 From: Amir Wilf Date: Sat, 25 Apr 2026 16:53:04 +0300 Subject: [PATCH] fix(knesset_committee_decisions): paginate OData responses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pipeline fetched the global KNS_DocumentCommitteeSession query in one shot and ignored the @odata.nextLink. OData v4 on knesset.gov.il returns at most 100 rows per response, so the pipeline was only ever looking at the first 100 documents — sorted by Id ascending, those are all from 2016 (Knesset 18/20). The output index.csv has been frozen at 7 rows from 2016 ever since. Smoke-test against the live API today shows 12.5M+ matching documents in the dataset and the most recent GroupTypeID=106 PDF is dated 2026-04-21. Walking @odata.nextLink and pushing the 'CommitteeSessionID eq …' predicate into OData (instead of an in-memory scan over the stale 100-row global list) recovers all of it. Other changes: - _odata_paged helper for re-use across the three pagination sites (committees, sessions, per-session documents). - Renamed the inner 'document' loop variable to 'pdf_resp' so it no longer shadows the outer dict. --- .../knesset/knesset_committee_decisions.py | 85 +++++++++++++------ 1 file changed, 57 insertions(+), 28 deletions(-) diff --git a/datapackage_pipelines_budgetkey/pipelines/knesset/knesset_committee_decisions.py b/datapackage_pipelines_budgetkey/pipelines/knesset/knesset_committee_decisions.py index 09b6c63c..1dcdc4c1 100644 --- a/datapackage_pipelines_budgetkey/pipelines/knesset/knesset_committee_decisions.py +++ b/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() \ No newline at end of file + flow()