Implement Pagination for SerialTitleSearch() (close #431)

Author: nils-herrmann

pybliometrics/scopus/tests/test_SerialTitleSearch.py

@@ -11,11 +11,8 @@
 ser2 = SerialTitleSearch({'issn': '1468-0262'}, refresh=30)
 # Search by publisher
 ser3 = SerialTitleSearch({'pub': 'Stellenbosch'}, refresh=30)
-# Search by subject abbreviation
-ser4 = SerialTitleSearch({'subj': 'COMP'}, refresh=30)
 # Search by subject area code
-ser5 = SerialTitleSearch({'subjCode': '2612'}, refresh=30)
-
+ser4 = SerialTitleSearch(query={'subjCode': '2708', 'content': 'journal'}, refresh=30)
 
 def test_deprecated_class():
     from pytest import deprecated_call
@@ -28,23 +25,22 @@ def test_deprecated_class():
 def test_results_title():
     assert len(ser1.results) == 1
     assert ser1.results[0]['title'] == 'SoftwareX'
+    assert ser1.get_results_size() == 1
 
 
 def test_results_issn():
     assert len(ser2.results) == 1
     assert ser2.results[0]['title'] == 'Econometrica'
+    assert ser1.get_results_size() == 1
 
 
 def test_results_pub():
     assert len(ser3.results) == 4
     assert ser3.results[0]['title'] == 'Akroterion'
-
-
-def test_results_subj():
-    ser4_subj_abbs = set(i['subject_area_abbrevs'] for i in ser4.results)
-    assert False not in ['COMP' in i for i in ser4_subj_abbs]
+    assert ser3.get_results_size() == 4
 
 
 def test_results_subjcode():
-    ser5_subj_codes = set(i['subject_area_codes'] for i in ser5.results)
-    assert False not in ['2612' in i for i in ser5_subj_codes]
+    ser4_subj_codes = set(i['subject_area_codes'] for i in ser4.results)
+    assert False not in ['2708' in i for i in ser4_subj_codes]
+    assert ser4.get_results_size() >= 255

pybliometrics/superclasses/base.py

@@ -4,6 +4,8 @@
 from math import ceil
 from time import localtime, strftime, time
 
+from urllib.parse import parse_qs, urlparse
+
 from tqdm import tqdm
 
 from pybliometrics.exception import ScopusQueryError
@@ -53,13 +55,18 @@ def __init__(self,
         ab_ref_retrieval = (api == 'AbstractRetrieval') and (params['view'] == 'REF')
         # Check if object retrieval
         obj_retrieval = (api == 'ObjectRetrieval')
+        # Check if serial title search (special pagination)
+        serial_search = (api == 'SerialTitleSearch')
 
         if fname.exists() and not self._refresh:
             self._mdate = mod_ts
             if search_request:
                 self._json = [loads(line) for line in
                               fname.read_text().split("\n") if line]
                 self._n = len(self._json)
+            elif serial_search:
+                self._json = loads(fname.read_text())
+                self._n = len(self._json['serial-metadata-response'].get('entry', []))
             elif obj_retrieval:
                 self._object = fname.read_bytes()
             else:
@@ -73,6 +80,11 @@ def __init__(self,
                 data = _get_all_refs(url, params, verbose, resp, **kwds)
                 self._json = data
                 data = [data]
+            elif serial_search:
+                entries = _get_all_serial_results(url, params, verbose, resp, **kwds)
+                self._json = {'serial-metadata-response': {'entry': entries}}
+                self._n = len(entries)
+                data = [self._json]
             elif search_request:
                 # Get number of results
                 res = resp.json()
@@ -210,3 +222,34 @@ def _get_all_refs(url: str, params: dict, verbose: bool, resp: dict, **kwds) ->
         print(f'Total data: {len(parse_content.chained_get(data, ["abstracts-retrieval-response", "references", "reference"]))}')
 
     return data
+
+
+def _get_all_serial_results(url: str, params: dict, verbose: bool, resp, **kwds) -> list:
+    """Get all results for `SerialTitleSearch` with pagination."""
+    res = resp.json()
+    data = res.get('serial-metadata-response', {}).get('entry', [])
+    
+    # Check for 'next' link to determine if pagination is needed
+    links = res.get('serial-metadata-response', {}).get('link', [])
+    next_link = next((l for l in links if l.get('@ref') == 'next'), None)
+    last_link = next((l for l in links if l.get('@ref') == 'last'), None)
+    
+    if not next_link or not last_link:
+        return data
+    
+    # Calculate total from last link (start + count)
+    last_url = last_link.get('@href', '')
+    parsed = parse_qs(urlparse(last_url).query)
+    last_start = int(parsed.get('start', [0])[0])
+    count = int(parsed.get('count', [params['count']])[0])
+    n_total = last_start + count
+    n_chunks = ceil(n_total / count)
+    
+    for i in tqdm(range(1, n_chunks), disable=not verbose, initial=1, total=n_chunks):
+        params['start'] = i * count
+        resp = get_content(url, 'SerialTitleSearch', params, **kwds)
+        res = resp.json()
+        entries = res.get('serial-metadata-response', {}).get('entry', [])
+        data.extend(entries)
+    
+    return data