6675 lines
270 KiB
Python
6675 lines
270 KiB
Python
#!/usr/bin/env python3
|
||
"""Local map service with disk caching for Map Studio.
|
||
|
||
The browser only talks to this server. The server serves static project files,
|
||
proxies map/DEM tiles and Overpass railway/water queries, and caches upstream data.
|
||
It uses only the Python standard library.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import gzip
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import math
|
||
import mimetypes
|
||
import os
|
||
import queue
|
||
import re
|
||
import secrets
|
||
import shutil
|
||
import struct
|
||
import sys
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
import zlib
|
||
from dataclasses import dataclass
|
||
from http import HTTPStatus
|
||
from http import cookies
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
DEFAULT_CACHE_DIR = ROOT / "cache"
|
||
|
||
# Tunable worker counts. Manual download workers fetch tiles concurrently; HTTP
|
||
# response threads and short-lived upstream timeout workers are separate.
|
||
CACHE_REFRESH_WORKERS = 16
|
||
VECTOR_BACKGROUND_WORKERS = 16
|
||
VECTOR_TILE_REQUEST_WORKERS = 24
|
||
RAIL_AGGREGATE_WORKERS = 4
|
||
RAIL_AGGREGATE_FETCH_WORKERS = 16
|
||
TIANDITU_DOWNLOAD_WORKERS = 8
|
||
OPENMAP_DOWNLOAD_WORKERS = 16
|
||
|
||
DOWNLOAD_STAGING_DIR_NAME = ".download-tmp"
|
||
STALE_DOWNLOAD_STAGING_SECONDS = 24 * 60 * 60
|
||
|
||
DEFAULT_TTL_SECONDS = 180 * 24 * 60 * 60
|
||
DEFAULT_RAIL_TTL_SECONDS = 180 * 24 * 60 * 60
|
||
DEFAULT_REFRESH_SECONDS = 180 * 24 * 60 * 60
|
||
USER_AGENT = "MapStudioLocalMapService/1.0"
|
||
LONG_BROWSER_CACHE = "public, max-age=15552000, immutable"
|
||
MAX_VECTOR_TILES = 1200
|
||
MAX_REFRESH_TILES = 512
|
||
OPENMAP_TILE_LOCK_SECONDS = 110
|
||
OPENMAP_VECTOR_FETCH_SECONDS = 52
|
||
OPENMAP_RASTER_FETCH_SECONDS = 24
|
||
OPENMAP_LONG_RASTER_FETCH_SECONDS = 40
|
||
OPENMAP_UPSTREAM_ATTEMPT_SECONDS = 8
|
||
OPENMAP_JOB_STALL_SECONDS = 150
|
||
CACHE_ONLY_SETTING_NAME = "cache-only.setting"
|
||
UPSTREAM_PROXY_SETTING_NAME = "upstream-proxy.setting"
|
||
CACHE_COVERAGE_INDEX_NAME = ".cache-admin-coverage.json"
|
||
CACHE_COVERAGE_INDEX_VERSION = 3
|
||
CACHE_COVERAGE_TILE_SUFFIXES = {
|
||
"raster": {".png"},
|
||
"tianditu": {".png"},
|
||
"vector": {".pbf"},
|
||
"rail-derived": {".json"},
|
||
"dynamic-vector": {".json"},
|
||
}
|
||
|
||
CACHE_COVERAGE_RESPONSE_LAYERS: dict[str, list[dict[str, str]]] = {
|
||
"rail": [
|
||
{
|
||
"id": "rail-national",
|
||
"key": "national",
|
||
"label": "国铁线路瓦片",
|
||
"description": "按 OpenFreeMap 来源 Z/X/Y 瓦片生成；空白只表示当前没有已索引瓦片。",
|
||
},
|
||
{
|
||
"id": "rail-metro",
|
||
"key": "metro",
|
||
"label": "地铁 / 轻轨瓦片",
|
||
"description": "按 OpenFreeMap 来源 Z/X/Y 瓦片生成；空白只表示当前没有已索引瓦片。",
|
||
},
|
||
],
|
||
"environment": [
|
||
{
|
||
"id": "environment-water",
|
||
"key": "water",
|
||
"label": "水域瓦片",
|
||
"description": "按 OpenFreeMap 来源 Z/X/Y 瓦片生成；空白只表示当前没有已索引瓦片。",
|
||
},
|
||
{
|
||
"id": "environment-vegetation",
|
||
"key": "vegetation",
|
||
"label": "植被瓦片",
|
||
"description": "按 OpenFreeMap 来源 Z/X/Y 瓦片生成；空白只表示当前没有已索引瓦片。",
|
||
},
|
||
{
|
||
"id": "environment-ports",
|
||
"key": "ports",
|
||
"label": "港口瓦片",
|
||
"description": "按 OpenFreeMap 来源 Z/X/Y 瓦片生成；空白只表示当前没有已索引瓦片。",
|
||
},
|
||
],
|
||
}
|
||
JSON_LATITUDE_RE = re.compile(rb'"lat"\s*:\s*(-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?)')
|
||
JSON_LONGITUDE_RE = re.compile(rb'"lon"\s*:\s*(-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?)')
|
||
|
||
|
||
class CacheOnlyMode(RuntimeError):
|
||
pass
|
||
|
||
|
||
def normalize_upstream_proxy(value: str) -> str:
|
||
value = str(value or "").strip()
|
||
if not value:
|
||
return ""
|
||
if any(ord(char) < 32 or ord(char) == 127 for char in value):
|
||
raise ValueError("代理地址包含无效控制字符")
|
||
parsed = urllib.parse.urlsplit(value)
|
||
if parsed.scheme not in ("http", "https") or not parsed.hostname:
|
||
raise ValueError("代理地址必须是 http:// 或 https:// 地址")
|
||
if parsed.path not in ("", "/") or parsed.query or parsed.fragment:
|
||
raise ValueError("代理地址不能包含路径、查询参数或片段")
|
||
try:
|
||
parsed.port
|
||
except ValueError as error:
|
||
raise ValueError("代理地址端口无效") from error
|
||
return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, "", "", "")).rstrip("/")
|
||
|
||
|
||
def display_upstream_proxy(value: str) -> str:
|
||
if not value:
|
||
return ""
|
||
parsed = urllib.parse.urlsplit(value)
|
||
host = parsed.hostname or ""
|
||
if ":" in host and not host.startswith("["):
|
||
host = f"[{host}]"
|
||
port = f":{parsed.port}" if parsed.port else ""
|
||
return f"{parsed.scheme}://{host}{port}"
|
||
|
||
|
||
def build_placeholder_tile(size: int = 256) -> bytes:
|
||
rows = bytearray()
|
||
for y in range(size):
|
||
rows.append(0)
|
||
for x in range(size):
|
||
checker = ((x // 32) + (y // 32)) % 2
|
||
color = (232, 236, 232) if checker else (219, 226, 220)
|
||
if x < 3 or y < 3 or x >= size - 3 or y >= size - 3:
|
||
color = (174, 188, 178)
|
||
in_cross = (112 <= x <= 143 and 72 <= y <= 183) or (72 <= y <= 103 and 52 <= x <= 203)
|
||
if in_cross:
|
||
color = (111, 132, 119)
|
||
rows.extend(color)
|
||
|
||
def chunk(tag: bytes, data: bytes) -> bytes:
|
||
return (
|
||
struct.pack(">I", len(data))
|
||
+ tag
|
||
+ data
|
||
+ struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
|
||
)
|
||
|
||
header = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0)
|
||
return (
|
||
b"\x89PNG\r\n\x1a\n"
|
||
+ chunk(b"IHDR", header)
|
||
+ chunk(b"IDAT", zlib.compress(bytes(rows), level=6))
|
||
+ chunk(b"IEND", b"")
|
||
)
|
||
|
||
|
||
PLACEHOLDER_TILE_PNG = build_placeholder_tile()
|
||
|
||
|
||
def resolve_gis_dependencies() -> Path:
|
||
candidates: list[Path] = []
|
||
configured = os.environ.get("MAP_STUDIO_GIS_DEPS", "").strip()
|
||
if configured:
|
||
candidate = Path(configured).expanduser()
|
||
if not candidate.is_absolute():
|
||
candidate = ROOT / candidate
|
||
candidates.append(candidate)
|
||
candidates.extend((ROOT / "_gisread", ROOT.parent / "_gisread"))
|
||
for candidate in candidates:
|
||
resolved = candidate.resolve()
|
||
if resolved.is_dir():
|
||
return resolved
|
||
return candidates[0].resolve()
|
||
|
||
|
||
DYNAMIC_VECTOR_DEPENDENCIES = resolve_gis_dependencies()
|
||
|
||
_dynamic_geometry_lock = threading.Lock()
|
||
_dynamic_geometry_backend: dict[str, Any] | None = None
|
||
|
||
PUBLIC_STATIC_FILES = {
|
||
"/index.html",
|
||
"/styles.css",
|
||
"/app.js",
|
||
"/coordinate-systems.js",
|
||
"/compliance-gate.js",
|
||
"/api-comparison.html",
|
||
"/readme.md",
|
||
"/project_documentation.md",
|
||
"/print-block.js",
|
||
"/sw.js",
|
||
"/tianditu-admin.html",
|
||
"/openmap-admin.html",
|
||
"/cache-admin.html",
|
||
"/regulations.html",
|
||
}
|
||
PUBLIC_STATIC_DIRS = {"vendor"}
|
||
|
||
TILE_PROVIDERS: dict[str, dict[str, Any]] = {
|
||
"clean": {
|
||
"urls": [
|
||
"https://server.arcgisonline.com/ArcGIS/rest/services/Canvas/World_Light_Gray_Base/MapServer/tile/{z}/{y}/{x}",
|
||
],
|
||
"content_type": "image/png",
|
||
},
|
||
"clean-labels": {
|
||
"urls": [
|
||
"https://server.arcgisonline.com/ArcGIS/rest/services/Canvas/World_Light_Gray_Reference/MapServer/tile/{z}/{y}/{x}",
|
||
],
|
||
"content_type": "image/png",
|
||
},
|
||
"osm": {
|
||
"urls": [
|
||
"https://tile.openstreetmap.de/{z}/{x}/{y}.png",
|
||
"https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
||
],
|
||
"content_type": "image/png",
|
||
},
|
||
"opentopo": {
|
||
"urls": [
|
||
"https://a.tile.opentopomap.org/{z}/{x}/{y}.png",
|
||
"https://b.tile.opentopomap.org/{z}/{x}/{y}.png",
|
||
"https://c.tile.opentopomap.org/{z}/{x}/{y}.png",
|
||
"https://server.arcgisonline.com/ArcGIS/rest/services/World_Topo_Map/MapServer/tile/{z}/{y}/{x}",
|
||
],
|
||
"content_type": "image/png",
|
||
},
|
||
"topo-preview": {
|
||
"urls": [
|
||
"https://server.arcgisonline.com/ArcGIS/rest/services/World_Topo_Map/MapServer/tile/{z}/{y}/{x}",
|
||
],
|
||
"content_type": "image/jpeg",
|
||
},
|
||
"satellite": {
|
||
"urls": [
|
||
"https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}",
|
||
"https://services.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}",
|
||
],
|
||
"content_type": "image/jpeg",
|
||
},
|
||
"railway": {
|
||
"urls": [
|
||
"https://tiles.openrailwaymap.org/standard/{z}/{x}/{y}.png",
|
||
],
|
||
"content_type": "image/png",
|
||
},
|
||
"dem": {
|
||
"urls": [
|
||
"https://elevation-tiles-prod.s3.amazonaws.com/terrarium/{z}/{x}/{y}.png",
|
||
],
|
||
"content_type": "image/png",
|
||
},
|
||
}
|
||
|
||
for provider_name, layer_name, label in (
|
||
("tdt-vec", "vec", "天地图矢量底图"),
|
||
("tdt-vec-label", "cva", "天地图矢量注记"),
|
||
("tdt-img", "img", "天地图影像底图"),
|
||
("tdt-img-label", "cia", "天地图影像注记"),
|
||
("tdt-ter", "ter", "天地图地形底图"),
|
||
("tdt-ter-label", "cta", "天地图地形注记"),
|
||
("tdt-boundary", "ibo", "天地图行政区划"),
|
||
):
|
||
TILE_PROVIDERS[provider_name] = {
|
||
"urls": [
|
||
"https://t0.tianditu.gov.cn/"
|
||
+ layer_name
|
||
+ "_w/wmts?SERVICE=WMTS&REQUEST=GetTile&VERSION=1.0.0&LAYER="
|
||
+ layer_name
|
||
+ "&STYLE=default&TILEMATRIXSET=w&FORMAT=tiles&TILEMATRIX={z}&TILEROW={y}&TILECOL={x}&tk={key}"
|
||
],
|
||
"content_type": "image/png",
|
||
"requires_key": True,
|
||
"manual_only": True,
|
||
"label": label,
|
||
"tianditu_layer": layer_name,
|
||
}
|
||
|
||
OVERPASS_ENDPOINTS = [
|
||
"https://overpass-api.de/api/interpreter",
|
||
"https://overpass.kumi.systems/api/interpreter",
|
||
"https://overpass.private.coffee/api/interpreter",
|
||
"https://overpass.monicz.dev/api/interpreter",
|
||
]
|
||
|
||
OPENFREEMAP_TILEJSON_URL = "https://tiles.openfreemap.org/planet"
|
||
OPENFREEMAP_FALLBACK_TILE_URL = (
|
||
"https://tiles.openfreemap.org/planet/20261101_080013_pt/{z}/{x}/{y}.pbf"
|
||
)
|
||
OPENFREEMAP_SOURCE_MIN_ZOOM = 3
|
||
OPENFREEMAP_SOURCE_MAX_ZOOM = 14
|
||
ENVIRONMENT_MIN_ZOOM = 3
|
||
ENVIRONMENT_MAX_ZOOM = 16
|
||
|
||
OPEN_MAP_PROVIDERS: dict[str, dict[str, Any]] = {
|
||
"osm": {
|
||
"label": "OpenStreetMap 标准地图",
|
||
"kind": "raster",
|
||
"provider": "osm",
|
||
"minZoom": 1,
|
||
"maxZoom": 19,
|
||
"source": "OpenStreetMap",
|
||
"license": "ODbL",
|
||
"description": "OpenStreetMap 标准栅格底图",
|
||
},
|
||
"opentopo": {
|
||
"label": "OpenTopoMap 地形图",
|
||
"kind": "raster",
|
||
"provider": "opentopo",
|
||
"minZoom": 1,
|
||
"maxZoom": 17,
|
||
"source": "OpenTopoMap / OpenStreetMap",
|
||
"license": "CC-BY-SA / ODbL",
|
||
"description": "开放地形、等高线与道路渲染",
|
||
},
|
||
"railway": {
|
||
"label": "OpenRailwayMap 铁路基础设施",
|
||
"kind": "raster",
|
||
"provider": "railway",
|
||
"minZoom": 1,
|
||
"maxZoom": 19,
|
||
"source": "OpenRailwayMap / OpenStreetMap",
|
||
"license": "ODbL",
|
||
"description": "铁路、站场、信号等基础设施叠加瓦片",
|
||
},
|
||
"dem": {
|
||
"label": "AWS Terrain Tiles 开放地形数据",
|
||
"kind": "raster",
|
||
"provider": "dem",
|
||
"minZoom": 3,
|
||
"maxZoom": 15,
|
||
"source": "AWS Open Data / Terrarium",
|
||
"license": "开放地形数据",
|
||
"description": "供等高线、地形晕渲和视线分析使用",
|
||
},
|
||
"openfreemap": {
|
||
"label": "OpenFreeMap 全球矢量专题",
|
||
"kind": "vector",
|
||
"provider": None,
|
||
"minZoom": OPENFREEMAP_SOURCE_MIN_ZOOM,
|
||
"maxZoom": OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
"source": "OpenFreeMap / OpenStreetMap",
|
||
"license": "ODbL",
|
||
"description": "水域、植被、港口、铁路和行政边界原始 MVT",
|
||
},
|
||
"clean": {
|
||
"label": "简洁底图（Esri）",
|
||
"kind": "raster",
|
||
"provider": "clean",
|
||
"minZoom": 1,
|
||
"maxZoom": 19,
|
||
"source": "Esri Light Gray Canvas",
|
||
"license": "Esri Terms of Use",
|
||
"description": "主地图简洁底图；离线使用时建议同时选择简洁底图注记",
|
||
},
|
||
"clean-labels": {
|
||
"label": "简洁底图注记（Esri）",
|
||
"kind": "raster",
|
||
"provider": "clean-labels",
|
||
"minZoom": 1,
|
||
"maxZoom": 19,
|
||
"source": "Esri Light Gray Reference",
|
||
"license": "Esri Terms of Use",
|
||
"description": "简洁底图上的道路和地名注记",
|
||
},
|
||
"topo-preview": {
|
||
"label": "地形图（Esri）",
|
||
"kind": "raster",
|
||
"provider": "topo-preview",
|
||
"minZoom": 1,
|
||
"maxZoom": 13,
|
||
"source": "Esri World Topographic Map",
|
||
"license": "Esri Terms of Use",
|
||
"description": "主地图地形底图；原生最高 Z13",
|
||
},
|
||
"satellite": {
|
||
"label": "卫星影像（Esri）",
|
||
"kind": "raster",
|
||
"provider": "satellite",
|
||
"minZoom": 1,
|
||
"maxZoom": 19,
|
||
"source": "Esri World Imagery",
|
||
"license": "Esri Terms of Use",
|
||
"description": "主地图卫星影像底图",
|
||
},
|
||
}
|
||
|
||
CACHE_ADMIN_CATEGORIES: dict[str, dict[str, Any]] = {
|
||
"raster": {
|
||
"label": "底图、注记与地形瓦片",
|
||
"description": "主地图和开放地图后台使用的非天地图栅格、DEM 与预览瓦片。",
|
||
"refreshable": True,
|
||
},
|
||
"tianditu": {
|
||
"label": "天地图手动离线瓦片",
|
||
"description": "由天地图下载后台手动获取的离线瓦片；更新请使用天地图下载后台。",
|
||
"refreshable": False,
|
||
},
|
||
"vector": {
|
||
"label": "OpenFreeMap 全球原始矢量瓦片",
|
||
"description": "水域、植被、港口、铁路和行政边界使用的全球 MVT 源瓦片；实际显示范围取决于已缓存瓦片。",
|
||
"refreshable": True,
|
||
},
|
||
"rail": {
|
||
"label": "铁路查询响应集合",
|
||
"description": "按视窗保存的铁路专题响应集合；管理页将国铁和地铁拆分为两个可单独管理的入口。",
|
||
"refreshable": True,
|
||
},
|
||
"environment": {
|
||
"label": "环境查询响应集合",
|
||
"description": "按视窗保存的环境专题响应集合；管理页将水域、植被和港口拆分为三个可单独管理的入口。",
|
||
"refreshable": True,
|
||
},
|
||
"boundaries": {
|
||
"label": "行政边界响应",
|
||
"description": "由本服务从 OpenFreeMap 行政边界矢量瓦片生成并缓存的视窗响应，不依赖外部本地 API。",
|
||
"refreshable": False,
|
||
},
|
||
"rail-derived": {
|
||
"label": "派生铁路瓦片",
|
||
"description": "由矢量源生成的国铁和地铁本地瓦片；删除后按地图请求自动重建。",
|
||
"refreshable": False,
|
||
},
|
||
"dynamic-vector": {
|
||
"label": "按需裁切矢量瓦片",
|
||
"description": "1:100 万和 1:25 万高缩放按需裁切缓存；删除后自动重建。",
|
||
"refreshable": False,
|
||
},
|
||
"pending": {
|
||
"label": "待重试任务",
|
||
"description": "铁路和环境查询失败后保存的待重试队列。",
|
||
"refreshable": False,
|
||
},
|
||
"staging": {
|
||
"label": "下载暂存残留",
|
||
"description": "下载中的 .part 临时文件；仅在没有运行中的下载任务时删除。",
|
||
"refreshable": False,
|
||
},
|
||
}
|
||
|
||
CACHE_ADMIN_QUERY_LAYERS: dict[str, dict[str, Any]] = {
|
||
"rail-national": {
|
||
"parent": "rail",
|
||
"key": "national",
|
||
"label": "国铁线路瓦片",
|
||
"description": "按 OpenFreeMap 全球铁路来源 Z/X/Y 瓦片索引；可查看、更新和下载这些来源瓦片。删除专题只移除本专题索引，共享原始瓦片仍由 OpenFreeMap 原始瓦片统一管理。",
|
||
"refreshable": True,
|
||
},
|
||
"rail-metro": {
|
||
"parent": "rail",
|
||
"key": "metro",
|
||
"label": "地铁 / 轻轨瓦片",
|
||
"description": "按 OpenFreeMap 全球交通来源 Z/X/Y 瓦片索引地铁、轻轨、单轨和有轨电车；可查看、更新和下载来源瓦片。",
|
||
"refreshable": True,
|
||
},
|
||
"environment-water": {
|
||
"parent": "environment",
|
||
"key": "water",
|
||
"label": "水域瓦片",
|
||
"description": "按 OpenFreeMap 全球水域来源 Z/X/Y 瓦片建立索引；可查看、更新和下载这些来源瓦片。",
|
||
"refreshable": True,
|
||
},
|
||
"environment-vegetation": {
|
||
"parent": "environment",
|
||
"key": "vegetation",
|
||
"label": "植被瓦片",
|
||
"description": "按 OpenFreeMap 全球地表覆盖来源 Z/X/Y 瓦片建立索引；可查看、更新和下载这些来源瓦片。",
|
||
"refreshable": True,
|
||
},
|
||
"environment-ports": {
|
||
"parent": "environment",
|
||
"key": "ports",
|
||
"label": "港口瓦片",
|
||
"description": "按 OpenFreeMap 港口来源 Z/X/Y 瓦片建立索引；可查看、更新和下载这些来源瓦片。",
|
||
"refreshable": True,
|
||
},
|
||
"boundaries": {
|
||
"parent": "boundaries",
|
||
"key": "boundaries",
|
||
"label": "行政边界瓦片",
|
||
"description": "按 OpenFreeMap 行政边界来源 Z/X/Y 瓦片建立索引，不依赖外部本地 API；可查看、更新和下载这些来源瓦片。",
|
||
"refreshable": True,
|
||
},
|
||
}
|
||
|
||
CACHE_ADMIN_MANAGEMENT_ORDER = [
|
||
"raster",
|
||
"tianditu",
|
||
"vector",
|
||
"rail-national",
|
||
"rail-metro",
|
||
"environment-water",
|
||
"environment-vegetation",
|
||
"environment-ports",
|
||
"boundaries",
|
||
"rail-derived",
|
||
"dynamic-vector",
|
||
"pending",
|
||
"staging",
|
||
]
|
||
|
||
TILE_PATH_RE = re.compile(r"^/tiles/([a-z0-9-]+)/(\d+)/(\d+)/(\d+)\.png$")
|
||
LOCAL_VECTOR_PATH_RE = re.compile(r"^/api/local-vector/(\d+)/(\d+)/(\d+)\.json$")
|
||
LOCAL_VECTOR_25_PATH_RE = re.compile(r"^/api/local-vector-25/(\d+)/(\d+)/(\d+)\.json$")
|
||
RAIL_TILE_PATH_RE = re.compile(r"^/api/rail-tiles/(\d+)/(\d+)/(\d+)\.json$")
|
||
KNOWN_TOP_LEVEL = {
|
||
"index.html",
|
||
"styles.css",
|
||
"app.js",
|
||
"compliance-gate.js",
|
||
"api-comparison.html",
|
||
"README.md",
|
||
"PROJECT_DOCUMENTATION.md",
|
||
"print-block.js",
|
||
"sw.js",
|
||
"tianditu-admin.html",
|
||
"openmap-admin.html",
|
||
"cache-admin.html",
|
||
"regulations.html",
|
||
"tianditu-key.example.txt",
|
||
"admin-password.example.txt",
|
||
"admin-password.example.txt",
|
||
"start-map-studio.bat",
|
||
"vendor",
|
||
}
|
||
|
||
|
||
class RailPending(Exception):
|
||
def __init__(self, cache_key: str) -> None:
|
||
super().__init__("Rail request queued for background retry")
|
||
self.cache_key = cache_key
|
||
|
||
|
||
class WaterPending(Exception):
|
||
def __init__(self, cache_key: str) -> None:
|
||
super().__init__("Water request queued for background retry")
|
||
self.cache_key = cache_key
|
||
|
||
|
||
class TileNotCached(Exception):
|
||
pass
|
||
|
||
|
||
class TiandituKeyMissing(Exception):
|
||
pass
|
||
|
||
|
||
ATOMIC_STAGING_DIR: Path | None = None
|
||
|
||
|
||
def configure_atomic_staging(cache_dir: Path) -> Path:
|
||
"""Place incomplete files outside the cache tree until they are complete."""
|
||
global ATOMIC_STAGING_DIR
|
||
staging_dir = cache_dir.resolve() / DOWNLOAD_STAGING_DIR_NAME
|
||
staging_dir.mkdir(parents=True, exist_ok=True)
|
||
cutoff = time.time() - STALE_DOWNLOAD_STAGING_SECONDS
|
||
for leftover in staging_dir.glob("*.part"):
|
||
try:
|
||
if leftover.stat().st_mtime < cutoff:
|
||
leftover.unlink()
|
||
except OSError:
|
||
pass
|
||
ATOMIC_STAGING_DIR = staging_dir
|
||
return staging_dir
|
||
|
||
|
||
def atomic_write(path: Path, data: bytes) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
staging_dir = ATOMIC_STAGING_DIR
|
||
temporary = (staging_dir or path.parent) / (
|
||
f"{path.name}.{os.getpid()}-{threading.get_ident()}-{uuid.uuid4().hex}.part"
|
||
)
|
||
try:
|
||
with temporary.open("wb") as handle:
|
||
handle.write(data)
|
||
handle.flush()
|
||
os.fsync(handle.fileno())
|
||
os.replace(temporary, path)
|
||
finally:
|
||
try:
|
||
temporary.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def detect_image_content_type(data: bytes, fallback: str) -> str:
|
||
if data.startswith(b"\x89PNG\r\n\x1a\n"):
|
||
return "image/png"
|
||
if data.startswith(b"\xff\xd8\xff"):
|
||
return "image/jpeg"
|
||
if data.startswith(b"GIF87a") or data.startswith(b"GIF89a"):
|
||
return "image/gif"
|
||
if len(data) > 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
||
return "image/webp"
|
||
return fallback
|
||
|
||
|
||
def _decode_varint(data: bytes, offset: int) -> tuple[int, int]:
|
||
value = 0
|
||
shift = 0
|
||
while offset < len(data) and shift <= 70:
|
||
byte = data[offset]
|
||
offset += 1
|
||
value |= (byte & 0x7F) << shift
|
||
if not byte & 0x80:
|
||
return value, offset
|
||
shift += 7
|
||
raise ValueError("Invalid protobuf varint")
|
||
|
||
|
||
def _decode_zigzag(value: int) -> int:
|
||
return (value >> 1) ^ -(value & 1)
|
||
|
||
|
||
def _protobuf_fields(data: bytes):
|
||
offset = 0
|
||
while offset < len(data):
|
||
key, offset = _decode_varint(data, offset)
|
||
field_number = key >> 3
|
||
wire_type = key & 7
|
||
if wire_type == 0:
|
||
value, offset = _decode_varint(data, offset)
|
||
elif wire_type == 1:
|
||
value = data[offset : offset + 8]
|
||
offset += 8
|
||
elif wire_type == 2:
|
||
length, offset = _decode_varint(data, offset)
|
||
value = data[offset : offset + length]
|
||
offset += length
|
||
elif wire_type == 5:
|
||
value = data[offset : offset + 4]
|
||
offset += 4
|
||
else:
|
||
raise ValueError(f"Unsupported protobuf wire type: {wire_type}")
|
||
yield field_number, wire_type, value
|
||
|
||
|
||
def _decode_mvt_value(data: bytes):
|
||
for field_number, wire_type, value in _protobuf_fields(data):
|
||
if field_number == 1 and wire_type == 2:
|
||
return bytes(value).decode("utf-8", errors="replace")
|
||
if field_number == 2 and wire_type == 5:
|
||
return struct.unpack("<f", value)[0]
|
||
if field_number == 3 and wire_type == 1:
|
||
return struct.unpack("<d", value)[0]
|
||
if field_number in (4, 5) and wire_type == 0:
|
||
return value
|
||
if field_number == 6 and wire_type == 0:
|
||
return _decode_zigzag(value)
|
||
if field_number == 7 and wire_type == 0:
|
||
return bool(value)
|
||
return None
|
||
|
||
|
||
def _decode_mvt_geometry(values: list[int], geometry_type: int) -> list[list[tuple[int, int]]]:
|
||
paths: list[list[tuple[int, int]]] = []
|
||
current: list[tuple[int, int]] | None = None
|
||
x = 0
|
||
y = 0
|
||
index = 0
|
||
while index < len(values):
|
||
command_integer = values[index]
|
||
index += 1
|
||
command = command_integer & 7
|
||
count = command_integer >> 3
|
||
if command in (1, 2):
|
||
if command == 1:
|
||
current = []
|
||
paths.append(current)
|
||
elif current is None:
|
||
current = []
|
||
paths.append(current)
|
||
for _ in range(count):
|
||
if index + 1 >= len(values):
|
||
break
|
||
x += _decode_zigzag(values[index])
|
||
y += _decode_zigzag(values[index + 1])
|
||
index += 2
|
||
current.append((x, y))
|
||
elif command == 7 and current and geometry_type == 3:
|
||
if current[0] != current[-1]:
|
||
current.append(current[0])
|
||
else:
|
||
break
|
||
return [path for path in paths if path]
|
||
|
||
|
||
def _decode_mvt_feature(
|
||
data: bytes,
|
||
keys: list[str],
|
||
values: list[Any],
|
||
extent: int,
|
||
) -> dict[str, Any] | None:
|
||
tags: list[int] = []
|
||
geometry: list[int] = []
|
||
geometry_type = 0
|
||
feature_id: int | None = None
|
||
for field_number, wire_type, value in _protobuf_fields(data):
|
||
if field_number == 1 and wire_type == 0:
|
||
feature_id = value
|
||
elif field_number == 2:
|
||
if wire_type == 2:
|
||
offset = 0
|
||
while offset < len(value):
|
||
item, offset = _decode_varint(value, offset)
|
||
tags.append(item)
|
||
elif wire_type == 0:
|
||
tags.append(value)
|
||
elif field_number == 3 and wire_type == 0:
|
||
geometry_type = value
|
||
elif field_number == 4:
|
||
if wire_type == 2:
|
||
offset = 0
|
||
while offset < len(value):
|
||
item, offset = _decode_varint(value, offset)
|
||
geometry.append(item)
|
||
elif wire_type == 0:
|
||
geometry.append(value)
|
||
properties: dict[str, Any] = {}
|
||
for index in range(0, len(tags) - 1, 2):
|
||
key_index = tags[index]
|
||
value_index = tags[index + 1]
|
||
if key_index < len(keys) and value_index < len(values):
|
||
properties[keys[key_index]] = values[value_index]
|
||
paths = _decode_mvt_geometry(geometry, geometry_type)
|
||
if not paths:
|
||
return None
|
||
return {
|
||
"id": feature_id,
|
||
"type": geometry_type,
|
||
"properties": properties,
|
||
"paths": paths,
|
||
"extent": extent,
|
||
}
|
||
|
||
|
||
def decode_vector_tile(
|
||
data: bytes,
|
||
required_layers: set[str] | None = None,
|
||
) -> dict[str, list[dict[str, Any]]]:
|
||
layers: dict[str, list[dict[str, Any]]] = {}
|
||
for field_number, wire_type, layer_data in _protobuf_fields(data):
|
||
if field_number != 3 or wire_type != 2:
|
||
continue
|
||
layer_name = ""
|
||
keys: list[str] = []
|
||
values: list[Any] = []
|
||
feature_messages: list[bytes] = []
|
||
extent = 4096
|
||
for layer_field, layer_wire, value in _protobuf_fields(layer_data):
|
||
if layer_field == 1 and layer_wire == 2:
|
||
layer_name = bytes(value).decode("utf-8", errors="replace")
|
||
elif layer_field == 2 and layer_wire == 2:
|
||
feature_messages.append(bytes(value))
|
||
elif layer_field == 3 and layer_wire == 2:
|
||
keys.append(bytes(value).decode("utf-8", errors="replace"))
|
||
elif layer_field == 4 and layer_wire == 2:
|
||
values.append(_decode_mvt_value(bytes(value)))
|
||
elif layer_field == 5 and layer_wire == 0:
|
||
extent = max(1, int(value))
|
||
features = []
|
||
if layer_name and (required_layers is None or layer_name in required_layers):
|
||
for feature_data in feature_messages:
|
||
feature = _decode_mvt_feature(feature_data, keys, values, extent)
|
||
if feature:
|
||
features.append(feature)
|
||
if layer_name and (required_layers is None or layer_name in required_layers):
|
||
layers[layer_name] = features
|
||
return layers
|
||
|
||
|
||
def clip_segment_to_extent(
|
||
start: tuple[int, int],
|
||
end: tuple[int, int],
|
||
extent: int,
|
||
) -> tuple[tuple[float, float], tuple[float, float]] | None:
|
||
x0, y0 = float(start[0]), float(start[1])
|
||
x1, y1 = float(end[0]), float(end[1])
|
||
dx = x1 - x0
|
||
dy = y1 - y0
|
||
lower = 0.0
|
||
upper = 1.0
|
||
for p, q in ((-dx, x0), (dx, extent - x0), (-dy, y0), (dy, extent - y0)):
|
||
if abs(p) < 1e-12:
|
||
if q < 0:
|
||
return None
|
||
continue
|
||
ratio = q / p
|
||
if p < 0:
|
||
lower = max(lower, ratio)
|
||
else:
|
||
upper = min(upper, ratio)
|
||
if lower > upper:
|
||
return None
|
||
def clamp(point: tuple[float, float]) -> tuple[float, float]:
|
||
return (
|
||
max(0.0, min(float(extent), point[0])),
|
||
max(0.0, min(float(extent), point[1])),
|
||
)
|
||
|
||
return clamp((x0 + lower * dx, y0 + lower * dy)), clamp(
|
||
(x0 + upper * dx, y0 + upper * dy)
|
||
)
|
||
|
||
|
||
def clip_line_path(path: list[tuple[int, int]], extent: int) -> list[list[tuple[float, float]]]:
|
||
result: list[list[tuple[float, float]]] = []
|
||
current: list[tuple[float, float]] = []
|
||
for start, end in zip(path, path[1:]):
|
||
clipped = clip_segment_to_extent(start, end, extent)
|
||
if clipped is None:
|
||
if len(current) >= 2:
|
||
result.append(current)
|
||
current = []
|
||
continue
|
||
segment_start, segment_end = clipped
|
||
if current and abs(current[-1][0] - segment_start[0]) < 1e-7 and abs(
|
||
current[-1][1] - segment_start[1]
|
||
) < 1e-7:
|
||
current.append(segment_end)
|
||
else:
|
||
if len(current) >= 2:
|
||
result.append(current)
|
||
current = [segment_start, segment_end]
|
||
if len(current) >= 2:
|
||
result.append(current)
|
||
return result
|
||
|
||
|
||
def clip_polygon_path(path: list[tuple[int, int]], extent: int) -> list[tuple[float, float]]:
|
||
points = [
|
||
(float(point[0]), float(point[1]))
|
||
for point in path
|
||
if -extent <= point[0] <= extent * 2 and -extent <= point[1] <= extent * 2
|
||
]
|
||
if len(points) > 1 and points[0] == points[-1]:
|
||
points.pop()
|
||
if len(points) < 3:
|
||
return []
|
||
|
||
def clip_axis(
|
||
source: list[tuple[float, float]],
|
||
axis: int,
|
||
boundary: float,
|
||
keep_greater: bool,
|
||
) -> list[tuple[float, float]]:
|
||
if not source:
|
||
return []
|
||
output: list[tuple[float, float]] = []
|
||
previous = source[-1]
|
||
previous_inside = previous[axis] >= boundary if keep_greater else previous[axis] <= boundary
|
||
for current in source:
|
||
current_inside = current[axis] >= boundary if keep_greater else current[axis] <= boundary
|
||
if current_inside != previous_inside:
|
||
delta = current[axis] - previous[axis]
|
||
ratio = 0.0 if abs(delta) < 1e-12 else (boundary - previous[axis]) / delta
|
||
intersection = (
|
||
previous[0] + ratio * (current[0] - previous[0]),
|
||
previous[1] + ratio * (current[1] - previous[1]),
|
||
)
|
||
intersection = (
|
||
max(0.0, min(float(extent), intersection[0])),
|
||
max(0.0, min(float(extent), intersection[1])),
|
||
)
|
||
output.append(intersection)
|
||
if current_inside:
|
||
output.append(
|
||
(
|
||
max(0.0, min(float(extent), current[0])),
|
||
max(0.0, min(float(extent), current[1])),
|
||
)
|
||
)
|
||
previous = current
|
||
previous_inside = current_inside
|
||
return output
|
||
|
||
for axis, boundary, keep_greater in (
|
||
(0, 0.0, True),
|
||
(0, float(extent), False),
|
||
(1, 0.0, True),
|
||
(1, float(extent), False),
|
||
):
|
||
points = clip_axis(points, axis, boundary, keep_greater)
|
||
if len(points) < 3:
|
||
return []
|
||
if points[0] != points[-1]:
|
||
points.append(points[0])
|
||
return points
|
||
|
||
|
||
def simplify_path(
|
||
points: list[tuple[float, float]],
|
||
tolerance: float,
|
||
) -> list[tuple[float, float]]:
|
||
if len(points) <= 3 or tolerance <= 0:
|
||
return points
|
||
keep = {0, len(points) - 1}
|
||
stack = [(0, len(points) - 1)]
|
||
tolerance_squared = tolerance * tolerance
|
||
while stack:
|
||
start_index, end_index = stack.pop()
|
||
start = points[start_index]
|
||
end = points[end_index]
|
||
dx = end[0] - start[0]
|
||
dy = end[1] - start[1]
|
||
length_squared = dx * dx + dy * dy
|
||
farthest_index = -1
|
||
farthest_distance = tolerance_squared
|
||
for index in range(start_index + 1, end_index):
|
||
point = points[index]
|
||
if length_squared == 0:
|
||
distance_squared = (
|
||
(point[0] - start[0]) ** 2 + (point[1] - start[1]) ** 2
|
||
)
|
||
else:
|
||
ratio = max(
|
||
0.0,
|
||
min(
|
||
1.0,
|
||
((point[0] - start[0]) * dx + (point[1] - start[1]) * dy)
|
||
/ length_squared,
|
||
),
|
||
)
|
||
projected_x = start[0] + ratio * dx
|
||
projected_y = start[1] + ratio * dy
|
||
distance_squared = (
|
||
(point[0] - projected_x) ** 2 + (point[1] - projected_y) ** 2
|
||
)
|
||
if distance_squared > farthest_distance:
|
||
farthest_distance = distance_squared
|
||
farthest_index = index
|
||
if farthest_index >= 0:
|
||
keep.add(farthest_index)
|
||
stack.append((start_index, farthest_index))
|
||
stack.append((farthest_index, end_index))
|
||
return [points[index] for index in sorted(keep)]
|
||
|
||
|
||
def clipped_feature_paths(feature: dict[str, Any]) -> list[list[tuple[float, float]]]:
|
||
extent = int(feature["extent"])
|
||
geometry_type = int(feature["type"])
|
||
polygon_tolerance = max(1.0, extent / 2048.0)
|
||
line_tolerance = max(0.75, extent / 4096.0)
|
||
result: list[list[tuple[float, float]]] = []
|
||
for path in feature["paths"]:
|
||
if geometry_type == 1:
|
||
if path and 0 <= path[0][0] <= extent and 0 <= path[0][1] <= extent:
|
||
result.append([(float(path[0][0]), float(path[0][1]))])
|
||
elif geometry_type == 3:
|
||
clipped = clip_polygon_path(path, extent)
|
||
if clipped:
|
||
simplified = simplify_path(clipped, polygon_tolerance)
|
||
if len(simplified) >= 4:
|
||
if simplified[0] != simplified[-1]:
|
||
simplified.append(simplified[0])
|
||
result.append(simplified)
|
||
else:
|
||
result.extend(
|
||
simplified
|
||
for clipped in clip_line_path(path, extent)
|
||
for simplified in [simplify_path(clipped, line_tolerance)]
|
||
if len(simplified) >= 2
|
||
)
|
||
return result
|
||
|
||
|
||
def normalize_request_path(path: str, accept_header: str = "") -> str:
|
||
decoded = urllib.parse.unquote(path)
|
||
if decoded in ("", "/"):
|
||
return "/"
|
||
parts = [part for part in decoded.split("/") if part]
|
||
for index in range(len(parts)):
|
||
candidate = "/" + "/".join(parts[index:])
|
||
if candidate.startswith(("/tiles/", "/api/")):
|
||
return candidate
|
||
first = parts[index]
|
||
if first in KNOWN_TOP_LEVEL:
|
||
return candidate
|
||
if "text/html" in accept_header.lower():
|
||
return "/"
|
||
return decoded
|
||
|
||
|
||
def vector_tile_bounds(x: int, y: int, z: int):
|
||
backend = dynamic_geometry_backend()
|
||
count = 1 << z
|
||
west = x / count * 360.0 - 180.0
|
||
east = (x + 1) / count * 360.0 - 180.0
|
||
north = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * y / count))))
|
||
south = math.degrees(
|
||
math.atan(math.sinh(math.pi * (1.0 - 2.0 * (y + 1) / count)))
|
||
)
|
||
return backend["box"](west, south, east, north)
|
||
|
||
|
||
def vector_tile_bbox(z: int, x: int, y: int) -> tuple[float, float, float, float]:
|
||
count = 1 << z
|
||
west = x / count * 360.0 - 180.0
|
||
east = (x + 1) / count * 360.0 - 180.0
|
||
north = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * y / count))))
|
||
south = math.degrees(
|
||
math.atan(math.sinh(math.pi * (1.0 - 2.0 * (y + 1) / count)))
|
||
)
|
||
return south, west, north, east
|
||
|
||
|
||
def rounded_coordinates(value: Any, precision: int = 6) -> Any:
|
||
if isinstance(value, (int, float)):
|
||
return round(float(value), precision)
|
||
if isinstance(value, (list, tuple)):
|
||
return [rounded_coordinates(item, precision) for item in value]
|
||
return value
|
||
|
||
|
||
def rounded_geometry_json(geometry: dict[str, Any]) -> dict[str, Any]:
|
||
if geometry.get("type") == "GeometryCollection":
|
||
return {
|
||
"type": "GeometryCollection",
|
||
"geometries": [
|
||
rounded_geometry_json(item)
|
||
for item in geometry.get("geometries", [])
|
||
if item
|
||
],
|
||
}
|
||
result = dict(geometry)
|
||
if "coordinates" in result:
|
||
result["coordinates"] = rounded_coordinates(result["coordinates"])
|
||
return result
|
||
|
||
|
||
def dynamic_geometry_backend() -> dict[str, Any]:
|
||
global _dynamic_geometry_backend
|
||
if _dynamic_geometry_backend is not None:
|
||
return _dynamic_geometry_backend
|
||
with _dynamic_geometry_lock:
|
||
if _dynamic_geometry_backend is not None:
|
||
return _dynamic_geometry_backend
|
||
dependency = str(DYNAMIC_VECTOR_DEPENDENCIES)
|
||
if DYNAMIC_VECTOR_DEPENDENCIES.is_dir() and dependency not in sys.path:
|
||
sys.path.append(dependency)
|
||
try:
|
||
from shapely.geometry import box, mapping, shape
|
||
except Exception as error:
|
||
raise RuntimeError(
|
||
"High-zoom vector generation requires shapely and the local GIS dependencies"
|
||
) from error
|
||
_dynamic_geometry_backend = {
|
||
"box": box,
|
||
"mapping": mapping,
|
||
"shape": shape,
|
||
}
|
||
return _dynamic_geometry_backend
|
||
|
||
|
||
def clip_local_vector_payload(
|
||
payload: dict[str, Any],
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
manifest_generated_at: str = "",
|
||
) -> dict[str, Any]:
|
||
backend = dynamic_geometry_backend()
|
||
clip_box = vector_tile_bounds(x, y, z)
|
||
features: list[dict[str, Any]] = []
|
||
for feature in payload.get("features", []):
|
||
raw_geometry = feature.get("geometry")
|
||
if not raw_geometry:
|
||
continue
|
||
try:
|
||
geometry = backend["shape"](raw_geometry)
|
||
clipped = geometry.intersection(clip_box)
|
||
except Exception:
|
||
continue
|
||
if clipped.is_empty:
|
||
continue
|
||
mapped = rounded_geometry_json(backend["mapping"](clipped))
|
||
features.append(
|
||
{
|
||
"type": "Feature",
|
||
"id": feature.get("id"),
|
||
"properties": feature.get("properties") or {},
|
||
"geometry": mapped,
|
||
}
|
||
)
|
||
return {
|
||
"type": "FeatureCollection",
|
||
"format": "local-geojson-vector-v1",
|
||
"generatedAt": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||
"z": z,
|
||
"x": x,
|
||
"y": y,
|
||
"dynamic": True,
|
||
"manifestGeneratedAt": manifest_generated_at,
|
||
"features": features,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class ServiceConfig:
|
||
host: str
|
||
port: int
|
||
cache_dir: Path
|
||
tile_ttl_seconds: int
|
||
rail_ttl_seconds: int
|
||
refresh_seconds: int
|
||
refresh_batch: int
|
||
tianditu_key: str = ""
|
||
admin_password: str = ""
|
||
cache_only: bool = False
|
||
upstream_proxy: str = ""
|
||
|
||
|
||
class DaemonTaskPool:
|
||
def __init__(self, max_workers: int, name: str) -> None:
|
||
self._queue: queue.Queue[Any] = queue.Queue()
|
||
self._stopped = threading.Event()
|
||
self._workers = [
|
||
threading.Thread(
|
||
target=self._worker,
|
||
name=f"{name}-{index}",
|
||
daemon=True,
|
||
)
|
||
for index in range(max(1, max_workers))
|
||
]
|
||
for worker in self._workers:
|
||
worker.start()
|
||
|
||
def _worker(self) -> None:
|
||
while not self._stopped.is_set():
|
||
item = self._queue.get()
|
||
if item is None:
|
||
return
|
||
function, args, kwargs = item
|
||
if self._stopped.is_set():
|
||
continue
|
||
try:
|
||
function(*args, **kwargs)
|
||
except Exception:
|
||
pass
|
||
|
||
def submit(self, function: Any, *args: Any, **kwargs: Any) -> None:
|
||
if not self._stopped.is_set():
|
||
self._queue.put((function, args, kwargs))
|
||
|
||
def shutdown(self, wait: bool = False, cancel_futures: bool = False) -> None:
|
||
self._stopped.set()
|
||
while True:
|
||
try:
|
||
self._queue.get_nowait()
|
||
except queue.Empty:
|
||
break
|
||
for _ in self._workers:
|
||
self._queue.put(None)
|
||
|
||
|
||
def run_with_timeout(
|
||
function: Any,
|
||
timeout_seconds: float,
|
||
label: str,
|
||
*args: Any,
|
||
**kwargs: Any,
|
||
) -> Any:
|
||
"""Bound an outbound request even when DNS or a remote socket stalls."""
|
||
result_queue: queue.Queue[tuple[bool, Any]] = queue.Queue(maxsize=1)
|
||
|
||
def run() -> None:
|
||
try:
|
||
result_queue.put((True, function(*args, **kwargs)), block=False)
|
||
except Exception as error:
|
||
try:
|
||
result_queue.put((False, error), block=False)
|
||
except queue.Full:
|
||
pass
|
||
|
||
threading.Thread(
|
||
target=run,
|
||
name="openmap-outbound-request",
|
||
daemon=True,
|
||
).start()
|
||
try:
|
||
succeeded, value = result_queue.get(timeout=max(0.1, timeout_seconds))
|
||
except queue.Empty as error:
|
||
raise TimeoutError(f"{label} did not finish within {timeout_seconds:g}s") from error
|
||
if not succeeded:
|
||
raise value
|
||
return value
|
||
|
||
|
||
class MapService:
|
||
def __init__(self, config: ServiceConfig) -> None:
|
||
self.config = config
|
||
self.cache_dir = config.cache_dir.resolve()
|
||
self.tile_cache_dir = self.cache_dir / "tiles"
|
||
self.rail_cache_dir = self.cache_dir / "rail"
|
||
self.pending_rail_dir = self.cache_dir / "rail-pending"
|
||
self.water_cache_dir = self.cache_dir / "water"
|
||
self.pending_water_dir = self.cache_dir / "water-pending"
|
||
self.boundary_cache_dir = self.cache_dir / "boundaries"
|
||
self.vector_cache_dir = self.cache_dir / "vector"
|
||
self.local_vector_dir = self.cache_dir / "local-vector"
|
||
self.local_vector_25_dir = self.cache_dir / "local-vector-25"
|
||
self.local_vector_dynamic_dir = self.cache_dir / "local-vector-dynamic"
|
||
self.local_vector_25_dynamic_dir = self.cache_dir / "local-vector-25-dynamic"
|
||
self.rail_tile_dir = self.cache_dir / "rail-tiles"
|
||
self.rail_metro_dir = self.cache_dir / "rail-metro-tiles"
|
||
self.tile_cache_dir.mkdir(parents=True, exist_ok=True)
|
||
self.rail_cache_dir.mkdir(parents=True, exist_ok=True)
|
||
self.pending_rail_dir.mkdir(parents=True, exist_ok=True)
|
||
self.water_cache_dir.mkdir(parents=True, exist_ok=True)
|
||
self.pending_water_dir.mkdir(parents=True, exist_ok=True)
|
||
self.boundary_cache_dir.mkdir(parents=True, exist_ok=True)
|
||
self.vector_cache_dir.mkdir(parents=True, exist_ok=True)
|
||
self.local_vector_dir.mkdir(parents=True, exist_ok=True)
|
||
self.local_vector_25_dir.mkdir(parents=True, exist_ok=True)
|
||
self.local_vector_dynamic_dir.mkdir(parents=True, exist_ok=True)
|
||
self.local_vector_25_dynamic_dir.mkdir(parents=True, exist_ok=True)
|
||
self.rail_tile_dir.mkdir(parents=True, exist_ok=True)
|
||
self.rail_metro_dir.mkdir(parents=True, exist_ok=True)
|
||
self.started_at = time.time()
|
||
self.stats_lock = threading.Lock()
|
||
self.stats = {
|
||
"tile_hits": 0,
|
||
"tile_stale": 0,
|
||
"tile_misses": 0,
|
||
"tile_errors": 0,
|
||
"rail_hits": 0,
|
||
"rail_stale": 0,
|
||
"rail_misses": 0,
|
||
"rail_errors": 0,
|
||
"rail_pending": 0,
|
||
"water_hits": 0,
|
||
"water_misses": 0,
|
||
"water_errors": 0,
|
||
"water_pending": 0,
|
||
"boundary_hits": 0,
|
||
"boundary_misses": 0,
|
||
"boundary_errors": 0,
|
||
"vector_hits": 0,
|
||
"vector_misses": 0,
|
||
"vector_errors": 0,
|
||
"local_vector_dynamic_hits": 0,
|
||
"local_vector_dynamic_generated": 0,
|
||
"local_vector_dynamic_errors": 0,
|
||
"rail_tile_hits": 0,
|
||
"rail_tiles_generated": 0,
|
||
"rail_tile_errors": 0,
|
||
"rail_metro_downloads": 0,
|
||
"rail_metro_errors": 0,
|
||
"manual_refreshes": 0,
|
||
}
|
||
self._lock_guard = threading.Lock()
|
||
self._tile_locks: dict[str, threading.Lock] = {}
|
||
self._refreshing: set[str] = set()
|
||
self._rail_aggregate_inflight: set[str] = set()
|
||
self._vector_template: str | None = None
|
||
self._vector_template_loaded_at = 0.0
|
||
self._vector_template_lock = threading.Lock()
|
||
self._upstream_proxy_lock = threading.Lock()
|
||
self._upstream_proxy_opener: Any = None
|
||
self._default_upstream_proxy = self.config.upstream_proxy
|
||
configure_atomic_staging(self.cache_dir)
|
||
self.set_upstream_proxy(self.config.upstream_proxy, persist=False)
|
||
self._cache_totals_lock = threading.Lock()
|
||
self._cache_totals_cached_at = 0.0
|
||
self._cache_totals_result = (0, 0)
|
||
self._cache_admin_stats_lock = threading.Lock()
|
||
self._cache_admin_stats_cached_at = 0.0
|
||
self._cache_admin_stats_result: dict[str, Any] | None = None
|
||
self._cache_coverage_lock = threading.Lock()
|
||
self._cache_coverage_result: dict[str, Any] | None = None
|
||
self._cache_coverage_cached_at = 0.0
|
||
self._refresh_executor = DaemonTaskPool(CACHE_REFRESH_WORKERS, "map-cache-refresh")
|
||
self._download_executor = DaemonTaskPool(VECTOR_BACKGROUND_WORKERS, "map-vector-download")
|
||
self._vector_tile_executor = DaemonTaskPool(VECTOR_TILE_REQUEST_WORKERS, "openfreemap-tile-fetch")
|
||
self._rail_aggregate_executor = DaemonTaskPool(RAIL_AGGREGATE_WORKERS, "rail-lowzoom-aggregate")
|
||
self._rail_aggregate_fetch_executor = DaemonTaskPool(RAIL_AGGREGATE_FETCH_WORKERS, "rail-lowzoom-tile-fetch")
|
||
self._tianditu_executor = DaemonTaskPool(TIANDITU_DOWNLOAD_WORKERS, "tianditu-manual-download")
|
||
self._openmap_executor = DaemonTaskPool(OPENMAP_DOWNLOAD_WORKERS, "openmap-manual-download")
|
||
self._tianditu_jobs: dict[str, dict[str, Any]] = {}
|
||
self._openmap_jobs: dict[str, dict[str, Any]] = {}
|
||
self._tianditu_cancel_events: dict[str, threading.Event] = {}
|
||
self._openmap_cancel_events: dict[str, threading.Event] = {}
|
||
self._admin_sessions: dict[str, float] = {}
|
||
self._rate_events: dict[str, list[float]] = {}
|
||
self._stop = threading.Event()
|
||
self._maintenance = threading.Thread(
|
||
target=self._maintenance_loop,
|
||
name="map-cache-maintenance",
|
||
daemon=True,
|
||
)
|
||
self._maintenance.start()
|
||
|
||
@staticmethod
|
||
def build_upstream_proxy_opener(proxy_url: str) -> Any:
|
||
if not proxy_url:
|
||
return None
|
||
return urllib.request.build_opener(
|
||
urllib.request.ProxyHandler({"http": proxy_url, "https": proxy_url})
|
||
)
|
||
|
||
def set_upstream_proxy(self, value: str, persist: bool = True) -> str:
|
||
normalized = normalize_upstream_proxy(value)
|
||
opener = self.build_upstream_proxy_opener(normalized)
|
||
self.config.upstream_proxy = normalized
|
||
with self._upstream_proxy_lock:
|
||
self._upstream_proxy_opener = opener
|
||
if persist:
|
||
self._default_upstream_proxy = normalized
|
||
atomic_write(
|
||
self.cache_dir / UPSTREAM_PROXY_SETTING_NAME,
|
||
(normalized + "\n").encode("utf-8"),
|
||
)
|
||
return normalized
|
||
|
||
def reset_upstream_proxy(self) -> str:
|
||
return self.set_upstream_proxy(self._default_upstream_proxy, persist=False)
|
||
|
||
def open_upstream_url(self, request: urllib.request.Request, timeout: float) -> Any:
|
||
with self._upstream_proxy_lock:
|
||
opener = self._upstream_proxy_opener
|
||
if opener is not None:
|
||
return opener.open(request, timeout=timeout)
|
||
return urllib.request.urlopen(request, timeout=timeout)
|
||
|
||
def upstream_proxy_status(self) -> dict[str, Any]:
|
||
current = display_upstream_proxy(self.config.upstream_proxy)
|
||
default = display_upstream_proxy(self._default_upstream_proxy)
|
||
return {
|
||
"configured": bool(self.config.upstream_proxy),
|
||
"proxy": current,
|
||
"defaultConfigured": bool(self._default_upstream_proxy),
|
||
"defaultProxy": default,
|
||
"temporary": self.config.upstream_proxy != self._default_upstream_proxy,
|
||
}
|
||
|
||
def set_cache_only(self, enabled: bool) -> bool:
|
||
enabled = bool(enabled)
|
||
self.config.cache_only = enabled
|
||
setting_path = self.cache_dir / CACHE_ONLY_SETTING_NAME
|
||
atomic_write(setting_path, b"1\n" if enabled else b"0\n")
|
||
if enabled:
|
||
now = time.time()
|
||
with self._lock_guard:
|
||
for events in (self._tianditu_cancel_events, self._openmap_cancel_events):
|
||
for event in events.values():
|
||
event.set()
|
||
events.clear()
|
||
for jobs in (self._tianditu_jobs, self._openmap_jobs):
|
||
for job in jobs.values():
|
||
if job.get("status") not in ("queued", "running"):
|
||
continue
|
||
remaining = max(
|
||
0,
|
||
int(job.get("total", 0)) - int(job.get("completed", 0)),
|
||
)
|
||
job["completed"] = int(job.get("total", 0))
|
||
job["failed"] = int(job.get("failed", 0)) + remaining
|
||
job["status"] = "partial"
|
||
job["finishedAt"] = now
|
||
job["updatedAt"] = now
|
||
if len(job.get("errors", [])) < 20:
|
||
job.setdefault("errors", []).append(
|
||
"服务器已切换为仅缓存模式，任务停止请求云端。"
|
||
)
|
||
return enabled
|
||
|
||
def increment(self, key: str, amount: int = 1) -> None:
|
||
with self.stats_lock:
|
||
self.stats[key] = self.stats.get(key, 0) + amount
|
||
|
||
def consume_rate_limit(
|
||
self,
|
||
bucket: str,
|
||
identity: str,
|
||
limit: int,
|
||
window_seconds: float,
|
||
) -> bool:
|
||
now = time.time()
|
||
key = f"{bucket}:{identity}"
|
||
with self._lock_guard:
|
||
timestamps = [
|
||
timestamp
|
||
for timestamp in self._rate_events.get(key, [])
|
||
if now - timestamp < window_seconds
|
||
]
|
||
if len(timestamps) >= limit:
|
||
self._rate_events[key] = timestamps
|
||
return False
|
||
timestamps.append(now)
|
||
self._rate_events[key] = timestamps
|
||
if len(self._rate_events) > 2048:
|
||
expired = [
|
||
event_key
|
||
for event_key, events in self._rate_events.items()
|
||
if not events or now - events[-1] >= 24 * 60 * 60
|
||
]
|
||
for event_key in expired:
|
||
self._rate_events.pop(event_key, None)
|
||
return True
|
||
|
||
def redact_secrets(self, value: Any) -> str:
|
||
text = str(value)
|
||
proxy = self.config.upstream_proxy
|
||
proxy_secrets: list[str] = []
|
||
if proxy:
|
||
parsed_proxy = urllib.parse.urlsplit(proxy)
|
||
proxy_secrets.extend(
|
||
secret for secret in (parsed_proxy.username, parsed_proxy.password) if secret
|
||
)
|
||
for secret in (self.config.tianditu_key, self.config.admin_password, proxy, *proxy_secrets):
|
||
if secret:
|
||
text = text.replace(secret, "***")
|
||
text = text.replace(urllib.parse.quote(secret, safe=""), "***")
|
||
return text
|
||
|
||
def local_vector_status(self, directory: Path | None = None) -> dict[str, Any]:
|
||
manifest_path = (directory or self.local_vector_dir) / "manifest.json"
|
||
if not manifest_path.exists():
|
||
return {
|
||
"ok": True,
|
||
"available": False,
|
||
"message": "Local vector tiles have not been built",
|
||
}
|
||
try:
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
except Exception as error:
|
||
return {"ok": False, "available": False, "error": str(error)}
|
||
manifest.pop("sources", None)
|
||
return {"ok": True, "available": True, **manifest}
|
||
|
||
def local_vector_tile(
|
||
self,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
directory: Path | None = None,
|
||
) -> dict[str, Any] | None:
|
||
if z < 0 or z > 22 or x < 0 or y < 0 or x >= 1 << z or y >= 1 << z:
|
||
return None
|
||
static_dir = (directory or self.local_vector_dir).resolve()
|
||
manifest_path = static_dir / "manifest.json"
|
||
try:
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
browse_max_zoom = int(manifest.get("browseMaxZoom") or manifest.get("maxZoom") or 0)
|
||
static_max_zoom = int(manifest.get("maxZoom") or 0)
|
||
static_zooms = sorted(
|
||
int(level)
|
||
for level in (manifest.get("tiles") or {})
|
||
if str(level).isdigit()
|
||
)
|
||
except Exception:
|
||
return None
|
||
if not static_zooms or z < min(static_zooms) or z > browse_max_zoom:
|
||
return None
|
||
static_path = static_dir / str(z) / str(x) / f"{y}.json"
|
||
if z in static_zooms:
|
||
if not static_path.exists():
|
||
return None
|
||
try:
|
||
return json.loads(static_path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return None
|
||
if z <= static_max_zoom:
|
||
return None
|
||
manifest_generated_at = str(manifest.get("generatedAt") or "")
|
||
dynamic_dir = (
|
||
self.local_vector_25_dynamic_dir
|
||
if static_dir == self.local_vector_25_dir.resolve()
|
||
else self.local_vector_dynamic_dir
|
||
)
|
||
dynamic_path = dynamic_dir / str(z) / str(x) / f"{y}.json"
|
||
if dynamic_path.exists():
|
||
try:
|
||
cached = json.loads(dynamic_path.read_text(encoding="utf-8"))
|
||
if cached.get("manifestGeneratedAt") == manifest_generated_at:
|
||
self.increment("local_vector_dynamic_hits")
|
||
return cached
|
||
except Exception:
|
||
pass
|
||
parent_zoom = max(level for level in static_zooms if level < z)
|
||
shift = z - parent_zoom
|
||
parent_x = x >> shift
|
||
parent_y = y >> shift
|
||
parent_path = static_dir / str(parent_zoom) / str(parent_x) / f"{parent_y}.json"
|
||
if not parent_path.exists():
|
||
return None
|
||
lock_key = f"local-vector-dynamic:{dynamic_dir}:{z}:{x}:{y}"
|
||
with self.tile_lock(lock_key):
|
||
if dynamic_path.exists():
|
||
try:
|
||
cached = json.loads(dynamic_path.read_text(encoding="utf-8"))
|
||
if cached.get("manifestGeneratedAt") == manifest_generated_at:
|
||
self.increment("local_vector_dynamic_hits")
|
||
return cached
|
||
except Exception:
|
||
pass
|
||
try:
|
||
parent_payload = json.loads(parent_path.read_text(encoding="utf-8"))
|
||
payload = clip_local_vector_payload(
|
||
parent_payload,
|
||
z,
|
||
x,
|
||
y,
|
||
manifest_generated_at,
|
||
)
|
||
dynamic_path.parent.mkdir(parents=True, exist_ok=True)
|
||
temporary = dynamic_path.with_suffix(".json.tmp")
|
||
temporary.write_text(
|
||
json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
|
||
encoding="utf-8",
|
||
)
|
||
os.replace(temporary, dynamic_path)
|
||
self.increment("local_vector_dynamic_generated")
|
||
return payload
|
||
except Exception:
|
||
self.increment("local_vector_dynamic_errors")
|
||
return None
|
||
|
||
def tile_lock(self, key: str) -> threading.Lock:
|
||
with self._lock_guard:
|
||
return self._tile_locks.setdefault(key, threading.Lock())
|
||
|
||
def tile_cache_path(self, provider: str, z: int, x: int, y: int) -> Path:
|
||
return self.tile_cache_dir / provider / str(z) / str(x) / f"{y}.png"
|
||
|
||
def tile_upstream_url(self, provider: str, z: int, x: int, y: int, url: str) -> str:
|
||
return url.format(
|
||
z=z,
|
||
x=x,
|
||
y=y,
|
||
key=urllib.parse.quote(self.config.tianditu_key, safe=""),
|
||
)
|
||
|
||
def fetch_tile_upstream(
|
||
self,
|
||
provider: str,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
upstream_timeout_seconds: float | None = None,
|
||
) -> tuple[bytes, str]:
|
||
provider_config = TILE_PROVIDERS[provider]
|
||
if self.config.cache_only:
|
||
raise CacheOnlyMode("Cloud tile requests are disabled by cache-only mode")
|
||
if provider_config.get("requires_key") and not self.config.tianditu_key:
|
||
raise TiandituKeyMissing("Tianditu API key is not configured")
|
||
errors: list[str] = []
|
||
for template in provider_config["urls"]:
|
||
url = self.tile_upstream_url(provider, z, x, y, template)
|
||
safe_url = self.redact_secrets(url)
|
||
request = urllib.request.Request(
|
||
url,
|
||
headers={
|
||
"User-Agent": USER_AGENT,
|
||
"Accept": "image/png,image/jpeg,image/webp,image/*;q=0.8",
|
||
"Referer": f"http://{self.config.host}:{self.config.port}/",
|
||
},
|
||
)
|
||
try:
|
||
request_timeout = max(
|
||
1.0,
|
||
float(
|
||
upstream_timeout_seconds
|
||
if upstream_timeout_seconds is not None
|
||
else provider_config.get("request_timeout_seconds", 20)
|
||
),
|
||
)
|
||
with self.open_upstream_url(request, request_timeout) as response:
|
||
data = response.read()
|
||
content_type = detect_image_content_type(
|
||
data, response.headers.get_content_type() or provider_config["content_type"]
|
||
)
|
||
if data and content_type.startswith("image/"):
|
||
return data, content_type
|
||
errors.append(f"{safe_url}: non-image response")
|
||
except Exception as error:
|
||
errors.append(f"{safe_url}: {self.redact_secrets(error)}")
|
||
raise RuntimeError("; ".join(errors) or "No tile upstream configured")
|
||
|
||
def get_tile(self, provider: str, z: int, x: int, y: int) -> tuple[bytes, str, str]:
|
||
if provider not in TILE_PROVIDERS:
|
||
raise KeyError(provider)
|
||
path = self.tile_cache_path(provider, z, x, y)
|
||
key = str(path)
|
||
lock = self.tile_lock(key)
|
||
with lock:
|
||
if path.exists():
|
||
age = time.time() - path.stat().st_mtime
|
||
data = path.read_bytes()
|
||
content_type = detect_image_content_type(
|
||
data, TILE_PROVIDERS[provider]["content_type"]
|
||
)
|
||
if age <= self.config.tile_ttl_seconds:
|
||
self.increment("tile_hits")
|
||
return data, content_type, "HIT"
|
||
self.increment("tile_stale")
|
||
if self.config.cache_only:
|
||
return data, content_type, "STALE"
|
||
if TILE_PROVIDERS[provider].get("manual_only"):
|
||
return data, content_type, "STALE"
|
||
self.schedule_tile_refresh(path)
|
||
return data, content_type, "STALE"
|
||
|
||
if self.config.cache_only:
|
||
self.increment("tile_offline_placeholders")
|
||
return PLACEHOLDER_TILE_PNG, "image/png", "PLACEHOLDER"
|
||
|
||
if TILE_PROVIDERS[provider].get("manual_only"):
|
||
raise TileNotCached(
|
||
f"Tile {provider}/{z}/{x}/{y} has not been downloaded"
|
||
)
|
||
|
||
try:
|
||
data, content_type = self.fetch_tile_upstream(provider, z, x, y)
|
||
atomic_write(path, data)
|
||
self.increment("tile_misses")
|
||
return data, content_type, "MISS"
|
||
except Exception:
|
||
self.increment("tile_errors")
|
||
raise
|
||
|
||
def refresh_tile_path(self, path: Path) -> bool:
|
||
if self.config.cache_only:
|
||
return False
|
||
try:
|
||
relative = path.relative_to(self.tile_cache_dir).parts
|
||
if len(relative) != 4:
|
||
return False
|
||
provider, z_text, x_text, y_text = relative
|
||
z = int(z_text)
|
||
x = int(x_text)
|
||
y = int(y_text[:-4])
|
||
data, _ = self.fetch_tile_upstream(provider, z, x, y)
|
||
atomic_write(path, data)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
finally:
|
||
with self._lock_guard:
|
||
self._refreshing.discard(str(path))
|
||
|
||
def schedule_tile_refresh(self, path: Path) -> None:
|
||
key = str(path)
|
||
with self._lock_guard:
|
||
if key in self._refreshing:
|
||
return
|
||
self._refreshing.add(key)
|
||
self._download_executor.submit(self.refresh_tile_path, path)
|
||
|
||
def tianditu_provider_names(self) -> list[str]:
|
||
return [
|
||
name for name, config in TILE_PROVIDERS.items()
|
||
if config.get("manual_only")
|
||
]
|
||
|
||
def tianditu_status(self) -> dict[str, Any]:
|
||
providers = []
|
||
for name in self.tianditu_provider_names():
|
||
config = TILE_PROVIDERS[name]
|
||
files = len(list((self.tile_cache_dir / name).rglob("*.png")))
|
||
providers.append({
|
||
"id": name,
|
||
"label": config.get("label", name),
|
||
"layer": config.get("tianditu_layer", ""),
|
||
"cachedTiles": files,
|
||
})
|
||
proxy_status = self.upstream_proxy_status()
|
||
return {
|
||
"ok": True,
|
||
"configured": bool(self.config.tianditu_key),
|
||
"cacheOnly": self.config.cache_only,
|
||
"upstreamProxy": proxy_status["proxy"],
|
||
"upstreamProxyConfigured": proxy_status["configured"],
|
||
"upstreamProxyDefault": proxy_status["defaultProxy"],
|
||
"upstreamProxyDefaultConfigured": proxy_status["defaultConfigured"],
|
||
"upstreamProxyTemporary": proxy_status["temporary"],
|
||
"providers": providers,
|
||
"downloadWorkers": TIANDITU_DOWNLOAD_WORKERS,
|
||
"activeJobs": sum(
|
||
1 for job in self._tianditu_jobs.values()
|
||
if job.get("status") in ("queued", "running")
|
||
),
|
||
}
|
||
|
||
def create_admin_session(self) -> str:
|
||
token = secrets.token_urlsafe(32)
|
||
now = time.time()
|
||
with self._lock_guard:
|
||
self._admin_sessions = {
|
||
key: expires for key, expires in self._admin_sessions.items()
|
||
if expires > now
|
||
}
|
||
self._admin_sessions[token] = now + 8 * 60 * 60
|
||
return token
|
||
|
||
def admin_session_valid(self, token: str) -> bool:
|
||
if not token:
|
||
return False
|
||
now = time.time()
|
||
with self._lock_guard:
|
||
expires = self._admin_sessions.get(token, 0)
|
||
if expires <= now:
|
||
self._admin_sessions.pop(token, None)
|
||
return False
|
||
return True
|
||
|
||
def destroy_admin_session(self, token: str) -> None:
|
||
if not token:
|
||
return
|
||
with self._lock_guard:
|
||
self._admin_sessions.pop(token, None)
|
||
|
||
def destroy_all_admin_sessions(self) -> None:
|
||
with self._lock_guard:
|
||
self._admin_sessions.clear()
|
||
|
||
def start_tianditu_download(
|
||
self,
|
||
providers: list[str],
|
||
force: bool = False,
|
||
*,
|
||
bounds: tuple[float, float, float, float] | None = None,
|
||
zoom_min: int | None = None,
|
||
zoom_max: int | None = None,
|
||
z: int | None = None,
|
||
x0: int | None = None,
|
||
x1: int | None = None,
|
||
y0: int | None = None,
|
||
y1: int | None = None,
|
||
) -> dict[str, Any]:
|
||
if self.config.cache_only:
|
||
raise ValueError("服务器处于仅缓存模式，已禁止提交云端下载任务")
|
||
if not self.config.tianditu_key:
|
||
raise TiandituKeyMissing("Tianditu API key is not configured")
|
||
allowed = set(self.tianditu_provider_names())
|
||
selected = list(dict.fromkeys(provider for provider in providers if provider in allowed))
|
||
if not selected:
|
||
raise ValueError("No Tianditu providers selected")
|
||
tile_ranges: list[tuple[int, int, int, int, int]] = []
|
||
if bounds is not None:
|
||
south, west, north, east = bounds
|
||
if not (-90 <= south < north <= 90 and -180 <= west < east <= 180):
|
||
raise ValueError("Invalid geographic bounds")
|
||
if zoom_min is None or zoom_max is None:
|
||
raise ValueError("Zoom range is required")
|
||
if not (1 <= zoom_min <= zoom_max <= 18):
|
||
raise ValueError("Zoom range must be between 1 and 18")
|
||
for tile_z in range(zoom_min, zoom_max + 1):
|
||
tile_x0 = self.lon_to_tile_x(west, tile_z)
|
||
tile_x1 = self.lon_to_tile_x(east, tile_z)
|
||
tile_y0 = self.lat_to_tile_y(north, tile_z)
|
||
tile_y1 = self.lat_to_tile_y(south, tile_z)
|
||
tile_ranges.append((tile_z, tile_x0, tile_x1, tile_y0, tile_y1))
|
||
else:
|
||
if z is None or x0 is None or x1 is None or y0 is None or y1 is None:
|
||
raise ValueError("Tile range is required")
|
||
if not (1 <= z <= 18):
|
||
raise ValueError("Zoom must be between 1 and 18")
|
||
count = 1 << z
|
||
if not (0 <= x0 <= x1 < count and 0 <= y0 <= y1 < count):
|
||
raise ValueError("Invalid tile range")
|
||
tile_ranges = [(z, x0, x1, y0, y1)]
|
||
total_per_provider = sum(
|
||
(tile_x1 - tile_x0 + 1) * (tile_y1 - tile_y0 + 1)
|
||
for _, tile_x0, tile_x1, tile_y0, tile_y1 in tile_ranges
|
||
)
|
||
total = len(selected) * total_per_provider
|
||
job_id = uuid.uuid4().hex
|
||
cancel_event = threading.Event()
|
||
job = {
|
||
"id": job_id,
|
||
"status": "queued",
|
||
"providers": selected,
|
||
"bounds": bounds,
|
||
"zoomMin": zoom_min if bounds is not None else z,
|
||
"zoomMax": zoom_max if bounds is not None else z,
|
||
"force": force,
|
||
"total": total,
|
||
"completed": 0,
|
||
"succeeded": 0,
|
||
"skipped": 0,
|
||
"failed": 0,
|
||
"downloadWorkers": TIANDITU_DOWNLOAD_WORKERS,
|
||
"startedAt": time.time(),
|
||
"updatedAt": time.time(),
|
||
"errors": [],
|
||
}
|
||
with self._lock_guard:
|
||
self._tianditu_jobs[job_id] = job
|
||
self._tianditu_cancel_events[job_id] = cancel_event
|
||
|
||
def download_tile(provider: str, tile_z: int, x: int, y: int) -> None:
|
||
if cancel_event.is_set():
|
||
return
|
||
path = self.tile_cache_path(provider, tile_z, x, y)
|
||
lock = self.tile_lock(f"tianditu:{provider}:{tile_z}:{x}:{y}")
|
||
with lock:
|
||
if cancel_event.is_set():
|
||
return
|
||
try:
|
||
if path.exists() and not force:
|
||
with self._lock_guard:
|
||
job["completed"] += 1
|
||
job["skipped"] += 1
|
||
job["updatedAt"] = time.time()
|
||
return
|
||
data, _ = self.fetch_tile_upstream(provider, tile_z, x, y)
|
||
if cancel_event.is_set():
|
||
return
|
||
atomic_write(path, data)
|
||
with self._lock_guard:
|
||
job["completed"] += 1
|
||
job["succeeded"] += 1
|
||
job["updatedAt"] = time.time()
|
||
except Exception as error:
|
||
if cancel_event.is_set():
|
||
return
|
||
with self._lock_guard:
|
||
job["completed"] += 1
|
||
job["failed"] += 1
|
||
job["updatedAt"] = time.time()
|
||
if len(job["errors"]) < 20:
|
||
job["errors"].append(
|
||
self.redact_secrets(
|
||
f"{provider}/{tile_z}/{x}/{y}: {error}"
|
||
)
|
||
)
|
||
finally:
|
||
with self._lock_guard:
|
||
if job["status"] == "running" and job["completed"] >= job["total"]:
|
||
job["status"] = "completed" if not job["failed"] else "partial"
|
||
job["finishedAt"] = time.time()
|
||
|
||
with self._lock_guard:
|
||
job["status"] = "running"
|
||
tile_iterator = (
|
||
(provider, tile_z, tile_x, tile_y)
|
||
for provider in selected
|
||
for tile_z, tile_x0, tile_x1, tile_y0, tile_y1 in tile_ranges
|
||
for tile_y in range(tile_y0, tile_y1 + 1)
|
||
for tile_x in range(tile_x0, tile_x1 + 1)
|
||
)
|
||
tile_dispatch_lock = threading.Lock()
|
||
|
||
def download_worker() -> None:
|
||
while not cancel_event.is_set():
|
||
with tile_dispatch_lock:
|
||
try:
|
||
provider, tile_z, tile_x, tile_y = next(tile_iterator)
|
||
except StopIteration:
|
||
return
|
||
download_tile(provider, tile_z, tile_x, tile_y)
|
||
|
||
for _ in range(TIANDITU_DOWNLOAD_WORKERS):
|
||
self._tianditu_executor.submit(download_worker)
|
||
return dict(job)
|
||
|
||
def tianditu_job_status(self, job_id: str) -> dict[str, Any] | None:
|
||
with self._lock_guard:
|
||
job = self._tianditu_jobs.get(job_id)
|
||
return dict(job) if job else None
|
||
|
||
def cancel_tianditu_job(self, job_id: str) -> dict[str, Any] | None:
|
||
with self._lock_guard:
|
||
job = self._tianditu_jobs.get(job_id)
|
||
if not job:
|
||
return None
|
||
if job.get("status") in ("queued", "running"):
|
||
event = self._tianditu_cancel_events.pop(job_id, None)
|
||
if event:
|
||
event.set()
|
||
job["status"] = "cancelled"
|
||
job["cancelledAt"] = time.time()
|
||
job["finishedAt"] = job["cancelledAt"]
|
||
job["updatedAt"] = job["cancelledAt"]
|
||
if len(job["errors"]) < 20:
|
||
job["errors"].append("下载任务已由管理员中断；已完成的完整瓦片会保留。")
|
||
return dict(job)
|
||
|
||
def openmap_status(self) -> dict[str, Any]:
|
||
providers = []
|
||
vector_count: int | None = None
|
||
for name, config in OPEN_MAP_PROVIDERS.items():
|
||
if config["kind"] == "vector":
|
||
if vector_count is None:
|
||
vector_count = sum(1 for _ in self.vector_cache_dir.rglob("*.pbf"))
|
||
cached_tiles = vector_count
|
||
else:
|
||
cached_tiles = sum(
|
||
1 for _ in (self.tile_cache_dir / str(config["provider"])).rglob("*.png")
|
||
)
|
||
providers.append({
|
||
"id": name,
|
||
"label": config["label"],
|
||
"kind": config["kind"],
|
||
"minZoom": config["minZoom"],
|
||
"maxZoom": config["maxZoom"],
|
||
"source": config["source"],
|
||
"license": config["license"],
|
||
"description": config["description"],
|
||
"cachedTiles": cached_tiles,
|
||
})
|
||
with self._lock_guard:
|
||
active_jobs = sorted(
|
||
(
|
||
dict(job)
|
||
for job in self._openmap_jobs.values()
|
||
if job.get("status") in ("queued", "running")
|
||
),
|
||
key=lambda job: float(job.get("startedAt", 0)),
|
||
reverse=True,
|
||
)
|
||
proxy_status = self.upstream_proxy_status()
|
||
return {
|
||
"ok": True,
|
||
"cacheOnly": self.config.cache_only,
|
||
"upstreamProxy": proxy_status["proxy"],
|
||
"upstreamProxyConfigured": proxy_status["configured"],
|
||
"upstreamProxyDefault": proxy_status["defaultProxy"],
|
||
"upstreamProxyDefaultConfigured": proxy_status["defaultConfigured"],
|
||
"upstreamProxyTemporary": proxy_status["temporary"],
|
||
"providers": providers,
|
||
"downloadWorkers": OPENMAP_DOWNLOAD_WORKERS,
|
||
"activeJobs": len(active_jobs),
|
||
"activeJob": active_jobs[0] if active_jobs else None,
|
||
}
|
||
|
||
def start_openmap_download(
|
||
self,
|
||
providers: list[str],
|
||
force: bool = False,
|
||
*,
|
||
bounds: tuple[float, float, float, float] | None = None,
|
||
zoom_min: int | None = None,
|
||
zoom_max: int | None = None,
|
||
z: int | None = None,
|
||
x0: int | None = None,
|
||
x1: int | None = None,
|
||
y0: int | None = None,
|
||
y1: int | None = None,
|
||
tiles: list[tuple[int, int, int]] | None = None,
|
||
) -> dict[str, Any]:
|
||
if self.config.cache_only:
|
||
raise ValueError("服务器处于仅缓存模式，已禁止提交云端下载任务")
|
||
selected = list(dict.fromkeys(providers))
|
||
unknown = [provider for provider in selected if provider not in OPEN_MAP_PROVIDERS]
|
||
if unknown:
|
||
raise ValueError("Unknown open map provider: " + ", ".join(unknown))
|
||
if not selected:
|
||
raise ValueError("No open map providers selected")
|
||
|
||
exact_tiles: list[tuple[int, int, int]] | None = None
|
||
tile_ranges: list[tuple[int, int, int, int, int]] = []
|
||
if tiles is not None:
|
||
if len(selected) != 1 or OPEN_MAP_PROVIDERS[selected[0]]["kind"] != "vector":
|
||
raise ValueError("Exact tile downloads require one vector provider")
|
||
config = OPEN_MAP_PROVIDERS[selected[0]]
|
||
exact_tiles = sorted(set(
|
||
(int(tile_z), int(tile_x), int(tile_y))
|
||
for tile_z, tile_x, tile_y in tiles
|
||
))
|
||
if not exact_tiles:
|
||
raise ValueError("No source tiles selected")
|
||
for tile_z, tile_x, tile_y in exact_tiles:
|
||
count = 1 << tile_z
|
||
if not (config["minZoom"] <= tile_z <= config["maxZoom"]):
|
||
raise ValueError(
|
||
f"{config['label']} supports Z{config['minZoom']}-Z{config['maxZoom']}"
|
||
)
|
||
if not (0 <= tile_x < count and 0 <= tile_y < count):
|
||
raise ValueError("Invalid source tile coordinate")
|
||
elif bounds is not None:
|
||
south, west, north, east = bounds
|
||
if not (-90 <= south < north <= 90 and -180 <= west < east <= 180):
|
||
raise ValueError("Invalid geographic bounds")
|
||
if zoom_min is None or zoom_max is None:
|
||
raise ValueError("Zoom range is required")
|
||
if zoom_min > zoom_max:
|
||
raise ValueError("Invalid zoom range")
|
||
for provider in selected:
|
||
config = OPEN_MAP_PROVIDERS[provider]
|
||
if zoom_min < config["minZoom"] or zoom_max > config["maxZoom"]:
|
||
raise ValueError(
|
||
f"{config['label']} supports Z{config['minZoom']}-Z{config['maxZoom']}"
|
||
)
|
||
for tile_z in range(zoom_min, zoom_max + 1):
|
||
tile_x0 = self.lon_to_tile_x(west, tile_z)
|
||
tile_x1 = self.lon_to_tile_x(east, tile_z)
|
||
tile_y0 = self.lat_to_tile_y(north, tile_z)
|
||
tile_y1 = self.lat_to_tile_y(south, tile_z)
|
||
tile_ranges.append((tile_z, tile_x0, tile_x1, tile_y0, tile_y1))
|
||
else:
|
||
if z is None or x0 is None or x1 is None or y0 is None or y1 is None:
|
||
raise ValueError("Tile range is required")
|
||
for provider in selected:
|
||
config = OPEN_MAP_PROVIDERS[provider]
|
||
if not (config["minZoom"] <= z <= config["maxZoom"]):
|
||
raise ValueError(
|
||
f"{config['label']} supports Z{config['minZoom']}-Z{config['maxZoom']}"
|
||
)
|
||
count = 1 << z
|
||
if not (0 <= x0 <= x1 < count and 0 <= y0 <= y1 < count):
|
||
raise ValueError("Invalid tile range")
|
||
tile_ranges = [(z, x0, x1, y0, y1)]
|
||
|
||
total_per_provider = sum(
|
||
(tile_x1 - tile_x0 + 1) * (tile_y1 - tile_y0 + 1)
|
||
for _, tile_x0, tile_x1, tile_y0, tile_y1 in tile_ranges
|
||
)
|
||
total = len(exact_tiles) if exact_tiles is not None else len(selected) * total_per_provider
|
||
job_id = uuid.uuid4().hex
|
||
cancel_event = threading.Event()
|
||
job = {
|
||
"id": job_id,
|
||
"status": "queued",
|
||
"providers": selected,
|
||
"bounds": bounds,
|
||
"zoomMin": min(tile_z for tile_z, _, _ in exact_tiles) if exact_tiles is not None else (zoom_min if bounds is not None else z),
|
||
"zoomMax": max(tile_z for tile_z, _, _ in exact_tiles) if exact_tiles is not None else (zoom_max if bounds is not None else z),
|
||
"force": force,
|
||
"total": total,
|
||
"completed": 0,
|
||
"succeeded": 0,
|
||
"skipped": 0,
|
||
"failed": 0,
|
||
"downloadWorkers": OPENMAP_DOWNLOAD_WORKERS,
|
||
"activeRequests": 0,
|
||
"activeTiles": [],
|
||
"startedAt": time.time(),
|
||
"updatedAt": time.time(),
|
||
"errors": [],
|
||
}
|
||
with self._lock_guard:
|
||
self._openmap_jobs[job_id] = job
|
||
self._openmap_cancel_events[job_id] = cancel_event
|
||
|
||
def record_job_result(kind: str, message: str = "") -> None:
|
||
with self._lock_guard:
|
||
if job["status"] not in ("queued", "running"):
|
||
return
|
||
if job["completed"] >= job["total"]:
|
||
return
|
||
job["completed"] += 1
|
||
job[kind] += 1
|
||
job["updatedAt"] = time.time()
|
||
if message and len(job["errors"]) < 20:
|
||
job["errors"].append(self.redact_secrets(message))
|
||
if job["completed"] >= job["total"]:
|
||
job["status"] = "completed" if not job["failed"] else "partial"
|
||
job["finishedAt"] = time.time()
|
||
|
||
def download_tile(provider: str, tile_z: int, x: int, y: int) -> None:
|
||
if cancel_event.is_set():
|
||
return
|
||
config = OPEN_MAP_PROVIDERS[provider]
|
||
tile_lock = self.tile_lock(f"openmap:{provider}:{tile_z}:{x}:{y}")
|
||
if not tile_lock.acquire(timeout=OPENMAP_TILE_LOCK_SECONDS):
|
||
if cancel_event.is_set():
|
||
return
|
||
record_job_result(
|
||
"failed",
|
||
f"{provider}/{tile_z}/{x}/{y}: waiting for tile lock timed out",
|
||
)
|
||
return
|
||
try:
|
||
if cancel_event.is_set():
|
||
return
|
||
with self._lock_guard:
|
||
if job["status"] not in ("queued", "running"):
|
||
return
|
||
if config["kind"] == "vector":
|
||
path = self.vector_tile_cache_path(tile_z, x, y)
|
||
else:
|
||
path = self.tile_cache_path(config["provider"], tile_z, x, y)
|
||
if path.exists() and not force:
|
||
record_job_result("skipped")
|
||
return
|
||
active_label = f"{provider}/{tile_z}/{x}/{y}"
|
||
with self._lock_guard:
|
||
if job["status"] not in ("queued", "running"):
|
||
return
|
||
job["activeRequests"] += 1
|
||
job["activeTiles"].append(active_label)
|
||
try:
|
||
if config["kind"] == "vector":
|
||
run_with_timeout(
|
||
self.fetch_vector_tile,
|
||
OPENMAP_VECTOR_FETCH_SECONDS,
|
||
f"OpenFreeMap tile {tile_z}/{x}/{y}",
|
||
tile_z,
|
||
x,
|
||
y,
|
||
force=force,
|
||
cancel_event=cancel_event,
|
||
)
|
||
else:
|
||
timeout_seconds = (
|
||
OPENMAP_LONG_RASTER_FETCH_SECONDS
|
||
if config["provider"] == "opentopo"
|
||
else OPENMAP_RASTER_FETCH_SECONDS
|
||
)
|
||
data, _ = run_with_timeout(
|
||
self.fetch_tile_upstream,
|
||
timeout_seconds,
|
||
f"{provider} tile {tile_z}/{x}/{y}",
|
||
config["provider"],
|
||
tile_z,
|
||
x,
|
||
y,
|
||
upstream_timeout_seconds=OPENMAP_UPSTREAM_ATTEMPT_SECONDS,
|
||
)
|
||
if cancel_event.is_set():
|
||
return
|
||
atomic_write(path, data)
|
||
finally:
|
||
with self._lock_guard:
|
||
job["activeRequests"] = max(0, job["activeRequests"] - 1)
|
||
if active_label in job["activeTiles"]:
|
||
job["activeTiles"].remove(active_label)
|
||
if not cancel_event.is_set():
|
||
record_job_result("succeeded")
|
||
except Exception as error:
|
||
if cancel_event.is_set():
|
||
return
|
||
record_job_result(
|
||
"failed",
|
||
f"{provider}/{tile_z}/{x}/{y}: {error}",
|
||
)
|
||
finally:
|
||
tile_lock.release()
|
||
|
||
with self._lock_guard:
|
||
job["status"] = "running"
|
||
if exact_tiles is not None:
|
||
tile_iterator = (
|
||
(selected[0], tile_z, tile_x, tile_y)
|
||
for tile_z, tile_x, tile_y in exact_tiles
|
||
)
|
||
else:
|
||
tile_iterator = (
|
||
(provider, tile_z, tile_x, tile_y)
|
||
for provider in selected
|
||
for tile_z, tile_x0, tile_x1, tile_y0, tile_y1 in tile_ranges
|
||
for tile_y in range(tile_y0, tile_y1 + 1)
|
||
for tile_x in range(tile_x0, tile_x1 + 1)
|
||
)
|
||
tile_dispatch_lock = threading.Lock()
|
||
|
||
def download_worker() -> None:
|
||
while not cancel_event.is_set():
|
||
with tile_dispatch_lock:
|
||
try:
|
||
provider, tile_z, tile_x, tile_y = next(tile_iterator)
|
||
except StopIteration:
|
||
return
|
||
download_tile(provider, tile_z, tile_x, tile_y)
|
||
|
||
for _ in range(OPENMAP_DOWNLOAD_WORKERS):
|
||
self._openmap_executor.submit(download_worker)
|
||
return dict(job)
|
||
|
||
def start_openmap_topic_download(
|
||
self,
|
||
category_id: str,
|
||
force: bool = False,
|
||
) -> dict[str, Any]:
|
||
if category_id not in CACHE_ADMIN_QUERY_LAYERS:
|
||
raise ValueError("Unknown OpenFreeMap source topic")
|
||
tiles = self.coverage_tiles_for_category(category_id)
|
||
if not tiles:
|
||
raise ValueError("当前专题没有可下载的来源瓦片，请先浏览或查询对应区域")
|
||
job = self.start_openmap_download(
|
||
providers=["openfreemap"],
|
||
force=force,
|
||
tiles=tiles,
|
||
)
|
||
job["sourceCategory"] = category_id
|
||
job["sourceLabel"] = CACHE_ADMIN_QUERY_LAYERS[category_id]["label"]
|
||
with self._lock_guard:
|
||
stored = self._openmap_jobs.get(job["id"])
|
||
if stored is not None:
|
||
stored["sourceCategory"] = job["sourceCategory"]
|
||
stored["sourceLabel"] = job["sourceLabel"]
|
||
return job
|
||
|
||
def openmap_job_status(self, job_id: str) -> dict[str, Any] | None:
|
||
with self._lock_guard:
|
||
job = self._openmap_jobs.get(job_id)
|
||
return dict(job) if job else None
|
||
|
||
def cancel_openmap_job(self, job_id: str) -> dict[str, Any] | None:
|
||
with self._lock_guard:
|
||
job = self._openmap_jobs.get(job_id)
|
||
if not job:
|
||
return None
|
||
if job.get("status") in ("queued", "running"):
|
||
event = self._openmap_cancel_events.pop(job_id, None)
|
||
if event:
|
||
event.set()
|
||
job["status"] = "cancelled"
|
||
job["cancelledAt"] = time.time()
|
||
job["finishedAt"] = job["cancelledAt"]
|
||
job["updatedAt"] = job["cancelledAt"]
|
||
if len(job["errors"]) < 20:
|
||
job["errors"].append("下载任务已由管理员中断；已完成的完整瓦片会保留。")
|
||
return dict(job)
|
||
|
||
def expire_stalled_openmap_jobs(
|
||
self,
|
||
stall_seconds: float = OPENMAP_JOB_STALL_SECONDS,
|
||
) -> None:
|
||
now = time.time()
|
||
with self._lock_guard:
|
||
for job in self._openmap_jobs.values():
|
||
if job.get("status") not in ("queued", "running"):
|
||
continue
|
||
idle_seconds = now - float(job.get("updatedAt", now))
|
||
if idle_seconds < stall_seconds:
|
||
continue
|
||
remaining = max(0, int(job.get("total", 0)) - int(job.get("completed", 0)))
|
||
job["completed"] = int(job.get("total", 0))
|
||
job["failed"] = int(job.get("failed", 0)) + remaining
|
||
job["status"] = "partial"
|
||
job["timedOut"] = True
|
||
job["finishedAt"] = now
|
||
job["updatedAt"] = now
|
||
if len(job["errors"]) < 20:
|
||
job["errors"].append(
|
||
f"下载任务连续 {int(idle_seconds)} 秒没有进展，已结束等待；"
|
||
f"{remaining} 个瓦片记为失败，可稍后重试。"
|
||
)
|
||
|
||
def vector_tile_cache_path(self, z: int, x: int, y: int) -> Path:
|
||
return self.vector_cache_dir / str(z) / str(x) / f"{y}.pbf"
|
||
|
||
def openfreemap_tile_template(self) -> str:
|
||
with self._vector_template_lock:
|
||
if self._vector_template and time.time() - self._vector_template_loaded_at < 6 * 60 * 60:
|
||
return self._vector_template
|
||
if self.config.cache_only:
|
||
return OPENFREEMAP_FALLBACK_TILE_URL
|
||
try:
|
||
request = urllib.request.Request(
|
||
OPENFREEMAP_TILEJSON_URL,
|
||
headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
|
||
)
|
||
with self.open_upstream_url(request, 15) as response:
|
||
payload = json.load(response)
|
||
tiles = payload.get("tiles") or []
|
||
if isinstance(tiles, list) and tiles and isinstance(tiles[0], str):
|
||
self._vector_template = tiles[0]
|
||
self._vector_template_loaded_at = time.time()
|
||
return self._vector_template
|
||
except Exception:
|
||
pass
|
||
self._vector_template = self._vector_template or OPENFREEMAP_FALLBACK_TILE_URL
|
||
return self._vector_template
|
||
|
||
def fetch_vector_tile(
|
||
self,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
force: bool = False,
|
||
cancel_event: threading.Event | None = None,
|
||
) -> tuple[bytes, bool]:
|
||
path = self.vector_tile_cache_path(z, x, y)
|
||
if path.exists() and not force:
|
||
self.increment("vector_hits")
|
||
return path.read_bytes(), True
|
||
if self.config.cache_only:
|
||
raise CacheOnlyMode(f"Vector tile {z}/{x}/{y} is not cached")
|
||
url = self.openfreemap_tile_template().format(z=z, x=x, y=y)
|
||
last_error: Exception | None = None
|
||
for attempt in range(3):
|
||
if cancel_event and cancel_event.is_set():
|
||
raise RuntimeError("Vector tile download cancelled")
|
||
request = urllib.request.Request(
|
||
url,
|
||
headers={
|
||
"User-Agent": USER_AGENT,
|
||
"Accept": "application/x-protobuf,application/octet-stream,*/*",
|
||
},
|
||
)
|
||
try:
|
||
with self.open_upstream_url(request, 10) as response:
|
||
data = response.read()
|
||
if not data:
|
||
raise RuntimeError("Empty vector tile")
|
||
if cancel_event and cancel_event.is_set():
|
||
raise RuntimeError("Vector tile download cancelled")
|
||
atomic_write(path, data)
|
||
self.increment("vector_misses")
|
||
return data, False
|
||
except Exception as error:
|
||
last_error = error
|
||
if attempt < 2:
|
||
time.sleep(0.25 * (attempt + 1))
|
||
if path.exists():
|
||
return path.read_bytes(), True
|
||
self.increment("vector_errors")
|
||
raise last_error or RuntimeError("Vector tile unavailable")
|
||
|
||
@staticmethod
|
||
def lon_to_tile_x(lon: float, z: int) -> int:
|
||
count = 1 << z
|
||
return max(0, min(count - 1, int(math.floor((lon + 180.0) / 360.0 * count))))
|
||
|
||
@staticmethod
|
||
def lat_to_tile_y(lat: float, z: int) -> int:
|
||
count = 1 << z
|
||
clipped = max(-85.05112878, min(85.05112878, lat))
|
||
radians = math.radians(clipped)
|
||
return max(
|
||
0,
|
||
min(
|
||
count - 1,
|
||
int(math.floor((1.0 - math.asinh(math.tan(radians)) / math.pi) / 2.0 * count)),
|
||
),
|
||
)
|
||
|
||
def vector_tile_coordinates(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
z: int,
|
||
limit: int = MAX_VECTOR_TILES,
|
||
) -> list[tuple[int, int]]:
|
||
south, west, north, east = bbox
|
||
x0 = self.lon_to_tile_x(west, z)
|
||
x1 = self.lon_to_tile_x(east, z)
|
||
y0 = self.lat_to_tile_y(north, z)
|
||
y1 = self.lat_to_tile_y(south, z)
|
||
count = (x1 - x0 + 1) * (y1 - y0 + 1)
|
||
if count > limit:
|
||
raise ValueError(f"Vector tile request too large: {count} tiles")
|
||
return [(x, y) for x in range(x0, x1 + 1) for y in range(y0, y1 + 1)]
|
||
|
||
def choose_vector_zoom(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
preferred: int = 14,
|
||
max_tiles: int = MAX_VECTOR_TILES,
|
||
) -> int:
|
||
for z in range(max(4, min(14, preferred)), 3, -1):
|
||
south, west, north, east = bbox
|
||
x0 = self.lon_to_tile_x(west, z)
|
||
x1 = self.lon_to_tile_x(east, z)
|
||
y0 = self.lat_to_tile_y(north, z)
|
||
y1 = self.lat_to_tile_y(south, z)
|
||
if (x1 - x0 + 1) * (y1 - y0 + 1) <= max_tiles:
|
||
return z
|
||
return 4
|
||
|
||
def load_vector_layers(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
tile_zoom: int | None,
|
||
force: bool = False,
|
||
required_layers: set[str] | None = None,
|
||
minimum_source_zoom: int = OPENFREEMAP_SOURCE_MIN_ZOOM,
|
||
tile_executor: DaemonTaskPool | None = None,
|
||
) -> dict[str, Any]:
|
||
requested_zoom = max(
|
||
OPENFREEMAP_SOURCE_MIN_ZOOM,
|
||
min(ENVIRONMENT_MAX_ZOOM, int(tile_zoom or 12)),
|
||
)
|
||
source_zoom = min(
|
||
OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
max(minimum_source_zoom, requested_zoom),
|
||
)
|
||
coordinates = self.vector_tile_coordinates(bbox, source_zoom)
|
||
if len(coordinates) > MAX_VECTOR_TILES:
|
||
raise RuntimeError(f"Vector tile request too large: {len(coordinates)} tiles")
|
||
results: queue.Queue[tuple[int, int, bytes | None, str]] = queue.Queue()
|
||
|
||
def fetch_one(x: int, y: int) -> None:
|
||
try:
|
||
data, _ = self.fetch_vector_tile(source_zoom, x, y, force=force)
|
||
results.put((x, y, data, ""))
|
||
except Exception as error:
|
||
results.put((x, y, None, str(error)))
|
||
|
||
executor = tile_executor or self._vector_tile_executor
|
||
for x, y in coordinates:
|
||
executor.submit(fetch_one, x, y)
|
||
|
||
layers: dict[str, list[dict[str, Any]]] = {}
|
||
succeeded = 0
|
||
errors = 0
|
||
deadline_seconds = max(60, min(300, (len(coordinates) + 2) // 3 + 20))
|
||
deadline = time.time() + deadline_seconds
|
||
for _ in coordinates:
|
||
remaining = deadline - time.time()
|
||
if remaining <= 0:
|
||
break
|
||
try:
|
||
x, y, data, error = results.get(timeout=remaining)
|
||
except queue.Empty:
|
||
break
|
||
if data is None:
|
||
errors += 1
|
||
continue
|
||
succeeded += 1
|
||
try:
|
||
tile_layers = decode_vector_tile(data, required_layers)
|
||
except Exception:
|
||
errors += 1
|
||
continue
|
||
for layer_name, features in tile_layers.items():
|
||
for feature in features:
|
||
feature["tile"] = (source_zoom, x, y)
|
||
layers.setdefault(layer_name, []).append(feature)
|
||
if not succeeded:
|
||
raise RuntimeError("OpenFreeMap vector tiles unavailable")
|
||
failed_tiles = len(coordinates) - succeeded
|
||
return {
|
||
"tileZoom": requested_zoom,
|
||
"sourceTileZoom": source_zoom,
|
||
"tileCount": len(coordinates),
|
||
"loadedTiles": succeeded,
|
||
"failedTiles": max(errors, failed_tiles),
|
||
"layers": layers,
|
||
}
|
||
|
||
@staticmethod
|
||
def vector_point_to_lonlat(
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
point: tuple[int, int],
|
||
extent: int,
|
||
) -> tuple[float, float]:
|
||
count = 1 << z
|
||
world_x = (x + point[0] / extent) / count
|
||
world_y = (y + point[1] / extent) / count
|
||
lon = world_x * 360.0 - 180.0
|
||
lat = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * world_y))))
|
||
return lon, lat
|
||
|
||
def vector_rail_response(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
types: list[str],
|
||
tile_zoom: int | None,
|
||
force: bool = False,
|
||
) -> dict[str, Any]:
|
||
result = self.load_vector_layers(
|
||
bbox,
|
||
tile_zoom,
|
||
force=force,
|
||
required_layers={"transportation"},
|
||
)
|
||
elements: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for feature in result["layers"].get("transportation", []):
|
||
properties = feature["properties"]
|
||
feature_class = properties.get("class")
|
||
subclass = properties.get("subclass") or feature_class
|
||
service = properties.get("service")
|
||
if feature_class not in ("rail", "transit") or feature["type"] != 2:
|
||
continue
|
||
if service and service not in ("main", None):
|
||
continue
|
||
is_metro = feature_class == "transit" and subclass in (
|
||
"subway", "light_rail", "monorail", "tram", "funicular"
|
||
)
|
||
if is_metro and "subway" not in types:
|
||
continue
|
||
if not is_metro and "rail" not in types:
|
||
continue
|
||
z, x, y = feature["tile"]
|
||
extent = feature["extent"]
|
||
for path in feature["paths"]:
|
||
geometry = []
|
||
for point in path:
|
||
lon, lat = self.vector_point_to_lonlat(z, x, y, point, extent)
|
||
geometry.append({"lat": round(lat, 7), "lon": round(lon, 7)})
|
||
if len(geometry) < 2:
|
||
continue
|
||
fingerprint = hashlib.sha1(
|
||
json.dumps(geometry, separators=(",", ":")).encode("utf-8")
|
||
).hexdigest()
|
||
if fingerprint in seen:
|
||
continue
|
||
seen.add(fingerprint)
|
||
elements.append(
|
||
{
|
||
"type": "way",
|
||
"id": "ofm:" + fingerprint[:16],
|
||
"tags": {"railway": "subway" if is_metro else "rail"},
|
||
"geometry": geometry,
|
||
}
|
||
)
|
||
return {
|
||
"version": 0.6,
|
||
"generator": "OpenFreeMap vector tiles",
|
||
"source": "openfreemap",
|
||
"tileZoom": result["tileZoom"],
|
||
"sourceTileZoom": result["sourceTileZoom"],
|
||
"tileCount": result["tileCount"],
|
||
"loadedTiles": result["loadedTiles"],
|
||
"failedTiles": result["failedTiles"],
|
||
"elements": elements,
|
||
}
|
||
|
||
def low_zoom_rail_cache_path(
|
||
self,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
selected: set[str],
|
||
) -> Path:
|
||
suffix = "-".join(sorted(selected)) or "none"
|
||
return self.rail_metro_dir / str(z) / str(x) / f"{y}-{suffix}-v2.json"
|
||
|
||
def read_low_zoom_rail_cache(
|
||
self,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
selected: set[str],
|
||
) -> list[dict[str, Any]] | None:
|
||
path = self.low_zoom_rail_cache_path(z, x, y, selected)
|
||
if not path.exists():
|
||
return None
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
elements = payload.get("elements")
|
||
return elements if isinstance(elements, list) else None
|
||
except Exception:
|
||
return None
|
||
|
||
def schedule_low_zoom_rail_aggregation(
|
||
self,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
selected: set[str],
|
||
force: bool = False,
|
||
) -> None:
|
||
key = f"{z}:{x}:{y}:{','.join(sorted(selected))}"
|
||
with self._lock_guard:
|
||
if key in self._rail_aggregate_inflight:
|
||
return
|
||
self._rail_aggregate_inflight.add(key)
|
||
self._rail_aggregate_executor.submit(
|
||
self._build_low_zoom_rail_aggregation,
|
||
key,
|
||
z,
|
||
x,
|
||
y,
|
||
set(selected),
|
||
force,
|
||
)
|
||
|
||
def _build_low_zoom_rail_aggregation(
|
||
self,
|
||
key: str,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
selected: set[str],
|
||
force: bool,
|
||
) -> None:
|
||
try:
|
||
self.low_zoom_rail_elements(z, x, y, selected, force=force)
|
||
finally:
|
||
with self._lock_guard:
|
||
self._rail_aggregate_inflight.discard(key)
|
||
|
||
def low_zoom_rail_elements(
|
||
self,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
selected: set[str],
|
||
force: bool = False,
|
||
) -> tuple[list[dict[str, Any]], bool]:
|
||
if z < 9 or z >= OPENFREEMAP_SOURCE_MAX_ZOOM:
|
||
return [], False
|
||
path = self.low_zoom_rail_cache_path(z, x, y, selected)
|
||
if path.exists() and not force:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
return payload.get("elements", []), True
|
||
except Exception:
|
||
pass
|
||
bbox = vector_tile_bbox(z, x, y)
|
||
try:
|
||
result = self.load_vector_layers(
|
||
bbox,
|
||
OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
force=force,
|
||
required_layers={"transportation"},
|
||
minimum_source_zoom=OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
tile_executor=self._rail_aggregate_fetch_executor,
|
||
)
|
||
elements: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for index, feature in enumerate(result["layers"].get("transportation", [])):
|
||
properties = feature.get("properties") or {}
|
||
feature_class = properties.get("class")
|
||
subclass = properties.get("subclass") or properties.get("class")
|
||
if feature.get("type") != 2:
|
||
continue
|
||
if feature_class == "transit" and subclass in (
|
||
"subway",
|
||
"light_rail",
|
||
"monorail",
|
||
):
|
||
if "metro" not in selected:
|
||
continue
|
||
railway = str(subclass)
|
||
elif feature_class == "rail":
|
||
if "rail" not in selected:
|
||
continue
|
||
railway = "rail"
|
||
else:
|
||
continue
|
||
tags: dict[str, Any] = {"railway": railway}
|
||
for source_key in (
|
||
"name",
|
||
"ref",
|
||
"operator",
|
||
"electrified",
|
||
"service",
|
||
"usage",
|
||
"railway:traffic_mode",
|
||
"highspeed",
|
||
):
|
||
if properties.get(source_key):
|
||
tags[source_key] = properties[source_key]
|
||
z_source, x_source, y_source = feature["tile"]
|
||
extent = feature["extent"]
|
||
for path_index, line_path in enumerate(clipped_feature_paths(feature)):
|
||
geometry = []
|
||
for point in line_path:
|
||
lon, lat = self.vector_point_to_lonlat(
|
||
z_source, x_source, y_source, point, extent
|
||
)
|
||
geometry.append({"lat": round(lat, 7), "lon": round(lon, 7)})
|
||
if len(geometry) < 2:
|
||
continue
|
||
fingerprint = hashlib.sha1(
|
||
json.dumps(
|
||
[geometry, tags],
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
sort_keys=True,
|
||
).encode("utf-8")
|
||
).hexdigest()
|
||
if fingerprint in seen:
|
||
continue
|
||
seen.add(fingerprint)
|
||
elements.append(
|
||
{
|
||
"type": "way",
|
||
"id": f"ofm-metro:{fingerprint[:20]}",
|
||
"tags": tags,
|
||
"geometry": geometry,
|
||
}
|
||
)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
complete = result.get("failedTiles", 0) == 0
|
||
if complete:
|
||
atomic_write(
|
||
path,
|
||
json.dumps(
|
||
{
|
||
"createdAt": time.time(),
|
||
"sourceTileZoom": OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
"elements": elements,
|
||
},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
).encode("utf-8"),
|
||
)
|
||
self.increment("rail_metro_downloads")
|
||
return elements, result.get("failedTiles", 0) == 0
|
||
except Exception as error:
|
||
print(
|
||
f"Low-zoom metro aggregation {z}/{x}/{y} failed: "
|
||
f"{self.redact_secrets(error)}",
|
||
flush=True,
|
||
)
|
||
self.increment("rail_metro_errors")
|
||
return [], False
|
||
|
||
def rail_tile(
|
||
self,
|
||
z: int,
|
||
x: int,
|
||
y: int,
|
||
force: bool = False,
|
||
include: set[str] | None = None,
|
||
) -> dict[str, Any] | None:
|
||
if (
|
||
z < ENVIRONMENT_MIN_ZOOM
|
||
or z > ENVIRONMENT_MAX_ZOOM
|
||
or x < 0
|
||
or y < 0
|
||
or x >= 1 << z
|
||
or y >= 1 << z
|
||
):
|
||
return None
|
||
selected = set(include or ("rail", "metro"))
|
||
low_zoom_tile = bool(selected and 9 <= z < OPENFREEMAP_SOURCE_MAX_ZOOM)
|
||
cache_suffix = ("-".join(sorted(selected)) or "none") + "-v7"
|
||
tile_path = (
|
||
self.rail_tile_dir / str(z) / str(x) / f"{y}-{cache_suffix}.json"
|
||
)
|
||
if tile_path.exists() and not force:
|
||
try:
|
||
cached_payload = json.loads(tile_path.read_text(encoding="utf-8"))
|
||
if not low_zoom_tile or int(cached_payload.get("sourceTileZoom", 0)) >= OPENFREEMAP_SOURCE_MAX_ZOOM:
|
||
self.increment("rail_tile_hits")
|
||
return cached_payload
|
||
except Exception:
|
||
pass
|
||
source_zoom = min(z, OPENFREEMAP_SOURCE_MAX_ZOOM)
|
||
source_x = x >> (z - source_zoom)
|
||
source_y = y >> (z - source_zoom)
|
||
lock_key = f"rail-tile:{z}:{x}:{y}"
|
||
with self.tile_lock(lock_key):
|
||
if tile_path.exists() and not force:
|
||
try:
|
||
cached_payload = json.loads(tile_path.read_text(encoding="utf-8"))
|
||
if not low_zoom_tile or int(cached_payload.get("sourceTileZoom", 0)) >= OPENFREEMAP_SOURCE_MAX_ZOOM:
|
||
self.increment("rail_tile_hits")
|
||
return cached_payload
|
||
except Exception:
|
||
pass
|
||
try:
|
||
data, _ = self.fetch_vector_tile(source_zoom, source_x, source_y, force=force)
|
||
layers = decode_vector_tile(data, {"transportation"})
|
||
source_features: list[dict[str, Any]] = []
|
||
for index, feature in enumerate(layers.get("transportation", [])):
|
||
if feature.get("type") != 2:
|
||
continue
|
||
properties = feature.get("properties") or {}
|
||
feature_class = properties.get("class")
|
||
subclass = properties.get("subclass") or feature_class
|
||
if feature_class == "transit" and subclass in (
|
||
"subway",
|
||
"light_rail",
|
||
"monorail",
|
||
):
|
||
if "metro" not in selected:
|
||
continue
|
||
railway = str(subclass)
|
||
elif feature_class == "rail":
|
||
if "rail" not in selected:
|
||
continue
|
||
railway = "rail"
|
||
else:
|
||
continue
|
||
tags: dict[str, Any] = {"railway": railway}
|
||
for source_key in (
|
||
"name",
|
||
"ref",
|
||
"operator",
|
||
"electrified",
|
||
"service",
|
||
"usage",
|
||
):
|
||
if properties.get(source_key):
|
||
tags[source_key] = properties[source_key]
|
||
z_source, x_source, y_source = source_zoom, source_x, source_y
|
||
extent = feature["extent"]
|
||
for path_index, path in enumerate(clipped_feature_paths(feature)):
|
||
coordinates = [
|
||
self.vector_point_to_lonlat(
|
||
z_source,
|
||
x_source,
|
||
y_source,
|
||
point,
|
||
extent,
|
||
)
|
||
for point in path
|
||
]
|
||
if len(coordinates) >= 2:
|
||
source_features.append(
|
||
{
|
||
"type": "Feature",
|
||
"id": f"ofm-rail:{source_zoom}:{source_x}:{source_y}:{index}:{path_index}",
|
||
"properties": tags,
|
||
"geometry": {
|
||
"type": "LineString",
|
||
"coordinates": [[lon, lat] for lon, lat in coordinates],
|
||
},
|
||
}
|
||
)
|
||
clipped = (
|
||
{"features": source_features}
|
||
if source_zoom == z
|
||
else clip_local_vector_payload(
|
||
{"features": source_features},
|
||
z,
|
||
x,
|
||
y,
|
||
"openfreemap-rail-tile-v1",
|
||
)
|
||
)
|
||
elements: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for feature in clipped.get("features", []):
|
||
geometry = feature.get("geometry") or {}
|
||
geometry_type = geometry.get("type")
|
||
if geometry_type == "LineString":
|
||
paths = [geometry.get("coordinates", [])]
|
||
elif geometry_type == "MultiLineString":
|
||
paths = geometry.get("coordinates", [])
|
||
else:
|
||
continue
|
||
tags = feature.get("properties") or {}
|
||
for path in paths:
|
||
points = [
|
||
{"lat": round(float(point[1]), 7), "lon": round(float(point[0]), 7)}
|
||
for point in path
|
||
if len(point) >= 2
|
||
]
|
||
if len(points) < 2:
|
||
continue
|
||
fingerprint = hashlib.sha1(
|
||
json.dumps(
|
||
[points, tags],
|
||
separators=(",", ":"),
|
||
sort_keys=True,
|
||
).encode("utf-8")
|
||
).hexdigest()
|
||
if fingerprint in seen:
|
||
continue
|
||
seen.add(fingerprint)
|
||
elements.append(
|
||
{
|
||
"type": "way",
|
||
"id": "ofm-rail:" + fingerprint[:16],
|
||
"tags": tags,
|
||
"geometry": points,
|
||
}
|
||
)
|
||
aggregating = False
|
||
if low_zoom_tile:
|
||
detailed_elements = None if force else self.read_low_zoom_rail_cache(
|
||
z,
|
||
x,
|
||
y,
|
||
selected,
|
||
)
|
||
if detailed_elements is None or force:
|
||
self.schedule_low_zoom_rail_aggregation(
|
||
z,
|
||
x,
|
||
y,
|
||
selected,
|
||
force=force,
|
||
)
|
||
aggregating = detailed_elements is None
|
||
if detailed_elements is not None:
|
||
elements = detailed_elements
|
||
source_zoom = OPENFREEMAP_SOURCE_MAX_ZOOM
|
||
payload = {
|
||
"format": "rail-tile-v1",
|
||
"source": "openfreemap",
|
||
"tileZoom": z,
|
||
"sourceTileZoom": source_zoom,
|
||
"include": sorted(selected),
|
||
"x": x,
|
||
"y": y,
|
||
"aggregating": aggregating,
|
||
"elements": elements,
|
||
}
|
||
if not aggregating:
|
||
tile_path.parent.mkdir(parents=True, exist_ok=True)
|
||
atomic_write(
|
||
tile_path,
|
||
json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
self.increment("rail_tiles_generated")
|
||
return payload
|
||
except Exception as error:
|
||
print(
|
||
f"Rail tile {z}/{x}/{y} generation failed: "
|
||
f"{self.redact_secrets(error)}",
|
||
flush=True,
|
||
)
|
||
self.increment("rail_tile_errors")
|
||
return None
|
||
|
||
def vector_water_response(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
tile_zoom: int | None,
|
||
force: bool = False,
|
||
include: list[str] | None = None,
|
||
) -> dict[str, Any]:
|
||
selected = set(include or ("water", "ports", "vegetation"))
|
||
required_layers: set[str] = set()
|
||
if "water" in selected:
|
||
required_layers.update(("water", "waterway"))
|
||
if "ports" in selected:
|
||
required_layers.update(("poi", "landuse"))
|
||
if "vegetation" in selected:
|
||
required_layers.update(("landcover", "landuse"))
|
||
minimum_source_zoom = 7 if "vegetation" in selected else OPENFREEMAP_SOURCE_MIN_ZOOM
|
||
result = self.load_vector_layers(
|
||
bbox,
|
||
tile_zoom,
|
||
force=force,
|
||
required_layers=required_layers,
|
||
minimum_source_zoom=minimum_source_zoom,
|
||
)
|
||
elements: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for layer_name in (("water", "waterway") if "water" in selected else ()):
|
||
for feature in result["layers"].get(layer_name, []):
|
||
properties = feature["properties"]
|
||
water_class = properties.get("class") or "water"
|
||
if layer_name == "water" and water_class == "ocean":
|
||
continue
|
||
z, x, y = feature["tile"]
|
||
extent = feature["extent"]
|
||
for path in clipped_feature_paths(feature):
|
||
geometry = []
|
||
for point in path:
|
||
lon, lat = self.vector_point_to_lonlat(z, x, y, point, extent)
|
||
geometry.append({"lat": round(lat, 7), "lon": round(lon, 7)})
|
||
if len(geometry) < 2:
|
||
continue
|
||
tags = (
|
||
{"waterway": water_class, "name": properties.get("name")}
|
||
if layer_name == "waterway"
|
||
else {"natural": "water", "water": water_class, "name": properties.get("name")}
|
||
)
|
||
fingerprint = hashlib.sha1(
|
||
json.dumps([geometry, tags], separators=(",", ":"), sort_keys=True).encode("utf-8")
|
||
).hexdigest()
|
||
if fingerprint in seen:
|
||
continue
|
||
seen.add(fingerprint)
|
||
elements.append(
|
||
{
|
||
"type": "way",
|
||
"id": "ofm-water:" + fingerprint[:16],
|
||
"tags": tags,
|
||
"geometry": geometry,
|
||
}
|
||
)
|
||
|
||
def convert_features(layer_names, allowed_classes, prefix, tag_builder):
|
||
converted = []
|
||
seen_features: set[str] = set()
|
||
for layer_name in layer_names:
|
||
for feature in result["layers"].get(layer_name, []):
|
||
properties = feature["properties"]
|
||
feature_class = properties.get("class") or ""
|
||
subclass = properties.get("subclass") or feature_class
|
||
if feature_class not in allowed_classes and subclass not in allowed_classes:
|
||
continue
|
||
z, x, y = feature["tile"]
|
||
extent = feature["extent"]
|
||
for path in clipped_feature_paths(feature):
|
||
geometry = []
|
||
for point in path:
|
||
lon, lat = self.vector_point_to_lonlat(z, x, y, point, extent)
|
||
geometry.append({"lat": round(lat, 7), "lon": round(lon, 7)})
|
||
if not geometry:
|
||
continue
|
||
tags = tag_builder(properties, feature_class, subclass)
|
||
fingerprint = hashlib.sha1(
|
||
json.dumps([geometry, tags], separators=(",", ":"), sort_keys=True).encode("utf-8")
|
||
).hexdigest()
|
||
if fingerprint in seen_features:
|
||
continue
|
||
seen_features.add(fingerprint)
|
||
converted.append(
|
||
{
|
||
"type": "point" if feature["type"] == 1 else "way",
|
||
"id": prefix + fingerprint[:16],
|
||
"tags": tags,
|
||
"geometry": geometry,
|
||
}
|
||
)
|
||
return converted
|
||
|
||
ports = convert_features(
|
||
("poi", "landuse"),
|
||
{"harbor", "ferry_terminal", "harbour", "port", "dock", "marina"},
|
||
"ofm-port:",
|
||
lambda props, cls, sub: {
|
||
"port": sub if cls == "poi" else cls,
|
||
"name": props.get("name"),
|
||
},
|
||
) if "ports" in selected else []
|
||
vegetation = convert_features(
|
||
("landcover", "landuse"),
|
||
{
|
||
"wood", "forest", "grass", "scrub", "heath", "wetland",
|
||
"meadow", "orchard", "vineyard", "farmland", "allotments",
|
||
"plant_nursery", "recreation_ground",
|
||
},
|
||
"ofm-vegetation:",
|
||
lambda props, cls, sub: {
|
||
"vegetation": sub,
|
||
"class": cls,
|
||
"name": props.get("name"),
|
||
},
|
||
) if "vegetation" in selected else []
|
||
return {
|
||
"version": 0.6,
|
||
"generator": "OpenFreeMap vector tiles",
|
||
"source": "openfreemap",
|
||
"tileZoom": result["tileZoom"],
|
||
"sourceTileZoom": result["sourceTileZoom"],
|
||
"tileCount": result["tileCount"],
|
||
"loadedTiles": result["loadedTiles"],
|
||
"failedTiles": result["failedTiles"],
|
||
"elements": elements,
|
||
"ports": ports,
|
||
"vegetation": vegetation,
|
||
}
|
||
|
||
def vector_boundary_response(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
tile_zoom: int | None,
|
||
force: bool = False,
|
||
) -> dict[str, Any]:
|
||
result = self.load_vector_layers(
|
||
bbox,
|
||
tile_zoom,
|
||
force=force,
|
||
required_layers={"boundary"},
|
||
)
|
||
elements: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for feature in result["layers"].get("boundary", []):
|
||
if feature["type"] != 2:
|
||
continue
|
||
properties = feature["properties"]
|
||
if properties.get("maritime"):
|
||
continue
|
||
try:
|
||
admin_level = int(properties.get("admin_level"))
|
||
except (TypeError, ValueError):
|
||
admin_level = 6
|
||
if not 2 <= admin_level <= 10:
|
||
continue
|
||
tags: dict[str, Any] = {
|
||
"boundary": "administrative",
|
||
"admin_level": admin_level,
|
||
}
|
||
name = properties.get("name") or properties.get("name:zh")
|
||
if name:
|
||
tags["name"] = name
|
||
if properties.get("disputed"):
|
||
tags["disputed"] = True
|
||
z, x, y = feature["tile"]
|
||
extent = feature["extent"]
|
||
for path in feature["paths"]:
|
||
geometry = []
|
||
for point in path:
|
||
lon, lat = self.vector_point_to_lonlat(z, x, y, point, extent)
|
||
geometry.append({"lat": round(lat, 7), "lon": round(lon, 7)})
|
||
if len(geometry) < 2:
|
||
continue
|
||
fingerprint = hashlib.sha1(
|
||
json.dumps(geometry, separators=(",", ":")).encode("utf-8")
|
||
).hexdigest()
|
||
if fingerprint in seen:
|
||
continue
|
||
seen.add(fingerprint)
|
||
elements.append(
|
||
{
|
||
"type": "way",
|
||
"id": "ofm-boundary:" + fingerprint[:16],
|
||
"tags": tags,
|
||
"geometry": geometry,
|
||
}
|
||
)
|
||
return {
|
||
"version": 0.6,
|
||
"generator": "OpenFreeMap vector tiles",
|
||
"source": "openfreemap",
|
||
"tileZoom": result["tileZoom"],
|
||
"tileCount": result["tileCount"],
|
||
"loadedTiles": result["loadedTiles"],
|
||
"failedTiles": result["failedTiles"],
|
||
"elements": elements,
|
||
}
|
||
|
||
def boundary_cache_path(self, cache_key: str) -> Path:
|
||
return self.boundary_cache_dir / f"{cache_key}.json"
|
||
|
||
@staticmethod
|
||
def cache_only_vector_response(
|
||
kind: str,
|
||
tile_zoom: int,
|
||
include: list[str] | None = None,
|
||
) -> dict[str, Any]:
|
||
payload: dict[str, Any] = {
|
||
"ok": True,
|
||
"version": 0.6,
|
||
"generator": "Map Studio cache-only placeholder",
|
||
"source": "cache-only",
|
||
"cacheOnly": True,
|
||
"placeholder": True,
|
||
"tileZoom": tile_zoom,
|
||
"tileCount": 0,
|
||
"loadedTiles": 0,
|
||
"failedTiles": 0,
|
||
"elements": [],
|
||
}
|
||
if kind == "water":
|
||
selected = set(include or ("water", "ports", "vegetation"))
|
||
payload["ports"] = []
|
||
payload["vegetation"] = []
|
||
payload["include"] = sorted(selected)
|
||
return payload
|
||
|
||
def get_boundaries(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
force: bool,
|
||
tile_zoom: int | None = None,
|
||
) -> tuple[dict[str, Any], str]:
|
||
vector_zoom = max(4, min(14, int(tile_zoom or 12)))
|
||
cache_key = self.vector_bbox_cache_key(bbox, vector_zoom, ["admin-boundaries-v1"])
|
||
path = self.boundary_cache_path(cache_key)
|
||
cached_response: dict[str, Any] | None = None
|
||
if path.exists() and not force:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
cached_response = payload["response"]
|
||
if time.time() - path.stat().st_mtime <= self.config.rail_ttl_seconds:
|
||
self.increment("boundary_hits")
|
||
return cached_response, "HIT"
|
||
except Exception:
|
||
cached_response = None
|
||
if self.config.cache_only:
|
||
if cached_response is None and path.exists():
|
||
try:
|
||
cached_response = json.loads(
|
||
path.read_text(encoding="utf-8")
|
||
)["response"]
|
||
except Exception:
|
||
cached_response = None
|
||
if cached_response is not None:
|
||
return cached_response, "STALE"
|
||
missing_tiles = self.vector_cache_missing(
|
||
bbox,
|
||
vector_zoom,
|
||
force=False,
|
||
)
|
||
if missing_tiles == 0:
|
||
try:
|
||
response = self.vector_boundary_response(
|
||
bbox,
|
||
vector_zoom,
|
||
force=False,
|
||
)
|
||
if (
|
||
response.get("failedTiles") == 0
|
||
and response.get("loadedTiles") == response.get("tileCount")
|
||
):
|
||
return response, "HIT"
|
||
except Exception:
|
||
pass
|
||
return self.cache_only_vector_response("boundary", vector_zoom), "PLACEHOLDER"
|
||
try:
|
||
response = self.vector_boundary_response(bbox, vector_zoom, force=force)
|
||
complete = (
|
||
response.get("failedTiles") == 0
|
||
and response.get("loadedTiles") == response.get("tileCount")
|
||
)
|
||
if complete:
|
||
atomic_write(
|
||
path,
|
||
json.dumps(
|
||
{
|
||
"createdAt": time.time(),
|
||
"query": "OpenFreeMap administrative boundary vector tiles",
|
||
"response": response,
|
||
},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
).encode("utf-8"),
|
||
)
|
||
if cached_response is not None and not complete:
|
||
return cached_response, "STALE"
|
||
self.increment("boundary_misses")
|
||
return response, "MISS"
|
||
except Exception:
|
||
self.increment("boundary_errors")
|
||
if cached_response is not None:
|
||
return cached_response, "STALE"
|
||
raise
|
||
|
||
@staticmethod
|
||
def compact_vector_response(response: dict[str, Any], kind: str) -> dict[str, Any]:
|
||
if response.get("source") != "openfreemap":
|
||
return response
|
||
def compact_group(source_elements: list[dict[str, Any]], group_kind: str) -> list[dict[str, Any]]:
|
||
compact_elements: list[dict[str, Any]] = []
|
||
for element in source_elements:
|
||
geometry = element.get("geometry") or []
|
||
if not geometry:
|
||
continue
|
||
first = geometry[0]
|
||
previous_lat = int(round(float(first["lat"]) * 10_000_000))
|
||
previous_lon = int(round(float(first["lon"]) * 10_000_000))
|
||
values = [previous_lat, previous_lon]
|
||
for point in geometry[1:]:
|
||
current_lat = int(round(float(point["lat"]) * 10_000_000))
|
||
current_lon = int(round(float(point["lon"]) * 10_000_000))
|
||
values.extend((current_lat - previous_lat, current_lon - previous_lon))
|
||
previous_lat = current_lat
|
||
previous_lon = current_lon
|
||
tags = element.get("tags") or {}
|
||
item: dict[str, Any] = {"i": element.get("id"), "g": values}
|
||
if group_kind == "rail":
|
||
item["r"] = tags.get("railway") or tags.get("route") or "rail"
|
||
for source_key, target_key in (
|
||
("name", "n"),
|
||
("ref", "f"),
|
||
("operator", "o"),
|
||
("electrified", "e"),
|
||
):
|
||
if tags.get(source_key):
|
||
item[target_key] = tags[source_key]
|
||
elif group_kind == "boundary":
|
||
item["a"] = int(tags.get("admin_level") or 6)
|
||
if tags.get("name"):
|
||
item["n"] = tags["name"]
|
||
if tags.get("disputed"):
|
||
item["d"] = 1
|
||
elif group_kind == "port":
|
||
item["p"] = tags.get("port") or "port"
|
||
if tags.get("name"):
|
||
item["n"] = tags["name"]
|
||
elif group_kind == "vegetation":
|
||
item["v"] = tags.get("vegetation") or tags.get("class") or "vegetation"
|
||
if tags.get("name"):
|
||
item["n"] = tags["name"]
|
||
else:
|
||
item["k"] = "L" if tags.get("waterway") else "P"
|
||
item["w"] = tags.get("waterway") or tags.get("water") or "water"
|
||
if tags.get("name"):
|
||
item["n"] = tags["name"]
|
||
compact_elements.append(item)
|
||
return compact_elements
|
||
|
||
compact = dict(response)
|
||
compact["format"] = "delta-v1"
|
||
compact["elements"] = compact_group(response.get("elements", []), kind)
|
||
if kind == "water":
|
||
compact["ports"] = compact_group(response.get("ports", []), "port")
|
||
compact["vegetation"] = compact_group(response.get("vegetation", []), "vegetation")
|
||
return compact
|
||
|
||
def rail_cache_path(self, cache_key: str) -> Path:
|
||
return self.rail_cache_dir / f"{cache_key}.json"
|
||
|
||
def pending_rail_path(self, cache_key: str) -> Path:
|
||
return self.pending_rail_dir / f"{cache_key}.json"
|
||
|
||
def rail_cache_key(self, bbox: tuple[float, float, float, float], types: list[str]) -> str:
|
||
key_source = json.dumps(
|
||
{
|
||
"schema": 2,
|
||
"bbox": [round(value, 5) for value in bbox],
|
||
"types": sorted(types),
|
||
},
|
||
separators=(",", ":"),
|
||
)
|
||
return hashlib.sha1(key_source.encode("utf-8")).hexdigest()
|
||
|
||
def vector_bbox_cache_key(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
tile_zoom: int,
|
||
types: list[str] | None = None,
|
||
minimum_source_zoom: int = OPENFREEMAP_SOURCE_MIN_ZOOM,
|
||
) -> str:
|
||
source_zoom = min(
|
||
OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
max(minimum_source_zoom, OPENFREEMAP_SOURCE_MIN_ZOOM, tile_zoom),
|
||
)
|
||
coordinates = self.vector_tile_coordinates(bbox, source_zoom)
|
||
xs = [item[0] for item in coordinates]
|
||
ys = [item[1] for item in coordinates]
|
||
key_source = json.dumps(
|
||
{
|
||
"schema": 3,
|
||
"z": tile_zoom,
|
||
"sourceZ": source_zoom,
|
||
"x": [min(xs), max(xs)],
|
||
"y": [min(ys), max(ys)],
|
||
"types": sorted(types or []),
|
||
},
|
||
separators=(",", ":"),
|
||
)
|
||
return hashlib.sha1(key_source.encode("utf-8")).hexdigest()
|
||
|
||
def queue_pending_rail(
|
||
self,
|
||
cache_key: str,
|
||
bbox: tuple[float, float, float, float],
|
||
types: list[str],
|
||
query: str,
|
||
error: str,
|
||
job_kind: str = "overpass",
|
||
tile_zoom: int | None = None,
|
||
force_tiles: bool = False,
|
||
) -> None:
|
||
path = self.pending_rail_path(cache_key)
|
||
existing: dict[str, Any] = {}
|
||
if path.exists():
|
||
try:
|
||
existing = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
existing = {}
|
||
payload = {
|
||
"cacheKey": cache_key,
|
||
"bbox": bbox,
|
||
"types": types,
|
||
"query": query,
|
||
"jobKind": job_kind,
|
||
"tileZoom": tile_zoom,
|
||
"forceTiles": force_tiles,
|
||
"createdAt": existing.get("createdAt", time.time()),
|
||
"updatedAt": time.time(),
|
||
"nextAttempt": time.time() + 30,
|
||
"attempts": int(existing.get("attempts", 0)) + 1,
|
||
"lastError": error,
|
||
}
|
||
atomic_write(path, json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
|
||
self.increment("rail_pending")
|
||
|
||
def vector_cache_missing(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
tile_zoom: int,
|
||
force: bool = False,
|
||
minimum_source_zoom: int = OPENFREEMAP_SOURCE_MIN_ZOOM,
|
||
) -> int:
|
||
source_zoom = min(
|
||
OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
max(minimum_source_zoom, OPENFREEMAP_SOURCE_MIN_ZOOM, tile_zoom),
|
||
)
|
||
coordinates = self.vector_tile_coordinates(bbox, source_zoom)
|
||
if force:
|
||
return len(coordinates)
|
||
return sum(
|
||
1 for x, y in coordinates
|
||
if not self.vector_tile_cache_path(source_zoom, x, y).exists()
|
||
)
|
||
|
||
def complete_vector_rail_pending(self, path: Path) -> bool:
|
||
if self.config.cache_only:
|
||
return False
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
cache_key = str(pending["cacheKey"])
|
||
bbox = tuple(float(value) for value in pending["bbox"])
|
||
types = [str(value) for value in pending.get("types", ["rail", "subway"])]
|
||
tile_zoom = int(pending.get("tileZoom") or 12)
|
||
response = self.vector_rail_response(
|
||
bbox, # type: ignore[arg-type]
|
||
types,
|
||
tile_zoom,
|
||
force=bool(pending.get("forceTiles", False)),
|
||
)
|
||
if response.get("failedTiles") or response.get("loadedTiles") != response.get("tileCount"):
|
||
raise RuntimeError("OpenFreeMap rail tiles incomplete")
|
||
atomic_write(
|
||
self.rail_cache_path(cache_key),
|
||
json.dumps(
|
||
{"createdAt": time.time(), "query": "vector-or-overpass", "response": response},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
).encode("utf-8"),
|
||
)
|
||
path.unlink(missing_ok=True)
|
||
return True
|
||
except Exception as error:
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
pending["updatedAt"] = time.time()
|
||
pending["nextAttempt"] = time.time() + 10
|
||
pending["attempts"] = int(pending.get("attempts", 0)) + 1
|
||
pending["lastError"] = str(error)
|
||
atomic_write(
|
||
path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
def pending_rail_status(self, cache_key: str) -> dict[str, Any]:
|
||
cache_path = self.rail_cache_path(cache_key)
|
||
if cache_path.exists():
|
||
return {
|
||
"ok": True,
|
||
"ready": True,
|
||
"cacheKey": cache_key,
|
||
"createdAt": cache_path.stat().st_mtime,
|
||
}
|
||
pending_path = self.pending_rail_path(cache_key)
|
||
if pending_path.exists():
|
||
try:
|
||
pending = json.loads(pending_path.read_text(encoding="utf-8"))
|
||
return {
|
||
"ok": True,
|
||
"ready": False,
|
||
"cacheKey": cache_key,
|
||
"attempts": pending.get("attempts", 0),
|
||
"lastError": pending.get("lastError", ""),
|
||
"updatedAt": pending.get("updatedAt"),
|
||
}
|
||
except Exception:
|
||
pass
|
||
return {"ok": False, "ready": False, "cacheKey": cache_key, "error": "Unknown rail cache key"}
|
||
|
||
def retry_pending_rail_jobs(self, limit: int | None = None) -> int:
|
||
if self.config.cache_only:
|
||
return 0
|
||
processed = 0
|
||
max_items = limit or self.config.refresh_batch
|
||
now = time.time()
|
||
for path in self.pending_rail_dir.glob("*.json"):
|
||
if processed >= max_items:
|
||
break
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
if now < float(pending.get("nextAttempt", 0)):
|
||
continue
|
||
if pending.get("jobKind") != "vector":
|
||
bbox = tuple(float(value) for value in pending["bbox"])
|
||
pending["jobKind"] = "vector"
|
||
pending["tileZoom"] = self.choose_vector_zoom(bbox) # type: ignore[arg-type]
|
||
pending["forceTiles"] = False
|
||
pending["nextAttempt"] = now
|
||
atomic_write(
|
||
path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
self.complete_vector_rail_pending(path)
|
||
processed += 1
|
||
continue
|
||
cache_key = str(pending["cacheKey"])
|
||
query = str(pending["query"])
|
||
response = self.fetch_overpass(query)
|
||
if not response.get("elements"):
|
||
raise RuntimeError("Overpass returned no elements")
|
||
cache_payload = {
|
||
"createdAt": time.time(),
|
||
"query": query,
|
||
"response": response,
|
||
}
|
||
atomic_write(
|
||
self.rail_cache_path(cache_key),
|
||
json.dumps(cache_payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
path.unlink(missing_ok=True)
|
||
except Exception as error:
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
pending["updatedAt"] = time.time()
|
||
pending["nextAttempt"] = time.time() + 120
|
||
pending["attempts"] = int(pending.get("attempts", 0)) + 1
|
||
pending["lastError"] = str(error)
|
||
atomic_write(
|
||
path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
except Exception:
|
||
pass
|
||
processed += 1
|
||
return processed
|
||
|
||
def water_cache_path(self, cache_key: str) -> Path:
|
||
return self.water_cache_dir / f"{cache_key}.json"
|
||
|
||
def build_water_query(
|
||
self,
|
||
south: float,
|
||
west: float,
|
||
north: float,
|
||
east: float,
|
||
) -> str:
|
||
bbox = f"{south:.6f},{west:.6f},{north:.6f},{east:.6f}"
|
||
return (
|
||
"[out:json][timeout:60];\n(\n"
|
||
f' way["natural"="water"]({bbox});\n'
|
||
f' way["water"="reservoir"]({bbox});\n'
|
||
f' way["waterway"="riverbank"]({bbox});\n'
|
||
f' way["waterway"~"^(river|canal|stream|drain)$"]({bbox});\n'
|
||
f' relation["natural"="water"]({bbox});\n'
|
||
f' relation["water"="reservoir"]({bbox});\n'
|
||
f' relation["waterway"="riverbank"]({bbox});\n'
|
||
");\nout geom;"
|
||
)
|
||
|
||
def refresh_water_path(self, path: Path) -> bool:
|
||
if self.config.cache_only:
|
||
return False
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
query = payload["query"]
|
||
response = self.fetch_overpass(query)
|
||
atomic_write(
|
||
path,
|
||
json.dumps(
|
||
{
|
||
"createdAt": time.time(),
|
||
"query": query,
|
||
"response": response,
|
||
},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
).encode("utf-8"),
|
||
)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
finally:
|
||
with self._lock_guard:
|
||
self._refreshing.discard(str(path))
|
||
|
||
def queue_pending_water(
|
||
self,
|
||
cache_key: str,
|
||
bbox: tuple[float, float, float, float],
|
||
query: str,
|
||
error: str,
|
||
job_kind: str = "overpass",
|
||
tile_zoom: int | None = None,
|
||
force_tiles: bool = False,
|
||
include: list[str] | None = None,
|
||
) -> None:
|
||
path = self.pending_water_dir / f"{cache_key}.json"
|
||
existing: dict[str, Any] = {}
|
||
if path.exists():
|
||
try:
|
||
existing = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
existing = {}
|
||
pending = {
|
||
"cacheKey": cache_key,
|
||
"bbox": bbox,
|
||
"query": query,
|
||
"jobKind": job_kind,
|
||
"tileZoom": tile_zoom,
|
||
"forceTiles": force_tiles,
|
||
"include": include or ["water", "ports", "vegetation"],
|
||
"createdAt": existing.get("createdAt", time.time()),
|
||
"updatedAt": time.time(),
|
||
"nextAttempt": time.time() + 30,
|
||
"attempts": int(existing.get("attempts", 0)) + 1,
|
||
"lastError": error,
|
||
}
|
||
atomic_write(
|
||
path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
self.increment("water_pending")
|
||
|
||
def complete_vector_water_pending(self, path: Path) -> bool:
|
||
if self.config.cache_only:
|
||
return False
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
cache_key = str(pending["cacheKey"])
|
||
bbox = tuple(float(value) for value in pending["bbox"])
|
||
tile_zoom = int(pending.get("tileZoom") or 12)
|
||
response = self.vector_water_response(
|
||
bbox, # type: ignore[arg-type]
|
||
tile_zoom,
|
||
force=bool(pending.get("forceTiles", False)),
|
||
include=[str(value) for value in pending.get("include", ["water", "ports", "vegetation"])],
|
||
)
|
||
if response.get("failedTiles") or response.get("loadedTiles") != response.get("tileCount"):
|
||
raise RuntimeError("OpenFreeMap water tiles incomplete")
|
||
atomic_write(
|
||
self.water_cache_path(cache_key),
|
||
json.dumps(
|
||
{"createdAt": time.time(), "query": "vector-or-overpass", "response": response},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
).encode("utf-8"),
|
||
)
|
||
path.unlink(missing_ok=True)
|
||
return True
|
||
except Exception as error:
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
pending["updatedAt"] = time.time()
|
||
pending["nextAttempt"] = time.time() + 10
|
||
pending["attempts"] = int(pending.get("attempts", 0)) + 1
|
||
pending["lastError"] = str(error)
|
||
atomic_write(
|
||
path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
def get_water(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
force: bool,
|
||
tile_zoom: int | None = None,
|
||
include: list[str] | None = None,
|
||
) -> tuple[dict[str, Any], str]:
|
||
selected_include = [
|
||
value for value in (include or ["water", "ports", "vegetation"])
|
||
if value in ("water", "ports", "vegetation")
|
||
]
|
||
if not selected_include:
|
||
raise ValueError("No environment layers selected")
|
||
vector_zoom = max(
|
||
ENVIRONMENT_MIN_ZOOM,
|
||
min(ENVIRONMENT_MAX_ZOOM, int(tile_zoom or 12)),
|
||
)
|
||
minimum_source_zoom = 7 if "vegetation" in selected_include else OPENFREEMAP_SOURCE_MIN_ZOOM
|
||
cache_key = self.vector_bbox_cache_key(
|
||
bbox,
|
||
vector_zoom,
|
||
["environment-v6", *sorted(selected_include)],
|
||
minimum_source_zoom=minimum_source_zoom,
|
||
)
|
||
pending_path = self.pending_water_dir / f"{cache_key}.json"
|
||
path = self.water_cache_path(cache_key)
|
||
has_vector_result = False
|
||
result_is_stale = False
|
||
if path.exists() and not force:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
response = payload["response"]
|
||
if response.get("source") == "openfreemap":
|
||
result_is_stale = time.time() - path.stat().st_mtime > self.config.rail_ttl_seconds
|
||
if not result_is_stale:
|
||
has_vector_result = True
|
||
self.increment("water_hits")
|
||
return response, "HIT"
|
||
except Exception:
|
||
pass
|
||
if self.config.cache_only:
|
||
if path.exists():
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
return payload["response"], "STALE"
|
||
except Exception:
|
||
pass
|
||
missing_tiles = self.vector_cache_missing(
|
||
bbox,
|
||
vector_zoom,
|
||
force=False,
|
||
minimum_source_zoom=minimum_source_zoom,
|
||
)
|
||
if missing_tiles == 0:
|
||
try:
|
||
response = self.vector_water_response(
|
||
bbox,
|
||
vector_zoom,
|
||
force=False,
|
||
include=selected_include,
|
||
)
|
||
if (
|
||
response.get("failedTiles") == 0
|
||
and response.get("loadedTiles") == response.get("tileCount")
|
||
):
|
||
return response, "HIT"
|
||
except Exception:
|
||
pass
|
||
return self.cache_only_vector_response(
|
||
"water",
|
||
vector_zoom,
|
||
include=selected_include,
|
||
), "PLACEHOLDER"
|
||
missing_tiles = self.vector_cache_missing(
|
||
bbox,
|
||
vector_zoom,
|
||
force=force,
|
||
minimum_source_zoom=minimum_source_zoom,
|
||
)
|
||
needs_vector_result = force or result_is_stale or not has_vector_result
|
||
if missing_tiles or needs_vector_result:
|
||
with self.tile_lock("water-pending:" + cache_key):
|
||
existing_kind = ""
|
||
if pending_path.exists():
|
||
try:
|
||
existing_kind = str(
|
||
json.loads(pending_path.read_text(encoding="utf-8")).get("jobKind", "")
|
||
)
|
||
except Exception:
|
||
existing_kind = ""
|
||
if existing_kind != "vector":
|
||
self.queue_pending_water(
|
||
cache_key,
|
||
bbox,
|
||
"",
|
||
f"Queued {missing_tiles} OpenFreeMap vector tiles",
|
||
job_kind="vector",
|
||
tile_zoom=vector_zoom,
|
||
force_tiles=force or result_is_stale,
|
||
include=selected_include,
|
||
)
|
||
self._download_executor.submit(self.complete_vector_water_pending, pending_path)
|
||
raise WaterPending(cache_key)
|
||
south, west, north, east = bbox
|
||
if path.exists() and not force:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
age = time.time() - path.stat().st_mtime
|
||
if age <= self.config.rail_ttl_seconds:
|
||
self.increment("water_hits")
|
||
return payload["response"], "HIT"
|
||
key = str(path)
|
||
with self._lock_guard:
|
||
needs_refresh = key not in self._refreshing
|
||
if needs_refresh:
|
||
self._refreshing.add(key)
|
||
if needs_refresh:
|
||
self._refresh_executor.submit(self.refresh_water_path, path)
|
||
return payload["response"], "STALE"
|
||
except Exception:
|
||
pass
|
||
|
||
if pending_path.exists() and not force:
|
||
raise WaterPending(cache_key)
|
||
|
||
query = self.build_water_query(south, west, north, east)
|
||
try:
|
||
response = self.fetch_overpass(query)
|
||
atomic_write(
|
||
path,
|
||
json.dumps(
|
||
{
|
||
"createdAt": time.time(),
|
||
"query": query,
|
||
"response": response,
|
||
},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
).encode("utf-8"),
|
||
)
|
||
self.increment("water_misses")
|
||
return response, "MISS"
|
||
except Exception as error:
|
||
self.increment("water_errors")
|
||
if path.exists():
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
return payload["response"], "STALE"
|
||
existing: dict[str, Any] = {}
|
||
if pending_path.exists():
|
||
try:
|
||
existing = json.loads(pending_path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
existing = {}
|
||
pending = {
|
||
"cacheKey": cache_key,
|
||
"bbox": bbox,
|
||
"query": query,
|
||
"createdAt": existing.get("createdAt", time.time()),
|
||
"updatedAt": time.time(),
|
||
"nextAttempt": time.time() + 30,
|
||
"attempts": int(existing.get("attempts", 0)) + 1,
|
||
"lastError": str(error),
|
||
}
|
||
atomic_write(
|
||
pending_path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
self.increment("water_pending")
|
||
raise WaterPending(cache_key)
|
||
|
||
def water_pending_status(self, cache_key: str) -> dict[str, Any]:
|
||
cache_path = self.water_cache_path(cache_key)
|
||
if cache_path.exists():
|
||
return {
|
||
"ok": True,
|
||
"ready": True,
|
||
"cacheKey": cache_key,
|
||
"createdAt": cache_path.stat().st_mtime,
|
||
}
|
||
pending_path = self.pending_water_dir / f"{cache_key}.json"
|
||
if pending_path.exists():
|
||
try:
|
||
pending = json.loads(pending_path.read_text(encoding="utf-8"))
|
||
return {
|
||
"ok": True,
|
||
"ready": False,
|
||
"cacheKey": cache_key,
|
||
"attempts": pending.get("attempts", 0),
|
||
"lastError": pending.get("lastError", ""),
|
||
"updatedAt": pending.get("updatedAt"),
|
||
}
|
||
except Exception:
|
||
pass
|
||
return {"ok": False, "ready": False, "cacheKey": cache_key}
|
||
|
||
def retry_pending_water_jobs(self, limit: int | None = None) -> int:
|
||
if self.config.cache_only:
|
||
return 0
|
||
processed = 0
|
||
max_items = limit or self.config.refresh_batch
|
||
now = time.time()
|
||
for path in self.pending_water_dir.glob("*.json"):
|
||
if processed >= max_items:
|
||
break
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
if now < float(pending.get("nextAttempt", 0)):
|
||
continue
|
||
if pending.get("jobKind") != "vector":
|
||
bbox = tuple(float(value) for value in pending["bbox"])
|
||
pending["jobKind"] = "vector"
|
||
pending["tileZoom"] = self.choose_vector_zoom(bbox) # type: ignore[arg-type]
|
||
pending["forceTiles"] = False
|
||
pending["nextAttempt"] = now
|
||
atomic_write(
|
||
path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
self.complete_vector_water_pending(path)
|
||
processed += 1
|
||
continue
|
||
response = self.fetch_overpass(str(pending["query"]))
|
||
if not response.get("elements"):
|
||
raise RuntimeError("Overpass returned no water elements")
|
||
atomic_write(
|
||
self.water_cache_path(str(pending["cacheKey"])),
|
||
json.dumps(
|
||
{
|
||
"createdAt": time.time(),
|
||
"query": pending["query"],
|
||
"response": response,
|
||
},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
).encode("utf-8"),
|
||
)
|
||
path.unlink(missing_ok=True)
|
||
except Exception as error:
|
||
try:
|
||
pending = json.loads(path.read_text(encoding="utf-8"))
|
||
pending["updatedAt"] = time.time()
|
||
pending["nextAttempt"] = time.time() + 120
|
||
pending["attempts"] = int(pending.get("attempts", 0)) + 1
|
||
pending["lastError"] = str(error)
|
||
atomic_write(
|
||
path,
|
||
json.dumps(pending, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
except Exception:
|
||
pass
|
||
processed += 1
|
||
return processed
|
||
|
||
def build_overpass_query(self, south: float, west: float, north: float, east: float, types: list[str]) -> str:
|
||
bbox = f"{south:.6f},{west:.6f},{north:.6f},{east:.6f}"
|
||
requests = []
|
||
if "rail" in types:
|
||
requests.append(f'way["railway"="rail"][!"service"]({bbox});')
|
||
if "subway" in types:
|
||
# Beijing uses subway for metro tracks, while other Chinese cities also
|
||
# use light_rail/monorail for parts of their urban rail network.
|
||
requests.append(
|
||
f'way["railway"~"^(subway|light_rail|monorail)$"][!"service"]({bbox});'
|
||
)
|
||
# Route relations reconnect OSM ways that are split at stations and
|
||
# junctions, so the frontend can draw continuous lines.
|
||
requests.append(
|
||
f'relation["route"~"^(subway|light_rail|monorail)$"]({bbox});'
|
||
)
|
||
return "[out:json][timeout:90];\n(\n " + "\n ".join(requests) + "\n);\nout geom;"
|
||
|
||
def fetch_overpass(self, query: str) -> dict[str, Any]:
|
||
if self.config.cache_only:
|
||
raise CacheOnlyMode("Overpass requests are disabled by cache-only mode")
|
||
body = urllib.parse.urlencode({"data": query}).encode("utf-8")
|
||
task_pool = DaemonTaskPool(len(OVERPASS_ENDPOINTS), "overpass-query")
|
||
results: queue.Queue[tuple[bool, Any]] = queue.Queue()
|
||
|
||
def fetch_endpoint(endpoint: str) -> None:
|
||
request = urllib.request.Request(
|
||
endpoint,
|
||
data=body,
|
||
method="POST",
|
||
headers={
|
||
"User-Agent": USER_AGENT,
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
|
||
},
|
||
)
|
||
try:
|
||
results.put((True, self._read_json(request, 35)))
|
||
except Exception as error:
|
||
results.put((False, error))
|
||
|
||
for endpoint in OVERPASS_ENDPOINTS:
|
||
task_pool.submit(fetch_endpoint, endpoint)
|
||
|
||
empty_result: dict[str, Any] | None = None
|
||
deadline = time.time() + 40
|
||
try:
|
||
for _ in OVERPASS_ENDPOINTS:
|
||
remaining = deadline - time.time()
|
||
if remaining <= 0:
|
||
break
|
||
try:
|
||
succeeded, value = results.get(timeout=remaining)
|
||
if not succeeded:
|
||
continue
|
||
payload = value
|
||
elements = payload.get("elements", [])
|
||
if elements:
|
||
return payload
|
||
if empty_result is None:
|
||
empty_result = payload
|
||
except queue.Empty:
|
||
continue
|
||
if empty_result is not None:
|
||
return empty_result
|
||
raise RuntimeError("All Overpass endpoints failed")
|
||
finally:
|
||
task_pool.shutdown()
|
||
|
||
def _read_json(self, request: urllib.request.Request, timeout: int) -> dict[str, Any]:
|
||
with self.open_upstream_url(request, timeout) as response:
|
||
return json.load(response)
|
||
|
||
def refresh_rail_path(self, path: Path) -> bool:
|
||
if self.config.cache_only:
|
||
return False
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
query = payload.get("query")
|
||
if not isinstance(query, str):
|
||
return False
|
||
response = self.fetch_overpass(query)
|
||
updated = {
|
||
"createdAt": time.time(),
|
||
"query": query,
|
||
"response": response,
|
||
}
|
||
atomic_write(
|
||
path,
|
||
json.dumps(updated, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
finally:
|
||
with self._lock_guard:
|
||
self._refreshing.discard(str(path))
|
||
|
||
def get_rail(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
types: list[str],
|
||
force: bool,
|
||
tile_zoom: int | None = None,
|
||
) -> tuple[dict[str, Any], str]:
|
||
vector_zoom = max(4, min(14, int(tile_zoom or 12)))
|
||
cache_key = self.vector_bbox_cache_key(bbox, vector_zoom, types)
|
||
pending_path = self.pending_rail_path(cache_key)
|
||
path = self.rail_cache_path(cache_key)
|
||
has_vector_result = False
|
||
result_is_stale = False
|
||
if path.exists() and not force:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
response = payload["response"]
|
||
if response.get("source") == "openfreemap":
|
||
result_is_stale = time.time() - path.stat().st_mtime > self.config.rail_ttl_seconds
|
||
if not result_is_stale:
|
||
has_vector_result = True
|
||
self.increment("rail_hits")
|
||
return response, "HIT"
|
||
except Exception:
|
||
pass
|
||
if self.config.cache_only:
|
||
if path.exists():
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
return payload["response"], "STALE"
|
||
except Exception:
|
||
pass
|
||
missing_tiles = self.vector_cache_missing(
|
||
bbox,
|
||
vector_zoom,
|
||
force=False,
|
||
)
|
||
if missing_tiles == 0:
|
||
try:
|
||
response = self.vector_rail_response(
|
||
bbox,
|
||
types,
|
||
vector_zoom,
|
||
force=False,
|
||
)
|
||
if (
|
||
response.get("failedTiles") == 0
|
||
and response.get("loadedTiles") == response.get("tileCount")
|
||
):
|
||
return response, "HIT"
|
||
except Exception:
|
||
pass
|
||
return self.cache_only_vector_response("rail", vector_zoom), "PLACEHOLDER"
|
||
missing_tiles = self.vector_cache_missing(bbox, vector_zoom, force=force)
|
||
needs_vector_result = force or result_is_stale or not has_vector_result
|
||
if missing_tiles or needs_vector_result:
|
||
with self.tile_lock("rail-pending:" + cache_key):
|
||
existing_kind = ""
|
||
if pending_path.exists():
|
||
try:
|
||
existing_kind = str(
|
||
json.loads(pending_path.read_text(encoding="utf-8")).get("jobKind", "")
|
||
)
|
||
except Exception:
|
||
existing_kind = ""
|
||
if existing_kind != "vector":
|
||
self.queue_pending_rail(
|
||
cache_key,
|
||
bbox,
|
||
types,
|
||
"",
|
||
f"Queued {missing_tiles} OpenFreeMap vector tiles",
|
||
job_kind="vector",
|
||
tile_zoom=vector_zoom,
|
||
force_tiles=force or result_is_stale,
|
||
)
|
||
self._download_executor.submit(self.complete_vector_rail_pending, pending_path)
|
||
raise RailPending(cache_key)
|
||
south, west, north, east = bbox
|
||
query = self.build_overpass_query(south, west, north, east, types)
|
||
if path.exists() and not force:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
age = time.time() - path.stat().st_mtime
|
||
response = payload["response"]
|
||
if age <= self.config.rail_ttl_seconds:
|
||
self.increment("rail_hits")
|
||
return response, "HIT"
|
||
self.increment("rail_stale")
|
||
refresh_key = str(path)
|
||
with self._lock_guard:
|
||
needs_refresh = refresh_key not in self._refreshing
|
||
if needs_refresh:
|
||
self._refreshing.add(refresh_key)
|
||
if needs_refresh:
|
||
self._refresh_executor.submit(self.refresh_rail_path, path)
|
||
return response, "STALE"
|
||
except Exception:
|
||
pass
|
||
|
||
if pending_path.exists() and not force:
|
||
raise RailPending(cache_key)
|
||
|
||
try:
|
||
response = self.fetch_overpass(query)
|
||
payload = {
|
||
"createdAt": time.time(),
|
||
"query": query,
|
||
"response": response,
|
||
}
|
||
atomic_write(
|
||
path,
|
||
json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
self.increment("rail_misses")
|
||
return response, "MISS"
|
||
except Exception as error:
|
||
self.increment("rail_errors")
|
||
if path.exists():
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
return payload["response"], "STALE"
|
||
self.queue_pending_rail(cache_key, bbox, types, query, str(error))
|
||
raise RailPending(cache_key)
|
||
|
||
def cache_category_paths(self) -> dict[str, list[Path]]:
|
||
raster_providers = [
|
||
name for name, provider in TILE_PROVIDERS.items()
|
||
if not provider.get("manual_only")
|
||
]
|
||
tianditu_providers = [
|
||
name for name, provider in TILE_PROVIDERS.items()
|
||
if provider.get("manual_only")
|
||
]
|
||
return {
|
||
"raster": [self.tile_cache_dir / name for name in raster_providers],
|
||
"tianditu": [self.tile_cache_dir / name for name in tianditu_providers],
|
||
"vector": [self.vector_cache_dir],
|
||
"rail": [self.rail_cache_dir],
|
||
"environment": [self.water_cache_dir],
|
||
"boundaries": [self.boundary_cache_dir],
|
||
"rail-derived": [self.rail_tile_dir, self.rail_metro_dir],
|
||
"dynamic-vector": [self.local_vector_dynamic_dir, self.local_vector_25_dynamic_dir],
|
||
"pending": [self.pending_rail_dir, self.pending_water_dir],
|
||
"staging": [self.cache_dir / DOWNLOAD_STAGING_DIR_NAME],
|
||
}
|
||
|
||
@staticmethod
|
||
def cache_admin_category_info(category_id: str) -> dict[str, Any] | None:
|
||
query_layer = CACHE_ADMIN_QUERY_LAYERS.get(category_id)
|
||
if query_layer is not None:
|
||
return query_layer
|
||
return CACHE_ADMIN_CATEGORIES.get(category_id)
|
||
|
||
def response_layer_has_features(self, category_id: str, layer_key: str, path: Path) -> bool:
|
||
return layer_key in self.response_layer_coverage(category_id, path)
|
||
|
||
def delete_response_layer_file(
|
||
self,
|
||
category_id: str,
|
||
layer_key: str,
|
||
path: Path,
|
||
) -> tuple[int, int]:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
response = payload.get("response") if isinstance(payload, dict) else None
|
||
if not isinstance(response, dict):
|
||
return 0, 0
|
||
grouped = self.grouped_response_features(category_id, response)
|
||
target_features = grouped.get(layer_key) or []
|
||
if not target_features:
|
||
return 0, 0
|
||
old_size = path.stat().st_size
|
||
target_ids = {id(feature) for feature in target_features}
|
||
if category_id == "rail":
|
||
response["elements"] = [
|
||
feature for feature in response.get("elements", [])
|
||
if id(feature) not in target_ids
|
||
]
|
||
elif category_id == "environment":
|
||
field = "elements" if layer_key == "water" else layer_key
|
||
response[field] = [
|
||
feature for feature in response.get(field, [])
|
||
if id(feature) not in target_ids
|
||
]
|
||
elif category_id == "boundaries":
|
||
response["elements"] = []
|
||
remaining = any(
|
||
response.get(field)
|
||
for field in ("elements", "vegetation", "ports")
|
||
)
|
||
if not remaining:
|
||
path.unlink()
|
||
return 1, old_size
|
||
atomic_write(
|
||
path,
|
||
json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
return 1, max(0, old_size - path.stat().st_size)
|
||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||
return 0, 0
|
||
|
||
@staticmethod
|
||
def parse_cache_bounds(value: Any) -> tuple[float, float, float, float] | None:
|
||
if value is None:
|
||
return None
|
||
if not isinstance(value, (list, tuple)) or len(value) != 4:
|
||
raise ValueError("Cache bounds must contain south, west, north and east")
|
||
try:
|
||
south, west, north, east = (float(item) for item in value)
|
||
except (TypeError, ValueError) as error:
|
||
raise ValueError("Cache bounds must be numeric") from error
|
||
if not all(math.isfinite(item) for item in (south, west, north, east)):
|
||
raise ValueError("Cache bounds must be finite")
|
||
if south < -85.05112878 or north > 85.05112878 or south >= north:
|
||
raise ValueError("Cache bounds have an invalid latitude range")
|
||
if west < -180.0 or east > 180.0 or west >= east:
|
||
raise ValueError("Cache bounds have an invalid longitude range")
|
||
return south, west, north, east
|
||
|
||
@staticmethod
|
||
def cache_bounds_intersect(
|
||
first: tuple[float, float, float, float] | list[float],
|
||
second: tuple[float, float, float, float] | list[float],
|
||
) -> bool:
|
||
return (
|
||
first[0] <= second[2]
|
||
and first[2] >= second[0]
|
||
and first[1] <= second[3]
|
||
and first[3] >= second[1]
|
||
)
|
||
|
||
@staticmethod
|
||
def cache_tile_coordinate(
|
||
category_id: str,
|
||
path: Path,
|
||
root: Path,
|
||
) -> tuple[int, int, int] | None:
|
||
suffixes = CACHE_COVERAGE_TILE_SUFFIXES.get(category_id)
|
||
if not suffixes or path.suffix.lower() not in suffixes:
|
||
return None
|
||
try:
|
||
relative = path.relative_to(root).parts
|
||
if len(relative) != 3:
|
||
return None
|
||
zoom = int(relative[0])
|
||
x = int(relative[1])
|
||
y_match = re.match(r"^(\d+)", path.stem)
|
||
if y_match is None:
|
||
return None
|
||
y = int(y_match.group(1))
|
||
world_tiles = 1 << zoom
|
||
if not 0 <= zoom <= 22 or not (0 <= x < world_tiles and 0 <= y < world_tiles):
|
||
return None
|
||
return zoom, x, y
|
||
except (OSError, ValueError, TypeError):
|
||
return None
|
||
|
||
def cache_files_in_bounds(
|
||
self,
|
||
category_ids: list[str],
|
||
bounds: tuple[float, float, float, float],
|
||
) -> tuple[dict[str, list[Path]], list[str]]:
|
||
path_groups = self.cache_category_paths()
|
||
response_categories = {"rail", "environment", "boundaries", "pending"}
|
||
matches: dict[str, list[Path]] = {}
|
||
skipped: list[str] = []
|
||
for category_id in category_ids:
|
||
category_files: list[Path] = []
|
||
seen: set[str] = set()
|
||
for root in path_groups[category_id]:
|
||
if not root.exists():
|
||
continue
|
||
if category_id in CACHE_COVERAGE_TILE_SUFFIXES:
|
||
try:
|
||
zoom_dirs = [item for item in root.iterdir() if item.is_dir() and item.name.isdigit()]
|
||
except OSError:
|
||
continue
|
||
for zoom_dir in zoom_dirs:
|
||
try:
|
||
zoom = int(zoom_dir.name)
|
||
if not 0 <= zoom <= 22:
|
||
continue
|
||
world_tiles = 1 << zoom
|
||
x_min = self.lon_to_tile_x(bounds[1], zoom)
|
||
x_max = self.lon_to_tile_x(bounds[3], zoom)
|
||
y_min = self.lat_to_tile_y(bounds[2], zoom)
|
||
y_max = self.lat_to_tile_y(bounds[0], zoom)
|
||
x_dirs = list(zoom_dir.iterdir())
|
||
except (OSError, ValueError):
|
||
continue
|
||
for x_dir in x_dirs:
|
||
if not x_dir.is_dir() or not x_dir.name.isdigit():
|
||
continue
|
||
x = int(x_dir.name)
|
||
if x < x_min or x > x_max or not 0 <= x < world_tiles:
|
||
continue
|
||
try:
|
||
candidates = list(x_dir.iterdir())
|
||
except OSError:
|
||
continue
|
||
for path in candidates:
|
||
coordinate = self.cache_tile_coordinate(category_id, path, root)
|
||
if coordinate is None:
|
||
continue
|
||
z, tile_x, tile_y = coordinate
|
||
if tile_y < y_min or tile_y > y_max:
|
||
continue
|
||
tile_bbox = vector_tile_bbox(z, tile_x, tile_y)
|
||
if not self.cache_bounds_intersect(tile_bbox, bounds):
|
||
continue
|
||
key = str(path)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
category_files.append(path)
|
||
elif category_id in response_categories:
|
||
for path in root.rglob("*.json"):
|
||
response_bbox = self.response_cache_bbox(category_id, path)
|
||
if response_bbox is None or not self.cache_bounds_intersect(response_bbox, bounds):
|
||
continue
|
||
key = str(path)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
category_files.append(path)
|
||
else:
|
||
if category_id not in skipped:
|
||
skipped.append(category_id)
|
||
matches[category_id] = category_files
|
||
return matches, skipped
|
||
|
||
@staticmethod
|
||
def summarize_cache_paths(paths: list[Path]) -> dict[str, Any]:
|
||
files = 0
|
||
total_bytes = 0
|
||
updated_at = 0.0
|
||
for root in paths:
|
||
if not root.exists():
|
||
continue
|
||
candidates = (root,) if root.is_file() else root.rglob("*")
|
||
for path in candidates:
|
||
try:
|
||
if not path.is_file():
|
||
continue
|
||
stat = path.stat()
|
||
except OSError:
|
||
continue
|
||
files += 1
|
||
total_bytes += stat.st_size
|
||
updated_at = max(updated_at, stat.st_mtime)
|
||
return {"files": files, "bytes": total_bytes, "updatedAt": updated_at or None}
|
||
|
||
def active_manual_download_count(self) -> int:
|
||
with self._lock_guard:
|
||
return sum(
|
||
1
|
||
for jobs in (self._tianditu_jobs, self._openmap_jobs)
|
||
for job in jobs.values()
|
||
if job.get("status") in ("queued", "running")
|
||
)
|
||
|
||
@staticmethod
|
||
def tile_runs(tiles: set[tuple[int, int]]) -> list[list[int]]:
|
||
rows: dict[int, list[int]] = {}
|
||
for x, y in tiles:
|
||
rows.setdefault(y, []).append(x)
|
||
runs: list[list[int]] = []
|
||
for y, values in sorted(rows.items()):
|
||
values.sort()
|
||
start = previous = values[0]
|
||
for value in values[1:]:
|
||
if value != previous + 1:
|
||
runs.append([y, start, previous])
|
||
start = value
|
||
previous = value
|
||
runs.append([y, start, previous])
|
||
return runs
|
||
|
||
@staticmethod
|
||
def coverage_bounds_for_tiles(z: int, tiles: set[tuple[int, int]]) -> list[float] | None:
|
||
if not tiles:
|
||
return None
|
||
min_x = min(x for x, _ in tiles)
|
||
max_x = max(x for x, _ in tiles)
|
||
min_y = min(y for _, y in tiles)
|
||
max_y = max(y for _, y in tiles)
|
||
north = vector_tile_bbox(z, min_x, min_y)[2]
|
||
south = vector_tile_bbox(z, max_x, max_y)[0]
|
||
west = min_x / (1 << z) * 360.0 - 180.0
|
||
east = (max_x + 1) / (1 << z) * 360.0 - 180.0
|
||
return [south, west, north, east]
|
||
|
||
@staticmethod
|
||
def response_cache_bbox(category_id: str, path: Path) -> list[float] | None:
|
||
try:
|
||
if category_id == "pending":
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
bbox = payload.get("bbox")
|
||
if not isinstance(bbox, list) or len(bbox) != 4:
|
||
return None
|
||
south, west, north, east = (float(value) for value in bbox)
|
||
else:
|
||
raw = path.read_bytes()
|
||
south, west, north, east = 90.0, 180.0, -90.0, -180.0
|
||
found = False
|
||
for match in JSON_LATITUDE_RE.finditer(raw):
|
||
value = float(match.group(1))
|
||
if -90.0 <= value <= 90.0:
|
||
south = min(south, value)
|
||
north = max(north, value)
|
||
found = True
|
||
for match in JSON_LONGITUDE_RE.finditer(raw):
|
||
value = float(match.group(1))
|
||
if -180.0 <= value <= 180.0:
|
||
west = min(west, value)
|
||
east = max(east, value)
|
||
found = True
|
||
if not found:
|
||
return None
|
||
if not all(math.isfinite(value) for value in (south, west, north, east)):
|
||
return None
|
||
if south > north or west > east or south < -90 or north > 90:
|
||
return None
|
||
return [south, west, north, east]
|
||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||
return None
|
||
|
||
def response_features_coverage(
|
||
self,
|
||
features: Any,
|
||
source_zoom: int | None,
|
||
) -> tuple[list[float] | None, set[tuple[int, int]]]:
|
||
lats: list[float] = []
|
||
lons: list[float] = []
|
||
tiles: set[tuple[int, int]] = set()
|
||
|
||
def collect(node: Any) -> None:
|
||
if isinstance(node, dict):
|
||
lat = node.get("lat")
|
||
lon = node.get("lon", node.get("lng"))
|
||
if lat is not None and lon is not None:
|
||
try:
|
||
lat_value = float(lat)
|
||
lon_value = float(lon)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
else:
|
||
if -90.0 <= lat_value <= 90.0 and -180.0 <= lon_value <= 180.0:
|
||
lats.append(lat_value)
|
||
lons.append(lon_value)
|
||
if source_zoom is not None:
|
||
tiles.add(
|
||
(
|
||
self.lon_to_tile_x(lon_value, source_zoom),
|
||
self.lat_to_tile_y(lat_value, source_zoom),
|
||
)
|
||
)
|
||
return
|
||
for key in ("geometry", "geometries", "coordinates"):
|
||
if key in node:
|
||
collect(node[key])
|
||
elif isinstance(node, list):
|
||
for value in node:
|
||
collect(value)
|
||
|
||
collect(features)
|
||
if not lats or not lons:
|
||
return None, tiles
|
||
return [min(lats), min(lons), max(lats), max(lons)], tiles
|
||
|
||
@staticmethod
|
||
def grouped_response_features(category_id: str, response: dict[str, Any]) -> dict[str, list[Any]]:
|
||
grouped: dict[str, list[Any]] = {}
|
||
if category_id == "rail":
|
||
national: list[Any] = []
|
||
metro: list[Any] = []
|
||
metro_modes = {"subway", "light_rail", "monorail", "tram"}
|
||
for element in response.get("elements") or []:
|
||
if not isinstance(element, dict):
|
||
continue
|
||
tags = element.get("tags") if isinstance(element.get("tags"), dict) else {}
|
||
mode = str(tags.get("railway") or tags.get("route") or "").lower()
|
||
if mode in metro_modes:
|
||
metro.append(element)
|
||
else:
|
||
national.append(element)
|
||
grouped = {"national": national, "metro": metro}
|
||
elif category_id == "environment":
|
||
grouped = {
|
||
"water": response.get("elements") or [],
|
||
"vegetation": response.get("vegetation") or [],
|
||
"ports": response.get("ports") or [],
|
||
}
|
||
elif category_id == "boundaries":
|
||
grouped = {"boundaries": response.get("elements") or []}
|
||
return grouped
|
||
|
||
def response_layer_coverage(
|
||
self,
|
||
category_id: str,
|
||
path: Path,
|
||
) -> dict[str, dict[str, Any]]:
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||
return {}
|
||
response = payload.get("response") if isinstance(payload, dict) else None
|
||
if not isinstance(response, dict):
|
||
return {}
|
||
try:
|
||
source_zoom = int(response.get("sourceTileZoom") or response.get("tileZoom"))
|
||
except (TypeError, ValueError):
|
||
source_zoom = 0
|
||
if not OPENFREEMAP_SOURCE_MIN_ZOOM <= source_zoom <= 22:
|
||
source_zoom = None
|
||
grouped = self.grouped_response_features(category_id, response)
|
||
result: dict[str, dict[str, Any]] = {}
|
||
for key, features in grouped.items():
|
||
bbox, tiles = self.response_features_coverage(features, source_zoom)
|
||
if bbox is not None:
|
||
result[key] = {
|
||
"bbox": [round(value, 6) for value in bbox],
|
||
"sourceZoom": source_zoom,
|
||
"tiles": tiles,
|
||
}
|
||
return result
|
||
|
||
def build_cache_coverage(self) -> dict[str, Any]:
|
||
started = time.time()
|
||
path_groups = self.cache_category_paths()
|
||
categories: list[dict[str, Any]] = []
|
||
for category_id, info in CACHE_ADMIN_CATEGORIES.items():
|
||
category: dict[str, Any] = {
|
||
"id": category_id,
|
||
"label": info["label"],
|
||
"description": info["description"],
|
||
"spatial": False,
|
||
"mode": "none",
|
||
"bounds": None,
|
||
"zooms": [],
|
||
"areas": [],
|
||
"mappedFiles": 0,
|
||
}
|
||
suffixes = CACHE_COVERAGE_TILE_SUFFIXES.get(category_id)
|
||
if suffixes:
|
||
by_zoom: dict[int, set[tuple[int, int]]] = {}
|
||
for root in path_groups[category_id]:
|
||
if not root.exists():
|
||
continue
|
||
for path in root.rglob("*"):
|
||
if not path.is_file() or path.suffix.lower() not in suffixes:
|
||
continue
|
||
try:
|
||
relative = path.relative_to(root).parts
|
||
if len(relative) != 3:
|
||
continue
|
||
z = int(relative[0])
|
||
x = int(relative[1])
|
||
y_match = re.match(r"^(\d+)", Path(relative[2]).stem)
|
||
if y_match is None:
|
||
continue
|
||
y = int(y_match.group(1))
|
||
world_tiles = 1 << z
|
||
if not 0 <= z <= 22 or not (0 <= x < world_tiles and 0 <= y < world_tiles):
|
||
continue
|
||
except (OSError, ValueError):
|
||
continue
|
||
by_zoom.setdefault(z, set()).add((x, y))
|
||
category["mappedFiles"] += 1
|
||
zoom_details: list[dict[str, Any]] = []
|
||
for zoom in sorted(by_zoom):
|
||
zoom_tiles = by_zoom[zoom]
|
||
zoom_details.append(
|
||
{
|
||
"zoom": zoom,
|
||
"tiles": len(zoom_tiles),
|
||
"runs": self.tile_runs(zoom_tiles),
|
||
"bounds": self.coverage_bounds_for_tiles(zoom, zoom_tiles),
|
||
}
|
||
)
|
||
zoom_bounds = [item["bounds"] for item in zoom_details if item["bounds"]]
|
||
category.update(
|
||
{
|
||
"spatial": bool(zoom_details),
|
||
"mode": "tiles",
|
||
"bounds": [
|
||
min(item[0] for item in zoom_bounds),
|
||
min(item[1] for item in zoom_bounds),
|
||
max(item[2] for item in zoom_bounds),
|
||
max(item[3] for item in zoom_bounds),
|
||
] if zoom_bounds else None,
|
||
"zooms": zoom_details,
|
||
"totalTiles": sum(item["tiles"] for item in zoom_details),
|
||
}
|
||
)
|
||
elif category_id in CACHE_COVERAGE_RESPONSE_LAYERS:
|
||
layer_specs = CACHE_COVERAGE_RESPONSE_LAYERS[category_id]
|
||
layer_areas: dict[str, list[list[float]]] = {spec["id"]: [] for spec in layer_specs}
|
||
layer_tiles: dict[str, dict[int, set[tuple[int, int]]]] = {
|
||
spec["id"]: {} for spec in layer_specs
|
||
}
|
||
for root in path_groups[category_id]:
|
||
if not root.exists():
|
||
continue
|
||
for path in root.rglob("*.json"):
|
||
found = self.response_layer_coverage(category_id, path)
|
||
for spec in layer_specs:
|
||
layer = found.get(spec["key"])
|
||
if not isinstance(layer, dict):
|
||
continue
|
||
bbox = layer.get("bbox")
|
||
if isinstance(bbox, list) and len(bbox) == 4:
|
||
layer_areas[spec["id"]].append(bbox)
|
||
source_zoom = layer.get("sourceZoom")
|
||
source_tiles = layer.get("tiles")
|
||
if isinstance(source_zoom, int) and isinstance(source_tiles, set) and source_tiles:
|
||
layer_tiles[spec["id"]].setdefault(source_zoom, set()).update(source_tiles)
|
||
for spec in layer_specs:
|
||
areas = layer_areas[spec["id"]]
|
||
if not areas:
|
||
continue
|
||
zoom_details: list[dict[str, Any]] = []
|
||
for zoom, zoom_tiles in sorted(layer_tiles[spec["id"]].items()):
|
||
zoom_details.append(
|
||
{
|
||
"zoom": zoom,
|
||
"tiles": len(zoom_tiles),
|
||
"runs": self.tile_runs(zoom_tiles),
|
||
"bounds": self.coverage_bounds_for_tiles(zoom, zoom_tiles),
|
||
}
|
||
)
|
||
categories.append(
|
||
{
|
||
"id": spec["id"],
|
||
"parentId": category_id,
|
||
"label": spec["label"],
|
||
"description": spec["description"],
|
||
"spatial": True,
|
||
"mode": "tiles" if zoom_details else "areas",
|
||
"bounds": [
|
||
min(area[0] for area in areas),
|
||
min(area[1] for area in areas),
|
||
max(area[2] for area in areas),
|
||
max(area[3] for area in areas),
|
||
],
|
||
"zooms": zoom_details,
|
||
"areas": areas,
|
||
"mappedFiles": len(areas),
|
||
"totalTiles": sum(item["tiles"] for item in zoom_details),
|
||
"responseOnly": True,
|
||
}
|
||
)
|
||
continue
|
||
elif category_id in ("boundaries", "pending"):
|
||
areas: list[list[float]] = []
|
||
by_zoom: dict[int, set[tuple[int, int]]] = {}
|
||
for root in path_groups[category_id]:
|
||
if not root.exists():
|
||
continue
|
||
for path in root.rglob("*.json"):
|
||
found = self.response_layer_coverage(category_id, path)
|
||
layer = found.get("boundaries") if category_id == "boundaries" else None
|
||
if isinstance(layer, dict):
|
||
bbox = layer.get("bbox")
|
||
if isinstance(bbox, list) and len(bbox) == 4:
|
||
areas.append(bbox)
|
||
source_zoom = layer.get("sourceZoom")
|
||
source_tiles = layer.get("tiles")
|
||
if isinstance(source_zoom, int) and isinstance(source_tiles, set) and source_tiles:
|
||
by_zoom.setdefault(source_zoom, set()).update(source_tiles)
|
||
else:
|
||
bbox = self.response_cache_bbox(category_id, path)
|
||
if bbox is not None:
|
||
areas.append([round(value, 6) for value in bbox])
|
||
if areas:
|
||
zoom_details = [
|
||
{
|
||
"zoom": zoom,
|
||
"tiles": len(zoom_tiles),
|
||
"runs": self.tile_runs(zoom_tiles),
|
||
"bounds": self.coverage_bounds_for_tiles(zoom, zoom_tiles),
|
||
}
|
||
for zoom, zoom_tiles in sorted(by_zoom.items())
|
||
]
|
||
category.update(
|
||
{
|
||
"spatial": True,
|
||
"mode": "tiles" if zoom_details else "areas",
|
||
"bounds": [
|
||
min(area[0] for area in areas),
|
||
min(area[1] for area in areas),
|
||
max(area[2] for area in areas),
|
||
max(area[3] for area in areas),
|
||
],
|
||
"zooms": zoom_details,
|
||
"areas": areas,
|
||
"mappedFiles": len(areas),
|
||
"totalTiles": sum(item["tiles"] for item in zoom_details),
|
||
"responseOnly": category_id == "boundaries",
|
||
}
|
||
)
|
||
categories.append(category)
|
||
return {
|
||
"ok": True,
|
||
"version": CACHE_COVERAGE_INDEX_VERSION,
|
||
"rangeOperations": True,
|
||
"generatedAt": time.time(),
|
||
"elapsedMs": round((time.time() - started) * 1000),
|
||
"categories": categories,
|
||
}
|
||
|
||
def cache_coverage(self, refresh: bool = False) -> dict[str, Any]:
|
||
index_path = self.cache_dir / CACHE_COVERAGE_INDEX_NAME
|
||
with self._cache_coverage_lock:
|
||
now = time.time()
|
||
if (
|
||
not refresh
|
||
and self._cache_coverage_result is not None
|
||
and now - self._cache_coverage_cached_at < 60
|
||
):
|
||
return self._cache_coverage_result
|
||
if not refresh:
|
||
try:
|
||
stored = json.loads(index_path.read_text(encoding="utf-8"))
|
||
if (
|
||
stored.get("version") == CACHE_COVERAGE_INDEX_VERSION
|
||
and isinstance(stored.get("categories"), list)
|
||
):
|
||
stored["fromIndex"] = True
|
||
stored["rangeOperations"] = True
|
||
self._cache_coverage_result = stored
|
||
self._cache_coverage_cached_at = now
|
||
return stored
|
||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||
pass
|
||
result = self.build_cache_coverage()
|
||
result["fromIndex"] = False
|
||
atomic_write(
|
||
index_path,
|
||
json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
|
||
)
|
||
self._cache_coverage_result = result
|
||
self._cache_coverage_cached_at = now
|
||
return result
|
||
|
||
def coverage_tiles_for_category(
|
||
self,
|
||
category_id: str,
|
||
bounds: tuple[float, float, float, float] | None = None,
|
||
) -> list[tuple[int, int, int]]:
|
||
coverage = self.cache_coverage(refresh=False)
|
||
category = next(
|
||
(
|
||
item for item in coverage.get("categories", [])
|
||
if isinstance(item, dict) and item.get("id") == category_id
|
||
),
|
||
None,
|
||
)
|
||
if not isinstance(category, dict):
|
||
return []
|
||
tiles: set[tuple[int, int, int]] = set()
|
||
for zoom_entry in category.get("zooms") or []:
|
||
if not isinstance(zoom_entry, dict):
|
||
continue
|
||
try:
|
||
zoom = int(zoom_entry.get("zoom"))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if not OPENFREEMAP_SOURCE_MIN_ZOOM <= zoom <= OPENFREEMAP_SOURCE_MAX_ZOOM:
|
||
continue
|
||
world_tiles = 1 << zoom
|
||
for run in zoom_entry.get("runs") or []:
|
||
if not isinstance(run, list) or len(run) != 3:
|
||
continue
|
||
try:
|
||
y, x0, x1 = (int(value) for value in run)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if not (0 <= y < world_tiles and 0 <= x0 <= x1 < world_tiles):
|
||
continue
|
||
for x in range(x0, x1 + 1):
|
||
if bounds is not None and not self.cache_bounds_intersect(
|
||
vector_tile_bbox(zoom, x, y), bounds
|
||
):
|
||
continue
|
||
tiles.add((zoom, x, y))
|
||
return sorted(tiles)
|
||
|
||
def invalidate_query_layer_responses(
|
||
self,
|
||
category_id: str,
|
||
bounds: tuple[float, float, float, float] | None = None,
|
||
) -> int:
|
||
info = CACHE_ADMIN_QUERY_LAYERS.get(category_id)
|
||
if not info:
|
||
return 0
|
||
parent_id = str(info["parent"])
|
||
layer_key = str(info["key"])
|
||
path_groups = self.cache_category_paths()
|
||
changed_files = 0
|
||
for root in path_groups.get(parent_id, []):
|
||
if not root.exists():
|
||
continue
|
||
for path in root.glob("*.json"):
|
||
if bounds is not None:
|
||
bbox = self.response_cache_bbox(parent_id, path)
|
||
if bbox is None or not self.cache_bounds_intersect(bbox, bounds):
|
||
continue
|
||
if not self.response_layer_has_features(parent_id, layer_key, path):
|
||
continue
|
||
changed, _ = self.delete_response_layer_file(parent_id, layer_key, path)
|
||
changed_files += changed
|
||
if changed_files:
|
||
self.invalidate_cache_admin_stats()
|
||
return changed_files
|
||
|
||
def invalidate_cache_admin_stats(self) -> None:
|
||
with self._cache_admin_stats_lock:
|
||
self._cache_admin_stats_cached_at = 0.0
|
||
self._cache_admin_stats_result = None
|
||
with self._cache_totals_lock:
|
||
self._cache_totals_cached_at = 0.0
|
||
self._cache_totals_result = (0, 0)
|
||
with self._cache_coverage_lock:
|
||
self._cache_coverage_cached_at = 0.0
|
||
self._cache_coverage_result = None
|
||
try:
|
||
(self.cache_dir / CACHE_COVERAGE_INDEX_NAME).unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
def cache_admin_status(self, refresh: bool = False) -> dict[str, Any]:
|
||
now = time.time()
|
||
with self._cache_admin_stats_lock:
|
||
if (
|
||
not refresh
|
||
and self._cache_admin_stats_result is not None
|
||
and now - self._cache_admin_stats_cached_at < 30
|
||
):
|
||
result = self._cache_admin_stats_result
|
||
else:
|
||
path_groups = self.cache_category_paths()
|
||
canonical_summaries = {
|
||
category_id: self.summarize_cache_paths(paths)
|
||
for category_id, paths in path_groups.items()
|
||
}
|
||
categories: list[dict[str, Any]] = []
|
||
total_files = sum(summary["files"] for summary in canonical_summaries.values())
|
||
total_bytes = sum(summary["bytes"] for summary in canonical_summaries.values())
|
||
total_tiles = 0
|
||
physical_tile_categories = {
|
||
"raster", "tianditu", "vector", "rail-derived", "dynamic-vector",
|
||
}
|
||
coverage_by_id: dict[str, dict[str, Any]] = {}
|
||
try:
|
||
stored_coverage = json.loads(
|
||
(self.cache_dir / CACHE_COVERAGE_INDEX_NAME).read_text(encoding="utf-8")
|
||
)
|
||
if stored_coverage.get("version") == CACHE_COVERAGE_INDEX_VERSION:
|
||
coverage_by_id = {
|
||
str(category.get("id")): category
|
||
for category in stored_coverage.get("categories", [])
|
||
if isinstance(category, dict) and category.get("id")
|
||
}
|
||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||
coverage_by_id = {}
|
||
if refresh or not coverage_by_id:
|
||
coverage_data = self.cache_coverage(refresh=refresh)
|
||
coverage_by_id = {
|
||
str(category.get("id")): category
|
||
for category in coverage_data.get("categories", [])
|
||
if isinstance(category, dict) and category.get("id")
|
||
}
|
||
for category_id in CACHE_ADMIN_MANAGEMENT_ORDER:
|
||
query_layer = CACHE_ADMIN_QUERY_LAYERS.get(category_id)
|
||
if query_layer:
|
||
coverage = coverage_by_id.get(category_id, {})
|
||
parent_id = str(query_layer["parent"])
|
||
parent_summary = canonical_summaries[parent_id]
|
||
windows = int(coverage.get("mappedFiles") or 0)
|
||
categories.append(
|
||
{
|
||
"id": category_id,
|
||
"parentId": parent_id,
|
||
"layerKey": query_layer["key"],
|
||
"label": query_layer["label"],
|
||
"description": query_layer["description"],
|
||
"refreshable": query_layer["refreshable"],
|
||
"queryResponse": True,
|
||
"files": windows,
|
||
"bytes": 0,
|
||
"updatedAt": parent_summary["updatedAt"],
|
||
"windows": windows,
|
||
"sourceTiles": int(coverage.get("totalTiles") or 0),
|
||
"tileCount": int(coverage.get("totalTiles") or 0),
|
||
"tileKind": "source",
|
||
"downloadProvider": "openfreemap",
|
||
}
|
||
)
|
||
continue
|
||
info = CACHE_ADMIN_CATEGORIES.get(category_id)
|
||
if info is None:
|
||
continue
|
||
canonical_summary = canonical_summaries[category_id]
|
||
tile_count = 0
|
||
if category_id in physical_tile_categories:
|
||
tile_count = int(
|
||
coverage_by_id.get(category_id, {}).get("totalTiles")
|
||
or canonical_summary["files"]
|
||
)
|
||
total_tiles += tile_count
|
||
categories.append(
|
||
{
|
||
"id": category_id,
|
||
"label": info["label"],
|
||
"description": info["description"],
|
||
"refreshable": info["refreshable"],
|
||
**canonical_summary,
|
||
"tileCount": tile_count,
|
||
"tileKind": "cache",
|
||
}
|
||
)
|
||
result = {
|
||
"ok": True,
|
||
"cacheOnly": self.config.cache_only,
|
||
"tileTtlSeconds": self.config.tile_ttl_seconds,
|
||
"railTtlSeconds": self.config.rail_ttl_seconds,
|
||
"totalFiles": total_files,
|
||
"totalBytes": total_bytes,
|
||
"totalTiles": total_tiles,
|
||
"categories": categories,
|
||
"excluded": [
|
||
"cache/local-vector",
|
||
"cache/local-vector-25",
|
||
"cache/cache-only.setting",
|
||
f"cache/{CACHE_COVERAGE_INDEX_NAME}",
|
||
"data/map-library",
|
||
"data/map-library-25",
|
||
],
|
||
"generatedAt": now,
|
||
}
|
||
self._cache_admin_stats_result = result
|
||
self._cache_admin_stats_cached_at = now
|
||
current = dict(result)
|
||
current["activeJobs"] = self.active_manual_download_count()
|
||
return current
|
||
|
||
def refresh_stale(self, force: bool = False, limit: int | None = None) -> int:
|
||
return self.refresh_cache_categories(
|
||
["raster", "vector", "rail", "environment"],
|
||
force=force,
|
||
limit=limit,
|
||
)
|
||
|
||
def refresh_cache_categories(
|
||
self,
|
||
category_ids: list[str],
|
||
force: bool = False,
|
||
limit: int | None = None,
|
||
) -> int:
|
||
if self.config.cache_only:
|
||
return 0
|
||
categories = list(dict.fromkeys(category_ids))
|
||
category_info = {
|
||
category_id: self.cache_admin_category_info(category_id)
|
||
for category_id in categories
|
||
}
|
||
invalid = [item for item, info in category_info.items() if info is None]
|
||
if invalid:
|
||
raise ValueError("Unknown cache categories: " + ", ".join(invalid))
|
||
blocked = [
|
||
item for item in categories
|
||
if not category_info[item]["refreshable"]
|
||
]
|
||
if blocked:
|
||
raise ValueError("Cache categories cannot be refreshed in place: " + ", ".join(blocked))
|
||
max_items = limit or self.config.refresh_batch
|
||
scheduled = 0
|
||
now = time.time()
|
||
path_groups = self.cache_category_paths()
|
||
|
||
for category_id in categories:
|
||
info = category_info[category_id]
|
||
if info is None:
|
||
continue
|
||
if category_id in CACHE_ADMIN_QUERY_LAYERS:
|
||
category_scheduled = 0
|
||
for zoom, x, y in self.coverage_tiles_for_category(category_id):
|
||
if scheduled >= max_items:
|
||
break
|
||
path = self.vector_tile_cache_path(zoom, x, y)
|
||
try:
|
||
stale = force or not path.exists() or now - path.stat().st_mtime > self.config.tile_ttl_seconds
|
||
except OSError:
|
||
stale = force
|
||
if not stale:
|
||
continue
|
||
self._download_executor.submit(self.refresh_vector_tile, zoom, x, y)
|
||
scheduled += 1
|
||
category_scheduled += 1
|
||
if category_scheduled:
|
||
self.invalidate_query_layer_responses(category_id)
|
||
continue
|
||
parent_id = str(info.get("parent", category_id))
|
||
layer_key = info.get("key")
|
||
if scheduled >= max_items:
|
||
break
|
||
for root in path_groups[parent_id]:
|
||
if scheduled >= max_items:
|
||
break
|
||
if not root.exists():
|
||
continue
|
||
if parent_id == "raster":
|
||
for path in root.rglob("*.png"):
|
||
if scheduled >= max_items:
|
||
break
|
||
try:
|
||
stale = force or now - path.stat().st_mtime > self.config.tile_ttl_seconds
|
||
except OSError:
|
||
continue
|
||
if stale:
|
||
self.schedule_tile_refresh(path)
|
||
scheduled += 1
|
||
elif parent_id == "vector":
|
||
for path in root.rglob("*.pbf"):
|
||
if scheduled >= max_items:
|
||
break
|
||
try:
|
||
stale = force or now - path.stat().st_mtime > self.config.tile_ttl_seconds
|
||
relative = path.relative_to(self.vector_cache_dir).parts
|
||
if len(relative) != 3:
|
||
continue
|
||
z = int(relative[0])
|
||
x = int(relative[1])
|
||
y = int(relative[2][:-4])
|
||
except (OSError, ValueError):
|
||
continue
|
||
if stale:
|
||
self._download_executor.submit(self.refresh_vector_tile, z, x, y)
|
||
scheduled += 1
|
||
elif parent_id in ("rail", "environment"):
|
||
refresh_path = self.refresh_rail_path if parent_id == "rail" else self.refresh_water_path
|
||
ttl = self.config.rail_ttl_seconds
|
||
for path in root.glob("*.json"):
|
||
if scheduled >= max_items:
|
||
break
|
||
if layer_key and not self.response_layer_has_features(parent_id, str(layer_key), path):
|
||
continue
|
||
try:
|
||
stale = force or now - path.stat().st_mtime > ttl
|
||
except OSError:
|
||
continue
|
||
if not stale:
|
||
continue
|
||
key = str(path)
|
||
with self._lock_guard:
|
||
if key in self._refreshing:
|
||
continue
|
||
self._refreshing.add(key)
|
||
self._refresh_executor.submit(refresh_path, path)
|
||
scheduled += 1
|
||
return scheduled
|
||
|
||
def delete_cache_categories(self, category_ids: list[str]) -> dict[str, Any]:
|
||
categories = list(dict.fromkeys(category_ids))
|
||
category_info = {
|
||
category_id: self.cache_admin_category_info(category_id)
|
||
for category_id in categories
|
||
}
|
||
invalid = [item for item, info in category_info.items() if info is None]
|
||
if invalid:
|
||
raise ValueError("Unknown cache categories: " + ", ".join(invalid))
|
||
if not categories:
|
||
raise ValueError("Select at least one cache category")
|
||
if self.active_manual_download_count():
|
||
raise RuntimeError("存在正在运行的下载任务，请先完成或中断任务后再删除缓存")
|
||
|
||
path_groups = self.cache_category_paths()
|
||
targets: list[Path] = []
|
||
response_layers: list[tuple[str, str]] = []
|
||
seen: set[str] = set()
|
||
for category_id in categories:
|
||
info = category_info[category_id]
|
||
if info is None:
|
||
continue
|
||
parent_id = str(info.get("parent", category_id))
|
||
layer_key = info.get("key")
|
||
if parent_id in ("rail", "environment") and parent_id != category_id and layer_key:
|
||
response_layers.append((parent_id, str(layer_key)))
|
||
continue
|
||
for path in path_groups[parent_id]:
|
||
resolved = path.resolve()
|
||
if resolved == self.cache_dir or self.cache_dir not in resolved.parents:
|
||
raise ValueError("Refusing to delete a path outside the managed cache")
|
||
key = str(resolved)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
targets.append(resolved)
|
||
|
||
before = self.summarize_cache_paths(targets)
|
||
response_targets = {
|
||
path
|
||
for parent_id, layer_key in response_layers
|
||
for root in path_groups[parent_id]
|
||
if root.exists()
|
||
for path in root.glob("*.json")
|
||
if self.response_layer_has_features(parent_id, layer_key, path)
|
||
}
|
||
response_before = self.summarize_cache_paths(list(response_targets))
|
||
deleted_layer_files = 0
|
||
deleted_layer_bytes = 0
|
||
for parent_id, layer_key in response_layers:
|
||
for path in response_targets:
|
||
changed, deleted_bytes = self.delete_response_layer_file(parent_id, layer_key, path)
|
||
deleted_layer_files += changed
|
||
deleted_layer_bytes += deleted_bytes
|
||
for target in targets:
|
||
if target.exists():
|
||
if target.is_dir():
|
||
shutil.rmtree(target)
|
||
else:
|
||
target.unlink()
|
||
target.mkdir(parents=True, exist_ok=True)
|
||
self.invalidate_cache_admin_stats()
|
||
return {
|
||
"ok": True,
|
||
"categories": categories,
|
||
"deletedFiles": before["files"] + deleted_layer_files,
|
||
"deletedBytes": before["bytes"] + deleted_layer_bytes,
|
||
"targets": [str(path.relative_to(self.cache_dir)) for path in targets],
|
||
"responseFilesTouched": response_before["files"],
|
||
}
|
||
|
||
def refresh_cache_scope(
|
||
self,
|
||
category_ids: list[str],
|
||
bounds: tuple[float, float, float, float],
|
||
force: bool = False,
|
||
) -> dict[str, Any]:
|
||
if self.config.cache_only:
|
||
return {
|
||
"scheduled": 0,
|
||
"matchedFiles": 0,
|
||
"skippedCategories": [],
|
||
"bounds": list(bounds),
|
||
}
|
||
categories = list(dict.fromkeys(category_ids))
|
||
category_info = {
|
||
category_id: self.cache_admin_category_info(category_id)
|
||
for category_id in categories
|
||
}
|
||
invalid = [item for item, info in category_info.items() if info is None]
|
||
if invalid:
|
||
raise ValueError("Unknown cache categories: " + ", ".join(invalid))
|
||
if not categories:
|
||
raise ValueError("Select at least one cache category")
|
||
blocked = [
|
||
item for item in categories
|
||
if not category_info[item]["refreshable"]
|
||
]
|
||
if blocked:
|
||
raise ValueError("Cache categories cannot be refreshed in place: " + ", ".join(blocked))
|
||
|
||
parent_categories = list(dict.fromkeys(
|
||
str(category_info[category_id].get("parent", category_id))
|
||
for category_id in categories
|
||
))
|
||
matches, skipped = self.cache_files_in_bounds(parent_categories, bounds)
|
||
scheduled = 0
|
||
matched_files = 0
|
||
matched_tiles = 0
|
||
now = time.time()
|
||
for category_id in categories:
|
||
info = category_info[category_id]
|
||
if info is None:
|
||
continue
|
||
if category_id in CACHE_ADMIN_QUERY_LAYERS:
|
||
source_tiles = self.coverage_tiles_for_category(category_id, bounds)
|
||
matched_tiles += len(source_tiles)
|
||
category_scheduled = 0
|
||
for zoom, x, y in source_tiles:
|
||
path = self.vector_tile_cache_path(zoom, x, y)
|
||
try:
|
||
stale = force or not path.exists() or now - path.stat().st_mtime > self.config.tile_ttl_seconds
|
||
except OSError:
|
||
stale = force
|
||
if not stale:
|
||
continue
|
||
self._download_executor.submit(self.refresh_vector_tile, zoom, x, y)
|
||
scheduled += 1
|
||
category_scheduled += 1
|
||
if category_scheduled:
|
||
self.invalidate_query_layer_responses(category_id, bounds)
|
||
continue
|
||
parent_id = str(info.get("parent", category_id))
|
||
layer_key = info.get("key")
|
||
paths = matches.get(parent_id, [])
|
||
for path in paths:
|
||
if layer_key and parent_id in ("rail", "environment") and not self.response_layer_has_features(parent_id, str(layer_key), path):
|
||
continue
|
||
matched_files += 1
|
||
try:
|
||
ttl = (
|
||
self.config.rail_ttl_seconds
|
||
if parent_id in ("rail", "environment")
|
||
else self.config.tile_ttl_seconds
|
||
)
|
||
stale = force or now - path.stat().st_mtime > ttl
|
||
except OSError:
|
||
continue
|
||
if not stale:
|
||
continue
|
||
if parent_id == "raster":
|
||
self.schedule_tile_refresh(path)
|
||
scheduled += 1
|
||
elif parent_id == "vector":
|
||
coordinate = self.cache_tile_coordinate(parent_id, path, self.vector_cache_dir)
|
||
if coordinate is None:
|
||
continue
|
||
zoom, x, y = coordinate
|
||
self._download_executor.submit(self.refresh_vector_tile, zoom, x, y)
|
||
scheduled += 1
|
||
elif parent_id in ("rail", "environment"):
|
||
refresh_path = self.refresh_rail_path if parent_id == "rail" else self.refresh_water_path
|
||
key = str(path)
|
||
with self._lock_guard:
|
||
if key in self._refreshing:
|
||
continue
|
||
self._refreshing.add(key)
|
||
self._refresh_executor.submit(refresh_path, path)
|
||
scheduled += 1
|
||
return {
|
||
"scheduled": scheduled,
|
||
"matchedFiles": matched_files,
|
||
"matchedTiles": matched_tiles + matched_files,
|
||
"skippedCategories": skipped,
|
||
"bounds": list(bounds),
|
||
}
|
||
|
||
def delete_cache_scope(
|
||
self,
|
||
category_ids: list[str],
|
||
bounds: tuple[float, float, float, float],
|
||
) -> dict[str, Any]:
|
||
categories = list(dict.fromkeys(category_ids))
|
||
category_info = {
|
||
category_id: self.cache_admin_category_info(category_id)
|
||
for category_id in categories
|
||
}
|
||
invalid = [item for item, info in category_info.items() if info is None]
|
||
if invalid:
|
||
raise ValueError("Unknown cache categories: " + ", ".join(invalid))
|
||
if not categories:
|
||
raise ValueError("Select at least one cache category")
|
||
if self.active_manual_download_count():
|
||
raise RuntimeError("存在正在运行的下载任务，请先完成或中断任务后再删除缓存")
|
||
|
||
parent_categories = list(dict.fromkeys(
|
||
str(category_info[category_id].get("parent", category_id))
|
||
for category_id in categories
|
||
))
|
||
matches, skipped = self.cache_files_in_bounds(parent_categories, bounds)
|
||
path_groups = self.cache_category_paths()
|
||
deleted_files = 0
|
||
deleted_bytes = 0
|
||
for category_id in categories:
|
||
info = category_info[category_id]
|
||
if info is None:
|
||
continue
|
||
parent_id = str(info.get("parent", category_id))
|
||
layer_key = info.get("key")
|
||
paths = matches.get(parent_id, [])
|
||
for path in paths:
|
||
resolved = path.resolve()
|
||
if self.cache_dir not in resolved.parents:
|
||
raise ValueError("Refusing to delete a path outside the managed cache")
|
||
if layer_key and parent_id in ("rail", "environment"):
|
||
if not self.response_layer_has_features(parent_id, str(layer_key), resolved):
|
||
continue
|
||
changed, released_bytes = self.delete_response_layer_file(
|
||
parent_id, str(layer_key), resolved
|
||
)
|
||
deleted_files += changed
|
||
deleted_bytes += released_bytes
|
||
continue
|
||
try:
|
||
size = resolved.stat().st_size
|
||
resolved.unlink()
|
||
except FileNotFoundError:
|
||
continue
|
||
except OSError:
|
||
continue
|
||
deleted_files += 1
|
||
deleted_bytes += size
|
||
for root in path_groups[parent_id]:
|
||
if not root.is_dir():
|
||
continue
|
||
for path in paths:
|
||
parent = path.parent
|
||
while parent != root and root in parent.parents:
|
||
try:
|
||
parent.rmdir()
|
||
except OSError:
|
||
break
|
||
parent = parent.parent
|
||
self.invalidate_cache_admin_stats()
|
||
return {
|
||
"ok": True,
|
||
"categories": categories,
|
||
"deletedFiles": deleted_files,
|
||
"deletedBytes": deleted_bytes,
|
||
"skippedCategories": skipped,
|
||
"bounded": True,
|
||
"bounds": list(bounds),
|
||
}
|
||
|
||
def raster_tile_coordinates(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
z: int,
|
||
limit: int = MAX_REFRESH_TILES,
|
||
) -> list[tuple[int, int]]:
|
||
south, west, north, east = bbox
|
||
x0 = self.lon_to_tile_x(west, z)
|
||
x1 = self.lon_to_tile_x(east, z)
|
||
y0 = self.lat_to_tile_y(north, z)
|
||
y1 = self.lat_to_tile_y(south, z)
|
||
count = (x1 - x0 + 1) * (y1 - y0 + 1)
|
||
if count > limit:
|
||
raise ValueError(f"Tile request too large: {count} tiles")
|
||
return [(x, y) for x in range(x0, x1 + 1) for y in range(y0, y1 + 1)]
|
||
|
||
def refresh_vector_tile(self, z: int, x: int, y: int) -> bool:
|
||
if self.config.cache_only:
|
||
return False
|
||
key = f"vector:{z}:{x}:{y}"
|
||
with self._lock_guard:
|
||
if key in self._refreshing:
|
||
return False
|
||
self._refreshing.add(key)
|
||
try:
|
||
self.fetch_vector_tile(z, x, y, force=True)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
finally:
|
||
with self._lock_guard:
|
||
self._refreshing.discard(key)
|
||
|
||
def refresh_current_view(
|
||
self,
|
||
bbox: tuple[float, float, float, float],
|
||
map_zoom: int,
|
||
force: bool = True,
|
||
) -> int:
|
||
if self.config.cache_only:
|
||
return 0
|
||
scheduled = 0
|
||
zoom = max(0, min(18, int(map_zoom)))
|
||
active_providers = [
|
||
(provider, provider_config)
|
||
for provider, provider_config in TILE_PROVIDERS.items()
|
||
if not provider_config.get("manual_only")
|
||
]
|
||
raster_coordinates = self.raster_tile_coordinates(bbox, zoom)
|
||
if len(raster_coordinates) * len(active_providers) > MAX_REFRESH_TILES:
|
||
raise ValueError("Current view is too large to refresh; zoom in")
|
||
for provider, provider_config in active_providers:
|
||
if provider_config.get("manual_only"):
|
||
continue
|
||
for x, y in raster_coordinates:
|
||
path = self.tile_cache_path(provider, zoom, x, y)
|
||
if path.exists():
|
||
self.schedule_tile_refresh(path)
|
||
scheduled += 1
|
||
elif force:
|
||
self._download_executor.submit(
|
||
self.refresh_tile_path,
|
||
self.tile_cache_path(provider, zoom, x, y),
|
||
)
|
||
scheduled += 1
|
||
vector_zoom = self.choose_vector_zoom(
|
||
bbox,
|
||
preferred=min(14, max(7, zoom + 2)),
|
||
max_tiles=MAX_REFRESH_TILES,
|
||
)
|
||
vector_coordinates = self.vector_tile_coordinates(bbox, vector_zoom)
|
||
if len(vector_coordinates) > MAX_REFRESH_TILES:
|
||
raise ValueError("Current view has too many vector tiles to refresh")
|
||
for x, y in vector_coordinates:
|
||
self._download_executor.submit(self.refresh_vector_tile, vector_zoom, x, y)
|
||
scheduled += 1
|
||
return scheduled
|
||
|
||
def cache_totals(self) -> tuple[int, int]:
|
||
now = time.time()
|
||
with self._cache_totals_lock:
|
||
if now - self._cache_totals_cached_at < 15:
|
||
return self._cache_totals_result
|
||
files = 0
|
||
total_bytes = 0
|
||
for path in self.cache_dir.rglob("*"):
|
||
if not path.is_file():
|
||
continue
|
||
files += 1
|
||
try:
|
||
total_bytes += path.stat().st_size
|
||
except OSError:
|
||
pass
|
||
self._cache_totals_result = (files, total_bytes)
|
||
self._cache_totals_cached_at = now
|
||
return self._cache_totals_result
|
||
|
||
def status(self) -> dict[str, Any]:
|
||
files, total_bytes = self.cache_totals()
|
||
with self.stats_lock:
|
||
stats = dict(self.stats)
|
||
pending_files = len(list(self.pending_rail_dir.glob("*.json")))
|
||
pending_water_files = len(list(self.pending_water_dir.glob("*.json")))
|
||
return {
|
||
"ok": True,
|
||
"uptimeSeconds": round(time.time() - self.started_at, 1),
|
||
"cacheOnly": self.config.cache_only,
|
||
"cacheDir": "cache",
|
||
"cacheFiles": files,
|
||
"cacheBytes": total_bytes,
|
||
"tileTtlSeconds": self.config.tile_ttl_seconds,
|
||
"railTtlSeconds": self.config.rail_ttl_seconds,
|
||
"refreshSeconds": self.config.refresh_seconds,
|
||
"pendingRailRequests": pending_files,
|
||
"pendingWaterRequests": pending_water_files,
|
||
"stats": stats,
|
||
"providers": sorted(TILE_PROVIDERS),
|
||
"environment": {
|
||
"minZoom": ENVIRONMENT_MIN_ZOOM,
|
||
"maxZoom": ENVIRONMENT_MAX_ZOOM,
|
||
"sourceMaxZoom": OPENFREEMAP_SOURCE_MAX_ZOOM,
|
||
"vegetationSourceMinZoom": 7,
|
||
"layers": ["water", "waterway", "landcover", "landuse", "poi"],
|
||
},
|
||
"tiandituConfigured": bool(self.config.tianditu_key),
|
||
"manualOnlyProviders": sorted(self.tianditu_provider_names()),
|
||
"downloadStagingDir": DOWNLOAD_STAGING_DIR_NAME,
|
||
"workers": {
|
||
"cacheRefresh": CACHE_REFRESH_WORKERS,
|
||
"vectorBackground": VECTOR_BACKGROUND_WORKERS,
|
||
"vectorTileRequest": VECTOR_TILE_REQUEST_WORKERS,
|
||
"railAggregate": RAIL_AGGREGATE_WORKERS,
|
||
"railAggregateFetch": RAIL_AGGREGATE_FETCH_WORKERS,
|
||
"tiandituManual": TIANDITU_DOWNLOAD_WORKERS,
|
||
"openmapManual": OPENMAP_DOWNLOAD_WORKERS,
|
||
},
|
||
}
|
||
|
||
def _maintenance_loop(self) -> None:
|
||
last_refresh = time.time()
|
||
while not self._stop.wait(10):
|
||
self.retry_pending_rail_jobs(limit=min(8, self.config.refresh_batch))
|
||
self.retry_pending_water_jobs(limit=min(8, self.config.refresh_batch))
|
||
self.expire_stalled_openmap_jobs()
|
||
now = time.time()
|
||
if now - last_refresh >= self.config.refresh_seconds:
|
||
self.refresh_stale(force=False, limit=self.config.refresh_batch)
|
||
last_refresh = now
|
||
|
||
def close(self) -> None:
|
||
self._stop.set()
|
||
self._refresh_executor.shutdown(wait=False, cancel_futures=True)
|
||
self._download_executor.shutdown(wait=False, cancel_futures=True)
|
||
self._vector_tile_executor.shutdown(wait=False, cancel_futures=True)
|
||
self._rail_aggregate_executor.shutdown(wait=False, cancel_futures=True)
|
||
self._rail_aggregate_fetch_executor.shutdown(wait=False, cancel_futures=True)
|
||
self._tianditu_executor.shutdown(wait=False, cancel_futures=True)
|
||
self._openmap_executor.shutdown(wait=False, cancel_futures=True)
|
||
|
||
|
||
SERVICE: MapService | None = None
|
||
|
||
|
||
class QuietThreadingHTTPServer(ThreadingHTTPServer):
|
||
daemon_threads = True
|
||
block_on_close = False
|
||
request_queue_size = 128
|
||
|
||
def handle_error(self, request: Any, client_address: Any) -> None:
|
||
error = sys.exc_info()[1]
|
||
if isinstance(error, (BrokenPipeError, ConnectionResetError, ConnectionAbortedError)):
|
||
return
|
||
super().handle_error(request, client_address)
|
||
|
||
|
||
class MapStudioRequestHandler(BaseHTTPRequestHandler):
|
||
server_version = "MapStudioMapService/1.0"
|
||
|
||
def _cors_headers(self) -> None:
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header("Access-Control-Allow-Methods", "GET, HEAD, POST, OPTIONS")
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type, Accept")
|
||
|
||
def _security_headers(self) -> None:
|
||
self.send_header("X-Content-Type-Options", "nosniff")
|
||
self.send_header("Referrer-Policy", "no-referrer")
|
||
self.send_header("X-Frame-Options", "SAMEORIGIN")
|
||
self.send_header("X-Robots-Tag", "noindex, nofollow")
|
||
self.send_header("Permissions-Policy", "camera=(), microphone=(), geolocation=(self)")
|
||
|
||
def _client_identity(self) -> str:
|
||
return str(self.client_address[0])
|
||
|
||
def _allow_request(
|
||
self,
|
||
bucket: str,
|
||
limit: int,
|
||
window_seconds: float,
|
||
) -> bool:
|
||
if SERVICE.consume_rate_limit(
|
||
bucket,
|
||
self._client_identity(),
|
||
limit,
|
||
window_seconds,
|
||
):
|
||
return True
|
||
self._send_json(
|
||
HTTPStatus.TOO_MANY_REQUESTS,
|
||
{"ok": False, "error": "Too many requests; retry later"},
|
||
extra_headers={"Retry-After": str(max(1, int(window_seconds)))},
|
||
)
|
||
return False
|
||
|
||
def _request_prefix(self) -> str:
|
||
raw_path = urllib.parse.urlsplit(self.path).path
|
||
marker = "/api/"
|
||
if raw_path.startswith(marker):
|
||
return "/"
|
||
index = raw_path.find(marker)
|
||
if index >= 0:
|
||
prefix = raw_path[:index]
|
||
return prefix.rstrip("/") + "/" if prefix else "/"
|
||
return "/"
|
||
|
||
def _send_json(
|
||
self,
|
||
status: int,
|
||
payload: dict[str, Any],
|
||
cache_state: str | None = None,
|
||
extra_headers: dict[str, str] | list[tuple[str, str]] | None = None,
|
||
) -> None:
|
||
data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||
use_gzip = len(data) > 1024 and "gzip" in self.headers.get("Accept-Encoding", "").lower()
|
||
body = gzip.compress(data, compresslevel=6, mtime=0) if use_gzip else data
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
if use_gzip:
|
||
self.send_header("Content-Encoding", "gzip")
|
||
self.send_header("Vary", "Accept-Encoding")
|
||
self.send_header(
|
||
"Cache-Control",
|
||
LONG_BROWSER_CACHE
|
||
if status == HTTPStatus.OK and cache_state in ("HIT", "MISS")
|
||
else "public, max-age=86400" if cache_state == "STALE" else "no-store",
|
||
)
|
||
self._cors_headers()
|
||
self._security_headers()
|
||
if cache_state:
|
||
self.send_header("X-Cache", cache_state)
|
||
header_items = (
|
||
extra_headers.items()
|
||
if isinstance(extra_headers, dict)
|
||
else (extra_headers or [])
|
||
)
|
||
for name, value in header_items:
|
||
self.send_header(name, value)
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(body)
|
||
|
||
def _require_admin(self) -> bool:
|
||
password = SERVICE.config.admin_password
|
||
if not password:
|
||
self._send_json(
|
||
HTTPStatus.SERVICE_UNAVAILABLE,
|
||
{
|
||
"ok": False,
|
||
"code": "ADMIN_PASSWORD_NOT_CONFIGURED",
|
||
"error": "地图下载后台密码尚未配置",
|
||
},
|
||
)
|
||
return False
|
||
try:
|
||
cookie_jar = cookies.SimpleCookie(self.headers.get("Cookie", ""))
|
||
session_token = cookie_jar.get("mapstudio_admin")
|
||
if session_token and SERVICE.admin_session_valid(session_token.value):
|
||
return True
|
||
except Exception:
|
||
pass
|
||
authorization = self.headers.get("Authorization", "")
|
||
valid = False
|
||
if authorization.startswith("Basic "):
|
||
if not self._allow_request("admin-basic-auth", 20, 600):
|
||
return False
|
||
try:
|
||
decoded = base64.b64decode(authorization[6:], validate=True).decode("utf-8")
|
||
username, supplied = decoded.split(":", 1)
|
||
valid = username == "admin" and hmac.compare_digest(supplied, password)
|
||
except Exception:
|
||
valid = False
|
||
if valid:
|
||
return True
|
||
self._send_json(
|
||
HTTPStatus.UNAUTHORIZED,
|
||
{"ok": False, "code": "ADMIN_AUTH_REQUIRED", "error": "需要管理员密码"},
|
||
)
|
||
return False
|
||
|
||
def _send_bytes(self, status: int, data: bytes, content_type: str, cache_state: str | None = None) -> None:
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.send_header(
|
||
"Cache-Control",
|
||
LONG_BROWSER_CACHE
|
||
if cache_state in ("HIT", "MISS")
|
||
else "public, max-age=86400" if cache_state == "STALE" else "no-store",
|
||
)
|
||
self._cors_headers()
|
||
self._security_headers()
|
||
if cache_state:
|
||
self.send_header("X-Cache", cache_state)
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(data)
|
||
|
||
def do_OPTIONS(self) -> None:
|
||
self.send_response(HTTPStatus.NO_CONTENT)
|
||
self._cors_headers()
|
||
self._security_headers()
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
def do_HEAD(self) -> None:
|
||
self.do_GET()
|
||
|
||
def do_GET(self) -> None:
|
||
parsed = urllib.parse.urlsplit(self.path)
|
||
path = normalize_request_path(parsed.path, self.headers.get("Accept", ""))
|
||
if (
|
||
path in ("/api/cache-policy", "/api/upstream-proxy")
|
||
or path.startswith(("/api/tianditu/", "/api/openmap/", "/api/cache-admin/"))
|
||
):
|
||
if not self._require_admin():
|
||
return
|
||
if path == "/api/status":
|
||
if not self._allow_request("status", 60, 60):
|
||
return
|
||
self._send_json(HTTPStatus.OK, SERVICE.status())
|
||
return
|
||
if path == "/api/cache-admin/status":
|
||
if not self._allow_request("cache-admin-status", 20, 60):
|
||
return
|
||
refresh = urllib.parse.parse_qs(parsed.query).get("refresh", ["0"])[0] == "1"
|
||
self._send_json(HTTPStatus.OK, SERVICE.cache_admin_status(refresh=refresh))
|
||
return
|
||
if path == "/api/cache-admin/coverage":
|
||
if not self._allow_request("cache-admin-coverage", 10, 60):
|
||
return
|
||
refresh = urllib.parse.parse_qs(parsed.query).get("refresh", ["0"])[0] == "1"
|
||
self._send_json(HTTPStatus.OK, SERVICE.cache_coverage(refresh=refresh))
|
||
return
|
||
if path == "/api/cache-policy":
|
||
if not self._allow_request("cache-policy", 60, 60):
|
||
return
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
{
|
||
"ok": True,
|
||
"cacheOnly": SERVICE.config.cache_only,
|
||
"placeholderTiles": True,
|
||
},
|
||
)
|
||
return
|
||
if path == "/api/upstream-proxy":
|
||
if not self._allow_request("upstream-proxy", 60, 60):
|
||
return
|
||
proxy = SERVICE.upstream_proxy_status()
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
{
|
||
"ok": True,
|
||
"upstreamProxyConfigured": proxy["configured"],
|
||
"upstreamProxy": proxy["proxy"],
|
||
"upstreamProxyDefaultConfigured": proxy["defaultConfigured"],
|
||
"upstreamProxyDefault": proxy["defaultProxy"],
|
||
"upstreamProxyTemporary": proxy["temporary"],
|
||
},
|
||
)
|
||
return
|
||
if path == "/api/local-vector/status":
|
||
self._send_json(HTTPStatus.OK, SERVICE.local_vector_status())
|
||
return
|
||
if path == "/api/local-vector-25/status":
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
SERVICE.local_vector_status(SERVICE.local_vector_25_dir),
|
||
)
|
||
return
|
||
local_vector_match = LOCAL_VECTOR_PATH_RE.match(path)
|
||
if local_vector_match:
|
||
if not self._allow_request("local-vector", 300, 60):
|
||
return
|
||
z, x, y = (int(value) for value in local_vector_match.groups())
|
||
tile = SERVICE.local_vector_tile(z, x, y)
|
||
if tile is None:
|
||
self._send_json(
|
||
HTTPStatus.NOT_FOUND,
|
||
{"ok": False, "error": "Local vector tile not found"},
|
||
)
|
||
else:
|
||
self._send_json(HTTPStatus.OK, tile, "HIT")
|
||
return
|
||
local_vector_25_match = LOCAL_VECTOR_25_PATH_RE.match(path)
|
||
if local_vector_25_match:
|
||
if not self._allow_request("local-vector-25", 300, 60):
|
||
return
|
||
z, x, y = (int(value) for value in local_vector_25_match.groups())
|
||
tile = SERVICE.local_vector_tile(
|
||
z,
|
||
x,
|
||
y,
|
||
SERVICE.local_vector_25_dir,
|
||
)
|
||
if tile is None:
|
||
self._send_json(
|
||
HTTPStatus.NOT_FOUND,
|
||
{"ok": False, "error": "Local 1:250,000 vector tile not found"},
|
||
)
|
||
else:
|
||
self._send_json(HTTPStatus.OK, tile, "HIT")
|
||
return
|
||
rail_tile_match = RAIL_TILE_PATH_RE.match(path)
|
||
if rail_tile_match:
|
||
if not self._allow_request("rail-tiles", 300, 60):
|
||
return
|
||
z, x, y = (int(value) for value in rail_tile_match.groups())
|
||
force = urllib.parse.parse_qs(parsed.query).get("refresh", ["0"])[0] == "1"
|
||
include = {
|
||
value
|
||
for value in urllib.parse.parse_qs(parsed.query).get(
|
||
"include", ["rail,metro"]
|
||
)[0].split(",")
|
||
if value in ("rail", "metro")
|
||
}
|
||
tile = SERVICE.rail_tile(z, x, y, force=force, include=include)
|
||
if tile is None:
|
||
if SERVICE.config.cache_only:
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
{
|
||
"ok": True,
|
||
"format": "rail-tile-v1",
|
||
"source": "cache-only",
|
||
"cacheOnly": True,
|
||
"placeholder": True,
|
||
"tileZoom": z,
|
||
"x": x,
|
||
"y": y,
|
||
"include": sorted(include),
|
||
"elements": [],
|
||
},
|
||
"PLACEHOLDER",
|
||
)
|
||
else:
|
||
self._send_json(
|
||
HTTPStatus.BAD_GATEWAY,
|
||
{"ok": False, "error": "Rail tile unavailable"},
|
||
)
|
||
else:
|
||
self._send_json(HTTPStatus.OK, tile, "HIT" if not force else "MISS")
|
||
return
|
||
if path == "/api/tianditu/status":
|
||
if not self._allow_request("tianditu-status", 180, 60):
|
||
return
|
||
self._send_json(HTTPStatus.OK, SERVICE.tianditu_status())
|
||
return
|
||
if path == "/api/tianditu/job":
|
||
if not self._allow_request("tianditu-status", 180, 60):
|
||
return
|
||
job_id = urllib.parse.parse_qs(parsed.query).get("id", [""])[0]
|
||
job = SERVICE.tianditu_job_status(job_id)
|
||
if not job:
|
||
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "Job not found"})
|
||
else:
|
||
self._send_json(HTTPStatus.OK, {"ok": True, "job": job})
|
||
return
|
||
if path == "/api/openmap/status":
|
||
if not self._allow_request("openmap-status", 180, 60):
|
||
return
|
||
self._send_json(HTTPStatus.OK, SERVICE.openmap_status())
|
||
return
|
||
if path == "/api/openmap/job":
|
||
if not self._allow_request("openmap-status", 180, 60):
|
||
return
|
||
job_id = urllib.parse.parse_qs(parsed.query).get("id", [""])[0]
|
||
job = SERVICE.openmap_job_status(job_id)
|
||
if not job:
|
||
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "Job not found"})
|
||
else:
|
||
self._send_json(HTTPStatus.OK, {"ok": True, "job": job})
|
||
return
|
||
if path == "/api/rail/status":
|
||
key = urllib.parse.parse_qs(parsed.query).get("key", [""])[0]
|
||
if not re.fullmatch(r"[a-f0-9]{40}", key):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid cache key"})
|
||
return
|
||
status = SERVICE.pending_rail_status(key)
|
||
self._send_json(
|
||
HTTPStatus.OK if status.get("ok") else HTTPStatus.NOT_FOUND,
|
||
status,
|
||
)
|
||
return
|
||
if path == "/api/water/status":
|
||
key = urllib.parse.parse_qs(parsed.query).get("key", [""])[0]
|
||
if not re.fullmatch(r"[a-f0-9]{40}", key):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid cache key"})
|
||
return
|
||
status = SERVICE.water_pending_status(key)
|
||
self._send_json(
|
||
HTTPStatus.OK if status.get("ok") else HTTPStatus.NOT_FOUND,
|
||
status,
|
||
)
|
||
return
|
||
if path == "/api/rail":
|
||
if not self._allow_request("rail-api", 40, 60):
|
||
return
|
||
self._handle_rail(parsed.query)
|
||
return
|
||
if path == "/api/water":
|
||
if not self._allow_request("water-api", 60, 60):
|
||
return
|
||
self._handle_water(parsed.query)
|
||
return
|
||
if path == "/api/boundaries":
|
||
if not self._allow_request("boundary-api", 60, 60):
|
||
return
|
||
self._handle_boundaries(parsed.query)
|
||
return
|
||
tile_match = TILE_PATH_RE.match(path)
|
||
if tile_match:
|
||
provider, z_text, x_text, y_text = tile_match.groups()
|
||
z, x, y = int(z_text), int(x_text), int(y_text)
|
||
if z < 0 or z > 22 or x < 0 or y < 0 or x >= 2**z or y >= 2**z:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid tile coordinates"})
|
||
return
|
||
try:
|
||
data, content_type, cache_state = SERVICE.get_tile(provider, z, x, y)
|
||
self._send_bytes(HTTPStatus.OK, data, content_type, cache_state)
|
||
except TileNotCached as error:
|
||
self._send_json(
|
||
HTTPStatus.NOT_FOUND,
|
||
{
|
||
"ok": False,
|
||
"code": "TILE_NOT_DOWNLOADED",
|
||
"error": "Tile is not available in the local manual cache",
|
||
"detail": str(error),
|
||
},
|
||
)
|
||
except TiandituKeyMissing as error:
|
||
self._send_json(
|
||
HTTPStatus.CONFLICT,
|
||
{
|
||
"ok": False,
|
||
"code": "TIANDITU_KEY_MISSING",
|
||
"error": "Tianditu API key is not configured",
|
||
"detail": str(error),
|
||
},
|
||
)
|
||
except KeyError:
|
||
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "Unknown tile provider"})
|
||
except Exception as error:
|
||
self._send_json(
|
||
HTTPStatus.BAD_GATEWAY,
|
||
{"ok": False, "error": "Tile unavailable", "detail": str(error)},
|
||
)
|
||
return
|
||
self._serve_static(path)
|
||
|
||
def do_POST(self) -> None:
|
||
parsed = urllib.parse.urlsplit(self.path)
|
||
path = normalize_request_path(parsed.path, self.headers.get("Accept", ""))
|
||
if path == "/api/admin/login":
|
||
if not self._allow_request("admin-login", 8, 600):
|
||
return
|
||
password = SERVICE.config.admin_password
|
||
if not password:
|
||
self._send_json(
|
||
HTTPStatus.SERVICE_UNAVAILABLE,
|
||
{
|
||
"ok": False,
|
||
"code": "ADMIN_PASSWORD_NOT_CONFIGURED",
|
||
"error": "地图下载后台密码尚未配置",
|
||
},
|
||
)
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length <= 0 or length > 16 * 1024:
|
||
raise ValueError("Invalid request body")
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("Invalid login request")
|
||
supplied = str(payload.get("password", ""))
|
||
except (ValueError, TypeError, json.JSONDecodeError):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid login request"})
|
||
return
|
||
if not hmac.compare_digest(supplied, password):
|
||
self._send_json(
|
||
HTTPStatus.UNAUTHORIZED,
|
||
{"ok": False, "code": "ADMIN_AUTH_REQUIRED", "error": "管理员密码错误"},
|
||
)
|
||
return
|
||
token = SERVICE.create_admin_session()
|
||
cookie_path = self._request_prefix()
|
||
secure_cookie = (
|
||
"; Secure"
|
||
if self.headers.get("X-Forwarded-Proto", "").lower() == "https"
|
||
or self.headers.get("X-Forwarded-Ssl", "").lower() == "on"
|
||
else ""
|
||
)
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
{"ok": True},
|
||
extra_headers={
|
||
"Set-Cookie": (
|
||
f"mapstudio_admin={token}; Path={cookie_path}; HttpOnly; "
|
||
f"SameSite=Strict; Max-Age=28800{secure_cookie}"
|
||
)
|
||
},
|
||
)
|
||
return
|
||
if path == "/api/admin/logout":
|
||
if not self._require_admin():
|
||
return
|
||
requested_cookie_path = ""
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if 0 < length <= 4096:
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
requested_cookie_path = str(payload.get("cookiePath", ""))
|
||
cookie_jar = cookies.SimpleCookie(self.headers.get("Cookie", ""))
|
||
session_token = cookie_jar.get("mapstudio_admin")
|
||
if session_token:
|
||
SERVICE.destroy_admin_session(session_token.value)
|
||
except Exception:
|
||
pass
|
||
SERVICE.destroy_all_admin_sessions()
|
||
raw_path = urllib.parse.urlsplit(self.path).path
|
||
suffix = "/api/admin/logout"
|
||
proxy_prefix = raw_path[:-len(suffix)] + "/" if raw_path.endswith(suffix) else "/"
|
||
deletion = (
|
||
"mapstudio_admin=; Max-Age=0; "
|
||
"Expires=Thu, 01 Jan 1970 00:00:00 GMT; HttpOnly; SameSite=Strict; Path="
|
||
)
|
||
cookie_paths = ["/", "/api/admin/"]
|
||
if proxy_prefix not in cookie_paths:
|
||
cookie_paths.append(proxy_prefix)
|
||
if (
|
||
requested_cookie_path.startswith("/")
|
||
and not any(char in requested_cookie_path for char in "\r\n;")
|
||
):
|
||
normalized_cookie_path = requested_cookie_path
|
||
if not normalized_cookie_path.endswith("/"):
|
||
normalized_cookie_path += "/"
|
||
if normalized_cookie_path not in cookie_paths:
|
||
cookie_paths.append(normalized_cookie_path)
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
{"ok": True},
|
||
extra_headers=[("Set-Cookie", deletion + cookie_path) for cookie_path in cookie_paths],
|
||
)
|
||
return
|
||
if (
|
||
path in ("/api/cache-policy", "/api/upstream-proxy")
|
||
or path.startswith(("/api/tianditu/", "/api/openmap/", "/api/cache-admin/"))
|
||
) and not self._require_admin():
|
||
return
|
||
if path in ("/api/cache-admin/refresh", "/api/cache-admin/delete"):
|
||
limit_key = "cache-admin-refresh" if path.endswith("/refresh") else "cache-admin-delete"
|
||
if not self._allow_request(limit_key, 3, 60):
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length <= 0 or length > 16 * 1024:
|
||
raise ValueError("Invalid request body")
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("Invalid cache management request")
|
||
categories = payload.get("categories")
|
||
if not isinstance(categories, list) or not categories or len(categories) > 20:
|
||
raise ValueError("Select at least one valid cache category")
|
||
category_ids = [str(item) for item in categories]
|
||
bounds = SERVICE.parse_cache_bounds(payload.get("bounds"))
|
||
if path.endswith("/refresh"):
|
||
if SERVICE.config.cache_only:
|
||
self._send_json(
|
||
HTTPStatus.CONFLICT,
|
||
{
|
||
"ok": False,
|
||
"code": "CACHE_ONLY_MODE",
|
||
"error": "服务器处于仅缓存模式，已禁止更新云端缓存",
|
||
},
|
||
)
|
||
return
|
||
force = bool(payload.get("force", False))
|
||
if bounds is None:
|
||
scheduled = SERVICE.refresh_cache_categories(category_ids, force=force)
|
||
scope = {
|
||
"scheduled": scheduled,
|
||
"matchedFiles": scheduled,
|
||
"skippedCategories": [],
|
||
}
|
||
else:
|
||
scope = SERVICE.refresh_cache_scope(category_ids, bounds, force=force)
|
||
self._send_json(
|
||
HTTPStatus.ACCEPTED,
|
||
{
|
||
"ok": True,
|
||
"categories": category_ids,
|
||
"scheduled": scope["scheduled"],
|
||
"scheduledTiles": scope["scheduled"],
|
||
"matchedFiles": scope.get("matchedFiles", 0),
|
||
"matchedTiles": scope.get("matchedTiles", scope.get("matchedFiles", 0)),
|
||
"skippedCategories": scope.get("skippedCategories", []),
|
||
"bounded": bounds is not None,
|
||
"bounds": list(bounds) if bounds is not None else None,
|
||
"force": force,
|
||
},
|
||
)
|
||
return
|
||
if payload.get("confirmation") != "DELETE_CACHE":
|
||
raise ValueError("Deletion confirmation is required")
|
||
result = (
|
||
SERVICE.delete_cache_scope(category_ids, bounds)
|
||
if bounds is not None
|
||
else SERVICE.delete_cache_categories(category_ids)
|
||
)
|
||
self._send_json(HTTPStatus.OK, result)
|
||
except RuntimeError as error:
|
||
self._send_json(
|
||
HTTPStatus.CONFLICT,
|
||
{"ok": False, "code": "CACHE_BUSY", "error": str(error)},
|
||
)
|
||
except (ValueError, TypeError, json.JSONDecodeError) as error:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": str(error)})
|
||
except OSError as error:
|
||
self._send_json(
|
||
HTTPStatus.INTERNAL_SERVER_ERROR,
|
||
{"ok": False, "error": "删除缓存失败", "detail": str(error)},
|
||
)
|
||
return
|
||
if path in ("/api/tianditu/job/cancel", "/api/openmap/job/cancel"):
|
||
job_id = urllib.parse.parse_qs(parsed.query).get("id", [""])[0]
|
||
if not job_id:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Job id is required"})
|
||
return
|
||
cancel = (
|
||
SERVICE.cancel_tianditu_job
|
||
if path.startswith("/api/tianditu/")
|
||
else SERVICE.cancel_openmap_job
|
||
)
|
||
job = cancel(job_id)
|
||
if not job:
|
||
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "Job not found"})
|
||
else:
|
||
self._send_json(HTTPStatus.OK, {"ok": True, "job": job})
|
||
return
|
||
if path == "/api/cache-policy":
|
||
if not self._allow_request("cache-policy", 60, 60):
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length <= 0 or length > 4096:
|
||
raise ValueError("Invalid request body")
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
if not isinstance(payload, dict) or not isinstance(payload.get("cacheOnly"), bool):
|
||
raise ValueError("cacheOnly must be a boolean")
|
||
enabled = SERVICE.set_cache_only(payload["cacheOnly"])
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
{
|
||
"ok": True,
|
||
"cacheOnly": enabled,
|
||
"placeholderTiles": True,
|
||
},
|
||
)
|
||
except (ValueError, TypeError, json.JSONDecodeError) as error:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": str(error)})
|
||
return
|
||
if path == "/api/upstream-proxy":
|
||
if not self._allow_request("upstream-proxy", 60, 60):
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length <= 0 or length > 4096:
|
||
raise ValueError("Invalid request body")
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("Invalid proxy request")
|
||
if payload.get("reset") is True:
|
||
proxy_url = SERVICE.reset_upstream_proxy()
|
||
else:
|
||
proxy_url = SERVICE.set_upstream_proxy(
|
||
str(payload.get("proxy", "")),
|
||
persist=bool(payload.get("persist", False)),
|
||
)
|
||
proxy = SERVICE.upstream_proxy_status()
|
||
self._send_json(
|
||
HTTPStatus.OK,
|
||
{
|
||
"ok": True,
|
||
"upstreamProxyConfigured": proxy["configured"],
|
||
"upstreamProxy": proxy["proxy"],
|
||
"upstreamProxyDefaultConfigured": proxy["defaultConfigured"],
|
||
"upstreamProxyDefault": proxy["defaultProxy"],
|
||
"upstreamProxyTemporary": proxy["temporary"],
|
||
},
|
||
)
|
||
except (ValueError, TypeError, json.JSONDecodeError) as error:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": str(error)})
|
||
return
|
||
if path == "/api/openmap/topic-download":
|
||
if not self._allow_request("openmap-topic-download", 12, 60):
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length <= 0 or length > 16 * 1024:
|
||
raise ValueError("Invalid request body")
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("Invalid topic download request")
|
||
category_id = str(payload.get("category", ""))
|
||
job = SERVICE.start_openmap_topic_download(
|
||
category_id,
|
||
force=bool(payload.get("force", False)),
|
||
)
|
||
self._send_json(HTTPStatus.ACCEPTED, {"ok": True, "job": job})
|
||
except (ValueError, TypeError, json.JSONDecodeError) as error:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": str(error)})
|
||
return
|
||
if path == "/api/openmap/download":
|
||
if not self._allow_request("openmap-download", 12, 60):
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length <= 0 or length > 1024 * 1024:
|
||
raise ValueError("Invalid request body")
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("Invalid download request")
|
||
providers = [str(value) for value in payload.get("providers", [])]
|
||
force = bool(payload.get("force", False))
|
||
if isinstance(payload.get("bounds"), dict):
|
||
raw_bounds = payload["bounds"]
|
||
bounds = (
|
||
float(raw_bounds["south"]),
|
||
float(raw_bounds["west"]),
|
||
float(raw_bounds["north"]),
|
||
float(raw_bounds["east"]),
|
||
)
|
||
job = SERVICE.start_openmap_download(
|
||
providers=providers,
|
||
bounds=bounds,
|
||
zoom_min=int(payload.get("zoomMin")),
|
||
zoom_max=int(payload.get("zoomMax")),
|
||
force=force,
|
||
)
|
||
else:
|
||
job = SERVICE.start_openmap_download(
|
||
providers=providers,
|
||
z=int(payload.get("z")),
|
||
x0=int(payload.get("x0")),
|
||
x1=int(payload.get("x1")),
|
||
y0=int(payload.get("y0")),
|
||
y1=int(payload.get("y1")),
|
||
force=force,
|
||
)
|
||
self._send_json(HTTPStatus.ACCEPTED, {"ok": True, "job": job})
|
||
except (ValueError, TypeError, KeyError, json.JSONDecodeError) as error:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": str(error)})
|
||
return
|
||
if path == "/api/tianditu/download":
|
||
if not self._allow_request("tianditu-download", 12, 60):
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length <= 0 or length > 1024 * 1024:
|
||
raise ValueError("Invalid request body")
|
||
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("Invalid download request")
|
||
providers = [str(value) for value in payload.get("providers", [])]
|
||
force = bool(payload.get("force", False))
|
||
if isinstance(payload.get("bounds"), dict):
|
||
raw_bounds = payload["bounds"]
|
||
bounds = (
|
||
float(raw_bounds["south"]),
|
||
float(raw_bounds["west"]),
|
||
float(raw_bounds["north"]),
|
||
float(raw_bounds["east"]),
|
||
)
|
||
job = SERVICE.start_tianditu_download(
|
||
providers=providers,
|
||
bounds=bounds,
|
||
zoom_min=int(payload.get("zoomMin")),
|
||
zoom_max=int(payload.get("zoomMax")),
|
||
force=force,
|
||
)
|
||
else:
|
||
job = SERVICE.start_tianditu_download(
|
||
providers=providers,
|
||
z=int(payload.get("z")),
|
||
x0=int(payload.get("x0")),
|
||
x1=int(payload.get("x1")),
|
||
y0=int(payload.get("y0")),
|
||
y1=int(payload.get("y1")),
|
||
force=force,
|
||
)
|
||
self._send_json(HTTPStatus.ACCEPTED, {"ok": True, "job": job})
|
||
except TiandituKeyMissing as error:
|
||
self._send_json(
|
||
HTTPStatus.CONFLICT,
|
||
{
|
||
"ok": False,
|
||
"code": "TIANDITU_KEY_MISSING",
|
||
"error": "请先配置天地图 API KEY",
|
||
"detail": str(error),
|
||
},
|
||
)
|
||
except (ValueError, TypeError, json.JSONDecodeError) as error:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": str(error)})
|
||
return
|
||
if path not in ("/api/cache/refresh", "/api/cache/refresh-all"):
|
||
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "Not found"})
|
||
return
|
||
if not self._require_admin():
|
||
return
|
||
query = urllib.parse.parse_qs(parsed.query)
|
||
force = query.get("force", ["0"])[0] == "1"
|
||
if SERVICE.config.cache_only:
|
||
self._send_json(
|
||
HTTPStatus.CONFLICT,
|
||
{
|
||
"ok": False,
|
||
"code": "CACHE_ONLY_MODE",
|
||
"error": "服务器处于仅缓存模式，已禁止刷新云端缓存",
|
||
},
|
||
)
|
||
return
|
||
if path == "/api/cache/refresh-all":
|
||
if not self._allow_request("cache-refresh-all", 2, 60):
|
||
return
|
||
scheduled = SERVICE.refresh_stale(force=force)
|
||
scope = "all"
|
||
else:
|
||
if not self._allow_request("cache-refresh-view", 4, 60):
|
||
return
|
||
try:
|
||
south = float(query["south"][0])
|
||
west = float(query["west"][0])
|
||
north = float(query["north"][0])
|
||
east = float(query["east"][0])
|
||
zoom = int(float(query.get("zoom", ["12"])[0]))
|
||
except (KeyError, ValueError, IndexError):
|
||
self._send_json(
|
||
HTTPStatus.BAD_REQUEST,
|
||
{"ok": False, "error": "Current view bbox and zoom are required"},
|
||
)
|
||
return
|
||
if not (-90 <= south < north <= 90 and -180 <= west < east <= 180):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid bbox order"})
|
||
return
|
||
if not 0 <= zoom <= 18:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid zoom"})
|
||
return
|
||
try:
|
||
scheduled = SERVICE.refresh_current_view(
|
||
(south, west, north, east),
|
||
zoom,
|
||
force=True,
|
||
)
|
||
except ValueError as error:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": str(error)})
|
||
return
|
||
scope = "view"
|
||
force = True
|
||
pending = len(list(SERVICE.pending_rail_dir.glob("*.json")))
|
||
if pending:
|
||
SERVICE._refresh_executor.submit(SERVICE.retry_pending_rail_jobs, min(8, pending))
|
||
pending_water = len(list(SERVICE.pending_water_dir.glob("*.json")))
|
||
if pending_water:
|
||
SERVICE._refresh_executor.submit(SERVICE.retry_pending_water_jobs, min(8, pending_water))
|
||
SERVICE.increment("manual_refreshes")
|
||
self._send_json(
|
||
HTTPStatus.ACCEPTED,
|
||
{
|
||
"ok": True,
|
||
"scheduled": scheduled,
|
||
"pendingRailRequests": pending,
|
||
"pendingWaterRequests": pending_water,
|
||
"force": force,
|
||
"scope": scope,
|
||
"message": "Current-view cache refresh scheduled in background"
|
||
if scope == "view" else "Bulk cache refresh scheduled in background",
|
||
},
|
||
)
|
||
|
||
def _handle_rail(self, raw_query: str) -> None:
|
||
query = urllib.parse.parse_qs(raw_query)
|
||
try:
|
||
south = float(query["south"][0])
|
||
west = float(query["west"][0])
|
||
north = float(query["north"][0])
|
||
east = float(query["east"][0])
|
||
except (KeyError, ValueError, IndexError):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Missing or invalid bbox"})
|
||
return
|
||
if not (-90 <= south < north <= 90 and -180 <= west < east <= 180):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid bbox order"})
|
||
return
|
||
area = (north - south) * (east - west)
|
||
if area > 64800:
|
||
self._send_json(
|
||
HTTPStatus.BAD_REQUEST,
|
||
{"ok": False, "error": "Rail query bbox is too large; zoom in"},
|
||
)
|
||
return
|
||
types = [
|
||
value
|
||
for value in query.get("types", ["rail", "subway"])[0].split(",")
|
||
if value in ("rail", "subway")
|
||
]
|
||
if not types:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "No valid rail types"})
|
||
return
|
||
force = query.get("refresh", ["0"])[0] == "1"
|
||
try:
|
||
tile_zoom = int(query.get("zoom", ["12"])[0])
|
||
except ValueError:
|
||
tile_zoom = 12
|
||
if "subway" in types:
|
||
tile_zoom = SERVICE.choose_vector_zoom(
|
||
(south, west, north, east),
|
||
preferred=14,
|
||
max_tiles=MAX_VECTOR_TILES,
|
||
)
|
||
else:
|
||
tile_zoom = min(14, max(7, tile_zoom + 2))
|
||
try:
|
||
response, cache_state = SERVICE.get_rail(
|
||
(south, west, north, east), types, force, tile_zoom=tile_zoom
|
||
)
|
||
if query.get("format", [""])[0] == "delta-v1":
|
||
response = SERVICE.compact_vector_response(response, "rail")
|
||
self._send_json(HTTPStatus.OK, response, cache_state)
|
||
except RailPending as pending:
|
||
self._send_json(
|
||
HTTPStatus.ACCEPTED,
|
||
{
|
||
"ok": False,
|
||
"pending": True,
|
||
"cacheKey": pending.cache_key,
|
||
"message": "Rail query queued for background retrieval",
|
||
},
|
||
)
|
||
except Exception as error:
|
||
self._send_json(
|
||
HTTPStatus.BAD_GATEWAY,
|
||
{"ok": False, "error": "Rail upstream unavailable", "detail": str(error)},
|
||
)
|
||
|
||
def _handle_water(self, raw_query: str) -> None:
|
||
query = urllib.parse.parse_qs(raw_query)
|
||
try:
|
||
south = float(query["south"][0])
|
||
west = float(query["west"][0])
|
||
north = float(query["north"][0])
|
||
east = float(query["east"][0])
|
||
except (KeyError, ValueError, IndexError):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Missing or invalid bbox"})
|
||
return
|
||
if not (-90 <= south < north <= 90 and -180 <= west < east <= 180):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid bbox order"})
|
||
return
|
||
area = (north - south) * (east - west)
|
||
if area > 64800:
|
||
self._send_json(
|
||
HTTPStatus.BAD_REQUEST,
|
||
{"ok": False, "error": "Water query bbox is too large; zoom in"},
|
||
)
|
||
return
|
||
force = query.get("refresh", ["0"])[0] == "1"
|
||
include = [
|
||
value
|
||
for value in query.get("include", ["water,ports,vegetation"])[0].split(",")
|
||
if value in ("water", "ports", "vegetation")
|
||
]
|
||
if not include:
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "No environment layers selected"})
|
||
return
|
||
try:
|
||
tile_zoom = int(query.get("zoom", ["12"])[0])
|
||
except ValueError:
|
||
tile_zoom = 12
|
||
tile_zoom = min(ENVIRONMENT_MAX_ZOOM, max(ENVIRONMENT_MIN_ZOOM, tile_zoom))
|
||
try:
|
||
response, cache_state = SERVICE.get_water(
|
||
(south, west, north, east), force, tile_zoom=tile_zoom, include=include
|
||
)
|
||
if query.get("format", [""])[0] == "delta-v1":
|
||
response = SERVICE.compact_vector_response(response, "water")
|
||
self._send_json(HTTPStatus.OK, response, cache_state)
|
||
except WaterPending as pending:
|
||
self._send_json(
|
||
HTTPStatus.ACCEPTED,
|
||
{
|
||
"ok": False,
|
||
"pending": True,
|
||
"cacheKey": pending.cache_key,
|
||
"message": "Water query queued for background retrieval",
|
||
},
|
||
)
|
||
except Exception as error:
|
||
self._send_json(
|
||
HTTPStatus.BAD_GATEWAY,
|
||
{"ok": False, "error": "Water upstream unavailable", "detail": str(error)},
|
||
)
|
||
|
||
def _handle_boundaries(self, raw_query: str) -> None:
|
||
query = urllib.parse.parse_qs(raw_query)
|
||
try:
|
||
south = float(query["south"][0])
|
||
west = float(query["west"][0])
|
||
north = float(query["north"][0])
|
||
east = float(query["east"][0])
|
||
except (KeyError, ValueError, IndexError):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Missing or invalid bbox"})
|
||
return
|
||
if not (-90 <= south < north <= 90 and -180 <= west < east <= 180):
|
||
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Invalid bbox order"})
|
||
return
|
||
try:
|
||
zoom = int(query.get("zoom", ["12"])[0])
|
||
except ValueError:
|
||
zoom = 12
|
||
preferred_zoom = min(14, max(4, zoom + 1))
|
||
tile_zoom = SERVICE.choose_vector_zoom(
|
||
(south, west, north, east),
|
||
preferred=preferred_zoom,
|
||
max_tiles=MAX_VECTOR_TILES,
|
||
)
|
||
force = query.get("refresh", ["0"])[0] == "1"
|
||
try:
|
||
response, cache_state = SERVICE.get_boundaries(
|
||
(south, west, north, east),
|
||
force,
|
||
tile_zoom=tile_zoom,
|
||
)
|
||
if query.get("format", [""])[0] == "delta-v1":
|
||
response = SERVICE.compact_vector_response(response, "boundary")
|
||
self._send_json(HTTPStatus.OK, response, cache_state)
|
||
except Exception as error:
|
||
self._send_json(
|
||
HTTPStatus.BAD_GATEWAY,
|
||
{"ok": False, "error": "Administrative boundary data unavailable", "detail": str(error)},
|
||
)
|
||
|
||
def _serve_static(self, request_path: str) -> None:
|
||
decoded = urllib.parse.unquote(request_path)
|
||
if decoded in ("", "/"):
|
||
decoded = "/index.html"
|
||
relative = decoded.lstrip("/")
|
||
candidate = (ROOT / relative).resolve()
|
||
try:
|
||
resolved_relative = candidate.relative_to(ROOT)
|
||
except ValueError:
|
||
self._send_json(HTTPStatus.FORBIDDEN, {"ok": False, "error": "Forbidden"})
|
||
return
|
||
public_path = "/" + resolved_relative.as_posix()
|
||
public_parts = public_path.split("/")
|
||
public_allowed = public_path.lower() in PUBLIC_STATIC_FILES or (
|
||
len(public_parts) > 2 and public_parts[1] in PUBLIC_STATIC_DIRS
|
||
)
|
||
if not public_allowed:
|
||
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "File not found"})
|
||
return
|
||
if not candidate.is_file():
|
||
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "File not found"})
|
||
return
|
||
content_type = mimetypes.guess_type(str(candidate))[0] or "application/octet-stream"
|
||
if content_type.startswith("text/") or content_type in (
|
||
"application/javascript",
|
||
"application/json",
|
||
):
|
||
content_type += "; charset=utf-8"
|
||
try:
|
||
data = candidate.read_bytes()
|
||
except OSError:
|
||
self._send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"ok": False, "error": "Read failed"})
|
||
return
|
||
compressible = (
|
||
content_type.startswith("text/")
|
||
or content_type in ("application/javascript", "application/json", "image/svg+xml")
|
||
)
|
||
use_gzip = (
|
||
compressible
|
||
and len(data) > 1024
|
||
and "gzip" in self.headers.get("Accept-Encoding", "").lower()
|
||
)
|
||
body = gzip.compress(data, compresslevel=6, mtime=0) if use_gzip else data
|
||
self.send_response(HTTPStatus.OK)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
if use_gzip:
|
||
self.send_header("Content-Encoding", "gzip")
|
||
self.send_header("Vary", "Accept-Encoding")
|
||
entry_file = candidate.name in (
|
||
"index.html",
|
||
"sw.js",
|
||
"tianditu-admin.html",
|
||
"openmap-admin.html",
|
||
"cache-admin.html",
|
||
"regulations.html",
|
||
)
|
||
versioned_asset = "?v=" in self.path or "/vendor/" in decoded
|
||
long_lived_asset = candidate.name in (
|
||
"app.js",
|
||
"styles.css",
|
||
"print-block.js",
|
||
"compliance-gate.js",
|
||
)
|
||
self.send_header(
|
||
"Cache-Control",
|
||
"no-cache"
|
||
if entry_file
|
||
else LONG_BROWSER_CACHE if versioned_asset or long_lived_asset else "public, max-age=86400",
|
||
)
|
||
self._cors_headers()
|
||
self._security_headers()
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(body)
|
||
|
||
def log_message(self, fmt: str, *args: Any) -> None:
|
||
print(
|
||
"%s - - [%s] %s"
|
||
% (self.client_address[0], self.log_date_time_string(), fmt % args),
|
||
flush=True,
|
||
)
|
||
|
||
|
||
def parse_args() -> ServiceConfig:
|
||
parser = argparse.ArgumentParser(description="Map Studio local map cache service")
|
||
parser.add_argument("--host", default="127.0.0.1")
|
||
parser.add_argument("--port", type=int, default=8081)
|
||
parser.add_argument("--cache-dir", type=Path, default=DEFAULT_CACHE_DIR)
|
||
parser.add_argument("--ttl-hours", type=float, default=DEFAULT_TTL_SECONDS / 3600)
|
||
parser.add_argument("--rail-ttl-hours", type=float, default=DEFAULT_RAIL_TTL_SECONDS / 3600)
|
||
parser.add_argument("--refresh-hours", type=float, default=DEFAULT_REFRESH_SECONDS / 3600)
|
||
parser.add_argument("--refresh-batch", type=int, default=128)
|
||
parser.add_argument("--tianditu-key", default="")
|
||
parser.add_argument("--admin-password", default="")
|
||
parser.add_argument(
|
||
"--upstream-proxy",
|
||
default=None,
|
||
help="HTTP/HTTPS proxy for all outbound map, DEM, and Overpass requests",
|
||
)
|
||
parser.add_argument(
|
||
"--disable-upstream-proxy",
|
||
dest="upstream_proxy",
|
||
action="store_const",
|
||
const="",
|
||
help="Disable a persisted or environment upstream proxy",
|
||
)
|
||
parser.add_argument(
|
||
"--cache-only",
|
||
"--offline-only",
|
||
dest="cache_only",
|
||
action="store_true",
|
||
default=None,
|
||
help="Disable all upstream requests and serve placeholder tiles for cache misses",
|
||
)
|
||
parser.add_argument(
|
||
"--allow-online",
|
||
dest="cache_only",
|
||
action="store_false",
|
||
help="Override a persisted or environment cache-only setting",
|
||
)
|
||
args = parser.parse_args()
|
||
tianditu_key = (args.tianditu_key or os.environ.get("TIANDITU_KEY", "")).strip()
|
||
if not tianditu_key:
|
||
key_file = ROOT / "tianditu-key.txt"
|
||
if key_file.is_file():
|
||
tianditu_key = key_file.read_text(encoding="utf-8").strip()
|
||
admin_password = (
|
||
args.admin_password or os.environ.get("MAP_STUDIO_ADMIN_PASSWORD", "")
|
||
).strip()
|
||
if not admin_password:
|
||
password_file = ROOT / "admin-password.txt"
|
||
if password_file.is_file():
|
||
admin_password = password_file.read_text(encoding="utf-8").strip()
|
||
cache_only = args.cache_only
|
||
if cache_only is None:
|
||
env_cache_only = os.environ.get("MAP_STUDIO_CACHE_ONLY", "").strip().lower()
|
||
if env_cache_only in ("1", "true", "yes", "on"):
|
||
cache_only = True
|
||
elif env_cache_only in ("0", "false", "no", "off"):
|
||
cache_only = False
|
||
else:
|
||
setting_file = Path(args.cache_dir) / CACHE_ONLY_SETTING_NAME
|
||
try:
|
||
cache_only = setting_file.read_text(encoding="utf-8").strip() == "1"
|
||
except OSError:
|
||
cache_only = False
|
||
upstream_proxy = args.upstream_proxy
|
||
if upstream_proxy is None:
|
||
upstream_proxy = os.environ.get("MAP_STUDIO_UPSTREAM_PROXY")
|
||
if upstream_proxy is None:
|
||
proxy_file = Path(args.cache_dir) / UPSTREAM_PROXY_SETTING_NAME
|
||
try:
|
||
upstream_proxy = proxy_file.read_text(encoding="utf-8").strip()
|
||
except OSError:
|
||
upstream_proxy = ""
|
||
upstream_proxy = normalize_upstream_proxy(upstream_proxy)
|
||
return ServiceConfig(
|
||
host=args.host,
|
||
port=args.port,
|
||
cache_dir=args.cache_dir,
|
||
tile_ttl_seconds=max(60, int(args.ttl_hours * 3600)),
|
||
rail_ttl_seconds=max(60, int(args.rail_ttl_hours * 3600)),
|
||
refresh_seconds=max(60, int(args.refresh_hours * 3600)),
|
||
refresh_batch=max(1, args.refresh_batch),
|
||
tianditu_key=tianditu_key,
|
||
admin_password=admin_password,
|
||
cache_only=bool(cache_only),
|
||
upstream_proxy=upstream_proxy,
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
global SERVICE
|
||
config = parse_args()
|
||
SERVICE = MapService(config)
|
||
server = QuietThreadingHTTPServer((config.host, config.port), MapStudioRequestHandler)
|
||
print(f"Map Studio server: http://{config.host}:{config.port}/", flush=True)
|
||
print(f"Cache directory: {config.cache_dir.resolve()}", flush=True)
|
||
print(f"Cache-only mode: {'ON' if config.cache_only else 'OFF'}", flush=True)
|
||
print(
|
||
"Manual download workers: "
|
||
f"Tianditu={TIANDITU_DOWNLOAD_WORKERS}, OpenMap={OPENMAP_DOWNLOAD_WORKERS}; "
|
||
f"staging={DOWNLOAD_STAGING_DIR_NAME}",
|
||
flush=True,
|
||
)
|
||
proxy_label = display_upstream_proxy(config.upstream_proxy)
|
||
print(f"Upstream proxy: {proxy_label or 'direct'}", flush=True)
|
||
if config.admin_password and len(config.admin_password) < 12:
|
||
print(
|
||
"WARNING: admin password is shorter than 12 characters; use a longer "
|
||
"password, especially when exposing the service through a proxy.",
|
||
flush=True,
|
||
)
|
||
print(
|
||
f"Tile TTL: {config.tile_ttl_seconds}s, rail TTL: {config.rail_ttl_seconds}s, "
|
||
f"refresh interval: {config.refresh_seconds}s",
|
||
flush=True,
|
||
)
|
||
try:
|
||
server.serve_forever(poll_interval=0.5)
|
||
except KeyboardInterrupt:
|
||
print("\nStopping Map Studio server...", flush=True)
|
||
finally:
|
||
server.server_close()
|
||
SERVICE.close()
|
||
print("Map Studio server stopped.", flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|