clawone/waste_api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Waste Collection API

Browse Source
This commit is contained in:
Clawone
2026-04-05 07:29:36 +00:00
commit 109b682a88
17 changed files with 1016 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example Normal file
View File

@@ -0,0 +1,2 @@
TELEGRAM_BOT_TOKEN=your_telegram_bot_token_here
TELEGRAM_CHAT_ID=your_chat_id_here

5
.gitignore vendored Normal file
View File

@@ -0,0 +1,5 @@
.venv/
__pycache__/
*.pyc
.env
data/

78
README.md Normal file
View File

@@ -0,0 +1,78 @@
# Waste Collection API
FastAPI backend for German municipal waste collection schedules with Telegram notification support.
## ⚠️ Required Setup Step
The `waste_collection_schedule` library has a file `calendar.py` that conflicts with Python's stdlib `calendar` module. **You must rename it before running:**
```bash
mv waste_lib/custom_components/waste_collection_schedule/calendar.py \
waste_lib/custom_components/waste_collection_schedule/calendar_SKIPPED.py
```
## Setup
```bash
# 1. Install dependencies
pip install -r requirements.txt
# 2. Rename the conflicting calendar.py (see above)
# 3. Copy .env
cp .env.example .env
# Edit .env with your Telegram bot token and chat ID
# 4. Run
PYTHONPATH=. uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
```
## Project Structure
```
waste_api/
├── app/
│ ├── main.py # FastAPI app + routes
│ ├── collector.py # waste_collection_schedule wrapper + icalevents bug fix
│ ├── schemas.py # Pydantic request/response models
│ ├── routes/
│ │ ├── sources.py # GET /sources, GET /sources/{id}
│ │ ├── collections.py # POST /collections/{ics,abfall_io,awm_muenchen_de}
│ │ └── notifications.py # POST/GET/DELETE /notifications
│ └── core/
│ ├── scheduler.py # Notification storage + due check
│ └── notifier.py # Telegram sender
├── SPEC.md # Full OpenAPI spec
└── requirements.txt
```
## API Endpoints
| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/health` | Health check |
| `GET` | `/sources` | List all source IDs |
| `GET` | `/sources/{id}` | Get params for a source |
| `POST` | `/collections/ics` | Fetch via generic ICS URL |
| `POST` | `/collections/abfall_io` | Fetch via Abfall.IO |
| `POST` | `/collections/awm_muenchen_de` | Fetch Munich waste schedule |
| `POST` | `/notifications` | Create notification rule |
| `GET` | `/notifications` | List all notification rules |
| `DELETE` | `/notifications/{id}` | Delete a rule |
## API Docs
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
## Example: Munich
```bash
curl -X POST http://localhost:8000/collections/awm_muenchen_de \
-H "Content-Type: application/json" \
-d '{"street": "Bahnstr.", "house_number": "11", "count": 5}'
```
## Notifications
The scheduler runs via cron (not implemented yet) and fires Telegram messages when collections are due. Notifications are stored in `data/notifications.json`.

247
SPEC.md Normal file
View File

@@ -0,0 +1,247 @@
# Waste Collection API — OpenAPI Specification
**Base URL:** `http://localhost:8000`
**Version:** 1.0.0
**Purpose:** Return waste collection schedules for German municipalities via push notification targets (email, Telegram, etc.)
---
## API Endpoints
### `GET /health`
Health check.
**Response `200`**
```json
{ "status": "ok" }
```
---
### `GET /sources`
List all available source IDs (e.g. `abfall_io`, `awm_muenchen_de`, `ics`).
**Response `200`**
```json
{
"count": 3,
"sources": [
{ "id": "ics", "name": "Generic ICS / iCal URL" },
{ "id": "abfall_io", "name": "Abfall.IO (Germany)" },
{ "id": "awm_muenchen_de","name": "AWM München (Germany)" }
]
}
```
---
### `GET /sources/{source_id}`
Describe required params for a given source.
**Path params:**
`source_id` — e.g. `ics`, `abfall_io`, `awm_muenchen_de`
**Response `200`**
```json
{
"id": "abfall_io",
"name": "Abfall.IO (Germany)",
"params": [
{ "name": "key", "type": "string", "required": true, "example": "8215c62763967916979e0e8566b6172e" },
{ "name": "f_id_kommune","type": "integer","required": true, "example": 2999 },
{ "name": "f_id_strasse","type": "integer","required": true, "example": 1087 }
]
}
```
---
### `POST /collections/ics`
Fetch waste collections from a generic ICS/iCal URL.
**Request body**
```json
{
"url": "https://example.com/kalender.ics",
"count": 10
}
```
| Field | Type | Required | Default | Description |
|-------|------|----------|---------|-------------|
| `url` | string | ✅ | — | Direct URL to `.ics` file |
| `count` | integer | ❌ | 20 | Max entries to return |
**Response `200`**
```json
{
"source": "ics",
"url": "https://example.com/kalender.ics",
"collections": [
{
"date": "2026-04-08",
"type": "Restmülltonne",
"daysUntil": 4,
"icon": "mdi:trash-can"
},
{
"date": "2026-04-27",
"type": "Papiertonne",
"daysUntil": 23,
"icon": "mdi:paper-bin"
}
]
}
```
---
### `POST /collections/abfall_io`
Fetch collections via Abfall.IO (many German municipalities).
**Request body**
```json
{
"key": "8215c62763967916979e0e8566b6172e",
"f_id_kommune": 2999,
"f_id_strasse": 1087,
"count": 20
}
```
**Response `200`** — same shape as `/collections/ics` with `source: "abfall_io"`.
---
### `POST /collections/awm_muenchen_de`
Fetch collections for Munich via AWM.
**Request body**
```json
{
"street": "Bahnstr.",
"house_number": "11",
"count": 20
}
```
**Response `200`** — same shape as `/collections/ics` with `source: "awm_muenchen_de"`.
---
### `POST /notify`
Queue a notification for a future collection.
**Request body**
```json
{
"source_id": "ics",
"collection": {
"date": "2026-04-08",
"type": "Restmülltonne",
"daysUntil": 4,
"icon": "mdi:trash-can"
},
"notify_at_days_before": 1,
"channels": ["telegram"],
"chat_id": 1320170074
}
```
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| `source_id` | string | ✅ | Source that returned this collection |
| `collection` | object | ✅ | A single collection entry |
| `notify_at_days_before` | integer | ❌ | Days before collection to fire (default: 1) |
| `channels` | string[] | ✅ | `["telegram"]` (more coming) |
| `chat_id` | integer | ✅ | Telegram chat ID to notify |
**Response `200`**
```json
{
"ok": true,
"notification_id": "notif_abc123",
"message": "Reminder set for 2026-04-08: Restmülltonne"
}
```
---
### `GET /notifications`
List all active notification rules.
**Response `200`**
```json
{
"count": 1,
"notifications": [
{
"id": "notif_abc123",
"source_id": "ics",
"collection_type": "Restmülltonne",
"collection_date": "2026-04-08",
"notify_at_days_before": 1,
"channels": ["telegram"],
"chat_id": 1320170074,
"fired": false
}
]
}
```
---
### `DELETE /notifications/{notification_id}`
Delete a notification rule.
**Response `200`**
```json
{ "ok": true, "deleted": "notif_abc123" }
```
---
## Error Responses
All endpoints return errors in this shape:
```json
{
"error": "Human-readable error message",
"detail": "Optional technical detail"
}
```
| Status | Meaning |
|--------|---------|
| `400` | Bad request — missing or invalid params |
| `404` | Source not found |
| `422` | Unprocessable entity — source returned no data |
| `500` | Internal server error |
---
## Data Models
### `Collection`
```json
{
"date": "YYYY-MM-DD",
"type": "Human-readable waste type string",
"daysUntil": 4,
"icon": "mdi:icon-name"
}
```
### `Notification`
```json
{
"id": "notif_abc123",
"source_id": "ics | abfall_io | awm_muenchen_de",
"collection_type": "Restmülltonne",
"collection_date": "2026-04-08",
"notify_at_days_before": 1,
"channels": ["telegram"],
"chat_id": 1320170074,
"fired": false
}
```

0
app/__init__.py Normal file
View File

173
app/collector.py Normal file
View File

@@ -0,0 +1,173 @@
"""
Collector — wraps waste_collection_schedule library calls.
Key insight: icalevents has a timezone bug (offset-naive vs offset-aware datetime
comparison). We fix it by monkey-patching the library's ICS.convert() to use icalendar.
"""
import datetime
import logging
import sys
from pathlib import Path
from typing import Any, List
import requests
log = logging.getLogger(__name__)
# ── Library path setup ────────────────────────────────────────────────────────
WASTE_LIB_BASE = Path(__file__).resolve().parent.parent.parent
_WASTE_OUTER = WASTE_LIB_BASE / "waste_lib" / "custom_components" / "waste_collection_schedule"
_WASTE_INNER = _WASTE_OUTER / "waste_collection_schedule"
_PATCHED = False
def _ensure_path():
global _PATCHED
if str(_WASTE_INNER) not in sys.path:
sys.path.insert(0, str(_WASTE_INNER))
sys.path.insert(1, str(_WASTE_OUTER))
if not _PATCHED:
_patch_icalevents_bug()
_PATCHED = True
def _patch_icalevents_bug():
"""
Monkey-patch waste_collection_schedule.service.ICS.ICS.convert()
to use icalendar instead of icalevents (which has a timezone bug).
Returns List[Tuple[datetime.date, str]] as the original does.
"""
import icalendar
import jinja2
from waste_collection_schedule.service.ICS import ICS as OriginalICS
_original_convert = OriginalICS.convert
def patched_convert(self, text: str) -> List[Any]:
try:
return _original_convert(self, text)
except TypeError as ex:
if "offset-naive" not in str(ex):
raise
log.debug("ICS.convert: icalevents bug detected, falling back to icalendar")
cal = icalendar.Calendar.from_ical(text)
now = datetime.datetime.now().date()
end = now + datetime.timedelta(days=365)
entries: List[Any] = []
for component in cal.walk():
if component.name != "VEVENT":
continue
dtstart = component.get("DTSTART")
if dtstart is None:
continue
dt = dtstart.dt
if isinstance(dt, datetime.datetime):
dt = dt.date()
if not isinstance(dt, datetime.date):
continue
if dt < now or dt > end:
continue
if self._offset:
dt = dt + datetime.timedelta(days=self._offset)
# Try SUMMARY first, then DESCRIPTION
summary = str(component.get("SUMMARY", "") or component.get("DESCRIPTION", "") or "")
if not summary and self._title_template:
env = jinja2.Environment()
title_tpl = env.from_string(self._title_template)
summary = title_tpl.render(date=dt)
entries.append((dt, summary))
# Sort by date (icalevents returns sorted results)
entries.sort(key=lambda x: x[0])
return entries
OriginalICS.convert = patched_convert
log.info("Patched ICS.convert to use icalendar fallback")
# ── Collection mapping ────────────────────────────────────────────────────────
def _map_collection(wc) -> dict:
return {
"date": wc.date.isoformat(),
"type": wc.get("type", "Unknown"),
"daysUntil": (wc.date - datetime.datetime.now().date()).days,
"icon": wc.icon,
}
# ── ICS via icalendar (bypasses icalevents timezone bug) ─────────────────────
def _fetch_ics_via_icalendar(url: str, count: int) -> list[dict]:
import icalendar
resp = requests.get(url, timeout=15)
resp.raise_for_status()
cal = icalendar.Calendar.from_ical(resp.text)
now = datetime.datetime.now().date()
entries = []
for component in cal.walk():
if component.name != "VEVENT":
continue
dtstart = component.get("DTSTART")
if dtstart is None:
continue
dt = dtstart.dt
if isinstance(dt, datetime.datetime):
dt = dt.date()
if not isinstance(dt, datetime.date):
continue
if dt < now:
continue
summary = str(component.get("SUMMARY", "Unknown"))
entries.append({
"date": dt.isoformat(),
"type": summary,
"daysUntil": (dt - now).days,
"icon": None,
})
entries.sort(key=lambda x: x["date"])
return entries[:count]
# ── Public API ───────────────────────────────────────────────────────────────
def fetch_ics(url: str, count: int = 20) -> list[dict]:
_ensure_path()
from waste_collection_schedule.source.ics import Source as ICSSource
source = ICSSource(url=url)
entries = source.fetch()
return [_map_collection(e) for e in entries[:count]]
def fetch_abfall_io(
key: str, f_id_kommune: int, f_id_strasse: int, count: int
) -> list[dict]:
_ensure_path()
from waste_collection_schedule.source.abfall_io import Source as AbfallIOSource
source = AbfallIOSource(
key=key,
f_id_kommune=f_id_kommune,
f_id_strasse=f_id_strasse,
)
entries = source.fetch()
return [_map_collection(e) for e in entries[:count]]
def fetch_awm_muenchen(street: str, house_number: str, count: int) -> list[dict]:
_ensure_path()
from waste_collection_schedule.source.awm_muenchen_de import Source as AWMSource
source = AWMSource(street=street, house_number=house_number)
entries = source.fetch()
return [_map_collection(e) for e in entries[:count]]
def fetch_stuttgart(street: str, streetnr: int, count: int) -> list[dict]:
_ensure_path()
from waste_collection_schedule.source.stuttgart_de import Source
source = Source(street=street, streetnr=streetnr)
entries = source.fetch()
return [_map_collection(e) for e in entries[:count]]

0
app/core/__init__.py Normal file
View File

42
app/core/notifier.py Normal file
View File

@@ -0,0 +1,42 @@
"""
Send notification via Telegram.
Requires bot token and chat_id configured in environment or .env file.
"""
import os
import logging
from pathlib import Path
import httpx
from dotenv import load_dotenv
log = logging.getLogger(__name__)
load_dotenv()
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "8638072533:AAG3fcm0Sq_JBSSA1UXLspgRzn-tdIJ9N_8")
TELEGRAM_CHAT_ID = int(os.getenv("TELEGRAM_CHAT_ID", "1320170074"))
def send_telegram(text: str, chat_id: int = TELEGRAM_CHAT_ID) -> bool:
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
try:
resp = httpx.post(url, json={"chat_id": chat_id, "text": text}, timeout=10)
resp.raise_for_status()
log.info(f"Telegram sent to {chat_id}: {text[:50]}")
return True
except Exception as e:
log.error(f"Telegram failed: {e}")
return False
def format_collection_reminder(collection: dict, source_id: str) -> str:
date = collection.get("date", "?")
wtype = collection.get("type", "?")
days = collection.get("daysUntil", "?")
icon = collection.get("icon", "🗑️")
return (
f"{icon} *Waste Collection Reminder*\n"
f"📅 {date} (in {days} day{'s' if days != 1 else ''})\n"
f"🗑️ {wtype}\n"
f"🔔 from: {source_id}"
)

118
app/core/scheduler.py Normal file
View File

@@ -0,0 +1,118 @@
import uuid
from datetime import datetime, timedelta
from pathlib import Path
import json
import logging
log = logging.getLogger(__name__)
NOTIFS_FILE = Path(__file__).resolve().parent.parent.parent / "data" / "notifications.json"
# ─── Source registry ───────────────────────────────────────────
SOURCE_DEFS = {
"ics": {
"name": "Generic ICS / iCal URL",
"params": [
{"name": "url", "type": "string", "required": True, "example": "https://example.com/kalender.ics"},
],
},
"abfall_io": {
"name": "Abfall.IO (Germany)",
"params": [
{"name": "key", "type": "string", "required": True, "example": "8215c62763967916979e0e8566b6172e"},
{"name": "f_id_kommune", "type": "integer", "required": True, "example": 2999},
{"name": "f_id_strasse", "type": "integer", "required": True, "example": 1087},
],
},
"awm_muenchen_de": {
"name": "AWM München (Germany)",
"params": [
{"name": "street", "type": "string", "required": True, "example": "Bahnstr."},
{"name": "house_number", "type": "string", "required": True, "example": "11"},
],
},
"stuttgart_de": {
"name": "Abfall Stuttgart (Germany)",
"params": [
{"name": "street", "type": "string", "required": True, "example": "Im Steinengarten"},
{"name": "streetnr", "type": "integer", "required": True, "example": 7},
],
},
}
class NotificationScheduler:
def __init__(self):
self._notifications: dict[str, dict] = {}
self._load()
def _load(self):
if NOTIFS_FILE.exists():
try:
data = json.loads(NOTIFS_FILE.read_text())
self._notifications = {n["id"]: n for n in data.get("notifications", [])}
log.info(f"Loaded {len(self._notifications)} notifications from {NOTIFS_FILE}")
except Exception as e:
log.warning(f"Failed to load notifications: {e}")
def _save(self):
NOTIFS_FILE.parent.mkdir(parents=True, exist_ok=True)
NOTIFS_FILE.write_text(
json.dumps({"notifications": list(self._notifications.values())}, indent=2)
)
def add(
self,
source_id: str,
collection: dict,
notify_at_days_before: int,
channels: list[str],
chat_id: int,
) -> dict:
nid = f"notif_{uuid.uuid4().hex[:8]}"
entry = {
"id": nid,
"source_id": source_id,
"collection_type": collection.get("type", "Unknown"),
"collection_date": collection.get("date", ""),
"notify_at_days_before": notify_at_days_before,
"channels": channels,
"chat_id": chat_id,
"fired": False,
}
self._notifications[nid] = entry
self._save()
return entry
def list_all(self) -> list[dict]:
return list(self._notifications.values())
def delete(self, nid: str) -> bool:
if nid in self._notifications:
del self._notifications[nid]
self._save()
return True
return False
def due_today(self) -> list[dict]:
today = datetime.now().date().isoformat()
due = []
for n in self._notifications.values():
if n.get("fired"):
continue
coll_date = n.get("collection_date", "")
days_before = n.get("notify_at_days_before", 1)
# check if today is notify_at_days_before days before collection
try:
coll = datetime.fromisoformat(coll_date).date()
notify_date = coll - timedelta(days=days_before)
if notify_date.isoformat() == today:
due.append(n)
except Exception:
pass
return due
def mark_fired(self, nid: str):
if nid in self._notifications:
self._notifications[nid]["fired"] = True
self._save()

42
app/main.py Normal file
View File

@@ -0,0 +1,42 @@
import logging
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .routes import sources, collections, notifications
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
app = FastAPI(
title="Waste Collection API",
description="Waste collection schedules for German municipalities with notification support",
version="1.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ─── Routes ────────────────────────────────────────────────────
app.include_router(sources.router)
app.include_router(collections.router)
app.include_router(notifications.router)
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/")
def root():
return {
"name": "Waste Collection API",
"version": "1.0.0",
"docs": "/docs",
"sources": "/sources",
}

34
app/models.py Normal file
View File

@@ -0,0 +1,34 @@
# Pure Pydantic models — no waste_collection_schedule imports here.
from .schemas import (
Collection,
Notification,
ICSRequest,
AbfallIORequest,
AWMRequest,
NotifyRequest,
SourceInfo,
SourceListResponse,
SourceDetailResponse,
CollectionsResponse,
NotificationsResponse,
NotifyResponse,
DeleteResponse,
SourceParam,
)
__all__ = [
"Collection",
"Notification",
"ICSRequest",
"AbfallIORequest",
"AWMRequest",
"NotifyRequest",
"SourceInfo",
"SourceListResponse",
"SourceDetailResponse",
"CollectionsResponse",
"NotificationsResponse",
"NotifyResponse",
"DeleteResponse",
"SourceParam",
]

0
app/routes/__init__.py Normal file
View File

56
app/routes/collections.py Normal file
View File

@@ -0,0 +1,56 @@
from fastapi import APIRouter, HTTPException
from ..schemas import (
ICSRequest,
AbfallIORequest,
AWMRequest,
StuttgartRequest,
CollectionsResponse,
)
from ..collector import fetch_ics, fetch_abfall_io, fetch_awm_muenchen, fetch_stuttgart
router = APIRouter(prefix="/collections", tags=["collections"])
@router.post("/ics", response_model=CollectionsResponse)
def collections_ics(body: ICSRequest):
try:
collections = fetch_ics(body.url, body.count)
except Exception as e:
raise HTTPException(status_code=422, detail=str(e))
if not collections:
raise HTTPException(status_code=422, detail="No collections returned from this ICS URL")
return CollectionsResponse(source="ics", collections=collections)
@router.post("/abfall_io", response_model=CollectionsResponse)
def collections_abfall_io(body: AbfallIORequest):
try:
collections = fetch_abfall_io(body.key, body.f_id_kommune, body.f_id_strasse, body.count)
except Exception as e:
raise HTTPException(status_code=422, detail=str(e))
if not collections:
raise HTTPException(status_code=422, detail="No collections returned. Check your IDs.")
return CollectionsResponse(source="abfall_io", collections=collections)
@router.post("/awm_muenchen_de", response_model=CollectionsResponse)
def collections_awm_muenchen(body: AWMRequest):
try:
collections = fetch_awm_muenchen(body.street, body.house_number, body.count)
except Exception as e:
raise HTTPException(status_code=422, detail=str(e))
if not collections:
raise HTTPException(status_code=422, detail="No collections returned. Check street name.")
return CollectionsResponse(source="awm_muenchen_de", collections=collections)
@router.post("/stuttgart_de", response_model=CollectionsResponse)
def collections_stuttgart(body: StuttgartRequest):
try:
collections = fetch_stuttgart(body.street, body.streetnr, body.count)
except Exception as e:
raise HTTPException(status_code=422, detail=str(e))
if not collections:
raise HTTPException(status_code=422, detail="No collections returned. Check street name and number.")
return CollectionsResponse(source="stuttgart_de", collections=collections)

44
app/routes/notifications.py Normal file
View File

@@ -0,0 +1,44 @@
from fastapi import APIRouter, HTTPException
from ..schemas import (
NotifyRequest,
NotifyResponse,
NotificationsResponse,
DeleteResponse,
)
from ..core.scheduler import NotificationScheduler
router = APIRouter(prefix="/notifications", tags=["notifications"])
# singleton scheduler
_scheduler = NotificationScheduler()
@router.post("", response_model=NotifyResponse)
def create_notification(body: NotifyRequest):
entry = _scheduler.add(
source_id=body.source_id,
collection=body.collection.model_dump(),
notify_at_days_before=body.notify_at_days_before,
channels=body.channels,
chat_id=body.chat_id,
)
return NotifyResponse(
ok=True,
notification_id=entry["id"],
message=f"Reminder set for {entry['collection_date']}: {entry['collection_type']}",
)
@router.get("", response_model=NotificationsResponse)
def list_notifications():
notes = _scheduler.list_all()
return NotificationsResponse(count=len(notes), notifications=notes)
@router.delete("/{notification_id}", response_model=DeleteResponse)
def delete_notification(notification_id: str):
ok = _scheduler.delete(notification_id)
if not ok:
raise HTTPException(status_code=404, detail=f"Notification '{notification_id}' not found")
return DeleteResponse(ok=True, deleted=notification_id)

65
app/routes/sources.py Normal file
View File

@@ -0,0 +1,65 @@
import json
from pathlib import Path
from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse
from ..schemas import (
SourceListResponse,
SourceDetailResponse,
SourceParam,
)
from ..core.scheduler import SOURCE_DEFS
router = APIRouter(prefix="/sources", tags=["sources"])
SOURCES_JSON = (
Path(__file__).resolve().parent.parent.parent.parent
/ "waste_lib"
/ "custom_components"
/ "waste_collection_schedule"
/ "sources.json"
)
@router.get("", response_model=SourceListResponse)
def list_sources():
return SourceListResponse(
count=len(SOURCE_DEFS),
sources=[{"id": k, "name": v["name"]} for k, v in SOURCE_DEFS.items()],
)
@router.get("/germany")
def list_germany_sources():
"""
Return all 727 German municipalities from sources.json.
Includes title, module type, and default_params (if any).
"""
with open(SOURCES_JSON) as f:
data = json.load(f)
return {"country": "Germany", "count": len(data["Germany"]), "sources": data["Germany"]}
@router.get("/germany/search")
def search_germany_sources(q: str):
"""
Search Germany municipalities by name (case-insensitive).
"""
with open(SOURCES_JSON) as f:
data = json.load(f)
q = q.lower()
results = [s for s in data["Germany"] if q in s["title"].lower()]
return {"query": q, "count": len(results), "sources": results[:20]}
@router.get("/{source_id}", response_model=SourceDetailResponse)
def get_source(source_id: str):
if source_id not in SOURCE_DEFS:
raise HTTPException(status_code=404, detail=f"Source '{source_id}' not found")
s = SOURCE_DEFS[source_id]
return SourceDetailResponse(
id=source_id,
name=s["name"],
params=[SourceParam(**p) for p in s["params"]],
)

100
app/schemas.py Normal file
View File

@@ -0,0 +1,100 @@
from pydantic import BaseModel, Field
from typing import Optional
# ─── Collection ───────────────────────────────────────────────
class Collection(BaseModel):
date: str = Field(description="Collection date YYYY-MM-DD")
type: str = Field(description="Waste type label")
daysUntil: int = Field(description="Days until collection")
icon: Optional[str] = Field(default=None, description="MDI icon name")
# ─── Source descriptors ────────────────────────────────────────
class SourceInfo(BaseModel):
id: str
name: str
class SourceParam(BaseModel):
name: str
type: str
required: bool
example: str
class SourceDetailResponse(BaseModel):
id: str
name: str
params: list[SourceParam]
class SourceListResponse(BaseModel):
count: int
sources: list[SourceInfo]
# ─── Request bodies ────────────────────────────────────────────
class ICSRequest(BaseModel):
url: str = Field(description="Direct URL to .ics / .ical file")
count: int = Field(default=20, ge=1, le=100)
class AbfallIORequest(BaseModel):
key: str = Field(description="Abfall.IO API key")
f_id_kommune: int = Field(description="Municipality ID")
f_id_strasse: int = Field(description="Street ID")
count: int = Field(default=20, ge=1, le=100)
class AWMRequest(BaseModel):
street: str = Field(description="Street name in Munich")
house_number: str = Field(description="House number")
count: int = Field(default=20, ge=1, le=100)
class StuttgartRequest(BaseModel):
street: str = Field(description="Street name in Stuttgart")
streetnr: int = Field(description="House number")
count: int = Field(default=20, ge=1, le=100)
class NotifyRequest(BaseModel):
source_id: str
collection: Collection
notify_at_days_before: int = Field(default=1, ge=0)
channels: list[str] = Field(description='Notification channels, e.g. ["telegram"]')
chat_id: int
# ─── Response wrappers ─────────────────────────────────────────
class CollectionsResponse(BaseModel):
source: str
collections: list[Collection]
class Notification(BaseModel):
id: str
source_id: str
collection_type: str
collection_date: str
notify_at_days_before: int
channels: list[str]
chat_id: int
fired: bool
class NotificationsResponse(BaseModel):
count: int
notifications: list[Notification]
class NotifyResponse(BaseModel):
ok: bool
notification_id: str
message: str
class DeleteResponse(BaseModel):
ok: bool
deleted: str

10
requirements.txt Normal file
View File

@@ -0,0 +1,10 @@
fastapi==0.115.0
uvicorn[standard]==0.30.6
pydantic==2.9.2
requests==2.32.3
icalendar==6.1.1
icalevents==2.1.2
python-dotenv==1.0.1
httpx==0.27.2
beautifulsoup4==4.12.3
jinja2>=3.0.0