"""
OSM-Beiträge: "Hund war willkommen" (dog=yes) erfassen — mit Anti-Fraud und
Gamification-Zähler.
Anti-Fraud (Defense in Depth, soweit serverseitig möglich):
- GPS-Beleg: eine kürzliche EIGENE Tour (routes.gps_track) muss am POI
vorbeiführen (≤ GPS_RADIUS_M) mit Verweil-Proxy (≥ DWELL_MIN_POINTS Punkte
im Radius — ohne Pro-Punkt-Zeitstempel der beste verfügbare Dwell-Proxy).
- Zeitkomponente: Tour-Recency (ROUTE_RECENCY_H) + Tages-Rate-Limit (DAILY_CAP).
- Dedup: 1× pro POI pro User. Positions-Sanity gegen die osm_pois-Koordinate.
NOCH NICHT hier (folgt separat, höheres Risiko): Geräte-Attestierung +
Sensor-Korroboration (nativ), tatsächliches OSM-Changeset-Upload, Revert-
Überleben/Konsens, und die echte Pro-Freischaltung. Beiträge werden daher als
status='pending' verifiziert erfasst; der Zähler ist provisorisch.
"""
import json
import logging
import xml.etree.ElementTree as ET
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
import httpx
from database import db
from auth import get_current_user
from math_utils import haversine_m
from routes.osm_auth import OSM_API_BASE, _decrypt
logger = logging.getLogger(__name__)
router = APIRouter()
# --- Anti-Fraud-Parameter ---
GPS_RADIUS_M = 50 # max. Abstand POI ↔ nächster Track-Punkt
DWELL_MIN_POINTS = 2 # mind. so viele Track-Punkte im Radius (Verweil-Proxy)
ROUTE_RECENCY_H = 48 # Tour darf max. so alt sein
POI_NEAR_M = 80 # eingereichte Position muss so nah am POI sein
DAILY_CAP = 20 # max. Beiträge pro Tag/User
# --- Gamification-Schwellen ---
BADGE_AT = 10 # "Kartograf"-Badge
PRO_AT = 100 # 100 geprüfte → 1 Jahr Pro (Freischaltung folgt separat)
class DogFriendlyIn(BaseModel):
osm_id: int
osm_type: str = Field('node', pattern='^(node|way)$')
poi_type: Optional[str] = None
lat: float
lon: float
welcome: bool = True # True → dog=yes, False → dog=no (Pächterwechsel)
def _verified_count(conn, uid: int) -> int:
return conn.execute(
"SELECT COUNT(*) FROM osm_contributions WHERE user_id=? AND status!='rejected'",
(uid,)
).fetchone()[0]
# ------------------------------------------------------------------
# OSM-Changeset-Upload (write_api): Element holen → dog=yes → Changeset.
# ------------------------------------------------------------------
def _changeset_xml(value: str) -> str:
note = "Hund willkommen" if value == "yes" else "Hund nicht willkommen"
return (''
''
f''
''
'')
def _mark_submitted(contrib_id: int, etype: str, changeset_id):
with db() as conn:
conn.execute(
"UPDATE osm_contributions SET status='submitted', osm_type=?, "
"changeset_id=?, submitted_at=datetime('now') WHERE id=?",
(etype, changeset_id, contrib_id)
)
async def submit_dog_tag(contrib_id: int, osm_id: int, osm_type: str, token: str, value: str) -> bool:
"""Setzt dog= (yes|no) am OSM-Element des Nutzers (eigener OAuth-Token).
Idempotent. Wirft bei Fehler → Beitrag bleibt 'pending' (Retry über den Job)."""
headers = {"Authorization": f"Bearer {token}"}
order = [osm_type, "way" if osm_type == "node" else "node"]
async with httpx.AsyncClient(timeout=20) as client:
# 1) Element holen (node/way auto-detect)
elem_xml = etype = None
for t in order:
r = await client.get(f"{OSM_API_BASE}/api/0.6/{t}/{osm_id}", headers=headers)
if r.status_code == 200:
elem_xml, etype = r.text, t
break
if elem_xml is None:
raise RuntimeError(f"OSM-Element {osm_id} nicht gefunden")
root = ET.fromstring(elem_xml)
el = root.find(etype)
existing = el.find("./tag[@k='dog']")
if existing is not None and existing.get("v") == value:
_mark_submitted(contrib_id, etype, None) # schon gesetzt → fertig
return True
# 2) Changeset öffnen
cs = await client.put(f"{OSM_API_BASE}/api/0.6/changeset/create",
headers=headers, content=_changeset_xml(value))
cs.raise_for_status()
changeset_id = cs.text.strip()
# 3) dog= setzen + Element hochladen (Geometrie/andere Tags bleiben)
if existing is not None:
existing.set("v", value)
else:
ET.SubElement(el, "tag", {"k": "dog", "v": value})
el.set("changeset", changeset_id)
up = await client.put(f"{OSM_API_BASE}/api/0.6/{etype}/{osm_id}",
headers=headers, content=ET.tostring(root, encoding="unicode"))
up.raise_for_status()
# 4) Changeset schließen
await client.put(f"{OSM_API_BASE}/api/0.6/changeset/{changeset_id}/close",
headers=headers)
_mark_submitted(contrib_id, etype, int(changeset_id))
return True
@router.post('/dog-friendly')
async def mark_dog_friendly(body: DogFriendlyIn, user=Depends(get_current_user)):
uid = user['id']
with db() as conn:
# 0) OSM verknüpft?
if not conn.execute("SELECT 1 FROM user_osm WHERE user_id=?", (uid,)).fetchone():
raise HTTPException(409, "Bitte zuerst dein OSM-Konto verknüpfen.")
value = 'yes' if body.welcome else 'no'
# 1) Vorhandene Markierung? Gleicher Wert → fertig. Anderer Wert →
# umdrehen erlaubt (Pächter wechseln → aus willkommen wird nicht mehr).
existing = conn.execute(
"SELECT id, tag_value FROM osm_contributions "
"WHERE user_id=? AND osm_id=? AND tag_key='dog'",
(uid, body.osm_id)
).fetchone()
if existing and existing['tag_value'] == value:
raise HTTPException(409, "Diesen Ort hast du schon so markiert.")
# 2) Zeitkomponente: Tages-Rate-Limit
today_n = conn.execute(
"SELECT COUNT(*) FROM osm_contributions "
"WHERE user_id=? AND created_at > datetime('now','-1 day')",
(uid,)
).fetchone()[0]
if today_n >= DAILY_CAP:
raise HTTPException(429, "Tageslimit erreicht — morgen geht's weiter.")
# 3) GPS-Beleg: kürzliche Tour, die am POI vorbeiführt (+ Verweil-Proxy)
routes = conn.execute(
"SELECT id, gps_track FROM routes "
"WHERE user_id=? AND created_at > datetime('now', ?) ORDER BY created_at DESC",
(uid, f'-{ROUTE_RECENCY_H} hours')
).fetchall()
best = None # (route_id, min_dist, points_near)
for r in routes:
try:
track = json.loads(r['gps_track'])
except Exception:
continue
near, mind = 0, float('inf')
for p in track:
d = haversine_m(body.lat, body.lon, p['lat'], p['lon'])
if d < mind:
mind = d
if d <= GPS_RADIUS_M:
near += 1
if mind <= GPS_RADIUS_M and near >= DWELL_MIN_POINTS:
if best is None or mind < best[1]:
best = (r['id'], mind, near)
if not best:
raise HTTPException(
422,
"Kein GPS-Beleg: In deinen letzten Touren ist kein Besuch an diesem Ort. "
"Geh mit deinem Hund dorthin, dann kannst du ihn eintragen."
)
# 4) Positions-Sanity gegen die bekannte POI-Koordinate
poi = conn.execute(
"SELECT lat, lon FROM osm_pois WHERE osm_id=? LIMIT 1", (body.osm_id,)
).fetchone()
if poi and haversine_m(body.lat, body.lon, poi['lat'], poi['lon']) > POI_NEAR_M:
raise HTTPException(422, "Position passt nicht zum gewählten Ort.")
# 5) verifiziert erfassen oder umdrehen (pending; OSM-Upload gleich best-effort)
if existing:
conn.execute(
"UPDATE osm_contributions SET tag_value=?, osm_type=?, poi_type=?, "
"lat=?, lon=?, route_id=?, gps_distance_m=?, gps_points_near=?, "
"status='pending', changeset_id=NULL, submitted_at=NULL, "
"created_at=datetime('now') WHERE id=?",
(value, body.osm_type, body.poi_type, body.lat, body.lon,
best[0], round(best[1], 1), best[2], existing['id'])
)
contrib_id = existing['id']
else:
cur = conn.execute(
"""INSERT INTO osm_contributions
(user_id, osm_id, osm_type, poi_type, tag_key, tag_value, lat, lon,
route_id, gps_distance_m, gps_points_near, status)
VALUES (?,?,?,?, 'dog',?, ?,?, ?,?,?, 'pending')""",
(uid, body.osm_id, body.osm_type, body.poi_type, value, body.lat, body.lon,
best[0], round(best[1], 1), best[2])
)
contrib_id = cur.lastrowid
total = _verified_count(conn, uid)
token_enc = conn.execute(
"SELECT token_enc FROM user_osm WHERE user_id=?", (uid,)
).fetchone()[0]
# 6) OSM-Upload best-effort — Fehler → bleibt 'pending', Job versucht erneut
submitted = False
try:
submitted = await submit_dog_tag(contrib_id, body.osm_id, body.osm_type, _decrypt(token_enc), value)
except Exception as e:
logger.warning("OSM-Upload später erneut (contrib %s): %s", contrib_id, e)
logger.info("dog=%s erfasst: user %s, osm %s, Tour %s (%.0fm, %d Pkt), submitted=%s",
value, uid, body.osm_id, best[0], best[1], best[2], submitted)
return {
"status": "erfasst", "value": value, "verified": True, "submitted": submitted,
"verified_count": total, "badge": total >= BADGE_AT,
"pro_progress": min(total, PRO_AT), "pro_at": PRO_AT,
}
@router.get('/status')
async def contrib_status(user=Depends(get_current_user)):
uid = user['id']
with db() as conn:
total = _verified_count(conn, uid)
by_status = {row[0]: row[1] for row in conn.execute(
"SELECT status, COUNT(*) FROM osm_contributions WHERE user_id=? GROUP BY status",
(uid,)
).fetchall()}
return {
"verified_count": total, "by_status": by_status,
"badge": total >= BADGE_AT,
"pro_progress": min(total, PRO_AT), "pro_at": PRO_AT,
}
# ------------------------------------------------------------------
# Confirm/Revert + Pro-Freischaltung (vom Scheduler-Job aufgerufen)
# ------------------------------------------------------------------
CONFIRM_AFTER_DAYS = 7 # Edit muss so lange in OSM ohne Revert überleben
def _grant_pro_if_earned(uid: int):
"""100 bestätigte Beiträge = 1 Jahr Pro. Idempotent über osm_pro_grants.
HINWEIS: setzt is_premium/subscription_* direkt — vor Produktion mit dem
Abo-/Billing-System abgleichen."""
with db() as conn:
confirmed = conn.execute(
"SELECT COUNT(*) FROM osm_contributions WHERE user_id=? AND status='confirmed'",
(uid,)).fetchone()[0]
granted = conn.execute(
"SELECT COUNT(*) FROM osm_pro_grants WHERE user_id=?", (uid,)).fetchone()[0]
for _ in range(confirmed // PRO_AT - granted):
conn.execute("INSERT INTO osm_pro_grants (user_id) VALUES (?)", (uid,))
conn.execute(
"UPDATE users SET is_premium=1, subscription_tier='pro', "
"subscription_expires_at=datetime("
" MAX(COALESCE(subscription_expires_at, datetime('now')), datetime('now')), '+1 year') "
"WHERE id=?", (uid,))
logger.info("OSM-Pro freigeschaltet: user %s (+1 Jahr)", uid)
async def run_confirmation_round():
"""Täglich: (1) hängengebliebene 'pending' erneut hochladen, (2) 'submitted'
nach CONFIRM_AFTER_DAYS auf Revert-Überleben prüfen → confirmed|rejected,
(3) Pro-Freischaltung prüfen."""
# (1) Pending-Retry
with db() as conn:
pend = conn.execute(
"SELECT c.id, c.osm_id, c.osm_type, c.tag_value, o.token_enc FROM osm_contributions c "
"JOIN user_osm o ON o.user_id=c.user_id WHERE c.status='pending' LIMIT 50"
).fetchall()
for r in pend:
try:
await submit_dog_tag(r["id"], r["osm_id"], r["osm_type"] or "node",
_decrypt(r["token_enc"]), r["tag_value"])
except Exception:
pass
# (2) Confirm/Revert
with db() as conn:
subs = conn.execute(
"SELECT id, user_id, osm_id, osm_type, tag_value FROM osm_contributions "
"WHERE status='submitted' AND submitted_at < datetime('now', ?)",
(f"-{CONFIRM_AFTER_DAYS} days",)
).fetchall()
affected = set()
async with httpx.AsyncClient(timeout=15) as client:
for r in subs:
etype = r["osm_type"] or "node"
try:
resp = await client.get(f"{OSM_API_BASE}/api/0.6/{etype}/{r['osm_id']}")
ok = False
if resp.status_code == 200:
el = ET.fromstring(resp.text).find(etype)
tag = el.find("./tag[@k='dog']") if el is not None else None
ok = tag is not None and tag.get("v") == r["tag_value"]
new_status = "confirmed" if ok else "rejected"
except Exception:
continue # nächste Runde erneut
with db() as conn:
conn.execute("UPDATE osm_contributions SET status=? WHERE id=?", (new_status, r["id"]))
affected.add(r["user_id"])
# (3) Pro-Freischaltung
for uid in affected:
_grant_pro_if_earned(uid)
if subs or pend:
logger.info("OSM-Confirm-Runde: %d pending-retry, %d geprüft, %d User betroffen",
len(pend), len(subs), len(affected))