CodeWhisperer：AWS AI编程

# CodeWhisperer 能识别 AWS SDK 上下文并生成建议import boto3 # 当开发者输入以下注释时，CodeWhisperer 会自动建议完整实现# Upload a file to S3 with progress trackingdef upload_to_s3(file_path, bucket_name, object_key):    s3_client = boto3.client('s3')    # CodeWhisperer 建议的完整实现：    with open(file_path, 'rb') as f:        s3_client.upload_fileobj(f, bucket_name, object_key)    print(f'Uploaded {file_path} to s3://{bucket_name}/{object_key}')

import boto3from botocore.exceptions import ClientError def create_dynamodb_table(table_name, partition_key, region='us-east-1'):    dynamodb = boto3.resource('dynamodb', region_name=region)    try:        table = dynamodb.create_table(            TableName=table_name,            KeySchema=[{'AttributeName': partition_key, 'KeyType': 'HASH'}],            AttributeDefinitions=[{'AttributeName': partition_key, 'AttributeType': 'S'}],            BillingMode='PAY_PER_REQUEST'        )        table.wait_until_exists()        return table    except ClientError as e:        print(f'Error creating table: {e}')        raise

import sqlite3def get_user_bad(username):    conn = sqlite3.connect('users.db')    query = f"SELECT * FROM users WHERE name = '{username}'"    return conn.execute(query).fetchall() def get_user_good(username):    conn = sqlite3.connect('users.db')    query = "SELECT * FROM users WHERE name = ?"    return conn.execute(query, (username,)).fetchall()

import jsonimport boto3import osfrom botocore.exceptions import ClientError def lambda_handler(event, context):    s3 = boto3.client('s3')    dynamodb = boto3.resource('dynamodb')    table = dynamodb.Table(os.environ['TABLE_NAME'])    for record in event.get('Records', []):        bucket = record['s3']['bucket']['name']        key = record['s3']['object']['key']        try:            response = s3.head_object(Bucket=bucket, Key=key)            table.put_item(Item={                'object_key': key,                'bucket': bucket,                'size': response['ContentLength'],                'last_modified': str(response['LastModified'])            })        except ClientError as e:            print(f'Error processing {key}: {e}')    return {'statusCode': 200, 'body': json.dumps({'processed': len(event.get('Records', []))})}

import boto3from botocore.config import Configimport osfrom botocore.exceptions import ClientError, NoCredentialsError config = Config(    retries={'max_attempts': 3, 'mode': 'standard'},    connect_timeout=5,    read_timeout=10)s3 = boto3.client('s3', config=config)BUCKET = os.environ.get('S3_BUCKET', 'default-bucket') def safe_s3_operation():    try:        response = s3.list_objects_v2(Bucket=BUCKET, MaxKeys=10)        return response.get('Contents', [])    except NoCredentialsError:        print('AWS credentials not found')        raise    except ClientError as e:        error_code = e.response['Error']['Code']        print(f'AWS error: {error_code}')        raise

import jsonimport random class CodeWhispererSimulator:    """模拟 CodeWhisperer 的代码建议生成与安全扫描流程"""     def __init__(self):        self.context_window = []        self.reference_db = self._init_reference_db()        self.security_rules = self._init_security_rules()     def _init_reference_db(self):        """初始化开源代码引用数据库"""        return [            {"repo": "https://github.com/aws/aws-sdk-python", "license": "Apache-2.0", "snippet": "boto3.client('s3')"},            {"repo": "https://github.com/requests/requests", "license": "Apache-2.0", "snippet": "requests.get(url, timeout=30)"},            {"repo": "https://github.com/pallets/flask", "license": "BSD-3-Clause", "snippet": "@app.route('/api')"},            {"repo": "https://github.com/gpl-project/gpl", "license": "GPL-3.0", "snippet": "def gpl_function()"},        ]     def _init_security_rules(self):        """初始化安全扫描规则集"""        return [            {"rule_id": "SEC001", "pattern": "f\"SELECT.*{", "severity": "HIGH", "description": "SQL注入风险"},            {"rule_id": "SEC002", "pattern": "aws_access_key_id", "severity": "CRITICAL", "description": "硬编码AWS凭证"},            {"rule_id": "SEC003", "pattern": "eval(", "severity": "HIGH", "description": "代码注入风险"},            {"rule_id": "SEC004", "pattern": "ACL='public-read'", "severity": "MEDIUM", "description": "S3公开访问风险"},        ]     def generate_suggestion(self, context_lines, comment_hint):        """根据上下文和注释生成代码建议"""        suggestions = []        if 's3' in context_lines.lower() or 's3' in comment_hint.lower():            suggestions.append({                "code": "s3_client = boto3.client('s3')\nresponse = s3_client.put_object(Bucket=bucket, Key=key, Body=data)",                "confidence": 0.92,                "reference": None            })        elif 'lambda' in context_lines.lower():            suggestions.append({                "code": "def lambda_handler(event, context):\n    return {'statusCode': 200}",                "confidence": 0.88,                "reference": None            })        for ref in self.reference_db:            if ref["snippet"] in context_lines:                suggestions.append({                    "code": f"# 参考自 {ref['repo']} ({ref['license']})\n{ref['snippet']}",                    "confidence": 0.75,                    "reference": {"repo": ref["repo"], "license": ref["license"]}                })        return suggestions     def security_scan(self, code):        """对代码执行安全扫描"""        findings = []        for rule in self.security_rules:            if rule["pattern"] in code:                findings.append({                    "rule_id": rule["rule_id"],                    "severity": rule["severity"],                    "description": rule["description"],                    "line": code.index(rule["pattern"]) // 40 + 1                })        return findings def run_demo():    """运行 CodeWhisperer 模拟器演示"""    cw = CodeWhispererSimulator()    print("=== 场景1: S3 上传代码建议 ===")    context = "import boto3\n# Upload file to S3 bucket\ndef upload_file(bucket, key, data):"    suggestions = cw.generate_suggestion(context, "Upload file to S3 bucket")    for i, s in enumerate(suggestions):        ref_info = f" [来源: {s['reference']['repo']}]" if s['reference'] else " [原创生成]"        print(f"建议{i+1} (置信度{s['confidence']:.0%}){ref_info}:")        print(f"  {s['code'][:60]}...")     print("\n=== 场景2: 安全扫描检测 ===")    risky_code = "query = f\"SELECT * FROM users WHERE id = {user_id}\"\naws_access_key_id = 'AKIA12345'"    findings = cw.security_scan(risky_code)    for f in findings:        print(f"[{f['severity']}] {f['rule_id']}: {f['description']}")     print("\n=== 场景3: 引用追踪与合规过滤 ===")    licenses_filter = {'GPL-3.0'}    all_suggestions = cw.generate_suggestion("def gpl_function()", "implement gpl logic")    for s in all_suggestions:        if s['reference'] and s['reference']['license'] in licenses_filter:            print(f"已过滤: GPL许可证建议 ({s['reference']['license']})")        else:            print(f"已采纳: {s['code'][:50]}...") if __name__ == "__main__":    run_demo()