#!/usr/bin/env python3 """Build the family ecological pasture dataset used by the Yak Industry Chain screen. The historical construction ledger is the authoritative source. The 2025 owner sheet only enriches matched 2025 ledger rows with map coordinates; it never changes the construction-record count. """ from __future__ import annotations import argparse import json import os import re import shutil import subprocess import tempfile from collections import defaultdict from pathlib import Path import openpyxl SCRIPT_ROOT = Path(__file__).resolve().parents[1] DEFAULT_SUMMARY = Path( "/Users/mocker/Documents/hy/数据清单 2/家庭牧场/红原县2023年至今建设家庭牧场汇总表.xls" ) DEFAULT_POINTS = Path( "/Users/mocker/Documents/hy/数据清单 2/家庭牧场/红原县2025年适度规模化（生态家庭牧场）家庭牧场项目-户主信息表.xlsx" ) DEFAULT_OUTPUT = SCRIPT_ROOT / "src/views/yakIndustryChain/familyPastureData.json" 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"), "/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_dms_pair(value: object) -> tuple[float, float] | None: text = clean(value).replace("\n", " ").replace("′", "'").replace("″", '"') if not text: return None longitude = re.search(r"[Ee]\s*(\d+)°\s*(\d+)'\s*(\d+(?:\.\d+)?)\"?", text) latitude = re.search(r"[Nn]\s*(\d+)°\s*(\d+)'\s*(\d+(?:\.\d+)?)\"?", text) if not longitude or not latitude: return None lng = int(longitude.group(1)) + int(longitude.group(2)) / 60 + float(longitude.group(3)) / 3600 lat = int(latitude.group(1)) + int(latitude.group(2)) / 60 + float(latitude.group(3)) / 3600 if not (101 <= lng <= 105 and 31 <= lat <= 35): return None return round(lng, 8), round(lat, 8) def parse_points(points_path: Path) -> list[dict]: workbook = openpyxl.load_workbook(points_path, data_only=True) sheet = workbook.worksheets[0] rows = [] town = "" for row in sheet.iter_rows(min_row=3, values_only=True): point_no = number(row[0] if len(row) > 0 else None) if point_no is None: continue town = clean(row[1]) or town point_name, owner = clean(row[2]), clean(row[3]) greenhouse_location = parse_dms_pair(row[5] if len(row) > 5 else None) management_location = parse_dms_pair(row[6] if len(row) > 6 else None) village = clean(row[7] if len(row) > 7 else None) rows.append( { "pointNo": int(point_no), "town": town, "pointName": point_name, "owner": owner, "village": village, "greenhouseLocation": greenhouse_location, "managementLocation": management_location, } ) return rows def record_key(record: dict) -> tuple[str, str, str]: return key_text(record["town"]), key_text(record["village"]), key_text(record["owner"]) def attach_locations(records: list[dict], point_rows: list[dict]) -> list[dict]: available_by_key: dict[tuple[str, str, str], list[dict]] = defaultdict(list) available_by_town_village: dict[tuple[str, str], list[dict]] = defaultdict(list) for record in records: if record["constructionYear"] != 2025: continue available_by_key[record_key(record)].append(record) available_by_town_village[(key_text(record["town"]), key_text(record["village"]))].append(record) source_coordinate_count = sum( 1 for point in point_rows if point["greenhouseLocation"] or point["managementLocation"] ) matched_records: set[int] = set() match_count = 0 for point in point_rows: if not (point["greenhouseLocation"] or point["managementLocation"]): continue candidates = available_by_key.get(record_key(point), []) target = next((candidate for candidate in candidates if id(candidate) not in matched_records), None) if target is None and not point["owner"]: town_village_key = key_text(point["town"]), key_text(point["village"]) target = next( (candidate for candidate in available_by_town_village.get(town_village_key, []) if id(candidate) not in matched_records), None, ) if target is None: raise RuntimeError( f"无法将有坐标的第 {point['pointNo']} 号点匹配至 2025 正式台账：" f"{point['town']} {point['village']} {point['owner'] or '未填写户主'}" ) matched_records.add(id(target)) match_count += 1 target["pointNo"] = point["pointNo"] target["pointName"] = point["pointName"] target["greenhouseLocation"] = point["greenhouseLocation"] target["managementLocation"] = point["managementLocation"] if match_count != source_coordinate_count: raise RuntimeError( f"定位补充匹配数量异常：坐标表有 {source_coordinate_count} 条有效坐标，" f"实际匹配 {match_count} 条。" ) return records def format_area(value: int | float | None) -> str: if value is None: return "--" return f"{value:g}" 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)}㎡活动场地及通道", ] ) locations = {} if record.get("greenhouseLocation"): locations["暖棚"] = { "longitude": record["greenhouseLocation"][0], "latitude": record["greenhouseLocation"][1], } if record.get("managementLocation"): locations["管理用房"] = { "longitude": record["managementLocation"][0], "latitude": record["managementLocation"][1], } primary = locations.get("管理用房") or locations.get("暖棚") source_no = record["sourceNo"] point_no = record.get("pointNo") identifier = ( f"YC-FAMILY-PASTURE-{point_no:03d}" if point_no is not None else f"YC-FAMILY-PASTURE-SUM-{source_no:03d}" ) owner = record["owner"] name = f"{owner}家庭生态牧场" if owner else f"{record['town']}{record['village']}{source_no}号家庭生态牧场" metrics = { "标准化暖棚 (㎡)": greenhouse, "管理用房 (㎡)": management, "饲草料库 (㎡)": fodder, "活动场地及通道 (㎡)": activity, } if support_mu is not None: metrics["配套草料地 (亩)"] = support_mu attributes = { "设施配置": facility_detail, "草料配套": support_text, } if owner: attributes["户主"] = owner return { "id": identifier, "name": name, "town": record["town"], "village": record["village"], "owner": owner, "constructionYear": record["constructionYear"], "sourceNo": source_no, "pointName": record.get("pointName", ""), "facilityBrief": facility_brief, "facilityDetail": facility_detail, "grassSupportType": record["grassSupportType"], "grassSupportMu": support_mu, "grassSupportLowerBound": record["grassSupportLowerBound"], "grassSupportText": support_text, "longitude": primary["longitude"] if primary else None, "latitude": primary["latitude"] if primary else None, "locations": locations, "metrics": metrics, "attributes": attributes, "sourceSheet": "红原县2023年至今建设家庭牧场汇总表", } def verify_records(records: list[dict], expected_coordinate_count: int) -> 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 != expected_coordinate_count: raise RuntimeError(f"可定位台账数量异常：期望 {expected_coordinate_count} 条，实际 {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 main() -> None: parser = argparse.ArgumentParser(description="生成家庭生态牧场大屏数据") parser.add_argument("--summary", type=Path, default=DEFAULT_SUMMARY) parser.add_argument("--points", type=Path, default=DEFAULT_POINTS) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) args = parser.parse_args() if not args.summary.exists(): raise FileNotFoundError(f"未找到历史台账：{args.summary}") if not args.points.exists(): raise FileNotFoundError(f"未找到 2025 定位表：{args.points}") point_rows = parse_points(args.points) expected_coordinate_count = sum( 1 for point in point_rows if point["greenhouseLocation"] or point["managementLocation"] ) records = attach_locations(parse_summary(args.summary), point_rows) display_records = [build_display_record(record) for record in records] verify_records(display_records, expected_coordinate_count) payload = { "datasetName": "红原县家庭生态牧场建设台账", "sourceSummary": args.summary.name, "sourceCoordinates": args.points.name, "recordUnit": "处", "recordCount": len(display_records), "locatedRecordCount": sum(1 for record in display_records if record.get("longitude") and record.get("latitude")), "records": display_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") print( f"已生成 {args.output}: {payload['recordCount']} 条正式台账，" f"{payload['locatedRecordCount']} 条可定位。" ) if __name__ == "__main__": main()