Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| from azure.cosmos import CosmosClient, exceptions | |
| import os | |
| import pandas as pd | |
| import traceback | |
| import shutil | |
| from github import Github | |
| from git import Repo | |
| from datetime import datetime | |
| import base64 | |
| import json | |
| import uuid # 🎲 For generating unique IDs | |
| from urllib.parse import quote # 🔗 For encoding URLs | |
| # 🎉 Welcome to our fun-filled Cosmos DB and GitHub Integration app! | |
| st.set_page_config(layout="wide") | |
| # 🌌 Cosmos DB configuration | |
| ENDPOINT = "https://acae-afd.documents.azure.com:443/" | |
| DATABASE_NAME = os.environ.get("COSMOS_DATABASE_NAME") | |
| CONTAINER_NAME = os.environ.get("COSMOS_CONTAINER_NAME") | |
| Key = os.environ.get("Key") # 🔑 Don't forget your key! | |
| # 🏠 Your local app URL (Change this to your app's URL) | |
| LOCAL_APP_URL = "http://localhost:8501" | |
| # 🐙 GitHub configuration | |
| def download_github_repo(url, local_path): | |
| # 🚚 Let's download that GitHub repo! | |
| if os.path.exists(local_path): | |
| shutil.rmtree(local_path) | |
| Repo.clone_from(url, local_path) | |
| def create_zip_file(source_dir, output_filename): | |
| # 📦 Zipping up files like a pro! | |
| shutil.make_archive(output_filename, 'zip', source_dir) | |
| def create_repo(g, repo_name): | |
| # 🛠️ Creating a new GitHub repo. Magic! | |
| user = g.get_user() | |
| return user.create_repo(repo_name) | |
| def push_to_github(local_path, repo, github_token): | |
| # 🚀 Pushing code to GitHub. Hold on tight! | |
| repo_url = f"https://{github_token}@github.com/{repo.full_name}.git" | |
| local_repo = Repo(local_path) | |
| if 'origin' in [remote.name for remote in local_repo.remotes]: | |
| origin = local_repo.remote('origin') | |
| origin.set_url(repo_url) | |
| else: | |
| origin = local_repo.create_remote('origin', repo_url) | |
| if not local_repo.heads: | |
| local_repo.git.checkout('-b', 'main') | |
| current_branch = 'main' | |
| else: | |
| current_branch = local_repo.active_branch.name | |
| local_repo.git.add(A=True) | |
| if local_repo.is_dirty(): | |
| local_repo.git.commit('-m', 'Initial commit') | |
| origin.push(refspec=f'{current_branch}:{current_branch}') | |
| def get_base64_download_link(file_path, file_name): | |
| # 🧙♂️ Generating a magical download link! | |
| with open(file_path, "rb") as file: | |
| contents = file.read() | |
| base64_encoded = base64.b64encode(contents).decode() | |
| return f'<a href="data:application/zip;base64,{base64_encoded}" download="{file_name}">⬇️ Download {file_name}</a>' | |
| # 🧭 New functions for dynamic sidebar navigation | |
| def get_databases(client): | |
| # 📚 Fetching list of databases. So many options! | |
| return [db['id'] for db in client.list_databases()] | |
| def get_containers(database): | |
| # 📂 Getting containers. Containers within containers! | |
| return [container['id'] for container in database.list_containers()] | |
| def get_documents(container, limit=None): | |
| # 📝 Retrieving documents. Shhh, don't tell anyone! | |
| query = "SELECT * FROM c ORDER BY c._ts DESC" | |
| items = list(container.query_items(query=query, enable_cross_partition_query=True, max_item_count=limit)) | |
| return items | |
| # 🌟 Cosmos DB functions | |
| def insert_record(container, record): | |
| try: | |
| container.create_item(body=record) | |
| return True, "Record inserted successfully! 🎉" | |
| except exceptions.CosmosHttpResponseError as e: | |
| return False, f"HTTP error occurred: {str(e)} 🚨" | |
| except Exception as e: | |
| return False, f"An unexpected error occurred: {str(e)} 😱" | |
| def update_record(container, updated_record): | |
| try: | |
| container.upsert_item(body=updated_record) | |
| return True, f"Record with id {updated_record['id']} successfully updated. 🛠️" | |
| except exceptions.CosmosHttpResponseError as e: | |
| return False, f"HTTP error occurred: {str(e)} 🚨" | |
| except Exception as e: | |
| return False, f"An unexpected error occurred: {traceback.format_exc()} 😱" | |
| def delete_record(container, name, id): | |
| try: | |
| container.delete_item(item=id, partition_key=id) | |
| return True, f"Successfully deleted record with name: {name} and id: {id} 🗑️" | |
| except exceptions.CosmosResourceNotFoundError: | |
| return False, f"Record with id {id} not found. It may have been already deleted. 🕵️♂️" | |
| except exceptions.CosmosHttpResponseError as e: | |
| return False, f"HTTP error occurred: {str(e)} 🚨" | |
| except Exception as e: | |
| return False, f"An unexpected error occurred: {traceback.format_exc()} 😱" | |
| # 🎲 Function to generate a unique UUID | |
| def generate_unique_id(): | |
| # 🧙♂️ Generating a unique UUID! | |
| return str(uuid.uuid4()) | |
| # 📦 Function to archive current container | |
| def archive_current_container(database_name, container_name, client): | |
| try: | |
| base_dir = "./cosmos_archive_current_container" | |
| if os.path.exists(base_dir): | |
| shutil.rmtree(base_dir) | |
| os.makedirs(base_dir) | |
| db_client = client.get_database_client(database_name) | |
| container_client = db_client.get_container_client(container_name) | |
| items = list(container_client.read_all_items()) | |
| container_dir = os.path.join(base_dir, container_name) | |
| os.makedirs(container_dir) | |
| for item in items: | |
| item_id = item.get('id', f"unknown_{datetime.now().strftime('%Y%m%d%H%M%S')}") | |
| with open(os.path.join(container_dir, f"{item_id}.json"), 'w') as f: | |
| json.dump(item, f, indent=2) | |
| archive_name = f"{container_name}_archive_{datetime.now().strftime('%Y%m%d%H%M%S')}" | |
| shutil.make_archive(archive_name, 'zip', base_dir) | |
| return get_base64_download_link(f"{archive_name}.zip", f"{archive_name}.zip") | |
| except Exception as e: | |
| return f"An error occurred while archiving data: {str(e)} 😢" | |
| # 🎈 Let's modify the main app to be more fun! | |
| def main(): | |
| st.title("🐙Git🌌Cosmos💫 - Azure Cosmos DB and Github Agent") | |
| # 🚦 Initialize session state | |
| if 'logged_in' not in st.session_state: | |
| st.session_state.logged_in = False | |
| if 'selected_records' not in st.session_state: | |
| st.session_state.selected_records = [] | |
| if 'client' not in st.session_state: | |
| st.session_state.client = None | |
| if 'selected_database' not in st.session_state: | |
| st.session_state.selected_database = None | |
| if 'selected_container' not in st.session_state: | |
| st.session_state.selected_container = None | |
| if 'selected_document_id' not in st.session_state: | |
| st.session_state.selected_document_id = None | |
| if 'current_index' not in st.session_state: | |
| st.session_state.current_index = 0 | |
| if 'cloned_doc' not in st.session_state: | |
| st.session_state.cloned_doc = None | |
| # 🔐 Automatic Login | |
| if Key: | |
| st.session_state.primary_key = Key | |
| st.session_state.logged_in = True | |
| else: | |
| st.error("Cosmos DB Key is not set in environment variables. 🔑❌") | |
| return # Can't proceed without a key | |
| if st.session_state.logged_in: | |
| # 🌌 Initialize Cosmos DB client | |
| try: | |
| if st.session_state.client is None: | |
| st.session_state.client = CosmosClient(ENDPOINT, credential=st.session_state.primary_key) | |
| # 🗄️ Sidebar for database, container, and document selection | |
| st.sidebar.title("🐙Git🌌Cosmos💫🗄️Navigator") | |
| databases = get_databases(st.session_state.client) | |
| selected_db = st.sidebar.selectbox("🗃️ Select Database", databases) | |
| if selected_db != st.session_state.selected_database: | |
| st.session_state.selected_database = selected_db | |
| st.session_state.selected_container = None | |
| st.session_state.selected_document_id = None | |
| st.session_state.current_index = 0 | |
| st.rerun() | |
| if st.session_state.selected_database: | |
| database = st.session_state.client.get_database_client(st.session_state.selected_database) | |
| containers = get_containers(database) | |
| selected_container = st.sidebar.selectbox("📁 Select Container", containers) | |
| if selected_container != st.session_state.selected_container: | |
| st.session_state.selected_container = selected_container | |
| st.session_state.selected_document_id = None | |
| st.session_state.current_index = 0 | |
| st.rerun() | |
| if st.session_state.selected_container: | |
| container = database.get_container_client(st.session_state.selected_container) | |
| # 📦 Add Export button | |
| if st.button("📦 Export Container Data"): | |
| download_link = archive_current_container(st.session_state.selected_database, st.session_state.selected_container, st.session_state.client) | |
| if download_link.startswith('<a'): | |
| st.markdown(download_link, unsafe_allow_html=True) | |
| else: | |
| st.error(download_link) | |
| # Fetch documents | |
| documents = get_documents(container) | |
| total_docs = len(documents) | |
| if total_docs > 5: | |
| documents_to_display = documents[:5] | |
| st.info("Showing top 5 most recent documents.") | |
| else: | |
| documents_to_display = documents | |
| st.info(f"Showing all {len(documents_to_display)} documents.") | |
| if documents_to_display: | |
| # 🎨 Add Viewer/Editor selection | |
| view_options = ['Show as Markdown', 'Show as Code Editor', 'Show as Edit and Save', 'Clone Document', 'New Record'] | |
| selected_view = st.selectbox("Select Viewer/Editor", view_options, index=2) | |
| if selected_view == 'Show as Markdown': | |
| # 🖌️ Show each record as Markdown with navigation | |
| total_docs = len(documents) | |
| doc = documents[st.session_state.current_index] | |
| st.markdown(f"#### Document ID: {doc.get('id', '')}") | |
| # 🕵️♂️ Let's extract values from the JSON that have at least one space | |
| values_with_space = [] | |
| def extract_values(obj): | |
| if isinstance(obj, dict): | |
| for k, v in obj.items(): | |
| extract_values(v) | |
| elif isinstance(obj, list): | |
| for item in obj: | |
| extract_values(item) | |
| elif isinstance(obj, str): | |
| if ' ' in obj: | |
| values_with_space.append(obj) | |
| extract_values(doc) | |
| # 🔗 Let's create a list of links for these values | |
| search_urls = { | |
| "🚀🌌ArXiv": lambda k: f"/?q={quote(k)}", | |
| "🃏Analyst": lambda k: f"/?q={quote(k)}-{quote('PromptPrefix')}", | |
| "📚PyCoder": lambda k: f"/?q={quote(k)}-{quote('PromptPrefix2')}", | |
| "🔬JSCoder": lambda k: f"/?q={quote(k)}-{quote('PromptPrefix3')}", | |
| "🏠": lambda k: f"{LOCAL_APP_URL}/?q={quote(k)}", | |
| "📖": lambda k: f"https://en.wikipedia.org/wiki/{quote(k)}", | |
| "🔍": lambda k: f"https://www.google.com/search?q={quote(k)}", | |
| "▶️": lambda k: f"https://www.youtube.com/results?search_query={quote(k)}", | |
| "🔎": lambda k: f"https://www.bing.com/search?q={quote(k)}", | |
| "🎥": lambda k: f"https://www.youtube.com/results?search_query={quote(k)}", | |
| "🐦": lambda k: f"https://twitter.com/search?q={quote(k)}", | |
| } | |
| st.markdown("#### 🔗 Links for Extracted Texts") | |
| for term in values_with_space: | |
| links_md = ' '.join([f"[{emoji}]({url(term)})" for emoji, url in search_urls.items()]) | |
| st.markdown(f"**{term}** <small>{links_md}</small>", unsafe_allow_html=True) | |
| # Show the document content as markdown | |
| content = json.dumps(doc, indent=2) | |
| st.markdown(f"```json\n{content}\n```") | |
| # Navigation buttons | |
| col_prev, col_next = st.columns([1, 1]) | |
| with col_prev: | |
| if st.button("⬅️ Previous", key='prev_markdown'): | |
| if st.session_state.current_index > 0: | |
| st.session_state.current_index -= 1 | |
| st.rerun() | |
| with col_next: | |
| if st.button("➡️ Next", key='next_markdown'): | |
| if st.session_state.current_index < total_docs - 1: | |
| st.session_state.current_index += 1 | |
| st.rerun() | |
| elif selected_view == 'Show as Code Editor': | |
| # 💻 Show each record in a code editor with navigation | |
| total_docs = len(documents) | |
| doc = documents[st.session_state.current_index] | |
| st.markdown(f"#### Document ID: {doc.get('id', '')}") | |
| doc_str = st.text_area("Edit Document", value=json.dumps(doc, indent=2), height=300, key=f'code_editor_{st.session_state.current_index}') | |
| col_prev, col_next = st.columns([1, 1]) | |
| with col_prev: | |
| if st.button("⬅️ Previous", key='prev_code'): | |
| if st.session_state.current_index > 0: | |