#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 工程研究中心数据完整导入脚本 用于导入assets/工程研究中心.json中的所有工程研究中心数据到数据库 使用方法: 1. 确保后端服务未运行 2. 在backend目录下执行: python import_lab_data_full.py 主要功能: - 导入所有工程研究中心的基本信息 - 导入多年度数据到annual_data JSON字段 - 自动生成lab_id - 支持数据更新（如果lab已存在） """ import json import sys import os from pathlib import Path # 添加父目录到路径以便导入模块 sys.path.append(os.path.dirname(os.path.abspath(__file__))) from sqlalchemy.orm import sessionmaker from database import SessionLocal, engine from models import Lab import uuid def safe_int(value): """安全转换为整数""" if value is None or value == "": return 0 if isinstance(value, (int, float)): return int(value) if isinstance(value, str): try: return int(float(value)) except (ValueError, TypeError): return 0 return 0 def safe_float(value): """安全转换为浮点数""" if value is None or value == "": return 0.0 if isinstance(value, (int, float)): return float(value) if isinstance(value, str): try: return float(value) except (ValueError, TypeError): return 0.0 return 0.0 def safe_str(value): """安全转换为字符串""" if value is None: return "" return str(value) def import_lab_data(): """导入工程研究中心数据""" # 检查JSON文件是否存在 json_file = Path(__file__).parent.parent / "src" / "assets" / "工程研究中心.json" if not json_file.exists(): print(f"❌ 错误：找不到数据文件 {json_file}") return False try: # 读取JSON数据 print(f"📖 正在读取数据文件: {json_file}") with open(json_file, 'r', encoding='utf-8') as f: labs_data = json.load(f) print(f"✅ 成功读取数据，共 {len(labs_data)} 个工程研究中心") # 创建数据库会话 db = SessionLocal() imported_count = 0 updated_count = 0 error_count = 0 try: for lab_info in labs_data: try: center_name = lab_info.get("中心名称", "") center_number = lab_info.get("中心编号", "") annual_data = lab_info.get("年度数据", []) if not center_name: print(f"⚠️ 跳过无名称的工程研究中心") continue print(f"\n🔄 正在处理工程研究中心: {center_name} (编号: {center_number})") # 检查是否已存在 existing_lab = None if center_number: existing_lab = db.query(Lab).filter(Lab.center_number == center_number).first() if not existing_lab and center_name: existing_lab = db.query(Lab).filter(Lab.name == center_name).first() if existing_lab: print(f" 📝 工程研究中心已存在，更新数据...") lab = existing_lab updated_count += 1 else: print(f" ➕ 创建新工程研究中心...") lab = Lab() lab.id = str(uuid.uuid4()) imported_count += 1 # 设置基本信息 lab.name = center_name lab.center_number = center_number lab.idcode = center_number # 用编号作为显示ID # 处理年度数据 if annual_data: # 提取基本信息（从最新年度数据） latest_data = max(annual_data, key=lambda x: x.get("归属年份", "0")) lab.school = latest_data.get("所属学校", "") lab.department = latest_data.get("主管部门", "") lab.field = latest_data.get("所属领域", "") lab.current_year = latest_data.get("归属年份", "") # 存储多年度数据到annual_data JSON字段 lab.annual_data = json.dumps(annual_data, ensure_ascii=False) # 从最新数据中提取主要信息字段 lab.innovation_situation = latest_data.get("技术攻关与创新情况", "") lab.overall_situation = latest_data.get("1.总体情况", "") lab.engineering_cases = latest_data.get("2.工程化案例", "") lab.industry_service = latest_data.get("3.行业服务情况", "") lab.discipline_support = latest_data.get("1.学科发展支撑情况", "") or latest_data.get("1.支撑学科发展情况", "") lab.talent_cultivation = latest_data.get("2.人才培养情况", "") lab.team_building = latest_data.get("3.研究队伍建设情况", "") lab.department_support = latest_data.get("1.主管部门、依托单位支持情况", "") lab.equipment_sharing = latest_data.get("2.仪器设备开放共享情况", "") lab.academic_style = latest_data.get("3.学风建设情况", "") lab.technical_committee = latest_data.get("4.技术委员会工作情况", "") lab.next_year_plan = latest_data.get("下一年度工作计划", "") lab.problems_suggestions = latest_data.get("问题与建议", "") lab.director_opinion = latest_data.get("1.工程中心负责人意见", "") lab.institution_opinion = latest_data.get("2.依托单位意见", "") lab.research_directions = latest_data.get("研究方向/学术带头人", "") # 统计数据字段 lab.national_awards_first = safe_int(latest_data.get("国家级科技奖励一等奖(项)", 0)) lab.national_awards_second = safe_int(latest_data.get("国家级科技奖励二等奖(项)", 0)) lab.provincial_awards_first = safe_int(latest_data.get("省、部级科技奖励一等奖(项)", 0)) lab.provincial_awards_second = safe_int(latest_data.get("省、部级科技奖励二等奖(项)", 0)) lab.valid_patents = safe_int(latest_data.get("有效专利(项)", 0)) lab.other_ip = safe_int(latest_data.get("其他知识产权(项)", 0)) lab.international_standards = safe_int(latest_data.get("国际/国家标准(项)", 0)) lab.industry_standards = safe_int(latest_data.get("行业/地方标准(项)", 0)) # 专利转化数据 lab.patent_transfer_contracts = safe_int(latest_data.get("合同项数(项)", 0)) lab.patent_transfer_amount = safe_float(latest_data.get("合同金额(万元)", 0)) lab.patent_license_contracts = safe_int(latest_data.get("合同项数(项)_1", 0)) lab.patent_license_amount = safe_float(latest_data.get("合同金额(万元)_1", 0)) lab.patent_valuation_contracts = safe_int(latest_data.get("合同项数(项)_2", 0)) lab.patent_valuation_amount = safe_float(latest_data.get("作价金额(万元)", 0)) # 项目合作 lab.project_contracts = safe_int(latest_data.get("项目合同项数(项)", 0)) lab.project_amount = safe_float(latest_data.get("项目合同金额(万元)", 0)) # 学科信息 lab.discipline_1 = latest_data.get("依托学科1", "") lab.discipline_2 = latest_data.get("依托学科2", "") lab.discipline_3 = latest_data.get("依托学科3", "") # 人才培养数据 lab.doctoral_students = safe_int(latest_data.get("在读博士生", 0)) lab.master_students = safe_int(latest_data.get("在读硕士生", 0)) lab.graduated_doctoral = safe_int(latest_data.get("当年毕业博士", 0)) lab.graduated_master = safe_int(latest_data.get("当年毕业硕士", 0)) lab.undergraduate_courses = safe_int(latest_data.get("承担本科课程", 0)) lab.graduate_courses = safe_int(latest_data.get("承担研究生课程", 0)) lab.textbooks = safe_int(latest_data.get("大专院校教材", 0)) # 人员结构 lab.professors = safe_int(latest_data.get("科技人才-教授(人)", 0)) lab.associate_professors = safe_int(latest_data.get("科技人才-副教授(人)", 0)) lab.lecturers = safe_int(latest_data.get("科技人才-讲师(人)", 0)) lab.domestic_visitors = safe_int(latest_data.get("访问学者-国内(人)", 0)) lab.foreign_visitors = safe_int(latest_data.get("访问学者-国外(人)", 0)) lab.postdoc_in = safe_int(latest_data.get("本年度进站博士后(人)", 0)) lab.postdoc_out = safe_int(latest_data.get("本年度出站博士后(人)", 0)) # 基础设施 lab.center_area = safe_float(latest_data.get("工程中心面积(m²)", 0)) lab.new_area = safe_float(latest_data.get("当年新增面积(m²)", 0)) lab.fixed_personnel = safe_int(latest_data.get("固定人员(人)", 0)) lab.mobile_personnel = safe_int(latest_data.get("流动人员(人)", 0)) # 经费情况 lab.total_funding = safe_float(latest_data.get("当年项目到账总经费(万元)", 0)) lab.vertical_funding = safe_float(latest_data.get("纵向经费(万元)", 0)) lab.horizontal_funding = safe_float(latest_data.get("横向经费(万元)", 0)) # 服务情况 lab.technical_consultations = safe_int(latest_data.get("技术咨询(次)", 0)) lab.training_services = safe_int(latest_data.get("培训服务(人次)", 0)) # 设置兼容性字段 lab.personnel = f"{lab.fixed_personnel + lab.mobile_personnel}人" lab.nationalProjects = str(lab.project_contracts) lab.otherProjects = "0" lab.achievements = str(lab.valid_patents) lab.image = "/image/实验室1.png" # 默认图片 print(f" ✅ 年度数据: {len(annual_data)} 年, 最新年份: {lab.current_year}") # 添加到数据库（如果是新记录） if lab not in db.query(Lab).all(): db.add(lab) except Exception as e: print(f" ❌ 处理工程研究中心 {center_name} 时出错: {str(e)}") error_count += 1 continue # 提交更改 print(f"\n💾 正在保存到数据库...") db.commit() print(f"✅ 数据导入完成!") print(f"📊 统计信息:") print(f" - 新增工程研究中心: {imported_count}") print(f" - 更新工程研究中心: {updated_count}") print(f" - 错误数量: {error_count}") print(f" - 总计处理: {imported_count + updated_count}") return True except Exception as e: print(f"❌ 数据库操作失败: {str(e)}") db.rollback() return False finally: db.close() except Exception as e: print(f"❌ 导入失败: {str(e)}") return False def main(): """主函数""" print("🚀 开始导入工程研究中心数据...") print("=" * 50) success = import_lab_data() print("=" * 50) if success: print("🎉 导入完成！") else: print("💥 导入失败！") input("\n按回车键退出...") if __name__ == "__main__": main()