reposync/issue_sync_module/clients/gitlink_parser.py

177 lines
5.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
GitLink Issue 数据解析器
基于官方API文档https://apifox.com/apidoc/shared/da30afb0-9d2e-429b-a4bc-a83209e06021/api-76027798
"""
from typing import Dict, List, Optional
from datetime import datetime
class GitLinkIssueParser:
"""GitLink Issue数据解析器"""
@staticmethod
def parse_issue(gitlink_issue: Dict) -> Dict:
"""
解析GitLink Issue数据为标准格式
GitLink API返回字段
- id: integer
- subject: string (标题)
- project_issues_index: integer (序号)
- description: string (描述)
- status: object (Issue状态)
- priority: object (Issue优先级)
- author: object (发布者)
- assigners: array (负责人)
- tags: array (标记)
- created_at, updated_at: string
"""
# 提取基本信息
parsed = {
'id': gitlink_issue.get('id'),
'title': gitlink_issue.get('subject', ''),
'body': gitlink_issue.get('description', ''),
'number': gitlink_issue.get('project_issues_index'),
'state': GitLinkIssueParser._parse_state(gitlink_issue.get('status', {})),
'created_at': gitlink_issue.get('created_at', ''),
'updated_at': gitlink_issue.get('updated_at', '')
}
# 解析作者信息
author = gitlink_issue.get('author', {})
parsed['author'] = {
'id': author.get('id'),
'login': author.get('login', ''),
'name': author.get('name', ''),
'avatar_url': author.get('image_url', '')
}
# 解析优先级
priority = gitlink_issue.get('priority', {})
parsed['priority'] = {
'id': priority.get('id'),
'name': priority.get('name', '')
}
# 解析指派人
assigners = gitlink_issue.get('assigners', [])
parsed['assignees'] = [
{
'id': assigner.get('id'),
'login': assigner.get('login', ''),
'name': assigner.get('name', ''),
'avatar_url': assigner.get('image_url', '')
}
for assigner in assigners
]
# 解析标签
tags = gitlink_issue.get('tags', [])
parsed['labels'] = [
{
'id': tag.get('id'),
'name': tag.get('name', ''),
'color': tag.get('color', '')
}
for tag in tags
]
# 解析里程碑
milestone = gitlink_issue.get('milestone', {})
if milestone and milestone.get('id'):
parsed['milestone'] = {
'id': milestone.get('id'),
'title': milestone.get('name', '')
}
else:
parsed['milestone'] = None
# 附加信息
parsed['comment_count'] = gitlink_issue.get('comment_journals_count', 0)
parsed['attachments'] = gitlink_issue.get('attachments', [])
return parsed
@staticmethod
def _parse_state(status: Dict) -> str:
"""解析GitLink状态为标准状态"""
status_name = status.get('name', '').lower()
# GitLink状态映射
status_mapping = {
'新建': 'open',
'已确认': 'open',
'已解决': 'closed',
'关闭': 'closed',
'重新打开': 'open',
'进行中': 'open',
'已分配': 'open'
}
return status_mapping.get(status.get('name', ''), 'open')
@staticmethod
def to_github_format(gitlink_issue: Dict) -> Dict:
"""转换GitLink Issue为GitHub Issue格式"""
parsed = GitLinkIssueParser.parse_issue(gitlink_issue)
return {
'number': parsed['number'],
'title': parsed['title'],
'body': parsed['body'],
'state': parsed['state'],
'user': parsed['author'],
'assignees': parsed['assignees'],
'labels': [{'name': label['name']} for label in parsed['labels']],
'created_at': parsed['created_at'],
'updated_at': parsed['updated_at']
}
@staticmethod
def to_gitee_format(gitlink_issue: Dict) -> Dict:
"""转换GitLink Issue为Gitee Issue格式"""
parsed = GitLinkIssueParser.parse_issue(gitlink_issue)
return {
'number': parsed['number'],
'title': parsed['title'],
'body': parsed['body'],
'state': parsed['state'],
'user': parsed['author'],
'assignee': parsed['assignees'][0] if parsed['assignees'] else None,
'labels': [{'name': label['name']} for label in parsed['labels']],
'priority': parsed['priority']['name'] if parsed['priority'] else None,
'created_at': parsed['created_at'],
'updated_at': parsed['updated_at']
}
@staticmethod
def get_priority_id(priority_name: str) -> int:
"""根据优先级名称获取ID基于常见的GitLink优先级"""
priority_mapping = {
'': 1,
'正常': 2,
'': 3,
'紧急': 4,
'立即': 5
}
return priority_mapping.get(priority_name, 2)
@staticmethod
def get_status_id(status_name: str) -> int:
"""根据状态名称获取ID"""
status_mapping = {
'新建': 1,
'已确认': 2,
'已分配': 3,
'进行中': 4,
'已解决': 5,
'关闭': 6,
'重新打开': 7
}
return status_mapping.get(status_name, 1)