def image_processor()

in gemini/autocal/image-processor-function/main.py [0:0]
67 lines of code
9 McCabe index (conditional complexity)

def image_processor(cloud_event: CloudEvent) -> None:
    """Triggered by a change to a Firestore document.

    Args:
        cloud_event: cloud event with information on the firestore event trigger
    """
    firestore_payload = firestoredata.DocumentEventData()
    # Not sure how to parse the protobuf without using this protected method
    # pylint: disable=protected-access
    firestore_payload._pb.ParseFromString(cloud_event.data)

    print(f"Function triggered by change to: {cloud_event['source']}")

    # Again, not sure how to do this without accessing the fields directly
    # pylint: disable=no-member
    gcs_url = firestore_payload.value.fields.get("image").string_value
    mime_type = firestore_payload.value.fields.get("type").string_value
    document_id = firestore_payload.value.fields.get("ID").string_value

    if not all([gcs_url, mime_type, document_id]):
        print(f"Missing required fields in document: {cloud_event.data}")
        return

    # Get the current date and time
    current_datetime = datetime.datetime.now().isoformat()

    # Format the prompt with the current date and time
    prompt = PROMPT_TEMPLATE.format(current_datetime=current_datetime)
    try:
        response = client.models.generate_content(
            model=MODEL_ID,
            contents=[
                Part.from_uri(file_uri=gcs_url, mime_type=mime_type),
                prompt,
            ],
            config=GenerateContentConfig(
                response_mime_type="application/json", response_schema=response_schema
            ),
        )
    except (InvalidArgument, PermissionDenied, NotFound, ResourceExhausted) as e:
        # Handle Gemini API errors
        print(f"Gemini API error: {e}")
        doc_ref = db.collection("state").document(document_id)
        firestore_document = {"error": True, "message": f"Gemini API error: {e}"}
        doc_ref.set(firestore_document, merge=True)
        raise e
    except ValueError as e:
        # Handle file/URI issues
        print(f"Invalid file URI or MIME type: {e}")
        doc_ref = db.collection("state").document(document_id)
        firestore_document = {
            "error": True,
            "message": f"Invalid file URI or MIME type: {e}",
        }
        doc_ref.set(firestore_document, merge=True)
        raise e
    except GoogleAPICallError as e:
        # Handle other Google API errors
        print(f"General Google API error: {e}")
        doc_ref = db.collection("state").document(document_id)
        firestore_document = {
            "error": True,
            "message": f"General Google API error: {e}",
        }
        doc_ref.set(firestore_document, merge=True)
        raise e
    except Exception as e:
        # Catch any other unexpected errors
        print(f"An unexpected error occurred: {e}")
        doc_ref = db.collection("state").document(document_id)
        firestore_document = {
            "error": True,
            "message": f"An unexpected error occurred: {e}",
        }
        doc_ref.set(firestore_document, merge=True)
        raise e

    print(f"Raw Gemini Response: {response.text}")
    event_data = json.loads(response.text)
    print(event_data)

    # firestore document
    firestore_document = {"processed": True, "event": event_data}

    # Write the event data to Firestore
    try:
        doc_ref = db.collection("state").document(document_id)
        doc_ref.set(firestore_document, merge=True)
        print(f"Successfully wrote data to Firestore document: {document_id}")
    except GoogleAPICallError as e:
        print(f"Error writing to Firestore: {e}")
        return