270 lines
13 KiB
Python
270 lines
13 KiB
Python
#!/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() |