|
#!/usr/bin/env python3 |
|
""" |
|
bb_fetch_links.py — download Bitbucket-hosted files referenced inside a Bitbucket |
|
issue-export JSON, store them next to the JSON, and rewrite the references to |
|
point at the local copies. |
|
|
|
What it does |
|
------------ |
|
* Scans every text field of the JSON for URLs on bitbucket.org. |
|
* DOWNLOADS the inline-image uploads (URLs shaped like |
|
https://bitbucket.org/repo/<hash>/images/<file>) |
|
into a folder next to the JSON file, then rewrites each such URL in the JSON to |
|
a path relative to the JSON's parent folder, e.g. |
|
content_files/1209406364-Screen%20Shot%202018-11-13%20at%2015.45.08.png |
|
* SKIPS the entries in the top-level "attachments" array — those already have a |
|
local "path" and downloaded file, so they are left untouched. |
|
* Only LISTS wiki-page links (https://bitbucket.org/<ws>/<repo>/wiki/<page>) and |
|
any other unrecognised bitbucket URLs in a report; these are pages, not files, |
|
and are not rewritten. |
|
|
|
The JSON is edited by exact textual replacement of the (ASCII, percent-encoded) |
|
image URLs, so the rest of the file is preserved byte-for-byte. A .bak backup is |
|
written unless --no-backup. Re-running is a no-op: once a URL has been rewritten |
|
there is nothing left on bitbucket.org to fetch. |
|
|
|
Authentication |
|
-------------- |
|
The legacy /repo/<hash>/images/ URLs are served by Bitbucket's web app and (for |
|
private repos) only accept a logged-in **browser session cookie** — app passwords |
|
get a 401 there. So the primary auth method is a cookie: |
|
|
|
1. In a browser signed in to bitbucket.org, open DevTools -> Network, click any |
|
request to bitbucket.org, and copy the whole "Cookie:" request-header value. |
|
2. Paste it into a file (e.g. cookie.txt) next to this script. |
|
3. Run with --cookie-file cookie.txt (keeps the secret out of shell history) |
|
|
|
Alternatively: env BITBUCKET_COOKIE="...", or --cookie "...". |
|
|
|
An app password is still accepted (--user/--app-password or env BITBUCKET_USER / |
|
BITBUCKET_APP_PASSWORD) in case a future export links to endpoints that allow it, |
|
but it will NOT work for the /repo/images/ URLs above. |
|
|
|
Usage |
|
----- |
|
python3 bb_fetch_links.py db-2.0.json --dry-run # preview |
|
python3 bb_fetch_links.py db-2.0.json --cookie-file cookie.txt --limit 1 # 1 test |
|
python3 bb_fetch_links.py db-2.0.json --cookie-file cookie.txt # all + rewrite |
|
""" |
|
|
|
import argparse |
|
import json |
|
import os |
|
import re |
|
import sys |
|
import urllib.parse |
|
import urllib.request |
|
import urllib.error |
|
|
|
# URLs we DOWNLOAD: Bitbucket inline-image uploads. |
|
IMAGE_URL_RE = re.compile(r"https?://bitbucket\.org/repo/[^\s\"')<>\]]+") |
|
# URLs we only REPORT: wiki pages (not files). |
|
WIKI_URL_RE = re.compile(r"https?://bitbucket\.org/[^\s\"')<>\]]*?/wiki/[^\s\"')<>\]]+") |
|
# Any other bitbucket URL (for the "unrecognised" report). |
|
ANY_BB_URL_RE = re.compile(r"https?://bitbucket\.org/[^\s\"')<>\]]+") |
|
# The already-downloaded attachment uploads live under this path and are skipped. |
|
ATTACHMENT_MARKER = "/issues/attachments/" |
|
# Top-level JSON arrays whose URLs are already handled elsewhere (attachments have |
|
# their own local "path" and downloaded file), so they are not scanned. |
|
SKIP_TOP_KEYS = {"attachments"} |
|
|
|
|
|
def url_filename(url): |
|
"""Decoded filename from the last path segment of a URL.""" |
|
path = urllib.parse.urlsplit(url).path |
|
seg = path.rsplit("/", 1)[-1] |
|
return urllib.parse.unquote(seg) |
|
|
|
|
|
def encoded_segment(url): |
|
"""The (already percent-encoded) last path segment, reused verbatim in the |
|
rewritten local reference so no re-encoding is needed.""" |
|
path = urllib.parse.urlsplit(url).path |
|
return path.rsplit("/", 1)[-1] |
|
|
|
|
|
def unique_path(folder, filename): |
|
"""A non-colliding path inside folder for filename.""" |
|
base = os.path.join(folder, filename) |
|
if not os.path.exists(base): |
|
return filename |
|
stem, ext = os.path.splitext(filename) |
|
n = 2 |
|
while os.path.exists(os.path.join(folder, f"{stem}-{n}{ext}")): |
|
n += 1 |
|
return f"{stem}-{n}{ext}" |
|
|
|
|
|
LOGIN_HOSTS = ("id.atlassian.com", "auth.atlassian.com") |
|
LOGIN_PATHS = ("/account/signin", "/login", "/plugins/servlet/samlsso") |
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler): |
|
def redirect_request(self, *a, **k): |
|
return None # turn 30x into an HTTPError we handle ourselves |
|
|
|
|
|
_opener = urllib.request.build_opener(_NoRedirect) |
|
|
|
|
|
def http_get(url, headers, timeout, depth=0): |
|
"""GET url with explicit redirect control. Follows redirects, but: |
|
* a redirect to a login page raises a clear auth error (no infinite loop); |
|
* sensitive headers (Cookie/Authorization) are dropped when the host changes |
|
(so a session cookie is never sent to the signed media/S3 host). |
|
Returns (data_bytes, content_type).""" |
|
if depth > 6: |
|
raise RuntimeError("too many redirects") |
|
req = urllib.request.Request(url, headers=headers) |
|
try: |
|
resp = _opener.open(req, timeout=timeout) |
|
return resp.read(), resp.headers.get("Content-Type", "") |
|
except urllib.error.HTTPError as e: |
|
if e.code in (301, 302, 303, 307, 308): |
|
loc = urllib.parse.urljoin(url, e.headers.get("Location", "")) |
|
host = urllib.parse.urlsplit(loc) |
|
if host.netloc in LOGIN_HOSTS or any(p in host.path for p in LOGIN_PATHS): |
|
raise RuntimeError("redirected to sign-in — the session cookie is " |
|
"missing/expired or lacks access") from None |
|
fwd = dict(headers) |
|
if host.netloc != urllib.parse.urlsplit(url).netloc: |
|
fwd.pop("Cookie", None) |
|
fwd.pop("Authorization", None) |
|
return http_get(loc, fwd, timeout, depth + 1) |
|
raise |
|
|
|
|
|
def download(url, auth_headers, timeout): |
|
"""Fetch url with the given auth headers. Returns (data_bytes, content_type).""" |
|
headers = {"User-Agent": "bb_fetch_links/1.0"} |
|
headers.update(auth_headers) |
|
return http_get(url, headers, timeout) |
|
|
|
|
|
def iter_strings(obj, skip_top_keys=SKIP_TOP_KEYS): |
|
"""Yield every string value in the parsed JSON, skipping the values under the |
|
given top-level keys (e.g. the already-handled 'attachments' array).""" |
|
if isinstance(obj, dict): |
|
for k, v in obj.items(): |
|
if k in skip_top_keys: |
|
continue |
|
yield from iter_strings(v, skip_top_keys=()) |
|
elif isinstance(obj, list): |
|
for v in obj: |
|
yield from iter_strings(v, skip_top_keys=()) |
|
elif isinstance(obj, str): |
|
yield obj |
|
|
|
|
|
def collect_urls(data): |
|
"""Return (image_urls, wiki_urls, other_bb_urls) from the parsed JSON, |
|
de-duplicated but order-preserving.""" |
|
images, wiki, other = [], [], [] |
|
seen = set() |
|
for s in iter_strings(data): |
|
for m in ANY_BB_URL_RE.finditer(s): |
|
u = m.group(0) |
|
if u in seen: |
|
continue |
|
seen.add(u) |
|
if ATTACHMENT_MARKER in u: |
|
continue # already-downloaded attachment: skip |
|
if IMAGE_URL_RE.fullmatch(u): |
|
images.append(u) |
|
elif WIKI_URL_RE.match(u): |
|
wiki.append(u) |
|
else: |
|
other.append(u) |
|
return images, wiki, other |
|
|
|
|
|
def main(): |
|
ap = argparse.ArgumentParser(description=__doc__, |
|
formatter_class=argparse.RawDescriptionHelpFormatter) |
|
ap.add_argument("json_file", nargs="?", default="db-2.0.json", |
|
help="the Bitbucket export JSON (default: db-2.0.json)") |
|
ap.add_argument("--dest", default="content_files", |
|
help="download folder name, next to the JSON (default: content_files)") |
|
ap.add_argument("--cookie-file", default=None, |
|
help="file containing the bitbucket.org 'Cookie:' header value " |
|
"(recommended — keeps the secret out of shell history)") |
|
ap.add_argument("--cookie", default=os.environ.get("BITBUCKET_COOKIE"), |
|
help="bitbucket.org Cookie header string (or env BITBUCKET_COOKIE)") |
|
ap.add_argument("--user", default=os.environ.get("BITBUCKET_USER"), |
|
help="Bitbucket username for app-password auth (or env BITBUCKET_USER)") |
|
ap.add_argument("--app-password", default=os.environ.get("BITBUCKET_APP_PASSWORD"), |
|
help="Bitbucket app password (or env BITBUCKET_APP_PASSWORD); note " |
|
"app passwords do NOT work for /repo/images/ URLs") |
|
ap.add_argument("--output", default=None, |
|
help="write updated JSON here (default: overwrite input in place)") |
|
ap.add_argument("--no-backup", action="store_true", |
|
help="do not write a .bak backup of the input when editing in place") |
|
ap.add_argument("--dry-run", action="store_true", |
|
help="report what would happen; download and edit nothing") |
|
ap.add_argument("--limit", type=int, default=None, |
|
help="download at most N images (handy for a first auth test)") |
|
ap.add_argument("--match", default=None, |
|
help="only process image URLs containing this substring (testing)") |
|
ap.add_argument("--timeout", type=int, default=60, help="per-request timeout (s)") |
|
args = ap.parse_args() |
|
|
|
json_path = os.path.abspath(args.json_file) |
|
if not os.path.exists(json_path): |
|
sys.exit(f"error: file not found: {json_path}") |
|
base_dir = os.path.dirname(json_path) |
|
dest_dir = os.path.join(base_dir, args.dest) |
|
|
|
# assemble auth headers: a session Cookie (preferred) or app-password Basic auth |
|
cookie = args.cookie |
|
if args.cookie_file: |
|
with open(args.cookie_file, encoding="utf-8") as f: |
|
cookie = f.read().strip() |
|
auth_headers = {} |
|
if cookie: |
|
if cookie.lower().startswith("cookie:"): |
|
cookie = cookie.split(":", 1)[1].strip() |
|
auth_headers["Cookie"] = cookie |
|
elif args.user and args.app_password: |
|
import base64 |
|
tok = base64.b64encode(f"{args.user}:{args.app_password}".encode()).decode() |
|
auth_headers["Authorization"] = "Basic " + tok |
|
|
|
with open(json_path, encoding="utf-8") as f: |
|
text = f.read() |
|
data = json.loads(text) |
|
|
|
images, wiki, other = collect_urls(data) |
|
if args.match: |
|
images = [u for u in images if args.match in u] |
|
|
|
print(f"Scanned {os.path.basename(json_path)}") |
|
print(f" inline images to download : {len(images)}") |
|
print(f" wiki-page links (listed) : {len(wiki)}") |
|
print(f" other bitbucket URLs : {len(other)}") |
|
print(f" download folder : {args.dest}/ (next to the JSON)") |
|
if args.dry_run: |
|
print(" MODE : DRY RUN (no downloads, no edits)") |
|
print() |
|
|
|
if wiki: |
|
print("Wiki-page links (not downloaded — handle manually):") |
|
for u in wiki: |
|
print(f" - {u}") |
|
print() |
|
if other: |
|
print("Other unrecognised bitbucket URLs (not downloaded):") |
|
for u in other: |
|
print(f" - {u}") |
|
print() |
|
|
|
if not images: |
|
print("No inline images to fetch. Done.") |
|
return |
|
|
|
if not args.dry_run: |
|
if not auth_headers: |
|
sys.exit("error: Bitbucket credentials required to download.\n" |
|
" preferred: --cookie-file FILE (a logged-in bitbucket.org Cookie header)\n" |
|
" or: --cookie / BITBUCKET_COOKIE\n" |
|
" or: --user + --app-password (won't work for /repo/images/)") |
|
os.makedirs(dest_dir, exist_ok=True) |
|
|
|
replacements = {} # old_url -> new_relative_reference |
|
ok = failed = 0 |
|
todo = images[:args.limit] if args.limit else images |
|
for url in todo: |
|
fname = url_filename(url) |
|
rel_ref = f"{args.dest}/{encoded_segment(url)}" # for the JSON (encoded) |
|
if args.dry_run: |
|
print(f" would download: {fname}") |
|
print(f" {url}") |
|
print(f" -> {rel_ref}") |
|
continue |
|
try: |
|
blob, ctype = download(url, auth_headers, args.timeout) |
|
except urllib.error.HTTPError as e: |
|
print(f" FAILED {fname}: HTTP {e.code} {e.reason}") |
|
failed += 1 |
|
continue |
|
except Exception as e: |
|
print(f" FAILED {fname}: {e}") |
|
failed += 1 |
|
continue |
|
# an auth failure / missing file typically comes back as an HTML error page |
|
if "text/html" in ctype.lower(): |
|
print(f" FAILED {fname}: server returned HTML (auth problem or file gone)") |
|
failed += 1 |
|
continue |
|
out_name = unique_path(dest_dir, fname) |
|
with open(os.path.join(dest_dir, out_name), "wb") as fh: |
|
fh.write(blob) |
|
# if a collision renamed the file, point the reference at the actual name |
|
if out_name != fname: |
|
rel_ref = f"{args.dest}/{urllib.parse.quote(out_name)}" |
|
replacements[url] = rel_ref |
|
ok += 1 |
|
print(f" saved {out_name} ({len(blob):,} bytes)") |
|
|
|
if args.dry_run: |
|
print("\nDry run complete.") |
|
return |
|
|
|
# rewrite the JSON by exact textual replacement of each downloaded URL. |
|
# This export escapes slashes as "\/", so match that form first; fall back to |
|
# plain slashes so the tool also works on non-slash-escaped JSON. |
|
new_text = text |
|
for old, new in replacements.items(): |
|
esc_old = old.replace("/", "\\/") |
|
esc_new = new.replace("/", "\\/") |
|
if esc_old in new_text: |
|
new_text = new_text.replace(esc_old, esc_new) |
|
else: |
|
new_text = new_text.replace(old, new) |
|
|
|
if replacements: |
|
out_path = os.path.abspath(args.output) if args.output else json_path |
|
if out_path == json_path and not args.no_backup: |
|
bak = json_path + ".bak" |
|
if not os.path.exists(bak): |
|
with open(bak, "w", encoding="utf-8") as fh: |
|
fh.write(text) |
|
print(f"\nBackup written: {os.path.basename(bak)}") |
|
with open(out_path, "w", encoding="utf-8") as fh: |
|
fh.write(new_text) |
|
print(f"Updated JSON : {os.path.basename(out_path)} " |
|
f"({len(replacements)} URL(s) rewritten)") |
|
|
|
print(f"\nDone. downloaded={ok} failed={failed} " |
|
f"wiki-links-listed={len(wiki)} other={len(other)}") |
|
if failed: |
|
sys.exit(1) |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |