a3_mapper¶

def transform_orcid(orcid: str) -> str:
    """
    Formats a bare 16-digit ORCID into its dashed `xxxx-xxxx-xxxx-xxxx` form.

    A3 stores ORCIDs with no dashes; iMarina expects them dashed. Values that
    already contain a dash are passed through unchanged (already formatted,
    or not a plain digit string this function can safely reformat).

    Args:
        orcid (str): The raw ORCID value from the A3 dump.

    Returns:
        str: The dashed ORCID, the input unchanged if it already has a dash,
            or `""` if `orcid` is falsy.
    """
    if not orcid or orcid == "":
        return ""
    if "-" in orcid:
        return orcid
    orcid = orcid.strip()
    ret = ""
    for counter, char in enumerate(orcid):
        if counter % 4 == 0 and counter != 0:
            ret += "-"
        ret += char
    logger.trace(f"Transform ORCID input is {orcid} and output is {ret}")
    return ret

def normalize_country_name(name: str) -> str:
    """
    Normalizes a country name into a stable lookup key for the translator dict.

    Strips non-breaking/zero-width spaces and parentheses, removes digits and
    accents, and lowercases the result, so that spelling/formatting
    variations in the A3 dump and the `countries.xlsx` dictionary still
    resolve to the same key.

    Args:
        name (str): The raw country name to normalize.

    Returns:
        str: The normalized name, or `""` if `name` is not a string.
    """
    if not isinstance(name, str):
        return ""
    name = (
        name.replace("\xa0", " ")
        .replace("\u200b", " ")
        .replace("(", "")
        .replace(")", "")
        .strip()
    )
    name = re.sub(r"\d+", "", name)  # remove numbers
    name = "".join(
        c
        for c in unicodedata.normalize("NFD", name)  # remove accents
        if unicodedata.category(c) != "Mn"
    )
    return name.lower().strip()

def parse_a3_row_data(row: Any, translator: Translator) -> Any:
    """
    Converts one A3 dump row into a `Researcher`, applying every A3→iMarina
    translation (country, unit group→entity, job description, sex, ORCID
    formatting, name normalization, date sanitizing).

    Args:
        row (Any): A row (`pandas.Series`) from the A3 input dataframe,
            indexed by `A3Field.value` column names.
        translator (Translator): The `{A3Field: {raw_value: translated_value}}`
            dictionaries built by `translations.build_translations()`.

    Returns:
        Researcher: The researcher built from this A3 row, with every field
            iMarina needs populated (fields with no A3 source default to `""`).

    Raises:
        KeyError: If `row`'s unit group, job description or sex value has no
            entry in the corresponding translator dictionary.
    """
    # translator[A3Field.COUNTRY] is pre-normalized by build_translations(), so it can
    # be used directly here without rebuilding it on every row.
    translator_countries = translator[A3Field.COUNTRY]

    born_country_raw = str(
        row[A3Field.BORN_COUNTRY.value]
    ).strip()  # Read the value of the BORN_COUNTRY column for this row. Remove any spaces
    born_country_clean = normalize_country_name(
        born_country_raw
    )  # born country normalize and find in translator_countries

    born_country_clean = MANUAL_COUNTRY_ALIASES.get(
        born_country_clean, born_country_clean
    )

    born_country = translator_countries.get(
        born_country_clean, born_country_clean.capitalize()
    )  # born country fully translated

    country_raw = str(row[A3Field.COUNTRY.value]).strip()
    country_clean = normalize_country_name(country_raw)  #  country normalized

    country_clean = MANUAL_COUNTRY_ALIASES.get(country_clean, country_clean)

    country = translator_countries.get(
        country_clean, country_clean.capitalize()
    )  # country fully translated

    logger.debug(f"Raw born_country: {born_country_raw}")
    logger.debug(f"Clean born_country: {born_country_clean}")
    logger.debug(f"Translated born_country: {born_country}")
    logger.debug(f"Raw country: {country_raw}")
    logger.debug(f"Clean country: {country_clean}")
    logger.debug(f"Translated country: {country}")

    email_val = get_val(row, A3Field.EMAIL.value)
    if email_val is not None:
        email_val = email_val.lower()

    # Translates unit_group into entity
    try:
        entity_val = translator[A3Field.UNIT_GROUP][row[A3Field.UNIT_GROUP.value]]
    except KeyError:
        logger.exception(f"KeyError in UNIT_GROUP: {row[A3Field.UNIT_GROUP.value]!r}")
        raise

    personal_web_val = translator[A3Field.PERSONAL_WEB][entity_val]

    orcid_val = get_val(row, A3Field.ORCID.value)
    if orcid_val is None:
        orcid_val = ""

    try:
        job_description_val = translator[A3Field.JOB_DESCRIPTION][
            row[A3Field.JOB_DESCRIPTION.value]
        ]
    except KeyError:
        logger.exception(
            f"KeyError in JOB DESCRIPTION: {row[A3Field.JOB_DESCRIPTION.value]!r}"
        )
        raise

    # The job description is one of the special job descriptions that are used to determine the entity
    if row[A3Field.JOB_DESCRIPTION.value] in translator[A3Field.JOB_DESCRIPTION_ENTITY]:
        logger.debug(
            f"Special job description found, translating to entity. entity_val was going to be: {entity_val!s}"
        )
        entity_val = translator[A3Field.JOB_DESCRIPTION_ENTITY][
            row[A3Field.JOB_DESCRIPTION.value]
        ]
        logger.debug(f"Entity translated is: {entity_val!s}")

    # Special case for ICREA group leaders, which needs also info from group unit field
    if row[A3Field.UNIT_GROUP.value] == "ICREA":
        job_description_val = JOB_TITLE_GROUP_LEADER_ICREA

    try:
        sex_val = translator[A3Field.SEX][row[A3Field.SEX.value]]
    except KeyError:
        logger.exception(f"KeyError in SEX: {row[A3Field.SEX.value]!r}")
        raise

    data = Researcher(
        code_center=row[A3Field.CODE_CENTER.value],
        dni=row[A3Field.DNI.value],
        email=email_val,
        orcid=transform_orcid(orcid_val),
        name=normalize_name(row[A3Field.NAME.value]),
        surname=normalize_name(row[A3Field.SURNAME.value]),
        second_surname=normalize_name(row[A3Field.SECOND_SURNAME.value]),
        ini_date=sanitize_date(row[A3Field.INI_DATE.value]),
        end_date=sanitize_date(row[A3Field.END_DATE.value]),
        ini_prorrog=sanitize_date(row[A3Field.INI_PRORROG.value]),
        end_prorrog=sanitize_date(row[A3Field.END_PRORROG.value]),
        date_termination=sanitize_date(row[A3Field.DATE_TERMINATION.value]),
        sex=sex_val,
        personal_web=personal_web_val,
        signature="",
        signature_custom="",
        country=country,
        born_country=born_country,
        job_description=job_description_val,
        unit_group=entity_val,
        entity_type=translator[A3Field.ENTITY_TYPE][entity_val],
        google_scholar_id="",
        adscription_type="",
        entity_country="",
        entity_community="",
        entity_province="",
        entity_city="",
        entity_postal_code="",
        entity_address="",
        entity_web="",
        contact_phone="",
        scopus_id="",
    )
    return data