XinJe_PLC_Skill/scripts/catalog_xinje_downloads.py

import argparse
import csv
import datetime as dt
import html
import json
import pathlib
import re
import time
import urllib.parse
import urllib.request
from collections import Counter


BASE = "https://www.xinje.com"
DOWNLOAD_ENDPOINT = "/web/downloadCenter/file"

DOCUMENT_CATEGORIES = {
    16: "产品手册",
    19: "彩页/样本",
    31: "产品图纸",
    33: "EPLAN",
    49: "认证证书",
}

PLC_HMI_KEYWORDS = [
    "PLC",
    "可编程",
    "控制器",
    "XDP",
    "XDPPro",
    "TouchWin",
    "人机界面",
    "HMI",
    "XD",
    "XL",
    "XG",
    "XC",
    "XS",
    "TS",
    "TG",
]


def fetch(params: dict[str, str | int]) -> str:
    query = urllib.parse.urlencode(params)
    url = f"{BASE}{DOWNLOAD_ENDPOINT}?{query}"
    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": "Mozilla/5.0 (Codex Xinje catalog)",
            "X-Requested-With": "XMLHttpRequest",
        },
    )
    with urllib.request.urlopen(request, timeout=40) as response:
        return response.read().decode("utf-8", "ignore")


def clean_text(value: str) -> str:
    value = re.sub(r"<[^>]+>", "", value)
    return html.unescape(re.sub(r"\s+", " ", value).strip())


def text_between(pattern: str, block: str) -> str:
    match = re.search(pattern, block, flags=re.I | re.S)
    return clean_text(match.group(1)) if match else ""


def parse_results(markup: str) -> list[dict[str, str]]:
    blocks = re.findall(r"(<li>\s*<div class=\"con\">.*?</li>)", markup, flags=re.I | re.S)
    results = []
    for block in blocks:
        href_match = re.search(r'href=["\']([^"\']+)["\']', block, flags=re.I)
        href = html.unescape(href_match.group(1)) if href_match else ""
        title = text_between(r"<h3[^>]*>(.*?)</h3>", block)
        download_name = text_between(r'download=["\']([^"\']+)["\']', block)
        results.append(
            {
                "id": text_between(r'id=["\']file-(\d+)["\']', block),
                "title": title or download_name,
                "download_name": download_name,
                "url": href,
                "size": text_between(r'class=["\']filesize["\'][^>]*>.*?<em>(.*?)</em>', block),
                "version": text_between(r'class=["\']filever["\'][^>]*>.*?<em>(.*?)</em>', block),
                "file_type": text_between(r'class=["\']filetype["\'][^>]*>.*?<em>(.*?)</em>', block),
                "update_date": text_between(r'class=["\']filedate["\'][^>]*>.*?<em>(.*?)</em>', block),
            }
        )
    return results


def parse_max_page(markup: str, one_id: int) -> int:
    pages = [1]
    pattern = rf"changePage\((\d+),\s*0,\s*{one_id},\s*0\)"
    pages.extend(int(value) for value in re.findall(pattern, markup))
    jump_match = re.search(r'id=["\']pageJump["\'][^>]+max=["\'](\d+)["\']', markup, flags=re.I)
    if jump_match:
        pages.append(int(jump_match.group(1)))
    return max(pages)


def collect_category(one_id: int, delay: float) -> list[dict[str, str]]:
    first_markup = fetch({"seriesId": 0, "oneId": one_id, "twoId": 0, "page": 1, "fileName": ""})
    max_page = parse_max_page(first_markup, one_id)
    items = []
    for page in range(1, max_page + 1):
        markup = first_markup if page == 1 else fetch(
            {"seriesId": 0, "oneId": one_id, "twoId": 0, "page": page, "fileName": ""}
        )
        for item in parse_results(markup):
            item["category_id"] = str(one_id)
            item["category"] = DOCUMENT_CATEGORIES.get(one_id, str(one_id))
            item["source"] = f"{BASE}/web/downloadCenter/index"
            items.append(item)
        if delay:
            time.sleep(delay)
    return items


def write_csv(path: pathlib.Path, rows: list[dict[str, str]]) -> None:
    fieldnames = [
        "id",
        "category_id",
        "category",
        "title",
        "download_name",
        "file_type",
        "size",
        "version",
        "update_date",
        "url",
        "source",
    ]
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            writer.writerow({field: row.get(field, "") for field in fieldnames})


def write_json(path: pathlib.Path, rows: list[dict[str, str]]) -> None:
    payload = {
        "cataloged_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        "source": f"{BASE}/web/downloadCenter/index",
        "scope": "Xinje download-center document-like categories; metadata only, no file downloads",
        "categories": DOCUMENT_CATEGORIES,
        "total_items": len(rows),
        "items": rows,
    }
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def write_markdown(path: pathlib.Path, rows: list[dict[str, str]]) -> None:
    by_category = Counter(row["category"] for row in rows)
    by_type = Counter((row.get("file_type") or "unknown").lower() for row in rows)
    plc_hmi = [
        row for row in rows
        if any(keyword.lower() in (row.get("title", "") + row.get("download_name", "")).lower()
               for keyword in PLC_HMI_KEYWORDS)
    ]
    recent = sorted(rows, key=lambda row: row.get("update_date", ""), reverse=True)[:20]

    lines = [
        "# Xinje Official Document Catalog",
        "",
        f"Cataloged at: {dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "",
        "Scope: metadata from Xinje download-center document-like categories. Files are not downloaded by this script.",
        "",
        "## Counts By Category",
        "",
        "| Category | Count |",
        "|---|---:|",
    ]
    for category, count in by_category.most_common():
        lines.append(f"| {category} | {count} |")

    lines.extend(["", "## Counts By File Type", "", "| File type | Count |", "|---|---:|"])
    for file_type, count in by_type.most_common():
        lines.append(f"| {file_type} | {count} |")

    lines.extend([
        "",
        "## Recent Items",
        "",
        "| ID | Category | Title | Type | Version | Update date |",
        "|---|---|---|---|---|---|",
    ])
    for row in recent:
        lines.append(
            f"| {row.get('id','')} | {row.get('category','')} | {row.get('title','')} | "
            f"{row.get('file_type','')} | {row.get('version','')} | {row.get('update_date','')} |"
        )

    lines.extend([
        "",
        "## PLC/HMI Candidates",
        "",
        "Use the full CSV/JSON catalog for exact URLs. This section highlights likely PLC/HMI-related entries.",
        "",
        "| ID | Category | Title | Type | Version | Update date |",
        "|---|---|---|---|---|---|",
    ])
    for row in plc_hmi[:120]:
        lines.append(
            f"| {row.get('id','')} | {row.get('category','')} | {row.get('title','')} | "
            f"{row.get('file_type','')} | {row.get('version','')} | {row.get('update_date','')} |"
        )

    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def main() -> None:
    parser = argparse.ArgumentParser(description="Catalog Xinje download-center document metadata.")
    parser.add_argument("--output-dir", default="assets/catalogs", help="Directory for JSON/CSV/MD outputs.")
    parser.add_argument("--delay", type=float, default=0.15, help="Delay between page requests.")
    parser.add_argument("--categories", default=",".join(str(key) for key in DOCUMENT_CATEGORIES))
    args = parser.parse_args()

    rows = []
    seen = set()
    for one_id in [int(value) for value in args.categories.split(",") if value.strip()]:
        for item in collect_category(one_id, args.delay):
            key = item.get("id") or item.get("url")
            if key in seen:
                continue
            seen.add(key)
            rows.append(item)

    rows.sort(key=lambda row: (row.get("category_id", ""), row.get("update_date", ""), row.get("id", "")), reverse=True)

    output_dir = pathlib.Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    write_json(output_dir / "xinje-official-document-catalog.json", rows)
    write_csv(output_dir / "xinje-official-document-catalog.csv", rows)
    write_markdown(output_dir / "xinje-official-document-catalog-summary.md", rows)
    print(f"cataloged {len(rows)} items into {output_dir}")


if __name__ == "__main__":
    main()