我来帮你搞定这个需求——让Lambda里的recurse_for_values函数打印的内容直接落到S3桶的文件里。这里有两种实用的方案，你可以按需选择：

方案一：实时重定向stdout到S3（适合小批量输出） #

这种方法是直接替换Python的sys.stdout为一个自定义写入器，每次print都会把内容追加到S3文件里。不过要注意，频繁调用S3 API可能会有性能开销，适合输出内容不多的场景。

代码实现 #

import sys
import boto3
from io import StringIO

class S3Writer:
    def __init__(self, bucket_name, s3_key):
        self.bucket = bucket_name
        self.key = s3_key
        self.s3 = boto3.client('s3')
        # 初始化时先创建空文件（可选，避免首次追加报错）
        try:
            self.s3.put_object(Bucket=self.bucket, Key=self.key, Body='')
        except Exception as e:
            print(f"初始化S3文件失败: {e}", file=sys.stderr)

    def write(self, content):
        # S3本身没有原生追加，这里实现"读取原内容+追加新内容"的逻辑
        try:
            response = self.s3.get_object(Bucket=self.bucket, Key=self.key)
            existing_content = response['Body'].read().decode('utf-8')
            new_content = existing_content + content
            self.s3.put_object(Bucket=self.bucket, Key=self.key, Body=new_content.encode('utf-8'))
        except self.s3.exceptions.NoSuchKey:
            # 文件不存在时直接写入
            self.s3.put_object(Bucket=self.bucket, Key=self.key, Body=content.encode('utf-8'))
        except Exception as e:
            # 出错时把内容输出到Lambda stderr，避免丢失
            print(f"写入S3失败: {e}, 内容: {content}", file=sys.stderr)

    def flush(self):
        # 实现stdout要求的flush接口
        pass

def lambda_handler(event, context):
    # 替换成你的S3桶名和目标文件路径
    S3_BUCKET = "your-bucket-name"
    S3_KEY = "outputs/function_stdout.txt"

    # 替换stdout为自定义S3写入器
    original_stdout = sys.stdout
    sys.stdout = S3Writer(S3_BUCKET, S3_KEY)

    try:
        # 调用你的函数，所有print内容都会流向S3
        recurse_for_values(top_vault_prefix="your-target-prefix", top_level_keys=["key1", "key2"])
    finally:
        # 恢复原stdout，避免影响后续逻辑
        sys.stdout = original_stdout

    return {
        'statusCode': 200,
        'body': '输出已成功写入S3'
    }

# 你的原函数示例
def recurse_for_values(top_vault_prefix, top_level_keys):
    print(f"开始处理前缀: {top_vault_prefix}")
    for key in top_level_keys:
        print(f"处理键: {key}，对应值为xxx")
    # 其他打印逻辑...

方案二：捕获所有输出后一次性上传（推荐，性能更好） #

如果你的函数输出量较大，频繁操作S3会拖慢速度，不如先把所有print内容捕获到内存缓冲区，函数执行完后一次性上传到S3，这样更高效，也能减少S3 API调用次数。

代码实现 #

import sys
import boto3
from io import StringIO

def lambda_handler(event, context):
    S3_BUCKET = "your-bucket-name"
    S3_KEY = "outputs/function_stdout.txt"

    # 创建内存缓冲区，用来捕获所有stdout输出
    buffer = StringIO()
    original_stdout = sys.stdout
    sys.stdout = buffer

    try:
        # 执行你的业务函数
        recurse_for_values(top_vault_prefix="your-target-prefix", top_level_keys=["key1", "key2"])
    finally:
        # 恢复原stdout
        sys.stdout = original_stdout

    # 把缓冲区内容一次性上传到S3
    s3 = boto3.client('s3')
    try:
        s3.put_object(
            Bucket=S3_BUCKET,
            Key=S3_KEY,
            Body=buffer.getvalue().encode('utf-8'),
            ContentType='text/plain; charset=utf-8'
        )
    except Exception as e:
        print(f"上传S3失败: {e}", file=sys.stderr)
        return {
            'statusCode': 500,
            'body': '输出写入S3失败'
        }

    buffer.close()
    return {
        'statusCode': 200,
        'body': '输出已成功写入S3'
    }

# 你的原函数
def recurse_for_values(top_vault_prefix, top_level_keys):
    print(f"开始处理前缀: {top_vault_prefix}")
    for key in top_level_keys:
        print(f"处理键: {key}，对应值为xxx")
    # 其他打印逻辑...

关键注意事项 #

• Lambda权限配置: 必须给Lambda的执行角色添加S3写入权限，策略示例如下：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::your-bucket-name/*"
        }
    ]
}

• 避免硬编码: 建议把S3桶名和文件路径放到Lambda的环境变量里，不要直接写在代码里，方便后续修改。
• 超大输出场景: 如果函数输出超过几百MB，内存缓冲区可能不够用，这时候可以考虑用Lambda的临时存储（/tmp目录）先写入本地文件，再上传到S3。
• 编码问题: 确保上传到S3的内容用UTF-8编码，避免出现乱码。

内容的提问来源于stack exchange，提问作者Donislav Belev