Gmail Pub/sub push -уведомления достигают облачного запуска, но не мое приложение Python webhook ⇐ Python

[Anonymous]

Моя цель состоит в том, чтобы разработать сценарии Python, которые читают текущие сообщения в почтовом ящике Gmail в Gmail, и прочитать живые сообщения, проходящие через почтовый ящик клиента Gmail. Эти данные в конечном итоге будут проанализированы с помощью другого сценария, в конечном итоге классифицируются с целью защиты клиента от вредоносных электронных писем. Pythonfileemp.py имеет дело с историческими сообщениями в почтовых ящиках, в то время как webhook.py имеет дело с живыми сообщениями, проходящими через почтовые ящики. Журналы Показывают пост 200 успех, когда отправляются в прямом эфире тестовых электронных писем (тестовые электронные письма регистрируются правильно). Кроме того, при тестировании WebHook локально через команду Curl в PowerShell, WebHook успешно запускается. По какой-то причине тестовые электронные письма просто не видны в моем журнале /выводе Webhook в Pycharm. < /P>
Информация, которая может быть полезна: < /p>

[*] Webhook развернут в Google Cloud. Настройка, чтобы указать на https://webhook-service-xxxx.australia- ... pp/webhook.
Webhook в контейнер с использованием Docker с Dockerfile и требованиями. /> < /ul>
Почему это происходит и как вы это исправляете? Заранее спасибо < /p>
Здесь я включил минимальную резо сценария. Этот сценарий фокусируется на основной функциональности, необходимой для получения и журнала Gmail Push -уведомлений через Google Cloud Pub/sub. Эта уменьшенная версия опускает потоки аутентификации, управление токенами и внешние зависимости. < /P>
from flask import Flask, request, jsonify
import base64
import logging
import json

app = Flask(__name__)
logging.basicConfig(level=logging.DEBUG)

@app.route('/webhook', methods=['POST'])
def gmail_webhook():
try:
# Log raw request data for debugging
logging.debug(f"Headers: {dict(request.headers)}")
logging.debug(f"Body: {request.data.decode('utf-8')}")

# Parse JSON payload
data = request.get_json(silent=True)
if not isinstance(data, dict):
logging.error("Invalid JSON format received.")
return jsonify({"error": "Invalid JSON format received."}), 400

# Extract and decode the message data
message_data = data.get("message", {}).get("data")
if message_data:
decoded_bytes = base64.urlsafe_b64decode(message_data)
decoded_str = decoded_bytes.decode("utf-8")
logging.info(f"Decoded message: {decoded_str}")
else:
logging.warning("No message data found in the request.")
return jsonify({"error": "No message data found."}), 400

return jsonify({"status": "Success", "message": decoded_str}), 200

except Exception as e:
logging.error(f"Unhandled error: {e}")
return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
app.run(host='0.0.0.0', port=8080, debug=True)
< /code>
output: < /p>

Приложение для обслуживания колбы 'abc' < /li>
Режим отладки: on
info: werkzeug: предупреждение: это сервер разработки. Не используйте его в развертывании производства. Вместо этого используйте производственный сервер WSGI. /> Информация: werkzeug: нажмите Ctrl+C, чтобы уйти
info: werkzeug: * перезагрузка со статистиком
предупреждение: werkzeug: * отладчик активен!
info: werkzeug: * Debugger Pin: 142-623-917 < /li>
< /ul: 142-623-917 < /li>
< /u>

Подробнее здесь: https://stackoverflow.com/questions/796 ... ebhook-app