API reference

Every public name in scopusflow is documented here, grouped along the path a search takes: describe it, size it, run it as a resumable harvest, normalise what comes back, see what a later retrieval changed, and summarise and plot the result. The groups are the ones the R package's reference index uses, and they come in the same order, so a function documented in both languages is filed under the same heading on either site.

Two of the group names differ from the R ones because the membership differs. The R group 'Plan and size' also holds the counting function, which is documented under Retrieve here alongside the other calls that spend quota, so the first group is called 'Plan and query' instead. Records is wider than its R namesake, because it absorbs the R groups 'Export and I/O' and 'Data'. The BibTeX and RIS writers and the bundled example harvest all work on the record table, and neither would fill a section of its own. Reporting holds the same one function in both languages. The two remaining R groups have no counterpart below. 'App' documents a function that launches the interface, where the Python app is started from the command line and covered by The code-free app, and 'Keys' documents a key check that scopusflow does not need in Python, since pybliometrics holds the key configuration.

Every name listed under a group heading is importable straight from scopusflow, conventionally as sf. The guides linked from the home page work the same functions through end-to-end examples.

Plan and query

Describe a search before running it, and build field-tagged Boolean queries.

scopus_query

scopus_query(*terms, op='AND', field=None)

Combine terms into one Scopus query, optionally field-wrapping each.

Parameters:

Name	Type	Description	Default
*terms	str	One or more non-empty search terms.	()
op	str	The boolean operator joining the terms: "AND", "OR" or "AND NOT".	'AND'
field	str | None	An optional field tag applied to every term (see :data:FIELD_TAGS).	None

Source code in src/scopusflow/query.py

def scopus_query(*terms: str, op: str = "AND", field: str | None = None) -> str:
    """Combine terms into one Scopus query, optionally field-wrapping each.

    Parameters
    ----------
    *terms:
        One or more non-empty search terms.
    op:
        The boolean operator joining the terms: ``"AND"``, ``"OR"`` or ``"AND NOT"``.
    field:
        An optional field tag applied to every term (see :data:`FIELD_TAGS`).
    """
    if op not in {"AND", "OR", "AND NOT"}:
        raise ValueError("op must be one of 'AND', 'OR', 'AND NOT'.")
    cleaned = [t.strip() for t in terms]
    if not cleaned or any(not t for t in cleaned):
        raise ValueError("All terms must be non-empty.")
    return f" {op} ".join(wrap_field(t, field) for t in cleaned)

wrap_field

wrap_field(query, field)

Wrap query in a field tag, e.g. TITLE-ABS-KEY(graphene).

Source code in src/scopusflow/query.py

def wrap_field(query: str, field: str | None) -> str:
    """Wrap ``query`` in a field tag, e.g. ``TITLE-ABS-KEY(graphene)``."""
    if field is None:
        return query
    field = field.strip().upper()
    if not _FIELD_RE.match(field):
        raise ValueError(f"Invalid field tag {field!r}; use letters and hyphens only.")
    return f"{field}({query})"

FIELD_TAGS module-attribute

FIELD_TAGS = {'TITLE': 'Words in the document title', 'TITLE-ABS-KEY': 'Title, abstract and keywords', 'TITLE-ABS-KEY-AUTH': 'Title, abstract, keywords and author names', 'ABS': 'Abstract text', 'KEY': 'Indexed and author keywords', 'AUTH': 'Author names', 'AUTHKEY': 'Author-supplied keywords', 'AFFIL': 'Affiliation, any part', 'AFFILORG': 'Affiliation organisation name', 'SRCTITLE': 'Source (publication) title', 'DOI': 'Digital Object Identifier', 'ALL': 'All available fields'}

SearchPlan dataclass

A fully specified, inspectable description of a Scopus search.

Splitting describing a search from executing it makes a workflow reproducible and lets a large retrieval be partitioned by year, so it can be cached and resumed.

page_size is the number of records asked for per request, None (the default) meaning the largest the view allows: 200 for STANDARD, 25 for COMPLETE. It is stored on the plan, and sent, because the search record :func:scopusflow.report.scopus_search_report writes has to state how the harvest was paged, and a figure taken from anywhere but the plan would be a guess.

Source code in src/scopusflow/plan.py

@dataclass(frozen=True)
class SearchPlan:
    """A fully specified, inspectable description of a Scopus search.

    Splitting *describing* a search from *executing* it makes a workflow
    reproducible and lets a large retrieval be partitioned by year, so it can be
    cached and resumed.

    ``page_size`` is the number of records asked for per request, ``None``
    (the default) meaning the largest the view allows: 200 for ``STANDARD``,
    25 for ``COMPLETE``. It is stored on the plan, and sent, because the search
    record :func:`scopusflow.report.scopus_search_report` writes has to state
    how the harvest was paged, and a figure taken from anywhere but the plan
    would be a guess.
    """

    query: str
    years: Sequence[int] | None = None
    field: str | None = None
    view: str = "STANDARD"
    partition: str = "none"  # "none" or "year"
    page_size: int | None = None

    def __post_init__(self) -> None:
        if not self.query or not self.query.strip():
            raise ValueError("query must be a non-empty string.")
        if self.view not in {"STANDARD", "COMPLETE"}:
            raise ValueError("view must be 'STANDARD' or 'COMPLETE'.")
        if self.partition not in {"none", "year"}:
            raise ValueError("partition must be 'none' or 'year'.")
        object.__setattr__(self, "page_size",
                           _check_page_size(self.page_size, self.view))
        # Store the validated integers, never the values as passed: cells() renders the
        # year into the cell's date, and str(2015.0) would reach the API as
        # "2015.0". A tuple, since a list would leave the frozen dataclass unhashable.
        # object.__setattr__ is how a frozen dataclass normalises a field.
        #
        # Sorted and de-duplicated, as every other caller of _check_years already
        # does with its result and as cells() does with this one, so that the
        # stored years say what the search will actually do. Without it two plans
        # describing the same search compared unequal on the order the years
        # happened to be typed in, and the reproduction snippet
        # scopus_search_report() emits (which renders them canonically) rebuilt a
        # plan that ran identically but failed an equality check against its own
        # original.
        checked = _check_years(self.years)
        object.__setattr__(self, "years",
                           None if checked is None else tuple(sorted(set(checked))))
        if self.partition == "year" and not self.years:
            raise ValueError("partition='year' requires years.")

    @property
    def wrapped_query(self) -> str:
        return wrap_field(self.query, self.field)

    def cells(self) -> list[PlanCell]:
        """Expand the plan into the cells that will be fetched."""
        q = self.wrapped_query
        size = int(self.page_size)  # type: ignore[arg-type]
        if self.partition == "year":
            years = sorted(set(self.years))  # type: ignore[arg-type]
            return [
                PlanCell(i + 1, q, str(y), y, self.view, size)
                for i, y in enumerate(years)
            ]
        date = None
        if self.years:
            lo, hi = min(self.years), max(self.years)
            date = str(lo) if lo == hi else f"{lo}-{hi}"
        return [PlanCell(1, q, date, None, self.view, size)]

cells

cells()

Expand the plan into the cells that will be fetched.

Source code in src/scopusflow/plan.py

def cells(self) -> list[PlanCell]:
    """Expand the plan into the cells that will be fetched."""
    q = self.wrapped_query
    size = int(self.page_size)  # type: ignore[arg-type]
    if self.partition == "year":
        years = sorted(set(self.years))  # type: ignore[arg-type]
        return [
            PlanCell(i + 1, q, str(y), y, self.view, size)
            for i, y in enumerate(years)
        ]
    date = None
    if self.years:
        lo, hi = min(self.years), max(self.years)
        date = str(lo) if lo == hi else f"{lo}-{hi}"
    return [PlanCell(1, q, date, None, self.view, size)]

PlanCell dataclass

One unit of work in a :class:SearchPlan.

Source code in src/scopusflow/plan.py

@dataclass(frozen=True)
class PlanCell:
    """One unit of work in a :class:`SearchPlan`."""

    cell: int
    query: str
    date: str | None
    year: int | None
    view: str
    page_size: int = 200

Retrieve

Size a search cheaply first, then execute a plan as a resumable, checkpointed harvest, and pull fuller records.

scopus_count

scopus_count(query, years=None, field=None, view='STANDARD', **kwargs)

Return how many records the (optionally year-filtered) query matches.

A single cheap request that does not download the records, so it is the right way to size a search before committing quota to a harvest.

Source code in src/scopusflow/count.py

def scopus_count(query: str, years: Sequence[int] | None = None,
                 field: str | None = None, view: str = "STANDARD",
                 **kwargs) -> int:
    """Return how many records the (optionally year-filtered) query matches.

    A single cheap request that does not download the records, so it is the right
    way to size a search before committing quota to a harvest.
    """
    if not query or not str(query).strip():
        raise ValueError("query must be a non-empty string.")
    q = _count_query(str(query).strip(), years, field)

    from pybliometrics.scopus import ScopusSearch  # imported lazily; needs a key

    return int(ScopusSearch(q, view=view, download=False, **kwargs).get_results_size())

fetch_plan

fetch_plan(plan, cache_dir=None, resume=True, format='parquet', should_stop=None, **kwargs)

Run every cell of plan and return one normalised DataFrame.

With cache_dir set, each cell is written to disk as it completes, so an interrupted or quota-limited run resumes without re-fetching finished cells. A cache_dir belongs to one plan: checkpoints are keyed by cell number, so on resume each checkpoint's own recorded query and view are compared against the cell's, and a checkpoint written by a different plan is warned about and refetched, never silently returned. Point each plan at its own directory. A checkpoint that cannot be read back is likewise treated as a miss, warned about and refetched, and the harvest carries on. format selects the checkpoint format ("parquet" or "csv"); parquet silently falls back to CSV when no parquet engine is installed. Pass a zero-argument should_stop callable to allow co-operative cancellation: it is checked before each cell and the harvest stops (returning what it has) when it returns True. Per-cell progress is emitted on the "scopusflow" logger.

Each cell's row count is compared against the total the API reports for that cell's query, and a shortfall is warned about, since a truncated or failed download otherwise arrives as a merely small result. The per-cell accounting is attached as result.attrs["cell_totals"], a frame of cell, date, n_records and reported_total, and their sum as result.attrs["total_results"], the attribute the R twin's scopus_fetch() also attaches. The sum is None unless every cell reported a total, since a partial sum would understate the search while looking like a real figure; a cell resumed from a checkpoint reports none, the count not being part of what a checkpoint stores.

The harvest also carries its provenance: the originating plan, retrieved_at (a timezone-aware UTC datetime), scopusflow_version and paging. These are what :func:scopusflow.report.scopus_search_report reads back, and they are omitted, never approximated, when any cell was resumed from a checkpoint, since a checkpoint carries no record of when it was taken and dating the whole from the cells that were fetched now would date it later than part of what it holds.

When plan.view == "COMPLETE", the output gains an authkeywords column (see :func:scopusflow.records.to_records) at no extra request cost beyond COMPLETE's own smaller page size, which already means more requests, and so more quota, for the same number of records. A plan with view="STANDARD" (the default) never carries this column, so existing code is unaffected: a checkpoint written under the other view is refetched like any other different-plan checkpoint. Resuming a cache written before this column existed is safe in the COMPLETE direction: pandas.concat fills the older cells' missing column with NA rather than erroring.

Source code in src/scopusflow/fetch.py

def fetch_plan(
    plan: SearchPlan,
    cache_dir: str | None = None,
    resume: bool = True,
    format: str = "parquet",
    should_stop=None,
    **kwargs,
) -> pd.DataFrame:
    """Run every cell of ``plan`` and return one normalised DataFrame.

    With ``cache_dir`` set, each cell is written to disk as it completes, so an
    interrupted or quota-limited run resumes without re-fetching finished cells.
    A cache_dir belongs to one plan: checkpoints are keyed by cell number, so
    on resume each checkpoint's own recorded query and view are compared
    against the cell's, and a checkpoint written by a different plan is warned
    about and refetched, never silently returned. Point each plan at its own
    directory. A checkpoint that cannot be read back is likewise treated as a
    miss, warned about and refetched, and the harvest carries on.
    ``format`` selects the checkpoint format ("parquet" or "csv"); parquet
    silently falls back to CSV when no parquet engine is installed. Pass a
    zero-argument ``should_stop`` callable to allow co-operative cancellation: it
    is checked before each cell and the harvest stops (returning what it has) when
    it returns ``True``. Per-cell progress is emitted on the ``"scopusflow"``
    logger.

    Each cell's row count is compared against the total the API reports for
    that cell's query, and a shortfall is warned about, since a truncated or
    failed download otherwise arrives as a merely small result. The per-cell
    accounting is attached as ``result.attrs["cell_totals"]``, a frame of
    ``cell``, ``date``, ``n_records`` and ``reported_total``, and their sum as
    ``result.attrs["total_results"]``, the attribute the R twin's
    ``scopus_fetch()`` also attaches. The sum is ``None`` unless every cell
    reported a total, since a partial sum would understate the search while
    looking like a real figure; a cell resumed from a checkpoint reports none,
    the count not being part of what a checkpoint stores.

    The harvest also carries its provenance: the originating ``plan``,
    ``retrieved_at`` (a timezone-aware UTC ``datetime``),
    ``scopusflow_version`` and ``paging``. These are what
    :func:`scopusflow.report.scopus_search_report` reads back, and they are
    omitted, never approximated, when any cell was resumed from a
    checkpoint, since a checkpoint carries no record of when it was taken and
    dating the whole from the cells that were fetched now would date it later
    than part of what it holds.

    When ``plan.view == "COMPLETE"``, the output gains an ``authkeywords``
    column (see :func:`scopusflow.records.to_records`) at no extra request cost
    beyond ``COMPLETE``'s own smaller page size, which already means more
    requests, and so more quota, for the same number of records. A plan with
    ``view="STANDARD"`` (the default) never carries this column, so existing
    code is unaffected: a checkpoint written under the other view is refetched
    like any other different-plan checkpoint. Resuming a cache written before
    this column existed is safe in the ``COMPLETE`` direction:
    ``pandas.concat`` fills the older cells' missing column with ``NA`` rather
    than erroring.
    """
    if not isinstance(plan, SearchPlan):
        raise ValueError("plan must be a SearchPlan.")
    if format not in _FORMATS:
        raise ValueError("format must be 'parquet' or 'csv'.")

    from pybliometrics.scopus import ScopusSearch  # imported lazily; needs a key

    cache = Path(cache_dir) if cache_dir else None
    if cache is not None:
        cache.mkdir(parents=True, exist_ok=True)

    cells = plan.cells()
    total = len(cells)
    frames: list[pd.DataFrame] = []
    accounting: list[dict] = []
    stamps: list[datetime | None] = []
    for cell in cells:
        if should_stop is not None and should_stop():
            logger.info("Stopped before cell %d/%d.", cell.cell, total)
            break

        query = _cell_query(cell.query, cell.year, cell.date)
        if cache is not None and resume:
            existing = _find_checkpoint(cache, cell.cell)
            cached = _read_checkpoint(existing) if existing is not None else None
            if existing is not None and cached is None:
                warnings.warn(
                    f"The checkpoint {existing} could not be read back, so it "
                    "was discarded and the cell refetched. An interrupted run "
                    "can leave a checkpoint half-written.",
                    stacklevel=2,
                )
            elif cached is not None:
                # Checkpoints are keyed by cell number alone, so a cache_dir
                # reused for a different plan would otherwise hand back the
                # wrong records silently. The frames carry the query they were
                # fetched with; a mismatch means the checkpoint belongs to
                # another plan and the cell is refetched. A zero-row checkpoint
                # carries no query values to compare and is accepted as is.
                # The view is compared too, since the query alone cannot tell
                # a STANDARD plan from a COMPLETE one, and a COMPLETE-written
                # checkpoint would hand a STANDARD resume an authkeywords
                # column the documentation promises it never carries.
                cached_queries = (
                    set(cached["query"].dropna().unique())
                    if "query" in cached.columns else set()
                )
                cached_view = _checkpoint_view(cached)
                if cached_queries and cached_queries != {query}:
                    warnings.warn(
                        f"Checkpoint for cell {cell.cell} in {cache} was written "
                        f"by a different plan (query {sorted(cached_queries)!r}, "
                        f"not {query!r}); refetching this cell. Use one cache_dir "
                        "per plan.",
                        stacklevel=2,
                    )
                elif cached_view is not None and cached_view != cell.view:
                    warnings.warn(
                        f"Checkpoint for cell {cell.cell} in {cache} was written "
                        f"by a different plan (view {cached_view!r}, "
                        f"not {cell.view!r}); refetching this cell. Use one "
                        "cache_dir per plan.",
                        stacklevel=2,
                    )
                else:
                    logger.info("Cell %d/%d: loaded from cache.", cell.cell, total)
                    served = cached.drop(columns=["view"], errors="ignore")
                    frames.append(served)
                    accounting.append({"cell": cell.cell, "date": cell.date,
                                       "n_records": len(served),
                                       "reported_total": None})
                    stamps.append(None)
                    continue

        logger.info("Cell %d/%d: fetching %s", cell.cell, total, query)
        # The page size is the plan's, so the harvest pages the way the plan (and
        # the search record built from it) says it does. setdefault, so a caller
        # passing count of their own still wins.
        kwargs.setdefault("count", cell.page_size)
        search = ScopusSearch(query, view=cell.view, cursor=True, **kwargs)
        frame = to_records(search.results, query=query, view=cell.view)

        # A download that returned nothing yields a zero-row frame and never
        # an error, so without this comparison a truncated or failed cell is
        # indistinguishable from one that matched only a few records.
        cell_total = _reported_total(search)
        accounting.append({"cell": cell.cell, "date": cell.date,
                           "n_records": len(frame), "reported_total": cell_total})
        stamps.append(datetime.now(timezone.utc))
        if cell_total is not None:
            if len(frame) < cell_total:
                warnings.warn(
                    f"Cell {cell.cell} retrieved {len(frame)} record(s), but the "
                    f"Scopus API reports {cell_total} for this query, so the "
                    "harvest may be incomplete. Check the key's remaining quota, "
                    "and consider partitioning the plan by year so each cell is "
                    "smaller.",
                    stacklevel=2,
                )

        if cache is not None:
            # The view travels with the checkpoint (and only the checkpoint;
            # resume strips it again) so a resume under the other view is
            # detectable in both directions, beyond the case where an authkeywords
            # column betrays a COMPLETE origin.
            _write_checkpoint(frame.assign(view=cell.view), cache, cell.cell, format)
        frames.append(frame)

    if not frames:
        columns = [*RECORD_COLUMNS, "authkeywords"] if plan.view == "COMPLETE" else RECORD_COLUMNS
        out = pd.DataFrame(columns=columns)
    else:
        out = pd.concat(frames, ignore_index=True)
        out["entry_number"] = range(1, len(out) + 1)
        logger.info("Retrieved %d records.", len(out))

    # The plan travels with its own harvest, as it does in the R twin, so the
    # search record can be written from the records alone.
    out.attrs["plan"] = plan
    out.attrs["cell_totals"] = pd.DataFrame(
        accounting, columns=["cell", "date", "n_records", "reported_total"]
    )
    reported = [row["reported_total"] for row in accounting]
    out.attrs["total_results"] = (
        sum(reported) if accounting and all(n is not None for n in reported) else None
    )
    out.attrs["paging"] = "cursor"
    if stamps and all(s is not None for s in stamps):
        # Imported inside the function, and never at module scope: this module
        # is imported
        # while the package's own __init__ is still executing, and __version__
        # is not bound until after that import returns.
        from . import __version__

        out.attrs["retrieved_at"] = min(stamps)
        out.attrs["scopusflow_version"] = __version__
    return out

scopus_abstract

scopus_abstract(ids, by='doi', view='META_ABS', include=(), cache_dir=None, resume=True, **kwargs)

Retrieve abstracts for one or many ids, resilient per id.

by selects the lookup type ("doi", "eid" or "scopus_id"). Any id that fails is warned about and yields an all-NA row that still records the id. view defaults to "META_ABS" (pybliometrics' own default), which carries the abstract text; the lighter "META" omits it and leaves the abstract column empty. include, when unused, leaves the column set exactly as before.

include names extra fields to retrieve in the same request: "references" and/or "keywords". Both require Abstract Retrieval's "FULL" or "REF" view (see view), an entitlement separate from ordinary abstract access and from Scopus Search access, and that, per Elsevier's own documentation, some fields (notably author keywords) may need to be requested from your Scopus/Elsevier account contact even when the view itself is otherwise accessible. In development, against a live key with full Abstract Retrieval access, "FULL" returned a complete, correctly counted reference list for every document tried, while "REF" returned the identical, complete list in one case but a truncated (paginated) subset in another on an otherwise identical request made moments apart; "FULL" is recommended when your entitlement allows it, and a mismatch between the number of references returned and the document's own reported reference count (refcount) is warned about, since the list may be an incomplete page of the bibliography.

When "keywords" is included, an authkeywords column is added: the document's author-supplied keywords, joined with "; ", or NA when the document has none, or when the API omits the field for a given key's entitlement. In this package's own development testing, against a live, otherwise fully-entitled key, this field did not populate for documents that do carry author keywords in Scopus itself, so an all-NA column is more likely an entitlement gap worth raising with your Scopus/Elsevier account contact than genuinely absent data.

When "references" is included, a references column is added: one DataFrame per document (one row per cited work), using pybliometrics' own native field set (position, id, doi, title, authors, authors_auid, authors_affiliationid, sourcetitle, publicationyear, coverDate, volume, issue, first, last, citedbycount, type, text, fulltext). A document with no resolvable references yields a zero-row DataFrame, so the column can always be unnested.

cache_dir and resume checkpoint per identifier, the way :func:scopusflow.fetch.fetch_plan checkpoints per cell, pickled rather than written as parquet/csv, since a row here can carry a nested DataFrame. Only a successful retrieval is checkpointed: a failed identifier still yields its warned-about NA row in the returned frame, but is retried on the next resumed run, never read back as data. Worth setting whenever include is used: Abstract Retrieval draws on its own weekly quota, smaller than and separate from Search's, and every identifier costs its own request, so re-running an interrupted batch without a cache re-spends quota already spent. Relying on pybliometrics' own on-disk response cache (its refresh parameter, keyed by identifier and view under its configured cache directory) is enough to avoid repeat network calls for the same identifier across script runs; this checkpoint is for batch-level progress and resumability across many identifiers, a separate concern.

The number of Abstract Retrieval requests made, and the most recently parsed remaining-quota figure (from pybliometrics' get_key_remaining_quota()), are attached as result.attrs["n_requests"] and result.attrs["quota"], since this is a materially more expensive operation than a search call.

A 403 (an entitlement gate, most often on the requested view or field) raises :class:scopusflow.exceptions.ScopusFlowForbiddenError and stops the batch immediately, naming the view and identifier, where a generic failure would leave the caller guessing. Repeating the identical failure for every remaining identifier would serve nobody: entitlement is a property of the account, so a retry cannot succeed.

Source code in src/scopusflow/abstract.py

def scopus_abstract(
    ids,
    by: str = "doi",
    view: str = "META_ABS",
    include: tuple[str, ...] = (),
    cache_dir: str | None = None,
    resume: bool = True,
    **kwargs,
) -> pd.DataFrame:
    """Retrieve abstracts for one or many ids, resilient per id.

    ``by`` selects the lookup type ("doi", "eid" or "scopus_id"). Any id that
    fails is warned about and yields an all-NA row that still records the id.
    ``view`` defaults to "META_ABS" (pybliometrics' own default), which
    carries the abstract text; the lighter "META" omits it and leaves the
    ``abstract`` column empty. ``include``, when unused, leaves the column
    set exactly as before.

    ``include`` names extra fields to retrieve in the same request:
    "references" and/or "keywords". Both require Abstract Retrieval's "FULL"
    or "REF" view (see ``view``), an entitlement separate from ordinary
    abstract access and from Scopus Search access, and that, per Elsevier's
    own documentation, some fields (notably author keywords) may need to be
    requested from your Scopus/Elsevier account contact even when the view
    itself is otherwise accessible. In development, against a live key with
    full Abstract Retrieval access, "FULL" returned a complete, correctly
    counted reference list for every document tried, while "REF" returned the
    identical, complete list in one case but a truncated (paginated) subset in
    another on an otherwise identical request made moments apart; "FULL" is
    recommended when your entitlement allows it, and a mismatch between the
    number of references returned and the document's own reported reference
    count (``refcount``) is warned about, since the list may be an incomplete
    page of the bibliography.

    When "keywords" is included, an ``authkeywords`` column is added: the
    document's author-supplied keywords, joined with "; ", or ``NA`` when the
    document has none, or when the API omits the field for a given key's
    entitlement. In this package's own development testing, against a live,
    otherwise fully-entitled key, this field did not populate for documents
    that do carry author keywords in Scopus itself, so an all-``NA`` column
    is more likely an entitlement gap worth raising with your Scopus/Elsevier
    account contact than genuinely absent data.

    When "references" is included, a ``references`` column is added: one
    DataFrame per document (one row per cited work), using pybliometrics' own
    native field set (``position``, ``id``, ``doi``, ``title``, ``authors``,
    ``authors_auid``, ``authors_affiliationid``, ``sourcetitle``,
    ``publicationyear``, ``coverDate``, ``volume``, ``issue``, ``first``,
    ``last``, ``citedbycount``, ``type``, ``text``, ``fulltext``). A document
    with no resolvable references yields a zero-row DataFrame, so
    the column can always be unnested.

    ``cache_dir`` and ``resume`` checkpoint per identifier, the way
    :func:`scopusflow.fetch.fetch_plan` checkpoints per cell, pickled rather
    than written as parquet/csv, since a row here can carry a nested
    DataFrame. Only a successful retrieval is checkpointed: a failed
    identifier still yields its warned-about NA row in the returned frame,
    but is retried on the next resumed run, never read back as data.
    Worth setting whenever ``include`` is used: Abstract Retrieval
    draws on its own weekly quota, smaller than and separate from Search's,
    and every identifier costs its own request, so re-running an interrupted
    batch without a cache re-spends quota already spent. Relying on
    pybliometrics' own on-disk response cache (its ``refresh`` parameter,
    keyed by identifier and view under its configured cache directory) is
    enough to avoid repeat network calls for the *same* identifier across
    script runs; this checkpoint is for batch-level progress and resumability
    across *many* identifiers, a separate concern.

    The number of Abstract Retrieval requests made, and the most recently
    parsed remaining-quota figure (from pybliometrics'
    ``get_key_remaining_quota()``), are attached as ``result.attrs["n_requests"]``
    and ``result.attrs["quota"]``, since this is a materially more expensive
    operation than a search call.

    A 403 (an entitlement gate, most often on the requested view or field)
    raises :class:`scopusflow.exceptions.ScopusFlowForbiddenError` and stops
    the batch immediately, naming the view and identifier, where a generic
    failure would leave the caller guessing. Repeating the identical failure
    for every remaining identifier would serve nobody: entitlement is a
    property of the account, so a retry cannot succeed.
    """
    if by not in _ID_TYPES:
        raise ValueError("by must be one of 'doi', 'eid', 'scopus_id'.")
    include = tuple(include) if include else ()
    if not set(include) <= _KNOWN_INCLUDE:
        raise ValueError("include must be made up of 'references' and/or 'keywords'.")
    if "references" in include and view not in _VIEWS_WITH_REFERENCES:
        raise ValueError('include="references" needs view="FULL" or view="REF".')
    id_type = _ID_TYPES[by]
    id_column = "doi" if by == "doi" else "scopus_id"

    if isinstance(ids, str):
        ids = [ids]

    columns = list(ABSTRACT_COLUMNS)
    if "keywords" in include:
        columns = [*columns, "authkeywords"]
    if "references" in include:
        columns = [*columns, "references"]

    cache = Path(cache_dir) if cache_dir else None
    if cache is not None:
        cache.mkdir(parents=True, exist_ok=True)

    # Resolve the dependency once: a missing pybliometrics is a setup error and
    # must be raised in its own right, where masquerading as every id failing
    # to retrieve would hide the account-level cause.
    from pybliometrics.scopus import AbstractRetrieval  # lazy; needs a key
    try:
        # A separate, defensive import: pybliometrics.exception is an internal
        # module, outside pybliometrics.scopus's own public surface, so a
        # minimal test double or an unexpected future reorganisation should
        # degrade to generic exception handling below, which stops it breaking
        # every call to this function.
        from pybliometrics.exception import Scopus403Error
    except ImportError:
        class Scopus403Error(Exception):  # never actually raised; see above
            pass

    n_requests = 0
    quota = None
    rows = []
    for i, ident in enumerate(ids, start=1):
        checkpoint = (
            _find_abstract_checkpoint(cache, view, include, ident)
            if cache is not None else None
        )
        if checkpoint is not None and resume:
            cached = _read_abstract_checkpoint(checkpoint)
            if cached is None:
                warnings.warn(
                    f"The checkpoint {checkpoint} could not be read back, so it "
                    "was discarded and the identifier retrieved again. An "
                    "interrupted run can leave a checkpoint half-written.",
                    UserWarning,
                    stacklevel=2,
                )
            else:
                logger.info("%d/%d: %s loaded from cache.", i, len(ids), ident)
                rows.append(cached)
                continue

        logger.info("Retrieving %d/%d: %s", i, len(ids), ident)
        try:
            ab = AbstractRetrieval(ident, id_type=id_type, view=view, **kwargs)
            n_requests += 1
            # Optional: an object without these methods (a plain dict, or a
            # minimal stand-in) simply reports no quota, and does not fail
            # the whole retrieval over a metadata nicety. Both lookups are
            # guarded, since an object carrying only the first would otherwise
            # fail here after a successful retrieval and record an NA row.
            get_quota = getattr(ab, "get_key_remaining_quota", None)
            get_reset = getattr(ab, "get_key_reset_time", None)
            remaining = get_quota() if callable(get_quota) else None
            if remaining is not None:
                reset = get_reset() if callable(get_reset) else None
                quota = {"remaining": remaining, "reset": reset}
            row = _abstract_row(ab, include=include)
        except Scopus403Error as exc:
            n_requests += 1
            remaining_ids = len(ids) - i
            # The FULL/REF alternative is only sensible advice when the failed
            # view was one of the two reference-carrying views; a 403 on a
            # plain META/META_ABS retrieval is an ordinary entitlement gap.
            alternative = (
                ' or, if you have not already, try the other of "FULL"/"REF"'
                if view in _VIEWS_WITH_REFERENCES else ""
            )
            raise ScopusFlowForbiddenError(
                f'Abstract Retrieval refused view="{view}" (HTTP 403) for {ident!r}. '
                "This usually means your Scopus API key's entitlement does not cover "
                "the requested view or field; contact your Scopus/Elsevier account "
                f"holder or institutional administrator to request access{alternative}. "
                "Stopping rather "
                f"than repeating the same failure for the remaining {remaining_ids} "
                "identifier(s) (this entitlement is an account-level property, not a "
                "per-document one, so it will not succeed on retry)."
            ) from exc
        except Exception:  # one bad id must not sink the batch
            n_requests += 1
            warnings.warn(
                f"Could not retrieve abstract for {ident!r}; recording NA row.",
                stacklevel=2,
            )
            row = {col: pd.NA for col in columns}
            row[id_column] = ident
            if "references" in include:
                row["references"] = _references_frame(None)
            # Never checkpointed: many failures are transient (a timeout, a
            # quota 429, a 5xx), and a persisted NA row would be read back as
            # data on every later resume. The identifier is retried instead,
            # at the cost of one request.
            rows.append(row)
            continue

        if cache is not None:
            _write_abstract_checkpoint(row, cache, view, include, ident)
        rows.append(row)

    out = pd.DataFrame(rows, columns=columns)
    out.attrs["n_requests"] = n_requests
    out.attrs["quota"] = quota
    return out

ABSTRACT_COLUMNS module-attribute

ABSTRACT_COLUMNS = ['scopus_id', 'doi', 'title', 'abstract', 'publication', 'date', 'year', 'citations']

corpus

corpus(records, by='doi', view='FULL', cache_dir=None, resume=True, **kwargs)

Enrich records (from :func:scopusflow.fetch.fetch_plan) with author keywords and structured references via Abstract Retrieval, returning a minimal, uniform shape close to what OpenAlex's works API already returns: id, title, year, keywords (a list of strings per row) and references (a DataFrame of cited works per row). This is meant for downstream tools that want to consume Scopus output without writing their own parsing layer, for example for keyword co-occurrence or citation-network analysis. It does not replace :func:to_bibtex/ :func:to_ris, which keep their own established interchange formats.

by selects which column of records ("doi" or "scopus_id") to look identifiers up by. view is passed to :func:scopus_abstract and defaults to "FULL", which in development returned a complete, correctly counted reference list for every document tried, unlike "REF", which returned an inconsistent, sometimes-truncated subset (see :func:scopus_abstract's documentation for the entitlement each view needs). cache_dir and resume are passed through unchanged, and are worth setting for anything beyond a handful of records, since this performs one Abstract Retrieval request per record, against its own, smaller weekly quota, separate from Search's. What that cost came to is carried through from :func:scopus_abstract: the number of requests made and the most recently parsed remaining-quota figure are attached as result.attrs["n_requests"] and result.attrs["quota"].

A record whose identifier is missing (NA/None) is dropped, with a warning naming how many.

The keywords column here is list[str] per row, split out of :func:scopus_abstract's joined authkeywords string, empty when the document has none or the field is unavailable. references carries pybliometrics' own native reference field set (see :func:scopus_abstract's documentation).

Source code in src/scopusflow/corpus.py

def corpus(
    records: pd.DataFrame,
    by: str = "doi",
    view: str = "FULL",
    cache_dir: str | None = None,
    resume: bool = True,
    **kwargs,
) -> pd.DataFrame:
    """Enrich ``records`` (from :func:`scopusflow.fetch.fetch_plan`) with author
    keywords and structured references via Abstract Retrieval, returning a
    minimal, uniform shape close to what OpenAlex's ``works`` API already
    returns: ``id``, ``title``, ``year``, ``keywords`` (a list of strings per
    row) and ``references`` (a DataFrame of cited works per row). This is
    meant for downstream tools that want to consume Scopus output without
    writing their own parsing layer, for example for keyword co-occurrence or
    citation-network analysis. It does not replace :func:`to_bibtex`/
    :func:`to_ris`, which keep their own established interchange formats.

    ``by`` selects which column of ``records`` ("doi" or "scopus_id") to look
    identifiers up by. ``view`` is passed to :func:`scopus_abstract` and
    defaults to "FULL", which in development returned a complete, correctly
    counted reference list for every document tried, unlike "REF", which
    returned an inconsistent, sometimes-truncated subset (see
    :func:`scopus_abstract`'s documentation for the entitlement each view
    needs). ``cache_dir`` and ``resume`` are passed through unchanged, and are
    worth setting for anything beyond a handful of records, since this
    performs one Abstract Retrieval request per record, against its own,
    smaller weekly quota, separate from Search's. What that cost came to is
    carried through from :func:`scopus_abstract`: the number of requests made
    and the most recently parsed remaining-quota figure are attached as
    ``result.attrs["n_requests"]`` and ``result.attrs["quota"]``.

    A record whose identifier is missing (``NA``/``None``) is dropped, with a
    warning naming how many.

    The `keywords` column here is `list[str]` per row, split out of
    :func:`scopus_abstract`'s joined `authkeywords` string, empty when the
    document has none or the field is unavailable. `references` carries
    pybliometrics' own native reference field set (see
    :func:`scopus_abstract`'s documentation).
    """
    required = {by, "title", "year"}
    if not required.issubset(records.columns):
        raise ValueError(
            f"records must have {sorted(required)} columns "
            "(as fetch_plan() returns)."
        )

    ids = records[by]
    keep = ids.notna()
    n_dropped = int((~keep).sum())
    if n_dropped:
        warnings.warn(
            f"Dropped {n_dropped} record(s) with no usable {by}.",
            stacklevel=2,
        )
    if not keep.any():
        raise ValueError("records has no usable identifiers to look up.")
    records = records.loc[keep].reset_index(drop=True)

    ab = scopus_abstract(
        list(records[by]), by=by, view=view, include=("references", "keywords"),
        cache_dir=cache_dir, resume=resume, **kwargs,
    )

    def _split(kw):
        if pd.isna(kw):
            return []
        return [k.strip() for k in kw.split(";")]

    out = pd.DataFrame({
        # The identifier used to look each record up, taken from `records`
        # directly, and never read back from `ab`: scopus_abstract() only
        # echoes the input identifier verbatim in its "doi"/"scopus_id"
        # column on a failed lookup (to keep the failing row identifiable);
        # on success that column holds whatever Scopus itself returns, which
        # is usually but not guaranteedly identical to the input.
        "id": list(records[by]),
        "title": records["title"],
        "year": records["year"],
        "keywords": [_split(kw) for kw in ab["authkeywords"]],
        "references": list(ab["references"]),
    })
    # A freshly constructed frame carries no attrs, so the wrapper that spends
    # the most quota would otherwise be the one reporting none of it.
    out.attrs.update(ab.attrs)
    return out

ScopusFlowForbiddenError

Bases: Exception

Raised when the Scopus API refuses a request (HTTP 403), most often because the configured key's entitlement does not cover the requested Abstract Retrieval view or field.

Source code in src/scopusflow/exceptions.py

class ScopusFlowForbiddenError(Exception):
    """Raised when the Scopus API refuses a request (HTTP 403), most often
    because the configured key's entitlement does not cover the requested
    Abstract Retrieval view or field."""

Records

Normalise results into one stable schema, tally the most frequent values, and export to reference-manager formats.

to_records

to_records(results, query=None, view=None)

Normalise a pybliometrics ScopusSearch().results list (named tuples) or a list of dicts into a tidy :data:RECORD_COLUMNS DataFrame.

Whatever the query type, the columns are the same, so the downstream DOI, diff and analysis helpers can rely on them.

When view="COMPLETE", an authkeywords column is added: the author- supplied keywords the Scopus Search API returns under that view, as a single string in Scopus's own " | "-delimited form (None when the document has none, or when the API omits the field for a given key's entitlement: this was observed directly against a live, otherwise fully-entitled key during development, on documents that do carry author keywords in Scopus itself, so an all-None column is more likely an entitlement gap worth raising with your Scopus/Elsevier account contact than genuinely absent data). Any other view, including the default None, reproduces :data:RECORD_COLUMNS exactly, so existing callers that never pass view see no change at all.

Source code in src/scopusflow/records.py

def to_records(results, query: str | None = None, view: str | None = None) -> pd.DataFrame:
    """Normalise a pybliometrics ``ScopusSearch().results`` list (named tuples)
    or a list of dicts into a tidy :data:`RECORD_COLUMNS` DataFrame.

    Whatever the query type, the columns are the same, so the downstream DOI,
    diff and analysis helpers can rely on them.

    When ``view="COMPLETE"``, an ``authkeywords`` column is added: the author-
    supplied keywords the Scopus Search API returns under that view, as a
    single string in Scopus's own ``" | "``-delimited form (``None`` when the
    document has none, or when the API omits the field for a given key's
    entitlement: this was observed directly against a live, otherwise
    fully-entitled key during development, on documents that do carry author
    keywords in Scopus itself, so an all-``None`` column is more likely an
    entitlement gap worth raising with your Scopus/Elsevier account contact
    than genuinely absent data). Any other ``view``, including the default
    ``None``, reproduces :data:`RECORD_COLUMNS` exactly, so existing callers
    that never pass ``view`` see no change at all.
    """
    add_keywords = view == "COMPLETE"
    columns = [*RECORD_COLUMNS, "authkeywords"] if add_keywords else RECORD_COLUMNS
    rows = []
    for i, r in enumerate(results or [], start=1):
        date = _get(r, "coverDate")
        row = {
            "entry_number": i,
            "scopus_id": _scopus_id(_get(r, "eid")),
            "doi": _get(r, "doi"),
            "title": _get(r, "title"),
            # pybliometrics joins multiple authors with ';' in author_names.
            "authors": _get(r, "author_names") or _get(r, "creator"),
            "year": _year(date),
            "date": date,
            "publication": _get(r, "publicationName"),
            "citations": _citations(_get(r, "citedby_count")),
            "query": query,
        }
        if add_keywords:
            row["authkeywords"] = _get(r, "authkeywords") or pd.NA
        rows.append(row)
    return pd.DataFrame(rows, columns=columns)

top

top(records, by='source', n=10)

Tally the most frequent sources or authors in a record set.

Source code in src/scopusflow/records.py

def top(records: pd.DataFrame, by: str = "source", n: int = 10) -> pd.DataFrame:
    """Tally the most frequent sources or authors in a record set."""
    if by == "source":
        values = records["publication"].dropna()
    elif by == "author":
        values = (
            records["authors"].dropna().str.split(";").explode().str.strip()
        )
        values = values[values != ""]
    else:
        raise ValueError("by must be 'source' or 'author'.")
    counts = values.value_counts().head(n)
    return counts.rename_axis("value").reset_index(name="n")

The tally below runs at build time over the bundled example harvest, so it needs no key.

records = sf.example_records()
out(sf.top(records, by="source", n=5))

value	n
ACS Applied Materials & Interfaces	8
Journal of Power Sources	5
Synthetic Metals	5
Electrochimica Acta	4
Scientific Reports	4

RECORD_COLUMNS module-attribute

RECORD_COLUMNS = ['entry_number', 'scopus_id', 'doi', 'title', 'authors', 'year', 'date', 'publication', 'citations', 'query']

scopus_combine

scopus_combine(*sets, dedupe=False)

Bind several record frames into one, renumbering entry_number.

This is the safe way to merge separate harvests: a plain concat leaves duplicate entry numbers, and nothing then records how many records went in.

Parameters:

Name	Type	Description	Default
*sets		Two or more record frames, or a single list of them.	()
dedupe	bool	When True, records sharing a Scopus identifier, or failing that a DOI (compared case-insensitively), are kept once.	False

Returns:

Type

Description

DataFrame

The merged records. Attributes describing a single retrieval, among them plan, total_results and cell_totals, are not carried over, since a set built from several harvests is none of them. The merge itself is recorded in attrs["combined"], a dict of n_in (records supplied), n_out (records kept), n_removed and deduplicated. That count exists only at the moment of the merge, and PRISMA-S asks for it (item 16), so :func:scopusflow.report.scopus_search_report reads it back from there so the item is answered.

Examples:

>>> import scopusflow as sf
>>> baseline = sf.example_records()
>>> later = sf.example_records()
>>> merged = sf.scopus_combine(baseline, later, dedupe=True)
>>> merged.attrs["combined"]["n_in"]
276
>>> len(merged)
149

Source code in src/scopusflow/combine.py

def scopus_combine(*sets, dedupe: bool = False) -> pd.DataFrame:
    """Bind several record frames into one, renumbering ``entry_number``.

    This is the safe way to merge separate harvests: a plain ``concat`` leaves
    duplicate entry numbers, and nothing then records how many records went in.

    Parameters
    ----------
    *sets:
        Two or more record frames, or a single list of them.
    dedupe:
        When ``True``, records sharing a Scopus identifier, or failing that a
        DOI (compared case-insensitively), are kept once.

    Returns
    -------
    pandas.DataFrame
        The merged records. Attributes describing a single retrieval, among
        them ``plan``, ``total_results`` and ``cell_totals``, are not carried
        over, since a set built from several harvests is none of them. The merge
        itself is recorded in ``attrs["combined"]``, a dict of ``n_in`` (records
        supplied), ``n_out`` (records kept), ``n_removed`` and
        ``deduplicated``. That count exists only at the moment
        of the merge, and PRISMA-S asks for it (item 16), so
        :func:`scopusflow.report.scopus_search_report` reads it back from there
        so the item is answered.

    Examples
    --------
    >>> import scopusflow as sf
    >>> baseline = sf.example_records()
    >>> later = sf.example_records()
    >>> merged = sf.scopus_combine(baseline, later, dedupe=True)
    >>> merged.attrs["combined"]["n_in"]
    276
    >>> len(merged)
    149
    """
    frames = list(sets)
    if len(frames) == 1 and isinstance(frames[0], (list, tuple)):
        frames = list(frames[0])
    if not frames or not all(isinstance(f, pd.DataFrame) for f in frames):
        raise ValueError("All inputs to scopus_combine() must be record frames.")

    out = pd.concat([_without_attrs(f) for f in frames], ignore_index=True)
    n_in = len(out)
    if dedupe:
        out = out[~_key(out).duplicated()].reset_index(drop=True)
    if len(out):
        out["entry_number"] = range(1, len(out) + 1)
    else:
        out = out.reindex(columns=list(out.columns) or list(RECORD_COLUMNS))
    out.attrs["combined"] = {
        "n_in": n_in,
        "n_out": len(out),
        "n_removed": n_in - len(out),
        "deduplicated": bool(dedupe),
    }
    return out

example_records

example_records()

Return the bundled example harvest as a records frame.

A fresh copy is returned on each call, so a caller may edit the result without disturbing anyone else's.

Returns:

Type	Description
DataFrame	138 records with the standard RECORD_COLUMNS schema, covering 2015 to 2024. The harvest is complete, so the rows per year are the real publications per year for the query. scopus_id is empty throughout, these records not having come from Scopus; eleven records carry no DOI and two no source title, exactly as they arrive.

Examples:

>>> import scopusflow as sf
>>> records = sf.example_records()
>>> len(records)
138
>>> sorted(records.columns) == sorted(sf.RECORD_COLUMNS)
True

Source code in src/scopusflow/data.py

def example_records() -> pd.DataFrame:
    """Return the bundled example harvest as a records frame.

    A fresh copy is returned on each call, so a caller may edit the result
    without disturbing anyone else's.

    Returns
    -------
    pandas.DataFrame
        138 records with the standard ``RECORD_COLUMNS`` schema, covering 2015
        to 2024. The harvest is complete, so the rows per
        year are the real publications per year for the query. ``scopus_id`` is
        empty throughout, these records not having come from Scopus; eleven
        records carry no DOI and two no source title, exactly as they arrive.

    Examples
    --------
    >>> import scopusflow as sf
    >>> records = sf.example_records()
    >>> len(records)
    138
    >>> sorted(records.columns) == sorted(sf.RECORD_COLUMNS)
    True
    """
    return _load().copy()

to_bibtex

to_bibtex(records)

Render records as a BibTeX string, one @article entry per row, with citation keys made unique within the export.

Source code in src/scopusflow/export.py

def to_bibtex(records: pd.DataFrame) -> str:
    """Render records as a BibTeX string, one ``@article`` entry per row, with
    citation keys made unique within the export."""
    if not isinstance(records, pd.DataFrame):
        raise ValueError("records must be a pandas DataFrame.")
    rows = [row for _, row in records.iterrows()]
    keys = _disambiguate(
        [_bibtex_key(r.get("authors"), r.get("year"), r.get("scopus_id")) for r in rows]
    )
    return "\n\n".join(_bibtex_entry(r, k) for r, k in zip(rows, keys, strict=True))

to_ris

to_ris(records)

Render records as an RIS string, one JOUR record per row.

Source code in src/scopusflow/export.py

def to_ris(records: pd.DataFrame) -> str:
    """Render records as an RIS string, one ``JOUR`` record per row."""
    if not isinstance(records, pd.DataFrame):
        raise ValueError("records must be a pandas DataFrame.")
    return "\n\n".join(_ris_entry(row) for _, row in records.iterrows())

Reporting

Write the search up for a methods section, to the PRISMA-S standard, from what the plan and the harvest already record.

scopus_search_report

scopus_search_report(x, plan=None, file=None)

Assemble a reproducible record of a Scopus search.

Turns a harvest, or a plan not yet run, into the search-strategy record a systematic review has to report: what was searched, exactly how, when, how much came back, and how much the API said there was. The record prints as a readable report, formats as a methods paragraph fit to paste into a manuscript, and writes as Markdown. The reporting standard it follows is PRISMA-S (Rethlefsen et al., 2021), together with the identification counts of the PRISMA 2020 flow diagram.

Everything in the record comes from the objects handed to it. The date of the search is the retrieved_at attribute :func:scopusflow.fetch.fetch_plan attaches, never the current time; the number of records the API reported as matching is the per-cell accounting in cell_totals, never an inference from the number of rows; and the duplicates removed are those :func:scopusflow.combine.scopus_combine recorded removing. Where an attribute is absent, as it is for a frame read back from CSV, for the bundled corpus, and for a harvest with a cell resumed from a checkpoint, the record says the field is unrecorded and fills nothing in. This matters most for completeness: a harvest whose reported total is unknown is never described as exhaustive.

The PRISMA-S map is decided the same way. Items the package holds evidence for (the database and platform, the full strategy, the limits, the date, the totals, and de-duplication where it was performed) are listed as supplied. The rest, among them peer review of the strategy, grey literature, other databases and citation searching, are listed as the author's to supply, because the package has no way to know them.

Parameters:

Name	Type	Description	Default
x		A records frame, which supplies the counts and the retrieval provenance and, through attrs["plan"], the plan; or a bare :class:~scopusflow.plan.SearchPlan for a search not yet run.	required
plan	SearchPlan | None	The plan describing x. Supply it when a records frame does not carry one, for instance one read back from CSV. An explicit plan takes precedence over the one x carries.	None
file		Path at which to write the record as Markdown. A file is written only when this is supplied, and only to the exact path given, so nothing is written to the working directory unless asked.	None

Returns:

Type	Description
SearchReport	The record. print shows it, report.format(style="paragraph") gives the methods paragraph.

References

Rethlefsen, M. L., Kirtley, S., Waffenschmidt, S., Ayala, A. P., Moher, D., Page, M. J., & Koffel, J. B. (2021). PRISMA-S: an extension to the PRISMA Statement for Reporting Literature Searches in Systematic Reviews. Systematic Reviews, 10, 39. https://doi.org/10.1186/s13643-020-01542-z

Examples:

A search described but not yet run. The record says so throughout rather than implying figures it cannot have.

>>> import scopusflow as sf
>>> plan = sf.SearchPlan("graphene supercapacitor", years=range(2015, 2025),
...                      field="TITLE-ABS-KEY", partition="year")
>>> report = sf.scopus_search_report(plan)
>>> report.n_records is None
True

The same search after a harvest. The bundled corpus stands in for one, since Scopus records may not be redistributed, so the attributes a live retrieval records are set here by hand.

>>> from datetime import datetime, timezone
>>> records = sf.example_records()
>>> records.attrs["plan"] = plan
>>> records.attrs["retrieved_at"] = datetime(2026, 7, 22, 9, 15,
...                                          tzinfo=timezone.utc)
>>> records.attrs["scopusflow_version"] = "0.3.0"
>>> report = sf.scopus_search_report(records)
>>> report.n_records
138
>>> "22 July 2026" in report.format(style="paragraph")
True

Source code in src/scopusflow/report.py

def scopus_search_report(x, plan: SearchPlan | None = None, file=None) -> SearchReport:
    """Assemble a reproducible record of a Scopus search.

    Turns a harvest, or a plan not yet run, into the search-strategy record a
    systematic review has to report: what was searched, exactly how, when, how
    much came back, and how much the API said there was. The record prints as a
    readable report, formats as a methods paragraph fit to paste into a
    manuscript, and writes as Markdown. The reporting standard it follows is
    PRISMA-S (Rethlefsen et al., 2021), together with the identification counts
    of the PRISMA 2020 flow diagram.

    Everything in the record comes from the objects handed to it. The date of
    the search is the ``retrieved_at`` attribute :func:`scopusflow.fetch.fetch_plan`
    attaches, never the current time; the number of records the API reported as
    matching is the per-cell accounting in ``cell_totals``, never an inference
    from the number of rows; and the duplicates removed are those
    :func:`scopusflow.combine.scopus_combine` recorded removing. Where an
    attribute is absent, as it is for a frame read back from CSV, for the
    bundled corpus, and for a harvest with a cell resumed from a checkpoint, the
    record says the field is unrecorded and fills nothing in. This matters
    most for completeness: a harvest whose reported total is unknown is never
    described as exhaustive.

    The PRISMA-S map is decided the same way. Items the package holds evidence
    for (the database and platform, the full strategy, the limits, the date, the
    totals, and de-duplication where it was performed) are listed as supplied.
    The rest, among them peer review of the strategy, grey literature, other
    databases and citation searching, are listed as the author's to supply,
    because the package has no way to know them.

    Parameters
    ----------
    x:
        A records frame, which supplies the counts and the retrieval provenance
        and, through ``attrs["plan"]``, the plan; or a bare
        :class:`~scopusflow.plan.SearchPlan` for a search not yet run.
    plan:
        The plan describing ``x``. Supply it when a records frame does not carry
        one, for instance one read back from CSV. An explicit plan takes
        precedence over the one ``x`` carries.
    file:
        Path at which to write the record as Markdown. A file is written only
        when this is supplied, and only to the exact path given, so nothing is
        written to the working directory unless asked.

    Returns
    -------
    SearchReport
        The record. ``print`` shows it, ``report.format(style="paragraph")``
        gives the methods paragraph.

    References
    ----------
    Rethlefsen, M. L., Kirtley, S., Waffenschmidt, S., Ayala, A. P., Moher, D.,
    Page, M. J., & Koffel, J. B. (2021). PRISMA-S: an extension to the PRISMA
    Statement for Reporting Literature Searches in Systematic Reviews.
    *Systematic Reviews*, *10*, 39. https://doi.org/10.1186/s13643-020-01542-z

    Examples
    --------
    A search described but not yet run. The record says so throughout rather
    than implying figures it cannot have.

    >>> import scopusflow as sf
    >>> plan = sf.SearchPlan("graphene supercapacitor", years=range(2015, 2025),
    ...                      field="TITLE-ABS-KEY", partition="year")
    >>> report = sf.scopus_search_report(plan)
    >>> report.n_records is None
    True

    The same search after a harvest. The bundled corpus stands in for one, since
    Scopus records may not be redistributed, so the attributes a live retrieval
    records are set here by hand.

    >>> from datetime import datetime, timezone
    >>> records = sf.example_records()
    >>> records.attrs["plan"] = plan
    >>> records.attrs["retrieved_at"] = datetime(2026, 7, 22, 9, 15,
    ...                                          tzinfo=timezone.utc)
    >>> records.attrs["scopusflow_version"] = "0.3.0"
    >>> report = sf.scopus_search_report(records)
    >>> report.n_records
    138
    >>> "22 July 2026" in report.format(style="paragraph")
    True
    """
    if plan is not None and not isinstance(plan, SearchPlan):
        raise ValueError("The plan must be a search plan.")
    if isinstance(x, SearchPlan):
        records = None
        plan = plan or x
    elif isinstance(x, pd.DataFrame):
        records = x
        if plan is None:
            carried = x.attrs.get("plan")
            plan = carried if isinstance(carried, SearchPlan) else None
    else:
        raise ValueError("A search report needs a record set or a search plan.")

    report = _build(records, plan)

    if file is not None:
        if not isinstance(file, (str, bytes)) and not hasattr(file, "__fspath__"):
            raise ValueError("The file must be a single non-empty path.")
        if isinstance(file, (str, bytes)) and not str(file).strip():
            raise ValueError("The file must be a single non-empty path.")
        # Newline fixed to LF, so the record is byte-identical on every platform
        # and against the R twin, which writes its text artefacts the same way.
        with open(file, "w", encoding="utf-8", newline="\n") as fh:
            fh.write(_render_markdown(report) + "\n")
    return report

SearchReport dataclass

The fields a search record is built from, and its three renderings.

Every attribute is either what the objects recorded or None, which the renderings spell out as unrecorded. cells is a frame of cell, limit, n_records and reported_total; prisma a frame of item, name, source ("record" or "author") and note.

Source code in src/scopusflow/report.py

@dataclass(eq=False)
class SearchReport:
    """The fields a search record is built from, and its three renderings.

    Every attribute is either what the objects recorded or ``None``, which the
    renderings spell out as unrecorded. ``cells`` is a frame of ``cell``,
    ``limit``, ``n_records`` and ``reported_total``; ``prisma`` a frame of
    ``item``, ``name``, ``source`` (``"record"`` or ``"author"``) and ``note``.
    """

    database: str = "Scopus"
    platform: str = "Elsevier Scopus Search API"
    query: str | None = None
    field: str | None = None
    expression: list[str] | None = None
    view: str | None = None
    page_size: int | None = None
    paging: str | None = None
    partition: str | None = None
    n_cells: int | None = None
    years: str | None = None
    cells: pd.DataFrame = dataclass_field(default_factory=pd.DataFrame)
    searched_at: datetime | None = None
    version: str | None = None
    n_records: int | None = None
    n_with_doi: int | None = None
    reported_total: int | None = None
    cells_reported: int = 0
    records_combined: int | None = None
    duplicates_removed: int | None = None
    deduplicated: bool | None = None
    snippet: str | None = None
    prisma: pd.DataFrame = dataclass_field(default_factory=pd.DataFrame)

    def format(self, style: str = "report") -> str:
        """Render the record: ``"report"`` for the readable record ``print``
        shows, ``"paragraph"`` for the methods paragraph, or ``"markdown"`` for
        the whole record as Markdown, which is what ``file`` writes."""
        if style == "report":
            return _render_report(self)
        if style == "paragraph":
            return _render_paragraph(self)
        if style == "markdown":
            return _render_markdown(self)
        raise ValueError("style must be 'report', 'paragraph' or 'markdown'.")

    def __str__(self) -> str:
        return _render_report(self)

    __repr__ = __str__

format

format(style='report')

Render the record: "report" for the readable record print shows, "paragraph" for the methods paragraph, or "markdown" for the whole record as Markdown, which is what file writes.

Source code in src/scopusflow/report.py

def format(self, style: str = "report") -> str:
    """Render the record: ``"report"`` for the readable record ``print``
    shows, ``"paragraph"`` for the methods paragraph, or ``"markdown"`` for
    the whole record as Markdown, which is what ``file`` writes."""
    if style == "report":
        return _render_report(self)
    if style == "paragraph":
        return _render_paragraph(self)
    if style == "markdown":
        return _render_markdown(self)
    raise ValueError("style must be 'report', 'paragraph' or 'markdown'.")

PRISMA_S_ITEMS module-attribute

PRISMA_S_ITEMS = ['Database name', 'Multi-database searching', 'Study registries', 'Online resources and browsing', 'Citation searching', 'Contacts', 'Other methods', 'Full search strategies', 'Limits and restrictions', 'Search filters', 'Prior work', 'Updates', 'Dates of searches', 'Peer review', 'Total records', 'Deduplication']

DOIs and change tracking

Extract clean DOIs and compare two retrievals to see exactly what changed.

extract_dois

extract_dois(records, dedupe=True)

Pull cleaned, optionally de-duplicated DOIs from records or a list.

Source code in src/scopusflow/diff.py

def extract_dois(records, dedupe: bool = True) -> list[str]:
    """Pull cleaned, optionally de-duplicated DOIs from records or a list."""
    dois = _clean(_as_dois(records))
    if not dedupe:
        return dois
    seen: set[str] = set()
    result: list[str] = []
    for d in dois:
        key = d.lower()
        if key not in seen:
            seen.add(key)
            result.append(d)
    return result

diff_dois

diff_dois(old, new)

Compare two retrievals; return a frame of (doi, status) where status is added, removed or unchanged (compared case-insensitively).

Source code in src/scopusflow/diff.py

def diff_dois(old, new) -> pd.DataFrame:
    """Compare two retrievals; return a frame of (doi, status) where status is
    ``added``, ``removed`` or ``unchanged`` (compared case-insensitively)."""
    old_d, new_d = extract_dois(old), extract_dois(new)
    old_keys = {d.lower() for d in old_d}
    new_keys = {d.lower() for d in new_d}
    rows = (
        [(d, "added") for d in new_d if d.lower() not in old_keys]
        + [(d, "removed") for d in old_d if d.lower() not in new_keys]
        + [(d, "unchanged") for d in new_d if d.lower() in old_keys]
    )
    df = pd.DataFrame(rows, columns=["doi", "status"])
    return df.sort_values(["status", "doi"]).reset_index(drop=True)

Analyse and visualise

Summarise a literature over time, compare topics within it, and turn the summaries into figures.

The plotting functions below are pure over the frame they are given, so the record examples in this section run at build time over the bundled example harvest, and the topic comparison over a frame of the documented shape. Nothing here contacts the Scopus API or needs a key. The functions that do retrieve counts are marked as such and are shown but not run. The guides linked from each one demonstrate them against a live key.

scopus_trend

scopus_trend(query, years, field=None, view='STANDARD', **kwargs)

Count Scopus hits for query in each of years without downloading them.

Each year is a cheap result-size lookup, so this gives a publication trend far faster than harvesting every record. field wraps the query in a Scopus field tag (see :data:scopusflow.query.FIELD_TAGS), the way :func:scopusflow.count.scopus_count and the R twin's scopus_trend() do; left None, the query is sent as given.

Source code in src/scopusflow/trend.py

def scopus_trend(
    query: str,
    years: Sequence[int],
    field: str | None = None,
    view: str = "STANDARD",
    **kwargs,
) -> pd.DataFrame:
    """Count Scopus hits for ``query`` in each of ``years`` without downloading them.

    Each year is a cheap result-size lookup, so this gives a publication trend
    far faster than harvesting every record. ``field`` wraps the query in a
    Scopus field tag (see :data:`scopusflow.query.FIELD_TAGS`), the way
    :func:`scopusflow.count.scopus_count` and the R twin's ``scopus_trend()``
    do; left ``None``, the query is sent as given.
    """
    if not query or not query.strip():
        raise ValueError("query must be a non-empty string.")
    years = list(years)
    if not years:
        raise ValueError("years must be a non-empty sequence.")
    # Validated before it reaches the query: the year used to be
    # interpolated raw, so a float year sent the API "PUBYEAR IS 2015.0" while
    # the result was filed under 2015.
    years = _check_years(years)
    # Wrapped once, before the loop: the tag applies to the query alone, never
    # to the year
    # filter folded in beside it.
    query = wrap_field(query, field)

    from pybliometrics.scopus import ScopusSearch  # imported lazily; needs a key

    counts: dict[int, int] = {}
    for y in years:
        search = ScopusSearch(
            f"{query} AND PUBYEAR IS {y}", view=view, download=False, **kwargs
        )
        counts[y] = int(search.get_results_size())
    return _trend_frame(counts)

This asks the API for a count per year, so it cannot run at build time. See Analysing a literature for the worked example, and year_counts below for the offline equivalent over records you already hold.

year_counts

year_counts(records)

Count records per publication year, dropping rows with a missing year.

Returns a :data:TREND_COLUMNS frame sorted ascending by year, with both columns as plain integers.

Source code in src/scopusflow/trend.py

def year_counts(records: pd.DataFrame) -> pd.DataFrame:
    """Count records per publication year, dropping rows with a missing year.

    Returns a :data:`TREND_COLUMNS` frame sorted ascending by year, with both
    columns as plain integers.
    """
    years = pd.to_numeric(records["year"], errors="coerce").dropna()
    counts = {int(y): int(n) for y, n in years.astype(int).value_counts().items()}
    return _trend_frame(counts)

out(sf.year_counts(records))

year	n
2015	15
2016	9
2017	10
2018	15
2019	19
2020	13
2021	13
2022	15
2023	15
2024	14

TREND_COLUMNS module-attribute

TREND_COLUMNS = ['year', 'n']

compare_topics

compare_topics(reference_query, comparison_terms, years, field=None, view='STANDARD', **kwargs)

Compare comparison topics against a reference topic over the years.

Returns a :data:COMPARISON_COLUMNS frame. One count request per term per year, plus one per year for the reference topic, so keep the term and year counts modest to stay within quota.

Source code in src/scopusflow/compare.py

def compare_topics(reference_query: str, comparison_terms, years: Sequence[int],
                   field: str | None = None, view: str = "STANDARD",
                   **kwargs) -> pd.DataFrame:
    """Compare comparison topics against a reference topic over the years.

    Returns a :data:`COMPARISON_COLUMNS` frame. One count request per term per
    year, plus one per year for the reference topic, so keep the term and year
    counts modest to stay within quota.
    """
    if not reference_query or not str(reference_query).strip():
        raise ValueError("reference_query must be a non-empty string.")
    if isinstance(comparison_terms, str):
        comparison_terms = [comparison_terms]
    terms = [str(t).strip() for t in comparison_terms]
    if not terms or any(not t for t in terms):
        raise ValueError("comparison_terms must be a non-empty list of non-empty terms.")
    if years is None or not list(years):
        raise ValueError("years must be a non-empty sequence.")
    ys = sorted(set(_check_years(years)))

    from pybliometrics.scopus import ScopusSearch  # imported lazily; needs a key

    def size(query: str, year: int) -> int:
        full = f"{query} AND PUBYEAR IS {year}"
        return int(ScopusSearch(full, view=view, download=False, **kwargs).get_results_size())

    ref_query = wrap_field(str(reference_query).strip(), field)
    # One count step per term, plus the reference; logged as "Cell k/N:" so the
    # app's progress parser can drive a bar (mirrors the R verbose output).
    total = len(terms) + 1
    logger.info("Cell 1/%d: counting reference across %d year(s)", total, len(ys))
    ref_counts = {y: size(ref_query, y) for y in ys}

    comparison = []
    for i, term in enumerate(terms):
        logger.info("Cell %d/%d: counting '%s'", i + 2, total, term)
        cmp_query = f"{ref_query} AND {wrap_field(term, field)}"
        comparison.append((term, cmp_query, {y: size(cmp_query, y) for y in ys}))

    return _assemble(str(reference_query).strip(), ref_query, ref_counts, comparison, ys)

This makes one count request per term per year, so it cannot run at build time. See Comparing topics for the worked example.

COMPARISON_COLUMNS module-attribute

COMPARISON_COLUMNS = ['query', 'query_type', 'abridged_query', 'year', 'n', 'reference_n', 'comparison_percentage', 'average_comparison_percentage']

scopus_intersections

scopus_intersections(concepts, intersections=None, abbrev=None, sep=' × ', years=None, field=None, view='STANDARD', verbose=False, **kwargs)

Count each concept and each requested intersection of concepts.

Parameters:

Name	Type	Description	Default
concepts	Mapping[str, str]	A mapping whose keys are display labels and whose values are search terms (wrapped in field when one is given) or complete field-tagged query expressions such as "TITLE(virtual reality)" (used as-is). The labels must be unique.	required
intersections	Sequence[Sequence[str]] | None	An optional sequence of sequences, each naming two or more distinct concept labels whose intersection should be counted, for example [["A", "B"], ["A", "B", "C"]]. A single flat sequence of labels is taken as one intersection.	None
abbrev	Mapping[str, str] | None	An optional mapping of short labels, keyed by concept label and used only when composing intersection labels, so those rows stay readable while the concept rows keep their full names.	None
sep	str	The separator joining member labels in an intersection label; defaults to a multiplication sign between spaces.	' × '
years	Sequence[int] | None	An optional inclusive year range applied to every count.	None
field	str | None	An optional Scopus field tag wrapped around each concept value that is not already a complete field-tagged expression (see :data:scopusflow.query.FIELD_TAGS).	None
view	str	The Scopus search view, "STANDARD" or "COMPLETE".	'STANDARD'
verbose	bool	When True, progress is reported on the scopusflow logger.	False

Returns:

Type	Description
DataFrame	One row per concept and per intersection, with columns label, query, n (the count, as a nullable integer), type ("concept" or "intersection"), size (the number of member concepts) and members (the member labels, joined by "; "). The years restriction, when given, is stored in df.attrs["years"].

Notes

This performs one count request per concept and per intersection, so it needs a valid API key and internet access, exactly as :func:scopusflow.count.scopus_count does.

Source code in src/scopusflow/intersections.py

def scopus_intersections(
    concepts: Mapping[str, str],
    intersections: Sequence[Sequence[str]] | None = None,
    abbrev: Mapping[str, str] | None = None,
    sep: str = " × ",
    years: Sequence[int] | None = None,
    field: str | None = None,
    view: str = "STANDARD",
    verbose: bool = False,
    **kwargs,
) -> pd.DataFrame:
    """Count each concept and each requested intersection of concepts.

    Parameters
    ----------
    concepts:
        A mapping whose keys are display labels and whose values are search terms
        (wrapped in ``field`` when one is given) or complete field-tagged query
        expressions such as ``"TITLE(virtual reality)"`` (used as-is). The labels
        must be unique.
    intersections:
        An optional sequence of sequences, each naming two or more distinct
        concept labels whose intersection should be counted, for example
        ``[["A", "B"], ["A", "B", "C"]]``. A single flat sequence of labels is
        taken as one intersection.
    abbrev:
        An optional mapping of short labels, keyed by concept label and used only
        when composing intersection labels, so those rows stay readable while the
        concept rows keep their full names.
    sep:
        The separator joining member labels in an intersection label; defaults to
        a multiplication sign between spaces.
    years:
        An optional inclusive year range applied to every count.
    field:
        An optional Scopus field tag wrapped around each concept value that is not
        already a complete field-tagged expression (see
        :data:`scopusflow.query.FIELD_TAGS`).
    view:
        The Scopus search view, ``"STANDARD"`` or ``"COMPLETE"``.
    verbose:
        When ``True``, progress is reported on the ``scopusflow`` logger.

    Returns
    -------
    pandas.DataFrame
        One row per concept and per intersection, with columns ``label``,
        ``query``, ``n`` (the count, as a nullable integer), ``type``
        (``"concept"`` or ``"intersection"``), ``size`` (the number of member
        concepts) and ``members`` (the member labels, joined by ``"; "``). The
        ``years`` restriction, when given, is stored in ``df.attrs["years"]``.

    Notes
    -----
    This performs one count request per concept and per intersection, so it needs
    a valid API key and internet access, exactly as
    :func:`scopusflow.count.scopus_count` does.
    """
    out = _intersection_rows(concepts, intersections, abbrev, sep, field)

    ys = sorted(set(_check_years(years))) if years else None
    if verbose:
        n_inter = int((out["type"] == "intersection").sum())
        logger.info(
            "Counting %d queries (%d intersection%s).",
            len(out), n_inter, "" if n_inter == 1 else "s",
        )

    counts: list[int] = []
    for label, query in zip(out["label"], out["query"], strict=True):
        if verbose:
            logger.info("Counting %s", label)
        counts.append(scopus_count(query, years=ys, view=view, **kwargs))
    out["n"] = pd.array(counts, dtype="Int64")

    out = out[["label", "query", "n", "type", "size", "members"]]
    out.attrs["years"] = ys
    return out

This makes one count request per concept and per intersection, so it cannot run at build time. See Analysing a literature for the worked example.

plot_trend

plot_trend(trend, ax=None)

Plot publication counts over time as a filled area, line and points.

trend has columns ["year", "n"]; returns the matplotlib Axes.

Source code in src/scopusflow/plots.py

def plot_trend(trend: pd.DataFrame, ax=None):
    """Plot publication counts over time as a filled area, line and points.

    ``trend`` has columns ``["year", "n"]``; returns the matplotlib ``Axes``.
    """
    import matplotlib.pyplot as plt

    if ax is None:
        _, ax = plt.subplots()

    years = trend["year"]
    counts = trend["n"]
    ax.fill_between(years, counts, alpha=0.16, color=_TREND_COLOUR)
    ax.plot(years, counts, color=_TREND_COLOUR, linewidth=2)
    ax.scatter(years, counts, color=_TREND_COLOUR, s=18, zorder=3)
    ax.set_ylim(bottom=0)
    ax.set_xlabel("Year")
    ax.set_ylabel("Records")
    _clean_axes(ax)
    return ax

sf.plot_trend(sf.year_counts(records))
show()

plot_top

plot_top(top, ax=None)

Plot a horizontal bar chart of the most frequent values, largest on top.

top has columns ["value", "n"] (from :func:scopusflow.records.top); each bar carries its count at its end, matching the R plot_scopus_top. Returns the matplotlib Axes.

Source code in src/scopusflow/plots.py

def plot_top(top: pd.DataFrame, ax=None):
    """Plot a horizontal bar chart of the most frequent values, largest on top.

    ``top`` has columns ``["value", "n"]`` (from :func:`scopusflow.records.top`);
    each bar carries its count at its end, matching the R ``plot_scopus_top``.
    Returns the matplotlib ``Axes``.
    """
    import matplotlib.pyplot as plt

    # An all-missing publication column tallies to no rows at all, which would
    # otherwise surface as an opaque max() error on the label widths below.
    if len(top) == 0:
        raise ValueError("The tally has no rows to plot.")

    if ax is None:
        _, ax = plt.subplots()

    # Reverse so the largest count sits at the top of the chart.
    ordered = top.iloc[::-1]
    bars = ax.barh(ordered["value"].astype(str), ordered["n"], color=_TOP_COLOUR)
    labels = [f"{int(n):,}" for n in ordered["n"]]
    ax.bar_label(bars, labels=labels, padding=3, fontsize=8, color="#4d4d4d")
    # Headroom derived from the widest label, so the count on the longest bar
    # stays inside the axes, where a wide one would otherwise clip at the
    # right edge. Mirrors the
    # R plot's label-width-derived axis expansion.
    ax.margins(x=0.04 + 0.024 * max(len(lab) for lab in labels))
    ax.set_xlim(left=0)
    ax.set_xlabel("Records")
    ax.set_ylabel("")
    _clean_axes(ax)
    return ax

sf.plot_top(sf.top(records, by="source"))
show()

plot_comparison

plot_comparison(comparison, highlight=None, interval=True, counts_in_legend=True, ax=None, legend_inside=False)

Plot each comparison topic's share of the reference literature over time.

comparison is the frame from :func:scopusflow.compare.compare_topics. With interval a shaded Wilson band shows how stable each yearly share is (illustrative, never a confidence interval, since Scopus counts are exact). highlight names one topic to draw in an accent colour, the rest in grey. With counts_in_legend (the default) each label carries the topic's total record count, for example machine learning (n = 1,204).

A legend is only drawn when there are too many topics to label the lines directly. legend_inside governs where it then sits: with the default False matplotlib chooses the least-obtrusive spot automatically (loc="best"); with True the legend is placed inside the axes, in whichever corner has the most free space, saving the width an outside legend would otherwise take. Returns the matplotlib Axes.

Source code in src/scopusflow/plots.py

def plot_comparison(comparison: pd.DataFrame, highlight=None, interval: bool = True,
                    counts_in_legend: bool = True, ax=None, legend_inside: bool = False):
    """Plot each comparison topic's share of the reference literature over time.

    ``comparison`` is the frame from :func:`scopusflow.compare.compare_topics`.
    With ``interval`` a shaded Wilson band shows how stable each yearly share is
    (illustrative, never a confidence interval, since Scopus counts are exact).
    ``highlight`` names one topic to draw in an accent colour, the rest in grey.
    With ``counts_in_legend`` (the default) each label carries the topic's total
    record count, for example ``machine learning (n = 1,204)``.

    A legend is only drawn when there are too many topics to label the lines
    directly. ``legend_inside`` governs where it then sits: with the default
    ``False`` matplotlib chooses the least-obtrusive spot automatically
    (``loc="best"``); with ``True`` the legend is placed inside the axes, in
    whichever corner has the most free space, saving the width an outside legend
    would otherwise take. Returns the matplotlib ``Axes``.
    """
    import matplotlib.pyplot as plt
    import matplotlib.ticker as mticker

    required = {"query_type", "abridged_query", "year", "comparison_percentage",
                "average_comparison_percentage"}
    if not required.issubset(comparison.columns):
        raise ValueError("comparison must be a topic-comparison frame.")

    comp_all = comparison[comparison["query_type"] == "comparison"]
    # A year whose reference has no records carries no defined share; it is
    # dropped and noted in the caption, mirroring the R plot.
    n_missing = int(comp_all["comparison_percentage"].isna().sum())
    df = comp_all[comp_all["comparison_percentage"].notna()].copy()
    if df.empty:
        raise ValueError("comparison has no comparison topics with a finite share to plot.")

    # The reference topic names the subtitle (it is the 100% denominator).
    ref_rows = comparison[comparison["query_type"] == "reference"]
    ref_names = ref_rows["abridged_query"].dropna().unique() if len(ref_rows) else []
    ref_label = str(ref_names[0]) if len(ref_names) == 1 else None

    order = (df.groupby("abridged_query")["average_comparison_percentage"].first()
             .reset_index()
             .sort_values(["average_comparison_percentage", "abridged_query"],
                          ascending=[False, True]))
    topics = list(order["abridged_query"])
    if highlight is not None and highlight not in topics:
        raise ValueError(f"highlight must be one of: {', '.join(topics)}.")

    # Optionally append each topic's total record count to its label.
    has_counts = counts_in_legend and "n" in df.columns
    totals = df.groupby("abridged_query")["n"].sum() if has_counts else None

    def _label(topic):
        return f"{topic} (n = {int(totals[topic]):,})" if has_counts else topic

    if ax is None:
        _, ax = plt.subplots()
    cmap = plt.get_cmap("viridis")
    spread = max(len(topics) - 1, 1)
    has_band = interval and {"n", "reference_n"}.issubset(df.columns)
    label_points = []

    for i, topic in enumerate(topics):
        sub = df[df["abridged_query"] == topic].sort_values("year")
        is_hi = highlight == topic
        if highlight is not None:
            colour = "#BB5566" if is_hi else "#BFBFBF"
            width = 1.6 if is_hi else 0.8
        else:
            colour = cmap(0.05 + 0.8 * i / spread)
            width = 1.4
        if has_band and (highlight is None or is_hi):
            lo, up = _wilson(sub["n"].to_numpy(), sub["reference_n"].to_numpy())
            ax.fill_between(sub["year"], lo, up, color=colour, alpha=0.16, linewidth=0)
        ax.plot(sub["year"], sub["comparison_percentage"], color=colour,
                linewidth=width, label=_label(topic))
        ax.scatter(sub["year"], sub["comparison_percentage"], color=colour, s=14, zorder=3)
        last = sub.iloc[-1]
        label_points.append((float(last["year"]), float(last["comparison_percentage"]),
                             _label(topic), colour, is_hi))

    # Cap the y-axis at the next 5% above the data (and bands), as the R plot
    # does, to remove dead headroom.
    if has_band:
        _, band_upper = _wilson(df["n"].to_numpy(), df["reference_n"].to_numpy())
        top = max(float(df["comparison_percentage"].max()), float(band_upper.max()))
    else:
        top = float(df["comparison_percentage"].max())
    ymax = min(100, math.ceil(top / 5) * 5)
    ax.set_ylim(0, ymax)

    # Label the lines directly when they fit legibly; otherwise fall back to a
    # legend. The gap is the minimum vertical separation between labels.
    gap = ymax * 0.055
    if highlight is not None:
        to_label = [p for p in label_points if p[4]]
    elif len(topics) <= 8 and (len(topics) - 1) * gap <= ymax:
        to_label = label_points
    else:
        to_label = []
        if legend_inside:
            legend_loc = _free_corner(
                df["year"], df["comparison_percentage"],
                (float(df["year"].min()), float(df["year"].max())), (0.0, ymax))
        else:
            legend_loc = "best"
        ax.legend(fontsize=8, loc=legend_loc, frameon=False,
                  ncol=2 if len(topics) > 8 else 1)

    # Place each label just past its line's endpoint. The labels are spread
    # apart at the end (see _decollide_once), once the final layout is known,
    # so none overlaps another.
    anns, label_xs, label_y_true = [], [], []
    if to_label:
        years = df["year"]
        dx = (float(years.max()) - float(years.min())) * 0.015 + 0.1
        for x, y_true, topic, colour, _is_hi in to_label:
            # No leader line: where labels converge they are nudged apart, and a
            # leader to a nudged label then cuts across the neighbouring labels'
            # text. The colour match and shared top-to-bottom order keep each
            # label tied to its line without one.
            anns.append(ax.annotate(
                topic, xy=(x, y_true), xytext=(x + dx, y_true), textcoords="data",
                va="center", ha="left", fontsize=8, color=colour,
                annotation_clip=False,
            ))
            label_xs.append(x + dx)
            label_y_true.append(y_true)
    ax.set_xlabel("")
    ax.set_ylabel("Share of reference records")
    ax.yaxis.set_major_formatter(mticker.PercentFormatter(xmax=100, decimals=0))
    ax.set_title("Topic share within a reference literature, over time",
                 loc="left", fontsize=12, pad=22)
    if ref_label:
        ax.annotate(
            f"Each line: % of '{ref_label}' records that also match the topic",
            xy=(0, 1), xycoords="axes fraction", xytext=(0, 6),
            textcoords="offset points", ha="left", va="bottom",
            fontsize=9, color="#555555",
        )
    # A caption that names the source and guards against reading the illustrative
    # Wilson band as an inferential confidence interval (it is not).
    caption = (f"Source: 'Scopus' Search API. Years {int(df['year'].min())} "
               f"to {int(df['year'].max())}.")
    if has_band:
        caption += ("\nShaded band: illustrative Wilson stability range "
                    "(not a confidence interval), wider where the reference set is small.")
    if n_missing > 0:
        plural = "" if n_missing == 1 else "s"
        caption += (f"\n{n_missing} year-topic value{plural} omitted for want "
                    "of reference records.")
    ax.annotate(caption, xy=(0, 0), xycoords="axes fraction", xytext=(0, -24),
                textcoords="offset points", ha="left", va="top",
                fontsize=7.5, color="#737373")
    _clean_axes(ax)

    if anns:
        # Re-spread the labels on every draw, measuring the rendered text height in
        # the layout that actually applies, so they never overlap however the lines
        # converge or the figure is sized. Running on the draw means it stays
        # correct after the caller tightens the layout. A guard stops the redraw it
        # requests from recursing once the positions have settled.
        _busy = {"on": False}

        def _on_draw(_event=None):
            if _busy["on"]:
                return
            _busy["on"] = True
            try:
                if _decollide_once(ax, anns, label_xs, label_y_true, ymax, gap):
                    ax.figure.canvas.draw_idle()
            finally:
                _busy["on"] = False

        ax.figure.canvas.mpl_connect("draw_event", _on_draw)
        # An initial draw so a one-shot render (savefig without a prior draw) still
        # gets de-collided labels.
        try:
            ax.figure.canvas.draw()
        except Exception:
            pass
    return ax

years = list(range(2013, 2022))
ref_n = [400, 550, 700, 850, 1000, 1150, 1300, 1450, 1600]
shares = {"computer vision": 34.0, "natural language processing": 24.0,
          "medical imaging": 11.0, "drug discovery": 6.0}

rows = [{"query": "deep learning", "query_type": "reference",
         "abridged_query": "deep learning", "year": year, "n": n,
         "reference_n": n, "comparison_percentage": 100.0,
         "average_comparison_percentage": 100.0}
        for year, n in zip(years, ref_n)]
for topic, end in shares.items():
    for i, (year, n) in enumerate(zip(years, ref_n)):
        pct = end * (0.45 + 0.55 * i / (len(years) - 1))
        rows.append({"query": topic, "query_type": "comparison",
                     "abridged_query": topic, "year": year,
                     "n": int(pct * n / 100), "reference_n": n,
                     "comparison_percentage": pct,
                     "average_comparison_percentage": end})

comparison = pd.DataFrame(rows, columns=sf.COMPARISON_COLUMNS)
sf.plot_comparison(comparison)
show()

plot_scopus_intersections

plot_scopus_intersections(x, highlight=None, highlight_label=None, ax=None)

Plot concept and intersection counts as a horizontal lollipop chart.

x is a frame from :func:scopusflow.intersections.scopus_intersections (columns label, n and type). Counts run along a logarithmic x axis, so fields spanning orders of magnitude sit together with their smaller intersections. Rows named in highlight are drawn in the focal colour, and highlight_label names them in the legend (derived from their type when unset). Returns the matplotlib Axes. Mirrors the R plot_scopus_intersections.

Source code in src/scopusflow/plots.py

def plot_scopus_intersections(x, highlight=None, highlight_label=None, ax=None):
    """Plot concept and intersection counts as a horizontal lollipop chart.

    ``x`` is a frame from :func:`scopusflow.intersections.scopus_intersections`
    (columns ``label``, ``n`` and ``type``). Counts run along a logarithmic x
    axis, so fields spanning orders of magnitude sit together with their smaller
    intersections. Rows named in ``highlight`` are drawn in the focal colour, and
    ``highlight_label`` names them in the legend (derived from their type when
    unset). Returns the matplotlib ``Axes``. Mirrors the R
    ``plot_scopus_intersections``.
    """
    import warnings

    import matplotlib.pyplot as plt
    from matplotlib.lines import Line2D
    from matplotlib.ticker import FuncFormatter, NullFormatter

    required = {"label", "n", "type"}
    if not required.issubset(x.columns):
        raise ValueError(
            "x must have columns 'label', 'n' and 'type' (see scopus_intersections())."
        )
    df = x.copy()
    if len(df) == 0:
        raise ValueError("The intersections frame has no rows to plot.")
    if df["label"].duplicated().any():
        raise ValueError(
            "The 'label' column must be unique to place each row on its own line."
        )

    if highlight is None:
        highlight = []
    elif isinstance(highlight, str):
        highlight = [highlight]
    else:
        highlight = list(highlight)
    known = set(df["label"])
    if highlight and any(h not in known for h in highlight):
        raise ValueError(
            "highlight must name rows to accent, among: "
            + ", ".join(map(str, df["label"])) + "."
        )

    counts = pd.to_numeric(df["n"], errors="coerce")
    keep = counts.notna() & (counts > 0)
    n_dropped = int((~keep).sum())
    if n_dropped:
        warnings.warn(
            f"{n_dropped} row(s) without a positive count cannot sit on a log axis "
            "and were dropped.",
            stacklevel=2,
        )
    df = df.loc[keep].copy()
    df["n"] = counts[keep].astype(float)
    if len(df) == 0:
        raise ValueError("No row has a positive count to place on the log axis.")

    if highlight_label is None:
        hi_types = set(df.loc[df["label"].isin(highlight), "type"])
        highlight_label = {
            frozenset({"intersection"}): "Focal intersection",
            frozenset({"concept"}): "Focal concept",
        }.get(frozenset(hi_types), "Focal set")

    df["grp"] = [
        "highlight" if lab in highlight else typ
        for lab, typ in zip(df["label"], df["type"], strict=True)
    ]
    df = df.sort_values("n", kind="stable").reset_index(drop=True)

    colours = {
        "concept": _CONCEPT_COLOUR,
        "intersection": _INTERSECTION_COLOUR,
        "highlight": _HIGHLIGHT_COLOUR,
    }
    legend_names = {
        "concept": "Concept",
        "intersection": "Intersection",
        "highlight": highlight_label,
    }

    lo = max(1.0, df["n"].min()) * 0.55  # the smallest point clears the axis
    hi = df["n"].max() * 4               # headroom for the widest count label
    # A constant ratio (not increment) beyond each point renders as a constant
    # pixel gap on a log axis, mirroring the R plot's label placement.
    gap_mult = 10 ** (0.024 * math.log10(hi / lo))

    if ax is None:
        _, ax = plt.subplots()

    for y, (_, row) in enumerate(df.iterrows()):
        colour = colours[row["grp"]]
        ax.hlines(y, lo, row["n"], color=colour, linewidth=1.6, zorder=1)
        ax.scatter(row["n"], y, color=colour, s=42, zorder=2)
        ax.text(row["n"] * gap_mult, y, f"{int(row['n']):,}",
                va="center", ha="left", fontsize=8, color="#4d4d4d")

    ax.set_yticks(range(len(df)))
    ax.set_yticklabels(df["label"])
    ax.set_xscale("log")
    ax.set_xlim(lo, hi)
    ax.xaxis.set_major_formatter(
        FuncFormatter(lambda v, _pos: f"{int(v):,}" if v >= 1 else f"{v:g}")
    )
    ax.xaxis.set_minor_formatter(NullFormatter())
    ax.set_xlabel("Records (log scale)")
    ax.set_title("Records matching each concept and intersection")
    _clean_axes(ax)

    present = [g for g in ("concept", "intersection", "highlight")
               if (df["grp"] == g).any()]
    handles = [
        Line2D([0], [0], marker="o", linestyle="", markersize=7,
               color=colours[g], label=legend_names[g])
        for g in present
    ]
    if handles:
        ax.legend(handles=handles, loc="upper left", frameon=False, fontsize=8)
    return ax

sets = pd.DataFrame({
    "label": ["semantic priming", "mental simulation",
              "semantic priming × mental simulation"],
    "query": ["TITLE-ABS-KEY(semantic priming)",
              "TITLE-ABS-KEY(mental simulation)",
              "(TITLE-ABS-KEY(semantic priming)) AND "
              "(TITLE-ABS-KEY(mental simulation))"],
    "n": pd.array([6600, 2600, 18], dtype="Int64"),
    "type": ["concept", "concept", "intersection"],
    "size": [1, 1, 2],
    "members": ["semantic priming", "mental simulation",
                "semantic priming; mental simulation"],
})
sf.plot_scopus_intersections(
    sets, highlight=["semantic priming × mental simulation"]
)
show()