martin/Sharepoint-Download-Tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add support for downloading all document libraries on a site

Browse Source 
- New get_all_drives() fetches all drives from a SharePoint site
- main() loops over all drives when DOCUMENT_LIBRARY is empty, placing
  each library in its own subfolder under LOCAL_PATH
- MSAL force_refresh now catches TypeError/ValueError for compatibility
  with older MSAL versions that don't support the parameter
- GUI: "Download alle biblioteker" checkbox disables Library Navn and
  Mapper fields; load_settings restores checkbox state from config

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martin Tranberg
2026-05-04 13:41:41 +02:00
parent d15b9afc03
commit 24665c5797
2 changed files with 81 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
download_sharepoint.py
+33 -12
View File
@@ -175,15 +175,19 @@ def verify_integrity(local_path, remote_hash, config):
def get_headers(app, force_refresh=False):
scopes = ["https://graph.microsoft.com/.default"]
# If force_refresh is True, we don't rely on the cache
result = None
if not force_refresh:
result = app.acquire_token_for_client(scopes=scopes)
if force_refresh or not result or "access_token" not in result:
logger.info("Refreshing Access Token...")
result = app.acquire_token_for_client(scopes=scopes, force_refresh=True)
try:
result = app.acquire_token_for_client(scopes=scopes, force_refresh=True)
except (TypeError, ValueError):
# Ældre MSAL-versioner understøtter ikke force_refresh på acquire_token_for_client.
# MSAL håndterer udløbne tokens automatisk, så et nyt kald er tilstrækkeligt.
result = app.acquire_token_for_client(scopes=scopes)
if "access_token" in result:
return {'Authorization': f'Bearer {result["access_token"]}'}
raise Exception(f"Auth failed: {result.get('error_description')}")
@@ -198,12 +202,12 @@ def get_drive_id(app, site_id, drive_name):
url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives"
response = safe_graph_get(app, url)
drives = response.json().get('value', [])
# Prøv præcis match
for drive in drives:
if drive['name'] == drive_name:
return drive['id']
# Prøv fallback til "Documents" hvis "Delte dokumenter" fejler (SharePoint standard)
if drive_name == "Delte dokumenter":
for drive in drives:
@@ -216,6 +220,11 @@ def get_drive_id(app, site_id, drive_name):
logger.error(f"Drive '{drive_name}' not found. Available drives on this site: {available_names}")
raise Exception(f"Drive {drive_name} not found. Check the log for available drive names.")
def get_all_drives(app, site_id):
url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives"
response = safe_graph_get(app, url)
return response.json().get('value', [])
# --- Punkt 2: Resume / Chunked Download logic ---
def get_fresh_download_url(app, drive_id, item_id):
"""Fetches a fresh download URL for a specific item ID with retries and robust error handling."""
@@ -414,20 +423,32 @@ def main(config=None, stop_event=None):
local_base = config.get('LOCAL_PATH', '').replace('\\', os.sep)
folders = [f.strip() for f in folders_str.split(',') if f.strip()] or [""]
logger.info("Initializing SharePoint Production Sync Tool...")
app = create_msal_app(tenant_id, client_id, client_secret)
site_id = get_site_id(app, site_url)
drive_id = get_drive_id(app, site_id, drive_name)
if not drive_name:
drives = get_all_drives(app, site_id)
logger.info(f"Downloading all {len(drives)} document libraries: {[d['name'] for d in drives]}")
use_subfolder = True
folders = [""] # Download fra rod af hvert bibliotek
else:
drives = [{'id': get_drive_id(app, site_id, drive_name), 'name': drive_name}]
use_subfolder = False
report = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS, thread_name_prefix="DL") as executor:
futures = {}
for folder in folders:
for drive in drives:
if stop_event and stop_event.is_set():
break
logger.info(f"Scanning: {folder or 'Root'}")
process_item_list(app, drive_id, folder, os.path.join(local_base, folder), report, executor, futures, config, stop_event)
drive_local_base = os.path.join(local_base, drive['name']) if use_subfolder else local_base
for folder in folders:
if stop_event and stop_event.is_set():
break
logger.info(f"[{drive['name']}] Scanning: {folder or 'Root'}")
process_item_list(app, drive['id'], folder, os.path.join(drive_local_base, folder), report, executor, futures, config, stop_event)
logger.info(f"Scan complete. Processing {len(futures)} tasks...")
for future in as_completed(futures):