neverSatRabbit

---
name: payment_skill
description: AI 原生支付解决方案 - 提供安全、可靠的支付功能
version: 1.0.3
author: Payment Institution
license: MIT
tags:
  - payment
  - finance
  - commerce
skill_type: code
has_server: false
has_install_scripts: true
security_level: high
requires:
  - python>=3.6
environment_variables:
  required:
    - PAYMENT_API_KEY: "API 密钥，用于身份认证"
    - PAYMENT_API_SECRET: "API 密钥，用于请求签名"
  optional:
    - PAYMENT_API_URL: "支付 API 地址，默认 https://api.zlclaw.com"
    - PAYMENT_ENCRYPTION_KEY: "32 字节加密密钥（可选）"
    - PAYMENT_LOG_LEVEL: "日志级别，默认 INFO"
tools:
  - name: create_payment
    description: 创建支付请求，触发用户授权流程
    parameters:
      - name: amount
        type: number
        description: 支付金额，最多2位小数
        required: true
        example: 299.00
      - name: currency
        type: string
        description: ISO 4217 货币代码
        required: false
        default: CNY
        example: CNY
      - name: merchant_id
        type: string
        description: 商户唯一标识
        required: true
        example: taobao_merchant_001
      - name: description
        type: string
        description: 支付描述，用户可见
        required: false
        example: 购买无线鼠标
    returns:
      type: object
      properties:
        success:
          type: boolean
          description: 是否成功
        transaction_id:
          type: string
          description: 交易ID
        status:
          type: string
          description: 支付状态
        amount:
          type: number
          description: 支付金额
        currency:
          type: string
          description: 货币代码
        verification_url:
          type: string
          description: 验证URL
        qr_code:
          type: string
          description: 二维码数据
  
  - name: query_payment
    description: 查询支付状态
    parameters:
      - name: transaction_id
        type: string
        description: 交易 ID
        required: true
        example: txn_20260315_001
    returns:
      type: object
      properties:
        success:
          type: boolean
          description: 是否成功
        transaction_id:
          type: string
          description: 交易ID
        status:
          type: string
          description: 支付状态
        amount:
          type: number
          description: 支付金额
  
  - name: refund_payment
    description: 发起退款
    parameters:
      - name: transaction_id
        type: string
        description: 原交易 ID
        required: true
        example: txn_20260315_001
      - name: amount
        type: number
        description: 退款金额（可选，不填则全额退款）
        required: false
        example: 100.00
      - name: reason
        type: string
        description: 退款原因
        required: false
        example: 商品质量问题
    returns:
      type: object
      properties:
        success:
          type: boolean
          description: 是否成功
        refund_id:
          type: string
          description: 退款ID
        status:
          type: string
          description: 退款状态
        amount:
          type: number
          description: 退款金额
---

# 支付 Skill

AI 原生支付解决方案，提供安全、可靠的支付功能。

## 工具调用方式

AI 通过执行 `skill_cli.py` 脚本来调用工具：

```bash
python skill_cli.py <tool_name> --param1=value1 --param2=value2
```

**示例**：
```bash
# 创建支付
python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001 --description="购买商品"

# 查询支付
python skill_cli.py query_payment --transaction_id=txn_001

# 发起退款
python skill_cli.py refund_payment --transaction_id=txn_001 --amount=100.00 --reason="质量问题"
```

**返回格式**：所有工具都返回 JSON 格式的结果
```json
{
  "success": true,
  "data": { ... }
}
```

## 快速开始

### 1. 安装依赖

```bash
# Linux/macOS
./scripts/setup.sh

# Windows
scripts\setup.bat
```

### 2. 配置环境变量

**必需**：
```bash
export PAYMENT_API_KEY=your_api_key
export PAYMENT_API_SECRET=your_api_secret
```

**可选**：
```bash
export PAYMENT_API_URL=https://api.zlclaw.com
```

### 3. 使用示例

```bash
# 创建支付
python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001

# 查询支付
python skill_cli.py query_payment --transaction_id=txn_001

# 发起退款
python skill_cli.py refund_payment --transaction_id=txn_001 --amount=100.00
```

## 核心功能

- **create_payment**: 创建支付请求
- **query_payment**: 查询支付状态
- **refund_payment**: 发起退款

## 安全特性

- TLS 加密通信
- HMAC-SHA256 签名
- 参数验证
- 审计日志

## 技术要求

- Python 3.6+
- 网络访问权限
- 环境变量配置

## 支持

- 📧 技术支持: [email protected]
- 🐛 问题报告: https://github.com/zlclaw/payment-skill/issues

---

**版本**: 1.0.3 | **许可证**: MIT

FILE:_meta.json
{
  "ownerId": "kn79xswxbje4zn43vw5yn021a180bsan",
  "slug": "payment-skill",
  "version": "1.0.3",
  "publishedAt": 1710768000000,
  "requiredEnvVars": [
    "PAYMENT_API_KEY",
    "PAYMENT_API_SECRET",
    "PAYMENT_API_URL"
  ],
  "optionalEnvVars": [
    "PAYMENT_ENCRYPTION_KEY",
    "PAYMENT_LOG_LEVEL",
    "PAYMENT_LOG_FILE"
  ]
}

FILE:config/production.env
# 支付 Skill 生产环境配置
# 
# 重要：API 凭证必须通过环境变量设置，不要在此文件中硬编码
# 
# 设置方式：
# export PAYMENT_API_KEY=your_real_key
# export PAYMENT_API_SECRET=your_real_secret
#
# PAYMENT_API_URL 有默认值 https://api.zlclaw.com，不需要设置
#
# 如果未设置必需的环境变量，Skill 会在初始化时报错

# ============================================
# 必需配置 (Required) - 通过环境变量设置
# ============================================

# API 配置（必须通过环境变量设置）
# 不要在此文件中设置 PAYMENT_API_KEY、PAYMENT_API_SECRET
# 这些凭证应该通过环境变量设置，以保护敏感信息
PAYMENT_API_URL=https://api.zlclaw.com
PAYMENT_API_TIMEOUT=30

# ============================================
# 可选配置 (Optional)
# ============================================

# 安全配置
PAYMENT_ENABLE_SIGNATURE=true
PAYMENT_ENABLE_ENCRYPTION=false
PAYMENT_TLS_VERIFY=true

# 日志配置
PAYMENT_LOG_LEVEL=INFO
PAYMENT_LOG_FILE=/var/log/payment-skill/payment-skill.log

# 功能配置
PAYMENT_ENABLE_CROSS_DEVICE_AUTH=true
PAYMENT_POLLING_INTERVAL=2000
PAYMENT_MAX_POLLING_ATTEMPTS=60

FILE:scripts/diagnose.py
#!/usr/bin/env python3
"""
支付 Skill 环境诊断工具

用于诊断 Python 3.6 环境中的依赖问题。
"""

import sys
import subprocess
import importlib
from pathlib import Path


def print_header(text):
    """打印标题"""
    print(f"\n{'='*60}")
    print(f"  {text}")
    print(f"{'='*60}\n")


def print_success(text):
    """打印成功信息"""
    print(f"✅ {text}")


def print_error(text):
    """打印错误信息"""
    print(f"❌ {text}")


def print_warning(text):
    """打印警告信息"""
    print(f"⚠️  {text}")


def print_info(text):
    """打印信息"""
    print(f"ℹ️  {text}")


def check_python_version():
    """检查 Python 版本"""
    print_header("Python 版本检查")
    
    version = sys.version_info
    version_str = f"{version.major}.{version.minor}.{version.micro}"
    
    print(f"Python 版本: {version_str}")
    print(f"Python 路径: {sys.executable}")
    
    if version.major == 3 and version.minor == 6:
        print_warning("Python 3.6 已停止支持，建议升级到 Python 3.8+")
        return True
    elif version.major == 3 and version.minor >= 8:
        print_success("Python 版本支持良好")
        return True
    else:
        print_error(f"不支持的 Python 版本: {version_str}")
        return False


def check_pip_version():
    """检查 pip 版本"""
    print_header("pip 版本检查")
    
    try:
        result = subprocess.run(
            [sys.executable, "-m", "pip", "--version"],
            capture_output=True,
            text=True
        )
        pip_version = result.stdout.strip()
        print(f"pip 版本: {pip_version}")
        
        # 提取版本号
        if "pip" in pip_version:
            print_success("pip 已安装")
            return True
    except Exception as e:
        print_error(f"无法检查 pip 版本: {e}")
        return False


def check_dependencies():
    """检查依赖"""
    print_header("依赖检查")
    
    dependencies = {
        "aiohttp": "异步 HTTP 客户端",
        "pydantic": "数据验证",
        "pyyaml": "YAML 配置",
        "python-dotenv": "环境变量管理",
        "cryptography": "加密库",
        "Crypto": "加密库 (pycryptodome)",
        "pytest": "测试框架",
        "pytest_asyncio": "异步测试支持",
        "redis": "缓存库",
    }
    
    installed = {}
    missing = []
    
    for module_name, description in dependencies.items():
        try:
            module = importlib.import_module(module_name)
            version = getattr(module, "__version__", "unknown")
            installed[module_name] = version
            print_success(f"{module_name} ({description}): {version}")
        except ImportError:
            missing.append((module_name, description))
            print_error(f"{module_name} ({description}): 未安装")
    
    return len(missing) == 0, missing


def check_asyncmock():
    """检查 AsyncMock 兼容性"""
    print_header("AsyncMock 兼容性检查")
    
    version = sys.version_info
    
    if version.major == 3 and version.minor >= 8:
        try:
            from unittest.mock import AsyncMock
            print_success("AsyncMock 原生支持 (Python 3.8+)")
            return True
        except ImportError:
            print_error("AsyncMock 导入失败")
            return False
    else:
        print_info("Python 3.6/3.7 使用兼容实现")
        try:
            from unittest.mock import Mock
            
            # 测试兼容实现
            class AsyncMock(Mock):
                async def __call__(self, *args, **kwargs):
                    return super(AsyncMock, self).__call__(*args, **kwargs)
            
            print_success("AsyncMock 兼容实现可用")
            return True
        except Exception as e:
            print_error(f"AsyncMock 兼容实现失败: {e}")
            return False


def check_project_structure():
    """检查项目结构"""
    print_header("项目结构检查")
    
    # 获取脚本目录和项目目录
    script_dir = Path(__file__).parent
    project_dir = script_dir.parent
    
    required_files = [
        "src/payment_skill.py",
        "src/payment_api_client.py",
        "src/security.py",
        "src/utils.py",
        "tests/test_payment_skill.py",
        "tests/test_integration.py",
        "scripts/requirements-py36.txt",
        "scripts/payment_skill.yaml",
    ]
    
    all_exist = True
    for file_path in required_files:
        full_path = project_dir / file_path
        if full_path.exists():
            print_success(f"✓ {file_path}")
        else:
            print_error(f"✗ {file_path} (缺失)")
            all_exist = False
    
    return all_exist


def check_requirements_file():
    """检查 requirements 文件"""
    print_header("Requirements 文件检查")
    
    # 获取脚本目录和项目目录
    script_dir = Path(__file__).parent
    project_dir = script_dir.parent
    
    py36_req = project_dir / "scripts" / "requirements-py36.txt"
    std_req = project_dir / "scripts" / "requirements.txt"
    
    if py36_req.exists():
        print_success(f"✓ requirements-py36.txt 存在")
        print_info("Python 3.6 环境应使用此文件")
    else:
        print_error("✗ requirements-py36.txt 缺失")
    
    if std_req.exists():
        print_success(f"✓ requirements.txt 存在")
        print_info("Python 3.8+ 环境应使用此文件")
    else:
        print_error("✗ requirements.txt 缺失")


def check_environment_variables():
    """检查环境变量"""
    print_header("环境变量检查")
    
    import os
    
    env_vars = [
        "PAYMENT_API_KEY",
        "PAYMENT_API_SECRET",
        "PAYMENT_API_URL",
        "PAYMENT_LOG_LEVEL",
    ]
    
    for var in env_vars:
        value = os.getenv(var)
        if value:
            # 隐藏敏感信息
            if "SECRET" in var or "KEY" in var:
                display_value = f"{value[:4]}...{value[-4:]}"
            else:
                display_value = value
            print_success(f"{var}: {display_value}")
        else:
            print_warning(f"{var}: 未设置")


def run_diagnostics():
    """运行所有诊断"""
    print("\n")
    print("╔" + "="*58 + "╗")
    print("║" + " "*58 + "║")
    print("║" + "  支付 Skill 环境诊断工具".center(58) + "║")
    print("║" + " "*58 + "║")
    print("╚" + "="*58 + "╝")
    
    results = {
        "Python 版本": check_python_version(),
        "pip 版本": check_pip_version(),
        "AsyncMock 兼容性": check_asyncmock(),
        "项目结构": check_project_structure(),
    }
    
    check_dependencies()
    check_requirements_file()
    check_environment_variables()
    
    # 总结
    print_header("诊断总结")
    
    all_passed = all(results.values())
    
    if all_passed:
        print_success("所有检查通过！")
        print_info("您可以运行: python3 -m pytest tests/ -v")
    else:
        print_error("发现问题，请参考上面的错误信息")
        print_info("查看 PYTHON36_SETUP.md 获取详细解决方案")
    
    print("\n")


if __name__ == "__main__":
    run_diagnostics()

FILE:scripts/requirements-py36.txt
# Python 3.6 兼容版本
# 注意: Python 3.6 已停止支持,建议升级到 Python 3.8+

# 异步 HTTP 客户端 (3.6 兼容版本)
aiohttp==3.6.3
multidict==4.7.6
yarl==1.4.2

# 数据验证
pydantic==1.8.2

# YAML 配置
pyyaml==5.4.1

# 环境变量管理
python-dotenv==0.19.0

# 加密库
cryptography==3.4.8
pycryptodome==3.10.4

# 测试框架
pytest==6.2.5
pytest-asyncio==0.14.0
pytest-cov==2.12.1

# 缓存
redis==3.5.3

FILE:scripts/requirements.txt
# 核心依赖（固定版本以防止供应链攻击）
aiohttp==3.9.5
pydantic==2.7.0
pyyaml==6.0.1
python-dotenv==1.0.1
cryptography==42.0.5
pycryptodome==3.19.1

# 测试依赖
pytest==8.2.0
pytest-asyncio==0.23.6
pytest-cov==5.0.0

# 可选依赖
redis==5.0.1

# 安全扫描工具（可选安装）
# safety==2.3.5
# bandit==1.7.9

FILE:scripts/setup.sh
#!/bin/bash

# Payment Skill Setup Script (Linux/macOS)
# Auto-detect Python version and install dependencies

set -e

SCRIPT_DIR="$(cd "$(dirname "BASH_SOURCE[0]")" && pwd)"
PROJECT_DIR="$(dirname "$SCRIPT_DIR")"

echo ""
echo "========================================"
echo "Payment Skill Setup"
echo "========================================"
echo ""

echo "Checking Python version..."
python3 --version 2>/dev/null || {
    echo "Error: Python 3 is not installed"
    exit 1
}

# Detect Python version
PYTHON_VERSION_TYPE=$(python3 -c "import sys; print('py38' if sys.version_info >= (3, 8) else 'py36')")

if [ "$PYTHON_VERSION_TYPE" = "py38" ]; then
    echo "Python 3.8+ detected (recommended)"
    REQUIREMENTS_FILE="$SCRIPT_DIR/requirements.txt"
else
    echo "Python 3.6 detected (deprecated)"
    REQUIREMENTS_FILE="$SCRIPT_DIR/requirements-py36.txt"
fi

echo ""
echo "Checking virtual environment..."
if [ -d "$PROJECT_DIR/venv" ]; then
    echo "Virtual environment already exists"
else
    echo "Creating virtual environment..."
    python3 -m venv "$PROJECT_DIR/venv"
    if [ $? -ne 0 ]; then
        echo "Error: Failed to create virtual environment"
        exit 1
    fi
    echo "Virtual environment created"
fi
echo ""

echo "Activating virtual environment..."
source "$PROJECT_DIR/venv/bin/activate"
echo "Virtual environment activated"
echo ""

echo "Upgrading pip..."
python3 -m pip install --upgrade pip
echo "pip upgraded"
echo ""

echo "Installing dependencies..."
if [ ! -f "$REQUIREMENTS_FILE" ]; then
    echo "Error: \"$REQUIREMENTS_FILE\" not found"
    exit 1
fi

echo "Installing from $REQUIREMENTS_FILE..."
pip install -r "$REQUIREMENTS_FILE"
if [ $? -ne 0 ]; then
    echo "Error: Failed to install dependencies"
    exit 1
fi
echo "Dependencies installed successfully!"
echo ""

# echo "Creating directories..."
# mkdir -p "$PROJECT_DIR/logs"
# mkdir -p "$PROJECT_DIR/data"
# echo "Directories created"

echo ""
echo "Running diagnostics..."
if [ -f "$SCRIPT_DIR/diagnose.py" ]; then
    python3 "$SCRIPT_DIR/diagnose.py"
else
    echo "Warning: diagnose.py not found, skipping diagnostics"
fi

echo ""
echo "========================================"
echo "Setup completed successfully!"
echo "========================================"
echo ""
echo "Next steps:"
echo "1. Activate virtual environment: source venv/bin/activate"
echo "2. Set environment variables:"
echo "   export PAYMENT_API_KEY=your_key"
echo "   export PAYMENT_API_SECRET=your_secret"
echo "3. Run tests: python3 -m pytest tests/ -v"
echo "4. Read documentation: cat README.md"
echo ""

FILE:skill_cli.py
#!/usr/bin/env python3
"""
支付 Skill 命令行入口 (v2 - 使用命令行参数)

龙虾通过 exec 工具调用此脚本来执行支付操作。

用法:
    python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001
    python skill_cli.py query_payment --transaction_id=txn_001
    python skill_cli.py refund_payment --transaction_id=txn_001 --amount=100.00
"""

import sys
import json
import os
import asyncio
import logging
import argparse
from pathlib import Path

# 配置日志：写入文件而不是 stdout，避免污染 JSON 输出
log_dir = Path(__file__).parent / "logs"
log_dir.mkdir(exist_ok=True)

logging.basicConfig(
    filename=str(log_dir / "skill.log"),
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def check_environment():
    """
    检查环境是否完整
    
    检查项：
    1. 虚拟环境是否存在
    2. 必需的依赖是否安装
    """
    project_root = Path(__file__).parent
    venv_path = project_root / "venv"
    
    # 检查虚拟环境
    if not venv_path.exists():
        print("❌ 错误: 虚拟环境未初始化", file=sys.stderr)
        print("\n请运行以下命令初始化环境:", file=sys.stderr)
        print("  Linux/Mac: ./scripts/setup.sh", file=sys.stderr)
        print("  Windows:   scripts\\setup.bat", file=sys.stderr)
        print("\n或者手动执行:", file=sys.stderr)
        print("  python -m venv venv", file=sys.stderr)
        print("  pip install -r requirements.txt", file=sys.stderr)
        sys.exit(1)
    
    # 检查必需的依赖
    try:
        import aiohttp
        import pydantic
    except ImportError as e:
        print(f"❌ 错误: 缺少依赖 - {e}", file=sys.stderr)
        print("\n请运行以下命令安装依赖:", file=sys.stderr)
        print("  Linux/Mac: ./scripts/setup.sh", file=sys.stderr)
        print("  Windows:   scripts\\setup.bat", file=sys.stderr)
        print("\n或者手动执行:", file=sys.stderr)
        print("  pip install -r requirements.txt", file=sys.stderr)
        sys.exit(1)


def create_parser():
    """创建命令行参数解析器"""
    parser = argparse.ArgumentParser(
        description='支付 Skill 命令行工具',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 创建支付
  python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001 --description="购买鼠标"
  
  # 查询支付
  python skill_cli.py query_payment --transaction_id=txn_20260315_001
  
  # 发起退款
  python skill_cli.py refund_payment --transaction_id=txn_20260315_001 --amount=100.00 --reason="商品质量问题"

环境变量:
  PAYMENT_API_KEY     - API 密钥（必需）
  PAYMENT_API_SECRET  - API 密钥（必需）
  PAYMENT_API_URL     - API 地址（可选）
  PAYMENT_ENV         - 环境 (development/production，默认 development)
        """
    )
    
    # 添加子命令
    subparsers = parser.add_subparsers(dest='command', help='可用命令')
    
    # create_payment 命令
    create_parser = subparsers.add_parser('create_payment', help='创建支付请求')
    create_parser.add_argument('--amount', type=float, required=True, help='支付金额（元）')
    create_parser.add_argument('--currency', type=str, default='CNY', help='货币代码（默认 CNY）')
    create_parser.add_argument('--merchant_id', type=str, required=True, help='商户 ID')
    create_parser.add_argument('--description', type=str, help='支付描述')
    
    # query_payment 命令
    query_parser = subparsers.add_parser('query_payment', help='查询支付状态')
    query_parser.add_argument('--transaction_id', type=str, required=True, help='交易 ID')
    
    # refund_payment 命令
    refund_parser = subparsers.add_parser('refund_payment', help='发起退款')
    refund_parser.add_argument('--transaction_id', type=str, required=True, help='原交易 ID')
    refund_parser.add_argument('--amount', type=float, help='退款金额（元，可选）')
    refund_parser.add_argument('--reason', type=str, help='退款原因')
    
    # 全局参数
    parser.add_argument('--env', type=str, choices=['development', 'production'], 
                       help='环境（默认从 PAYMENT_ENV 读取，或 development）')
    
    return parser


async def execute_tool(tool_name: str, params: dict, env: str = "development") -> dict:
    """
    执行工具
    
    Args:
        tool_name: 工具名称
        params: 工具参数
        env: 环境名称 (development 或 production)
    
    Returns:
        执行结果
    """
    # 延迟导入 PaymentSkill，确保环境检查已通过
    src_dir = Path(__file__).parent / "src"
    if str(src_dir) not in sys.path:
        sys.path.insert(0, str(src_dir))
    
    from payment_skill import PaymentSkill
    
    skill = None
    try:
        # 初始化 Skill
        skill = PaymentSkill(env=env)
        await skill.initialize()
        
        # 执行工具
        result = await skill.execute(tool_name, params)
        return {
            "success": True,
            "data": result
        }
    except Exception as e:
        logger.error(f"执行失败: {e}")
        return {
            "success": False,
            "error": str(e)
        }
    finally:
        # 清理资源
        if skill:
            try:
                await skill.cleanup()
            except:
                pass


def main():
    """主函数"""
    # 检查环境
    check_environment()
    
    parser = create_parser()
    args = parser.parse_args()
    
    # 检查是否提供了命令
    if not args.command:
        parser.print_help()
        sys.exit(1)
    
    # 确定环境
    env = args.env or os.getenv("PAYMENT_ENV", "development")
    
    # 构造参数字典
    params = {}
    if args.command == 'create_payment':
        params = {
            'amount': args.amount,
            'currency': args.currency,
            'merchant_id': args.merchant_id,
        }
        if args.description:
            params['description'] = args.description
    
    elif args.command == 'query_payment':
        params = {
            'transaction_id': args.transaction_id
        }
    
    elif args.command == 'refund_payment':
        params = {
            'transaction_id': args.transaction_id
        }
        if args.amount:
            params['amount'] = args.amount
        if args.reason:
            params['reason'] = args.reason
    
    # 执行工具
    try:
        # 兼容 Python 3.6 的异步执行方式
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(execute_tool(args.command, params, env))
        
        # 输出结果（JSON 格式）
        print(json.dumps(result, ensure_ascii=False, indent=2))
        
        # 根据结果设置退出码
        sys.exit(0 if result["success"] else 1)
        
    except KeyboardInterrupt:
        print("\n操作已取消", file=sys.stderr)
        sys.exit(130)
    except Exception as e:
        print(f"错误: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

FILE:src/__init__.py
"""
支付 Skill 包

提供 AI 原生支付能力的龙虾 Skill 实现。
"""

from .payment_skill import PaymentSkill
from .payment_api_client import PaymentAPIClient
from .security import InputValidator, DataEncryption
from .utils import (
    generate_transaction_id,
    format_currency,
    validate_amount,
    validate_currency,
)
from .skill_entry import SkillEntry, get_skill_entry

__version__ = "1.0.3"
__author__ = "Payment Institution"
__all__ = [
    "PaymentSkill",
    "PaymentAPIClient",
    "InputValidator",
    "DataEncryption",
    "generate_transaction_id",
    "format_currency",
    "validate_amount",
    "validate_currency",
    "SkillEntry",
    "get_skill_entry",
]

FILE:src/config_loader.py
"""
配置加载器 - 从 config 目录加载环境配置
"""

import os
import logging
from pathlib import Path
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)


class ConfigLoader:
    """配置加载器"""
    
    @staticmethod
    def get_config_dir() -> Path:
        """获取配置目录路径"""
        # 获取项目根目录（payment-skill-demo）
        current_file = Path(__file__)
        project_root = current_file.parent.parent
        config_dir = project_root / "config"
        return config_dir
    
    @staticmethod
    def load_env_file(env_name: str = "development") -> Dict[str, str]:
        """
        加载环境配置文件
        
        参数:
            env_name: 环境名称 (development 或 production)
        
        返回:
            配置字典
        """
        config_dir = ConfigLoader.get_config_dir()
        env_file = config_dir / f"{env_name}.env"
        
        if not env_file.exists():
            logger.warning(f"配置文件不存在: {env_file}")
            return {}
        
        config = {}
        try:
            with open(env_file, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    # 跳过注释和空行
                    if not line or line.startswith('#'):
                        continue
                    
                    # 解析 KEY=VALUE 格式
                    if '=' in line:
                        key, value = line.split('=', 1)
                        key = key.strip()
                        value = value.strip()
                        
                        # 移除引号
                        if value.startswith('"') and value.endswith('"'):
                            value = value[1:-1]
                        elif value.startswith("'") and value.endswith("'"):
                            value = value[1:-1]
                        
                        config[key] = value
            
            logger.info(f"成功加载配置文件: {env_file}")
            return config
        
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
            return {}
    
    @staticmethod
    def get_api_config(env_name: str = "development") -> Dict[str, Any]:
        """
        获取 API 配置
        
        参数:
            env_name: 环境名称
        
        返回:
            API 配置字典
        """
        # 首先从环境变量读取
        config = {
            "api_key": os.getenv("PAYMENT_API_KEY"),
            "api_secret": os.getenv("PAYMENT_API_SECRET"),
            "api_url": os.getenv("PAYMENT_API_URL"),
            "timeout": int(os.getenv("PAYMENT_API_TIMEOUT", "30")) if os.getenv("PAYMENT_API_TIMEOUT") else 30,
        }
        
        # 如果环境变量未设置，从配置文件读取
        env_config = ConfigLoader.load_env_file(env_name)
        
        if not config["api_key"]:
            config["api_key"] = env_config.get("PAYMENT_API_KEY")
        if not config["api_secret"]:
            config["api_secret"] = env_config.get("PAYMENT_API_SECRET")
        if not config["api_url"]:
            config["api_url"] = env_config.get("PAYMENT_API_URL")
        
        # PAYMENT_API_URL 有默认值
        if not config["api_url"]:
            config["api_url"] = "https://api.zlclaw.com"
        
        # 读取超时配置
        if not config["timeout"] or config["timeout"] == 30:
            timeout_str = env_config.get("PAYMENT_API_TIMEOUT", "30")
            try:
                config["timeout"] = int(timeout_str)
            except ValueError:
                config["timeout"] = 30
        
        # 验证必需的配置
        if not config["api_key"]:
            raise ValueError("PAYMENT_API_KEY 未设置")
        if not config["api_secret"]:
            raise ValueError("PAYMENT_API_SECRET 未设置")
        
        return config
    
    @staticmethod
    def get_encryption_config(env_name: str = "development") -> Optional[str]:
        """
        获取加密密钥配置
        
        参数:
            env_name: 环境名称
        
        返回:
            加密密钥或 None
        """
        # 首先从环境变量读取
        encryption_key = os.getenv("PAYMENT_ENCRYPTION_KEY")
        
        # 如果环境变量未设置，从配置文件读取
        if not encryption_key:
            env_config = ConfigLoader.load_env_file(env_name)
            encryption_key = env_config.get("PAYMENT_ENCRYPTION_KEY")
        
        return encryption_key
    
    @staticmethod
    def get_logging_config(env_name: str = "development") -> Dict[str, Any]:
        """
        获取日志配置
        
        参数:
            env_name: 环境名称
        
        返回:
            日志配置字典
        """
        # 首先从环境变量读取
        config = {
            "level": os.getenv("PAYMENT_LOG_LEVEL", "INFO"),
            "file": os.getenv("PAYMENT_LOG_FILE"),
        }
        
        # 如果环境变量未设置，从配置文件读取
        if not config["file"]:
            env_config = ConfigLoader.load_env_file(env_name)
            config["file"] = env_config.get("PAYMENT_LOG_FILE")
        
        return config

FILE:src/payment_api_client.py
"""
支付 API 客户端
"""

import aiohttp
import json
import logging
import time
import hmac
import hashlib
import base64
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)


class PaymentAPIClient:
    """支付 API 客户端"""
    
    def __init__(self, api_key: str, api_secret: str, api_url: str, timeout: int = 30):
        """
        初始化 API 客户端
        
        参数:
            api_key: API 密钥（用于身份认证）
            api_secret: API 密钥（用于请求签名）
            api_url: API 地址
            timeout: 超时时间(秒)
        """
        self.api_key = api_key
        self.api_secret = api_secret
        self.api_url = api_url.rstrip('/')
        self.timeout = timeout
        self.session: Optional[aiohttp.ClientSession] = None
    
    async def _ensure_session(self):
        """确保会话已创建"""
        if not self.session:
            self.session = aiohttp.ClientSession()
    
    async def _request(self, method: str, endpoint: str,
                      data: Dict[str, Any] = None) -> Dict[str, Any]:
        """
        发送 HTTP 请求
        
        参数:
            method: HTTP 方法
            endpoint: API 端点
            data: 请求数据
        
        返回:
            响应数据
        """
        await self._ensure_session()
        
        url = f"{self.api_url}/{endpoint}"
        
        # 生成请求签名
        signature = self._generate_signature(method, endpoint, data)
        timestamp = str(int(time.time()))
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "X-Signature": signature,
            "X-Timestamp": timestamp,
            "Content-Type": "application/json"
        }
        
        try:
            async with self.session.request(
                method, url,
                json=data,
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=self.timeout)
            ) as response:
                response_data = await response.json()
                
                if response.status == 200:
                    logger.debug(f"API 请求成功: {method} {endpoint}")
                    return response_data
                else:
                    error_msg = response_data.get("error", "未知错误")
                    logger.error(f"API 请求失败: {response.status} - {error_msg}")
                    raise Exception(f"API 错误: {response.status} - {error_msg}")
        
        except aiohttp.ClientError as e:
            logger.error(f"网络错误: {e}")
            raise Exception(f"网络错误: {str(e)}")
    
    def _generate_signature(self, method: str, endpoint: str,
                           data: Dict[str, Any] = None) -> str:
        """
        生成请求签名
        
        参数:
            method: HTTP 方法
            endpoint: API 端点
            data: 请求数据
        
        返回:
            签名字符串
        """
        timestamp = str(int(time.time()))
        message = f"{method}:{endpoint}:{timestamp}"
        
        if data:
            message += f":{json.dumps(data, sort_keys=True)}"
        
        # 使用 HMAC-SHA256 签名（使用 api_secret）
        signature = hmac.new(
            key=self.api_secret.encode(),
            msg=message.encode(),
            digestmod=hashlib.sha256
        ).hexdigest()
        
        return signature
    
    async def health_check(self) -> bool:
        """
        健康检查
        
        返回:
            是否健康
        """
        try:
            result = await self._request("GET", "health")
            return result.get("status") == "ok"
        except Exception as e:
            logger.error(f"健康检查失败: {e}")
            return False
    
    async def create_payment(self, amount: float, currency: str,
                            merchant_id: str, description: str = "") -> Dict[str, Any]:
        """
        创建支付
        
        参数:
            amount: 支付金额
            currency: 货币代码
            merchant_id: 商户 ID
            description: 支付描述
        
        返回:
            支付结果
        """
        payload = {
            "amount": amount,
            "currency": currency,
            "merchant_id": merchant_id,
            "description": description
        }
        
        result = await self._request("POST", "v1/payments", payload)
        
        return {
            "id": result.get("id"),
            "status": result.get("status"),
            "amount": result.get("amount"),
            "currency": result.get("currency"),
            "verification_url": result.get("verification_url"),
            "qr_code": result.get("qr_code"),
            "expires_at": result.get("expires_at")
        }
    
    async def query_payment(self, transaction_id: str) -> Dict[str, Any]:
        """
        查询支付状态
        
        参数:
            transaction_id: 交易 ID
        
        返回:
            支付状态
        """
        result = await self._request("GET", f"v1/payments/{transaction_id}")
        
        return {
            "id": result.get("id"),
            "status": result.get("status"),
            "amount": result.get("amount"),
            "currency": result.get("currency"),
            "created_at": result.get("created_at"),
            "completed_at": result.get("completed_at")
        }
    
    async def refund_payment(self, transaction_id: str,
                            amount: float = None) -> Dict[str, Any]:
        """
        发起退款
        
        参数:
            transaction_id: 原交易 ID
            amount: 退款金额(可选)
        
        返回:
            退款结果
        """
        payload = {}
        if amount is not None:
            payload["amount"] = amount
        
        result = await self._request(
            "POST",
            f"v1/payments/{transaction_id}/refund",
            payload
        )
        
        return {
            "id": result.get("id"),
            "status": result.get("status"),
            "amount": result.get("amount"),
            "currency": result.get("currency"),
            "created_at": result.get("created_at")
        }
    
    async def close(self):
        """关闭连接"""
        if self.session:
            await self.session.close()

FILE:src/payment_skill.py
"""
支付 Skill - 主类
"""

import asyncio
import json
import logging
from typing import Any, Dict, Optional
from datetime import datetime

try:
    # Try relative imports first (when used as a package)
    from .payment_api_client import PaymentAPIClient
    from .security import InputValidator, DataEncryption
    from .utils import generate_transaction_id, format_currency
    from .config_loader import ConfigLoader
except ImportError:
    # Fall back to absolute imports (when run directly)
    from payment_api_client import PaymentAPIClient
    from security import InputValidator, DataEncryption
    from utils import generate_transaction_id, format_currency
    from config_loader import ConfigLoader

logger = logging.getLogger(__name__)


class PaymentSkill:
    """支付 Skill - 封装支付 API"""

    def __init__(self, config: Dict[str, Any] = None, env: str = "development"):
        """
        初始化支付 Skill
        
        参数:
            config: 配置字典（如果提供，将覆盖从文件加载的配置）
            env: 环境名称 (development 或 production)
        """
        self.id = "payment_skill"
        self.name = "支付服务"
        self.description = "提供支付相关的功能"
        self.version = "1.0.0"
        self.env = env

        # 配置 - 优先使用传入的配置，否则从文件加载
        if config:
            self.config = config
        else:
            try:
                self.config = ConfigLoader.get_api_config(env)
            except ValueError as e:
                logger.warning(f"从配置文件加载失败: {e}，将使用空配置")
                self.config = {}
        
        self.api_client: Optional[PaymentAPIClient] = None
        self.encryption = None  # 延迟初始化，在 initialize() 中创建
        self.validator = InputValidator()

        # 工具注册
        self.tools = {
            "create_payment": self.create_payment,
            "query_payment": self.query_payment,
            "refund_payment": self.refund_payment,
        }

        # 状态
        self.is_initialized = False
        self.transactions = {}  # 本地缓存

    async def initialize(self):
        """初始化 Skill"""
        try:
            logger.info("初始化支付 Skill...")

            # 创建 API 客户端
            self.api_client = PaymentAPIClient(
                api_key=self.config.get("api_key"),
                api_secret=self.config.get("api_secret"),
                api_url=self.config.get("api_url"),
                timeout=self.config.get("timeout", 30)
            )

            # 健康检查
            is_healthy = await self.api_client.health_check()
            if not is_healthy:
                raise Exception("支付 API 健康检查失败")

            # 初始化加密器（如果启用）
            if self.config.get("enable_encryption"):
                try:
                    self.encryption = DataEncryption.from_env()
                    logger.info("加密器初始化成功")
                except Exception as e:
                    logger.warning(f"加密器初始化失败: {e}，将继续运行但不使用加密")

            self.is_initialized = True
            logger.info("支付 Skill 初始化成功")

        except Exception as e:
            logger.error(f"支付 Skill 初始化失败: {e}")
            raise

    async def create_payment(self, amount: float, currency: str,
                            merchant_id: str, description: str = "") -> Dict[str, Any]:
        """
        创建支付请求

        参数:
            amount: 支付金额
            currency: 货币代码 (CNY, USD, EUR)
            merchant_id: 商户 ID
            description: 支付描述

        返回:
            支付结果
        """
        try:
            logger.info(f"创建支付: {amount} {currency} 给商户 {merchant_id}")

            # 参数验证
            if amount <= 0:
                raise ValueError("金额必须大于 0")
            if amount > 1000000:
                raise ValueError("单笔金额不能超过 100 万")
            if currency not in ["CNY", "USD", "EUR"]:
                raise ValueError(f"不支持的货币: {currency}")
            if not merchant_id:
                raise ValueError("商户 ID 不能为空")

            # 生成交易 ID
            transaction_id = generate_transaction_id()

            # 调用 API
            result = await self.api_client.create_payment(
                amount=amount,
                currency=currency,
                merchant_id=merchant_id,
                description=description
            )

            # 缓存交易信息
            self.transactions[transaction_id] = {
                "amount": amount,
                "currency": currency,
                "merchant_id": merchant_id,
                "status": "pending_auth",
                "created_at": datetime.utcnow().isoformat()
            }

            # 返回结果
            response = {
                "success": True,
                "transaction_id": transaction_id,
                "status": result.get("status", "pending_auth"),
                "amount": amount,
                "currency": currency,
                "verification_url": result.get("verification_url"),
                "qr_code": result.get("qr_code"),
                "expires_at": result.get("expires_at")
            }

            logger.info(f"支付创建成功: {transaction_id}")
            return response

        except ValueError as e:
            logger.warning(f"参数验证失败: {e}")
            return {
                "success": False,
                "error": str(e),
                "error_code": "INVALID_PARAMS"
            }
        except Exception as e:
            logger.error(f"创建支付失败: {e}")
            return {
                "success": False,
                "error": "创建支付失败,请稍后重试",
                "error_code": "PAYMENT_ERROR"
            }

    async def query_payment(self, transaction_id: str) -> Dict[str, Any]:
        """
        查询支付状态

        参数:
            transaction_id: 交易 ID

        返回:
            支付状态
        """
        try:
            logger.info(f"查询支付状态: {transaction_id}")

            # 参数验证
            if not transaction_id:
                raise ValueError("交易 ID 不能为空")

            # 调用 API
            result = await self.api_client.query_payment(transaction_id)

            # 更新本地缓存
            if transaction_id in self.transactions:
                self.transactions[transaction_id]["status"] = result.get("status")

            # 返回结果
            response = {
                "success": True,
                "transaction_id": transaction_id,
                "status": result.get("status"),
                "amount": result.get("amount"),
                "currency": result.get("currency"),
                "created_at": result.get("created_at"),
                "completed_at": result.get("completed_at")
            }

            logger.info(f"支付状态查询成功: {transaction_id} - {result.get('status')}")
            return response

        except ValueError as e:
            logger.warning(f"参数验证失败: {e}")
            return {
                "success": False,
                "error": str(e),
                "error_code": "INVALID_PARAMS"
            }
        except Exception as e:
            logger.error(f"查询支付失败: {e}")
            return {
                "success": False,
                "error": "查询支付失败,请稍后重试",
                "error_code": "QUERY_ERROR"
            }

    async def refund_payment(self, transaction_id: str,
                            amount: float = None) -> Dict[str, Any]:
        """
        发起退款

        参数:
            transaction_id: 原交易 ID
            amount: 退款金额(可选,不填则全额退款)

        返回:
            退款结果
        """
        try:
            logger.info(f"发起退款: {transaction_id}, 金额: {amount}")

            # 参数验证
            if not transaction_id:
                raise ValueError("交易 ID 不能为空")

            if amount is not None and amount <= 0:
                raise ValueError("退款金额必须大于 0")

            # 调用 API
            result = await self.api_client.refund_payment(
                transaction_id=transaction_id,
                amount=amount
            )

            # 返回结果
            response = {
                "success": True,
                "refund_id": result.get("id"),
                "status": result.get("status"),
                "amount": result.get("amount"),
                "currency": result.get("currency"),
                "created_at": result.get("created_at")
            }

            logger.info(f"退款发起成功: {result.get('id')}")
            return response

        except ValueError as e:
            logger.warning(f"参数验证失败: {e}")
            return {
                "success": False,
                "error": str(e),
                "error_code": "INVALID_PARAMS"
            }
        except Exception as e:
            logger.error(f"发起退款失败: {e}")
            return {
                "success": False,
                "error": "发起退款失败,请稍后重试",
                "error_code": "REFUND_ERROR"
            }

    async def execute(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        执行工具

        参数:
            tool_name: 工具名称
            params: 工具参数

        返回:
            执行结果
        """
        if not self.is_initialized:
            return {
                "success": False,
                "error": "Skill 未初始化",
                "error_code": "NOT_INITIALIZED"
            }

        if tool_name not in self.tools:
            return {
                "success": False,
                "error": f"未知的工具: {tool_name}",
                "error_code": "UNKNOWN_TOOL"
            }

        try:
            tool_func = self.tools[tool_name]
            result = await tool_func(**params)
            return result
        except Exception as e:
            logger.error(f"执行工具 {tool_name} 失败: {e}")
            return {
                "success": False,
                "error": "工具执行失败",
                "error_code": "EXECUTION_ERROR"
            }

    async def cleanup(self):
        """清理资源"""
        try:
            logger.info("清理支付 Skill...")
            if self.api_client:
                await self.api_client.close()
            self.is_initialized = False
            logger.info("支付 Skill 清理完成")
        except Exception as e:
            logger.error(f"清理支付 Skill 失败: {e}")

    def get_tool_schema(self, tool_name: str) -> Dict[str, Any]:
        """获取工具的 JSON Schema"""
        schemas = {
            "create_payment": {
                "type": "object",
                "properties": {
                    "amount": {
                        "type": "number",
                        "description": "支付金额",
                        "minimum": 0.01,
                        "maximum": 1000000
                    },
                    "currency": {
                        "type": "string",
                        "description": "货币代码",
                        "enum": ["CNY", "USD", "EUR"]
                    },
                    "merchant_id": {
                        "type": "string",
                        "description": "商户 ID"
                    },
                    "description": {
                        "type": "string",
                        "description": "支付描述"
                    }
                },
                "required": ["amount", "currency", "merchant_id"]
            },
            "query_payment": {
                "type": "object",
                "properties": {
                    "transaction_id": {
                        "type": "string",
                        "description": "交易 ID"
                    }
                },
                "required": ["transaction_id"]
            },
            "refund_payment": {
                "type": "object",
                "properties": {
                    "transaction_id": {
                        "type": "string",
                        "description": "原交易 ID"
                    },
                    "amount": {
                        "type": "number",
                        "description": "退款金额(可选)"
                    }
                },
                "required": ["transaction_id"]
            }
        }
        return schemas.get(tool_name, {})

FILE:src/payment_skill.yaml
id: payment_skill
name: 支付服务
description: 提供支付相关的功能,支持创建支付、查询状态、发起退款
version: 1.0.0
author: zlclaw Team

# 依赖
dependencies:
  - aiohttp>=3.8.0
  - pydantic>=2.0.0
  - pyyaml>=6.0

# 必需环境变量
required_env_vars:
  - PAYMENT_API_KEY: "API 密钥，用于身份认证"
  - PAYMENT_API_SECRET: "API 密钥，用于请求签名"
  - PAYMENT_API_URL: "支付 API 地址"

# 可选环境变量
optional_env_vars:
  - PAYMENT_ENCRYPTION_KEY: "32 字节加密密钥（可选）"
  - PAYMENT_LOG_LEVEL: "日志级别，默认 INFO"
  - PAYMENT_LOG_FILE: "日志文件路径"
  - PAYMENT_ENABLE_CROSS_DEVICE_AUTH: "启用跨设备验证，默认 true"

# 配置
config:
  api_key: PAYMENT_API_KEY
  api_url: https://api.alipay.com
  timeout: 30
  retry_count: 3
  enable_logging: true
  enable_encryption: false

# 工具列表
tools:
  - name: create_payment
    description: 创建支付请求
    timeout: 10000
    rate_limit:
      requests_per_minute: 60
    requires_approval: false

  - name: query_payment
    description: 查询支付状态
    timeout: 5000
    rate_limit:
      requests_per_minute: 100
    requires_approval: false

  - name: refund_payment
    description: 发起退款
    timeout: 10000
    rate_limit:
      requests_per_minute: 30
    requires_approval: true

# 权限要求
permissions:
  - network.http
  - storage.read
  - storage.write
  - crypto.sign

# 健康检查
health_check:
  enabled: true
  interval: 60
  timeout: 5000

# 日志配置
logging:
  level: INFO
  format: json
  output: stdout

FILE:src/security.py
"""
安全工具
"""

import json
import re
from typing import Dict, Any
from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes
import base64


class InputValidator:
    """输入验证器"""
    
    @staticmethod
    def validate_and_sanitize(params: Dict[str, Any],
                             schema: Dict[str, Any]) -> Dict[str, Any]:
        """
        验证和清理输入参数
        
        参数:
            params: 输入参数
            schema: JSON Schema
        
        返回:
            验证后的参数
        """
        validated = {}
        
        for field, field_schema in schema.get("properties", {}).items():
            if field not in params:
                if field in schema.get("required", []):
                    raise ValueError(f"缺少必需字段: {field}")
                continue
            
            value = params[field]
            field_type = field_schema.get("type")
            
            # 类型检查
            if field_type == "string":
                if not isinstance(value, str):
                    raise ValueError(f"字段 {field} 必须是字符串")
                # 防止 SQL 注入
                value = InputValidator.sanitize_string(value)
            
            elif field_type == "number":
                if not isinstance(value, (int, float)):
                    raise ValueError(f"字段 {field} 必须是数字")
                # 检查范围
                if "minimum" in field_schema and value < field_schema["minimum"]:
                    raise ValueError(f"字段 {field} 小于最小值")
                if "maximum" in field_schema and value > field_schema["maximum"]:
                    raise ValueError(f"字段 {field} 大于最大值")
            
            elif field_type == "object":
                if not isinstance(value, dict):
                    raise ValueError(f"字段 {field} 必须是对象")
            
            validated[field] = value
        
        return validated
    
    @staticmethod
    def sanitize_string(value: str) -> str:
        """
        清理字符串,防止注入攻击
        
        参数:
            value: 输入字符串
        
        返回:
            清理后的字符串
        """
        # 移除危险字符
        dangerous_patterns = [
            r"'",
            r'"',
            r";",
            r"--",
            r"/\*",
            r"\*/",
            r"xp_",
            r"sp_"
        ]
        
        for pattern in dangerous_patterns:
            value = re.sub(pattern, "", value)
        
        # 限制长度
        max_length = 1000
        if len(value) > max_length:
            value = value[:max_length]
        
        return value


class DataEncryption:
    """数据加密"""
    
    def __init__(self, key: bytes):
        """
        初始化加密器
        
        参数:
            key: 加密密钥(32 字节)，生产环境必须提供
        """
        if key is None or len(key) != 32:
            raise ValueError("加密密钥必须为 32 字节，生产环境必须提供有效密钥")
        self.key = key
    
    @classmethod
    def from_env(cls, env_key: str = "PAYMENT_ENCRYPTION_KEY") -> "DataEncryption":
        """
        从环境变量创建加密器
        
        参数:
            env_key: 环境变量名称
        """
        import os
        key_str = os.environ.get(env_key)
        if not key_str:
            raise ValueError(f"环境变量 {env_key} 未设置，生产环境必须提供加密密钥")
        key_bytes = key_str.encode()[:32].ljust(32, b'0')
        return cls(key_bytes)
    
    def encrypt_sensitive_data(self, data: Dict[str, Any]) -> str:
        """
        加密敏感数据
        
        参数:
            data: 数据字典
        
        返回:
            加密后的数据(Base64 编码)
        """
        # 标记敏感字段
        sensitive_fields = ["api_key", "password", "token", "card_number"]
        
        encrypted_data = data.copy()
        
        for field in sensitive_fields:
            if field in encrypted_data:
                # 加密敏感字段
                plaintext = str(encrypted_data[field]).encode()
                
                # 生成随机 nonce
                nonce = get_random_bytes(16)
                
                # 创建加密器
                cipher = AES.new(self.key, AES.MODE_GCM, nonce=nonce)
                
                # 加密
                ciphertext, tag = cipher.encrypt_and_digest(plaintext)
                
                # 存储加密后的数据
                encrypted_data[field] = {
                    "encrypted": base64.b64encode(ciphertext).decode(),
                    "nonce": base64.b64encode(nonce).decode(),
                    "tag": base64.b64encode(tag).decode()
                }
        
        return json.dumps(encrypted_data)
    
    def decrypt_sensitive_data(self, encrypted_data: str) -> Dict[str, Any]:
        """
        解密敏感数据
        
        参数:
            encrypted_data: 加密后的数据(JSON 字符串)
        
        返回:
            解密后的数据
        """
        data = json.loads(encrypted_data)
        
        for field, value in data.items():
            if isinstance(value, dict) and "encrypted" in value:
                # 解密敏感字段
                ciphertext = base64.b64decode(value["encrypted"])
                nonce = base64.b64decode(value["nonce"])
                tag = base64.b64decode(value["tag"])
                
                # 创建解密器
                cipher = AES.new(self.key, AES.MODE_GCM, nonce=nonce)
                
                # 解密
                plaintext = cipher.decrypt_and_verify(ciphertext, tag)
                data[field] = plaintext.decode()
        
        return data

FILE:src/utils.py
"""
工具函数模块

提供支付 Skill 所需的各种工具函数。
"""

import uuid
import re
from datetime import datetime
from typing import Optional, Tuple
from decimal import Decimal, InvalidOperation


def generate_transaction_id(prefix: str = "txn") -> str:
    """
    生成唯一的交易 ID。
    
    Args:
        prefix: 交易 ID 前缀，默认为 "txn"
    
    Returns:
        格式为 "{prefix}_{timestamp}_{uuid}" 的交易 ID
    
    Example:
        >>> txn_id = generate_transaction_id()
        >>> print(txn_id)
        txn_20260315_550e8400e29b41d4a716446655440000
    """
    timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")
    unique_id = str(uuid.uuid4()).replace("-", "")[:16]
    return f"{prefix}_{timestamp}_{unique_id}"


def generate_verification_id(prefix: str = "ver") -> str:
    """
    生成跨设备验证 ID。
    
    Args:
        prefix: 验证 ID 前缀，默认为 "ver"
    
    Returns:
        格式为 "{prefix}_{uuid}" 的验证 ID
    """
    unique_id = str(uuid.uuid4()).replace("-", "")[:16]
    return f"{prefix}_{unique_id}"


def format_currency(amount: float, currency: str = "CNY") -> str:
    """
    格式化货币显示。
    
    Args:
        amount: 金额
        currency: 货币代码
    
    Returns:
        格式化后的货币字符串
    
    Example:
        >>> format_currency(299.00, "CNY")
        '¥299.00'
        >>> format_currency(99.99, "USD")
        '$99.99'
    """
    currency_symbols = {
        "CNY": "¥",
        "USD": "$",
        "EUR": "€",
        "GBP": "£",
        "JPY": "¥",
    }
    symbol = currency_symbols.get(currency, currency)
    return f"{symbol}{amount:.2f}"


def validate_amount(amount: float) -> Tuple[bool, Optional[str]]:
    """
    验证支付金额。
    
    Args:
        amount: 支付金额
    
    Returns:
        (是否有效, 错误信息)
    
    Example:
        >>> validate_amount(299.00)
        (True, None)
        >>> validate_amount(-100)
        (False, '金额必须大于0')
        >>> validate_amount(299.999)
        (False, '金额最多2位小数')
    """
    try:
        # 转换为 Decimal 以精确处理浮点数
        decimal_amount = Decimal(str(amount))
        
        # 检查金额范围
        if decimal_amount <= 0:
            return False, "金额必须大于0"
        
        if decimal_amount > 1000000:
            return False, "金额不能超过1,000,000"
        
        # 检查小数位数
        if decimal_amount.as_tuple().exponent < -2:
            return False, "金额最多2位小数"
        
        return True, None
    except (InvalidOperation, ValueError):
        return False, "金额格式无效"


def validate_currency(currency: str) -> Tuple[bool, Optional[str]]:
    """
    验证货币代码。
    
    Args:
        currency: ISO 4217 货币代码
    
    Returns:
        (是否有效, 错误信息)
    
    Example:
        >>> validate_currency("CNY")
        (True, None)
        >>> validate_currency("INVALID")
        (False, '无效的货币代码')
    """
    valid_currencies = {
        "CNY", "USD", "EUR", "GBP", "JPY", "AUD", "CAD", "CHF",
        "HKD", "SGD", "INR", "KRW", "MXN", "NZD", "ZAR"
    }
    
    if not isinstance(currency, str):
        return False, "货币代码必须是字符串"
    
    currency_upper = currency.upper()
    
    if currency_upper not in valid_currencies:
        return False, f"无效的货币代码: {currency}"
    
    return True, None


def validate_merchant_id(merchant_id: str) -> Tuple[bool, Optional[str]]:
    """
    验证商户 ID。
    
    Args:
        merchant_id: 商户 ID
    
    Returns:
        (是否有效, 错误信息)
    """
    if not isinstance(merchant_id, str):
        return False, "商户 ID 必须是字符串"
    
    if len(merchant_id) == 0:
        return False, "商户 ID 不能为空"
    
    if len(merchant_id) > 255:
        return False, "商户 ID 长度不能超过255"
    
    # 只允许字母、数字、下划线和连字符
    if not re.match(r"^[a-zA-Z0-9_-]+$", merchant_id):
        return False, "商户 ID 只能包含字母、数字、下划线和连字符"
    
    return True, None


def validate_description(description: str) -> Tuple[bool, Optional[str]]:
    """
    验证支付描述。
    
    Args:
        description: 支付描述
    
    Returns:
        (是否有效, 错误信息)
    """
    if not isinstance(description, str):
        return False, "描述必须是字符串"
    
    if len(description) == 0:
        return False, "描述不能为空"
    
    if len(description) > 500:
        return False, "描述长度不能超过500"
    
    return True, None


def validate_reference_id(reference_id: str) -> Tuple[bool, Optional[str]]:
    """
    验证幂等性标识。
    
    Args:
        reference_id: 幂等性标识
    
    Returns:
        (是否有效, 错误信息)
    """
    if not isinstance(reference_id, str):
        return False, "幂等性标识必须是字符串"
    
    if len(reference_id) == 0:
        return False, "幂等性标识不能为空"
    
    if len(reference_id) > 255:
        return False, "幂等性标识长度不能超过255"
    
    # UUID 或订单号格式
    if not re.match(r"^[a-zA-Z0-9_-]+$", reference_id):
        return False, "幂等性标识只能包含字母、数字、下划线和连字符"
    
    return True, None


def parse_timestamp(timestamp: int) -> str:
    """
    将时间戳转换为可读的日期时间字符串。
    
    Args:
        timestamp: Unix 时间戳（秒）
    
    Returns:
        格式为 "YYYY-MM-DD HH:MM:SS" 的日期时间字符串
    
    Example:
        >>> parse_timestamp(1710489600)
        '2026-03-15 12:00:00'
    """
    try:
        dt = datetime.utcfromtimestamp(timestamp)
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    except (ValueError, OSError):
        return "Invalid timestamp"


def get_current_timestamp() -> int:
    """
    获取当前 Unix 时间戳。
    
    Returns:
        当前时间戳（秒）
    """
    return int(datetime.utcnow().timestamp())


def calculate_expiry_time(ttl_seconds: int) -> int:
    """
    计算过期时间戳。
    
    Args:
        ttl_seconds: 生存时间（秒）
    
    Returns:
        过期时间戳
    
    Example:
        >>> current = get_current_timestamp()
        >>> expiry = calculate_expiry_time(300)
        >>> expiry - current
        300
    """
    return get_current_timestamp() + ttl_seconds


def is_expired(expiry_time: int) -> bool:
    """
    检查是否已过期。
    
    Args:
        expiry_time: 过期时间戳
    
    Returns:
        是否已过期
    """
    return get_current_timestamp() > expiry_time


def truncate_string(text: str, max_length: int = 100) -> str:
    """
    截断字符串以防止日志过长。
    
    Args:
        text: 原始字符串
        max_length: 最大长度
    
    Returns:
        截断后的字符串
    """
    if len(text) <= max_length:
        return text
    return text[:max_length - 3] + "..."


def mask_sensitive_data(data: str, show_chars: int = 4) -> str:
    """
    掩盖敏感数据（如 API 密钥）。
    
    Args:
        data: 原始数据
        show_chars: 显示的字符数
    
    Returns:
        掩盖后的数据
    
    Example:
        >>> mask_sensitive_data("sk_live_abc123def456", 4)
        'sk_l...456'
    """
    if len(data) <= show_chars * 2:
        return "*" * len(data)
    
    start = data[:show_chars]
    end = data[-show_chars:]
    middle = "*" * (len(data) - show_chars * 2)
    return f"{start}{middle}{end}"

---
name: payment_skill_demo
description: AI 原生支付解决方案 - 提供安全、可靠的支付功能
version: 1.0.3
author: zlclaw_dev
license: MIT
tags:
  - payment
  - finance
  - commerce
skill_type: code
has_server: false
has_install_scripts: true
security_level: high
requires:
  - python>=3.6
environment_variables:
  required:
    - PAYMENT_API_KEY: "API 密钥，用于身份认证"
    - PAYMENT_API_SECRET: "API 密钥，用于请求签名"
  optional:
    - PAYMENT_ENCRYPTION_KEY: "32 字节加密密钥（可选）"
    - PAYMENT_LOG_LEVEL: "日志级别，默认 INFO"
tools:
  - name: create_payment
    description: 创建支付请求，触发用户授权流程
    parameters:
      - name: amount
        type: number
        description: 支付金额，最多2位小数
        required: true
        example: 299.00
      - name: currency
        type: string
        description: ISO 4217 货币代码
        required: false
        default: CNY
        example: CNY
      - name: merchant_id
        type: string
        description: 商户唯一标识
        required: true
        example: taobao_merchant_001
      - name: description
        type: string
        description: 支付描述，用户可见
        required: false
        example: 购买无线鼠标
    returns:
      type: object
      properties:
        success:
          type: boolean
          description: 是否成功
        transaction_id:
          type: string
          description: 交易ID
        status:
          type: string
          description: 支付状态
        amount:
          type: number
          description: 支付金额
        currency:
          type: string
          description: 货币代码
        verification_url:
          type: string
          description: 验证URL
        qr_code:
          type: string
          description: 二维码数据
  
  - name: query_payment
    description: 查询支付状态
    parameters:
      - name: transaction_id
        type: string
        description: 交易 ID
        required: true
        example: txn_20260315_001
    returns:
      type: object
      properties:
        success:
          type: boolean
          description: 是否成功
        transaction_id:
          type: string
          description: 交易ID
        status:
          type: string
          description: 支付状态
        amount:
          type: number
          description: 支付金额
  
  - name: refund_payment
    description: 发起退款
    parameters:
      - name: transaction_id
        type: string
        description: 原交易 ID
        required: true
        example: txn_20260315_001
      - name: amount
        type: number
        description: 退款金额（可选，不填则全额退款）
        required: false
        example: 100.00
      - name: reason
        type: string
        description: 退款原因
        required: false
        example: 商品质量问题
    returns:
      type: object
      properties:
        success:
          type: boolean
          description: 是否成功
        refund_id:
          type: string
          description: 退款ID
        status:
          type: string
          description: 退款状态
        amount:
          type: number
          description: 退款金额
---

# 支付 Skill

AI 原生支付解决方案，提供安全、可靠的支付功能。

## 工具调用方式

AI 通过执行 `skill_cli.py` 脚本来调用工具：

```bash
python skill_cli.py <tool_name> --param1=value1 --param2=value2
```

**示例**：
```bash
# 创建支付
python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001 --description="购买商品"

# 查询支付
python skill_cli.py query_payment --transaction_id=txn_001

# 发起退款
python skill_cli.py refund_payment --transaction_id=txn_001 --amount=100.00 --reason="质量问题"
```

**返回格式**：所有工具都返回 JSON 格式的结果
```json
{
  "success": true,
  "data": { ... }
}
```

## 快速开始

### 1. 安装依赖

```bash
# Linux/macOS
./scripts/setup.sh

# Windows
scripts\setup.bat
```

### 2. 配置环境变量

**必需**：
```bash
export PAYMENT_API_KEY=your_api_key
export PAYMENT_API_SECRET=your_api_secret
```


### 3. 使用示例

```bash
# 创建支付
python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001

# 查询支付
python skill_cli.py query_payment --transaction_id=txn_001

# 发起退款
python skill_cli.py refund_payment --transaction_id=txn_001 --amount=100.00
```

## 核心功能

- **create_payment**: 创建支付请求
- **query_payment**: 查询支付状态
- **refund_payment**: 发起退款

## 安全特性

- TLS 加密通信
- HMAC-SHA256 签名
- 参数验证
- 审计日志

## 技术要求

- Python 3.6+
- 网络访问权限
- 环境变量配置

## 支持

- 📧 技术支持: [email protected]
- 🐛 问题报告: https://github.com/zlclaw/payment-skill-demo/issues

---

**版本**: 1.0.3 | **许可证**: MIT

FILE:_meta.json
{
  "ownerId": "kn79xswxbje4zn43vw5yn021a180bsan",
  "slug": "payment-skill-demo",
  "version": "1.0.3",
  "publishedAt": 1710768000000,
  "requiredEnvVars": [
    "PAYMENT_API_KEY",
    "PAYMENT_API_SECRET"
  ],
  "optionalEnvVars": [
    "PAYMENT_ENCRYPTION_KEY",
    "PAYMENT_LOG_LEVEL"
  ]
}

FILE:config/production.env
# 支付 Skill 生产环境配置
# 
# 重要：API 凭证必须通过环境变量设置，不要在此文件中硬编码
# 
# 设置方式：
# export PAYMENT_API_KEY=your_real_key
# export PAYMENT_API_SECRET=your_real_secret
#
# PAYMENT_API_URL 有默认值 https://api.zlclaw.com，不需要设置
#
# 如果未设置必需的环境变量，Skill 会在初始化时报错

# ============================================
# 必需配置 (Required) - 通过环境变量设置
# ============================================

# API 配置（必须通过环境变量设置）
# 不要在此文件中设置 PAYMENT_API_KEY、PAYMENT_API_SECRET
# 这些凭证应该通过环境变量设置，以保护敏感信息
PAYMENT_API_URL=https://api.zlclaw.com
PAYMENT_API_TIMEOUT=30

# ============================================
# 可选配置 (Optional)
# ============================================

# 安全配置
PAYMENT_ENABLE_SIGNATURE=true
PAYMENT_ENABLE_ENCRYPTION=false
PAYMENT_TLS_VERIFY=true


# 功能配置
PAYMENT_ENABLE_CROSS_DEVICE_AUTH=true
PAYMENT_POLLING_INTERVAL=2000
PAYMENT_MAX_POLLING_ATTEMPTS=60

FILE:scripts/diagnose.py
#!/usr/bin/env python3
"""
支付 Skill 环境诊断工具

用于诊断 Python 3.6 环境中的依赖问题。
"""

import sys
import subprocess
import importlib
from pathlib import Path


def print_header(text):
    """打印标题"""
    print(f"\n{'='*60}")
    print(f"  {text}")
    print(f"{'='*60}\n")


def print_success(text):
    """打印成功信息"""
    print(f"✅ {text}")


def print_error(text):
    """打印错误信息"""
    print(f"❌ {text}")


def print_warning(text):
    """打印警告信息"""
    print(f"⚠️  {text}")


def print_info(text):
    """打印信息"""
    print(f"ℹ️  {text}")


def check_python_version():
    """检查 Python 版本"""
    print_header("Python 版本检查")
    
    version = sys.version_info
    version_str = f"{version.major}.{version.minor}.{version.micro}"
    
    print(f"Python 版本: {version_str}")
    print(f"Python 路径: {sys.executable}")
    
    if version.major == 3 and version.minor == 6:
        print_warning("Python 3.6 已停止支持，建议升级到 Python 3.8+")
        return True
    elif version.major == 3 and version.minor >= 8:
        print_success("Python 版本支持良好")
        return True
    else:
        print_error(f"不支持的 Python 版本: {version_str}")
        return False


def check_pip_version():
    """检查 pip 版本"""
    print_header("pip 版本检查")
    
    try:
        result = subprocess.run(
            [sys.executable, "-m", "pip", "--version"],
            capture_output=True,
            text=True
        )
        pip_version = result.stdout.strip()
        print(f"pip 版本: {pip_version}")
        
        # 提取版本号
        if "pip" in pip_version:
            print_success("pip 已安装")
            return True
    except Exception as e:
        print_error(f"无法检查 pip 版本: {e}")
        return False


def check_dependencies():
    """检查依赖"""
    print_header("依赖检查")
    
    dependencies = {
        "aiohttp": "异步 HTTP 客户端",
        "pydantic": "数据验证",
        "pyyaml": "YAML 配置",
        "python-dotenv": "环境变量管理",
        "cryptography": "加密库",
        "Crypto": "加密库 (pycryptodome)",
        "pytest": "测试框架",
        "pytest_asyncio": "异步测试支持",
        "redis": "缓存库",
    }
    
    installed = {}
    missing = []
    
    for module_name, description in dependencies.items():
        try:
            module = importlib.import_module(module_name)
            version = getattr(module, "__version__", "unknown")
            installed[module_name] = version
            print_success(f"{module_name} ({description}): {version}")
        except ImportError:
            missing.append((module_name, description))
            print_error(f"{module_name} ({description}): 未安装")
    
    return len(missing) == 0, missing


def check_asyncmock():
    """检查 AsyncMock 兼容性"""
    print_header("AsyncMock 兼容性检查")
    
    version = sys.version_info
    
    if version.major == 3 and version.minor >= 8:
        try:
            from unittest.mock import AsyncMock
            print_success("AsyncMock 原生支持 (Python 3.8+)")
            return True
        except ImportError:
            print_error("AsyncMock 导入失败")
            return False
    else:
        print_info("Python 3.6/3.7 使用兼容实现")
        try:
            from unittest.mock import Mock
            
            # 测试兼容实现
            class AsyncMock(Mock):
                async def __call__(self, *args, **kwargs):
                    return super(AsyncMock, self).__call__(*args, **kwargs)
            
            print_success("AsyncMock 兼容实现可用")
            return True
        except Exception as e:
            print_error(f"AsyncMock 兼容实现失败: {e}")
            return False


def check_project_structure():
    """检查项目结构"""
    print_header("项目结构检查")
    
    # 获取脚本目录和项目目录
    script_dir = Path(__file__).parent
    project_dir = script_dir.parent
    
    required_files = [
        "src/payment_skill.py",
        "src/payment_api_client.py",
        "src/security.py",
        "src/utils.py",
        "tests/test_payment_skill.py",
        "tests/test_integration.py",
        "scripts/requirements-py36.txt",
        "scripts/payment_skill.yaml",
    ]
    
    all_exist = True
    for file_path in required_files:
        full_path = project_dir / file_path
        if full_path.exists():
            print_success(f"✓ {file_path}")
        else:
            print_error(f"✗ {file_path} (缺失)")
            all_exist = False
    
    return all_exist


def check_requirements_file():
    """检查 requirements 文件"""
    print_header("Requirements 文件检查")
    
    # 获取脚本目录和项目目录
    script_dir = Path(__file__).parent
    project_dir = script_dir.parent
    
    py36_req = project_dir / "scripts" / "requirements-py36.txt"
    std_req = project_dir / "scripts" / "requirements.txt"
    
    if py36_req.exists():
        print_success(f"✓ requirements-py36.txt 存在")
        print_info("Python 3.6 环境应使用此文件")
    else:
        print_error("✗ requirements-py36.txt 缺失")
    
    if std_req.exists():
        print_success(f"✓ requirements.txt 存在")
        print_info("Python 3.8+ 环境应使用此文件")
    else:
        print_error("✗ requirements.txt 缺失")


def check_environment_variables():
    """检查环境变量"""
    print_header("环境变量检查")
    
    import os
    
    env_vars = [
        "PAYMENT_API_KEY",
        "PAYMENT_API_SECRET",
        "PAYMENT_API_URL",
        "PAYMENT_LOG_LEVEL",
    ]
    
    for var in env_vars:
        value = os.getenv(var)
        if value:
            # 隐藏敏感信息
            if "SECRET" in var or "KEY" in var:
                display_value = f"{value[:4]}...{value[-4:]}"
            else:
                display_value = value
            print_success(f"{var}: {display_value}")
        else:
            print_warning(f"{var}: 未设置")


def run_diagnostics():
    """运行所有诊断"""
    print("\n")
    print("╔" + "="*58 + "╗")
    print("║" + " "*58 + "║")
    print("║" + "  支付 Skill 环境诊断工具".center(58) + "║")
    print("║" + " "*58 + "║")
    print("╚" + "="*58 + "╝")
    
    results = {
        "Python 版本": check_python_version(),
        "pip 版本": check_pip_version(),
        "AsyncMock 兼容性": check_asyncmock(),
        "项目结构": check_project_structure(),
    }
    
    check_dependencies()
    check_requirements_file()
    check_environment_variables()
    
    # 总结
    print_header("诊断总结")
    
    all_passed = all(results.values())
    
    if all_passed:
        print_success("所有检查通过！")
        print_info("您可以运行: python3 -m pytest tests/ -v")
    else:
        print_error("发现问题，请参考上面的错误信息")
        print_info("查看 PYTHON36_SETUP.md 获取详细解决方案")
    
    print("\n")


if __name__ == "__main__":
    run_diagnostics()

FILE:scripts/requirements-py36.txt
# Python 3.6 兼容版本
# 注意: Python 3.6 已停止支持,建议升级到 Python 3.8+

# 异步 HTTP 客户端 (3.6 兼容版本)
aiohttp==3.6.3
multidict==4.7.6
yarl==1.4.2

# 数据验证
pydantic==1.8.2

# YAML 配置
pyyaml==5.4.1

# 环境变量管理
python-dotenv==0.19.0

# 加密库
cryptography==3.4.8
pycryptodome==3.10.4

# 测试框架
pytest==6.2.5
pytest-asyncio==0.14.0
pytest-cov==2.12.1

# 缓存
redis==3.5.3

FILE:scripts/requirements.txt
# 核心依赖（固定版本以防止供应链攻击）
aiohttp==3.9.5
pydantic==2.7.0
pyyaml==6.0.1
python-dotenv==1.0.1
cryptography==42.0.5
pycryptodome==3.19.1

# 测试依赖
pytest==8.2.0
pytest-asyncio==0.23.6
pytest-cov==5.0.0

# 可选依赖
redis==5.0.1

# 安全扫描工具（可选安装）
# safety==2.3.5
# bandit==1.7.9

FILE:scripts/setup.sh
#!/bin/bash

# Payment Skill Setup Script (Linux/macOS)
# Auto-detect Python version and install dependencies

set -e

SCRIPT_DIR="$(cd "$(dirname "BASH_SOURCE[0]")" && pwd)"
PROJECT_DIR="$(dirname "$SCRIPT_DIR")"

echo ""
echo "========================================"
echo "Payment Skill Setup"
echo "========================================"
echo ""

echo "Checking Python version..."
python3 --version 2>/dev/null || {
    echo "Error: Python 3 is not installed"
    exit 1
}

# Detect Python version
PYTHON_VERSION_TYPE=$(python3 -c "import sys; print('py38' if sys.version_info >= (3, 8) else 'py36')")

if [ "$PYTHON_VERSION_TYPE" = "py38" ]; then
    echo "Python 3.8+ detected (recommended)"
    REQUIREMENTS_FILE="$SCRIPT_DIR/requirements.txt"
else
    echo "Python 3.6 detected (deprecated)"
    REQUIREMENTS_FILE="$SCRIPT_DIR/requirements-py36.txt"
fi

echo ""
echo "Checking virtual environment..."
if [ -d "$PROJECT_DIR/venv" ]; then
    echo "Virtual environment already exists"
else
    echo "Creating virtual environment..."
    python3 -m venv "$PROJECT_DIR/venv"
    if [ $? -ne 0 ]; then
        echo "Error: Failed to create virtual environment"
        exit 1
    fi
    echo "Virtual environment created"
fi
echo ""

echo "Activating virtual environment..."
source "$PROJECT_DIR/venv/bin/activate"
echo "Virtual environment activated"
echo ""

echo "Upgrading pip..."
python3 -m pip install --upgrade pip
echo "pip upgraded"
echo ""

echo "Installing dependencies..."
if [ ! -f "$REQUIREMENTS_FILE" ]; then
    echo "Error: \"$REQUIREMENTS_FILE\" not found"
    exit 1
fi

echo "Installing from $REQUIREMENTS_FILE..."
pip install -r "$REQUIREMENTS_FILE"
if [ $? -ne 0 ]; then
    echo "Error: Failed to install dependencies"
    exit 1
fi
echo "Dependencies installed successfully!"
echo ""

# echo "Creating directories..."
# mkdir -p "$PROJECT_DIR/logs"
# mkdir -p "$PROJECT_DIR/data"
# echo "Directories created"

echo ""
echo "Running diagnostics..."
if [ -f "$SCRIPT_DIR/diagnose.py" ]; then
    python3 "$SCRIPT_DIR/diagnose.py"
else
    echo "Warning: diagnose.py not found, skipping diagnostics"
fi

echo ""
echo "========================================"
echo "Setup completed successfully!"
echo "========================================"
echo ""
echo "Next steps:"
echo "1. Activate virtual environment: source venv/bin/activate"
echo "2. Set environment variables:"
echo "   export PAYMENT_API_KEY=your_key"
echo "   export PAYMENT_API_SECRET=your_secret"
echo "3. Run tests: python3 -m pytest tests/ -v"
echo "4. Read documentation: cat README.md"
echo ""

FILE:skill_cli.py
#!/usr/bin/env python3
"""
支付 Skill 命令行入口 (v2 - 使用命令行参数)

龙虾通过 exec 工具调用此脚本来执行支付操作。

用法:
    python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001
    python skill_cli.py query_payment --transaction_id=txn_001
    python skill_cli.py refund_payment --transaction_id=txn_001 --amount=100.00
"""

import sys
import json
import os
import asyncio
import logging
import argparse
from pathlib import Path

# 配置日志：写入文件而不是 stdout，避免污染 JSON 输出
log_dir = Path(__file__).parent / "logs"
log_dir.mkdir(exist_ok=True)

logging.basicConfig(
    filename=str(log_dir / "skill.log"),
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def check_environment():
    """
    检查环境是否完整
    
    检查项：
    1. 虚拟环境是否存在
    2. 必需的依赖是否安装
    """
    project_root = Path(__file__).parent
    venv_path = project_root / "venv"
    
    # 检查虚拟环境
    if not venv_path.exists():
        print("❌ 错误: 虚拟环境未初始化", file=sys.stderr)
        print("\n请运行以下命令初始化环境:", file=sys.stderr)
        print("  Linux/Mac: ./scripts/setup.sh", file=sys.stderr)
        print("  Windows:   scripts\\setup.bat", file=sys.stderr)
        print("\n或者手动执行:", file=sys.stderr)
        print("  python -m venv venv", file=sys.stderr)
        print("  pip install -r requirements.txt", file=sys.stderr)
        sys.exit(1)
    
    # 检查必需的依赖
    try:
        import aiohttp
        import pydantic
    except ImportError as e:
        print(f"❌ 错误: 缺少依赖 - {e}", file=sys.stderr)
        print("\n请运行以下命令安装依赖:", file=sys.stderr)
        print("  Linux/Mac: ./scripts/setup.sh", file=sys.stderr)
        print("  Windows:   scripts\\setup.bat", file=sys.stderr)
        print("\n或者手动执行:", file=sys.stderr)
        print("  pip install -r requirements.txt", file=sys.stderr)
        sys.exit(1)


def create_parser():
    """创建命令行参数解析器"""
    parser = argparse.ArgumentParser(
        description='支付 Skill 命令行工具',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 创建支付
  python skill_cli.py create_payment --amount=299.00 --currency=CNY --merchant_id=taobao_001 --description="购买鼠标"
  
  # 查询支付
  python skill_cli.py query_payment --transaction_id=txn_20260315_001
  
  # 发起退款
  python skill_cli.py refund_payment --transaction_id=txn_20260315_001 --amount=100.00 --reason="商品质量问题"

环境变量:
  PAYMENT_API_KEY     - API 密钥（必需）
  PAYMENT_API_SECRET  - API 密钥（必需）
  PAYMENT_API_URL     - API 地址（可选）
  PAYMENT_ENV         - 环境 (development/production，默认 development)
        """
    )
    
    # 添加子命令
    subparsers = parser.add_subparsers(dest='command', help='可用命令')
    
    # create_payment 命令
    create_parser = subparsers.add_parser('create_payment', help='创建支付请求')
    create_parser.add_argument('--amount', type=float, required=True, help='支付金额（元）')
    create_parser.add_argument('--currency', type=str, default='CNY', help='货币代码（默认 CNY）')
    create_parser.add_argument('--merchant_id', type=str, required=True, help='商户 ID')
    create_parser.add_argument('--description', type=str, help='支付描述')
    
    # query_payment 命令
    query_parser = subparsers.add_parser('query_payment', help='查询支付状态')
    query_parser.add_argument('--transaction_id', type=str, required=True, help='交易 ID')
    
    # refund_payment 命令
    refund_parser = subparsers.add_parser('refund_payment', help='发起退款')
    refund_parser.add_argument('--transaction_id', type=str, required=True, help='原交易 ID')
    refund_parser.add_argument('--amount', type=float, help='退款金额（元，可选）')
    refund_parser.add_argument('--reason', type=str, help='退款原因')
    
    # 全局参数
    parser.add_argument('--env', type=str, choices=['development', 'production'], 
                       help='环境（默认从 PAYMENT_ENV 读取，或 development）')
    
    return parser


async def execute_tool(tool_name: str, params: dict, env: str = "development") -> dict:
    """
    执行工具
    
    Args:
        tool_name: 工具名称
        params: 工具参数
        env: 环境名称 (development 或 production)
    
    Returns:
        执行结果
    """
    # 延迟导入 PaymentSkill，确保环境检查已通过
    src_dir = Path(__file__).parent / "src"
    if str(src_dir) not in sys.path:
        sys.path.insert(0, str(src_dir))
    
    from payment_skill import PaymentSkill
    
    skill = None
    try:
        # 初始化 Skill
        skill = PaymentSkill(env=env)
        await skill.initialize()
        
        # 执行工具
        result = await skill.execute(tool_name, params)
        return {
            "success": True,
            "data": result
        }
    except Exception as e:
        logger.error(f"执行失败: {e}")
        return {
            "success": False,
            "error": str(e)
        }
    finally:
        # 清理资源
        if skill:
            try:
                await skill.cleanup()
            except:
                pass


def main():
    """主函数"""
    # 检查环境
    check_environment()
    
    parser = create_parser()
    args = parser.parse_args()
    
    # 检查是否提供了命令
    if not args.command:
        parser.print_help()
        sys.exit(1)
    
    # 确定环境
    env = args.env or os.getenv("PAYMENT_ENV", "development")
    
    # 构造参数字典
    params = {}
    if args.command == 'create_payment':
        params = {
            'amount': args.amount,
            'currency': args.currency,
            'merchant_id': args.merchant_id,
        }
        if args.description:
            params['description'] = args.description
    
    elif args.command == 'query_payment':
        params = {
            'transaction_id': args.transaction_id
        }
    
    elif args.command == 'refund_payment':
        params = {
            'transaction_id': args.transaction_id
        }
        if args.amount:
            params['amount'] = args.amount
        if args.reason:
            params['reason'] = args.reason
    
    # 执行工具
    try:
        # 兼容 Python 3.6 的异步执行方式
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(execute_tool(args.command, params, env))
        
        # 输出结果（JSON 格式）
        print(json.dumps(result, ensure_ascii=False, indent=2))
        
        # 根据结果设置退出码
        sys.exit(0 if result["success"] else 1)
        
    except KeyboardInterrupt:
        print("\n操作已取消", file=sys.stderr)
        sys.exit(130)
    except Exception as e:
        print(f"错误: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

FILE:src/__init__.py
"""
支付 Skill 包

提供 AI 原生支付能力的龙虾 Skill 实现。
"""

from .payment_skill import PaymentSkill
from .payment_api_client import PaymentAPIClient
from .security import InputValidator, DataEncryption
from .utils import (
    generate_transaction_id,
    format_currency,
    validate_amount,
    validate_currency,
)
from .skill_entry import SkillEntry, get_skill_entry

__version__ = "1.0.3"
__author__ = "Payment Institution"
__all__ = [
    "PaymentSkill",
    "PaymentAPIClient",
    "InputValidator",
    "DataEncryption",
    "generate_transaction_id",
    "format_currency",
    "validate_amount",
    "validate_currency",
    "SkillEntry",
    "get_skill_entry",
]

FILE:src/config_loader.py
"""
配置加载器 - 从 config 目录加载环境配置
"""

import os
import logging
from pathlib import Path
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)


class ConfigLoader:
    """配置加载器"""
    
    @staticmethod
    def get_config_dir() -> Path:
        """获取配置目录路径"""
        # 获取项目根目录（payment-skill-demo）
        current_file = Path(__file__)
        project_root = current_file.parent.parent
        config_dir = project_root / "config"
        return config_dir
    
    @staticmethod
    def load_env_file(env_name: str = "development") -> Dict[str, str]:
        """
        加载环境配置文件
        
        参数:
            env_name: 环境名称 (development 或 production)
        
        返回:
            配置字典
        """
        config_dir = ConfigLoader.get_config_dir()
        env_file = config_dir / f"{env_name}.env"
        
        if not env_file.exists():
            logger.warning(f"配置文件不存在: {env_file}")
            return {}
        
        config = {}
        try:
            with open(env_file, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    # 跳过注释和空行
                    if not line or line.startswith('#'):
                        continue
                    
                    # 解析 KEY=VALUE 格式
                    if '=' in line:
                        key, value = line.split('=', 1)
                        key = key.strip()
                        value = value.strip()
                        
                        # 移除引号
                        if value.startswith('"') and value.endswith('"'):
                            value = value[1:-1]
                        elif value.startswith("'") and value.endswith("'"):
                            value = value[1:-1]
                        
                        config[key] = value
            
            logger.info(f"成功加载配置文件: {env_file}")
            return config
        
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
            return {}
    
    @staticmethod
    def get_api_config(env_name: str = "development") -> Dict[str, Any]:
        """
        获取 API 配置
        
        参数:
            env_name: 环境名称
        
        返回:
            API 配置字典
        """
        # 首先从环境变量读取
        config = {
            "api_key": os.getenv("PAYMENT_API_KEY"),
            "api_secret": os.getenv("PAYMENT_API_SECRET"),
            "api_url": os.getenv("PAYMENT_API_URL"),
            "timeout": int(os.getenv("PAYMENT_API_TIMEOUT", "30")) if os.getenv("PAYMENT_API_TIMEOUT") else 30,
        }
        
        # 如果环境变量未设置，从配置文件读取
        env_config = ConfigLoader.load_env_file(env_name)
        
        if not config["api_key"]:
            config["api_key"] = env_config.get("PAYMENT_API_KEY")
        if not config["api_secret"]:
            config["api_secret"] = env_config.get("PAYMENT_API_SECRET")
        if not config["api_url"]:
            config["api_url"] = env_config.get("PAYMENT_API_URL")
        
        # PAYMENT_API_URL 有默认值
        if not config["api_url"]:
            config["api_url"] = "https://api.zlclaw.com"
        
        # 读取超时配置
        if not config["timeout"] or config["timeout"] == 30:
            timeout_str = env_config.get("PAYMENT_API_TIMEOUT", "30")
            try:
                config["timeout"] = int(timeout_str)
            except ValueError:
                config["timeout"] = 30
        
        # 验证必需的配置
        if not config["api_key"]:
            raise ValueError("PAYMENT_API_KEY 未设置")
        if not config["api_secret"]:
            raise ValueError("PAYMENT_API_SECRET 未设置")
        
        return config
    
    @staticmethod
    def get_encryption_config(env_name: str = "development") -> Optional[str]:
        """
        获取加密密钥配置
        
        参数:
            env_name: 环境名称
        
        返回:
            加密密钥或 None
        """
        # 首先从环境变量读取
        encryption_key = os.getenv("PAYMENT_ENCRYPTION_KEY")
        
        # 如果环境变量未设置，从配置文件读取
        if not encryption_key:
            env_config = ConfigLoader.load_env_file(env_name)
            encryption_key = env_config.get("PAYMENT_ENCRYPTION_KEY")
        
        return encryption_key
    
    @staticmethod
    def get_logging_config(env_name: str = "development") -> Dict[str, Any]:
        """
        获取日志配置
        
        参数:
            env_name: 环境名称
        
        返回:
            日志配置字典
        """
        # 首先从环境变量读取
        config = {
            "level": os.getenv("PAYMENT_LOG_LEVEL", "INFO"),
            "file": os.getenv("PAYMENT_LOG_FILE"),
        }
        
        # 如果环境变量未设置，从配置文件读取
        if not config["file"]:
            env_config = ConfigLoader.load_env_file(env_name)
            config["file"] = env_config.get("PAYMENT_LOG_FILE")
        
        return config

FILE:src/payment_api_client.py
"""
支付 API 客户端
"""

import aiohttp
import json
import logging
import time
import hmac
import hashlib
import base64
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)


class PaymentAPIClient:
    """支付 API 客户端"""
    
    def __init__(self, api_key: str, api_secret: str, api_url: str, timeout: int = 30):
        """
        初始化 API 客户端
        
        参数:
            api_key: API 密钥（用于身份认证）
            api_secret: API 密钥（用于请求签名）
            api_url: API 地址
            timeout: 超时时间(秒)
        """
        self.api_key = api_key
        self.api_secret = api_secret
        self.api_url = api_url.rstrip('/')
        self.timeout = timeout
        self.session: Optional[aiohttp.ClientSession] = None
    
    async def _ensure_session(self):
        """确保会话已创建"""
        if not self.session:
            self.session = aiohttp.ClientSession()
    
    async def _request(self, method: str, endpoint: str,
                      data: Dict[str, Any] = None) -> Dict[str, Any]:
        """
        发送 HTTP 请求
        
        参数:
            method: HTTP 方法
            endpoint: API 端点
            data: 请求数据
        
        返回:
            响应数据
        """
        await self._ensure_session()
        
        url = f"{self.api_url}/{endpoint}"
        
        # 生成请求签名
        signature = self._generate_signature(method, endpoint, data)
        timestamp = str(int(time.time()))
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "X-Signature": signature,
            "X-Timestamp": timestamp,
            "Content-Type": "application/json"
        }
        
        try:
            async with self.session.request(
                method, url,
                json=data,
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=self.timeout)
            ) as response:
                response_data = await response.json()
                
                if response.status == 200:
                    logger.debug(f"API 请求成功: {method} {endpoint}")
                    return response_data
                else:
                    error_msg = response_data.get("error", "未知错误")
                    logger.error(f"API 请求失败: {response.status} - {error_msg}")
                    raise Exception(f"API 错误: {response.status} - {error_msg}")
        
        except aiohttp.ClientError as e:
            logger.error(f"网络错误: {e}")
            raise Exception(f"网络错误: {str(e)}")
    
    def _generate_signature(self, method: str, endpoint: str,
                           data: Dict[str, Any] = None) -> str:
        """
        生成请求签名
        
        参数:
            method: HTTP 方法
            endpoint: API 端点
            data: 请求数据
        
        返回:
            签名字符串
        """
        timestamp = str(int(time.time()))
        message = f"{method}:{endpoint}:{timestamp}"
        
        if data:
            message += f":{json.dumps(data, sort_keys=True)}"
        
        # 使用 HMAC-SHA256 签名（使用 api_secret）
        signature = hmac.new(
            key=self.api_secret.encode(),
            msg=message.encode(),
            digestmod=hashlib.sha256
        ).hexdigest()
        
        return signature
    
    async def health_check(self) -> bool:
        """
        健康检查
        
        返回:
            是否健康
        """
        try:
            result = await self._request("GET", "health")
            return result.get("status") == "ok"
        except Exception as e:
            logger.error(f"健康检查失败: {e}")
            return False
    
    async def create_payment(self, amount: float, currency: str,
                            merchant_id: str, description: str = "") -> Dict[str, Any]:
        """
        创建支付
        
        参数:
            amount: 支付金额
            currency: 货币代码
            merchant_id: 商户 ID
            description: 支付描述
        
        返回:
            支付结果
        """
        payload = {
            "amount": amount,
            "currency": currency,
            "merchant_id": merchant_id,
            "description": description
        }
        
        result = await self._request("POST", "v1/payments", payload)
        
        return {
            "id": result.get("id"),
            "status": result.get("status"),
            "amount": result.get("amount"),
            "currency": result.get("currency"),
            "verification_url": result.get("verification_url"),
            "qr_code": result.get("qr_code"),
            "expires_at": result.get("expires_at")
        }
    
    async def query_payment(self, transaction_id: str) -> Dict[str, Any]:
        """
        查询支付状态
        
        参数:
            transaction_id: 交易 ID
        
        返回:
            支付状态
        """
        result = await self._request("GET", f"v1/payments/{transaction_id}")
        
        return {
            "id": result.get("id"),
            "status": result.get("status"),
            "amount": result.get("amount"),
            "currency": result.get("currency"),
            "created_at": result.get("created_at"),
            "completed_at": result.get("completed_at")
        }
    
    async def refund_payment(self, transaction_id: str,
                            amount: float = None) -> Dict[str, Any]:
        """
        发起退款
        
        参数:
            transaction_id: 原交易 ID
            amount: 退款金额(可选)
        
        返回:
            退款结果
        """
        payload = {}
        if amount is not None:
            payload["amount"] = amount
        
        result = await self._request(
            "POST",
            f"v1/payments/{transaction_id}/refund",
            payload
        )
        
        return {
            "id": result.get("id"),
            "status": result.get("status"),
            "amount": result.get("amount"),
            "currency": result.get("currency"),
            "created_at": result.get("created_at")
        }
    
    async def close(self):
        """关闭连接"""
        if self.session:
            await self.session.close()

FILE:src/payment_skill.py
"""
支付 Skill - 主类
"""

import asyncio
import json
import logging
from typing import Any, Dict, Optional
from datetime import datetime

try:
    # Try relative imports first (when used as a package)
    from .payment_api_client import PaymentAPIClient
    from .security import InputValidator, DataEncryption
    from .utils import generate_transaction_id, format_currency
    from .config_loader import ConfigLoader
except ImportError:
    # Fall back to absolute imports (when run directly)
    from payment_api_client import PaymentAPIClient
    from security import InputValidator, DataEncryption
    from utils import generate_transaction_id, format_currency
    from config_loader import ConfigLoader

logger = logging.getLogger(__name__)


class PaymentSkill:
    """支付 Skill - 封装支付 API"""

    def __init__(self, config: Dict[str, Any] = None, env: str = "development"):
        """
        初始化支付 Skill
        
        参数:
            config: 配置字典（如果提供，将覆盖从文件加载的配置）
            env: 环境名称 (development 或 production)
        """
        self.id = "payment_skill"
        self.name = "支付服务"
        self.description = "提供支付相关的功能"
        self.version = "1.0.0"
        self.env = env

        # 配置 - 优先使用传入的配置，否则从文件加载
        if config:
            self.config = config
        else:
            try:
                self.config = ConfigLoader.get_api_config(env)
            except ValueError as e:
                logger.warning(f"从配置文件加载失败: {e}，将使用空配置")
                self.config = {}
        
        self.api_client: Optional[PaymentAPIClient] = None
        self.encryption = None  # 延迟初始化，在 initialize() 中创建
        self.validator = InputValidator()

        # 工具注册
        self.tools = {
            "create_payment": self.create_payment,
            "query_payment": self.query_payment,
            "refund_payment": self.refund_payment,
        }

        # 状态
        self.is_initialized = False
        self.transactions = {}  # 本地缓存

    async def initialize(self):
        """初始化 Skill"""
        try:
            logger.info("初始化支付 Skill...")

            # 创建 API 客户端
            self.api_client = PaymentAPIClient(
                api_key=self.config.get("api_key"),
                api_secret=self.config.get("api_secret"),
                api_url=self.config.get("api_url"),
                timeout=self.config.get("timeout", 30)
            )

            # 健康检查
            is_healthy = await self.api_client.health_check()
            if not is_healthy:
                raise Exception("支付 API 健康检查失败")

            # 初始化加密器（如果启用）
            if self.config.get("enable_encryption"):
                try:
                    self.encryption = DataEncryption.from_env()
                    logger.info("加密器初始化成功")
                except Exception as e:
                    logger.warning(f"加密器初始化失败: {e}，将继续运行但不使用加密")

            self.is_initialized = True
            logger.info("支付 Skill 初始化成功")

        except Exception as e:
            logger.error(f"支付 Skill 初始化失败: {e}")
            raise

    async def create_payment(self, amount: float, currency: str,
                            merchant_id: str, description: str = "") -> Dict[str, Any]:
        """
        创建支付请求

        参数:
            amount: 支付金额
            currency: 货币代码 (CNY, USD, EUR)
            merchant_id: 商户 ID
            description: 支付描述

        返回:
            支付结果
        """
        try:
            logger.info(f"创建支付: {amount} {currency} 给商户 {merchant_id}")

            # 参数验证
            if amount <= 0:
                raise ValueError("金额必须大于 0")
            if amount > 1000000:
                raise ValueError("单笔金额不能超过 100 万")
            if currency not in ["CNY", "USD", "EUR"]:
                raise ValueError(f"不支持的货币: {currency}")
            if not merchant_id:
                raise ValueError("商户 ID 不能为空")

            # 生成交易 ID
            transaction_id = generate_transaction_id()

            # 调用 API
            result = await self.api_client.create_payment(
                amount=amount,
                currency=currency,
                merchant_id=merchant_id,
                description=description
            )

            # 缓存交易信息
            self.transactions[transaction_id] = {
                "amount": amount,
                "currency": currency,
                "merchant_id": merchant_id,
                "status": "pending_auth",
                "created_at": datetime.utcnow().isoformat()
            }

            # 返回结果
            response = {
                "success": True,
                "transaction_id": transaction_id,
                "status": result.get("status", "pending_auth"),
                "amount": amount,
                "currency": currency,
                "verification_url": result.get("verification_url"),
                "qr_code": result.get("qr_code"),
                "expires_at": result.get("expires_at")
            }

            logger.info(f"支付创建成功: {transaction_id}")
            return response

        except ValueError as e:
            logger.warning(f"参数验证失败: {e}")
            return {
                "success": False,
                "error": str(e),
                "error_code": "INVALID_PARAMS"
            }
        except Exception as e:
            logger.error(f"创建支付失败: {e}")
            return {
                "success": False,
                "error": "创建支付失败,请稍后重试",
                "error_code": "PAYMENT_ERROR"
            }

    async def query_payment(self, transaction_id: str) -> Dict[str, Any]:
        """
        查询支付状态

        参数:
            transaction_id: 交易 ID

        返回:
            支付状态
        """
        try:
            logger.info(f"查询支付状态: {transaction_id}")

            # 参数验证
            if not transaction_id:
                raise ValueError("交易 ID 不能为空")

            # 调用 API
            result = await self.api_client.query_payment(transaction_id)

            # 更新本地缓存
            if transaction_id in self.transactions:
                self.transactions[transaction_id]["status"] = result.get("status")

            # 返回结果
            response = {
                "success": True,
                "transaction_id": transaction_id,
                "status": result.get("status"),
                "amount": result.get("amount"),
                "currency": result.get("currency"),
                "created_at": result.get("created_at"),
                "completed_at": result.get("completed_at")
            }

            logger.info(f"支付状态查询成功: {transaction_id} - {result.get('status')}")
            return response

        except ValueError as e:
            logger.warning(f"参数验证失败: {e}")
            return {
                "success": False,
                "error": str(e),
                "error_code": "INVALID_PARAMS"
            }
        except Exception as e:
            logger.error(f"查询支付失败: {e}")
            return {
                "success": False,
                "error": "查询支付失败,请稍后重试",
                "error_code": "QUERY_ERROR"
            }

    async def refund_payment(self, transaction_id: str,
                            amount: float = None) -> Dict[str, Any]:
        """
        发起退款

        参数:
            transaction_id: 原交易 ID
            amount: 退款金额(可选,不填则全额退款)

        返回:
            退款结果
        """
        try:
            logger.info(f"发起退款: {transaction_id}, 金额: {amount}")

            # 参数验证
            if not transaction_id:
                raise ValueError("交易 ID 不能为空")

            if amount is not None and amount <= 0:
                raise ValueError("退款金额必须大于 0")

            # 调用 API
            result = await self.api_client.refund_payment(
                transaction_id=transaction_id,
                amount=amount
            )

            # 返回结果
            response = {
                "success": True,
                "refund_id": result.get("id"),
                "status": result.get("status"),
                "amount": result.get("amount"),
                "currency": result.get("currency"),
                "created_at": result.get("created_at")
            }

            logger.info(f"退款发起成功: {result.get('id')}")
            return response

        except ValueError as e:
            logger.warning(f"参数验证失败: {e}")
            return {
                "success": False,
                "error": str(e),
                "error_code": "INVALID_PARAMS"
            }
        except Exception as e:
            logger.error(f"发起退款失败: {e}")
            return {
                "success": False,
                "error": "发起退款失败,请稍后重试",
                "error_code": "REFUND_ERROR"
            }

    async def execute(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        执行工具

        参数:
            tool_name: 工具名称
            params: 工具参数

        返回:
            执行结果
        """
        if not self.is_initialized:
            return {
                "success": False,
                "error": "Skill 未初始化",
                "error_code": "NOT_INITIALIZED"
            }

        if tool_name not in self.tools:
            return {
                "success": False,
                "error": f"未知的工具: {tool_name}",
                "error_code": "UNKNOWN_TOOL"
            }

        try:
            tool_func = self.tools[tool_name]
            result = await tool_func(**params)
            return result
        except Exception as e:
            logger.error(f"执行工具 {tool_name} 失败: {e}")
            return {
                "success": False,
                "error": "工具执行失败",
                "error_code": "EXECUTION_ERROR"
            }

    async def cleanup(self):
        """清理资源"""
        try:
            logger.info("清理支付 Skill...")
            if self.api_client:
                await self.api_client.close()
            self.is_initialized = False
            logger.info("支付 Skill 清理完成")
        except Exception as e:
            logger.error(f"清理支付 Skill 失败: {e}")

    def get_tool_schema(self, tool_name: str) -> Dict[str, Any]:
        """获取工具的 JSON Schema"""
        schemas = {
            "create_payment": {
                "type": "object",
                "properties": {
                    "amount": {
                        "type": "number",
                        "description": "支付金额",
                        "minimum": 0.01,
                        "maximum": 1000000
                    },
                    "currency": {
                        "type": "string",
                        "description": "货币代码",
                        "enum": ["CNY", "USD", "EUR"]
                    },
                    "merchant_id": {
                        "type": "string",
                        "description": "商户 ID"
                    },
                    "description": {
                        "type": "string",
                        "description": "支付描述"
                    }
                },
                "required": ["amount", "currency", "merchant_id"]
            },
            "query_payment": {
                "type": "object",
                "properties": {
                    "transaction_id": {
                        "type": "string",
                        "description": "交易 ID"
                    }
                },
                "required": ["transaction_id"]
            },
            "refund_payment": {
                "type": "object",
                "properties": {
                    "transaction_id": {
                        "type": "string",
                        "description": "原交易 ID"
                    },
                    "amount": {
                        "type": "number",
                        "description": "退款金额(可选)"
                    }
                },
                "required": ["transaction_id"]
            }
        }
        return schemas.get(tool_name, {})

FILE:src/payment_skill.yaml
id: payment_skill
name: 支付服务
description: 提供支付相关的功能,支持创建支付、查询状态、发起退款
version: 1.0.0
author: zlclaw Team

# 依赖
dependencies:
  - aiohttp>=3.8.0
  - pydantic>=2.0.0
  - pyyaml>=6.0

# 必需环境变量
required_env_vars:
  - PAYMENT_API_KEY: "API 密钥，用于身份认证"
  - PAYMENT_API_SECRET: "API 密钥，用于请求签名"
  - PAYMENT_API_URL: "支付 API 地址"

# 可选环境变量
optional_env_vars:
  - PAYMENT_ENCRYPTION_KEY: "32 字节加密密钥（可选）"
  - PAYMENT_LOG_LEVEL: "日志级别，默认 INFO"
  - PAYMENT_LOG_FILE: "日志文件路径"
  - PAYMENT_ENABLE_CROSS_DEVICE_AUTH: "启用跨设备验证，默认 true"

# 配置
config:
  api_key: PAYMENT_API_KEY
  api_url: https://api.zlclaw.com
  timeout: 30
  retry_count: 3
  enable_logging: true
  enable_encryption: false

# 工具列表
tools:
  - name: create_payment
    description: 创建支付请求
    timeout: 10000
    rate_limit:
      requests_per_minute: 60
    requires_approval: false

  - name: query_payment
    description: 查询支付状态
    timeout: 5000
    rate_limit:
      requests_per_minute: 100
    requires_approval: false

  - name: refund_payment
    description: 发起退款
    timeout: 10000
    rate_limit:
      requests_per_minute: 30
    requires_approval: true

# 权限要求
permissions:
  - network.http
  - storage.read
  - storage.write
  - crypto.sign

# 健康检查
health_check:
  enabled: true
  interval: 60
  timeout: 5000

# 日志配置
logging:
  level: INFO
  format: json
  output: stdout

FILE:src/security.py
"""
安全工具
"""

import json
import re
from typing import Dict, Any
from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes
import base64


class InputValidator:
    """输入验证器"""
    
    @staticmethod
    def validate_and_sanitize(params: Dict[str, Any],
                             schema: Dict[str, Any]) -> Dict[str, Any]:
        """
        验证和清理输入参数
        
        参数:
            params: 输入参数
            schema: JSON Schema
        
        返回:
            验证后的参数
        """
        validated = {}
        
        for field, field_schema in schema.get("properties", {}).items():
            if field not in params:
                if field in schema.get("required", []):
                    raise ValueError(f"缺少必需字段: {field}")
                continue
            
            value = params[field]
            field_type = field_schema.get("type")
            
            # 类型检查
            if field_type == "string":
                if not isinstance(value, str):
                    raise ValueError(f"字段 {field} 必须是字符串")
                # 防止 SQL 注入
                value = InputValidator.sanitize_string(value)
            
            elif field_type == "number":
                if not isinstance(value, (int, float)):
                    raise ValueError(f"字段 {field} 必须是数字")
                # 检查范围
                if "minimum" in field_schema and value < field_schema["minimum"]:
                    raise ValueError(f"字段 {field} 小于最小值")
                if "maximum" in field_schema and value > field_schema["maximum"]:
                    raise ValueError(f"字段 {field} 大于最大值")
            
            elif field_type == "object":
                if not isinstance(value, dict):
                    raise ValueError(f"字段 {field} 必须是对象")
            
            validated[field] = value
        
        return validated
    
    @staticmethod
    def sanitize_string(value: str) -> str:
        """
        清理字符串,防止注入攻击
        
        参数:
            value: 输入字符串
        
        返回:
            清理后的字符串
        """
        # 移除危险字符
        dangerous_patterns = [
            r"'",
            r'"',
            r";",
            r"--",
            r"/\*",
            r"\*/",
            r"xp_",
            r"sp_"
        ]
        
        for pattern in dangerous_patterns:
            value = re.sub(pattern, "", value)
        
        # 限制长度
        max_length = 1000
        if len(value) > max_length:
            value = value[:max_length]
        
        return value


class DataEncryption:
    """数据加密"""
    
    def __init__(self, key: bytes):
        """
        初始化加密器
        
        参数:
            key: 加密密钥(32 字节)，生产环境必须提供
        """
        if key is None or len(key) != 32:
            raise ValueError("加密密钥必须为 32 字节，生产环境必须提供有效密钥")
        self.key = key
    
    @classmethod
    def from_env(cls, env_key: str = "PAYMENT_ENCRYPTION_KEY") -> "DataEncryption":
        """
        从环境变量创建加密器
        
        参数:
            env_key: 环境变量名称
        """
        import os
        key_str = os.environ.get(env_key)
        if not key_str:
            raise ValueError(f"环境变量 {env_key} 未设置，生产环境必须提供加密密钥")
        key_bytes = key_str.encode()[:32].ljust(32, b'0')
        return cls(key_bytes)
    
    def encrypt_sensitive_data(self, data: Dict[str, Any]) -> str:
        """
        加密敏感数据
        
        参数:
            data: 数据字典
        
        返回:
            加密后的数据(Base64 编码)
        """
        # 标记敏感字段
        sensitive_fields = ["api_key", "password", "token", "card_number"]
        
        encrypted_data = data.copy()
        
        for field in sensitive_fields:
            if field in encrypted_data:
                # 加密敏感字段
                plaintext = str(encrypted_data[field]).encode()
                
                # 生成随机 nonce
                nonce = get_random_bytes(16)
                
                # 创建加密器
                cipher = AES.new(self.key, AES.MODE_GCM, nonce=nonce)
                
                # 加密
                ciphertext, tag = cipher.encrypt_and_digest(plaintext)
                
                # 存储加密后的数据
                encrypted_data[field] = {
                    "encrypted": base64.b64encode(ciphertext).decode(),
                    "nonce": base64.b64encode(nonce).decode(),
                    "tag": base64.b64encode(tag).decode()
                }
        
        return json.dumps(encrypted_data)
    
    def decrypt_sensitive_data(self, encrypted_data: str) -> Dict[str, Any]:
        """
        解密敏感数据
        
        参数:
            encrypted_data: 加密后的数据(JSON 字符串)
        
        返回:
            解密后的数据
        """
        data = json.loads(encrypted_data)
        
        for field, value in data.items():
            if isinstance(value, dict) and "encrypted" in value:
                # 解密敏感字段
                ciphertext = base64.b64decode(value["encrypted"])
                nonce = base64.b64decode(value["nonce"])
                tag = base64.b64decode(value["tag"])
                
                # 创建解密器
                cipher = AES.new(self.key, AES.MODE_GCM, nonce=nonce)
                
                # 解密
                plaintext = cipher.decrypt_and_verify(ciphertext, tag)
                data[field] = plaintext.decode()
        
        return data

FILE:src/utils.py
"""
工具函数模块

提供支付 Skill 所需的各种工具函数。
"""

import uuid
import re
from datetime import datetime
from typing import Optional, Tuple
from decimal import Decimal, InvalidOperation


def generate_transaction_id(prefix: str = "txn") -> str:
    """
    生成唯一的交易 ID。
    
    Args:
        prefix: 交易 ID 前缀，默认为 "txn"
    
    Returns:
        格式为 "{prefix}_{timestamp}_{uuid}" 的交易 ID
    
    Example:
        >>> txn_id = generate_transaction_id()
        >>> print(txn_id)
        txn_20260315_550e8400e29b41d4a716446655440000
    """
    timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")
    unique_id = str(uuid.uuid4()).replace("-", "")[:16]
    return f"{prefix}_{timestamp}_{unique_id}"


def generate_verification_id(prefix: str = "ver") -> str:
    """
    生成跨设备验证 ID。
    
    Args:
        prefix: 验证 ID 前缀，默认为 "ver"
    
    Returns:
        格式为 "{prefix}_{uuid}" 的验证 ID
    """
    unique_id = str(uuid.uuid4()).replace("-", "")[:16]
    return f"{prefix}_{unique_id}"


def format_currency(amount: float, currency: str = "CNY") -> str:
    """
    格式化货币显示。
    
    Args:
        amount: 金额
        currency: 货币代码
    
    Returns:
        格式化后的货币字符串
    
    Example:
        >>> format_currency(299.00, "CNY")
        '¥299.00'
        >>> format_currency(99.99, "USD")
        '$99.99'
    """
    currency_symbols = {
        "CNY": "¥",
        "USD": "$",
        "EUR": "€",
        "GBP": "£",
        "JPY": "¥",
    }
    symbol = currency_symbols.get(currency, currency)
    return f"{symbol}{amount:.2f}"


def validate_amount(amount: float) -> Tuple[bool, Optional[str]]:
    """
    验证支付金额。
    
    Args:
        amount: 支付金额
    
    Returns:
        (是否有效, 错误信息)
    
    Example:
        >>> validate_amount(299.00)
        (True, None)
        >>> validate_amount(-100)
        (False, '金额必须大于0')
        >>> validate_amount(299.999)
        (False, '金额最多2位小数')
    """
    try:
        # 转换为 Decimal 以精确处理浮点数
        decimal_amount = Decimal(str(amount))
        
        # 检查金额范围
        if decimal_amount <= 0:
            return False, "金额必须大于0"
        
        if decimal_amount > 1000000:
            return False, "金额不能超过1,000,000"
        
        # 检查小数位数
        if decimal_amount.as_tuple().exponent < -2:
            return False, "金额最多2位小数"
        
        return True, None
    except (InvalidOperation, ValueError):
        return False, "金额格式无效"


def validate_currency(currency: str) -> Tuple[bool, Optional[str]]:
    """
    验证货币代码。
    
    Args:
        currency: ISO 4217 货币代码
    
    Returns:
        (是否有效, 错误信息)
    
    Example:
        >>> validate_currency("CNY")
        (True, None)
        >>> validate_currency("INVALID")
        (False, '无效的货币代码')
    """
    valid_currencies = {
        "CNY", "USD", "EUR", "GBP", "JPY", "AUD", "CAD", "CHF",
        "HKD", "SGD", "INR", "KRW", "MXN", "NZD", "ZAR"
    }
    
    if not isinstance(currency, str):
        return False, "货币代码必须是字符串"
    
    currency_upper = currency.upper()
    
    if currency_upper not in valid_currencies:
        return False, f"无效的货币代码: {currency}"
    
    return True, None


def validate_merchant_id(merchant_id: str) -> Tuple[bool, Optional[str]]:
    """
    验证商户 ID。
    
    Args:
        merchant_id: 商户 ID
    
    Returns:
        (是否有效, 错误信息)
    """
    if not isinstance(merchant_id, str):
        return False, "商户 ID 必须是字符串"
    
    if len(merchant_id) == 0:
        return False, "商户 ID 不能为空"
    
    if len(merchant_id) > 255:
        return False, "商户 ID 长度不能超过255"
    
    # 只允许字母、数字、下划线和连字符
    if not re.match(r"^[a-zA-Z0-9_-]+$", merchant_id):
        return False, "商户 ID 只能包含字母、数字、下划线和连字符"
    
    return True, None


def validate_description(description: str) -> Tuple[bool, Optional[str]]:
    """
    验证支付描述。
    
    Args:
        description: 支付描述
    
    Returns:
        (是否有效, 错误信息)
    """
    if not isinstance(description, str):
        return False, "描述必须是字符串"
    
    if len(description) == 0:
        return False, "描述不能为空"
    
    if len(description) > 500:
        return False, "描述长度不能超过500"
    
    return True, None


def validate_reference_id(reference_id: str) -> Tuple[bool, Optional[str]]:
    """
    验证幂等性标识。
    
    Args:
        reference_id: 幂等性标识
    
    Returns:
        (是否有效, 错误信息)
    """
    if not isinstance(reference_id, str):
        return False, "幂等性标识必须是字符串"
    
    if len(reference_id) == 0:
        return False, "幂等性标识不能为空"
    
    if len(reference_id) > 255:
        return False, "幂等性标识长度不能超过255"
    
    # UUID 或订单号格式
    if not re.match(r"^[a-zA-Z0-9_-]+$", reference_id):
        return False, "幂等性标识只能包含字母、数字、下划线和连字符"
    
    return True, None


def parse_timestamp(timestamp: int) -> str:
    """
    将时间戳转换为可读的日期时间字符串。
    
    Args:
        timestamp: Unix 时间戳（秒）
    
    Returns:
        格式为 "YYYY-MM-DD HH:MM:SS" 的日期时间字符串
    
    Example:
        >>> parse_timestamp(1710489600)
        '2026-03-15 12:00:00'
    """
    try:
        dt = datetime.utcfromtimestamp(timestamp)
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    except (ValueError, OSError):
        return "Invalid timestamp"


def get_current_timestamp() -> int:
    """
    获取当前 Unix 时间戳。
    
    Returns:
        当前时间戳（秒）
    """
    return int(datetime.utcnow().timestamp())


def calculate_expiry_time(ttl_seconds: int) -> int:
    """
    计算过期时间戳。
    
    Args:
        ttl_seconds: 生存时间（秒）
    
    Returns:
        过期时间戳
    
    Example:
        >>> current = get_current_timestamp()
        >>> expiry = calculate_expiry_time(300)
        >>> expiry - current
        300
    """
    return get_current_timestamp() + ttl_seconds


def is_expired(expiry_time: int) -> bool:
    """
    检查是否已过期。
    
    Args:
        expiry_time: 过期时间戳
    
    Returns:
        是否已过期
    """
    return get_current_timestamp() > expiry_time


def truncate_string(text: str, max_length: int = 100) -> str:
    """
    截断字符串以防止日志过长。
    
    Args:
        text: 原始字符串
        max_length: 最大长度
    
    Returns:
        截断后的字符串
    """
    if len(text) <= max_length:
        return text
    return text[:max_length - 3] + "..."


def mask_sensitive_data(data: str, show_chars: int = 4) -> str:
    """
    掩盖敏感数据（如 API 密钥）。
    
    Args:
        data: 原始数据
        show_chars: 显示的字符数
    
    Returns:
        掩盖后的数据
    
    Example:
        >>> mask_sensitive_data("sk_live_abc123def456", 4)
        'sk_l...456'
    """
    if len(data) <= show_chars * 2:
        return "*" * len(data)
    
    start = data[:show_chars]
    end = data[-show_chars:]
    middle = "*" * (len(data) - show_chars * 2)
    return f"{start}{middle}{end}"