纯技术干货|Python全自动网站巡检工具|分块代码精讲+落地教程

# -*- coding: utf-8 -*-# 内置基础库import osimport timeimport loggingimport socketimport tracebackimport smtplib# 加密与网络请求库import requestsimport hmacimport hashlibimport base64# 图片处理库from PIL import Image, ImageDraw, ImageFontfrom io import BytesIO# 浏览器自动化库from selenium import webdriverfrom selenium.webdriver.chrome.options import Optionsfrom selenium.webdriver.chrome.service import Servicefrom selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECfrom selenium.webdriver.common.by import By

# ===================== 全局日志配置（运维标准化） =====================logging.basicConfig(    level=logging.INFO,    format="%(asctime)s - %(levelname)s - %(message)s",    handlers=[logging.StreamHandler()])logger = logging.getLogger(__name__)# ===================== 自定义核心配置区（直接修改此处） =====================# 浏览器驱动路径CHROME_PATH = r"C:\Program Files\Google\Chrome\Application\chrome.exe"CHROME_DRIVER_PATH = r"C:\Program Files (x86)\Google\chromedriver.exe"# 截图保存目录（自动创建，无需手动新建）LOCAL_SCREENSHOT_DIR = r"D:\网站巡检截图"os.makedirs(LOCAL_SCREENSHOT_DIR, exist_ok=True)# 待巡检网站列表（支持无限增删）WEBSITES = [    {"name": "网易", "url": "https://www.163.com/#/"},    {"name": "百度", "url": "https://www.baidu.com/"},    {"name": "网站2", "url": "https://ynmuseum.org/index.html"},]# 钉钉机器人推送配置DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=xxx"DINGTALK_SECRET = "SECxxx"# 截图参数配置RETRY_TIMES = 2            # 访问失败重试次数SCREEN_WIDTH = 1920        # 截图宽度（高清大屏）IMAGE_QUALITY = 95         # 截图清晰度MIN_SCREEN_HEIGHT = 800    # 最小页面高度MAX_SCREEN_HEIGHT = 10000  # 最大页面高度，防止超长异常# 邮件归档配置QQ_EMAIL_USER = "你的QQ邮箱"QQ_EMAIL_PWD = "邮箱授权码"TO_EMAIL = "接收邮箱"QQ_SMTP_SERVER = "smtp.qq.com"QQ_SMTP_PORT = 465

def init_chrome_driver():    # 浏览器参数配置    chrome_options = Options()    # 新版高性能无头模式（无弹窗、后台静默运行）    chrome_options.add_argument('--headless=new')    # 初始化窗口尺寸    chrome_options.add_argument(f'--window-size={SCREEN_WIDTH},{MIN_SCREEN_HEIGHT}')    # 页面语言设置中文    chrome_options.add_argument('--lang=zh-CN')    # 忽略HTTPS证书报错    chrome_options.add_argument('--ignore-certificate-errors')    # 关闭GPU加速，规避服务器运行报错    chrome_options.add_argument('--disable-gpu')    # 核心：屏蔽自动化特征，绕过政务网站爬虫检测    chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])    chrome_options.add_experimental_option("useAutomationExtension", False)    # 绑定本地Chrome程序路径    chrome_options.binary_location = CHROME_PATH    # 静默启动驱动，关闭冗余日志    service = Service(executable_path=CHROME_DRIVER_PATH, log_path=os.devnull)    driver = webdriver.Chrome(service=service, options=chrome_options)    # 页面超时与隐式等待设置，防止卡死    driver.set_page_load_timeout(60)    driver.implicitly_wait(5)    logger.info("✅ Chrome浏览器驱动初始化完成")    return driver

def get_windows_font():    """获取系统中文字体，防止水印乱码"""    font_paths = [r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simhei.ttf"]    for path in font_paths:        if os.path.exists(path):            try:                return ImageFont.truetype(path, 36)            except Exception:                continue    return ImageFont.load_default(size=36)def add_time_watermark(img, website_name):    """为截图添加时间+站点水印"""    draw = ImageDraw.Draw(img)    font = get_windows_font()    # 水印文本内容    water_text = f"{website_name} | 巡检时间：{time.strftime('%Y-%m-%d %H:%M:%S')}"    # 自动计算水印位置（右上角）    text_bbox = draw.textbbox((0, 0), water_text, font=font)    text_width, text_height = text_bbox[2], text_bbox[3]    x = img.width - text_width - 20    y = 20    # 半透明白色背景，兼顾美观与辨识度    draw.rectangle(        [(x-10, y-10), (x+text_width+10, y+text_height+10)],        fill=(255, 255, 255, 180)    )    # 绘制水印文字    draw.text((x, y), water_text, font=font, fill=(0, 0, 0))    return img

def capture_full_page(driver, website_name):    try:        # 1. 滚动至页面底部，加载所有懒加载内容        driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")        time.sleep(2)        # 2. 多维度精准获取页面真实高度        page_height = driver.execute_script("""            return Math.min(                document.body.scrollHeight,                document.documentElement.scrollHeight,                document.body.offsetHeight,                document.documentElement.offsetHeight,                %d            );        """ % MAX_SCREEN_HEIGHT)        # 3. 限制高度区间，规避极端异常页面        page_height = max(page_height, MIN_SCREEN_HEIGHT)        page_height = min(page_height, MAX_SCREEN_HEIGHT)        # 4. 适配页面尺寸，滚回顶部（核心：保证截图从页头开始）        driver.set_window_size(SCREEN_WIDTH, page_height)        time.sleep(2)        driver.execute_script("window.scrollTo(0, 0);")        time.sleep(1)        # 5. 截图+格式处理        screenshot = driver.get_screenshot_as_png()        img = Image.open(BytesIO(screenshot))        img = img.convert("RGB") if img.mode in ("RGBA", "P") else img        # 添加水印        img = add_time_watermark(img, website_name)        logger.info(f"✅ {website_name} 截图成功，尺寸：{SCREEN_WIDTH}×{page_height}")        return img    except Exception as e:        # 异常兜底：固定高度截图，绝不空手        logger.warning(f"⚠️ {website_name} 动态截图失败，启用兜底方案：{str(e)}")        driver.execute_script("window.scrollTo(0, 0);")        driver.set_window_size(SCREEN_WIDTH, MIN_SCREEN_HEIGHT)        time.sleep(2)        screenshot = driver.get_screenshot_as_png()        img = Image.open(BytesIO(screenshot)).convert("RGB")        return add_time_watermark(img, website_name)

def capture_and_save(driver, name, url):    err_msg = ""    # 循环重试访问    for retry in range(RETRY_TIMES + 1):        try:            logger.info(f"[{retry+1}/{RETRY_TIMES+1}] 正在巡检：{name}")            # 访问目标网站            driver.get(url)            # 等待页面DOM加载完成            WebDriverWait(driver, 30).until(EC.presence_of_element_located((By.TAG_NAME, "body")))            # 预留页面渲染缓冲时间            time.sleep(8)            # 执行整页截图            img = capture_full_page(driver, name)            # 时间戳命名，防止文件覆盖            timestamp = time.strftime('%Y%m%d_%H%M%S')            filename = f"{name}_{timestamp}.jpg"            filepath = os.path.join(LOCAL_SCREENSHOT_DIR, filename)            # 保存高清截图            img.save(filepath, format="JPEG", quality=IMAGE_QUALITY)            return filepath, filename, "成功", ""        except Exception as e:            err_msg = str(e)[:100]            logger.error(f"{name} 第{retry+1}次巡检失败：{err_msg}")            time.sleep(2)    # 多次失败返回失败状态    fail_filename = f"{name}_失败_{time.strftime('%Y%m%d_%H%M%S')}.jpg"    return "", fail_filename, "失败", err_msg

def get_dingtalk_sign():    """生成钉钉官方加密签名"""    timestamp = str(round(time.time() * 1000))    sign_str = f"{timestamp}\n{DINGTALK_SECRET}"    hmac_obj = hmac.new(DINGTALK_SECRET.encode("utf-8"), sign_str.encode("utf-8"), hashlib.sha256)    sign = base64.b64encode(hmac_obj.digest()).decode("utf-8")    return timestamp, signdef send_dingtalk_summary(results):    timestamp, sign = get_dingtalk_sign()    # 标准无转义接口地址，彻底修复43002报错    api_url = f"{DINGTALK_WEBHOOK}&timestamp={timestamp}&sign={sign}"    # 构建Markdown精美报告    msg = f"### 【网站自动巡检汇总】{time.strftime('%Y-%m-%d %H:%M:%S')}\n"    msg += f"**截图配置**：1920px宽屏 | 动态高度适配 | 时间水印归档\n\n"    # 遍历拼接所有站点巡检结果    for res in results:        msg += f"#### {res['name']} - {res['status']}\n"        msg += f"> 访问地址：{res['url']}\n"        msg += f"> 文件名称：{res['filename']}\n"        if res["error_msg"]:            msg += f"> 异常信息：{res['error_msg']}\n\n"        else:            msg += "\n"    # 统计成功/失败数量    success_num = sum(1 for r in results if r["status"] == "成功")    fail_num = len(results) - success_num    msg += f"**巡检统计**：总计{len(results)}个网站 | 成功{success_num}个 | 失败{fail_num}个"    # POST请求推送    try:        res = requests.post(            api_url,            json={"msgtype": "markdown", "markdown": {"title": "网站巡检报告", "text": msg}, "at": {"isAtAll": fail_num > 0}},            timeout=30        )        if res.json().get("errcode") == 0:            logger.info("✅ 钉钉巡检报告推送成功")        else:            logger.error(f"❌ 钉钉推送失败：{res.json().get('errmsg')}")    except Exception as e:        logger.error(f"❌ 钉钉推送异常：{str(e)}")

def send_email_with_screenshots(results):    logger.info("开始构建巡检邮件报告")    # 筛选有效截图文件    success_list = [r for r in results if r["status"] == "成功" and os.path.exists(r["filepath"])]    # 无截图则跳过发送    if not success_list:        logger.info("无有效截图，跳过邮件发送")        return    try:        # 构建邮件主体        msg = MIMEMultipart()        msg["From"] = QQ_EMAIL_USER        msg["To"] = TO_EMAIL        msg["Subject"] = f"网站巡检截图归档-{time.strftime('%Y-%m-%d %H:%M:%S')}"        # 邮件正文内容        body = f"""网站自动化巡检归档报告巡检时间：{time.strftime('%Y-%m-%d %H:%M:%S')}巡检总数：{len(results)}个 | 成功{len(success_list)}个 | 失败{len(results)-len(success_list)}个巡检明细："""        for res in results:            body += f"\n- {res['name']}：{res['status']}（{res['error_msg'] if res['error_msg'] else'无异常'}）"        msg.attach(MIMEText(body, "plain", "utf-8"))        # 绑定所有截图为附件        for item in success_list:            with open(item["filepath"], "rb") as f:                img_attach = MIMEImage(f.read())                img_attach.add_header("Content-Disposition", "attachment", filename=item["filename"])                msg.attach(img_attach)        logger.info(f"✅ 成功绑定{len(success_list)}个截图附件")        # 双重重试发送        send_ok = False        for retry in range(2):            try:                smtp = smtplib.SMTP_SSL(QQ_SMTP_SERVER, QQ_SMTP_PORT, timeout=20)                smtp.login(QQ_EMAIL_USER, QQ_EMAIL_PWD)                smtp.sendmail(QQ_EMAIL_USER, TO_EMAIL, msg.as_string())                smtp.quit()                send_ok = True                break            except Exception as e:                logger.error(f"邮件发送重试{retry+1}失败：{str(e)}")                time.sleep(3)        logger.info("✅ 邮件发送成功" if send_ok else "❌ 邮件发送最终失败")    except Exception as e:        logger.error(f"❌ 邮件构建失败：{str(e)}")        traceback.print_exc()

def main():    """全自动巡检主流程"""    logger.info("===== 启动全自动网站巡检任务 =====")    # 初始化浏览器    driver = init_chrome_driver()    inspect_results = []    # 批量循环巡检所有网站    for site in WEBSITES:        filepath, filename, status, err = capture_and_save(driver, site["name"], site["url"])        inspect_results.append({            "name": site["name"], "url": site["url"], "status": status,            "filepath": filepath, "filename": filename, "error_msg": err        })        time.sleep(3)    # 强制释放浏览器资源，杜绝进程残留    driver.quit()    logger.info("✅ 浏览器资源已释放")    # 执行钉钉推送+邮件备份    send_dingtalk_summary(inspect_results)    send_email_with_screenshots(inspect_results)    # 控制台输出最终汇总结果    logger.info("\n===== 巡检任务全部完成 =====")    for res in inspect_results:        logger.info(f"{res['name']} | {res['status']} | {res['filename']}")# 程序运行入口if __name__ == "__main__":    main()