#!/usr/bin/env python3 import re from collections import defaultdict from datetime import datetime, date from decimal import Decimal from pathlib import Path import openpyxl import pymysql BASE = Path("/Users/liuhuayong/Documents/文档/其他/figs") TENANT_ID = "000000" CREATE_BY = 1 CREATE_DEPT = 100 NOW = datetime.now() DB_CONFIG = { "host": "100.122.151.35", "user": "root", "password": "root123", "database": "figs_education_db", "charset": "utf8mb4", "autocommit": False, "cursorclass": pymysql.cursors.DictCursor, } ROLE_SALES = "SALES" ROLE_HOUSEKEEPER = "HOUSEKEEPER" ROLE_DOC_WRITER = "DOC_WRITER" ROLE_APPLY_TEACHER = "APPLY_TEACHER" def clean_text(value): if value is None: return None text = str(value).replace("\xa0", " ").strip() return text or None def clean_phone(value): """取首个11位手机号，避免 A/B 格式超长。""" text = clean_text(value) if not text: return None m = re.search(r"1[3-9]\d{9}", text) return m.group(0) if m else text[:20] def clip(text, length): """按数据库列宽裁剪，防止 Data too long。""" if not text: return None return text[:length] if len(text) > length else text def normalize_gender(value): text = clean_text(value) if text in ("男", "0", "M", "male", "Male"): return "0" if text in ("女", "1", "F", "female", "Female"): return "1" return "2" def normalize_yes_no(value): text = clean_text(value) if not text: return "N" if text in ("是", "Y", "y", "YES", "Yes", "yes", "有", "通过", "确认参与", "确认参加", "参加"): return "Y" if text in ("否", "N", "n", "NO", "No", "no", "无", "不申请", "不参加"): return "N" return "Y" def first_decimal(text): text = clean_text(text) if not text: return None m = re.search(r"\d+(?:\.\d+)?", text.replace(",", "")) if not m: return None return Decimal(m.group(0)) def parse_date(text): if isinstance(text, datetime): return text.date() if isinstance(text, date): return text text = clean_text(text) if not text: return None for fmt in ("%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d", "%Y年%m月%d日"): try: return datetime.strptime(text, fmt).date() except ValueError: pass return None class Importer: def __init__(self): self.conn = pymysql.connect(**DB_CONFIG) self.student_cache = {} self.staff_cache = {} self.school_cache = {} self.stats = defaultdict(int) def close(self): self.conn.close() def fetch_existing(self): with self.conn.cursor() as cur: cur.execute("select id, name from edu_student where tenant_id=%s and del_flag='0'", (TENANT_ID,)) for row in cur.fetchall(): self.student_cache[row["name"]] = row["id"] cur.execute("select id, name, role from edu_staff where tenant_id=%s and del_flag='0'", (TENANT_ID,)) for row in cur.fetchall(): self.staff_cache[(row["name"], row["role"])] = row["id"] cur.execute("select id, name_cn from edu_school where tenant_id=%s and del_flag='0'", (TENANT_ID,)) for row in cur.fetchall(): self.school_cache[row["name_cn"]] = row["id"] def next_manual_id(self, table): with self.conn.cursor() as cur: cur.execute(f"select coalesce(max(id), 0) as max_id from {table}") return cur.fetchone()["max_id"] + 1 def ensure_student(self, name, **fields): name = clean_text(name) if not name or re.fullmatch(r"[0-9]{10,}", name): return None if name in self.student_cache: return self.student_cache[name] student_id = self.next_manual_id("edu_student") student_no = fields.get("student_no") or f"S{student_id:06d}" with self.conn.cursor() as cur: cur.execute( """ insert into edu_student ( id, tenant_id, student_no, name, english_name, gender, id_card, phone, school, grade, target_region, admission_pathway, target_major, exam_language, exam_identity, nationality, exam_location, curriculum_system, selected_subjects, has_health_condition, health_condition_note, need_political_review, application_plan, academic_test_score, personal_experience, english_score, listening_score, source, status, remark, special_remark, create_dept, create_by, create_time, update_by, update_time ) values ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) """, ( student_id, TENANT_ID, student_no, name, fields.get("english_name"), fields.get("gender", "2"), clip(fields.get("id_card"), 32), clean_phone(fields.get("phone")), fields.get("school"), fields.get("grade"), fields.get("target_region"), fields.get("admission_pathway"), fields.get("target_major"), fields.get("exam_language"), fields.get("exam_identity"), fields.get("nationality"), fields.get("exam_location"), fields.get("curriculum_system"), fields.get("selected_subjects"), fields.get("has_health_condition", "N"), fields.get("health_condition_note"), fields.get("need_political_review", "N"), fields.get("application_plan"), fields.get("academic_test_score"), fields.get("personal_experience"), fields.get("english_score"), fields.get("listening_score"), fields.get("source"), fields.get("status", "2"), fields.get("remark"), fields.get("special_remark"), CREATE_DEPT, CREATE_BY, NOW, CREATE_BY, NOW ) ) self.student_cache[name] = student_id self.stats["students_inserted"] += 1 return student_id def ensure_staff(self, name, role): name = clean_text(name) if not name or re.fullmatch(r"[0-9]{10,}", name): return None key = (name, role) if key in self.staff_cache: return self.staff_cache[key] staff_id = self.next_manual_id("edu_staff") staff_no = f"T{staff_id:06d}" with self.conn.cursor() as cur: cur.execute( """ insert into edu_staff ( id, tenant_id, staff_no, name, role, status, del_flag, create_dept, create_by, create_time, update_by, update_time ) values (%s, %s, %s, %s, %s, '0', '0', %s, %s, %s, %s, %s) """, (staff_id, TENANT_ID, staff_no, name, role, CREATE_DEPT, CREATE_BY, NOW, CREATE_BY, NOW) ) self.staff_cache[key] = staff_id self.stats["staff_inserted"] += 1 return staff_id def bind_staff_student(self, staff_id, student_id, role): if not staff_id or not student_id: return with self.conn.cursor() as cur: cur.execute( """ select id from edu_staff_student where tenant_id=%s and staff_id=%s and student_id=%s and staff_role=%s """, (TENANT_ID, staff_id, student_id, role) ) if cur.fetchone(): return relation_id = self.next_manual_id("edu_staff_student") cur.execute( """ insert into edu_staff_student ( id, tenant_id, staff_id, student_id, staff_role, bind_time, is_primary, status, del_flag, create_dept, create_by, create_time, update_by, update_time ) values (%s, %s, %s, %s, %s, %s, 'Y', '0', '0', %s, %s, %s, %s, %s) """, (relation_id, TENANT_ID, staff_id, student_id, role, NOW, CREATE_DEPT, CREATE_BY, NOW, CREATE_BY, NOW) ) self.stats["staff_student_inserted"] += 1 def ensure_school(self, row): name_cn = clean_text(row["name_cn"]) if not name_cn: return None if name_cn in self.school_cache: return self.school_cache[name_cn] with self.conn.cursor() as cur: cur.execute( """ insert into edu_school ( tenant_id, name_cn, name_en, country, nature, ranking, intro, history, advantages, fees, language_req, materials, official_url, contact_info, brochure_url, del_flag, remark, create_dept, create_by, create_time, update_by, update_time ) values ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, '0', %s, %s, %s, %s, %s, %s ) """, ( TENANT_ID, name_cn, clip(row.get("name_en"), 256), clip(row.get("country"), 32), clip(row.get("nature"), 32), clip(row.get("ranking"), 64), clip(row.get("intro"), 1000), clip(row.get("history"), 1000), clip(row.get("advantages"), 1000), clip(row.get("fees"), 500), clip(row.get("language_req"), 255), clip(row.get("materials"), 1000), clip(row.get("official_url"), 500), clip(row.get("contact_info"), 500), clip(row.get("brochure_url"), 500), clip(row.get("remark"), 500), CREATE_DEPT, CREATE_BY, NOW, CREATE_BY, NOW ) ) school_id = cur.lastrowid self.school_cache[name_cn] = school_id self.stats["school_inserted"] += 1 return school_id def insert_admission(self, row): with self.conn.cursor() as cur: cur.execute( """ select id from edu_admission where tenant_id=%s and student_id=%s and school_name=%s and del_flag='0' """, (TENANT_ID, row["student_id"], row["school_name"]) ) if cur.fetchone(): return cur.execute( """ insert into edu_admission ( tenant_id, student_id, school_id, school_name, enrolled, major, is_comprehensive, exam_score, exam_ranking, english_score, scholarship, admit_date, admit_term, del_flag, remark, create_dept, create_by, create_time, update_by, update_time ) values ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, '0', %s, %s, %s, %s, %s, %s ) """, ( TENANT_ID, row["student_id"], row.get("school_id"), row["school_name"], row.get("enrolled", "Y"), row.get("major"), row.get("is_comprehensive", "N"), row.get("exam_score"), row.get("exam_ranking"), row.get("english_score"), row.get("scholarship"), row.get("admit_date"), row.get("admit_term"), row.get("remark"), CREATE_DEPT, CREATE_BY, NOW, CREATE_BY, NOW ) ) self.stats["admission_inserted"] += 1 def insert_activity(self, row): with self.conn.cursor() as cur: cur.execute( """ select id from edu_student_activity where tenant_id=%s and student_id=%s and activity_type=%s and title=%s and del_flag='0' """, (TENANT_ID, row["student_id"], row["activity_type"], row["title"]) ) if cur.fetchone(): return cur.execute( """ insert into edu_student_activity ( tenant_id, student_id, activity_type, title, provider, period, description, achievement, del_flag, remark, create_dept, create_by, create_time, update_by, update_time ) values ( %s, %s, %s, %s, %s, %s, %s, %s, '0', %s, %s, %s, %s, %s, %s ) """, ( TENANT_ID, row["student_id"], row["activity_type"], row["title"], row.get("provider"), row.get("period"), row.get("description"), row.get("achievement"), row.get("remark"), CREATE_DEPT, CREATE_BY, NOW, CREATE_BY, NOW ) ) self.stats["activity_inserted"] += 1 def insert_score(self, row): with self.conn.cursor() as cur: cur.execute( """ select id from edu_student_score where tenant_id=%s and student_id=%s and exam_name=%s and exam_type=%s and del_flag='0' limit 1 """, (TENANT_ID, row["student_id"], row["exam_name"], row["exam_type"]) ) if cur.fetchone(): return cur.execute( """ insert into edu_student_score ( tenant_id, student_id, exam_name, exam_type, subject, score, total_score, ranking, grade_level, exam_date, del_flag, remark, create_dept, create_by, create_time, update_by, update_time ) values ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, '0', %s, %s, %s, %s, %s, %s ) """, ( TENANT_ID, row["student_id"], row["exam_name"], row["exam_type"], row.get("subject"), row.get("score"), row.get("total_score"), row.get("ranking"), row.get("grade_level"), row.get("exam_date"), row.get("remark"), CREATE_DEPT, CREATE_BY, NOW, CREATE_BY, NOW ) ) self.stats["score_inserted"] += 1 def import_students_and_bindings(self): wb = openpyxl.load_workbook(BASE / "副本汇总-导入学生信息241009.xlsx", data_only=True) ws = wb["学生信息"] for r in range(2, ws.max_row + 1): name = clean_text(ws.cell(r, 1).value) if not name: continue student_id = self.ensure_student( name, english_name=clean_text(ws.cell(r, 2).value), school=clean_text(ws.cell(r, 3).value), gender=normalize_gender(ws.cell(r, 4).value), id_card=clean_text(ws.cell(r, 5).value), phone=clean_text(ws.cell(r, 6).value), exam_language=clean_text(ws.cell(r, 7).value), exam_identity=clean_text(ws.cell(r, 8).value), nationality=clean_text(ws.cell(r, 9).value), exam_location=clean_text(ws.cell(r, 10).value), curriculum_system=clean_text(ws.cell(r, 11).value), grade=clean_text(ws.cell(r, 12).value), selected_subjects=clean_text(ws.cell(r, 13).value), has_health_condition=normalize_yes_no(ws.cell(r, 14).value), health_condition_note=clean_text(ws.cell(r, 15).value), need_political_review=normalize_yes_no(ws.cell(r, 16).value), application_plan=clean_text(ws.cell(r, 17).value), academic_test_score=clean_text(ws.cell(r, 18).value), personal_experience=clean_text(ws.cell(r, 19).value), english_score=clean_text(ws.cell(r, 20).value), listening_score=clean_text(ws.cell(r, 21).value), target_region=clean_text(ws.cell(r, 22).value), admission_pathway=clean_text(ws.cell(r, 23).value), target_major=clean_text(ws.cell(r, 24).value), special_remark=clean_text(ws.cell(r, 26).value), source=clean_text(ws.cell(r, 35).value), remark=clean_text(ws.cell(r, 44).value), ) for col, role in ((27, ROLE_APPLY_TEACHER), (28, ROLE_SALES), (29, ROLE_DOC_WRITER), (30, ROLE_HOUSEKEEPER)): staff_name = clean_text(ws.cell(r, col).value) if staff_name: staff_id = self.ensure_staff(staff_name, role) self.bind_staff_student(staff_id, student_id, role) wb.close() def import_schools(self): wb = openpyxl.load_workbook(BASE / "综评、港、澳、新、4+0、副学士学校简介（系统对外版）_副本.xlsx", data_only=True) for ws in wb.worksheets: headers = [clean_text(ws.cell(1, c).value) for c in range(1, ws.max_column + 1)] for r in range(2, ws.max_row + 1): first = clean_text(ws.cell(r, 1).value) if not first: continue row = {"name_cn": first, "remark": ws.title} mapping = {h: idx + 1 for idx, h in enumerate(headers) if h} def gv(name): idx = mapping.get(name) return clean_text(ws.cell(r, idx).value) if idx else None row["name_en"] = gv("学校英文名") or gv("项目名称") row["country"] = gv("所在城市或地区") or gv("所在国家/地区") row["nature"] = gv("学校性质") or gv("项目性质") row["ranking"] = gv("排名") or gv("合作学校排名") row["intro"] = gv("学校简介") or gv("项目简介") row["history"] = gv("校史沿革") row["advantages"] = gv("热门专业") or gv("招生专业") or gv("优势专业") row["fees"] = gv("综评学费") or gv("高考生学费") or gv("学费") or gv("费用") row["language_req"] = gv("语言成绩要求（最低参考）") or gv("报名条件") or gv("小语种是否可报") row["materials"] = gv("常规申请材料清单") or gv("学部设置") row["official_url"] = gv("学校官网") row["contact_info"] = gv("学校联系方式") row["brochure_url"] = gv("综评招生简章") or gv("高考生招生简章") or gv("自主招生简章") or gv("招生简章") self.ensure_school(row) wb.close() def import_admissions(self): wb = openpyxl.load_workbook(BASE / "FIGS-【2024届】高三申请工作汇报记录表.xlsx", data_only=True) ws = wb["申请老师工作表"] for r in range(2, ws.max_row + 1): student_name = clean_text(ws.cell(r, 13).value) school_name = clean_text(ws.cell(r, 14).value) if not student_name or not school_name: continue student_id = self.ensure_student(student_name, grade=clean_text(ws.cell(r, 1).value), school=clean_text(ws.cell(r, 9).value)) school_id = self.school_cache.get(school_name) self.insert_admission( { "student_id": student_id, "school_id": school_id, "school_name": school_name, "major": clean_text(ws.cell(r, 15).value), "is_comprehensive": normalize_yes_no(ws.cell(r, 16).value), "exam_score": clean_text(ws.cell(r, 17).value), "exam_ranking": clean_text(ws.cell(r, 18).value), "english_score": clean_text(ws.cell(r, 19).value), "remark": clean_text(ws.cell(r, 7).value), "enrolled": "Y", } ) wb.close() def import_activities(self): wb = openpyxl.load_workbook(BASE / "FIGS-【2024届】高三申请工作汇报记录表.xlsx", data_only=True) ws = wb["背景安排情况"] activity_cols = [ (5, "NUS在线课"), (6, "BPC比赛"), (7, "香港研习项目"), (8, "MS副总裁培训"), (9, "IFPC竞赛"), (10, "香港case比赛"), (11, "新加坡case推荐信"), (12, "巴厘岛支教"), (13, "联合国感谢信"), (14, "GVI"), ] for r in range(2, ws.max_row + 1): student_name = clean_text(ws.cell(r, 2).value) if not student_name: continue student_id = self.ensure_student( student_name, english_name=clean_text(ws.cell(r, 3).value), gender=normalize_gender(ws.cell(r, 4).value), school=clean_text(ws.cell(r, 25).value), grade=clean_text(ws.cell(r, 18).value), selected_subjects=clean_text(ws.cell(r, 26).value), english_score=clean_text(ws.cell(r, 29).value), personal_experience=clean_text(ws.cell(r, 30).value), ) for col, title in activity_cols: raw = clean_text(ws.cell(r, col).value) if raw and raw not in ("否", "N"): self.insert_activity( { "student_id": student_id, "activity_type": title, "title": title, "period": clean_text(ws.cell(r, 18).value), "description": clean_text(ws.cell(r, 20).value), "achievement": raw if raw not in ("Y", "是") else None, "remark": clean_text(ws.cell(r, 30).value), } ) wb.close() def import_scores(self): wb = openpyxl.load_workbook(BASE / "操作手册/副本FIGS 学员明细-服务中.xlsx", data_only=True) ws = wb["服务中所有学员明细"] for r in range(2, ws.max_row + 1): student_name = clean_text(ws.cell(r, 4).value) if not student_name: continue student_id = self.ensure_student( student_name, gender=normalize_gender(ws.cell(r, 6).value), grade=clean_text(ws.cell(r, 7).value), school=clean_text(ws.cell(r, 8).value), selected_subjects=clean_text(ws.cell(r, 11).value), ) recent_exam = clean_text(ws.cell(r, 13).value) recent_score = clean_text(ws.cell(r, 14).value) recent_rank = clean_text(ws.cell(r, 15).value) if recent_exam or recent_score or recent_rank: self.insert_score( { "student_id": student_id, "exam_name": recent_exam or "最近考试", "exam_type": "SCHOOL_EXAM", "score": first_decimal(recent_score), "ranking": recent_rank, "remark": recent_score, } ) lang = clean_text(ws.cell(r, 16).value) if lang: self.insert_score( { "student_id": student_id, "exam_name": "雅思托福成绩", "exam_type": "LANGUAGE", "score": first_decimal(lang), "remark": lang, } ) academic = clean_text(ws.cell(r, 17).value) if academic: self.insert_score( { "student_id": student_id, "exam_name": "学水成绩", "exam_type": "ACADEMIC_TEST", "score": first_decimal(academic), "remark": academic, } ) wb.close() def run(self): self.fetch_existing() self.import_students_and_bindings() self.import_schools() self.import_admissions() self.import_activities() self.import_scores() self.conn.commit() for k in sorted(self.stats): print(f"{k}={self.stats[k]}") if __name__ == "__main__": importer = Importer() try: importer.run() except Exception: importer.conn.rollback() raise finally: importer.close()