import os import time from datetime import datetime import folium import pandas as pd import requests import streamlit as st from folium import plugins from huggingface_hub import HfApi 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 from streamlit_folium import st_folium 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") # 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 # Session for Requests session = requests.Session() 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 # Utility functions @st.cache_data(persist=True) def parse_latlng_from_link(url): try: # extract latitude and longitude from gmaps link if "@" not in url: resp = session.head(url, allow_redirects=True) url = resp.url latlng = url.split("@")[1].split(",")[0:2] return [float(latlng[0]), float(latlng[1])] except Exception as e: print(f"Error parsing latlng from link: {e}") return None def parse_gg_sheet_interventions(url): url = url.replace("edit#gid=", "export?format=csv&gid=") print(url) df = pd.read_csv(url, on_bad_lines="skip") return df.assign(latlng=df.iloc[:, 3].apply(parse_latlng_from_link)) # 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(): """Display the google sheet tables for requests and interventions""" st.subheader("📝 **Table of requests / جدول الطلبات**") st.markdown( f"""""", unsafe_allow_html=True, ) st.subheader("📝 **Table of interventions / جدول التدخلات**") st.markdown( f"""""", 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) # Language tabs st.sidebar.title("Language / اللغة") 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) # Load data and initialize map with plugins df = parse_gg_sheet(REQUESTS_URL) interventions_df = parse_gg_sheet_interventions(INTERVENTIONS_URL) m = init_map() # 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**") 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: display_interventions(interventions_df, m) # Show requests show_requests(filtered_df, m) st_data = st_folium(m, use_container_width=True) # Google Sheet Tables display_google_sheet_tables() # Submit an id for review id_review_submission() # Credits st.markdown( CREDITS_TEXT, unsafe_allow_html=True, ) if auto_refresh: time.sleep(number) st.experimental_rerun()