bigscreen/scripts/plane_sync.py

#!/usr/bin/env python3
"""
Plane 项目管理集成脚本
同步 Gitea CI/CD 状态到 Plane 项目管理系统
"""

import os
import sys
import json
import requests
from datetime import datetime
from typing import Dict, Any, Optional

# Plane API 配置
PLANE_API_URL = os.getenv("PLANE_API_URL", "http://192.168.1.83:8088/api")
PLANE_API_KEY = os.getenv("PLANE_API_KEY", "")
PLANE_WORKSPACE = os.getenv("PLANE_WORKSPACE", "ljwx")
PLANE_PROJECT_ID = os.getenv("PLANE_PROJECT_ID", "BIGSCREEN")

# Gitea 事件信息
GITHUB_EVENT_NAME = os.getenv("GITHUB_EVENT_NAME", "")
GITHUB_REF = os.getenv("GITHUB_REF", "")
GITHUB_SHA = os.getenv("GITHUB_SHA", "")
GITHUB_ACTOR = os.getenv("GITHUB_ACTOR", "")
WORKFLOW_STATUS = os.getenv("WORKFLOW_STATUS", "success")
WORKFLOW_NAME = os.getenv("WORKFLOW_NAME", "")


class PlaneClient:
    """Plane API 客户端"""

    def __init__(self, api_url: str, api_key: str, workspace: str):
        self.api_url = api_url.rstrip("/")
        self.api_key = api_key
        self.workspace = workspace
        self.headers = {
            "X-API-Key": api_key,
            "Content-Type": "application/json",
        }

    def get_project(self, project_identifier: str) -> Optional[Dict[str, Any]]:
        """获取项目信息 - 通过identifier查找项目"""
        # Plane API 不支持直接通过 identifier 获取项目，需要先获取列表
        url = f"{self.api_url}/v1/workspaces/{self.workspace}/projects/"
        try:
            response = requests.get(url, headers=self.headers, timeout=10)
            if response.status_code == 200:
                data = response.json()
                # 在结果中查找匹配的项目
                for project in data.get("results", []):
                    if project.get("identifier") == project_identifier:
                        return project
                print(f"⚠️ 未找到标识符为 {project_identifier} 的项目")
                return None
            else:
                print(f"⚠️ 获取项目列表失败: {response.status_code} - {response.text}")
                return None
        except Exception as e:
            print(f"❌ API 请求失败: {e}")
            return None

    def create_issue(
        self, project_id: str, title: str, description: str, **kwargs
    ) -> Optional[Dict[str, Any]]:
        """创建 Issue"""
        url = f"{self.api_url}/v1/workspaces/{self.workspace}/projects/{project_id}/issues/"

        data = {
            "name": title,
            "description_html": description,
            "priority": kwargs.get("priority", "medium"),
        }

        # 只有当提供了有效的 state 和 labels 时才添加
        if kwargs.get("state"):
            data["state"] = kwargs.get("state")

        # labels 需要是 UUID 数组，暂时不使用
        # if kwargs.get("labels"):
        #     data["labels"] = kwargs.get("labels")

        # 移除 None 值
        data = {k: v for k, v in data.items() if v is not None}

        try:
            response = requests.post(url, headers=self.headers, json=data, timeout=10)
            if response.status_code in [200, 201]:
                return response.json()
            else:
                print(f"⚠️ 创建 Issue 失败: {response.status_code} - {response.text}")
                return None
        except Exception as e:
            print(f"❌ 创建 Issue 失败: {e}")
            return None

    def add_comment(
        self, project_id: str, issue_id: str, comment: str
    ) -> Optional[Dict[str, Any]]:
        """添加评论"""
        url = f"{self.api_url}/v1/workspaces/{self.workspace}/projects/{project_id}/issues/{issue_id}/comments/"

        data = {"comment_html": comment}

        try:
            response = requests.post(url, headers=self.headers, json=data, timeout=10)
            if response.status_code in [200, 201]:
                return response.json()
            else:
                print(f"⚠️ 添加评论失败: {response.status_code} - {response.text}")
                return None
        except Exception as e:
            print(f"❌ 添加评论失败: {e}")
            return None


def generate_build_report() -> str:
    """生成构建报告"""
    status_emoji = "✅" if WORKFLOW_STATUS == "success" else "❌"

    report = f"""
<h3>{status_emoji} CI/CD 构建报告</h3>

<p><strong>工作流:</strong> {WORKFLOW_NAME}</p>
<p><strong>状态:</strong> {WORKFLOW_STATUS}</p>
<p><strong>分支:</strong> {GITHUB_REF}</p>
<p><strong>提交:</strong> <code>{GITHUB_SHA[:7]}</code></p>
<p><strong>触发者:</strong> {GITHUB_ACTOR}</p>
<p><strong>时间:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>

<h4>🔗 相关链接</h4>
<ul>
  <li><a href="http://192.168.1.83:33000/gao/bigscreen">Gitea 仓库</a></li>
  <li><a href="http://192.168.1.83:33000/gao/bigscreen/actions">CI/CD 记录</a></li>
  <li><a href="http://192.168.1.83:15000/v2/bigscreen/tags/list">镜像仓库</a></li>
</ul>
"""
    return report


def main():
    """主函数"""
    print("=" * 60)
    print("🚀 Plane 项目管理集成")
    print("=" * 60)

    # 检查 API Key
    if not PLANE_API_KEY:
        print("⚠️ 未设置 PLANE_API_KEY，跳过同步")
        print("💡 请在 Gitea 仓库设置中添加 PLANE_API_KEY secret")
        sys.exit(0)

    # 初始化 Plane 客户端
    client = PlaneClient(PLANE_API_URL, PLANE_API_KEY, PLANE_WORKSPACE)

    # 检查项目是否存在
    print(f"\n📋 检查项目: {PLANE_PROJECT_ID}")
    project = client.get_project(PLANE_PROJECT_ID)

    if not project:
        print(f"❌ 项目 {PLANE_PROJECT_ID} 不存在，请先在 Plane 中创建")
        print(f"🔗 访问: http://192.168.1.83:8088/{PLANE_WORKSPACE}")
        sys.exit(1)

    print(f"✅ 找到项目: {project.get('name', PLANE_PROJECT_ID)}")

    # 生成构建报告
    build_report = generate_build_report()

    # 根据事件类型处理
    if GITHUB_EVENT_NAME == "push":
        # Push 事件：创建构建记录 Issue
        title = f"🏗️ 构建 {GITHUB_REF.split('/')[-1]} - {GITHUB_SHA[:7]}"

        print(f"\n📝 创建构建记录 Issue: {title}")
        issue = client.create_issue(
            project_id=project.get("id"),
            title=title,
            description=build_report,
            priority="low",
        )

        if issue:
            print(f"✅ Issue 创建成功: {issue.get('name')}")
            print(f"🔗 查看: http://192.168.1.83:8088/{PLANE_WORKSPACE}/projects/{PLANE_PROJECT_ID}/issues/{issue.get('id')}")
        else:
            print("❌ Issue 创建失败")

    elif GITHUB_EVENT_NAME == "pull_request":
        # PR 事件：创建 PR 跟踪 Issue
        title = f"🔀 Pull Request - {GITHUB_SHA[:7]}"

        print(f"\n📝 创建 PR 跟踪 Issue: {title}")
        issue = client.create_issue(
            project_id=project.get("id"),
            title=title,
            description=build_report,
            priority="medium",
        )

        if issue:
            print(f"✅ Issue 创建成功")
        else:
            print("❌ Issue 创建失败")

    elif GITHUB_EVENT_NAME == "workflow_run":
        # Workflow 完成事件
        print(f"\n📊 工作流完成: {WORKFLOW_NAME}")
        print(f"📌 状态: {WORKFLOW_STATUS}")

    print("\n" + "=" * 60)
    print("✅ Plane 同步完成")
    print("=" * 60)


if __name__ == "__main__":
    main()