JSON转Protobuf遇时间格式不匹配ValueError问题求助

问题背景 #

将指定JSON结构转换为Protobuf时，编写的Python转换函数触发ValueError: time data '2024-05' does not match format '%Y-%m-%dT%H:%M:%S'错误。尽管打印provided_at_str显示为"2024-05-16T22:20:28"，但错误提示时间数据为'2024-05'。

示例JSON片段 #

{
  "answerUpdateRequest": {
    "entity": {
      "type": "ORGANIZATION",
      "id": "UU1234321234ID"
    },
    "answers": [
      {
        "key": "legal_company_name",
        "source": {
          "type": "DOCUMENT",
          "id": "3ea20440-83fb-43c0-b409-1dd8f68e73ec | DocumentType.application"
        },
        "provided_at": "2024-05-02T15:54:15.941988",
        "received_at": "2024-05-02T15:54:15.945350",
        "type": "TEXT",
        "value": {
          "text": "Ciccone Law, LLC"
        }
      }
    ]
  },
  "documentKey": "3ea20440-83fb-43c0-b409-1dd8f68e73ec",
  "applicationId": "1343245432",
  "activityId": "1111"
}

出错的Python转换函数片段 #

def create_answer_update_request(json_data):
    data = json_data
    print("Data is "+str(data))
    answer_update_request = events_pb2.AnswerUpdateRequest()
    answer_update_request.entity.type = model_pb2.Entity.Type.Value(data["answerUpdateRequest"]["entity"]["type"])
    answer_update_request.entity.id = data["answerUpdateRequest"]["entity"]["id"]

    # Convert answers to protobuf format
    for answer_data in data["answerUpdateRequest"]["answers"]:
        answer = answer_update_request.answers.add()
        answer.key = answer_data["key"]
        answer.source.type = answer_data["source"]["type"]
        answer.source.id = answer_data["source"]["id"]

        # Handle provided_at and received_at fields
        provided_at_str = answer_data["provided_at"].split('.')[0]
        print(provided_at_str)

        provided_at_datetime = datetime.strptime(provided_at_str, "%Y-%m-%dT%H:%M:%S")

        provided_at = Timestamp()
        provided_at.FromJsonString(provided_at_datetime.isoformat())
        print("##########provided_at############")
        answer.provided_at.CopyFrom(provided_at_datetime.isoformat())

        received_at_str = answer_data["received_at"]
        if len(received_at_str) == 7:  # Only contains 'YYYY-MM'
            received_at_str += "-01T00:00:00"  # Append default day and time
        received_at = Timestamp()
        received_at.FromJsonString(received_at_str)
        answer.received_at.CopyFrom(received_at)

        # ... 其他代码省略

错误原因分析 #

1. 时间格式兼容缺失：错误提示的'2024-05'来自某个received_at字段，当前代码仅在received_at_str长度为7时补全日期时间，若数据中存在YYYY-MM格式但长度不符合（或代码逻辑判断失误），会跳过补全直接传入Timestamp.FromJsonString()，导致解析失败。
2. 参数类型错误：answer.provided_at.CopyFrom(provided_at_datetime.isoformat())中，CopyFrom需要传入Timestamp对象，而非字符串，这会引发潜在错误。
3. 布尔值转换遗漏：JSON中布尔类型的value存储为"Yes"字符串，但代码直接赋值answer.value.boolean = answer_data["value"]["boolean"]，会因键不存在报错。

解决方案 #

修复后的完整转换函数 #

from datetime import datetime
from google.protobuf.timestamp_pb2 import Timestamp

def parse_time_str(time_str):
    # 兼容多种时间格式，自动补全缺失部分
    formats = [
        "%Y-%m-%dT%H:%M:%S",
        "%Y-%m-%d",
        "%Y-%m"
    ]
    for fmt in formats:
        try:
            dt = datetime.strptime(time_str, fmt)
            # 补全缺失的时分秒或日期
            if fmt == "%Y-%m":
                return dt.replace(day=1, hour=0, minute=0, second=0)
            elif fmt == "%Y-%m-%d":
                return dt.replace(hour=0, minute=0, second=0)
            return dt
        except ValueError:
            continue
    raise ValueError(f"Unsupported time format: {time_str}")

def create_answer_update_request(json_data):
    data = json_data
    answer_update_request = events_pb2.AnswerUpdateRequest()
    answer_update_request.entity.type = model_pb2.Entity.Type.Value(data["answerUpdateRequest"]["entity"]["type"])
    answer_update_request.entity.id = data["answerUpdateRequest"]["entity"]["id"]

    for answer_data in data["answerUpdateRequest"]["answers"]:
        answer = answer_update_request.answers.add()
        answer.key = answer_data["key"]
        answer.source.type = answer_data["source"]["type"]
        answer.source.id = answer_data["source"]["id"]

        # 处理provided_at字段
        provided_at_str = answer_data["provided_at"].split('.')[0]
        try:
            provided_at_dt = parse_time_str(provided_at_str)
        except ValueError as e:
            raise ValueError(f"Invalid provided_at for key {answer_data['key']}: {str(e)}")
        provided_at_ts = Timestamp()
        provided_at_ts.FromDatetime(provided_at_dt)
        answer.provided_at.CopyFrom(provided_at_ts)

        # 处理received_at字段
        received_at_str = answer_data["received_at"].split('.')[0]
        try:
            received_at_dt = parse_time_str(received_at_str)
        except ValueError as e:
            raise ValueError(f"Invalid received_at for key {answer_data['key']}: {str(e)}")
        received_at_ts = Timestamp()
        received_at_ts.FromDatetime(received_at_dt)
        answer.received_at.CopyFrom(received_at_ts)

        # 处理answer类型校验
        if answer_data["type"] not in events_pb2.Answer.Type.keys():
            raise ValueError(f"Invalid answer type: {answer_data['type']}")
        answer.type = events_pb2.Answer.Type.Value(answer_data["type"])

        # 处理不同类型的value
        if answer_data["type"] == "TEXT":
            answer.value.text = answer_data["value"]["text"]
        elif answer_data["type"] == "BOOLEAN":
            # 将字符串转换为布尔值
            answer.value.boolean = answer_data["value"]["text"].strip().lower() == "yes"
        # 可扩展其他类型的处理逻辑

    serialized_data = answer_update_request.SerializeToString()
    return serialized_data

关键修复点说明 #

• 统一时间解析逻辑：新增parse_time_str函数，支持YYYY-MM、YYYY-MM-DD、YYYY-MM-DDTHH:MM:SS三种格式，自动补全缺失的日期或时分秒。
• 修复Timestamp赋值错误：使用Timestamp.FromDatetime()将datetime对象转换为Protobuf Timestamp，再通过CopyFrom赋值，避免字符串类型不匹配问题。
• 布尔值转换修复：将JSON中的"Yes"/"No"字符串转换为布尔值，适配Protobuf的布尔类型字段。
• 错误定位优化：报错时包含当前处理的answer key，方便快速定位问题数据。

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