pbf/build_semantic_delivery_assets.py

from __future__ import annotations

import csv
import json
import os
import re
import shutil
import subprocess
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass
from pathlib import Path

import mapbox_vector_tile
from PIL import Image

from navsea_tile_reencode_guard import (
    assert_reencoded_tile_safe,
    decode_tile_with_extents,
    encode_layers_preserving_extents,
)

REPO_ROOT = Path(__file__).resolve().parent
BASE_AUDIT_CSV = REPO_ROOT / "tasks/pbf/NavSea_Sprite_Audit_newpec_style_with_pbf_2026-04-16.csv"
T_OVERRIDE_CSV = REPO_ROOT / "tasks/pbf/NavSea_Sprite_Basis_T_建议修正表_2026-04-16.csv"
MAPPING_JSON = REPO_ROOT / "src/pbf/semantic_sprite_key_map_2026-04-16.json"

SOURCE_SPRITE_JSON_2X = Path("/mnt/sda1/www/newpec/sprite/sprite@2x.json")
SOURCE_SPRITE_PNG_2X = Path("/mnt/sda1/www/newpec/sprite/sprite@2x.png")
OUTPUT_SPRITE_DIR = Path("/mnt/sda1/www/newpec/sprite-semantic")

SOURCE_STYLE = REPO_ROOT / "src/pbf/style.navsea-delivery-full.json"
OUTPUT_STYLE = REPO_ROOT / "src/pbf/style.navsea-delivery-full-semantic.json"
DEPLOYED_STYLE = Path("/mnt/sda1/www/newpec/domain/style.navsea-delivery-full-semantic.json")

SOURCE_PBF_ROOT = Path("/home/wwwroot/pbf-delivery-full-20260415")
OUTPUT_PBF_ROOT = Path("/home/wwwroot/pbf-delivery-full-semantic-20260416")

SEMANTIC_SPRITE_BASE_URL = "http://192.168.200.184/newpec/sprite-semantic/sprite"
SEMANTIC_PBF_TILE_URL = "http://192.168.200.184/pbf-delivery-full-semantic-20260416/{z}/{x}/{y}.pbf"
STYLE_VERSION = "full-semantic-style-r1-20260416-1628"
PBF_VERSION = "full-semantic-pbf-r1-20260416-1628"
SPRITE_VERSION = "full-semantic-sprite-r1-20260416-1628"

PBF_PROPERTY_FIELDS = ("chart_icon_image", "chart_fill_pattern")


@dataclass(frozen=True)
class AuditRow:
    old_sprite_key: str
    used_in_newpec_style: int
    used_in_pbf: int
    new_sprite_key: str


def normalize_key(value: str) -> str:
    value = re.sub(r"\s*\[[^\]]+\]\s*", "", value.strip())
    value = value.replace("-", "_").replace(" ", "_").lower()
    value = (
        value.replace("easte", "east")
        .replace("sourth", "south")
        .replace("varian", "variant")
        .replace("cardianal", "cardinal")
    )
    value = re.sub(r"_+", "_", value).strip("_")
    return value


def load_audit_rows() -> list[AuditRow]:
    overrides: dict[str, str] = {}
    with T_OVERRIDE_CSV.open(encoding="utf-8-sig") as handle:
        for row in csv.DictReader(handle):
            overrides[row["old_sprite_key"].strip()] = row["suggested_new_sprite_key"].strip()

    rows: list[AuditRow] = []
    with BASE_AUDIT_CSV.open(encoding="utf-8-sig") as handle:
        for row in csv.DictReader(handle):
            old_key = row["old_sprite_key"].strip()
            new_key = overrides.get(old_key) or row["new_sprite_key"].strip()
            rows.append(
                AuditRow(
                    old_sprite_key=old_key,
                    used_in_newpec_style=int(row["used_in_newpec_style"] or 0),
                    used_in_pbf=int(row["used_in_pbf"] or 0),
                    new_sprite_key=normalize_key(new_key) if new_key else "",
                )
            )
    return rows


def build_mapping(rows: list[AuditRow]) -> dict[str, str]:
    mapping: dict[str, str] = {}
    for row in rows:
        if row.new_sprite_key:
            mapping[row.old_sprite_key] = row.new_sprite_key
    return mapping


def build_used_sprite_entries(rows: list[AuditRow], mapping: dict[str, str]) -> list[tuple[str, str]]:
    used: list[tuple[str, str]] = []
    for row in rows:
        if row.used_in_newpec_style <= 0 and row.used_in_pbf <= 0:
            continue
        used.append((row.old_sprite_key, mapping.get(row.old_sprite_key, row.old_sprite_key)))
    return used


def write_mapping_json(mapping: dict[str, str]) -> None:
    MAPPING_JSON.write_text(
        json.dumps(dict(sorted(mapping.items())), ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def pack_entries(
    entries: list[tuple[str, str]],
    source_meta: dict[str, dict],
    scale: float,
    max_width: int,
) -> tuple[Image.Image, dict[str, dict]]:
    positions: dict[str, dict] = {}
    cursor_x = 0
    cursor_y = 0
    row_height = 0
    packed: list[tuple[str, tuple[int, int, int, int]]] = []

    for old_key, new_key in entries:
        meta = source_meta[old_key]
        width = max(1, round(meta["width"] * scale))
        height = max(1, round(meta["height"] * scale))
        if cursor_x and cursor_x + width > max_width:
            cursor_x = 0
            cursor_y += row_height
            row_height = 0
        packed.append((old_key, (cursor_x, cursor_y, width, height)))
        record = {
            "x": cursor_x,
            "y": cursor_y,
            "width": width,
            "height": height,
            "pixelRatio": 1 if scale == 0.5 else 2,
        }
        positions[new_key] = record
        if old_key != new_key:
            positions[old_key] = dict(record)
        cursor_x += width
        row_height = max(row_height, height)

    atlas_height = cursor_y + row_height
    image = Image.new("RGBA", (max_width, atlas_height), (0, 0, 0, 0))
    source_image = Image.open(SOURCE_SPRITE_PNG_2X).convert("RGBA")

    for old_key, (x, y, width, height) in packed:
        meta = source_meta[old_key]
        crop = source_image.crop(
            (
                meta["x"],
                meta["y"],
                meta["x"] + meta["width"],
                meta["y"] + meta["height"],
            )
        )
        if scale != 1.0:
            crop = crop.resize((width, height), Image.LANCZOS)
        image.paste(crop, (x, y))

    return image, positions


def build_semantic_sprite(entries: list[tuple[str, str]]) -> None:
    OUTPUT_SPRITE_DIR.mkdir(parents=True, exist_ok=True)
    source_meta = json.loads(SOURCE_SPRITE_JSON_2X.read_text(encoding="utf-8"))

    atlas_2x, meta_2x = pack_entries(entries, source_meta, scale=1.0, max_width=2048)
    atlas_1x, meta_1x = pack_entries(entries, source_meta, scale=0.5, max_width=1024)

    atlas_2x.save(OUTPUT_SPRITE_DIR / "sprite@2x.png")
    atlas_1x.save(OUTPUT_SPRITE_DIR / "sprite.png")
    (OUTPUT_SPRITE_DIR / "sprite@2x.json").write_text(
        json.dumps(meta_2x, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )
    (OUTPUT_SPRITE_DIR / "sprite.json").write_text(
        json.dumps(meta_1x, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )


def replace_sprite_keys(value, mapping: dict[str, str]):
    if isinstance(value, dict):
        return {k: replace_sprite_keys(v, mapping) for k, v in value.items()}
    if isinstance(value, list):
        return [replace_sprite_keys(item, mapping) for item in value]
    if isinstance(value, str):
        return mapping.get(value, value)
    return value


def build_semantic_style(mapping: dict[str, str]) -> None:
    style = json.loads(SOURCE_STYLE.read_text(encoding="utf-8"))
    style = replace_sprite_keys(style, mapping)
    style["name"] = "NavSea Delivery Full Semantic"
    style["sprite"] = SEMANTIC_SPRITE_BASE_URL
    if style.get("sources", {}).get("navsea_delivery"):
        style["sources"]["navsea_delivery"]["tiles"] = [SEMANTIC_PBF_TILE_URL]
    OUTPUT_STYLE.write_text(json.dumps(style, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    DEPLOYED_STYLE.write_text(json.dumps(style, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def hardlink_or_copy(src: Path, dst: Path) -> None:
    dst.parent.mkdir(parents=True, exist_ok=True)
    if dst.exists():
        dst.unlink()
    try:
        os.link(src, dst)
    except OSError:
        dst.write_bytes(src.read_bytes())


def remap_tile(src: Path, dst: Path, value_map: dict[str, str], needles: list[bytes]) -> tuple[bool, int]:
    raw = src.read_bytes()
    if not any(needle in raw for needle in needles):
        hardlink_or_copy(src, dst)
        return False, 0

    decoded, source_extents = decode_tile_with_extents(raw)
    changed_features = 0
    changed = False
    layers = []

    for layer_name, layer in decoded.items():
        features = []
        for feature in layer["features"]:
            props = feature.get("properties", {})
            feature_changed = False
            for field in PBF_PROPERTY_FIELDS:
                value = props.get(field)
                if isinstance(value, str) and value in value_map:
                    props[field] = value_map[value]
                    feature_changed = True
            if feature_changed:
                changed_features += 1
                changed = True
            output_feature = {
                "geometry": feature["geometry"],
                "properties": props,
            }
            if feature.get("id") is not None:
                output_feature["id"] = feature["id"]
            features.append(output_feature)
        layers.append({"name": layer_name, "features": features})

    if changed:
        dst.parent.mkdir(parents=True, exist_ok=True)
        if dst.exists():
            dst.unlink()
        dst.write_bytes(encode_layers_preserving_extents(layers, source_extents))
        assert_reencoded_tile_safe(src, dst)
    else:
        hardlink_or_copy(src, dst)
    return changed, changed_features


def remap_tile_job(args: tuple[str, str, dict[str, str], list[bytes]]) -> tuple[bool, int]:
    src_str, dst_str, value_map, needles = args
    return remap_tile(Path(src_str), Path(dst_str), value_map, needles)


def backfill_missing_tiles(src_root: Path, dst_root: Path) -> int:
    missing = 0
    for src in src_root.rglob("*.pbf"):
        dst = dst_root / src.relative_to(src_root)
        if dst.exists():
            continue
        hardlink_or_copy(src, dst)
        missing += 1
    return missing


def build_semantic_pbf(rows: list[AuditRow], mapping: dict[str, str]) -> dict[str, int]:
    value_map = {
        row.old_sprite_key: mapping[row.old_sprite_key]
        for row in rows
        if row.used_in_pbf > 0 and row.old_sprite_key in mapping
    }
    needles = [old.encode("utf-8") for old in value_map]
    if OUTPUT_PBF_ROOT.exists():
        shutil.rmtree(OUTPUT_PBF_ROOT)
    OUTPUT_PBF_ROOT.parent.mkdir(parents=True, exist_ok=True)
    subprocess.run(
        ["cp", "-al", f"{SOURCE_PBF_ROOT}/.", str(OUTPUT_PBF_ROOT)],
        check=True,
    )

    rg_cmd = ["rg", "-a", "-l", "-F"]
    for old_key in value_map:
        rg_cmd.extend(["-e", old_key])
    rg_cmd.append(str(SOURCE_PBF_ROOT))
    result = subprocess.run(rg_cmd, check=True, capture_output=True, text=True)
    tile_paths = [Path(line) for line in result.stdout.splitlines() if line.strip()]

    stats = {
        "tiles_total": len(list(SOURCE_PBF_ROOT.rglob("*.pbf"))),
        "tiles_changed": 0,
        "features_changed": 0,
        "tiles_scanned_for_rewrite": len(tile_paths),
    }

    jobs = [
        (
            str(src),
            str(OUTPUT_PBF_ROOT / src.relative_to(SOURCE_PBF_ROOT)),
            value_map,
            needles,
        )
        for src in tile_paths
    ]

    with ProcessPoolExecutor(max_workers=min(os.cpu_count() or 4, 12)) as executor:
        for changed, feature_count in executor.map(remap_tile_job, jobs, chunksize=128):
            if changed:
                stats["tiles_changed"] += 1
                stats["features_changed"] += feature_count

    stats["tiles_backfilled"] = backfill_missing_tiles(SOURCE_PBF_ROOT, OUTPUT_PBF_ROOT)
    stats["tiles_total_after_backfill"] = len(list(OUTPUT_PBF_ROOT.rglob("*.pbf")))

    return stats


def main() -> None:
    rows = load_audit_rows()
    mapping = build_mapping(rows)
    used_entries = build_used_sprite_entries(rows, mapping)

    write_mapping_json(mapping)
    build_semantic_sprite(used_entries)
    build_semantic_style(mapping)
    pbf_stats = build_semantic_pbf(rows, mapping)

    summary = {
        "mapping_count": len(mapping),
        "used_sprite_entries": len(used_entries),
        "style_output": str(OUTPUT_STYLE),
        "deployed_style": str(DEPLOYED_STYLE),
        "sprite_output_dir": str(OUTPUT_SPRITE_DIR),
        "pbf_output_dir": str(OUTPUT_PBF_ROOT),
        **pbf_stats,
        "style_version": STYLE_VERSION,
        "pbf_version": PBF_VERSION,
        "sprite_version": SPRITE_VERSION,
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()