diff options
Diffstat (limited to 'generate_config.py')
| -rw-r--r-- | generate_config.py | 206 |
1 files changed, 206 insertions, 0 deletions
diff --git a/generate_config.py b/generate_config.py new file mode 100644 index 0000000..bba424e --- /dev/null +++ b/generate_config.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +"""Generate rclone combine configs for Microsoft Teams channel drives.""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + +GRAPH = "https://graph.microsoft.com/v1.0" +HERE = Path(__file__).resolve().parent +DEFAULT_CONFIG = HERE / "config.json" +REMOTE_TMPL = (HERE / "remote-template.conf").read_text() +COMBINE_TMPL = (HERE / "combine-template.conf").read_text() + + +def sanitize_remote(name: str) -> str: + return re.sub(r"[^\w.-]+", "_", name).strip("_") + + +def get_token(tenant: str, client_id: str, client_secret: str) -> str: + data = urllib.parse.urlencode( + { + "client_id": client_id, + "client_secret": client_secret, + "scope": "https://graph.microsoft.com/.default", + "grant_type": "client_credentials", + } + ).encode() + req = urllib.request.Request( + f"https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token", + data=data, + method="POST", + ) + with urllib.request.urlopen(req) as resp: + return json.load(resp)["access_token"] + + +def graph_get(url: str, token: str) -> dict: + req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}) + with urllib.request.urlopen(req) as resp: + return json.load(resp) + + +def graph_paged(url: str, token: str) -> list: + items: list = [] + while url: + payload = graph_get(url, token) + items.extend(payload.get("value", [])) + url = payload.get("@odata.nextLink") + return items + + +def find_team(teams: list, name: str, exact: bool) -> dict | None: + for team in teams: + dn = team.get("displayName") or "" + if exact and dn == name: + return team + if not exact and dn.lower().startswith(name.lower()): + return team + return None + + +def collect_channels(team_id: str, token: str) -> list[dict]: + channels = graph_paged(f"{GRAPH}/teams/{team_id}/channels", token) + results = [] + for ch in channels: + time.sleep(0.25) + folder = graph_get( + f"{GRAPH}/teams/{team_id}/channels/{ch['id']}/filesFolder", token + ) + results.append( + { + "displayName": ch["displayName"], + "driveId": folder["parentReference"]["driveId"], + "membershipType": ch.get("membershipType", "standard"), + } + ) + print(f" {ch['membershipType']}: {ch['displayName']} -> {folder['parentReference']['driveId']}") + return results + + +def build_config(team_name: str, channels: list[dict], creds: dict) -> str: + team_remote = sanitize_remote(team_name) + standard = [c for c in channels if c["membershipType"] == "standard"] + other = [c for c in channels if c["membershipType"] != "standard"] + + blocks: list[str] = [] + upstreams: list[str] = [] + written_remotes: set[str] = set() + + def remote_block(name: str, drive_id: str) -> str: + return REMOTE_TMPL.format( + name=name, + drive_id=drive_id, + client_id=creds["client_id"], + client_secret=creds["client_secret"], + tenant=creds["tenant"], + ) + + if standard: + blocks.append(remote_block(team_remote, standard[0]["driveId"])) + written_remotes.add(team_remote) + for ch in standard: + name = ch["displayName"] + upstreams.append(f'"{name}={team_remote}:{name}"') + + for ch in other: + remote = sanitize_remote(ch["displayName"]) + if remote in written_remotes: + remote = f"{remote}_{ch['membershipType']}" + blocks.append(remote_block(remote, ch["driveId"])) + written_remotes.add(remote) + name = ch["displayName"] + # private drives nest files under a folder named after the channel + if ch["membershipType"] == "private": + upstreams.append(f'"{name}={remote}:{name}"') + else: + # shared / unknownFutureValue: files sit at drive root + upstreams.append(f'"{name}={remote}:"') + + if upstreams: + blocks.append(COMBINE_TMPL.format(f"{team_remote}_combine", " ".join(upstreams))) + + return "\n".join(blocks) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--team-name", nargs="+", help="Team display name prefix(es)") + parser.add_argument("--team-id", help="Team id (skips name lookup)") + parser.add_argument("--exact", action="store_true", help="Exact displayName match") + parser.add_argument( + "--config", + type=Path, + default=DEFAULT_CONFIG, + help=f"Path to config.json (default: {DEFAULT_CONFIG})", + ) + parser.add_argument( + "-o", + "--outdir", + type=Path, + default=HERE, + help="Output directory for .conf files", + ) + args = parser.parse_args() + + if not args.team_id and not args.team_name: + parser.error("provide --team-name and/or --team-id") + + cfg = json.loads(args.config.read_text()) + creds = { + "tenant": cfg["tenantId"], + "client_id": cfg["clientID"], + "client_secret": cfg["clientSecret"], + } + + print("Authenticating...") + token = get_token(creds["tenant"], creds["client_id"], creds["client_secret"]) + args.outdir.mkdir(parents=True, exist_ok=True) + + targets = [] + if args.team_id: + team = graph_get(f"{GRAPH}/teams/{args.team_id}", token) + print(f"Team: {team['displayName']} ({args.team_id})") + targets.append(team) + + if args.team_name: + print("Listing teams...") + all_teams = graph_paged(f"{GRAPH}/teams/", token) + for requested in args.team_name: + team = find_team(all_teams, requested, args.exact) + if not team: + print(f"ERROR: no team matched '{requested}'", file=sys.stderr) + continue + print(f"Found: {team['displayName']} ({team['id']})") + targets.append(team) + + for team in targets: + print(f"\nChannels for {team['displayName']}:") + try: + channels = collect_channels(team["id"], token) + except urllib.error.HTTPError as e: + print(f"ERROR fetching channels: {e}", file=sys.stderr) + continue + + if not channels: + print(f"ERROR: no channels for {team['displayName']}", file=sys.stderr) + continue + + conf = build_config(team["displayName"], channels, creds) + out = args.outdir / f"{sanitize_remote(team['displayName'])}.conf" + out.write_text(conf) + print(f"Wrote {out}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) |