“就跑一下”的代码，不该为它单独开一台服务器！直接上亚马逊云 Lambda!

project/
├── lambda_function.py        # 核心逻辑
├── requirements.txt          # Pandas 依赖
└── deploy.zip                # 打包上传

pandas
openpyxl
boto3
requests

import json
import boto3
import pandas as pd
import requests
import os
from io import BytesIO

dynamodb = boto3.resource('dynamodb')
s3 = boto3.client('s3')

# 环境变量提前设置
TABLE_NAME = os.environ.get("TABLE_NAME")
WECHAT_WEBHOOK = os.environ.get("WECHAT_WEBHOOK")

def lambda_handler(event, context):
    # 1. 从 S3 事件中提取 Bucket 和 Key
    record = event['Records'][0]['s3']
    bucket = record['bucket']['name']
    key = record['object']['key']
    
    print(f"Processing file: s3://{bucket}/{key}")
    
    # 2. 下载 Excel 文件
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response['Body'].read()
    df = pd.read_excel(BytesIO(content), engine='openpyxl')
    
    print(f"Loaded DataFrame with shape: {df.shape}")
    
    # 3. 简单分析示例（比如统计每个“部门”的数量）
    summary = df['部门'].value_counts().to_dict()
    print(f"Summary: {summary}")
    
    # 4. 写入 DynamoDB（每个部门写一行）
    table = dynamodb.Table(TABLE_NAME)
    for department, count in summary.items():
        table.put_item(Item={
            'Department': department,
            'Count': count,
            'UploadedFile': key
        })
    
    # 5. 推送到企业微信机器人
    content_lines = [f"{dept}: {cnt}人" for dept, cnt in summary.items()]
    msg = {
        "msgtype": "text",
        "text": {
            "content": f"📊 新文件分析完成：{key}\n\n" + "\n".join(content_lines)
        }
    }
    r = requests.post(WECHAT_WEBHOOK, json=msg)
    print(f"WeChat notification sent: {r.status_code}")
    
    return {
        'statusCode': 200,
        'body': json.dumps('Excel 分析完成，通知已发送')
    }

pip install -r requirements.txt -t .
zip -r deploy.zip .
# 上传到 AWS Lambda 控制台