543 lines
19 KiB
Python
Executable File
543 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Download PlantTFDB species TF binding motif sets.
|
|
|
|
This script parses the official PlantTFDB download page and downloads links from
|
|
the "Sets of TF binding motifs for 156 species" motif section. It uses only the
|
|
Python standard library so it can run in a clean environment.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import concurrent.futures
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass, asdict
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
|
|
PLANTTFDB_DOWNLOAD_URL = "https://planttfdb.gao-lab.org/download.php"
|
|
DEFAULT_OUTDIR = Path("/home/xia-lab/下载/PlantTFDB_TF_binding_motifs_156_species")
|
|
USER_AGENT = "Mozilla/5.0 (compatible; bbav5-PlantTFDB-motif-downloader/1.0)"
|
|
SYSTEM_BIN_DIRS = ("/usr/bin", "/bin", "/usr/local/bin")
|
|
PREFER_EXTERNAL_TRANSPORT = False
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MotifLink:
|
|
url: str
|
|
filename: str
|
|
category: str
|
|
species: str = ""
|
|
|
|
|
|
class AnchorParser(HTMLParser):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.anchors: list[tuple[str, str]] = []
|
|
self._href: str | None = None
|
|
self._text: list[str] = []
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
if tag.lower() != "a":
|
|
return
|
|
attrs_dict = {k.lower(): v for k, v in attrs}
|
|
href = attrs_dict.get("href")
|
|
if href:
|
|
self._href = href
|
|
self._text = []
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
if self._href is not None:
|
|
self._text.append(data)
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
if tag.lower() == "a" and self._href is not None:
|
|
text = " ".join(x.strip() for x in self._text if x.strip())
|
|
self.anchors.append((self._href, text))
|
|
self._href = None
|
|
self._text = []
|
|
|
|
|
|
def urlopen(url: str, timeout: int, insecure: bool = False) -> bytes:
|
|
headers = {
|
|
"User-Agent": USER_AGENT,
|
|
"Accept": "*/*",
|
|
}
|
|
request = urllib.request.Request(url, headers=headers)
|
|
context = None
|
|
if insecure:
|
|
context = ssl._create_unverified_context()
|
|
with urllib.request.urlopen(request, timeout=timeout, context=context) as response:
|
|
return response.read()
|
|
|
|
|
|
def downgrade_to_http(url: str) -> str:
|
|
parsed = urllib.parse.urlparse(url)
|
|
if parsed.scheme != "https":
|
|
return url
|
|
return urllib.parse.urlunparse(parsed._replace(scheme="http"))
|
|
|
|
|
|
def clean_subprocess_env() -> dict[str, str]:
|
|
env = os.environ.copy()
|
|
for key in ("LD_LIBRARY_PATH", "LD_PRELOAD", "PYTHONHOME"):
|
|
env.pop(key, None)
|
|
env["PATH"] = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
|
|
return env
|
|
|
|
|
|
def preferred_executable(name: str) -> str | None:
|
|
for directory in SYSTEM_BIN_DIRS:
|
|
candidate = Path(directory) / name
|
|
if candidate.exists() and os.access(candidate, os.X_OK):
|
|
return str(candidate)
|
|
return shutil.which(name)
|
|
|
|
|
|
def curl_available() -> bool:
|
|
return preferred_executable("curl") is not None
|
|
|
|
|
|
def wget_available() -> bool:
|
|
return preferred_executable("wget") is not None
|
|
|
|
|
|
def wget_base_command(url: str, timeout: int, insecure: bool = False) -> list[str]:
|
|
wget = preferred_executable("wget")
|
|
if not wget:
|
|
raise RuntimeError("wget is not available")
|
|
command = [
|
|
wget,
|
|
"--inet4-only",
|
|
"--tries=6",
|
|
"--waitretry=2",
|
|
"--retry-connrefused",
|
|
"--no-http-keep-alive",
|
|
f"--timeout={timeout}",
|
|
f"--user-agent={USER_AGENT}",
|
|
]
|
|
if insecure:
|
|
command.append("--no-check-certificate")
|
|
command.append(url)
|
|
return command
|
|
|
|
|
|
def wget_fetch(url: str, timeout: int, insecure: bool = False) -> bytes:
|
|
command = wget_base_command(url, timeout=timeout, insecure=insecure)
|
|
command[-1:-1] = ["-O", "-"]
|
|
proc = subprocess.run(
|
|
command,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=clean_subprocess_env(),
|
|
)
|
|
if proc.returncode != 0:
|
|
stderr = proc.stderr.decode("utf-8", errors="replace").strip()
|
|
raise RuntimeError(f"wget failed with exit code {proc.returncode}: {stderr}")
|
|
return proc.stdout
|
|
|
|
|
|
def wget_download(url: str, dest: Path, timeout: int, insecure: bool = False) -> None:
|
|
command = wget_base_command(url, timeout=timeout, insecure=insecure)
|
|
command[-1:-1] = ["-O", str(dest)]
|
|
proc = subprocess.run(
|
|
command,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=clean_subprocess_env(),
|
|
)
|
|
if proc.returncode != 0:
|
|
stderr = proc.stderr.decode("utf-8", errors="replace").strip()
|
|
raise RuntimeError(f"wget failed with exit code {proc.returncode}: {stderr}")
|
|
|
|
|
|
def curl_base_command(url: str, timeout: int, insecure: bool = False) -> list[str]:
|
|
curl = preferred_executable("curl")
|
|
if not curl:
|
|
raise RuntimeError("curl is not available")
|
|
command = [
|
|
curl,
|
|
"-4",
|
|
"-L",
|
|
"--http1.1",
|
|
"--compressed",
|
|
"--fail",
|
|
"--retry",
|
|
"2",
|
|
"--connect-timeout",
|
|
str(min(timeout, 30)),
|
|
"--max-time",
|
|
str(timeout),
|
|
"-A",
|
|
USER_AGENT,
|
|
]
|
|
if insecure:
|
|
command.append("-k")
|
|
command.append(url)
|
|
return command
|
|
|
|
|
|
def curl_fetch(url: str, timeout: int, insecure: bool = False) -> bytes:
|
|
command = curl_base_command(url, timeout=timeout, insecure=insecure)
|
|
proc = subprocess.run(
|
|
command,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=clean_subprocess_env(),
|
|
)
|
|
if proc.returncode != 0:
|
|
stderr = proc.stderr.decode("utf-8", errors="replace").strip()
|
|
raise RuntimeError(f"curl failed with exit code {proc.returncode}: {stderr}")
|
|
return proc.stdout
|
|
|
|
|
|
def curl_download(url: str, dest: Path, timeout: int, insecure: bool = False) -> None:
|
|
command = curl_base_command(url, timeout=timeout, insecure=insecure)
|
|
command[-1:-1] = ["-o", str(dest)]
|
|
proc = subprocess.run(
|
|
command,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=clean_subprocess_env(),
|
|
)
|
|
if proc.returncode != 0:
|
|
stderr = proc.stderr.decode("utf-8", errors="replace").strip()
|
|
raise RuntimeError(f"curl failed with exit code {proc.returncode}: {stderr}")
|
|
|
|
|
|
def urlopen_with_http_fallback(url: str, timeout: int, insecure: bool = False) -> tuple[bytes, str]:
|
|
errors: list[str] = []
|
|
if not PREFER_EXTERNAL_TRANSPORT:
|
|
try:
|
|
return urlopen(url, timeout=timeout, insecure=insecure), url
|
|
except Exception as exc: # noqa: BLE001 - network libraries raise several exception families
|
|
errors.append(f"HTTPS/Python: {exc}")
|
|
http_url = downgrade_to_http(url)
|
|
if http_url != url:
|
|
print(
|
|
f"[PlantTFDB] HTTPS/Python fetch failed ({exc}); retrying with HTTP/Python: {http_url}",
|
|
file=sys.stderr,
|
|
)
|
|
try:
|
|
return urlopen(http_url, timeout=timeout, insecure=insecure), http_url
|
|
except Exception as http_exc: # noqa: BLE001 - fall through to external transport
|
|
errors.append(f"HTTP/Python: {http_exc}")
|
|
|
|
if wget_available():
|
|
for candidate in dict.fromkeys([url, downgrade_to_http(url)]):
|
|
try:
|
|
return wget_fetch(candidate, timeout=timeout, insecure=insecure), candidate
|
|
except Exception as exc: # noqa: BLE001 - include all transport errors
|
|
errors.append(f"{candidate}/wget: {exc}")
|
|
|
|
if curl_available():
|
|
for candidate in dict.fromkeys([url, downgrade_to_http(url)]):
|
|
try:
|
|
return curl_fetch(candidate, timeout=timeout, insecure=insecure), candidate
|
|
except Exception as exc: # noqa: BLE001 - include all transport errors
|
|
errors.append(f"{candidate}/curl: {exc}")
|
|
|
|
raise RuntimeError("; ".join(errors))
|
|
|
|
|
|
def download_with_transport_fallback(url: str, part: Path, timeout: int, insecure: bool = False) -> str:
|
|
errors: list[str] = []
|
|
try:
|
|
payload, used_url = urlopen_with_http_fallback(url, timeout=timeout, insecure=insecure)
|
|
if not payload:
|
|
raise RuntimeError("empty response")
|
|
with part.open("wb") as handle:
|
|
handle.write(payload)
|
|
return used_url
|
|
except Exception as exc: # noqa: BLE001 - fall through to direct curl download
|
|
errors.append(f"Python/fetch: {exc}")
|
|
|
|
if wget_available():
|
|
for candidate in dict.fromkeys([url, downgrade_to_http(url)]):
|
|
try:
|
|
wget_download(candidate, part, timeout=timeout, insecure=insecure)
|
|
if part.stat().st_size == 0:
|
|
raise RuntimeError("empty response")
|
|
return candidate
|
|
except Exception as exc: # noqa: BLE001 - include all transport errors
|
|
errors.append(f"{candidate}/wget-download: {exc}")
|
|
|
|
if curl_available():
|
|
for candidate in dict.fromkeys([url, downgrade_to_http(url)]):
|
|
try:
|
|
curl_download(candidate, part, timeout=timeout, insecure=insecure)
|
|
if part.stat().st_size == 0:
|
|
raise RuntimeError("empty response")
|
|
return candidate
|
|
except Exception as exc: # noqa: BLE001 - include all transport errors
|
|
errors.append(f"{candidate}/curl-download: {exc}")
|
|
|
|
raise RuntimeError("; ".join(errors))
|
|
|
|
|
|
|
|
def classify_filename(filename: str) -> str:
|
|
lower = filename.lower()
|
|
if lower.startswith("planttfdb_tf_binding_motifs_"):
|
|
return "global_experimental_motifs"
|
|
if lower.endswith("_tf_binding_motifs_information.txt") or lower.endswith("_information.txt"):
|
|
return "motif_information"
|
|
if lower.endswith(".meme.gz") or lower.endswith(".meme"):
|
|
return "all_motifs_meme"
|
|
if lower.endswith((".zip", ".tar.gz", ".tgz", ".tar")):
|
|
return "individual_motifs"
|
|
return "other"
|
|
|
|
|
|
def parse_motif_links(html: str, page_url: str) -> list[MotifLink]:
|
|
parser = AnchorParser()
|
|
parser.feed(html)
|
|
|
|
links: dict[str, MotifLink] = {}
|
|
current_species = ""
|
|
for href, _text in parser.anchors:
|
|
url = urllib.parse.urljoin(page_url, href)
|
|
parsed = urllib.parse.urlparse(url)
|
|
text = " ".join((_text or "").split())
|
|
if text and text.lower() != "download" and "/download/motif/" not in parsed.path:
|
|
current_species = text
|
|
if "/download/motif/" not in parsed.path:
|
|
continue
|
|
filename = Path(urllib.parse.unquote(parsed.path)).name
|
|
if not filename:
|
|
continue
|
|
category = classify_filename(filename)
|
|
links[url] = MotifLink(url=url, filename=filename, category=category, species=current_species)
|
|
|
|
return sorted(links.values(), key=lambda x: (x.species, x.category, x.filename, x.url))
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def target_path(outdir: Path, link: MotifLink) -> Path:
|
|
return outdir / link.category / link.filename
|
|
|
|
|
|
def download_one(
|
|
link: MotifLink,
|
|
outdir: Path,
|
|
timeout: int,
|
|
retries: int,
|
|
force: bool,
|
|
insecure: bool,
|
|
) -> dict:
|
|
dest = target_path(outdir, link)
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if dest.exists() and dest.stat().st_size > 0 and not force:
|
|
return {
|
|
**asdict(link),
|
|
"used_url": "",
|
|
"path": str(dest),
|
|
"status": "exists",
|
|
"size": dest.stat().st_size,
|
|
"sha256": sha256_file(dest),
|
|
"error": "",
|
|
}
|
|
|
|
last_error = ""
|
|
part = dest.with_suffix(dest.suffix + ".part")
|
|
for attempt in range(1, retries + 2):
|
|
try:
|
|
used_url = download_with_transport_fallback(
|
|
link.url,
|
|
part,
|
|
timeout=timeout,
|
|
insecure=insecure,
|
|
)
|
|
part.replace(dest)
|
|
return {
|
|
**asdict(link),
|
|
"used_url": used_url,
|
|
"path": str(dest),
|
|
"status": "downloaded",
|
|
"size": dest.stat().st_size,
|
|
"sha256": sha256_file(dest),
|
|
"error": "",
|
|
}
|
|
except Exception as exc: # noqa: BLE001 - keep download report complete
|
|
last_error = f"{type(exc).__name__}: {exc}"
|
|
if attempt <= retries:
|
|
time.sleep(min(10, attempt * 2))
|
|
|
|
try:
|
|
part.unlink()
|
|
except OSError:
|
|
pass
|
|
return {
|
|
**asdict(link),
|
|
"used_url": "",
|
|
"path": str(dest),
|
|
"status": "failed",
|
|
"size": 0,
|
|
"sha256": "",
|
|
"error": last_error,
|
|
}
|
|
|
|
|
|
def write_manifest(outdir: Path, rows: Iterable[dict], name: str) -> None:
|
|
rows = list(rows)
|
|
outdir.mkdir(parents=True, exist_ok=True)
|
|
|
|
json_path = outdir / f"{name}.json"
|
|
json_path.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
tsv_path = outdir / f"{name}.tsv"
|
|
fields = ["species", "category", "filename", "url", "used_url", "path", "status", "size", "sha256", "error"]
|
|
with tsv_path.open("w", encoding="utf-8") as handle:
|
|
handle.write("\t".join(fields) + "\n")
|
|
for row in rows:
|
|
handle.write("\t".join(str(row.get(field, "")).replace("\t", " ") for field in fields) + "\n")
|
|
|
|
|
|
def summarize_links(links: list[MotifLink]) -> dict[str, int]:
|
|
summary: dict[str, int] = {}
|
|
for link in links:
|
|
summary[link.category] = summary.get(link.category, 0) + 1
|
|
return dict(sorted(summary.items()))
|
|
|
|
|
|
def main() -> int:
|
|
global PREFER_EXTERNAL_TRANSPORT
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Download PlantTFDB Sets of TF binding motifs for 156 species.",
|
|
)
|
|
parser.add_argument("--out", type=Path, default=DEFAULT_OUTDIR, help="Output directory.")
|
|
parser.add_argument("--page-url", default=PLANTTFDB_DOWNLOAD_URL, help="PlantTFDB download page URL.")
|
|
parser.add_argument("--jobs", type=int, default=4, help="Parallel downloads.")
|
|
parser.add_argument("--timeout", type=int, default=120, help="Network timeout seconds.")
|
|
parser.add_argument("--retries", type=int, default=2, help="Retries per file.")
|
|
parser.add_argument("--force", action="store_true", help="Re-download existing files.")
|
|
parser.add_argument("--dry-run", action="store_true", help="Only parse links and write link manifest.")
|
|
parser.add_argument("--insecure", action="store_true", help="Disable TLS certificate verification.")
|
|
parser.add_argument(
|
|
"--prefer-external",
|
|
action="store_true",
|
|
help="Use system wget/curl first. Useful when Python TLS fails for PlantTFDB.",
|
|
)
|
|
parser.add_argument(
|
|
"--species",
|
|
default="",
|
|
help="Download only species whose name contains this text, e.g. 'Arabidopsis thaliana'.",
|
|
)
|
|
parser.add_argument(
|
|
"--category",
|
|
choices=[
|
|
"all",
|
|
"motif_information",
|
|
"all_motifs_meme",
|
|
"individual_motifs",
|
|
"global_experimental_motifs",
|
|
"other",
|
|
],
|
|
default="all",
|
|
help="Download only one category.",
|
|
)
|
|
parser.add_argument(
|
|
"--include-global-experimental",
|
|
action="store_true",
|
|
help="Include the global PlantTFDB experimental motif files in addition to species-projected sets.",
|
|
)
|
|
args = parser.parse_args()
|
|
PREFER_EXTERNAL_TRANSPORT = args.prefer_external
|
|
|
|
args.out.mkdir(parents=True, exist_ok=True)
|
|
print(f"[PlantTFDB] Fetching official download page: {args.page_url}", file=sys.stderr)
|
|
page_bytes, used_page_url = urlopen_with_http_fallback(
|
|
args.page_url,
|
|
timeout=args.timeout,
|
|
insecure=args.insecure,
|
|
)
|
|
html = page_bytes.decode("utf-8", errors="replace")
|
|
links = parse_motif_links(html, used_page_url)
|
|
if args.species:
|
|
needle = args.species.lower()
|
|
links = [link for link in links if needle in link.species.lower()]
|
|
if args.category == "all" and not args.include_global_experimental:
|
|
links = [link for link in links if link.category != "global_experimental_motifs"]
|
|
if args.category != "all":
|
|
links = [link for link in links if link.category == args.category]
|
|
|
|
if not links:
|
|
print("[PlantTFDB] No motif links found. The page structure may have changed.", file=sys.stderr)
|
|
return 2
|
|
|
|
link_rows = [
|
|
{
|
|
**asdict(link),
|
|
"used_url": "",
|
|
"path": str(target_path(args.out, link)),
|
|
"status": "planned",
|
|
"size": "",
|
|
"sha256": "",
|
|
"error": "",
|
|
}
|
|
for link in links
|
|
]
|
|
write_manifest(args.out, link_rows, "planttfdb_motif_links")
|
|
print(f"[PlantTFDB] Found {len(links)} motif files: {summarize_links(links)}", file=sys.stderr)
|
|
print(f"[PlantTFDB] Link manifest: {args.out / 'planttfdb_motif_links.tsv'}", file=sys.stderr)
|
|
|
|
if args.dry_run:
|
|
print("[PlantTFDB] Dry run only; no files downloaded.", file=sys.stderr)
|
|
return 0
|
|
|
|
results: list[dict] = []
|
|
max_workers = max(1, args.jobs)
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
future_to_link = {
|
|
executor.submit(download_one, link, args.out, args.timeout, args.retries, args.force, args.insecure): link
|
|
for link in links
|
|
}
|
|
for i, future in enumerate(concurrent.futures.as_completed(future_to_link), start=1):
|
|
result = future.result()
|
|
results.append(result)
|
|
print(
|
|
f"[{i}/{len(links)}] {result['status']}: {result['category']}/{result['filename']}",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
results = sorted(results, key=lambda x: (x["species"], x["category"], x["filename"], x["url"]))
|
|
write_manifest(args.out, results, "planttfdb_motif_download_manifest")
|
|
failed = [row for row in results if row["status"] == "failed"]
|
|
if failed:
|
|
write_manifest(args.out, failed, "planttfdb_motif_failed")
|
|
print(f"[PlantTFDB] Failed downloads: {len(failed)}. See planttfdb_motif_failed.tsv", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"[PlantTFDB] Complete. Manifest: {args.out / 'planttfdb_motif_download_manifest.tsv'}", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|