yolov26_3d/tools/feishu_project/export_feishu_view_issues.py

#!/usr/bin/env python3
"""Export Feishu view issues and enrich them with selected detail fields."""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from datetime import datetime
from pathlib import Path


DEFAULT_EXTRA_FIELDS = [
    "缺陷标签池",
    "问题数据地址",
    "问题数据地址_PDCL",
    "问题发生frameid",
]


def run_fp(args: list[str]) -> dict:
    result = subprocess.run(
        ["fp", *args],
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    if result.returncode != 0:
        stderr = result.stderr.strip()
        stdout = result.stdout.strip()
        detail = stderr or stdout or "unknown error"
        raise RuntimeError(f"fp command failed: {' '.join(args)}\n{detail}")
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        raise RuntimeError(
            f"fp returned non-JSON output for command: {' '.join(args)}"
        ) from exc


def fetch_view(project_key: str, user_key: str, view_name: str, work_item_type: str) -> dict | None:
    payload = run_fp(
        [
            "view",
            "list",
            "-o",
            "json",
            "-p",
            project_key,
            "-u",
            user_key,
            "-t",
            work_item_type,
            "--name",
            view_name,
        ]
    )
    views = payload.get("data", [])
    if not views:
        return None
    return views[0]


def fetch_items(project_key: str, user_key: str, view_name: str, work_item_type: str) -> dict:
    return run_fp(
        [
            "workitem",
            "list",
            "-o",
            "json",
            "--all",
            "-p",
            project_key,
            "-u",
            user_key,
            "-t",
            work_item_type,
            "--view",
            view_name,
        ]
    )


def fetch_item_detail(project_key: str, user_key: str, item_id: int, work_item_type: str) -> dict:
    payload = run_fp(
        [
            "workitem",
            "get",
            str(item_id),
            "-o",
            "json",
            "-p",
            project_key,
            "-u",
            user_key,
            "-t",
            work_item_type,
        ]
    )
    return payload["data"]


def enrich_items(
    base_items: list[dict],
    project_key: str,
    user_key: str,
    work_item_type: str,
    extra_fields: list[str],
) -> list[dict]:
    enriched = []
    for item in base_items:
        detail = fetch_item_detail(project_key, user_key, item["id"], work_item_type)
        detail_fields = detail.get("fields", {})
        merged = dict(item)
        for field_name in extra_fields:
            merged[field_name] = detail_fields.get(field_name)
        enriched.append(merged)
    return enriched


def build_output(
    project_key: str,
    user_key: str,
    view_name: str,
    view: dict | None,
    items_payload: dict,
    extra_fields: list[str],
    work_item_type: str,
) -> dict:
    return {
        "exported_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "source": {
            "project_key": project_key,
            "user_key": user_key,
            "view_name": view_name,
            "view_id": None if view is None else view.get("view_id"),
            "work_item_type": work_item_type,
            "included_detail_fields": extra_fields,
        },
        "summary": {
            "success": items_payload.get("success", False),
            "page": items_payload.get("page"),
            "total": items_payload.get("total"),
        },
        "items": items_payload["data"],
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--project-key", required=True)
    parser.add_argument("--user-key", required=True)
    parser.add_argument("--view-name", required=True)
    parser.add_argument(
        "--output",
        required=True,
        type=Path,
        help="Path to the exported JSON file.",
    )
    parser.add_argument(
        "--extra-field",
        action="append",
        dest="extra_fields",
        help="Extra detail field to include. Can be repeated.",
    )
    parser.add_argument(
        "--work-item-type",
        default="issue",
        help="Feishu work item type. Defaults to issue.",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    extra_fields = args.extra_fields or list(DEFAULT_EXTRA_FIELDS)

    view = fetch_view(args.project_key, args.user_key, args.view_name, args.work_item_type)
    items_payload = fetch_items(args.project_key, args.user_key, args.view_name, args.work_item_type)
    items_payload["data"] = enrich_items(
        items_payload.get("data", []),
        args.project_key,
        args.user_key,
        args.work_item_type,
        extra_fields,
    )

    output = build_output(
        args.project_key,
        args.user_key,
        args.view_name,
        view,
        items_payload,
        extra_fields,
        args.work_item_type,
    )

    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(output, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return 0


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