数据工程

在 第一部分，我们探讨了核心搜索逻辑和 API 结构。本篇深入 数据层——如何规范化、存储和检索数据。我们还看看驱动后端功能的实用脚本。 1. 数据"胶水"：utils/normalize_name.py 从多个来源（QS、US News、Niche）聚合数据的最大挑战是它们命名大学的方式不一。“MIT” 在一个数据集中是"Massachusetts Institute of Technology"，在另一个可能是"Mass Inst of Tech"。 这个文件包含标准化名称的逻辑，以便我们能将数据链接在一起。 函数：normalize_name 输入：name（字符串），如"University of California, Berkeley (UCB)" 逻辑： 转小写：全部转小写避免大小写不匹配 清理：删除括号内文本（如缩写）和标点 别名检查：检查已知别名字典（ALIASES）。若输入"mit"，自动转换为"massachusetts institute of technology" 输出：清洁、标准化的字符串（如"university of california berkeley"） def normalize_name(name: str) -> str: # 转小写 name = name.lower() # 删除括号内文本，如"(MIT)" name = re.sub(r'\(.*?\)', '', name) # 删除标点 name = re.sub(r'[^\w\s]', '', name) # 规范化空格 name = re.sub(r'\s+', ' ', name) if name in ALIASES: # 若名称是别名，替换为全名 name = ALIASES[name] return name.strip() 2. 详细资料逻辑：models/university.py 虽然 models/universities.py（复数）处理列表和搜索，但这个文件处理 单一 大学资料的深度挖掘。 函数：get_university_by_id 这个函数很有趣，因为它必须从 许多 不同表收集数据构建完整资料。 输入：univ_id（整数） 逻辑： 基础信息：从 Universities 表获取主要信息（名称、位置、照片） 翻译：与 University_names_en_to_zh 联接获中文名称 动态排名检索： 查询 sqlite_master 查找数据库中所有以 _Rankings 结尾的表 遍历每张排名表检查该大学是否在其中 这使系统易于扩展——若添加新排名表（如"Mars_University_Rankings"），这函数自动使用它，无需代码改动 统计数据：从 UniversityStats 获取统计信息（学生数等） 输出：包含该大学已知所有信息的大字典 def get_university_by_id(univ_id): conn = get_db_connection() # 第一步：获取大学基础信息 cur = conn.execute(""" SELECT Universities.*, T.chinese_name FROM Universities LEFT JOIN University_names_en_to_zh AS T ON Universities.id = T.id WHERE Universities.id = ? """, (univ_id,)) row = cur.fetchone() if not row: return None university = dict(row) normalized_name = university['normalized_name'] # 第二步：从所有 *_Rankings 表获取排名 cur = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE '%_Rankings'") ranking_tables = [r["name"] for r in cur.fetchall()] rankings = [] for table in ranking_tables: try: cur = conn.execute( f"""SELECT subject, source, rank_value FROM "{table}" WHERE normalized_name = ?""", (normalized_name,) ) rankings += [dict(r) for r in cur.fetchall()] except Exception as e: # 跳过格式错误或不匹配的表 continue # 第三步：从 UniversityStats 获取统计数据 cur = conn.execute( "SELECT type, count, year FROM UniversityStats WHERE normalized_name = ?", (normalized_name,) ) stats = [dict(r) for r in cur.fetchall()] conn.close() rankings = sorted(rankings, key= lambda x: -1 if "global" in x["subject"] or "World" in x["subject"] else x["rank_value"]) # 合并所有信息 university["rankings"] = rankings university["stats"] = stats return university 3. 学科排名：routes/ranking_detail.py 这个路由处理特定学科排名，如"计算机科学"或"医学"。 ...