XinJe_PLC_Skill/scripts/download_catalog_ids.py

import argparse
import csv
import pathlib
import re
import sys

from download_xinje_file import request, quote_url
import urllib.request


INVALID_FILENAME_CHARS = r'<>:"/\|?*'


def clean_filename(value: str) -> str:
    value = "".join("_" if ch in INVALID_FILENAME_CHARS else ch for ch in value)
    value = re.sub(r"\s+", " ", value).strip(" .")
    return value or "xinje-download"


def build_filename(row: dict[str, str]) -> str:
    base = row.get("download_name") or row.get("title") or row["id"]
    base = clean_filename(base)
    version = clean_filename(row.get("version") or "")
    update_date = clean_filename(row.get("update_date") or "")
    suffix_parts = []
    if version:
        suffix_parts.append(f"v{version}")
    if update_date:
        suffix_parts.append(update_date)
    suffix = f" ({', '.join(suffix_parts)})" if suffix_parts else ""
    extension = (row.get("file_type") or "pdf").lower().lstrip(".")
    return f"{base}{suffix}.{extension}"


def load_rows(catalog: pathlib.Path) -> dict[str, dict[str, str]]:
    with catalog.open("r", encoding="utf-8-sig", newline="") as handle:
        return {row["id"]: row for row in csv.DictReader(handle)}


def download(row: dict[str, str], output_dir: pathlib.Path, skip_existing: bool) -> pathlib.Path:
    output = output_dir / build_filename(row)
    if output.exists() and skip_existing:
        print(f"skip existing {row['id']}: {output}")
        return output

    url = quote_url(row["url"])
    output.parent.mkdir(parents=True, exist_ok=True)
    with urllib.request.urlopen(request(url), timeout=120) as response:
        output.write_bytes(response.read())
    print(f"downloaded {row['id']}: {output} ({output.stat().st_size} bytes)")
    return output


def main() -> None:
    parser = argparse.ArgumentParser(description="Download Xinje catalog files by ID.")
    parser.add_argument("--catalog", required=True, type=pathlib.Path)
    parser.add_argument("--output-dir", required=True, type=pathlib.Path)
    parser.add_argument("--ids", required=True, nargs="+")
    parser.add_argument("--overwrite", action="store_true")
    args = parser.parse_args()

    rows = load_rows(args.catalog)
    missing = [item for item in args.ids if item not in rows]
    if missing:
        raise SystemExit(f"catalog IDs not found: {', '.join(missing)}")

    for item in args.ids:
        download(rows[item], args.output_dir, skip_existing=not args.overwrite)


if __name__ == "__main__":
    sys.exit(main())