#!/usr/bin/env python3
"""生成当前 NavSea PBF 的可取舍数据字典和分类统计。"""

from __future__ import annotations

import argparse
import csv
import json
from collections import Counter, defaultdict
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from typing import Any

import mapbox_vector_tile


TRACKED_FIELDS = (
    "class_code",
    "display_code",
    "shape_class_code",
    "canonical_object_type",
    "object_family",
    "detection_class",
    "chart_symbol_code",
    "chart_label_position_code",
    "chart_label_subtext",
    "chart_icon_image",
    "chart_fill_pattern",
    "chart_fill_style",
    "chart_text_color",
    "chart_text_style",
    "chart_line_color",
    "chart_line_width",
    "chart_line_pattern",
    "arc_id",
    "arc_source_code",
    "light_color_code",
    "light_sector_mode",
    "icon_id",
    "source_layer",
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--pbf-root",
        type=Path,
        default=Path("/home/wwwroot/pbf-delivery-full-semantic-unified-20260808"),
    )
    parser.add_argument(
        "--style",
        type=Path,
        default=Path("src/pbf/style.navsea-delivery-full-semantic-unified-20260808.json"),
    )
    parser.add_argument(
        "--report-root",
        type=Path,
        default=Path("report/current_pbf_inventory_full_semantic_20260809"),
    )
    parser.add_argument("--workers", type=int, default=4)
    return parser.parse_args()


def empty_result() -> dict[str, Any]:
    return {
        "tiles": Counter(),
        "layers": Counter(),
        "geometries": Counter(),
        "property_keys": Counter(),
        "property_key_layers": Counter(),
        "values": Counter(),
        "value_layers": Counter(),
    }


def scan_chunk(paths: list[str]) -> dict[str, Any]:
    result = empty_result()
    for raw_path in paths:
        path = Path(raw_path)
        rel = path.relative_to(path.parents[2])
        zoom = rel.parts[0]
        result["tiles"][f"z{zoom}"] += 1
        decoded = mapbox_vector_tile.decode(path.read_bytes())
        for layer_name, layer in decoded.items():
            features = layer.get("features", [])
            result["layers"][layer_name] += len(features)
            for feature in features:
                geometry_type = (feature.get("geometry") or {}).get("type", "<none>")
                result["geometries"][(layer_name, geometry_type)] += 1
                properties = feature.get("properties") or {}
                for key, value in properties.items():
                    result["property_keys"][key] += 1
                    result["property_key_layers"][(key, layer_name)] += 1
                    if key not in TRACKED_FIELDS or value is None:
                        continue
                    value_text = str(value)
                    result["values"][(key, value_text)] += 1
                    result["value_layers"][(key, value_text, layer_name)] += 1
    return result


def merge(results: list[dict[str, Any]]) -> dict[str, Counter]:
    merged = empty_result()
    for result in results:
        for key in merged:
            merged[key].update(result[key])
    return merged


def style_get_fields(style: dict[str, Any]) -> set[str]:
    fields: set[str] = set()

    def walk(value: Any) -> None:
        if isinstance(value, dict):
            for child in value.values():
                walk(child)
        elif isinstance(value, list):
            if (
                len(value) >= 2
                and isinstance(value[0], str)
                and value[0] in {"get", "has"}
                and isinstance(value[1], str)
            ):
                fields.add(value[1])
            for child in value:
                walk(child)

    walk(style)
    return fields


def write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, Any]]) -> None:
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    args = parse_args()
    paths = sorted(args.pbf_root.glob("*/*/*.pbf"))
    if not paths:
        raise SystemExit(f"no PBF tiles under {args.pbf_root}")
    chunk_size = 128
    chunks = [
        [str(path) for path in paths[start : start + chunk_size]]
        for start in range(0, len(paths), chunk_size)
    ]
    with ProcessPoolExecutor(max_workers=args.workers) as executor:
        results = list(executor.map(scan_chunk, chunks))
    stats = merge(results)

    args.report_root.mkdir(parents=True, exist_ok=True)
    style = json.loads(args.style.read_text(encoding="utf-8"))
    style_fields = style_get_fields(style)

    layer_rows = []
    for layer, feature_count in stats["layers"].most_common():
        geometries = Counter(
            {
                geometry: count
                for (layer_name, geometry), count in stats["geometries"].items()
                if layer_name == layer
            }
        )
        fields = sorted(
            key
            for (key, layer_name), count in stats["property_key_layers"].items()
            if layer_name == layer and count
        )
        layer_rows.append(
            {
                "layer": layer,
                "feature_count": feature_count,
                "geometry_counts": json.dumps(geometries, ensure_ascii=False),
                "property_count": len(fields),
                "properties": ",".join(fields),
            }
        )
    write_csv(
        args.report_root / "图层分类.csv",
        ["layer", "feature_count", "geometry_counts", "property_count", "properties"],
        layer_rows,
    )

    field_rows = []
    for key, count in stats["property_keys"].most_common():
        layer_counts = Counter(
            {
                layer: layer_count
                for (field, layer), layer_count in stats["property_key_layers"].items()
                if field == key
            }
        )
        field_rows.append(
            {
                "field": key,
                "feature_value_count": count,
                "layer_count": len(layer_counts),
                "layers": json.dumps(layer_counts, ensure_ascii=False),
                "used_by_current_style": "yes" if key in style_fields else "no",
            }
        )
    write_csv(
        args.report_root / "字段分类.csv",
        ["field", "feature_value_count", "layer_count", "layers", "used_by_current_style"],
        field_rows,
    )

    value_rows = []
    for (field, value), count in stats["values"].most_common():
        layer_counts = Counter(
            {
                layer: layer_count
                for (tracked_field, tracked_value, layer), layer_count in stats["value_layers"].items()
                if tracked_field == field and tracked_value == value
            }
        )
        value_rows.append(
            {
                "field": field,
                "value": value,
                "feature_count": count,
                "layers": json.dumps(layer_counts, ensure_ascii=False),
            }
        )
    write_csv(
        args.report_root / "字段值分类.csv",
        ["field", "value", "feature_count", "layers"],
        value_rows,
    )

    total_features = sum(stats["layers"].values())
    summary = {
        "pbf_root": str(args.pbf_root),
        "tile_count": len(paths),
        "tiles_by_zoom": dict(sorted(stats["tiles"].items())),
        "feature_count": total_features,
        "layer_count": len(stats["layers"]),
        "property_count": len(stats["property_keys"]),
        "style_get_fields": sorted(style_fields),
    }
    (args.report_root / "summary.json").write_text(
        json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )

    md: list[str] = [
        "# 当前统一 Full PBF 数据分类报告",
        "",
        f"- PBF 根目录：`{args.pbf_root}`",
        f"- 瓦片数：`{len(paths)}`",
        f"- feature 总数：`{total_features}`",
        f"- 图层数：`{len(stats['layers'])}`",
        f"- 属性字段数：`{len(stats['property_keys'])}`",
        "",
        "## 按 zoom 的瓦片数量",
        "",
        "| zoom | 瓦片数 |",
        "|---|---:|",
    ]
    md.extend(f"| {zoom} | {count} |" for zoom, count in sorted(stats["tiles"].items()))
    md.extend(["", "## 图层分类", "", "| 图层 | feature 数 | 几何类型 | 字段数 |", "|---|---:|---|---:|"])
    for row in layer_rows:
        md.append(
            f"| `{row['layer']}` | {row['feature_count']} | `{row['geometry_counts']}` | {row['property_count']} |"
        )
    md.extend(
        [
            "",
            "## 取舍说明",
            "",
            "- `geometry`、图层名、`canonical_object_type`、分类/显示码、`chart_*`、`arc_id`/`arc_source_code` 是渲染、对象审计或灯弧追溯的核心数据，不能在未做专项审计前删除。",
            "- 只被点击详情或外部业务使用、但不被当前 style 读取的字段，属于可选保留项；应先确认 pickup、审计和后续重建是否依赖。",
            "- 仅存在于 PBF、当前 style 不读取且不参与对象身份/分类/追溯的字段，才是候选删除项；具体字段和数量见 `字段分类.csv`。",
            "- 所有细分值见 `字段值分类.csv`，不要仅依据当前出现次数为 0 的 style 分支删除上游分类能力。",
            "",
            "## 明细文件",
            "",
            "- `图层分类.csv`：每个 source-layer 的 feature、几何和字段分布。",
            "- `字段分类.csv`：每个属性字段的出现量、所在图层和当前 style 是否读取。",
            "- `字段值分类.csv`：分类码、显示码、对象类型、图片、纹理和灯弧等字段的逐值统计。",
        ]
    )
    (args.report_root / "当前PBF数据分类报告.md").write_text("\n".join(md) + "\n", encoding="utf-8")
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()