Research Evidence Collector

Build a Python collector that extracts scholarly sections and exports evidence.

Introduction

30 Second Summary

Finding a useful research paper is only the start. The evidence worth keeping may be buried deep inside the article.

In this project, you will build a Windows-friendly Python collector that turns any scholarly topic into structured evidence from the OpenAlex API. Each record keeps its abstract plus available results and conclusion text.

What You'll Build

Your finished collector turns a topic into an output folder filled with traceable evidence text and a clear extraction status for every article.

By the end of this project, you'll have:

  • A topic-driven collection that retrieves up to 90 relevant scholarly articles while keeping the source details needed to trace each one.
  • Section-aware evidence text that combines the abstract with Results or Findings plus Conclusion when both target sections exist. Missing target sections receive the explicit abstract_fallback label.
  • Portable exports in CSV, JSONL, and one readable text file per paper.
  • Secret Mission: Add resumable cursor paging so one topic can continue toward 500 papers across multiple daily API budgets.

Are there any prerequisites?

You need a Windows computer with internet access plus Python 3.10 or newer and VS Code.

The guide covers the Requests 2.34.2 environment and OpenAlex account setup.

Before We Start

A clear topic gives your collector a real purpose before the hands-on work begins. Take a moment to choose what you want to investigate and how reusable Abstract, Results or Findings, and Conclusion text could support your research or knowledge-base workflow.

Prepare the Windows Project

Your collector uses Python to request scholarly content from OpenAlex. Those metered requests need a reproducible environment before the collector can use them safely.

An isolated environment pins Requests to version 2.34.2 inside the collector folder. A verified account gives you a private API key for authenticated content requests.

In this step, get ready to:
  • Verify Python 3.10 or newer.
  • Prepare an isolated Windows workspace for the collector.
  • Activate private access to OpenAlex content.
Prepare the Python workspace

Requests 2.34.2 requires Python 3.10 or newer. The version check prevents an incompatible interpreter from reaching the installation step.

  • Press the Windows key to open Windows search.
  • Type PowerShell and press Enter to open it.
  • Check your installed Python version by running this command:
py --version

What Does This Command Check?

The Windows Python launcher reports the interpreter version it can run. That result determines whether your current installation supports this project.

✔️ I see a supported version

Your Python installation is ready. You have cleared the project’s first compatibility check.

ⓧ I see an older version

The installed interpreter cannot run Requests 2.34.2. Install a current Python 3 release before continuing.

  • Visit python.org in your browser.
  • Download a current Python 3 release for Windows.
  • Complete the Windows installation with the Python launcher available.
  • Restart PowerShell through Windows search.
  • Check the updated version by running this command:
py --version

What Should You See?

The command should now report Python 3.10 or newer. Continue once the installed version meets that minimum.

Still Seeing the Older Version?

  • Restart VS Code after the Python installation finishes.
  • Restart PowerShell so the launcher can detect the new installation.
  • Ask for help with the Windows launcher if it still reports the previous installation.

ⓧ Command not found

Windows cannot currently find the Python launcher. Installing a current Python 3 release provides the launcher this project uses.

  • Visit python.org in your browser.
  • Download a current Python 3 release for Windows.
  • Complete the Windows installation with the Python launcher available.
  • Restart PowerShell through Windows search.
  • Check the installation by running this command:
py --version

What Should You See?

The command should report Python 3.10 or newer. That confirms the launcher is available.

Still Missing the Python Launcher?

  • Restart Windows if the completed installation is still unavailable.
  • Run the Python installer again if the launcher was omitted.
  • Ask for help if PowerShell still cannot locate Python.

The project folder gives every script and export a predictable home on your Desktop.

  • Press the Windows key to open Windows search.
  • Type File Explorer and press Enter to open it.
  • Use the left sidebar to display your Desktop.
  • Use the toolbar’s new-folder control to create a folder.
  • Name the folder topic-evidence-collector.

You should now see the topic-evidence-collector folder on your Desktop.

  • Press the Windows key to open Windows search.
  • Type VS Code and press Enter to open it.
  • Use the top menu’s folder-opening option to display the folder picker.
  • Select the topic-evidence-collector folder from your Desktop.

VS Code should now show topic-evidence-collector in its Explorer sidebar.

The requirements.txt file records the exact dependency version for this collector. That pin makes later installations reproducible.

  • Use the new-file control in the VS Code Explorer sidebar to create requirements.txt.
  • Add the dependency pin by pasting this line:
requests==2.34.2

What Does This File Do?

The version pin tells Python to install the same Requests release every time. Your collector gets a known HTTP client instead of an unpredictable future version.

  • Save requirements.txt.
  • Confirm the VS Code Explorer sidebar lists requirements.txt inside topic-evidence-collector.

Cannot Find the Requirements File?

  • Confirm VS Code shows topic-evidence-collector at the top of the Explorer sidebar.
  • Check that the file name is exactly requirements.txt.
  • Ask for help if the file appears outside the collector folder.

✔️ Awesome, I've got everything!

Great. Double check that you saved requirements.txt inside topic-evidence-collector.

ⓧ I'd like to double check the full code

Compare your saved file with this complete version.

requests==2.34.2
Create the isolated environment

A virtual environment keeps this project’s packages separate from other Python projects on your computer. Activating it makes the installation apply to .venv.

  • Use the VS Code terminal menu to open a new integrated terminal.
  • Use the terminal profile selector to choose PowerShell if another shell opens.
  • Prepare the complete isolated environment by running these commands:
py -m venv .venv
.venv\Scripts\activate
py -m pip install -r requirements.txt

What Do These Commands Do?

  • The first line creates the isolated environment in .venv.
  • The second line activates that environment in the current PowerShell terminal.
  • The third line installs the dependency pinned in requirements.txt.
  • Confirm the VS Code terminal prompt includes the active environment name.
  • Confirm the Explorer sidebar lists .venv inside topic-evidence-collector.

That is a key setup milestone complete. Requests is now isolated inside the collector’s own environment.

Environment Setup Failed?

  • Confirm the terminal is inside topic-evidence-collector before rerunning the commands.
  • Check that requirements.txt contains requests==2.34.2.
  • Ask for help with the exact PowerShell message you received.
Activate private OpenAlex access

OpenAlex uses an account key to authenticate content requests. The collector asks for that key through a hidden terminal prompt in the next step.

Handling a credential can feel risky. This key stays outside your project files so it cannot be copied into source code by mistake.

  • Visit openalex.org in your browser.
  • Create a free OpenAlex account.
  • Verify the account through the email OpenAlex sends you.
  • Return to OpenAlex after verification.
  • Open Settings in your account.
  • Select API key.
  • Activate the key on that page.
  • Copy the active key to a private credential manager outside topic-evidence-collector.

Your key is ready for hidden terminal input. No project file contains the credential.

  • Switch back to the VS Code PowerShell terminal from earlier.

Before you run the final check, which two version signals do you expect from the active environment?

  • Confirm the environment is ready by running these commands:
py --version
py -m pip freeze

What Confirms the Setup?

  • The first line should report Python 3.10 or newer.
  • The package list should include requests==2.34.2.

That is the project foundation locked in. Your isolated environment can now support authenticated OpenAlex requests without exposing the key in source code.

Your Windows workspace is ready. Next up, you’ll search for one paper and reconstruct its abstract from OpenAlex metadata.

Retrieve a Paper and Rebuild Its Abstract

Your Windows environment can now run Requests 2.34.2. Your private OpenAlex API key is ready for authenticated requests.

A metadata-only search gives you an immediate paper to inspect. The record stores its abstract as word positions.

This step rebuilds those positions into readable text. The output also exposes the missing named Results and Conclusion sections.

In this step, get ready to:
  • Create the first working version of collector.py.
  • Retrieve one matching article from the OpenAlex Works API.
  • Reconstruct the article abstract from abstract_inverted_index.
Create the collector scaffold

A REST API accepts structured requests at a web address. Your scaffold stores that address with the metadata fields each search should return.

  • Select the new-file icon beside the topic-evidence-collector folder in the VS Code file sidebar.
  • Name the new file collector.py.
  • Add the imports with the OpenAlex configuration by pasting this code:
import getpass
import time

import requests

API_URL = "https://api.openalex.org/works"
SEARCH_FIELDS = (
    "id,doi,display_name,publication_year,authorships,"
    "abstract_inverted_index,content_urls,best_oa_location"
)


def main():
    print("Topic-to-Evidence Collector")


if __name__ == "__main__":
    main()

What does this code do?

  • The getpass import provides hidden terminal input for your API key.
  • The requests import gives the collector an HTTP client for the OpenAlex request.
  • API_URL points to the OpenAlex Works endpoint.
  • SEARCH_FIELDS limits each response to the metadata used by this collector.
  • main() provides the script entry point.
  • Save collector.py.
  • Confirm the scaffold runs by entering this command in the activated PowerShell terminal:
python collector.py

What does this check prove?

The command loads the imports and reaches main(). You should see Topic-to-Evidence Collector before the terminal returns to its prompt.

Good start. Your collector now has a runnable entry point for the search workflow.

Does the script stop before the banner?

  • Confirm the terminal is still inside the topic-evidence-collector folder.
  • Check that the file is named collector.py.
  • Confirm the activated environment still contains Requests 2.34.2.
  • Help me diagnose why collector.py stops before printing its banner.
Search for one matching article

Network requests can fail temporarily. A retry helper gives brief server problems another chance before the collector reports the failure.

  • Find def main(): in collector.py.
  • Paste this retry helper above that line:
def request_with_retry(url, params, timeout):
    last_error = None
    for attempt in range(4):
        try:
            response = requests.get(url, params=params, timeout=timeout)
            if response.status_code == 429 or 500 <= response.status_code < 600:
                if attempt < 3:
                    time.sleep(2 ** attempt)
                    continue
            response.raise_for_status()
            return response
        except requests.RequestException as error:
            last_error = error
            if attempt == 3:
                break
            time.sleep(2 ** attempt)
    if last_error is not None:
        raise last_error
    raise RuntimeError("Request failed without a response.")

How does retrying work?

  • request_with_retry() sends the URL with its query parameters and timeout.
  • The loop permits up to four request attempts.
  • Temporary server failures trigger progressively longer pauses through time.sleep().
  • raise_for_status() turns an unsuccessful HTTP response into an exception.
  • The final exception preserves the failure for the calling code.
  • Save collector.py.
  • Check the updated file for syntax problems by running this command:
python collector.py

What does this run confirm?

You should see the collector banner again. A clean return to the terminal prompt confirms that Python can load the new helper.

Seeing a syntax problem in the helper?

  • Check that every line inside request_with_retry() uses consistent four-space indentation.
  • Confirm each try block has the matching except block shown above.
  • Help me fix the request_with_retry function in collector.py.

The next function builds the search parameters. It searches titles with abstracts plus named keywords while limiting the response to articles that have an abstract and GROBID XML.

  • Find def main(): again.
  • Paste this search function directly above it:
def search_page(topic, api_key, per_page, cursor=None):
    params = {
        "search.title_abstract_keywords": topic,
        "filter": "type:article,has_abstract:true,has_content.grobid_xml:true",
        "per_page": per_page,
        "select": SEARCH_FIELDS,
        "api_key": api_key,
    }
    if cursor is not None:
        params["cursor"] = cursor
    response = request_with_retry(API_URL, params=params, timeout=30)
    return response.json()

What does the search function do?

  • search.title_abstract_keywords searches the topic across title text with abstracts plus named keywords.
  • type:article,has_abstract:true,has_content.grobid_xml:true keeps articles that have the two content formats this project needs.
  • per_page controls how many works the request returns.
  • select asks OpenAlex for the fields listed in SEARCH_FIELDS.
  • api_key authenticates the request without placing the key in the URL constant.
  • Save collector.py.
  • Confirm Python can load the search function by running this command:
python collector.py

What should this run show?

You should see the collector banner with no traceback. The search function is now available to main().

Entering a credential can feel risky. The hidden prompt keeps your OpenAlex key off the screen and out of collector.py.

  • Find the block starting at def main(): near the bottom of collector.py.
  • Replace that block through the final main() call with this code:
def main():
    print("Topic-to-Evidence Collector")
    api_key = getpass.getpass("OpenAlex API key: ").strip()
    topic = input("Research topic: ").strip()

    if not api_key or not topic:
        raise SystemExit("An API key and topic are required.")

    payload = search_page(topic, api_key, 1)
    works = payload.get("results") or []

    for work in works:
        print(work.get("display_name") or "")
        break


if __name__ == "__main__":
    main()

How does the first search run?

  • getpass.getpass() collects the API key without echoing it in the terminal.
  • input() collects the scholarly topic.
  • search_page() requests one matching article.
  • payload.get("results") reads the returned works list.
  • display_name supplies the article title for the first visible result.
  • Save collector.py.

Before you run it, predict this outcome: the request prints a paper title or stops before receiving a result.

  • Send the first metadata request by running this command:
python collector.py

What should you enter?

  • Paste your private API key at the hidden prompt.
  • Enter your selected scholarly topic at the topic prompt.
  • Look for one matching article title beneath the prompts.

That is your first live search result. Your collector can now turn a topic into a relevant article title.

Did the search return no title?

  • Confirm your OpenAlex account email is verified.
  • Paste the activated API key again at the hidden prompt.
  • Try a broader topic if the filters return no matching article.
  • Help me diagnose why my OpenAlex search returns no paper title.
Rebuild the abstract from word positions

An inverted index maps each word to the positions where it appears. Sorting those positions restores the abstract's reading order.

  • Find def main(): in collector.py.
  • Paste this reconstruction function directly above it:
def abstract_from_index(inverted_index):
    positioned_words = []
    for word, positions in (inverted_index or {}).items():
        for position in positions:
            positioned_words.append((position, word))
    positioned_words.sort(key=lambda item: item[0])
    return " ".join(word for _, word in positioned_words)

How is the abstract reconstructed?

  • positioned_words collects each word beside every stored position.
  • sort() arranges those pairs by position.
  • join() combines the sorted words into readable text.
  • inverted_index or {} lets the function return an empty string when no abstract index is available.
  • Save collector.py.
  • Confirm the new function loads by running the collector again:
python collector.py

What does this check confirm?

Enter the hidden API key with your topic again. You should still see one article title with no traceback from abstract_from_index().

  • Find the block starting at def main():.
  • Replace that block through the final main() call with this completed version:
def main():
    print("Topic-to-Evidence Collector")
    api_key = getpass.getpass("OpenAlex API key: ").strip()
    topic = input("Research topic: ").strip()

    if not api_key or not topic:
        raise SystemExit("An API key and topic are required.")

    payload = search_page(topic, api_key, 1)
    works = payload.get("results") or []

    for work in works:
        abstract = abstract_from_index(work.get("abstract_inverted_index"))
        print(work.get("display_name") or "")
        print(abstract)
        break


if __name__ == "__main__":
    main()

What changed in the main flow?

  • work.get("abstract_inverted_index") reads the stored word positions from the first result.
  • abstract_from_index() rebuilds those positions into a string.
  • print(abstract) displays the reconstructed text beneath the article title.
  • Save collector.py.

Before the final run, predict whether a metadata record can separate named Results and Conclusion sections from its reconstructed abstract.

  • Verify the completed metadata collector by running this command:
python collector.py

What should you see?

  • Enter your private API key at the hidden prompt.
  • Enter the topic you selected for this project.
  • Look for one article title followed by a readable abstract.

The output ends after the reconstructed abstract. It contains no named Results or Conclusion fields.

That shortfall is intentional. You have proved that metadata alone cannot supply the section-aware evidence your collector needs.

Is the abstract empty or scrambled?

  • Confirm the search filter still includes has_abstract:true.
  • Check that positioned_words.sort() runs before the final join().
  • Verify that abstract_from_index() receives abstract_inverted_index.
  • Help me debug an empty or scrambled reconstructed abstract.

✔️ Awesome, I've got everything!

Your saved collector.py now retrieves one title and rebuilds its abstract.

ⓧ I'd like to double check the full code

Compare your saved collector.py with this complete version for the current step.

import getpass
import time

import requests

API_URL = "https://api.openalex.org/works"
SEARCH_FIELDS = (
    "id,doi,display_name,publication_year,authorships,"
    "abstract_inverted_index,content_urls,best_oa_location"
)


def request_with_retry(url, params, timeout):
    last_error = None
    for attempt in range(4):
        try:
            response = requests.get(url, params=params, timeout=timeout)
            if response.status_code == 429 or 500 <= response.status_code < 600:
                if attempt < 3:
                    time.sleep(2 ** attempt)
                    continue
            response.raise_for_status()
            return response
        except requests.RequestException as error:
            last_error = error
            if attempt == 3:
                break
            time.sleep(2 ** attempt)
    if last_error is not None:
        raise last_error
    raise RuntimeError("Request failed without a response.")


def search_page(topic, api_key, per_page, cursor=None):
    params = {
        "search.title_abstract_keywords": topic,
        "filter": "type:article,has_abstract:true,has_content.grobid_xml:true",
        "per_page": per_page,
        "select": SEARCH_FIELDS,
        "api_key": api_key,
    }
    if cursor is not None:
        params["cursor"] = cursor
    response = request_with_retry(API_URL, params=params, timeout=30)
    return response.json()


def abstract_from_index(inverted_index):
    positioned_words = []
    for word, positions in (inverted_index or {}).items():
        for position in positions:
            positioned_words.append((position, word))
    positioned_words.sort(key=lambda item: item[0])
    return " ".join(word for _, word in positioned_words)


def main():
    print("Topic-to-Evidence Collector")
    api_key = getpass.getpass("OpenAlex API key: ").strip()
    topic = input("Research topic: ").strip()

    if not api_key or not topic:
        raise SystemExit("An API key and topic are required.")

    payload = search_page(topic, api_key, 1)
    works = payload.get("results") or []

    for work in works:
        abstract = abstract_from_index(work.get("abstract_inverted_index"))
        print(work.get("display_name") or "")
        print(abstract)
        break


if __name__ == "__main__":
    main()

How should the file be ordered?

The imports and constants come first. The request helper, search function, abstract reconstruction function, and main() follow in that order.

Your collector now turns a research topic into a paper title with readable abstract text. Next, you will recover named sections from the article's structured content.

Extract Target Sections from GROBID XML

Your OpenAlex collector can already find a relevant article and reconstruct its abstract. The metadata still leaves the paper's named Results or Findings section out of reach.

In this step, you will download GROBID TEI XML for one paper. Your collector will extract both target sections when available. It will use the abstract when either section is missing.

In this step, get ready to:
  • Download one paper's GROBID XML through the existing retry helper.
  • Parse Results or Findings plus Conclusion sections across supported XML shapes.
  • Report whether the paper uses complete target sections or the abstract fallback.
Prepare the XML download

Each matching work contains a URL for cached XML. Requests can retrieve that file through the timeout logic already in your collector.

Why Use GROBID XML?

GROBID XML preserves document structure through elements for headings and paragraphs. This structure gives your collector named sections to inspect.

OpenAlex provides the prepared XML file through its content API. This keeps your work focused on structured extraction and transparent fallback rules.

  • Switch back to collector.py in VS Code.
  • Replace the import and constant section at the top of collector.py with this expanded section:
import getpass
import re
import time
import xml.etree.ElementTree as ET

import requests

API_URL = "https://api.openalex.org/works"
SEARCH_FIELDS = (
    "id,doi,display_name,publication_year,authorships,"
    "abstract_inverted_index,content_urls,best_oa_location"
)
RESULT_HEADINGS = {
    "results",
    "findings",
    "results and discussion",
    "results and findings",
}
CONCLUSION_HEADINGS = {
    "conclusion",
    "conclusions",
    "discussion and conclusion",
    "discussion and conclusions",
    "summary and conclusion",
    "summary and conclusions",
}

What Does This Section Add?

  • The re module removes numbering from section headings before classification.
  • ElementTree converts the downloaded XML into elements that Python can inspect.
  • RESULT_HEADINGS contains the accepted Results and Findings variants.
  • CONCLUSION_HEADINGS contains the accepted Conclusion variants.
  • Save collector.py.
  • Confirm VS Code shows both new imports without a syntax warning.

Seeing an Import Warning?

Check that xml.etree.ElementTree uses dots between each name. Confirm that its alias is uppercase ET.

Make sure both heading sets use matching braces. Help me check the new imports and heading sets.

Parse headings across XML shapes

OpenAlex documents multiple archive-era XML shapes. Matching tags by their local names lets one parser handle elements with or without namespace prefixes.

  • Find the end of abstract_from_index() in collector.py.
  • Add the tag and heading helpers below abstract_from_index() by pasting this code:
def local_name(tag):
    return tag.rsplit("}", 1)[-1].casefold()


def element_text(element):
    return " ".join("".join(element.itertext()).split())


def normalize_heading(heading):
    normalized = heading.casefold().strip()
    normalized = re.sub(
        r"^\s*(?:[ivxlcdm]+|\d+(?:\.\d+)*)[\s.)\-:]+",
        "",
        normalized,
    )
    normalized = re.sub(r"\s+", " ", normalized)
    return normalized.strip(" .:-")


def heading_kind(heading):
    normalized = normalize_heading(heading)
    if normalized in RESULT_HEADINGS or normalized.startswith("results "):
        return "results"
    if normalized in CONCLUSION_HEADINGS or normalized.startswith("conclusion"):
        return "conclusion"
    return None

How Are Headings Classified?

  • local_name() removes any namespace prefix before comparing a tag name.
  • element_text() joins nested text into a clean string with consistent spacing.
  • normalize_heading() converts headings to lowercase text. It also removes leading section numbers.
  • heading_kind() returns a target category only when the normalized heading matches an accepted variant.
  • Add the XML scope helpers directly below heading_kind() by pasting this code:
def find_scope(root):
    body = next(
        (element for element in root.iter() if local_name(element.tag) == "body"),
        None,
    )
    if body is not None:
        return body
    text = next(
        (element for element in root.iter() if local_name(element.tag) == "text"),
        None,
    )
    return text if text is not None else root


def extract_div_text(div):
    parts = []
    for child in list(div):
        child_name = local_name(child.tag)
        if child_name in {"head", "div"}:
            continue
        for element in child.iter():
            if local_name(element.tag) in {"p", "item"}:
                text = element_text(element)
                if text:
                    parts.append(text)
    return "\n\n".join(parts)

How Is the XML Scope Chosen?

  • find_scope() prefers the document's body element when one exists.
  • The helper falls back to the broader text element before using the XML root.
  • extract_div_text() collects text from p and item descendants. It skips nested section containers.
  • Add the initial parse_target_sections() structure below extract_div_text() by pasting this code:
def parse_target_sections(xml_bytes):
    root = ET.fromstring(xml_bytes)
    scope = find_scope(root)
    sections = {"results": [], "conclusion": []}

    return (
        "\n\n".join(sections["results"]),
        "\n\n".join(sections["conclusion"]),
    )

What Does the Parser Structure Hold?

  • ET.fromstring() converts the downloaded bytes into an XML root element.
  • sections keeps Results and Conclusion paragraphs separate while the parser scans the document.
  • The return value joins each category into one readable block of text.
  • Insert the section-based scan between the sections dictionary and the return statement:
    for element in scope.iter():
        if local_name(element.tag) != "div":
            continue
        heading = next(
            (
                child
                for child in list(element)
                if local_name(child.tag) == "head"
            ),
            None,
        )
        if heading is None:
            continue
        kind = heading_kind(element_text(heading))
        if kind is None:
            continue
        text = extract_div_text(element)
        if text:
            sections[kind].append(text)

How Does the Section Scan Work?

  • The loop inspects each div as a possible document section.
  • A direct head child supplies the section heading for classification.
  • Recognized sections pass to extract_div_text() so their paragraphs can be collected.
  • Insert the flat-heading fallback immediately after the div scan. Keep it before the return statement:
    if not sections["results"] and not sections["conclusion"]:
        current_kind = None
        for element in scope.iter():
            name = local_name(element.tag)
            if name == "head":
                current_kind = heading_kind(element_text(element))
            elif name in {"p", "item"} and current_kind is not None:
                text = element_text(element)
                if text:
                    sections[current_kind].append(text)

Why Is There a Second Scan?

Some archived documents place headings and paragraphs in a flatter sequence. This fallback remembers the latest recognized heading while collecting its following paragraphs.

The second scan runs only when the structured div scan found neither target section. This keeps the extraction path predictable.

  • Save collector.py.
  • Confirm parse_target_sections() ends with one tuple containing the joined Results text plus the joined Conclusion text.

Seeing Indentation Problems?

Keep both scans inside parse_target_sections() with four leading spaces. Keep the final return at the same indentation level as the scans.

Check that the fallback begins after the first loop finishes. Help me fix the indentation in parse_target_sections().

Apply the fallback and test one paper

A paper qualifies for combined section text only when Results or Findings contains text and Conclusion contains text. Any missing target section sets the source basis to abstract_fallback.

  • Add the author helper below parse_target_sections() by pasting this code:
def authors_from_work(work):
    names = []
    for authorship in work.get("authorships") or []:
        name = (authorship.get("author") or {}).get("display_name")
        if name:
            names.append(name)
    return "; ".join(names)

What Does the Author Helper Do?

authors_from_work() collects each available display name from the work record. It joins the names into one export-ready string.

  • Add the first complete version of build_record() directly below authors_from_work() by pasting this code:
def build_record(work, topic, results, conclusion, extraction_status):
    abstract = abstract_from_index(work.get("abstract_inverted_index"))
    complete_sections = bool(results and conclusion)
    if complete_sections:
        combined_text = (
            f"ABSTRACT\n{abstract}\n\n"
            f"RESULTS OR FINDINGS\n{results}\n\n"
            f"CONCLUSION\n{conclusion}"
        )
        content_basis = "results_and_conclusion"
    else:
        combined_text = f"ABSTRACT\n{abstract}"
        content_basis = "abstract_fallback"

    return {
        "topic": topic,
        "title": work.get("display_name") or "",
        "content_basis": content_basis,
        "extraction_status": extraction_status,
        "abstract": abstract,
        "results": results,
        "conclusion": conclusion,
        "combined_text": combined_text,
    }

How Does the Fallback Contract Work?

  • complete_sections becomes true only when Results and Conclusion both contain text.
  • results_and_conclusion combines the abstract with both extracted target sections.
  • abstract_fallback keeps only the reconstructed abstract when either target section is absent.
  • The record stores the source basis beside the text. Downstream users can identify every fallback.
  • Insert the extraction status and source metadata immediately before the return statement in build_record():
    if extraction_status == "parsed":
        extraction_status = (
            "sections_complete" if complete_sections else "missing_target_section"
        )

    best_location = work.get("best_oa_location") or {}
    openalex_id = work.get("id") or ""
    doi = work.get("doi") or ""

What Metadata Is Prepared?

  • A parsed paper receives a status that distinguishes complete sections from a missing target section.
  • best_location supplies any available license value.
  • openalex_id preserves the OpenAlex source identifier.
  • doi preserves the article's DOI when one is available.
  • Insert these metadata fields after the title entry in the dictionary returned by build_record():
        "authors": authors_from_work(work),
        "publication_year": work.get("publication_year") or "",
        "doi": doi,
        "openalex_id": openalex_id,
        "source_url": doi or openalex_id,
        "license": best_location.get("license") or "",

Why Keep the Source Fields?

These fields connect the extracted text to its article and available license metadata. The record stays traceable when it later moves into exported files.

  • Add the content download function directly below build_record() by pasting this code:
def process_work(work, topic, api_key):
    xml_url = (work.get("content_urls") or {}).get("grobid_xml")
    if not xml_url:
        return build_record(work, topic, "", "", "no_grobid_xml_url")
    response = request_with_retry(
        xml_url,
        params={"api_key": api_key},
        timeout=60,
    )
    results, conclusion = parse_target_sections(response.content)
    return build_record(work, topic, results, conclusion, "parsed")

What Does the Download Function Do?

  • process_work() reads the paper's content_urls.grobid_xml URL.
  • The existing retry helper sends the private key through the api_key parameter. It uses a 60-second timeout.
  • response.content supplies the XML bytes to the parser.
  • A missing content URL produces an abstract-based record without attempting a download.
  • Find the existing line that assigns works from the search response inside main().
  • Replace the old title and abstract printing logic below that line with this processing loop and entry point:
    records = []
    for position, work in enumerate(works, start=1):
        record = process_work(work, topic, api_key)
        records.append(record)
        print(
            f"[{position}/{len(works)}] {record['content_basis']}: "
            f"{record['title']}"
        )


if __name__ == "__main__":
    main()

What Changes in the Terminal?

The loop sends the single search result through the XML extraction path. The printed line includes the paper's content_basis followed by its title.

  • Save collector.py.

The next run downloads one cached content file. That content request costs $0.01 of API usage. The one-paper limit keeps this test contained.

Before you run the collector, which source basis do you expect the paper to use?

  • Start the updated extraction path by running this command in the activated VS Code terminal:
python collector.py

What Does This Run Test?

The command searches for one matching article. It downloads the article's available GROBID XML before printing the chosen source basis.

  • Enter your private OpenAlex API key at the hidden prompt.
  • Enter the research topic you used in the previous step.

You will see one article line containing either results_and_conclusion or abstract_fallback before the title. That confirms the XML parser and fallback decision both executed.

That is the section-extraction path working. Your collector can now distinguish complete target sections from an explicit abstract fallback.

The Extraction Run Did Not Finish?

  • Check that the activated environment still appears at the start of the terminal prompt.
  • Confirm your internet connection can reach OpenAlex before retrying the request.
  • Check that process_work() appears below build_record().
  • Confirm process_work() appears above main().
  • Help me debug my one-paper GROBID extraction run.

✔️ Awesome, I've got everything!

Great. Double check that you saved collector.py after seeing the extraction basis and article title.

ⓧ I'd like to double check the full code

Compare your one-paper collector.py with this complete version:

import getpass
import re
import time
import xml.etree.ElementTree as ET

import requests

API_URL = "https://api.openalex.org/works"
SEARCH_FIELDS = (
    "id,doi,display_name,publication_year,authorships,"
    "abstract_inverted_index,content_urls,best_oa_location"
)
RESULT_HEADINGS = {
    "results",
    "findings",
    "results and discussion",
    "results and findings",
}
CONCLUSION_HEADINGS = {
    "conclusion",
    "conclusions",
    "discussion and conclusion",
    "discussion and conclusions",
    "summary and conclusion",
    "summary and conclusions",
}


def request_with_retry(url, params, timeout):
    last_error = None
    for attempt in range(4):
        try:
            response = requests.get(url, params=params, timeout=timeout)
            if response.status_code == 429 or 500 <= response.status_code < 600:
                if attempt < 3:
                    time.sleep(2 ** attempt)
                    continue
            response.raise_for_status()
            return response
        except requests.RequestException as error:
            last_error = error
            if attempt == 3:
                break
            time.sleep(2 ** attempt)
    if last_error is not None:
        raise last_error
    raise RuntimeError("Request failed without a response.")


def search_page(topic, api_key, per_page, cursor=None):
    params = {
        "search.title_abstract_keywords": topic,
        "filter": "type:article,has_abstract:true,has_content.grobid_xml:true",
        "per_page": per_page,
        "select": SEARCH_FIELDS,
        "api_key": api_key,
    }
    if cursor is not None:
        params["cursor"] = cursor
    response = request_with_retry(API_URL, params=params, timeout=30)
    return response.json()


def abstract_from_index(inverted_index):
    positioned_words = []
    for word, positions in (inverted_index or {}).items():
        for position in positions:
            positioned_words.append((position, word))
    positioned_words.sort(key=lambda item: item[0])
    return " ".join(word for _, word in positioned_words)


def local_name(tag):
    return tag.rsplit("}", 1)[-1].casefold()


def element_text(element):
    return " ".join("".join(element.itertext()).split())


def normalize_heading(heading):
    normalized = heading.casefold().strip()
    normalized = re.sub(
        r"^\s*(?:[ivxlcdm]+|\d+(?:\.\d+)*)[\s.)\-:]+",
        "",
        normalized,
    )
    normalized = re.sub(r"\s+", " ", normalized)
    return normalized.strip(" .:-")


def heading_kind(heading):
    normalized = normalize_heading(heading)
    if normalized in RESULT_HEADINGS or normalized.startswith("results "):
        return "results"
    if normalized in CONCLUSION_HEADINGS or normalized.startswith("conclusion"):
        return "conclusion"
    return None


def find_scope(root):
    body = next(
        (element for element in root.iter() if local_name(element.tag) == "body"),
        None,
    )
    if body is not None:
        return body
    text = next(
        (element for element in root.iter() if local_name(element.tag) == "text"),
        None,
    )
    return text if text is not None else root


def extract_div_text(div):
    parts = []
    for child in list(div):
        child_name = local_name(child.tag)
        if child_name in {"head", "div"}:
            continue
        for element in child.iter():
            if local_name(element.tag) in {"p", "item"}:
                text = element_text(element)
                if text:
                    parts.append(text)
    return "\n\n".join(parts)


def parse_target_sections(xml_bytes):
    root = ET.fromstring(xml_bytes)
    scope = find_scope(root)
    sections = {"results": [], "conclusion": []}

    for element in scope.iter():
        if local_name(element.tag) != "div":
            continue
        heading = next(
            (
                child
                for child in list(element)
                if local_name(child.tag) == "head"
            ),
            None,
        )
        if heading is None:
            continue
        kind = heading_kind(element_text(heading))
        if kind is None:
            continue
        text = extract_div_text(element)
        if text:
            sections[kind].append(text)

    if not sections["results"] and not sections["conclusion"]:
        current_kind = None
        for element in scope.iter():
            name = local_name(element.tag)
            if name == "head":
                current_kind = heading_kind(element_text(element))
            elif name in {"p", "item"} and current_kind is not None:
                text = element_text(element)
                if text:
                    sections[current_kind].append(text)

    return (
        "\n\n".join(sections["results"]),
        "\n\n".join(sections["conclusion"]),
    )


def authors_from_work(work):
    names = []
    for authorship in work.get("authorships") or []:
        name = (authorship.get("author") or {}).get("display_name")
        if name:
            names.append(name)
    return "; ".join(names)


def build_record(work, topic, results, conclusion, extraction_status):
    abstract = abstract_from_index(work.get("abstract_inverted_index"))
    complete_sections = bool(results and conclusion)
    if complete_sections:
        combined_text = (
            f"ABSTRACT\n{abstract}\n\n"
            f"RESULTS OR FINDINGS\n{results}\n\n"
            f"CONCLUSION\n{conclusion}"
        )
        content_basis = "results_and_conclusion"
    else:
        combined_text = f"ABSTRACT\n{abstract}"
        content_basis = "abstract_fallback"

    if extraction_status == "parsed":
        extraction_status = (
            "sections_complete" if complete_sections else "missing_target_section"
        )

    best_location = work.get("best_oa_location") or {}
    openalex_id = work.get("id") or ""
    doi = work.get("doi") or ""

    return {
        "topic": topic,
        "title": work.get("display_name") or "",
        "authors": authors_from_work(work),
        "publication_year": work.get("publication_year") or "",
        "doi": doi,
        "openalex_id": openalex_id,
        "source_url": doi or openalex_id,
        "license": best_location.get("license") or "",
        "content_basis": content_basis,
        "extraction_status": extraction_status,
        "abstract": abstract,
        "results": results,
        "conclusion": conclusion,
        "combined_text": combined_text,
    }


def process_work(work, topic, api_key):
    xml_url = (work.get("content_urls") or {}).get("grobid_xml")
    if not xml_url:
        return build_record(work, topic, "", "", "no_grobid_xml_url")
    response = request_with_retry(
        xml_url,
        params={"api_key": api_key},
        timeout=60,
    )
    results, conclusion = parse_target_sections(response.content)
    return build_record(work, topic, results, conclusion, "parsed")


def main():
    print("Topic-to-Evidence Collector")
    api_key = getpass.getpass("OpenAlex API key: ").strip()
    topic = input("Research topic: ").strip()

    payload = search_page(topic, api_key, 1)
    works = payload.get("results") or []

    records = []
    for position, work in enumerate(works, start=1):
        record = process_work(work, topic, api_key)
        records.append(record)
        print(
            f"[{position}/{len(works)}] {record['content_basis']}: "
            f"{record['title']}"
        )


if __name__ == "__main__":
    main()

Your one-paper collector now makes its extraction decision visible. Next, you will apply the same rule across a bounded batch and write portable output files.

Export a Batch Collection

Your collector now searches OpenAlex and parses GROBID TEI XML for one paper. That proves the extraction path can identify complete sections or choose the abstract fallback.

A useful research collector needs to apply those rules across a bounded batch. This step adds validated batch prompts. It also creates portable exports with enough metadata to trace every paper.

In this step, get ready to:
  • Validate a target from 1 to 90 papers before making requests.
  • Handle content limits without losing completed records.
  • Export each batch as CSV, JSONL, and individual TXT files.
Set the batch limits and confirmation

Each batch uses one search request plus one content download per candidate paper. The collector calculates the planned usage before any search or download begins.

How is the estimate calculated?

An OpenAlex search costs $0.001. Each cached content file costs $0.01.

A 90-paper run therefore plans about $0.901 of usage. This stays within the $1 daily account budget when that allowance is otherwise unused.

  • Add the missing import lines in collector.py by using this final import group as your position guide:
import csv
import getpass
import json
import re
import time
import xml.etree.ElementTree as ET
from pathlib import Path

import requests

What do the new imports provide?

  • The csv module writes spreadsheet-compatible rows with a fixed header order.
  • The json module serializes each complete record for the JSONL export.
  • Path builds the output directories and writes the per-paper TXT files.
  • Save collector.py.
  • Check the import group for red syntax underlines.

Seeing an import warning?

Confirm that every import begins at the far left of the file. Keep import requests below the standard-library imports.

Use this prompt if the warning remains. Help me check the import order in collector.py without changing any function names.

  • Add OUTPUT_DIR below API_URL.
  • Add FIELDNAMES below SEARCH_FIELDS by matching this final constants section:
API_URL = "https://api.openalex.org/works"
OUTPUT_DIR = Path("output")
SEARCH_FIELDS = (
    "id,doi,display_name,publication_year,authorships,"
    "abstract_inverted_index,content_urls,best_oa_location"
)
FIELDNAMES = [
    "topic",
    "title",
    "authors",
    "publication_year",
    "doi",
    "openalex_id",
    "source_url",
    "license",
    "content_basis",
    "extraction_status",
    "abstract",
    "results",
    "conclusion",
    "combined_text",
]

What do these constants control?

  • OUTPUT_DIR keeps every generated file beneath the output directory.
  • FIELDNAMES fixes the CSV column order so metadata appears before extracted text.
  • content_basis and extraction_status preserve the decisions made during extraction.
  • Save collector.py.
  • Confirm that FIELDNAMES contains fourteen quoted field names.

Seeing a bracket error?

Check that SEARCH_FIELDS ends with a closing parenthesis. Check that FIELDNAMES ends with a closing square bracket.

Use this prompt to compare the delimiters. Help me find an unmatched bracket in the constants section of collector.py.

The search function now needs to accept the chosen batch size. Its optional cursor also prepares the same search path for later paging without changing this batch workflow.

  • Replace the existing search_page function in collector.py with this version:
def search_page(topic, api_key, per_page, cursor=None):
    params = {
        "search.title_abstract_keywords": topic,
        "filter": "type:article,has_abstract:true,has_content.grobid_xml:true",
        "per_page": per_page,
        "select": SEARCH_FIELDS,
        "api_key": api_key,
    }
    if cursor is not None:
        params["cursor"] = cursor
    response = request_with_retry(API_URL, params=params, timeout=30)
    return response.json()

How does batch search work?

  • per_page carries the validated paper count into the OpenAlex request.
  • The filter keeps only articles with abstracts and available GROBID XML.
  • The optional cursor is omitted during the main one-page batch run.
  • Replace the existing main function with this prompt and validation section:
def main():
    print("Topic-to-Evidence Collector")
    api_key = getpass.getpass("OpenAlex API key: ").strip()
    topic = input("Research topic: ").strip()
    count_text = input("Number of papers, 1 to 90: ").strip()

    if not api_key or not topic:
        raise SystemExit("An API key and topic are required.")
    if not count_text.isdigit() or not 1 <= int(count_text) <= 90:
        raise SystemExit("Enter a whole number from 1 to 90.")

    count = int(count_text)
    planned_usage = 0.001 + (count * 0.01)
    print(f"Planned API usage is about ${planned_usage:.3f}.")
    if input("Continue? Type y: ").strip().casefold() != "y":
        print("Cancelled before any search or content download.")
        return

What does the prompt validation protect?

  • getpass.getpass accepts the API key without echoing it in the terminal.
  • count_text.isdigit() rejects decimals and other non-numeric input.
  • The range check caps one batch at 90 papers.
  • The confirmation prompt exits before any search when the learner does not enter y.

Before you test the guard, do you think a count of 91 reaches the cost confirmation?

  • Test the upper limit by running this command:
python collector.py

Why is this a safe validation test?

The count check runs before the search request. A rejected value therefore uses no OpenAlex search or content budget.

  • Enter your private OpenAlex API key at the hidden prompt.
  • Enter your chosen research topic.
  • Enter 91 for the paper count.

You will see Enter a whole number from 1 to 90. The program exits before displaying a cost estimate.

Did the collector continue past 91?

Confirm that the second validation uses not 1 <= int(count_text) <= 90. Keep that check above the conversion to count.

Use this prompt if the guard still accepts the value. Help me debug the 1-to-90 count validation in collector.py.

Write the CSV, JSONL, and TXT files

CSV supports spreadsheet analysis. JSONL keeps one complete JSON record per line for streaming into another program.

The TXT export gives each paper a readable file containing its metadata and chosen combined text. Together, the three formats support inspection and downstream processing.

  • Add the per-paper TXT export loop directly below the JSONL loop inside write_outputs by pasting this code:
def write_outputs(records, output_dir=OUTPUT_DIR):
    output_dir.mkdir(parents=True, exist_ok=True)
    text_dir = output_dir / "text"
    text_dir.mkdir(parents=True, exist_ok=True)

    for old_file in text_dir.glob("*.txt"):
        old_file.unlink()

    csv_path = output_dir / "papers.csv"
    with csv_path.open("w", newline="", encoding="utf-8-sig") as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=FIELDNAMES)
        writer.writeheader()
        writer.writerows(records)

    jsonl_path = output_dir / "papers.jsonl"
    with jsonl_path.open("w", encoding="utf-8") as jsonl_file:
        for record in records:
            jsonl_file.write(json.dumps(record, ensure_ascii=False) + "\n")

How are CSV and JSONL written?

  • mkdir creates the output directories when they do not already exist.
  • glob("*.txt") finds TXT files from a previous batch so the folder reflects only the current run.
  • csv.DictWriter writes the fixed header followed by one dictionary per record.
  • json.dumps preserves each complete record on one JSONL line.
  • Save collector.py.
  • Check that the body of write_outputs remains indented by four spaces.

Seeing an indentation warning?

Keep the directory setup and both file-writing blocks inside write_outputs. Keep each nested for or with body indented one additional level.

Use this prompt to inspect the nesting. Help me correct the indentation in write_outputs without changing its statements.

  • Add the per-paper text loop directly below the JSONL loop inside write_outputs by pasting this code:
    for record in records:
        work_id = record["openalex_id"].rsplit("/", 1)[-1] or "unknown-work"
        text = (
            f"Title: {record['title']}\n"
            f"Authors: {record['authors']}\n"
            f"Year: {record['publication_year']}\n"
            f"DOI: {record['doi']}\n"
            f"OpenAlex ID: {record['openalex_id']}\n"
            f"Source: {record['source_url']}\n"
            f"License: {record['license']}\n"
            f"Content basis: {record['content_basis']}\n"
            f"Extraction status: {record['extraction_status']}\n\n"
            f"{record['combined_text']}\n"
        )
        (text_dir / f"{work_id}.txt").write_text(text, encoding="utf-8")

How is each TXT file built?

  • rsplit keeps the final part of the OpenAlex ID for the filename.
  • The metadata header preserves authorship and publication details beside the extracted content.
  • Content basis records whether the file contains both target sections or the abstract fallback.
  • write_text saves each paper with UTF-8 encoding.
  • Save collector.py.
  • Confirm that the TXT export loop aligns with the earlier CSV and JSONL blocks.

Is the TXT export loop nested inside JSONL?

The final for record in records line needs four leading spaces. It should align with csv_path and jsonl_path.

Use this prompt if the nesting is unclear. Help me align the per-paper TXT export loop inside write_outputs.

Process and verify a batch

A batch can encounter a content limit or a malformed XML file after earlier papers have succeeded. The processing loop converts recoverable failures into traceable records and preserves completed work.

  • Add the first part of the processing loop below the cancellation return inside main by pasting this code:
    payload = search_page(topic, api_key, count)
    works = payload.get("results") or []
    print(f"OpenAlex returned {len(works)} candidate papers.")

    records = []
    for position, work in enumerate(works, start=1):
        try:
            record = process_work(work, topic, api_key)
        except requests.HTTPError as error:
            status = error.response.status_code if error.response is not None else None
            if status == 429:
                print("OpenAlex reported a rate or daily-budget limit. Saving progress.")
                break
            record = build_record(
                work,
                topic,
                "",
                "",
                f"download_http_error_{status}",
            )

How does the batch handle HTTP limits?

  • search_page requests up to the validated count in one metadata search.
  • enumerate tracks the current position for visible progress messages.
  • A 429 stops further content downloads while retaining earlier records in memory.
  • Other HTTP failures become records with a status-specific extraction label.
  • Continue the same processing loop immediately below the HTTP error block by pasting this code:
        except requests.RequestException:
            record = build_record(
                work,
                topic,
                "",
                "",
                "download_request_error",
            )
        except ET.ParseError:
            record = build_record(
                work,
                topic,
                "",
                "",
                "xml_parse_error",
            )

        records.append(record)
        print(
            f"[{position}/{len(works)}] {record['content_basis']}: "
            f"{record['title']}"
        )

How are recoverable failures recorded?

  • RequestException covers network failures that remain after the retry attempts.
  • ET.ParseError labels content that cannot be parsed as XML.
  • records.append retains both successful records and fallback records for export.
  • The progress line displays the paper position and its chosen content basis.
  • Finish main by adding these lines after the processing loop:
    write_outputs(records)
    print(f"Saved {len(records)} papers under {OUTPUT_DIR.resolve()}.")

What completes the run?

write_outputs writes every accumulated record after the loop stops. The final message displays the resolved output location.

  • Save collector.py.
  • Confirm that write_outputs(records) aligns with payload = search_page(topic, api_key, count) inside main.

Seeing an except or scope error?

Keep every except aligned with its matching try. Keep records.append(record) inside the for loop and outside the exception blocks.

Use this prompt to inspect the loop structure. Help me fix the batch loop indentation in collector.py.

✔️ Awesome, I've got everything!

Your batch prompts and export paths are ready. Double check that you saved collector.py.

ⓧ I'd like to double check the full code

Compare your saved collector.py with this complete batch collector:

import csv
import getpass
import json
import re
import time
import xml.etree.ElementTree as ET
from pathlib import Path

import requests

API_URL = "https://api.openalex.org/works"
OUTPUT_DIR = Path("output")
SEARCH_FIELDS = (
    "id,doi,display_name,publication_year,authorships,"
    "abstract_inverted_index,content_urls,best_oa_location"
)
FIELDNAMES = [
    "topic",
    "title",
    "authors",
    "publication_year",
    "doi",
    "openalex_id",
    "source_url",
    "license",
    "content_basis",
    "extraction_status",
    "abstract",
    "results",
    "conclusion",
    "combined_text",
]
RESULT_HEADINGS = {
    "results",
    "findings",
    "results and discussion",
    "results and findings",
}
CONCLUSION_HEADINGS = {
    "conclusion",
    "conclusions",
    "discussion and conclusion",
    "discussion and conclusions",
    "summary and conclusion",
    "summary and conclusions",
}


def request_with_retry(url, params, timeout):
    last_error = None
    for attempt in range(4):
        try:
            response = requests.get(url, params=params, timeout=timeout)
            if response.status_code == 429 or 500 <= response.status_code < 600:
                if attempt < 3:
                    time.sleep(2 ** attempt)
                    continue
            response.raise_for_status()
            return response
        except requests.RequestException as error:
            last_error = error
            if attempt == 3:
                break
            time.sleep(2 ** attempt)
    if last_error is not None:
        raise last_error
    raise RuntimeError("Request failed without a response.")


def search_page(topic, api_key, per_page, cursor=None):
    params = {
        "search.title_abstract_keywords": topic,
        "filter": "type:article,has_abstract:true,has_content.grobid_xml:true",
        "per_page": per_page,
        "select": SEARCH_FIELDS,
        "api_key": api_key,
    }
    if cursor is not None:
        params["cursor"] = cursor
    response = request_with_retry(API_URL, params=params, timeout=30)
    return response.json()


def abstract_from_index(inverted_index):
    positioned_words = []
    for word, positions in (inverted_index or {}).items():
        for position in positions:
            positioned_words.append((position, word))
    positioned_words.sort(key=lambda item: item[0])
    return " ".join(word for _, word in positioned_words)


def local_name(tag):
    return tag.rsplit("}", 1)[-1].casefold()


def element_text(element):
    return " ".join("".join(element.itertext()).split())


def normalize_heading(heading):
    normalized = heading.casefold().strip()
    normalized = re.sub(
        r"^\s*(?:[ivxlcdm]+|\d+(?:\.\d+)*)[\s.)\-:]+",
        "",
        normalized,
    )
    normalized = re.sub(r"\s+", " ", normalized)
    return normalized.strip(" .:-")


def heading_kind(heading):
    normalized = normalize_heading(heading)
    if normalized in RESULT_HEADINGS or normalized.startswith("results "):
        return "results"
    if normalized in CONCLUSION_HEADINGS or normalized.startswith("conclusion"):
        return "conclusion"
    return None


def find_scope(root):
    body = next(
        (element for element in root.iter() if local_name(element.tag) == "body"),
        None,
    )
    if body is not None:
        return body
    text = next(
        (element for element in root.iter() if local_name(element.tag) == "text"),
        None,
    )
    return text if text is not None else root


def extract_div_text(div):
    parts = []
    for child in list(div):
        child_name = local_name(child.tag)
        if child_name in {"head", "div"}:
            continue
        for element in child.iter():
            if local_name(element.tag) in {"p", "item"}:
                text = element_text(element)
                if text:
                    parts.append(text)
    return "\n\n".join(parts)


def parse_target_sections(xml_bytes):
    root = ET.fromstring(xml_bytes)
    scope = find_scope(root)
    sections = {"results": [], "conclusion": []}

    for element in scope.iter():
        if local_name(element.tag) != "div":
            continue
        heading = next(
            (
                child
                for child in list(element)
                if local_name(child.tag) == "head"
            ),
            None,
        )
        if heading is None:
            continue
        kind = heading_kind(element_text(heading))
        if kind is None:
            continue
        text = extract_div_text(element)
        if text:
            sections[kind].append(text)

    if not sections["results"] and not sections["conclusion"]:
        current_kind = None
        for element in scope.iter():
            name = local_name(element.tag)
            if name == "head":
                current_kind = heading_kind(element_text(element))
            elif name in {"p", "item"} and current_kind is not None:
                text = element_text(element)
                if text:
                    sections[current_kind].append(text)

    return (
        "\n\n".join(sections["results"]),
        "\n\n".join(sections["conclusion"]),
    )


def authors_from_work(work):
    names = []
    for authorship in work.get("authorships") or []:
        name = (authorship.get("author") or {}).get("display_name")
        if name:
            names.append(name)
    return "; ".join(names)


def build_record(work, topic, results, conclusion, extraction_status):
    abstract = abstract_from_index(work.get("abstract_inverted_index"))
    complete_sections = bool(results and conclusion)
    if complete_sections:
        combined_text = (
            f"ABSTRACT\n{abstract}\n\n"
            f"RESULTS OR FINDINGS\n{results}\n\n"
            f"CONCLUSION\n{conclusion}"
        )
        content_basis = "results_and_conclusion"
    else:
        combined_text = f"ABSTRACT\n{abstract}"
        content_basis = "abstract_fallback"

    if extraction_status == "parsed":
        extraction_status = (
            "sections_complete" if complete_sections else "missing_target_section"
        )

    best_location = work.get("best_oa_location") or {}
    openalex_id = work.get("id") or ""
    doi = work.get("doi") or ""

    return {
        "topic": topic,
        "title": work.get("display_name") or "",
        "authors": authors_from_work(work),
        "publication_year": work.get("publication_year") or "",
        "doi": doi,
        "openalex_id": openalex_id,
        "source_url": doi or openalex_id,
        "license": best_location.get("license") or "",
        "content_basis": content_basis,
        "extraction_status": extraction_status,
        "abstract": abstract,
        "results": results,
        "conclusion": conclusion,
        "combined_text": combined_text,
    }


def process_work(work, topic, api_key):
    xml_url = (work.get("content_urls") or {}).get("grobid_xml")
    if not xml_url:
        return build_record(work, topic, "", "", "no_grobid_xml_url")
    response = request_with_retry(
        xml_url,
        params={"api_key": api_key},
        timeout=60,
    )
    results, conclusion = parse_target_sections(response.content)
    return build_record(work, topic, results, conclusion, "parsed")


def write_outputs(records, output_dir=OUTPUT_DIR):
    output_dir.mkdir(parents=True, exist_ok=True)
    text_dir = output_dir / "text"
    text_dir.mkdir(parents=True, exist_ok=True)

    for old_file in text_dir.glob("*.txt"):
        old_file.unlink()

    csv_path = output_dir / "papers.csv"
    with csv_path.open("w", newline="", encoding="utf-8-sig") as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=FIELDNAMES)
        writer.writeheader()
        writer.writerows(records)

    jsonl_path = output_dir / "papers.jsonl"
    with jsonl_path.open("w", encoding="utf-8") as jsonl_file:
        for record in records:
            jsonl_file.write(json.dumps(record, ensure_ascii=False) + "\n")

    for record in records:
        work_id = record["openalex_id"].rsplit("/", 1)[-1] or "unknown-work"
        text = (
            f"Title: {record['title']}\n"
            f"Authors: {record['authors']}\n"
            f"Year: {record['publication_year']}\n"
            f"DOI: {record['doi']}\n"
            f"OpenAlex ID: {record['openalex_id']}\n"
            f"Source: {record['source_url']}\n"
            f"License: {record['license']}\n"
            f"Content basis: {record['content_basis']}\n"
            f"Extraction status: {record['extraction_status']}\n\n"
            f"{record['combined_text']}\n"
        )
        (text_dir / f"{work_id}.txt").write_text(text, encoding="utf-8")


def main():
    print("Topic-to-Evidence Collector")
    api_key = getpass.getpass("OpenAlex API key: ").strip()
    topic = input("Research topic: ").strip()
    count_text = input("Number of papers, 1 to 90: ").strip()

    if not api_key or not topic:
        raise SystemExit("An API key and topic are required.")
    if not count_text.isdigit() or not 1 <= int(count_text) <= 90:
        raise SystemExit("Enter a whole number from 1 to 90.")

    count = int(count_text)
    planned_usage = 0.001 + (count * 0.01)
    print(f"Planned API usage is about ${planned_usage:.3f}.")
    if input("Continue? Type y: ").strip().casefold() != "y":
        print("Cancelled before any search or content download.")
        return

    payload = search_page(topic, api_key, count)
    works = payload.get("results") or []
    print(f"OpenAlex returned {len(works)} candidate papers.")

    records = []
    for position, work in enumerate(works, start=1):
        try:
            record = process_work(work, topic, api_key)
        except requests.HTTPError as error:
            status = error.response.status_code if error.response is not None else None
            if status == 429:
                print("OpenAlex reported a rate or daily-budget limit. Saving progress.")
                break
            record = build_record(
                work,
                topic,
                "",
                "",
                f"download_http_error_{status}",
            )
        except requests.RequestException:
            record = build_record(
                work,
                topic,
                "",
                "",
                "download_request_error",
            )
        except ET.ParseError:
            record = build_record(
                work,
                topic,
                "",
                "",
                "xml_parse_error",
            )

        records.append(record)
        print(
            f"[{position}/{len(works)}] {record['content_basis']}: "
            f"{record['title']}"
        )

    write_outputs(records)
    print(f"Saved {len(records)} papers under {OUTPUT_DIR.resolve()}.")


if __name__ == "__main__":
    main()

How should you use this reference?

Compare function names and indentation from top to bottom. Preserve the exact field names because every export depends on those dictionary keys.

Before you run the completed collector, which files do you expect one three-paper batch to create?

  • Run the completed batch collector with this command:
python collector.py

Why is the API key absent from the screen?

The hidden prompt accepts your key without echoing its characters. The key is sent as the api_key request parameter and is never written to an export.

  • Enter your private OpenAlex API key at the hidden prompt.
  • Enter any scholarly topic.
  • Enter 3 for the paper count.
  • Review the displayed planned usage.
  • Enter y to approve the run.

You will see a planned usage of about $0.031. The terminal then prints each title with either results_and_conclusion or abstract_fallback.

  • Expand the output folder in your VS Code project.
  • Open output/papers.csv to confirm it contains a header and three records.
  • Open output/papers.jsonl to confirm it contains three JSON lines.
  • Expand output/text to confirm it contains three .txt files.
  • Open one .txt file to confirm it includes metadata and the selected combined text.

You now have a batch collector that turns a topic into traceable CSV, JSONL, and TXT exports. Each record also reveals whether named sections or the abstract supplied its evidence text.

Missing an output file?

  • Confirm that write_outputs(records) sits after the processing loop inside main.
  • Check the terminal for a request or XML error recorded during the run.
  • Confirm that you entered lowercase y at the confirmation prompt.
  • Help me diagnose missing CSV, JSONL, or TXT export files.

Secret mission

Resume a Multi-Day Collection

Large evidence collections can exceed one OpenAlex daily API budget. Build a resumable collector that checkpoints every completed paper, skips saved OpenAlex IDs, and continues the same topic across multiple days.

Clean Up Your Resources

Clean Up Your Resources

OpenAlex meters usage only when a collector sends requests. Choose whether to keep, pause, or delete your local resources.

Resources you used:

  • The .venv virtual environment with Requests 2.34.2.
  • The standard batch export directory at output.
  • The topic-specific resumable export directory at output/resume-<topic-slug>.
  • The private OpenAlex API key attached to your free account.

Respect Downloaded Text Rights

Downloaded article text retains its original copyright. OpenAlex grants no additional content rights.

Keep the source metadata with each export. Keep the recorded license metadata with it.

Keep everything running

No action is needed. This option keeps the collector ready for more batch or resumable runs.

  • Retain .venv so the pinned dependency stays available.
  • Retain output so your standard exports remain available.
  • Retain output/resume-<topic-slug> so the resumable collector can continue from its checkpoint.
  • Continue entering your OpenAlex API key through the hidden terminal prompt.

Pause - I'll come back to this later

Pause the project by closing its active tools. Your code, environment, checkpoints, and exports stay available.

  • Return to the terminal panel for topic-evidence-collector from earlier.
  • Deactivate the virtual environment by running:
deactivate

What Does This Command Do?

This command removes the virtual environment from the current terminal session. The .venv directory stays on your computer for your next session.

  • Close VS Code.
  • Keep the topic-evidence-collector folder for your next collection session.

Delete - I don't want to use this again

Remove the generated environment and every standard or resumable export. Your collector source files remain available.

  • Return to the terminal panel for topic-evidence-collector from earlier.
  • Deactivate the virtual environment by running:
deactivate

What Does This Command Do?

This command releases the active virtual environment from the terminal. The environment remains on disk until the next cleanup command removes it.

  • Delete the generated local resources with PowerShell by running these commands:
Remove-Item -Recurse -Force .venv
Remove-Item -Recurse -Force output

What Do These Commands Remove?

The first command removes .venv with its installed dependency.

The second command removes the complete output directory. Your source files remain.

  • Return to the VS Code file list from earlier.

You should no longer see .venv or output inside topic-evidence-collector.

Revoking the key only blocks future API requests made with that key. Your OpenAlex account remains available.

  • Return to your OpenAlex account.
  • Select Settings.
  • Select API key.
  • Use the API key controls to revoke the key.

The key can no longer authorize future collector requests.

Nice Work!

Nice Work!

You did it! Your OpenAlex collector now turns a research topic into a traceable local evidence dataset with section-aware text.

You've learned how to:

  • Build a topic-driven search that collects up to 90 relevant scholarly articles in one run.
  • Reconstruct readable abstracts from abstract_inverted_index. Parse GROBID TEI XML for named Results or Findings plus Conclusion sections. Apply an explicit abstract fallback whenever either target section is missing.
  • Export evidence as CSV, JSONL, plus per-paper text files. Preserve traceable metadata beside every extracted record.
  • Complete the optional Secret Mission by adding resumable cursor paging for multi-day collections of up to 500 papers. Continue from topic-specific checkpoints without downloading completed papers again.

Ready to quiz yourself?