import os import time from datetime import datetime import folium import pandas as pd import streamlit as st from huggingface_hub import HfApi from streamlit_folium import st_folium from src.utils import add_latlng_col from src.text_content import ( COLOR_MAPPING, CREDITS_TEXT, HEADERS_MAPPING, ICON_MAPPING, INTRO_TEXT_AR, INTRO_TEXT_EN, INTRO_TEXT_FR, LOGO, REVIEW_TEXT, SLOGAN, ) from src.utils import init_map, parse_gg_sheet TOKEN = os.environ.get("HF_TOKEN", None) REQUESTS_URL = "https://docs.google.com/spreadsheets/d/1gYoBBiBo1L18IVakHkf3t1fOGvHWb23loadyFZUeHJs/edit#gid=966953708" INTERVENTIONS_URL = ( "https://docs.google.com/spreadsheets/d/1eXOTqunOWWP8FRdENPs4cU9ulISm4XZWYJJNR1-SrwY/edit#gid=2089222765" ) api = HfApi(TOKEN) # Initialize Streamlit Config st.set_page_config(layout="wide", initial_sidebar_state="collapsed", page_icon="🤝", page_title="Nt3awnou نتعاونو") # Initialize States if "sleep_time" not in st.session_state: st.session_state.sleep_time = 2 if "auto_refresh" not in st.session_state: st.session_state.auto_refresh = False auto_refresh = st.sidebar.checkbox("Auto Refresh?", st.session_state.auto_refresh) if auto_refresh: number = st.sidebar.number_input("Refresh rate in seconds", value=st.session_state.sleep_time) st.session_state.sleep_time = number # Streamlit functions def display_interventions(interventions_df, m): """Display NGO interventions on the map""" for index, row in interventions_df.iterrows(): status = ( "Done ✅" if row[interventions_df.columns[5]] != "Intervention prévue dans le futur / Planned future intervention" else "Planned ⌛" ) color_mk = ( "green" if row[interventions_df.columns[5]] != "Intervention prévue dans le futur / Planned future intervention" else "pink" ) intervention_type = row[interventions_df.columns[6]].split("/")[0].strip() org = row[interventions_df.columns[1]] city = row[interventions_df.columns[9]] date = row[interventions_df.columns[4]] intervention_info = f"Status: {status}
Org: {org}
Intervention: {intervention_type}
📅 Date: {date}" if row["latlng"] is None: continue folium.Marker( location=row["latlng"], tooltip=city, popup=folium.Popup(intervention_info, max_width=300), icon=folium.Icon(color=color_mk), ).add_to(m) def show_requests(filtered_df, m): """Display victim requests on the map""" for index, row in filtered_df.iterrows(): request_type = row["ما هي احتياجاتك؟ (أضفها إذا لم يتم ذكرها)"] long_lat = row[ "هل يمكنك تقديم الإحداثيات الدقيقة للموقع؟ (ادا كنت لا توجد بعين المكان) متلاً \n31.01837503440344, -6.781405948842175" ] maps_url = f"https://maps.google.com/?q={long_lat}" display_text = f'Request Type: {request_type}
Id: {row["id"]}
Google Maps' icon_name = ICON_MAPPING.get(request_type, "info-sign") if row["latlng"] is None: continue folium.Marker( location=row["latlng"], tooltip=row[" لأي جماعة / قيادة / دوار تنتمون ؟"] if not pd.isna(row[" لأي جماعة / قيادة / دوار تنتمون ؟"]) else None, popup=folium.Popup(display_text, max_width=300), icon=folium.Icon(color=COLOR_MAPPING.get(request_type, "blue"), icon=icon_name), ).add_to(m) def display_google_sheet_tables(data_url): """Display the google sheet tables for requests and interventions""" st.markdown( f"""""", unsafe_allow_html=True, ) def display_dataframe(df, drop_cols, data_url, search_id=True, status=False): """Display the dataframe in a table""" col_1, col_2 = st.columns([1, 1]) with col_1: query = st.text_input( "🔍 Search for information / بحث عن المعلومات", key=f"search_requests_{int(search_id)}", ) with col_2: if search_id: id_number = st.number_input( "🔍 Search for an id / بحث عن رقم", min_value=0, max_value=len(filtered_df), value=0, step=1, ) if status: selected_status = st.selectbox( "🗓️ Status / حالة", ["all / الكل", "Done / تم", "Planned / مخطط لها"], key="status", ) if query: # Filtering the dataframe based on the query mask = df.apply(lambda row: row.astype(str).str.contains(query).any(), axis=1) display_df = df[mask] else: display_df = df display_df = display_df.drop(drop_cols, axis=1) if search_id and id_number: display_df = display_df[display_df["id"] == id_number] if status: target = "Pouvez-vous nous préciser si vous êtes déjà intervenus ou si vous prévoyez de le faire | Tell us if you already made the intervention, or if you're planning to do it" if selected_status == "Done / تم": display_df = display_df[display_df[target] == "Intervention déjà passée / Past intevention"] elif selected_status == "Planned / مخطط لها": display_df = display_df[display_df[target] != "Intervention déjà passée / Past intevention"] st.dataframe(display_df, height=500) st.markdown( f"To view the full Google Sheet for advanced filtering go to: {data_url} **لعرض الورقة كاملة، اذهب إلى**" ) # if we want to check hidden contact information st.markdown( f"We are hiding contact information to protect the privacy of the victims. If you are an NGO and want to contact the victims, please contact us at nt3awnoumorocco@gmail.com", ) # arabic needs rtl st.markdown( f"""
nt3awnoumorocco@gmail.com نحن نخفي معلومات الاتصال لحماية خصوصية الضحايا. إذا كنت جمعية وتريد الاتصال بالضحايا، يرجى الاتصال بنا على
""", unsafe_allow_html=True, ) def id_review_submission(): """Id review submission form""" st.subheader("🔍 Review of requests") st.markdown(REVIEW_TEXT) id_to_review = st.number_input("Enter id / أدخل الرقم", min_value=0, max_value=len(df), value=0, step=1) reason_for_review = st.text_area("Explain why / أدخل سبب المراجعة") if st.button("Submit / أرسل"): if reason_for_review == "": st.error("Please enter a reason / الرجاء إدخال سبب") else: filename = f"review_id_{id_to_review}_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.txt" with open(filename, "w") as f: f.write(f"id: {id_to_review}, explanation: {reason_for_review}\n") api.upload_file( path_or_fileobj=filename, path_in_repo=filename, repo_id="nt3awnou/review_requests", repo_type="dataset", ) st.success("Submitted at https://huggingface.co/datasets/nt3awnou/review_requests/ تم الإرسال") # Logo and Title st.markdown(LOGO, unsafe_allow_html=True) # st.title("Nt3awnou نتعاونو") st.markdown(SLOGAN, unsafe_allow_html=True) # Load data and initialize map with plugins df = parse_gg_sheet(REQUESTS_URL) df = add_latlng_col(df, process_column=15) interventions_df = parse_gg_sheet(INTERVENTIONS_URL) interventions_df = add_latlng_col(interventions_df, process_column=12) # Get current location: from bokeh.models.widgets import Button from bokeh.models import CustomJS from streamlit_bokeh_events import streamlit_bokeh_events loc_button = Button(label="Get Location") loc_button.js_on_event("button_click", CustomJS(code=""" navigator.geolocation.getCurrentPosition( (loc) => { document.dispatchEvent(new CustomEvent("GET_LOCATION", {detail: {lat: loc.coords.latitude, lon: loc.coords.longitude}})) } ) """)) current_location = streamlit_bokeh_events( loc_button, events="GET_LOCATION", key="get_location", refresh_on_update=False, override_height=75, debounce_time=0) m = init_map(current_location=current_location) # Selection of requests options = [ "إغاثة", "مساعدة طبية", "مأوى", "طعام وماء", "مخاطر (تسرب الغاز، تلف في الخدمات العامة...)", ] selected_options = [] # with tab_en: # st.markdown("👉 **Choose request type**") # with tab_ar: # st.markdown("👉 **اختر نوع الطلب**") # with tab_fr: # st.markdown("👉 **Choisissez le type de demande**") st.markdown("👉 **Choose request type | Choissisez le type de demande | اختر نوع الطلب**") col1, col2, col3, col4, col5 = st.columns([2, 3, 2, 3, 4]) cols = [col1, col2, col3, col4, col5] for i, option in enumerate(options): checked = cols[i].checkbox(HEADERS_MAPPING[option], value=True) if checked: selected_options.append(option) df["id"] = df.index filtered_df = df[df["ما هي احتياجاتك؟ (أضفها إذا لم يتم ذكرها)"].isin(selected_options)] selected_headers = [HEADERS_MAPPING[request] for request in selected_options] # Selection of interventions show_interventions = st.checkbox( "Display Interventions | عرض عمليات المساعدة | Afficher les interventions", value=True, ) if show_interventions: # print(interventions_df.columns) display_interventions(interventions_df, m) # Show requests show_requests(filtered_df, m) st_data = st_folium(m, use_container_width=True) tab_ar, tab_en, tab_fr = st.tabs(["العربية", "English", "Français"]) with tab_en: st.markdown(INTRO_TEXT_EN, unsafe_allow_html=True) with tab_ar: st.markdown(INTRO_TEXT_AR, unsafe_allow_html=True) with tab_fr: st.markdown(INTRO_TEXT_FR, unsafe_allow_html=True) # Requests table st.divider() st.subheader("📝 **Table of requests / جدول الطلبات**") drop_cols = [ "(عند الامكان) رقم هاتف شخص موجود في عين المكان", "الرجاء الضغط على الرابط التالي لمعرفة موقعك إذا كان متاحا", "GeoStamp", "GeoCode", "GeoAddress", "Status", ] display_dataframe(filtered_df, drop_cols, REQUESTS_URL, search_id=True) # Interventions table st.divider() st.subheader("📝 **Table of interventions / جدول التدخلات**") display_dataframe( interventions_df, [], # We show NGOs contact information INTERVENTIONS_URL, search_id=False, status=True, ) # Submit an id for review st.divider() id_review_submission() # Credits st.markdown( CREDITS_TEXT, unsafe_allow_html=True, ) if auto_refresh: time.sleep(number) st.experimental_rerun()