Python验收工具案例如何封装项目验收

本文目录导读：

1. 验收工具核心功能设计
2. 配置文件示例
3. 使用示例
4. 运行方式
5. 扩展建议

我来为您详细介绍Python项目验收工具的封装方法和实战案例。

验收工具核心功能设计

基础验收模块封装

# acceptance_tool.py
import os
import sys
import json
import yaml
import hashlib
import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
class AcceptanceLevel(Enum):
    """验收级别"""
    PASS = "通过"
    WARNING = "警告"
    ERROR = "失败"
    CRITICAL = "严重"
@dataclass
class AcceptanceResult:
    """验收结果数据类"""
    module: str
    item: str
    level: AcceptanceLevel
    message: str
    timestamp: datetime.datetime = None
    details: Dict = None
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.datetime.now()
        if self.details is None:
            self.details = {}
class ProjectAcceptance:
    """项目验收主类"""
    def __init__(self, project_path: str, config_file: str = "acceptance_config.yaml"):
        self.project_path = Path(project_path)
        self.config = self._load_config(config_file)
        self.results: List[AcceptanceResult] = []
        self._check_project_path()
    def _load_config(self, config_file: str) -> Dict:
        """加载配置文件"""
        config_path = Path(__file__).parent / config_file
        if not config_path.exists():
            return self._get_default_config()
        with open(config_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)
    def _get_default_config(self) -> Dict:
        """获取默认配置"""
        return {
            "project_structure": {
                "required_dirs": ["src", "tests", "docs", "config"],
                "optional_dirs": ["scripts", "data", "examples"]
            },
            "code_quality": {
                "min_coverage": 80,
                "max_complexity": 10,
                "min_docstring_ratio": 0.3
            },
            "security_checks": {
                "check_vulnerabilities": True,
                "check_secrets": True
            }
        }
    def _check_project_path(self):
        """检查项目路径"""
        if not self.project_path.exists():
            raise FileNotFoundError(f"项目路径不存在: {self.project_path}")

文件结构检查模块

class ProjectStructureChecker:
    """项目结构检查器"""
    def __init__(self, project_path: Path, config: Dict):
        self.project_path = project_path
        self.config = config
        self.results = []
    def check_required_directories(self) -> List[AcceptanceResult]:
        """检查必需目录是否存在"""
        results = []
        required_dirs = self.config.get("required_dirs", [])
        for dir_name in required_dirs:
            dir_path = self.project_path / dir_name
            if dir_path.exists() and dir_path.is_dir():
                results.append(AcceptanceResult(
                    module="项目结构",
                    item=f"目录: {dir_name}",
                    level=AcceptanceLevel.PASS,
                    message=f"必需目录 {dir_name} 存在"
                ))
            else:
                results.append(AcceptanceResult(
                    module="项目结构",
                    item=f"目录: {dir_name}",
                    level=AcceptanceLevel.ERROR,
                    message=f"必需目录 {dir_name} 不存在"
                ))
        return results
    def check_file_naming_convention(self) -> List[AcceptanceResult]:
        """检查文件命名规范"""
        results = []
        conventions = self.config.get("naming_conventions", {
            "python_files": "snake_case",
            "config_files": "snake_case",
            "test_files": "test_*.py"
        })
        for pattern, convention in conventions.items():
            matched_files = list(self.project_path.rglob(pattern))
            for file_path in matched_files:
                if not self._check_naming(file_path.name, convention):
                    results.append(AcceptanceResult(
                        module="命名规范",
                        item=f"文件: {file_path}",
                        level=AcceptanceLevel.WARNING,
                        message=f"文件命名不符合 {convention} 规范"
                    ))
        return results
    def _check_naming(self, filename: str, convention: str) -> bool:
        """检查文件命名是否符合规范"""
        if convention == "snake_case":
            name_without_ext = filename.rsplit('.', 1)[0]
            return name_without_ext.islower() and '_' in name_without_ext
        return True

代码质量检查模块

import ast
import radon
from radon.complexity import cc_visit
from radon.metrics import mi_visit
import coverage
class CodeQualityChecker:
    """代码质量检查器"""
    def __init__(self, project_path: Path, config: Dict):
        self.project_path = project_path
        self.config = config
        self.results = []
    def check_code_complexity(self) -> List[AcceptanceResult]:
        """检查代码复杂度"""
        results = []
        max_complexity = self.config.get("max_complexity", 10)
        for python_file in self.project_path.rglob("*.py"):
            try:
                with open(python_file, 'r', encoding='utf-8') as f:
                    code = f.read()
                complexity_results = cc_visit(code)
                for func_info in complexity_results:
                    if func_info.complexity > max_complexity:
                        results.append(AcceptanceResult(
                            module="代码复杂度",
                            item=f"{python_file}:{func_info.name}",
                            level=AcceptanceLevel.WARNING,
                            message=f"函数复杂度 {func_info.complexity} 超过限制 {max_complexity}"
                        ))
            except Exception as e:
                results.append(AcceptanceResult(
                    module="代码复杂度",
                    item=str(python_file),
                    level=AcceptanceLevel.ERROR,
                    message=f"检查失败: {str(e)}"
                ))
        return results
    def check_docstring_coverage(self) -> List[AcceptanceResult]:
        """检查文档字符串覆盖率"""
        results = []
        min_ratio = self.config.get("min_docstring_ratio", 0.3)
        total_functions = 0
        documented_functions = 0
        for python_file in self.project_path.rglob("*.py"):
            try:
                with open(python_file, 'r', encoding='utf-8') as f:
                    tree = ast.parse(f.read())
                for node in ast.walk(tree):
                    if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
                        total_functions += 1
                        if ast.get_docstring(node):
                            documented_functions += 1
            except Exception as e:
                continue
        coverage_ratio = documented_functions / total_functions if total_functions > 0 else 1
        if coverage_ratio < min_ratio:
            results.append(AcceptanceResult(
                module="文档覆盖率",
                item="整体项目",
                level=AcceptanceLevel.WARNING,
                message=f"文档覆盖率 {coverage_ratio:.1%} 低于标准 {min_ratio:.1%}"
            ))
        else:
            results.append(AcceptanceResult(
                module="文档覆盖率",
                item="整体项目",
                level=AcceptanceLevel.PASS,
                message=f"文档覆盖率 {coverage_ratio:.1%} 符合标准"
            ))
        return results

安全审查模块

import re
import secrets
from pathlib import Path
class SecurityChecker:
    """安全审查器"""
    def __init__(self, project_path: Path, config: Dict):
        self.project_path = project_path
        self.config = config
        self.secret_patterns = [
            r'password\s*=\s*["\'](.*?)["\']',
            r'api_key\s*=\s*["\'](.*?)["\']',
            r'secret_key\s*=\s*["\'](.*?)["\']',
            r'token\s*=\s*["\'](.*?)["\']',
            r'DATABASE_URL\s*=\s*["\'](.*?)["\']'
        ]
    def check_hardcoded_secrets(self) -> List[AcceptanceResult]:
        """检查硬编码的密钥"""
        results = []
        for file_path in self.project_path.rglob("*"):
            if file_path.suffix not in ['.py', '.env', '.json', '.yaml', '.yml', '.toml']:
                continue
            try:
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = f.read()
                # 忽略测试文件
                if 'test' in file_path.stem.lower():
                    continue
                for i, pattern in enumerate(self.secret_patterns):
                    matches = re.finditer(pattern, content, re.IGNORECASE)
                    for match in matches:
                        value = match.group(1)
                        if not self._is_placeholder(value):
                            results.append(AcceptanceResult(
                                module="安全审查",
                                item=f"{file_path}:{match.group(0)[:30]}...",
                                level=AcceptanceLevel.CRITICAL,
                                message="发现硬编码的敏感信息"
                            ))
            except Exception as e:
                continue
        return results
    def _is_placeholder(self, value: str) -> bool:
        """检查是否为占位符"""
        placeholders = ['your_', 'xxx', '***', 'password', 'changeme']
        return any(placeholder.lower() in value.lower() for placeholder in placeholders)
    def check_dependency_vulnerabilities(self) -> List[AcceptanceResult]:
        """检查依赖漏洞"""
        results = []
        requirements_files = list(self.project_path.glob("requirements*.txt"))
        requirements_files.extend(self.project_path.glob("Pipfile"))
        requirements_files.extend(self.project_path.glob("poetry.lock"))
        if not requirements_files:
            results.append(AcceptanceResult(
                module="依赖安全",
                item="依赖文件",
                level=AcceptanceLevel.WARNING,
                message="未找到依赖管理文件"
            ))
        for req_file in requirements_files:
            # 这里可以集成安全数据库进行漏洞检查
            # 简化版：检查是否有已知的不安全包
            unsafe_packages = ['cryptography<3.4', 'django<3.2']
            with open(req_file, 'r') as f:
                content = f.read()
            for unsafe_pkg in unsafe_packages:
                if unsafe_pkg in content:
                    results.append(AcceptanceResult(
                        module="依赖安全",
                        item=str(req_file),
                        level=AcceptanceLevel.ERROR,
                        message=f"发现已知漏洞依赖: {unsafe_pkg}"
                    ))
        return results

测试覆盖率检查

import subprocess
import xml.etree.ElementTree as ET
class TestCoverageChecker:
    """测试覆盖率检查器"""
    def __init__(self, project_path: Path, config: Dict):
        self.project_path = project_path
        self.config = config
        self.cov = coverage.Coverage()
    def check_coverage(self) -> List[AcceptanceResult]:
        """检查测试覆盖率"""
        results = []
        min_coverage = self.config.get("min_coverage", 80)
        try:
            # 运行测试并收集覆盖率
            os.chdir(self.project_path)
            subprocess.run(['pytest', '--cov=./', '--cov-report=xml'], 
                         capture_output=True, text=True, check=True)
            # 解析覆盖率报告
            coverage_xml = self.project_path / 'coverage.xml'
            if coverage_xml.exists():
                tree = ET.parse(str(coverage_xml))
                root = tree.getroot()
                coverage_value = float(root.attrib.get('line-rate', 0)) * 100
                if coverage_value >= min_coverage:
                    results.append(AcceptanceResult(
                        module="测试覆盖率",
                        item="整体项目",
                        level=AcceptanceLevel.PASS,
                        message=f"测试覆盖率 {coverage_value:.2f}% 符合最低要求 {min_coverage}%"
                    ))
                else:
                    results.append(AcceptanceResult(
                        module="测试覆盖率",
                        item="整体项目",
                        level=AcceptanceLevel.ERROR,
                        message=f"测试覆盖率 {coverage_value:.2f}% 低于最低要求 {min_coverage}%"
                    ))
                # 清理覆盖率文件
                coverage_xml.unlink()
        except subprocess.CalledProcessError as e:
            results.append(AcceptanceResult(
                module="测试覆盖率",
                item="运行测试",
                level=AcceptanceLevel.ERROR,
                message=f"运行测试失败: {e.stderr}"
            ))
        except Exception as e:
            results.append(AcceptanceResult(
                module="测试覆盖率",
                item="覆盖率检查",
                level=AcceptanceLevel.ERROR,
                message=f"覆盖率检查失败: {str(e)}"
            ))
        return results

报告生成器

class AcceptanceReportGenerator:
    """验收报告生成器"""
    def __init__(self, results: List[AcceptanceResult]):
        self.results = results
    def generate_html_report(self, output_path: str = "acceptance_report.html") -> str:
        """生成HTML格式报告"""
        html_content = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>项目验收报告</title>
            <style>
                body {{ font-family: Arial, sans-serif; margin: 20px; }}
                h1 {{ color: #333; }}
                .summary {{ background: #f5f5f5; padding: 15px; border-radius: 5px; }}
                .pass {{ color: green; }}
                .warning {{ color: orange; }}
                .error {{ color: red; }}
                .critical {{ color: darkred; }}
                table {{ width: 100%; border-collapse: collapse; }}
                th, td {{ padding: 10px; text-align: left; border-bottom: 1px solid #ddd; }}
                th {{ background-color: #f2f2f2; }}
            </style>
        </head>
        <body>
            <h1>项目验收报告</h1>
            <div class="summary">
                <h2>概览</h2>
                <p>生成时间: {datetime.datetime.now()}</p>
                <p>总检查项: {len(self.results)}</p>
                <p>通过: {self._count_by_level(AcceptanceLevel.PASS)}</p>
                <p>警告: {self._count_by_level(AcceptanceLevel.WARNING)}</p>
                <p>失败: {self._count_by_level(AcceptanceLevel.ERROR)}</p>
                <p>严重: {self._count_by_level(AcceptanceLevel.CRITICAL)}</p>
            </div>
            <h2>详细结果</h2>
            <table>
                <thead>
                    <tr>
                        <th>模块</th>
                        <th>检查项</th>
                        <th>级别</th>
                        <th>描述</th>
                        <th>时间</th>
                    </tr>
                </thead>
                <tbody>
        """
        for result in self.results:
            level_class = result.level.name.lower()
            html_content += f"""
                    <tr>
                        <td>{result.module}</td>
                        <td>{result.item}</td>
                        <td class="{level_class}">{result.level.value}</td>
                        <td>{result.message}</td>
                        <td>{result.timestamp}</td>
                    </tr>
            """
        html_content += """
                </tbody>
            </table>
        </body>
        </html>
        """
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(html_content)
        return output_path
    def generate_json_report(self, output_path: str = "acceptance_report.json") -> str:
        """生成JSON格式报告"""
        report_data = {
            "report_time": datetime.datetime.now().isoformat(),
            "total_checks": len(self.results),
            "summary": {
                "passed": self._count_by_level(AcceptanceLevel.PASS),
                "warnings": self._count_by_level(AcceptanceLevel.WARNING),
                "errors": self._count_by_level(AcceptanceLevel.ERROR),
                "critical": self._count_by_level(AcceptanceLevel.CRITICAL)
            },
            "results": [asdict(r) for r in self.results]
        }
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(report_data, f, indent=2, ensure_ascii=False, default=str)
        return output_path
    def _count_by_level(self, level: AcceptanceLevel) -> int:
        """按级别统计结果数量"""
        return len([r for r in self.results if r.level == level])

主验收流程集成

class AcceptanceRunner:
    """验收运行器"""
    def __init__(self, project_path: str, config_file: str = "acceptance_config.yaml"):
        self.project_path = Path(project_path)
        self.config = self._load_config(config_file)
        self.results = []
    def _load_config(self, config_file: str) -> Dict:
        """加载配置"""
        config_path = self.project_path / config_file
        if config_path.exists():
            with open(config_path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f)
        return self._get_default_config()
    def _get_default_config(self) -> Dict:
        """获取默认配置"""
        return {
            "modules": {
                "structure": True,
                "code_quality": True,
                "security": True,
                "tests": True
            },
            "output": {
                "html_report": True,
                "json_report": True,
                "report_dir": "acceptance_reports"
            },
            "filters": {
                "exclude_dirs": [".git", "__pycache__", "venv", ".env"],
                "exclude_files": ["*.pyc", "*.pyo"]
            }
        }
    def run_acceptance(self) -> Dict:
        """运行完整的验收流程"""
        print(f"开始项目验收: {self.project_path}")
        # 初始化检查器
        structure_checker = ProjectStructureChecker(self.project_path, self.config)
        quality_checker = CodeQualityChecker(self.project_path, self.config)
        security_checker = SecurityChecker(self.project_path, self.config)
        test_checker = TestCoverageChecker(self.project_path, self.config)
        # 执行检查
        checks = []
        if self.config.get("modules", {}).get("structure", True):
            print("[1/4] 检查项目结构...")
            checks.append(structure_checker.check_required_directories())
            checks.append(structure_checker.check_file_naming_convention())
        if self.config.get("modules", {}).get("code_quality", True):
            print("[2/4] 检查代码质量...")
            checks.append(quality_checker.check_code_complexity())
            checks.append(quality_checker.check_docstring_coverage())
        if self.config.get("modules", {}).get("security", True):
            print("[3/4] 安全检查...")
            checks.append(security_checker.check_hardcoded_secrets())
            checks.append(security_checker.check_dependency_vulnerabilities())
        if self.config.get("modules", {}).get("tests", True):
            print("[4/4] 检查测试覆盖率...")
            checks.append(test_checker.check_coverage())
        # 合并所有结果
        for check_results in checks:
            self.results.extend(check_results)
        # 生成报告
        report_generator = AcceptanceReportGenerator(self.results)
        report_dir = self.project_path / self.config.get("output", {}).get("report_dir", "acceptance_reports")
        report_dir.mkdir(exist_ok=True)
        if self.config.get("output", {}).get("html_report", True):
            html_path = report_dir / "acceptance_report.html"
            report_generator.generate_html_report(str(html_path))
            print(f"HTML报告已生成: {html_path}")
        if self.config.get("output", {}).get("json_report", True):
            json_path = report_dir / "acceptance_report.json"
            report_generator.generate_json_report(str(json_path))
            print(f"JSON报告已生成: {json_path}")
        # 返回摘要
        return self._get_summary()
    def _get_summary(self) -> Dict:
        """获取验收摘要"""
        return {
            "project_path": str(self.project_path),
            "timestamp": datetime.datetime.now().isoformat(),
            "total_checks": len(self.results),
            "passed": len([r for r in self.results if r.level == AcceptanceLevel.PASS]),
            "warnings": len([r for r in self.results if r.level == AcceptanceLevel.WARNING]),
            "errors": len([r for r in self.results if r.level == AcceptanceLevel.ERROR]),
            "critical": len([r for r in self.results if r.level == AcceptanceLevel.CRITICAL]),
            "passed_checks": sum(1 for r in self.results if r.level in [AcceptanceLevel.PASS]),
            "failed_checks": sum(1 for r in self.results if r.level in [AcceptanceLevel.ERROR, AcceptanceLevel.CRITICAL])
        }

配置文件示例

# acceptance_config.yaml
project:
  name: "MyProject"
  version: "1.0.0"
  language: "python"
modules:
  structure: true
  code_quality: true
  security: true
  tests: true
project_structure:
  required_dirs:
    - src
    - tests
    - docs
    - config
  optional_dirs:
    - scripts
    - data
    - examples
  naming_conventions:
    python_files: snake_case
    config_files: snake_case
    test_files: "test_*.py"
code_quality:
  min_coverage: 80
  max_complexity: 10
  min_docstring_ratio: 0.3
  max_file_length: 500
  max_function_length: 50
security:
  check_vulnerabilities: true
  check_secrets: true
  allowed_unsafe_packages: []
output:
  html_report: true
  json_report: true
  report_dir: "acceptance_reports"
  verbose: true
filters:
  exclude_dirs:
    - .git
    - __pycache__
    - venv
    - .env
    - node_modules
  exclude_files:
    - "*.pyc"
    - "*.pyo"
    - "*.so"

使用示例

# main.py
import sys
from pathlib import Path
# 添加工具路径
sys.path.append(str(Path(__file__).parent))
from acceptance_tool import AcceptanceRunner
def main():
    """主函数"""
    import argparse
    parser = argparse.ArgumentParser(description="Python项目验收工具")
    parser.add_argument("project_path", help="要验收的项目路径")
    parser.add_argument("--config", default="acceptance_config.yaml",
                       help="配置文件路径")
    parser.add_argument("--debug", action="store_true",
                       help="启用调试模式")
    args = parser.parse_args()
    try:
        # 创建验收运行器
        runner = AcceptanceRunner(args.project_path, args.config)
        # 运行验收
        summary = runner.run_acceptance()
        # 输出结果
        print("\n=== 验收结果摘要 ===")
        print(f"总检查项: {summary['total_checks']}")
        print(f"通过: {summary['passed']}")
        print(f"警告: {summary['warnings']}")
        print(f"失败: {summary['errors']}")
        print(f"严重: {summary['critical']}")
        print(f"通过率: {summary['passed_checks']/summary['total_checks']*100:.2f}%")
        # 根据结果设置退出码
        if summary['critical'] > 0:
            sys.exit(2)
        elif summary['errors'] > 0:
            sys.exit(1)
        else:
            sys.exit(0)
    except Exception as e:
        print(f"验收过程发生错误: {str(e)}")
        if args.debug:
            import traceback
            traceback.print_exc()
        sys.exit(1)
if __name__ == "__main__":
    main()

运行方式

# 基本使用
python main.py /path/to/your/project
# 指定配置文件
python main.py /path/to/your/project --config custom_config.yaml
# 调试模式
python main.py /path/to/your/project --debug

扩展建议

1. 添加自定义检查器：通过继承基类实现新的检查模块
2. 支持CI/CD集成：生成JSON格式输出，方便Jenkins/GitLab CI解析
3. 添加性能监控：集成代码性能分析工具
4. 支持插件机制：通过插件系统扩展功能
5. 缓存机制：对于大型项目，缓存检查结果提高效率

这个验收工具封装了完整的项目验收流程,可以根据实际需求灵活配置和扩展。