#!/usr/bin/env python3 """Build the family ecological pasture dataset used by the Yak Industry Chain screen. The construction ledger is the authoritative record list. All display coordinates come from the map marker workbook, then get matched back to ledger records by administrative township and village. """ from __future__ import annotations import argparse import json import math import os import re import shutil import subprocess import tempfile from collections import Counter, defaultdict from datetime import datetime, timezone, timedelta from pathlib import Path import openpyxl SCRIPT_ROOT = Path(__file__).resolve().parents[1] DEFAULT_SUMMARY = Path( "/Users/mocker/Documents/hy/数据清单(1)/家庭牧场/红原县2023年至今建设家庭牧场汇总表.xls" ) DEFAULT_OWNER_POINTS = Path( "/Users/mocker/Documents/hy/数据清单(1)/家庭牧场/红原县2025年适度规模化（生态家庭牧场）家庭牧场项目-户主信息表.xlsx" ) DEFAULT_MARKER_POINTS = Path( "/Users/mocker/Library/Containers/com.tencent.xinWeChat/Data/Documents/xwechat_files/" "wxid_k77f4cxbgk8y52_8fdc/msg/file/2026-07/地图标记家庭牧场点位.xlsx" ) DEFAULT_OUTPUT = SCRIPT_ROOT / "src/views/yakIndustryChain/familyPastureData.json" DEFAULT_GEOJSON_OUTPUT = SCRIPT_ROOT / "public/datas/yak-industry-chain/yak_industry_points.geojson" DEFAULT_ADMIN_VILLAGES = SCRIPT_ROOT / "public/datas/geojsons/hongyuan-villages.geojson" DEFAULT_ADMIN_TOWNSHIPS = SCRIPT_ROOT / "public/datas/geojsons/hongyuan-townships.geojson" VILLAGE_ALIASES = { ("瓦切镇", "色永村"): ("瓦切镇", "色尔永村"), ("色地镇", "让里村"): ("色地镇", "壤里村"), ("龙日镇", "龙日坝村"): ("龙日镇", "四川省龙日种畜场"), } def clean(value: object) -> str: return "" if value is None else str(value).strip() def key_text(value: object) -> str: return re.sub(r"\s+", "", clean(value)) def number(value: object) -> int | float | None: if value is None or value == "": return None try: value = float(value) except (TypeError, ValueError): return None return int(value) if value.is_integer() else value def extract_area(text: str, patterns: list[str]) -> int | float | None: for pattern in patterns: matched = re.search(pattern, text, re.IGNORECASE) if matched: return number(matched.group(1)) return None def parse_facilities(text: str) -> dict: normalized = re.sub(r"\s+", "", text or "") greenhouse = extract_area( normalized, [ r"(?:标准化|标准分类)?(?:养殖)?暖棚(?:达到)?(\d+(?:\.\d+)?)(?:㎡|平方米)", r"暖棚(?:达到)?(\d+(?:\.\d+)?)(?:㎡|平方米)", ], ) management = extract_area(normalized, [r"(?:生产)?管理用房(?:达到)?(\d+(?:\.\d+)?)(?:㎡|平方米)"]) fodder = extract_area(normalized, [r"(?:饲草料库|草料库|贮草库|储草料库)(?:达到)?(\d+(?:\.\d+)?)(?:㎡|平方米)"]) activity = extract_area( normalized, [ r"运动场[^0-9]*(\d+(?:\.\d+)?)(?:㎡|平方米)", r"巷道圈(\d+(?:\.\d+)?)(?:㎡|平方米)", r"分区敞圈(\d+(?:\.\d+)?)(?:㎡|平方米)", ], ) support_type = "" support_mu = None support_lower_bound = False cutting = extract_area(normalized, [r"割草地(\d+(?:\.\d+)?)(?:亩)"]) storage = extract_area(normalized, [r"(?:储草基地|储草料基地)[^0-9]*(\d+(?:\.\d+)?)(?:亩)"]) net_fence = extract_area(normalized, [r"(?:种草)?网围栏(\d+(?:\.\d+)?)(?:亩)"]) if cutting is not None: support_type, support_mu = "割草地", cutting elif storage is not None: support_type, support_mu, support_lower_bound = "储草基地", storage, "不低于" in normalized elif net_fence is not None: support_type, support_mu = "网围栏种草", net_fence return { "greenhouseArea": greenhouse, "managementArea": management, "fodderArea": fodder, "activityArea": activity, "grassSupportType": support_type, "grassSupportMu": support_mu, "grassSupportLowerBound": support_lower_bound, } def find_soffice() -> str: candidates = [ os.getenv("SOFFICE_PATH"), "/Users/mocker/.cache/codex-runtimes/codex-primary-runtime/dependencies/bin/override/soffice", "/Applications/LibreOffice.app/Contents/MacOS/soffice", shutil.which("soffice"), ] for candidate in candidates: if candidate and Path(candidate).exists(): return candidate raise RuntimeError("未找到 LibreOffice soffice，无法读取 .xls 历史台账。") def convert_xls_to_xlsx(summary_path: Path, work_dir: Path) -> Path: output_path = work_dir / f"{summary_path.stem}.xlsx" subprocess.run( [find_soffice(), "--headless", "--convert-to", "xlsx", "--outdir", str(work_dir), str(summary_path)], check=True, capture_output=True, text=True, ) if not output_path.exists(): raise RuntimeError("历史台账转换失败，未生成 xlsx 文件。") return output_path def parse_summary(summary_path: Path) -> list[dict]: with tempfile.TemporaryDirectory(prefix="hy-family-pasture-") as temp_dir: converted = convert_xls_to_xlsx(summary_path, Path(temp_dir)) workbook = openpyxl.load_workbook(converted, data_only=True) sheet = workbook.worksheets[0] records = [] for row in sheet.iter_rows(min_row=4, values_only=True): source_no = number(row[0] if len(row) > 0 else None) if source_no is None: continue town, village, owner = clean(row[1]), clean(row[2]), clean(row[3]) year = number(row[4] if len(row) > 4 else None) content = clean(row[5] if len(row) > 5 else None) if not town or not village or not year: continue facilities = parse_facilities(content) records.append( { "sourceNo": int(source_no), "town": town, "village": village, "owner": owner, "constructionYear": int(year), **facilities, } ) return records def parse_marker_points(marker_path: Path) -> list[dict]: workbook = openpyxl.load_workbook(marker_path, data_only=True) for sheet in workbook.worksheets: rows = list(sheet.iter_rows(values_only=True)) for header_index, header_row in enumerate(rows): headers = [key_text(value) for value in header_row] if "序号" not in headers or ("经纬度" not in headers and not {"经度", "纬度"}.issubset(set(headers))): continue header_map = {name: index for index, name in enumerate(headers) if name} marker_rows = [] for row in rows[header_index + 1:]: point_no = number(value_at(row, header_map.get("序号"))) if point_no is None: continue coordinate = parse_marker_coordinate(row, header_map) if coordinate is None: continue longitude, latitude = coordinate marker_rows.append( { "markerNo": int(point_no), "longitude": longitude, "latitude": latitude, "sourceSheet": sheet.title, } ) if marker_rows: return marker_rows raise RuntimeError(f"未能在地图标记文件中识别有效经纬度：{marker_path}") def value_at(row: tuple, index: int | None) -> object: if index is None or index >= len(row): return None return row[index] def parse_marker_coordinate(row: tuple, header_map: dict[str, int]) -> tuple[float, float] | None: longitude = number(value_at(row, header_map.get("经度"))) latitude = number(value_at(row, header_map.get("纬度"))) if longitude is None or latitude is None: text = clean(value_at(row, header_map.get("经纬度"))) matched = re.search(r"point\(\s*([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s*\)", text, re.IGNORECASE) if matched: longitude, latitude = float(matched.group(1)), float(matched.group(2)) if longitude is None or latitude is None: return None longitude, latitude = float(longitude), float(latitude) if not (101 <= longitude <= 105 and 31 <= latitude <= 35): return None return round(longitude, 8), round(latitude, 8) def load_geojson(path: Path) -> dict: if not path.exists(): return {"type": "FeatureCollection", "features": []} return json.loads(path.read_text(encoding="utf-8")) def load_admin_areas(villages_path: Path, townships_path: Path) -> tuple[list[dict], list[dict]]: village_areas = [] for feature in load_geojson(villages_path).get("features", []): geometry = feature.get("geometry") if not geometry: continue props = feature.get("properties") or {} town = clean(props.get("town") or props.get("townName") or props.get("town_name")) village = clean(props.get("name") or props.get("village") or props.get("villageName")) if not town or not village: continue village_areas.append( { "town": town, "village": village, "code": clean(props.get("code")), "geometry": geometry, "bbox": geometry_bbox(geometry), "center": geometry_center(geometry), } ) township_areas = [] for feature in load_geojson(townships_path).get("features", []): geometry = feature.get("geometry") if not geometry: continue props = feature.get("properties") or {} town = clean(props.get("name") or props.get("town") or props.get("townName")) if not town: continue center = props.get("center") or props.get("cp") or geometry_center(geometry) township_areas.append( { "town": town, "geometry": geometry, "bbox": geometry_bbox(geometry), "center": normalize_point(center) or geometry_center(geometry), } ) return village_areas, township_areas def normalize_point(value: object) -> tuple[float, float] | None: if not isinstance(value, (list, tuple)) or len(value) < 2: return None try: return float(value[0]), float(value[1]) except (TypeError, ValueError): return None def geometry_bbox(geometry: dict) -> tuple[float, float, float, float]: points = list(iter_geometry_points(geometry)) if not points: return (math.inf, math.inf, -math.inf, -math.inf) longitudes = [point[0] for point in points] latitudes = [point[1] for point in points] return min(longitudes), min(latitudes), max(longitudes), max(latitudes) def geometry_center(geometry: dict) -> tuple[float, float] | None: west, south, east, north = geometry_bbox(geometry) if not all(math.isfinite(value) for value in (west, south, east, north)): return None return (west + east) / 2, (south + north) / 2 def iter_geometry_points(geometry: dict): geometry_type = geometry.get("type") coordinates = geometry.get("coordinates") or [] if geometry_type == "Point": yield coordinates elif geometry_type == "Polygon": for ring in coordinates: yield from ring elif geometry_type == "MultiPolygon": for polygon in coordinates: for ring in polygon: yield from ring def bbox_contains(bbox: tuple[float, float, float, float], longitude: float, latitude: float) -> bool: west, south, east, north = bbox return west <= longitude <= east and south <= latitude <= north def point_in_geometry(geometry: dict, longitude: float, latitude: float) -> bool: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates") or [] point = (longitude, latitude) if geometry_type == "Polygon": return point_in_polygon(point, coordinates) if geometry_type == "MultiPolygon": return any(point_in_polygon(point, polygon) for polygon in coordinates) return False def point_in_polygon(point: tuple[float, float], polygon: list) -> bool: if not polygon or not point_in_ring(point, polygon[0]): return False return not any(point_in_ring(point, hole) for hole in polygon[1:]) def point_in_ring(point: tuple[float, float], ring: list) -> bool: x, y = point inside = False if len(ring) < 3: return False previous = ring[-1] for current in ring: x1, y1 = float(previous[0]), float(previous[1]) x2, y2 = float(current[0]), float(current[1]) if point_on_segment(x, y, x1, y1, x2, y2): return True intersects = (y1 > y) != (y2 > y) if intersects: x_intersection = (x2 - x1) * (y - y1) / ((y2 - y1) or 1e-30) + x1 if x <= x_intersection: inside = not inside previous = current return inside def point_on_segment(x: float, y: float, x1: float, y1: float, x2: float, y2: float) -> bool: cross = (x - x1) * (y2 - y1) - (y - y1) * (x2 - x1) if abs(cross) > 1e-10: return False return min(x1, x2) - 1e-10 <= x <= max(x1, x2) + 1e-10 and min(y1, y2) - 1e-10 <= y <= max(y1, y2) + 1e-10 def assign_admin_to_markers(marker_points: list[dict], village_areas: list[dict], township_areas: list[dict]) -> list[dict]: assigned = [] for marker in marker_points: longitude, latitude = marker["longitude"], marker["latitude"] village_area = first_containing_area(village_areas, longitude, latitude) township_area = None if village_area else first_containing_area(township_areas, longitude, latitude) admin_town = village_area["town"] if village_area else township_area["town"] if township_area else "" admin_village = village_area["village"] if village_area else "" assigned.append( { **marker, "adminTown": admin_town, "adminVillage": admin_village, "adminVillageCode": village_area["code"] if village_area else "", "adminMatchLevel": "village" if village_area else "town" if township_area else "outside", } ) return assigned def first_containing_area(areas: list[dict], longitude: float, latitude: float) -> dict | None: for area in areas: if bbox_contains(area["bbox"], longitude, latitude) and point_in_geometry(area["geometry"], longitude, latitude): return area return None def aliased_location(town: str, village: str) -> tuple[str, str]: return VILLAGE_ALIASES.get((clean(town), clean(village)), (clean(town), clean(village))) def town_key(town: str) -> str: return key_text(town) def town_village_key(town: str, village: str) -> tuple[str, str]: aliased_town, aliased_village = aliased_location(town, village) return town_key(aliased_town), key_text(aliased_village) def marker_town_key(marker: dict) -> str: return town_key(marker.get("adminTown", "")) def marker_town_village_key(marker: dict) -> tuple[str, str]: return town_key(marker.get("adminTown", "")), key_text(marker.get("adminVillage", "")) def build_center_indexes(village_areas: list[dict], township_areas: list[dict]) -> tuple[dict[tuple[str, str], tuple[float, float]], dict[str, tuple[float, float]]]: village_centers = { town_village_key(area["town"], area["village"]): area["center"] for area in village_areas if area.get("center") } town_centers = { town_key(area["town"]): area["center"] for area in township_areas if area.get("center") } return village_centers, town_centers def attach_marker_locations(records: list[dict], markers: list[dict], village_areas: list[dict], township_areas: list[dict]) -> tuple[list[dict], Counter]: if len(markers) < len(records): raise RuntimeError(f"地图标记点数量不足：台账 {len(records)} 条，标记点 {len(markers)} 个。") by_town_village: dict[tuple[str, str], list[int]] = defaultdict(list) by_town: dict[str, list[int]] = defaultdict(list) for index, marker in enumerate(markers): if marker.get("adminTown"): by_town_village[marker_town_village_key(marker)].append(index) by_town[marker_town_key(marker)].append(index) village_centers, town_centers = build_center_indexes(village_areas, township_areas) assignments: list[tuple[dict, str] | None] = [None] * len(records) unused = set(range(len(markers))) stats: Counter = Counter() def assign(record_index: int, candidates: list[int], level: str, center: tuple[float, float] | None) -> None: candidate = choose_marker(candidates, markers, unused, center) if candidate is None: return unused.remove(candidate) assignments[record_index] = (markers[candidate], level) stats[level] += 1 for record_index, record in enumerate(records): key = town_village_key(record["town"], record["village"]) assign(record_index, by_town_village.get(key, []), "townVillage", village_centers.get(key)) for record_index, record in enumerate(records): if assignments[record_index] is not None: continue key = town_key(record["town"]) assign(record_index, by_town.get(key, []), "town", town_centers.get(key)) for record_index, record in enumerate(records): if assignments[record_index] is not None: continue key = town_key(record["town"]) assign(record_index, list(unused), "fallback", town_centers.get(key)) matched_records = [] for record, assignment in zip(records, assignments): if assignment is None: raise RuntimeError(f"台账第 {record['sourceNo']} 条未能匹配地图标记点。") marker, match_level = assignment matched_records.append( { **record, "markerNo": marker["markerNo"], "markerLongitude": marker["longitude"], "markerLatitude": marker["latitude"], "markerAdminTown": marker.get("adminTown", ""), "markerAdminVillage": marker.get("adminVillage", ""), "markerAdminVillageCode": marker.get("adminVillageCode", ""), "markerAdminMatchLevel": marker.get("adminMatchLevel", ""), "coordinateMatchLevel": match_level, } ) return matched_records, stats def choose_marker(candidates: list[int], markers: list[dict], unused: set[int], center: tuple[float, float] | None) -> int | None: available = [index for index in candidates if index in unused] if not available: return None if center: return min(available, key=lambda index: (distance_sq(markers[index], center), markers[index]["markerNo"])) return min(available, key=lambda index: markers[index]["markerNo"]) def distance_sq(marker: dict, center: tuple[float, float]) -> float: longitude, latitude = center return (marker["longitude"] - longitude) ** 2 + (marker["latitude"] - latitude) ** 2 def format_area(value: int | float | None) -> str: if value is None: return "--" return f"{value:g}" def format_match_text(record: dict) -> str: labels = { "townVillage": "同镇同村", "town": "同镇", "fallback": "回退匹配", } marker_area = " · ".join([record.get("markerAdminTown", ""), record.get("markerAdminVillage", "")]).strip(" ·") marker_label = f"地图标记{record['markerNo']}号点" if marker_area: marker_label = f"{marker_label}（{marker_area}）" return f"{labels.get(record.get('coordinateMatchLevel'), '匹配')}：{marker_label}" def build_display_record(record: dict) -> dict: greenhouse = record["greenhouseArea"] management = record["managementArea"] fodder = record["fodderArea"] activity = record["activityArea"] support_mu = record["grassSupportMu"] support_prefix = "≥" if record["grassSupportLowerBound"] else "" support_text = ( f"{record['grassSupportType']} {support_prefix}{format_area(support_mu)}亩" if record["grassSupportType"] and support_mu is not None else "未配置" ) facility_brief = f"{format_area(greenhouse)}㎡暖棚 / {format_area(management)}㎡管理房" facility_detail = " / ".join( [ f"{format_area(greenhouse)}㎡标准化暖棚", f"{format_area(management)}㎡管理用房", f"{format_area(fodder)}㎡饲草料库", f"{format_area(activity)}㎡活动场地及通道", ] ) source_no = record["sourceNo"] marker_no = record["markerNo"] owner = record["owner"] identifier = f"YC-FAMILY-PASTURE-{source_no:03d}" name = f"{owner}家庭生态牧场" if owner else f"{record['town']}{record['village']}{source_no}号家庭生态牧场" match_text = format_match_text(record) locations = { "地图标记": { "longitude": record["markerLongitude"], "latitude": record["markerLatitude"], "markerNo": marker_no, } } metrics = { "标准化暖棚 (㎡)": greenhouse, "管理用房 (㎡)": management, "饲草料库 (㎡)": fodder, "活动场地及通道 (㎡)": activity, } if support_mu is not None: metrics["配套草料地 (亩)"] = support_mu attributes = { "设施配置": facility_detail, "草料配套": support_text, "地图标记": f"{marker_no}号点", "点位匹配": match_text, } marker_area = " · ".join([record.get("markerAdminTown", ""), record.get("markerAdminVillage", "")]).strip(" ·") if marker_area: attributes["标记归属"] = marker_area if owner: attributes["户主"] = owner return { "id": identifier, "name": name, "town": record["town"], "village": record["village"], "owner": owner, "constructionYear": record["constructionYear"], "sourceNo": source_no, "pointName": f"地图标记{marker_no}号点", "matchedPointNo": marker_no, "matchedPointTown": record.get("markerAdminTown", ""), "matchedPointVillage": record.get("markerAdminVillage", ""), "coordinateMatchLevel": record.get("coordinateMatchLevel", ""), "coordinateMatchText": match_text, "facilityBrief": facility_brief, "facilityDetail": facility_detail, "grassSupportType": record["grassSupportType"], "grassSupportMu": support_mu, "grassSupportLowerBound": record["grassSupportLowerBound"], "grassSupportText": support_text, "longitude": record["markerLongitude"], "latitude": record["markerLatitude"], "locations": locations, "metrics": metrics, "attributes": attributes, "sourceSheet": "红原县2023年至今建设家庭牧场汇总表", } def build_marker_record(marker: dict, matched_marker_numbers: set[int]) -> dict: marker_no = marker["markerNo"] town = marker.get("adminTown", "") or "未识别乡镇" village = marker.get("adminVillage", "") marker_label = f"地图标记{marker_no}号点" return { "id": f"YC-FAMILY-PASTURE-MARKER-{marker_no:03d}", "name": marker_label, "town": town, "village": village, "markerNo": marker_no, "longitude": marker["longitude"], "latitude": marker["latitude"], "markerOnly": marker_no not in matched_marker_numbers, "adminMatchLevel": marker.get("adminMatchLevel", ""), "sourceSheet": marker.get("sourceSheet", ""), } def verify_records(records: list[dict]) -> None: if len(records) != 252: raise RuntimeError(f"正式建设台账数量异常：期望 252 条，实际 {len(records)} 条。") coordinate_count = sum(1 for record in records if record.get("longitude") and record.get("latitude")) if coordinate_count != len(records): raise RuntimeError(f"可定位台账数量异常：期望 {len(records)} 条，实际 {coordinate_count} 条。") expected_totals = { "标准化暖棚 (㎡)": 61600, "管理用房 (㎡)": 11920, "饲草料库 (㎡)": 12000, "活动场地及通道 (㎡)": 108800, } for metric, expected in expected_totals.items(): actual = sum(record["metrics"].get(metric, 0) or 0 for record in records) if actual != expected: raise RuntimeError(f"{metric} 汇总异常：期望 {expected}，实际 {actual}。") def build_family_features(records: list[dict], markers: list[dict]) -> list[dict]: features = [] matched_marker_numbers = set() for record in records: matched_marker_numbers.add(record["matchedPointNo"]) features.append( { "type": "Feature", "properties": { "id": record["id"], "name": record["name"], "owner": record["owner"], "constructionYear": record["constructionYear"], "stageKey": "family-pasture", "stageName": "家庭生态牧场", "categoryCode": "YC-FAMILY-PASTURE", "categoryName": "家庭生态牧场", "town": record["town"], "village": record["village"], "matchedPointNo": record["matchedPointNo"], "pointName": record["pointName"], "matchedPointTown": record.get("matchedPointTown", ""), "matchedPointVillage": record.get("matchedPointVillage", ""), "coordinateMatchLevel": record.get("coordinateMatchLevel", ""), "coordinateMatchText": record.get("coordinateMatchText", ""), "address": record["village"], "capacity": 1, "capacityUnit": "处", "contactName": record["owner"], "contactPhone": "", "status": f"{record['constructionYear']}年建设", "locationType": "地图标记", "sourceSheet": "地图标记家庭牧场点位", }, "geometry": { "type": "Point", "coordinates": [record["longitude"], record["latitude"]], }, } ) for marker in sorted(markers, key=lambda item: item["markerNo"]): if marker["markerNo"] in matched_marker_numbers: continue marker_area = " · ".join([marker.get("adminTown", ""), marker.get("adminVillage", "")]).strip(" ·") marker_label = f"地图标记{marker['markerNo']}号点" features.append( { "type": "Feature", "properties": { "id": f"YC-FAMILY-PASTURE-MARKER-{marker['markerNo']:03d}", "name": marker_label, "stageKey": "family-pasture", "stageName": "家庭生态牧场", "categoryCode": "YC-FAMILY-PASTURE", "categoryName": "家庭生态牧场", "town": marker.get("adminTown", "") or "未识别乡镇", "village": marker.get("adminVillage", ""), "address": marker.get("adminVillage", "") or marker.get("adminTown", ""), "matchedPointNo": marker["markerNo"], "pointName": marker_label, "matchedPointTown": marker.get("adminTown", ""), "matchedPointVillage": marker.get("adminVillage", ""), "coordinateMatchLevel": "markerOnly", "coordinateMatchText": f"地图原始标记点（未匹配建设台账）{f'：{marker_area}' if marker_area else ''}", "capacity": 1, "capacityUnit": "处", "contactName": "", "contactPhone": "", "status": "地图标记点位", "locationType": "地图标记", "markerOnly": True, "markerAdminMatchLevel": marker.get("adminMatchLevel", ""), "sourceSheet": "地图标记家庭牧场点位", }, "geometry": { "type": "Point", "coordinates": [marker["longitude"], marker["latitude"]], }, } ) return features def write_industry_points_geojson(output_path: Path, records: list[dict], markers: list[dict]) -> int: family_features = build_family_features(records, markers) existing = load_geojson(output_path) preserved = [ feature for feature in existing.get("features", []) if feature.get("properties", {}).get("categoryCode") != "YC-FAMILY-PASTURE" ] payload = { "type": "FeatureCollection", "name": existing.get("name") or "yak_industry_points", "generatedAt": generated_at(), "features": [*family_features, *preserved], } output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return len(family_features) def generated_at() -> str: china_tz = timezone(timedelta(hours=8)) return datetime.now(china_tz).isoformat(timespec="seconds") def main() -> None: parser = argparse.ArgumentParser(description="生成家庭生态牧场大屏数据") parser.add_argument("--summary", type=Path, default=DEFAULT_SUMMARY) parser.add_argument("--owner-points", type=Path, default=DEFAULT_OWNER_POINTS) parser.add_argument("--markers", type=Path, default=DEFAULT_MARKER_POINTS) parser.add_argument("--admin-villages", type=Path, default=DEFAULT_ADMIN_VILLAGES) parser.add_argument("--admin-townships", type=Path, default=DEFAULT_ADMIN_TOWNSHIPS) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) parser.add_argument("--geojson-output", type=Path, default=DEFAULT_GEOJSON_OUTPUT) args = parser.parse_args() if not args.summary.exists(): raise FileNotFoundError(f"未找到历史台账：{args.summary}") if not args.markers.exists(): raise FileNotFoundError(f"未找到地图标记点位表：{args.markers}") village_areas, township_areas = load_admin_areas(args.admin_villages, args.admin_townships) marker_points = assign_admin_to_markers(parse_marker_points(args.markers), village_areas, township_areas) records, match_stats = attach_marker_locations(parse_summary(args.summary), marker_points, village_areas, township_areas) display_records = [build_display_record(record) for record in records] verify_records(display_records) matched_marker_numbers = {record["matchedPointNo"] for record in display_records} marker_records = [build_marker_record(marker, matched_marker_numbers) for marker in marker_points] payload = { "datasetName": "红原县家庭生态牧场建设台账", "sourceSummary": args.summary.name, "sourceCoordinates": args.markers.name, "sourceOwnerCoordinates": args.owner_points.name if args.owner_points.exists() else "", "sourceAdminAreas": args.admin_villages.name if args.admin_villages.exists() else args.admin_townships.name, "recordUnit": "处", "recordCount": len(display_records), "locatedRecordCount": sum(1 for record in display_records if record.get("longitude") and record.get("latitude")), "markerPointCount": len(marker_points), "renderedMarkerPointCount": len(marker_points), "unmatchedMarkerPointCount": max(0, len(marker_points) - len(display_records)), "coordinateMatchSummary": { "townVillage": match_stats.get("townVillage", 0), "town": match_stats.get("town", 0), "fallback": match_stats.get("fallback", 0), }, "records": display_records, "markerPoints": marker_records, } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") family_feature_count = write_industry_points_geojson(args.geojson_output, display_records, marker_points) print( f"已生成 {args.output}: {payload['recordCount']} 条正式台账，" f"{payload['locatedRecordCount']} 条使用地图标记点定位。" ) print( "点位匹配：" f"同镇同村 {payload['coordinateMatchSummary']['townVillage']} 条，" f"同镇 {payload['coordinateMatchSummary']['town']} 条，" f"回退 {payload['coordinateMatchSummary']['fallback']} 条。" ) print( f"已更新 {args.geojson_output}: {family_feature_count} 条家庭生态牧场地图点位，" f"其中 {payload['unmatchedMarkerPointCount']} 条为未匹配台账的地图原始标记。" ) if __name__ == "__main__": main()