11 — Coercing OneDrive API

#!/usr/bin/env python3
import requests

TOKEN = "[..token..]"
USERAGENT = "[..user agent..]"

def get_me():
    headers = {"authorization": f"bearer {TOKEN}", "user-agent": USERAGENT}
    r = requests.get("https://graph.microsoft.com/v1.0/me/drive", headers=headers)
    if r.status_code == 200:
        return r.json()
    else:
        print(r.json())
        r.raise_for_status()

if __name__ == "__main__":
    print(get_me())

#!/usr/bin/env python3
import requests
from os.path import expanduser

CLIENTID = "[..client id..]"
USERAGENT = "[..user agent..]"
REFRESHTOKENFILE = expanduser("~/.config/onedrive/refresh_token")
refreshtoken = ""

with open(REFRESHTOKENFILE, "r") as f:
    refreshtoken = f.read()
    if not refreshtoken:
        raise RuntimeError("No refresh token")

def get_token():
    d = {"client_id": CLIENTID,
         "redirect_uri": "https://login.microsoftonline.com/common/oauth2/nativeclient",
         "refresh_token": refreshtoken,
         "grant_type": "refresh_token"}
    r = requests.post("https://login.microsoftonline.com/common/oauth2/v2.0/token",
                      headers={"user-agent": USERAGENT}, data=d)
    if r.status_code == 200:
        return r.json()
    else:
        print(r.json())
        r.raise_for_status()

if __name__ == "__main__":
    print(get_token())

{'token_type': 'Bearer', 'scope': 'Files.ReadWrite Files.ReadWrite.All', 'expires_in': 3599, 'ext_expires_in': 3599, 'access_token': '[..access token..]', 'refresh_token': '[..refresh token..]'}

#!/usr/bin/env python3
import re
import requests

CLIENTID = "[..client id..]"
USERAGENT = "[..user agent..]"
SCOPE = "Files.ReadWrite Files.ReadWrite.All Sites.ReadWrite.All offline_access"
DOMAIN = "https://login.microsoftonline.com/common/oauth2"

def auth():
    print("Please authorise this application by visiting the following URL:")
    d = {"client_id": CLIENTID,
         "redirect_uri": f"{DOMAIN}/nativeclient",
         "scope": SCOPE.replace(" ", "%20"),
         "response_type": "code",
         "prompt": "login"}
    print(f"{DOMAIN}/v2.0/authorize?" +
          "&".join("%s=%s" % (k, v) for k, v in d.items()))
    print("After completing the authorisation in your browser, copy the full "
          "redirect URI (from the address bar) and paste it below.")
    response = input("Paste redirect URI here: ")
    code = re.search(r"(?:[?&]code=)([^&]+)", response.strip())
    if not code:
        raise RuntimeError("An empty or invalid redirect URI was entered")
    d = {"client_id": CLIENTID,
         "redirect_uri": f"{DOMAIN}/nativeclient",
         "code": code.group(1),
         "grant_type": "authorization_code"}
    r = requests.post(f"{DOMAIN}/v2.0/token",
                      headers={"user-agent": USERAGENT},
                      data="&".join("%s=%s" % (k, v) for k, v in d.items()))
    if r.status_code == 200:
        return r.json()
    else:
        print(r.json())
        r.raise_for_status()

if __name__ == "__main__":
    print(auth())

{'token_type': 'Bearer', 'scope': 'Files.ReadWrite Files.ReadWrite.All', 'expires_in': 3599, 'ext_expires_in': 3599, 'access_token': '[..access token..]', 'refresh_token': '[..refresh token..]'}

GRAPHDOMAIN = "https://graph.microsoft.com/v1.0/me/drive"

def headers(token=None):
    """Return standard headers to send with requests."""
    if token:
        return {"authorization": f"bearer {token}", "user-agent": USERAGENT}
    return {"user-agent": USERAGENT}

def get_res(r, json=True):
    """Get json response of a request or print details and raise exception
    if it failed."""
    if r.status_code == 200:
        return r.json() if json else r.content
    else:
        try:
            print(r.json())
        except requests.exceptions.JSONDecodeError:
            print(f"No json body. Status code: {r.status_code}")
        r.raise_for_status()

def get_me(token):
    """Drive details."""
    r = requests.get(f"{GRAPHDOMAIN}", headers=headers(token))
    return get_res(r)

def get_me(token, sub=""):
    """Drive details."""
    r = requests.get(f"{GRAPHDOMAIN}{sub}", headers=headers(token))
    return get_res(r)

def get_children(token, subpath=""):
    """Drive top-level files, or files in SUBPATH.
    SUBPATH must not begin or end with /."""
    p = f"/root:/{subpath}:/children" if subpath else "/root/children"
    return get_me(token, p)

def get_next(token, data):
    """Resolve next link in data. Used for pagination."""
    r = requests.get(data['@odata.nextLink'], headers=headers(token))
    return get_res(r)

def get_all_children(token, subpath=""):
    """get_children wrapper with pagination handling."""
    data = get_children(token, subpath)
    entries = data.get("value", [])
    while "@odata.nextLink" in data:  # handle pagination
        data = get_next(token, data)
        entries.extend(data.get("value", []))
    return entries

def list_children(token, subpath=""):
    """List files/subfolders in root or under SUBPATH.
    Adapted from https://www.merge.dev/blog/onedrive-api-python"""
    entries = get_all_children(token, subpath)
    for entry in entries:
        what = "Folder" if "folder" in entry else "File"
        print(f"- {what}: {entry['name']} (ID: {entry['id']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")

def get_versions(token, itemid):
    """Data with versions of a file with given ITEMID."""
    return get_me(token, f"/items/{itemid}/versions")

def get_all_versions(token, itemid):
    """get_versions wrapper with pagination handling."""
    data = get_versions(token, itemid)
    entries = data.get("value", [])
    while "@odata.nextLink" in data:  # handle pagination
        data = get_next(token, data)
        entries.extend(data.get("value", []))
    return entries

def list_versions(token, itemid):
    """List versions for file with given itemid."""
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"No versions found for {itemid}")
        return
    for entry in entries:
        print(f"- {entry['id']} (ID) (Size: {entry['size']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")

def get_file_info(token, path):
    """Get information object for file at PATH.
    (Path relative to root, without starting or ending /)"""
    return get_me(token, f"/root:/{path}")

def get_itemid(token, path):
    """Get itemid of file at PATH.
    (Path relative to root, without starting or ending /)"""
    data = get_file_info(token, path)
    return data["id"]

def list_path_versions(token, path):
    """list_versions convenience wrapper to be able to give a path
    instead of itemid.
    (Path relative to root, without starting or ending /)"""
    list_versions(token, get_itemid(token, path))

def download_url(token, url, to):
    """Function for downloading a @microsoft.graph.downloadUrl URL."""
    r = requests.get(url)
    with open(to, 'wb') as f:
        f.write(get_res(r, json=False))
        print(f"Downloaded to '{to}'")

def download_file(token, path, to):
    """Download file at OneDrive PATH to local path TO.
    (PATH relative to the root of the drive, without starting or ending /)"""
    data = get_file_info(token, path)
    download_url(token, data["@microsoft.graph.downloadUrl"], to)

def download_file_version(token, itemid, versionid, to):
    """Download version VERSIONID of item with given ITEMID to local path
    TO."""
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"No versions found for {itemid}")
        return
    for entry in entries:
        if entry["id"] == versionid:
            download_url(token, entry["@microsoft.graph.downloadUrl"], to)
            return
    raise RuntimeError(f"No version with ID {versionid} found")

def download_file_version_before_date(token, itemid, date, to):
    """Download version before DATE of item with given ITEMID to local path
    TO. DATE should be in a format supported by dateutil.parser.parse, e.g.
    '2026-04-04 10:30'. If timezone is not provided it's assumed to be the
    local timezone."""
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"No versions found for {itemid}")
        return
    date = dateutil.parser.parse(date)
    if not date.tzinfo:    # can't compare timezone-aware with unaware
        date = date.astimezone()
    for entry in entries:
        if dateutil.parser.parse(entry["lastModifiedDateTime"]) < date:
            print(f"Downloading version {entry['id']} from",
                  entry["lastModifiedDateTime"])
            download_url(token, entry["@microsoft.graph.downloadUrl"], to)
            return
    raise RuntimeError(f"No version before date {date} found")

def list_versions(token, of):
    """List versions for file with given itemid or path.
    (Path relative to root, with starting /)"""
    is_path = of[0] == "/"
    itemid = get_itemid(token, of) if is_path else of
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"No versions found for {itemid}")
        return
    for entry in entries:
        print(f"- {entry['id']} (ID) (Size: {entry['size']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")

def download_file_version(token, of, versionid, to):
    """Download version VERSIONID of item with given itemid or path OF
    to local path TO.
    (Path relative to root, with starting /)"""
    is_path = of[0] == "/"
    itemid = get_itemid(token, of) if is_path else of
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"No versions found for {itemid}")
        return
    for entry in entries:
        if entry["id"] == versionid:
            download_url(token, entry["@microsoft.graph.downloadUrl"], to)
            return
    raise RuntimeError(f"No version with ID {versionid} found")

def download_file_version_before_date(token, of, date, to):
    """Download version before DATE of item with given itemid or path OF to
    local path TO. DATE should be in a format supported by
    dateutil.parser.parse, e.g. '2026-04-04 10:30'. If timezone is not
    provided it's assumed to be the local timezone."""
    is_path = of[0] == "/"
    itemid = get_itemid(token, of) if is_path else of
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"No versions found for {itemid}")
        return
    date = dateutil.parser.parse(date)
    if not date.tzinfo:    # can't compare timezone-aware with unaware
        date = date.astimezone()
    for entry in entries:
        if dateutil.parser.parse(entry["lastModifiedDateTime"]) < date:
            print(f"Downloading version {entry['id']} from",
                  entry["lastModifiedDateTime"])
            download_url(token, entry["@microsoft.graph.downloadUrl"], to)
            return
    raise RuntimeError(f"No version before date {date} found")

def find_versionid_in_entries(token, versionid, entries):
    """Return entry with version VERSIONID in entries array ENTRIES."""
    for entry in entries:
        if entry["id"] == versionid:
            return entry
    raise RuntimeError(f"No version with ID {versionid} found")

def find_version_before_date_in_entries(token, date, entries):
    """Return entry before DATE in entries array ENTRIES.
    (Date in a format supported by dateutil.parser.parse, e.g.
    '2026-04-04 10:30'. If timezone is not provided it's assumed to be the
    local timezone.)"""
    date = dateutil.parser.parse(date)
    if not date.tzinfo:  # can't compare timezone-aware with unaware
        date = date.astimezone()
    for entry in entries:
        if dateutil.parser.parse(entry["lastModifiedDateTime"]) < date:
            return entry
    raise RuntimeError(f"No version before date {date} found")

def download_file_version(token, of, ver, to):
    """Download version VER of item with given itemid or path OF
    to local path TO.
    VER can be either the version id or a date which the version is older than
    (downloads version before date).
    (Path relative to root, with starting /)
    (Date in a format supported by dateutil.parser.parse, e.g.
    '2026-04-04 10:30'. If timezone is not provided it's assumed to be the
    local timezone.)"""
    is_path = of[0] == "/"
    itemid = get_itemid(token, of) if is_path else of
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"No versions found for {itemid}")
        return
    ver_is_date = re.match(r"[0-9]+\.0", ver) is None
    entry = find_version_before_date_in_entries(token, ver, entries) \
        if ver_is_date \
        else find_versionid_in_entries(token, ver, entries)
    print(f"Downloading version {entry['id']} from",
          entry["lastModifiedDateTime"])
    download_url(token, entry["@microsoft.graph.downloadUrl"], to)

def download_file(token, file, dest="", ver=None):
    """Download FILE given its id or path. It will be downloaded to pwd with
    the same name as on OneDrive unless a destination path DEST is provided.
    A versionid or date VER can be specified to download that version or most
    recent version before date.
    PATH should be relative to root and start with a /.
    DATE should be in a format supported by dateutil.parser.parse, e.g.
    '2026-04-04 10:30'. If timezone is not provided it's assumed to be
    the local timezone."""
    entry = None
    if not ver:
        entry = get_file_info(token, file)
    else:
        is_path = file[0] == "/"
        itemid = get_itemid(token, file) if is_path else file
        entries = get_all_versions(token, itemid)
        if not entries:
            print(f"No versions found for {itemid}")
            return
        ver_is_date = re.match(r"[0-9]+\.0", ver) is None
        entry = find_version_before_date_in_entries(token, ver, entries) \
            if ver_is_date \
            else find_versionid_in_entries(token, ver, entries)
    if not len(dest):
        dest = entry["name"]
    if "@microsoft.graph.downloadUrl" not in entry:
        raise RuntimeError("No '@microsoft.graph.downloadUrl' in entry. This \
can happen if the path provided is a folder instead of a file.")
    what = f"version {entry['id']}" if ver else f"file {entry['name']}"
    print(f"Downloading {what} (Last Modified: "
          f"{entry['lastModifiedDateTime']}) to {dest}")
    download_url(token, entry["@microsoft.graph.downloadUrl"], dest)

def download_url(token, url, dest, overwrite=False):
    """Function for downloading a @microsoft.graph.downloadUrl URL."""
    r = requests.get(url)
    flags = "wb" if overwrite else "xb"
    with open(to, flags) as f:
        f.write(get_res(r, json=False))
        print(f"Downloaded to '{to}'")

def list_children(token, path=""):
    """List files/subfolders in root or under PATH."""
    entries = get_all_children(token, path)
    for i, entry in enumerate(entries):
        pre = "└──" if i == len(entries)-1 else "├──"
        is_folder = "folder" in entry
        what = "+" if is_folder else ""
        print(f"{pre}{what} {entry['name']} (ID: {entry['id']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")

def list_children(token, path="", recurse=False, rlevel=0):
    """List files/subfolders in root or under PATH.
    If RECURSE, list descendants. RLEVEL tracks depth."""
    entries = get_all_children(token, path)
    for i, entry in enumerate(entries):
        pre = "└──" if i == len(entries)-1 else "├──"
        pre = "│   " * rlevel + pre
        is_folder = "folder" in entry
        what = "+" if is_folder else ""
        print(f"{pre}{what} {entry['name']} (ID: {entry['id']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")
        if recurse and is_folder:
            p = f"{path}/{entry['name']}" if path else entry['name']
            list_children(token, p, True, rlevel+1)

def list_children(token, path="", recurse=False, pre=""):
    """List files/subfolders in root or under PATH.
    If RECURSE, list descendants. PRE is an extra prefix before each item."""
    entries = get_all_children(token, path)
    for i, entry in enumerate(entries):
        is_last = i == len(entries)-1
        connector = "└──" if is_last else "├──"
        is_folder = "folder" in entry
        what = "+" if is_folder else " "
        print(f"{pre}{connector}{what}{entry['name']} (ID: {entry['id']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")
        if recurse and is_folder:
            p = f"{path}/{entry['name']}" if path else entry['name']
            pre_ = pre + ("    " if is_last else "│   ")
            list_children(token, p, True, pre_)

def list_children(token, path="", recurse=False, pre="", callback=None):
    """List files/subfolders in root or under PATH.
    If RECURSE, list descendants. PRE is an extra prefix before each item.
    If CALLBACK is provided, it will be called with info of each descendant."""
    entries = get_all_children(token, path)
    for i, entry in enumerate(entries):
        is_last = i == len(entries)-1
        connector = "└──" if is_last else "├──"
        is_folder = "folder" in entry
        what = "+" if is_folder else " "
        print(f"{pre}{connector}{what}{entry['name']} (ID: {entry['id']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")
        if (callback):
            callback(entry)
        if recurse and is_folder:
            p = f"{path}/{entry['name']}" if path else entry['name']
            pre_ = pre + ("    " if is_last else "│   ")
            list_children(token, p, True, pre_, callback)

def download_folder(token, path, dest="", ver=None, dry_run=False):
    def cb(entry):
        p = f'{entry["parentReference"]["path"].split(":")[1]}/{entry["name"]}'
        dest_ = (dest if dest else ".") + p
        if "folder" in entry:
            print(f"Creating folder {dest_}")
            if not dry_run:
                makedirs(dest_)
        else:
            download_file(token, p, dest_, ver, dry_run=dry_run)
    list_children(token, path, True, callback=cb)

def download_folder(token, path, dest="", ver=None, dry_run=False):
    entry = get_file_info(token, path)
    if "folder" not in entry:
        raise ValueError(f"{path} is not a folder")
    dest = dest if dest else entry["name"]
    top = f'{entry["parentReference"]["path"].split(":")[1]}/{entry["name"]}'

    def cb(entry):
        p = f'{entry["parentReference"]["path"].split(":")[1]}/{entry["name"]}'
        dest_ = dest + (p[len(top):] if p.startswith(top) else p)
        if "folder" in entry:
            print(f"Creating folder {dest_}")
            if not dry_run:
                makedirs(dest_)
        else:
            download_file(token, p, dest_, ver, dry_run=dry_run)

    list_children(token, path, True, callback=cb)

        entry = find_version_before_date_in_entries(token, ver, entries) \
            if ver_is_date \
            else find_versionid_in_entries(token, ver, entries)
        if not entry:
            return

#!/usr/bin/env python3
# OneDrive verrer
# 2026-04-02 13:24
"""Script to download an older version of a OneDrive file or folder.

Examples:
    onedriveverrer --list-children ''
    onedriveverrer --download-file '/test.txt'  # current version
    onedriveverrer --download-file '/test.txt' 2026-03-01 ~/test.txt
    onedriveverrer --download-folder '/test' 2026-03-01 ~/test
"""

import re  # re.search for extracting auth code from url, re.match versionid
import sys                      # sys.argv
import json                     # load and dump in get_tokens and save_tokens
import argparse                 # CLI
import requests                 # all communication with server
import dateutil                 # dateutil.parser for version_before_date
from time import time           # tokens expiration calculation
from os import makedirs         # save_tokens, download_folder
from os.path import (
    expanduser,                 # expand ~
    dirname,                    # with makedirs in save_tokens
    getmtime)                   # tokens expiration calculation

NAME = "onedriveverrer"
DESCRIPTION = "Download an older version of a OneDrive file or folder."
ARGS = [(["--auth"], {"action": "store_true", "help": "\
Use our own refresh token or do an auth flow to generate one instead of using \
the assumed existing refresh token of a OneDrive client in \
~/.config/onedrive/refresh_token."}),
        (["--dry-run"], {"action": "store_true", "help": "\
With download operations, only log without downloading."}),
        (["--dump"], {"help": "\
Get file information object for given PATH, which should be relative to root \
and start with a /.", "metavar": "PATH"}),
        (["-l", "--list-children"], {"help": "\
List files and subfolders under given PATH, which should be relative to root \
and start with a /. Empty string will list top-level.", "metavar": "PATH"}),
        (["-a", "--list-arborescence"], {"help": "\
Like --list-children but recursive.", "metavar": "PATH"}),
        (["--list-versions"], {"help": "\
List versions of a file given its ID or PATH. PATH should be relative to \
root and start with a /. ID should be passed in single quotes in shells like \
Bash or it will try to interpret the ! in it.", "metavar": "(ID|PATH)"}),
        (["--download-file"], {"nargs": "+", "help": "\
Download a file given its ID or PATH. It will be downloaded to pwd with the \
same name as on OneDrive unless a destination path is provided as second \
argument DEST. A VERSIONID or DATE can be specified as the third argument to \
download the version specified or most recent version before DATE. \
PATH should be relative to root and start with a /. \
DATE should be in a format supported by dateutil.parser.parse, e.g. \
'2026-04-04 10:30'. If timezone is not provided it's assumed to be the local \
timezone.", "metavar": "(ID|PATH) [DEST] [VERSIONID|DATE]"}),
        (["--download-folder"], {"nargs": "+", "help": "\
Download a folder given its PATH. It will be downloaded to pwd with the \
same name as on OneDrive unless a destination path is provided as second \
argument DEST. A DATE can be specified as the third argument to \
download for each file the most recent version before DATE. \
PATH should be relative to root and start with a /. \
DATE should be in a format supported by dateutil.parser.parse, e.g. \
'2026-04-04 10:30'. If timezone is not provided it's assumed to be the local \
timezone.", "metavar": "PATH [DEST] [DATE]"})]

# Used if no --auth
EXISTINGRTOKENPATH = expanduser("~/.config/onedrive/refresh_token")  # expand ~
# Used always (if nothing else, to store the access token)
OWNTOKENSPATH = expanduser("~/.config/onedriveverrer/tokens")

CLIENTID = "[..client id..]"
USERAGENT = "[..user agent..]"
SCOPE = "\
Files.ReadWrite Files.ReadWrite.All Sites.ReadWrite.All offline_access"
AUTHDOMAIN = "https://login.microsoftonline.com/common/oauth2"
REDIRECTURI = f"{AUTHDOMAIN}/nativeclient"
AUTHURI = f"{AUTHDOMAIN}/v2.0/authorize"
TOKENURI = f"{AUTHDOMAIN}/v2.0/token"
GRAPHDOMAIN = "https://graph.microsoft.com/v1.0/me/drive"


def get_res(r, json=True):
    """Get json response of a request or print details and raise exception
    if it failed."""
    if r.status_code == 200:
        return r.json() if json else r.content
    else:
        try:
            print(r.json())
        except requests.exceptions.JSONDecodeError:
            print(f"{NAME}: No json body. Status code: {r.status_code}")
        r.raise_for_status()


def headers(token=None):
    """Return standard headers to send with requests."""
    if token:
        return {"authorization": f"bearer {token}", "user-agent": USERAGENT}
    return {"user-agent": USERAGENT}


def auth():
    """Interactive auth flow to generate a refresh token.
    Requires opening a page in the browser, logging in, and pasting the URL of
    the page one got redirected to, which should contain ?code= parameter."""
    print("Please authorise this application by visiting the following URL:")
    d = {"client_id": CLIENTID,
         "redirect_uri": REDIRECTURI,
         "scope": SCOPE.replace(" ", "%20"),
         "response_type": "code",
         "prompt": "login"}
    print(AUTHURI + "?" + "&".join("%s=%s" % (k, v) for k, v in d.items()))
    print("After completing the authorisation in your browser, copy the full "
          "redirect URI (from the address bar) and paste it below.")
    response = input("Paste redirect URI here: ")
    code = re.search(r"(?:[?&]code=)([^&]+)", response.strip())
    if not code:
        raise RuntimeError("An empty or invalid redirect URI was entered")
    d = {"client_id": CLIENTID,
         "redirect_uri": REDIRECTURI,
         "code": code.group(1),
         "grant_type": "authorization_code"}
    r = requests.post(TOKENURI, headers=headers(),
                      data="&".join("%s=%s" % (k, v) for k, v in d.items()))
    return get_res(r)


def save_tokens(tokens, path):
    """Save object TOKENS at PATH, creating containing directories if
    necessary."""
    makedirs(dirname(path), exist_ok=True)
    with open(path, "w") as f:
        json.dump(tokens, f)
        print(f"{NAME}: Wrote tokens in {path}")


def get_tokens(path, arg_auth=False):
    """Get tokens from file at PATH.
    If arg_auth is True and there are no tokens on disk, trigger
    interactive auth flow to generate them."""
    tokens = None
    try:
        with open(path, "r") as f:
            tokens = json.load(f)
    except json.decoder.JSONDecodeError as e:
        print(f"{NAME}: Something wrong with existing tokens file {path}: {e}")
        print(f"{NAME}: Will attempt to regenerate tokens.")
    except FileNotFoundError:
        print(f"{NAME}: No tokens in {path}. "
              "This is normal if we have not generated any yet.")
        if arg_auth:            # FIXME: Not tested this codepath
            print(f"{NAME}: Triggering interactive auth flow...")
            tokens = auth()
            save_tokens(tokens, path)
    return tokens


def read_refreshtoken(path):
    """Get refresh token from file at PATH.
    This is only used if --auth is not provided which means we are using an
    existing refresh token generated by another application. It must exist or
    it's a fatal error."""
    refreshtoken = None
    try:
        with open(path, "r") as f:
            refreshtoken = f.read()
            if not refreshtoken:
                raise RuntimeError(f"Empty refresh token file in {path}")
    except FileNotFoundError:
        raise RuntimeError(f"No refresh token file in {path}")
    return refreshtoken


def fetch_tokens(refreshtoken):
    """Fetch new tokens using a refresh token."""
    d = {"client_id": CLIENTID,
         "redirect_uri": REDIRECTURI,
         "refresh_token": refreshtoken,
         "grant_type": "refresh_token"}
    r = requests.post(TOKENURI, headers=headers(), data=d)
    return get_res(r)


def get_me(token, sub=""):
    """Drive details."""
    r = requests.get(f"{GRAPHDOMAIN}{sub}", headers=headers(token))
    return get_res(r)


def get_file_info(token, path):
    """Get information object for file at PATH.
    (Path relative to root, with starting /)"""
    return get_me(token, f"/root:{path}")


def get_itemid(token, path):
    """Get itemid of file at PATH.
    (Path relative to root, with starting /)"""
    data = get_file_info(token, path)
    return data["id"]


def get_children(token, path=""):
    """Drive top-level files, or files in PATH.
    (Path relative to root, with starting /)"""
    if path and path[0] != "/":
        raise ValueError("PATH must either be empty or begin with a /")
    p = f"/root:{path}:/children" if path else "/root/children"
    return get_me(token, p)


def get_next(token, data):
    """Resolve next link in data. Used for pagination."""
    r = requests.get(data['@odata.nextLink'], headers=headers(token))
    return get_res(r)


def get_all_children(token, path=""):
    """get_children wrapper with pagination handling."""
    data = get_children(token, path)
    entries = data.get("value", [])
    while "@odata.nextLink" in data:  # handle pagination
        data = get_next(token, data)
        entries.extend(data.get("value", []))
    return entries


def list_children(token, path="", recurse=False, pre="", callback=None):
    """List files/subfolders in root or under PATH.
    If RECURSE, list descendants. PRE is an extra prefix before each item.
    If CALLBACK is provided, it will be called with info of each descendant."""
    entries = get_all_children(token, path)
    for i, entry in enumerate(entries):
        is_last = i == len(entries)-1
        connector = "└──" if is_last else "├──"
        is_folder = "folder" in entry
        what = "+" if is_folder else " "
        print(f"{pre}{connector}{what}{entry['name']} (ID: {entry['id']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")
        if (callback):
            callback(entry)
        if recurse and is_folder:
            p = f"{path}/{entry['name']}" if path else entry['name']
            pre_ = pre + ("    " if is_last else "│   ")
            list_children(token, p, True, pre_, callback)


def get_versions(token, itemid):
    """Data with versions of a file with given ITEMID."""
    return get_me(token, f"/items/{itemid}/versions")


def get_all_versions(token, itemid):
    """get_versions wrapper with pagination handling."""
    data = get_versions(token, itemid)
    entries = data.get("value", [])
    while "@odata.nextLink" in data:  # handle pagination
        data = get_next(token, data)
        entries.extend(data.get("value", []))
    return entries


def list_versions(token, of):
    """List versions for file with given itemid or path.
    (Path relative to root, with starting /)"""
    is_path = of[0] == "/"
    itemid = get_itemid(token, of) if is_path else of
    entries = get_all_versions(token, itemid)
    if not entries:
        print(f"{NAME}: No versions found for {itemid}")
        return
    for entry in entries:
        print(f"- {entry['id']} (ID) (Size: {entry['size']}, "
              f"Last Modified: {entry['lastModifiedDateTime']})")


def download_url(token, url, dest, overwrite=False):
    """Function for downloading a @microsoft.graph.downloadUrl URL."""
    r = requests.get(url)
    flags = "wb" if overwrite else "xb"
    with open(dest, flags) as f:
        f.write(get_res(r, json=False))
        print(f"{NAME}: Downloaded to '{dest}'")


def find_versionid_in_entries(token, versionid, entries):
    """Return entry with version VERSIONID in entries array ENTRIES."""
    for entry in entries:
        if entry["id"] == versionid:
            return entry
    print(f"{NAME}: No version with ID {versionid} found")


def find_version_before_date_in_entries(token, date, entries):
    """Return entry before DATE in entries array ENTRIES.
    (Date in a format supported by dateutil.parser.parse, e.g.
    '2026-04-04 10:30'. If timezone is not provided it's assumed to be the
    local timezone.)"""
    date = dateutil.parser.parse(date)
    if not date.tzinfo:  # can't compare timezone-aware with unaware
        date = date.astimezone()
    for entry in entries:
        if dateutil.parser.parse(entry["lastModifiedDateTime"]) < date:
            return entry
    print(f"{NAME}: No version before date {date} found")


def download_file(token, file, dest="", ver=None, dry_run=False):
    """Download FILE given its id or path. It will be downloaded to pwd with
    the same name as on OneDrive unless a destination path DEST is provided.
    A versionid or date VER can be specified to download that version or most
    recent version before date.
    PATH should be relative to root and start with a /.
    DATE should be in a format supported by dateutil.parser.parse, e.g.
    '2026-04-04 10:30'. If timezone is not provided it's assumed to be
    the local timezone."""
    entry = None
    if not ver:
        entry = get_file_info(token, file)
    else:
        is_path = file[0] == "/"
        itemid = get_itemid(token, file) if is_path else file
        entries = get_all_versions(token, itemid)
        if not entries:
            print(f"{NAME}: No versions found for {itemid}")
            return
        ver_is_date = re.match(r"[0-9]+\.0", ver) is None
        entry = find_version_before_date_in_entries(token, ver, entries) \
            if ver_is_date \
            else find_versionid_in_entries(token, ver, entries)
        if not entry:
            return
    if not len(dest):
        dest = entry["name"]
    if "@microsoft.graph.downloadUrl" not in entry:
        raise RuntimeError("No '@microsoft.graph.downloadUrl' in entry. This \
can happen if the path provided is a folder instead of a file.")
    what = f"version {entry['id']}" if ver else f"file {entry['name']}"
    print(f"{NAME}: Downloading {what} (Last Modified: "
          f"{entry['lastModifiedDateTime']}) to {dest}")
    if not dry_run:
        download_url(token, entry["@microsoft.graph.downloadUrl"], dest)


def download_folder(token, path, dest="", ver=None, dry_run=False):
    """Download folder given its PATH. It will be downloaded to pwd with
    the same name as on OneDrive unless a destination path DEST is provided.
    A date VER can be specified to download for each file the most recent
    version before given date.
    PATH should be relative to root and start with a /.
    DATE should be in a format supported by dateutil.parser.parse, e.g.
    '2026-04-04 10:30'. If timezone is not provided it's assumed to be
    the local timezone."""
    entry = get_file_info(token, path)
    if "folder" not in entry:
        raise ValueError(f"{path} is not a folder")
    dest = dest if dest else entry["name"]
    top = f'{entry["parentReference"]["path"].split(":")[1]}/{entry["name"]}'

    def cb(entry):
        p = f'{entry["parentReference"]["path"].split(":")[1]}/{entry["name"]}'
        dest_ = dest + (p[len(top):] if p.startswith(top) else p)
        if "folder" in entry:
            print(f"{NAME}: Creating folder {dest_}")
            if not dry_run:
                makedirs(dest_)
        else:
            download_file(token, p, dest_, ver, dry_run=dry_run)

    list_children(token, path, True, callback=cb)


def parse_args():
    """CLI."""
    parser = argparse.ArgumentParser(
        prog=NAME,
        description=DESCRIPTION)
    parser.suggest_on_error = True
    for args, kwargs in ARGS:
        parser.add_argument(*args, **kwargs)
    # parse args / print help and quit if no args
    # (Primer https://stackoverflow.com/a/47440202/18396947)
    return parser.parse_args(sys.argv[1:] or ['--help'])


if __name__ == "__main__":
    args = parse_args()

    tokens = get_tokens(OWNTOKENSPATH, args.auth)
    accesstoken = refreshtoken = None
    if not tokens:
        if args.auth:
            print(f"{NAME}: This should be unreachable, because when --auth "
                  "is provided, if getting tokens fails it does the "
                  "interactive auth flow, and if that fails an exception "
                  "would have kicked you out.")
            exit(1)
        else:
            refreshtoken = read_refreshtoken(EXISTINGRTOKENPATH)
            tokens = fetch_tokens(refreshtoken)
            save_tokens(tokens, OWNTOKENSPATH)
    else:
        if time() > getmtime(OWNTOKENSPATH) + tokens["expires_in"]:
            # CONCERN: Maybe it should be fetched with the refresh
            # token in EXISTINGRTOKENPATH if not args.auth, in case
            # this refresh token is also too old
            tokens = fetch_tokens(tokens["refresh_token"])
            save_tokens(tokens, OWNTOKENSPATH)
    refreshtoken = tokens['refresh_token']
    token = tokens['access_token']

    dry_run = args.dry_run
    if args.dump is not None:
        print(get_file_info(token, args.dump))
    if args.list_children is not None:
        list_children(token, args.list_children)
    if args.list_arborescence is not None:
        list_children(token, args.list_arborescence, True)
    if args.list_versions is not None:
        list_versions(token, args.list_versions)
    if args.download_file is not None:
        download_file(token, *args.download_file, dry_run=dry_run)
    if args.download_folder is not None:
        download_folder(token, *args.download_folder, dry_run=dry_run)