DataOps-platform/scripts/import_hopms_dataset.py

"""Parse HOPMs DOCX table definitions and prepare DataOps import payloads.

The default mode is dry-run: parse the Word document, generate a JSON report,
and do not call the DataOps platform.
"""

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path
from typing import Any, Iterable

import requests
from docx import Document
from docx.oxml.ns import qn
from docx.table import Table
from docx.text.paragraph import Paragraph

FIELD_HEADER_ALIASES = {
    "字段中文名": "name_zh",
    "字段名": "name_en",
    "类型": "type",
    "字节": "length",
    "填报/要求": "requirement",
    "填报要求": "requirement",
    "说明": "comment",
}

DEFAULT_CATEGORY = "HOPMs标准数据集"
TABLE_NAME_OVERRIDES = {
    "患者基本信息": "patient_demographics",
}


def _clean_text(value: Any) -> str:
    text = "" if value is None else str(value)
    text = text.replace("\u3000", " ")
    text = re.sub(r"\s+", " ", text)
    return text.strip()


def _compact_header(value: str) -> str:
    return _clean_text(value).replace(" ", "").replace("/", "")


def _iter_blocks(document: Document) -> Iterable[tuple[str, Any]]:
    for child in document.element.body.iterchildren():
        if child.tag == qn("w:p"):
            yield "paragraph", Paragraph(child, document)
        elif child.tag == qn("w:tbl"):
            yield "table", Table(child, document)


def _table_rows(table: Table) -> list[list[str]]:
    return [
        [_clean_text(cell.text) for cell in row.cells]
        for row in table.rows
    ]


def _header_map(header_row: list[str]) -> dict[str, int] | None:
    mapping: dict[str, int] = {}
    for idx, header in enumerate(header_row):
        normalized = _compact_header(header)
        field_name = FIELD_HEADER_ALIASES.get(normalized)
        if field_name:
            mapping[field_name] = idx

    required = {"name_zh", "name_en", "type"}
    if not required.issubset(mapping):
        return None
    return mapping


def _is_field_definition_table(rows: list[list[str]]) -> bool:
    if not rows:
        return False
    return _header_map(rows[0]) is not None


def _find_table_title(recent_paragraphs: list[str]) -> tuple[str, str, str]:
    """Return table Chinese name, English/code name, and description."""
    title_pattern = re.compile(r"(.+?)[（(]\s*([A-Za-z][A-Za-z0-9_]*)\s*[）)]")

    for idx in range(len(recent_paragraphs) - 1, -1, -1):
        paragraph = recent_paragraphs[idx]
        if paragraph in TABLE_NAME_OVERRIDES:
            description = ""
            if idx + 1 < len(recent_paragraphs):
                description = _clean_text(recent_paragraphs[idx + 1])
            return paragraph, TABLE_NAME_OVERRIDES[paragraph], description

    for idx in range(len(recent_paragraphs) - 1, -1, -1):
        paragraph = recent_paragraphs[idx]
        match = title_pattern.search(paragraph)
        if match:
            name_zh = _clean_text(match.group(1))
            name_en = _clean_text(match.group(2))
            description = ""
            if idx + 1 < len(recent_paragraphs):
                description = _clean_text(recent_paragraphs[idx + 1])
            return name_zh, name_en, description

    if recent_paragraphs:
        return _clean_text(recent_paragraphs[-1]), "", ""
    return "", "", ""


def _normalize_data_type(raw_type: str, raw_length: str) -> str:
    data_type = _clean_text(raw_type).lower()
    length = _clean_text(raw_length)

    if "日期时间" in data_type or "datetime" in data_type or "timestamp" in data_type:
        return "timestamp"
    if data_type == "日期" or data_type == "date":
        return "date"
    if "时间" in data_type and "日期" not in data_type:
        return "time"
    if "整数" in data_type or data_type in {"int", "integer"}:
        return "integer"
    if "长整数" in data_type or data_type == "bigint":
        return "bigint"
    if "数字" in data_type or "数值" in data_type or "金额" in data_type:
        return "numeric"
    if "布尔" in data_type or data_type == "boolean":
        return "boolean"
    if "字符" in data_type or "string" in data_type or "varchar" in data_type:
        if length.isdigit() and int(length) > 0:
            return f"varchar({int(length)})"
        return "varchar(255)"
    if data_type:
        return data_type
    return "varchar(255)"


def _is_primary_key(requirement: str, comment: str) -> bool:
    joined = f"{requirement} {comment}"
    return "主键" in joined


def _nullable(requirement: str) -> bool:
    requirement = _clean_text(requirement)
    return "必填" not in requirement


def _get_mapped_cell(row: list[str], mapping: dict[str, int], field: str) -> str:
    idx = mapping.get(field)
    if idx is None or idx >= len(row):
        return ""
    return _clean_text(row[idx])


def _parse_columns(rows: list[list[str]], mapping: dict[str, int]) -> list[dict[str, Any]]:
    columns: list[dict[str, Any]] = []
    for row in rows[1:]:
        if not any(_clean_text(cell) for cell in row):
            continue

        name_zh = _get_mapped_cell(row, mapping, "name_zh")
        name_en = _get_mapped_cell(row, mapping, "name_en")
        raw_type = _get_mapped_cell(row, mapping, "type")
        length = _get_mapped_cell(row, mapping, "length")
        requirement = _get_mapped_cell(row, mapping, "requirement")
        comment = _get_mapped_cell(row, mapping, "comment")

        if not name_zh and not name_en:
            continue

        columns.append(
            {
                "name_zh": name_zh,
                "name_en": name_en,
                "data_type": _normalize_data_type(raw_type, length),
                "is_primary": _is_primary_key(requirement, comment),
                "nullable": _nullable(requirement),
                "comment": comment,
                "requirement": requirement,
                "length": length,
            }
        )
    return columns


def parse_hopms_docx(docx_path: Path) -> list[dict[str, Any]]:
    document = Document(str(docx_path))
    recent_paragraphs: list[str] = []
    parsed_tables: list[dict[str, Any]] = []
    physical_table_index = 0

    for block_type, block in _iter_blocks(document):
        if block_type == "paragraph":
            paragraph_text = _clean_text(block.text)
            if paragraph_text:
                recent_paragraphs.append(paragraph_text)
                recent_paragraphs = recent_paragraphs[-8:]
            continue

        physical_table_index += 1
        rows = _table_rows(block)
        if not _is_field_definition_table(rows):
            continue

        mapping = _header_map(rows[0])
        if mapping is None:
            continue

        name_zh, name_en, description = _find_table_title(recent_paragraphs)
        columns = _parse_columns(rows, mapping)
        if not columns:
            continue

        if not name_en:
            name_en = f"HOPMS_TABLE_{len(parsed_tables) + 1:03d}"

        parsed_tables.append(
            {
                "source_table_index": physical_table_index,
                "table_info": {
                    "name_zh": name_zh,
                    "name_en": name_en,
                    "description": description,
                },
                "columns": columns,
            }
        )

    return parsed_tables


def _column_to_backend_meta(column: dict[str, Any]) -> dict[str, str]:
    return {
        "name_zh": _clean_text(column.get("name_zh")),
        "name_en": _clean_text(column.get("name_en")),
        "data_type": _clean_text(column.get("data_type")) or "varchar(255)",
        "describe": _clean_text(column.get("comment")),
    }


def _table_to_backend_payload(table: dict[str, Any]) -> dict[str, Any]:
    info = table["table_info"]
    return {
        "name_zh": info["name_zh"],
        "name_en": info["name_en"],
        "describe": info.get("description", ""),
        "type": "table",
        "category": DEFAULT_CATEGORY,
        "parsed_data": [_column_to_backend_meta(col) for col in table["columns"]],
    }


def build_dry_run_report(
    tables: list[dict[str, Any]],
    source_path: Path,
) -> dict[str, Any]:
    tables_missing_name_en = [
        table for table in tables
        if re.fullmatch(r"HOPMS_TABLE_\d{3}", table["table_info"].get("name_en", ""))
    ]
    empty_column_names = [
        {
            "table": table["table_info"]["name_en"],
            "source_table_index": table["source_table_index"],
            "column_index": idx + 1,
            "column": column,
        }
        for table in tables
        for idx, column in enumerate(table["columns"])
        if not column.get("name_zh") or not column.get("name_en")
    ]

    report_tables = []
    for table in tables:
        report_tables.append(
            {
                **table,
                "backend_payload": _table_to_backend_payload(table),
            }
        )

    return {
        "mode": "dry-run",
        "source": str(source_path),
        "summary": {
            "table_count": len(tables),
            "column_count": sum(len(table["columns"]) for table in tables),
            "tables_missing_name_en": len(tables_missing_name_en),
            "columns_missing_name": len(empty_column_names),
        },
        "warnings": {
            "tables_missing_name_en": [
                {
                    "source_table_index": table["source_table_index"],
                    "name_zh": table["table_info"]["name_zh"],
                    "generated_name_en": table["table_info"]["name_en"],
                }
                for table in tables_missing_name_en
            ],
            "columns_missing_name": empty_column_names,
        },
        "tables": report_tables,
    }


def _api_url(base_url: str, path: str) -> str:
    return f"{base_url.rstrip('/')}/{path.lstrip('/')}"


def _response_json(response: Any) -> dict[str, Any]:
    response.raise_for_status()
    data = response.json()
    if not isinstance(data, dict):
        raise ValueError(f"API returned non-object JSON: {data!r}")
    return data


def _ensure_success(data: dict[str, Any], operation: str) -> None:
    if data.get("code") != 200:
        message = data.get("message") or data.get("error") or data
        raise RuntimeError(f"{operation} failed: {message}")


def login_dataops(
    session: Any,
    base_url: str,
    username: str,
    password: str,
    timeout: int = 30,
) -> dict[str, Any]:
    response = session.post(
        _api_url(base_url, "/api/system/auth/login"),
        json={"username": username, "password": password},
        timeout=timeout,
    )
    data = _response_json(response)
    _ensure_success(data, "login")
    return data


def find_existing_domain_id(
    session: Any,
    base_url: str,
    name_en: str,
    timeout: int = 60,
) -> int | None:
    response = session.post(
        _api_url(base_url, "/api/bd/list"),
        json={"current": 1, "size": 10, "name_en": name_en},
        timeout=timeout,
    )
    data = _response_json(response)
    _ensure_success(data, "find existing business domain")
    records = ((data.get("data") or {}).get("records") or [])
    for record in records:
        if record.get("name_en") == name_en and record.get("id") is not None:
            return int(record["id"])
    return None


def import_tables(
    report_tables: list[dict[str, Any]],
    base_url: str,
    username: str = "",
    password: str = "",
    session: Any | None = None,
    timeout: int = 60,
    skip_login: bool = True,
    update_existing: bool = False,
) -> dict[str, Any]:
    http = session or requests.Session()
    if not skip_login:
        login_dataops(http, base_url, username, password, timeout=timeout)

    results: list[dict[str, Any]] = []
    for index, table in enumerate(report_tables, start=1):
        payload = table["backend_payload"]
        table_info = table["table_info"]
        result_item = {
            "index": index,
            "source_table_index": table.get("source_table_index"),
            "name_zh": table_info.get("name_zh"),
            "name_en": table_info.get("name_en"),
            "column_count": len(table.get("columns") or []),
            "status": "pending",
        }
        try:
            payload = dict(payload)
            existing_id = find_existing_domain_id(
                http,
                base_url,
                str(payload.get("name_en") or ""),
                timeout=timeout,
            )
            if existing_id is not None:
                result_item["existing_id"] = existing_id
                if not update_existing:
                    result_item["operation"] = "skip_existing"
                    result_item["status"] = "skipped"
                    results.append(result_item)
                    continue
                payload["id"] = existing_id
                result_item["operation"] = "update"
                endpoint = "/api/bd/update"
            else:
                result_item["operation"] = "create"
                endpoint = "/api/bd/save"
            response = http.post(
                _api_url(base_url, endpoint),
                json=payload,
                timeout=timeout,
            )
            data = _response_json(response)
            _ensure_success(data, "save business domain")
            result_item["status"] = "success"
            result_item["response"] = data.get("data")
        except Exception as exc:  # noqa: BLE001 - keep batch import moving.
            result_item["status"] = "failed"
            result_item["error"] = str(exc)
        results.append(result_item)

    succeeded = sum(1 for item in results if item["status"] == "success")
    failed = sum(1 for item in results if item["status"] == "failed")
    skipped = sum(1 for item in results if item["status"] == "skipped")
    return {
        "mode": "commit",
        "base_url": base_url.rstrip("/"),
        "summary": {
            "attempted": len(results),
            "succeeded": succeeded,
            "failed": failed,
            "skipped": skipped,
        },
        "results": results,
    }


def _find_default_docx() -> Path:
    matches = [
        path for path in Path("docs").glob("HOPMs*.docx")
        if not path.name.startswith("~$")
    ]
    if not matches:
        raise FileNotFoundError("未在 docs 目录找到 HOPMs*.docx")
    return matches[0]


def _write_report(report: dict[str, Any], output_path: Path) -> None:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(
        json.dumps(report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Parse HOPMs DOCX definitions and generate a DataOps dry-run report."
    )
    parser.add_argument("--docx", type=Path, default=None, help="HOPMs DOCX path")
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("docs/generated/hopms_dry_run.json"),
        help="Dry-run JSON output path",
    )
    parser.add_argument("--limit", type=int, default=0, help="Only include first N tables")
    parser.add_argument(
        "--commit",
        action="store_true",
        help="Submit parsed table definitions to DataOps.",
    )
    parser.add_argument(
        "--base-url",
        default="https://company.citupro.com:18183",
        help="DataOps platform base URL",
    )
    parser.add_argument("--username", default=None, help="DataOps username")
    parser.add_argument("--password", default=None, help="DataOps password")
    parser.add_argument(
        "--login",
        action="store_true",
        help="Validate username/password before importing. Default is direct API import.",
    )
    parser.add_argument(
        "--update-existing",
        action="store_true",
        help="Call /api/bd/update for existing domains. Default is to skip existing domains.",
    )
    parser.add_argument(
        "--import-output",
        type=Path,
        default=Path("docs/generated/hopms_import_result.json"),
        help="Commit-mode import result JSON output path",
    )
    args = parser.parse_args()

    docx_path = args.docx or _find_default_docx()
    tables = parse_hopms_docx(docx_path)
    if args.limit and args.limit > 0:
        tables = tables[: args.limit]

    report = build_dry_run_report(tables, source_path=docx_path)
    _write_report(report, args.output)

    summary = report["summary"]
    print(f"Dry-run report written: {args.output}")
    print(f"Tables: {summary['table_count']}")
    print(f"Columns: {summary['column_count']}")
    print(f"Generated table names: {summary['tables_missing_name_en']}")
    print(f"Columns missing name: {summary['columns_missing_name']}")

    if args.commit:
        if args.login and (not args.username or not args.password):
            raise SystemExit("--commit --login requires --username and --password")
        import_result = import_tables(
            report["tables"],
            base_url=args.base_url,
            username=args.username or "",
            password=args.password or "",
            skip_login=not args.login,
            update_existing=args.update_existing,
        )
        _write_report(import_result, args.import_output)
        import_summary = import_result["summary"]
        print(f"Import result written: {args.import_output}")
        print(f"Attempted: {import_summary['attempted']}")
        print(f"Succeeded: {import_summary['succeeded']}")
        print(f"Skipped: {import_summary['skipped']}")
        print(f"Failed: {import_summary['failed']}")
        if import_summary["failed"]:
            return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())