Go to Qiita Advent Calendar 2024 Top
LoginSignup
14
12

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@saurus12(サウルス１２)

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

Last updated at Posted at 2022-08-04

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

概要

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

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

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

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/

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

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

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

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

  • 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"
}

アクセストークンの利用

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)

👇関連記事

👇参考URL

本記事へのリンク

test.png

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

[keywords]
OneDrive Python

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

更新日:2024年01月14日

14
12
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
14
12

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?