airflow added

Author: Chaimaekit

.gitignore

@@ -2,3 +2,5 @@ __pycache__/
 .venv
 .env
 env
+Evently_notes.md
+struct_events/coords_cache.json

Dockerfile.airflow

@@ -0,0 +1,3 @@
+FROM apache/airflow:2.8.1
+COPY requirements.airflow.txt .
+RUN pip install --no-cache-dir -r requirements.airflow.txt

README.md

@@ -1,20 +1,32 @@
-# Evently – Event Scraper + FastAPI + Elasticsearch + GitHub Actions Cron
+# Evently – Event Scraper + FastAPI + Elasticsearch + Airflow / Github Actions
 
-Evently is a platform that scrapes event data from multiple Moroccan sources, enriches each event with nearby bus and tramway transport info, indexes everything into Elasticsearch, and serves it through a FastAPI backend with a frontend UI.  
-All scraping and indexing are automated using GitHub Actions on a scheduled cron job.
+Evently is a platform that scrapes event data from multiple Moroccan sources, geocodes each event venue using **Nominatim (OpenStreetMap)**, enriches each event with nearby bus and tramway transport info, indexes everything into Elasticsearch, and serves it through a FastAPI backend with a frontend UI.
+
+Scraping and indexing are automated on a daily scheduled cron job at midnight, 
+with **Slack alerts** on failure.
+
+## Scheduling
+- **Apache Airflow** — used for local/self-hosted deployments (runs inside Docker)
+- **GitHub Actions** — alternative for cloud deployments using Elastic Cloud
 
 ---
 
 ## 📌 Features
 
 ### 🔍 Scraping
-- Collects events in parallel from multiple sources (Casaevents, Eventbrite, Events.ma, Guichet)
+- Collects events in parallel from multiple Moroccan sources
 - Parallel scraping for faster data collection
 
+### 📍 Geocoding
+- Each event venue is geocoded using the **Nominatim API** (OpenStreetMap) via `geopy`
+- Converts venue names and addresses into **coordinates (longitude, latitude)**
+- Coordinates are used downstream for transport enrichment via KDTree spatial indexing
+
 ### 🚌 Transport Integration
 - Each event is automatically enriched with nearby **Casabus** and **Tramway** lines at indexing time
-- Uses a **KDTree spatial index** for near-instant transport lookups
-- Transport data is precomputed and stored directly in the event document — no per-request calculation
+- Uses a **KDTree spatial index** built from precomputed local transport data to find nearest lines
+- Transport data is fetched **once** from the **Overpass API** (OpenStreetMap), stored locally as JSON, and reused at every indexing run — no live API calls during the pipeline
+- Transport lines are stored directly in the event document — no per-request spatial computation
 
 ### 🧠 Elasticsearch
 - Events are indexed with duplicate prevention using hashed unique IDs (MD5)
@@ -43,10 +55,20 @@ All scraping and indexing are automated using GitHub Actions on a scheduled cron
 
 ### 🎨 Frontend
 - Dynamic event loading from API
-- Pagination, search, and filter support
-
-### 🔄 GitHub Actions Automation
-A scheduled workflow runs all scrapers and Elasticsearch indexing automatically on a cron schedule.
+- Search bar with debounce and clear button
+- Filters panel: city, category, date, upcoming only, sort
+- Grid and list view toggle
+- Event detail modal with transport info
+- Favorites system with save/remove
+- Share events via Twitter, Facebook, WhatsApp, or copy link
+- Pagination with ellipsis
+
+### 🔄 Airflow Scheduling
+- Daily scraping and indexing triggered automatically at **midnight**
+- DAG: `evently_daily_scraping`
+- Retries up to **2 times** on failure with a 5 minute delay
+- **Slack notifications** on both success and failure
+- Airflow UI accessible at `http://localhost:8080`
 
 ---
 
@@ -56,85 +78,112 @@ A scheduled workflow runs all scrapers and Elasticsearch indexing automatically
 Events_Project/
 │
 ├── elastic/
-│   ├── elastic_script.py     # indexing + transport enrichment
-│   └── elastic_client.py     # Elasticsearch client setup
+│   ├── elastic_script.py         # indexing + transport enrichment
+│   └── elastic_client.py         # Elasticsearch client setup
 │
-├── main.py                   # FastAPI routes + lifespan startup
+├── main.py                       # FastAPI routes + lifespan startup
 │
 ├── static/
 │   ├── script.js
 │   └── style.css
 │
 ├── templates/
-│   └── index.html            # Frontend UI
+│   └── index.html                # Frontend UI
 │
-├── requirements.txt
-├── .env                      # Environment variables (not committed)
+├── requirements.txt              # FastAPI app dependencies
+├── requirements.airflow.txt      # Airflow container dependencies
+├── .env                          # Environment variables (not committed)
 │
-├── .github/
-│   └── workflows/
-│       └── build.yml         # GitHub Actions cron workflow
+├── Dockerfile                    # FastAPI container
+├── Dockerfile.airflow            # Airflow container with scraping deps
+├── docker-compose.yml            # All services
 │
-├── Dockerfile
-├── docker-compose.yml
+├── dags/
+│   └── evently_scraping.py       # Airflow DAG — daily scraping pipeline
 │
-├── scrape/                   # Scrapers (run in parallel)
+├── scrape/                       # Scrapers (run in parallel)
 │   ├── casaevents.py
 │   ├── eventbrit.py
 │   ├── eventsma.py
 │   └── guichet.py
 │
 ├── struct_events/
-│   ├── get_coords.py         # Geocoding utility
-│   └── models.py             # Events Pydantic model
+│   ├── get_coords.py             # Nominatim geocoding (venue → coordinates)
+│   └── models.py                 # Events Pydantic model
 │
 ├── transport/
-│   ├── bus.py                # KDTree transport lookup
-│   ├── cleaned_lines/
-│   └── overpass/
+│   ├── bus.py                    # KDTree transport lookup
+│   ├── cleaned_lines/            # Preprocessed transport data (JSON)
+│   └── overpass/                 # Raw Overpass API data (one-time fetch)
 ```
 
 ---
 
 ## ⚙️ Installation & Setup
 
-### 1️⃣ Pull the Docker image OR Clone the repo
+### 1️⃣ Clone the repo
 
-```bash
-docker pull chaimaaeljerrar/scraping-image:latest
-```
-OR
 ```bash
 git clone <repo_url>
+cd Events_Project
 ```
 
-### 2️⃣ (Optional) Create a `.env` file
-
-Only needed if connecting to Elastic Cloud or if security is enabled:
+### 2️⃣ Create a `.env` file
 
-```
+```env
 ELASTIC_PASSWORD=your_password_here
-CLOUD_ID=your_cloud_id_here
+SLACK_WEBHOOK_URL=https://hooks.slack.com/services/XXX/YYY/ZZZ
 ```
 
-> When running locally with Docker Compose, Elasticsearch security is disabled by default — no `.env` is required.
+> `SLACK_WEBHOOK_URL` is required for Airflow failure alerts. Get it from https://api.slack.com/apps
 
-### 3️⃣ Build and run the container
+### 3️⃣ Build and run all services
 
 ```bash
-docker compose up
+docker compose up --build
 ```
 
-The app will:
-1. Wait for Elasticsearch to be healthy
-2. Automatically start scraping and indexing events in the background
-3. Serve the frontend at `http://localhost:8000`
+This will start:
+- **Elasticsearch** at `http://localhost:9200`
+- **FastAPI** at `http://localhost:8000`
+- **Airflow webserver** at `http://localhost:8080`
+- **Airflow scheduler** running the daily DAG
+- **Postgres** for Airflow metadata (internal)
+
+> Wait about 60 seconds on first run for all services to initialize.
 
 ### 4️⃣ Access the app
 
-```
-http://localhost:8000
-```
+| Service | URL |
+|---------|-----|
+| Frontend | http://localhost:8000 |
+| Airflow UI | http://localhost:8080 |
+| Elasticsearch | http://localhost:9200 |
+
+---
+
+## 🔄 Airflow Setup
+
+### Login to Airflow UI
+- URL: `http://localhost:8080`
+- Username: `admin`
+- Password: `admin123`
+
+### Activate the DAG
+1. Find `evently_daily_scraping` in the DAGs list
+2. Toggle it **ON** (it's paused by default)
+3. Click ▶️ to trigger a manual run first to verify everything works
+4. After that it runs automatically every night at **midnight**
+
+### DAG Details
+
+| Property | Value |
+|----------|-------|
+| DAG ID | `evently_daily_scraping` |
+| Schedule | Daily at midnight (`0 0 * * *`) |
+| Retries | 2 (5 min delay between retries) |
+| Slack alert on success | ✅ |
+| Slack alert on failure | ✅ |
 
 ---
 
@@ -152,15 +201,53 @@ http://localhost:8000/health
 
 # Manually trigger reindex
 http://localhost:8000/reindex
+
+# Check Airflow
+http://localhost:8080
 ```
 
 ---
 
-## 🔧 How Transport Works
+## 🔧 How the Pipeline Works
 
-At indexing time, each event's coordinates are used to query a **KDTree** built from all bus and tramway stop coordinates. The nearest lines (within 0.5km) are stored directly on the event document in Elasticsearch.
+```
+Scrapers (parallel)
+      ↓
+Nominatim API — geocode venue name → coordinates
+      ↓
+KDTree spatial index — coordinates → nearest bus/tramway lines
+(from precomputed local JSON, originally fetched from Overpass API)
+      ↓
+Elasticsearch — event + coordinates + transport lines indexed together
+      ↓
+FastAPI — serves events, search, filters, transport, favorites
+      ↓
+Frontend UI — search, modal, pagination, favorites, share
+```
+
+At indexing time, each event's venue is geocoded to coordinates, which are used to query a **KDTree** built from all bus and tramway stop coordinates. The nearest lines within 0.5km are stored directly on the event document in Elasticsearch.
 
-This means `/transport/{event_id}` is just a simple document read — no spatial computation happens at request time.
+This means `/transport/{event_id}` is just a simple document read — no spatial computation or API call happens at request time.
+
+---
+
+## 🐳 Docker Commands Reference
+
+```bash
+# First run or after changing Dockerfile/requirements
+docker compose up --build
+
+# After changing only Python/JS/HTML files
+docker compose up
+
+# Full clean restart (wipes data volumes)
+docker compose down -v
+docker compose up --build
+
+# View logs for a specific service
+docker compose logs airflow-scheduler
+docker compose logs fastapi
+```
 
 ---
 
@@ -170,4 +257,6 @@ This means `/transport/{event_id}` is just a simple document read — no spatial
 2. Store favorites in Elasticsearch instead of in-memory
 3. Add AI-based event deduplication
 4. Add user authentication
-5. Deploy to cloud with security enabled
+5. Deploy to cloud with security enabled
+6. Add map view with Leaflet.js
+7. Add more Moroccan event sources

dags/evently_scraping.py

@@ -0,0 +1,56 @@
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.hooks.base import BaseHook
+from datetime import datetime, timedelta
+import requests
+import os
+import sys
+
+sys.path.insert(0, '/opt/airflow')
+
+default_args = {
+    'owner': 'evently',
+    'retries': 2,
+    'retry_delay': timedelta(minutes=5),
+}
+
+def run_indexing():
+    from elastic.elastic_script import indexing
+    result = indexing()
+    if not result:
+        raise ValueError("Indexing returned no results — possible scraping failure")
+    return result
+
+def send_slack_success(context):
+    webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
+    if not webhook_url:
+        return
+    requests.post(webhook_url, json={
+        "text": f"*Evently scraping succeeded!*\nRun: `{context['ds']}`\nAll sources indexed successfully."
+    })
+
+def send_slack_failure(context):
+    webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
+    if not webhook_url:
+        return
+    exception = context.get('exception', 'Unknown error')
+    requests.post(webhook_url, json={
+        "text": f"*Evently scraping FAILED!!!!*\nRun: `{context['ds']}`\nError: `{exception}`\nCheck Airflow at http://localhost:8080"
+    })
+
+with DAG(
+    dag_id='evently_daily_scraping',
+    default_args=default_args,
+    description='Daily scraping and indexing of Evently events',
+    schedule_interval='0 0 * * *', 
+    start_date=datetime(2026, 1, 1),
+    catchup=False,
+    on_success_callback=send_slack_success,
+    on_failure_callback=send_slack_failure,
+    tags=['evently', 'scraping']
+) as dag:
+
+    scrape_and_index = PythonOperator(
+        task_id='scrape_and_index_events',
+        python_callable=run_indexing,
+    )