CourseBot_a1.0.py

# -*- coding: utf-8 -*-
# Copyright (C) 2026 Your Name
# SPDX-License-Identifier: GPL-3.0-or-later

import os
import sys
import time
import json
import base64
import logging
from io import BytesIO

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from webdriver_manager.chrome import ChromeDriverManager
from openai import OpenAI
from dotenv import load_dotenv

# 加载 .env 文件（若存在）
load_dotenv()

# ================== 配置读取（环境变量优先，缺省则交互式输入） ==================
def get_config():
    config = {}

    # 智谱 API Key
    api_key = os.getenv("ZHIPU_API_KEY")
    if not api_key:
        api_key = input("智谱AI API Key: ").strip()
    config["api_key"] = api_key

    # 目标 URL
    target_url = os.getenv("TARGET_URL")
    if not target_url:
        target_url = input("刷课平台首页 URL: ").strip()
    # 修复 URL 重复协议头
    if target_url.startswith("https://https://"):
        target_url = target_url.replace("https://https://", "https://")
    elif target_url.startswith("http://http://"):
        target_url = target_url.replace("http://http://", "http://")
    config["target_url"] = target_url

    # 账号
    username = os.getenv("USERNAME")
    if not username:
        username = input("登录账号: ").strip()
    config["username"] = username

    # 密码
    password = os.getenv("PASSWORD")
    if not password:
        password = input("登录密码: ").strip()
    config["password"] = password

    # 最大课程数
    max_courses = os.getenv("MAX_COURSES")
    if max_courses:
        try:
            max_courses = int(max_courses)
        except ValueError:
            max_courses = 10
    else:
        max_courses = 10
    config["max_courses"] = max_courses

    # 最大步骤数
    max_steps = os.getenv("MAX_STEPS")
    if max_steps:
        try:
            max_steps = int(max_steps)
        except ValueError:
            max_steps = 200
    else:
        max_steps = 200
    config["max_steps"] = max_steps

    return config

# ================== 日志配置 ==================
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)

# ================== AI 客户端 ==================
def create_ai_client(api_key):
    return OpenAI(
        api_key=api_key,
        base_url="https://open.bigmodel.cn/api/paas/v4/"
    )

# ================== 浏览器驱动（自动管理驱动） ==================
def init_driver():
    chrome_options = Options()
    chrome_options.add_argument("--headless=new")
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--disable-dev-shm-usage")
    chrome_options.add_argument("--disable-gpu")
    chrome_options.add_argument("--window-size=1920,1080")
    # 自动下载并管理 ChromeDriver
    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=chrome_options)
    return driver

# ================== AI 决策函数（截图直接传字节流） ==================
def get_ai_action(client, image_bytes, user_prompt):
    # 将图片字节流编码为 Base64
    image_base64 = base64.b64encode(image_bytes).decode('utf-8')

    try:
        response = client.chat.completions.create(
            model="glm-4v-flash",
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": user_prompt},
                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_base64}"}}
                    ]
                }
            ],
            temperature=0.1,
        )
        content = response.choices[0].message.content
        try:
            action = json.loads(content)
            return action
        except json.JSONDecodeError:
            logger.error(f"AI返回非JSON格式: {content}")
            return None
    except Exception as e:
        logger.error(f"调用AI API失败: {e}")
        return None

# ================== 主程序 ==================
def main():
    config = get_config()
    api_key = config["api_key"]
    target_url = config["target_url"]
    username = config["username"]
    password = config["password"]
    max_courses = config["max_courses"]
    max_steps = config["max_steps"]

    client = create_ai_client(api_key)
    driver = init_driver()
    driver.get(target_url)
    time.sleep(3)

    completed_courses = 0
    consecutive_done = 0          # 连续收到 done 的次数
    consecutive_failures = 0      # 连续执行失败（AI无有效动作或异常）的次数
    step = 0
    exit_flag = False

    while step < max_steps and not exit_flag:
        step += 1
        logger.info(f"===== 第 {step} 步 =====")

        # 截图（直接获取 PNG 字节流，不落盘）
        try:
            screenshot_bytes = driver.get_screenshot_as_png()
        except Exception as e:
            logger.error(f"截图失败: {e}")
            consecutive_failures += 1
            if consecutive_failures >= 3:
                logger.error("连续截图失败，退出程序。")
                break
            time.sleep(3)
            continue

        # 构建 prompt
        prompt = f"""
你是一个专业的网页自动化助手，当前任务是在线刷课平台。请严格按以下规则操作：

1. **登录阶段**：
   - 如果页面出现登录框，请使用账号「{username}」和密码「{password}」进行登录。
   - 如果存在验证码图片（通常是一个包含字母/数字的图片），请**识别验证码内容**，并将其填入验证码输入框。
   - 如果登录失败（如验证码错误），请点击“刷新验证码”按钮，然后重新识别并提交。

2. **课程列表阶段**：
   - 如果当前是课程列表页，请**选择一个尚未完成**的课程（通常标记为“未学习”、“进行中”或进度<100%）点击进入。
   - 如果没有更多未完成课程，请返回 {{"action": "done", "message": "所有课程已完成"}}。

3. **课程学习阶段**：
   - 如果是视频播放页，请点击播放按钮（如“播放”、“继续”），并等待视频自动播放。
   - 如果出现题目/测验，请根据题目内容选择正确答案（若无法判断可随机选）。
   - 如果出现“下一节”、“下一章”或“完成”按钮，请点击以推进进度。
   - 当课程进度达到100%或出现“已完成”提示时，表示该课程结束，请返回课程列表。

4. **返回动作格式（仅JSON）**：
   - 点击：{{"action": "click", "selector": "CSS或XPath"}}
   - 输入：{{"action": "send_keys", "selector": "...", "value": "内容"}}
   - 等待：{{"action": "wait", "seconds": 3}}
   - 滚动：{{"action": "scroll", "direction": "down", "pixels": 500}}
   - 刷新验证码：{{"action": "refresh_captcha", "selector": "刷新按钮选择器"}}
   - 完成：{{"action": "done", "message": "..."}}

当前已完成课程数：{completed_courses} / {max_courses}（达到上限后自动结束）。
请分析截图，返回下一步操作，只输出JSON。
"""

        action = get_ai_action(client, screenshot_bytes, prompt)
        if not action:
            logger.warning("AI未返回有效指令，等待3秒重试...")
            consecutive_failures += 1
            if consecutive_failures >= 5:
                logger.error("连续5次AI返回无效，退出程序。")
                break
            time.sleep(3)
            continue

        # 重置失败计数
        consecutive_failures = 0

        logger.info(f"AI决策: {action}")
        act = action.get("action")

        # ---------- 执行动作 ----------
        try:
            if act == "done":
                logger.info(f"AI报告完成: {action.get('message')}")
                completed_courses += 1
                logger.info(f"当前已完成课程数: {completed_courses}")
                if completed_courses >= max_courses:
                    logger.info("已达到最大课程数，结束任务。")
                    exit_flag = True
                    break
                else:
                    # 可能AI认为所有课程完成，但实际还有，我们继续循环让AI重新分析
                    # 但如果连续多次 done，说明确实没有更多课程了
                    consecutive_done += 1
                    if consecutive_done >= 2:
                        logger.info("连续两次收到 done，认为所有课程已完成，退出。")
                        exit_flag = True
                        break
                    logger.info("继续尝试选择下一门课程...")
                    time.sleep(2)
                    # 不退出循环，让AI重新分析当前页面

            elif act == "click":
                selector = action["selector"]
                if selector.startswith("//") or selector.startswith("(//"):
                    element = driver.find_element(By.XPATH, selector)
                else:
                    element = driver.find_element(By.CSS_SELECTOR, selector)
                element.click()
                consecutive_done = 0   # 重置 done 计数
                time.sleep(2)

            elif act == "send_keys":
                selector = action["selector"]
                value = action["value"]
                if selector.startswith("//") or selector.startswith("(//"):
                    element = driver.find_element(By.XPATH, selector)
                else:
                    element = driver.find_element(By.CSS_SELECTOR, selector)
                element.clear()
                element.send_keys(value)
                consecutive_done = 0
                time.sleep(1)

            elif act == "wait":
                seconds = action.get("seconds", 3)
                logger.info(f"等待 {seconds} 秒...")
                consecutive_done = 0
                time.sleep(seconds)

            elif act == "scroll":
                direction = action.get("direction", "down")
                pixels = action.get("pixels", 500)
                if direction == "down":
                    driver.execute_script(f"window.scrollBy(0, {pixels});")
                else:
                    driver.execute_script(f"window.scrollBy(0, -{pixels});")
                consecutive_done = 0
                time.sleep(1)

            elif act == "refresh_captcha":
                selector = action["selector"]
                if selector.startswith("//") or selector.startswith("(//"):
                    element = driver.find_element(By.XPATH, selector)
                else:
                    element = driver.find_element(By.CSS_SELECTOR, selector)
                element.click()
                logger.info("已刷新验证码，等待新验证码加载...")
                consecutive_done = 0
                time.sleep(2)

            else:
                logger.warning(f"未知动作: {act}，跳过")
                consecutive_done = 0

        except Exception as e:
            logger.error(f"执行动作 {act} 失败: {e}")
            consecutive_failures += 1
            if consecutive_failures >= 3:
                logger.error("连续执行动作失败，退出程序。")
                exit_flag = True
                break
            time.sleep(2)
            continue

        # 每次操作后间隔
        time.sleep(2)

    driver.quit()
    logger.info("刷课流程结束。")

if __name__ == "__main__":
    main()