256 lines
8.8 KiB
Python
256 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
阳光高考院校收费数据爬虫 v2
|
||
使用 Playwright 批量抓取学校收费数据
|
||
|
||
流程：
|
||
1. 从 gaokao_henan.db 读取学校列表
|
||
2. 用 Playwright 搜索学校获取 schId
|
||
3. 访问收费公示页，解析学费/住宿费
|
||
4. 结果存入 JSON + 更新 schools 表 tuition 字段
|
||
"""
|
||
|
||
import sqlite3
|
||
import json
|
||
import time
|
||
import random
|
||
import re
|
||
import os
|
||
import sys
|
||
|
||
DB_PATH = "/www/wwwroot/gaokao/gaokao_henan.db"
|
||
OUTPUT_JSON = "/tmp/school_tuition.json"
|
||
SCHID_CACHE = "/tmp/school_schid_cache.json"
|
||
BATCH_SIZE = 100
|
||
DELAY = 1.5 # 秒
|
||
|
||
def get_schools_from_db():
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cur = conn.cursor()
|
||
cur.execute("""
|
||
SELECT DISTINCT school_name
|
||
FROM schools
|
||
WHERE school_name IS NOT NULL AND school_name != ''
|
||
ORDER BY school_name
|
||
""")
|
||
schools = [row[0] for row in cur.fetchall()]
|
||
conn.close()
|
||
return schools
|
||
|
||
def load_schid_cache():
|
||
if os.path.exists(SCHID_CACHE):
|
||
with open(SCHID_CACHE) as f:
|
||
return json.load(f)
|
||
return {}
|
||
|
||
def save_schid_cache(cache):
|
||
with open(SCHID_CACHE, 'w') as f:
|
||
json.dump(cache, f, ensure_ascii=False, indent=2)
|
||
|
||
def add_tuition_column():
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cur = conn.cursor()
|
||
try:
|
||
cur.execute("ALTER TABLE schools ADD COLUMN tuition TEXT")
|
||
print("[+] 已添加 tuition 列")
|
||
except Exception:
|
||
pass
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
def save_tuition_result(results):
|
||
with open(OUTPUT_JSON, 'w', encoding='utf-8') as f:
|
||
json.dump(results, f, ensure_ascii=False, indent=2)
|
||
|
||
def parse_fee_page(html_content):
|
||
"""
|
||
解析收费公示页HTML，提取学费和住宿费信息
|
||
返回格式: {"tuition": "5000-8000元/年", "dorm": "800-1200元/年", "raw": "..."}
|
||
"""
|
||
tuition_patterns = [
|
||
r'学费[：:]\s*(\d+\.?\d*[\u4e00-\u9fa5]*/\u5b66\u751f[\u00B7\u2022]?\u5e74?)',
|
||
r'(\d{4})\u5e74\u5b66\u8d39\u6807\u51c6[\uff1a:]\s*(\d+)',
|
||
r'\u5b66\u8d39\u6807\u51c6[\uff1a:（:）\s]*(\d+)',
|
||
r'(\d{4})[\u5e74]*\u5b66\u8d39[：:]\s*(\d+)',
|
||
r'\u5b66\u8d39\s*(\d+)\s*[\u5143\u5147]',
|
||
]
|
||
|
||
result = {"tuition": None, "dorm": None, "raw": ""}
|
||
|
||
# 尝试提取学费信息
|
||
for pattern in tuition_patterns:
|
||
matches = re.findall(pattern, html_content)
|
||
if matches:
|
||
result["tuition"] = str(matches[0])
|
||
break
|
||
|
||
# 如果没找到，返回页面中包含"元/年"的片段作为参考
|
||
if not result["tuition"]:
|
||
yuan_nian = re.findall(r'[\d\.]+\s*[\u5143\u5147]\s*/\s*\u5b66\u5e74|[一二三<E4BA8C>\u4e00-\u9fa5]{2,4}\u5b66\u8d39[\uff1a:]\s*\d+', html_content)
|
||
if yuan_nian:
|
||
result["raw"] = "; ".join(yuan_nian[:5])
|
||
|
||
return result
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print(" 阳光高考院校收费数据爬虫 v2")
|
||
print("=" * 60)
|
||
|
||
# 1. 初始化数据库
|
||
add_tuition_column()
|
||
|
||
# 2. 加载学校列表和缓存
|
||
schools = get_schools_from_db()
|
||
schid_cache = load_schid_cache()
|
||
print(f"[i] 数据库共 {len(schools)} 所学校")
|
||
print(f"[i] 已有 schId 缓存: {len(schid_cache)} 所")
|
||
|
||
# 3. 检查 Playwright
|
||
try:
|
||
from playwright.sync_api import sync_playwright
|
||
print("[+] Playwright 可用")
|
||
except ImportError:
|
||
print("[x] 未安装 Playwright，正在安装...")
|
||
os.system("pip install playwright -q")
|
||
os.system("playwright install chromium --with-deps")
|
||
from playwright.sync_api import sync_playwright
|
||
print("[+] Playwright 安装完成")
|
||
|
||
# 4. 启动 Playwright 浏览器
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(headless=True)
|
||
context = browser.new_context(
|
||
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||
locale="zh-CN",
|
||
)
|
||
page = context.new_page()
|
||
|
||
# 设置超时
|
||
page.set_default_timeout(15000)
|
||
|
||
results = {}
|
||
skip_count = 0
|
||
fetch_count = 0
|
||
|
||
for i, school_name in enumerate(schools):
|
||
if i < skip_count:
|
||
continue
|
||
|
||
if (i + 1) % BATCH_SIZE == 0:
|
||
print(f" 进度 {i+1}/{len(schools)} ({fetch_count} 所已抓取)")
|
||
save_schid_cache(schid_cache)
|
||
time.sleep(2)
|
||
|
||
# 检查缓存
|
||
if school_name in schid_cache:
|
||
sch_id = schid_cache[school_name]
|
||
else:
|
||
# 搜索学校获取 schId
|
||
try:
|
||
search_url = f"https://gaokao.chsi.com.cn/sch/search.do"
|
||
page.goto(search_url, wait_until="domcontentloaded")
|
||
time.sleep(0.8)
|
||
|
||
# 输入学校名
|
||
page.fill("input[type='text'], input[placeholder*='院校名称']", school_name)
|
||
time.sleep(0.3)
|
||
page.click("button:has-text('搜索'), button[type='submit']")
|
||
time.sleep(2.0)
|
||
|
||
# 从当前 URL 提取 schId
|
||
url = page.url
|
||
match = re.search(r'schId[=-](\d+)', url)
|
||
if match:
|
||
sch_id = match.group(1)
|
||
schid_cache[school_name] = sch_id
|
||
else:
|
||
# 从页面内容找 schId
|
||
content = page.content()
|
||
match = re.search(r'schId[=-](\d+)', content)
|
||
if match:
|
||
sch_id = match.group(1)
|
||
schid_cache[school_name] = sch_id
|
||
else:
|
||
results[school_name] = {"error": "未找到 schId", "schId": None}
|
||
time.sleep(DELAY)
|
||
continue
|
||
|
||
except Exception as e:
|
||
results[school_name] = {"error": str(e), "schId": None}
|
||
time.sleep(DELAY)
|
||
continue
|
||
|
||
time.sleep(DELAY + random.uniform(0, 0.5))
|
||
|
||
# 访问收费公示页
|
||
try:
|
||
# 收费项目页 URL 格式
|
||
fee_url = f"https://gaokao.chsi.com.cn/sch/schoolInfo--schId-{sch_id},categoryId-7721443883,mindex-7.dhtml"
|
||
page.goto(fee_url, wait_until="domcontentloaded")
|
||
time.sleep(1.0)
|
||
|
||
html = page.content()
|
||
parsed = parse_fee_page(html)
|
||
|
||
results[school_name] = {
|
||
"schId": sch_id,
|
||
"fee_url": fee_url,
|
||
"tuition": parsed["tuition"],
|
||
"dorm": parsed.get("dorm"),
|
||
"raw_note": parsed.get("raw"),
|
||
"status": "ok" if parsed["tuition"] else "no_tuition_data"
|
||
}
|
||
fetch_count += 1
|
||
|
||
# 定期保存
|
||
if fetch_count % 20 == 0:
|
||
save_tuition_result(results)
|
||
print(f" [saved] 已保存 {len(results)} 条结果")
|
||
|
||
except Exception as e:
|
||
results[school_name] = {"error": str(e), "schId": sch_id, "status": "error"}
|
||
|
||
time.sleep(DELAY + random.uniform(0, 0.5))
|
||
|
||
browser.close()
|
||
|
||
# 5. 保存结果
|
||
save_tuition_result(results)
|
||
save_schid_cache(schid_cache)
|
||
|
||
# 6. 统计
|
||
ok = sum(1 for v in results.values() if v.get("status") == "ok")
|
||
no_data = sum(1 for v in results.values() if v.get("status") == "no_tuition_data")
|
||
errors = sum(1 for v in results.values() if v.get("status") == "error")
|
||
|
||
print()
|
||
print("=" * 60)
|
||
print(" 抓取完成")
|
||
print(f" 总计处理: {len(results)} 所")
|
||
print(f" 成功获取学费: {ok} 所")
|
||
print(f" 无学费数据: {no_data} 所")
|
||
print(f" 错误: {errors} 所")
|
||
print(f" 结果文件: {OUTPUT_JSON}")
|
||
print(f" schId 缓存: {SCHID_CACHE}")
|
||
print("=" * 60)
|
||
|
||
# 7. 更新 schools 表
|
||
print()
|
||
print("更新 schools 表 tuition 字段...")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cur = conn.cursor()
|
||
updated = 0
|
||
for school_name, data in results.items():
|
||
if data.get("tuition"):
|
||
cur.execute(
|
||
"UPDATE schools SET tuition = ? WHERE school_name = ? AND tuition IS NULL",
|
||
(data["tuition"], school_name)
|
||
)
|
||
updated += cur.rowcount
|
||
conn.commit()
|
||
conn.close()
|
||
print(f"[+] 已更新 {updated} 条 tuition 记录")
|
||
|
||
if __name__ == "__main__":
|
||
main() |