MSGraphAPI / Files.Backup.AuthWorksapp.py
awacke1's picture
Rename app.py to Files.Backup.AuthWorksapp.py
320f734 verified
raw
history blame
6.26 kB
import os
import streamlit as st
import requests
import msal
import urllib.parse
import base64
import hashlib
import secrets
# πŸ€“ 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'
REDIRECT_URI = 'https://huggingface.co/spaces/awacke1/MSGraphAPI'
# Define product to scope mapping
PRODUCT_SCOPES = {
"πŸ“§ Outlook": ['Mail.Read', 'Mail.Send', 'Calendars.ReadWrite'],
"πŸ“’ OneNote": ['Notes.Read', 'Notes.Create'],
"πŸ“Š Excel": ['Files.ReadWrite.All'],
"πŸ“„ Word": ['Files.ReadWrite.All'],
"πŸ—ƒοΈ SharePoint": ['Sites.Read.All', 'Sites.ReadWrite.All'],
"πŸ“… Teams": ['Team.ReadBasic.All', 'Channel.ReadBasic.All'],
"πŸ’¬ Viva": ['Analytics.Read'],
"πŸš€ Power Platform": ['Flow.Read.All'],
"🧠 Copilot": ['Cognitive.Read'],
"πŸ—‚οΈ OneDrive": ['Files.ReadWrite.All'],
"πŸ’‘ PowerPoint": ['Files.ReadWrite.All'],
"πŸ“š Microsoft Bookings": ['Bookings.Read.All', 'Bookings.ReadWrite.All'],
"πŸ““ Loop": ['Files.ReadWrite.All'],
"πŸ—£οΈ Translator": ['Translation.Read'],
"πŸ“‹ To Do & Planner": ['Tasks.ReadWrite'],
"πŸ”— Azure OpenAI Service": ['AzureAIServices.ReadWrite.All']
}
# Always include these base scopes
BASE_SCOPES = ['User.Read', 'openid', 'profile', 'offline_access']
# Function to generate PKCE code verifier and challenge
def generate_pkce_codes():
code_verifier = secrets.token_urlsafe(128)[:128]
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).decode().rstrip('=')
return code_verifier, code_challenge
def get_msal_app(code_challenge=None):
return msal.PublicClientApplication(
client_id=APPLICATION_ID_KEY,
authority=AUTHORITY_URL
)
# Function to get access token
def get_access_token(code, code_verifier):
client_instance = get_msal_app()
try:
result = client_instance.acquire_token_by_authorization_code(
code=code,
scopes=st.session_state.get('scopes', BASE_SCOPES),
redirect_uri=REDIRECT_URI,
code_verifier=code_verifier
)
if 'access_token' in result:
return result['access_token']
else:
error_description = result.get('error_description', 'No error description provided')
raise Exception(f"Error acquiring token: {error_description}")
except Exception as e:
st.error(f"Exception in get_access_token: {str(e)}")
raise
# Main application function
def main():
st.title("πŸ¦„ MS Graph API with AI & Cloud Integration for M365")
# Sidebar for product selection
st.sidebar.title("πŸ“ M365 Products")
st.sidebar.write("Select products to integrate:")
selected_products = {}
for product in PRODUCT_SCOPES.keys():
selected = st.sidebar.checkbox(product)
if selected:
selected_products[product] = True
# Dynamically build scopes based on selected products
scopes = BASE_SCOPES.copy()
for product in selected_products:
scopes.extend(PRODUCT_SCOPES[product])
scopes = list(set(scopes)) # Remove duplicates
st.session_state['scopes'] = scopes
# Authentication flow
if 'access_token' not in st.session_state:
if 'code_verifier' not in st.session_state:
code_verifier, code_challenge = generate_pkce_codes()
st.session_state['code_verifier'] = code_verifier
else:
code_verifier = st.session_state['code_verifier']
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).decode().rstrip('=')
client_instance = get_msal_app()
auth_url = client_instance.get_authorization_request_url(
scopes=scopes,
redirect_uri=REDIRECT_URI,
code_challenge=code_challenge,
code_challenge_method="S256"
)
st.write('πŸ‘‹ Please [click here]({}) to log in and authorize the app.'.format(auth_url))
# Check for authorization code in query parameters
query_params = st.query_params
if 'code' in query_params:
code = query_params.get('code')
st.write('πŸ”‘ Authorization Code Obtained:', code[:10] + '...')
try:
access_token = get_access_token(code, code_verifier)
st.session_state['access_token'] = access_token
st.success("Access token acquired successfully!")
st.rerun()
except Exception as e:
st.error(f"Error acquiring access token: {str(e)}")
st.stop()
else:
# User is authenticated, proceed with the app
access_token = st.session_state['access_token']
# Greet the user
user_info = get_user_info(access_token)
if user_info:
st.sidebar.write(f"πŸ‘‹ Hello, {user_info.get('displayName', 'User')}!")
# Handle selected products
if selected_products:
st.header("🧩 M365 Product Integrations")
for product in selected_products:
st.subheader(f"{product}")
handle_product_integration(access_token, product)
else:
st.write("No products selected. Please select products from the sidebar.")
# Function to get user info
def get_user_info(access_token):
headers = {'Authorization': f'Bearer {access_token}'}
response = requests.get('https://graph.microsoft.com/v1.0/me', headers=headers)
if response.status_code == 200:
return response.json()
else:
st.error('Failed to fetch user info.')
return None
# Function to handle product integration
def handle_product_integration(access_token, product):
# Implement the integration logic for each product here
# This is a placeholder - you'll need to implement the actual API calls
st.write(f"Integrating {product}...")
# Example: if product == "πŸ“§ Outlook": get_outlook_data(access_token)
# Run the main function
if __name__ == "__main__":
main()