AI 时代，请放下你的“技术思维”

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
上海高中录取分数和升学率数据收集系统
基于豆包AI的自动化数据爬虫

功能：
1. 自动收集上海各高中录取分数线数据
2. 收集C9、985、211大学升学率数据
3. 支持历史数据分析和未来预测
4. 数据导出和可视化分析
"""

import asyncio
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import re
import json
import time
from typing import List, Dict, Optional, Tuple
from playwright.async_api import async_playwright, Page, Browser
import logging
from pathlib import Path
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import PolynomialFeatures
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False

class ShanghaiSchoolDataCollector:
    """上海高中数据收集器"""

    def __init__(self, headless: bool = False, slow_mo: int = 1000):
        self.headless = headless
        self.slow_mo = slow_mo
        self.browser: Optional[Browser] = None
        self.page: Optional[Page] = None

        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('shanghai_school_scraper.log', encoding='utf-8'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

        # 目标学校配置
        self.target_schools = {
            "四校": [
                "上海中学", "华师大二附中", "复旦附中", "交大附中"
            ],
            "八大": [
                "南洋模范中学", "格致中学", "大同中学", "控江中学",
                "七宝中学", "延安中学", "建平中学", "复兴高级中学"
            ],
            "市重点": [
                "向明中学", "卢湾高级中学", "敬业中学", "光明中学",
                "市西中学", "市北中学", "回民中学", "六十中学",
                "曹杨二中", "宜川中学", "晋元高级中学", "新中高级中学",
                "华师大一附中", "中原中学", "华东理工大学附中", "上理工附中",
                "上大附中", "上海师大附中", "行知中学", "吴淞中学",
                "杨浦高级中学", "同济一附中", "理工大附中", "复旦实验中学",
                "进才中学", "洋泾中学", "东昌中学", "川沙中学",
                "南汇中学", "华师大三附中", "上南中学", "金山中学",
                "张堰中学", "朱家角中学", "青浦高级中学", "白鹤中学",
                "崇明中学", "民本中学", "扬子中学"
            ],
            "闵行区重点": [
                "闵行中学", "莘庄中学", "闵行二中", "闵行三中",
                "文来中学", "协和双语学校", "星河湾双语学校"
            ]
        }

        # 查询年份
        self.target_years = [2021, 2022, 2023, 2024, 2025]

        # 数据存储
        self.admission_data = []
        self.university_data = []

        # 豆包网站配置
        self.doubao_url = "https://www.doubao.com/chat/"
        self.wait_time = 3  # 等待响应时间（秒）

    async def start_browser(self):
        """启动浏览器"""
        playwright = await async_playwright().start()
        self.browser = await playwright.chromium.launch(
            headless=self.headless,
            slow_mo=self.slow_mo
        )

        context = await self.browser.new_context(
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        )

        self.page = await context.new_page()
        self.logger.info("浏览器启动成功")

    async def close_browser(self):
        """关闭浏览器"""
        if self.browser:
            await self.browser.close()
            self.logger.info("浏览器已关闭")

    async def navigate_to_doubao(self):
        """导航到豆包网站"""
        try:
            await self.page.goto(self.doubao_url, wait_until="networkidle")
            await self.page.wait_for_timeout(2000)
            self.logger.info("成功访问豆包网站")
            return True
        except Exception as e:
            self.logger.error(f"访问豆包网站失败: {e}")
            return False

    async def send_query_and_get_response(self, query: str) -> Optional[str]:
        """发送查询并获取响应"""
        try:
            # 查找输入框
            input_selector = "textarea, input[type='text'], [contenteditable='true']"
            await self.page.wait_for_selector(input_selector, timeout=10000)

            # 清空并输入查询
            await self.page.fill(input_selector, "")
            await self.page.fill(input_selector, query)
            await self.page.wait_for_timeout(500)

            # 发送查询（尝试多种方式）
            try:
                await self.page.press(input_selector, "Enter")
            except:
                # 如果Enter不工作，尝试点击发送按钮
                send_button = await self.page.query_selector("button[type='submit'], button:has-text('发送'), button:has-text('提交')")
                if send_button:
                    await send_button.click()

            self.logger.info(f"已发送查询: {query}")

            # 等待响应
            await self.page.wait_for_timeout(self.wait_time * 1000)

            # 获取响应内容
            response_selectors = [
                ".message-content", ".response", ".answer", ".reply",
                "[data-testid='message']", ".chat-message", ".ai-response"
            ]

            response_text = ""
            for selector in response_selectors:
                elements = await self.page.query_selector_all(selector)
                if elements:
                    # 获取最后一个响应
                    last_element = elements[-1]
                    text = await last_element.inner_text()
                    if text and len(text) > 50:  # 确保是有意义的响应
                        response_text = text
                        break

            if not response_text:
                # 如果没有找到特定选择器，尝试获取页面中的所有文本
                page_text = await self.page.inner_text("body")
                # 简单的响应提取逻辑
                lines = page_text.split('\n')
                for i, line in enumerate(lines):
                    if query[:10] in line and i + 1 < len(lines):
                        response_text = '\n'.join(lines[i+1:i+20])  # 获取查询后的20行
                        break

            if response_text:
                self.logger.info(f"获取到响应，长度: {len(response_text)}")
                return response_text
            else:
                self.logger.warning("未能获取到有效响应")
                return None

        except Exception as e:
            self.logger.error(f"查询过程出错: {e}")
            return None

    def extract_admission_scores(self, text: str, school: str, year: int) -> Dict:
        """从响应文本中提取录取分数线信息"""
        data = {
            "学校": school,
            "年份": year,
            "录取分数线": None,
            "最低分": None,
            "平均分": None,
            "学生来源": [],
            "原始文本": text[:500]  # 保存前500字符用于调试
        }

        try:
            # 提取分数线（各种可能的表达方式）
            score_patterns = [
                r'录取分数[线]?[：:]?\s*(\d{3,4})',
                r'分数线[：:]?\s*(\d{3,4})',
                r'最低分[：:]?\s*(\d{3,4})',
                r'投档线[：:]?\s*(\d{3,4})',
                r'(\d{3,4})分',
                r'分数[：:]?\s*(\d{3,4})'
            ]

            for pattern in score_patterns:
                matches = re.findall(pattern, text)
                if matches:
                    scores = [int(score) for score in matches if 400 <= int(score) <= 700]
                    if scores:
                        data["录取分数线"] = max(scores)  # 取最高分作为录取线
                        data["最低分"] = min(scores)
                        data["平均分"] = sum(scores) / len(scores)
                        break

            # 提取学生来源信息
            source_patterns = [
                r'来自([^，。]+区)',
                r'([^，。]+区)学生',
                r'生源[：:]?([^，。]+)',
                r'主要来源[：:]?([^，。]+)'
            ]

            sources = []
            for pattern in source_patterns:
                matches = re.findall(pattern, text)
                sources.extend(matches)

            data["学生来源"] = list(set(sources))  # 去重

        except Exception as e:
            self.logger.error(f"提取录取分数信息时出错: {e}")

        return data

    def extract_university_rates(self, text: str, school: str, year: int) -> Dict:
        """从响应文本中提取大学升学率信息"""
        data = {
            "学校": school,
            "年份": year,
            "C9升学率": None,
            "985升学率": None,
            "211升学率": None,
            "一本率": None,
            "本科率": None,
            "原始文本": text[:500]
        }

        try:
            # 提取各类升学率
            rate_patterns = {
                "C9升学率": [r'C9[^\d]*(\d+(?:\.\d+)?)%', r'九校联盟[^\d]*(\d+(?:\.\d+)?)%'],
                "985升学率": [r'985[^\d]*(\d+(?:\.\d+)?)%', r'985院校[^\d]*(\d+(?:\.\d+)?)%'],
                "211升学率": [r'211[^\d]*(\d+(?:\.\d+)?)%', r'211院校[^\d]*(\d+(?:\.\d+)?)%'],
                "一本率": [r'一本[^\d]*(\d+(?:\.\d+)?)%', r'重点大学[^\d]*(\d+(?:\.\d+)?)%'],
                "本科率": [r'本科[^\d]*(\d+(?:\.\d+)?)%', r'本科院校[^\d]*(\d+(?:\.\d+)?)%']
            }

            for rate_type, patterns in rate_patterns.items():
                for pattern in patterns:
                    matches = re.findall(pattern, text)
                    if matches:
                        rates = [float(rate) for rate in matches if 0 <= float(rate) <= 100]
                        if rates:
                            data[rate_type] = max(rates)  # 取最高值
                            break
                if data[rate_type] is not None:
                    break

        except Exception as e:
            self.logger.error(f"提取升学率信息时出错: {e}")

        return data

    async def collect_school_data(self, school: str, year: int) -> Tuple[Dict, Dict]:
        """收集单个学校的数据"""
        admission_data = None
        university_data = None

        try:
            # 查询1: 录取分数及学生来源
            query1 = f"{year}年上海{school}录取分数及学生来源"
            response1 = await self.send_query_and_get_response(query1)

            if response1:
                admission_data = self.extract_admission_scores(response1, school, year)
                self.logger.info(f"已收集 {school} {year}年 录取分数数据")

            await self.page.wait_for_timeout(2000)  # 等待2秒避免请求过快

            # 查询2: C9、985、211升学率
            query2 = f"{year}年上海{school}C9、985、211入线率"
            response2 = await self.send_query_and_get_response(query2)

            if response2:
                university_data = self.extract_university_rates(response2, school, year)
                self.logger.info(f"已收集 {school} {year}年 升学率数据")

        except Exception as e:
            self.logger.error(f"收集 {school} {year}年 数据时出错: {e}")

        return admission_data, university_data

    async def collect_all_data(self):
        """收集所有学校的数据"""
        await self.start_browser()

        if not await self.navigate_to_doubao():
            await self.close_browser()
            return

        total_schools = sum(len(schools) for schools in self.target_schools.values())
        total_queries = total_schools * len(self.target_years)
        current_query = 0

        try:
            for category, schools in self.target_schools.items():
                self.logger.info(f"开始收集 {category} 数据")

                for school in schools:
                    for year in self.target_years:
                        current_query += 1
                        self.logger.info(f"进度: {current_query}/{total_queries} - 收集 {school} {year}年 数据")

                        admission_data, university_data = await self.collect_school_data(school, year)

                        if admission_data:
                            admission_data["学校类别"] = category
                            self.admission_data.append(admission_data)

                        if university_data:
                            university_data["学校类别"] = category
                            self.university_data.append(university_data)

                        # 保存临时结果
                        if current_query % 10 == 0:
                            await self.save_temporary_results(current_query)

                        await self.page.wait_for_timeout(3000)  # 等待3秒

        except Exception as e:
            self.logger.error(f"数据收集过程中出错: {e}")
        finally:
            await self.close_browser()

    async def save_temporary_results(self, query_count: int):
        """保存临时结果"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

        if self.admission_data:
            df_admission = pd.DataFrame(self.admission_data)
            filename = f"上海高中录取分数_临时结果_{query_count}条_{timestamp}.xlsx"
            df_admission.to_excel(filename, index=False, engine='openpyxl')

        if self.university_data:
            df_university = pd.DataFrame(self.university_data)
            filename = f"上海高中升学率_临时结果_{query_count}条_{timestamp}.xlsx"
            df_university.to_excel(filename, index=False, engine='openpyxl')

        self.logger.info(f"已保存临时结果，共 {query_count} 条查询")

    def predict_2026_scores(self) -> pd.DataFrame:
        """预测2026年录取分数线"""
        if not self.admission_data:
            self.logger.warning("没有录取分数数据，无法进行预测")
            return pd.DataFrame()

        df = pd.DataFrame(self.admission_data)
        df = df[df['录取分数线'].notna()]

        predictions = []

        for school in df['学校'].unique():
            school_data = df[df['学校'] == school].sort_values('年份')

            if len(school_data) >= 3:  # 至少需要3年数据进行预测
                X = school_data['年份'].values.reshape(-1, 1)
                y = school_data['录取分数线'].values

                # 使用线性回归进行预测
                model = LinearRegression()
                model.fit(X, y)

                # 预测2026年
                predicted_score = model.predict([[2026]])[0]

                # 获取学校类别
                category = school_data.iloc[0]['学校类别']

                predictions.append({
                    '学校': school,
                    '学校类别': category,
                    '2026年预测分数线': round(predicted_score, 1),
                    '历史平均分': round(school_data['录取分数线'].mean(), 1),
                    '数据年数': len(school_data),
                    '趋势': '上升' if model.coef_[0] > 0 else '下降'
                })

        return pd.DataFrame(predictions)

    def generate_analysis_report(self):
        """生成分析报告"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

        with pd.ExcelWriter(f"上海高中数据分析报告_{timestamp}.xlsx", engine='openpyxl') as writer:
            # 录取分数数据
            if self.admission_data:
                df_admission = pd.DataFrame(self.admission_data)
                df_admission.to_excel(writer, sheet_name='录取分数数据', index=False)

            # 升学率数据
            if self.university_data:
                df_university = pd.DataFrame(self.university_data)
                df_university.to_excel(writer, sheet_name='升学率数据', index=False)

            # 2026年预测
            df_predictions = self.predict_2026_scores()
            if not df_predictions.empty:
                df_predictions.to_excel(writer, sheet_name='2026年预测', index=False)

            # 统计分析
            if self.admission_data:
                df_admission = pd.DataFrame(self.admission_data)
                df_stats = df_admission.groupby(['学校类别', '年份'])['录取分数线'].agg([
                    'count', 'mean', 'median', 'std'
                ]).round(2)
                df_stats.to_excel(writer, sheet_name='分数统计分析')

        self.logger.info(f"分析报告已生成: 上海高中数据分析报告_{timestamp}.xlsx")

    async def run(self):
        """运行完整的数据收集流程"""
        self.logger.info("开始上海高中数据收集")
        self.logger.info(f"目标学校总数: {sum(len(schools) for schools in self.target_schools.values())}")
        self.logger.info(f"目标年份: {self.target_years}")

        # 收集数据
        await self.collect_all_data()

        # 生成报告
        self.generate_analysis_report()

        self.logger.info("数据收集完成")

        # 打印统计信息
        print(f"\n=== 数据收集统计 ===")
        print(f"录取分数数据条数: {len(self.admission_data)}")
        print(f"升学率数据条数: {len(self.university_data)}")

        if self.admission_data:
            df_admission = pd.DataFrame(self.admission_data)
            valid_scores = df_admission[df_admission['录取分数线'].notna()]
            print(f"有效录取分数数据: {len(valid_scores)}")

        if self.university_data:
            df_university = pd.DataFrame(self.university_data)
            valid_rates = df_university[df_university['985升学率'].notna()]
            print(f"有效升学率数据: {len(valid_rates)}")

# 使用示例
async def main():
    """主函数"""
    collector = ShanghaiSchoolDataCollector(
        headless=False,  # 设置为True可以无头模式运行
        slow_mo=1000     # 操作间隔时间（毫秒）
    )

    await collector.run()

if __name__ == "__main__":
    print("上海高中数据收集系统")
    print("=" * 50)
    print("功能说明：")
    print("1. 自动收集上海各高中录取分数线数据")
    print("2. 收集C9、985、211大学升学率数据")
    print("3. 支持历史数据分析和未来预测")
    print("4. 生成详细的数据分析报告")
    print("=" * 50)

    # 运行数据收集
    asyncio.run(main())

pip install playwright pandas numpy matplotlib seaborn scikit-learn openpyxl
playwright install chromium

python shanghai_school_scraper.py