[Python] onedrive file upload example(파이썬 원 드라이브 파일 업로드 예제)

 

일단 본인은 조직 계정으로 개인 계정과 다를 수도 있음, 링크나 메뉴 위치가 안보이면 빠르게 다른 글 가서 확인하길

 

Windows에 내장된 OneDrive를 통해 백업 동기화를 할때, 특정 폴더를 지정할 수 없는게 짜증나서 찾아 봄

 

내용은 azure portal에서 앱 등록 이후 코드로 넘어감

- 귀찮아도 토큰이 필수

 

 

1. azure portal, login

https://azure.microsoft.com/ko-kr/get-started/azure-portal

Microsoft Azure Portal | Microsoft Azure

 

2. 앱 등록(application register)

https://portal.azure.com/#view/Microsoft_AAD_RegisteredApps/ApplicationsListBlade

Microsoft Azure

 

3.새 등록, 이름 원하는대로, 본인 요구사항에 맞춰 선택, 플랫폼:웹, http://localhost:8080, 등록

 

4. 좌측의 인증서 및 암호 > 새 클라이언트 암호(설명, 기간) > 만들어지면 값(secret) 보관하기, 페이지 나가면 다시 보여주지 않기 문에 보관하지 않으면 다시 새로 만들어야함

- 코드에서는 "값"을 사용함

 

5. 좌측의 API 사용 권한 > 권한 추가 > Microsoft Graph > 위임된 권한(delegated permission?) 선택 > 검색 필드에 "Files" > Files의 FIles.ReadWrite.All 체크 > 아래 권한 추가

- 조직의 sharepoint에 업로드시에는 Sites로 검색하여 권한 추가

 

 

oauth login으로 auth code를 얻고, 해당 auth code로 토큰을 얻는 것 같음

- 사실 oauth 인증으로 토큰을 얻는게 확실히 이해는 안갔음

 

일반적인 프로세스면 auth code로 url을 열고 id, pw을 치고 로그인 해야하는데, 이거 귀찮아서 selenium으로 해놨음

- auth_url에서 진행되는 과정이 살짝식 달라지기도 해서 수동으로 해야할 수 있음

- 수동은 get_new_token()에서 auth_url을 직접 브라우저로 열고 로그인 하면 됨

 

토큰은 응답 자체를 json으로 저장시켜두고 expire 체크하도록 함

 

그리고 업로드가 당황했던게 토큰과 함께 바로 post하는게 아니라 업로드 세션을 얻고, 그 세션을 통해서 업로드하는게 낯설었음

- 4MB 이하는 세션 없이 바로 업로드가 가능한 것 같던데, 4MB이하는 나한테는 의미가 없음

 

create_folder_if_not_exist() 이거 좀 이상해서 수정 필요함, 폴더가 있으면 폴더명 N으로 생성돼서 앞에 체크하도록 하는 과정 필요함

 

main.py

import requests
import os, sys
import time
import re
import json
from datetime import datetime, timedelta

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.service import Service as ChromeService
from webdriver_manager.chrome import ChromeDriverManager


tenant_id = "앱의 개요 메뉴에 디렉터리(테넌트) ID"
client_id = "앱의 개요 메뉴에 애플리케이션(클라이언트) ID"
redirect_uri = "아까 앱 등록하면서 넣었던 url, http://localhost:8080"
scope = "https://graph.microsoft.com/.default"
client_secret = "4번 인증서 및 암호에서 보관한 값, 비밀 ID아님"
graph_api_endpoint = "https://graph.microsoft.com/v1.0"

email = "your_email@dot.com"
password = "password"

class TokenManager:
    def __init__(self):
        self.token_file_path = "token.json"

    def load_token_info(self):
        if os.path.exists(self.token_file_path):
            with open(self.token_file_path, "r") as file:
                try:
                    return json.load(file)
                except json.JSONDecodeError:
                    return self.get_new_token()
        return None

    def save_token_info(self, token_info):
        token_info["timestamp"] = datetime.now().timestamp()
        with open(self.token_file_path, "w") as file:
            json.dump(token_info, file)

    def is_token_expired(self, token_info):
        expires_in = token_info.get("expires_in")
        expiration = datetime.fromtimestamp(token_info.get("timestamp") + expires_in)
        return datetime.now() >= expiration

    def get_new_token(self):
        auth_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize?client_id={client_id}&response_type=code&redirect_uri={redirect_uri}&response_mode=query&scope={scope}&state=12345"

        # 해당 auth_url을 직접 들어가서 사용자 계정으로 로그인 하는게 원래 맞음
        # 이거 귀찮아서 selenium으로 내 계정 입력하도록 함
        options = webdriver.ChromeOptions()
        # options.add_argument("headless")

        driver = webdriver.Chrome(
            service=ChromeService(ChromeDriverManager().install()), options=options
        )
        print(auth_url)
        driver.get(auth_url)

        driver.implicitly_wait(5)

        elements = driver.find_elements(By.ID, "i0116")
        if len(elements) > 0:
            elements[0].send_keys(email)
        else:
            print("해당 요소가 페이지에 존재하지 않습니다.")

        driver.implicitly_wait(30)
        time.sleep(1)

        elements = driver.find_elements(By.ID, "idSIButton9")
        if len(elements) > 0:
            elements[0].click()
        else:
            print("해당 요소가 페이지에 존재하지 않습니다.")

        driver.implicitly_wait(30)
        time.sleep(1)

        elements = driver.find_elements(By.ID, "i0118")
        if len(elements) > 0:
            elements[0].send_keys(password)
        else:
            print("해당 요소가 페이지에 존재하지 않습니다.")

        driver.implicitly_wait(30)
        time.sleep(1)
        driver.implicitly_wait(30)

        elements = driver.find_elements(By.ID, "idSIButton9")
        if len(elements) > 0:
            elements[0].click()
        else:
            print("해당 요소가 페이지에 존재하지 않습니다.")

        time.sleep(1)
        elements = driver.find_elements(By.ID, "KmsiCheckboxField")
        if len(elements) > 0:
            elements[0].click()
        else:
            print("해당 요소가 페이지에 존재하지 않습니다.")
        driver.implicitly_wait(30)
        time.sleep(1)
        
        elements = driver.find_elements(By.ID, "idSIButton9")
        if len(elements) > 0:
            elements[0].click()
        else:
            print("해당 요소가 페이지에 존재하지 않습니다.")

        print(driver.current_url)
        # url에서 code 부분만 얻기
        match = re.search(r"code=([^&]+)", driver.current_url)
        authorization_code = None
        if match:
            authorization_code = match.group(1)
        else:
            print("Authorization code not found.")

        driver.quit()

        token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
        token_data = {
            "grant_type": "authorization_code",
            "client_id": client_id,
            "scope": scope,
            "code": authorization_code,
            "redirect_uri": redirect_uri,
            "client_secret": client_secret,
        }

        token_headers = {"Content-Type": "application/x-www-form-urlencoded"}

        token_response = requests.post(
            token_url, data=token_data, headers=token_headers
        )
        token_json = token_response.json()
        # token_json = json.loads(token_response.text)
        access_token = token_json.get("access_token")

        self.save_token_info(token_json)
        print(token_response)
        print(token_response.text)
        print("Access Token:", access_token)
        return token_json


def create_folder_if_not_exists(access_token, path):
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }

    folder_name = os.path.basename(path)
    parent_path = os.path.dirname(path)
    
    if parent_path == "":
        url = f"{graph_api_endpoint}/me/drive/root/children"
    else:
        url = f"{graph_api_endpoint}/me/drive/root:/{parent_path}:/children"

    data = {
        "name": folder_name,
        "folder": {},
        "@microsoft.graph.conflictBehavior": "rename",
    }
    response = requests.post(url, headers=headers, json=data)

    if response.status_code == 201:
        return response.json()["id"]
    else:
        print(response.json())
        return None

def get_upload_session(access_token, file_path, file_name):
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }
    
    upload_session_url = f"{graph_api_endpoint}/me/drive/root:/{file_path}{file_name}:/createUploadSession"

    response = requests.post(
        upload_session_url,
        headers=headers,
        json={
            "item": {
                "@microsoft.graph.conflictBehavior": "rename",
                "name": file_name,
            }
        },
    )


    # print(response)
    # print(response.text)
    upload_url = response.json().get("uploadUrl")

    # print("업로드 세션 URL:", upload_url)
    return upload_url


def upload_file(upload_url, local_path):
    file_size = os.path.getsize(local_path)
    # chunk_size = 262144  # 이거 너무 오래걸림
    chunk_size = 62914560  # 60MB

    headers = {"Content-Type": "application/octet-stream"}

    with open(local_path, "rb") as file_data:
        chunk_number = 0
        while True:
            start_byte = chunk_number * chunk_size
            end_byte = min(file_size - 1, start_byte + chunk_size - 1)

            file_data.seek(start_byte)
            chunk_data = file_data.read(chunk_size)
            if not chunk_data:
                break

            content_range = f"bytes {start_byte}-{end_byte}/{file_size}"
            headers["Content-Range"] = content_range

            response = requests.put(upload_url, headers=headers, data=chunk_data)

            if response.status_code in [200, 201, 202]:
                print(f"Chunk {chunk_number + 1} uploaded successfully")
            else:
                print(f"Failed to upload chunk {chunk_number + 1}")
                print(response.json())
                break

            chunk_number += 1

            if end_byte == file_size - 1:
                print("File uploaded successfully.")
                break

if __name__ == "__main__":
    tm = TokenManager()

    access_token = None
    token_info = tm.load_token_info()

    if tm.is_token_expired(token_info):
        token_info = tm.get_new_token()
        access_token = token_info.get("access_token")
    else:
        access_token = token_info.get("access_token")

    onedrive_path = f"folder/folder/folder"
    filename = os.path.basename(installer_path)

    # 폴더 체크 하고
    create_folder_if_not_exists(access_token, onedrive_path)
    # session 만들고
    upload_url = get_upload_session(access_token, onedrive_path, filename)
    # 업로드하고
    upload_file(upload_url, installer_path)

 

 

token.json

{"token_type": "Bearer", "scope": "profile openid email https://graph.microsoft.com/Files.ReadWrite https://graph.microsoft.com/User.Read https://graph.microsoft.com/.default", "expires_in": 3800, "ext_expires_in": 3800, "access_token": "eyJ0eXAiOiJKV1QiLCJub25jZSI6IkpnOW4tX01mQjk4UHdNZXRwWHc5TUk1cUJrd0pKQ09LcU9adEVnMnN4Y0UiLCJhbGciOiJSUzI1NiIsIng1dCI6ImtXYmthYTZxczh3c1RuQndpaU5ZT2hIYm5BdyIsImtpZCI6ImtXYmthYTZxczh3c1RuQndpaU5ZT2hIYm5BdyJ9.eyJhdWQiOiJodHRwczovL2dyYXBoLm1pY3Jvc29mdC5jb20iLCJpc3MiOiJodHRwczovL3N0cy53aW5kb3dzLm5ldC8xOWYzNTMwOS05OTE4LTQyZTMtODJiNy01MmFhMmZmNjY4ZjIvIiwiaWF0IjoxNzA3Mjg2NDAwLCJuYmYiOjE3MDcyODY0MDAsImV4cCI6MTcwNzI5MDUwMSwiYWNjdCI6MCwiYWNyIjoiMSIsImFpbyI6IkFUUUF5LzhWQUFBQUphUUxMWEQ5Ty85TDJGNU5KYXI3V1BGS2Y0VEJIejRpZCtZditqQnQ0U1BmVTlrUmxTeWlvYkw1NEJpN3dBRloiLCJhbXIiOlsicHdkIl0sImFwcF9kaXNwbGF5bmFtZSI6InB5IG9uZWRyaXZlIHVwbG9hZCIsImFwcGlkIjoiZjhmZTFmMGYtM2E3My00NWJiLThjNjItYmFhYjBhZmJmOWZiIiwiYXBwaWRhY3IiOiIxIiwiZmFtaWx5X25hbWUiOiLsnbQiLCJnaXZlbl9uYW1lIjoi7KKF66-8IiwiaWR0eXAiOiJ1c2VyIiwiaXBhZGRyIjoiMjEwLjk5LjExMC40MiIsIm5hbWUiOiLsnbQg7KKF66-8Iiwib2lkIjoiNjAzZjA0MWMtMTg0YS00Mjg4LWFjMjItMWMxMTc4YWM0NWU1IiwicGxhdGYiOiIzIiwicHVpZCI6IjEwMDMyMDAxMDE4QTBBNUUiLCJyaCI6IjAuQVhFQUNWUHpHUmlaNDBLQ3QxS3FMX1pvOGdNQUFBQUFBQUFBd0FBQUFBQUFBQURBQUYwLiIsInNjcCI6IkZpbGVzLlJlYWRXcml0ZSBVc2VyLlJlYWQgcHJvZmlsZSBvcGVuaWQgZW1haWwiLCJzdWIiOiJ4bncwVktxd0JrVl95TWYwdlhfc1VVcGhtcmNXTHh1V3UtWEJHakRWcE9JIiwidGVuYW50X3JlZ2lvbl9zY29wZSI6IkFTIiwidGlkIjoiMTlmMzUzMDktOTkxOC00MmUzLTgyYjctNTJhYTJmZjY2OGYyIiwidW5pcXVlX25hbWUiOiJqbWxlZUBwYWNzLmNvLmtyIiwidXBuIjoiam1sZWVAcGFjcy5jby5rciIsInV0aSI6IlRfYnZYSlhOZGtxeUhjUmhoeWhIQUEiLCJ2ZXIiOiIxLjAiLCJ3aWRzIjpbImI3OWZiZjRkLTNlZjktNDY4OS04MTQzLTc2YjE5NGU4NTUwOSJdLCJ4bXNfc3QiOnsic3ViIjoiWFlqYlduZDIweGFNX0JUdDQzN1RMcHJkTnFVdWp1ZHdoZmF5TDhRWW1YYyJ9LCJ4bXNfdGNkdCI6MTYwNzQyNDgzM30.OdcTZz7zwT97Q2_u7iHcO2ZNajvGFWtcPRbjb3zoC6NOaAdiL3bx0UrDkT3gUlQlp3N6nvH_1tcpKzaT0E4kaQw9Ez4cteW_DnoegTwA2bDyYAsqVXSJiNBQyZKuWZNo_-dc84t6BbByb6Ga0PJZ-pPFjjtTtUEdrEBfBVoM8_Z0rEIwikdcNr23H9wz6J0M4-hNTGrG2JLPEvR1Dx_0fcis9rj5E4fJ4xTx5ssmsDO_TNMrDYl_auPdWV3kmjDUxdvzML45NtydCrUGpKShEsLnbZF2FqdXVq6Vut6WQiIn8IH_dt3mP0aKAWzLVSAXT9YPnf1moIlboJ8Yessawg", "timestamp": 1707286732.111058}

 

webdriver_manager

selenium

requests

 

아래 sharepoint는 access denied로 안돼서 확인중...

- scope를 바꾸니까 잘 되는것 같음

scope = "https://graph.microsoft.com/Sites.ReadWrite.All"

 

sharepoint에 upload 해야 하기 때문에 위 코드는 내가 하려던 거랑은 거리가 좀 있음

정확히는 url이랑 권한이

 

access token으로 site-id를 얻어야 하고, url도 좀 달라짐

def get_site_id(access_token):
    headers = {"Authorization": f"Bearer {access_token}"}
    url = "https://graph.microsoft.com/v1.0/sites/이름.sharepoint.com:/sites/이름2"
    response = requests.get(url, headers=headers)

    print(response.text)
    #reponse.text에 key가 id인게 site-id임, url에 콤마로 나눠져 있을 텐데 다 합쳐서 id임
    
url example for sharepoint

url = f"{graph_api_endpoint}/sites/{site_id}/drive/root/children"
url = f"{graph_api_endpoint}/sites/{site_id}/drive/root:/{parent_path}:/children"
upload_session_url = f"{graph_api_endpoint}/sites/{site_id}/drive/root:/{onedrive_path}{file_name}:/createUploadSession"

 

그리고 앱의 권한도 추가하야함, sites > sites.readwrite.all, 혹시 몰라서manage.all도 추가함