Skip to content

imarina_excel

imarina_excel

normalized_dni

normalized_dni(dni: str) -> str

Normalizes a DNI/NIE/NIF into a comparable canonical form.

Strips everything but letters and digits, uppercases the result, and drops leading zeros, so equivalent DNIs written with different formatting (dashes, spaces, leading zeros) compare equal.

Parameters:

Name Type Description Default
dni str

The raw DNI/NIE/NIF value.

required

Returns:

Name Type Description
str str

The normalized value, or "" if dni is falsy.

Source code in source/src/imarina_load_researchers/core/imarina_excel.py
def normalized_dni(dni: str) -> str:
    """
    Normalizes a DNI/NIE/NIF into a comparable canonical form.

    Strips everything but letters and digits, uppercases the result, and
    drops leading zeros, so equivalent DNIs written with different
    formatting (dashes, spaces, leading zeros) compare equal.

    Args:
        dni (str): The raw DNI/NIE/NIF value.

    Returns:
        str: The normalized value, or `""` if `dni` is falsy.
    """
    if not dni:
        return ""
    dni = re.sub(r"[^0-9A-Za-z]", "", str(dni)).upper()
    dni = dni.lstrip("0")  # remove leading zeros
    return dni

build_upload_excel

build_upload_excel(output_path: Path, imarina_path: Path, a3_path: Path, translation_paths: TranslationDictionaryPaths) -> None

Build the next iMarina upload as an update on top of the previous one.

See _match_last_upload_against_a3's docstring for the exact per-case policy (left ICIQ / changed position) and _find_new_researchers_in_a3 for new hires.

Source code in source/src/imarina_load_researchers/core/imarina_excel.py
def build_upload_excel(
    output_path: Path,
    imarina_path: Path,
    a3_path: Path,
    translation_paths: TranslationDictionaryPaths,
) -> None:
    """Build the next iMarina upload as an update on top of the previous one.

    See _match_last_upload_against_a3's docstring for the exact per-case
    policy (left ICIQ / changed position) and _find_new_researchers_in_a3
    for new hires.
    """

    # Get A3 data
    a3_data = Excel(a3_path, skiprows=2, header=0)

    # Get iMarina last upload data
    im_data = Excel(imarina_path, header=0)

    # load the translators fields: country, job_description
    translator = build_translations(translation_paths)

    im_researchers = []
    for _index, row in im_data.dataframe.iterrows():
        im_researchers.append(parse_imarina_row_data(row))

    a3_researchers = []
    for _index, row in a3_data.dataframe.iterrows():
        a3_researchers.append(parse_a3_row_data(row, translator))

    logger.info(
        "Phase 1: Check if the researchers in last upload to iMarina are still in A3"
    )
    researchers_left, researchers_changed, researchers_output = (
        _match_last_upload_against_a3(im_researchers, a3_researchers)
    )

    logger.info("Phase 2: Add researchers in A3 that are not present in iMarina")
    researchers_new, researchers_output_phase2 = _find_new_researchers_in_a3(
        a3_researchers, im_researchers
    )
    researchers_output.extend(researchers_output_phase2)

    # The A3 snapshot has more people than iMarina needs, because visitors
    # aren't meant to be loaded into iMarina - see Researcher.is_visitor for
    # how a visitor is identified (center code 4, with ICREA/CSC-predoc
    # exceptions).
    #
    # NOTE: researchers_visitor is currently only used to report a count in
    # the log line below; visitors are not actually excluded from
    # researchers_output/the generated file today. Flagging this since it
    # doesn't match the "visitors aren't loaded" policy above - fix if
    # that's unintentional.
    researchers_visitor = [
        researcher for researcher in researchers_output if researcher.is_visitor()
    ]

    num_changed = len(researchers_changed)
    num_left = len(researchers_left)
    num_new = len(researchers_new)
    num_visitors = len(researchers_visitor)

    logger.info(
        f"Since the last upload, {num_changed} researchers have changed its position within ICIQ."
    )
    logger.info(f"Since the last upload, {num_visitors} researchers have visited ICIQ.")
    logger.info(f"Since the last upload, {num_left} researchers have left ICIQ.")
    logger.info(f"Since the last upload, {num_new} researchers have entered ICIQ.")

    # IF GROUP UNIT = DIRECCIO OR GROUP UNIT = GESTIO OR GROUP UNIT = OUTREACH DELETE OF OUTPUT

    # Build the output file starting from an empty copy of the previous
    # iMarina upload, so it retains its columns, types, and headers.
    im_data_any = cast(
        Any, im_data
    )  # data cast pass variable type to any type to use copy method
    im_data_empty: Any = im_data_any.__copy__()
    im_data_empty.empty()

    excel_output = im_data_empty.__copy__()
    append_researchers_to_output_data(researchers_output, excel_output)
    excel_output.to_excel(Path(output_path))