feat(microsoft-graph-files): O365 file management integration (Graph API)
Microsoft Graph file management for OneDrive / SharePoint / Teams with app-only (client credentials) authentication, stateless over urllib (no dependency). 19 commands: browse sites/drives/content, create folders, delete/upload/replace/ download files (content passed via base64 or source URL; download returns the pre-authenticated Graph URL), site permission management (list/create/update/ delete), SharePoint list reading (lists/items/get-item), and Excel worksheet editing (append row, read range, update cell). Each script obtains a bearer token via the client-credentials grant and calls Graph directly. Re-implemented cleanly from a customized source: dropped the hosted-proxy/auth-code/managed-identity/certificate flows, the duplicated and broken Excel helpers, and the platform-specific file-entry handling. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,164 @@
|
||||
import json, os, re, ssl, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
S = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
I = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
GRAPH = (S.get("host") or "https://graph.microsoft.com").rstrip("/")
|
||||
LOGIN = (S.get("login_url") or "https://login.microsoftonline.com").rstrip("/")
|
||||
|
||||
|
||||
def out(value):
|
||||
print(json.dumps(value, default=str))
|
||||
|
||||
|
||||
def fail(message, **extra):
|
||||
payload = {"error": message}
|
||||
payload.update(extra)
|
||||
print(json.dumps(payload, default=str))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _bool(v):
|
||||
return str(v).lower() in ("1", "true", "yes")
|
||||
|
||||
|
||||
def _list(v):
|
||||
if isinstance(v, list):
|
||||
return v
|
||||
return [x.strip() for x in str(v).split(",") if x.strip()]
|
||||
|
||||
|
||||
def _ctx():
|
||||
return ssl._create_unverified_context() if _bool(S.get("insecure")) else ssl.create_default_context()
|
||||
|
||||
|
||||
_TOKEN = None
|
||||
|
||||
|
||||
def token():
|
||||
global _TOKEN
|
||||
if _TOKEN is not None:
|
||||
return _TOKEN
|
||||
data = urllib.parse.urlencode({
|
||||
"client_id": S.get("client_id", ""),
|
||||
"client_secret": S.get("client_secret", ""),
|
||||
"grant_type": "client_credentials",
|
||||
"scope": GRAPH + "/.default",
|
||||
}).encode("utf-8")
|
||||
url = LOGIN + "/" + urllib.parse.quote(S.get("tenant_id", "")) + "/oauth2/v2.0/token"
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST")
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=60, context=_ctx()) as r:
|
||||
_TOKEN = json.loads(r.read())["access_token"]
|
||||
except urllib.error.HTTPError as e:
|
||||
raise Exception("Authentication failed: HTTP " + str(e.code) + " " + e.read().decode("utf-8", "replace"))
|
||||
return _TOKEN
|
||||
|
||||
|
||||
def graph(method, path, params=None, json_body=None, data=None, headers=None, resp="json"):
|
||||
url = path if path.startswith("http") else (GRAPH + "/v1.0" + path)
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean)
|
||||
h = {"Authorization": "Bearer " + token(), "Accept": "application/json"}
|
||||
body = None
|
||||
if json_body is not None:
|
||||
body = json.dumps(json_body).encode("utf-8")
|
||||
h["Content-Type"] = "application/json"
|
||||
elif data is not None:
|
||||
body = data
|
||||
h["Content-Type"] = "application/octet-stream"
|
||||
if headers:
|
||||
h.update(headers)
|
||||
req = urllib.request.Request(url, data=body, headers=h, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=180, context=_ctx()) as r:
|
||||
raw = r.read()
|
||||
if resp == "response":
|
||||
return {"status": r.status}
|
||||
if resp == "content":
|
||||
return raw
|
||||
return json.loads(raw) if raw else {}
|
||||
except urllib.error.HTTPError as e:
|
||||
raise Exception("Graph HTTP " + str(e.code) + ": " + e.read().decode("utf-8", "replace"))
|
||||
|
||||
|
||||
def fetch_url(url):
|
||||
"""Fetch raw bytes from an arbitrary URL (no Graph auth header)."""
|
||||
req = urllib.request.Request(url, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=180, context=_ctx()) as r:
|
||||
return r.read()
|
||||
|
||||
|
||||
def input_bytes():
|
||||
"""Resolve upload content from content_base64 or source_url."""
|
||||
if I.get("content_base64"):
|
||||
return base64.b64decode(I["content_base64"])
|
||||
if I.get("source_url"):
|
||||
return fetch_url(I["source_url"])
|
||||
fail("Provide either content_base64 or source_url")
|
||||
|
||||
|
||||
def col_letter(n):
|
||||
s = ""
|
||||
n = int(n)
|
||||
while n > 0:
|
||||
n, rem = divmod(n - 1, 26)
|
||||
s = chr(65 + rem) + s
|
||||
return s
|
||||
|
||||
|
||||
def resolve_site_id():
|
||||
if I.get("site_id"):
|
||||
return I["site_id"]
|
||||
name = I.get("site_name")
|
||||
if not name:
|
||||
fail("Provide site_id or site_name")
|
||||
res = graph("GET", "/sites", params={"search": name})
|
||||
value = res.get("value", [])
|
||||
if not value:
|
||||
fail("Site not found: " + name)
|
||||
return value[0]["id"]
|
||||
|
||||
|
||||
def resolve_list_id(site_id):
|
||||
if I.get("list_id"):
|
||||
return I["list_id"]
|
||||
name = I.get("list_name")
|
||||
if not name:
|
||||
fail("Provide list_id or list_name")
|
||||
res = graph("GET", "/sites/" + site_id + "/lists", params={"$top": 999})
|
||||
for lst in res.get("value", []):
|
||||
if (lst.get("displayName") or lst.get("name") or "").lower() == name.lower():
|
||||
return lst["id"]
|
||||
fail("List not found: " + name)
|
||||
|
||||
|
||||
def drive_item_base(object_type, object_type_id, item_id):
|
||||
if object_type == "drives":
|
||||
return "/drives/" + object_type_id + "/items/" + item_id
|
||||
return "/" + object_type + "/" + object_type_id + "/drive/items/" + item_id
|
||||
|
||||
|
||||
def worksheet_base(object_type, object_type_id, item_id, worksheet):
|
||||
ws = urllib.parse.quote(worksheet)
|
||||
if object_type == "drives":
|
||||
return "/drives/" + object_type_id + "/items/" + item_id + "/workbook/worksheets/" + ws
|
||||
return "/" + object_type + "/" + object_type_id + "/drive/items/" + item_id + "/workbook/worksheets/" + ws
|
||||
|
||||
|
||||
|
||||
def run():
|
||||
site_id = resolve_site_id()
|
||||
uri = "/sites/" + site_id + "/permissions/" + I["permission_id"]
|
||||
out(graph("PATCH", uri, json_body={"roles": _list(I["role"])}))
|
||||
|
||||
|
||||
try:
|
||||
run()
|
||||
except SystemExit:
|
||||
raise
|
||||
except Exception as e:
|
||||
fail(str(e))
|
||||
Reference in New Issue
Block a user