-
-
Notifications
You must be signed in to change notification settings - Fork 101
/
hook.py
105 lines (91 loc) · 4.25 KB
/
hook.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
import os
import logging
from heyoo import WhatsApp
from dotenv import load_dotenv
from flask import Flask, request, make_response
# Initialize Flask App
app = Flask(__name__)
# Load .env file
load_dotenv()
messenger = WhatsApp(os.getenv("TOKEN"), phone_number_id=os.getenv("PHONE_NUMBER_ID"))
VERIFY_TOKEN = "30cca545-3838-48b2-80a7-9e43b1ae8ce4"
# Logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
@app.get("/")
def verify_token():
if request.args.get("hub.verify_token") == VERIFY_TOKEN:
logging.info("Verified webhook")
response = make_response(request.args.get("hub.challenge"), 200)
response.mimetype = "text/plain"
return response
logging.error("Webhook Verification failed")
return "Invalid verification token"
@app.post("/")
def hook():
# Handle Webhook Subscriptions
data = request.get_json()
logging.info("Received webhook data: %s", data)
changed_field = messenger.changed_field(data)
if changed_field == "messages":
new_message = messenger.is_message(data)
if new_message:
mobile = messenger.get_mobile(data)
name = messenger.get_name(data)
message_type = messenger.get_message_type(data)
logging.info(
f"New Message; sender:{mobile} name:{name} type:{message_type}"
)
if message_type == "text":
message = messenger.get_message(data)
name = messenger.get_name(data)
logging.info("Message: %s", message)
messenger.send_message(f"Hi {name}, nice to connect with you", mobile)
elif message_type == "interactive":
message_response = messenger.get_interactive_response(data)
interactive_type = message_response.get("type")
message_id = message_response[interactive_type]["id"]
message_text = message_response[interactive_type]["title"]
logging.info(f"Interactive Message; {message_id}: {message_text}")
elif message_type == "location":
message_location = messenger.get_location(data)
message_latitude = message_location["latitude"]
message_longitude = message_location["longitude"]
logging.info("Location: %s, %s", message_latitude, message_longitude)
elif message_type == "image":
image = messenger.get_image(data)
image_id, mime_type = image["id"], image["mime_type"]
image_url = messenger.query_media_url(image_id)
image_filename = messenger.download_media(image_url, mime_type)
logging.info(f"{mobile} sent image {image_filename}")
elif message_type == "video":
video = messenger.get_video(data)
video_id, mime_type = video["id"], video["mime_type"]
video_url = messenger.query_media_url(video_id)
video_filename = messenger.download_media(video_url, mime_type)
logging.info(f"{mobile} sent video {video_filename}")
elif message_type == "audio":
audio = messenger.get_audio(data)
audio_id, mime_type = audio["id"], audio["mime_type"]
audio_url = messenger.query_media_url(audio_id)
audio_filename = messenger.download_media(audio_url, mime_type)
logging.info(f"{mobile} sent audio {audio_filename}")
elif message_type == "document":
file = messenger.get_document(data)
file_id, mime_type = file["id"], file["mime_type"]
file_url = messenger.query_media_url(file_id)
file_filename = messenger.download_media(file_url, mime_type)
logging.info(f"{mobile} sent file {file_filename}")
else:
logging.info(f"{mobile} sent {message_type} ")
logging.info(data)
else:
delivery = messenger.get_delivery(data)
if delivery:
logging.info(f"Message : {delivery}")
else:
logging.info("No new message")
return "OK", 200
if __name__ == "__main__":
app.run(port=5000, debug=False)