API 触发分析

from time import time
from hashlib import sha256

def get_headers(user_id, token):
    timestamp = int(time())
    token_sig = "%s%s#%s#%s%s" % (timestamp, user_id, token, user_id, timestamp)
    ticket = sha256(token_sig.encode("utf-8")).hexdigest().upper()
    return {
        "TCA-USERID": user_id,
        "TCA-TIMESTAMP": str(timestamp),
        "TCA-TICKET": ticket
    }

import requests

url = f"{base_url}/server/main/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/scans/create/"
headers = get_headers(user_id, token)

resp = requests.post(url, json=post_data, headers=headers)
result = resp.json()
print(result)

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/scaninfos/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
results = result.get("data", {}).get("results", [])
print(result[0])

lint = result[0].get("lintscan", {})
total = lint.get("total", {})
print({
    "active": total.get("state_detail", {}).get("active"),                               # 未处理问题数
    "fatal_active": total.get("severity_detail", {}).get("fatal", {}).get("active"),    # 严重问题数
    "error_active": total.get("severity_detail", {}).get("error", {}).get("active"),    # 错误问题数
    "warning_active": total.get("severity_detail", {}).get("warning", {}).get("active"), # 警告问题数
    "info_active": total.get("severity_detail", {}).get("info", {}).get("active"),      # 提示问题数
    "issue_detail_num": lint.get("issue_detail_num"),   # 问题总数（未聚合）
    "issue_open_num": lint.get("issue_open_num"),       # 新增缺陷数
    "issue_fix_num": lint.get("issue_fix_num"),         # 修复缺陷数
})

cc = result[0].get("cyclomaticcomplexityscan", {})
custom_summary = cc.get("custom_summary", {})
print({
    "over_cc_func_count": custom_summary.get("over_cc_func_count"),     # 该分支所有超标方法个数
    "over_cc_func_average": custom_summary.get("over_cc_func_average"), # 该分支所有超标方法平均复杂度
    "min_ccn": cc.get("min_ccn"),                                       # 圈复杂度标准（超标阈值）
})

dup = result[0].get("duplicatescan", {})
print({
    "duplicate_block_count": dup.get("duplicate_block_count"),             # 重复块数
    "total_duplicate_line_count": dup.get("total_duplicate_line_count"),   # 重复行数
    "duplicate_file_count": dup.get("duplicate_file_count"),               # 重复文件数（新增字段）
})

cloc = result[0].get("clocscan", {})
print({
    "code_line_num": cloc.get("code_line_num"),         # 代码行数
    "comment_line_num": cloc.get("comment_line_num"),   # 注释行数
    "blank_line_num": cloc.get("blank_line_num"),       # 空白行数
    "total_line_num": cloc.get("total_line_num"),       # 总行数
})

quality = result[0].get("qualityscan", {})
status = quality.get("status")  # "success": 门禁通过，"failure": 门禁未通过, "close": 门禁未开启
description = quality.get("description")

if status == "success":
    print("质量门禁通过")
elif status == "close":
    print("质量门禁未开启")
else:
    print(f"质量门禁未通过：{description}")

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
results = result.get("data", {})
print(results)

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/latestscan/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
results = result.get("data", {})
quality = results.get("qualityscan", {})
status = quality.get("status")  # "success": 门禁通过，"failure": 门禁未通过, "close": 门禁未开启
description = quality.get("description")

if status == "success":
    print("质量门禁通过")
elif status == "close":
    print("质量门禁未开启")
else:
    print(f"质量门禁未通过：{description}")

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/lintscans/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
lint_data = result.get("data", {}).get("results", [{}])[0]
total = lint_data.get("total", {})
print({
    "active": total.get("state_detail", {}).get("active"),                               # 未处理问题数
    "fatal_active": total.get("severity_detail", {}).get("fatal", {}).get("active"),    # 严重问题数
    "error_active": total.get("severity_detail", {}).get("error", {}).get("active"),    # 错误问题数
    "warning_active": total.get("severity_detail", {}).get("warning", {}).get("active"), # 警告问题数
    "info_active": total.get("severity_detail", {}).get("info", {}).get("active"),      # 提示问题数
    "issue_detail_num": lint_data.get("issue_detail_num"),  # 问题总数（未聚合）
    "issue_open_num": lint_data.get("issue_open_num"),      # 新增缺陷数
    "issue_fix_num": lint_data.get("issue_fix_num"),        # 修复缺陷数
})

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/cycscans/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
cyc_data = result.get("data", {}).get("results", [{}])[0]
custom_summary = cyc_data.get("custom_summary", {})
print({
    "over_cc_func_count": custom_summary.get("over_cc_func_count"),     # 该分支所有超标方法个数
    "over_cc_func_average": custom_summary.get("over_cc_func_average"), # 该分支所有超标方法平均复杂度
    "min_ccn": custom_summary.get("min_ccn") or cyc_data.get("min_ccn"), # 圈复杂度标准（超标阈值）
    "scan_revision": cyc_data.get("scan_revision"),                      # 对应的代码版本（commit）
})

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/dupscans/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
dup_data = result.get("data", {}).get("results", [{}])[0]
print({
    "duplicate_block_count": dup_data.get("duplicate_block_count"),             # 重复块数
    "total_duplicate_line_count": dup_data.get("total_duplicate_line_count"),   # 重复行数
    "duplicate_file_count": dup_data.get("duplicate_file_count"),               # 重复文件数
    "scan_revision": dup_data.get("scan_revision"),                             # 对应的代码版本（commit）
})

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/clocscans/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
cloc_data = result.get("data", {}).get("results", [{}])[0]
print({
    "code_line_num": cloc_data.get("code_line_num"),        # 代码行数
    "comment_line_num": cloc_data.get("comment_line_num"),  # 注释行数
    "blank_line_num": cloc_data.get("blank_line_num"),      # 空白行数
    "total_line_num": cloc_data.get("total_line_num"),      # 总行数
    "scan_revision": cloc_data.get("scan_revision"),        # 对应的代码版本（commit）
})

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/sca/scans/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
sca_data = result.get("data", {}).get("results", [{}])[0]
meta_data = sca_data.get("meta_data") or {}

def risk_count(data_list, risk_key):
    """从 [{risk: ..., count: ...}, ...] 列表中提取指定 risk 的 count"""
    for item in (data_list or []):
        if isinstance(item, dict) and item.get("risk") == risk_key:
            return item.get("count", 0)
    return 0

malware_data = sca_data.get("malware_data")
copyright_data = sca_data.get("copyright_tampering_data")
license_tampering_data = sca_data.get("license_tampering_data")
base_audit_data = sca_data.get("base_audit_data")
cve_data = sca_data.get("cve_data")
license_data = sca_data.get("license_data")
source_reuse_rate = meta_data.get("source_reuse_rate") or 0

print({
    "file_num": sca_data.get("file_num"),                                       # 文件总数
    "self_rate": round((1 - source_reuse_rate) * 100, 2),                       # 自研率（%）
    "source_reuse_rate": round(source_reuse_rate * 100, 2),                     # 开源代码占比（%）
    # 病毒检测
    "malware_count": risk_count(malware_data, "Malware"),                       # 病毒数
    "malware_pass_count": risk_count(malware_data, "Pass"),                     # 非病毒数
    # Copyright 检测（copyright_tampering_data）
    "copyright_has": risk_count(copyright_data, "Has"),                         # 声明 Copyright
    "copyright_without": risk_count(copyright_data, "Without"),                 # 未声明 Copyright
    "copyright_modified": risk_count(copyright_data, "Modified"),               # Copyright 篡改
    # License 篡改检测（license_tampering_data）
    "license_has": risk_count(license_tampering_data, "Has"),                   # 声明 License
    "license_without": risk_count(license_tampering_data, "Without"),           # 未声明 License
    "license_modified": risk_count(license_tampering_data, "Modified"),         # License 篡改
    # 安全审计（base_audit_data）
    "audit_high": risk_count(base_audit_data, "High"),                          # 安全审计-高危
    "audit_medium": risk_count(base_audit_data, "Medium"),                      # 安全审计-中危
    "audit_warning": risk_count(base_audit_data, "Warning"),                    # 安全审计-警告
    "audit_pass": risk_count(base_audit_data, "Pass"),                          # 安全审计-通过
    "audit_na": risk_count(base_audit_data, "N/A"),                             # 安全审计-N/A
    # 漏洞审计（cve_data）
    "cve_critical": risk_count(cve_data, "Critical"),                           # 漏洞审计-严重
    "cve_high": risk_count(cve_data, "High"),                                   # 漏洞审计-高危
    "cve_medium": risk_count(cve_data, "Medium"),                               # 漏洞审计-中危
    "cve_low": risk_count(cve_data, "Low"),                                     # 漏洞审计-低危
    "cve_na": risk_count(cve_data, "NotAvailable"),                             # 漏洞审计-N/A
    # License 审计（license_data）
    "license_high": risk_count(license_data, "High"),                           # License 审计-高风险
    "license_middle": risk_count(license_data, "Middle"),                       # License 审计-中风险
    "license_low": risk_count(license_data, "Low"),                             # License 审计-低风险
    "license_na": risk_count(license_data, "NotAvailable"),                     # License 审计-N/A
})

import requests
import json

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/codelint/issues/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
print(json.dumps(result, indent=2, ensure_ascii=False))

import requests

url = f"{base_url}/server/main/api/orgs/{org_sid}/teams/{team_name}/artifact/repos/{repo_id}/projects/{project_id}/scans/"
headers = get_headers(user_id, token)
post_data = {
    "artifact_url": "artifact_url",  # 制品文件的下载地址，需自行填写，例如：https://example.com/path/to/artifact.zip
    "created_from": "api"
}
resp = requests.post(url, json=post_data, headers=headers)
result = resp.json()
print(result)

import requests

url = f"{base_url}/server/main/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/jobs/{job_id}/detail/"
headers = get_headers(user_id, token)

resp = requests.get(url, headers=headers)
result = resp.json()
print(result)

import requests

url = f"{base_url}/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/sca/scans/"
headers = get_headers(user_id, token)
resp = requests.get(url, headers=headers)
result = resp.json()
print(result)

sca_data = result.get("data", {}).get("results", [{}])[0]
meta_data = sca_data.get("meta_data") or {}

def risk_count(data_list, risk_key):
    """从 [{risk: ..., count: ...}, ...] 列表中提取指定 risk 的 count"""
    for item in (data_list or []):
        if isinstance(item, dict) and item.get("risk") == risk_key:
            return item.get("count", 0)
    return 0

malware_data = sca_data.get("malware_data")
base_audit_data = sca_data.get("base_audit_data")
cve_data = sca_data.get("cve_data")
license_data = sca_data.get("license_data")
source_reuse_rate = meta_data.get("source_reuse_rate") or 0

print({
    "file_num": sca_data.get("file_num"),                                   # 文件总数
    "source_reuse_rate": round(source_reuse_rate * 100, 2),                 # 源码复用率（%）
    "self_rate": round((1 - source_reuse_rate) * 100, 2),                   # 自研率（%）
    # 病毒检测
    "malware_count": risk_count(malware_data, "Malware"),                   # 病毒数
    "malware_pass_count": risk_count(malware_data, "Pass"),                 # 非病毒数
    # 安全审计（base_audit_data）
    "audit_high": risk_count(base_audit_data, "High"),                      # 安全审计-高危
    "audit_medium": risk_count(base_audit_data, "Medium"),                  # 安全审计-中危
    "audit_warning": risk_count(base_audit_data, "Warning"),                # 安全审计-警告
    "audit_pass": risk_count(base_audit_data, "Pass"),                      # 安全审计-通过
    "audit_na": risk_count(base_audit_data, "N/A"),                         # 安全审计-N/A
    # 漏洞审计（cve_data）
    "cve_critical": risk_count(cve_data, "Critical"),                       # 漏洞审计-严重
    "cve_high": risk_count(cve_data, "High"),                               # 漏洞审计-高危
    "cve_medium": risk_count(cve_data, "Medium"),                           # 漏洞审计-中危
    "cve_low": risk_count(cve_data, "Low"),                                 # 漏洞审计-低危
    "cve_na": risk_count(cve_data, "NotAvailable"),                         # 漏洞审计-N/A
    # License 审计（license_data）
    "license_high": risk_count(license_data, "High"),                       # License 审计-高风险
    "license_middle": risk_count(license_data, "Middle"),                   # License 审计-中风险
    "license_low": risk_count(license_data, "Low"),                         # License 审计-低风险
    "license_na": risk_count(license_data, "NotAvailable"),                 # License 审计-N/A
})

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
通过 API 启动代码库分析
"""

import os
import sys
import json
import requests
import logging
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass
from time import time, sleep
from hashlib import sha256

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler('tca_ci_git_saas.log')
    ]
)
logger = logging.getLogger(__name__)

# 开放API路径映射
TCA_OPEN_APIS = {
    "project_team_list": "%s/server/main/api/orgs/{org_sid}/teams/",
    "pt_repo_list": "%s/server/main/api/orgs/{org_sid}/teams/{team_name}/repos/",
    "project_list": "%s/server/main/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/",
    "project_scan_list": "%s/server/main/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/scans/create/",
    "job_detail": "%s/server/main/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/jobs/{job_id}/detail/",
    "project_analysis_scan_list": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/scaninfos/",
    "project_issue_list": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/codelint/issues/",
    "project_overview": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/",
    "project_overview_latest_scan": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/latestscan/",
    "project_sca_scans": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/sca/scans/",
    "overview_cycscans": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/cycscans/",
    "overview_clocscans": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/clocscans/",
    "overview_dupscans": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/dupscans/",
    "overview_lintscans": "%s/server/analysis/api/orgs/{org_sid}/teams/{team_name}/repos/{repo_id}/projects/{project_id}/overview/lintscans/",
}


@dataclass
class TCAConfig:
    """TCA配置类"""
    base_url: str
    tca_token: str
    user_id: str
    org_sid: str
    team_name: str
    repo_url: str
    repo_branch: str
    repo_mr_branch: str
    scm_account_id: int
    scheme_id: int
    incr_scan: bool
    force_create: bool
    ignore_branch_issue: str
    timeout: int = 30
    max_retries: int = 3
    max_wait_time: int = 600


class TCAIntegrationError(Exception):
    """TCA集成异常"""
    pass


class TCAClient:
    """TCA API客户端"""

    def __init__(self, config: TCAConfig):
        self.config = config
        self.session = requests.Session()
        self._repo_id: Optional[int] = None
        self._project_id: Optional[int] = None

    def get_headers(self, user_id: Optional[str] = None, token: Optional[str] = None):
        timestamp = int(time())
        user_id = user_id or self.config.user_id
        token = token or self.config.tca_token
        token_sig = "%s%s#%s#%s%s" % (timestamp, user_id, token, user_id, timestamp)
        ticket = sha256(token_sig.encode("utf-8")).hexdigest().upper()
        return {
            "TCA-USERID": user_id,
            "TCA-TIMESTAMP": str(timestamp),
            "TCA-TICKET": ticket
        }

    def _is_success_status(self, status_code: int) -> bool:
        """判断HTTP状态码是否表示成功（2xx）"""
        return 200 <= status_code < 300

    def _make_request(self, method: str, url: str, data: Optional[Dict] = None,
                      params: Optional[Dict] = None) -> Dict:
        """发送API请求，url需为完整路径"""

        for attempt in range(self.config.max_retries):
            try:
                headers = self.get_headers()
                response = self.session.request(
                    method=method,
                    url=url,
                    headers=headers,
                    json=data,
                    params=params,
                    timeout=self.config.timeout
                )

                if self._is_success_status(response.status_code):
                    return response.json()
                elif response.status_code == 401:
                    raise TCAIntegrationError("TCA Token无效或已过期")
                elif response.status_code == 403:
                    raise TCAIntegrationError("权限不足，请检查团队ID和权限设置")
                elif response.status_code == 404:
                    raise TCAIntegrationError("请求的资源不存在")
                elif response.status_code >= 500:
                    logger.warning(f"服务器错误，第{attempt + 1}次重试...")
                    if attempt == self.config.max_retries - 1:
                        raise TCAIntegrationError("服务器内部错误，请稍后重试")
                    sleep(2 ** attempt)  # 指数退避
                else:
                    raise TCAIntegrationError(f"API请求失败: {response.status_code} - {response.text}")

            except requests.exceptions.Timeout:
                logger.warning(f"请求超时，第{attempt + 1}次重试...")
                if attempt == self.config.max_retries - 1:
                    raise TCAIntegrationError("请求超时，请检查网络连接")
            except requests.exceptions.ConnectionError:
                logger.warning(f"连接错误，第{attempt + 1}次重试...")
                if attempt == self.config.max_retries - 1:
                    raise TCAIntegrationError("网络连接错误，请检查网络设置")

    def _build_url(self, key: str, **kwargs: Any) -> str:
        base = self.config.base_url.rstrip('/')
        tpl = TCA_OPEN_APIS[key] % base
        return tpl.format(**kwargs)

    def get_project_teams(self) -> Dict:
        url = self._build_url("project_team_list", org_sid=self.config.org_sid)
        return self._make_request("GET", url)

    def get_pt_repos(self) -> Dict:
        url = self._build_url("pt_repo_list", org_sid=self.config.org_sid, team_name=self.config.team_name)
        return self._make_request("GET", url)

    def get_pt_repo_id(self, repo_url: Optional[str] = None) -> int:
        """根据仓库地址获取团队下对应仓库的ID（仅精确匹配）。
        精确比较 `scm_url`，找不到则抛出 TCAIntegrationError。
        首次获取后缓存，后续调用直接返回缓存值。
        """
        if self._repo_id is not None:
            return self._repo_id
        result = self.get_pt_repos()
        data = result.get("data") or {}
        results = data.get("results") or []
        repo_url = repo_url or self.config.repo_url
        if not isinstance(results, list):
            raise TCAIntegrationError("获取团队仓库列表失败，返回格式不正确")
        for item in results:
            if not isinstance(item, dict):
                continue
            if repo_url == item.get("scm_url") or repo_url == item.get("format_url"):
                repo_id_val = item.get("id")
                if repo_id_val is not None:
                    self._repo_id = int(repo_id_val)
                    return self._repo_id
        raise TCAIntegrationError(f"未在项目 {self.config.team_name} 下找到仓库: {repo_url}")

    def create_project_team(self) -> Dict:
        """创建项目"""
        logger.info(f"创建项目: {self.config.team_name}")
        url = self._build_url("project_team_list", org_sid=self.config.org_sid)
        post_data = {
            'name': self.config.team_name,
            'display_name': self.config.team_name
        }
        result = self._make_request("POST", url, data=post_data)
        return result

    def register_pt_repo(self) -> Dict:
        """登记代码库（项目维度）"""
        logger.info(f"项目: {self.config.team_name}, 创建代码库: {self.config.repo_url}")
        url = self._build_url("pt_repo_list", org_sid=self.config.org_sid, team_name=self.config.team_name)
        post_data = {
            'scm_url': self.config.repo_url,
            'scm_type': "git",
            'scm_auth': {
                "auth_type": "password",
                "scm_account": self.config.scm_account_id
            },
            'created_from': "api"
        }
        result = self._make_request("POST", url, data=post_data)
        return result

    def create_project(self) -> Dict:
        """创建分析项目"""
        logger.info(f"项目: {self.config.team_name}, 创建代码库: {self.config.repo_url}的分析项目")
        repo_id = self.get_pt_repo_id(self.config.repo_url)
        url = self._build_url("project_list", org_sid=self.config.org_sid, team_name=self.config.team_name,
                              repo_id=repo_id)
        post_data = {
            'branch': self.config.repo_branch,
            'global_scheme_id': self.config.scheme_id,
            'use_scheme_template': True,
            'created_from': "api"
        }
        result = self._make_request("POST", url, data=post_data)

        return result

    def get_projects(self) -> Dict:
        repo_id = self.get_pt_repo_id(self.config.repo_url)
        url = self._build_url("project_list", org_sid=self.config.org_sid, team_name=self.config.team_name,
                              repo_id=repo_id)
        return self._make_request("GET", url)

    def get_project_id(self, scheme_id: Optional[int] = None, branch: Optional[str] = None) -> int:
        """根据 scan_scheme.id 和 branch 获取分析项目ID。
        - 从项目列表中精确匹配 `scan_scheme.id` 与 `branch`
        首次获取后缓存，后续调用直接返回缓存值。
        """
        if self._project_id is not None:
            return self._project_id
        scheme_id = scheme_id or self.config.scheme_id
        branch = branch or self.config.repo_branch
        logger.info(f"scheme_id={scheme_id}, branch={branch}")
        result = self.get_projects()
        data = result.get("data") or {}
        results = data.get("results") or []
        if not isinstance(results, list):
            raise TCAIntegrationError("获取项目列表失败，返回格式不正确")
        for item in results:
            if not isinstance(item, dict):
                continue
            item_scheme = (item.get("scan_scheme") or {}).get("id")
            item_branch = item.get("branch")
            if int(item_scheme) == int(scheme_id) and item_branch == branch:
                proj_id = item.get("id")
                if proj_id is not None:
                    self._project_id = int(proj_id)
                    return self._project_id

        raise TCAIntegrationError(f"未找到匹配的分析项目，scheme_id={scheme_id}, branch={branch}")

    def start_scan(self) -> Dict:
        """启动扫描任务"""
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        url = self._build_url("project_scan_list", org_sid=self.config.org_sid,
                              team_name=self.config.team_name, repo_id=repo_id, project_id=project_id)
        incr_scan = self.config.incr_scan
        force_create = self.config.force_create
        ignore_branch_issue = self.config.ignore_branch_issue

        post_data = {
            'incr_scan': incr_scan,
            'force_create': force_create,
            'ignore_branch_issue': ignore_branch_issue
        }
        result = self._make_request("POST", url, data=post_data)
        logger.info(f"启动扫描任务结果: {result}")
        return result

    def wait_for_scan_completion(self, job_id: int) -> Dict:
        """等待分析任务状态变为 2（已完成），并处理重定向 job_id"""
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        url = self._build_url("job_detail", org_sid=self.config.org_sid,
                              team_name=self.config.team_name, repo_id=repo_id, project_id=project_id, job_id=job_id)
        max_wait_time = self.config.max_wait_time
        interval = 30  # 每隔 30秒请求一次
        start_time = int(time())
        while True:
            results = self._make_request("GET", url)
            data = results.get("data", {}) or {}
            state = data.get("state")
            result_code = data.get("result_code")

            if state == 2:
                if result_code == 0 or result_code == 1:
                    logging.info("扫描结果已出，结束等待")
                    break
                if result_code == 2:
                    result_msg_raw = data.get("result_msg") or {}
                    if isinstance(result_msg_raw, str):
                        try:
                            result_msg = json.loads(result_msg_raw)
                        except Exception as e:
                            raise TCAIntegrationError(f"扫描结果重定向但 result_msg 无法解析: {result_msg_raw}")
                    else:
                        result_msg = result_msg_raw
                    new_job_id = result_msg.get("job_id") if isinstance(result_msg, dict) else None
                    if not new_job_id:
                        raise TCAIntegrationError("扫描结果重定向但未返回新的 job_id")
                    job_id = int(new_job_id)
                    url = self._build_url("job_detail", org_sid=self.config.org_sid,
                                          team_name=self.config.team_name, repo_id=repo_id, project_id=project_id, job_id=job_id)
                    logging.info(f"扫描结果重定向，使用新的 job_id={job_id} 继续轮询")
                else:
                    raise TCAIntegrationError(f"扫描失败: result_code={result_code}, result_msg={data.get('result_msg')}")

            elapsed = int(time()) - start_time
            if elapsed > max_wait_time:
                logger.info(f"扫描超过{max_wait_time}秒，扫描状态仍未改变，扫描失败")
                raise TCAIntegrationError(f"扫描超过{max_wait_time}秒，扫描状态仍未改变，扫描失败")

            logging.info(
                f"代码正在扫描中（当前 state: {state if state is not None else '无数据'}），等待{interval}秒后重试...")
            sleep(interval)

        return results

    def get_scan_info(self, job_id: int) -> Dict:
        """获取指定 job_id 的分析结果信息"""
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        url = self._build_url("project_analysis_scan_list", org_sid=self.config.org_sid,
                              team_name=self.config.team_name, repo_id=repo_id, project_id=project_id)
        response = self._make_request("GET", url, params={"id": job_id})
        results = response.get("data", {}).get("results", [])
        if not results:
            raise TCAIntegrationError(f"未找到 job_id={job_id} 的扫描结果")
        return results[0]

    def get_project_overview(self) -> Dict:
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        url = self._build_url("project_overview", org_sid=self.config.org_sid,
                              team_name=self.config.team_name, repo_id=repo_id, project_id=project_id)
        result = self._make_request("GET", url)
        return result.get("data", {}) or {}

    def get_project_overview_latest_scan(self) -> Dict:
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        url = self._build_url("project_overview_latest_scan", org_sid=self.config.org_sid,
                              team_name=self.config.team_name, repo_id=repo_id, project_id=project_id)
        result = self._make_request("GET", url)
        return result.get("data", {}) or {}

    def get_issue_list(self) -> Dict:
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        url = self._build_url("project_issue_list", org_sid=self.config.org_sid,
                              team_name=self.config.team_name, repo_id=repo_id, project_id=project_id)
        result = self._make_request("GET", url)
        return result

    def _get_latest_scan_data(self, api_key: str) -> Optional[Dict]:
        """通用方法：获取指定扫描列表接口的最新一条数据（results[0]），无数据返回 None。
        支持的 api_key: project_sca_scans, overview_cycscans, overview_clocscans, overview_dupscans, overview_lintscans
        """
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        url = self._build_url(api_key, org_sid=self.config.org_sid,
                              team_name=self.config.team_name, repo_id=repo_id, project_id=project_id)
        response = self._make_request("GET", url)
        results = response.get("data", {}).get("results", [])
        if not results:
            return None
        return results[0]

    def get_sca_scan_info(self) -> Optional[Dict]:
        return self._get_latest_scan_data("project_sca_scans")

    def get_cyc_scan_info(self) -> Optional[Dict]:
        return self._get_latest_scan_data("overview_cycscans")

    def get_cloc_scan_info(self) -> Optional[Dict]:
        return self._get_latest_scan_data("overview_clocscans")

    def get_dup_scan_info(self) -> Optional[Dict]:
        return self._get_latest_scan_data("overview_dupscans")

    def get_lint_scan_info(self) -> Optional[Dict]:
        return self._get_latest_scan_data("overview_lintscans")

    # ---- 页面链接拼接 ----

    def _build_page_url(self, suffix: str) -> str:
        """通用页面链接拼接：{base_url}/t/{org_sid}/p/{team_name}/repos/{repo_id}/projects/{project_id}/{suffix}"""
        repo_id = self.get_pt_repo_id()
        project_id = self.get_project_id()
        base_url = self.config.base_url.rstrip('/')
        return (f"{base_url}/t/{self.config.org_sid}/p/{self.config.team_name}"
                f"/repos/{repo_id}/projects/{project_id}/{suffix}")

    def get_job_dtail_url(self, job_id: int) -> str:
        return self._build_page_url(f"jobs/{job_id}/progress")

    def get_job_result_url(self, job_id: int) -> str:
        return self._build_page_url(f"jobs/{job_id}/result")

    def get_scan_overview_url(self) -> str:
        return self._build_page_url("overview")

    def get_scan_lint_url(self) -> str:
        return self._build_page_url("lints?state=1")

    def get_scan_sca_url(self, sca_scan_id: int) -> str:
        return self._build_page_url(f"sca/{sca_scan_id}/overview")

    def get_cycs_url(self) -> str:
        return self._build_page_url("cycs/funcs")

    def get_scan_dupfile_url(self) -> str:
        return self._build_page_url("dupfiles?issue_state=1")

    def get_scan_cloc_url(self) -> str:
        return self._build_page_url("clocs")


class TCAPipeline:
    """TCA流水线集成类"""

    def __init__(self, config: TCAConfig):
        self.client = TCAClient(config)
        self.config = config

    def run_full_analysis(self) -> bool:
        """运行完整分析流程"""

        # 1. 创建项目（或复用已有 project_team）
        exist_project_teams = self.client.get_project_teams()
        if not self.has_team_name(exist_project_teams, self.config.team_name):
            project_team = self.client.create_project_team()
        # 2. 登记代码库（或复用已有 repo）
        try:
            self.client.get_pt_repo_id(self.config.repo_url)
        except TCAIntegrationError as e:
            logger.info(f"未找到已有代码库: {self.config.repo_url}, 创建新代码库")
            self.client.register_pt_repo()

        # 3. 在该 repo 下创建项目（或复用已有 project)
        try:
            self.client.get_project_id()
        except TCAIntegrationError as e:
            logger.info(f"未找到已有项目: {self.config.repo_url}, 创建新项目")
            self.client.create_project()

        # 4. 启动扫描
        start_scan_result = self.client.start_scan()

        job_id = self.get_job_id(start_scan_result)

        # 5. 等待完成
        try:
            scan_result = self.client.wait_for_scan_completion(job_id)
        except TCAIntegrationError as e:
            logger.error(f"扫描失败: {e}")
            logger.error(f"扫描失败详情链接：{self.client.get_job_dtail_url(job_id)}")
            return False
        # 5. 如果代码无变更，触发增量扫描，会显示result_code为1，显示无需扫描
        result_code, result_code_msg, result_msg = self.get_result_msg(scan_result)
        if result_code == 1:
            logger.info(
                f"代码无变更，result_code: {result_code}, result_code_msg: {result_code_msg}, result_msg: {result_msg} 代码扫描结果为上一次扫描信息")
            scan_info = self.client.get_project_overview()
            quality_scan_source = self.client.get_project_overview_latest_scan()
        elif result_code == 0:
            logger.info(
                f"全量扫描或代码有变更，result_code: {result_code}, result_code_msg: {result_code_msg}, result_msg: {result_msg} 代码扫描结果为最新扫描信息")
            scan_info = self.client.get_scan_info(job_id)
            quality_scan_source = scan_info
        elif result_code >= 100:
            logger.error(f"扫描失败: {result_code}, {result_code_msg}, {result_msg}")
            logger.error(f"扫描失败详情链接：{self.client.get_job_dtail_url(job_id)}")
            return False

        # 5. 获取各维度扫描结果（overview 优先，为空时 fallback 到独立接口）
        scan_collectors = [
            {
                "name": "代码扫描",
                "overview_key": "lintscan",
                "overview_empty_check": lambda info: not (info.get("lintscan") or {}).get("total"),
                "overview_parser": self.get_lint_scan_result,
                "fallback_fetcher": self.client.get_lint_scan_info,
                "fallback_parser": self.get_lint_scan_detail_result,
                "scan_id_extractor": lambda d: (d.get("scan") or {}).get("id"),
                "revision_extractor": lambda d: (d.get("scan") or {}).get("current_revision", ""),
            },
            {
                "name": "圈复杂度",
                "overview_key": "cyclomaticcomplexityscan",
                "overview_empty_check": lambda info: not ((info.get("cyclomaticcomplexityscan") or {}).get("custom_summary")),
                "overview_parser": self.get_cyclomatic_complexity_result,
                "fallback_fetcher": self.client.get_cyc_scan_info,
                "fallback_parser": self.get_cyc_scan_result,
                "scan_id_extractor": lambda d: d.get("scan") or d.get("id"),
                "revision_extractor": lambda d: d.get("scan_revision", ""),
            },
            {
                "name": "重复代码",
                "overview_key": "duplicatescan",
                "overview_empty_check": lambda info: self._is_all_none(info.get("duplicatescan") or {},
                                                                       ["duplicate_block_count", "total_duplicate_line_count", "duplicate_file_count"]),
                "overview_parser": self.get_duplicate_code_result,
                "fallback_fetcher": self.client.get_dup_scan_info,
                "fallback_parser": self.get_dup_scan_result,
                "scan_id_extractor": lambda d: d.get("scan") or d.get("id"),
                "revision_extractor": lambda d: d.get("scan_revision", ""),
            },
            {
                "name": "代码统计",
                "overview_key": "clocscan",
                "overview_empty_check": lambda info: self._is_all_none(info.get("clocscan") or {},
                                                                       ["code_line_num", "comment_line_num", "blank_line_num", "total_line_num"]),
                "overview_parser": self.get_cloc_result,
                "fallback_fetcher": self.client.get_cloc_scan_info,
                "fallback_parser": self.get_cloc_scan_result,
                "scan_id_extractor": lambda d: d.get("scan") or d.get("id"),
                "revision_extractor": lambda d: d.get("scan_revision", ""),
            },
        ]

        for collector in scan_collectors:
            self._collect_scan_result(collector, scan_info, job_id)

        # 9. 获取 SCA 组成成分分析结果
        sca_data = self.client.get_sca_scan_info()
        sca_scan_id = None
        if sca_data:
            sca_scan_id = sca_data.get("id")
            sca_scan_result = self.get_sca_scan_result(sca_data)
            if sca_scan_id and int(sca_scan_id) != int(job_id):
                # SCA 结果非本次扫描产生，显示对应的 commit_id
                current_revision = (sca_data.get("scan") or {}).get("current_revision", "")
                short_commit = current_revision[:8] if current_revision else "unknown"
                logger.info(f"SCA 扫描结果（非本次扫描，当前任务 job_id: {sca_scan_id}，来自 commit: {short_commit}）: {sca_scan_result}")
            else:
                logger.info(f"SCA 扫描结果: {sca_scan_result}")
        else:
            logger.info("未获取到 SCA 扫描数据，跳过 SCA 结果输出")

        # 10. 获取代码门禁结果
        quality_gate_result = self.get_quality_gate_result(quality_scan_source)
        logger.info(f"代码门禁结果: {quality_gate_result}")

        # 11. 输出相关链接
        logger.info(f"分析概览链接：{self.client.get_scan_overview_url()}")
        logger.info(f"问题列表链接：{self.client.get_scan_lint_url()}")
        if sca_scan_id:
            logger.info(f"SCA 组成成分链接：{self.client.get_scan_sca_url(sca_scan_id)}")
        logger.info(f"圈复杂度链接：{self.client.get_cycs_url()}")
        logger.info(f"重复代码链接：{self.client.get_scan_dupfile_url()}")
        logger.info(f"代码统计链接：{self.client.get_scan_cloc_url()}")
        logger.info(f"任务结果链接：{self.client.get_job_result_url(job_id)}")

        if quality_gate_result.get("status") == "success" or quality_gate_result.get("status") == "close":
            return True
        else:
            return False

    def get_job_id(self, result: dict) -> int:
        try:
            job_id = result["data"]["job"]["id"]
        except (KeyError, TypeError):
            raise TCAIntegrationError("响应格式不正确，未找到 data.job.id")
        return job_id

    def get_result_msg(self, result: dict) -> Tuple[int, str, str]:
        try:
            result_code = result["data"]["result_code"]
            result_code_msg = result["data"]["result_code_msg"]
            result_msg = result["data"]["result_msg"]
        except (KeyError, TypeError):
            raise TCAIntegrationError("响应格式不正确，未找到 data.result_code相关信息")
        return result_code, result_code_msg, result_msg

    @staticmethod
    def _safe_get(data: dict, *keys):
        """安全地逐层获取嵌套字典值，任意一层为 None 则返回 None"""
        for key in keys:
            if isinstance(data, dict):
                data = data.get(key)
            else:
                return None
        return data

    def get_lint_scan_result(self, results: dict) -> dict:
        """获取代码扫描结果（lintscan）"""
        lint = results.get("lintscan") or {}
        return {
            "未处理问题数": self._safe_get(lint, "total", "state_detail", "active"),
            "严重问题数": self._safe_get(lint, "total", "severity_detail", "fatal", "active") or 0,
            "错误问题数": self._safe_get(lint, "total", "severity_detail", "error", "active") or 0,
            "警告问题数": self._safe_get(lint, "total", "severity_detail", "warning", "active") or 0,
            "提示问题数": self._safe_get(lint, "total", "severity_detail", "info", "active") or 0,
            "问题总数（未聚合）": lint.get("issue_detail_num"),
            "新增缺陷数": lint.get("issue_open_num"),
            "修复缺陷数": lint.get("issue_fix_num"),
        }

    def get_lint_scan_detail_result(self, lint_data: dict) -> dict:
        """从 overview/lintscans 接口返回的单条数据中解析代码扫描结果"""
        total = lint_data.get("total") or {}
        return {
            "未处理问题数": self._safe_get(total, "state_detail", "active") or 0,
            "严重问题数": self._safe_get(total, "severity_detail", "fatal", "active") or 0,
            "错误问题数": self._safe_get(total, "severity_detail", "error", "active") or 0,
            "警告问题数": self._safe_get(total, "severity_detail", "warning", "active") or 0,
            "提示问题数": self._safe_get(total, "severity_detail", "info", "active") or 0,
            "问题总数（未聚合）": lint_data.get("issue_detail_num", 0),
            "新增缺陷数": lint_data.get("issue_open_num", 0),
            "修复缺陷数": lint_data.get("issue_fix_num", 0),
        }

    def get_cyclomatic_complexity_result(self, results: dict) -> dict:
        """获取圈复杂度结果（cyclomaticcomplexityscan）"""
        cc = results.get("cyclomaticcomplexityscan") or {}
        return {
            "该分支所有超标方法个数": self._safe_get(cc, "custom_summary", "over_cc_func_count"),
            "该分支所有超标方法平均复杂度": self._safe_get(cc, "custom_summary", "over_cc_func_average"),
            "圈复杂度标准": cc.get("min_ccn"),
        }

    def get_cyc_scan_result(self, cyc_data: dict) -> dict:
        """从 overview/cycscans 接口返回的单条数据中解析圈复杂度结果"""
        custom_summary = cyc_data.get("custom_summary") or {}
        return {
            "该分支所有超标方法个数": custom_summary.get("over_cc_func_count", 0),
            "该分支所有超标方法平均复杂度": custom_summary.get("over_cc_func_average", 0),
            "圈复杂度标准": custom_summary.get("min_ccn") or cyc_data.get("min_ccn", 0),
        }

    def get_duplicate_code_result(self, results: dict) -> dict:
        """获取重复代码结果（duplicatescan）"""
        dup = results.get("duplicatescan") or {}
        return {
            "重复块数": dup.get("duplicate_block_count"),
            "重复行数": dup.get("total_duplicate_line_count"),
            "重复文件数": dup.get("duplicate_file_count"),
        }

    def get_dup_scan_result(self, dup_data: dict) -> dict:
        """从 overview/dupscans 接口返回的单条数据中解析重复代码结果"""
        return {
            "重复块数": dup_data.get("duplicate_block_count", 0),
            "重复行数": dup_data.get("total_duplicate_line_count", 0),
            "重复文件数": dup_data.get("duplicate_file_count", 0),
        }

    def get_cloc_result(self, results: dict) -> dict:
        """获取代码统计结果（clocscan）"""
        cloc = results.get("clocscan") or {}
        return {
            "代码行数": cloc.get("code_line_num"),
            "注释行数": cloc.get("comment_line_num"),
            "空白行数": cloc.get("blank_line_num"),
            "总行数": cloc.get("total_line_num"),
        }

    def get_cloc_scan_result(self, cloc_data: dict) -> dict:
        """从 overview/clocscans 接口返回的单条数据中解析代码统计结果"""
        return {
            "代码行数": cloc_data.get("code_line_num", 0),
            "注释行数": cloc_data.get("comment_line_num", 0),
            "空白行数": cloc_data.get("blank_line_num", 0),
            "总行数": cloc_data.get("total_line_num", 0),
        }

    @staticmethod
    def _risk_count(data_list: Optional[list], risk_key: str) -> int:
        """从 [{risk: ..., count: ...}, ...] 列表中提取指定 risk 的 count，未找到返回 0"""
        if not data_list or not isinstance(data_list, list):
            return 0
        for item in data_list:
            if isinstance(item, dict) and item.get("risk") == risk_key:
                return item.get("count", 0)
        return 0

    def get_sca_scan_result(self, sca_data: dict) -> dict:
        """解析 SCA 扫描结果，输出文件信息、病毒、Copyright、License、安全审计、漏洞审计、License审计"""
        file_num = sca_data.get("file_num", 0)
        meta_data = sca_data.get("meta_data") or {}
        source_reuse_rate = meta_data.get("source_reuse_rate", 0)
        # 自研率 = 1 - 开源代码占比
        source_reuse_rate_pct = round(source_reuse_rate * 100, 2)
        self_rate_pct = round((1 - source_reuse_rate) * 100, 2)

        malware_data = sca_data.get("malware_data")
        copyright_data = sca_data.get("copyright_tampering_data")
        license_tampering_data = sca_data.get("license_tampering_data")
        base_audit_data = sca_data.get("base_audit_data")
        cve_data = sca_data.get("cve_data")
        license_data = sca_data.get("license_data")

        return {
            "文件总数": file_num,
            "自研率": f"{self_rate_pct}%",
            "开源代码占比": f"{source_reuse_rate_pct}%",
            "病毒数": self._risk_count(malware_data, "Malware"),
            "非病毒数": self._risk_count(malware_data, "Pass"),
            "声明Copyright": self._risk_count(copyright_data, "Has"),
            "未声明Copyright": self._risk_count(copyright_data, "Without"),
            "Copyright篡改": self._risk_count(copyright_data, "Modified"),
            "声明License": self._risk_count(license_tampering_data, "Has"),
            "未声明License": self._risk_count(license_tampering_data, "Without"),
            "License篡改": self._risk_count(license_tampering_data, "Modified"),
            "安全审计-高危": self._risk_count(base_audit_data, "High"),
            "安全审计-中危": self._risk_count(base_audit_data, "Medium"),
            "安全审计-警告": self._risk_count(base_audit_data, "Warning"),
            "安全审计-通过": self._risk_count(base_audit_data, "Pass"),
            "安全审计-N/A": self._risk_count(base_audit_data, "N/A"),
            "漏洞审计-严重": self._risk_count(cve_data, "Critical"),
            "漏洞审计-高危": self._risk_count(cve_data, "High"),
            "漏洞审计-中危": self._risk_count(cve_data, "Medium"),
            "漏洞审计-低危": self._risk_count(cve_data, "Low"),
            "漏洞审计-N/A": self._risk_count(cve_data, "NotAvailable"),
            "License审计-高风险": self._risk_count(license_data, "High"),
            "License审计-中风险": self._risk_count(license_data, "Middle"),
            "License审计-低风险": self._risk_count(license_data, "Low"),
            "License审计-N/A": self._risk_count(license_data, "NotAvailable"),
        }

    @staticmethod
    def _is_all_none(data: dict, keys: list) -> bool:
        """判断字典中指定的所有 key 是否全为 None（或字典本身为空）"""
        if not data:
            return True
        return all(data.get(k) is None for k in keys)

    def _collect_scan_result(self, collector: dict, scan_info: dict, job_id: int):
        """通用的扫描结果收集逻辑：
        1. 优先从 overview（scan_info）中取数据
        2. overview 为空时 fallback 到独立接口
        3. fallback 数据与 job_id 不一致时，标注"非本次扫描"
        """
        name = collector["name"]
        is_empty = collector["overview_empty_check"](scan_info)

        if not is_empty:
            result = collector["overview_parser"](scan_info)
            logger.info(f"{name}结果: {result}")
            return

        fallback_data = collector["fallback_fetcher"]()
        if not fallback_data:
            logger.info(f"未获取到{name}扫描数据，跳过{name}结果输出")
            return

        scan_id = collector["scan_id_extractor"](fallback_data)
        result = collector["fallback_parser"](fallback_data)

        if scan_id and int(scan_id) != int(job_id):
            revision = collector["revision_extractor"](fallback_data)
            short_commit = revision[:8] if revision else "unknown"
            logger.info(f"{name}结果（非本次扫描，当前任务 job_id: {scan_id}，来自 commit: {short_commit}）: {result}")
        else:
            logger.info(f"{name}结果: {result}")

    def get_quality_gate_result(self, results: dict) -> dict:
        """获取质量门禁状态（qualityscan）"""
        quality = results.get("qualityscan") or {}
        return {
            "status": quality.get("status"),
            "description": quality.get("description"),
        }

    def has_team_name(self, response: dict, team_name: str) -> bool:
        """
        判断接口返回的 project_team 数据中是否存在 name == team_name 的项目

        :param response: 接口返回的完整字典数据
        :return: True 表示存在名为 team_name 的团队，否则 False
        """
        try:
            teams = response.get("data", [])
            for team in teams:
                if team.get("name") == team_name:
                    return True
            return False
        except Exception as e:
            TCAIntegrationError(f"解析 team 数据时出错: {e}")
            return False


def _manual_load_dotenv(candidates: List[str]) -> int:
    """简易 .env 解析器：在未安装 python-dotenv 时，手动加载 KEY=VALUE 到 os.environ。
  仅在变量未设置时覆盖。返回成功加载的变量数量。"""
    loaded = 0
    for path in candidates:
        if not path:
            continue
        if os.path.exists(path) and os.path.isfile(path):
            try:
                with open(path, 'r', encoding='utf-8') as f:
                    for raw_line in f:
                        line = raw_line.strip()
                        if not line or line.startswith('#'):
                            continue
                        if '=' not in line:
                            continue
                        key, val = line.split('=', 1)
                        key = key.strip()
                        val = val.strip().strip('"').strip("'")
                        if key and key not in os.environ:
                            os.environ[key] = val
                            loaded += 1
                logger.info(f"已从 .env 加载 {loaded} 个变量: {path}")
            except Exception as e:
                logger.warning(f"读取 .env 失败: {path} - {e}")
    return loaded


def str_to_bool(value: str) -> bool:
    """将字符串转换为布尔值（忽略大小写）"""
    return str(value).strip().lower() in ['true', '1', 't', 'y', 'yes']


def load_config_from_env() -> TCAConfig:
    """仅从环境变量与 .env 文件加载配置（不再读取 config.py）。
    优先级：进程环境变量 > .env > 默认值
    支持两处 .env:
        - 与本脚本同目录的 .env
        - 当前工作目录的 .env
    """
    # 1) 优先尝试 python-dotenv
    dotenv_loaded = False
    try:
        from dotenv import load_dotenv  # type: ignore
        from pathlib import Path
        dotenv_paths = [
            Path(__file__).with_name('.env'),
            Path.cwd() / '.env'
        ]
        for p in dotenv_paths:
            if p.exists():
                load_dotenv(dotenv_path=str(p), override=False)
                dotenv_loaded = True
                logger.info(f"已使用 python-dotenv 加载 .env: {p}")
                break
    except Exception:
        pass

    # 2) 若未通过 python-dotenv 加载，则手动加载
    if not dotenv_loaded:
        script_env = os.path.join(os.path.dirname(__file__), '.env')
        cwd_env = os.path.join(os.getcwd(), '.env')
        _manual_load_dotenv([script_env, cwd_env])

    # 3) 读取环境变量
    base_url = os.getenv('TCA_BASE_URL', 'https://tca.tencent.com')
    tca_token = os.getenv('TCA_TOKEN', '')
    user_id = os.getenv('TCA_USER_ID', '')
    org_sid = os.getenv('TCA_ORG_SID', '')
    team_name = os.getenv('TCA_TEAM_NAME', '')
    repo_url = os.getenv('TCA_REPO_URL', '')
    repo_branch = os.getenv('TCA_REPO_BRANCH', '')
    repo_mr_branch = os.getenv('TCA_REPO_MR_BRANCH', '')
    scm_account_id = os.getenv('TCA_SCM_ACCOUNT_ID', '')
    scheme_id = int(os.getenv('TCA_SCHEME_ID', '0'))
    incr_scan = os.getenv('TCA_INCR_SCAN', False)
    force_create = os.getenv('TCA_FORCE_CREATE', False)
    ignore_branch_issue = os.getenv('TCA_IGNORE_BRANCH_ISSUE', '')
    max_wait_time = os.getenv('TCA_MAX_WAIT_TIME', '600')
    return TCAConfig(
        base_url=base_url,
        tca_token=tca_token,
        user_id=user_id,
        org_sid=org_sid,
        team_name=team_name,
        repo_url=repo_url,
        scm_account_id=int(scm_account_id),
        repo_branch=repo_branch,
        repo_mr_branch=repo_mr_branch,
        scheme_id=scheme_id,
        incr_scan=str_to_bool(incr_scan),
        force_create=str_to_bool(force_create),
        ignore_branch_issue=ignore_branch_issue,
        max_wait_time=int(max_wait_time),
    )


def main():
    """主函数 - 示例用法"""
    # 加载配置（优先级：环境变量 > config.py > 默认值）
    config = load_config_from_env()
    logger.info(
        f"已加载配置: base_url={config.base_url}, "
        f"org_sid={'***' if config.org_sid else ''}, "
        f"tca_token={'***' if config.tca_token else ''}, "
        f"user_id={'***' if config.user_id else ''}, "
        f"team_name={config.team_name}, "
        f"repo_url={config.repo_url}, "
        f"repo_branch={config.repo_branch}, "
        f"repo_mr_branch={config.repo_mr_branch}, "
        f"scm_account_id={config.scm_account_id}, "
        f"scheme_id={config.scheme_id}, "
        f"incr_scan={config.incr_scan}, "
        f"force_create={config.force_create}, "
        f"ignore_branch_issue={config.ignore_branch_issue}"
    )
    # 验证必要配置
    if (not config.org_sid \
            or not config.tca_token \
            or not config.user_id \
            or not config.team_name \
            or not config.repo_url \
            or not config.repo_branch \
            or not config.scm_account_id \
            or not config.scheme_id):
        logger.error(
            "缺少必要配置，请设置 TCA_USER_ID、TCA_TOKEN、TCA_ORG_SID、TCA_TEAM_NAME、TCA_REPO_URL、TCA_REPO_BRANCH、TCA_SCM_ACCOUNT_ID、TCA_SCHEME_ID")
        sys.exit(1)

    # 创建流水线实例
    pipeline = TCAPipeline(config)

    scan_result = pipeline.run_full_analysis()

    if scan_result:
        logger.info("流水线执行成功")
        sys.exit(0)
    else:
        logger.error("流水线执行失败")
        sys.exit(1)


if __name__ == "__main__":
    main()