Download OneDrive File

onedrive_download_file.py

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "aiohttp",
#     "yarl",
# ]
# ///
from __future__ import annotations

import argparse
import asyncio
import logging
import json
from dataclasses import dataclass
from pathlib import Path

from aiohttp import ClientSession, ClientTimeout
from yarl import URL

logging.basicConfig(level=logging.DEBUG, format="%(levelname)s - %(message)s")

API_ENTRYPOINT = URL("https://api.onedrive.com/v1.0/drives/")
PERSONAL_API_ENTRYPOINT = URL("https://my.microsoftpersonalcontent.com/_api/v2.0/shares/")
SHARE_LINK_HOST = "1drv.ms"
DEFAULT_HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:135.0) Gecko/20100101 Firefox/135.0"}
TIMEOUTS = ClientTimeout(total=60, connect=30)
DOWNLOAD_FOLDER = Path("OneDrive_Downloads")
BADGER_URL = URL("https://api-badgerp.svc.ms/v1.0/token")

# Default app details used in browsers by unautenticated sessions
APP_ID = "1141147648"
APP_UUID = "5cbed6ac-a083-4e14-b191-b4ba07653de2"


@dataclass(frozen=True, slots=True)
class AccessDetails:
    container_id: str
    resid: str
    auth_key: str
    redeem: str

    @classmethod
    def from_url(cls, direct_url: URL) -> AccessDetails:
        resid = direct_url.query.get("resid") or ""  # ex: ABCXYZ000!12345
        redeem = direct_url.query.get("redeem") or ""
        auth_key = direct_url.query.get("authkey") or ""
        id_ = direct_url.query.get("id") or ""
        container_id = direct_url.query.get("cid")
        if not resid and "!" in id_:
            resid = id_
        if not container_id:
            container_id = resid.split("!")[0]

        return AccessDetails(container_id, resid, auth_key, redeem)


async def process_url(url: URL) -> None:
    try:
        async with ClientSession(
            headers=DEFAULT_HEADERS,
            raise_for_status=False,
            timeout=TIMEOUTS,
        ) as client_session:
            # ex: https://1drv.ms/t/s!ABCJKL-ABCJKL?e=ABC123 or  https://1drv.ms/t/c/a12345678/aTOKEN?e=ABC123
            if is_share_link(url):
                async with client_session.get(url, allow_redirects=True) as response:
                    url = response.url

            # ex: https://onedrive.live.com/?id=ABCXYZ!12345&cid=ABC0123BVC
            await download(client_session, url)

    except Exception as e:
        msg = f"Download Failed: {e}"
        logging.error(msg)


async def download(client_session: ClientSession, url: URL) -> None:
    access_details = AccessDetails.from_url(url)
    if access_details.redeem:
        await get_badger_token(client_session)

    api_url = create_api_url(access_details)
    async with client_session.get(api_url, raise_for_status=True) as response:
        json_resp: dict = await response.json()

    filename: str = json_resp["name"]
    download_url = URL(json_resp["@content.downloadUrl"])
    download_path = DOWNLOAD_FOLDER / filename
    await download_file(client_session, download_url, download_path)


async def get_badger_token(client_session: ClientSession) -> None:
    new_headers = dict(client_session.headers) | {"AppId": APP_ID}
    data = {"appId": APP_UUID}

    async with client_session.post(BADGER_URL, headers=new_headers, raise_for_status=True, json=data) as response:
        json_resp: dict = await response.json()

    logging.info(f"\n Badger token response: \n{json.dumps(json_resp, indent=4)}")
    token: str = json_resp["token"]
    auth_headers = {"Prefer": "autoredeem", "Authorization": f"Badger {token}"}
    client_session.headers.update(auth_headers)


async def download_file(client_session: ClientSession, url: URL, output: Path) -> None:
    async with client_session.get(url, raise_for_status=True) as response:
        output.parent.mkdir(parents=True, exist_ok=True)
        with output.open("wb") as f:
            async for chunk in response.content.iter_chunked(1024):
                f.write(chunk)
        msg = f"File downloaded successfully to {output.resolve()}"
        logging.info(msg)


# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~


def create_api_url(access_details: AccessDetails) -> URL:
    if access_details.redeem:
        return PERSONAL_API_ENTRYPOINT / f"u!{access_details.redeem}" / "driveitem"
    api_url = API_ENTRYPOINT / access_details.container_id / "items" / access_details.resid
    if access_details.auth_key:
        return api_url.with_query(authkey=access_details.auth_key)
    return api_url


def is_share_link(url: URL) -> bool:
    return url.host == SHARE_LINK_HOST and any(p in url.parts for p in ("f", "t", "u"))


def main() -> None:
    try:
        parser = argparse.ArgumentParser()
        parser.add_argument("url", type=str, help="URL to download")
        args = parser.parse_args()
        url = URL(args.url)
        asyncio.run(process_url(url))
    except KeyboardInterrupt:
        pass
    except Exception:
        logging.exception("Download Failed")


if __name__ == "__main__":
    main()