Files
YunNanProject/Tools/T1_RenKou.py
2025-09-10 13:53:34 +08:00

93 lines
3.3 KiB
Python

import openpyxl
import os
from typing import List, Dict, Any
from Config.Config import EXCEL_PATH
from Util.DataUtil import (
init_directories, process_value, print_conversion_stats,
convert_area_name, save_to_json, load_workbook_sheet
)
# ======================= 配置常量 =======================
DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'Data')
JSON_PATH = os.path.join(DATA_DIR, 'RenKou.json')
SHEET_NAME = '人口' # 工作表名称
REGION_NAME_COLUMN = 'A' # 区域名称所在列
START_ROW = 3 # 数据起始行
YEAR_RANGE = range(2015, 2025) # 年份范围
# 数据列配置 (指标: (起始列, 结束列))
DATA_COLUMNS = {
'total_population': ('B', 'K'), # 年末总人口
'urban_population': ('L', 'U'), # 城镇人口
'rural_population': ('V', 'AE'), # 乡村人口
'urbanization_rate': ('AF', 'AO'), # 城镇化率
'birth_population': ('AP', 'AY') # 出生人口
}
# ======================= 核心逻辑 =======================
def extract_area_data(sheet: openpyxl.worksheet.worksheet.Worksheet) -> List[Dict[str, Any]]:
"""从工作表提取区域数据"""
population_data: List[Dict[str, Any]] = []
conversion_records: List[Dict[str, str]] = []
name_conversion_errors: List[str] = []
# 计算区域名称列索引
region_col_idx = openpyxl.utils.column_index_from_string(REGION_NAME_COLUMN) - 1
# 遍历数据行
for row_num in range(START_ROW, sheet.max_row + 1):
row = sheet[row_num]
# 获取区域名称并转换
raw_name = row[region_col_idx].value
area_name, area_code, conv_records, errors = convert_area_name(raw_name, row_num)
conversion_records.extend(conv_records)
name_conversion_errors.extend(errors)
if not area_name:
continue
# 构建区域数据
area_data = {
'area_name': area_name,
'area_code': area_code,
'raw_name': str(raw_name).strip() if raw_name else '未知地区'
}
# 提取各指标年度数据
for metric, (start_col, end_col) in DATA_COLUMNS.items():
start_idx = openpyxl.utils.column_index_from_string(start_col) - 1
end_idx = openpyxl.utils.column_index_from_string(end_col) - 1
year_data = {}
for col_idx, year in zip(range(start_idx, end_idx + 1), YEAR_RANGE):
cell_value = row[col_idx].value
year_data[str(year)] = process_value(cell_value)
area_data[metric] = year_data
population_data.append(area_data)
# 输出转换统计
print_conversion_stats(conversion_records, name_conversion_errors)
return population_data
# ======================= 主函数 =======================
def main() -> None:
"""人口数据提取主函数"""
init_directories(DATA_DIR)
# 加载工作表
sheet = load_workbook_sheet(EXCEL_PATH, SHEET_NAME)
if not sheet:
return
# 提取并处理数据
population_data = extract_area_data(sheet)
# 保存结果
save_to_json(population_data, JSON_PATH)
if __name__ == '__main__':
main()