Spaces:
Sleeping
Sleeping
| import os | |
| import streamlit as st | |
| import requests | |
| import msal | |
| # π€ Load environment variables (Ensure these are set!) | |
| APPLICATION_ID_KEY = os.getenv('APPLICATION_ID_KEY') | |
| CLIENT_SECRET_KEY = os.getenv('CLIENT_SECRET_KEY') | |
| AUTHORITY_URL = 'https://login.microsoftonline.com/common' # Use 'common' for multi-tenant apps | |
| REDIRECT_URI = 'http://localhost:8501' # π Make sure this matches your app's redirect URI | |
| # π― Define the scopes your app will need | |
| SCOPES = ['User.Read', 'Calendars.ReadWrite'] | |
| # π οΈ Initialize the MSAL client | |
| def get_msal_app(): | |
| return msal.ConfidentialClientApplication( | |
| client_id=APPLICATION_ID_KEY, | |
| client_credential=CLIENT_SECRET_KEY, | |
| authority=AUTHORITY_URL | |
| ) | |
| # π Acquire access token using authorization code | |
| def get_access_token(code): | |
| client_instance = get_msal_app() | |
| result = client_instance.acquire_token_by_authorization_code( | |
| code=code, | |
| scopes=SCOPES, | |
| redirect_uri=REDIRECT_URI | |
| ) | |
| if 'access_token' in result: | |
| return result['access_token'] | |
| else: | |
| st.error('Error acquiring token: ' + str(result.get('error_description'))) | |
| st.stop() | |
| # πββοΈ Main application function | |
| def main(): | |
| st.title("π¦ Simple Streamlit App with MS Graph API") | |
| # π Sidebar navigation | |
| st.sidebar.title("Navigation") | |
| menu = st.sidebar.radio("Go to", [ | |
| "1οΈβ£ Dashboard with Widgets", | |
| "π Landing Page", | |
| "π Upcoming Events", | |
| "π Schedule", | |
| "π Agenda", | |
| "π Event Details", | |
| "β Add Event", | |
| "π Filter By" | |
| ]) | |
| # π Authentication | |
| if 'access_token' not in st.session_state: | |
| # π΅οΈββοΈ Check for authorization code in query parameters | |
| query_params = st.get_query_params() | |
| if 'code' in query_params: | |
| code = query_params['code'][0] | |
| st.write('π Acquiring access token...') | |
| access_token = get_access_token(code) | |
| st.session_state['access_token'] = access_token | |
| st.rerun() # Reload the app to clear the code from URL | |
| else: | |
| # π’ Prompt user to log in | |
| client_instance = get_msal_app() | |
| authorization_url = client_instance.get_authorization_request_url( | |
| scopes=SCOPES, | |
| redirect_uri=REDIRECT_URI | |
| ) | |
| st.write('π Please [click here]({}) to log in and authorize the app.'.format(authorization_url)) | |
| st.stop() | |
| else: | |
| # π₯³ User is authenticated, proceed with the app | |
| access_token = st.session_state['access_token'] | |
| headers = {'Authorization': 'Bearer ' + access_token} | |
| # π€ Greet the user | |
| response = requests.get('https://graph.microsoft.com/v1.0/me', headers=headers) | |
| if response.status_code == 200: | |
| user_info = response.json() | |
| st.sidebar.write(f"π Hello, {user_info['displayName']}!") | |
| else: | |
| st.error('Failed to fetch user info.') | |
| st.write(response.text) | |
| # ποΈ Handle menu options | |
| if menu == "1οΈβ£ Dashboard with Widgets": | |
| st.header("1οΈβ£ Dashboard with Widgets") | |
| st.write("Widgets will be displayed here. ποΈ") | |
| # Add your widgets here | |
| elif menu == "π Landing Page": | |
| st.header("π Landing Page") | |
| st.write("Welcome to the app! π₯³") | |
| # Add landing page content here | |
| elif menu == "π Upcoming Events": | |
| st.header("π Upcoming Events") | |
| events = get_upcoming_events(access_token) | |
| for event in events: | |
| st.write(f"π {event['subject']} on {event['start']['dateTime']}") | |
| # Display upcoming events | |
| elif menu == "π Schedule": | |
| st.header("π Schedule") | |
| schedule = get_schedule(access_token) | |
| st.write(schedule) | |
| # Display schedule | |
| elif menu == "π Agenda": | |
| st.header("π Agenda") | |
| st.write("Your agenda for today. π") | |
| # Display agenda | |
| elif menu == "π Event Details": | |
| st.header("π Event Details") | |
| event_id = st.text_input("Enter Event ID") | |
| if event_id: | |
| event_details = get_event_details(access_token, event_id) | |
| st.write(event_details) | |
| # Display event details based on ID | |
| elif menu == "β Add Event": | |
| st.header("β Add Event") | |
| event_subject = st.text_input("Event Subject") | |
| event_start = st.date_input("Event Start Date") | |
| event_start_time = st.time_input("Event Start Time") | |
| event_end = st.date_input("Event End Date") | |
| event_end_time = st.time_input("Event End Time") | |
| if st.button("Add Event"): | |
| event_details = { | |
| "subject": event_subject, | |
| "start": { | |
| "dateTime": f"{event_start}T{event_start_time}", | |
| "timeZone": "UTC" | |
| }, | |
| "end": { | |
| "dateTime": f"{event_end}T{event_end_time}", | |
| "timeZone": "UTC" | |
| } | |
| } | |
| add_event(access_token, event_details) | |
| st.success("Event added successfully! π") | |
| # Form to add new event | |
| elif menu == "π Filter By": | |
| st.header("π Filter Events") | |
| filter_criteria = st.text_input("Enter filter criteria") | |
| if filter_criteria: | |
| filtered_events = filter_events(access_token, filter_criteria) | |
| for event in filtered_events: | |
| st.write(f"π {event['subject']} on {event['start']['dateTime']}") | |
| # Filter events based on criteria | |
| else: | |
| st.write("Please select a menu option.") | |
| # π Function to get upcoming events | |
| def get_upcoming_events(access_token): | |
| headers = {'Authorization': 'Bearer ' + access_token} | |
| response = requests.get('https://graph.microsoft.com/v1.0/me/events?$orderby=start/dateTime&$top=10', headers=headers) | |
| if response.status_code == 200: | |
| events = response.json().get('value', []) | |
| return events | |
| else: | |
| st.error('Failed to fetch upcoming events.') | |
| st.write(response.text) | |
| return [] | |
| # π Function to get schedule (Placeholder) | |
| def get_schedule(access_token): | |
| # Implement API call to get schedule | |
| return "π Your schedule goes here." | |
| # β Function to add a new event | |
| def add_event(access_token, event_details): | |
| headers = { | |
| 'Authorization': 'Bearer ' + access_token, | |
| 'Content-Type': 'application/json' | |
| } | |
| response = requests.post('https://graph.microsoft.com/v1.0/me/events', headers=headers, json=event_details) | |
| if response.status_code == 201: | |
| st.success('Event created successfully! π') | |
| else: | |
| st.error('Failed to create event.') | |
| st.write(response.text) | |
| # π Function to get event details | |
| def get_event_details(access_token, event_id): | |
| headers = {'Authorization': 'Bearer ' + access_token} | |
| response = requests.get(f'https://graph.microsoft.com/v1.0/me/events/{event_id}', headers=headers) | |
| if response.status_code == 200: | |
| event = response.json() | |
| return event | |
| else: | |
| st.error('Failed to fetch event details.') | |
| st.write(response.text) | |
| return {} | |
| # π Function to filter events | |
| def filter_events(access_token, filter_criteria): | |
| headers = {'Authorization': 'Bearer ' + access_token} | |
| # Implement filtering logic based on criteria | |
| response = requests.get(f"https://graph.microsoft.com/v1.0/me/events?$filter=startswith(subject,'{filter_criteria}')", headers=headers) | |
| if response.status_code == 200: | |
| events = response.json().get('value', []) | |
| return events | |
| else: | |
| st.error('Failed to filter events.') | |
| st.write(response.text) | |
| return [] | |
| # π Run the main function | |
| if __name__ == "__main__": | |
| main() | |