You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
390 lines
15 KiB
390 lines
15 KiB
#!/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()
|
|
|