ec8d25d5bd0eac0a4f5172603e704106cbf6c66b max Sun Oct 4 06:37:38 2026 -0700 methbase2 otto: ssh to hgdownload as the invoking user, not as qateam, refs #34246 diff --git src/hg/utils/otto/methbase2/methbaseDownload src/hg/utils/otto/methbase2/methbaseDownload index 88bc53891f4..8f39723590a 100755 --- src/hg/utils/otto/methbase2/methbaseDownload +++ src/hg/utils/otto/methbase2/methbaseDownload @@ -1,538 +1,538 @@ #!/usr/bin/env python3 """Mirror the MethBase2 track hub from smithlab.usc.edu to local disk. hubClone cannot do this job. The hub's bigDataUrls are absolute and point into http://smithlab.usc.edu/methbase/data/, a tree that is not under the hub directory at all, and hubClone -download writes every data file into a single flat per-assembly directory named after the URL's basename. That would collapse the SRA project/sample/experiment hierarchy into ~20,000 files per assembly directory, and it only skips a file if it already exists, so it can never repair a truncated download or pick up a file that changed upstream. It is also single-threaded. This script keeps the remote directory layout, downloads in parallel, and re-fetches a file only when the remote Content-Length differs from what is on disk, so it works both for the initial mirror and for later updates. Layout under --outDir: hub/ the hub's own text files hub.txt genomes.txt <db>/trackDb.txt <db>/mb2_<db>_metadata.tsv <db>/mb2_<db>_colors.json data/ the bigDataUrl tree with --stripPrefix removed <SRA project>/<sample>/<experiment>/<experiment>.hmr.bb ... Every bigDataUrl in the mirrored trackDb files is rewritten so it no longer reaches back to smithlab.usc.edu. By default it becomes an absolute url under --baseUrl (https://hgdownload.soe.ucsc.edu/hubs/methbase/v3/data/...), which is what makes the hub loadable by url from anywhere. Pass --relativeUrls for the "../../data/..." form instead, which keeps the tree self-contained so it can be moved or served from a different host, at the cost of only working when hub/ and data/ stay siblings. """ import argparse import os import sys import threading import time from concurrent.futures import ThreadPoolExecutor from urllib.parse import urljoin, urlparse import requests # The faceted hub. Note this is NOT the hub at .../methylation/hub.txt, which is # the older superTrack-per-study layout. methbase_hub/ is an alias for the # latest methbase_hub_<date>/, and those dated directories are rewritten in # place, so the date in the name buys no reproducibility - prefer the alias. DEFAULT_HUB_URL = "http://smithlab.usc.edu/trackdata/methylation/methbase_hub/hub.txt" # The mirror lives on hgdownload, served as https://hgdownload.soe.ucsc.edu/hubs/ -# and reached as qateam@hgdownload. Each refresh gets its own vN directory, so +# served from hgdownload. Each refresh gets its own vN directory, so # bump this (or pass -o) when v4 is cut. Do not update the directory that is # being served: hardlink it first (cp -al v3 v4), update the copy, then swap. DEFAULT_OUT_DIR = "/mirrordata/hubs/methbase/v3" # Where that directory is served from. Rewritten bigDataUrls point here, so the # mirrored trackDb works for anyone who loads it by url, not only for a copy # sitting next to the data. Keep this and DEFAULT_OUT_DIR pointing at the same # vN directory. DEFAULT_BASE_URL = "https://hgdownload.soe.ucsc.edu/hubs/methbase/v3" DEFAULT_STRIP_PREFIX = "http://smithlab.usc.edu/methbase/data/" # trackDb settings whose value is a big binary file that lives outside the hub BIG_DATA_TAGS = frozenset([ "bigDataUrl", "bigDataIndex", "barChartMatrixUrl", "barChartSampleUrl", "linkDataUrl", "frames", "summary", "searchTrix", ]) # trackDb settings whose value is a small text file. When the value is a # relative name the file belongs to the hub and is mirrored under hub/ and # always re-fetched; when it is an absolute url it is treated as a data file # instead (see scanTrackDb). HUB_TEXT_TAGS = frozenset([ "metaDataUrl", "colorSettingsUrl", "html", "groups", ]) # how long to wait for the server, in seconds: (connect, read) TIMEOUT = (30, 300) # thread-local requests.Session, so each worker keeps its own keep-alive # connection to the server instead of reconnecting for every file _local = threading.local() def session(): "return this thread's requests.Session, making it on first use" s = getattr(_local, "session", None) if s is None: s = requests.Session() s.headers["User-Agent"] = "UCSC-methbaseDownload/1.0" _local.session = s return s def fetchText(url, retries=3): "GET url and return the body as text, retrying on transient errors" lastErr = None for attempt in range(retries): try: r = session().get(url, timeout=TIMEOUT) r.raise_for_status() return r.text except Exception as e: lastErr = e time.sleep(2 ** attempt) raise IOError("cannot fetch %s: %s" % (url, lastErr)) def remoteSize(url, retries=3): """return the Content-Length of url, or None if the server does not say. Raises on a hard error, e.g. a 404, so a vanished file gets reported instead of silently skipped.""" lastErr = None for attempt in range(retries): try: r = session().head(url, timeout=TIMEOUT, allow_redirects=True) r.raise_for_status() size = r.headers.get("Content-Length") return int(size) if size is not None else None except Exception as e: lastErr = e time.sleep(2 ** attempt) raise IOError("HEAD failed for %s: %s" % (url, lastErr)) def writeTextAtomic(localPath, text): """write a text file via a temp file and a rename. Never write in place: the mirror is served to the public while it updates, so a reader must not be able to catch a half-written trackDb, and updating a hardlinked staging copy (cp -al v3 v4) must not write through into the directory still being served.""" os.makedirs(os.path.dirname(localPath), exist_ok=True) tmp = localPath + ".tmp" with open(tmp, "w") as fh: fh.write(text) os.replace(tmp, localPath) def humanSize(n): "short byte count, e.g. 21K or 64.1M, so log lines stay narrow" for unit in ("B", "K", "M", "G", "T"): if abs(n) < 1024 or unit == "T": return "%d%s" % (n, unit) if unit == "B" else "%.1f%s" % (n, unit) n /= 1024.0 def isAbsoluteUrl(val): "True if a trackDb setting names a remote file rather than a hub-local one" return val.startswith("http://") or val.startswith("https://") def parseStanzas(text): """yield (tag, value) pairs from a trackDb/hub text file. Stanza boundaries do not matter here - we only need every setting that names a file - so this is a flat scan. Continuation lines ending in a backslash are joined.""" pending = "" for line in text.splitlines(): line = pending + line.rstrip("\n") pending = "" if line.endswith("\\"): pending = line[:-1] continue line = line.strip() if not line or line.startswith("#"): continue parts = line.split(None, 1) if len(parts) != 2: continue yield parts[0], parts[1].strip() class Mirror(object): "collects the download list from the hub text files, then fetches the data" def __init__(self, args): self.args = args self.hubDir = os.path.join(args.outDir, "hub") self.dataDir = os.path.join(args.outDir, "data") # url -> local path, deduplicated: the same file can be referenced from # more than one assembly self.dataFiles = {} self.lock = threading.Lock() self.nDone = 0 self.nSkipped = 0 self.nDownloaded = 0 self.bytesDownloaded = 0 self.failures = [] self.skippedGenomes = [] # url -> assembly that first referenced it, so the log can name the # assemblies a run actually touched self.fileDb = {} # (path, oldSize or None, newSize) for every file this run wrote self.changes = [] # ---------------------------------------------------------------- hub text def mirrorText(self, url, localPath): "fetch a hub text file, write it under hub/, and return its content" text = fetchText(url, self.args.retries) writeTextAtomic(localPath, text) return text def dataPath(self, url): """map a bigDataUrl to a path under data/. Normally the URL starts with --stripPrefix and the rest is used as-is. Anything else is filed under data/_other/<host>/<path> so a stray URL still lands somewhere sane instead of aborting the run.""" if url.startswith(self.args.stripPrefix): rel = url[len(self.args.stripPrefix):] else: p = urlparse(url) rel = os.path.join("_other", p.netloc, p.path.lstrip("/")) sys.stderr.write("warning: url outside --stripPrefix, storing as " "data/%s: %s\n" % (rel, url)) rel = rel.split("?")[0].lstrip("/") return os.path.join(self.dataDir, rel) def linkFor(self, target, outDir): """what a rewritten bigDataUrl should point at. By default an absolute url under --baseUrl, so the mirrored trackDb works for anyone who loads it from hgdownload rather than only from a copy sitting next to the data. --relativeUrls gives the "../../data/.." form instead, which keeps the tree self-contained and relocatable.""" rel = os.path.relpath(target, self.dataDir) if self.args.relativeUrls: return os.path.relpath(target, outDir) return "%s/data/%s" % (self.args.baseUrl.rstrip("/"), rel) def scanTrackDb(self, url, localPath, db, seen): """mirror one trackDb.txt, noting and relinking every file it names. The copy written to disk has each bigDataUrl replaced by a relative path to where that file will land under data/, which is what makes the mirror loadable on its own. Follows include lines; `seen` guards against an include cycle.""" if url in seen: return seen.add(url) text = fetchText(url, self.args.retries) outDir = os.path.dirname(localPath) os.makedirs(outDir, exist_ok=True) includes = [] out = [] for line in text.splitlines(True): parts = line.strip().split(None, 1) tag = parts[0] if parts else None val = parts[1].strip() if len(parts) == 2 else None # A hub-text setting given as an absolute url is not hub-local # at all - MethBase2 writes its description pages as # "html http://smithlab.usc.edu/methbase/data/.../x.html" - so # file it under data/ and relink it like any other data file. # Left alone it would both mangle the local path and leave the # mirrored trackDb reaching back to smithlab for the page. if val and (tag in BIG_DATA_TAGS or (tag in HUB_TEXT_TAGS and isAbsoluteUrl(val))): dataUrl = urljoin(url, val) target = self.dataPath(dataUrl) self.dataFiles.setdefault(dataUrl, target) self.fileDb.setdefault(dataUrl, db) indent = line[:len(line) - len(line.lstrip())] out.append("%s%s %s\n" % (indent, tag, self.linkFor(target, outDir))) continue # include and the small hub-local text files keep their # relative names, so they need no rewriting - only fetching if tag == "include" and val: includes.append(val) elif tag in HUB_TEXT_TAGS and val: try: self.mirrorText(urljoin(url, val), os.path.join(outDir, val)) except IOError as e: sys.stderr.write("warning: %s\n" % e) out.append(line) writeTextAtomic(localPath, "".join(out)) for val in includes: self.scanTrackDb(urljoin(url, val), os.path.join(outDir, val), db, seen) def scanHub(self): "mirror hub.txt, genomes.txt and every trackDb, filling self.dataFiles" hubUrl = self.args.hubUrl hubText = self.mirrorText(hubUrl, os.path.join(self.hubDir, "hub.txt")) genomesFile = None for tag, val in parseStanzas(hubText): if tag == "genomesFile": genomesFile = val if genomesFile is None: raise SystemExit("no genomesFile setting in %s" % hubUrl) genomesUrl = urljoin(hubUrl, genomesFile) genomesText = self.mirrorText(genomesUrl, os.path.join(self.hubDir, genomesFile)) # genomes.txt is stanzas of "genome <db>" followed by "trackDb <path>" genomes = [] db = None for tag, val in parseStanzas(genomesText): if tag == "genome": db = val elif tag == "trackDb" and db is not None: genomes.append((db, val)) db = None if self.args.assemblies: wanted = set(self.args.assemblies) missing = wanted - set(d for d, _ in genomes) if missing: raise SystemExit("assemblies not in this hub: %s" % ", ".join(sorted(missing))) genomes = [g for g in genomes if g[0] in wanted] print("hub has %d assemblies, reading their trackDb files" % len(genomes)) for db, tdbPath in genomes: before = len(self.dataFiles) try: self.scanTrackDb(urljoin(genomesUrl, tdbPath), os.path.join(self.hubDir, tdbPath), db, set()) except IOError as e: # An assembly listed in genomes.txt whose directory was never # published upstream must not sink the whole mirror - as of # 2026-09-15 the MethBase2 faceted hub declares aplCal1 but # serves no aplCal1/trackDb.txt. Skipped assemblies are counted # and make the exit status non-zero, so this is never silent. sys.stderr.write("warning: skipping assembly %s: %s\n" % (db, e)) self.skippedGenomes.append(db) continue print(" %-10s %6d data files (%d new)" % (db, len(self.dataFiles) - before, len(self.dataFiles) - before)) # ----------------------------------------------------------------- fetching def needsDownload(self, url, path): """(needed, oldSize) - oldSize is None when the file is new. oldSize is what the log reports the size difference against. No HEAD is sent when the file does not exist yet, which keeps the initial mirror down to one request per file.""" if not os.path.exists(path): return True, None localSize = os.path.getsize(path) if self.args.noHead: return False, localSize rSize = remoteSize(url, self.args.retries) if rSize is None: # server did not give a length, so we cannot compare: keep what we # have rather than re-downloading everything on every run return False, localSize return rSize != localSize, localSize def download(self, url, path): "stream url into path, via a .tmp file so a crash leaves no partial" os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".tmp" lastErr = None for attempt in range(self.args.retries): try: nBytes = 0 with session().get(url, timeout=TIMEOUT, stream=True) as r: r.raise_for_status() with open(tmp, "wb") as fh: for chunk in r.iter_content(chunk_size=1024 * 1024): fh.write(chunk) nBytes += len(chunk) expected = r.headers.get("Content-Length") if expected is not None and int(expected) != nBytes: raise IOError("short read: got %d of %s bytes" % (nBytes, expected)) os.replace(tmp, path) return nBytes except Exception as e: lastErr = e if os.path.exists(tmp): os.remove(tmp) time.sleep(2 ** attempt) raise IOError("download failed for %s: %s" % (url, lastErr)) def fetchOne(self, item): url, path = item try: needed, oldSize = self.needsDownload(url, path) if not needed: with self.lock: self.nSkipped += 1 self.tick() return if self.args.dryRun: print("would download %s -> %s" % (url, path)) with self.lock: self.nDownloaded += 1 self.tick() return nBytes = self.download(url, path) with self.lock: self.nDownloaded += 1 self.bytesDownloaded += nBytes self.changes.append((url, path, oldSize, nBytes)) self.tick() except Exception as e: with self.lock: self.failures.append((url, str(e))) self.tick() sys.stderr.write("ERROR %s: %s\n" % (url, e)) def tick(self): "progress line; caller must hold self.lock" self.nDone += 1 if self.nDone % self.args.progressEvery == 0 or self.nDone == self.total: elapsed = time.time() - self.startTime print("%d/%d done %d downloaded (%.1f GB) %d unchanged " "%d failed %.0fs" % (self.nDone, self.total, self.nDownloaded, self.bytesDownloaded / 1e9, self.nSkipped, len(self.failures), elapsed), flush=True) def fetchAll(self): items = sorted(self.dataFiles.items()) self.total = len(items) self.startTime = time.time() print("%d distinct data files, %d threads" % (self.total, self.args.threads)) with ThreadPoolExecutor(max_workers=self.args.threads) as pool: list(pool.map(self.fetchOne, items)) def writeLog(self): """append one short record of this run to download.log. Kept terse on purpose: a separator, the date, the assemblies this run actually changed, then one line per file written - the new size, or the old and new size and the delta for a file that was replaced.""" logPath = os.path.join(self.args.outDir, "download.log") dbs = sorted({self.fileDb.get(url, "?") for url, _, _, _ in self.changes}) with open(logPath, "a") as fh: fh.write("%s\n" % ("-" * 64)) fh.write("%s %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), " ".join(dbs) if dbs else "no changes")) for url, path, oldSize, newSize in sorted(self.changes, key=lambda c: c[1]): rel = os.path.relpath(path, self.args.outDir) if oldSize is None: fh.write("new %s %s\n" % (rel, humanSize(newSize))) else: fh.write("upd %s %s -> %s (%+d B)\n" % (rel, humanSize(oldSize), humanSize(newSize), newSize - oldSize)) if self.skippedGenomes: fh.write("skipped assemblies: %s\n" % " ".join(self.skippedGenomes)) if self.failures: fh.write("failed: %d (see failures.txt)\n" % len(self.failures)) fh.write("%d written, %.1f GB, %d unchanged\n" % (self.nDownloaded, self.bytesDownloaded / 1e9, self.nSkipped)) def report(self): print("\ndownloaded %d files (%.1f GB), %d already up to date, %d failed" % (self.nDownloaded, self.bytesDownloaded / 1e9, self.nSkipped, len(self.failures))) retVal = 0 if self.skippedGenomes: print("SKIPPED %d assemblies listed in genomes.txt whose trackDb " "could not be read: %s" % (len(self.skippedGenomes), ", ".join(self.skippedGenomes))) retVal = 1 if self.failures: failFile = os.path.join(self.args.outDir, "failures.txt") with open(failFile, "w") as fh: for url, err in self.failures: fh.write("%s\t%s\n" % (url, err)) print("failed urls written to %s - rerun the script to retry them" % failFile) retVal = 1 return retVal def main(): parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--hubUrl", default=DEFAULT_HUB_URL, help="hub.txt url (default: %(default)s)") parser.add_argument("-o", "--outDir", default=DEFAULT_OUT_DIR, help="output directory (default: %(default)s)") parser.add_argument("-t", "--threads", type=int, default=8, help="parallel downloads (default: %(default)s)") parser.add_argument("--stripPrefix", default=DEFAULT_STRIP_PREFIX, help="url prefix removed to get the path under data/ " "(default: %(default)s)") parser.add_argument("--baseUrl", default=DEFAULT_BASE_URL, help="public url the mirror is served at; rewritten " "bigDataUrls point under its data/ " "(default: %(default)s)") parser.add_argument("--relativeUrls", action="store_true", help="rewrite bigDataUrls as relative paths " "(../../data/...) instead of urls under " "--baseUrl, making the tree relocatable") parser.add_argument("-a", "--assemblies", nargs="+", metavar="DB", help="only these assemblies, e.g. -a hg38 mm39") parser.add_argument("-n", "--dryRun", action="store_true", help="list what would be downloaded, fetch nothing") parser.add_argument("--noHead", action="store_true", help="do not HEAD existing files, assume they are " "current; faster but misses upstream changes") parser.add_argument("--retries", type=int, default=3, help="attempts per url (default: %(default)s)") parser.add_argument("--progressEvery", type=int, default=200, help="print progress every N files (default: %(default)s)") args = parser.parse_args() # line buffering, so a run redirected to a log file shows progress as it # happens instead of in one block at the end sys.stdout.reconfigure(line_buffering=True) os.makedirs(args.outDir, exist_ok=True) mirror = Mirror(args) mirror.scanHub() mirror.fetchAll() if not args.dryRun: mirror.writeLog() return mirror.report() if __name__ == "__main__": sys.exit(main())