停车场AWS Rekognition项目：无法更新DynamoDB车辆记录的exit_time

核心问题分析 #

你当前的代码使用put_item写入数据，无论车牌是否存在都会生成新记录（或覆盖旧记录），根本原因有两个：

1. 你的DynamoDB表未将reg_plate设为分区键（主键），导致同一车牌无法被识别为同一条记录。
2. 缺少「判断记录存在性，选择插入/更新」的逻辑。

第一步：修正DynamoDB表结构 #

必须将表的分区键设置为reg_plate，类型选为「字符串」。这是同一车牌能对应同一条记录的基础。

第二步：修改代码逻辑 #

以下是修正后的完整代码，实现「入场插入、出场更新」的核心逻辑：

import boto3
import time

def detect_text(photo, bucket):
    client = boto3.client('rekognition')
    response = client.detect_text(Image={'S3Object':{'Bucket':bucket,'Name':photo}})
                        
    textDetections = response['TextDetections']
    for text in textDetections:
        if text['Type'] == 'LINE':
            return text['DetectedText']
    return False

def main(event, context): 
    # 替换为你的S3桶名称
    bucket = 'your-bucket-name'
    photo = 'regtest.jpg'
    text_detected = detect_text(photo, bucket)
    
    if not text_detected:
        return
    
    print(f"Text detected: {text_detected}")
    
    current_time = str(int(time.time()))
    dynamodb = boto3.client('dynamodb')
    table_name = 'Customer_Plate_DB'
    
    # 先尝试更新出场时间：针对已存在的入场记录
    try:
        dynamodb.update_item(
            TableName=table_name,
            Key={'reg_plate': {'S': text_detected}},
            UpdateExpression='SET exit_time = :current',
            ExpressionAttributeValues={':current': {'S': current_time}},
            # 可选条件：仅允许更新未出场的记录（exit_time为0）
            ConditionExpression='exit_time = :zero',
            ExpressionAttributeValues={
                ':current': {'S': current_time},
                ':zero': {'S': '0'}
            }
        )
        print(f"更新车牌 {text_detected} 的出场时间成功")
    except dynamodb.exceptions.ConditionalCheckFailedException:
        # 条件失败：要么记录不存在，要么已出场，尝试插入新入场记录
        try:
            dynamodb.put_item(
                TableName=table_name,
                Item={
                    'reg_plate': {'S': text_detected},
                    'entry_time': {'S': current_time},
                    'exit_time': {'S': '0'}
                },
                # 确保仅当车牌不存在时才插入，避免并发重复
                ConditionExpression='attribute_not_exists(reg_plate)'
            )
            print(f"插入车牌 {text_detected} 的入场记录成功")
        except dynamodb.exceptions.ConditionalCheckFailedException:
            print(f"车牌 {text_detected} 的记录已存在，无需重复操作")

关键逻辑说明 #

1. 优先更新：先尝试更新对应车牌的exit_time，如果记录存在且未出场（exit_time为0），则更新为当前时间。
2. 异常处理插入：如果更新失败（记录不存在或已出场），则尝试插入新的入场记录，同时通过attribute_not_exists条件避免重复插入。
3. 原子操作：使用DynamoDB的条件表达式实现原子操作，避免并发场景下的数据冲突。

额外提示 #

• 务必填写正确的S3桶名称，否则文本识别会失败。
• 如果不需要限制「仅未出场记录可更新」，可以删除ConditionExpression和对应的ExpressionAttributeValues部分。
• 测试时可以直接在DynamoDB控制台查看记录变化，验证逻辑是否符合预期。

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