Spaces:
Sleeping
Sleeping
| import requests | |
| import hashlib | |
| import datetime | |
| import streamlit as st | |
| import pandas as pd | |
| import openai | |
| import os | |
| from langchain.chat_models import ChatOpenAI | |
| from langchain.chains import LLMChain | |
| from langchain.prompts import PromptTemplate | |
| # Make sure OpenAI's API key is set | |
| openai.api_key = os.getenv("OPENAI_API_KEY") | |
| if not openai.api_key: | |
| raise ValueError("OpenAI API key not set. Check your environment variables.") | |
| # Hotelbeds API credentials | |
| hotelbeds_api_key = "95ce3e45a02fc6fd9720ecc013a6f674" | |
| hotelbeds_api_secret = "785150201f" | |
| # Function to generate the API signature | |
| def generate_signature(): | |
| local_timestamp = int(datetime.datetime.now().timestamp()) | |
| assemble = hotelbeds_api_key + hotelbeds_api_secret + str(local_timestamp) | |
| signature = hashlib.sha256(assemble.encode()).hexdigest() | |
| return signature | |
| # Function to get geo-coordinates from a location name using OpenAI API through Langchain | |
| import re | |
| def get_coordinates(location_name): | |
| prompt_template = PromptTemplate( | |
| input_variables=["location"], | |
| template="What are the exact latitude and longitude of {location}? Please provide the values in decimal degrees." | |
| ) | |
| llm = ChatOpenAI(model="gpt-4", openai_api_key=openai.api_key) | |
| chain = LLMChain(llm=llm, prompt=prompt_template) | |
| result = chain.run(location=location_name).strip() | |
| try: | |
| # Use regular expressions to find latitude and longitude in the text | |
| lat_lon_match = re.findall(r"(-?\d+\.\d+)°?\s*([NSEW])", result) | |
| if lat_lon_match and len(lat_lon_match) == 2: | |
| # Extract latitude and longitude from regex matches | |
| latitude, lat_dir = lat_lon_match[0] | |
| longitude, lon_dir = lat_lon_match[1] | |
| # Convert the strings to float and adjust sign according to the hemisphere | |
| latitude = float(latitude) * (-1 if lat_dir in ['S', 's'] else 1) | |
| longitude = float(longitude) * (-1 if lon_dir in ['W', 'w'] else 1) | |
| return latitude, longitude | |
| else: | |
| raise ValueError("Could not find both latitude and longitude in the response") | |
| except Exception as e: | |
| st.error(f"Unable to parse coordinates from the AI's response: '{result}'. Error: {str(e)}") | |
| return None, None | |
| # Function to make the POST request to the Hotelbeds API and handle pagination | |
| def fetch_activities(latitude, longitude, from_date, to_date, language="en", paxes=[{"age": 5}, {"age": 70}], order="DEFAULT"): | |
| url = "https://api.test.hotelbeds.com/activity-api/3.0/activities/availability" | |
| signature = generate_signature() | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Api-Key": hotelbeds_api_key, | |
| "X-Signature": signature, | |
| "Accept": "application/json", | |
| } | |
| payload = { | |
| "filters": [ | |
| { | |
| "searchFilterItems": [ | |
| { | |
| "type": "gps", | |
| "latitude": latitude, | |
| "longitude": longitude | |
| } | |
| ] | |
| } | |
| ], | |
| "from": from_date, | |
| "to": to_date, | |
| "language": language, | |
| "paxes": paxes, | |
| "order": order | |
| } | |
| activities = [] | |
| page = 1 | |
| items_per_page = 100 | |
| total_items = 0 | |
| while True: | |
| payload['pagination'] = {"page": page, "itemsPerPage": items_per_page} | |
| response = requests.post(url, headers=headers, json=payload) | |
| data = response.json() | |
| if response.status_code != 200 or 'errors' in data and data['errors']: | |
| st.error(f"Error fetching activities: {response.status_code}, Details: {data['errors']}") | |
| return None | |
| if 'pagination' in data: | |
| total_items = data['pagination']['totalItems'] | |
| items_per_page = data['pagination']['itemsPerPage'] | |
| activities.extend(data.get('activities', [])) | |
| if 'pagination' not in data or total_items <= items_per_page * page: | |
| break | |
| page += 1 | |
| return activities | |
| # Streamlit app | |
| st.title("Hotelbeds Activity Availability Checker") | |
| location_name = st.text_input("Location Name", value="New York City") | |
| from_date = st.date_input("From Date", value=pd.to_datetime("2024-06-20")) | |
| to_date = st.date_input("To Date", value=pd.to_datetime("2024-06-24")) | |
| if st.button("Check Availability"): | |
| latitude, longitude = get_coordinates(location_name) | |
| activities = fetch_activities(latitude, longitude, from_date.isoformat(), to_date.isoformat()) | |
| if activities: | |
| activity_details = [] | |
| for activity in activities: | |
| name = activity['content']['name'] | |
| adult_price = next( | |
| (amount['amount'] for amount in activity['modalities'][0]['amountsFrom'] if amount['paxType'] == 'ADULT'), | |
| None | |
| ) | |
| if adult_price is not None: | |
| activity_details.append(f"{name} - ${adult_price:.2f}") | |
| else: | |
| activity_details.append(name) | |
| st.json(activity_details) | |
| st.write(f"Total number of activities: {len(activity_details)}") | |
| else: | |
| st.error("No activities found or an error occurred.") |