API Documentation

Core modules

pyreslib.koha

explicit_abbreviations_from_marc(record, abbreviations_dir=os.path.join('data', 'mappings', 'abbreviations'), abbreviation_codes={'relationships': {'fields': ['100$4', '700$4']}, 'languages': {'fields': ['041$a']}, 'item_types': {'fields': ['942$c']}, 'musical_instruments': {'fields': ['059$a']}})

The method returns a MARC-in-JSON record where MARC abbreviation codes are made explicit.

Parameters:

Name	Type	Description	Default
record	dict	MARC-in-JSON record from Koha API.	required
abbreviations_dir	str	Directory path (in data/mappings) containing a series of JSON mappings for standard MARC abbreviation codes.	join('data', 'mappings', 'abbreviations')
abbreviation_codes	dict	Dictionary of abbreviation codes tied to each type.	{'relationships': {'fields': ['100$4', '700$4']}, 'languages': {'fields': ['041$a']}, 'item_types': {'fields': ['942$c']}, 'musical_instruments': {'fields': ['059$a']}}

Returns: explicit_record (dict): MARC-in-JSON record with explicit abbreviations.

Source code in pyreslib/koha.py

def explicit_abbreviations_from_marc(
    record: dict,
    abbreviations_dir=os.path.join("data", "mappings", "abbreviations"),
    abbreviation_codes={
        "relationships": {"fields": ["100$4", "700$4"]},
        "languages": {"fields": ["041$a"]},
        "item_types": {"fields": ["942$c"]},
        "musical_instruments": {"fields": ["059$a"]},
    },
) -> dict:
    """
    The method returns a MARC-in-JSON record where MARC abbreviation codes are made explicit.

    Args:
        record (dict): MARC-in-JSON record from Koha API.
        abbreviations_dir (str): Directory path (in data/mappings) containing a series of JSON mappings for standard MARC abbreviation codes.
        abbreviation_codes (dict): Dictionary of abbreviation codes tied to each type.
    Returns:
        explicit_record (dict): MARC-in-JSON record with explicit abbreviations.


    """
    # load abbreviation jsons
    for f in os.scandir(abbreviations_dir):
        if f.path.endswith(".json"):
            with open(f.path) as json_file:
                abbreviation = json.load(json_file)
                abbreviation_name = f.name.split(".")[0]
                abbreviation_codes[abbreviation_name]["values"] = abbreviation

    # explicit abbreviations according to abbreviation_codes

    for key in abbreviation_codes.keys():
        for tag in abbreviation_codes[key]["fields"]:
            field = tag.split("$")[0]
            subfield = tag.split("$")[1]
            # retrieve field and subfield in record
            query_field = list(filter(lambda x: field in x.keys(), record["fields"]))
            if len(query_field) > 0:
                for statement in query_field:
                    query_subfield = list(
                        filter(
                            lambda x: subfield in x.keys(),
                            statement[field]["subfields"],
                        )
                    )
                    # print(query_subfield)
                    if len(query_subfield) > 0:
                        for sub_statement in query_subfield:
                            # retrive abbreviation code
                            # split values based on space
                            sub_statements = sub_statement[subfield].split(" ")
                            explicit_codes = []
                            found_codes = False
                            for sub in sub_statements:
                                retrieved_code = list(
                                    filter(
                                        lambda x: x["code"] == sub,
                                        abbreviation_codes[key]["values"],
                                    )
                                )
                                if len(retrieved_code) > 0:
                                    # explicit subfield value
                                    explicit_codes.append(retrieved_code[0]["label"])
                                else:
                                    # keep original value
                                    explicit_codes.append(sub)

                            # update substatement with explicit codes
                            sub_statement[subfield] = "|".join(explicit_codes)

    # print(f"Explicit MARC-in-JSON record: {record}")

    return record

fetch_and_process_authority(auth_id, session, base_url)

Fetch a single authority from API and extract metadata. This function runs in parallel worker processes.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
auth_id	int	Authority ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
result	dict	Dictionary containing authority ID, Wikidata ID and MARC-in-JSON record.

Source code in pyreslib/koha.py

def fetch_and_process_authority(auth_id: int, session, base_url: str) -> Dict[str, Any]:
    """
    Fetch a single authority from API and extract metadata.
    This function runs in parallel worker processes.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        auth_id (int): Authority ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        result (dict): Dictionary containing authority ID, Wikidata ID and MARC-in-JSON record.

    """
    try:
        record = get_authority_marc(session, auth_id, base_url)

        # Extract Wikidata ID
        wd_id = extract_wikidata_id(record)

        return {"auth_id": auth_id, "wd_id": wd_id, "record": record}
    except Exception as e:
        return {"auth_id": auth_id, "wd_id": [], "record": None}

fetch_and_process_biblio(biblio_id, session, base_url)

Fetch a single bibliographic record from API and extract metadata. This function runs in parallel worker processes.

Args: session (oauth2): Oauth2 session provided by pyreslib.koha.oauth2_session method. biblio_id (int): Biblio ID for the requested record. base_url (str): Koha API url from credentials.

Returns: result (dict): Dictionary containing biblio ID, and MARC-in-JSON record.

Source code in pyreslib/koha.py

def fetch_and_process_biblio(biblio_id: int, session, base_url: str) -> Dict[str, Any]:
    """
    Fetch a single bibliographic record from API and extract metadata.
    This function runs in parallel worker processes.

    Args:
    session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
    biblio_id (int): Biblio ID for the requested record.
    base_url (str): Koha API url from credentials.

    Returns:
    result (dict): Dictionary containing biblio ID, and MARC-in-JSON record.

    """
    try:
        record = get_biblio_marc(session, biblio_id, base_url)

        return {
            "biblio_id": biblio_id,
            "record": record,
        }
    except Exception as e:
        return {"biblio_id": biblio_id, "wd_id": [], "record": None}

get_alternative_headings_from_authority(record, headings={'PERSO_NAME': '400', 'CORPO_NAME': '410', 'CHRON_TERM': '448', 'TOPIC_TERM': '450', 'GEOGR_NAME': '451', 'KOOPM_KEYW': '450'}, heading_subfield='a')

This function returns a list of alternative headings (equivalent terms, pseudonyms,...) of a given record.

Parameters:

Name	Type	Description	Default
record	dict	MARC-in-JSON record of the authority, conform with Koha API.	required
headings	dict	Dictionary of main heading fields according to authority type. Default is {"PERSO_NAME": "100","CORPO_NAME": "110","CHRON_TERM": "148","TOPIC_TERM": "150","GEOGR_NAME": "151"}	{'PERSO_NAME': '400', 'CORPO_NAME': '410', 'CHRON_TERM': '448', 'TOPIC_TERM': '450', 'GEOGR_NAME': '451', 'KOOPM_KEYW': '450'}
heading_subfield	str	Subfield of the main heading. Default is "a".	'a'

Returns:

Name	Type	Description
alternative_headings	list	List of alternative headings.

Examples:

>>> pyreslib.koha.get_alternative_headings_from_authority(record=marc_json_authority)
>>> ["Example Person", "Example Pseudonym", "Example Equivalent Term"]

Source code in pyreslib/koha.py

def get_alternative_headings_from_authority(
    record: dict,
    headings={
        "PERSO_NAME": "400",
        "CORPO_NAME": "410",
        "CHRON_TERM": "448",
        "TOPIC_TERM": "450",
        "GEOGR_NAME": "451",
        "KOOPM_KEYW": "450",
    },
    heading_subfield="a",
) -> list:
    """
    This function returns a list of alternative headings (equivalent terms, pseudonyms,...) of a given record.

    Args:
        record (dict): MARC-in-JSON record of the authority, conform with Koha API.
        headings (dict): Dictionary of main heading fields according to authority type. Default is {"PERSO_NAME": "100","CORPO_NAME": "110","CHRON_TERM": "148","TOPIC_TERM": "150","GEOGR_NAME": "151"}
        heading_subfield (str): Subfield of the main heading. Default is "a".


    Returns:
        alternative_headings (list): List of alternative headings.

    Examples:
        >>> pyreslib.koha.get_alternative_headings_from_authority(record=marc_json_authority)
        >>> ["Example Person", "Example Pseudonym", "Example Equivalent Term"]
    """
    # get authority type
    auth_type = get_authority_type(record)
    heading_field = headings[auth_type]

    # get alternative headings
    alternative_headings = []
    try:
        query_field = list(
            filter(lambda x: heading_field in x.keys(), record["fields"])
        )
        for statement in query_field:
            alt_heading = list(
                filter(
                    lambda x: heading_subfield in x.keys(),
                    statement[heading_field]["subfields"],
                )
            )[0][heading_subfield]
            alternative_headings.append(alt_heading)

        return alternative_headings

    except Exception:
        return None

get_auth_list_from_csv_report(auth_id_csv_filepath='./data/mappings/koha/authority_list.csv', auth_id_field='authid')

Returns a list of all authorities in the catalogue.

Parameters:

Name	Type	Description	Default
auth_id_csv_filepath	str	File path of the csv file coming from report. By default this is data/mappings/authority_list.csv	'./data/mappings/koha/authority_list.csv'
auth_id_field	str	Koha report field name for authority ID. Default is authid.	'authid'

Returns:

Name	Type	Description
authority_list	list	List of authority IDs.

Source code in pyreslib/koha.py

def get_auth_list_from_csv_report(
    auth_id_csv_filepath: str = "./data/mappings/koha/authority_list.csv",
    auth_id_field="authid",
) -> list:
    """
    Returns a list of all authorities in the catalogue.

    Args:
        auth_id_csv_filepath (str): File path of the csv file coming from report. By default this is `data/mappings/authority_list.csv`
        auth_id_field (str): Koha report field name for authority ID. Default is `authid`.

    Returns:
        authority_list (list): List of authority IDs.

    """

    # import csv file and return list
    csv_list = utilities.csv2dict(auth_id_csv_filepath)

    print(csv_list[:4])
    return [int(auth[auth_id_field]) for auth in csv_list]

get_authority_json(session, auth_id, base_url)

Get JSON data for authority from Koha API. The amount of metadata is very limited compared to the MARC-in-JSON response.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
auth_id	int	Authority ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
response	dict	JSON serialization of the record.

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> json_authority = pyreslib.koha.get_authority_json(my_session,407,base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> {'authority_id': 11721, 'created_date': '2021-01-12', 'framework_id': 'PERSO_NAME', 'heading': 'Bach, Anna Magdalena 1701-1760', 'modified_date': '2026-03-30T09:54:31+02:00'}

Source code in pyreslib/koha.py

def get_authority_json(session, auth_id: int, base_url: str) -> dict:
    """Get JSON data for authority from Koha API. The amount of metadata is very limited compared to the MARC-in-JSON response.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        auth_id (int): Authority ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        response (dict): JSON serialization of the record.

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> json_authority = pyreslib.koha.get_authority_json(my_session,407,base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> {'authority_id': 11721, 'created_date': '2021-01-12', 'framework_id': 'PERSO_NAME', 'heading': 'Bach, Anna Magdalena 1701-1760', 'modified_date': '2026-03-30T09:54:31+02:00'}


    """
    headers = {"Accept": "application/json"}
    response = session.get(f"{base_url}/authorities/{str(auth_id)}", headers=headers)
    return response.json()

get_authority_marc(session, auth_id, base_url)

Get MARC in JSON data for authority from Koha API

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
auth_id	int	Authority ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
response	dict	MARC in JSON serialization of the record.

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> marc_json_authority = pyreslib.koha.get_authority_marc(my_session,407,base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> marc_json_authority[0:145] # returns the first 145 characters of the JSON dictionary
>>>{'leader': '00744nam a2200217Ia 4500', 'fields': [{'000': '00373cz  a2200157n  4500'}, {'001': '407'}, {'003': 'OI'}, {'005': '20260330105335.0'} ...

Source code in pyreslib/koha.py

def get_authority_marc(session, auth_id: int, base_url: str) -> dict:
    """Get MARC in JSON data for authority from Koha API

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        auth_id (int): Authority ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        response (dict): MARC in JSON serialization of the record.

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> marc_json_authority = pyreslib.koha.get_authority_marc(my_session,407,base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> marc_json_authority[0:145] # returns the first 145 characters of the JSON dictionary
        >>>{'leader': '00744nam a2200217Ia 4500', 'fields': [{'000': '00373cz  a2200157n  4500'}, {'001': '407'}, {'003': 'OI'}, {'005': '20260330105335.0'} ...


    """
    headers = {"Accept": "application/marc-in-json"}
    response = session.get(f"{base_url}/authorities/{str(auth_id)}", headers=headers)
    return response.json()

get_authority_type(record, auth_type_field=['942', 'a'])

This function returns the authority type code of a given record.

Parameters:

Name	Type	Description	Default
record	dict	MARC-in-JSON record of the authority, conform with Koha API.	required
auth_type_field	list	MARC field holding the authority type code. Default is 942$a.	['942', 'a']

Returns:

Name	Type	Description
auth_type	str	Authority type code or None if not found.

Examples:

>>> pyreslib.koha.get_authority_type(record=marc_json_authority)
>>> "PERSO_NAME"

Source code in pyreslib/koha.py

def get_authority_type(record: dict, auth_type_field=["942", "a"]) -> str:
    """
    This function returns the authority type code of a given record.

    Args:
        record (dict): MARC-in-JSON record of the authority, conform with Koha API.
        auth_type_field (list): MARC field holding the authority type code. Default is 942$a.


    Returns:
        auth_type (str): Authority type code or None if not found.

    Examples:
        >>> pyreslib.koha.get_authority_type(record=marc_json_authority)
        >>> "PERSO_NAME"
    """
    try:
        # get authority type
        query_field = list(
            filter(lambda x: auth_type_field[0] in x.keys(), record["fields"])
        )
        # the field should be unique
        auth_type = list(
            filter(
                lambda x: auth_type_field[1] in x.keys(),
                query_field[0][auth_type_field[0]]["subfields"],
            )
        )[0][auth_type_field[1]]

        return auth_type

    except Exception:
        return None

get_biblio_json(session, biblio_id, base_url)

Get JSON data for biblio from Koha API. The amount of metadata is very limited compared to the MARC-in-JSON response.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
biblio_id	int	Biblio ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
response	dict	JSON serialization of the record.

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> json_record = pyreslib.koha.get_biblio_json(my_session,12,base_url="https://{KOHA_STAFF_URL}/api/v1")
>>>{'abstract': None, 'age_restriction': None, 'author': 'Playford, Henry', 'biblio_id': 1637, ... }

Source code in pyreslib/koha.py

def get_biblio_json(session, biblio_id: int, base_url: str) -> dict:
    """Get JSON data for biblio from Koha API. The amount of metadata is very limited compared to the MARC-in-JSON response.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        biblio_id (int): Biblio ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        response (dict): JSON serialization of the record.

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> json_record = pyreslib.koha.get_biblio_json(my_session,12,base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>>{'abstract': None, 'age_restriction': None, 'author': 'Playford, Henry', 'biblio_id': 1637, ... }


    """
    headers = {"Accept": "application/json"}
    response = session.get(f"{base_url}/biblios/{str(biblio_id)}", headers=headers)
    return response.json()

get_biblio_list_from_csv_report(biblio_id_csv_filepath='./data/mappings/koha/biblio_list.csv', biblio_id_field='biblionumber')

Returns a list of all bibliographic records in the catalogue.

Parameters:

Name	Type	Description	Default
biblio_id_csv_filepath	str	File path of the csv file coming from report. By default this is data/mappings/biblio_list.csv	'./data/mappings/koha/biblio_list.csv'
biblio_id_field	str	Koha report field name for biblio ID. Default is biblionumber.	'biblionumber'

Returns:

Name	Type	Description
biblio_list	list	List of biblio IDs.

Source code in pyreslib/koha.py

def get_biblio_list_from_csv_report(
    biblio_id_csv_filepath: str = "./data/mappings/koha/biblio_list.csv",
    biblio_id_field="biblionumber",
) -> list:
    """
    Returns a list of all bibliographic records in the catalogue.

    Args:
        biblio_id_csv_filepath (str): File path of the csv file coming from report. By default this is `data/mappings/biblio_list.csv`
        biblio_id_field (str): Koha report field name for biblio ID. Default is `biblionumber`.

    Returns:
        biblio_list (list): List of biblio IDs.

    """

    # import csv file and return list
    return [
        int(biblio[biblio_id_field])
        for biblio in utilities.csv2dict(biblio_id_csv_filepath)
    ]

get_biblio_marc(session, biblio_id, base_url)

Get MARC in JSON data for biblio from Koha API. Note: this method does not return items metadata, use pyreslib.koha.get_items_from_biblio_json in order to extract metadata at item level.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
biblio_id	int	Biblio ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
response	dict	MARC in JSON serialization of the record.

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> marc_json_record = pyreslib.koha.get_biblio_marc(my_session,12,base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> {"leader": "01574acm a2200301   4500","fields": [{"001": "1"},{"005": "20251104114244.0"},{"008": "210318s1730       ||le| |||| 00| 0 fre d"},{"041": {"ind1": " ","ind2": " ","subfields": [{"a": "fre"}]}}, ...] }

Source code in pyreslib/koha.py

def get_biblio_marc(session, biblio_id: int, base_url: str) -> dict:
    """Get MARC in JSON data for biblio from Koha API. **Note**: this method does not return items metadata, use [pyreslib.koha.get_items_from_biblio_json][] in order to extract metadata at item level.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        biblio_id (int): Biblio ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        response (dict): MARC in JSON serialization of the record.

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> marc_json_record = pyreslib.koha.get_biblio_marc(my_session,12,base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> {"leader": "01574acm a2200301   4500","fields": [{"001": "1"},{"005": "20251104114244.0"},{"008": "210318s1730       ||le| |||| 00| 0 fre d"},{"041": {"ind1": " ","ind2": " ","subfields": [{"a": "fre"}]}}, ...] }



    """
    headers = {"Accept": "application/marc-in-json"}
    response = session.get(f"{base_url}/biblios/{str(biblio_id)}", headers=headers)
    return response.json()

get_biblio_type(record, biblio_type_field=['942', 'c'])

This function returns the biblio type code of a given record.

Parameters:

Name	Type	Description	Default
record	dict	MARC-in-JSON record of the biblio, conform with Koha API.	required
biblio_type_field	list	MARC field holding the biblio type code. Default is 942$c.	['942', 'c']

Returns:

Name	Type	Description
biblio_type	str	Biblio type code.

Examples:

>>> pyreslib.koha.get_biblio_type(record=marc_json_biblio)
>>> "BOO"

Source code in pyreslib/koha.py

def get_biblio_type(record: dict, biblio_type_field=["942", "c"]) -> str:
    """
    This function returns the biblio type code of a given record.

    Args:
        record (dict): MARC-in-JSON record of the biblio, conform with Koha API.
        biblio_type_field (list): MARC field holding the biblio type code. Default is 942$c.


    Returns:
        biblio_type (str): Biblio type code.

    Examples:
        >>> pyreslib.koha.get_biblio_type(record=marc_json_biblio)
        >>> "BOO"
    """
    try:
        # get biblio type
        query_field = list(
            filter(lambda x: biblio_type_field[0] in x.keys(), record["fields"])
        )
        # the field should be unique
        biblio_type = list(
            filter(
                lambda x: biblio_type_field[1] in x.keys(),
                query_field[0][biblio_type_field[0]]["subfields"],
            )
        )[0][biblio_type_field[1]]

        return biblio_type

    except Exception:
        return None

get_framework_id_authority(session, auth_id, base_url)

Get the authority record framework ID .

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
auth_id	int	Authority ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
framework	str	Koha Framework ID for the record.

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , user_agent="{USER_AGENT}", base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> pyres;lib.koha.get_framework_id_authority(my_session,1,base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> "1"

Source code in pyreslib/koha.py

def get_framework_id_authority(session, auth_id: int, base_url: str) -> str:
    """Get the authority record framework ID .

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        auth_id (int): Authority ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        framework (str): Koha Framework ID for the record.

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , user_agent="{USER_AGENT}", base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> pyres;lib.koha.get_framework_id_authority(my_session,1,base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> "1"

    """
    try:
        return get_authority_json(session=session, auth_id=auth_id, base_url=base_url)[
            "framework_id"
        ]
    except KeyError:
        return ""

get_framework_id_biblioitem(session, biblio_id, base_url)

Get the biblio record framework ID .

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
biblio_id	int	Biblio ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
framework	str	Koha Framework ID for the record.

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , user_agent="{USER_AGENT}", base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> pyres;lib.koha.get_framework_id_biblioitem(my_session,12,base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> "1"

Source code in pyreslib/koha.py

def get_framework_id_biblioitem(session, biblio_id: int, base_url: str) -> str:
    """Get the biblio record framework ID .

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        biblio_id (int): Biblio ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        framework (str): Koha Framework ID for the record.

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , user_agent="{USER_AGENT}", base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> pyres;lib.koha.get_framework_id_biblioitem(my_session,12,base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> "1"

    """
    try:
        return get_biblio_json(session=session, biblio_id=biblio_id, base_url=base_url)[
            "framework_id"
        ]
    except KeyError:
        return ""

get_items_from_biblio_json(session, biblio_id, base_url)

Get JSON item data for biblio from Koha API.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
biblio_id	int	Biblio ID for the requested record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Name	Type	Description
response	dict	JSON serialization of the record's items.

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> json_items = pyreslib.koha.get_items_from_biblio_json(my_session,12)
>>>

Source code in pyreslib/koha.py

def get_items_from_biblio_json(session, biblio_id: int, base_url: str) -> dict:
    """Get JSON item data for biblio from Koha API.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        biblio_id (int): Biblio ID for the requested record.
        base_url (str): Koha API url from credentials.

    Returns:
        response (dict): JSON serialization of the record's items.

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> json_items = pyreslib.koha.get_items_from_biblio_json(my_session,12)
        >>>



    """
    headers = {"Accept": "application/json"}
    response = session.get(
        f"{base_url}/biblios/{str(biblio_id)}/items", headers=headers
    )
    return response.json()

get_main_heading_from_authority(record, headings={'PERSO_NAME': '100', 'CORPO_NAME': '110', 'CHRON_TERM': '148', 'TOPIC_TERM': '150', 'GEOGR_NAME': '151', 'KOOPM_KEYW': '150'}, heading_subfield='a')

This function returns the authority main heading of a given record.

Parameters:

Name	Type	Description	Default
record	dict	MARC-in-JSON record of the authority, conform with Koha API.	required
headings	dict	Dictionary of main heading fields according to authority type. Default is {"PERSO_NAME": "100","CORPO_NAME": "110","CHRON_TERM": "148","TOPIC_TERM": "150","GEOGR_NAME": "151"}	{'PERSO_NAME': '100', 'CORPO_NAME': '110', 'CHRON_TERM': '148', 'TOPIC_TERM': '150', 'GEOGR_NAME': '151', 'KOOPM_KEYW': '150'}
heading_subfield	str	Subfield of the main heading. Default is "a".	'a'

Returns:

Name	Type	Description
main_heading	str	Main heading string.

Examples:

>>> pyreslib.koha.get_main_heading_from_authority(record=marc_json_authority)
>>> "Example Person"

Source code in pyreslib/koha.py

def get_main_heading_from_authority(
    record: dict,
    headings={
        "PERSO_NAME": "100",
        "CORPO_NAME": "110",
        "CHRON_TERM": "148",
        "TOPIC_TERM": "150",
        "GEOGR_NAME": "151",
        "KOOPM_KEYW": "150",
    },
    heading_subfield="a",
) -> str:
    """
    This function returns the authority main heading of a given record.

    Args:
        record (dict): MARC-in-JSON record of the authority, conform with Koha API.
        headings (dict): Dictionary of main heading fields according to authority type. Default is {"PERSO_NAME": "100","CORPO_NAME": "110","CHRON_TERM": "148","TOPIC_TERM": "150","GEOGR_NAME": "151"}
        heading_subfield (str): Subfield of the main heading. Default is "a".


    Returns:
        main_heading (str): Main heading string.

    Examples:
        >>> pyreslib.koha.get_main_heading_from_authority(record=marc_json_authority)
        >>> "Example Person"
    """
    # get authority type
    auth_type = get_authority_type(record)
    heading_field = headings[auth_type]

    # get main heading
    try:
        query_field = list(
            filter(lambda x: heading_field in x.keys(), record["fields"])
        )
        # the field should be unique
        main_heading = list(
            filter(
                lambda x: heading_subfield in x.keys(),
                query_field[0][heading_field]["subfields"],
            )
        )[0][heading_subfield]

        return main_heading

    except Exception:
        return None

get_umbrella_terms_from_authority(record, umbrella_fields={'PERSO_NAME': '500', 'CORPO_NAME': '510', 'CHRON_TERM': '548', 'TOPIC_TERM': '550', 'GEOGR_NAME': '551'}, umbrella_subfields={'label': 'a', 'id': '9'})

This function returns a list of umbrella terms of a given record.

Parameters:

Name	Type	Description	Default
record	dict	MARC-in-JSON record of the authority, conform with Koha API.	required
umbrella_fields	dict	Dictionary of umbrella fields for the authority type, according to MARC21 guidelines.	{'PERSO_NAME': '500', 'CORPO_NAME': '510', 'CHRON_TERM': '548', 'TOPIC_TERM': '550', 'GEOGR_NAME': '551'}
umbrella_subfields	str	label and authority id for each umbrella term. Default is {"label": "a","id": "9"}.	{'label': 'a', 'id': '9'}

Returns:

Name	Type	Description
umbrella_terms	list	List of umbrella terms.

Examples:

>>> pyreslib.koha.get_umbrella_terms_from_authority(record=marc_json_authority)
>>> [{"type": "PERSO_NAME", "label": "Example Person", "id": "434"},{...}]

Source code in pyreslib/koha.py

def get_umbrella_terms_from_authority(
    record: dict,
    umbrella_fields={
        "PERSO_NAME": "500",
        "CORPO_NAME": "510",
        "CHRON_TERM": "548",
        "TOPIC_TERM": "550",
        "GEOGR_NAME": "551",
    },
    umbrella_subfields={"label": "a", "id": "9"},
) -> list:
    """
    This function returns a list of umbrella terms of a given record.

    Args:
        record (dict): MARC-in-JSON record of the authority, conform with Koha API.
        umbrella_fields (dict): Dictionary of umbrella fields for the authority type, according to MARC21 guidelines.
        umbrella_subfields (str): label and authority id for each umbrella term. Default is {"label": "a","id": "9"}.


    Returns:
        umbrella_terms (list): List of umbrella terms.

    Examples:
        >>> pyreslib.koha.get_umbrella_terms_from_authority(record=marc_json_authority)
        >>> [{"type": "PERSO_NAME", "label": "Example Person", "id": "434"},{...}]
    """
    # get umbrella terms
    umbrella_terms = []
    try:
        for auth_type in umbrella_fields.keys():
            query_field = list(
                filter(
                    lambda x: umbrella_fields[auth_type] in x.keys(), record["fields"]
                )
            )
            for statement in query_field:
                umbrella_term = {"type": auth_type}
                for umbrella_key in umbrella_subfields.keys():
                    metadata = list(
                        filter(
                            lambda x: umbrella_subfields[umbrella_key] in x.keys(),
                            statement[umbrella_fields[auth_type]]["subfields"],
                        )
                    )
                    if len(metadata) > 0:
                        umbrella_term[umbrella_key] = metadata[0][
                            umbrella_subfields[umbrella_key]
                        ]
                    else:
                        metadata = None

                umbrella_terms.append(umbrella_term)

        return umbrella_terms

    except Exception:
        return None

get_wd_authority_list(report_csv_file=os.path.join('data', 'mappings', 'wikidata', 'wd_authority_list.csv'), separator='|')

This method returns a list of authority IDs and their corresponding Wikidata entities from a CSV report exported from Koha.

Parameters:

Name	Type	Description	Default
report_csv_file	str	File path to the CSV report exported from Koha. See Reports page for the SQL query.	join('data', 'mappings', 'wikidata', 'wd_authority_list.csv')
separator	str	Separator for multiple values. Pipe is default.	'|'

Returns:

Name	Type	Description
authority_wd_list	list	A list of dictionaries, each containing an authority ID and its corresponding Wikidata entities.

Examples:

>>> authority_wd_list = pyreslib.koha.get_authority_wd_list(report_csv_file="path/to/authorities_wd_list.csv")
>>> [{"auth_id": 1, "type": "GEOGR_NAME", "main_heading": "Venice", "qid": ["Q641"], "wd_uri": ["http://www.wikidata.org/entity/Q641"], "wd_label": ["Venice","Venezia"]}, ...]

Source code in pyreslib/koha.py

def get_wd_authority_list(
    report_csv_file=os.path.join(
        "data", "mappings", "wikidata", "wd_authority_list.csv"
    ),
    separator="|",
) -> list:
    """
    This method returns a list of authority IDs and their corresponding Wikidata entities from a CSV report exported from Koha.

    Args:
        report_csv_file (str): File path to the CSV report exported from Koha. See [Reports](reports) page for the SQL query.
        separator (str): Separator for multiple values. Pipe is default.

    Returns:
        authority_wd_list (list): A list of dictionaries, each containing an authority ID and its corresponding Wikidata entities.

    Examples:
        >>> authority_wd_list = pyreslib.koha.get_authority_wd_list(report_csv_file="path/to/authorities_wd_list.csv")
        >>> [{"auth_id": 1, "type": "GEOGR_NAME", "main_heading": "Venice", "qid": ["Q641"], "wd_uri": ["http://www.wikidata.org/entity/Q641"], "wd_label": ["Venice","Venezia"]}, ...]


    """
    authority_wd_list = []
    with open(report_csv_file, mode="r", encoding="utf-8") as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            authority_wd_list.append(
                {
                    "auth_id": int(row["authid"]),
                    "type": row["authtypecode"],
                    "main_heading": row("main_heading"),
                    "qid": row["qid"].split(separator),
                    "wd_uri": row["wd_uri"].split(separator),
                    "wd_label": row["wd_label"].split(separator),
                }
            )

    return authority_wd_list

get_wikidata_entities_from_authority(record, wikidata_field='024', wikidata_subfields={'uri': '1', 'label': '9'})

This function returns a list of wikidata entities linked with a given record.

Parameters:

Name	Type	Description	Default
record	dict	MARC-in-JSON record of the authority, conform with Koha API.	required
wikidata_field	str	Default is "024", according to MARC21 guidelines.	'024'
wikidata_subfields	str	label and authority id for each umbrella term. Default is {"label": "a","id": "9"}.	{'uri': '1', 'label': '9'}

Returns:

Name	Type	Description
wikidata_entities	list	List of Wikidata entities.

Examples:

>>> pyreslib.koha.get_wikidata_entities_from_authority(record=marc_json_authority)
>>> [{"uri": "http://www.wikidata.org/entity/Q42", "label": "Douglas Adams"}, {"uri": "http://www.wikidata.org/entity/Q12345", "label": "Example Entity"}]

Source code in pyreslib/koha.py

def get_wikidata_entities_from_authority(
    record: dict, wikidata_field="024", wikidata_subfields={"uri": "1", "label": "9"}
) -> list:
    """
    This function returns a list of wikidata entities linked with a given record.

    Args:
        record (dict): MARC-in-JSON record of the authority, conform with Koha API.
        wikidata_field (str): Default is "024", according to MARC21 guidelines.
        wikidata_subfields (str): label and authority id for each umbrella term. Default is {"label": "a","id": "9"}.


    Returns:
        wikidata_entities (list): List of Wikidata entities.

    Examples:
        >>> pyreslib.koha.get_wikidata_entities_from_authority(record=marc_json_authority)
        >>> [{"uri": "http://www.wikidata.org/entity/Q42", "label": "Douglas Adams"}, {"uri": "http://www.wikidata.org/entity/Q12345", "label": "Example Entity"}]

    """
    # get wikidata entities
    wd_entities = []
    try:
        query_field = list(
            filter(lambda x: wikidata_field in x.keys(), record["fields"])
        )
        for statement in query_field:
            wikidata_entity = {}
            for wikidata_key in wikidata_subfields.keys():
                metadata = list(
                    filter(
                        lambda x: wikidata_subfields[wikidata_key] in x.keys(),
                        statement[wikidata_field]["subfields"],
                    )
                )
                if len(metadata) > 0:
                    subfield = wikidata_subfields[wikidata_key]
                    wikidata_entity[wikidata_key] = metadata[0][subfield]
                else:
                    metadata = None

            wd_entities.append(wikidata_entity)

        return wd_entities

    except Exception:
        return None

import_koha_authorities_from_api(session, base_url, output_directory='./data/koha_auth/json', num_workers=None)

Generate authority dictionary using parallel processing.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
base_url	str	Koha API url from credentials.	required
output_directory	str	Output directory for the generated JSON. The default filename is "auth_dict-{yyyy-mm-dd}.json". Default is data/koha_auth/json. Set to None if you don't want to save the JSON result.	'./data/koha_auth/json'
num_workers	int	Number of parallel workers (default: CPU count)	None

Returns:

Name	Type	Description
auth_dict	list	List of authority dictionaries with auth_id, wd_id, and record

Source code in pyreslib/koha.py

def import_koha_authorities_from_api(
    session,
    base_url: str,
    output_directory: str = "./data/koha_auth/json",
    num_workers: int = None,
) -> List[Dict[str, Any]]:
    """
    Generate authority dictionary using parallel processing.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        base_url (str): Koha API url from credentials.
        output_directory (str): Output directory for the generated JSON. The default filename is "auth_dict-{yyyy-mm-dd}.json". Default is `data/koha_auth/json`. Set to `None` if you don't want to save the JSON result.
        num_workers: Number of parallel workers (default: CPU count)

    Returns:
        auth_dict (list): List of authority dictionaries with auth_id, wd_id, and record
    """

    start_time = time()

    # get authority_list
    authority_list = get_auth_list_from_csv_report()

    # Set number of workers (default: number of CPU cores)
    if num_workers is None:
        num_workers = max(1, cpu_count() - 1)  # Leave one core free

    print(f"Using {num_workers} parallel workers")

    # Use Pool for parallel processing
    with Pool(num_workers) as pool:
        # Map function across all authority IDs
        # using partial to fix other parameters, such as session and base_url
        fetch_auth = partial(
            fetch_and_process_authority, session=session, base_url=base_url
        )
        results = pool.map(fetch_auth, authority_list)

    # Filter successful results
    auth_dict = [result for result in results if result["record"] is not None]

    print(
        f"Successfully imported {len(auth_dict)} authorities out of {len(authority_list)} in {float(time()-start_time)/60} minutes."
    )

    # Save results to JSON file
    output_filepath = os.path.join(
        output_directory, f"auth_dict-{utilities.get_current_date()}.json"
    )
    if output_filepath is not None:
        utilities.dict2json(auth_dict, output_filepath)

    return auth_dict

import_koha_authorities_from_marc(marc_filepath, output_directory='./data/koha_auth/json')

Returns a MARC-in-JSON dictionaries from a MARC import of the authority catalogue.

Parameters:

Name	Type	Description	Default
marc_filepath	str	Location of the MARC file to be processed. Default directory is ./data/koha_auth/marc.	required
output_directory	str	Output directory for the generated JSON. The default filename is "auth_dict-{yyyy-mm-dd}.json". Default is data/koha_auth/json. Set to None if you wish not to save the list as JSON.	'./data/koha_auth/json'

Returns: auth_dict (list): List of authority dictionaries with auth_id, wd_id, and record.

Source code in pyreslib/koha.py

def import_koha_authorities_from_marc(
    marc_filepath: str, output_directory: str = "./data/koha_auth/json"
) -> list:
    """
    Returns a MARC-in-JSON dictionaries from a MARC import of the authority catalogue.

    Args:
        marc_filepath (str): Location of the MARC file to be processed. Default directory is `./data/koha_auth/marc`.
        output_directory (str): Output directory for the generated JSON. The default filename is "auth_dict-{yyyy-mm-dd}.json". Default is `data/koha_auth/json`. Set to `None` if you wish not to save the list as JSON.
    Returns:
        auth_dict (list): List of authority dictionaries with auth_id, wd_id, and record.

    """
    # extract all records as dictionaries.
    record_dict = marc.generate_record_dict(
        marc_filepath=marc_filepath, json_filepath=None, id_name="auth_id"
    )

    auth_dict = []
    for record in record_dict:
        # extract wikidata_qids
        wd_id = []
        wd_id = extract_wikidata_id(record["record"])
        auth_dict.append(
            {"auth_id": record["auth_id"], "wd_id": wd_id, "record": record["record"]}
        )

    # Save results to JSON file
    output_filepath = os.path.join(
        output_directory, f"auth_dict-{utilities.get_current_date()}.json"
    )
    if output_directory is not None:
        utilities.dict2json(auth_dict, output_filepath)

    return auth_dict

import_koha_biblios_from_api(session, base_url, output_directory='./data/koha_biblio/json', num_workers=None)

Generate authority dictionary using parallel processing.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
base_url	str	Koha API url from credentials.	required
output_directory	str	Output directory for the generated JSON. The default filename is "biblio_dict-{yyyy-mm-dd}.json". Default is data/koha_biblio/json.	'./data/koha_biblio/json'
num_workers	int	Number of parallel workers (default: CPU count)	None

Returns:

Type	Description
List[Dict[str, Any]]	List of bibliographic record dictionaries with biblio_id and record

Source code in pyreslib/koha.py

def import_koha_biblios_from_api(
    session,
    base_url: str,
    output_directory: str = "./data/koha_biblio/json",
    num_workers: int = None,
) -> List[Dict[str, Any]]:
    """
    Generate authority dictionary using parallel processing.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        base_url (str): Koha API url from credentials.
        output_directory (str): Output directory for the generated JSON. The default filename is "biblio_dict-{yyyy-mm-dd}.json". Default is `data/koha_biblio/json`.
        num_workers: Number of parallel workers (default: CPU count)

    Returns:
        List of bibliographic record dictionaries with biblio_id and record
    """

    start_time = time()

    # get authority_list
    biblio_list = get_biblio_list_from_csv_report()

    # Set number of workers (default: number of CPU cores)
    if num_workers is None:
        num_workers = max(1, cpu_count() - 1)  # Leave one core free

    print(f"Using {num_workers} parallel workers")

    # Use Pool for parallel processing
    with Pool(num_workers) as pool:
        # Map function across all authority IDs
        # using partial to fix other parameters, such as session and base_url
        fetch_biblio = partial(
            fetch_and_process_biblio, session=session, base_url=base_url
        )
        results = pool.map(fetch_biblio, biblio_list)

    # Filter successful results
    biblio_dict = [result for result in results if result["record"] is not None]

    print(
        f"Successfully imported {len(biblio_dict)} biblios out of {len(biblio_list)} in {float(time()-start_time)/60} minutes."
    )

    # Save results to JSON file
    output_filepath = os.path.join(
        output_directory, f"biblio_dict-{utilities.get_current_date()}.json"
    )

    utilities.dict2json(biblio_dict, output_filepath)

    return biblio_dict

import_koha_biblios_from_marc(marc_filepath, output_directory='./data/koha_biblio/json')

Returns a MARC-in-JSON dictionaries from a MARC import of the bibliographic catalogue.

Parameters:

Name	Type	Description	Default
marc_filepath	str	Location of the MARC file to be processed. Default directory is ./data/koha_biblio/marc.	required
output_directory	str	Output directory for the generated JSON. The default filename is "biblio_dict-{yyyy-mm-dd}.json". Default directory is ./data/koha_biblio/json. Set to None if you wish not to save the list as JSON.	'./data/koha_biblio/json'

Retruns

biblio_dict (list): List of bibliopgraphic record dictionaries with biblio_id and record

Source code in pyreslib/koha.py

def import_koha_biblios_from_marc(
    marc_filepath: str, output_directory: str = "./data/koha_biblio/json"
) -> list:
    """
    Returns a MARC-in-JSON dictionaries from a MARC import of the bibliographic catalogue.

    Args:
        marc_filepath (str): Location of the MARC file to be processed. Default directory is `./data/koha_biblio/marc`.
        output_directory (str): Output directory for the generated JSON. The default filename is "biblio_dict-{yyyy-mm-dd}.json". Default directory is `./data/koha_biblio/json`. Set to `None` if you wish not to save the list as JSON.

    Retruns:
        biblio_dict (list): List of bibliopgraphic record dictionaries with biblio_id and record

    """
    # extract all records as dictionaries.
    record_dict = marc.generate_record_dict(
        marc_filepath=marc_filepath, json_filepath=None, id_name="biblio_id"
    )

    biblio_dict = []
    for record in record_dict:
        biblio_dict.append({"biblio_id": record["auth_id"], "record": record["record"]})

    # Save results to JSON file
    output_filepath = os.path.join(
        output_directory, f"biblio_dict-{utilities.get_current_date()}.json"
    )
    if output_directory is not None:
        utilities.dict2json(biblio_dict, output_filepath)

    return biblio_dict

koha_session(client_id, client_secret, user_agent, base_url, scope='all')

Returns an OAuth2 session, given client ID and secret key of your Koha account.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID associated with your Koha Admin User.	required
client_secret	str	Secret key provided by the Koha Administration.	required
user_agent	str	User-Agent string. Default is None.	required
base_url	str	Base URL for your Koha Staff interface	required

Returns:

Name	Type	Description
session	oauth2	OAuth2 session

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")

Source code in pyreslib/koha.py

def koha_session(
    client_id: str, client_secret: str, user_agent: str, base_url: str, scope="all"
):
    """Returns an OAuth2 session, given client ID and secret key of your Koha account.

    Args:
        client_id (str): Client ID associated with your Koha Admin User.
        client_secret (str): Secret key provided by the Koha Administration.
        user_agent (str): User-Agent string. Default is None.
        base_url (str): Base URL for your Koha Staff interface

    Returns:
        session (oauth2): OAuth2 session

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")

    """
    token_url = f"{base_url}/oauth/token"  # This may be different for your endpoint

    oauth2client = OAuth2Client(
        token_endpoint=token_url, client_id=client_id, client_secret=client_secret
    )
    # Force the User-Agent before authorization
    if user_agent is not None:
        oauth2client.session.headers.update({"User-Agent": user_agent})
    auth = OAuth2ClientCredentialsAuth(oauth2client, scope=scope, resource=base_url)
    session = requests.Session()
    session.auth = auth
    # And also later, why not?
    if user_agent is not None:
        session.headers.update({"User-Agent": user_agent})
    return session

query_biblio_marc(session, base_url, q, limit=20, offset=0)

Returns a list of MARC-in-JSON biblio record matching a query according to Koha standards. This method exploits the ListBiblio operation.

Unfortunately the method does not yet work.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
base_url	str	Koha API url from credentials.	required
q	str	Query string according to Koha standards. For example 'title:"Orlando furioso" AND author:"Ariosto"'	required

Examples:

Source code in pyreslib/koha.py

def query_biblio_marc(
    session, base_url: str, q: str, limit: int = 20, offset: int = 0
) -> list:
    """
    Returns a list of MARC-in-JSON biblio record matching a query according to Koha standards. This method exploits the [ListBiblio](https://api.koha-community.org/#tag/biblios/operation/listBiblio) operation.

    **Unfortunately the method does not yet work.**

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        base_url (str): Koha API url from credentials.
        q (str): Query string according to Koha standards. For example 'title:"Orlando furioso" AND author:"Ariosto"'


    Examples:


    """
    headers = {"Accept": "application/json"}
    params = {"limit": limit, "offset": offset, "q": q}

    try:
        response = session.get(
            f"{base_url}/biblios",
            headers=headers,
            params=params,
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"API Error: {e.response.status_code} - {e.response.text}")
        raise

update_authority_marc(session, auth_id, marc_json, base_url)

Update authority record given a MARC-in-JSON payload via Koha API.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
auth_id	int	Authority ID for the requested record.	required
marc_json	dict	JSON dictionary serialization of the record.	required

Returns:

Type	Description
dict	None

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> marc_json_authority = {'leader': '00744nam a2200217Ia 4500', 'fields': [{'000': '00373cz  a2200157n  4500'}, {'001': '407'}, {'003': 'OI'}, {'005': '20260330105335.0'} ...
>>> update_authority_marc(my_session,407,marc_json_authority)

Source code in pyreslib/koha.py

def update_authority_marc(
    session, auth_id: int, marc_json: dict, base_url: str
) -> dict:
    """Update authority record given a MARC-in-JSON payload via Koha API.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        auth_id (int): Authority ID for the requested record.
        marc_json (dict): JSON dictionary serialization of the record.

    Returns:
        `None`

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> marc_json_authority = {'leader': '00744nam a2200217Ia 4500', 'fields': [{'000': '00373cz  a2200157n  4500'}, {'001': '407'}, {'003': 'OI'}, {'005': '20260330105335.0'} ...
        >>> update_authority_marc(my_session,407,marc_json_authority)


    """
    framework_id = get_framework_id_authority(
        session=session, auth_id=auth_id, base_url=base_url
    )
    if str(framework_id) != "":
        headers = {
            "Accept": "application/json",
            "Content-type": "application/marc-in-json",
            "x-authority-type": str(framework_id),
        }
    else:  # default framework
        headers = {
            "Accept": "application/json",
            "Content-type": "application/marc-in-json",
        }
    response = session.put(
        f"{base_url}/authorities/{str(auth_id)}",
        headers=headers,
        data=json.dumps(
            marc_json
        ),  # dumps serializes the Python dictionary into JSON string
    )

    return None

update_biblio_marc(session, biblio_id, marc_json, base_url)

Update biblio record given a MARC-in-JSON payload via Koha API.

Parameters:

Name	Type	Description	Default
session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
biblio_id	int	Biblio ID for the requested record.	required
marc_json	dict	JSON dictionary serialization of the record.	required
base_url	str	Koha API url from credentials.	required

Returns:

Type	Description
dict	None

Examples:

>>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
>>> marc_json_record = {'leader': '00744nam a2200217Ia 4500', 'fields': [{'000': '00373cz  a2200157n  4500'}, {'001': '12'}, {'003': 'OI'}, {'005': '20260330105335.0'} ...
>>> update_biblio_marc(my_session,12,marc_json_record)

Source code in pyreslib/koha.py

def update_biblio_marc(session, biblio_id: int, marc_json: dict, base_url: str) -> dict:
    """Update biblio record given a MARC-in-JSON payload via Koha API.

    Args:
        session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        biblio_id (int): Biblio ID for the requested record.
        marc_json (dict): JSON dictionary serialization of the record.
        base_url (str): Koha API url from credentials.

    Returns:
        `None`

    Examples:
        >>> my_session = pyreslib.koha.oauth2_session(client_id="{CLIENT_ID}"", client_secret="{SECRET_KEY}" , base_url="https://{KOHA_STAFF_URL}/api/v1")
        >>> marc_json_record = {'leader': '00744nam a2200217Ia 4500', 'fields': [{'000': '00373cz  a2200157n  4500'}, {'001': '12'}, {'003': 'OI'}, {'005': '20260330105335.0'} ...
        >>> update_biblio_marc(my_session,12,marc_json_record)


    """
    framework_id = get_framework_id_biblioitem(
        session=session, biblio_id=biblio_id, base_url=base_url
    )
    if str(framework_id) != "":
        headers = {
            "Accept": "application/json",
            "Content-type": "application/marc-in-json",
            "x-framework-id": str(framework_id),
        }
    else:  # dafault framework
        headers = {
            "Accept": "application/json",
            "Content-type": "application/marc-in-json",
        }
    response = session.put(
        f"{base_url}/biblios/{str(biblio_id)}",
        headers=headers,
        data=json.dumps(
            marc_json
        ),  # dumps serializes the Python dictionary into JSON string
    )
    return None

Enhancing modules

pyreslib.wikidata

add_qids_to_authorities(koha_session, koha_base_url, auth_qid_csv='./data/wikidata/ingest_qid_to_auth/add_qids_to_authorities.csv', field='024', subfield='1')

Given a CSV file with fields authid, main_heading, wd_uri and type, generated from Koha report (see reports section) and enriched via OpenRefine, the method updates the metadata authorities back to Koha via API.

Parameters:

Name	Type	Description	Default
koha_session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
koha_base_url	str	Koha API url from credentials.	required
auth_qid_csv	str	filepath for the CSV file. Default is "./data/wikidata/ingest_qid_to_auth/add_qids_to_authorities.csv".	'./data/wikidata/ingest_qid_to_auth/add_qids_to_authorities.csv'
field	str	Wikidata field. Default is "024", according to MARC21 framework.	'024'
subfield	str	Wikidata URI subfield. Default is "1", according to MARC21 framework.	'1'

Returns:

Type	Description
	None

Source code in pyreslib/wikidata.py

def add_qids_to_authorities(koha_session,koha_base_url: str, auth_qid_csv: str = "./data/wikidata/ingest_qid_to_auth/add_qids_to_authorities.csv", field: str = "024", subfield:str = "1"):
	"""
	Given a CSV file with fields `authid`, `main_heading`, `wd_uri` and `type`, generated from Koha report (see [reports](reports.md) section)
	and enriched via [OpenRefine](https://openrefine.org/docs/manual/reconciling), the method updates the metadata authorities back to Koha via API.

	Args:
		 koha_session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
		 koha_base_url (str): Koha API url from credentials.
		 auth_qid_csv(str): filepath for the CSV file. Default is "./data/wikidata/ingest_qid_to_auth/add_qids_to_authorities.csv".
		 field (str): Wikidata field. Default is "024", according to MARC21 framework.
		 subfield (str): Wikidata URI subfield. Default is "1", according to MARC21 framework.

	Returns:
		`None`
	"""

	# import csv as dictionary
	auth_qid_list = utilities.csv2dict(auth_qid_csv)

	# ingest for each authority the new URI value in field$subfield position
	for auth in auth_qid_list:
		auth_id = auth["authid"]
		main_heading = auth["main_heading"]
		wd_uri = auth["wd_uri"]
		#print(f"Current authority: {auth_id}")

		modified = False

		if "wikidata.org" in wd_uri:
			print(f"Getting {auth_id} record from API...")
			record = koha.get_authority_marc(session=koha_session,base_url=koha_base_url,auth_id=auth_id)
			# enrich 024 record
			field_query = list(filter(lambda x: field in x.keys(), record["fields"]))
			if len(field_query) >0:
				# check if QID already present in subfield
				found = False
				for statement in field_query:
					substatement_query = list(filter(lambda x: subfield in x.keys(), statement[field]["subfields"]))
					if len(substatement_query) >0:
						if substatement_query[0][subfield] == wd_uri:
							print(f"Wikidata URI {wd_uri} already present. Skipping...")
							found = True
							break
				if found is not True:
					modified = True
					# append new statement after last one
					print(f"Appending Wikidata URI to new {field} field...")
					record["fields"].insert(
						record["fields"].index(field_query[-1]),
						{field: {"ind1": ' ', "ind2": ' ', "subfields": [{subfield: wd_uri}]}}
						)
			else:
				modified = True
				print(f"Create new {field} field for URI...")
				# the field 024 is not present in record, create a new one and append it in the right position
				for statement in record["fields"]:
					statement_field = int(list(statement.keys())[0])
					if statement_field > int(field):
						# append before statement
						record["fields"].insert(record["fields"].index(statement)-1,
							{field: {"ind1": ' ', "ind2": ' ', "subfields": [{subfield: wd_uri}]}})
						break

			print(modified)
			if modified:
				print(f"Updating Wikidata URI to {auth_id} authority...")
				koha.update_authority_marc(session=koha_session,auth_id=auth_id,marc_json=record,base_url=koha_base_url)

append_qid_to_qid_log(qid, qid_log, wb)

Appends metadata and occurrence information from a Wikidata entity which might become a candidate authority for the catalogue.

The user can evaluate the final result in data/wikidata/qid_log folder.

Parameters:

Name	Type	Description	Default
qid	str	Wikdiata QID value for entity.	required
qid_log	list	List of dictionaries holding metadata on Wikidata entities that could be part of the Koha thesaurus and their occurrence in statements.	required
wb		Wikibase Integrator session generated via pyreslib.wikidata.wikibase_integrator_session_basic or pyreslib.wikidata.wikibase_integrator_session_oauth2.	required

Returns:

Name	Type	Description
qid_log	list	Updated qid_log list.

Source code in pyreslib/wikidata.py

def append_qid_to_qid_log(qid: str,qid_log: list, wb ):
	"""
	Appends metadata and occurrence information from a Wikidata entity which might become a candidate authority for the catalogue.

	The user can evaluate the final result in `data/wikidata/qid_log` folder.

	Args:
		qid(str): Wikdiata QID value for entity.
		qid_log (list): List of dictionaries holding metadata on Wikidata entities that could be part of the Koha thesaurus and their occurrence in statements.
		wb: Wikibase Integrator session generated via [pyreslib.wikidata.wikibase_integrator_session_basic][] or [pyreslib.wikidata.wikibase_integrator_session_oauth2][].

	Returns:
		qid_log (list): Updated qid_log list.
	"""

	if qid != "":
		# search if qid is already in log
		query_qid = list(filter(lambda x: x["qid"] == qid,qid_log))
		if len(query_qid) > 0:
			# append value
			query_qid[0]["occurrence"] += 1
		else:
			# add new qid to log list
			wd_entity = wb.item.get(qid,max_retries=15,retry_after=10)
			try:
				instance_of = wd_entity.claims.get("P31")[0].mainsnak.datavalue['value']["id"]
			except (IndexError,AttributeError):
				instance_of = ""
			try:
				description = wd_entity.descriptions.get("en").value
			except (IndexError,AttributeError):
				description = ""

			try:
				label = wd_entity.labels.get("en").value
			except (IndexError,AttributeError):
				label = ""
			qid_log.append({"qid": qid, "occurrence": 1,
			 "label": label,
			  "description": description ,
			  "instance_of": instance_of,
			   "uri": "http://wikidata.org/entity/" + qid } 
			   )

	return qid_log 		   	

convert_point_in_time_to_date(point_in_time, date_format=f'%Y-%m-%d')

Returns a formatted date given a Wikibase [Point in time] string. The default format is %y-%m-%d according to the datetime syntaxt. For DD/MM/YYYY format use %d/%m/%y instead.

Parameters:

Name	Type	Description	Default
point_in_time	str	Wikibase Point in Time string similar to ISO 8061 date, with the exception of + sign. The conversion only works for AD dates.	required
date_format	str	datetime formatting string. Default set to ISO 8601 date.	f'%Y-%m-%d'

Returns:

Name	Type	Description
date	str	Formatted date.

Examples:

>>> point_in_time = "+2016-01-01T00:00:00Z"
>>> date_value = convert_point_in_time_to_date(point_in_time)
>>> print(date_value)
>>> '2016-01-01'

Source code in pyreslib/wikidata.py

def convert_point_in_time_to_date(point_in_time: str, date_format=f"%Y-%m-%d") -> str:
	"""Returns a formatted date given a Wikibase [Point in time] string. The default format is `%y-%m-%d` according to the [datetime](https://docs.python.org/3/library/datetime.html) syntaxt. For `DD/MM/YYYY` format use `%d/%m/%y` instead.

	Args:
		point_in_time: Wikibase Point in Time string similar to ISO 8061 date, with the exception of + sign. The conversion only works for AD dates.
		date_format (str): `datetime` formatting string. Default set to ISO 8601 date.


	Returns:
		date (str): Formatted date.

	Examples:
		>>> point_in_time = "+2016-01-01T00:00:00Z"
		>>> date_value = convert_point_in_time_to_date(point_in_time)
		>>> print(date_value)
		>>> '2016-01-01'
	"""
	# extract date from Wikibase Point in time
	extracted_date = point_in_time[1:11]
	year = extracted_date.split("-")[0]
	month = extracted_date.split("-")[1]
	day = extracted_date.split("-")[2]
	date = ""
	# Convert missing date information to fit Koha date format.
	if day == "00":
		if month == "00":
			date = datetime.date(int(year), 1, 1).strftime(date_format)
		else:
			date = datetime.date(int(year), int(month), 1).strftime(date_format)
	else:
		date = datetime.date(int(year), int(month), int(day)).strftime(date_format)

	return date

convert_qid_to_URI(qid, base_uri='http://wikidata.org/')

Returns a QID string into the entity URI. By default the method uses the Wikidata Concept URI.

Parameters:

Name	Type	Description	Default
qid	str	Wikibase Point in Time string similar to ISO 8061 date, with the exception of + sign. The conversion only works for AD dates.	required
base_uri	str	By default set to http://wikidata.org/.	'http://wikidata.org/'

Returns:

Name	Type	Description
URI	str	URI of the entity.

Examples:

>>> qid = "Q1296"
>>> print(convert_qid_to_URI(qid,base_uri="http://my.wikibase.org/"))
>>> 'http://my.wikibase.org/Q1296'

Source code in pyreslib/wikidata.py

def convert_qid_to_URI(qid: str, base_uri="http://wikidata.org/") -> str:
	"""Returns a QID string into the entity URI. By default the method uses the Wikidata Concept URI.

	Args:
		qid (str): Wikibase Point in Time string similar to ISO 8061 date, with the exception of + sign. The conversion only works for AD dates.
		base_uri (str): By default set to `http://wikidata.org/`.


	Returns:
		URI (str): URI of the entity.

	Examples:
		>>> qid = "Q1296"
		>>> print(convert_qid_to_URI(qid,base_uri="http://my.wikibase.org/"))
		>>> 'http://my.wikibase.org/Q1296'
	"""
	return f"{base_uri}{qid}"

enhance_authorities_via_wikidata(auth_dict, wb, auth_id_range={'min': 1, 'max': None}, backup_frequency=10, qid_log_dir=os.path.join('data', 'wikidata', 'qid_log'), wikidata_koha_mapping_filepath=os.path.join('data', 'mappings', 'wikidata', 'wikidata-koha-properties.csv'), wikibase_base_url='http://wwww.wikidata.org/entity/', backup_auth_dir=os.path.join('data', 'wikidata', 'backup_auth'), changed_auth_dir=os.path.join('data', 'wikidata', 'changed_auth'), in_progress=False, exclude_types=[], koha_fields={'PERSO_NAME': '500', 'CORPO_NAME': '510', 'CHRON_TERM': '548', 'TOPIC_TERM': '550', 'GEOGR_NAME': '551'}, headings={'PERSO_NAME': '100', 'CORPO_NAME': '110', 'CHRON_TERM': '148', 'TOPIC_TERM': '150', 'GEOGR_NAME': '151'})

Generates a list of dictionaries of enhanced authorities and their backups to be later submitted via the Koha API.

Parameters:

Name	Type	Description	Default
auth_dict	list	Authorities list of dictionaries formatted according to pyreslib.koha.import_koha_authorities_from_marc or pyreslib.koha.import_koha_authorities_from_api.	required
wb		Wikibase Integrator session generated via pyreslib.wikidata.wikibase_integrator_session_basic or pyreslib.wikidata.wikibase_integrator_session_oauth2.	required
auth_id_range	dict	Minimal and maximal authority id number for enhancement, previous and subsequent ones will be ignored. Default is {"min": 1,"max": None}, meaning all catalogue is going to be exposed to enhancement.	{'min': 1, 'max': None}
backup_frequency	int	Backup frequency.	10
wikidata_koha_mapping_filepath	str	Path to Wikidata-Koha mapping CSV.	join('data', 'mappings', 'wikidata', 'wikidata-koha-properties.csv')
wikibase_base_url	str	Base URL for Wikibase instance. Default is wikidata URI "http://wwww.wikidata.org/entity/".	'http://wwww.wikidata.org/entity/'
backup_auth_dir	str	Destination path for backup enhanced authorities as JSON.	join('data', 'wikidata', 'backup_auth')
changed_auth_dir	str	Destination path for changed enhanced authorities as JSON.	join('data', 'wikidata', 'changed_auth')
in_progress	bool	Retrieves latest files in data/wikidata/changed_auth and data/wikidata/backup_auth. Default is False.	False
exclude_types	list	Authority type codes to be excluded from the process. Default is empty list.	[]
koha_fields	dict	Fields for umbrella terms statements for authority record. Default list is definined according to the 5XX MARC21 framework.	{'PERSO_NAME': '500', 'CORPO_NAME': '510', 'CHRON_TERM': '548', 'TOPIC_TERM': '550', 'GEOGR_NAME': '551'}
headings	dict	Fields for main heading statements for authority record. Default list is definined according to the 1XX MARC21 framework.	{'PERSO_NAME': '100', 'CORPO_NAME': '110', 'CHRON_TERM': '148', 'TOPIC_TERM': '150', 'GEOGR_NAME': '151'}

Returns:

Type	Description
	changed_authorities, backup_authorities (list): Lists of dictionaries for the enhanced authorities and their backups. You can locally store, or submit the changes to the catalogue via the API later on.

Examples:

>>> auth_dict = pyreslib.koha.import_koha_authorities_from_api(session=koha_session,base_url=koha_base_url)
>>> changed_authorities, backup_authorities  = pyreslib.wikidata.external_sources_metadata_authorities(auth_dict=auth_dict)
>>> changed_authorities
>>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": [...,{ ... } ...] }
>>> backup_authorities
>>> >>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": [..., { ... } ] }

Source code in pyreslib/wikidata.py

def enhance_authorities_via_wikidata(
	auth_dict: list,
	wb,
	auth_id_range: list = {"min": 1,"max": None},
	backup_frequency: int = 10,
	qid_log_dir: str = os.path.join("data","wikidata","qid_log"),
	wikidata_koha_mapping_filepath: str = os.path.join("data","mappings","wikidata","wikidata-koha-properties.csv"),
	wikibase_base_url: str = "http://wwww.wikidata.org/entity/",
	backup_auth_dir: str = os.path.join("data","wikidata","backup_auth"),
	changed_auth_dir: str = os.path.join("data","wikidata","changed_auth"),
	in_progress: bool = False,
	exclude_types: list = [],
	koha_fields: dict = {"PERSO_NAME": "500","CORPO_NAME": "510","CHRON_TERM": "548","TOPIC_TERM": "550","GEOGR_NAME": "551"},
	headings: dict = {"PERSO_NAME": "100","CORPO_NAME": "110","CHRON_TERM": "148","TOPIC_TERM": "150","GEOGR_NAME": "151"}
	):
	"""
	Generates a list of dictionaries of enhanced authorities and their backups to be later submitted via the Koha API.

	Args:
		auth_dict (list): Authorities list of dictionaries formatted according to [pyreslib.koha.import_koha_authorities_from_marc][] or [pyreslib.koha.import_koha_authorities_from_api][].
		wb: Wikibase Integrator session generated via [pyreslib.wikidata.wikibase_integrator_session_basic][] or [pyreslib.wikidata.wikibase_integrator_session_oauth2][].
		auth_id_range (dict): Minimal and maximal authority id number for enhancement, previous and subsequent ones will be ignored. Default is {"min": 1,"max": None}, meaning all catalogue is going to be exposed to enhancement.
		backup_frequency (int): Backup frequency.
		wikidata_koha_mapping_filepath (str): Path to Wikidata-Koha mapping CSV.
		wikibase_base_url(str): Base URL for Wikibase instance. Default is wikidata URI "http://wwww.wikidata.org/entity/".
		backup_auth_dir (str): Destination path for backup enhanced authorities as JSON.
		changed_auth_dir (str): Destination path for changed enhanced authorities as JSON.
		in_progress(bool): Retrieves latest files in `data/wikidata/changed_auth` and `data/wikidata/backup_auth`. Default is `False`.
		exclude_types (list): Authority type codes to be excluded from the process. Default is empty list.
		koha_fields (dict): Fields for umbrella terms statements for authority record. Default list is definined according to the 5XX MARC21 framework.
		headings (dict): Fields for main heading statements for authority record. Default list is definined according to the 1XX MARC21 framework. 

	Returns:
		changed_authorities, backup_authorities (list): Lists of dictionaries for the enhanced authorities and their backups. You can locally store, or submit the changes to the catalogue via the API later on.		

	Examples:
		>>> auth_dict = pyreslib.koha.import_koha_authorities_from_api(session=koha_session,base_url=koha_base_url)
		>>> changed_authorities, backup_authorities  = pyreslib.wikidata.external_sources_metadata_authorities(auth_dict=auth_dict)
		>>> changed_authorities
		>>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": [...,{ ... } ...] }
		>>> backup_authorities
		>>> >>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": [..., { ... } ] }
	"""


	# import Wikidata-Koha properties mapping from CSV
	print(f"Importing mapping CSV {wikidata_koha_mapping_filepath}...")
	wikidata_koha_mapping = utilities.csv2dict(wikidata_koha_mapping_filepath)

	n_authorities = len(auth_dict)
	print(f"Number of authorities: {n_authorities}")
	current_auth = 0
	if in_progress is not True:
		backup_authorities = []
		changed_authorities = []
		qid_log = []
	else:
		# get latest backup and changed
		print(f"Importing latest backup, changed and qid_log dictionaries...")
		backup_authorities = utilities.json2dict(get_latest_file(backup_auth_dir))
		changed_authorities = utilities.json2dict(get_latest_file(changed_auth_dir))   
		qid_log = utilities.csv2dict(get_latest_file(qid_log_dir))
		# transform qid_log occurrence field to int
		for entry in qid_log:
			entry["occurrence"] = int(entry["occurrence"])

	print("Sorting authorities according to their QID...")
	auth_dict_sorted_qid,auth_qid_list = generate_qid_sorted_dict_and_list(auth_dict,exclude_types=exclude_types)

	# Counters for partial backups, and measuring operation.
	backup_counter = 0
	start_time = time()

	if auth_id_range["max"] is None:
		# assign last authority id
		auth_id_range["max"] = int(auth_dict[-1]["auth_id"]) +1

	print("Enhancing authorities with Wikidata data...")

	for auth in auth_dict:
		current_auth +=1
		if int(auth["auth_id"]) < auth_id_range["min"]:
			continue
		elif int(auth["auth_id"]) > auth_id_range["max"]:
			break	
		else:
			print(f"\n\n####### CURRENT AUTHORITY: {auth["auth_id"]} {float(current_auth)/n_authorities*100}%")
			# check if authority is already in backup_authorities
			if in_progress is True:
				backup_query = list(filter(lambda x: x["auth_id"] == auth["auth_id"], backup_authorities))
				if len(backup_query) > 0:
					print("Authority already processed, skipping...")
					continue

			# exclude if belongs to exclude_types
			auth_type = koha.get_authority_type(auth["record"])
			if auth_type in exclude_types + [None]:
					print("Excluding authority from enhancement...")
					continue

			backup_auth = deepcopy(auth)
			changed_record = False
			if len(auth["wd_id"]) >0:
				for qid in auth["wd_id"]:
					qid = "Q"+str(qid)
					print(f"QID: {qid}")
					# get wikidata entity associated with authority
					try:
						entity = wb.item.get(qid,max_retries=10,retry_after=5)
					except Exception:
						entity = None
					# get properties from mapping according to auth_type (source field)
					properties = list(filter(lambda x: x["type_source"] == auth_type, wikidata_koha_mapping))

					if entity is not None:
						# query Wikidata for each property
						for prop in properties:
							pid = prop["pid"]
							#print(f"CURRENT PROPERTY: {pid}")
							query = wb_get_property_data(entity, pid)
							for value in [q["value"] for q in query]:
								if value != "":
									print(f"Queried value: {value}")
									if "wikidata.org" in value:
										value_uri = value
										value_qid = value.split("/")[-1]
									else:
										value_uri = wikibase_base_url + value
										value_qid = value

									if value_qid != qid: # exclude reflective statements.	
										#1. value is already in authority record, but no $i subfield --> Add value_i_subfield
										# search QID value in authority record field
										try:
											field = koha_fields[prop["type_target"]]
											field_query = list(filter(lambda x: field in x.keys(),auth["record"]["fields"]))

											if len(field_query) > 0: # statement(s) in authority

												for statement in field_query:
													#print(statement)
													#print(statement[field])
													subfield_query = list(filter(lambda x: "1" in x.keys(),statement[field]["subfields"]))
													for subfield_1 in subfield_query:
														if subfield_1["1"] == value_uri:
															# check if $i subfield is already filled
															subfield_i_query = list(filter(lambda x: "i" in x.keys(),statement[field]["subfields"]))
															try:
																if subfield_i_query[0]["i"] != "":
																	#2. value is already in authority record, and $i is filled --> skip
																	continue
																else:
																	#value is already in authority record, but no $i subfield --> Add value_i_subfield
																	subfield_i_query[0]["i"] = prop["value_i_subfield"]
																	changed_record = True


															except IndexError:
																#value is already in authority record, but no $i subfield --> Add value_i_subfield
																statement[field]["subfields"].append({"i": prop["value_i_subfield"]})
																changed_record = True

											else: # statement not in authority
												# add authority statement, if QID matches an existing authority
												retrieved_authority = retrieve_authority_from_qid(value_qid,auth_qid_list,auth_dict_sorted_qid)
												if retrieved_authority is None:
													print(f"Value {value_qid} not found in Koha thesaurus. Adding it to log list...")
													append_qid_to_qid_log(value_qid,qid_log,wb)
												else:
													# value is in the authority
													print(f"Found authority {retrieved_authority["auth_id"]}. Adding it as statement for {auth["auth_id"]}...")
													try:
														retrieved_authority_heading = list(filter(lambda x: headings[prop["type_target"]] in x.keys(), retrieved_authority["record"]["fields"] ))
														retrieved_authority_heading = retrieved_authority_heading[0][headings[prop["type_target"]]]["subfields"][0]["a"]
														#print(retrieved_authority_heading)
														#input()
														print(f"a: {retrieved_authority_heading} \n 9: {retrieved_authority["auth_id"]} \n i: {prop["value_i_subfield"]} ")
														auth["record"]["fields"].append({field: {"ind2": " ","ind1": " ", "subfields": [{"a": retrieved_authority_heading ,"9": retrieved_authority["auth_id"] ,"i": prop["value_i_subfield"] }]}})
														changed_record = True

													except Exception:
														pass

										except KeyError:
											continue

									else:
										print(f"Excluding reflective statement for {qid}")

			else:
				continue

			print(f"Changed record: {changed_record}")
			if changed_record is True:
				backup_authorities.append(backup_auth)
				changed_authorities.append(auth)
				print(changed_record)
				backup_counter +=1
				if backup_counter == backup_frequency:
					print("\n\n BACKING UP...")
					# Saving to JSON...
					utilities.dict2json(
						backup_authorities,
						os.path.join(
							backup_auth_dir, "backup_auth-" + utilities.get_current_date() + ".json"
						),
					)
					utilities.dict2json(
						changed_authorities,
						os.path.join(
							changed_auth_dir, "changed_auth-" + utilities.get_current_date() + ".json"
						),
					)
					# Saving to CSV
					utilities.dict2csv(qid_log,os.path.join(qid_log_dir,f"qid_log-{utilities.get_current_date()}.csv"))

					backup_counter = 0

				#print(f"Changed authority: {changed_authorities[-1]} \n\n")
				#input()




	# Saving qid_log
	print(f"Saving QIDs log file to {qid_log_dir}")
	utilities.dict2csv(qid_log,os.path.join(qid_log_dir,f"qid_log-{utilities.get_current_date()}.csv"))

	print(f"Enhancing completed in {float(time() - start_time)/60} minutes.")

	# return backup and changed_authorities
	return backup_authorities,changed_authorities

external_sources_metadata_authorities(auth_dict, wb, backup_auth_dir=os.path.join('data', 'wikidata', 'backup_auth'), changed_auth_dir=os.path.join('data', 'wikidata', 'changed_auth'), external_ids_mapping_filepath=os.path.join('data', 'mappings', 'lod', 'external_identifiers.json'), uri_field=['024', '1'], source_subfield='2', label_subfield='9', id_subfield='a')

Given a list of External identifiers, by default stored in data/mappins/lod/external_identifiers.json, authorities' 024 fields are enhanced with extra metadata such as source, label and id.

Parameters:

Name	Type	Description	Default
auth_dict	list	Authorities list of dictionaries formatted according to pyreslib.koha.import_koha_authorities_from_marc or pyreslib.koha.import_koha_authorities_from_api.	required
wb		Wikibase Integrator session generated via pyreslib.wikidata.wikibase_integrator_session_basic or pyreslib.wikidata.wikibase_integrator_session_oauth2.	required
external_ids_mapping_filepath	str	Path to external ids mapping JSON.	join('data', 'mappings', 'lod', 'external_identifiers.json')
source_subfield	str	Koha subfield where the name of the external source is recorded. Default is "2", but you can change it according to your MARC framework.	'2'
label_subfield	str	Koha subfield where the label of the external source is recorded. Default is "9", but you can change it according to your MARC framework.	'9'
id_subfield	str	Koha subfield where the id of the external source is recorded. Default is "a", but you can change it according to your MARC framework.	'a'

Returns:

Type	Description
	changed_authorities, backup_authorities (list): Lists of dictionaries for the enhanced authorities and their backups. You can locally store, or submit the changes to the catalogue via the API later on.

Examples:

>>> auth_dict = pyreslib.koha.import_koha_authorities_from_api(session=koha_session,base_url=koha_base_url)
>>> changed_authorities, backup_authorities  = pyreslib.wikidata.external_sources_metadata_authorities(auth_dict=auth_dict)
>>> changed_authorities
>>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": {"024": {... "subfields": [{"1": "http://www.wikidata.org/entity/Q254"},{"9": "Wolfgang Amadeus Mozart"},{"a": "Q254"},{"2": "Wikidata"} } ... ]} ...]
>>> backup_authorities
>>> >>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": {"024": {... "subfields": [{"1": "http://www.wikidata.org/entity/Q254"}] ...} ]

Source code in pyreslib/wikidata.py

def external_sources_metadata_authorities(
	auth_dict,
	wb,
	backup_auth_dir: str = os.path.join("data","wikidata","backup_auth"),
	changed_auth_dir: str = os.path.join("data","wikidata","changed_auth"),
	external_ids_mapping_filepath: str = os.path.join("data","mappings","lod","external_identifiers.json"),
	uri_field: list =["024","1"],
	source_subfield: str ="2",
	label_subfield: str ="9",
	id_subfield: str="a"):
	"""
	Given a list of External identifiers, by default stored in `data/mappins/lod/external_identifiers.json`, authorities' 024 fields are enhanced with extra metadata such as source, label and id.

	Args:
		auth_dict (list): Authorities list of dictionaries formatted according to [pyreslib.koha.import_koha_authorities_from_marc][] or [pyreslib.koha.import_koha_authorities_from_api][].
		wb: Wikibase Integrator session generated via [pyreslib.wikidata.wikibase_integrator_session_basic][] or [pyreslib.wikidata.wikibase_integrator_session_oauth2][].
		external_ids_mapping_filepath (str): Path to external ids mapping JSON.
		source_subfield(str): Koha subfield where the name of the external source is recorded. Default is "2", but you can change it according to your MARC framework.
		label_subfield(str): Koha subfield where the label of the external source is recorded. Default is "9", but you can change it according to your MARC framework.
		id_subfield(str): Koha subfield where the id of the external source is recorded. Default is "a", but you can change it according to your MARC framework.

	Returns:
		changed_authorities, backup_authorities (list): Lists of dictionaries for the enhanced authorities and their backups. You can locally store, or submit the changes to the catalogue via the API later on.		

	Examples:
		>>> auth_dict = pyreslib.koha.import_koha_authorities_from_api(session=koha_session,base_url=koha_base_url)
		>>> changed_authorities, backup_authorities  = pyreslib.wikidata.external_sources_metadata_authorities(auth_dict=auth_dict)
		>>> changed_authorities
		>>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": {"024": {... "subfields": [{"1": "http://www.wikidata.org/entity/Q254"},{"9": "Wolfgang Amadeus Mozart"},{"a": "Q254"},{"2": "Wikidata"} } ... ]} ...]
		>>> backup_authorities
		>>> >>> [{{"auth_id": 34922,"wd_id": [254], "record": {... "fields": {"024": {... "subfields": [{"1": "http://www.wikidata.org/entity/Q254"}] ...} ]
	"""
	# import external mapping file
	external_ids_mapping = utilities.json2dict(external_ids_mapping_filepath)

	# initialize changed authorities and their backup lists.
	changed_authorities = []
	backup_authorities = []

	for auth in auth_dict:
		modified = False
		# look for URI field
		backup_auth = deepcopy(auth)
		query_uri_field = list(filter(lambda x: uri_field[0] in x.keys(), auth["record"]["fields"] ))
		if len(query_uri_field) > 0:
			for uri in query_uri_field:
				# get current subfields
				try:
					uri_value = list(filter(lambda x: uri_field[1] in x.keys(), uri[uri_field[0]]["subfields"]))[0][uri_field[1]]
					if uri_value.split("/")[-1] != "": # exclude empty URI, like wikidata.org/entity/
						source_value = list(filter(lambda x: source_subfield in x.keys(), uri[uri_field[0]]["subfields"]))
						label_value = list(filter(lambda x: label_subfield in x.keys(), uri[uri_field[0]]["subfields"]))
						id_value = list(filter(lambda x: id_subfield in x.keys(), uri[uri_field[0]]["subfields"]))
						# if incomplete information
						if any(len(subfield) == 0 for subfield in [source_value,label_value,id_value]):
							# query values from URI
							for external_source in external_ids_mapping:
								if external_source["domain_name"] in uri_value:
									# found source
									# add source
									if len(source_value) > 0:
										modified = True
										source_value[0][source_subfield] = external_source["source"]
									else:
										modified = True
										uri[uri_field[0]]["subfields"].append({source_subfield: external_source["source"]})
									# add identifier
									if len(id_value) > 0:
										modified = True
										id_value[0][id_subfield] = uri_value.split("/")[-1]
									else:
										modified = True
										uri[uri_field[0]]["subfields"].append({id_subfield: uri_value.split("/")[-1]})
									# add label (only wikidata)
									if external_source["source"] == "Wikidata":
										try:
											entity = wb.item.get(uri_value.split("/")[-1])
											try:
												modified = True
												label = entity.labels.get('en').value
												if len(label_value) > 0:
													label_value[0][label_subfield] = label								
												else:
													modified = True
													uri[uri_field[0]]["subfields"].append({label_subfield: label})
											except AttributeError:
												pass
										except ValueError:
											pass

									break


				except IndexError:
					continue

		if modified:
			# adding changes
			print(f"Adding changes for {auth["auth_id"]}...")
			backup_authorities.append(backup_auth)
			changed_authorities.append(auth)
			#print(backup_authorities[-1])
			#print(changed_authorities[-1])
			#input()


	print("Saving to JSON...")

	utilities.dict2json(
		backup_authorities,
		os.path.join(
			backup_auth_dir,
			f"backup_auth-{utilities.get_current_date()}.json"
		),
	)

	utilities.dict2json(
		changed_authorities,
		os.path.join(
			changed_auth_dir,
			f"changed_auth-{utilities.get_current_date()}.json"
		)
	)			


	return backup_authorities,changed_authorities

generate_qid_sorted_dict_and_list(auth_dict, exclude_types=[])

Generates a list of dictionaries and a sorted list of IDs from Koha authorities.

Parameters:

Name	Type	Description	Default
auth_dict	list	Authorities list of dictionaries formatted according to pyreslib.koha.import_koha_authorities_from_marc or pyreslib.koha.import_koha_authorities_from_api.	required
exclude_types	list	Authority type codes to be excluded from the process. Default is empty list.	[]

Returns:

Type	Description
list	auth_dict_sorted_qid, auth_qid_list (list): Sorted list of dictionaries for authorities and list of ids for bisect search.

Examples:

>>> auth_dict = pyreslib.koha.import_koha_authorities_from_api(session=koha_session,base_url=koha_base_url)
>>> auth_dict_sorted_qid, auth_qid_list = pyreslib.wikidata.generate_qid_sorted_dict_and_list(auth_dict,exclude_types=["GEOGR_NAME","CHRON_TERM"])
>>> auth_dict_sorted_qid
>>> [{"auth_id": 20935, "wd_id": 6955, "record": {"leader": ... , "fields": [...]}}, ...]
>>> auth_qid_list
>>> [6955,...]

Source code in pyreslib/wikidata.py

def generate_qid_sorted_dict_and_list(auth_dict: list,exclude_types: list = [] ) -> list:
	"""
	Generates a list of dictionaries and a sorted list of IDs from Koha authorities.

	Args:
		auth_dict (list): Authorities list of dictionaries formatted according to [pyreslib.koha.import_koha_authorities_from_marc][] or [pyreslib.koha.import_koha_authorities_from_api][].
		exclude_types (list): Authority type codes to be excluded from the process. Default is empty list.

	Returns:
		auth_dict_sorted_qid, auth_qid_list (list): Sorted list of dictionaries for authorities and list of ids for bisect search.

	Examples:
		>>> auth_dict = pyreslib.koha.import_koha_authorities_from_api(session=koha_session,base_url=koha_base_url)
		>>> auth_dict_sorted_qid, auth_qid_list = pyreslib.wikidata.generate_qid_sorted_dict_and_list(auth_dict,exclude_types=["GEOGR_NAME","CHRON_TERM"])
		>>> auth_dict_sorted_qid
		>>> [{"auth_id": 20935, "wd_id": 6955, "record": {"leader": ... , "fields": [...]}}, ...]
		>>> auth_qid_list
		>>> [6955,...]

	"""
	filtered_auth_dict = []
	print(f"Excluding the following authority types: {", ".join([auth_type for auth_type in exclude_types])}")
	for auth in auth_dict:
		auth_type = koha.get_authority_type(record=auth["record"])
		if auth_type not in exclude_types + [None] :
			filtered_auth_dict.append(auth)


	filtered_dict_duplicate_qids = []
	for auth in filtered_auth_dict:
		if len(auth["wd_id"]) == 1:
			# single case
			filtered_dict_duplicate_qids.append({"auth_id": auth["auth_id"], "wd_id": auth["wd_id"][0], "record": auth["record"]})
		elif len(auth["wd_id"]) > 1:
			# multiple qids for authority, duplicate record in list
			for wd_id in auth["wd_id"]:
				filtered_dict_duplicate_qids.append({"auth_id": auth["auth_id"], "wd_id": wd_id, "record": auth["record"]})
		else:
			filtered_dict_duplicate_qids.append({"auth_id": auth["auth_id"], "wd_id": 0, "record": auth["record"]})

	auth_dict_sorted_qid = sorted(filtered_dict_duplicate_qids, key=lambda x: x["wd_id"])
	auth_qid_list = [auth["wd_id"] for auth in auth_dict_sorted_qid]

	print(f"Number of filtered authorities (with multiple QIDs duplicates: {len(auth_dict_sorted_qid)}")

	return auth_dict_sorted_qid, auth_qid_list

generate_statistics_authorities_wikidata_enhancement(changed_authorities, statistics_filepath=os.path.join('data', 'wikidata', 'statistics', f'statistics-{utilities.get_current_date()}.json'), wikidata_koha_mapping_filepath=os.path.join('data', 'mappings', 'wikidata', 'wikidata-koha-properties.csv'), koha_fields={'PERSO_NAME': '500', 'CORPO_NAME': '510', 'CHRON_TERM': '548', 'TOPIC_TERM': '550', 'GEOGR_NAME': '551'}, authority_types=['PERSO_NAME', 'CORPO_NAME', 'CHRON_TERM', 'TOPIC_TERM', 'GEOGR_NAME'], wd_property_subfield='i')

Generates a JSON file that records statistics for the Wikidata enhanchement, such as number of authorities modified and the type of statements ingested.

Parameters:

Name	Type	Description	Default
changed_authorities	list	Lists of dictionaries for the enhanced authorities.	required
statistics_filepath	str	Path to output JSON file. Default is data/wikidata/statistics.	join('data', 'wikidata', 'statistics', f'statistics-{get_current_date()}.json')
wikidata_koha_mapping_filepath	str	CSV file with Wikidata-Koha mapping of properties and fields. Default is data/mappings/wikidata/wikidata-koha-properties.csv.	join('data', 'mappings', 'wikidata', 'wikidata-koha-properties.csv')
koha_fields	dict	Fields for umbrella terms statements for authority record. Default list is defined according to the 5XX MARC21 framework.	{'PERSO_NAME': '500', 'CORPO_NAME': '510', 'CHRON_TERM': '548', 'TOPIC_TERM': '550', 'GEOGR_NAME': '551'}
authority_types	list	List of authority type codes, according to MARC21 framework.	['PERSO_NAME', 'CORPO_NAME', 'CHRON_TERM', 'TOPIC_TERM', 'GEOGR_NAME']
wd_property_subfield	str	Subfield where the Wikidata property URI is stored. Default is "i", to be set in MARC framework for every authority.	'i'

Returns:

Name	Type	Description
statistics	dict	Dictiorary of the statistics.

Source code in pyreslib/wikidata.py

def generate_statistics_authorities_wikidata_enhancement(
	changed_authorities: list,
	statistics_filepath: str = os.path.join("data","wikidata","statistics",f"statistics-{utilities.get_current_date()}.json"),
	wikidata_koha_mapping_filepath: str = os.path.join("data","mappings","wikidata","wikidata-koha-properties.csv"),
	koha_fields: dict ={"PERSO_NAME": "500","CORPO_NAME": "510","CHRON_TERM": "548","TOPIC_TERM": "550","GEOGR_NAME": "551"},
	authority_types:list =["PERSO_NAME","CORPO_NAME","CHRON_TERM","TOPIC_TERM","GEOGR_NAME"],
	wd_property_subfield: str = "i"):
	"""
	Generates a JSON file that records statistics for the Wikidata enhanchement, such as number of authorities modified and the type of statements ingested.

	Args:
		changed_authorities(list): Lists of dictionaries for the enhanced authorities.
		statistics_filepath (str): Path to output JSON file. Default is `data/wikidata/statistics`.
		wikidata_koha_mapping_filepath (str): CSV file with Wikidata-Koha mapping of properties and fields. Default is `data/mappings/wikidata/wikidata-koha-properties.csv`.
		koha_fields (dict): Fields for umbrella terms statements for authority record. Default list is defined according to the 5XX MARC21 framework.
		authority_types (list): List of authority type codes, according to MARC21 framework.
		wd_property_subfield (str): Subfield where the Wikidata property URI is stored. Default is "i", to be set in MARC framework for every authority.

	Returns:
		statistics (dict): Dictiorary of the statistics.
	"""

	# import mapping from CSV
	wikidata_koha_mapping = utilities.csv2dict(wikidata_koha_mapping_filepath)

	statistics = {"authorities": {"n_changed_authorities": 0, "authority_types": [{auth_type: 0} for auth_type in authority_types] }, "wikidata_properties": [{wd_property["value_i_subfield"]: 0} for wd_property in wikidata_koha_mapping] }

	for auth in changed_authorities:
		statistics["authorities"]["n_changed_authorities"] += 1

		# get authority type
		auth_type = koha.get_authority_type(auth["record"])

		if auth_type is not None:

			# increment occurrence of authority type:
			list(filter(lambda x: auth_type in x.keys(),statistics["authorities"]["authority_types"] ))[0][auth_type] +=1

			# count wikidata property statements
			for field in list(koha_fields.values()):
				field_query = list(filter(lambda x: field in x.keys(),auth["record"]["fields"]))
				if len(field_query) > 0 :
					for statement in field_query:
						i_subfield_query = list(filter(lambda x: wd_property_subfield in x.keys(),statement[field]["subfields"]))
						if len(i_subfield_query) >0:
							wd_property = i_subfield_query[0][wd_property_subfield]
							# found wikidata property statement, append to statistics
							property_query = list(filter(lambda x: wd_property in x.keys(),statistics["wikidata_properties"]))
							if len(property_query) > 0:
								# append occurence
								property_query[0][wd_property] +=1

	# saving statistics
	print(f"Saving statistics to {statistics_filepath}")
	utilities.dict2json(statistics,statistics_filepath)


	return statistics

retrieve_authority_from_qid(qid, auth_qid_list, auth_dict_sorted_qid)

Retrieves Koha authority metadata given a Wikidata QID

Parameters:

Name	Type	Description	Default
qid	str	Wikdiata QID value for entity.	required
auth_qid_list	list	Sorted list of ids for bisect search generated from pyreslib.wikidata.generate_qid_sorted_dict_and_list.	required
auth_dict_sorted_qid	list	Sorted list of dictionaries for authorities for bisect search generated from pyreslib.wikidata.generate_qid_sorted_dict_and_list.	required

Returns:

Name	Type	Description
auth	dict	Authority metadata or None if QID not present in catalogue.

Examples:

>>> auth_dict_sorted_qid, auth_qid_list = pyreslib.wikidata.generate_qid_sorted_dict_and_list(auth_dict,exclude_types=[])
>>> qid = "Q6927"
>>> pyreslib.wikidata.retrieve_authority_from_qid(qid,auth_qid_list,auth_dict_sorted_qid)
>>> {"auth_id": 20936, "wd_id": [6927], "record": {"leader": ... , "fields": [...]}}

Source code in pyreslib/wikidata.py

def retrieve_authority_from_qid(qid: str,auth_qid_list: list ,auth_dict_sorted_qid: list ) -> dict:
	"""
	Retrieves Koha authority metadata given a Wikidata QID

	Args:
		qid(str): Wikdiata QID value for entity.
		auth_qid_list (list): Sorted list of ids for bisect search generated from [pyreslib.wikidata.generate_qid_sorted_dict_and_list][].
		auth_dict_sorted_qid (list): Sorted list of dictionaries for authorities for bisect search generated from [pyreslib.wikidata.generate_qid_sorted_dict_and_list][].

	Returns:
		auth (dict): Authority metadata or None if QID not present in catalogue.

	Examples:
		>>> auth_dict_sorted_qid, auth_qid_list = pyreslib.wikidata.generate_qid_sorted_dict_and_list(auth_dict,exclude_types=[])
		>>> qid = "Q6927"
		>>> pyreslib.wikidata.retrieve_authority_from_qid(qid,auth_qid_list,auth_dict_sorted_qid)
		>>> {"auth_id": 20936, "wd_id": [6927], "record": {"leader": ... , "fields": [...]}}	

	"""
	try:
		qid = int(qid.replace("Q",""))
		index = bisect_left(auth_qid_list,qid)
		if index != len(auth_qid_list) and auth_qid_list[index] == qid:
			return auth_dict_sorted_qid[index]
		else:
			print(f"Wikidata entity {qid} not found in Koha thesaurus")
			return None
	except ValueError:
		return None

update_enhanced_authorities(koha_session, koha_base_url, changed_auth_filepath=os.path.join('data', 'wikidata', 'changed_auth', f'changed_auth-{utilities.get_current_date()}.json'), ckeck_updates=5)

Submit enhanced authorities via Koha API from JSON list after review.

Parameters:

Name	Type	Description	Default
changed_auth_filepath	str	Filepath of the changed_auth-{yyyy-mm-dd}.json file. Default is data/wikidata/changed_auth directory.	join('data', 'wikidata', 'changed_auth', f'changed_auth-{get_current_date()}.json')
koha_session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
koha_base_url	str	Koha API url from credentials.	required
check_updates	int	Number of checks before batch. To be manually done by user in Koha Staff interface.	required

Returns: None

Examples:

Source code in pyreslib/wikidata.py

def update_enhanced_authorities(
	koha_session,
	koha_base_url: str,
	changed_auth_filepath: str = os.path.join("data","wikidata","changed_auth",f"changed_auth-{utilities.get_current_date()}.json"),
	ckeck_updates: int = 5
	):
	"""
	Submit enhanced authorities via Koha API from JSON list after review.

	Args:
		changed_auth_filepath(str): Filepath of the `changed_auth-{yyyy-mm-dd}.json` file. Default is `data/wikidata/changed_auth` directory.
		koha_session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
		koha_base_url (str): Koha API url from credentials.
		check_updates (int): Number of checks before batch. To be manually done by user in Koha Staff interface.
	Returns:
		None

	Examples:

	"""
	# import changed authorities
	changed_auth = utilities.json2dict(changed_auth_filepath)
	# put each authority back to Koha via API
	n_auth = len(changed_auth)
	check = 0
	for auth in changed_auth: 
		print(f"Current authority {((changed_auth.index(auth) +1) / n_auth)*100 } % : {auth["auth_id"]}")
		koha.update_authority_marc(session=koha_session, auth_id= auth["auth_id"], marc_json= auth["record"], base_url= koha_base_url)
		if check <= ckeck_updates:
			check +=1
			input("Check the authority, then press ENTER to continue update.")
		else:
			pass 

	return None

wb_get_property_data(wb_entity, pid, wikibase_URI=False, base_uri='http://wikidata.org/entity/', date_format=f'%Y-%m-%d')

Returns a dictionary with information related to statements, given a WikibaseIntegrator entity tied to a QID and a specific property PID.

Parameters:

Name	Type	Description	Default
wb_entity		ItemEntity object retrieved by using the item.get() WikibaseIntegrator method.	required
pid	str	String indicating the PID of the property.	required
wikibase_URI	bool	Returns full URI of Wikibase item. False by default.	False
base_uri	str	Base Concept URI for the item. By default http://wikidata.org/entity/.	'http://wikidata.org/entity/'
date_format	str	datetime formatting string. Default set to ISO 8601 date.	f'%Y-%m-%d'

Returns:

Name	Type	Description
statements	dict	List of dictionaries with statement values, qualifiers and references.

Examples:

>>> qid = "Q1296"
>>> pid = "P1082"
>>> wb_entity = wb.item.get(qid)
>>> statements = wb_get_property_data(wb_entity,pid)
>>> print(statements)
>>> [{"value": "Q257029","references": [], "qualifiers": [{"pid": "P585","value": "2016-01-01"}]}, {"value": 230951,"references": [{"pid": "P143", "value": ""http://wikidata.org/"Q206855"}], "qualifiers": [{"pid": "P585","value": "2005-01-01"}]}  ]

Source code in pyreslib/wikidata.py

def wb_get_property_data(wb_entity, pid: str, wikibase_URI=False, base_uri="http://wikidata.org/entity/", date_format: str = f"%Y-%m-%d") -> dict:
	"""
	Returns a dictionary with information related to statements, given a `WikibaseIntegrator` entity tied to a QID and a specific property PID.

	Args:
		wb_entity: `ItemEntity` object retrieved by using the `item.get()` WikibaseIntegrator method.
		pid (str): String indicating the PID of the property.
		wikibase_URI (bool): Returns full URI of Wikibase item. False by default.
		base_uri (str): Base Concept URI for the item. By default `http://wikidata.org/entity/`.
		date_format (str): `datetime` formatting string. Default set to ISO 8601 date.


	Returns:
		statements: List of dictionaries with statement values, qualifiers and references.

	Examples:
		>>> qid = "Q1296"
		>>> pid = "P1082"
		>>> wb_entity = wb.item.get(qid)
		>>> statements = wb_get_property_data(wb_entity,pid)
		>>> print(statements)
		>>> [{"value": "Q257029","references": [], "qualifiers": [{"pid": "P585","value": "2016-01-01"}]}, {"value": 230951,"references": [{"pid": "P143", "value": ""http://wikidata.org/"Q206855"}], "qualifiers": [{"pid": "P585","value": "2005-01-01"}]}  ]

	"""
	try:
		statements = []
		prop_values = wb_entity.claims.get(pid)
		for prop_value in prop_values:
			statements.append({"value": "","qualifiers": [], "references": []})
			# add statement value
			value = ""
			if prop_value.mainsnak.datavalue["type"] == "wikibase-entityid":
				if wikibase_URI:
					value = f"{base_uri}{prop_value.mainsnak.datavalue["value"]["id"]}"
				else:
					value = prop_value.mainsnak.datavalue["value"]["id"]

			elif prop_value.mainsnak.datavalue["type"] == "quantity":
				# get rid of + sing. This means that negative quantities are not supported!
				value = prop_value.mainsnak.datavalue["value"]["amount"][1:]
			elif prop_value.mainsnak.datavalue["type"] == "time":
				# convert to ISO 8061 date, meaning that point in time without day or month are set to 1.
				value = convert_point_in_time_to_date(point_in_time=prop_value.datavalue["value"]["time"],date_format=date_format)

			else:
				# strings, such as URLs
				value = prop_value.mainsnak.datavalue["value"]

			# append value
			statements[-1]["value"] = value


			# add references
			for reference in prop_value.references:
				value = ""
				reference_pid = list(reference.snaks.snaks.keys())[0]
				reference_dict = reference.snaks.snaks[reference_pid][0]
				if reference_dict.datavalue["type"] == "wikibase-entityid":
					if wikibase_URI:
						value = f"{base_uri}{reference_dict.datavalue["value"]["id"]}"
					else:
						value = reference_dict.datavalue["value"]["id"]

				elif reference_dict.datavalue["type"] == "quantity":
					# get rid of + sing. This means that negative quantities are not supported!
					value = reference_dict.datavalue["value"]["amount"][1:]
				elif reference_dict.datavalue["type"] == "time":
					# convert to ISO 8061 date, meaning that point in time without day or month are set to 1.
					value = convert_point_in_time_to_date(point_in_time=reference_dict.datavalue["value"]["time"],date_format=date_format)

				else:
					# strings, such as URLs
					value = reference_dict.datavalue["value"]

				# append reference
				statements[-1]["references"].append({"pid": reference_dict.property_number, "value": value})

			# add qualifiers
			for qualifier in prop_value.qualifiers:
				value = ""
				if qualifier.datavalue["type"] == "wikibase-entityid":
					if wikibase_URI:
						value = f"{base_uri}{qualifier.datavalue["value"]["id"]}"
					else:
						value = qualifier.datavalue["value"]["id"]

				elif qualifier.datavalue["type"] == "quantity":
					# get rid of + sing. This means that negative quantities are not supported!
					value = qualifier.datavalue["value"]["amount"][1:]
				elif qualifier.datavalue["type"] == "time":
					# convert to ISO 8061 date, meaning that point in time without day or month are set to 1.
					value = convert_point_in_time_to_date(point_in_time=qualifier.datavalue["value"]["time"],date_format=date_format)

				else:
					# strings, such as URLs
					value = qualifier.datavalue["value"]

				# append qualifier
				statements[-1]["qualifiers"].append({"pid": qualifier.property_number, "value": value})




		return statements
	except Exception:
		return []

wikibase_integrator_session_basic(username, password, mediawiki_api_url='https://www.wikidata.org/w/api.php')

Returns an wikibase_integrator session, given user and password of your Wikidata account. We advice to create a Wikidata user in oder to prevent a "Too Many Requests HTTP 429" code.

Parameters:

Name	Type	Description	Default
username	str	Username associated with your Wikidata user.	required
password	str	Password associated with your Wikidata user.	required
mediawiki_api_url	str	Wikidata REST API by default. You can change it by replacing the https://www.wikidata.org/w/api.php with your own Wikibase URL.	'https://www.wikidata.org/w/api.php'

Returns:

Name	Type	Description
wb	WikibaseIntegrator	WikibaseIntegrator session

Examples:

>>> wb = pyreslib.koha.wikibase_integrator_session_basic(username="{USERNAME}"", password="{PASSWORD}" , mediawiki_api_url="https://www.wikidata.org/w/rest.php/wikibase/v1")

Source code in pyreslib/wikidata.py

def wikibase_integrator_session_basic(
	username: str,
	password: str,
	mediawiki_api_url="https://www.wikidata.org/w/api.php",
):
	"""Returns an wikibase_integrator session, given user and password of your Wikidata account. We advice to create a Wikidata user in oder to prevent a "Too Many Requests HTTP 429" code.

	Args:
		username (str): Username associated with your Wikidata user.
		password (str): Password associated with your Wikidata user.
		mediawiki_api_url (str): Wikidata REST API by default. You can change it by replacing the `https://www.wikidata.org/w/api.php` with your own Wikibase URL.

	Returns:
		wb (WikibaseIntegrator): WikibaseIntegrator session

	Examples:
		>>> wb = pyreslib.koha.wikibase_integrator_session_basic(username="{USERNAME}"", password="{PASSWORD}" , mediawiki_api_url="https://www.wikidata.org/w/rest.php/wikibase/v1")

	"""

	# Set up user-agent type
	wbi_config[
		"USER_AGENT"
	] = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.64 Safari/537.11"

	# login using basic credentials
	login_instance = wbi_login.Clientlogin(
		user=username, password=password, mediawiki_api_url=mediawiki_api_url
	)

	wb = WikibaseIntegrator(login=login_instance)

	return wb

wikibase_integrator_session_oauth2(consumer_token, consumer_secret, mediawiki_api_url='https://www.wikidata.org/w/api.php')

Returns an wikibase_integrator session, given API consumer token and secret keys of your Wikidata account. You have to make a request to Wikimedia via this URL in order to have such credentials. For advanced use only.

Parameters:

Name	Type	Description	Default
consumer_token	str	API consumer key associated with your Wikidata user.	required
consumer_secret	str	API secret key associated with your Wikidata user.	required
mediawiki_api_url	str	Wikidata REST API by default. You can change it by replacing the https://www.wikidata.org/w/api.php with your own Wikibase URL.	'https://www.wikidata.org/w/api.php'

Returns:

Name	Type	Description
wb	WikibaseIntegrator	Wikibase Integrator session

Examples:

>>> wb = pyreslib.koha.wikibase_integrator_session_oauth2(consumer_token="{TOKEN}"", consumer_secret="{SECRET_TOKEN}" , mediawiki_api_url="https://www.wikidata.org/w/api.php")

Source code in pyreslib/wikidata.py

def wikibase_integrator_session_oauth2(
	consumer_token: str,
	consumer_secret: str,
	mediawiki_api_url:str ="https://www.wikidata.org/w/api.php",
):
	"""Returns an wikibase_integrator session, given API consumer token and secret keys of your Wikidata account. You have to make a request to Wikimedia via this [URL](https://meta.wikimedia.org/wiki/Special:OAuthConsumerRegistration/propose/oauth2) in order to have such credentials. For advanced use only.

	Args:
		consumer_token (str): API consumer key associated with your Wikidata user.
		consumer_secret (str): API secret key associated with your Wikidata user.
		mediawiki_api_url (str): Wikidata REST API by default. You can change it by replacing the `https://www.wikidata.org/w/api.php` with your own Wikibase URL.

	Returns:
		wb (WikibaseIntegrator): Wikibase Integrator session

	Examples:
		>>> wb = pyreslib.koha.wikibase_integrator_session_oauth2(consumer_token="{TOKEN}"", consumer_secret="{SECRET_TOKEN}" , mediawiki_api_url="https://www.wikidata.org/w/api.php")

	"""

	# Set up user-agent type
	wbi_config[
		"USER_AGENT"
	] = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.64 Safari/537.11"

	# login using basic credentials
	try:
		login_instance = wbi_login.OAuth2(
			consumer_token=consumer_token,
			consumer_secret=consumer_secret,
			mediawiki_api_url=mediawiki_api_url,
		)
	except KeyError as e:
		print(f"KeyError during OAuth2 init: {e}")
		print(f"Consumer token: {consumer_token[:10]}..." if consumer_token else "None")
		print(f"API URL: {mediawiki_api_url}")
		raise

	wb = WikibaseIntegrator(login=login_instance)

	return wb

pyreslib.google_books

create_marc_field(tag, subfields, ind1, ind2)

Create a MARC field in Koha API format.

Parameters:

Name	Type	Description	Default
tag	str	The MARC field tag	required
ind1	str	First indicator	required
ind2	str	Second indicator	required
subfields	list	List of dictionaries with subfield code and value.	required

Returns:

Type	Description
dict	MARC field dictionary

Source code in pyreslib/google_books.py

def create_marc_field(
    tag: str,
    subfields: list,
    ind1: str,
    ind2: str,
) -> dict:
    """
    Create a MARC field in Koha API format.

    Args:
        tag: The MARC field tag
        ind1: First indicator
        ind2: Second indicator
        subfields: List of dictionaries with subfield code and value.

    Returns:
        MARC field dictionary
    """
    if ind1 in ["", " "]:
        ind1 = " "
    if ind2 in ["", " "]:
        ind2 = " "
    field = {tag: {"ind1": ind1, "ind2": ind2, "subfields": subfields}}

    return field

enhance_biblio_record_from_isbn(biblio_id, koha_session, base_url, google_api_key)

Enhance a Koha biblio record with metadata from Google Books API using the ISBN number.

Parameters:

Name	Type	Description	Default
biblio_id	int	The biblio_id of the book in Koha.	required
koha_session		An authenticated Koha session object.	required
base_url	str	The base URL of the Koha API.	required
google_api_key	str	Your Google API key for accessing the Books API.	required

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise

Source code in pyreslib/google_books.py

def enhance_biblio_record_from_isbn(
    biblio_id: int, koha_session, base_url: str, google_api_key: str
) -> bool:
    """
    Enhance a Koha biblio record with metadata from Google Books API using the ISBN number.

    Args:
        biblio_id (int): The biblio_id of the book in Koha.
        koha_session: An authenticated Koha session object.
        base_url (str): The base URL of the Koha API.
        google_api_key (str): Your Google API key for accessing the Books API.

    Returns:
        bool: True if successful, False otherwise
    """
    try:
        # Get ISBN from biblio record
        isbn = get_isbn_from_biblio_id(
            biblio_id=biblio_id, koha_session=koha_session, base_url=base_url
        )

        if isbn is None:
            print(f"ISBN is missing for biblio ID {biblio_id}")
            return False

        # Get Google Books metadata
        try:
            google_response = get_metadata_from_google_api(
                isbn=isbn, google_api_key=google_api_key
            )

            if "items" not in google_response or len(google_response["items"]) == 0:
                print(f"No results found for ISBN {isbn} (biblio ID {biblio_id})")
                return False

            google_metadata = google_response["items"][0]["volumeInfo"]
        except (KeyError, IndexError) as e:
            print(f"Error parsing Google API response for ISBN {isbn}: {e}")
            return False

        # Load mappings
        mappings = get_google_api_koha_mapping()

        # Prepare fields to update
        fields_to_add = []

        # Process each mapping
        for mapping in mappings:
            json_key = mapping["json_key"]
            field_tag = mapping["field"]
            subfield_code = mapping["subfield"]
            format_type = mapping["format"]
            field_type = mapping["type"]
            ind1 = mapping["ind1"]
            ind2 = mapping["ind2"]

            # Get value from Google metadata
            google_value = google_metadata.get(json_key)

            if google_value is None:
                continue

            # Apply transformations
            if format_type:
                if field_type == "list":
                    # For list types, apply transformation to each element
                    google_value = [
                        transform_value(item, format_type) for item in google_value
                    ]
                    google_value = [v for v in google_value if v]  # Remove None values
                else:
                    google_value = transform_value(google_value, format_type)

            if not google_value:
                continue

            # Prepare regular fields
            if field_type == "list":
                # For lists, create multiple fields
                for item in google_value:
                    fields_to_add.append(
                        create_marc_field(
                            tag=field_tag,
                            subfields=[{subfield_code: item}],
                            ind1=ind1,
                            ind2=ind2,
                        )
                    )
            else:
                # Single value
                fields_to_add.append(
                    create_marc_field(
                        tag=field_tag,
                        subfields=[{subfield_code: google_value}],
                        ind1=ind1,
                        ind2=ind2,
                    )
                )

        # print(f"New fields to be added: {fields_to_add}")

        # Get current biblionumber from API
        biblio = koha.get_biblio_marc(
            session=koha_session, biblio_id=biblio_id, base_url=base_url
        )
        # print(f"Old biblio: {biblio}")

        # append new fields to biblio in order!

        for new_field in fields_to_add:
            new_tag = int(list(new_field.keys())[0])
            # search right position in biblio record
            for old_field in biblio["fields"]:
                old_tag = int(list(old_field.keys())[0])
                if old_tag >= new_tag:
                    # append new field hier
                    biblio["fields"].insert(
                        biblio["fields"].index(old_field) + 1, new_field
                    )
                    break

        # print(f"New biblio: {biblio}")

        # importing new biblio back to Koha via API
        koha.update_biblio_marc(
            session=koha_session,
            biblio_id=biblio_id,
            marc_json=biblio,
            base_url=base_url,
        )

    except ValueError as e:
        print(f"Error enhancing biblio ID {biblio_id}: {e}")
        return False

get_google_api_koha_mapping(google_koha_mapping_filepath=os.path.join('data', 'mappings', 'google', 'google_books-koha_mapping.csv'))

Get the mapping between Google Books API metadata fields and Koha metadata fields from a CSV file.

Parameters:

Name	Type	Description	Default
google_koha_mapping_filepath	str	The file path to the CSV file containing the mapping.	join('data', 'mappings', 'google', 'google_books-koha_mapping.csv')

Returns:

Name	Type	Description
mapping	list	A list of dictionaries of the mappins.

Source code in pyreslib/google_books.py

def get_google_api_koha_mapping(
    google_koha_mapping_filepath=os.path.join(
        "data", "mappings", "google", "google_books-koha_mapping.csv"
    )
) -> list:
    """
    Get the mapping between Google Books API metadata fields and Koha metadata fields from a CSV file.

    Args:
        google_koha_mapping_filepath (str): The file path to the CSV file containing the mapping.

    Returns:
        mapping (list): A list of dictionaries of the mappins.
    """
    mapping = []
    with open(google_koha_mapping_filepath, mode="r", encoding="utf-8") as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            mapping.append(
                {
                    "json_key": row["json_key"],
                    "type": row["type"],
                    "field": row["field"],
                    "subfield": row["subfield"],
                    "format": row["format"],
                    "ind1": row["ind1"],
                    "ind2": row["ind2"],
                }
            )

    return mapping

get_isbn_from_biblio_id(biblio_id, koha_session, base_url)

Get the ISBN number from a Koha biblio_id.

Parameters:

Name	Type	Description	Default
biblio_id	int	The biblio_id of the book in Koha.	required
koha_session		An authenticated Koha session object.	required
base_url	str	The base URL of the Koha API.	required

Returns:

Name	Type	Description
isbn	str	The ISBN number of the book as string.

Source code in pyreslib/google_books.py

def get_isbn_from_biblio_id(biblio_id: int, koha_session, base_url: str) -> str:
    """
    Get the ISBN number from a Koha biblio_id.

    Args:
        biblio_id (int): The biblio_id of the book in Koha.
        koha_session: An authenticated Koha session object.
        base_url (str): The base URL of the Koha API.

    Returns:
        isbn (str): The ISBN number of the book as string.
    """
    biblio_metadata = koha.get_biblio_marc(
        session=koha_session, biblio_id=biblio_id, base_url=base_url
    )
    isbn = None
    for field in biblio_metadata["fields"]:
        if "020" in field.keys():
            isbn = field["020"]["subfields"][0]["a"]
            break
    return isbn

get_metadata_from_google_api(isbn, google_api_key)

Get book metadata from Google Books API. Args: isbn (str): The ISBN number of the book as string. google_api_key (str): Your Google API key for accessing the Books API from credentials.

Returns:

Name	Type	Description
response	dict	JSON dictionary of the Google API metadata

Examples:

>>> get_metadata_from_google_api(isbn="9780674970472",goole_api_key=credentals["google"]["api_key"])
>>> {'kind': 'books#volumes', 'totalItems': 1, 'items': [{'kind': 'books#volume', 'id': 'IXV-EAAAQBAJ', 'etag': 'J8cdkA9cXTg', 'selfLink': 'https://www.googleapis.com/books/v1/volumes/IXV-EAAAQBAJ', 'volumeInfo': {'title': 'In Praise of Failure', 'subtitle': 'Four Lessons in Humility', 'authors': ['Costica Bradatan'], 'publisher': 'Harvard University Press' ...}

Source code in pyreslib/google_books.py

def get_metadata_from_google_api(isbn: str, google_api_key: str) -> dict:
    """
    Get book metadata from Google Books API.
    Args:
        isbn (str): The ISBN number of the book as string.
        google_api_key (str): Your Google API key for accessing the Books API from credentials.

    Returns:
        response (dict): JSON dictionary of the Google API metadata

    Examples:
        >>> get_metadata_from_google_api(isbn="9780674970472",goole_api_key=credentals["google"]["api_key"])
        >>> {'kind': 'books#volumes', 'totalItems': 1, 'items': [{'kind': 'books#volume', 'id': 'IXV-EAAAQBAJ', 'etag': 'J8cdkA9cXTg', 'selfLink': 'https://www.googleapis.com/books/v1/volumes/IXV-EAAAQBAJ', 'volumeInfo': {'title': 'In Praise of Failure', 'subtitle': 'Four Lessons in Humility', 'authors': ['Costica Bradatan'], 'publisher': 'Harvard University Press' ...}

    """
    GB_API_URL = "https://www.googleapis.com/books/v1/volumes"
    session = requests.Session()
    session.headers.update(
        {"User-Agent": "google-books-metadata/1.0 (+https://example.org)"}
    )
    params = {"q": f"isbn:{isbn}", "key": google_api_key}
    response = session.get(GB_API_URL, params=params, timeout=10)
    """# Debugging
    print("REQUEST URL:", response.request.url)
    print("REQUEST HEADERS:", response.request.headers)
    print("REQUEST BODY (bytes):", response.request.body)
    print("STATUS:", response.status_code)
    print("RESPONSE HEADERS:", response.headers)
    print("RESPONSE TEXT:", (response.text or ""))
    """

    return response.json()

transform_value(value, format_type)

Apply format transformations to values based on the format field.

Parameters:

Name	Type	Description	Default
value		The value to transform	required
format_type	str	The format type from the mapping (e.g., 'ymd_date2year', 'name_surname2surname,_name')	required

Returns:

Type	Description
	Transformed value

Source code in pyreslib/google_books.py

def transform_value(value, format_type: str):
    """
    Apply format transformations to values based on the format field.

    Args:
        value: The value to transform
        format_type: The format type from the mapping (e.g., 'ymd_date2year', 'name_surname2surname,_name')

    Returns:
        Transformed value
    """
    if value is None or value == "":
        return None

    if format_type == "ymd_date2year":
        # Extract year from YYYY-MM-DD format
        try:
            return str(value)[:4]
        except Exception:
            return None

    elif format_type == "page_number2page_number_pages":
        # Convert page count to MARC format (e.g., "289" -> "289 pages")
        try:
            return f"{value} pages"
        except Exception:
            return None

    elif format_type == "name_surname2surname,_name":
        # Transform "John Doe" to "Doe, John"
        try:
            parts = str(value).strip().split()
            if len(parts) == 1:
                return parts[0]
            else:
                surname = parts[-1]
                name = " ".join(parts[:-1])
                return f"{surname}, {name}"
        except Exception:
            return None

    return value

pyreslib.transkribus

api_login(user, password)

Parameters:

Name	Type	Description	Default
user	user	Client Username	required
password	str	Client secret Password	required

Returns: session: requests.Session() for Transkribus API

Source code in pyreslib/transkribus.py

def api_login(user: str, password: str):
    """
    Args:
                                                                    user (user): Client Username
                                                                    password (str): Client secret Password
    Returns:
                                                                    session: requests.Session() for Transkribus API
    """
    session = requests.Session()
    response = session.post(
        "https://transkribus.eu/TrpServer/rest/auth/login",
        data={"user": user, "pw": password},
    )
    if response.status_code == requests.codes.ok:
        return session
    else:
        print(r)
        print("Login failed.")
        return None

classify_region_two_columns(region, page_center, error_margin)

Classify a region based o page_center x-coordinates, returning left, center or right

Source code in pyreslib/transkribus.py

def classify_region_two_columns(
    region: ET.Element, page_center: float, error_margin: float
) -> str:
    """
    Classify a region based o page_center x-coordinates, returning left, center or right
    """
    # Extract centroids for region
    coordinates = extract_region_polygonal_coordinates(region)
    centroids = get_polygonal_centroids(coordinates)
    x_centroid = centroids[0]
    y_centroid = centroids[1]
    region_length = max([p[0] for p in coordinates]) - min([p[0] for p in coordinates])

    x_left = x_centroid - region_length / 2
    x_right = x_centroid + region_length / 2

    # Thresholds for the central 'neutral' column
    left_threshold = page_center - error_margin
    right_threshold = page_center + error_margin

    if x_left < (left_threshold):
        if x_right < (right_threshold):
            return "left"
        else:
            return "center"
    else:
        return "right"

classify_regions_two_colums(page_xml_path, page_center_method='reference_region', reference_type='page-number', PAGEXML_NS='http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15')

Classifies regions according to two-column layout. You can use a reference region to define the center. It returns a list of dictionaries, with XML element and classification tag.

Parameters:

Name	Type	Description	Default
page_xml_path	str	Path to PAGEXML file.	required
page_center_method	str	Reference method in order to determine page center. Options are "image_width" (get half of the whole image length) and "reference_region" (a specific region is used as center).	'reference_region'
reference_type	str	Region tag used for "reference_region" as page_center_method parameter. Default is page-number.	'page-number'
PAGEXML_NS	str	Namespace URI for PAGEXML schema. Default is "http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15".	'http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15'

Returns: ordered_regions(list): List of blocks

Source code in pyreslib/transkribus.py

def classify_regions_two_colums(
    page_xml_path: str | Path,
    page_center_method: str = "reference_region",
    reference_type: str = "page-number",
    PAGEXML_NS: str = "http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15",
) -> list:
    """
    Classifies regions according to two-column layout. You can use a reference region to define the center.
    It returns a list of dictionaries, with XML element and classification tag.

    Args:
                    page_xml_path(str): Path to PAGEXML file.
                    page_center_method (str): Reference method in order to determine page center. Options are "image_width" (get half of the whole image length) and "reference_region" (a specific region is used as center).
                    reference_type (str): Region tag used for "reference_region" as `page_center_method` parameter. Default is `page-number`.
                    PAGEXML_NS(str): Namespace URI for PAGEXML schema. Default is "http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15".
    Returns:
                    ordered_regions(list): List of blocks

    """
    # import PAGEXML
    with open(page_xml_path, "rb") as f:
        page_xml = f.read()
        f.close()

    root = ET.fromstring(page_xml)
    ET.register_namespace("", PAGEXML_NS)

    # initialize classified_regions list
    region_blocks = []

    # get TextRegions
    regions = root.findall(f".//{{{PAGEXML_NS}}}TextRegion")
    # sort regions according to two-colum layout and reference region

    # Get image width
    image_width = root.find(f".//{{{PAGEXML_NS}}}Page").attrib["imageWidth"]

    # Calculate page center
    if page_center_method == "image_width":
        page_center = float(image_width) / 2
    elif page_center_method == "reference_region":
        # get region with reference_type tag:
        found = False
        for region in regions:
            try:
                if (
                    region.attrib["custom"].split("type:")[1].split(";")[0]
                    == reference_type
                ):
                    page_center = get_polygonal_centroids(
                        extract_region_polygonal_coordinates(region)
                    )[0]
                    found = True
                    break
            except IndexError:
                # exclude regions without structural tag
                pass
        if found is False:
            page_center = float(image_width) / 2

    print(f"Image width: {image_width}")
    error_margin = float(image_width) / 50
    print(f"Error margin 2%: {error_margin}")
    print(f"Page center: {page_center}")

    # sort regions in blocks based on their y_centroid coordinate. Everytime a centered region comes, we generate a new block.
    # left regions are stored in order in left column, while right regions in right one.

    for region in regions:
        # classify region
        try:
            region_type = region.attrib["custom"].split("type:")[1].split(";")[0]
            if (
                region.attrib["custom"].split("type:")[1].split(";")[0]
                == reference_type
            ):
                region_class = "center"
            else:
                region_class = classify_region_two_columns(
                    region, page_center=page_center, error_margin=error_margin
                )
        except IndexError:
            continue

        region_y_centroid = get_polygonal_centroids(
            extract_region_polygonal_coordinates(region)
        )[1]

        region_data = {
            "id": region.attrib["id"],
            "class": region_class,
            "y_centroid": region_y_centroid,
            "xml_element": region,
        }

        # 2. Block Generation Logic (The Fix)
        if region_class == "center":
            # Always initiate a new isolated block for centered elements
            region_blocks.append([region_data])

        else:
            # For 'left' or 'right' regions:
            # Start a NEW block if:
            # - No blocks exist yet
            # - OR the previous block contains a 'center' region
            if not region_blocks or region_blocks[-1][-1]["class"] == "center":
                region_blocks.append([region_data])
            else:
                # Otherwise, it's a left/right region following another left/right;
                # append to the current 'column' group
                region_blocks[-1].append(region_data)

    # Order each block based on y_centroid and left/right priority
    ordered_regions = []
    for i, block in enumerate(region_blocks, 1):
        # Separate regions in the current block by class
        center_regs = [r for r in block if r["class"] == "center"]
        left_regs = [r for r in block if r["class"] == "left"]
        right_regs = [r for r in block if r["class"] == "right"]

        # Sort the left and right columns vertically (y_centroid)
        left_regs_sorted = sorted(left_regs, key=lambda x: x["y_centroid"])
        right_regs_sorted = sorted(right_regs, key=lambda x: x["y_centroid"])

        # Construction of the final sequence for this block
        # 1. The 'divider' (Center) usually comes first (e.g., a header)
        for r in center_regs:
            r["block"] = i
            ordered_regions.append(r)

        # 2. Entire left column content
        for r in left_regs_sorted:
            r["block"] = i
            ordered_regions.append(r)

        # 3. Entire right column content
        for r in right_regs_sorted:
            r["block"] = i
            ordered_regions.append(r)

    return ordered_regions

extract_reading_order_index(text_region)

Extract the readingOrder index from the custom attribute.

Returns:

Type	Description
int	Index value, or float('inf') if not found

Source code in pyreslib/transkribus.py

def extract_reading_order_index(text_region: ET.Element) -> int:
    """
    Extract the readingOrder index from the custom attribute.

    Returns:
                                                                    Index value, or float('inf') if not found
    """
    custom = text_region.get("custom", "")

    # Parse readingOrder {index:N;} pattern
    if "readingOrder {index:" in custom:
        try:
            start = custom.index("readingOrder {index:") + len("readingOrder {index:")
            end = custom.index(";", start)
            return int(custom[start:end])
        except (ValueError, IndexError):
            return float("inf")

    return float("inf")

extract_region_polygonal_coordinates(region)

Returns a list of coordinates given the region id.

Parameters:

Name	Type	Description	Default
region	Element	ET XML element of the region.	required

Returns:

Name	Type	Description
coordinates	list	List of (x,y) tuples of coordinates.

Source code in pyreslib/transkribus.py

def extract_region_polygonal_coordinates(
    region: ET.Element,
) -> list:
    """
    Returns a list of coordinates given the region id.

    Args:
                                                                    region (ET.Element): ET XML element of the region.

    Returns:
                                                                    coordinates (list): List of (x,y) tuples of coordinates.

    """
    # initialize coordinates
    coordinates = []

    coordinates_element = region.find(
        "./{http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15}Coords"
    )
    points_string = coordinates_element.attrib["points"]
    points_list = points_string.split(" ")
    for point in points_list:
        x = int(point.split(",")[0])
        y = int(point.split(",")[1])
        coordinates.append((x, y))

    return coordinates

fit_text_line_in_region(text_line, root, TRANSKRIBUS_NS_URL='http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15')

Fit text_line into best region, according to area intersection

Parameters:

Name	Type	Description	Default
text_line	Element	XML element of text line to be inserted.	required
root	Element	XML root of the page document to be enriched.	required

Returns:

Type	Description
Element	root

Source code in pyreslib/transkribus.py

def fit_text_line_in_region(
    text_line: ET.Element,
    root: ET.Element,
    TRANSKRIBUS_NS_URL: str = "http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15",
) -> ET.Element:
    """
    Fit text_line into best region, according to area intersection

    Args:
                                                                    text_line (ET.Element): XML element of text line to be inserted.
                                                                    root (ET.Element): XML root of the page document to be enriched.

    Returns:
                                                                    root

    """
    # get all regions in root
    ET.register_namespace("", TRANSKRIBUS_NS_URL)
    regions = root.findall(
        f"./{{{TRANSKRIBUS_NS_URL}}}Page/{{{TRANSKRIBUS_NS_URL}}}TextRegion"
    )

    # select best region according to area interception
    text_line_coords = extract_region_polygonal_coordinates(text_line)
    best_region = None
    for region in regions:
        # print(f"Current region {region.attrib["id"]}")
        region_coords = extract_region_polygonal_coordinates(region)
        int_area = intersection_area(text_line_coords, region_coords)
        # print(f"Intersection area: {int_area}")

        if best_region is None:
            best_region = region
        else:
            if int_area > intersection_area(
                text_line_coords, extract_region_polygonal_coordinates(best_region)
            ):
                best_region = region

    # append line_text to  best region
    if (
        intersection_area(
            text_line_coords, extract_region_polygonal_coordinates(best_region)
        )
        > 0
    ):
        best_region.append(text_line)

    return root

get_document_metadata(session, collection_id, document_id)

Returns a document metadata from Transkribus API

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login() method.	required
collection_id	int	Collection ID identifier from Transkribus.	required
document_id	int	Document ID identifier from Transkribus.	required

Returns:

Name	Type	Description
document	dict	Document metadata

Examples:

                                                            >>> session = pyreslib.transkribus.api_login(user,password)
                                                            >>> document = pyreslib.transkribus.get_document_metadata(session,collection_id=2792,document_id=16606356)
                                                            >>> {"md": {...}, "pageList": {"pages":[...]}, ... }

Source code in pyreslib/transkribus.py

def get_document_metadata(session, collection_id: int, document_id: int) -> dict:
    """
    Returns a document metadata from Transkribus API

    Args:
                                                                    session: Transkribus session from `pyreslib.transkribus.api_login()` method.
                                                                    collection_id (int): Collection ID identifier from Transkribus.
                                                                    document_id (int): Document ID identifier from Transkribus.


    Returns:
                                                                    document (dict): Document metadata

    Examples:

                                                                    >>> session = pyreslib.transkribus.api_login(user,password)
                                                                    >>> document = pyreslib.transkribus.get_document_metadata(session,collection_id=2792,document_id=16606356)
                                                                    >>> {"md": {...}, "pageList": {"pages":[...]}, ... }
    """

    # Get document metadata
    headers = {"Accept": "application/json"}
    document = session.get(
        f"https://transkribus.eu/TrpServer/rest/collections/{collection_id}/{document_id}/fulldoc",
        headers=headers,
    ).json()

    return document

get_documents_metadata(session, collection_id)

Returns a full dictionary of documents, including pages metadata.

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login() method.	required
collection_id	int	Collection ID identifier from Transkribus.	required

Returns:

Name	Type	Description
documents	list	List of documents, with metadata and pagelist.

Examples:

                                                            >>> session = pyreslib.transkribus.api_login(user,password)
                                                            >>> documents = pyreslib.transkribus.get_documents_metadata(session,collection_id=2792)
                                                            >>> [{"md": {...}, "pageList": {"pages":[...]},"collection": {...}, ... }]

Source code in pyreslib/transkribus.py

def get_documents_metadata(session, collection_id: int) -> list:
    """
    Returns a full dictionary of documents, including pages metadata.

    Args:
                                                                    session: Transkribus session from `pyreslib.transkribus.api_login()` method.
                                                                    collection_id (int): Collection ID identifier from Transkribus.

    Returns:
                                                                    documents (list): List of documents, with metadata and pagelist.

    Examples:

                                                                    >>> session = pyreslib.transkribus.api_login(user,password)
                                                                    >>> documents = pyreslib.transkribus.get_documents_metadata(session,collection_id=2792)
                                                                    >>> [{"md": {...}, "pageList": {"pages":[...]},"collection": {...}, ... }]
    """
    headers = {"Accept": "application/json"}
    collection = session.get(
        f"https://transkribus.eu/TrpServer/rest/collections/{collection_id}/list",
        headers=headers,
    ).json()

    documents_metadata = []

    for document in collection:
        document = session.get(
            f"https://transkribus.eu/TrpServer/rest/collections/{collection_id}/{document['docId']}/fulldoc",
            headers=headers,
        )

        documents_metadata.append(document.json())

    return documents_metadata

get_jpg_image(session, collection_id, document_id, page_number, output_dir)

Extract the JPG image of a given page from Transkribus API

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login() method.	required
collection_id	int	Collection ID identifier from Transkribus.	required
document_id	int	Document ID identifier from Transkribus.	required
page_number	int	Internal page number identifier from Transkribus.	required
output_dir	str	local directory for download. The filename is equal to the one stored in Transkribus metadata.	required

Returns:

Type	Description
	None

Examples:

>>> session = pyreslib.transkribus.api_login(user,password)
>>> output_filepath = f"./data/kraken/transcriptions/{str(document_id)}"
>>> get_jpg_image(session,collection_id=2792,document_id=145869,page_number=14,output_dir=output_dir)

Source code in pyreslib/transkribus.py

def get_jpg_image(
    session, collection_id: int, document_id: int, page_number: int, output_dir: str
):
    """
    Extract the JPG image of a given page from Transkribus API

    Args:
            session (): Transkribus session from `pyreslib.transkribus.api_login()` method.
            collection_id (int): Collection ID identifier from Transkribus.
            document_id (int): Document ID identifier from Transkribus.
            page_number (int): Internal page number identifier from Transkribus.
            output_dir (str): local directory for download. The filename is equal to the one stored in Transkribus metadata.

    Returns:
            `None`

    Examples:
            >>> session = pyreslib.transkribus.api_login(user,password)
            >>> output_filepath = f"./data/kraken/transcriptions/{str(document_id)}"
            >>> get_jpg_image(session,collection_id=2792,document_id=145869,page_number=14,output_dir=output_dir)
    """

    # Get document metadata
    document = get_document_metadata(
        session=session, collection_id=collection_id, document_id=document_id
    )

    # Find page metadata
    page_metadata = list(
        filter(lambda x: x["pageNr"] == page_number, document["pageList"]["pages"])
    )
    if len(page_metadata) > 0:
        output_filepath = os.path.join(output_dir, page_metadata[0]["imgFileName"])
        # page found, get image URL
        image_url = page_metadata[0]["url"]
        urllib.request.urlretrieve(image_url, output_filepath)
    else:
        print(f"Page not found for {document_id}/{page_number}")

    return None

get_page_status(session, collection_id, document_id, page_number)

Retrieves the PAGEXML transcription status of a given page of a document as string.

Source code in pyreslib/transkribus.py

def get_page_status(
    session, collection_id: int, document_id: int, page_number: int
) -> str:
    """
    Retrieves the PAGEXML transcription status of a given page of a document as string.
    """
    # get document metadata
    document_metadata = get_document_metadata(
        session=session, collection_id=collection_id, document_id=document_id
    )

    # get page metadata
    page_metadata = list(
        filter(
            lambda x: x["pageNr"] == page_number, document_metadata["pageList"]["pages"]
        )
    )[0]

    # get latest transcript status
    return page_metadata["tsList"]["transcripts"][0]["status"]

get_page_txt(session, collection_id, document_id, page_number)

Retrieves the TXT transcription of a given page of a document as string.

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login() method.	required
collection_id	int	Collection ID identifier from Transkribus.	required
collection_id	int	Document ID identifier from Transkribus.	required
page_number	int	Internal page number identifier from Transkribus.	required

Returns:

Name	Type	Description
plain_text	str	string of the plaintext transcription of the page by Transkribus.

Examples:

>>> session = pyreslib.transkribus.api_login(user,password)
>>> plain_text = pyreslib.transkribus.get_page_txt(session,collection_id=2792,document_id=145869,page_number=14)

Source code in pyreslib/transkribus.py

def get_page_txt(
    session, collection_id: int, document_id: int, page_number: int
) -> str:
    """
    Retrieves the TXT transcription of a given page of a document as string.

    Args:
                                                                    session: Transkribus session from `pyreslib.transkribus.api_login()` method.
                                                                    collection_id (int): Collection ID identifier from Transkribus.
                                                                    collection_id (int): Document ID identifier from Transkribus.
                                                                    page_number (int): Internal page number identifier from Transkribus.

    Returns:
                                                                    plain_text (str): string of the plaintext transcription of the page by Transkribus.

    Examples:
                                                                    >>> session = pyreslib.transkribus.api_login(user,password)
                                                                    >>> plain_text = pyreslib.transkribus.get_page_txt(session,collection_id=2792,document_id=145869,page_number=14)
    """

    # get PAGEXML
    page_xml = get_page_xml(
        session=session,
        collection_id=collection_id,
        document_id=document_id,
        page_number=page_number,
    )
    # parse PAGEXML and extract plain text
    root = ET.fromstring(page_xml)
    text_regions = root.findall(
        "./{http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15}Page/{http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15}TextRegion/{http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15}TextLine"
    )
    plain_text = ""
    for region in text_regions:
        text_elements = region.findall(
            "./{http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15}TextEquiv/{http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15}Unicode"
        )

        for text_elem in text_elements:
            try:
                plain_text += text_elem.text + "\n"
            except TypeError:
                pass

    return plain_text

get_page_xml(session, collection_id, document_id, page_number)

Retrieves the PAGEXML transcription of a given page of a document as string.

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login() method.	required
collection_id	int	Collection ID identifier from Transkribus.	required
document_id	int	Document ID identifier from Transkribus.	required
page_number	int	Internal page number identifier from Transkribus.	required

Returns:

Name	Type	Description
page_xml	str	string serialization of the PAGEXML transcription of the page by Transkribus. You can parse it later by usin xml.etree.ElementTree.fromstring() method.

Examples:

>>> import xml.etree.ElementTree as ET
>>> session = pyreslib.transkribus.api_login(user,password)
>>> page_xml = pyreslib.transkribus.get_page_xml(session,collection_id=2792,document_id=145869,page_number=14)
>>> root = ET.fromstring(page_xml)

Source code in pyreslib/transkribus.py

def get_page_xml(
    session, collection_id: int, document_id: int, page_number: int
) -> str:
    """
    Retrieves the PAGEXML transcription of a given page of a document as string.

    Args:
                                                                    session: Transkribus session from `pyreslib.transkribus.api_login()` method.
                                                                    collection_id (int): Collection ID identifier from Transkribus.
                                                                    document_id (int): Document ID identifier from Transkribus.
                                                                    page_number (int): Internal page number identifier from Transkribus.

    Returns:
                                                                    page_xml (str): string serialization of the PAGEXML transcription of the page by Transkribus. You can parse it later by usin `xml.etree.ElementTree.fromstring()` method.

    Examples:
                                                                    >>> import xml.etree.ElementTree as ET
                                                                    >>> session = pyreslib.transkribus.api_login(user,password)
                                                                    >>> page_xml = pyreslib.transkribus.get_page_xml(session,collection_id=2792,document_id=145869,page_number=14)
                                                                    >>> root = ET.fromstring(page_xml)
    """

    headers = {"Accept": "application/xml"}
    page_xml = session.get(
        f"https://transkribus.eu/TrpServer/rest/collections/{collection_id}/{document_id}/{page_number}/text",
        headers=headers,
    )

    if page_xml.status_code in [200, 201]:
        pass
    else:
        print(f"Failed with Status Code: {page_xml.status_code}")
        print(f"Server Response: {page_xml.text}")

    return str(page_xml.text).encode("utf-8")

get_polygonal_centroids(coordinate_list)

Returns the centroids of a lsit of coordinates.

Parameters:

Name	Type	Description	Default
coordinate_list	list	List of coordinates [(x_1,y_1), ... (x_n,x_n)].	required

Returns:

Name	Type	Description
centroids		Tuple of x and y centroids.

Source code in pyreslib/transkribus.py

def get_polygonal_centroids(
    coordinate_list: list,
):
    """
    Returns the centroids of a lsit of coordinates.

    Args:
                                                                    coordinate_list (list): List of coordinates [(x_1,y_1), ... (x_n,x_n)].

    Returns:
                                                                    centroids: Tuple of x and y centroids.


    """
    n = len(coordinate_list)
    if n > 0:
        sum_x = float(0)
        sum_y = float(0)
        for point in coordinate_list:
            sum_x += point[0]
            sum_y += point[1]

        return (sum_x / n, sum_y / n)
    else:
        return (0, 0)

import_jpg_from_document(session, collection_id, document_id, output_dir)

Downloads JPG images from transkribus, giving them the same basename. as the page filename. Args: session(requests.Session): Transkribus session from pyreslib.transkribus.api_login() method. collection_id (int): Collection ID identifier from Transkribus. document_id (int): Document ID identifier from Transkribus. output_dir(str | Path): Directory path for JPG files.

Source code in pyreslib/transkribus.py

def import_jpg_from_document(
    session, collection_id: int, document_id: int, output_dir: str | Path
):
    """
    Downloads JPG images from transkribus, giving them the same basename. as the page filename.
    Args:
            session(requests.Session): Transkribus session from `pyreslib.transkribus.api_login()` method.
            collection_id (int): Collection ID identifier from Transkribus.
            document_id (int): Document ID identifier from Transkribus.
            output_dir(str | Path): Directory path for JPG files.

    """

    # initialize path
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    # get document metadata
    document_metadata = get_document_metadata(
        session=session, collection_id=collection_id, document_id=document_id
    )
    pages = document_metadata.get("pageList", {}).get("pages", [])
    print(
        f"Found {len(pages)} pages to download for document {document_id}. Press Enter to continue."
    )
    input()
    # loop each page and extract pagexml
    for page in pages:
        # Extract relevant metadata from page element
        page_number = page["pageNr"]
        base_name = Path(page["imgFileName"]).stem
        print(f"Processing page {page_number}")
        get_jpg_image(
            session=session,
            collection_id=collection_id,
            document_id=document_id,
            page_number=page_number,
            output_dir=output_dir,
        )

        print(f"JPG saved as {output_dir}/{base_name}.jpg")

import_pagexml_from_document(session, collection_id, document_id, output_dir)

Downloads PAGEXML transcriptions from transkribus, giving them the same basename. as the page filename. Args: session(requests.Session): Transkribus session from pyreslib.transkribus.api_login() method. collection_id (int): Collection ID identifier from Transkribus. document_id (int): Document ID identifier from Transkribus. output_dir(str | Path): Directory path for PAGEXML files.

Source code in pyreslib/transkribus.py

def import_pagexml_from_document(
    session: requests.Session,
    collection_id: int,
    document_id: int,
    output_dir: str | Path,
):
    """
    Downloads PAGEXML transcriptions from transkribus, giving them the same basename. as the page filename.
    Args:
            session(requests.Session): Transkribus session from `pyreslib.transkribus.api_login()` method.
            collection_id (int): Collection ID identifier from Transkribus.
            document_id (int): Document ID identifier from Transkribus.
            output_dir(str | Path): Directory path for PAGEXML files.

    """

    # initialize path
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    # get document metadata
    document_metadata = get_document_metadata(
        session=session, collection_id=collection_id, document_id=document_id
    )
    pages = document_metadata.get("pageList", {}).get("pages", [])
    print(
        f"Found {len(pages)} pages to download for document {document_id}. Press Enter to continue."
    )
    input()
    # loop each page and extract pagexml
    for page in pages:
        # Extract relevant metadata from page element
        page_number = page["pageNr"]
        base_name = Path(page["imgFileName"]).stem
        print(f"Processing page {page_number}")
        # get pagexml as string from Transkribus API
        page_xml = get_page_xml(
            session=session,
            collection_id=collection_id,
            document_id=document_id,
            page_number=page_number,
        )
        file_path = output_path / f"{base_name}.xml"
        # save pagexml to output directory
        with open(file_path, "wb") as f:
            f.write(page_xml)

        print(f"PAGEXML saved as {file_path}")

import_text_transcription_to_koha_field(text, biblio_id, koha_session, koha_base_url, field=['520', 'a'])

Creates a new field statement for a bibliographic record with the given Transkribus transcription.

Parameters:

Name	Type	Description	Default
text	str	String to be ingested in Koha.	required
biblio_id	int	Biblio ID for the record.	required
koha_session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
koha_base_url	str	Koha API url from credentials.	required
field	list	Field and subfield where the text has to be added to record. Default is Summary/Abstract (520$a) field = ["520","a"]	['520', 'a']

Returns:

Type	Description
	None

Source code in pyreslib/transkribus.py

def import_text_transcription_to_koha_field(
    text: str,
    biblio_id: int,
    koha_session,
    koha_base_url: str,
    field: list = ["520", "a"],
):
    """
    Creates a new field statement for a bibliographic record with the given Transkribus transcription.

    Args:
                                                                    text (str): String to be ingested in Koha.
                                                                    biblio_id (int): Biblio ID for the record.
                                                                    koha_session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
                                                                    koha_base_url (str): Koha API url from credentials.
                                                                    field (list): Field and subfield where the text has to be added to record. Default is Summary/Abstract (520$a) field = ["520","a"]

    Returns:
                                                                    None
    """

    # Import record via API
    print(f"Importing record {biblio_id} from Koha API...")
    record = koha.get_biblio_marc(
        biblio_id=biblio_id, session=koha_session, base_url=koha_base_url
    )
    print(f"Number of fields: {len(record['fields'])}")
    # Add new statement for given field
    query_field = list(filter(lambda x: field[0] in x.keys(), record["fields"]))
    if len(query_field) > 0:  # append new statement
        pos = record["fields"].index(query_field[0])
        new_statement = {
            field[0]: {"ind2": " ", "ind1": " ", "subfields": [{field[1]: text}]}
        }
        # append the field statement after the first found with the same tag.
        record["fields"].insert(pos, new_statement)
    else:
        # append the field according to order
        field_n = int(field[0])
        for statement in record["fields"]:
            stat_n = int(list(statement.keys())[0])
            if field_n < stat_n:
                pos = record["fields"].index(statement)
                new_statement = {
                    field[0]: {
                        "ind2": " ",
                        "ind1": " ",
                        "subfields": [{field[1]: text}],
                    }
                }

                record["fields"].insert(pos - 1, new_statement)
                break

    # update record back to Koha catalogue
    print(f"Updating transcription to biblionumber {biblio_id}")
    koha.update_biblio_marc(
        session=koha_session,
        biblio_id=biblio_id,
        marc_json=record,
        base_url=koha_base_url,
    )

import_txt_from_document(session, collection_id, document_id, output_dir)

Downloads TXT transcriptions from transkribus, giving them the same basename. as the page filename. Args: session(requests.Session): Transkribus session from pyreslib.transkribus.api_login() method. collection_id (int): Collection ID identifier from Transkribus. document_id (int): Document ID identifier from Transkribus. output_dir(str | Path): Directory path for TXT files.

Source code in pyreslib/transkribus.py

def import_txt_from_document(
    session, collection_id: int, document_id: int, output_dir: str | Path
):
    """
    Downloads TXT transcriptions from transkribus, giving them the same basename. as the page filename.
    Args:
            session(requests.Session): Transkribus session from `pyreslib.transkribus.api_login()` method.
            collection_id (int): Collection ID identifier from Transkribus.
            document_id (int): Document ID identifier from Transkribus.
            output_dir(str | Path): Directory path for TXT files.

    """

    # initialize path
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    # get document metadata
    document_metadata = get_document_metadata(
        session=session, collection_id=collection_id, document_id=document_id
    )
    pages = document_metadata.get("pageList", {}).get("pages", [])
    print(
        f"Found {len(pages)} pages to download for document {document_id}. Press Enter to continue."
    )
    input()
    # loop each page and extract pagexml
    for page in pages:
        # Extract relevant metadata from page element
        page_number = page["pageNr"]
        base_name = Path(page["imgFileName"]).stem
        print(f"Processing page {page_number}")
        txt = get_page_txt(
            session=session,
            collection_id=collection_id,
            document_id=document_id,
            page_number=page_number,
        )

        # saving txt string to file
        with open(output_path / f"{base_name}.txt", "w") as f:
            f.write(txt)

        print(f"TXT saved as {output_dir}/{base_name}.txt")

intersection_area(coords1, coords2)

Calcuate the intesection area between two 2D-coordinate lists.

Parameters:

Name	Type	Description	Default
coords1	list	First list of float coordinates.	required
coords2	list	Second list of float coordinates.	required

Returns:

Name	Type	Description
intersection_area	float	Size of intersection between two polygonal shapes.

Source code in pyreslib/transkribus.py

def intersection_area(coords1: list, coords2: list) -> float:
    """
    Calcuate the intesection area between two 2D-coordinate lists.

    Args:
                                                                    coords1 (list): First list of float coordinates.
                                                                    coords2 (list): Second list of float coordinates.

    Returns:
                                                                    intersection_area (float): Size of intersection between two polygonal shapes.

    """
    # import coordinates as polygons using Shapely
    polygon1 = Polygon(coords1)
    polygon2 = Polygon(coords2)

    intersection_area = polygon1.intersection(polygon2).area

    return intersection_area

post_page_xml(session, page_xml, collection_id, document_id, page_number, filepath=True)

Post a PAGEXML file back to Transkribus, either from string or local file. Args: session: Transkribus session from pyreslib.transkribus.api_login() method. page_xml (str): Filepath or XML string. Set filepath=True if you wish to interpret it as a path. collection_id (int): Collection ID identifier from Transkribus. document_id (int): Document ID identifier from Transkribus. page_number (int): Internal page number identifier from Transkribus. filepath (bool): True if page_xml is a filepath, False for page_xml as string.

Returns: None

Examples:

session = pyreslib.transkribus.api_login(user,password)

Source code in pyreslib/transkribus.py

def post_page_xml(
    session,
    page_xml: str,
    collection_id: int,
    document_id: int,
    page_number: int,
    filepath=True,
):
    """
    Post a PAGEXML file back to Transkribus, either from string or local file.
    Args:
    session: Transkribus session from `pyreslib.transkribus.api_login()` method.
    page_xml (str): Filepath or XML string. Set filepath=True if you wish to interpret it as a path.
    collection_id (int): Collection ID identifier from Transkribus.
    document_id (int): Document ID identifier from Transkribus.
    page_number (int): Internal page number identifier from Transkribus.
    filepath (bool): `True` if page_xml is a filepath, `False` for page_xml as string.

    Returns:
    None

    Examples:
    >>> session = pyreslib.transkribus.api_login(user,password)
    >>>
    """
    headers = {"Content-Type": "application/xml; charset=UTF-8"}
    if filepath:
        page_xml_file = open(page_xml, "rb")
        page_xml_data = page_xml_file.read()
        page_xml_file.close()
    else:
        page_xml_data = page_xml

    page_xml_response = session.post(
        f"https://transkribus.eu/TrpServer/rest/collections/{collection_id}/{document_id}/{page_number}/text",
        headers=headers,
        data=page_xml_data,
    )

    if page_xml_response.status_code in [200, 201]:
        print("Success! PAGEXML has been updated to Transkribus.")
    else:
        print(f"Failed with Status Code: {page_xml_response.status_code}")
        print(f"Server Response: {page_xml_response.text}")

    return None

reading_order_document(session, collection_id, document_id, n_columns=2, page_center_method='image_width', reference_type='page-number', min_page_status='FINAL')

Applies reading order to whole document, only for pages with status equal or better that given parameter.

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login method.	required
collection_id	int	Collection ID identifier from Transkribus.	required
doocument_id	int	Document ID identifier from Transkribus.	required
n_columns	int	Number of columns. The method accepts only 1 or 2. Default is 2.	2
page_center_method	str	Reference method in order to determine page center. Options are "image_width" (get half of the whole image length) and "reference_region" (a specific region is used as center).	'image_width'
reference_type	str	Region tag used for "reference_region" as page_center_method parameter. Default is None.	'page-number'
min_page_status	str	Minimal status in order to apply reordering to page. Default is FINAL, but you can use NEW,IN_PROGRESS, DONE and GT instead.	'FINAL'

Retuns

None

Source code in pyreslib/transkribus.py

def reading_order_document(
    session,
    collection_id: int,
    document_id: int,
    n_columns: int = 2,
    page_center_method: str = "image_width",
    reference_type: str = "page-number",
    min_page_status="FINAL",
):
    """
    Applies reading order to whole document, only for pages with status equal or better that given parameter.

    Args:
                                                                    session: Transkribus session from [pyreslib.transkribus.api_login][] method.
                                                                    collection_id (int): Collection ID identifier from Transkribus.
                                                                    doocument_id (int): Document ID identifier from Transkribus.
                                                                    n_columns (int): Number of columns. The method accepts only 1 or 2. Default is 2.
                                                                    page_center_method (str): Reference method in order to determine page center. Options are "image_width" (get half of the whole image length) and "reference_region" (a specific region is used as center).
                                                                    reference_type (str): Region tag used for "reference_region" as `page_center_method` parameter. Default is `None`.
                                                                    min_page_status (str): Minimal status in order to apply reordering to page. Default is `FINAL`, but you can use `NEW`,`IN_PROGRESS`, `DONE` and `GT` instead.

    Retuns:
                                                                    `None`

    """
    # Transkribus statusses:
    tr_statuses = ["NEW", "IN_PROGRESS", "DONE", "FINAL", "GT"]
    # Get minimal status for reading order application
    index_min_status = tr_statuses.index(min_page_status)

    # retrieve collection metadata
    print(f"Importing metadata for document {document_id}")
    documents_metadata = get_documents_metadata(
        session=session, collection_id=collection_id
    )

    # get document metadata
    document_metadata = list(
        filter(lambda x: x["md"]["docId"] == document_id, documents_metadata)
    )[0]

    print(f"Reference type: {reference_type}")
    for page in document_metadata["pageList"]["pages"]:
        page_status = get_page_status(
            session=session,
            collection_id=collection_id,
            document_id=document_id,
            page_number=page["pageNr"],
        )

        if tr_statuses.index(page_status) >= index_min_status:
            print(f"Current page: {page['pageNr']}")
            reading_order_regions(
                session=session,
                collection_id=collection_id,
                document_id=document_id,
                page_number=page["pageNr"],
                n_columns=n_columns,
                page_center_method=page_center_method,
                reference_type=reference_type,
            )

    print(
        f"Reading order completed, Check your collection at https://app.transkribus.org/collection/{collection_id}/doc/{document_id}"
    )

reading_order_regions(session, collection_id, document_id, page_number, n_columns=2, page_center_method='image_width', reference_type=None)

Update region order via Transkribus API.

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login method.	required
collection_id	int	Collection ID identifier from Transkribus.	required
doocument_id	int	Document ID identifier from Transkribus.	required
page_number	int	Internal page number identifier from Transkribus.	required
n_columns	int	Number of columns. The method accepts only 1 or 2. Default is 2.	2
page_center_method	str	Reference method in order to determine page center. Options are "image_width" (get half of the whole image length) and "reference_region" (a specific region is used as center).	'image_width'
reference_type	str	Region tag used for "reference_region" as page_center_method parameter. Default is None.	None

Returns:

Name	Type	Description
page_xml_string	str	String serialization of the modified PAGEXML file. The new XML file is automatically updated back to Transkribus via pyreslib.transkribus.post_page_xml method.

Examples:

>>> ordered_page_xml = reading_order_regions(
session=transkribus_session,
collection_id=2353709,document_id=14756063,page_number=29,
n_columns=2,page_center_method="reference_region",reference_type="page-number")
>>> Exporting back to Transkribus...

Source code in pyreslib/transkribus.py

def reading_order_regions(
    session,
    collection_id: int,
    document_id: int,
    page_number: int,
    n_columns: int = 2,
    page_center_method: str = "image_width",
    reference_type: str = None,
):
    """
    Update region order via Transkribus API.

    Args:
                                                                    session: Transkribus session from [pyreslib.transkribus.api_login][] method.
                                                                    collection_id (int): Collection ID identifier from Transkribus.
                                                                    doocument_id (int): Document ID identifier from Transkribus.
                                                                    page_number (int): Internal page number identifier from Transkribus.
                                                                    n_columns (int): Number of columns. The method accepts only 1 or 2. Default is 2.
                                                                    page_center_method (str): Reference method in order to determine page center. Options are "image_width" (get half of the whole image length) and "reference_region" (a specific region is used as center).
                                                                    reference_type (str): Region tag used for "reference_region" as `page_center_method` parameter. Default is `None`.

    Returns:
                                                                    page_xml_string (str): String serialization of the modified PAGEXML file. The new XML file is automatically updated back to Transkribus via [pyreslib.transkribus.post_page_xml][] method.

    Examples:
                                                                    >>> ordered_page_xml = reading_order_regions(
                                                                    session=transkribus_session,
                                                                    collection_id=2353709,document_id=14756063,page_number=29,
                                                                    n_columns=2,page_center_method="reference_region",reference_type="page-number")
                                                                    >>> Exporting back to Transkribus...



    """

    # Import page_xml from Transkribus API
    print(f"Importing PAGEXML from API...")
    print(f"page_number: {page_number}")
    page_xml = get_page_xml(
        session=session,
        collection_id=collection_id,
        document_id=document_id,
        page_number=page_number,
    )
    # Parse via ET
    root = ET.fromstring(page_xml)
    PAGEXML_NS = "http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15"
    ET.register_namespace("", PAGEXML_NS)
    # Get image width
    image_width = root.find(f"{{{PAGEXML_NS}}}Page").attrib["imageWidth"]
    # Calculate page center
    if page_center_method == "image_width":
        page_center = float(image_width) / 2
    elif page_center_method == "reference_region":
        # get region with reference_type tag:
        found = False
        for region in root.findall(f"{{{PAGEXML_NS}}}Page/{{{PAGEXML_NS}}}TextRegion"):
            try:
                if (
                    region.attrib["custom"].split("type:")[1].split(";")[0]
                    == reference_type
                ):
                    page_center = get_polygonal_centroids(
                        extract_region_polygonal_coordinates(region)
                    )[0]
                    found = True
                    break
            except IndexError:
                # exclude regions without structural tag
                pass

        if found is False:
            print("Reference region not found, Use image_width instead")
            page_center = float(image_width) / 2

    print(f"Image width: {image_width}")
    error_margin = float(image_width) / 50
    print(f"Error margin 2%: {error_margin}")
    print(f"Page center: {page_center}")

    # Get regions:
    regions = []
    for region in root.findall(f"{{{PAGEXML_NS}}}Page/{{{PAGEXML_NS}}}TextRegion"):
        coordinates = extract_region_polygonal_coordinates(region)
        try:
            regions.append(
                {
                    "id": region.attrib["id"],
                    "type": region.attrib["custom"].split("type:")[1].split(";")[0],
                    "centroids": get_polygonal_centroids(coordinates),
                    "region_length": (
                        max([p[0] for p in coordinates])
                        - min([p[0] for p in coordinates])
                    ),
                }
            )
        except IndexError:  # no type tag for region
            regions.append(
                {
                    "id": region.attrib["id"],
                    "type": "",
                    "centroids": get_polygonal_centroids(coordinates),
                    "region_length": (
                        max([p[0] for p in coordinates])
                        - min([p[0] for p in coordinates])
                    ),
                }
            )

    # classify region
    for region in regions:
        x_left = region["centroids"][0] - region["region_length"] / 2
        x_right = region["centroids"][0] + region["region_length"] / 2

        # print(region["id"],region["type"])

        if region["type"] == reference_type:
            region["cluster"] = "center"
        else:
            if x_left < (page_center - error_margin):
                if x_right < (page_center + error_margin):
                    region["cluster"] = "left"
                else:
                    region["cluster"] = "center"
            else:
                region["cluster"] = "right"

    # order regions according to y-centroids
    y_orderded_regions = sorted(regions, key=lambda x: x["centroids"][1])

    if n_columns == 2:
        # generate correct reading order, assuming Western left-to-right reading.
        r = -1
        right_regions_cache = []
        for region in y_orderded_regions:
            if region["cluster"] == "center":
                # append right_regions chache
                for r_region in right_regions_cache:
                    r += 1
                    r_region["reading_order"] = r

                # delete cache
                right_regions_cache = []
                # append new centered region
                r += 1
                region["reading_order"] = r

            elif region["cluster"] == "left":
                r += 1
                region["reading_order"] = r
            else:  # right case, add to cache
                right_regions_cache.append(region)

        # append last right regions
        for r_region in right_regions_cache:
            r += 1
            r_region["reading_order"] = r

        reading_ordered_regions = sorted(
            y_orderded_regions, key=lambda x: x["reading_order"]
        )

    else:  # 1 column case.
        reading_ordered_regions = y_orderded_regions

    # change reading order values in PAGEXML
    reading_order = root.find(
        f"{{{PAGEXML_NS}}}Page/{{{PAGEXML_NS}}}ReadingOrder/{{{PAGEXML_NS}}}OrderedGroup"
    )
    if reading_order is not None:
        # delete all subelements
        reading_order.clear()
        # add new order
        for region in reading_ordered_regions:
            new_element = ET.SubElement(
                reading_order,
                "RegionRefIndexed",
                attrib={
                    "index": str(region["reading_order"]),
                    "regionRef": str(region["id"]),
                },
            )
        # change reading_order value in each region
    else:
        print("ReadingOder element not present.")
        pass

    for region in root.findall(f"{{{PAGEXML_NS}}}Page/{{{PAGEXML_NS}}}TextRegion"):
        # get new reading order
        ordered_region = list(
            filter(lambda x: x["id"] == region.attrib["id"], reading_ordered_regions)
        )[0]
        # substitute reading order in custom attribute
        old_reading_order = region.attrib["custom"].split("index:")[1].split(";")[0]
        region.attrib["custom"] = region.attrib["custom"].replace(
            f"index:{old_reading_order}", f"index:{ordered_region['reading_order']}"
        )

    # string export
    page_xml_string = ET.tostring(
        root, encoding="utf-8", xml_declaration=True, method="xml"
    )
    # Export back to Transkribus API
    print("Exporting back to Transkribus...")
    post_page_xml(
        session=session,
        page_xml=page_xml_string,
        collection_id=collection_id,
        document_id=document_id,
        page_number=page_number,
        filepath=False,
    )
    return page_xml_string

pyreslib.kraken

serialization_prediction(segmentation, file_path, image, export_dir, xml_namespace)

Serializes recognition results to PageXML and plain text files.

Parameters:

Name	Type	Description	Default
segmentation		Segmentation object with recognition results	required
file_path	str	Path of the original image.	required
image		PIL.Image object	required
export_dir	str	Export directory.	required
xml_namespace	str	XML namespace for PageXML output.	required

Returns:

Type	Description
	None

Source code in pyreslib/kraken.py

def serialization_prediction(
	segmentation, file_path: str, image, export_dir: str, xml_namespace: str
):
	"""
	Serializes recognition results to PageXML and plain text files.

	Args:
			segmentation: Segmentation object with recognition results
			file_path (str): Path of the original image.
			image: PIL.Image object
			export_dir (str): Export directory.
			xml_namespace (str): XML namespace for PageXML output.

	Returns:
			None
	"""
	# Serialize to PageXML
	page_xml = serialization.serialize(
		segmentation,
		image_size=image.size,
		template="pagexml",
		sub_line_segmentation=False,
	)

	# Get base filename without extension
	base_filename = os.path.splitext(os.path.basename(file_path))[0]

	# Save XML to file
	xml_output_path = os.path.join(export_dir, f"{base_filename}.xml")
	print(f"Serializing image into {xml_output_path}...")
	with open(xml_output_path, "w") as f:
		f.write(page_xml)

	# Parse PageXML and extract plain text
	root = ET.fromstring(page_xml)
	plain_text = ""

	# Find all TextLine elements (using wildcard for namespace)
	for line in root.iter():
		if line.tag.endswith("TextLine"):
			for child in line.iter():
				if child.tag.endswith("Unicode") and child.text:
					plain_text += child.text
			plain_text += "\n"

	# Save TXT to file
	txt_output_path = os.path.join(export_dir, f"{base_filename}.txt")
	print(f"Saving text to {txt_output_path}...")
	with open(txt_output_path, "w") as f:
		f.write(plain_text)

transcribe_directory(dir_path='./data/kraken/transcriptions', segmentation_model_path=None, recognition_model_path='./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel', export_dir='./data/kraken/transcriptions', xml_namespace='http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15')

Transcribe image files within directory (no subdirectories allowed) using Kraken.

Parameters:

Name	Type	Description	Default
dir_path	str	Directory path where images are stored. Default is ./data/kraken/transcriptions	'./data/kraken/transcriptions'
segmentation_model_path	str	Path to segmentation model. Default is None, but you can install dfine_kraken plugin and upload your custom model in "./data/kraken/models/.	None
recognition_model_path	str	Path to recognition model. Default is the CATMUS Print large "./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel"	'./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel'
export_dir	str	Export directory for the PAGEXML files. Default is ./data/kraken/transcriptions	'./data/kraken/transcriptions'
xml_namespace	str	PAGEXML namespace used by Kraken 7: http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15.	'http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15'

Returns:

Type	Description
	None

Source code in pyreslib/kraken.py

def transcribe_directory(
	dir_path: str = "./data/kraken/transcriptions",
	segmentation_model_path: str = None,
	recognition_model_path: str = "./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel",
	export_dir: str = "./data/kraken/transcriptions",
	xml_namespace: str = "http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15",
):
	"""
	Transcribe image files within directory (no subdirectories allowed) using Kraken.

	Args:
		dir_path(str): Directory path where images are stored. Default is `./data/kraken/transcriptions`
		segmentation_model_path (str): Path to segmentation model. Default is `None`, but you can install dfine_kraken plugin and upload your custom model in `"./data/kraken/models/`.
		recognition_model_path(str): Path to recognition model. Default is the [CATMUS Print large](https://zenodo.org/records/10592716) "./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel"
		export_dir(str): Export directory for the PAGEXML files. Default is `./data/kraken/transcriptions`
		xml_namespace(str): PAGEXML namespace used by Kraken 7: `http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15`.

	Returns:
		`None`
	"""
	# Load models once outside the loop
	print("Loading segmentation model...")
	try:
		seg_model = SegmentationTaskModel.load_model(segmentation_model_path)
	except ValueError as e:
		print(f"Error loading segmentation model: {e}")
		print("Falling back to default BLLA segmentation model...")
		seg_model = SegmentationTaskModel.load_model()  # Load default model

	print("Loading recognition model...")
	rec_model = RecognitionTaskModel.load_model(recognition_model_path)

	seg_config = SegmentationInferenceConfig()
	rec_config = RecognitionInferenceConfig()

	for file in os.scandir(dir_path):
		if file.name.endswith((".png", ".jpg", ".jpeg", ".tif", ".gif")):
			print(f"Processing {file.name}...")

			try:
				# Open and binarize image
				img = Image.open(file.path)
				bw_img = binarization.nlbin(img)

				# Perform segmentation
				segmentation = seg_model.predict(bw_img, seg_config)

				# Perform recognition
				records = list(rec_model.predict(bw_img, segmentation, rec_config))

				# Build segmentation with recognition results
				recognized_segmentation = Segmentation(
					lines=records,
					imagename=file.path,
					type=segmentation.type,
					text_direction=segmentation.text_direction,
					script_detection=segmentation.script_detection,
					regions=segmentation.regions,
				)

				# Serialize and save results
				serialization_prediction(
					segmentation=recognized_segmentation,
					file_path=file.path,
					image=bw_img,
					export_dir=export_dir,
					xml_namespace=xml_namespace,
				)
			except Exception as e:
				print(f"Error processing {file.name}: {e}")
				continue

transcribe_image(image_filepath, export_dir, segmentation_model_path=None, recognition_model_path='./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel', xml_namespace='http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15')

Transcribe image file using Kraken, returning a PAGEXML and TXT transcriptions.

Parameters:

Name	Type	Description	Default
dir_path	str	Directory path where images are stored. Default is ./data/kraken/transcriptions	required
segmentation_model_path	str	Path to segmentation model. Default is None, but you can install dfine_kraken plugin and upload your custom model in "./data/kraken/models/.	None
recognition_model_path	str	Path to recognition model. Default is the CATMUS Print large "./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel"	'./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel'
xml_namespace	str	PAGEXML namespace used by Kraken 7: http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15.	'http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15'

Returns:

Type	Description
	None

Source code in pyreslib/kraken.py

def transcribe_image(
	image_filepath: str,
	export_dir: str,
	segmentation_model_path: str = None,
	recognition_model_path: str = "./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel",
	xml_namespace: str = "http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15",
):
	"""
	Transcribe image file using Kraken, returning a PAGEXML and TXT transcriptions.

	Args:
		dir_path(str): Directory path where images are stored. Default is `./data/kraken/transcriptions`
		segmentation_model_path (str): Path to segmentation model. Default is `None`, but you can install dfine_kraken plugin and upload your custom model in `"./data/kraken/models/`.
		recognition_model_path(str): Path to recognition model. Default is the [CATMUS Print large](https://zenodo.org/records/10592716) "./data/kraken/models/catmus-large/catmus-print-fondue-large.mlmodel"
		xml_namespace(str): PAGEXML namespace used by Kraken 7: `http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15`.

	Returns:
		`None`
	"""
	# Load models once outside the loop
	print("Loading segmentation model...")
	try:
		seg_model = SegmentationTaskModel.load_model(segmentation_model_path)
	except ValueError as e:
		print(f"Error loading segmentation model: {e}")
		print("Falling back to default BLLA segmentation model...")
		seg_model = SegmentationTaskModel.load_model()  # Load default model

	print("Loading recognition model...")
	rec_model = RecognitionTaskModel.load_model(recognition_model_path)

	seg_config = SegmentationInferenceConfig()
	rec_config = RecognitionInferenceConfig()

	print(f"Processing {image_filepath}...")

	try:
		# Open and binarize image
		img = Image.open(image_filepath)
		bw_img = binarization.nlbin(img)

		# Perform segmentation
		segmentation = seg_model.predict(bw_img, seg_config)

		# Perform recognition
		records = list(rec_model.predict(bw_img, segmentation, rec_config))

		# Build segmentation with recognition results
		recognized_segmentation = Segmentation(
			lines=records,
			imagename=image_filepath,
			type=segmentation.type,
			text_direction=segmentation.text_direction,
			script_detection=segmentation.script_detection,
			regions=segmentation.regions,
		)

		# Serialize and save results
		serialization_prediction(
			segmentation=recognized_segmentation,
			file_path=image_filepath,
			image=bw_img,
			export_dir=export_dir,
			xml_namespace=xml_namespace,
		)
	except Exception as e:
		print(f"Error processing {image_filepath}: {e}")

transcription_dir_to_transkribus(transcription_dir, session, collection_id, document_id)

Transcribes a whole Transkribus document by:

1. Getting all images from API using pyreslib.transkribus.get_jpg_image
2. Transcribing images using kraken pyreslib.kraken.transcribe_image
3. Updating transcriptions backt to Transkribus via [pyreslib.kraken].update_kraken_XML_to_transkribus][].

Source code in pyreslib/kraken.py

def transcription_dir_to_transkribus(transcription_dir: str, session, collection_id:int, document_id:int):
	"""
	Transcribes a whole Transkribus document by:

	1. Getting all images from API using [pyreslib.transkribus.get_jpg_image][]
	2. Transcribing images using kraken [pyreslib.kraken.transcribe_image][] 
	3. Updating transcriptions backt to Transkribus via [pyreslib.kraken].update_kraken_XML_to_transkribus][].

	"""
	# 1. images

	# document_metadata
	print("Importing document metadata...")
	document_metadata = transkribus.get_document_metadata(session=session,collection_id=collection_id,document_id=document_id)

	print("Getting images from Transkribus API...")
	for page in document_metadata["pageList"]["pages"]:
		# import images in directory
		print(f"Current page: {page["pageNr"]}")
		transkribus.get_jpg_image(session=session, collection_id=collection_id,document_id=document_id,page_number=page["pageNr"], output_dir=transcription_dir)

	# 2. Kraken
	print("Transcribing images using Kraken, by default it skips the image if an XML file is present.")
	print("It might take a while...")

	all_files = list(os.scandir(transcription_dir))
	existing_xml_basenames = {
		f.name.replace(".xml", "") 
		for f in all_files 
		if f.name.endswith(".xml")
		}

	for f in all_files:
		if f.path.endswith(".jpg"):
			img_basename = f.name.replace(".jpg","")
			if img_basename in existing_xml_basenames:
				print("Skipping image. Transcription is already present.")
			else:
				transcribe_image(image_filepath=f.path,export_dir=transcription_dir)


	# 3. Transkribus
	print("Converting and updating transcriptions back to Transkribus")
	for f in all_files:
		if f.path.endswith(".xml"):
			# retrieve page number from filename
			xml_page_number = list(filter(lambda x: x["imgFileName"] == f.name.replace(".xml",".jpg"), document_metadata["pageList"]["pages"]))[0]["pageNr"]
			update_kraken_XML_to_transkribus(session=session,collection_id=collection_id,document_id=document_id, page_number=xml_page_number,page_xml_filepath=f.path)

update_kraken_XML_to_transkribus(session, collection_id, document_id, page_number, page_xml_filepath, keep_regions=True)

Converts and updates PAGEXML generated by Kraken back to Transkribus via API.

Parameters:

Name	Type	Description	Default
session		Transkribus session from pyreslib.transkribus.api_login() method.	required
collection_id	int	Collection ID identifier from Transkribus.	required
document_id	int	Document ID identifier from Transkribus.	required
page_number	int	Internal page number identifier from Transkribus.	required
page_xml_filepath	str	Filepath for Kraken generated PAGEXML file.	required
keep_regions	bool	If True, it will fit the transcribed baselines into existing TextRegions in transcribus, otherwise it overwrites the whole page layout. Default is True.	True

Returns:

Type	Description
	None

Source code in pyreslib/kraken.py

def update_kraken_XML_to_transkribus(session,collection_id: int,document_id:int, page_number: int,page_xml_filepath: str, keep_regions=True):

	"""
	Converts and updates PAGEXML generated by Kraken back to Transkribus via API.

	Args:
		session: Transkribus session from `pyreslib.transkribus.api_login()` method.
		collection_id (int): Collection ID identifier from Transkribus.
		document_id (int): Document ID identifier from Transkribus.
		page_number (int): Internal page number identifier from Transkribus.
		page_xml_filepath (str): Filepath for Kraken generated PAGEXML file.
		keep_regions (bool): If `True`, it will fit the transcribed baselines into existing TextRegions in transcribus, otherwise it overwrites the whole page layout. Default is `True`.

	Returns:
		`None`

	"""
	# convert PAGEXML kraken file for Transkribus
	TRANSKRIBUS_NS_URL = "http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15"
	KRAKEN_NS_URL = "http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15"
	print(f"Parsing PAGEXML from Kraken transcription {page_xml_filepath}...")
	tree = ET.parse(page_xml_filepath)
	root = tree.getroot()
	# PAGE XML namespace map for searching elements
	ns = {'xmlns': KRAKEN_NS_URL}

	if 'xsi:schemaLocation' in root.attrib:
		del root.attrib['{http://w3.org}schemaLocation']

	print("Converting to Transkribus XML format...")
	# 3. Patch TextRegions: Add readingOrder and change layout structural names
	for idx, region in enumerate(root.findall('.//xmlns:TextRegion', ns)):
		# Transkribus reads layouts via 'structure {type:heading/paragraph/etc}'
		region.set('custom', f"readingOrder {{index:{idx};}} structure {{type:paragraph;}}")

	# 4. Patch TextLines: Transkribus expects 'layout' definitions rather than 'default'
	for idx, line in enumerate(root.findall('.//xmlns:TextLine', ns)):
		line.set('custom', f"readingOrder {{index:{idx};}}")

	# namespaces
	#ET.register_namespace('', 'http://schema.primaresearch.org/PAGE/gts/pagecontent/2019-07-15')

	#ET.register_namespace('xsi', "http://www.w3.org/2001/XMLSchema-instance")
	#ET.register_namespace('schemaLocation',"http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15 http://schema.primaresearch.org/PAGE/gts/pagecontent/2013-07-15/pagecontent.xsd")


	if keep_regions is False:
		# 5. Convert back to string bytes 
		ET.register_namespace("",TRANSKRIBUS_NS_URL) # use Transkribus version
		# Transkribus explicitly requires the XML declaration at the top
		xml_patched_bytes = ET.tostring(root, encoding="utf-8", xml_declaration=True)
		# replace Kraken namespace with transkribus one
		xml_patched_bytes = xml_patched_bytes.replace(KRAKEN_NS_URL.encode('utf-8'), TRANSKRIBUS_NS_URL.encode('utf-8'))

		# post file back to Transkribus via API, overwriting existing layout
		print(f"Updating transcription from {page_xml_filepath} back to Transkribus")
		transkribus.post_page_xml(session=session,page_xml=xml_patched_bytes,collection_id=collection_id,document_id=document_id,page_number=page_number,filepath=False)
	else:
		# get existing pagexml from trankribus
		print("Getting existing PAGEXML layout from Transkribus API")
		page_xml_transkribus = transkribus.get_page_xml(session=session,collection_id=collection_id,document_id=document_id,page_number=page_number)
		root_transkribus = ET.fromstring(page_xml_transkribus)

		# change kraken namespace to transkribus
		# 2. Iterate through all elements and replace the namespace
		for elem in root.iter():
			if elem.tag.startswith(f"{{{KRAKEN_NS_URL}}}"):
				# Extract the local tag name and prepend the new namespace
				tag_name = elem.tag.split('}', 1)[1]
				elem.tag = f"{{{TRANSKRIBUS_NS_URL}}}{tag_name}"
		ET.register_namespace("",TRANSKRIBUS_NS_URL) # use Transkribus version
		text_lines = root.findall(f".//{{{TRANSKRIBUS_NS_URL}}}TextLine")
		for text_line in text_lines:
			# get best regional fit base on coordinates
			root_transkribus = transkribus.fit_text_line_in_region(text_line,root_transkribus)

		xml_patched_bytes = ET.tostring(root_transkribus, encoding="utf-8", xml_declaration=True)
		# post file back to Transkribus via API, overwriting existing layout
		print(f"Updating transcription from {page_xml_filepath} back to Transkribus")
		transkribus.post_page_xml(session=session,page_xml=xml_patched_bytes,collection_id=collection_id,document_id=document_id,page_number=page_number,filepath=False)

Conversion modules

pyreslib.marc

csv2marc(csv_filepath, marc_filepath, separator='|')

Converts a record CSV into MARC as dictionary file.

Parameters:

Name	Type	Description	Default
csv_filepath	str	Filepath for the input CSV file. See template in data/koha_biblio/csv and data/koha_auth/csv.	required
marc_filepath	str	Filepath for the MARC file serialized using pymarc.	required
separator	str	CSV file separator for multiple values. Default is pipe.	'|'

Returns:

Name	Type	Description
marc_as_dict	dict	MARC record as dictionary for Koha API.

Examples:

>>> csv_filepath = "./data/koha_biblio/csv/records-yyyy-mm-dd.csv"
>>> marc_filepath = "./data/koha_biblio/marc/records-yyyy-mm-dd.marc"
>>> csv2marc(csv_filepath,marc_filepath)
>>> {"leader": ... "fields": [{"001": ... }, ...]}

Source code in pyreslib/marc.py

def csv2marc(csv_filepath: str, marc_filepath: str, separator: str = "|") -> dict:
    """
    Converts a record CSV into MARC as dictionary file.

    Args:
        csv_filepath (str): Filepath for the input CSV file. See template in `data/koha_biblio/csv` and `data/koha_auth/csv`.
        marc_filepath (str): Filepath for the MARC file serialized using pymarc.
        separator (str): CSV file separator for multiple values. Default is pipe.

    Returns:
        marc_as_dict (dict): MARC record as dictionary for Koha API.

    Examples:
        >>> csv_filepath = "./data/koha_biblio/csv/records-yyyy-mm-dd.csv"
        >>> marc_filepath = "./data/koha_biblio/marc/records-yyyy-mm-dd.marc"
        >>> csv2marc(csv_filepath,marc_filepath)
        >>> {"leader": ... "fields": [{"001": ... }, ...]}

    """
    # import CSV file as dictionary
    csv_dict = utilities.csv2dict(csv_filepath)

    # generate MARC records file via pymarc
    # new record (overwrite!)
    record_dict = []
    f = open(marc_filepath, "wb")
    record = None
    for row in csv_dict:
        if row["leader"] != "":
            # initate new record
            if record is None:
                # new record
                record = Record()
            else:
                # write previous record
                f.write(record.as_marc())
                record_dict.append(record.as_dict())
                # new record
                record = Record()
            # add leader

            record.leader = str(row["leader"])
            # add 001 field
            record.add_field(Field(tag=row["field"], data=row["values"]))

        else:  # other fields
            # split tags and values based on separator
            field_tag = row["field"]
            list_subfields = row["subfields"].split(separator)
            list_values = row["values"].split(separator)
            if list_subfields == [""]:  # no subfields, only field tag
                record.add_field(Field(tag=field_tag, data=row["values"]))
            else:
                record.add_field(
                    Field(
                        tag=field_tag,
                        indicators=Indicators(row["ind1"], row["ind2"]),
                        subfields=[
                            Subfield(code=list_subfields[i], value=list_values[i])
                            for i in range(len(list_subfields))
                        ],
                    )
                )

    # save MARC file to marc_filepath
    f.write(record.as_marc())
    record_dict.append(record.as_dict())
    f.close()

    return record_dict

generate_record_dict(marc_filepath, json_filepath, id_name)

Returns a list of dictionaries from a MARC record downloaded via Koha Cataloging batch import tool.

Parameters:

Name	Type	Description	Default
marc_filepath	str	Location of the MARC file to be processed. Default directory is ./data/koha_auth/marc for authorities and ./data/koha_biblio/marc for bibliographic records.	required
json_filepath	str	Destination file. Default directory is ./data/koha_auth/json for authorities and ./data/koha_biblio/json for bibliographic records. Set to None if you wish not to save the JSON file locally.	required
id_name	str	Field name of ID value. Use biblio_id for bibliographic records and auth_id for authorities.	required

Returns:

Name	Type	Description
record_dict	list	List of dictionaries in MARC-in-JSON format.

Source code in pyreslib/marc.py

def generate_record_dict(marc_filepath: str, json_filepath: str, id_name: str) -> list:
    """
    Returns a list of dictionaries from a MARC record downloaded via Koha Cataloging batch import tool.

    Args:
        marc_filepath (str): Location of the MARC file to be processed. Default directory is `./data/koha_auth/marc` for authorities and `./data/koha_biblio/marc` for bibliographic records.
        json_filepath (str): Destination file. Default directory is `./data/koha_auth/json` for authorities and `./data/koha_biblio/json` for bibliographic records. Set to `None` if you wish not to save the JSON file locally.
        id_name (str): Field name of ID value. Use `biblio_id` for bibliographic records and `auth_id` for authorities.

    Returns:
        record_dict (list): List of dictionaries in MARC-in-JSON format.
    """
    n = 0
    success = 0
    f = open(marc_filepath, "rb")
    reader = MARCReader(f)
    record_dict = []
    for record in reader:
        n += 1
        record_as_dict = record.as_dict()
        try:
            record_id = list(
                filter(lambda x: "001" in x.keys(), record_as_dict["fields"])
            )[0]["001"]
            record_dict.append(
                {
                    id_name: record_id,
                    "record": record_as_dict,
                }
            )
            success += 1
        except KeyError:
            pass

    print(f"{success} records out of {n} have been successfully extracted.")

    # Saving dictionary to JSON
    if json_filepath is not None:
        utilities.dict2json(record_dict, json_filepath)

    return record_dict

marc2csv(marc_filepath, csv_filepath, separator='|')

Converts a MARC record into CSV.

Parameters:

Name	Type	Description	Default
marc_filepath	str	Filepath for the MARC file serialized using pymarc.	required
csv_filepath	str	Filepath for the output CSV file. See template in data/koha_biblio/csv and data/koha_auth/csv.	required
separator	str	CSV file separator for multiple values. Default is pipe.	'|'

Returns: None

Examples:

>>> csv_filepath = "./data/koha_biblio/csv/records-yyyy-mm-dd.csv"
>>> marc_filepath = "./data/koha_biblio/marc/records-yyyy-mm-dd.marc"
>>> marc2csv(marc_filepath,csv_filepath)

Source code in pyreslib/marc.py

def marc2csv(marc_filepath: str, csv_filepath: str, separator: str = "|"):
    """
    Converts a MARC record into CSV.

    Args:
        marc_filepath (str): Filepath for the MARC file serialized using pymarc.
        csv_filepath (str): Filepath for the output CSV file. See template in `data/koha_biblio/csv` and `data/koha_auth/csv`.
        separator (str): CSV file separator for multiple values. Default is pipe.
    Returns:
        `None`

    Examples:
        >>> csv_filepath = "./data/koha_biblio/csv/records-yyyy-mm-dd.csv"
        >>> marc_filepath = "./data/koha_biblio/marc/records-yyyy-mm-dd.marc"
        >>> marc2csv(marc_filepath,csv_filepath)
    """
    # import MARC file via pymarc as dictionary
    f = open(marc_filepath, "rb")
    reader = MARCReader(f)
    record_dict = []
    for record in reader:
        record_dict.append(record.as_dict())

    # initiate csv dictionary
    csv_dict = []
    # add fields and subfields values
    for record in record_dict:
        # initialize new record

        # continue adding fields and subfields
        for field in record["fields"]:
            print(f"Current field: {field}")
            field_tag = list(field.keys())[0]
            # print(field_tag)
            if field_tag == "001":  # eclude 001 id field
                csv_dict.append(
                    {
                        "leader": record["leader"],
                        "ind1": " ",
                        "ind2": " ",
                        "field": "001",
                        "subfields": "",
                        "values": record["fields"][0]["001"],
                    }
                )
            else:
                try:
                    ind1 = field[field_tag]["ind1"]
                except Exception:
                    ind1 = ""
                try:
                    ind2 = field[field_tag]["ind2"]
                except Exception:
                    ind2 = ""

                try:
                    subfield_tags = separator.join(
                        [
                            list(subfield.keys())[0]
                            for subfield in field[field_tag]["subfields"]
                        ]
                    )
                    subfield_values = separator.join(
                        [
                            item.get(list(item.keys())[0])
                            for item in field[field_tag]["subfields"]
                        ]
                    )

                except TypeError:
                    # the field statement has no subfields
                    subfield_values = field.get(field_tag)
                    subfield_tags = ""
                csv_dict.append(
                    {
                        "leader": "",
                        "ind1": ind1,
                        "ind2": ind2,
                        "field": field_tag,
                        "subfields": subfield_tags,
                        "values": subfield_values,
                    }
                )
            # print(csv_dict)
            # input()

    # save csv_dict as CSV file
    utilities.dict2csv(csv_dict, csv_filepath)

pyreslib.bibtex

convert_biblio_to_bibtex(biblio_id, bibtex_filepath, koha_session, base_url)

Converts MARC-in-JSON record from Koha API into BibTeX file.

Parameters:

Name	Type	Description	Default
biblio_id	int	Unique idenfier for the record in Koha catalogue.	required
bibtex_filepath	str	String filepath indicating the location where the .bib output has to be stored.	required
koha_session	oauth2	Oauth2 session provided by pyreslib.koha.oauth2_session method.	required
base_url	str	Koha API url from credentials.	required

Returns:

Type	Description
	None

Examples:

Source code in pyreslib/bibtex.py

def convert_biblio_to_bibtex(
    biblio_id: int,
    bibtex_filepath: str,
    koha_session,
    base_url: str,
):
    """
    Converts MARC-in-JSON record from Koha API into BibTeX file.

    Args:
        biblio_id (int): Unique idenfier for the record in Koha catalogue.
        bibtex_filepath (str): String filepath indicating the location where the .bib output has to be stored.
        koha_session (oauth2): Oauth2 session provided by `pyreslib.koha.oauth2_session` method.
        base_url (str): Koha API url from credentials.

    Returns:
        `None`

    Examples:

    """

    # get MARC-in-JSON record
    record = koha.get_biblio_marc(
        session=koha_session, biblio_id=biblio_id, base_url=base_url
    )

    # get marc biblio type
    entry_type = entry_type_mapping[koha.get_biblio_type(record)]

    # generate BibTeX entry according to type

    entry = {}

    entry_id = (
        get_authors(record).split(" ")[0].lower()
        + "_"
        + get_title(record).replace(" ", "").lower()[:10]
        + "_"
        + get_year(record)
    )

    if entry_type == "article":
        entry = {
            "ENTRYTYPE": entry_type,
            "ID": entry_id,
            "author": get_authors(record),
            "title": get_title(record),
            "journal": get_journal(record),
            "year": str(get_year(record)),
            "volume": get_volume(record),
            "number": get_issue_number(record),
            "issn": get_issn(record),
            "pages": get_pages(record),
            "url": get_url(record),
            "doi": get_doi(record),
            "keywords": get_keywords(record),
            "abstract": get_abstract(record),
            "note": get_note(record),
            "type": entry_type,
            "howpublished": get_howpublished(record),
        }

    elif entry_type == "book":
        entry = {
            "ENTRYTYPE": entry_type,
            "ID": entry_id,
            "author": get_authors(record),
            "title": get_title(record),
            "publisher": get_publisher(record),
            "address": get_address(record),
            "year": str(get_year(record)),
            "isbn": get_isbn(record),
            "pages": get_pages(record),
            "url": get_url(record),
            "doi": get_doi(record),
            "keywords": get_keywords(record),
            "abstract": get_abstract(record),
            "note": get_note(record),
            "type": entry_type,
            "howpublished": get_howpublished(record),
        }

    elif entry_type == "incollection":
        entry = {
            "ENTRYTYPE": entry_type,
            "ID": entry_id,
            "author": get_authors(record),
            "title": get_title(record),
            "booktitle": get_booktitle(record),
            "publisher": get_publisher(record),
            "address": get_address(record),
            "year": str(get_year(record)),
            "isbn": get_isbn(record),
            "pages": get_pages(record),
            "url": get_url(record),
            "doi": get_doi(record),
            "keywords": get_keywords(record),
            "abstract": get_abstract(record),
            "note": get_note(record),
            "type": entry_type,
            "howpublished": get_howpublished(record),
        }

    elif entry_type == "misc":
        entry = {
            "ENTRYTYPE": entry_type,
            "ID": entry_id,
            "author": get_authors(record),
            "title": get_title(record),
            "publisher": get_publisher(record),
            "address": get_address(record),
            "year": str(get_year(record)),
            "url": get_url(record),
            "doi": get_doi(record),
            "keywords": get_keywords(record),
            "abstract": get_abstract(record),
            "note": get_note(record),
            "type": entry_type,
            "howpublished": get_howpublished(record),
        }

    elif entry_type == "phdthesis":
        entry = {
            "ENTRYTYPE": entry_type,
            "ID": entry_id,
            "author": get_authors(record),
            "title": get_title(record),
            "school": get_school(record),
            "address": get_address(record),
            "year": str(get_year(record)),
            "pages": get_pages(record),
            "url": get_url(record),
            "doi": get_doi(record),
            "keywords": get_keywords(record),
            "abstract": get_abstract(record),
            "note": get_note(record),
            "type": entry_type,
            "howpublished": get_howpublished(record),
        }
    else:
        raise ValueError(f"No entry_type found for biblio_id {biblio_id}")

    # generate single bibtex file
    bibtex_db = BibDatabase()
    # initialize BibTeX writer
    bibtex_writer = BibTexWriter()

    # add entries to database
    bibtex_db.entries = [entry]

    with open(bibtex_filepath, "w") as bibfile:
        bibfile.write(bibtex_writer.write(bibtex_db))

pyreslib.rdf

This module exports bibliographic and authority records as RDF data structures, according to a given mapping.

auth_records_to_rdf(auth_ids, session, base_url, koha_namespace, auth_rdf_mapping_filepath='./data/mappings/lod/koha-rdf_mapping-auth.csv', namespaces_filepath='./data/mappings/lod/rdf_namespaces.json', output_dir='./data/rdf/auth/', serialization_format='turtle', explicit_abbreviations=True)

Convert authority records to RDF graph.

Given a list of authority IDs, returns an rdflib.Graph object with RDF serializations of the authority records according to the provided mapping.

Parameters:

Name	Type	Description	Default
auth_ids	list	List of authority record IDs as integers.	required
session	Session	OAuth2 session provided by koha_session().	required
base_url	str	Koha API URL from credentials.	required
koha_namespace	str	Prefix used for your Koha namespace in the namespaces JSON file (e.g., "oikoha"). Omit "auth" suffix.	required
auth_rdf_mapping_filepath	str	File path for MARC-to-RDF mapping CSV. Default is ./data/mappings/lod/koha-rdf_mapping-auth.csv.	'./data/mappings/lod/koha-rdf_mapping-auth.csv'
namespaces_filepath	str	JSON file with RDF namespaces. Default is ./data/mappings/lod/rdf_namespaces.json.	'./data/mappings/lod/rdf_namespaces.json'
output_dir	str	Output directory for RDF files. Files are named {auth_id}.ttl (or other extension based on format). Default is ./data/rdf/auth/. Set to None to skip file output.	'./data/rdf/auth/'
serialization_format	str	RDF serialization format. Options: "turtle", "json-ld", "pretty-xml", "xml". Default is "turtle".	'turtle'
explicit_abbreviations	bool	Whether to expand MARC abbreviation codes using koha.explicit_abbreviations_from_marc(). Default is True.	True

Returns:

Type	Description
Graph	rdflib.Graph: Combined RDF graph containing all authority records.

Raises:

Type	Description
FileNotFoundError	If mapping or namespaces files not found.
ValueError	If required namespaces not found in namespaces file.

Examples:

>>> auth_ids = [1, 342, 17]
>>> session = koha.koha_session(client_id="...", ...)
>>> g = auth_records_to_rdf(
...     auth_ids=auth_ids,
...     session=session,
...     base_url="https://koha.example.com/api/v1",
...     koha_namespace="oikoha"
... )
>>> g.serialize(format="turtle", destination="authorities.ttl")

Source code in pyreslib/rdf.py

def auth_records_to_rdf(
    auth_ids: list,
    session,
    base_url: str,
    koha_namespace: str,
    auth_rdf_mapping_filepath: str = "./data/mappings/lod/koha-rdf_mapping-auth.csv",
    namespaces_filepath: str = "./data/mappings/lod/rdf_namespaces.json",
    output_dir: str = "./data/rdf/auth/",
    serialization_format: str = "turtle",
    explicit_abbreviations: bool = True,
) -> Graph:
    """Convert authority records to RDF graph.

    Given a list of authority IDs, returns an rdflib.Graph object with RDF
    serializations of the authority records according to the provided mapping.

    Args:
        auth_ids (list): List of authority record IDs as integers.
        session (requests.Session): OAuth2 session provided by `koha_session()`.
        base_url (str): Koha API URL from credentials.
        koha_namespace (str): Prefix used for your Koha namespace in the
            namespaces JSON file (e.g., "oikoha"). Omit "auth" suffix.
        auth_rdf_mapping_filepath (str): File path for MARC-to-RDF mapping CSV.
            Default is `./data/mappings/lod/koha-rdf_mapping-auth.csv`.
        namespaces_filepath (str): JSON file with RDF namespaces.
            Default is `./data/mappings/lod/rdf_namespaces.json`.
        output_dir (str): Output directory for RDF files. Files are named
            `{auth_id}.ttl` (or other extension based on format).
            Default is `./data/rdf/auth/`. Set to None to skip file output.
        serialization_format (str): RDF serialization format. Options: "turtle",
            "json-ld", "pretty-xml", "xml". Default is "turtle".
        explicit_abbreviations (bool): Whether to expand MARC abbreviation codes
            using `koha.explicit_abbreviations_from_marc()`. Default is True.

    Returns:
        rdflib.Graph: Combined RDF graph containing all authority records.

    Raises:
        FileNotFoundError: If mapping or namespaces files not found.
        ValueError: If required namespaces not found in namespaces file.

    Examples:
        >>> auth_ids = [1, 342, 17]
        >>> session = koha.koha_session(client_id="...", ...)
        >>> g = auth_records_to_rdf(
        ...     auth_ids=auth_ids,
        ...     session=session,
        ...     base_url="https://koha.example.com/api/v1",
        ...     koha_namespace="oikoha"
        ... )
        >>> g.serialize(format="turtle", destination="authorities.ttl")
    """
    # Initialize main RDF graph
    g = Graph()

    # Load namespaces and bind to main graph
    namespaces = utilities.json2dict(namespaces_filepath)
    namespace_map = _setup_namespaces(g, namespaces_filepath)

    # Load MARC-to-RDF mapping
    auth_rdf_mapping = utilities.csv2dict(auth_rdf_mapping_filepath)

    # Get Koha authority namespace URI
    koha_auth_namespace_key = koha_namespace + "auth"
    koha_auth_base_url = _get_namespace_uri(koha_auth_namespace_key, namespaces)

    # Create output directory if needed
    if output_dir is not None and not os.path.exists(output_dir):
        os.makedirs(output_dir, exist_ok=True)

    # Determine file extension
    extension_map = {
        "turtle": ".ttl",
        "json-ld": ".json",
        "pretty-xml": ".xml",
        "xml": ".xml",
    }
    extension = extension_map.get(serialization_format, "")

    for auth_id in auth_ids:
        print(f"Processing authority {auth_id}...")

        # Define subject URI
        subject_uri = URIRef(f"{koha_auth_base_url}{auth_id}")

        # Create graph for this record
        g_record = Graph()

        # Bind all namespaces to this record's graph
        for ns in namespaces:
            g_record.bind(ns["namespace"], Namespace(ns["base_URI"]))

        # Add type statement
        g_record.add((subject_uri, RDF.type, RDFS.Resource))

        try:
            # Fetch authority record from Koha API
            auth = koha.get_authority_marc(
                auth_id=auth_id, session=session, base_url=base_url
            )

            # Expand abbreviations if requested
            if explicit_abbreviations:
                auth = koha.explicit_abbreviations_from_marc(record=auth)

            # Process each mapping rule
            for mapping in auth_rdf_mapping:
                try:
                    field = str(mapping["field"]).zfill(3)
                    subfield = str(mapping.get("subfield", ""))

                    # Get predicate URI
                    namespace_key = mapping.get("namespace", "")
                    property_name = mapping.get("property", "")

                    if not namespace_key or not property_name:
                        continue

                    try:
                        base_uri = _get_namespace_uri(namespace_key, namespaces)
                    except ValueError:
                        print(
                            f"⚠ Warning: Namespace '{namespace_key}' not found, skipping mapping"
                        )
                        continue

                    predicate_uri = URIRef(f"{base_uri}{property_name}")

                    # Find matching MARC fields
                    field_query = list(
                        filter(lambda x: field in x.keys(), auth.get("fields", []))
                    )

                    for field_statement in field_query:
                        # If subfield specified, filter by subfield
                        if subfield:
                            subfield_query = list(
                                filter(
                                    lambda x: subfield in x.keys(),
                                    field_statement[field].get("subfields", []),
                                )
                            )
                        else:
                            # No subfield specified, use all subfields
                            subfield_query = field_statement[field].get("subfields", [])

                        for subfield_statement in subfield_query:
                            # Extract value based on data type
                            value = (
                                subfield_statement.get(subfield)
                                if subfield
                                else subfield_statement
                            )

                            if value is None:
                                continue

                            # Create object based on data type
                            data_type = mapping.get("data_type", "Text")
                            is_authority = mapping.get("is_authority", False)

                            if data_type == "Text":
                                obj = Literal(value, datatype=XSD.string)
                            else:  # URI type
                                if is_authority:
                                    obj = URIRef(f"{koha_auth_base_url}{value}")
                                else:
                                    obj = URIRef(value)

                            # Add statement to graph
                            g_record.add((subject_uri, predicate_uri, obj))

                except KeyError as e:
                    print(f"⚠ Warning: Missing key in mapping: {e}")
                    continue

            # Serialize individual record if output directory specified
            if output_dir is not None:
                output_path = os.path.join(output_dir, f"{auth_id}{extension}")
                g_record.serialize(format=serialization_format, destination=output_path)
                print(f"✓ Saved to {output_path}")

            # Append to main graph
            g += g_record

        except Exception as e:
            print(f"✗ Error processing authority {auth_id}: {e}")
            continue

    print(f"✓ Successfully processed {len(auth_ids)} authorities")
    return g

biblio_records_to_rdf(biblio_ids, session, base_url, koha_namespace, biblio_rdf_mapping_filepath='./data/mappings/lod/koha-rdf_mapping-biblio.csv', namespaces_filepath='./data/mappings/lod/rdf_namespaces.json', output_dir='./data/rdf/biblio/', serialization_format='turtle', explicit_abbreviations=True)

Convert bibliographic records to RDF graph.

Given a list of biblio IDs, returns an rdflib.Graph object with RDF serializations of the bibliographic records according to the provided mapping.

Parameters:

Name	Type	Description	Default
biblio_ids	list	List of bibliographic record IDs as integers.	required
session	Session	OAuth2 session provided by koha_session().	required
base_url	str	Koha API URL from credentials.	required
koha_namespace	str	Prefix used for your Koha namespace in the namespaces JSON file (e.g., "oikoha"). Omit "biblio"/"auth" suffix.	required
biblio_rdf_mapping_filepath	str	File path for MARC-to-RDF mapping CSV. Default is ./data/mappings/lod/koha-rdf_mapping-biblio.csv.	'./data/mappings/lod/koha-rdf_mapping-biblio.csv'
namespaces_filepath	str	JSON file with RDF namespaces. Default is ./data/mappings/lod/rdf_namespaces.json.	'./data/mappings/lod/rdf_namespaces.json'
output_dir	str	Output directory for RDF files. Files are named {biblio_id}.ttl (or other extension based on format). Default is ./data/rdf/biblio/. Set to None to skip file output.	'./data/rdf/biblio/'
serialization_format	str	RDF serialization format. Options: "turtle", "json-ld", "pretty-xml", "xml". Default is "turtle".	'turtle'
explicit_abbreviations	bool	Whether to expand MARC abbreviation codes using koha.explicit_abbreviations_from_marc(). Default is True.	True

Returns:

Type	Description
Graph	rdflib.Graph: Combined RDF graph containing all bibliographic records.

Raises:

Type	Description
FileNotFoundError	If mapping or namespaces files not found.
ValueError	If required namespaces not found in namespaces file.

Examples:

>>> biblio_ids = [1, 342, 17]
>>> koha_session = koha.koha_session(client_id="...", ...)
>>> g = biblio_records_to_rdf(
...     biblio_ids=biblio_ids,
...     session=koha_session,
...     base_url="https://koha.example.com/api/v1",
...     koha_namespace="oikoha"
... )
>>> g.serialize(format="turtle", destination="biblios.ttl")

Source code in pyreslib/rdf.py

def biblio_records_to_rdf(
    biblio_ids: list,
    session,
    base_url: str,
    koha_namespace: str,
    biblio_rdf_mapping_filepath: str = "./data/mappings/lod/koha-rdf_mapping-biblio.csv",
    namespaces_filepath: str = "./data/mappings/lod/rdf_namespaces.json",
    output_dir: str = "./data/rdf/biblio/",
    serialization_format: str = "turtle",
    explicit_abbreviations: bool = True,
) -> Graph:
    """Convert bibliographic records to RDF graph.

    Given a list of biblio IDs, returns an rdflib.Graph object with RDF
    serializations of the bibliographic records according to the provided mapping.

    Args:
        biblio_ids (list): List of bibliographic record IDs as integers.
        session (requests.Session): OAuth2 session provided by `koha_session()`.
        base_url (str): Koha API URL from credentials.
        koha_namespace (str): Prefix used for your Koha namespace in the
            namespaces JSON file (e.g., "oikoha"). Omit "biblio"/"auth" suffix.
        biblio_rdf_mapping_filepath (str): File path for MARC-to-RDF mapping CSV.
            Default is `./data/mappings/lod/koha-rdf_mapping-biblio.csv`.
        namespaces_filepath (str): JSON file with RDF namespaces.
            Default is `./data/mappings/lod/rdf_namespaces.json`.
        output_dir (str): Output directory for RDF files. Files are named
            `{biblio_id}.ttl` (or other extension based on format).
            Default is `./data/rdf/biblio/`. Set to None to skip file output.
        serialization_format (str): RDF serialization format. Options: "turtle",
            "json-ld", "pretty-xml", "xml". Default is "turtle".
        explicit_abbreviations (bool): Whether to expand MARC abbreviation codes
            using `koha.explicit_abbreviations_from_marc()`. Default is True.

    Returns:
        rdflib.Graph: Combined RDF graph containing all bibliographic records.

    Raises:
        FileNotFoundError: If mapping or namespaces files not found.
        ValueError: If required namespaces not found in namespaces file.

    Examples:
        >>> biblio_ids = [1, 342, 17]
        >>> koha_session = koha.koha_session(client_id="...", ...)
        >>> g = biblio_records_to_rdf(
        ...     biblio_ids=biblio_ids,
        ...     session=koha_session,
        ...     base_url="https://koha.example.com/api/v1",
        ...     koha_namespace="oikoha"
        ... )
        >>> g.serialize(format="turtle", destination="biblios.ttl")
    """
    # Initialize main RDF graph
    g = Graph()

    # Load namespaces and bind to main graph
    namespaces = utilities.json2dict(namespaces_filepath)
    namespace_map = _setup_namespaces(g, namespaces_filepath)

    # Load MARC-to-RDF mapping
    biblio_rdf_mapping = utilities.csv2dict(biblio_rdf_mapping_filepath)

    # Get Koha namespace URIs
    koha_biblio_namespace_key = koha_namespace + "biblio"
    koha_auth_namespace_key = koha_namespace + "auth"

    koha_biblio_base_url = _get_namespace_uri(koha_biblio_namespace_key, namespaces)
    koha_auth_base_url = _get_namespace_uri(koha_auth_namespace_key, namespaces)

    # Create output directory if needed
    if output_dir is not None and not os.path.exists(output_dir):
        os.makedirs(output_dir, exist_ok=True)

    # Determine file extension
    extension_map = {
        "turtle": ".ttl",
        "json-ld": ".json",
        "pretty-xml": ".xml",
        "xml": ".xml",
    }
    extension = extension_map.get(serialization_format, "")

    for biblio_id in biblio_ids:
        print(f"Processing biblio {biblio_id}...")

        # Define subject URI
        subject_uri = URIRef(f"{koha_biblio_base_url}{biblio_id}")

        # Create graph for this record
        g_record = Graph()

        # Bind all namespaces to this record's graph
        for ns in namespaces:
            g_record.bind(ns["namespace"], Namespace(ns["base_URI"]))

        # Add type statement
        g_record.add((subject_uri, RDF.type, RDFS.Resource))

        try:
            # Fetch bibliographic record and items from Koha API
            biblio = koha.get_biblio_marc(
                biblio_id=biblio_id, session=session, base_url=base_url
            )
            biblio_items = koha.get_items_from_biblio_json(
                session=session, base_url=base_url, biblio_id=biblio_id
            )

            # Expand abbreviations if requested
            if explicit_abbreviations:
                biblio = koha.explicit_abbreviations_from_marc(record=biblio)

            # Process each mapping rule
            for mapping in biblio_rdf_mapping:
                try:
                    field = mapping.get("field", "")
                    subfield = str(mapping.get("subfield", ""))

                    # Get predicate URI
                    namespace_key = mapping.get("namespace", "")
                    property_name = mapping.get("property", "")

                    if not namespace_key or not property_name:
                        continue

                    try:
                        base_uri = _get_namespace_uri(namespace_key, namespaces)
                    except ValueError:
                        print(
                            f"⚠ Warning: Namespace '{namespace_key}' not found, skipping mapping"
                        )
                        continue

                    predicate_uri = URIRef(f"{base_uri}{property_name}")

                    # Determine if this is a MARC field or item metadata
                    try:
                        # Try to parse as MARC field (numeric)
                        field_num = str(int(field)).zfill(3)

                        # Find matching MARC fields
                        field_query = list(
                            filter(
                                lambda x: field_num in x.keys(),
                                biblio.get("fields", []),
                            )
                        )

                        for field_statement in field_query:
                            # Filter by subfield if specified
                            if subfield:
                                subfield_query = list(
                                    filter(
                                        lambda x: subfield in x.keys(),
                                        field_statement[field_num].get("subfields", []),
                                    )
                                )
                            else:
                                subfield_query = field_statement[field_num].get(
                                    "subfields", []
                                )

                            for subfield_statement in subfield_query:
                                value = (
                                    subfield_statement.get(subfield)
                                    if subfield
                                    else subfield_statement
                                )

                                if value is None:
                                    continue

                                # Create object based on data type
                                data_type = mapping.get("data_type", "Text")
                                is_authority = mapping.get("is_authority", False)
                                is_biblionumber = mapping.get("is_biblionumber", False)

                                if data_type == "Text":
                                    obj = Literal(value, datatype=XSD.string)
                                else:  # URI type
                                    if is_authority:
                                        obj = URIRef(f"{koha_auth_base_url}{value}")
                                    elif is_biblionumber:
                                        obj = URIRef(f"{koha_biblio_base_url}{value}")
                                    else:
                                        obj = URIRef(value)

                                g_record.add((subject_uri, predicate_uri, obj))

                    except ValueError:
                        # Not a MARC field, try item metadata
                        if biblio_items:
                            for item in biblio_items:
                                value = item.get(field)
                                if value is not None:
                                    obj = Literal(value, datatype=XSD.string)
                                    g_record.add((subject_uri, predicate_uri, obj))

                except Exception as e:
                    print(f"⚠ Warning: Error in mapping {mapping}: {e}")
                    continue

            # Serialize individual record if output directory specified
            if output_dir is not None:
                output_path = os.path.join(output_dir, f"{biblio_id}{extension}")
                g_record.serialize(format=serialization_format, destination=output_path)
                print(f"✓ Saved to {output_path}")

            # Append to main graph
            g += g_record

        except Exception as e:
            print(f"✗ Error processing biblio {biblio_id}: {e}")
            continue

    print(f"✓ Successfully processed {len(biblio_ids)} biblios")
    return g

pyreslib.omekas

change_item_property_value(session, value, property_name, resource_id, resource_type, value_position=0)

Change the value of a specific statement of a property of a given resource. Args: session: omeka_s_tools session generated by [pyreslib.omekas.omekas_session]. value (str): new value for the property. property_name (str): name of the property to be updated, such as dcterms:title. resource_id (int): ID of the item set to be added to the site. resource_type (str): Resource type, such as items, item_sets and media. value_position (int): position of the value to be change, in case of multiple statements for property. 0 is default.

Returns:

Type	Description
	None

Source code in pyreslib/omekas.py

def change_item_property_value(
    session: OmekaAPIClient,
    value: str,
    property_name: str,
    resource_id: int,
    resource_type: str,
    value_position: int = 0,
):
    """Change the value of a specific statement of a property of a given resource.
    Args:
        session: `omeka_s_tools` session generated by [pyreslib.omekas.omekas_session].
        value (str): new value for the property.
        property_name (str): name of the property to be updated, such as `dcterms:title`.
        resource_id (int): ID of the item set to be added to the site.
        resource_type (str): Resource type, such as `items`, `item_sets` and `media`.
        value_position (int): position of the value to be change, in case of multiple statements for property. 0 is default.

    Returns:
        None

    """
    data = session.get_resource_by_id(resource_id, resource_type=resource_type)
    data[property_name][value_position]["@value"] = value
    session.update_resource(data, resource_type=resource_type)

generate_omekas_mapping(mappings_directory)

Generates a dictionary from a series of CSV files (auth,biblio,media...). Args: mappings_directory (str): Path of csv mapping files. Returns: A dictionary of mappings.

Source code in pyreslib/omekas.py

def generate_omekas_mapping(mappings_directory: str) -> dict:
    """
    Generates a dictionary from a series of CSV files (auth,biblio,media...).
    Args:
        mappings_directory (str): Path of csv mapping files.
    Returns:
        A dictionary of mappings.
    """
    omekas_mapping = {}

    # get authorities mapping
    omekas_mapping["auth"] = utilities.csv2dict(
        os.path.join(mappings_directory, "koha-omekas_mapping - auth.csv")
    )
    # biblio mapping
    omekas_mapping["biblio"] = utilities.csv2dict(
        os.path.join(mappings_directory, "koha-omekas_mapping - biblio.csv")
    )
    # get media mapping
    omekas_mapping["media"] = utilities.csv2dict(
        os.path.join(mappings_directory, "koha-omekas_mapping - media.csv")
    )
    # get locations
    omekas_mapping["locations"] = utilities.csv2dict(
        os.path.join(mappings_directory, "koha-omekas_mapping - locations.csv")
    )
    # get research groups
    omekas_mapping["research_groups"] = utilities.csv2dict(
        os.path.join(mappings_directory, "koha-omekas_mapping - research_groups.csv")
    )

    # get researchers
    omekas_mapping["researchers"] = utilities.csv2dict(
        os.path.join(mappings_directory, "koha-omekas_mapping - researchers.csv")
    )

    # get projects
    omekas_mapping["projects"] = utilities.csv2dict(
        os.path.join(mappings_directory, "koha-omekas_mapping - projects.csv")
    )

    return omekas_mapping

omekas_session(api_url, key_identity, key_credential)

Parameters:

Name	Type	Description	Default
api_url	str	API URL for your Omeka S instance.	required
key_identity	str	Omeka S key identity for the user.	required
key_credential	str	Omeka S key credential for the user.	required

Returns:

Name	Type	Description
omekas_session		omeka_s_tools session

Source code in pyreslib/omekas.py

def omekas_session(api_url: str, key_identity: str, key_credential: str):
    """
    Args:
        api_url (str): API URL for your Omeka S instance.
        key_identity (str): Omeka S key identity for the user.
        key_credential (str): Omeka S key credential for the user.

    Returns:
        omekas_session: `omeka_s_tools` session
    """
    omekas_session = OmekaAPIClient(
        api_url=omekas_credentials["api_url"],
        key_identity=omekas_credentials["key_identity"],
        key_credential=omekas_credentials["key_credential"],
    )
    return omekas_session