#!pip install -q specklepy # lone github repro & add to sys paths import gradio as gr import os import requests from specklepy.api.client import SpeckleClient from specklepy.api.credentials import get_default_account, get_local_accounts from specklepy.transports.server import ServerTransport from specklepy.api import operations from specklepy.objects.geometry import Polyline, Point from huggingface_hub import webhook_endpoint, WebhookPayload from fastapi import Request import requests import datetime import json # dictionary that collects tags that can be added to commit messages. # key -> column name in notion # value -> possible values tag_dict = {"status-message":["WIP", "ReviewNeeded", "Final"]} # tag part seperator -> #+ , other_tags def extract_branch_info(stream_id, server_url=None, token=None): if server_url and token: client = SpeckleClient(host=server_url) client.authenticate_with_token(token=token) branches = client.branch.list(stream_id, 100) branch_info = [] for branch in branches: print (branch.name) branch_data = { "Name": branch.name, "description": branch.description, "url": f"{server_url}/streams/{stream_id}/branches/{branch.name.replace('/', '%2F').replace('+', '%2B')}" } # Determine sub-variant #if branch.name.startswith(("template_geometry", "graph_geometry")): if '+' in branch.name: slash_index = branch.name.rfind('/') plus_index = branch.name.find('+', slash_index) sub_variant = branch.name[slash_index + 1:plus_index] else: sub_variant = "default" branch_data["sub-variant"] = sub_variant # Get commits for the branch #try: commits = client.branch.get(stream_id, branch.name).commits.items if commits: latest_commit = commits[0] branch_data["updated"] = latest_commit.createdAt branch_data["commit_url"] = f"{server_url}streams/{stream_id}/commits/{latest_commit.id}" branch_data["commit_message"] = latest_commit.message branch_data["author"] = latest_commit.authorName # Check if the commit message contains '#+' and then extract tags if '#+' in latest_commit.message: tags_part = latest_commit.message.split("#+")[1] branch_data["status-message"] = [tg for tg in tag_dict["status-message"] if tg in tags_part] else: branch_data["status-message"] = [] #except Exception as e: # print(f"Error fetching commits for branch {branch.name}: {str(e)}") branch_info.append(branch_data) return branch_info def update_select_options(database_id, headers, branch_names): update_payload = { "properties": { "depends-on": { "multi_select": { "options": [{"name": name} for name in branch_names] } } } } response = requests.patch(f"https://api.notion.com/v1/databases/{database_id}", headers=headers, json=update_payload) response.raise_for_status() def sync_to_notion(token, database_id, branch_data): base_url = "https://api.notion.com/v1" headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28", "Content-Type": "application/json" } # Extract all branch names for the "depends-on" options branch_names = [branch['Name'] for branch in branch_data] # Update the "depends-on" multi-select options to match branch names update_select_options(database_id, headers, branch_names) try: # Fetch existing data from Notion database response = requests.post(f"{base_url}/databases/{database_id}/query", headers=headers) response.raise_for_status() pages = response.json().get('results', []) # Create a dictionary for the latest update time of each branch branch_details = {branch['Name']: {'updated': branch.get('updated', ''), 'commit_message': branch.get('commit_message', '')} for branch in branch_data} # Status color mapping status_colors = { "outdated": "red", "up-to-date": "green", "empty": "gray" } # Process each branch and update or create Notion rows print("branchData", branch_data) for branch in branch_data: # Find the corresponding page in Notion branch_update_time = branch_details.get(branch['Name'], {}).get('updated', '') page_id = None existing_depends_on = [] for page in pages: notion_name = page['properties']['Name']['title'][0]['text']['content'] if page['properties']['Name']['title'] else '' if notion_name == branch['Name']: page_id = page['id'] # Retain the existing value of "depends-on" existing_depends_on = [dep['name'] for dep in page['properties'].get('depends-on', {}).get('multi_select', [])] break # Determine status based on dependencies status_tag = "up-to-date" # Default status for dependency_name in existing_depends_on: dependency_info = branch_details.get(dependency_name, {}) branch_update_time = branch_details.get(branch['Name'], {}).get('updated', '') # Check if the dependency is more recent and if its commit message does not contain 'isConnected' if dependency_info.get('updated', '') > branch_update_time and "isConnected" not in dependency_info.get('commit_message', ''): status_tag = "outdated" break # If there's no update information, set status to 'empty' if not branch_update_time: status_tag = "empty" # Prepare data for updating or creating a page print (branch.get("url", "")) updated_value = branch.get("updated") if isinstance(updated_value, datetime.datetime): # Use datetime.datetime here updated_isoformat = updated_value.isoformat() else: updated_isoformat = datetime.datetime.now().isoformat() # Use datetime.datetime.now() here # Prepare status-message tags for multi-select status_messages = branch.get("status-message", []) status_message_data = [{"name": sm} for sm in status_messages] url_clean = branch.get("url", "") url_clean = url_clean.replace("xyz//","xyz/") page_data = { "properties": { "Name": {"title": [{"text": {"content": branch['Name']}}]}, "status-tag": {"select": {"name": status_tag}}, # Assuming 'select' type "status": {"rich_text": [{"text": {"content": "Status: " + status_tag}}]}, "url": {"url": url_clean}, "updated": {"date": {"start": updated_isoformat}}, "description": {"rich_text": [{"text": {"content": str(branch.get("description", ""))}}]}, "sub-variant": {"rich_text": [{"text": {"content": branch.get("sub-variant", "")}}]}, "author": {"rich_text": [{"text": {"content": branch.get("author", "")}}]}, "commit_message": {"rich_text": [{"text": {"content": branch.get("commit_message", "")}}]}, "depends-on": {"multi_select": [{"name": d} for d in existing_depends_on]}, "status-message": {"multi_select": status_message_data} } } # Update an existing page or create a new one if page_id: update_url = f"{base_url}/pages/{page_id}" response = requests.patch(update_url, headers=headers, data=json.dumps(page_data)) else: create_page_data = { "parent": {"database_id": database_id}, **page_data } response = requests.post(f"{base_url}/pages", headers=headers, data=json.dumps(create_page_data)) response.raise_for_status() print("Data synced to Notion successfully.") except requests.exceptions.HTTPError as errh: print("Http Error:", errh) except requests.exceptions.ConnectionError as errc: print("Error Connecting:", errc) except requests.exceptions.Timeout as errt: print("Timeout Error:", errt) except requests.exceptions.RequestException as err: print("Something went wrong", err) # Predefined dictionaries of streams to update streams = { "2B_100_batch": { "stream": "287b605243", "notionDB": '402d60c9c1ef4980a3d85c5a4f4a0c97' }, "2B_U100_batch": { "stream": "ebcfc50abe", "notionDB": '00c3a26a119f487fa028c1ed404f22ba' } } # Function to run on button click @webhook_endpoint # Define a unique endpoint URL async def update_streams(request: Request): # Read the request body as JSON payload = await request.json() print(str(payload)) # Accessing nested data in the 'event' object speckle_token = os.environ.get("SPECKLE_TOKEN") notion_token = os.environ.get("NOTION_TOKEN") #client = SpeckleClient(host="https://speckle.xyz") #client.authenticate_with_token(token=speckle_token) if 'payload' in payload and 'stream' in payload['payload']: stream_info = payload['payload']['stream'] if 'name' in stream_info: stream_name = stream_info['name'] print("update dat for: " + stream_name) # retrieve specific stream stream_id = streams[stream_name]["stream"] notion_db = streams[stream_name]["notionDB"] branch_data = extract_branch_info(stream_id, server_url="https://speckle.xyz/", token=speckle_token) sync_to_notion(notion_token, notion_db, branch_data) else: print("updating data for all streams") for key, value in streams.items(): stream_id = value["stream"] notion_db = value["notionDB"] branch_data = extract_branch_info(stream_id, server_url="https://speckle.xyz/", token=speckle_token) sync_to_notion(notion_token, notion_db, branch_data) return "All streams updated successfully!" """ # Create Gradio interface iface = gr.Interface( fn=update_streams, inputs=gr.components.Button(value="Update Streams"), outputs="text", ) # Launch the app iface.launch() """