feat(gps-tracks): GPX download from public track popup
Some checks failed
CI / lint (push) Failing after 5s
CI / test (push) Failing after 6s
CI / build (push) Has been skipped
CI / lint (pull_request) Failing after 4s
CI / test (pull_request) Failing after 5s
CI / build (pull_request) Has been skipped

Реализация ET-011: кнопка «Скачать GPX» в popup публичного GPS-трека и
новый эндпоинт GET /api/gps-tracks/{track_id}/download (GPX 1.1 +
Content-Disposition с UTF-8 именем по RFC 5987). Реэкспорт защищён
per-source флагом `download_allowed` в `config/gps_sources.yaml`
(default-deny, MVP whitelist = `osm`).

Backend:
- `src/api/gps_tracks/export.py` — чистый stdlib-builder GPX 1.1
  (`build_gpx`) + санитизация имени файла (`safe_filename`, RFC 5987).
- `src/api/gps_tracks/endpoint.py` — новый route с проверками
  400 / 403 / 404 / 413; cap 200 000 точек (REQ-NF-02).
- `src/api/gps_tracks/config.py` — `load_download_allowed_sources()`
  читает YAML, default-deny при отсутствии поля; fallback на `{"osm"}`
  при отсутствии конфига.
- `src/api/main.py` — пробрасывает `GPS_SOURCES_CONFIG_PATH` в router.

Frontend:
- `src/web/gps_tracks.js` — кнопка в `_renderTrackPopupHtml`,
  обработчик `_downloadPublicTrack` (fetch + Blob + a.download — тот же
  паттерн, что в `app.js::downloadGPX`, R-1 митигирован), парсер
  `_parseFilenameFromCD` для RFC 5987, маппинг ошибок
  `_handleDownloadError` (403/404/413/5xx → showToast).
- `src/web/app.css` — стиль кнопки, 32×32 CSS px (REQ-NF-04).

Тесты:
- 13 unit для GPX-builder (UT-01/02/03/05; XSD-валидация против
  `tests/fixtures/gpx-1.1/gpx.xsd`).
- 10 unit для `safe_filename` (UT-04).
- 11 integration для download-эндпоинта (IT-01..08 +
  ANY-rule license check + default-deny без конфига).

ADR-014 (gpx-download-endpoint), ADR-015 (source-redistribution-policy).
Refs: ET-011

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-03 20:59:53 +00:00
parent 6fe2ecf12b
commit eea6c846c2
12 changed files with 2284 additions and 8 deletions

View File

@@ -1,6 +1,16 @@
"""Загрузка и валидация конфигурации GPS-источников и регионов (ET-008)."""
import logging
import os
import yaml
logger = logging.getLogger(__name__)
# ET-011 / ADR-015: дефолтный whitelist для скачивания, если конфиг недоступен
# (например, в unit-тестах). Совпадает с production-выбором "только OSM".
DEFAULT_DOWNLOAD_ALLOWED_SOURCES = frozenset({"osm"})
def load_sources_config(path: str) -> list:
"""Загружает конфигурацию источников GPS-треков.
@@ -87,3 +97,47 @@ def load_regions_config(path: str) -> list:
)
return regions
def load_download_allowed_sources(path: str | None) -> set[str]:
"""ET-011 / ADR-015: возвращает whitelist source_id с download_allowed=true.
Семантика default-deny: если поле `download_allowed` отсутствует,
источник **не** добавляется в whitelist.
Args:
path: путь к `config/gps_sources.yaml` либо None.
Returns:
set[str] — id источников, для которых разрешено скачивание.
При path=None / отсутствии файла / ошибке парсинга — возвращает
`DEFAULT_DOWNLOAD_ALLOWED_SOURCES` (`{"osm"}`) и логирует WARNING.
"""
if not path:
return set(DEFAULT_DOWNLOAD_ALLOWED_SOURCES)
if not os.path.exists(path):
logger.warning(
"gps_sources config not found at %s; falling back to default "
"download whitelist=%s",
path,
sorted(DEFAULT_DOWNLOAD_ALLOWED_SOURCES),
)
return set(DEFAULT_DOWNLOAD_ALLOWED_SOURCES)
try:
sources = load_sources_config(path)
except (ValueError, OSError) as exc:
logger.warning(
"failed to load gps_sources config from %s (%s); falling back to "
"default download whitelist=%s",
path,
exc,
sorted(DEFAULT_DOWNLOAD_ALLOWED_SOURCES),
)
return set(DEFAULT_DOWNLOAD_ALLOWED_SOURCES)
allowed: set[str] = set()
for src in sources:
# Дефолт `False` — default-deny (ADR-015 §C).
if src.get("download_allowed") is True:
allowed.add(src["id"])
return allowed

View File

@@ -1,13 +1,17 @@
"""FastAPI router для GPS-треков (ET-008)."""
"""FastAPI router для GPS-треков (ET-008, расширен в ET-011)."""
import json
import logging
import os
from typing import Optional
from fastapi import APIRouter, HTTPException, Query, Response
from fastapi import APIRouter, HTTPException, Path, Query, Response
from src.api.gps_tracks.config import load_download_allowed_sources
from src.api.gps_tracks.db import get_tracks_in_bbox, init_db, open_db
from src.api.gps_tracks.export import build_gpx, safe_filename
from src.api.gps_tracks.mvt import (
_gps_tile_cache,
_wkb_to_coords,
build_gps_mvt,
clear_gps_tile_cache,
get_gps_cached_tile,
@@ -15,6 +19,13 @@ from src.api.gps_tracks.mvt import (
_tile_to_bbox,
)
logger = logging.getLogger("uvicorn.access")
# ET-011 / ADR-014:
ALLOWED_DOWNLOAD_FORMATS = {"gpx"}
MAX_POINTS_FOR_DOWNLOAD = 200_000 # REQ-NF-02
GPX_MEDIA_TYPE = "application/gpx+xml; charset=utf-8"
def _parse_bbox(bbox_str: str) -> tuple:
"""Парсит и валидирует bbox строку 'west,south,east,north'.
@@ -52,8 +63,6 @@ def _parse_bbox(bbox_str: str) -> tuple:
def _row_to_geojson_feature(row) -> dict:
"""Конвертирует sqlite3.Row в GeoJSON Feature."""
from src.api.gps_tracks.mvt import _wkb_to_coords
coords = _wkb_to_coords(row["geom"])
sources = json.loads(row["sources_json"] or "[]")
@@ -94,17 +103,29 @@ def _row_to_geojson_feature(row) -> dict:
}
def create_gps_router(db_path: str) -> APIRouter:
def create_gps_router(
db_path: str,
sources_config_path: Optional[str] = None,
) -> APIRouter:
"""Создаёт FastAPI router для GPS-треков.
Args:
db_path: путь к SQLite БД для GPS-треков
db_path: путь к SQLite БД для GPS-треков.
sources_config_path: путь к ``config/gps_sources.yaml``.
Если None — для ET-011 download-эндпоинта используется
default-deny whitelist ``{"osm"}`` (см. ADR-015).
Returns:
APIRouter с prefix="/api/gps-tracks"
"""
router = APIRouter(prefix="/api/gps-tracks", tags=["gps-tracks"])
# ET-011 / ADR-015: whitelist source_id, для которых разрешено
# скачивание GPX. Читается один раз при старте router'а.
allowed_download_sources: set[str] = load_download_allowed_sources(
sources_config_path
)
def _get_conn():
conn = open_db(db_path)
init_db(conn)
@@ -307,4 +328,110 @@ def create_gps_router(db_path: str) -> APIRouter:
clear_gps_tile_cache()
return {"status": "ok", "cleared": True}
# ─── ET-011: скачивание GPX из popup ──────────────────────────
@router.get("/{track_id}/download")
async def download_track(
track_id: int = Path(..., ge=1),
format: str = Query("gpx", description="Формат файла (только 'gpx' в MVP)"),
):
"""Отдаёт GPX-файл для трека с правильным Content-Disposition.
Реализует ADR-014 / ADR-015 для ET-011.
Порядок проверок (ADR-014 §H):
1. format ∈ whitelist (иначе 400).
2. SELECT по id (иначе 404).
3. points_count <= MAX (иначе 413).
4. licence policy по sources (иначе 403).
5. Сборка GPX → 200.
"""
if format not in ALLOWED_DOWNLOAD_FORMATS:
raise HTTPException(
status_code=400,
detail="unsupported_format",
)
try:
conn = _get_conn()
cur = conn.cursor()
cur.execute(
"""
SELECT id, name, description, activity_type, user, created_at,
length_m, points_count, geom, sources_json,
external_urls_json
FROM tracks
WHERE id = ?
""",
(track_id,),
)
row = cur.fetchone()
conn.close()
except Exception as exc:
raise HTTPException(500, f"DB error: {exc}")
if row is None:
raise HTTPException(status_code=404, detail="track_not_found")
points_count = row["points_count"] or 0
if points_count > MAX_POINTS_FOR_DOWNLOAD:
raise HTTPException(status_code=413, detail="track_too_large")
sources = json.loads(row["sources_json"] or "[]")
external_urls = json.loads(row["external_urls_json"] or "[]")
# ADR-015 §B1: разрешение по принципу ANY — хотя бы один разрешённый.
if not any(s in allowed_download_sources for s in sources):
raise HTTPException(
status_code=403,
detail={
"detail": "source_forbidden",
"external_urls": external_urls,
},
)
coords = _wkb_to_coords(row["geom"]) or []
try:
xml_str = build_gpx(
track_id=row["id"],
name=row["name"],
description=row["description"],
activity_type=row["activity_type"],
user=row["user"],
created_at=row["created_at"],
sources=sources,
external_urls=external_urls,
coords=coords,
)
except Exception as exc:
logger.exception("build_gpx failed for track_id=%s", track_id)
raise HTTPException(500, f"GPX build error: {exc}")
ascii_name, utf8_quoted = safe_filename(row["name"], track_id)
content_disposition = (
f'attachment; filename="{ascii_name}.gpx"; '
f"filename*=UTF-8''{utf8_quoted}.gpx"
)
xml_bytes = xml_str.encode("utf-8")
# REQ-F-07: лёгкое журналирование успешной отдачи.
logger.info(
"track_download id=%d sources=%s size_bytes=%d",
track_id,
",".join(sources) if sources else "",
len(xml_bytes),
)
return Response(
content=xml_bytes,
media_type=GPX_MEDIA_TYPE,
headers={
"Content-Disposition": content_disposition,
"Cache-Control": "private, max-age=3600",
"Access-Control-Allow-Origin": "*",
"Access-Control-Expose-Headers": "Content-Disposition",
},
)
return router

View File

@@ -0,0 +1,265 @@
"""GPX-экспорт публичных GPS-треков (ET-011, ADR-014).
Сборка GPX 1.1 из метаданных трека + санитизация имени файла для
HTTP Content-Disposition с поддержкой RFC 5987 (UTF-8 filename*).
Чистый stdlib-модуль, без I/O — легко тестируется юнитами.
"""
from __future__ import annotations
import urllib.parse
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
# OSM-license URL для блока <copyright> (ADR-014 §G, ODbL).
_OSM_LICENSE_URL = "https://www.openstreetmap.org/copyright"
# Запрещённые в FAT/NTFS символы (ADR-014 §F.2).
_FORBIDDEN_NAME_CHARS = set('/\\:*?"<>|')
# Лимит длины ASCII-fallback по байтам UTF-8 (ADR-014 §F.5; RFC 5987 — 254
# на параметр, минус префикс "filename*=UTF-8''" и расширение).
_MAX_NAME_BYTES = 80
def _format_utc(iso_str: str | None) -> str | None:
"""Нормализует ISO-8601 datetime → 'YYYY-MM-DDTHH:MM:SSZ' (UTC).
Поддерживает входные строки с/без таймзоны. None / нераспарсимое — None.
"""
if not iso_str:
return None
try:
# Python 3.11+ fromisoformat понимает 'Z'-суффикс; для надёжности
# делаем явную замену.
normalized = iso_str.replace("Z", "+00:00")
dt = datetime.fromisoformat(normalized)
except (ValueError, TypeError):
return None
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
else:
dt = dt.astimezone(timezone.utc)
return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
def build_gpx(
*,
track_id: int,
name: str | None,
description: str | None,
activity_type: str | None,
user: str | None,
created_at: str | None,
sources: list[str],
external_urls: list[str],
coords: list[tuple[float, float]],
) -> str:
"""Собирает GPX 1.1 как XML-строку (с XML-declaration).
Args:
track_id: id трека (используется только в fallback-имени).
name: tracks.name (если пусто — в `<name>` ставится «Без названия»).
description: tracks.description (если пусто — `<desc>` опускается).
activity_type: tracks.activity_type, попадает в `<trk><type>`.
user: tracks.user — попадает в `<metadata><author><name>`.
created_at: ISO-8601 строка → нормализуется в UTC `<metadata><time>`.
sources: список source_id (для `<copyright>` и `<link><text>`).
external_urls: список внешних URL → `<metadata><link>` по одному.
coords: список (lon, lat) — точки трека.
Returns:
XML-строка (включает `<?xml …?>`-декларацию).
"""
# GPX namespace должен быть default — иначе ET создаёт префикс ns0:gpx.
gpx_ns = "http://www.topografix.com/GPX/1/1"
xsi_ns = "http://www.w3.org/2001/XMLSchema-instance"
ET.register_namespace("", gpx_ns)
ET.register_namespace("xsi", xsi_ns)
root = ET.Element(
f"{{{gpx_ns}}}gpx",
{
"version": "1.1",
"creator": "Enduro Trails",
f"{{{xsi_ns}}}schemaLocation": (
"http://www.topografix.com/GPX/1/1 "
"http://www.topografix.com/GPX/1/1/gpx.xsd"
),
},
)
# ─── <metadata> ───────────────────────────────────────────────
# Порядок дочерних элементов фиксирован XSD-схемой GPX 1.1:
# name, desc, author, copyright, link*, time, keywords, bounds, extensions.
# Любое отклонение → DocumentInvalid (см. UT-03).
metadata = ET.SubElement(root, f"{{{gpx_ns}}}metadata")
meta_name = ET.SubElement(metadata, f"{{{gpx_ns}}}name")
meta_name.text = (name or "").strip() or "Без названия"
desc_clean = (description or "").strip()
if desc_clean:
desc_el = ET.SubElement(metadata, f"{{{gpx_ns}}}desc")
desc_el.text = desc_clean
user_clean = (user or "").strip()
if user_clean:
author_el = ET.SubElement(metadata, f"{{{gpx_ns}}}author")
author_name = ET.SubElement(author_el, f"{{{gpx_ns}}}name")
author_name.text = user_clean
# <copyright>: OSM → официальная ODbL-ссылка (ADR-014 §G).
# Для не-OSM источников: license = первый external_url (если есть),
# иначе блок опускаем.
sources_list = list(sources or [])
if "osm" in sources_list:
cr_el = ET.SubElement(
metadata, f"{{{gpx_ns}}}copyright", {"author": "Enduro Trails"}
)
lic_el = ET.SubElement(cr_el, f"{{{gpx_ns}}}license")
lic_el.text = _OSM_LICENSE_URL
elif external_urls:
first_url = next((u for u in external_urls if u), None)
if first_url:
cr_el = ET.SubElement(
metadata, f"{{{gpx_ns}}}copyright", {"author": "Enduro Trails"}
)
lic_el = ET.SubElement(cr_el, f"{{{gpx_ns}}}license")
lic_el.text = first_url
# <link> на каждый external_url; <text> = "Источник: <source_id>".
# ADR-014 §G: по одному `<link>` на каждый элемент external_urls.
src_for_link = list(sources or [])
for idx, url in enumerate(external_urls or []):
if not url:
continue
link_el = ET.SubElement(metadata, f"{{{gpx_ns}}}link", {"href": url})
text_el = ET.SubElement(link_el, f"{{{gpx_ns}}}text")
src_label = src_for_link[idx] if idx < len(src_for_link) else (
src_for_link[0] if src_for_link else ""
)
text_el.text = (
f"Источник: {src_label}" if src_label else "Источник"
)
time_str = _format_utc(created_at)
if time_str:
time_el = ET.SubElement(metadata, f"{{{gpx_ns}}}time")
time_el.text = time_str
# ─── <trk> ────────────────────────────────────────────────────
trk = ET.SubElement(root, f"{{{gpx_ns}}}trk")
trk_name = ET.SubElement(trk, f"{{{gpx_ns}}}name")
trk_name.text = (name or "").strip() or "Без названия"
act_clean = (activity_type or "").strip()
if act_clean:
trk_type = ET.SubElement(trk, f"{{{gpx_ns}}}type")
trk_type.text = act_clean
trkseg = ET.SubElement(trk, f"{{{gpx_ns}}}trkseg")
# Координаты приходят как (lon, lat) из _wkb_to_coords (см. mvt.py).
# GPX: lat/lon атрибуты с фиксированной точностью 6 знаков
# (~0.11 м, ADR-014 §G).
for lon, lat in coords or []:
ET.SubElement(
trkseg,
f"{{{gpx_ns}}}trkpt",
{"lat": f"{lat:.6f}", "lon": f"{lon:.6f}"},
)
# ET.tostring с xml_declaration=True даёт нужный prolog.
xml_bytes = ET.tostring(
root,
encoding="utf-8",
xml_declaration=True,
)
return xml_bytes.decode("utf-8")
def _sanitize_for_filesystem(name: str) -> str:
"""Заменяет запрещённые / управляющие символы на '_'.
Затем триммит пробелы и точки по краям (Windows-нюанс).
"""
out_chars: list[str] = []
for ch in name:
code = ord(ch)
if ch in _FORBIDDEN_NAME_CHARS:
out_chars.append("_")
elif code < 0x20 or code == 0x7F:
out_chars.append("_")
else:
out_chars.append(ch)
return "".join(out_chars).strip(" .")
def _truncate_utf8(name: str, max_bytes: int) -> str:
"""Триммит строку так, чтобы её UTF-8-длина не превышала max_bytes."""
encoded = name.encode("utf-8")
if len(encoded) <= max_bytes:
return name
# Декодируем с ignore, чтобы не обрезать середину code-point'а.
return encoded[:max_bytes].decode("utf-8", errors="ignore")
def _ascii_fallback(name: str) -> str:
"""ASCII-fallback для параметра `filename=` (без `*`).
ADR-014 §F.7: транслитерации **не делаем**; non-ASCII / non-printable
символы заменяем на '_'. Если результат пуст — caller подставит
'track-<id>'.
"""
out: list[str] = []
for ch in name:
code = ord(ch)
# 0x20..0x7E — printable ASCII, исключая запрещённые ФС-символы
# (они уже подменены в _sanitize_for_filesystem, но на всякий случай).
if 0x20 <= code <= 0x7E and ch not in _FORBIDDEN_NAME_CHARS:
out.append(ch)
else:
out.append("_")
return "".join(out).strip(" .")
def safe_filename(name: str | None, track_id: int) -> tuple[str, str]:
"""Возвращает (ascii_fallback, utf8_percent_quoted) без расширения.
Алгоритм (ADR-014 §F):
1. Пустой/None → 'track-<id>'.
2. Запрещённые / управляющие символы → '_'.
3. Триммим пробелы и точки.
4. Триммим до 80 байт UTF-8.
5. Пустой результат → 'track-<id>'.
6. ASCII-fallback: только printable ASCII; non-ASCII → '_'.
7. UTF-8 quoted: urllib.parse.quote(name, safe='', encoding='utf-8').
Args:
name: исходное имя (tracks.name) — может быть None / пустым.
track_id: id трека для fallback-имени.
Returns:
Кортеж (ascii_fallback, utf8_percent_quoted). Оба без расширения.
"""
fallback = f"track-{track_id}"
raw = (name or "").strip()
if not raw:
return fallback, urllib.parse.quote(fallback, safe="", encoding="utf-8")
sanitized = _sanitize_for_filesystem(raw)
if not sanitized:
return fallback, urllib.parse.quote(fallback, safe="", encoding="utf-8")
truncated = _truncate_utf8(sanitized, _MAX_NAME_BYTES).strip(" .")
if not truncated:
return fallback, urllib.parse.quote(fallback, safe="", encoding="utf-8")
utf8_quoted = urllib.parse.quote(truncated, safe="", encoding="utf-8")
ascii_ok = _ascii_fallback(truncated)
if not ascii_ok:
ascii_ok = fallback
return ascii_ok, utf8_quoted

View File

@@ -1252,7 +1252,14 @@ async def terrain_tile(layer: str, z: int, x: int, y: int):
# ─── Static files ─────────────────────────────────────────────────────────────
from src.api.gps_tracks.endpoint import create_gps_router
gps_router = create_gps_router(GPS_TRACKS_DB_PATH)
# ET-011 / ADR-015: путь к config/gps_sources.yaml — содержит per-source
# флаг `download_allowed`, который router читает один раз при старте.
GPS_SOURCES_CONFIG_PATH = os.environ.get(
"GPS_SOURCES_CONFIG_PATH",
os.path.join(os.path.dirname(__file__), "../../config/gps_sources.yaml"),
)
gps_router = create_gps_router(GPS_TRACKS_DB_PATH, GPS_SOURCES_CONFIG_PATH)
app.include_router(gps_router)
if os.path.exists(STATIC_DIR):