'commit'

2025-09-10 14:00:21 +08:00
parent 54fae48629
commit 4d77d0a881
1 changed files with 109 additions and 142 deletions
--- a/Tools/T5_ZhaoShengCount.py
+++ b/Tools/T5_ZhaoShengCount.py
@@ -1,22 +1,34 @@
-import openpyxl
-import json
 import os
 import traceback
-from openpyxl.utils import column_index_from_string
+from typing import List, Dict, Any, Tuple
+import openpyxl

 from Config.Config import EXCEL_PATH
-from Util.AreaUtil import query_area_info
+from Util.DataUtil import (
+    init_directories, process_value, print_conversion_stats,
+    convert_area_name, save_to_json, load_workbook_sheet
+)

-# 创建数据保存目录
+# ======================== 配置常量 ========================
+# 数据目录和JSON路径
 DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'Data')
-os.makedirs(DATA_DIR, exist_ok=True)
 JSON_PATH = os.path.join(DATA_DIR, 'ZhaoShengCount.json')

-# 教育阶段配置 - 招生数(2015-2024年)
-education_stages = [
+# 工作表名称
+SHEET_NAME = '招生数'
+
+# 起始行（跳过前4行表头）
+START_ROW = 5
+
+# 年份范围
+YEAR_RANGE = [2015, 2016, 2017, 2018, 2019, 2020, 2021, 2022, 2023, 2024]
+
+# 教育阶段配置
+EDUCATION_STAGES = [
    {
        'name': 'preschool',
        'chinese_name': '学前教育',
+        'categories': ['urban', 'town', 'rural', 'total'],
        'columns': [
            {'year': 2015, 'urban': 'D', 'town': 'E', 'rural': 'F', 'total': 'G'},
            {'year': 2016, 'urban': 'H', 'town': 'I', 'rural': 'J', 'total': 'K'},
@@ -33,6 +45,7 @@ education_stages = [
    {
        'name': 'primary',
        'chinese_name': '小学教育',
+        'categories': ['urban', 'town', 'rural', 'total'],
        'columns': [
            {'year': 2015, 'urban': 'AR', 'town': 'AS', 'rural': 'AT', 'total': 'AU'},
            {'year': 2016, 'urban': 'AV', 'town': 'AW', 'rural': 'AX', 'total': 'AY'},
@@ -49,6 +62,7 @@ education_stages = [
    {
        'name': 'junior',
        'chinese_name': '初中教育',
+        'categories': ['urban', 'town', 'rural', 'total'],
        'columns': [
            {'year': 2015, 'urban': 'CF', 'town': 'CG', 'rural': 'CH', 'total': 'CI'},
            {'year': 2016, 'urban': 'CJ', 'town': 'CK', 'rural': 'CL', 'total': 'CM'},
@@ -65,6 +79,7 @@ education_stages = [
    {
        'name': 'senior',
        'chinese_name': '高中教育',
+        'categories': ['urban', 'town', 'rural', 'total'],
        'columns': [
            {'year': 2015, 'urban': 'DT', 'town': 'DU', 'rural': 'DV', 'total': 'DW'},
            {'year': 2016, 'urban': 'DX', 'town': 'DY', 'rural': 'DZ', 'total': 'EA'},
@@ -81,6 +96,7 @@ education_stages = [
    {
        'name': 'vocational',
        'chinese_name': '中职教育',
+        'categories': ['total'],
        'columns': [
            {'year': 2015, 'total': 'FH'},
            {'year': 2016, 'total': 'FI'},
@@ -123,148 +139,99 @@ def process_value(value):
        return 0


-def main():
-    file_name = EXCEL_PATH
-    enrollment_data = []
-    name_conversion_errors = []
-    conversion_records = []
+def extract_enrollment_data(sheet: openpyxl.worksheet.worksheet.Worksheet) -> Tuple[List[Dict[str, Any]], List[Dict[str, str]], List[str]]:
+    """提取招生数据并处理区域名称转换
+
+    Args:
+        sheet: 工作表对象
+
+    Returns:
+        Tuple包含:
+        - enrollment_data: 提取的招生数据列表
+        - conversion_records: 区域名称转换记录
+        - errors: 错误信息列表
+    """
+    enrollment_data: List[Dict[str, Any]] = []
+    conversion_records: List[Dict[str, str]] = []
+    errors: List[str] = []
    processed_count = 0

-    try:
-        # 加载工作簿并选择招生数Sheet
-        workbook = openpyxl.load_workbook(file_name, data_only=True)
-        if '招生数' not in workbook.sheetnames:
-            print("❌ 错误：未找到'招生数'Sheet")
-            return
-        sheet = workbook['招生数']
+    for row_num, row in enumerate(sheet.iter_rows(min_row=START_ROW, values_only=True), start=START_ROW):
+        # 获取区域名称（B列，索引1）
+        raw_name = row[1] if (len(row) > 1 and row[1] is not None) else '未知地区'
+        if not raw_name:  # 跳过空行
+            continue

-        print(f"✅ 成功加载Excel文件：{file_name}")
+        # 转换区域名称
+        area_name, area_code, row_conversions, row_errors = convert_area_name(raw_name, row_num)
+        conversion_records.extend(row_conversions)
+        errors.extend(row_errors)
+
+        # 创建区域数据对象
+        area_info = {
+            'area_name': area_name,
+            'area_code': area_code,
+            'raw_name': raw_name,
+            'education_data': {}
+        }
+
+        # 提取各教育阶段招生数据
+        for stage in EDUCATION_STAGES:
+            stage_name = stage['name']
+            stage_data = {}
+            categories = stage['categories']
+
+            for year_config in stage['columns']:
+                year = year_config['year']
+                year_data = {}
+
+                for category in categories:
+                    col_name = year_config[category]
+                    col_idx = openpyxl.utils.column_index_from_string(col_name) - 1
+                    value = row[col_idx] if col_idx < len(row) else None
+                    year_data[category] = process_value(value)
+
+                stage_data[str(year)] = year_data
+
+            area_info['education_data'][stage_name] = stage_data
+
+        enrollment_data.append(area_info)
+        processed_count += 1
+
+        # 进度提示（每10行）
+        if processed_count % 10 == 0:
+            print(f"🔄 已处理 {processed_count} 条数据...")
+
+    return enrollment_data, conversion_records, errors
+
+# ======================== 主函数 ========================
+
+def main() -> None:
+    """主函数：加载数据并执行提取流程"""
+    try:
+        # 初始化目录
+        init_directories(DATA_DIR)
+
+        # 加载工作表
+        sheet = load_workbook_sheet(EXCEL_PATH, SHEET_NAME)
+        if not sheet:
+            return
+
+        print(f"✅ 成功加载Excel文件：{EXCEL_PATH}")
        print(f"✅ 开始处理招生数数据，共{sheet.max_row}行数据")

-        # 遍历行数据
-        for row_idx, row in enumerate(sheet.iter_rows(values_only=True), start=1):
-            # 跳过前4行表头
-            if row_idx < 5:
-                continue
+        # 提取招生数据
+        enrollment_data, conversion_records, errors = extract_enrollment_data(sheet)

-            # 从B列获取区域名称（索引1）
-            try:
-                # 检查行是否有足够的列
-                if len(row) < 2:
-                    print(f"⚠️ 第{row_idx}行数据不足，跳过")
-                    continue
+        # 打印转换统计
+        print_conversion_stats(conversion_records, errors)

-                raw_name = row[1]
-                if raw_name is None:
-                    print(f"⚠️ 第{row_idx}行B列区域名称为空，跳过")
-                    continue
+        # 保存数据到JSON
+        save_to_json(enrollment_data, JSON_PATH)

-                raw_name = str(raw_name).strip()
-                if not raw_name:
-                    print(f"⚠️ 第{row_idx}行B列区域名称为空字符串，跳过")
-                    continue
-
-                # 查询区域信息
-                area_info = query_area_info(raw_name)
-                area_name = raw_name
-                area_code = 'unknown'
-
-                # 验证区域信息
-                if isinstance(area_info, dict):
-                    if 'full_name' in area_info and 'area_code' in area_info:
-                        area_name = area_info['full_name']
-                        area_code = area_info['area_code']
-                        conversion_records.append(f"✅ 第{row_idx}行: {raw_name} → {area_name}")
-                        processed_count += 1
-                    else:
-                        name_conversion_errors.append(f"第{row_idx}行: {raw_name} (缺少必要字段)")
-                        conversion_records.append(f"❌ 第{row_idx}行: {raw_name} (格式错误)")
-                else:
-                    name_conversion_errors.append(f"第{row_idx}行: {raw_name}")
-                    conversion_records.append(f"❌ 第{row_idx}行: {raw_name} (未找到匹配)")
-
-                # 创建区域数据对象
-                area_data = {
-                    'area_name': area_name,
-                    'area_code': area_code,
-                    'raw_name': raw_name,
-                    'education_data': {}
-                }
-
-                # 处理各教育阶段数据
-                for stage in education_stages:
-                    stage_name = stage['name']
-                    stage_data = {}
-
-                    for year_config in stage['columns']:
-                        year = year_config['year']
-                        year_data = {}
-
-                        # 处理学前、小学、初中、高中（有城区/镇区/乡村分类）
-                        if 'urban' in year_config:
-                            # 城区
-                            urban_col = column_index_from_string(year_config['urban']) - 1
-                            urban_val = row[urban_col] if len(row) > urban_col else None
-                            year_data['urban'] = process_value(urban_val)
-
-                            # 镇区
-                            town_col = column_index_from_string(year_config['town']) - 1
-                            town_val = row[town_col] if len(row) > town_col else None
-                            year_data['town'] = process_value(town_val)
-
-                            # 乡村
-                            rural_col = column_index_from_string(year_config['rural']) - 1
-                            rural_val = row[rural_col] if len(row) > rural_col else None
-                            year_data['rural'] = process_value(rural_val)
-
-                            # 总计
-                            total_col = column_index_from_string(year_config['total']) - 1
-                            total_val = row[total_col] if len(row) > total_col else None
-                            year_data['total'] = process_value(total_val)
-                        # 处理中职教育（单列）
-                        else:
-                            total_col = column_index_from_string(year_config['total']) - 1
-                            total_val = row[total_col] if len(row) > total_col else None
-                            year_data['total'] = process_value(total_val)
-
-                        stage_data[str(year)] = year_data
-
-                    area_data['education_data'][stage_name] = stage_data
-
-                enrollment_data.append(area_data)
-
-                # 进度提示
-                if processed_count % 10 == 0 and processed_count > 0:
-                    print(f"🔄 已处理{processed_count}条数据...")
-
-            except Exception as e:
-                print(f"🔴 处理第{row_idx}行时发生错误：{str(e)}")
-                continue
-
-        # 保存数据到JSON文件
-        with open(JSON_PATH, 'w', encoding='utf-8') as f:
-            json.dump(enrollment_data, f, ensure_ascii=False, indent=2)
-
-        print("\n=== 数据处理完成 ===")
-        print(f"📊 共处理 {processed_count} 条地区数据")
-        print(f"✅ 区域名称转换成功: {processed_count - len(name_conversion_errors)}")
-        if name_conversion_errors:
-            print(f"❌ 区域名称转换失败: {len(name_conversion_errors)}个")
-            for error in name_conversion_errors[:5]:
-                print(f"  - {error}")
-            if len(name_conversion_errors) > 5:
-                print(f"  - ... 等{len(name_conversion_errors) - 5}个错误")
-
-        print(f"💾 数据已保存至 {JSON_PATH}")
-
-    except FileNotFoundError:
-        print(f"🔴 错误：Excel文件 '{file_name}' 不存在")
    except Exception as e:
-        print(f"🔴 处理数据时发生错误：{str(e)}{traceback.format_exc()}")
-    finally:
-        if 'workbook' in locals():
-            workbook.close()
+        print(f"🔴 处理数据时发生错误：{str(e)}")
+        print(traceback.format_exc())

-
-if __name__ == "__main__":
+if __name__ == '__main__':
    main()