sharepoint¶

@functools.cache
def get_token_manager() -> TokenManager:
    """
    Returns the process-wide `TokenManager` singleton, creating it on first call.

    Returns:
        TokenManager: The cached `TokenManager` instance.
    """
    manager = _create_token_manager()
    return manager

def get_list_id(token_manager: TokenManager, site_id: str, list_name: str) -> str:
    """Return the GUID of the named SharePoint list.

    Args:
        token_manager: Authenticated token manager.
        site_id: SharePoint site identifier.
        list_name: Display name of the target list.

    Returns:
        The list's Graph API GUID string.
    """
    url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_name}"
    headers = {"Authorization": f"Bearer {token_manager.get_token()}"}
    response = requests.get(url, headers=headers, timeout=60)
    response.raise_for_status()
    return cast(str, response.json()["id"])

def get_site_id(token_manager: TokenManager, domain: str, site_name: str) -> Any:
    """Return the GUID of a SharePoint site given its domain and site path.

    Args:
        token_manager: Authenticated token manager.
        domain: SharePoint tenant domain (e.g. `contoso.sharepoint.com`).
        site_name: The site's path segment, as used in its URL.

    Returns:
        The site's Graph API GUID string.
    """
    url = f"https://graph.microsoft.com/v1.0/sites/{domain}:/sites/{site_name}"  # Obtain the ID of site from SharePoint
    headers = {"Authorization": f"Bearer {token_manager.get_token()}"}
    response = requests.get(url, headers=headers, timeout=60)
    response.raise_for_status()
    return response.json()["id"]

def upload_file(
    token_manager: TokenManager,
    drive_id: str,
    target_folder: Path,
    local_file_path: Path,
    remote_file_name: str | None = None,
) -> str:
    """Upload a local file to SharePoint, returning the uploaded item's id
    (needed by callers that go on to generate a sharing link for it).

    The uploaded item is named `local_file_path.name` unless `remote_file_name`
    is given — needed when the local filename doesn't match what the remote
    folder's naming convention expects (e.g. `download`'s iMarina fallback,
    which reads under the fixed local name `iMarina.xlsx` but re-uploads it
    under the original datetime-encoded filename so `runtime/imarina` stays
    consistent with the other remote folders' naming)."""

    remote_path = target_folder / (remote_file_name or local_file_path.name)
    logger.info(f"Uploading from local path {local_file_path} to {remote_path}")
    url = f"https://graph.microsoft.com/v1.0/drives/{drive_id}/root:/{remote_path}:/content?%40microsoft.graph.conflictBehavior=replace"
    headers = {
        "Authorization": f"Bearer {token_manager.get_token()}",
        "Content-Type": "application/octet-stream",
    }
    try:
        with open(local_file_path, "rb") as f:

            response = requests.put(url, headers=headers, data=f, timeout=300)

        if response.status_code in (200, 201):
            logger.info(
                f"File '{local_file_path.name}' uploaded successfully to {remote_path}."
            )
        else:
            response.raise_for_status()
    except requests.exceptions.HTTPError:
        if response.status_code == HTTPStatus.NOT_FOUND:
            logger.exception(
                f"Destination folder does not exist ({remote_path}) in SharePoint."
            )
        else:
            logger.exception(f"HTTP error uploading '{local_file_path.name}'")
        raise
    except Exception:
        logger.exception(f"Unexpected error uploading '{local_file_path.name}'")
        raise

    logger.info("Upload done")
    return cast(str, response.json()["id"])

def download_item_content(
    token_manager: TokenManager, drive_id: str, item_id: str, destination: Path
) -> None:
    """Download a driveItem's content by id, writing it to `destination`."""
    url = f"https://graph.microsoft.com/v1.0/drives/{drive_id}/items/{item_id}/content"
    headers = {"Authorization": f"Bearer {token_manager.get_token()}"}
    response = requests.get(url, headers=headers, timeout=300)
    response.raise_for_status()
    with open(destination, "wb") as f:
        f.write(response.content)

def download_shared_link_content(
    token_manager: TokenManager, url: str, destination: Path
) -> None:
    """Download the file behind an MS List "sharing link" field value (the
    encoding Graph expects for its `/shares/u!{...}` endpoint), writing it to
    `destination`. Used for the two user-supplied input links (`download`)
    and for sourcing `publish`'s file from a request's output link.
    """
    encoded = base64.b64encode(url.encode()).decode()
    encoded = encoded.rstrip("=").replace("/", "_").replace("+", "-")
    download_url = (
        f"https://graph.microsoft.com/v1.0/shares/u!{encoded}/driveItem/content"
    )
    headers = {"Authorization": f"Bearer {token_manager.get_token()}"}
    response = requests.get(
        download_url, headers=headers, allow_redirects=True, timeout=300
    )
    response.raise_for_status()
    with open(destination, "wb") as f:
        f.write(response.content)

def download_files_in_folder_from_sharepoint(
    drive_id: str, local_destiny_folder: Path, remote_origin_folder: Path
) -> Any:
    """Downloads every file in a SharePoint folder into a local folder, flat.

    This is `download`'s bulk-sync mechanism for the static translation-
    dictionary files (`SHAREPOINT_INPUT_DIR`, see CLAUDE.md's `download`
    section) — it lists and downloads all `.xlsx` files as-is, unconditionally.
    A per-file download failure is logged and skipped rather than aborting
    the whole sync.

    Args:
        drive_id: SharePoint drive identifier the folders live under.
        local_destiny_folder: Local folder to download files into (created
            if it doesn't exist).
        remote_origin_folder: Remote SharePoint folder path to list and
            download files from.

    Raises:
        SharePointError: If listing the remote folder's children fails.
    """
    token_manager = get_token_manager()

    local_destiny_folder.mkdir(parents=True, exist_ok=True)

    url_list = f"https://graph.microsoft.com/v1.0/drives/{drive_id}/root:/{remote_origin_folder}:/children"
    headers = {"Authorization": f"Bearer {token_manager.get_token()}"}

    response = requests.get(url_list, headers=headers, timeout=30)

    if response.status_code != HTTPStatus.OK:
        raise SharePointError(response.status_code, response.text)

    items = response.json().get("value", [])
    files_to_download = [f for f in items if f.get("file")]

    if not files_to_download:
        logger.warning("No files to download in the SharePoint path.")
        return

    logger.info(f"Found {len(files_to_download)} files. Downloading...")

    for remote_file in files_to_download:
        name = remote_file["name"]
        try:
            download_item_content(
                token_manager, drive_id, remote_file["id"], local_destiny_folder / name
            )
            logger.debug(f"{name} saved successfully.")
        except requests.exceptions.HTTPError:
            logger.exception(f"Error downloading {name}")

def get_parameters_list(operation_id: str) -> tuple[str | None, str | None]:
    """Reads an MS List item's A3/iMarina input sharing links, by Operation ID.

    Args:
        operation_id: The MS List item's ID (the workflow's Operation ID).

    Returns:
        A `(a3_link, imarina_link)` tuple; either may be `None` if that
        field isn't set (both are optional per CLAUDE.md's "Where the raw
        inputs come from" section, and `download` falls back to selecting
        the latest remote file when they are).
    """
    fields = _get_list_item_fields(operation_id)

    return fields.get(FIELD_A3_EXCEL_INPUT_LINK), fields.get(
        FIELD_IMARINA_EXCEL_INPUT_LINK
    )

def get_list_item_link_field(operation_id: str, field_name: str) -> str | None:
    """Read a single Text-type link field off an MS List item, or None if
    that field isn't set. Used by `publish` to source its file from the
    request's output-link field when given an ID (see CLAUDE.md's `publish`
    section)."""
    return cast(str | None, _get_list_item_fields(operation_id).get(field_name))

def update_list_item_fields(operation_id: str, fields: dict[str, str]) -> None:
    """PATCH one or more fields on an MS List item.

    Graph's `/fields` sub-resource takes the field dict directly as the PATCH
    body (unlike `_get_list_item_fields`'s GET, which reads the whole item and
    pulls a `fields` object back out of the response).
    """
    token_manager = get_token_manager()
    site_id, list_name = _resolve_site_and_list()

    url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_name}/items/{operation_id}/fields"
    response = requests.patch(
        url,
        headers={
            "Authorization": f"Bearer {token_manager.get_token()}",
            "Content-Type": "application/json",
        },
        json=fields,
        timeout=60,
    )
    response.raise_for_status()

def create_sharing_link(
    token_manager: TokenManager, drive_id: str, item_id: str
) -> str:
    """Create an organization-scoped, view-only sharing link for a SharePoint
    driveItem. Scope is deliberately "organization", not "anonymous" -- these
    links point at personnel data (see CLAUDE.md's "GDPR implications" section)."""
    url = (
        f"https://graph.microsoft.com/v1.0/drives/{drive_id}/items/{item_id}/createLink"
    )
    headers = {"Authorization": f"Bearer {token_manager.get_token()}"}
    response = requests.post(
        url,
        headers=headers,
        json={"type": "view", "scope": "organization"},
        timeout=60,
    )
    response.raise_for_status()
    return cast(str, response.json()["link"]["webUrl"])

def select_latest_remote_file(
    token_manager: TokenManager, drive_id: str, remote_folder: Path, suffix: str
) -> RemoteFile:
    """List `remote_folder`'s children and pick the latest .xlsx by the
    filename-encoded datetime (CLAUDE.md's "What this does" section: "the
    'latest' of a group of files is always deduced from the file's name").
    Remote counterpart of `select_file_to_upload` (core/file_select.py),
    which does the same thing against local files -- unlike that function,
    there is no modification-time fallback here: the rule for remote folders
    is filename-datetime only, and a SharePoint item's
    `lastModifiedDateTime` doesn't carry the same "file was produced at"
    meaning local mtime does for the local fallback.
    """
    url = f"https://graph.microsoft.com/v1.0/drives/{drive_id}/root:/{remote_folder}:/children"
    headers = {"Authorization": f"Bearer {token_manager.get_token()}"}
    response = requests.get(url, headers=headers, timeout=30)
    if response.status_code != HTTPStatus.OK:
        raise SharePointError(response.status_code, response.text)

    items = response.json().get("value", [])
    excel_files = [f for f in items if f.get("file") and f["name"].endswith(".xlsx")]
    if not excel_files:
        raise NoExcelFilesFoundError

    dated_files: list[tuple[datetime.datetime, dict[str, Any]]] = []
    for item in excel_files:
        parsed_dt = parse_datetime_from_filename(item["name"], suffix)
        if parsed_dt is not None:
            dated_files.append((parsed_dt, item))

    if not dated_files:
        raise NoExcelFilesFoundError

    dated_files.sort(key=lambda x: x[0], reverse=True)
    chosen = dated_files[0][1]
    return RemoteFile(id=chosen["id"], name=chosen["name"])