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.
278 lines
10 KiB
278 lines
10 KiB
#!/usr/bin/env python3
|
|
"""Build the artificial-grassland business summary used by the big screen.
|
|
|
|
Usage:
|
|
python3 scripts/prepare-artificial-grassland-business-data.py <source.xlsx>
|
|
|
|
The source workbook contains merged project rows. This script keeps the
|
|
workbook's total row as the reporting baseline and only uses the detail rows
|
|
to identify project coverage towns.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from collections import defaultdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
TARGET = ROOT / "src/views/yakIndustryChain/artificialGrasslandBusinessSummary.json"
|
|
TOTAL_ROW_LABEL = "合计面积"
|
|
TOWNS = [
|
|
("邛溪镇", ("邛溪",)),
|
|
("刷经寺镇", ("刷经寺",)),
|
|
("安曲镇", ("安曲",)),
|
|
("龙日镇", ("龙日",)),
|
|
("江茸乡", ("江茸",)),
|
|
("查尔玛乡", ("查尔玛",)),
|
|
("瓦切镇", ("瓦切",)),
|
|
("阿木乡", ("阿木",)),
|
|
("麦洼乡", ("麦洼",)),
|
|
("色地镇", ("色地",)),
|
|
]
|
|
|
|
|
|
def text(value: object) -> str:
|
|
return str(value or "").replace("\n", " ").strip()
|
|
|
|
|
|
def number(value: object) -> float:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def display_number(value: float) -> int | float:
|
|
rounded = round(value, 2)
|
|
return int(rounded) if rounded.is_integer() else rounded
|
|
|
|
|
|
def first_text(ws, rows, column: int) -> str:
|
|
for row in rows:
|
|
value = text(ws.cell(row, column).value)
|
|
if value:
|
|
return value
|
|
return ""
|
|
|
|
|
|
def first_number(ws, rows, column: int) -> float:
|
|
for row in rows:
|
|
value = number(ws.cell(row, column).value)
|
|
if value:
|
|
return value
|
|
return 0.0
|
|
|
|
|
|
def sum_numbers(ws, rows, column: int) -> float:
|
|
return sum(number(ws.cell(row, column).value) for row in rows)
|
|
|
|
|
|
def project_year(name: str, notes: list[str]) -> str:
|
|
year_match = re.search(r"20(2[0-9])", name)
|
|
if year_match:
|
|
return f"20{year_match.group(1)}"
|
|
note_text = " ".join(notes)
|
|
year_match = re.search(r"20(2[0-9])", note_text)
|
|
if year_match:
|
|
return f"20{year_match.group(1)}"
|
|
short_match = re.search(r"(?<!\d)(2[0-9])年", note_text)
|
|
if short_match:
|
|
return f"20{short_match.group(1)}"
|
|
return "待核定"
|
|
|
|
|
|
def project_type(name: str) -> str:
|
|
if "草种基地" in name:
|
|
return "草种基地"
|
|
if "草地资源保护" in name:
|
|
return "草地资源保护"
|
|
return "高产稳产饲草基地"
|
|
|
|
|
|
def project_short_name(name: str, year: str, category: str) -> str:
|
|
if year in {"2022", "2023", "2024"}:
|
|
if category == "高产稳产饲草基地":
|
|
return f"{year}饲草基地"
|
|
return f"{year}{category}"
|
|
return category
|
|
|
|
|
|
def contract_area(requirement: str) -> float:
|
|
values = [number(item) for item in re.findall(r"(\d+(?:\.\d+)?)\s*亩", requirement)]
|
|
return sum(values)
|
|
|
|
|
|
def locate_towns(locations: list[str]) -> list[str]:
|
|
found = []
|
|
for location in locations:
|
|
# A location can mention a direction or a park in another town. Prefer
|
|
# the explicit town name first, then use a prefix only when no town is
|
|
# written in the location itself.
|
|
explicit = []
|
|
for town, aliases in TOWNS:
|
|
if any(f"{alias}镇" in location or f"{alias}乡" in location for alias in aliases):
|
|
explicit.append(town)
|
|
candidates = explicit
|
|
if not candidates:
|
|
normalized = location.lstrip("（( ")
|
|
candidates = [
|
|
town
|
|
for town, aliases in TOWNS
|
|
if any(normalized.startswith(alias) for alias in aliases)
|
|
]
|
|
for town in candidates:
|
|
if town not in found:
|
|
found.append(town)
|
|
return found
|
|
|
|
|
|
def make_row(key: str, name: str, value: float, unit: str, extra: str = "") -> dict:
|
|
return {
|
|
"key": key,
|
|
"name": name,
|
|
"value": display_number(value),
|
|
"unit": unit,
|
|
"extra": extra,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) < 2:
|
|
raise SystemExit("Usage: python3 scripts/prepare-artificial-grassland-business-data.py <source.xlsx>")
|
|
|
|
source = Path(sys.argv[1]).expanduser().resolve()
|
|
workbook = load_workbook(source, data_only=True)
|
|
sheet = workbook.active
|
|
total_row = next(
|
|
(
|
|
row
|
|
for row in range(2, sheet.max_row + 1)
|
|
if TOTAL_ROW_LABEL in text(sheet.cell(row, 1).value)
|
|
),
|
|
None,
|
|
)
|
|
if not total_row:
|
|
raise RuntimeError("Cannot find the workbook total row.")
|
|
|
|
starts = [
|
|
row
|
|
for row in range(2, total_row)
|
|
if text(sheet.cell(row, 1).value)
|
|
]
|
|
projects = []
|
|
for index, start in enumerate(starts):
|
|
end = starts[index + 1] - 1 if index + 1 < len(starts) else total_row - 1
|
|
rows = list(range(start, end + 1))
|
|
name = text(sheet.cell(start, 1).value)
|
|
requirement = first_text(sheet, rows, 5)
|
|
locations = [text(sheet.cell(row, 6).value) for row in rows if text(sheet.cell(row, 6).value)]
|
|
notes = [text(sheet.cell(row, 13).value) for row in rows if text(sheet.cell(row, 13).value)]
|
|
year = project_year(name, notes)
|
|
category = project_type(name)
|
|
actual_area = sum_numbers(sheet, rows, 7)
|
|
contract = contract_area(requirement)
|
|
project = {
|
|
"key": f"artificial-grassland-project-{index + 1}",
|
|
"name": project_short_name(name, year, category),
|
|
"fullName": name,
|
|
"year": year,
|
|
"category": category,
|
|
"value": display_number(actual_area),
|
|
"unit": "亩",
|
|
"actualAreaMu": display_number(actual_area),
|
|
"contractAreaMu": display_number(contract),
|
|
"investmentWanYuan": display_number(first_number(sheet, rows, 4)),
|
|
"owner": first_text(sheet, rows, 2),
|
|
"contractor": first_text(sheet, rows, 3),
|
|
"requirement": requirement,
|
|
"towns": locate_towns(locations),
|
|
"locations": locations,
|
|
"notes": notes,
|
|
"canLocate": False,
|
|
"searchText": " ".join([name, year, category, *locations, *notes]),
|
|
}
|
|
projects.append(project)
|
|
|
|
actual_total = number(sheet.cell(total_row, 7).value)
|
|
contract_total = sum(number(project["contractAreaMu"]) for project in projects)
|
|
investment_total = number(sheet.cell(total_row, 4).value)
|
|
town_project_count = defaultdict(int)
|
|
for project in projects:
|
|
for town in project["towns"]:
|
|
town_project_count[town] += 1
|
|
|
|
year_area = defaultdict(float)
|
|
type_area = defaultdict(float)
|
|
for project in projects:
|
|
if project["year"] in {"2022", "2023", "2024"}:
|
|
year_area[project["year"]] += number(project["actualAreaMu"])
|
|
type_area[project["category"]] += number(project["actualAreaMu"])
|
|
|
|
investment_rows = [
|
|
make_row(project["key"], project["name"], number(project["investmentWanYuan"]), "万元", project["year"])
|
|
for project in projects
|
|
if number(project["investmentWanYuan"]) > 0
|
|
]
|
|
investment_rows.sort(key=lambda item: number(item["value"]), reverse=True)
|
|
|
|
summary = {
|
|
"generatedAt": datetime.now(timezone.utc).isoformat(),
|
|
"sourceLabel": source.name,
|
|
"sourceSheet": sheet.title,
|
|
"sourceNote": "按汇总表合计行作为大屏面积口径；图斑资料仅用于地图定位，不作为统计主口径。",
|
|
"summary": {
|
|
"actualAreaMu": display_number(actual_total),
|
|
"contractAreaMu": display_number(contract_total),
|
|
"areaDifferenceMu": display_number(actual_total - contract_total),
|
|
"projectCount": len(projects),
|
|
"coveredTownCount": len(town_project_count),
|
|
"constructionYearCount": len(year_area),
|
|
"investmentRecordCount": len(investment_rows),
|
|
"investmentWanYuan": display_number(investment_total),
|
|
"tillageDisturbanceMu": display_number(number(sheet.cell(total_row, 8).value)),
|
|
"slightDisturbanceMu": display_number(number(sheet.cell(total_row, 9).value)),
|
|
"tillageMu": display_number(number(sheet.cell(total_row, 10).value)),
|
|
"drillSowingMu": display_number(number(sheet.cell(total_row, 11).value)),
|
|
"noTillOverseedingMu": display_number(number(sheet.cell(total_row, 12).value)),
|
|
},
|
|
"projectRows": projects,
|
|
"yearRows": [
|
|
make_row(f"year-{year}", f"{year}年", year_area[year], "亩", "实测种草面积")
|
|
for year in sorted(year_area)
|
|
],
|
|
"typeRows": [
|
|
make_row(f"type-{category}", category, value, "亩", "实测种草面积")
|
|
for category, value in sorted(type_area.items(), key=lambda item: item[1], reverse=True)
|
|
],
|
|
"operationRows": [
|
|
make_row("tillage", "实际翻耕", number(sheet.cell(total_row, 10).value), "亩", "表内实测记录"),
|
|
make_row("drill-sowing", "条播", number(sheet.cell(total_row, 11).value), "亩", "表内实测记录"),
|
|
make_row("no-till-overseeding", "免耕补播", number(sheet.cell(total_row, 12).value), "亩", "表内实测记录"),
|
|
],
|
|
"disturbanceRows": [
|
|
make_row("disturbance-total", "翻耕及扰动", number(sheet.cell(total_row, 8).value), "亩", "施工扰动面积"),
|
|
make_row("disturbance-light", "轻微扰动", number(sheet.cell(total_row, 9).value), "亩", "施工扰动面积"),
|
|
make_row("disturbance-tillage", "实际翻耕", number(sheet.cell(total_row, 10).value), "亩", "施工扰动面积"),
|
|
],
|
|
"townRows": [
|
|
make_row(f"town-{town}", town, town_project_count[town], "项", "涉及种草项目")
|
|
for town in town_project_count
|
|
],
|
|
"investmentRows": investment_rows,
|
|
}
|
|
|
|
TARGET.parent.mkdir(parents=True, exist_ok=True)
|
|
TARGET.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(f"Generated {TARGET}")
|
|
print(json.dumps(summary["summary"], ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|