| """Pull the CWA open-data significant-earthquake report (E-A0015-001). |
| |
| The API needs a free authorization code. Provide it either in a file next to this |
| script: |
| |
| .cwa_api_key (one line, e.g. CWA-XXXXXXXX...) |
| |
| or via the environment variable CWA_API_KEY. |
| |
| Usage: |
| python fetch_cwa_api.py --inspect # 1 call, print the response shape |
| python fetch_cwa_api.py --from 2020-01-01 --to 2020-12-31 |
| python fetch_cwa_api.py --datetime "2020-06-14T00:00:00" |
| """ |
| import argparse |
| import json |
| import os |
| import urllib.parse |
| import urllib.request |
|
|
| from palert.paths import DATA_DIR, ROOT |
|
|
| ENDPOINT = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001" |
| KEY_FILE = os.path.join(ROOT, ".cwa_api_key") |
| RAW_DIR = os.path.join(DATA_DIR, "raw", "cwa_api") |
| os.makedirs(RAW_DIR, exist_ok=True) |
|
|
|
|
| def load_key(): |
| key = os.environ.get("CWA_API_KEY") |
| if not key and os.path.exists(KEY_FILE): |
| with open(KEY_FILE, encoding="utf-8") as fh: |
| key = fh.read().strip() |
| if not key: |
| raise SystemExit( |
| "No API key. Put it in .cwa_api_key (one line) or set CWA_API_KEY." |
| ) |
| return key |
|
|
|
|
| def call(params, key=None): |
| query = dict(params) |
| query.setdefault("format", "JSON") |
| query["Authorization"] = key or load_key() |
| url = ENDPOINT + "?" + urllib.parse.urlencode(query, doseq=True) |
| req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) |
| with urllib.request.urlopen(req, timeout=60) as resp: |
| return json.loads(resp.read().decode("utf-8")) |
|
|
|
|
| def summarize(obj, depth=0, path="", max_depth=4): |
| pad = " " * depth |
| if isinstance(obj, dict): |
| for k, v in obj.items(): |
| if isinstance(v, (dict, list)) and depth < max_depth: |
| n = f"[{len(v)}]" if isinstance(v, list) else "" |
| print(f"{pad}{k}: {type(v).__name__}{n}") |
| summarize(v, depth + 1, f"{path}.{k}", max_depth) |
| else: |
| print(f"{pad}{k}: {repr(v)[:70]}") |
| elif isinstance(obj, list) and obj: |
| print(f"{pad}(first element)") |
| summarize(obj[0], depth + 1, path + "[0]", max_depth) |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--inspect", action="store_true", help="single call, print the structure") |
| ap.add_argument("--datetime", dest="datetime_", help="single origin time yyyy-MM-ddThh:mm:ss") |
| ap.add_argument("--from", dest="date_from", help="start date yyyy-MM-dd") |
| ap.add_argument("--to", dest="date_to", help="end date yyyy-MM-dd") |
| ap.add_argument("--limit", type=int, default=1000) |
| ap.add_argument("--max-pages", type=int, default=50) |
| args = ap.parse_args() |
|
|
| if args.inspect: |
| data = call({"limit": 1}) |
| print(json.dumps(data, ensure_ascii=False)[:400]) |
| print("\n--- structure ---") |
| summarize(data) |
| return |
|
|
| if args.datetime_: |
| data = call({"OriginTime": args.datetime_, "limit": args.limit}) |
| out = os.path.join(RAW_DIR, "single.json") |
| with open(out, "w", encoding="utf-8") as fh: |
| json.dump(data, fh, ensure_ascii=False, indent=2) |
| print("saved", out) |
| return |
|
|
| if not (args.date_from and args.date_to): |
| ap.error("need --datetime, --inspect, or --from/--to") |
|
|
| collected = [] |
| for page in range(args.max_pages): |
| params = {"limit": args.limit, "offset": page * args.limit} |
| data = call(params) |
| out = os.path.join(RAW_DIR, f"page_{page:03d}.json") |
| with open(out, "w", encoding="utf-8") as fh: |
| json.dump(data, fh, ensure_ascii=False, indent=2) |
| recs = (data.get("records") or {}) |
| batch = [] |
| for v in recs.values(): |
| if isinstance(v, list): |
| batch = v |
| break |
| print(f"page {page}: {len(batch)} records") |
| if not batch: |
| break |
| collected.extend(batch) |
| if len(batch) < args.limit: |
| break |
|
|
| print(f"[total] {len(collected)} records across pages") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
