reposync/test_gitlink_sync_debug.py

#!/usr/bin/env python3
"""
调试GitLink同步失败问题
"""

import asyncio
import json
import requests
from loguru import logger

# 模拟Gitee webhook数据
mock_gitee_webhook_data = {
    "source": "Gitee",
    "issue": {
        "title": "newissue8",
        "body": "这是一个测试Issue",
        "number": "ICMMM6",
        "state": "open"
    }
}

async def test_gitlink_sync_function():
    """测试GitLink同步函数"""
    
    # GitLink配置
    token = "your_gitlink_token"  # 需要替换为实际的token
    owner = "yuyebo22"
    repo = "fix"
    
    # 从其他平台的webhook数据中提取issue信息
    issue = mock_gitee_webhook_data.get("issue", {})
    
    # 根据来源平台正确映射字段
    if mock_gitee_webhook_data.get("source") == "Gitee":
        title = issue.get("title", "")  # Gitee用title
        body = issue.get("body", "")  # Gitee用body
    elif mock_gitee_webhook_data.get("source") == "GitHub":
        title = issue.get("title", "")  # GitHub用title
        body = issue.get("body", "")  # GitHub用body
    else:
        title = issue.get("subject", "")  # GitLink用subject
        body = issue.get("description", "")  # GitLink用description
    
    logger.info(f"📝 提取的标题: {title}")
    logger.info(f"📝 提取的内容: {body}")
    
    # 构造GitLink API请求
    url = f"https://www.gitlink.org.cn/api/v1/{owner}/{repo}/issues.json"
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json'
    }
    payload = {
        "subject": title,  # GitLink使用subject字段
        "description": body,
        "status_id": 1,  # 新建状态
        "priority_id": 2,  # 普通优先级
    }
    
    logger.info(f"📤 向GitLink发送请求: POST {url}")
    logger.info(f"📤 请求头: {headers}")
    logger.info(f"📤 请求体: {json.dumps(payload, ensure_ascii=False, indent=2)}")
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=30)
        logger.info(f"📥 响应状态码: {response.status_code}")
        logger.info(f"📥 响应头: {dict(response.headers)}")
        logger.info(f"📥 响应内容: {response.text}")
        
        if response.status_code == 200:
            result = response.json()
            logger.info(f"✅ 成功解析响应: {json.dumps(result, ensure_ascii=False, indent=2)}")
            return {
                "success": True,
                "issue_id": result.get("id"),
                "url": result.get("html_url")
            }
        else:
            logger.error(f"❌ API请求失败，状态码: {response.status_code}")
            return {
                "success": False,
                "error": f"API请求失败，状态码: {response.status_code}",
                "message": response.text
            }
            
    except requests.exceptions.RequestException as e:
        logger.error(f"❌ GitLink请求异常: {str(e)}")
        logger.error(f"❌ 异常类型: {type(e)}")
        if hasattr(e, 'response') and e.response is not None:
            logger.error(f"❌ 响应状态码: {e.response.status_code}")
            logger.error(f"❌ 响应内容: {e.response.text}")
        return {
            "success": False,
            "error": f"GitLink API请求失败: {str(e)}",
            "message": str(e)
        }
    except json.JSONDecodeError as e:
        logger.error(f"❌ GitLink JSON解析异常: {str(e)}")
        return {
            "success": False,
            "error": f"GitLink响应解析失败: {str(e)}",
            "message": str(e)
        }
    except Exception as e:
        logger.error(f"❌ GitLink同步失败: {str(e)}")
        logger.error(f"❌ 异常类型: {type(e)}")
        return {
            "success": False,
            "error": f"GitLink同步失败: {str(e)}",
            "message": str(e)
        }

async def test_webhook_exception_handling():
    """测试webhook中的异常处理"""
    
    logger.info("🔍 测试webhook中的异常处理...")
    
    try:
        # 模拟调用GitLink同步函数
        result = await test_gitlink_sync_function()
        
        if result.get('success'):
            logger.info(f"✅ GitLink同步成功: {result.get('issue_id')}")
            return {
                'status': 'success',
                'issue_id': result.get('issue_id'),
                'message': result.get('message', '')
            }
        else:
            logger.error(f"❌ GitLink同步失败: {result.get('error')}")
            return {
                'status': 'error',
                'error': result.get('error'),
                'message': result.get('message', '')
            }
            
    except Exception as e:
        logger.error(f"❌ GitLink同步异常: {str(e)}")
        return {
            'status': 'error',
            'error': str(e),
            'message': f'Exception during sync: {str(e)}'
        }

async def main():
    """主函数"""
    logger.info("🚀 开始调试GitLink同步问题...")
    
    # 测试GitLink同步函数
    result = await test_webhook_exception_handling()
    
    logger.info("📊 测试结果:")
    logger.info(f"  状态: {result.get('status')}")
    if result.get('error'):
        logger.info(f"  错误: {result.get('error')}")
    if result.get('message'):
        logger.info(f"  消息: {result.get('message')}")
    if result.get('issue_id'):
        logger.info(f"  Issue ID: {result.get('issue_id')}")

if __name__ == "__main__":
    asyncio.run(main())