157 lines
5.1 KiB
Python
157 lines
5.1 KiB
Python
===================== 配置（只改这里）=====================
|
||
MYSQL_CONFIG = {
|
||
"host": "101.132.169.38",
|
||
"port": 13307,
|
||
"user": "xyuser2023",
|
||
"password": "tkS,72yu&cbT",
|
||
"db": "figs",
|
||
"charset": "utf8mb4"
|
||
}
|
||
import json
|
||
import datetime
|
||
import decimal
|
||
import pymysql
|
||
from pymongo import MongoClient
|
||
|
||
|
||
|
||
|
||
MONGO_URI = "mongodb://localhost:27017/"
|
||
MONGO_DB = "学生数据库"
|
||
MONGO_COLLECTION = "学生全量信息"
|
||
# ============================================================
|
||
# 格式化：时间 + Decimal 全兼容
|
||
def format_val(val):
|
||
if isinstance(val, datetime.datetime):
|
||
return val.strftime("%Y-%m-%d %H:%M:%S")
|
||
if isinstance(val, datetime.date):
|
||
return val.strftime("%Y-%m-%d")
|
||
if isinstance(val, decimal.Decimal):
|
||
return float(val)
|
||
return val
|
||
|
||
# 表名 → 中文
|
||
TABLE_CN = {
|
||
"student_info": "学生基础信息",
|
||
"student_remark_info": "学生附加信息",
|
||
"student_book_lesson": "学生约课记录",
|
||
"student_grades": "学生成绩信息",
|
||
"student_materials_info": "学生材料信息",
|
||
"student_feedback_info": "学生反馈信息",
|
||
"student_yuetan_info": "学生约谈信息",
|
||
"student_yuyue_info": "学生预约信息",
|
||
"student_service_report": "学生服务月报",
|
||
"student_activity_registration": "学生活动报名",
|
||
"college_admission_details": "大学录取明细",
|
||
"college_application_process": "大学申请流程",
|
||
"college_application_scheme": "大学申请方案",
|
||
"high_school_admission": "高中录取信息",
|
||
"high_school_application_info": "高中申请信息",
|
||
"middle_school_admission": "中学录取信息",
|
||
"document_info": "文书信息",
|
||
"contracts": "合同信息",
|
||
"contract_payments": "合同付款记录",
|
||
"contract_refund_info": "合同退款信息",
|
||
"contract_service_contents": "合同服务内容",
|
||
"contract_personalized_services": "合同个性化服务",
|
||
"course_changes": "课时变更记录",
|
||
"course_exchange": "课程兑换",
|
||
"course_exchange_record": "课程兑换记录"
|
||
}
|
||
|
||
# 字段名 → 中文（全覆盖你的库）
|
||
FIELD_CN = {
|
||
"id": "主键ID",
|
||
"student_id": "学生ID",
|
||
"name": "姓名",
|
||
"en_name": "英文名",
|
||
"school_name": "学校",
|
||
"gender": "性别",
|
||
"id_card_num": "身份证号",
|
||
"parent_mobile_phone": "家长电话",
|
||
"grade": "年级",
|
||
"student_status": "状态",
|
||
"contract_number": "合同编号",
|
||
"contract_signed_date": "签订日期",
|
||
"contract_amount": "合同金额",
|
||
"contract_status": "合同状态",
|
||
"payment_amount": "应付金额",
|
||
"actual_payment_amount": "实付金额",
|
||
"refund_amount": "退款金额",
|
||
"service_type": "服务类型",
|
||
"service_quantity": "课时数量",
|
||
"remaining_lessons": "剩余课时",
|
||
"appointment_start_time": "约课开始",
|
||
"appointment_end_time": "约课结束",
|
||
"use_count": "扣减课时",
|
||
"admitted_school": "录取学校",
|
||
"admitted_program": "录取专业",
|
||
"exam_name": "考试名称",
|
||
"exam_date": "考试日期",
|
||
"chinese": "语文",
|
||
"math": "数学",
|
||
"english": "英语",
|
||
"physics": "物理",
|
||
"chemistry": "化学",
|
||
"biology": "生物",
|
||
"history": "历史",
|
||
"politics": "政治",
|
||
"geography": "地理"
|
||
}
|
||
|
||
# 子表列表
|
||
CHILD_TABLES = list(TABLE_CN.keys())
|
||
|
||
# 获取单表学生数据
|
||
def get_table_data(table, student_id):
|
||
try:
|
||
conn = pymysql.connect(**MYSQL_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
sql = f"SELECT * FROM {table} WHERE student_id = %s AND del_flag = 0"
|
||
cursor.execute(sql, (student_id,))
|
||
rows = cursor.fetchall()
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
cn_rows = []
|
||
for row in rows:
|
||
new_row = {FIELD_CN.get(k, k): format_val(v) for k, v in row.items()}
|
||
cn_rows.append(new_row)
|
||
return cn_rows
|
||
except Exception as e:
|
||
return []
|
||
|
||
# 主流程：逐条学生、逐表同步
|
||
if __name__ == "__main__":
|
||
client = MongoClient(MONGO_URI)
|
||
coll = client[MONGO_DB][MONGO_COLLECTION]
|
||
coll.create_index("学生ID", unique=True)
|
||
|
||
# 读取主表
|
||
conn = pymysql.connect(**MYSQL_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
cursor.execute("SELECT * FROM student_info WHERE del_flag = 0")
|
||
student_list = cursor.fetchall()
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
total = len(student_list)
|
||
print(f"共 {total} 名学生，开始逐条同步...")
|
||
|
||
for idx, stu in enumerate(student_list, 1):
|
||
stu_id = str(stu["id"])
|
||
name = stu.get("name", "未命名")
|
||
print(f"[{idx}/{total}] 处理：{name}({stu_id})")
|
||
|
||
# 主信息转中文
|
||
main_cn = {FIELD_CN.get(k, k): format_val(v) for k, v in stu.items()}
|
||
doc = {"学生ID": stu_id, "学生基础信息": main_cn}
|
||
|
||
# 逐表拓展
|
||
for table in CHILD_TABLES:
|
||
doc[TABLE_CN[table]] = get_table_data(table, stu_id)
|
||
|
||
# 写入 MongoDB
|
||
coll.update_one({"学生ID": stu_id}, {"$set": doc}, upsert=True)
|
||
|
||
print("✅ 全部同步完成！全中文字段、无报错")# |