app.py

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 streamlit_folium import st_folium

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,
    REVIEW_TEXT_2,
    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")

# 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:
        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"<b>Status:</b> {status}<br><b>Org:</b> {org}<br><b>Intervention:</b> {intervention_type}<br><b>📅 Date:</b> {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'<b>Request Type:</b> {request_type}<br><b>Id:</b> {row["id"]}<br><a href="{maps_url}" target="_blank" rel="noopener noreferrer"><b>Google Maps</b></a>'
        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"""<iframe src="{REQUESTS_URL}" width="100%" height="600px"></iframe>""",
        unsafe_allow_html=True,
    )

    st.subheader("📝 **Table of interventions / جدول التدخلات**")
    st.markdown(
        f"""<iframe src="{INTERVENTIONS_URL}" width="100%" height="600px"></iframe>""",
        unsafe_allow_html=True,
    )


def id_review_submission():
    """Id review submission form"""
    st.subheader("🔍 Review of requests")
    st.markdown(REVIEW_TEXT)
    st.markdown(REVIEW_TEXT_2)

    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()