download_controller(ctx: Context, id_element: OperationIdOpt, input_dir: DirectoryOpt = LOCAL_INPUT_DIR) -> None
Implements the download CLI command: populates input_dir with every file build expects.
Bulk-syncs the static translation-dictionary files from SharePoint, then downloads the A3 dump and previous iMarina upload from the MS List item's sharing links, falling back to the latest file in the corresponding SharePoint folder (_fallback_a3/_fallback_imarina) for whichever link is missing. Exits with code 1 if any of REQUIRED_INPUT_FILES is still missing afterward.
Parameters:
| Name | Type | Description | Default |
ctx | Context | Typer's invocation context (unused directly; required so Typer's --help machinery can populate it). | required |
id_element | OperationIdOpt | The request's Operation ID; used to look up its input links and to keep its Workflow State field in sync (best-effort). | required |
input_dir | DirectoryOpt | Local directory to populate. | LOCAL_INPUT_DIR |
Raises:
| Type | Description |
Exit | With code 1 if any required input file is still missing once every download/fallback attempt has run. |
Source code in source/src/imarina_load_researchers/commands/download/cli.py
| def download_controller(
ctx: typer.Context,
id_element: OperationIdOpt,
input_dir: DirectoryOpt = LOCAL_INPUT_DIR,
) -> None:
"""
Implements the `download` CLI command: populates `input_dir` with every
file `build` expects.
Bulk-syncs the static translation-dictionary files from SharePoint, then
downloads the A3 dump and previous iMarina upload from the MS List
item's sharing links, falling back to the latest file in the
corresponding SharePoint folder (`_fallback_a3`/`_fallback_imarina`) for
whichever link is missing. Exits with code 1 if any of
`REQUIRED_INPUT_FILES` is still missing afterward.
Args:
ctx (typer.Context): Typer's invocation context (unused directly;
required so Typer's `--help` machinery can populate it).
id_element (OperationIdOpt): The request's Operation ID; used to
look up its input links and to keep its Workflow State field
in sync (best-effort).
input_dir (DirectoryOpt): Local directory to populate.
Raises:
typer.Exit: With code 1 if any required input file is still missing
once every download/fallback attempt has run.
"""
logger.info(f"Starting download of input files from SharePoint into: {input_dir}")
try:
update_list_item_fields(
str(id_element), {FIELD_WORKFLOW_STATE: WorkflowState.PREPARING}
)
except Exception:
# Best-effort bookkeeping: must not block the actual download below.
logger.exception("Error updating Workflow State to Preparing")
drive_id = read_secret(SecretName.DRIVE_ID)
try:
download_files_in_folder_from_sharepoint(
drive_id, input_dir, Path(SHAREPOINT_INPUT_DIR)
)
logger.info(
f"DONE: Input files successfully downloaded to local directory: {input_dir}"
)
except Exception:
# Intentionally broad: this step's real success/failure is verified by the
# missing-file check below, which is what actually fails the pipeline.
logger.exception("Error downloading input files from SharePoint")
try:
# Function get_parameters_list and download the links(url) of Excels (A3 Excel and iMarina Excel)
a3_link, imarina_link = get_parameters_list(str(id_element))
token_manager = get_token_manager() # get token
if a3_link:
download_shared_link_content(token_manager, a3_link, input_dir / "A3.xlsx")
logger.info("A3.xlsx download successful")
else:
logger.warning("URL not found for A3.xlsx, falling back to latest A3 dump")
_fallback_a3(token_manager, drive_id, id_element, input_dir)
if imarina_link:
download_shared_link_content(
token_manager, imarina_link, input_dir / "iMarina.xlsx"
)
logger.info("iMarina.xlsx download successful")
else:
logger.warning(
"URL not found for iMarina.xlsx, falling back to latest published file"
)
_fallback_imarina(token_manager, drive_id, id_element, input_dir)
except Exception:
# Intentionally broad: same rationale as above, the missing-file check below
# is the actual pass/fail signal for this pipeline stage.
logger.exception("Error getting parameters for MS List")
missing = [
filename
for filename in REQUIRED_INPUT_FILES.values()
if not (input_dir / filename).exists()
]
if missing:
logger.error(f"Missing required input file(s) in {input_dir}:")
for filename in missing:
logger.error(f" - {filename}")
raise typer.Exit(code=1)
logger.info(f"All required input files are present in {input_dir}")
|