275 lines
8.6 KiB
Python
275 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
migration_add_tuition_schid.py
|
||
高考志愿网站 v3 数据迁移
|
||
|
||
功能：
|
||
1. schools 表新增 sch_id（TEXT）和 tuition（TEXT）字段
|
||
2. 从最新25年招生章程链接.xlsx 导入 schId（2891所学校）
|
||
3. 根据学校名称计算并写入 tuition 字段（静态学费标准）
|
||
|
||
数据来源：
|
||
- schId: 最新25年招生章程链接.xlsx（阳光高考官方）
|
||
- tuition: 河南省发改委高等教育收费标准 + 学校名称关键字推断
|
||
|
||
用法（本地测试）：
|
||
python3 migration_add_tuition_schid.py --local
|
||
|
||
用法（服务器执行）：
|
||
python3 migration_add_tuition_schid.py
|
||
（默认连接远程数据库）
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import sqlite3
|
||
import json
|
||
import re
|
||
import argparse
|
||
|
||
# 导入学费数据模块
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from tuition_data import get_tuition_for_school, ALL_TUITION_RATES
|
||
|
||
# ============================================================
|
||
# 配置
|
||
# ============================================================
|
||
REMOTE_DB = "/www/wwwroot/gaokao/gaokao_henan.db"
|
||
SCHID_XLSX = "/tmp/最新25年招生章程链接.xlsx"
|
||
LOCAL_DB = os.path.expanduser("~/mc/gaokao-site/gaokao_henan.db")
|
||
|
||
|
||
def get_db_path(local=False):
|
||
if local:
|
||
if os.path.exists(LOCAL_DB):
|
||
return LOCAL_DB
|
||
# 尝试其他本地路径
|
||
alt = "/www/wwwroot/gaokao/gaokao_henan.db"
|
||
if os.path.exists(alt):
|
||
return alt
|
||
print(f"[警告] 本地数据库不存在: {LOCAL_DB}")
|
||
return None
|
||
return REMOTE_DB
|
||
|
||
|
||
def add_columns(conn):
|
||
"""给 schools 表添加 sch_id 和 tuition 列"""
|
||
cur = conn.cursor()
|
||
for col, dtype in [("sch_id", "TEXT"), ("tuition", "TEXT")]:
|
||
try:
|
||
cur.execute(f"ALTER TABLE schools ADD COLUMN {col} {dtype}")
|
||
print(f" [+] 新增列: {col}")
|
||
except sqlite3.OperationalError as e:
|
||
if "duplicate column" in str(e).lower() or "already exists" in str(e).lower():
|
||
print(f" [=] 列已存在: {col}")
|
||
else:
|
||
print(f" [!] 新增列失败: {col}: {e}")
|
||
conn.commit()
|
||
|
||
|
||
def load_schid_xlsx(path):
|
||
"""
|
||
从 xlsx 读取 schId 映射
|
||
返回: {(学校名, 省份): sch_id}
|
||
"""
|
||
try:
|
||
import openpyxl
|
||
except ImportError:
|
||
print("[!] 需要 openpyxl: pip install openpyxl")
|
||
return {}
|
||
|
||
if not os.path.exists(path):
|
||
print(f"[!] xlsx 文件不存在: {path}")
|
||
return {}
|
||
|
||
wb = openpyxl.load_workbook(path, data_only=True, read_only=True)
|
||
ws = wb.active
|
||
|
||
result = {}
|
||
for i, row in enumerate(ws.iter_rows(values_only=True)):
|
||
if i < 2: # 跳过表头
|
||
continue
|
||
if not row or not row[0]:
|
||
continue
|
||
|
||
school_name = str(row[0]).strip() if row[0] else ""
|
||
province = str(row[1]).strip() if len(row) > 1 and row[1] else ""
|
||
|
||
if not school_name or school_name == '院校名称':
|
||
continue
|
||
|
||
# 提取 schId（从 URL 中 schId-数字 格式）
|
||
sch_id = None
|
||
if len(row) > 4 and row[4]:
|
||
url = str(row[4])
|
||
m = re.search(r'schId[=-]?(\d+)', url)
|
||
if m:
|
||
sch_id = m.group(1)
|
||
|
||
if sch_id:
|
||
result[(school_name, province)] = sch_id
|
||
|
||
wb.close()
|
||
print(f" [i] 从 xlsx 加载 {len(result)} 条 schId 映射")
|
||
return result
|
||
|
||
|
||
def write_schid_to_db(conn, schid_map):
|
||
"""将 schId 写入 schools 表"""
|
||
cur = conn.cursor()
|
||
|
||
# 先查一下现有学校
|
||
cur.execute("SELECT DISTINCT school_name FROM schools LIMIT 10")
|
||
print(f" [i] 数据库现有学校示例: {[r[0] for r in cur.fetchall()]}")
|
||
|
||
updated = 0
|
||
not_found = []
|
||
|
||
cur.execute("SELECT DISTINCT school_name FROM schools")
|
||
all_schools = [r[0] for r in cur.fetchall()]
|
||
|
||
for school in all_schools:
|
||
# 多轮匹配：精确 → 含郑州/河南前缀/后缀
|
||
sch_id = schid_map.get((school, "河南")) or \
|
||
schid_map.get((school, ""))
|
||
|
||
if not sch_id:
|
||
# 尝试模糊匹配（去掉"大学""学院"后缀）
|
||
aliases = [
|
||
school,
|
||
school.replace("大学", ""),
|
||
school.replace("学院", ""),
|
||
]
|
||
for alias in aliases:
|
||
for (sn, prov), sid in schid_map.items():
|
||
if sn == alias or alias in sn:
|
||
sch_id = sid
|
||
break
|
||
if sch_id:
|
||
break
|
||
|
||
if sch_id:
|
||
conn.execute(
|
||
"UPDATE schools SET sch_id = ? WHERE school_name = ? AND (sch_id IS NULL OR sch_id = '')",
|
||
(sch_id, school)
|
||
)
|
||
updated += 1
|
||
else:
|
||
not_found.append(school)
|
||
|
||
conn.commit()
|
||
print(f" [+] 写入 schId: {updated} 所学校")
|
||
if not_found[:10]:
|
||
print(f" [!] 未找到 schId (前10): {not_found[:10]}")
|
||
return updated, not_found
|
||
|
||
|
||
def write_tuition_to_db(conn):
|
||
"""根据学校名称计算并写入 tuition 字段"""
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT DISTINCT school_name FROM schools")
|
||
schools = [r[0] for r in cur.fetchall()]
|
||
|
||
updated = 0
|
||
for school in schools:
|
||
tuition_info = get_tuition_for_school(school)
|
||
tuition_json = json.dumps(tuition_info, ensure_ascii=False)
|
||
conn.execute(
|
||
"UPDATE schools SET tuition = ? WHERE school_name = ? AND (tuition IS NULL OR tuition = '')",
|
||
(tuition_json, school)
|
||
)
|
||
updated += 1
|
||
|
||
conn.commit()
|
||
print(f" [+] 写入 tuition: {updated} 所学校")
|
||
return updated
|
||
|
||
|
||
def get_school_tuition_preview(conn, n=15):
|
||
"""打印 tuition 写入预览（不实际写入）"""
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT DISTINCT school_name FROM schools ORDER BY school_name LIMIT ?", (n,))
|
||
schools = [r[0] for r in cur.fetchall()]
|
||
|
||
print(f"\n 学费预览（前{n}所）：")
|
||
for s in schools:
|
||
info = get_tuition_for_school(s)
|
||
print(f" {s:<28} → {info['annual']:,}元/年 ({info['category']})")
|
||
|
||
|
||
def main(local=False, preview_only=False):
|
||
db_path = get_db_path(local)
|
||
if not db_path:
|
||
print("[!] 无法找到数据库，退出")
|
||
return
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f" 高考志愿网站 v3 - 学费 & schId 迁移")
|
||
print(f" 数据库: {db_path}")
|
||
print(f"{'='*60}")
|
||
|
||
conn = sqlite3.connect(db_path)
|
||
|
||
# Step 1: 添加列
|
||
print("\n[1/3] 添加 sch_id / tuition 列...")
|
||
add_columns(conn)
|
||
|
||
# Step 2: 导入 schId
|
||
print("\n[2/3] 导入 schId（从 xlsx）...")
|
||
schid_map = load_schid_xlsx(SCHID_XLSX)
|
||
|
||
# 手动添加几所河南重点学校的 schId（已知数据）
|
||
manual_schid = {
|
||
("郑州大学", "河南"): "361",
|
||
("河南大学", "河南"): "419",
|
||
("河南师范大学", "河南"): "471",
|
||
("河南科技大学", "河南"): "512",
|
||
("河南理工大学", "河南"): "478",
|
||
("华北水利水电大学", "河南"): "509",
|
||
("河南农业大学", "河南"): "473",
|
||
("河南工业大学", "河南"): "510",
|
||
("郑州轻工业大学", "河南"): "515",
|
||
("河南中医药大学", "河南"): "481",
|
||
("新乡医学院", "河南"): "527",
|
||
("河南财经政法大学", "河南"): "521",
|
||
}
|
||
schid_map.update(manual_schid)
|
||
print(f" [i] 合并后共 {len(schid_map)} 条 schId")
|
||
|
||
write_schid_to_db(conn, schid_map)
|
||
|
||
# Step 3: 写入 tuition
|
||
print("\n[3/3] 写入 tuition（静态学费标准）...")
|
||
get_school_tuition_preview(conn)
|
||
if preview_only:
|
||
print("\n [preview_only] 预览模式，未写入数据库")
|
||
conn.close()
|
||
return
|
||
write_tuition_to_db(conn)
|
||
|
||
# 验证
|
||
print("\n[验证] 随机抽查 5 所学校的写入结果：")
|
||
cur = conn.cursor()
|
||
cur.execute("""
|
||
SELECT school_name, sch_id, tuition
|
||
FROM schools
|
||
WHERE tuition IS NOT NULL AND tuition != ''
|
||
GROUP BY school_name
|
||
LIMIT 5
|
||
""")
|
||
for row in cur.fetchall():
|
||
sch, sid, tui = row
|
||
tui_parsed = json.loads(tui) if tui else {}
|
||
print(f" {sch}: schId={sid}, tuition={tui_parsed.get('annual')}元/年 ({tui_parsed.get('category', '')})")
|
||
|
||
conn.close()
|
||
print("\n[完成] 迁移成功！")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description="高考志愿网站学费 & schId迁移")
|
||
parser.add_argument("--local", action="store_true", help="使用本地数据库而非远程")
|
||
parser.add_argument("--preview", action="store_true", help="仅预览，不写入数据库")
|
||
args = parser.parse_args()
|
||
|
||
main(local=args.local, preview_only=args.preview) |