239 lines
7.5 KiB
Python
239 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
tuition_scanner.py
|
||
扫描阳光高考学校收费数据可用性
|
||
|
||
流程：
|
||
1. 从 schools 表取不重复学校列表
|
||
2. 对每所学校搜索并获取 schId
|
||
3. 尝试多个收费数据 URL 模式
|
||
4. 统计覆盖率，决定后续方案
|
||
"""
|
||
|
||
import sqlite3
|
||
import json
|
||
import time
|
||
import random
|
||
import re
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# ========== 配置 ==========
|
||
DB_PATH = "/www/wwwroot/gaokao/gaokao_henan.db"
|
||
SCHID_CACHE = "/tmp/schid_map.json"
|
||
SCAN_RESULT = "/tmp/tuition_scan_result.json"
|
||
BATCH = 100
|
||
DELAY = 1.2
|
||
# ==========================
|
||
|
||
def get_top_schools(n=50):
|
||
"""取概率计算器最常推荐的学校（按出现频次）"""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cur = conn.cursor()
|
||
# 优先取河南省学校（概率计算器的主要推荐范围）
|
||
cur.execute("""
|
||
SELECT school_name, COUNT(*) as cnt
|
||
FROM schools
|
||
WHERE school_name LIKE '%郑州%' OR school_name LIKE '%河南%'
|
||
OR school_name LIKE '%师范%' OR school_name LIKE '%大学'
|
||
GROUP BY school_name
|
||
ORDER BY cnt DESC
|
||
LIMIT ?
|
||
""", (n,))
|
||
result = [(row[0], row[1]) for row in cur.fetchall()]
|
||
conn.close()
|
||
return result
|
||
|
||
def get_all_schools():
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT DISTINCT school_name FROM schools ORDER BY school_name")
|
||
result = [row[0] for row in cur.fetchall()]
|
||
conn.close()
|
||
return result
|
||
|
||
def load_schid_cache():
|
||
if os.path.exists(SCHID_CACHE):
|
||
with open(SCHID_CACHE) as f:
|
||
return json.load(f)
|
||
return {}
|
||
|
||
def save_cache(cache):
|
||
with open(SCHID_CACHE, 'w') as f:
|
||
json.dump(cache, f, ensure_ascii=False, indent=2)
|
||
|
||
def save_results(data):
|
||
with open(SCAN_RESULT, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def get_schid_for_school(browser, school_name, cache):
|
||
"""用 Playwright 搜索学校返回 schId"""
|
||
if school_name in cache:
|
||
return cache[school_name]
|
||
|
||
try:
|
||
browser.goto("https://gaokao.chsi.com.cn/sch/", wait_until="load")
|
||
time.sleep(0.8)
|
||
browser.click("#yxmc")
|
||
browser.fill("#yxmc", school_name)
|
||
browser.press("#yxmc", "Enter")
|
||
time.sleep(2.0)
|
||
|
||
content = browser.content()
|
||
matches = re.findall(r'/sch/schoolInfo--schId-(\d+)\.dhtml', content)
|
||
if matches:
|
||
sch_id = matches[0]
|
||
cache[school_name] = sch_id
|
||
return sch_id
|
||
except Exception as e:
|
||
pass
|
||
|
||
time.sleep(DELAY + random.uniform(0, 0.3))
|
||
return None
|
||
|
||
|
||
def check_fee_urls(browser, sch_id):
|
||
"""尝试多个 URL 模式检查收费数据"""
|
||
results = {}
|
||
|
||
# 模式1：schoolInfo 收费项目页（mindex-7）
|
||
# 模式2：listBulletin 公示栏
|
||
# 模式3：直接查学校页是否有收费信息
|
||
|
||
patterns = [
|
||
# (name, url_template)
|
||
("schoolInfo_fee", f"https://gaokao.chsi.com.cn/sch/schoolInfo--schId-{sch_id},categoryId-87835,mindex-7.dhtml"),
|
||
("listBulletin", f"https://gaokao.chsi.com.cn/sch/listBulletin--schId-{sch_id},categoryId-87835,mindex-12.dhtml"),
|
||
]
|
||
|
||
for name, url in patterns:
|
||
try:
|
||
browser.goto(url, wait_until="load")
|
||
time.sleep(1.0)
|
||
|
||
# 检查页面内容是否有实质数据
|
||
body_text = browser.inner_text("body")
|
||
|
||
# 查找关键词
|
||
has_fee_data = False
|
||
fee_sample = ""
|
||
|
||
keywords = ['学', '费', '元/', '年', '收费', '标准', '公示']
|
||
for kw in keywords:
|
||
if kw in body_text:
|
||
# 找包含关键词的句子
|
||
for line in body_text.split('\n'):
|
||
if kw in line and len(line.strip()) > 4:
|
||
has_fee_data = True
|
||
fee_sample = line.strip()[:100]
|
||
break
|
||
if has_fee_data:
|
||
break
|
||
|
||
# 排除只有标题的情况
|
||
if body_text.count('收费项目') == 1 and body_text.count('学') <= 3:
|
||
has_fee_data = False
|
||
|
||
results[name] = {
|
||
"has_data": has_fee_data,
|
||
"url": url,
|
||
"sample": fee_sample if has_fee_data else "",
|
||
"page_len": len(body_text)
|
||
}
|
||
except Exception as e:
|
||
results[name] = {"has_data": False, "error": str(e), "url": url}
|
||
|
||
return results
|
||
|
||
|
||
def run_scan(school_list, max_schools=None):
|
||
"""执行扫描"""
|
||
if max_schools:
|
||
school_list = school_list[:max_schools]
|
||
|
||
# 启动 Playwright
|
||
try:
|
||
from playwright.sync_api import sync_playwright
|
||
except ImportError:
|
||
print("[x] 需要安装: pip install playwright && playwright install chromium")
|
||
sys.exit(1)
|
||
|
||
cache = load_schid_cache()
|
||
all_results = {}
|
||
stats = {"total": 0, "with_data": 0, "no_data": 0, "schid_error": 0}
|
||
|
||
print(f"扫描 {len(school_list)} 所学校...")
|
||
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(
|
||
headless=True,
|
||
args=["--no-sandbox", "--disable-dev-shm-usage"]
|
||
)
|
||
context = browser.new_context(viewport={"width": 1280, "height": 900})
|
||
page = context.new_page()
|
||
page.set_default_timeout(20000)
|
||
|
||
for i, (school_name, *_) in enumerate(school_list):
|
||
if (i + 1) % 10 == 0:
|
||
print(f" 进度 {i+1}/{len(school_list)}, 已找到数据: {stats['with_data']} 所")
|
||
save_cache(cache)
|
||
|
||
stats["total"] += 1
|
||
|
||
# 获取 schId
|
||
sch_id = get_schid_for_school(page, school_name, cache)
|
||
|
||
if not sch_id:
|
||
all_results[school_name] = {"status": "schid_not_found"}
|
||
stats["schid_error"] += 1
|
||
continue
|
||
|
||
# 检查收费数据
|
||
fee_results = check_fee_urls(page, sch_id)
|
||
|
||
has_any_data = any(r.get("has_data") for r in fee_results.values())
|
||
|
||
all_results[school_name] = {
|
||
"schId": sch_id,
|
||
"fee_results": fee_results,
|
||
"has_fee_data": has_any_data
|
||
}
|
||
|
||
if has_any_data:
|
||
stats["with_data"] += 1
|
||
print(f" ✓ {school_name} (schId={sch_id}): 有收费数据")
|
||
else:
|
||
stats["no_data"] += 1
|
||
|
||
time.sleep(DELAY + random.uniform(0, 0.3))
|
||
|
||
browser.close()
|
||
|
||
save_cache(cache)
|
||
save_results(all_results)
|
||
|
||
print()
|
||
print("=" * 60)
|
||
print(f" 扫描完成：共 {stats['total']} 所")
|
||
print(f" ✓ 有收费数据：{stats['with_data']} 所")
|
||
print(f" ✗ 无收费数据：{stats['no_data']} 所")
|
||
print(f" ! schId 未找到：{stats['schid_error']} 所")
|
||
print(f" 覆盖率：{stats['with_data']/max(stats['total'],1)*100:.1f}%")
|
||
print(f" 结果文件：{SCAN_RESULT}")
|
||
print("=" * 60)
|
||
|
||
return all_results, stats
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 先做一个小规模测试（20所）
|
||
print("先测试20所高校收费数据覆盖率...")
|
||
schools = get_all_schools()
|
||
run_scan(schools, max_schools=20)
|
||
|
||
print("\n然后扫描完整列表? (y/n)")
|
||
answer = input("> ").strip().lower()
|
||
if answer == 'y':
|
||
run_scan(schools) |