aseimel
241 lines
9.8 KiB
Python
241 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Download all script-accessible raw sources into _local/raw/.
|
|
|
|
The downloader keeps every file inside PARTY2D_RAW_DATA_DIR (default:
|
|
_local/raw). It does not write to committed data/. Sources that require an API
|
|
key or provider form are handled explicitly and reported rather than silently
|
|
skipped.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import html.parser
|
|
import json
|
|
import http.cookiejar
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
|
|
UA = "party2d-data-setup/1.0 (+https://git.seimel.app/armin/party2d)"
|
|
|
|
|
|
def raw_dir() -> Path:
|
|
return Path(os.environ.get("PARTY2D_RAW_DATA_DIR", Path.cwd() / "_local" / "raw")).resolve()
|
|
|
|
|
|
def report_dir() -> Path:
|
|
return Path(os.environ.get("PARTY2D_REPORT_DIR", Path.cwd() / "_local" / "reports")).resolve()
|
|
|
|
|
|
def request(url: str, data: bytes | None = None, headers: dict[str, str] | None = None):
|
|
h = {"User-Agent": UA}
|
|
if headers:
|
|
h.update(headers)
|
|
return urllib.request.Request(url, data=data, headers=h)
|
|
|
|
|
|
def download(url: str, dest: Path, *, overwrite: bool = False, headers: dict[str, str] | None = None) -> bool:
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if dest.exists() and dest.stat().st_size > 0 and not overwrite:
|
|
print(f"OK existing: {dest}")
|
|
return True
|
|
tmp = dest.with_suffix(dest.suffix + ".tmp")
|
|
print(f"Downloading {url} -> {dest}")
|
|
try:
|
|
with urllib.request.urlopen(request(url, headers=headers), timeout=180) as r, tmp.open("wb") as fh:
|
|
shutil.copyfileobj(r, fh)
|
|
tmp.replace(dest)
|
|
return True
|
|
except Exception as e:
|
|
if tmp.exists():
|
|
tmp.unlink()
|
|
print(f"FAILED {url}: {type(e).__name__}: {e}")
|
|
return False
|
|
|
|
|
|
def dataverse_file_id(doi: str, filename: str, directory: str | None = None) -> int | None:
|
|
api = "https://dataverse.harvard.edu/api/datasets/:persistentId/?persistentId=" + urllib.parse.quote(doi, safe="")
|
|
with urllib.request.urlopen(request(api), timeout=120) as r:
|
|
data = json.load(r)
|
|
for f in data["data"]["latestVersion"]["files"]:
|
|
df = f["dataFile"]
|
|
if df["filename"] == filename and (directory is None or f.get("directoryLabel") == directory):
|
|
return int(df["id"])
|
|
return None
|
|
|
|
|
|
def download_dataverse(doi: str, filename: str, dest: Path, directory: str | None = None) -> bool:
|
|
try:
|
|
fid = dataverse_file_id(doi, filename, directory)
|
|
except Exception as e:
|
|
print(f"FAILED Dataverse lookup {doi} {filename}: {type(e).__name__}: {e}")
|
|
return False
|
|
if fid is None:
|
|
print(f"FAILED Dataverse lookup {doi}: file not found: {directory or ''}/{filename}")
|
|
return False
|
|
return download(f"https://dataverse.harvard.edu/api/access/datafile/{fid}", dest)
|
|
|
|
|
|
class InputParser(html.parser.HTMLParser):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.inputs: dict[str, str] = {}
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
if tag != "input":
|
|
return
|
|
d = dict(attrs)
|
|
name = d.get("name")
|
|
if name:
|
|
self.inputs[name] = d.get("value", "")
|
|
|
|
|
|
def download_vparty(dest: Path) -> bool:
|
|
"""Download V-Party R zip through V-Dem's required form.
|
|
|
|
V-Dem requires email, gender, privacy acceptance, and format selection. We do
|
|
not invent personal info: set PARTY2D_VDEM_EMAIL and PARTY2D_VDEM_GENDER
|
|
(woman/man/non_binary/empty for prefer-not-to-say) to enable this download.
|
|
"""
|
|
email = os.environ.get("PARTY2D_VDEM_EMAIL")
|
|
if not email:
|
|
print("SKIP V-Party: set PARTY2D_VDEM_EMAIL to use V-Dem's required download form")
|
|
return False
|
|
gender = os.environ.get("PARTY2D_VDEM_GENDER", "")
|
|
page = "https://www.v-dem.net/data/v-party-dataset/country-party-date-v2/"
|
|
cj = http.cookiejar.CookieJar()
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
|
|
try:
|
|
with opener.open(request(page), timeout=60) as r:
|
|
html = r.read().decode("utf-8", "replace")
|
|
except Exception as e:
|
|
print(f"FAILED V-Party form load: {type(e).__name__}: {e}")
|
|
return False
|
|
p = InputParser(); p.feed(html)
|
|
csrf = p.inputs.get("csrfmiddlewaretoken", "")
|
|
fields = {
|
|
"csrfmiddlewaretoken": csrf,
|
|
"email": email,
|
|
"gender": gender,
|
|
"accept_terms": "on",
|
|
"dataset_file": "17", # R format in current V-Dem form
|
|
"website": "",
|
|
}
|
|
data = urllib.parse.urlencode(fields).encode()
|
|
headers = {
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Referer": page,
|
|
"X-CSRFToken": csrf,
|
|
"X-Requested-With": "XMLHttpRequest",
|
|
}
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = dest.with_suffix(".zip")
|
|
print(f"Submitting V-Dem form -> {tmp}")
|
|
try:
|
|
with opener.open(request(page + "#dataset-download", data=data, headers=headers), timeout=180) as r:
|
|
payload = r.read()
|
|
ctype = r.headers.get("content-type", "")
|
|
# The AJAX form may return JSON with a download URL, or the file itself.
|
|
if b"download" in payload[:1000] and "json" in ctype:
|
|
js = json.loads(payload.decode("utf-8"))
|
|
url = js.get("download_url") or js.get("url") or js.get("file")
|
|
if url and url.startswith("/"):
|
|
url = urllib.parse.urljoin(page, url)
|
|
if not url:
|
|
print(f"FAILED V-Party form response did not include download URL: {js}")
|
|
return False
|
|
if not download(url, tmp, overwrite=True, headers={"Referer": page}):
|
|
return False
|
|
else:
|
|
text = payload.decode("utf-8", "replace")
|
|
m = re.search(r'href="([^"]*CPD_V-Party_R_v2\.zip[^"]*)"', text)
|
|
if m:
|
|
url = urllib.parse.urljoin(page, m.group(1))
|
|
if not download(url, tmp, overwrite=True, headers={"Referer": page}):
|
|
return False
|
|
else:
|
|
tmp.write_bytes(payload)
|
|
if zipfile.is_zipfile(tmp):
|
|
with zipfile.ZipFile(tmp) as z:
|
|
r_files = [n for n in z.namelist() if n.lower().endswith((".rds", ".rda", ".rdata"))]
|
|
if not r_files:
|
|
print("FAILED V-Party zip contains no R data file")
|
|
return False
|
|
member = r_files[0]
|
|
with z.open(member) as src, dest.open("wb") as out:
|
|
shutil.copyfileobj(src, out)
|
|
print(f"Extracted V-Party R data: {dest}")
|
|
return True
|
|
print("FAILED V-Party download was not a zip file")
|
|
return False
|
|
except Exception as e:
|
|
print(f"FAILED V-Party form submit: {type(e).__name__}: {e}")
|
|
return False
|
|
|
|
|
|
def download_manifesto(dest: Path) -> bool:
|
|
key = os.environ.get("MANIFESTO_API_KEY") or os.environ.get("PARTY2D_MANIFESTO_API_KEY")
|
|
if not key:
|
|
print("SKIP Manifesto: set MANIFESTO_API_KEY or PARTY2D_MANIFESTO_API_KEY")
|
|
return False
|
|
url = "https://manifesto-project.wzb.eu/api/v1/get_core?" + urllib.parse.urlencode({"key": "MPDS2025a", "raw": "true"})
|
|
return download(
|
|
url,
|
|
dest,
|
|
overwrite=True,
|
|
headers={"Referer": "https://manifesto-project.wzb.eu/datasets", "API_KEY": key},
|
|
)
|
|
|
|
|
|
def write_report(status: dict[str, bool]) -> None:
|
|
rd = report_dir(); rd.mkdir(parents=True, exist_ok=True)
|
|
path = rd / "download_sources_report.md"
|
|
lines = ["# Download sources report", "", "| source | status |", "| --- | --- |"]
|
|
for k, v in status.items():
|
|
lines.append(f"| {k} | {'ok' if v else 'missing/failed'} |")
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
print(f"Wrote download report: {path}")
|
|
|
|
|
|
def main() -> int:
|
|
root = raw_dir()
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
status: dict[str, bool] = {}
|
|
|
|
status["PolDem"] = download("https://poldem.eui.eu/downloads/cosa/poldem-election_all.csv", root / "poldem" / "poldem-election_all.csv")
|
|
status["PartyFacts external parties"] = download("https://partyfacts.herokuapp.com/download/external-parties-csv/", root / "partyfacts" / "partyfacts-external-parties.csv")
|
|
status["Manifesto MPDS 2025a"] = download_manifesto(root / "manifesto" / "MPDataset_MPDS2025a.csv")
|
|
|
|
ches_base = "https://www.chesdata.eu/s/"
|
|
ches_files = {
|
|
"1999-2019_CHES_dataset_meansv3.csv": "1999-2019_CHES_dataset_means(v3).csv",
|
|
"1999-2024_CHES_dataset_meansV2-3k4l.csv": "1999-2024_CHES_dataset_meansV2-3k4l.csv",
|
|
"CHES_2024_final_v2.csv": "CHES_2024_final_v2.csv",
|
|
"CHES_2024_ALL_Stacked_Expert.csv": "CHES_2024_expert_level.csv",
|
|
"CHES_CA2023.csv": "CHES_CA2023.csv",
|
|
"CHES_CA2023_expert-level.csv": "CHES_CA2023_expert_level.csv",
|
|
"ches_la_2020_aggregate_level_v01.csv": "ches_la_2020_aggregate_level_v01.csv",
|
|
"ches_la_2020_expert_level_v01.csv": "CHES_LA2020_expert_level.csv",
|
|
"CHES_ISRAEL_means_2021_2022.csv": "CHES_ISRAEL_means_2021_2022.csv",
|
|
"CHES_ISRAEL_expert_level_2021_2022.csv": "CHES_IL_expert_level.csv",
|
|
}
|
|
for remote, local in ches_files.items():
|
|
status[f"CHES {local}"] = download(ches_base + urllib.parse.quote(remote), root / "ches" / local)
|
|
|
|
status["POPPA integrated v2"] = download_dataverse("doi:10.7910/DVN/RMQREQ", "poppa_integrated_v2.rds", root / "poppa" / "poppa_integrated_v2.rds", "final_data_v2")
|
|
status["GPS 2019 party"] = download_dataverse("doi:10.7910/DVN/WMGTNS", "Global Party Survey by Party SPSS V2_1_Apr_2020-2.tab", root / "gps" / "Global Party Survey by Party SPSS V2_1_Apr_2020-2.tab")
|
|
status["V-Party"] = download_vparty(root / "vparty" / "V-Dem-CPD-Party-V2.rds")
|
|
|
|
write_report(status)
|
|
return 0 if all(status.values()) else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|