codex-bot
4845826deb
Shanhe source and offline package checks / verify (pull_request) Failing after 5s
225 lines
11 KiB
Python
225 lines
11 KiB
Python
"""Download/audit county GIS inputs without changing the released game maps.
|
|
|
|
Raw source geometry, reproducible per-city game coordinates and a provenance
|
|
manifest live in ignored data/. Four cities without county children retain
|
|
their whole-city outline instead of fabricated county or township polygons.
|
|
"""
|
|
import argparse
|
|
import concurrent.futures
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import pathlib
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import zipfile
|
|
|
|
from map_geometry import make_level, place_labels
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
BASE = "https://geo.datav.aliyun.com/areas_v3/bound/"
|
|
EXCEPTIONS = {
|
|
"441900": "https://www.dg.gov.cn/zjdz/dzgk/xzqh/content/post_2824924.html",
|
|
"442000": "https://www.zs.gov.cn/zjzs/zsgk/content/post_1669002.html",
|
|
"460400": "https://www.danzhou.gov.cn/danzhou/index.html",
|
|
"620200": "https://wap.jyg.gov.cn/zjjyg/",
|
|
}
|
|
|
|
|
|
def read_json(path):
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def inspect_features(data, parent, whole=False):
|
|
if data.get("type") != "FeatureCollection" or not data.get("features"):
|
|
raise ValueError("Expected a non-empty FeatureCollection")
|
|
entries, seen = [], set()
|
|
for feature in data["features"]:
|
|
prop = feature.get("properties", {})
|
|
code = str(prop.get("adcode", ""))
|
|
if not code.isdigit() or code in seen or not prop.get("name"):
|
|
raise ValueError("Missing/duplicate feature id or name")
|
|
seen.add(code)
|
|
if whole:
|
|
if code != parent:
|
|
raise ValueError("Whole-city feature does not match city")
|
|
elif (str(prop.get("parent", {}).get("adcode")) != parent
|
|
or prop.get("level") != "district"):
|
|
raise ValueError("County is not an immediate district child")
|
|
geo = feature.get("geometry", {})
|
|
if geo.get("type") not in ("Polygon", "MultiPolygon"):
|
|
raise ValueError("Unexpected non-polygon geometry")
|
|
polys = geo["coordinates"] if geo["type"] == "MultiPolygon" else [geo["coordinates"]]
|
|
rings = holes = points = 0
|
|
for polygon in polys:
|
|
if not polygon:
|
|
raise ValueError("Empty polygon")
|
|
rings += 1
|
|
holes += len(polygon) - 1
|
|
for ring in polygon:
|
|
if len(ring) < 4 or ring[0] != ring[-1]:
|
|
raise ValueError("Open or degenerate source ring")
|
|
for point in ring:
|
|
if (len(point) < 2 or not all(isinstance(v, (int, float))
|
|
and math.isfinite(v) for v in point[:2])
|
|
or not -180 <= point[0] <= 180 or not -90 <= point[1] <= 90):
|
|
raise ValueError("Invalid geographic coordinate")
|
|
if abs(sum(a[0]*b[1]-b[0]*a[1] for a, b in zip(ring, ring[1:]))) < 1e-15:
|
|
raise ValueError("Zero-area source ring")
|
|
points += len(ring)
|
|
entries.append({"id": code, "name": prop["name"], "rings": rings,
|
|
"holes": holes, "points": points})
|
|
return entries
|
|
|
|
|
|
def get_source(code, folder, whole=False):
|
|
suffix = ".json" if whole else "_full.json"
|
|
url = BASE + code + suffix
|
|
path = folder / (code + ".json")
|
|
if path.exists():
|
|
raw = path.read_bytes()
|
|
data = json.loads(raw)
|
|
regions = inspect_features(data, code, whole)
|
|
return raw, data, regions, url
|
|
for attempt in range(3):
|
|
try:
|
|
request = urllib.request.Request(url, headers={"User-Agent": "ShanhePuzzle-GIS/1.0"})
|
|
with urllib.request.urlopen(request, timeout=35) as response:
|
|
raw = response.read()
|
|
data = json.loads(raw)
|
|
regions = inspect_features(data, code, whole)
|
|
tmp = path.with_suffix(".partial")
|
|
tmp.write_bytes(raw)
|
|
tmp.replace(path)
|
|
return raw, data, regions, url
|
|
except urllib.error.HTTPError as error:
|
|
if error.code == 404 or attempt == 2:
|
|
raise
|
|
except Exception:
|
|
if attempt == 2:
|
|
raise
|
|
time.sleep(.5 * (attempt + 1))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--snapshot", required=True, help="Local YYYY-MM-DD snapshot label")
|
|
parser.add_argument("--only", help="Optional comma-separated city ids for a small probe")
|
|
args = parser.parse_args()
|
|
datetime.date.fromisoformat(args.snapshot)
|
|
index = read_json(ROOT / "research/prefecture-map-index.json")
|
|
cities = [dict(region, province=province["id"], provinceName=province["name"])
|
|
for province in index["levels"] for region in province["regions"]
|
|
if region["unitType"] == "prefecture"]
|
|
if args.only:
|
|
selected = set(args.only.split(","))
|
|
cities = [city for city in cities if city["id"] in selected]
|
|
if len(cities) != len(selected):
|
|
raise ValueError("Unknown city id in --only")
|
|
dest = ROOT / "data/city-county-gis" / args.snapshot
|
|
for name in ["raw", "unsplit", "levels"]:
|
|
(dest / name).mkdir(parents=True, exist_ok=True)
|
|
|
|
def fetch(city):
|
|
code = city["id"]
|
|
entry = {"id": code, "name": city["name"], "province": city["province"],
|
|
"provinceName": city["provinceName"]}
|
|
try:
|
|
raw, data, regions, url = get_source(code, dest / "raw")
|
|
entry.update(status="available", path="raw/"+code+".json", source=url,
|
|
sha256=hashlib.sha256(raw).hexdigest(), bytes=len(raw), regions=regions)
|
|
except urllib.error.HTTPError as error:
|
|
if error.code != 404 or code not in EXCEPTIONS:
|
|
entry.update(status="failed", reason=str(error))
|
|
return entry
|
|
try:
|
|
raw, data, regions, url = get_source(code, dest / "unsplit", whole=True)
|
|
prop = data["features"][0]["properties"]
|
|
if len(regions) != 1 or prop.get("childrenNum") != 0:
|
|
raise ValueError("Cannot confirm the no-county-child exception")
|
|
entry.update(status="no_county_children", path="unsplit/"+code+".json",
|
|
source=url, unavailableChildSource=BASE+code+"_full.json",
|
|
sourceChildrenNum=0, reference=EXCEPTIONS[code],
|
|
sha256=hashlib.sha256(raw).hexdigest(), bytes=len(raw), regions=regions,
|
|
note="Whole-city outline only; township boundaries are not supplied by this source.")
|
|
except Exception as fallback_error:
|
|
entry.update(status="failed", reason=str(fallback_error))
|
|
except Exception as error:
|
|
entry.update(status="failed", reason=str(error))
|
|
return entry
|
|
|
|
entries = []
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
|
|
futures = [pool.submit(fetch, city) for city in cities]
|
|
for future in concurrent.futures.as_completed(futures):
|
|
entries.append(future.result())
|
|
if len(entries) % 25 == 0 or len(entries) == len(cities):
|
|
print("GIS fetched:", len(entries), "/", len(cities), flush=True)
|
|
entries.sort(key=lambda entry: entry["id"])
|
|
failed = [entry for entry in entries if entry["status"] == "failed"]
|
|
for entry in entries:
|
|
if entry["status"] != "available":
|
|
continue
|
|
data = read_json(dest / entry["path"])
|
|
level = make_level(data["features"], int(entry["id"]), entry["name"], int(entry["province"]))
|
|
if not level:
|
|
raise ValueError("No playable subdivision for " + entry["id"])
|
|
level.update(kind="county", parent=entry["province"], playLabel="县区拼图")
|
|
level.pop("projection", None)
|
|
place_labels(level)
|
|
source_by_id = {region["id"]: region for region in entry["regions"]}
|
|
assert {region["id"] for region in level["regions"]} == set(source_by_id)
|
|
for region in level["regions"]:
|
|
expected = source_by_id[region["id"]]
|
|
assert region["name"] == expected["name"]
|
|
assert len(region["rings"]) == expected["rings"]
|
|
assert len(region.get("holes", [])) == expected["holes"]
|
|
for ring in region["rings"] + region.get("holes", []):
|
|
assert len(ring) >= 3 and all(all(math.isfinite(v) for v in p) for p in ring)
|
|
assert abs(sum(ring[i][0]*ring[(i+1)%len(ring)][1]
|
|
-ring[(i+1)%len(ring)][0]*ring[i][1]
|
|
for i in range(len(ring)))) > 1e-10
|
|
encoded = json.dumps(level, ensure_ascii=False, separators=(",", ":")).encode()
|
|
path = dest / "levels" / (entry["id"] + ".json")
|
|
path.write_bytes(encoded)
|
|
entry["preparedPath"] = "levels/" + path.name
|
|
entry["preparedSha256"] = hashlib.sha256(encoded).hexdigest()
|
|
available = [entry for entry in entries if entry["status"] == "available"]
|
|
manifest = {"source": "DataV.GeoAtlas", "snapshotDate": args.snapshot,
|
|
"scope": "Existing game's prefectural areas; provider snapshot, not a guarantee of current administrative boundaries.",
|
|
"rawCoordinates": "Source longitude/latitude values preserved; datum not declared by the source, no WGS84/GCJ02 assumption.",
|
|
"preparedCoordinates": "Simplified game-plane coordinates, longest extent 1000; not geographic GIS coordinates.",
|
|
"releaseMapsChanged": False, "requested": len(cities), "available": len(available),
|
|
"countyPieces": sum(len(entry["regions"]) for entry in available),
|
|
"noCountyChildren": sum(entry["status"] == "no_county_children" for entry in entries),
|
|
"failed": len(failed), "cities": entries}
|
|
(dest / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2)+"\n")
|
|
(dest / "README.md").write_text("# City county GIS material snapshot\n\n"
|
|
"raw/: original per-city GeoJSON county outlines; longitude/latitude values are preserved.\n"
|
|
"levels/: simplified per-city game coordinates, with every source outer ring and hole retained.\n"
|
|
"unsplit/: whole-city outlines for four cities that report no county children.\n"
|
|
"manifest.json: source URLs, county names/codes, raw/prepared checksums, coverage and exceptions.\n\n"
|
|
"Prepared files still require collision decomposition, topology/playability and mobile performance validation before game integration. "
|
|
"They are materials, not an uploadable mini-tool ZIP. No released game map is modified.\n")
|
|
print(json.dumps({k: manifest[k] for k in ["requested", "available", "countyPieces", "noCountyChildren", "failed"]}), flush=True)
|
|
if failed:
|
|
print(json.dumps(failed, ensure_ascii=False), flush=True)
|
|
raise SystemExit(1)
|
|
artifact = ROOT / "artifacts" / ("city-county-gis-"+args.snapshot.replace("-", "")+".zip")
|
|
artifact.parent.mkdir(exist_ok=True)
|
|
paths = [dest / "manifest.json", dest / "README.md"]
|
|
for entry in entries:
|
|
paths.append(dest / entry["path"])
|
|
if entry.get("preparedPath"):
|
|
paths.append(dest / entry["preparedPath"])
|
|
with zipfile.ZipFile(artifact, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as bundle:
|
|
for path in sorted(paths):
|
|
bundle.write(path, path.relative_to(dest).as_posix())
|
|
print("GIS material ZIP:", artifact, "|", artifact.stat().st_size, "bytes", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|