the code :
# define a route for translation
@app.route('/webhook', methods=['POST'])
def webhook():
req = request.get_json()
if 'fulfillmentInfo' not in req or 'tag' not in req['fulfillmentInfo']:
return jsonify({"error": "Invalid request, tag missing"}), 400
tag = req['fulfillmentInfo']['tag']
parameters = req.get('sessionInfo', {}).get('parameters', {}) # Correct parameters extraction
if tag == "TranslateText_fulfillment":
return handle_translate_request(parameters)
else:
return jsonify({"error": "Unknown tag"}), 400
def handle_translate_request(parameters):
text = parameters.get('textToTranslate')
language_code = parameters.get('language')
if not text or not language_code:
return jsonify({"error": "Missing text or language code"}), 400
# Check if the language code is valid
valid_languages = ['en', 'tw', 'gaa', 'ee', 'fat', 'dag', 'gur', 'yo', 'ki', 'luo', 'mer']
if language_code not in valid_languages:
return jsonify({"error": "Invalid language code"}), 400
return translate_text(text, language_code)
def translate_text(text, language_code):
api_url = "https://translation-api.ghananlp.org/v1/translate"
headers = {
'Content-Type': 'application/json',
'Ocp-Apim-Subscription-Key': TRANSLATION_API_KEY
}
payload = {
"in": text,
"lang": f"{language_code}-en" # Assuming translation to English
}
try:
response = requests.post(api_url, json=payload, headers=headers)
response.raise_for_status()
response_json = response.json()
logging.debug(f"API response JSON: {response_json}")
if isinstance(response_json, str):
translated_text = response_json
elif 'translatedText' in response_json:
translated_text = response_json['translatedText']
else:
logging.error("Translation API did not return 'translatedText' or a direct string.")
return jsonify({"error": "Translation failed"}), 500
return jsonify({
"fulfillment_response": {
"messages": [{
"text": {
"text": [translated_text]
}
}]
}
})
except requests.exceptions.RequestException as e:
logging.error(f"Error during translation API request: {e}")
return jsonify({"error": str(e)}), 500
def test_webhook():
try:
# Adjust the test payload to match how Dialogflow CX sends parameters
test_payload = {
"fulfillmentInfo": {
"tag": "TranslateText_fulfillment"
},
"sessionInfo": { # Ensure parameters are nested within sessionInfo
"parameters": {
"textToTranslate": "Hello, how are you?",
"language": "tw"
}
}
}
# Using test_request_context to simulate a POST request
with app.test_request_context('/webhook', method='POST', json=test_payload):
response = webhook()
if isinstance(response, tuple):
response, status = response # Handling the tuple response
else:
status = 200 # Default status if response isn't a tuple
# Extracting JSON data from the response
response_data = response.get_json()
print(response_data, status)
except Exception as e:
print(f"Error during webhook testing: {e}")
