DropBoxの権限調査用スクリプト
もはや自分でコードを書くことも少なく生成AIに任せきってます。DropBoxでディレクトリを確認する必要が出てきて、ひとまず必要になるかなということでGoogleColabから実行するためにpythonスクリプトを生成してもらったやつ。
ディレクトリを再帰的にたどって全体取得するパターンと、階層を指定してその階層分取ってくるパターンにしてます。使い捨てスクリプトなので、かなり雑いです。
1.APIKEYの発行
ブラウザで Dropbox App Console（公式） にアクセスします。
右上の 「Create app」 ボタンを押します。
設定を以下のように選びます：
Choose an API: Scoped access commons を選択
Choose the type of access: Full Dropbox（または必要に応じて App folder）を選択
Name your app: 適当な名前（例：colab-folder-get など）を入力
下の 「Create app」 をクリックします。
アプリの管理画面が開いたら、上部タブの 「Permissions（権限）」 をクリックし、以下の項目にチェックを入れて一番下の「Submit」を押します：
files.metadata.read（ファイルやフォルダの情報を読み取る権限）
上部タブの 「Settings」 に戻り、少し下にスクロールして 「Generated access token」 という項目にある 「Generate」 ボタンを押します。
表示された長い英数字の文字列（これがAPIキーです）をコピーします。
💡 コピーしたAPIキーは、Colabのシークレットに、名前 DROPBOX_API_KEY としてバリューに貼り付けて保存してください。
2.OAUTH発行
! pip install dropbox
import dropbox from google.colab import userdata APP_KEY = userdata.get('DROPBOX_APP_KEY') APP_SECRET = userdata.get('DROPBOX_APP_SECRET') # 永続トークン（リフレッシュトークン）を要求する認証URLを生成 auth_flow = dropbox.DropboxOAuth2FlowNoRedirect( APP_KEY, APP_SECRET, token_access_type='offline' # ←これが「永続（リフレッシュトークン）」を指定する魔法のキーワードです ) authorize_url = auth_flow.start() print("1. 以下のURLをブラウザで開き、Dropboxアカウントへのアクセスを『許可』してください：") print(authorize_url) print("\n2. 画面に表示された「アクセスコード」をコピーして、下の入力欄に貼り付けて Enter を押してください。") auth_code = input("アクセスコードを入力: ").strip() try: oauth_result = auth_flow.finish(auth_code) print("\n🎉 成功しました！【永続トークン（Refresh Token）】を取得しました。") print("--------------------------------------------------") print(oauth_result.refresh_token) print("--------------------------------------------------") print("上記の長い文字列をすべてコピーし、Colabの左メニュー（鍵マーク）に") print("名前：【DROPBOX_REFRESH_TOKEN】として新規追加し、値を貼り付けて保存してください。") except Exception as e: print(f"❌ エラーが発生しました: {e}")
取得したAPIキーをColabのTokenに埋め込んで使用します。
3.プログラム本体
# 💡 dropboxライブラリの自動インストール
!pip install dropbox -q
import dropbox
from dropbox.exceptions import ApiError, RateLimitError
from google.colab import userdata
import pandas as pd
import time
import re
import os
import pickle
import logging
import threading
from contextlib import contextmanager
from concurrent.futures import ThreadPoolExecutor, as_completed
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.utils import get_column_letter
from tqdm.auto import tqdm
# 接続プール満杯の警告(動作には影響しない)を非表示にする
logging.getLogger("urllib3.connectionpool").setLevel(logging.ERROR)
# ==========================================
# ⚙️ 設定項目
# ==========================================
TARGET_ADMIN_EMAIL = "admin@example.com"
TARGET_FOLDERS = [
"Dropbox フォルダー1",
"Dropbox フォルダー2",
]
EXCEL_FILE = "dropbox_folder_permissions_recursive.xlsx"
# --- 取得する範囲 ---
# None : 全階層を再帰的に取得（従来どおり）
# 整数N: 対象フォルダの直下を1階層目として、N階層目までだけ取得（例: 3 → 3階層目まで）
# N階層目より下は取得しないので、浅いほど大幅に速くなります。
# この場合、出力ファイル名の末尾に「_depthN」が付きます（全階層版を上書きしないため）。
MAX_DEPTH = None # 再帰で全階層取得 # MAX_DEPTH = 5 # 5階層目まで取得
# --- 権限の出力 ---
PERM_CHECK_DEPTHS = list(range(1, 31)) # 権限列を付ける階層（実際の最大階層まで作られる）
SHOW_PERMISSION_DETAILS = True # True: メンバー詳細まで取得 / False: 「あり/なし」のみ
# --- フォルダ一覧の取得（高速化まわり）---
PARALLEL_LISTING = True # True: 並列で取得 / False: 単一スレッドで取得
LIST_WORKERS = 8 # 並列数（Dropbox SDKの接続プールが8のため、8以下を推奨）
PERM_WORKERS = 8 # 権限取得の並列数（同上）
LIST_PAGE_LIMIT = 2000 # 1ページあたりの取得件数（APIの上限は2000）
SPLIT_DEPTH = 3 # 【全階層取得のとき】何階層目までを非再帰で展開してから、再帰取得を並列化するか
# 大きいほど負荷の偏りは減るが、最初の展開に呼び出しが増える
MAX_SPLIT_TASKS = 3000 # 【全階層取得のとき】分割後のサブツリー数がこれを超えたら、それ以上深く展開しない
SLOW_TOP_N = 10 # 【全階層取得のとき】所要時間が長かったサブツリーを何件表示するか
# --- キャッシュ（フォルダ構造のみ）---
# 権限情報は毎回取得し直します（共有設定の変更を取りこぼさないため。数十秒で終わります）
# 全階層のキャッシュがあれば、階層限定の実行はそこから取り出すのでAPIを呼びません。
# 逆に、階層限定のキャッシュは、それより深い範囲の実行には使いません。
USE_CACHE = True # True: 前回取得したフォルダ構造があれば再利用 / False: 取り直す
CACHE_FILE = "dropbox_scan_cache.pkl"
# ==========================================
def validate_settings():
if MAX_DEPTH is not None and (isinstance(MAX_DEPTH, bool) or not isinstance(MAX_DEPTH, int) or MAX_DEPTH < 1):
raise ValueError("MAX_DEPTH は None（全階層）または 1 以上の整数で指定してください。")
validate_settings()
try:
APP_KEY = userdata.get('DROPBOX_APP_KEY')
APP_SECRET = userdata.get('DROPBOX_APP_SECRET')
REFRESH_TOKEN = userdata.get('DROPBOX_REFRESH_TOKEN')
except Exception as e:
print("❌ Colabの鍵マークから APP_KEY, APP_SECRET, REFRESH_TOKEN が正しく設定されているか確認してください。")
raise e
# ------------------------------------------
# 共通ユーティリティ
# ------------------------------------------
def clean_for_excel(text):
if not isinstance(text, str):
return text
return re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text)
def call_with_retry(fn, *args, max_retries=6, **kwargs):
"""レート制限(429)は指定秒数待って再試行。通信系の一時エラーも指数バックオフで再試行。"""
backoff = 1.0
for attempt in range(max_retries):
try:
return fn(*args, **kwargs)
except RateLimitError as e:
time.sleep((e.backoff or backoff) + 0.5)
backoff *= 1.5
except ApiError:
raise # パス不正・権限なし等は再試行しても無駄
except Exception:
if attempt == max_retries - 1:
raise
time.sleep(backoff)
backoff *= 1.5
raise RuntimeError("リトライ上限に達しました（レート制限が続いています）")
@contextmanager
def heartbeat(msg, interval=15):
"""応答待ちの間、一定間隔で経過時間を表示（固まって見えないように）"""
stop = threading.Event()
t0 = time.time()
def run():
while not stop.wait(interval):
print(f" ⏳ {msg} ... {int(time.time() - t0)}秒経過", flush=True)
th = threading.Thread(target=run, daemon=True)
th.start()
try:
yield
finally:
stop.set()
# ------------------------------------------
# キャッシュ（取得した範囲も一緒に記録する）
# ------------------------------------------
def load_cache():
if USE_CACHE and os.path.exists(CACHE_FILE):
try:
with open(CACHE_FILE, "rb") as f:
data = pickle.load(f)
folders = data.get("folders", {})
saved = time.strftime('%Y-%m-%d %H:%M', time.localtime(os.path.getmtime(CACHE_FILE)))
print(f"💾 キャッシュを読み込みました（{saved} 保存 / {len(folders)} 件のスキャン結果）: {CACHE_FILE}")
print(" ※ フォルダ構造は保存時点の内容です。最新化したい場合は USE_CACHE=False にしてください。")
return {"folders": folders}
except Exception:
print("⚠️ キャッシュの読み込みに失敗したため、新規取得します。")
return {"folders": {}}
def save_cache(cache):
try:
with open(CACHE_FILE, "wb") as f:
pickle.dump({"folders": cache["folders"]}, f)
except Exception as e:
print(f"⚠️ キャッシュ保存に失敗: {e}")
CACHE = load_cache()
perm_cache = {} # 権限情報はこの実行中だけ保持（毎回取得し直す）
def _cache_entry(entry):
"""(取得した階層[None=全階層], フォルダ一覧) を返す。旧形式(listそのもの)は全階層として扱う。"""
if isinstance(entry, dict):
return entry.get("depth"), entry.get("folders", [])
return None, entry
def _coverage(depth):
return float("inf") if depth is None else depth
def cache_lookup(display_name, path_lower):
"""今回の MAX_DEPTH を満たすキャッシュがあればフォルダ一覧を返す。なければ None。"""
entry = CACHE["folders"].get(display_name)
if entry is None:
return None
depth, folders = _cache_entry(entry)
if _coverage(depth) < _coverage(MAX_DEPTH):
return None # キャッシュの方が浅い → 使えない
if MAX_DEPTH is None:
return folders
base = path_lower.count('/')
return [f for f in folders if f[1].count('/') - base <= MAX_DEPTH]
def cache_store(display_name, folders):
"""既存キャッシュの方が広い範囲(深い)なら上書きしない。"""
old = CACHE["folders"].get(display_name)
if old is not None:
old_depth, _ = _cache_entry(old)
if _coverage(old_depth) > _coverage(MAX_DEPTH):
print(" ℹ️ 既存のキャッシュの方が深い範囲を含むため、キャッシュは上書きしません。")
return
CACHE["folders"][display_name] = {"depth": MAX_DEPTH, "folders": folders}
save_cache(CACHE)
# ------------------------------------------
# フォルダ一覧の取得
# ------------------------------------------
def to_tuple(entry):
sid = entry.sharing_info.shared_folder_id if entry.sharing_info else None
return (entry.path_lower, entry.path_display, sid)
def list_children(dbx, path_lower):
"""path_lower 直下のフォルダだけ返す（非再帰）。"""
out = []
res = call_with_retry(dbx.files_list_folder, path_lower, recursive=False, limit=LIST_PAGE_LIMIT)
while True:
for e in res.entries:
if isinstance(e, dropbox.files.FolderMetadata):
out.append(to_tuple(e))
if not res.has_more:
break
res = call_with_retry(dbx.files_list_folder_continue, res.cursor)
return out
def list_limited(dbx, folder, remaining):
"""folder 配下を remaining 階層分だけ、非再帰で順に展開する（単一スレッド）。"""
out = []
if remaining <= 0:
return out
kids = list_children(dbx, folder[0])
out.extend(kids)
for k in kids:
out.extend(list_limited(dbx, k, remaining - 1))
return out
def list_subtree(dbx, path_lower, pbar=None):
"""path_lower 配下を再帰的に列挙し、フォルダだけ返す（path自体は含まない）。"""
out = []
res = call_with_retry(dbx.files_list_folder, path_lower, recursive=True, limit=LIST_PAGE_LIMIT)
while True:
n = 0
for e in res.entries:
if isinstance(e, dropbox.files.FolderMetadata):
out.append(to_tuple(e))
n += 1
if pbar is not None and n:
pbar.update(n)
if not res.has_more:
break
res = call_with_retry(dbx.files_list_folder_continue, res.cursor)
return out
def timed_subtree(dbx, folder, pbar):
t0 = time.time()
res = list_subtree(dbx, folder[0], pbar)
return res, time.time() - t0
def collect_folders(dbx, root_lower):
"""
戻り値: (フォルダ一覧, 取得に失敗したフォルダ[(表示パス, エラー内容)])
MAX_DEPTH=None（全階層）:
1) 浅い階層(SPLIT_DEPTH階層目まで)を非再帰で並列に展開
2) 展開した最下段のフォルダごとに、再帰取得を並列実行
MAX_DEPTH=N（階層限定）:
直下から N 階層目まで、1階層ずつ非再帰で並列展開（N階層目の中は取得しない）
"""
limited = MAX_DEPTH is not None
root = (root_lower, root_lower, None)
if not PARALLEL_LISTING:
if limited:
print(f" 🔍 フォルダ構造を取得中（単一スレッド / {MAX_DEPTH}階層目まで）...")
with heartbeat("Dropbox側で走査中"):
return list_limited(dbx, root, MAX_DEPTH), []
print(" 🔍 フォルダ構造を取得中（単一スレッド）...")
with heartbeat("Dropbox側で走査中"):
with tqdm(desc=" 検出フォルダ数", unit="件") as pbar:
return list_subtree(dbx, root_lower, pbar), []
all_folders, failed, timings = [], [], []
t_start = time.time()
levels = MAX_DEPTH if limited else SPLIT_DEPTH
if limited:
print(f" 🔍 {MAX_DEPTH}階層目までを {LIST_WORKERS} 並列で取得します...")
else:
print(f" 🔍 サブツリーに分割して {LIST_WORKERS} 並列で取得します（分割階層: {SPLIT_DEPTH}）...")
with tqdm(desc=" 検出フォルダ数", unit="件") as pbar:
# ---- 浅い階層を非再帰で1階層ずつ展開 ----
frontier = [root]
for lvl in range(1, levels + 1):
if not frontier or (not limited and len(frontier) > MAX_SPLIT_TASKS):
break
next_frontier = []
with ThreadPoolExecutor(max_workers=LIST_WORKERS) as ex:
futs = {ex.submit(list_children, dbx, f[0]): f for f in frontier}
for fut in as_completed(futs):
f = futs[fut]
try:
kids = fut.result()
except Exception as e:
# 限定取得: 残りの階層分だけ後で再試行 / 全階層: 配下をまるごと再帰で再試行
failed.append((f, e, (levels - lvl + 1) if limited else None))
continue
all_folders.extend(kids)
next_frontier.extend(kids)
pbar.update(len(kids))
frontier = next_frontier
t1 = time.time() - t_start
if limited:
pbar.write(f" ✅ {MAX_DEPTH}階層目までの取得が完了（{len(all_folders):,}件 / {t1:.0f}秒）")
else:
pbar.write(f" ✅ 浅い階層の展開が完了（{len(all_folders):,}件 / {t1:.0f}秒）"
f" → {len(frontier):,} 個のサブツリーを並列で再帰取得します")
# ---- 各サブツリーを再帰取得 ----
done = 0
with ThreadPoolExecutor(max_workers=LIST_WORKERS) as ex:
futs = {ex.submit(timed_subtree, dbx, f, pbar): f for f in frontier}
for fut in as_completed(futs):
f = futs[fut]
try:
sub, sec = fut.result()
all_folders.extend(sub)
timings.append((sec, len(sub), f[1]))
except Exception as e:
failed.append((f, e, None))
done += 1
pbar.set_postfix(サブツリー=f"{done}/{len(frontier)}")
# ---- 失敗分を1回だけ単独で再試行 ----
still_failed = []
if failed:
print(f" 🔁 取得に失敗した {len(failed)} 件を再試行します...")
for f, _, remaining in failed:
try:
if remaining is None:
all_folders.extend(list_subtree(dbx, f[0]))
else:
all_folders.extend(list_limited(dbx, f, remaining))
except Exception as e:
still_failed.append((f[1], str(e)))
# 重複除去（念のため）
uniq = {}
for f in all_folders:
uniq[f[0]] = f
all_folders = list(uniq.values())
total = time.time() - t_start
print(f" ✅ フォルダ構造の検出が完了しました（合計: {len(all_folders):,} 件 / {total / 60:.1f}分）")
# ---- 所要時間が長かったサブツリー（全階層取得のみ）----
if timings:
timings.sort(reverse=True)
print(f" 🐢 所要時間が長かったサブツリー上位{min(SLOW_TOP_N, len(timings))}件:")
for sec, n, disp in timings[:SLOW_TOP_N]:
print(f" {sec / 60:6.1f}分 {n:>8,}件 {disp}")
print(f" ※ 最長サブツリーが全体の所要時間に近い場合は、そこで待たされています。"
f"SPLIT_DEPTH を増やすと分割が細かくなります。")
return all_folders, still_failed
# ------------------------------------------
# 権限情報の取得
# ------------------------------------------
def fetch_permission(dbx, shared_id):
if shared_id in perm_cache:
return
try:
info = call_with_retry(dbx.sharing_list_folder_members, shared_id, limit=1000)
groups, users = list(info.groups), list(info.users)
while info.cursor: # 1000件超のメンバーにも対応
info = call_with_retry(dbx.sharing_list_folder_members_continue, info.cursor)
groups.extend(info.groups)
users.extend(info.users)
targets = [f"[グループ]{g.group.group_name}({g.access_type._tag})" for g in groups]
targets += [f"[ユーザー]{u.user.display_name}({u.access_type._tag})" for u in users]
perm_cache[shared_id] = ("あり: " + ", ".join(targets)) if targets else "あり（詳細なし）"
except Exception:
perm_cache[shared_id] = "あり（照会不可）"
def batch_fetch_permissions(dbx, items):
if not SHOW_PERMISSION_DETAILS:
return
todo = {it['shared_id'] for it in items
if it['depth'] in PERM_CHECK_DEPTHS and it['shared_id'] and it['shared_id'] not in perm_cache}
if not todo:
return
print(f" ⚡ 権限情報を並列取得中... ({len(todo)} 件の shared_id)")
with ThreadPoolExecutor(max_workers=PERM_WORKERS) as ex:
futures = [ex.submit(fetch_permission, dbx, sid) for sid in todo]
for _ in tqdm(as_completed(futures), total=len(futures), desc=" 権限情報取得進捗"):
pass
def perm_text(item):
if item['depth'] not in PERM_CHECK_DEPTHS:
return None
if not item['shared_id']:
return "なし（継承）"
if not SHOW_PERMISSION_DETAILS:
return "あり"
return perm_cache.get(item['shared_id'], "あり（照会不可）")
# ------------------------------------------
# スキャン本体
# ------------------------------------------
def scan_target_zone(dbx, path_lower, display_name):
scope = "全階層" if MAX_DEPTH is None else f"{MAX_DEPTH}階層目まで"
print(f"\n📂 【{display_name}】のスキャンを開始します（取得範囲: {scope}）...")
folders = cache_lookup(display_name, path_lower) if USE_CACHE else None
if folders is not None:
print(f" 💾 キャッシュを使用します（{len(folders):,} 件のフォルダ）")
else:
try:
folders, failed = collect_folders(dbx, path_lower)
except Exception as e:
print(f"\n ※アクセスエラー: {e}")
return [], 0
if failed:
print(f"\n ⚠️ {len(failed)} 件のフォルダ配下を取得できませんでした（結果に含まれません）:")
for disp, err in failed[:10]:
print(f" - {disp}: {err}")
print(" ⚠️ 不完全なデータのため、キャッシュには保存しません。")
else:
cache_store(display_name, folders)
if not folders:
return [], 0
print(" ⚙️ 階層の計算と構造化を実行中...")
# パス区切り(/)単位で並べる。文字列全体で並べると「foo」「foo bar」「foo/baz」の順で
# 親の塊が分断され、グループ化や件数集計がずれるため。
folders = sorted(folders, key=lambda x: x[0].split('/'))
base_depth = path_lower.count('/')
items = []
max_depth = 0
for _, disp, sid in folders:
parts = [clean_for_excel(p) for p in disp.strip('/').split('/') if p][base_depth:]
depth = len(parts)
max_depth = max(max_depth, depth)
items.append({'parts': parts, 'depth': depth, 'shared_id': sid})
n_shared = sum(1 for it in items if it['shared_id'])
deep_shared = sum(1 for it in items if it['shared_id'] and it['depth'] > 7)
print(f" 📏 最大階層: {max_depth} / フォルダ: {len(items):,} 件 / "
f"共有フォルダ: {n_shared:,} 件（うち8階層目以降: {deep_shared:,} 件）")
batch_fetch_permissions(dbx, items)
for it in items:
it['perm'] = perm_text(it)
return items, max_depth
def build_table(items, max_depth):
"""行データ・重複判定フラグ・白字にするセルを、Excelに触れる前に純Pythonで一括計算する。"""
rows, flags, dups, depths = [], [], [], []
prev = [""] * max_depth
for it in items:
vals = it['parts'] + [""] * (max_depth - len(it['parts']))
k = 0 # 直前行と最初に異なる階層の位置
while k < max_depth and vals[k] == prev[k]:
k += 1
flags.append([1 if (vals[d] and d >= k) else 0 for d in range(max_depth)])
dups.append([d + 1 for d in range(k) if vals[d]]) # 直前行と同じ親・同じ名前 → 白字
depths.append(it['depth'])
row = []
for d in range(1, max_depth + 1):
row.append(vals[d - 1])
if d in PERM_CHECK_DEPTHS:
row.append(it['perm'] if d == it['depth'] else "")
rows.append(row)
prev = vals
return rows, flags, dups, depths
def write_sheet(writer, sheet_name, items, max_depth):
max_depth = max(max_depth, 1)
columns, folder_col, perm_cols = [], {}, []
col = 1
for i in range(1, max_depth + 1):
columns.append(f"{i}階層目")
folder_col[i] = col
col += 1
if i in PERM_CHECK_DEPTHS:
columns.append(f"{i}階層目の権限")
perm_cols.append(col)
col += 1
data_cols = len(columns)
flag_names = [f"_flag_{i}" for i in range(1, max_depth + 1)]
if items:
rows, flags, dups, depths = build_table(items, max_depth)
table = [r + f for r, f in zip(rows, flags)]
else:
dummy = []
for d in range(1, max_depth + 1):
dummy.append("対象データなし")
if d in PERM_CHECK_DEPTHS:
dummy.append("-")
rows, dups, depths = [dummy], [[]], [1]
table = [dummy + [0] * max_depth]
clean_name = re.sub(r'[\\*?:/\[\]]', '', sheet_name)[:31]
pd.DataFrame(table, columns=columns + flag_names).to_excel(
writer, sheet_name=clean_name, index=False, startrow=1)
ws = writer.book[clean_name]
last_row = len(table) + 2 # データは3行目から
# 1行目：階層ごとのユニーク件数（SUBTOTALなのでフィルタに追従）
for d in range(1, max_depth + 1):
letter = get_column_letter(data_cols + d)
c = ws.cell(row=1, column=folder_col[d], value=f"=SUBTOTAL(109,{letter}3:{letter}{last_row})")
c.font = Font(bold=True, color="002060")
c.fill = PatternFill(start_color="FFF2CC", end_color="FFF2CC", fill_type="solid")
c.alignment = Alignment(horizontal="center")
# ヘッダー
header_fill = PatternFill(start_color="D9E1F2", end_color="D9E1F2", fill_type="solid")
perm_fill = PatternFill(start_color="E2EFDA", end_color="E2EFDA", fill_type="solid")
for c in range(1, data_cols + 1):
h = ws.cell(row=2, column=c)
h.font = Font(bold=True)
h.fill = perm_fill if c in perm_cols else header_fill
ws.auto_filter.ref = f"A2:{get_column_letter(data_cols)}{last_row}"
ws.freeze_panes = "A3"
# 重複セルを白字に（該当セルだけ触る）
white = Font(color="FFFFFF")
for i, ds in enumerate(dups):
r = 3 + i
for d in ds:
ws.cell(row=r, column=folder_col[d]).font = white
# グループ化（アウトライン）。Excelの上限は7段階のため、8階層目以降は同じレベルにまとまる
if items:
for i, depth in enumerate(depths):
if depth > 1:
ws.row_dimensions[3 + i].outlineLevel = min(depth - 1, 7)
ws.sheet_properties.outlinePr.summaryBelow = False
# 列幅（最大60）
for c in range(1, data_cols + 1):
longest = len(str(columns[c - 1]))
for r in rows:
v = r[c - 1]
if v:
longest = max(longest, len(str(v)))
ws.column_dimensions[get_column_letter(c)].width = min(longest * 1.6 + 2, 60)
# 補助フラグ列は非表示
for d in range(1, max_depth + 1):
idx = data_cols + d
ws.cell(row=2, column=idx, value="")
ws.column_dimensions[get_column_letter(idx)].hidden = True
def output_file():
"""階層限定のときは、全階層版を上書きしないよう末尾に _depthN を付ける。"""
if MAX_DEPTH is None:
return EXCEL_FILE
root, ext = os.path.splitext(EXCEL_FILE)
return f"{root}_depth{MAX_DEPTH}{ext}"
# ------------------------------------------
# メイン処理
# ------------------------------------------
try:
start_time = time.time()
dbx_team = dropbox.DropboxTeam(
app_key=APP_KEY,
app_secret=APP_SECRET,
oauth2_refresh_token=REFRESH_TOKEN
)
members_res = dbx_team.team_members_list()
all_members = list(members_res.members)
while members_res.has_more:
members_res = dbx_team.team_members_list_continue(members_res.cursor)
all_members.extend(members_res.members)
target_member_id, matched_email = None, None
clean_email = (TARGET_ADMIN_EMAIL or "").strip().lower()
for m in all_members:
if not m.profile.status.is_active():
continue
if not clean_email or m.profile.email.lower() == clean_email:
target_member_id = m.profile.team_member_id
matched_email = m.profile.email
break
if not target_member_id:
raise ValueError(f"指定されたメールアドレス '{TARGET_ADMIN_EMAIL}' を持つアクティブなチームメンバーが見つかりませんでした。")
print(f"👤 実行ユーザー: {matched_email} (ID: {target_member_id})")
dbx_user = dbx_team.as_user(target_member_id)
root_id = dbx_user.users_get_current_account().root_info.root_namespace_id
dbx = dbx_user.with_path_root(dropbox.common.PathRoot('namespace_id', root_id))
print("🔄 最上位ルートに正常接続しました。")
# ルート直下から対象フォルダを検索
target_paths = {}
res_root = call_with_retry(dbx.files_list_folder, '', recursive=False)
while True:
for entry in res_root.entries:
if isinstance(entry, dropbox.files.FolderMetadata):
name = entry.path_display.strip("/").split("/")[-1]
for target_name in TARGET_FOLDERS:
if target_name.lower() in name.lower():
target_paths[target_name] = (entry.path_lower, name)
if not res_root.has_more:
break
res_root = call_with_retry(dbx.files_list_folder_continue, res_root.cursor)
all_sheet_data = {}
for target_name in TARGET_FOLDERS:
if target_name not in target_paths:
print(f"⚠️ 対象フォルダが見つかりません: {target_name}")
continue
path_lower, display_name = target_paths[target_name]
items, max_d = scan_target_zone(dbx, path_lower, display_name)
all_sheet_data[display_name] = (items, max_d)
out_file = output_file()
print("\n📊 データをExcelファイルに書き込み、フォーマットを行っています...")
with pd.ExcelWriter(out_file, engine='openpyxl') as writer:
for sheet_name, (items, max_d) in all_sheet_data.items():
write_sheet(writer, sheet_name, items, max_d)
elapsed = time.time() - start_time
print("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
print("🎉 【処理完了】階層別権限列付きExcelが完成しました！")
print(f"⏱️ 総処理時間: {int(elapsed // 60)} 分 {int(elapsed % 60)} 秒")
print(f"📁 出力ファイル: {out_file}")
print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
except Exception as e:
print(f"\n❌ エラーが発生しました: {e}")
あまり需要はないかもしれないけどDropBoxのフォルダを整理するときに使えるんじゃないかとは思います。
コメント