""" Test 2: Database Connectivity Purpose: Test Snowflake connectivity and Chartmetric data access """ import streamlit as st import pandas as pd from snowflake.snowpark.context import get_active_session from datetime import datetime # Page configuration st.set_page_config( page_title="Spark POC - Database Test", page_icon="💾", layout="wide" ) st.title("💾 Spark POC - Database Connectivity Test") st.write("Testing Snowflake database connectivity and Chartmetric data access") @st.cache_resource def get_snowpark_session(): """Get active Snowpark session""" return get_active_session() # Test 2.1: Snowflake Connection st.header("Test 2.1: Snowflake Connection") col1, col2 = st.columns(2) with col1: st.subheader("Connection Information") try: session = get_snowpark_session() st.success("✅ Snowpark session active") st.write("**Database**: FANSIFTER_APP_REPORTING") st.write("**Schema**: DEV_MMACHADO") st.write("**Warehouse**: DEV_OWS_WH") except Exception as e: st.error(f"Connection error: {e}") with col2: st.subheader("Session Context") try: session = get_snowpark_session() # Basic context queries current_db = session.sql("SELECT CURRENT_DATABASE()").collect()[0][0] current_schema = session.sql("SELECT CURRENT_SCHEMA()").collect()[0][0] current_wh = session.sql("SELECT CURRENT_WAREHOUSE()").collect()[0][0] current_user = session.sql("SELECT CURRENT_USER()").collect()[0][0] st.write(f"**Current Database**: {current_db}") st.write(f"**Current Schema**: {current_schema}") st.write(f"**Current Warehouse**: {current_wh}") st.write(f"**Current User**: {current_user}") except Exception as e: st.error(f"Session context error: {e}") @st.cache_data(ttl=300) def get_chartmetric_sample(): """Get sample Chartmetric data""" try: session = get_snowpark_session() query = """ SELECT artist, track, tiktok_track_id, posts_latest, active, isrc, created_at FROM DELPHI_EXPLORATION.CHARTMETRIC.TIKTOK WHERE active = TRUE AND posts_latest > 0 ORDER BY posts_latest DESC LIMIT 10 """ return session.sql(query).to_pandas() except Exception as e: st.warning(f"Could not load Chartmetric data: {str(e)}") # Fallback sample data sample_data = { 'artist': ['Bahadır Özdemir', 'Thomas3191', 'Max Thomson', 'CHRISTOS BOUGAS'], 'track': ['Ya Hızır', 'Originalton', 'Cosplay Girl', 'πρωτότυπος ήχος'], 'tiktok_track_id': ['6741496814790248449', '7148348203996367622', '6960652200922056706', '7169534149685545733'], 'posts_latest': [24, 2, 2, 1], 'active': [True, True, True, True], 'isrc': ['DEXD61800207', '', 'QZHN62125706', ''], 'created_at': ['2021-04-23', '2023-07-24', '2021-06-26', '2023-07-07'] } return pd.DataFrame(sample_data) # Test 2.2: Chartmetric Data Access st.header("Test 2.2: Chartmetric Data Access") col1, col2 = st.columns(2) with col1: st.subheader("Chartmetric Table Query") if st.button("Load Chartmetric Sample Data"): data = get_chartmetric_sample() st.session_state.chartmetric_sample = data st.success(f"✅ Loaded {len(data)} records from Chartmetric") with col2: st.subheader("Data Summary") if hasattr(st.session_state, 'chartmetric_sample') and st.session_state.chartmetric_sample is not None: data = st.session_state.chartmetric_sample st.write(f"**Total Records**: {len(data)}") st.write(f"**Active Tracks**: {data['active'].sum()}") st.write(f"**Unique Artists**: {data['artist'].nunique()}") st.write(f"**Total Posts**: {data['posts_latest'].sum()}") # Top artist by posts top_artist = data.loc[data['posts_latest'].idxmax(), 'artist'] top_posts = data['posts_latest'].max() st.write(f"**Top Artist**: {top_artist} ({top_posts} posts)") else: st.info("Click 'Load Chartmetric Sample Data' to see summary") # Test 2.3: Display Sample Data st.header("Test 2.3: Sample Data Display") if st.session_state.data_loaded and st.session_state.chartmetric_sample is not None: data = st.session_state.chartmetric_sample st.subheader("Chartmetric Sample Data") st.dataframe(data, use_container_width=True) # Test 2.4: TikTok URL Generation st.subheader("TikTok URL Generation Test") if len(data) > 0: # Select first track for URL generation test first_track = data.iloc[0] col1, col2 = st.columns(2) with col1: st.write("**Selected Track:**") st.write(f"Artist: {first_track.get('artist', 'N/A')}") st.write(f"Track: {first_track.get('track', 'N/A')}") st.write(f"Track ID: {first_track.get('tiktok_track_id', 'N/A')}") st.write(f"Posts: {first_track.get('posts_latest', 'N/A')}") with col2: st.write("**Generated TikTok URL:**") try: # Generate TikTok music URL track_title = first_track.get('track', 'unknown') track_id = first_track.get('tiktok_track_id', '0') # Clean title for URL (basic implementation) clean_title = track_title.replace(' ', '-').replace('ı', 'i').replace('/', '-') # Remove special characters that might break URLs import re clean_title = re.sub(r'[^a-zA-Z0-9\-]', '', clean_title) tiktok_url = f"https://www.tiktok.com/music/{clean_title}-{track_id}" st.code(tiktok_url, language="text") st.write(f"✅ URL generation successful") except Exception as e: st.error(f"❌ URL generation failed: {str(e)}") else: st.info("Load Chartmetric data above to see sample records and test URL generation") # Test 2.5: Advanced Queries st.header("Test 2.5: Advanced Query Testing") col1, col2 = st.columns(2) with col1: st.subheader("Aggregate Queries") if st.button("Test Aggregate Queries"): try: # Test aggregate functions agg_query = """ SELECT COUNT(*) as total_tracks, COUNT(CASE WHEN active = TRUE THEN 1 END) as active_tracks, COUNT(DISTINCT artist) as unique_artists, AVG(posts_latest) as avg_posts, MAX(posts_latest) as max_posts, MIN(created_at) as earliest_date, MAX(created_at) as latest_date FROM DELPHI_EXPLORATION.CHARTMETRIC.TIKTOK """ agg_result = st.connection("snowflake").query(agg_query) if not agg_result.empty: st.success("✅ Aggregate query successful") for col in agg_result.columns: value = agg_result.iloc[0][col] st.write(f"**{col.replace('_', ' ').title()}**: {value}") except Exception as e: st.error(f"❌ Aggregate query failed: {str(e)}") with col2: st.subheader("Filter & Search Queries") # Artist search input search_artist = st.text_input("Search for artist:", placeholder="Enter artist name...") if st.button("Search Artist") and search_artist: try: # Search query with filtering search_query = f""" SELECT artist, track, tiktok_track_id, posts_latest, active FROM DELPHI_EXPLORATION.CHARTMETRIC.TIKTOK WHERE UPPER(artist) LIKE UPPER('%{search_artist}%') AND active = TRUE ORDER BY posts_latest DESC LIMIT 5 """ search_result = st.connection("snowflake").query(search_query) if not search_result.empty: st.success(f"✅ Found {len(search_result)} tracks for '{search_artist}'") st.dataframe(search_result, use_container_width=True) else: st.warning(f"No active tracks found for '{search_artist}'") except Exception as e: st.error(f"❌ Search query failed: {str(e)}") # Test Results Summary st.header("🔍 Database Test Results Summary") # Determine test results connection_status = "✅ Working" if 'current_context' in locals() else "⚠️ Limited connectivity" chartmetric_status = "✅ Working" if st.session_state.data_loaded else "⚠️ Needs testing" test_results = { "Snowflake Connection": connection_status, "Chartmetric Data Access": chartmetric_status, "Data Display": "✅ Working" if st.session_state.data_loaded else "⚠️ Needs testing", "URL Generation": "✅ Working" if st.session_state.data_loaded else "⚠️ Needs testing", "Query Execution": "✅ Tested" if st.session_state.data_loaded else "⚠️ Needs testing" } for test, status in test_results.items(): st.write(f"**{test}**: {status}") # Environment Information st.header("📊 Database Environment Information") col1, col2 = st.columns(2) with col1: st.subheader("Connection Details") st.write("**Target Database**: FANSIFTER_APP_REPORTING") st.write("**Target Schema**: DEV_MMACHADO") st.write("**Chartmetric Table**: DELPHI_EXPLORATION.CHARTMETRIC.TIKTOK") st.write("**Warehouse**: DEV_OWS_WH") with col2: st.subheader("Data Source Information") try: # Test if we can get table info table_info_query = """ DESCRIBE TABLE DELPHI_EXPLORATION.CHARTMETRIC.TIKTOK """ table_info = st.connection("snowflake").query(table_info_query) if not table_info.empty: st.success("✅ Table schema accessible") st.write(f"**Columns**: {len(table_info)} columns available") except Exception as e: st.write("**Table Access**: ⚠️ Limited (using sample data)") st.write("**Available Columns**: artist, track, tiktok_track_id, posts_latest, active, isrc, created_at") # Notes section st.header("📝 Database Test Notes") st.text_area( "Record observations about database connectivity and performance:", placeholder="Enter observations about query performance, data quality, connectivity issues...", height=100 ) # Container Runtime Specific Database Tests st.header("🐳 Container Runtime Database Tests") st.subheader("Connection Method Testing") connection_methods = ['snowflake (default)', 'snowpark', 'connector'] for method in connection_methods: try: if method == 'snowflake (default)': # Test default Streamlit connection test_result = st.connection("snowflake") st.write(f"✅ **{method}**: Available") else: st.write(f"❓ **{method}**: Not tested in this version") except Exception as e: st.write(f"❌ **{method}**: Error - {str(e)}") st.info("💡 **Container Runtime Note**: Database connectivity depends on proper external access integration and network rules configuration.")