scopus_api.py

import requests
import pandas as pd
import time
import json
import os

from dotenv import load_dotenv

load_dotenv(dotenv_path=os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
API_KEY = os.environ.get("SCOPUS_API_KEY", "")
if not API_KEY:
    print("ERROR: SCOPUS_API_KEY is not set.")
    print("  Create a .env file with: SCOPUS_API_KEY=your_key_here")
    raise SystemExit(1)

# ---------------------------------------------------------------------------
# Load search configuration from search_config.json
# ---------------------------------------------------------------------------

_CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "search_config.json")
if not os.path.exists(_CONFIG_FILE):
    print(f"ERROR: search_config.json not found at {_CONFIG_FILE}")
    print("  Create it with: base_query, year_start, year_end, output_csv")
    raise SystemExit(1)

with open(_CONFIG_FILE) as _f:
    _cfg = json.load(_f)

try:
    BASE_QUERY = _cfg["base_query"]
    YEARS      = range(int(_cfg["year_start"]), int(_cfg["year_end"]) + 1)
    OUT_CSV    = os.path.join(os.path.dirname(os.path.abspath(__file__)), _cfg["output_csv"])
except KeyError as _e:
    print(f"ERROR: search_config.json is missing required field: {_e}")
    print("  Required fields: base_query, year_start, year_end, output_csv")
    raise SystemExit(1)

COUNT = 25          # confirmed working for this API key
SLEEP = 0.35
MAX_RETRIES = 5
MAX_OFFSET = 4600   # confirmed limit: original script got 400 at start=2025
API_LIMIT = 20_000
PROGRESS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scopus_progress.json")

FIELDS = (
    "eid,dc:title,prism:coverDate,prism:publicationName,"
    "dc:creator,authname,authid,affilname,affiliation,"
    "authkeywords,idxterms,prism:publisher,language,"
    "prism:issn,prism:eIssn,prism:url,prism:aggregationType,"
    "subtypeDescription,citedby-count,openaccessFlag,source-id"
)

HEADERS = {
    "X-ELS-APIKey": API_KEY,
    "Accept": "application/json"
}

# Both dimensions are exhaustive — every record has exactly one SRCTYPE and
# one DOCTYPE, so splitting by them guarantees 100 % coverage.
SPLIT_DIMS = [
    ("SRCTYPE", ["j", "p", "b", "k", "r", "d"]),
    ("DOCTYPE", ["ar", "re", "cp", "ip", "ed", "le", "no", "sh", "er",
                 "bk", "ch", "ab", "cr", "dp"]),
]

api_calls = 0
_quota_warned = False


# ---------------------------------------------------------------------------
# HTTP — counted, retrying, timeout-safe
# ---------------------------------------------------------------------------

def _get(params):
    """
    Single GET with per-attempt quota tracking and auto-retry on:
      429  — exponential back-off (10 s, 20 s …)
      5xx  — exponential back-off (5 s, 10 s …)
      network errors — exponential back-off
    Raises PermissionError immediately on 401/403 (bad API key).
    Raises RuntimeError after MAX_RETRIES consecutive failures.
    """
    global api_calls, _quota_warned
    last_exc = None

    for attempt in range(MAX_RETRIES):
        if api_calls % 100 == 0 and api_calls > 0:
            print(f"  [quota] {api_calls}/{API_LIMIT} "
                  f"({api_calls / API_LIMIT * 100:.1f}%)")
        if not _quota_warned and api_calls >= int(API_LIMIT * 0.95):
            _quota_warned = True
            print(f"  ⚠️  APPROACHING QUOTA: {api_calls}/{API_LIMIT}")

        try:
            api_calls += 1
            r = requests.get(
                "https://api.elsevier.com/content/search/scopus",
                headers=HEADERS,
                params=params,
                timeout=30
            )
            if r.status_code in (401, 403):
                raise PermissionError(
                    f"HTTP {r.status_code} — check your API key / institution access"
                )
            if r.status_code == 429:
                wait = 10 * (2 ** attempt)
                print(f"  Rate limited (429) — waiting {wait}s "
                      f"(attempt {attempt + 1}/{MAX_RETRIES})")
                time.sleep(wait)
                continue
            if r.status_code in (500, 502, 503, 504):
                wait = 5 * (2 ** attempt)
                print(f"  Server error ({r.status_code}) — waiting {wait}s "
                      f"(attempt {attempt + 1}/{MAX_RETRIES})")
                time.sleep(wait)
                continue
            # Show real remaining quota from response headers (more accurate than counting)
            remaining = r.headers.get("X-RateLimit-Remaining")
            try:
                if remaining is not None and int(remaining) < 2000:
                    print(f"  ⚠️  API quota remaining this week: {remaining}")
            except ValueError:
                pass
            return r

        except PermissionError:
            raise
        except requests.exceptions.RequestException as exc:
            wait = 5 * (2 ** attempt)
            print(f"  Network error: {exc} — waiting {wait}s "
                  f"(attempt {attempt + 1}/{MAX_RETRIES})")
            time.sleep(wait)
            last_exc = exc

    raise RuntimeError(f"All {MAX_RETRIES} retries failed. Last: {last_exc}")


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _as_list(value):
    """
    Scopus returns "entry" as a plain dict (not a list) when there is exactly
    ONE result. Wrap it so we always iterate over records, never over keys.
    """
    if value is None:
        return []
    if isinstance(value, dict):
        return [value]
    return value


def _parse_sr(r, label, page_desc):
    """Parse response → search-results dict, or None on any error."""
    try:
        data = r.json()
        if "search-results" not in data:
            print(f"  [{label}] Unexpected response at {page_desc}: "
                  f"keys={list(data.keys())}")
            return None
        return data["search-results"]
    except (ValueError, KeyError) as exc:
        print(f"  [{label}] Bad JSON at {page_desc}: {exc}")
        return None


def parse_entry(e):
    """Return row dict, or None if the entry is a Scopus error object."""
    if not isinstance(e, dict) or "error" in e:
        return None
    return {
        "Authors":                           e.get("dc:creator"),
        "Author full names":                 e.get("authname"),
        "Author(s) ID":                      e.get("authid"),
        "Title":                             e.get("dc:title"),
        "Year":                              (e.get("prism:coverDate") or "")[:4],
        "Source title":                      e.get("prism:publicationName"),
        "Cited by":                          e.get("citedby-count"),
        "Link":                              e.get("prism:url"),
        "Affiliations (basic)":              e.get("affilname"),
        "Authors with affiliations (basic)": e.get("affiliation"),
        "Author Keywords":                   e.get("authkeywords"),
        "Index Keywords":                    e.get("idxterms"),
        "Publisher":                         e.get("prism:publisher"),
        "Language of Original Document":     e.get("language"),
        "Abbreviated Source Title":          e.get("prism:issn") or e.get("prism:eIssn"),
        "Document Type":                     e.get("subtypeDescription"),
        "Publication Stage":                 e.get("prism:aggregationType"),
        "Open Access":                       e.get("openaccessFlag"),
        "Source":                            e.get("source-id"),
        "EID":                               e.get("eid"),
    }


# ---------------------------------------------------------------------------
# Count probe — cheap single-record request to get total before fetching
# ---------------------------------------------------------------------------

def _get_total(query, label):
    """
    Returns total record count for a query using a 1-record probe.
    Raises ValueError if the probe itself returns 400 (bad query / key issue).
    Raises RuntimeError / PermissionError on fatal HTTP errors.
    """
    r = _get({"query": query, "count": 1, "start": 0, "field": "eid"})

    if r.status_code == 400:
        raise ValueError(
            f"[{label}] HTTP 400 on count probe — "
            f"query may be invalid or API key may lack access"
        )
    if not r.ok:
        raise RuntimeError(f"[{label}] HTTP {r.status_code} on count probe")

    sr = _parse_sr(r, label, "count probe")
    if sr is None:
        raise ValueError(f"[{label}] Could not parse count probe response")

    return int(sr.get("opensearch:totalResults", 0))


# ---------------------------------------------------------------------------
# Offset pagination — fetch up to MAX_OFFSET records for one query
# ---------------------------------------------------------------------------

def _fetch_offset(query, label):
    """
    Fetch all records for a query known to have <= MAX_OFFSET total records.
    Returns list of row dicts.
    Raises on any unrecoverable error.
    """
    rows = []
    start = 0
    total = None

    while True:
        r = _get({"query": query, "start": start, "count": COUNT, "field": FIELDS})

        if r.status_code == 400:
            if start == 0:
                raise ValueError(
                    f"[{label}] HTTP 400 on first page (start=0) — "
                    f"unexpected since count probe succeeded"
                )
            # 400 at start > 0 = hit the offset cap
            print(f"  [{label}] Offset cap hit at start={start} "
                  f"(fetched {len(rows)} of {total})")
            break

        if not r.ok:
            raise RuntimeError(f"[{label}] HTTP {r.status_code} at start={start}")

        sr = _parse_sr(r, label, f"start={start}")
        if sr is None:
            if total is None:
                raise ValueError(f"[{label}] Could not parse first page")
            print(f"  [{label}] Parse error at start={start}, stopping")
            break

        if total is None:
            total = int(sr.get("opensearch:totalResults", 0))
            if total == 0:
                return []

        entries = _as_list(sr.get("entry"))
        if not entries:
            break

        for e in entries:
            parsed = parse_entry(e)
            if parsed:
                rows.append(parsed)

        start += len(entries)
        if start >= total:
            break

        time.sleep(SLEEP)

    return rows


# ---------------------------------------------------------------------------
# Main fetch — probe first, then offset or recursive splits
# ---------------------------------------------------------------------------

def fetch_all(query, label="", _depth=0):
    """
    Retrieve ALL records with zero data loss.

    1. Probe total count (1 cheap request).
    2. If total <= MAX_OFFSET: fetch directly with offset pagination.
    3. If total > MAX_OFFSET: split by SRCTYPE, then DOCTYPE.
       Each dimension is exhaustive → no records ever missed.
    """
    total = _get_total(query, label)

    if total == 0:
        return [], 0

    if total <= MAX_OFFSET:
        rows = _fetch_offset(query, label)
        return rows, total

    # Too many records for offset — split
    if _depth >= len(SPLIT_DIMS):
        print(f"  ⚠️  [{label}] {total} records, splits exhausted — "
              f"fetching first {MAX_OFFSET} only (contact Elsevier for full access)")
        rows = _fetch_offset(query, label)
        return rows, total

    field, values = SPLIT_DIMS[_depth]
    print(f"  [{label}] {total} > {MAX_OFFSET} → splitting by {field}")

    all_rows = []
    for v in values:
        sub_label = f"{label}/{field}({v})"
        sub_rows, sub_total = fetch_all(
            f"{query} AND {field}({v})",
            label=sub_label,
            _depth=_depth + 1
        )
        if sub_total:
            print(f"    {sub_label}: {len(sub_rows)}/{sub_total}")
        all_rows.extend(sub_rows)

    return all_rows, total


# ---------------------------------------------------------------------------
# Progress tracking — crash-safe incremental save
# ---------------------------------------------------------------------------

def load_progress():
    if not os.path.exists(PROGRESS_FILE):
        return set()
    try:
        with open(PROGRESS_FILE) as f:
            return set(json.load(f).get("completed", []))
    except (json.JSONDecodeError, IOError) as exc:
        print(f"⚠️  Progress file corrupted ({exc}) — starting fresh")
        return set()


def save_progress(completed):
    """Atomic write: temp file + rename (never corrupts on crash)."""
    tmp = PROGRESS_FILE + ".tmp"
    with open(tmp, "w") as f:
        json.dump({"completed": sorted(completed)}, f, indent=2)
    os.replace(tmp, PROGRESS_FILE)


def append_rows_to_csv(rows):
    if not rows:
        return
    df = pd.DataFrame(rows)
    write_header = not os.path.exists(OUT_CSV)
    df.to_csv(OUT_CSV, mode="a", index=False, header=write_header)


# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    failed_years = []
    completed = load_progress()

    if completed:
        print(f"Resuming — {len(completed)} years already saved: {sorted(completed)}\n")

    for year in YEARS:
        if year in completed:
            print(f"📅 Year {year} — skipping (already saved)")
            continue

        print(f"\n📅 Year {year}")
        year_query = f"{BASE_QUERY} AND PUBYEAR = {year}"

        try:
            year_rows, total = fetch_all(year_query, label=str(year))
            append_rows_to_csv(year_rows)
            completed.add(year)
            save_progress(completed)
            print(f"  → saved {len(year_rows)}/{total} | "
                  f"API calls: {api_calls}/{API_LIMIT}")

        except PermissionError as exc:
            print(f"\n❌ FATAL: {exc}")
            print("   Fix your API key and re-run. Progress so far is saved.")
            break

        except Exception as exc:
            print(f"  ❌ Year {year} FAILED: {exc} — will retry on next run")
            failed_years.append(year)

    # ---------------------------------------------------------------------------
    # Final dedup — atomic write so the CSV is never corrupted
    # ---------------------------------------------------------------------------

    if failed_years:
        print(f"\n⚠️  {len(failed_years)} year(s) failed: {failed_years}")
        print("   Re-run the script to retry them.")

    if not os.path.exists(OUT_CSV):
        print("\n⚠️  No output file created — all years may have failed.")
    else:
        print("\nDeduplicating final CSV...")
        df = pd.read_csv(OUT_CSV, dtype=str)
        before = len(df)
        df.drop_duplicates(subset=["EID"], inplace=True)
        tmp_csv = OUT_CSV + ".tmp"
        df.to_csv(tmp_csv, index=False)
        os.replace(tmp_csv, OUT_CSV)
        print(f"✅ DONE: {len(df)} unique records saved to {OUT_CSV}")
        if before != len(df):
            print(f"   (removed {before - len(df)} duplicate rows)")
        print(f"   API calls used: {api_calls}/{API_LIMIT} "
              f"({api_calls / API_LIMIT * 100:.1f}%)")