티스토리 뷰
일단 본인은 조직 계정으로 개인 계정과 다를 수도 있음, 링크나 메뉴 위치가 안보이면 빠르게 다른 글 가서 확인하길
windows에 내장된 onedrive가 특정 폴더 지정이 안되는게 열 받아서 찾아 봄
내용은 azure portal에서 앱 등록 이후 코드로 넘어감
- 정말 귀찮아도 토큰이 필수
1. azure portal, login
https://azure.microsoft.com/ko-kr/get-started/azure-portal
2. 앱 등록(application register)
https://portal.azure.com/#view/Microsoft_AAD_RegisteredApps/ApplicationsListBlade
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으로 해놨음
- selenium이 귀찮으면 get_new_token()에서 auth_url을 직접 브라우저로 열고 로그인 하면 됨
- 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}
아래 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도 추가함
'Python' 카테고리의 다른 글
python class 생성 시 사용한 변수의 값을 내부에서 참조함 (0) | 2023.12.02 |
---|---|
ModuleNotFoundError: No module named 'MySQLdb' (0) | 2023.08.09 |
mac m1, m2 anaconda locust ValueError: greenlet.greenlet size changed, may indicate binary incompatibility (0) | 2023.08.06 |
클라이언트 간 메시지 송수신 스터디(nginx + fastapi + celery + rabbitMQ) (0) | 2023.08.04 |
Python datetime 날짜의 월이 영어일때, 12시 pm, am처리 (0) | 2020.07.06 |
티스토리 방명록
- Total
- Today
- Yesterday
Contact: j0n9m1n1@gmail.com