From 1289d20ea47463ac22b9259c9eed8b838f108cee Mon Sep 17 00:00:00 2001 From: jerubball Date: Fri, 30 Oct 2026 11:14:46 -0400 Subject: [PATCH 1/4] run formatter and organize import to pass black and flake8 checks --- examples/confluence-diagnostics.py | 442 ++++++++++-------- .../confluence-space-attachments-download.py | 303 ++++++------ 2 files changed, 414 insertions(+), 331 deletions(-) diff --git a/examples/confluence-diagnostics.py b/examples/confluence-diagnostics.py index e27f451e1..e14a06456 100644 --- a/examples/confluence-diagnostics.py +++ b/examples/confluence-diagnostics.py @@ -14,14 +14,6 @@ The token is never written to the log in full. """ -# Used only when confluence-space-attachments-download.py is not next to this file -CONFLUENCE_URL = 'https://confluence.example.com' -CONFLUENCE_TOKEN = 'PASTE-YOUR-TOKEN-HERE' -SPACE_KEY = 'DOCS' -OUTPUT_DIR = './confluence-export' -VERIFY_SSL = True -DIRECT_CONNECTION = True - import ast import json import os @@ -36,60 +28,79 @@ try: # Python 3 only, but give a readable message on Python 2 from urllib.parse import urlparse, urlencode from urllib.request import Request, urlopen, getproxies - from urllib.error import HTTPError, URLError + from urllib.error import HTTPError, URLError # noqa: F401 except ImportError: - print('ERROR: this script needs Python 3, you are running Python %s' % sys.version.split()[0]) + print("ERROR: this script needs Python 3, you are running Python %s" % sys.version.split()[0]) raise SystemExit(1) + +# Used only when confluence-space-attachments-download.py is not next to this file +CONFLUENCE_URL = "https://confluence.example.com" +CONFLUENCE_TOKEN = "PASTE-YOUR-TOKEN-HERE" +SPACE_KEY = "DOCS" +OUTPUT_DIR = "./confluence-export" +VERIFY_SSL = True +DIRECT_CONNECTION = True + + SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) -MAIN_SCRIPT = os.path.join(SCRIPT_DIR, 'confluence-space-attachments-download.py') -LOG_FILE = os.path.join(SCRIPT_DIR, 'confluence-diagnostics.log') +MAIN_SCRIPT = os.path.join(SCRIPT_DIR, "confluence-space-attachments-download.py") +LOG_FILE = os.path.join(SCRIPT_DIR, "confluence-diagnostics.log") TIMEOUT = 30 -_log = open(LOG_FILE, 'w', encoding='utf-8') +_log = open(LOG_FILE, "w", encoding="utf-8") results = [] # (step, status) -def out(msg=''): - line = '%s %s' % (time.strftime('%H:%M:%S'), msg) if msg else '' +def out(msg=""): + line = "%s %s" % (time.strftime("%H:%M:%S"), msg) if msg else "" print(line) - _log.write(line + '\n') + _log.write(line + "\n") _log.flush() def step(title): out() - out('=== %s ===' % title) + out("=== %s ===" % title) def ok(name, msg): - results.append((name, 'OK')) - out('[OK] ' + msg) + results.append((name, "OK")) + out("[OK] " + msg) def warn(name, msg): - results.append((name, 'WARN')) - out('[WARN] ' + msg) + results.append((name, "WARN")) + out("[WARN] " + msg) def fail(name, msg): - results.append((name, 'FAIL')) - out('[FAIL] ' + msg) + results.append((name, "FAIL")) + out("[FAIL] " + msg) def mask(token): if not token: - return '' - return '%s...%s (length %d)' % (token[:4], token[-2:], len(token)) if len(token) > 8 else '' % len(token) + return "" + return ( + "%s...%s (length %d)" % (token[:4], token[-2:], len(token)) + if len(token) > 8 + else "" % len(token) + ) def read_config(): - cfg = {'CONFLUENCE_URL': CONFLUENCE_URL, 'CONFLUENCE_TOKEN': CONFLUENCE_TOKEN, - 'SPACE_KEY': SPACE_KEY, 'OUTPUT_DIR': OUTPUT_DIR, 'VERIFY_SSL': VERIFY_SSL, - 'DIRECT_CONNECTION': DIRECT_CONNECTION} + cfg = { + "CONFLUENCE_URL": CONFLUENCE_URL, + "CONFLUENCE_TOKEN": CONFLUENCE_TOKEN, + "SPACE_KEY": SPACE_KEY, + "OUTPUT_DIR": OUTPUT_DIR, + "VERIFY_SSL": VERIFY_SSL, + "DIRECT_CONNECTION": DIRECT_CONNECTION, + } if not os.path.exists(MAIN_SCRIPT): - return cfg, 'this file (main script not found next to it)' - with open(MAIN_SCRIPT, encoding='utf-8-sig') as f: + return cfg, "this file (main script not found next to it)" + with open(MAIN_SCRIPT, encoding="utf-8-sig") as f: tree = ast.parse(f.read()) for node in tree.body: if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): @@ -99,8 +110,8 @@ def read_config(): cfg[name] = ast.literal_eval(node.value) except ValueError: pass - if os.environ.get('CONFLUENCE_TOKEN'): - cfg['CONFLUENCE_TOKEN'] = os.environ['CONFLUENCE_TOKEN'] + if os.environ.get("CONFLUENCE_TOKEN"): + cfg["CONFLUENCE_TOKEN"] = os.environ["CONFLUENCE_TOKEN"] return cfg, MAIN_SCRIPT @@ -110,28 +121,29 @@ def system_ca_bundle(): so company CAs (e.g. an internal "Issuing CA") are rejected. Build a PEM bundle with certifi + trusted Windows ROOT/CA certificates and return its path (None if not Windows). """ - if not hasattr(ssl, 'enum_certificates'): # only exists on Windows + if not hasattr(ssl, "enum_certificates"): # only exists on Windows return None, 0 pems, count = [], 0 try: import certifi - with open(certifi.where(), encoding='utf-8') as f: + + with open(certifi.where(), encoding="utf-8") as f: pems.append(f.read()) except (ImportError, OSError): pass - for store in ('ROOT', 'CA'): + for store in ("ROOT", "CA"): try: certificates = ssl.enum_certificates(store) except OSError: continue for der, encoding, trust in certificates: # trust is True (all purposes) or a set of OIDs; 1.3.6.1.5.5.7.3.1 = server authentication - if encoding == 'x509_asn' and (trust is True or '1.3.6.1.5.5.7.3.1' in trust): + if encoding == "x509_asn" and (trust is True or "1.3.6.1.5.5.7.3.1" in trust): pems.append(ssl.DER_cert_to_PEM_cert(der)) count += 1 - path = os.path.join(tempfile.gettempdir(), 'confluence-ca-bundle.pem') - with open(path, 'w', encoding='utf-8') as f: - f.write('\n'.join(pems)) + path = os.path.join(tempfile.gettempdir(), "confluence-ca-bundle.pem") + with open(path, "w", encoding="utf-8") as f: + f.write("\n".join(pems)) return path, count @@ -146,36 +158,38 @@ def resolve_verify(setting): def bypass_proxy(url): """Connect to Confluence directly, ignoring the system/company proxy, for this host only.""" host = urlparse(url).hostname - for key in ('NO_PROXY', 'no_proxy'): - current = os.environ.get(key, '') - os.environ[key] = ','.join(x for x in (current, host) if x) + for key in ("NO_PROXY", "no_proxy"): + current = os.environ.get(key, "") + os.environ[key] = ",".join(x for x in (current, host) if x) class Http: """Same requests as the main script; uses 'requests' if installed, else urllib.""" def __init__(self, base, token, verify): - self.base = base.rstrip('/') + self.base = base.rstrip("/") self.token = token self.verify = verify try: import requests + self.requests = requests self.session = requests.Session() except ImportError: self.requests = None def get(self, path, params=None, auth=True, max_bytes=None): - url = path if path.startswith('http') else self.base + path + url = path if path.startswith("http") else self.base + path if params: - url += '?' + urlencode(params) - headers = {'Accept': 'application/json'} + url += "?" + urlencode(params) + headers = {"Accept": "application/json"} if auth: - headers['Authorization'] = 'Bearer ' + self.token + headers["Authorization"] = "Bearer " + self.token started = time.time() if self.requests: - r = self.session.get(url, headers=headers, timeout=TIMEOUT, stream=True, allow_redirects=False, - verify=self.verify) + r = self.session.get( + url, headers=headers, timeout=TIMEOUT, stream=True, allow_redirects=False, verify=self.verify + ) body = r.raw.read(max_bytes) if max_bytes else r.content status, hdrs = r.status_code, r.headers r.close() @@ -189,293 +203,327 @@ def get(self, path, params=None, auth=True, max_bytes=None): body = resp.read(max_bytes) if max_bytes else resp.read() except HTTPError as e: status, hdrs, body = e.code, e.headers, e.read() - out(' GET %s -> HTTP %s (%.1fs)' % (url, status, time.time() - started)) + out(" GET %s -> HTTP %s (%.1fs)" % (url, status, time.time() - started)) return status, hdrs, body def body_preview(body, n=300): - text = body.decode('utf-8', 'replace') if isinstance(body, bytes) else str(body) - text = ' '.join(text.split()) - return text[:n] + ('...' if len(text) > n else '') + text = body.decode("utf-8", "replace") if isinstance(body, bytes) else str(body) + text = " ".join(text.split()) + return text[:n] + ("..." if len(text) > n else "") def main(): - out('Confluence diagnostics - log file: %s' % LOG_FILE) + out("Confluence diagnostics - log file: %s" % LOG_FILE) # 1 ----------------------------------------------------------------- - step('1. Python environment') - out('Python: %s' % sys.version.replace('\n', ' ')) - out('Executable: %s' % sys.executable) - out('System: %s %s (%s)' % (platform.system(), platform.release(), platform.machine())) - out('Script dir: %s' % SCRIPT_DIR) - out('Working dir:%s' % os.getcwd()) - out('OpenSSL: %s' % ssl.OPENSSL_VERSION) + step("1. Python environment") + out("Python: %s" % sys.version.replace("\n", " ")) + out("Executable: %s" % sys.executable) + out("System: %s %s (%s)" % (platform.system(), platform.release(), platform.machine())) + out("Script dir: %s" % SCRIPT_DIR) + out("Working dir:%s" % os.getcwd()) + out("OpenSSL: %s" % ssl.OPENSSL_VERSION) if sys.version_info < (3, 6): - fail('python', 'Python 3.6+ required') + fail("python", "Python 3.6+ required") else: - ok('python', 'Python version is fine') + ok("python", "Python version is fine") try: import requests - ok('requests', 'requests %s installed (%s)' % (requests.__version__, os.path.dirname(requests.__file__))) + + ok("requests", "requests %s installed (%s)" % (requests.__version__, os.path.dirname(requests.__file__))) try: import certifi - out(' certifi CA bundle: %s' % certifi.where()) + + out(" certifi CA bundle: %s" % certifi.where()) except ImportError: pass except ImportError: - fail('requests', 'requests NOT installed - the main script cannot start. Run: "%s" -m pip install requests' - % sys.executable) + fail( + "requests", + 'requests NOT installed - the main script cannot start. Run: "%s" -m pip install requests' + % sys.executable, + ) # 2 ----------------------------------------------------------------- - step('2. Configuration') + step("2. Configuration") cfg, source = read_config() - out('Read from: %s' % source) - out('Main script present: %s' % os.path.exists(MAIN_SCRIPT)) - url, token, space_key = cfg['CONFLUENCE_URL'], cfg['CONFLUENCE_TOKEN'], cfg['SPACE_KEY'] - out('URL: %r' % url) - out('Token: %s' % mask(token)) - out('Space key: %r' % space_key) - out('Output dir: %r -> %s' % (cfg['OUTPUT_DIR'], os.path.abspath(cfg['OUTPUT_DIR']))) - out('VERIFY_SSL: %r' % (cfg['VERIFY_SSL'],)) - out('DIRECT_CONNECTION: %r' % (cfg['DIRECT_CONNECTION'],)) + out("Read from: %s" % source) + out("Main script present: %s" % os.path.exists(MAIN_SCRIPT)) + url, token, space_key = cfg["CONFLUENCE_URL"], cfg["CONFLUENCE_TOKEN"], cfg["SPACE_KEY"] + out("URL: %r" % url) + out("Token: %s" % mask(token)) + out("Space key: %r" % space_key) + out("Output dir: %r -> %s" % (cfg["OUTPUT_DIR"], os.path.abspath(cfg["OUTPUT_DIR"]))) + out("VERIFY_SSL: %r" % (cfg["VERIFY_SSL"],)) + out("DIRECT_CONNECTION: %r" % (cfg["DIRECT_CONNECTION"],)) problems = [] - if 'example.com' in url: - problems.append('CONFLUENCE_URL still has the example value') - if not url.startswith(('http://', 'https://')): - problems.append('CONFLUENCE_URL must start with https:// (or http://)') - if '/display/' in url or '/pages/' in url or '/spaces/' in url: - problems.append('CONFLUENCE_URL looks like a page link - use only the base address, e.g. https://wiki.company.com') - if not token or token.startswith('PASTE-'): - problems.append('CONFLUENCE_TOKEN is not filled in') + if "example.com" in url: + problems.append("CONFLUENCE_URL still has the example value") + if not url.startswith(("http://", "https://")): + problems.append("CONFLUENCE_URL must start with https:// (or http://)") + if "/display/" in url or "/pages/" in url or "/spaces/" in url: + problems.append( + "CONFLUENCE_URL looks like a page link - use only the base address, e.g. https://wiki.company.com" + ) + if not token or token.startswith("PASTE-"): + problems.append("CONFLUENCE_TOKEN is not filled in") elif token != token.strip(): - problems.append('CONFLUENCE_TOKEN has spaces/newlines at the beginning or end') + problems.append("CONFLUENCE_TOKEN has spaces/newlines at the beginning or end") if not space_key or space_key != space_key.strip(): - problems.append('SPACE_KEY is empty or has spaces') - if isinstance(cfg['VERIFY_SSL'], str) and not os.path.exists(cfg['VERIFY_SSL']): - problems.append('VERIFY_SSL points to a file that does not exist: %s' % cfg['VERIFY_SSL']) + problems.append("SPACE_KEY is empty or has spaces") + if isinstance(cfg["VERIFY_SSL"], str) and not os.path.exists(cfg["VERIFY_SSL"]): + problems.append("VERIFY_SSL points to a file that does not exist: %s" % cfg["VERIFY_SSL"]) for p in problems: - fail('config', p) + fail("config", p) if problems: return - ok('config', 'configuration looks complete') + ok("config", "configuration looks complete") parsed = urlparse(url) host = parsed.hostname - port = parsed.port or (443 if parsed.scheme == 'https' else 80) + port = parsed.port or (443 if parsed.scheme == "https" else 80) # 3 ----------------------------------------------------------------- - step('3. Proxy settings') + step("3. Proxy settings") proxies = getproxies() - out('Proxies seen by Python: %s' % (proxies or 'none')) - for name in ('HTTP_PROXY', 'HTTPS_PROXY', 'NO_PROXY', 'REQUESTS_CA_BUNDLE', 'SSL_CERT_FILE'): + out("Proxies seen by Python: %s" % (proxies or "none")) + for name in ("HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "REQUESTS_CA_BUNDLE", "SSL_CERT_FILE"): for key in (name, name.lower()): if os.environ.get(key): - out(' env %s=%s' % (key, os.environ[key])) - if cfg['DIRECT_CONNECTION']: + out(" env %s=%s" % (key, os.environ[key])) + if cfg["DIRECT_CONNECTION"]: bypass_proxy(url) - out(' DIRECT_CONNECTION=True -> proxy bypassed for %s' % host) + out(" DIRECT_CONNECTION=True -> proxy bypassed for %s" % host) try: import requests + used = requests.utils.get_environ_proxies(url) - out(' proxy that requests will use for Confluence: %s' % (used.get(parsed.scheme) or 'none (direct)')) + out(" proxy that requests will use for Confluence: %s" % (used.get(parsed.scheme) or "none (direct)")) except ImportError: pass - ok('proxy', 'proxy info collected') + ok("proxy", "proxy info collected") # 4 ----------------------------------------------------------------- - step('4. DNS: %s' % host) + step("4. DNS: %s" % host) try: addrs = sorted({a[4][0] for a in socket.getaddrinfo(host, port)}) - ok('dns', '%s resolves to %s' % (host, ', '.join(addrs))) + ok("dns", "%s resolves to %s" % (host, ", ".join(addrs))) except socket.gaierror as e: - fail('dns', 'cannot resolve %s: %s - wrong address, or VPN not connected?' % (host, e)) + fail("dns", "cannot resolve %s: %s - wrong address, or VPN not connected?" % (host, e)) return # 5 ----------------------------------------------------------------- - step('5. TCP connection to %s:%d' % (host, port)) + step("5. TCP connection to %s:%d" % (host, port)) try: started = time.time() socket.create_connection((host, port), timeout=TIMEOUT).close() - ok('tcp', 'port %d open (%.2fs)' % (port, time.time() - started)) + ok("tcp", "port %d open (%.2fs)" % (port, time.time() - started)) except OSError as e: if proxies: - warn('tcp', 'direct connection failed (%s) - may be fine if traffic goes through the proxy' % e) + warn("tcp", "direct connection failed (%s) - may be fine if traffic goes through the proxy" % e) else: - fail('tcp', 'cannot connect: %s - firewall/VPN?' % e) + fail("tcp", "cannot connect: %s - firewall/VPN?" % e) return # 6 ----------------------------------------------------------------- - step('6. HTTPS / SSL certificate') - if parsed.scheme == 'https': + step("6. HTTPS / SSL certificate") + if parsed.scheme == "https": try: ctx = ssl.create_default_context() with socket.create_connection((host, port), timeout=TIMEOUT) as sock: with ctx.wrap_socket(sock, server_hostname=host) as tls: cert = tls.getpeercert() - issuer = dict(x[0] for x in cert.get('issuer', ())) - out(' %s, issuer: %s, valid until: %s' % (tls.version(), issuer.get('organizationName') or issuer.get('commonName'), cert.get('notAfter'))) - ok('ssl-system', 'certificate trusted by the system (OpenSSL) store') + issuer = dict(x[0] for x in cert.get("issuer", ())) + out( + " %s, issuer: %s, valid until: %s" + % ( + tls.version(), + issuer.get("organizationName") or issuer.get("commonName"), + cert.get("notAfter"), + ) + ) + ok("ssl-system", "certificate trusted by the system (OpenSSL) store") except ssl.SSLError as e: - warn('ssl-system', 'system store does not trust the certificate: %s' % e) + warn("ssl-system", "system store does not trust the certificate: %s" % e) except OSError as e: - warn('ssl-system', 'could not test directly: %s' % e) + warn("ssl-system", "could not test directly: %s" % e) else: - warn('ssl-system', 'plain http:// - no TLS') + warn("ssl-system", "plain http:// - no TLS") - verify = resolve_verify(cfg['VERIFY_SSL']) - if parsed.scheme == 'https': - step('6b. requests: certifi only vs. Windows certificate store') + verify = resolve_verify(cfg["VERIFY_SSL"]) + if parsed.scheme == "https": + step("6b. requests: certifi only vs. Windows certificate store") bundle, count = system_ca_bundle() - out(' Windows store: %s' % ('%d trusted certificates -> %s' % (count, bundle) if bundle else 'not available (not Windows)')) + out( + " Windows store: %s" + % ("%d trusted certificates -> %s" % (count, bundle) if bundle else "not available (not Windows)") + ) try: import requests - variants = [('certifi only', True)] + ([('certifi + Windows store', bundle)] if bundle else []) + + variants = [("certifi only", True)] + ([("certifi + Windows store", bundle)] if bundle else []) for label, value in variants: try: - requests.get(url.rstrip('/') + '/status', verify=value, timeout=TIMEOUT) - out(' %-24s -> OK' % label) + requests.get(url.rstrip("/") + "/status", verify=value, timeout=TIMEOUT) + out(" %-24s -> OK" % label) except requests.exceptions.SSLError as e: - out(' %-24s -> SSL ERROR: %s' % (label, str(e)[:200])) + out(" %-24s -> SSL ERROR: %s" % (label, str(e)[:200])) except requests.RequestException as e: - out(' %-24s -> %s: %s' % (label, type(e).__name__, str(e)[:200])) + out(" %-24s -> %s: %s" % (label, type(e).__name__, str(e)[:200])) except ImportError: pass - ok('ssl-compare', 'comparison done') + ok("ssl-compare", "comparison done") http = Http(url, token, verify) - out('HTTP library used for next steps: %s' % ('requests' if http.requests else 'urllib (requests missing)')) - out('TLS verification used: %s' % (verify,)) + out("HTTP library used for next steps: %s" % ("requests" if http.requests else "urllib (requests missing)")) + out("TLS verification used: %s" % (verify,)) # 7 ----------------------------------------------------------------- - step('7. Confluence reachable (/status, no login)') + step("7. Confluence reachable (/status, no login)") try: - status, hdrs, body = http.get('/status', auth=False) - out(' response: %s' % body_preview(body)) - if status == 200 and b'RUNNING' in body: - ok('status', 'Confluence is RUNNING') + status, hdrs, body = http.get("/status", auth=False) + out(" response: %s" % body_preview(body)) + if status == 200 and b"RUNNING" in body: + ok("status", "Confluence is RUNNING") elif status in (301, 302, 303, 307, 308): - warn('status', 'redirect to %s - CONFLUENCE_URL may need a different form (context path, https?)' % hdrs.get('Location')) + warn( + "status", + "redirect to %s - CONFLUENCE_URL may need a different form (context path, https?)" + % hdrs.get("Location"), + ) else: - warn('status', 'unexpected answer - is CONFLUENCE_URL the Confluence base address?') + warn("status", "unexpected answer - is CONFLUENCE_URL the Confluence base address?") except Exception as e: - if 'SSL' in type(e).__name__ or 'CERTIFICATE' in str(e).upper(): - fail('status', 'SSL error: %s\n' - ' -> company certificate not trusted. Export the company root + issuing CA (Base-64 .cer) and set VERIFY_SSL to that file path.' % e) + if "SSL" in type(e).__name__ or "CERTIFICATE" in str(e).upper(): + fail( + "status", + "SSL error: %s\n" + " -> company certificate not trusted. Export the company root + issuing CA (Base-64 .cer) and set VERIFY_SSL to that file path." + % e, + ) else: - fail('status', '%s: %s' % (type(e).__name__, e)) + fail("status", "%s: %s" % (type(e).__name__, e)) return # 8 ----------------------------------------------------------------- - step('8. Token / login (/rest/api/user/current)') - status, hdrs, body = http.get('/rest/api/user/current') + step("8. Token / login (/rest/api/user/current)") + status, hdrs, body = http.get("/rest/api/user/current") if status == 200: - user = json.loads(body.decode('utf-8')) - if user.get('type') == 'anonymous': - fail('token', 'Confluence treats you as ANONYMOUS - token not accepted (check it was copied fully)') + user = json.loads(body.decode("utf-8")) + if user.get("type") == "anonymous": + fail("token", "Confluence treats you as ANONYMOUS - token not accepted (check it was copied fully)") else: - ok('token', 'logged in as: %s (%s)' % (user.get('displayName'), user.get('username'))) + ok("token", "logged in as: %s (%s)" % (user.get("displayName"), user.get("username"))) else: - out(' response: %s' % body_preview(body)) - fail('token', 'HTTP %s - %s' % (status, 'token invalid or expired' if status == 401 else 'unexpected')) + out(" response: %s" % body_preview(body)) + fail("token", "HTTP %s - %s" % (status, "token invalid or expired" if status == 401 else "unexpected")) if status in (401, 403): return # 9 ----------------------------------------------------------------- - step('9. Space %s' % space_key) - status, hdrs, body = http.get('/rest/api/space/' + space_key) + step("9. Space %s" % space_key) + status, hdrs, body = http.get("/rest/api/space/" + space_key) if status == 200: - space = json.loads(body.decode('utf-8')) - ok('space', 'space found: %s (%s)' % (space.get('name'), space.get('key'))) + space = json.loads(body.decode("utf-8")) + ok("space", "space found: %s (%s)" % (space.get("name"), space.get("key"))) else: - out(' response: %s' % body_preview(body)) - fail('space', 'HTTP %s - space key wrong (case matters!) or no permission to view it' % status) + out(" response: %s" % body_preview(body)) + fail("space", "HTTP %s - space key wrong (case matters!) or no permission to view it" % status) return # 10 ---------------------------------------------------------------- - step('10. Listing pages of the space') - status, hdrs, body = http.get('/rest/api/content', {'spaceKey': space_key, 'type': 'page', 'status': 'current', - 'expand': 'ancestors', 'limit': 25}) + step("10. Listing pages of the space") + status, hdrs, body = http.get( + "/rest/api/content", + {"spaceKey": space_key, "type": "page", "status": "current", "expand": "ancestors", "limit": 25}, + ) if status != 200: - out(' response: %s' % body_preview(body)) - fail('pages', 'HTTP %s while listing pages' % status) + out(" response: %s" % body_preview(body)) + fail("pages", "HTTP %s while listing pages" % status) return - data = json.loads(body.decode('utf-8')) - pages = data.get('results', []) - out(' first batch: %d pages, more available: %s' % (len(pages), 'next' in data.get('_links', {}))) + data = json.loads(body.decode("utf-8")) + pages = data.get("results", []) + out(" first batch: %d pages, more available: %s" % (len(pages), "next" in data.get("_links", {}))) for p in pages[:5]: - path = ' / '.join([a['title'] for a in p.get('ancestors', [])] + [p['title']]) - out(' - [%s] %s' % (p['id'], path)) + path = " / ".join([a["title"] for a in p.get("ancestors", [])] + [p["title"]]) + out(" - [%s] %s" % (p["id"], path)) if pages: - ok('pages', 'pages are visible') + ok("pages", "pages are visible") else: - fail('pages', 'no pages visible in the space for this user') + fail("pages", "no pages visible in the space for this user") return # 11 ---------------------------------------------------------------- - step('11. Attachments (checking up to 25 pages)') + step("11. Attachments (checking up to 25 pages)") sample = None for p in pages: - status, hdrs, body = http.get('/rest/api/content/%s/child/attachment' % p['id'], {'limit': 5}) + status, hdrs, body = http.get("/rest/api/content/%s/child/attachment" % p["id"], {"limit": 5}) if status != 200: - out(' response: %s' % body_preview(body)) - fail('attachments', 'HTTP %s while listing attachments of page %s' % (status, p['id'])) + out(" response: %s" % body_preview(body)) + fail("attachments", "HTTP %s while listing attachments of page %s" % (status, p["id"])) return - atts = json.loads(body.decode('utf-8')).get('results', []) + atts = json.loads(body.decode("utf-8")).get("results", []) if atts: sample = atts[0] - out(' page "%s" has attachments, e.g. %s (%s bytes)' - % (p['title'], sample['title'], sample.get('extensions', {}).get('fileSize'))) + out( + ' page "%s" has attachments, e.g. %s (%s bytes)' + % (p["title"], sample["title"], sample.get("extensions", {}).get("fileSize")) + ) break if sample: - ok('attachments', 'attachments can be listed') + ok("attachments", "attachments can be listed") else: - warn('attachments', 'none of the first %d pages has attachments (listing works)' % len(pages)) + warn("attachments", "none of the first %d pages has attachments (listing works)" % len(pages)) # 12 ---------------------------------------------------------------- - step('12. Downloading a file (first 64 KB)') + step("12. Downloading a file (first 64 KB)") if sample: - link = sample['_links']['download'] - dl = link if link.startswith('http') else sample['_links'].get('base', url.rstrip('/')) + link + link = sample["_links"]["download"] + dl = link if link.startswith("http") else sample["_links"].get("base", url.rstrip("/")) + link status, hdrs, body = http.get(dl, max_bytes=65536) if status == 200: - ok('download', 'download works (%d bytes read, type %s)' % (len(body), hdrs.get('Content-Type'))) + ok("download", "download works (%d bytes read, type %s)" % (len(body), hdrs.get("Content-Type"))) else: - out(' response: %s' % body_preview(body)) - fail('download', 'HTTP %s while downloading' % status) + out(" response: %s" % body_preview(body)) + fail("download", "HTTP %s while downloading" % status) else: - warn('download', 'skipped - no attachment found to test') + warn("download", "skipped - no attachment found to test") # 13 ---------------------------------------------------------------- - step('13. Write access to output folder') - target = os.path.abspath(cfg['OUTPUT_DIR']) + step("13. Write access to output folder") + target = os.path.abspath(cfg["OUTPUT_DIR"]) try: os.makedirs(target, exist_ok=True) - test_file = os.path.join(target, '_write_test.tmp') - with open(test_file, 'w') as f: - f.write('test') + test_file = os.path.join(target, "_write_test.tmp") + with open(test_file, "w") as f: + f.write("test") os.remove(test_file) - ok('write', 'can write to %s' % target) + ok("write", "can write to %s" % target) except OSError as e: - fail('write', 'cannot write to %s: %s' % (target, e)) + fail("write", "cannot write to %s: %s" % (target, e)) -if __name__ == '__main__': +if __name__ == "__main__": try: main() except Exception: - results.append(('unexpected', 'FAIL')) - out('[FAIL] unexpected error:') + results.append(("unexpected", "FAIL")) + out("[FAIL] unexpected error:") for line in traceback.format_exc().splitlines(): - out(' ' + line) + out(" " + line) out() - out('=== SUMMARY ===') + out("=== SUMMARY ===") for name, status in results: - out(' %-12s %s' % (name, status)) - failed = [n for n, s in results if s == 'FAIL'] - out('RESULT: %s' % ('PROBLEMS FOUND: ' + ', '.join(failed) if failed else 'ALL CHECKS PASSED - the main script should work')) - out('Full log: %s' % LOG_FILE) + out(" %-12s %s" % (name, status)) + failed = [n for n, s in results if s == "FAIL"] + out( + "RESULT: %s" + % ("PROBLEMS FOUND: " + ", ".join(failed) if failed else "ALL CHECKS PASSED - the main script should work") + ) + out("Full log: %s" % LOG_FILE) _log.close() try: - input('\nPress Enter to close...') + input("\nPress Enter to close...") except (EOFError, KeyboardInterrupt): pass diff --git a/examples/confluence-space-attachments-download.py b/examples/confluence-space-attachments-download.py index 28bd0dae7..6bae97025 100644 --- a/examples/confluence-space-attachments-download.py +++ b/examples/confluence-space-attachments-download.py @@ -16,21 +16,33 @@ Requires: requests (pip install requests) """ +import argparse +import csv +import logging +import os +import re +import ssl +import sys +import tempfile +import time +from urllib.parse import urlparse + + # ============================================================================= # CONFIGURATION - fill in before running # ============================================================================= # Confluence base URL incl. context path, e.g. https://wiki.example.com/confluence -CONFLUENCE_URL = 'https://confluence.example.com' +CONFLUENCE_URL = "https://confluence.example.com" # Personal Access Token: Avatar (top right) -> Profile -> Personal Access Tokens -> Create token -CONFLUENCE_TOKEN = 'PASTE-YOUR-TOKEN-HERE' +CONFLUENCE_TOKEN = "PASTE-YOUR-TOKEN-HERE" # Space key (visible in the URL: .../display/KEY/... or in Space tools -> Overview) -SPACE_KEY = 'DOCS' +SPACE_KEY = "DOCS" # Where to save files (a subfolder is created inside) -OUTPUT_DIR = './confluence-export' +OUTPUT_DIR = "./confluence-export" # Also download attachments of blog posts (saved under "_Blog posts") INCLUDE_BLOGPOSTS = False @@ -50,29 +62,21 @@ # ============================================================================= -import argparse -import csv -import logging -import os -import re -import ssl -import sys -import tempfile -import time -from urllib.parse import urlparse - try: import requests except ImportError: print('ERROR: the "requests" library is missing. Install it with:') print(' "%s" -m pip install requests' % sys.executable) if PAUSE_AT_END: - input('\nPress Enter to close...') + input("\nPress Enter to close...") sys.exit(1) + PAGE_LIMIT = 100 INVALID_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]') -RESERVED_NAMES = {'CON', 'PRN', 'AUX', 'NUL'} | {'COM%d' % i for i in range(1, 10)} | {'LPT%d' % i for i in range(1, 10)} +RESERVED_NAMES = ( + {"CON", "PRN", "AUX", "NUL"} | {"COM%d" % i for i in range(1, 10)} | {"LPT%d" % i for i in range(1, 10)} +) def system_ca_bundle(): @@ -81,28 +85,29 @@ def system_ca_bundle(): so company CAs (e.g. an internal "Issuing CA") are rejected. Build a PEM bundle with certifi + trusted Windows ROOT/CA certificates and return its path (None if not Windows). """ - if not hasattr(ssl, 'enum_certificates'): # only exists on Windows + if not hasattr(ssl, "enum_certificates"): # only exists on Windows return None, 0 pems, count = [], 0 try: import certifi - with open(certifi.where(), encoding='utf-8') as f: + + with open(certifi.where(), encoding="utf-8") as f: pems.append(f.read()) except (ImportError, OSError): pass - for store in ('ROOT', 'CA'): + for store in ("ROOT", "CA"): try: certificates = ssl.enum_certificates(store) except OSError: continue for der, encoding, trust in certificates: # trust is True (all purposes) or a set of OIDs; 1.3.6.1.5.5.7.3.1 = server authentication - if encoding == 'x509_asn' and (trust is True or '1.3.6.1.5.5.7.3.1' in trust): + if encoding == "x509_asn" and (trust is True or "1.3.6.1.5.5.7.3.1" in trust): pems.append(ssl.DER_cert_to_PEM_cert(der)) count += 1 - path = os.path.join(tempfile.gettempdir(), 'confluence-ca-bundle.pem') - with open(path, 'w', encoding='utf-8') as f: - f.write('\n'.join(pems)) + path = os.path.join(tempfile.gettempdir(), "confluence-ca-bundle.pem") + with open(path, "w", encoding="utf-8") as f: + f.write("\n".join(pems)) return path, count @@ -117,46 +122,46 @@ def resolve_verify(setting): def bypass_proxy(url): """Connect to Confluence directly, ignoring the system/company proxy, for this host only.""" host = urlparse(url).hostname - for key in ('NO_PROXY', 'no_proxy'): - current = os.environ.get(key, '') - os.environ[key] = ','.join(x for x in (current, host) if x) + for key in ("NO_PROXY", "no_proxy"): + current = os.environ.get(key, "") + os.environ[key] = ",".join(x for x in (current, host) if x) -log = logging.getLogger('confluence-download') +log = logging.getLogger("confluence-download") def safe_name(name, max_len=120): """Make a title usable as a file/folder name on Windows, macOS and Linux.""" - name = INVALID_CHARS.sub('_', name).strip().rstrip('.') + name = INVALID_CHARS.sub("_", name).strip().rstrip(".") if not name: - name = '_' - if name.split('.')[0].upper() in RESERVED_NAMES: - name = '_' + name + name = "_" + if name.split(".")[0].upper() in RESERVED_NAMES: + name = "_" + name if len(name) > max_len: root, ext = os.path.splitext(name) - name = root[:max_len - len(ext)] + ext + name = root[: max_len - len(ext)] + ext return name def human_size(num): - for unit in ('B', 'KB', 'MB', 'GB'): - if num < 1024 or unit == 'GB': - return ('%d %s' if unit == 'B' else '%.1f %s') % (num, unit) + for unit in ("B", "KB", "MB", "GB"): + if num < 1024 or unit == "GB": + return ("%d %s" if unit == "B" else "%.1f %s") % (num, unit) num /= 1024.0 class Confluence: def __init__(self, url, token, verify=True, timeout=120): - self.url = url.rstrip('/') + self.url = url.rstrip("/") self.session = requests.Session() # passed on every request: a session-level verify is overridden by REQUESTS_CA_BUNDLE/CURL_CA_BUNDLE self.verify = verify self.timeout = timeout - self.session.headers['Accept'] = 'application/json' - self.session.headers['Authorization'] = 'Bearer ' + token + self.session.headers["Accept"] = "application/json" + self.session.headers["Authorization"] = "Bearer " + token def get(self, path_or_url, params=None, stream=False, retries=4): - url = path_or_url if path_or_url.startswith('http') else self.url + path_or_url + url = path_or_url if path_or_url.startswith("http") else self.url + path_or_url for attempt in range(retries + 1): wait = 2 ** (attempt + 1) try: @@ -166,7 +171,7 @@ def get(self, path_or_url, params=None, stream=False, retries=4): except requests.RequestException as e: if attempt == retries: raise - log.warning('Connection problem (%s), retrying in %ds...', e.__class__.__name__, wait) + log.warning("Connection problem (%s), retrying in %ds...", e.__class__.__name__, wait) else: # 429 = rate limiting (DC 7.7+), 5xx = transient server errors if response.status_code != 429 and response.status_code < 500: @@ -174,9 +179,9 @@ def get(self, path_or_url, params=None, stream=False, retries=4): return response if attempt == retries: response.raise_for_status() - retry_after = response.headers.get('Retry-After', '') + retry_after = response.headers.get("Retry-After", "") wait = int(retry_after) if retry_after.isdigit() else wait - log.warning('Server answered HTTP %s, retrying in %ds...', response.status_code, wait) + log.warning("Server answered HTTP %s, retrying in %ds...", response.status_code, wait) time.sleep(wait) def paged(self, path, params): @@ -184,49 +189,55 @@ def paged(self, path, params): start = 0 while True: data = self.get(path, params=dict(params, start=start, limit=PAGE_LIMIT)).json() - results = data.get('results', []) + results = data.get("results", []) for item in results: yield item # Confluence may cap 'limit' server-side, so rely on the 'next' link, not on the page size - if not results or 'next' not in data.get('_links', {}): + if not results or "next" not in data.get("_links", {}): return start += len(results) def check_space(self, space_key): try: - return self.get('/rest/api/space/' + space_key).json() + return self.get("/rest/api/space/" + space_key).json() except requests.exceptions.SSLError as e: - raise SystemExit('ERROR: SSL certificate not trusted: %s\n' - 'Run confluence-diagnostics.py, or set VERIFY_SSL to your company CA file (.pem/.cer)' % e) + raise SystemExit( + "ERROR: SSL certificate not trusted: %s\n" + "Run confluence-diagnostics.py, or set VERIFY_SSL to your company CA file (.pem/.cer)" % e + ) except requests.HTTPError as e: code = e.response.status_code if e.response is not None else None if code == 401: - raise SystemExit('ERROR: token rejected (HTTP 401) - check CONFLUENCE_TOKEN') + raise SystemExit("ERROR: token rejected (HTTP 401) - check CONFLUENCE_TOKEN") if code in (403, 404): - raise SystemExit('ERROR: space "%s" does not exist or you have no access to it (HTTP %s)' - % (space_key, code)) + raise SystemExit( + 'ERROR: space "%s" does not exist or you have no access to it (HTTP %s)' % (space_key, code) + ) raise def space_content(self, space_key, content_type): - return self.paged('/rest/api/content', { - 'spaceKey': space_key, - 'type': content_type, - 'status': 'current', - 'expand': 'ancestors', - }) + return self.paged( + "/rest/api/content", + { + "spaceKey": space_key, + "type": content_type, + "status": "current", + "expand": "ancestors", + }, + ) def attachments(self, content_id): - return self.paged('/rest/api/content/%s/child/attachment' % content_id, {'expand': 'version'}) + return self.paged("/rest/api/content/%s/child/attachment" % content_id, {"expand": "version"}) def page_folder(page, folder_cache, base_dir): """Folder path built from the page's ancestors + its own title.""" - parts = [safe_name(a['title']) for a in page.get('ancestors', [])] + [safe_name(page['title'])] + parts = [safe_name(a["title"]) for a in page.get("ancestors", [])] + [safe_name(page["title"])] folder = os.path.join(base_dir, *parts) # Titles are unique per space, but sanitising may produce collisions (e.g. "a/b" and "a_b"). - owner = folder_cache.setdefault(folder, page['id']) - if owner != page['id']: - folder = '%s (%s)' % (folder, page['id']) + owner = folder_cache.setdefault(folder, page["id"]) + if owner != page["id"]: + folder = "%s (%s)" % (folder, page["id"]) return folder @@ -236,16 +247,16 @@ def unique_path(path, used): return path root, ext = os.path.splitext(path) i = 1 - while '%s (%d)%s' % (root, i, ext) in used or os.path.exists('%s (%d)%s' % (root, i, ext)): + while "%s (%d)%s" % (root, i, ext) in used or os.path.exists("%s (%d)%s" % (root, i, ext)): i += 1 - path = '%s (%d)%s' % (root, i, ext) + path = "%s (%d)%s" % (root, i, ext) used.add(path) return path def download(confluence, url, target): - tmp = target + '.part' - with confluence.get(url, stream=True) as response, open(tmp, 'wb') as f: + tmp = target + ".part" + with confluence.get(url, stream=True) as response, open(tmp, "wb") as f: for chunk in response.iter_content(chunk_size=1024 * 1024): f.write(chunk) os.replace(tmp, target) @@ -253,31 +264,33 @@ def download(confluence, url, target): def setup_logging(log_file): log.setLevel(logging.INFO) - fmt = logging.Formatter('%(asctime)s %(levelname)-7s %(message)s', '%H:%M:%S') - for handler in (logging.StreamHandler(sys.stdout), logging.FileHandler(log_file, encoding='utf-8')): + fmt = logging.Formatter("%(asctime)s %(levelname)-7s %(message)s", "%H:%M:%S") + for handler in (logging.StreamHandler(sys.stdout), logging.FileHandler(log_file, encoding="utf-8")): handler.setFormatter(fmt) log.addHandler(handler) def parse_args(): """Optional command-line overrides of the CONFIGURATION section.""" - parser = argparse.ArgumentParser(description='Download all attachments of a Confluence space into a page-tree folder structure.') - parser.add_argument('--url', default=CONFLUENCE_URL) - parser.add_argument('--token', default=os.environ.get('CONFLUENCE_TOKEN') or CONFLUENCE_TOKEN) - parser.add_argument('--space', default=SPACE_KEY) - parser.add_argument('--out', default=OUTPUT_DIR) - parser.add_argument('--blogposts', action='store_true', default=INCLUDE_BLOGPOSTS) - parser.add_argument('--no-skip-existing', dest='skip_existing', action='store_false', default=SKIP_EXISTING) - parser.add_argument('--ca-bundle', default=VERIFY_SSL if isinstance(VERIFY_SSL, str) else None) + parser = argparse.ArgumentParser( + description="Download all attachments of a Confluence space into a page-tree folder structure." + ) + parser.add_argument("--url", default=CONFLUENCE_URL) + parser.add_argument("--token", default=os.environ.get("CONFLUENCE_TOKEN") or CONFLUENCE_TOKEN) + parser.add_argument("--space", default=SPACE_KEY) + parser.add_argument("--out", default=OUTPUT_DIR) + parser.add_argument("--blogposts", action="store_true", default=INCLUDE_BLOGPOSTS) + parser.add_argument("--no-skip-existing", dest="skip_existing", action="store_false", default=SKIP_EXISTING) + parser.add_argument("--ca-bundle", default=VERIFY_SSL if isinstance(VERIFY_SSL, str) else None) return parser.parse_args() def main(): args = parse_args() - if not args.token or args.token.startswith('PASTE-'): - raise SystemExit('ERROR: set CONFLUENCE_TOKEN in the CONFIGURATION section at the top of the script') - if 'example.com' in args.url: - raise SystemExit('ERROR: set CONFLUENCE_URL in the CONFIGURATION section at the top of the script') + if not args.token or args.token.startswith("PASTE-"): + raise SystemExit("ERROR: set CONFLUENCE_TOKEN in the CONFIGURATION section at the top of the script") + if "example.com" in args.url: + raise SystemExit("ERROR: set CONFLUENCE_URL in the CONFIGURATION section at the top of the script") verify = args.ca_bundle or resolve_verify(VERIFY_SSL) if DIRECT_CONNECTION: @@ -285,111 +298,133 @@ def main(): confluence = Confluence(args.url, args.token, verify=verify) space = confluence.check_space(args.space) - base_dir = os.path.join(args.out, safe_name(space['key'])) + base_dir = os.path.join(args.out, safe_name(space["key"])) os.makedirs(base_dir, exist_ok=True) - setup_logging(os.path.join(base_dir, '_download.log')) + setup_logging(os.path.join(base_dir, "_download.log")) started = time.time() - log.info('TLS verification: %s | proxy: %s', verify, 'bypassed' if DIRECT_CONNECTION else 'system settings') - log.info('Space: %s (%s)', space['name'], space['key']) - log.info('Output folder: %s', os.path.abspath(base_dir)) + log.info("TLS verification: %s | proxy: %s", verify, "bypassed" if DIRECT_CONNECTION else "system settings") + log.info("Space: %s (%s)", space["name"], space["key"]) + log.info("Output folder: %s", os.path.abspath(base_dir)) - sources = [('page', base_dir)] + sources = [("page", base_dir)] if args.blogposts: - sources.append(('blogpost', os.path.join(base_dir, '_Blog posts'))) + sources.append(("blogpost", os.path.join(base_dir, "_Blog posts"))) - log.info('Collecting the list of pages...') + log.info("Collecting the list of pages...") items = [] for content_type, root in sources: for page in confluence.space_content(args.space, content_type): items.append((page, root)) if len(items) % 500 == 0: - log.info(' ...%d found so far', len(items)) - log.info('Found %d %s. Starting download.', len(items), 'pages/blog posts' if args.blogposts else 'pages') + log.info(" ...%d found so far", len(items)) + log.info("Found %d %s. Starting download.", len(items), "pages/blog posts" if args.blogposts else "pages") folder_cache, used_paths = {}, set() - stats = {'pages_with_files': 0, 'files': 0, 'skipped': 0, 'errors': 0, 'bytes': 0} - manifest_path = os.path.join(base_dir, '_attachments_manifest.csv') + stats = {"pages_with_files": 0, "files": 0, "skipped": 0, "errors": 0, "bytes": 0} + manifest_path = os.path.join(base_dir, "_attachments_manifest.csv") width = len(str(len(items))) - with open(manifest_path, 'w', newline='', encoding='utf-8-sig') as manifest_file: - manifest = csv.writer(manifest_file, delimiter=';') - manifest.writerow(['page_id', 'page_title', 'attachment_id', 'file_name', 'version', 'size', 'local_path', 'status']) + with open(manifest_path, "w", newline="", encoding="utf-8-sig") as manifest_file: + manifest = csv.writer(manifest_file, delimiter=";") + manifest.writerow( + ["page_id", "page_title", "attachment_id", "file_name", "version", "size", "local_path", "status"] + ) for index, (page, root) in enumerate(items, 1): - progress = '[%*d/%d]' % (width, index, len(items)) - attachments = list(confluence.attachments(page['id'])) + progress = "[%*d/%d]" % (width, index, len(items)) + attachments = list(confluence.attachments(page["id"])) if not attachments: - log.info('%s %s - no attachments', progress, page['title']) + log.info("%s %s - no attachments", progress, page["title"]) continue - stats['pages_with_files'] += 1 + stats["pages_with_files"] += 1 folder = page_folder(page, folder_cache, root) os.makedirs(folder, exist_ok=True) - log.info('%s %s - %d attachment(s) -> %s', progress, page['title'], len(attachments), - os.path.relpath(folder, base_dir)) + log.info( + "%s %s - %d attachment(s) -> %s", + progress, + page["title"], + len(attachments), + os.path.relpath(folder, base_dir), + ) for att in attachments: - size = att.get('extensions', {}).get('fileSize') - target = os.path.join(folder, safe_name(att['title'], max_len=200)) - if args.skip_existing and os.path.exists(target) and size is not None and os.path.getsize(target) == size: + size = att.get("extensions", {}).get("fileSize") + target = os.path.join(folder, safe_name(att["title"], max_len=200)) + if ( + args.skip_existing + and os.path.exists(target) + and size is not None + and os.path.getsize(target) == size + ): used_paths.add(target) - status = 'skipped' - stats['skipped'] += 1 - log.info(' = %s (already downloaded)', att['title']) + status = "skipped" + stats["skipped"] += 1 + log.info(" = %s (already downloaded)", att["title"]) else: target = unique_path(target, used_paths) - link = att['_links']['download'] - url = link if link.startswith('http') else att['_links'].get('base', confluence.url) + link + link = att["_links"]["download"] + url = link if link.startswith("http") else att["_links"].get("base", confluence.url) + link try: download(confluence, url, target) - status = 'ok' - stats['files'] += 1 - stats['bytes'] += os.path.getsize(target) - log.info(' + %s (%s)', att['title'], human_size(os.path.getsize(target))) + status = "ok" + stats["files"] += 1 + stats["bytes"] += os.path.getsize(target) + log.info(" + %s (%s)", att["title"], human_size(os.path.getsize(target))) except Exception as e: # keep going, report at the end - status = 'error: %s' % e - stats['errors'] += 1 - log.error(' ! %s - %s', att['title'], e) - - manifest.writerow([page['id'], page['title'], att['id'], att['title'], - att.get('version', {}).get('number'), size, - os.path.relpath(target, base_dir), status]) + status = "error: %s" % e + stats["errors"] += 1 + log.error(" ! %s - %s", att["title"], e) + + manifest.writerow( + [ + page["id"], + page["title"], + att["id"], + att["title"], + att.get("version", {}).get("number"), + size, + os.path.relpath(target, base_dir), + status, + ] + ) elapsed = int(time.time() - started) - log.info('=' * 60) - log.info('DONE in %dm %02ds', elapsed // 60, elapsed % 60) - log.info('Pages scanned: %d (with attachments: %d)', len(items), stats['pages_with_files']) - log.info('Files downloaded: %d (%s)', stats['files'], human_size(stats['bytes'])) - log.info('Skipped (existing): %d', stats['skipped']) - log.info('Errors: %d', stats['errors']) - log.info('Manifest: %s', os.path.abspath(manifest_path)) - return 1 if stats['errors'] else 0 + log.info("=" * 60) + log.info("DONE in %dm %02ds", elapsed // 60, elapsed % 60) + log.info("Pages scanned: %d (with attachments: %d)", len(items), stats["pages_with_files"]) + log.info("Files downloaded: %d (%s)", stats["files"], human_size(stats["bytes"])) + log.info("Skipped (existing): %d", stats["skipped"]) + log.info("Errors: %d", stats["errors"]) + log.info("Manifest: %s", os.path.abspath(manifest_path)) + return 1 if stats["errors"] else 0 def pause(): if PAUSE_AT_END: try: - input('\nPress Enter to close...') + input("\nPress Enter to close...") except (EOFError, KeyboardInterrupt): pass -if __name__ == '__main__': - print('Starting Confluence attachments download...') +if __name__ == "__main__": + print("Starting Confluence attachments download...") exit_code = 1 try: exit_code = main() except KeyboardInterrupt: - print('\nInterrupted - run again to resume (already downloaded files will be skipped).') + print("\nInterrupted - run again to resume (already downloaded files will be skipped).") exit_code = 130 except SystemExit as e: if e.code not in (None, 0): - print(e.code if isinstance(e.code, str) else 'Exit code %s' % e.code) + print(e.code if isinstance(e.code, str) else "Exit code %s" % e.code) exit_code = e.code if isinstance(e.code, int) else 1 except Exception: import traceback - print('\nUNEXPECTED ERROR - run confluence-diagnostics.py and send its log:') + + print("\nUNEXPECTED ERROR - run confluence-diagnostics.py and send its log:") traceback.print_exc(file=sys.stdout) pause() sys.exit(exit_code) From e43447dff6dae743917fb9b6fbba00fc98513b14 Mon Sep 17 00:00:00 2001 From: jerubball Date: Fri, 30 Oct 2026 11:32:04 -0400 Subject: [PATCH 2/4] also remove unused imports --- examples/confluence-diagnostics.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/confluence-diagnostics.py b/examples/confluence-diagnostics.py index e14a06456..80521dd48 100644 --- a/examples/confluence-diagnostics.py +++ b/examples/confluence-diagnostics.py @@ -28,7 +28,7 @@ try: # Python 3 only, but give a readable message on Python 2 from urllib.parse import urlparse, urlencode from urllib.request import Request, urlopen, getproxies - from urllib.error import HTTPError, URLError # noqa: F401 + from urllib.error import HTTPError except ImportError: print("ERROR: this script needs Python 3, you are running Python %s" % sys.version.split()[0]) raise SystemExit(1) From 5f859339e6463f3a2e1f303bd217095694fcc630 Mon Sep 17 00:00:00 2001 From: jerubball Date: Fri, 30 Oct 2026 11:38:45 -0400 Subject: [PATCH 3/4] fix CodeQL url pattern warning --- examples/confluence-diagnostics.py | 2 +- examples/confluence-space-attachments-download.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/confluence-diagnostics.py b/examples/confluence-diagnostics.py index 80521dd48..3a74d37be 100644 --- a/examples/confluence-diagnostics.py +++ b/examples/confluence-diagnostics.py @@ -258,7 +258,7 @@ def main(): out("VERIFY_SSL: %r" % (cfg["VERIFY_SSL"],)) out("DIRECT_CONNECTION: %r" % (cfg["DIRECT_CONNECTION"],)) problems = [] - if "example.com" in url: + if url.startswith("https://confluence.example.com"): problems.append("CONFLUENCE_URL still has the example value") if not url.startswith(("http://", "https://")): problems.append("CONFLUENCE_URL must start with https:// (or http://)") diff --git a/examples/confluence-space-attachments-download.py b/examples/confluence-space-attachments-download.py index 6bae97025..fa5600c70 100644 --- a/examples/confluence-space-attachments-download.py +++ b/examples/confluence-space-attachments-download.py @@ -289,7 +289,7 @@ def main(): args = parse_args() if not args.token or args.token.startswith("PASTE-"): raise SystemExit("ERROR: set CONFLUENCE_TOKEN in the CONFIGURATION section at the top of the script") - if "example.com" in args.url: + if args.url.startswith("https://confluence.example.com"): raise SystemExit("ERROR: set CONFLUENCE_URL in the CONFIGURATION section at the top of the script") verify = args.ca_bundle or resolve_verify(VERIFY_SSL) From 1d1d311f4f157329e5f86fd60ff1092f496adc2a Mon Sep 17 00:00:00 2001 From: jerubball Date: Fri, 30 Oct 2026 11:42:15 -0400 Subject: [PATCH 4/4] fix CodeQL url pattern warning --- examples/confluence-diagnostics.py | 2 +- examples/confluence-space-attachments-download.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/confluence-diagnostics.py b/examples/confluence-diagnostics.py index 3a74d37be..f4ac27abb 100644 --- a/examples/confluence-diagnostics.py +++ b/examples/confluence-diagnostics.py @@ -258,7 +258,7 @@ def main(): out("VERIFY_SSL: %r" % (cfg["VERIFY_SSL"],)) out("DIRECT_CONNECTION: %r" % (cfg["DIRECT_CONNECTION"],)) problems = [] - if url.startswith("https://confluence.example.com"): + if url == "https://confluence.example.com": problems.append("CONFLUENCE_URL still has the example value") if not url.startswith(("http://", "https://")): problems.append("CONFLUENCE_URL must start with https:// (or http://)") diff --git a/examples/confluence-space-attachments-download.py b/examples/confluence-space-attachments-download.py index fa5600c70..b88d9e8a5 100644 --- a/examples/confluence-space-attachments-download.py +++ b/examples/confluence-space-attachments-download.py @@ -289,7 +289,7 @@ def main(): args = parse_args() if not args.token or args.token.startswith("PASTE-"): raise SystemExit("ERROR: set CONFLUENCE_TOKEN in the CONFIGURATION section at the top of the script") - if args.url.startswith("https://confluence.example.com"): + if args.url == "https://confluence.example.com": raise SystemExit("ERROR: set CONFLUENCE_URL in the CONFIGURATION section at the top of the script") verify = args.ca_bundle or resolve_verify(VERIFY_SSL)