OneDrive のファイル操作（登録／取得）を自動化する

Pythonを使って「OneDriveへファイルアップロード」「OneDriveからのファイルダウンロード」を実現する手順を解説します。

概要

• OneDriveへ自動アクセスするためAzure アプリを登録します。
• OneDriveへログインし、認証コードを取得します。
• 認証コードからアクセストークンを取得します。
• アクセストークンを利用してMicrosoft Graphにより「OneDriveへファイルアップロード」「OneDriveからのファイルダウンロード」を実現します。

OneDriveへ自動アクセスするためAzure アプリを登録

• 下記手順を参考にAzure アプリを登録します。
  Microsoft にアプリを登録する
  操作はAzure Portalから実施します。
  
• リダイレクトURLにはhttp://localhost:4200/を入力します。
  
• 概要を選択し、クライアントIDを取得します。
  
• 証明書とシークレットを選択し、新しいクライアントシークレットを選択します。
  
• 任意の有効期限を指定します。
  
• クライアントシークレットを取得します。
  ※ クライアントシークレットとして取得するのは「値」の方です。(シークレットIDの方ではありません。)

OneDriveへログインし、認証コードを取得

• 下記手順を参考にOneDriveへのサインインを組み込みます。
  ユーザーのサインイン

トークン取得用URLを作成し、ブラウザへ貼り付けます。

• 下記のclient_id=xxxxxxxxxxxxxxxxxxxxxxxxには上記で取得したクライアントIDを指定します。

https://login.microsoftonline.com/common/oauth2/v2.0/authorize?
client_id=xxxxxxxxxxxxxxxxxxxxxxxx&scope=offline_access%20files.readwrite.all&response_type=code&redirect_uri=http://localhost:4200/

• サインインが完了すると認証コードが発行されるので取得します。
  

• code=xxxxxxxxxxxxxと表示されます。このxxxxxxxxxxxxxが認証コードです。
  

認証コードからアクセストークンを取得

• 下記手順を参考に認証コードからアクセストークンを取得します。
  アクセストークンの取得

• aaaaaaaaaaaaaaaaaには上記で取得した認証コードを指定します。

• bbbbbbbbbbbbbbbbbには上記で取得したクライアントIDを指定します。

• cccccccccccccccccには上記で取得したクライアントシークレットを指定します。

import urllib.parse
import urllib.request
import json

def get_onedrive_token(code: str):

    apl_client_id= "bbbbbbbbbbbbbbbbb"
    client_secret = "ccccccccccccccccccc"
    redirect_url = "http://localhost:4200/"

    url = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
    method = "POST"
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
    }

    params = {
        "client_id": apl_client_id
        ,"redirect_uri": redirect_url
        ,"client_secret": client_secret
        ,"code": code
        ,"grant_type": "authorization_code"
    }

    encoded_param = urllib.parse.urlencode(params).encode()

    request = urllib.request.Request(url, data=encoded_param, method=method, headers=headers)
    with urllib.request.urlopen(request) as res:
        body = res.read()
        dat = json.loads(body)
        print(dat)

if __name__ == '__main__':

    code = "aaaaaaaaaaaaaaaaaaaaaaaa"
    get_onedrive_token(code)

• 下記のようなjsonが取得できます。
• この中に以降の処理で必要なアクセストークン(access_token)及びリフレッシュトークン(refresh_token)が含まれます。

{
  "token_type":"bearer",
  "expires_in": 3600,
  "scope":"wl.basic onedrive.readwrite",
  "access_token":"EwCo...AA==",
  "refresh_token":"eyJh...9323"
}

アクセストークンの更新

• 下記手順を参考にリフレッシュトークンからアクセストークンを取得します。
  新しいアクセストークンの取得

• aaaaaaaaaaaaaaaaaには上記で取得したリフレッシュトークンを指定します。

• bbbbbbbbbbbbbbbbbには上記で取得したクライアントIDを指定します。

• cccccccccccccccccには上記で取得したクライアントシークレットを指定します。

import urllib.parse
import urllib.request
import json

def get_onedrive_reflesh_token(refresh_token: str):

    apl_client_id= "bbbbbbbbbbbbbbbbbbbb"
    client_secret = "ccccccccccccccccccc"
    redirect_url = "http://localhost:4200/"

    url = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
    method = "POST"
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
    }

    params = {
        "client_id": apl_client_id
        ,"redirect_uri": redirect_url
        ,"client_secret": client_secret
        ,"refresh_token": refresh_token
        ,"grant_type": "refresh_token"
    }

    encoded_param = urllib.parse.urlencode(params).encode()

    request = urllib.request.Request(url, data=encoded_param, method=method, headers=headers)
    with urllib.request.urlopen(request) as res:
        body = res.read()
        dat = json.loads(body)
        print("refresh_token:" + dat["refresh_token"])
        print("access_token:" + dat["access_token"])

if __name__ == '__main__':

    refresh_token = "aaaaaaaaaaaaaaaaaaaaa"
    get_onedrive_reflesh_token(refresh_token)

• 下記のようなjsonが取得できます。
• この中に以降の処理で必要なアクセストークン(access_token)及びリフレッシュトークン(refresh_token)が含まれます。

{
  "token_type":"bearer",
  "expires_in": 3600,
  "scope": "wl.basic onedrive.readwrite wl.offline_access",
  "access_token":"EwCo...AA==",
  "refresh_token":"eyJh...9323"
}

アクセストークンの利用

• 下記手順を参考にアクセストークンをHTTP ヘッダーへ設定します。
  アクセストークンの利用

OneDriveのフォルダ内容一覧取得

• 下記手順を参考にOneDriveのフォルダ内容一覧を取得します。
  driveItem の子を一覧表示する

• aaaaaaaaaaaaaaaaaには上記で取得したアクセストークンを指定します。

import urllib.parse
import urllib.request
import json

def get_onedrive_list(access_token: str):

    url = "https://graph.microsoft.com/v1.0/me/drive/root/children"
    method = "GET"
    headers = {
        'Authorization': 'bearer ' + access_token
    }

    request = urllib.request.Request(url, method=method, headers=headers)
    with urllib.request.urlopen(request) as res:
        body = res.read()
        dat = json.loads(body)
        if dat["value"]:
            list = dat["value"]
            for item in list:
                print(item)
                print("  ")

if __name__ == '__main__':

    access_token = "aaaaaaaaaaaaaaaaaa"
    get_onedrive_list(access_token)

• 下記のようなjsonが取得できます。

HTTP/1.1 200 OK
Content-type: application/json

{
  "value": [
    {"name": "myfile.jpg", "size": 2048, "id": "bbbbbbbbbbbbbb", "file": {} },
    {"name": "Documents", "id": "bbbbbbbbbbbbbb", "folder": { "childCount": 4} },
    {"name": "Photos", "id": "bbbbbbbbbbbbbb", "folder": { "childCount": 203} },
    {"name": "my sheet(1).xlsx", "id": "bbbbbbbbbbbbbb", "size": 197 }
  ],
  "@odata.nextLink": "https://..."
}

OneDriveへのファイル登録

• 下記手順を参考にOneDriveへファイルを登録します。
  ファイルをアップロードする

• aaaaaaaaaaaaaaaaaには上記で取得したアクセストークンを指定します。

• bbbbbbbbbbbbbbbbbには上記で取得した格納対象フォルダのidを指定します。

import pathlib
import urllib.parse
import urllib.request
import json
import os

def upload_file_session(access_token: str, upload_url:str, target_file:pathlib.Path):

    contentLength = os.path.getsize(target_file)
    print("contentLength:" + str(contentLength))

    with open(target_file, 'rb') as f:

        file_split_size = 1024 * 1024 * 10

        current_pos = 0

        while current_pos < contentLength:

            remain = contentLength - current_pos
            next_size = file_split_size
            if remain <= file_split_size:
                next_size = remain

            file_body = f.read(next_size)

            reange_string = "bytes " + str(current_pos) + "-" + str(current_pos + next_size -1) + "/" + str(contentLength)
            print(reange_string)

            method = "PUT"
            headers = {
                'Content-Type': 'application/octet-stream'
                ,'Authorization': "Bearer " + access_token
                ,'Content-Length': next_size
                ,'Content-Range': reange_string
            }

            request = urllib.request.Request(upload_url, data=file_body, method=method, headers=headers)
            with urllib.request.urlopen(request) as res:
                body = res.read()
                print(body)

            current_pos = current_pos + next_size

def upload_file(access_token: str, parent_id: str, target_file:pathlib.Path):

    url = "https://graph.microsoft.com/v1.0/me/drive/items/" + urllib.parse.quote(parent_id) + ":/"+ urllib.parse.quote(target_file.name) + ":/createUploadSession"
    method = "POST"
    headers = {
        'Content-Type': 'application/json'
        ,'Authorization': 'bearer ' + access_token
    }

    params = {
        "item": {
            "@odata.type": "microsoft.graph.driveItemUploadableProperties",
            "@microsoft.graph.conflictBehavior": "replace",
            "name": target_file.name
        }
      }

    encoded_param = json.dumps(params).encode("utf-8")

    request = urllib.request.Request(url, data=encoded_param, method=method, headers=headers)

    with urllib.request.urlopen(request) as res:
        body = res.read()
        dat = json.loads(body)
        upload_url = dat["uploadUrl"]
        
        upload_file_session(access_token,upload_url,target_file)

OneDriveからのファイルダウンロード

• 下記手順を参考にOneDriveからファイルをダウンロードします。 DriveItem のコンテンツをダウンロードする

• aaaaaaaaaaaaaaaaaには上記で取得したアクセストークンを指定します。

• bbbbbbbbbbbbbbbbbには上記で取得した格納対象フォルダのidを指定します。

import urllib.parse
import urllib.request
import json

def download_file(access_token: str, item_id:str):

    url = "https://graph.microsoft.com/v1.0/me/drive/items/" + urllib.parse.quote(item_id)

    method = "GET"
    headers = {
        'Authorization': 'bearer ' + access_token
    }

    request = urllib.request.Request(url, method=method, headers=headers)
    with urllib.request.urlopen(request) as res:
        body = res.read()
        dat = json.loads(body)
        item_name = dat["name"]
        
        download_url = dat["@microsoft.graph.downloadUrl"]
        print("download_url:" + download_url)
        
        save_name = "c:/tmp/" + item_name
        urllib.request.urlretrieve(download_url, save_name)        
        

if __name__ == '__main__':

    access_token = "aaaaaaaaaaaaaaaaaaaaaaaaaaa"
    item_id = "bbbbbbbbbbbbbbbbbbbbbbb"

    download_file(access_token,item_id)

👇関連記事

• PythonでQRコードを作成する

👇参考URL

• Microsoft にアプリを登録する
• Azure アプリ登録
• 認証
• 新しいアクセス トークンまたは更新トークンを取得する
• アクセストークンの利用
• エンドポイント
• driveItem の子を一覧表示する
• DriveItem のコンテンツをダウンロードする
• ファイルをアップロードする

本記事へのリンク

https://docs.saurus12.com/python/onedrive

[keywords]
OneDrive Python

OneDrive のファイル操作（登録／取得）を自動化する

更新日:2024年01月14日