226 lines
8.9 KiB
Python
226 lines
8.9 KiB
Python
import json
|
||
from sqlalchemy.orm import Session
|
||
from database import engine, SessionLocal, Base
|
||
import models
|
||
import main # 导入原有假数据
|
||
import logging
|
||
|
||
# 设置日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 默认仪表盘数据（如果main.py中没有定义）
|
||
default_dashboard_data = {
|
||
"paperCount": 3500,
|
||
"patentCount": 2000,
|
||
"highImpactPapers": 100,
|
||
"keyProjects": 50,
|
||
"fundingAmount": "500万元",
|
||
"researcherStats": {
|
||
"academician": 12,
|
||
"chiefScientist": 28,
|
||
"distinguishedProfessor": 56,
|
||
"youngScientist": 120
|
||
},
|
||
"newsData": [
|
||
{"title": "我校科研团队在量子计算领域取得重大突破", "date": "2023-05-15"},
|
||
{"title": "张教授团队论文被Nature收录", "date": "2023-04-28"},
|
||
{"title": "校长率团访问美国麻省理工学院商讨合作事宜", "date": "2023-04-15"},
|
||
{"title": "我校获批3项国家重点研发计划", "date": "2023-03-20"},
|
||
{"title": "2023年度国家自然科学基金申请工作启动", "date": "2023-02-10"}
|
||
]
|
||
}
|
||
|
||
# 创建所有表
|
||
def create_tables():
|
||
Base.metadata.create_all(bind=engine)
|
||
logger.info("数据库表已创建")
|
||
|
||
# 导入用户数据
|
||
def import_users(db: Session):
|
||
# 检查是否已存在用户
|
||
existing_users = db.query(models.User).count()
|
||
if existing_users == 0:
|
||
for username, user_data in main.fake_users_db.items():
|
||
db_user = models.User(
|
||
username=user_data["username"],
|
||
email=user_data.get("email"),
|
||
full_name=user_data.get("full_name"),
|
||
hashed_password=user_data["hashed_password"],
|
||
disabled=user_data.get("disabled", False)
|
||
)
|
||
db.add(db_user)
|
||
|
||
db.commit()
|
||
logger.info("用户数据已导入")
|
||
else:
|
||
logger.info("用户数据已存在，跳过导入")
|
||
|
||
# 导入人才数据
|
||
def import_talents(db: Session):
|
||
# 检查是否已存在人才数据
|
||
existing_talents = db.query(models.Talent).count()
|
||
if existing_talents == 0:
|
||
for talent_data in main.talents:
|
||
# 处理evaluationData字段 - 确保是JSON格式
|
||
evaluation_data = talent_data.get("evaluationData")
|
||
|
||
db_talent = models.Talent(
|
||
id=talent_data["id"],
|
||
name=talent_data["name"],
|
||
gender=talent_data.get("gender"),
|
||
birthDate=talent_data.get("birthDate"),
|
||
title=talent_data.get("title"),
|
||
position=talent_data.get("position"),
|
||
education=talent_data.get("education"),
|
||
address=talent_data.get("address"),
|
||
academicDirection=talent_data.get("academicDirection"),
|
||
talentPlan=talent_data.get("talentPlan"),
|
||
officeLocation=talent_data.get("officeLocation"),
|
||
email=talent_data.get("email"),
|
||
phone=talent_data.get("phone"),
|
||
tutorType=talent_data.get("tutorType"),
|
||
papers=talent_data.get("papers"),
|
||
projects=talent_data.get("projects"),
|
||
photo=talent_data.get("photo"),
|
||
eduWorkHistory=talent_data.get("eduWorkHistory"),
|
||
researchDirection=talent_data.get("researchDirection"),
|
||
recentProjects=talent_data.get("recentProjects"),
|
||
representativePapers=talent_data.get("representativePapers"),
|
||
patents=talent_data.get("patents"),
|
||
evaluationData=evaluation_data
|
||
)
|
||
db.add(db_talent)
|
||
|
||
db.commit()
|
||
logger.info("人才数据已导入")
|
||
else:
|
||
logger.info("人才数据已存在，跳过导入")
|
||
|
||
# 导入工程研究中心数据
|
||
def import_labs(db: Session):
|
||
# 检查是否已存在工程研究中心数据
|
||
existing_labs = db.query(models.Lab).count()
|
||
if existing_labs == 0:
|
||
for lab_data in main.labs:
|
||
# 处理evaluationData字段 - 确保是JSON格式
|
||
evaluation_data = lab_data.get("evaluationData")
|
||
|
||
db_lab = models.Lab(
|
||
id=lab_data["id"],
|
||
name=lab_data["name"],
|
||
personnel=lab_data.get("personnel"),
|
||
nationalProjects=lab_data.get("nationalProjects"),
|
||
otherProjects=lab_data.get("otherProjects"),
|
||
achievements=lab_data.get("achievements"),
|
||
labAchievements=lab_data.get("labAchievements"),
|
||
image=lab_data.get("image"),
|
||
score=lab_data.get("score"),
|
||
evaluationData=evaluation_data
|
||
)
|
||
db.add(db_lab)
|
||
|
||
db.commit()
|
||
logger.info("工程研究中心数据已导入")
|
||
else:
|
||
logger.info("工程研究中心数据已存在，跳过导入")
|
||
|
||
# 导入仪表盘数据
|
||
def import_dashboard(db: Session):
|
||
# 检查是否已存在仪表盘数据
|
||
existing_dashboard = db.query(models.DashboardData).count()
|
||
if existing_dashboard == 0:
|
||
# 优先使用main.py中的数据，如果不存在则使用默认数据
|
||
dashboard_data = getattr(main, "dashboard_data", default_dashboard_data)
|
||
|
||
# 创建仪表盘记录
|
||
db_dashboard = models.DashboardData(
|
||
id=1, # 主键ID设为1
|
||
paperCount=dashboard_data["paperCount"],
|
||
patentCount=dashboard_data["patentCount"],
|
||
highImpactPapers=dashboard_data["highImpactPapers"],
|
||
keyProjects=dashboard_data["keyProjects"],
|
||
fundingAmount=dashboard_data["fundingAmount"],
|
||
researcherStats=dashboard_data["researcherStats"]
|
||
)
|
||
db.add(db_dashboard)
|
||
db.flush() # 立即写入数据库，获取ID
|
||
|
||
# 创建新闻数据记录
|
||
for news_item in dashboard_data["newsData"]:
|
||
db_news = models.News(
|
||
title=news_item["title"],
|
||
date=news_item["date"],
|
||
dashboard_id=db_dashboard.id
|
||
)
|
||
db.add(db_news)
|
||
|
||
db.commit()
|
||
logger.info("仪表盘和新闻数据已导入")
|
||
else:
|
||
logger.info("仪表盘数据已存在，跳过导入")
|
||
|
||
# 导入默认维度
|
||
def import_dimensions(db: Session):
|
||
# 人才评估维度
|
||
talent_dimensions = [
|
||
{"name": "学术成果", "weight": 1.0, "category": "talent", "description": "包括论文发表、专著、专利等"},
|
||
{"name": "科研项目", "weight": 1.0, "category": "talent", "description": "承担的科研项目数量和级别"},
|
||
{"name": "人才引进", "weight": 1.0, "category": "talent", "description": "引进的人才数量和质量"},
|
||
{"name": "学术影响力", "weight": 1.0, "category": "talent", "description": "学术引用和影响力指标"},
|
||
{"name": "教学质量", "weight": 1.0, "category": "talent", "description": "教学评估和学生反馈"},
|
||
{"name": "社会服务", "weight": 1.0, "category": "talent", "description": "社会服务和贡献"}
|
||
]
|
||
|
||
# 工程研究中心评估维度
|
||
lab_dimensions = [
|
||
{"name": "科研产出", "weight": 1.0, "category": "lab", "description": "工程研究中心科研成果产出"},
|
||
{"name": "人才培养", "weight": 1.0, "category": "lab", "description": "培养的研究生和博士后数量"},
|
||
{"name": "项目承担", "weight": 1.0, "category": "lab", "description": "承担的科研项目数量和级别"},
|
||
{"name": "设备利用", "weight": 1.0, "category": "lab", "description": "设备利用率和效益"},
|
||
{"name": "学术交流", "weight": 1.0, "category": "lab", "description": "国内外学术交流和合作"},
|
||
{"name": "社会服务", "weight": 1.0, "category": "lab", "description": "社会服务和贡献"}
|
||
]
|
||
|
||
# 检查是否已存在维度
|
||
existing_dimensions = db.query(models.Dimension).count()
|
||
if existing_dimensions == 0:
|
||
# 添加人才评估维度
|
||
for dim in talent_dimensions:
|
||
db_dimension = models.Dimension(**dim)
|
||
db.add(db_dimension)
|
||
|
||
# 添加工程研究中心评估维度
|
||
for dim in lab_dimensions:
|
||
db_dimension = models.Dimension(**dim)
|
||
db.add(db_dimension)
|
||
|
||
db.commit()
|
||
logger.info("默认评估维度已导入")
|
||
else:
|
||
logger.info("评估维度已存在，跳过导入")
|
||
|
||
# 主函数
|
||
def init_db():
|
||
# 创建表结构
|
||
create_tables()
|
||
|
||
# 获取数据库会话
|
||
db = SessionLocal()
|
||
try:
|
||
# 导入各类数据
|
||
import_users(db)
|
||
import_talents(db)
|
||
import_labs(db)
|
||
import_dashboard(db)
|
||
import_dimensions(db)
|
||
|
||
logger.info("数据库初始化完成！")
|
||
except Exception as e:
|
||
logger.error(f"数据库初始化失败: {e}")
|
||
finally:
|
||
db.close()
|
||
|
||
# 直接运行脚本时执行初始化
|
||
if __name__ == "__main__":
|
||
init_db() |