Python中如何检测某个页面是否打开正常并记录到数据库,请看看代码写对了吗?


Python中如何检测某个页面是否打开正常并记录到数据库,请看看代码写对了吗?

7 回复

:) 运行结果正确吗?


import requests
from datetime import datetime
import sqlite3
import time

def check_website_status(url, timeout=10):
    """检查网站状态并返回结果字典"""
    try:
        response = requests.get(url, timeout=timeout)
        status_code = response.status_code
        is_success = 200 <= status_code < 400
        return {
            'url': url,
            'status_code': status_code,
            'is_success': is_success,
            'response_time': response.elapsed.total_seconds(),
            'timestamp': datetime.now(),
            'error_message': None
        }
    except requests.exceptions.RequestException as e:
        return {
            'url': url,
            'status_code': None,
            'is_success': False,
            'response_time': None,
            'timestamp': datetime.now(),
            'error_message': str(e)
        }

def init_database():
    """初始化SQLite数据库"""
    conn = sqlite3.connect('website_monitor.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS website_checks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            url TEXT NOT NULL,
            status_code INTEGER,
            is_success BOOLEAN,
            response_time REAL,
            timestamp DATETIME,
            error_message TEXT
        )
    ''')
    conn.commit()
    return conn

def save_to_database(conn, check_result):
    """将检查结果保存到数据库"""
    cursor = conn.cursor()
    cursor.execute('''
        INSERT INTO website_checks 
        (url, status_code, is_success, response_time, timestamp, error_message)
        VALUES (?, ?, ?, ?, ?, ?)
    ''', (
        check_result['url'],
        check_result['status_code'],
        check_result['is_success'],
        check_result['response_time'],
        check_result['timestamp'],
        check_result['error_message']
    ))
    conn.commit()

def monitor_website(url, interval=60):
    """持续监控网站状态"""
    conn = init_database()
    
    print(f"开始监控: {url}")
    print("按 Ctrl+C 停止监控\n")
    
    try:
        while True:
            result = check_website_status(url)
            
            # 保存到数据库
            save_to_database(conn, result)
            
            # 输出当前状态
            status = "✓ 正常" if result['is_success'] else "✗ 异常"
            print(f"[{result['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}] {status} "
                  f"状态码: {result['status_code'] or 'N/A'} "
                  f"响应时间: {result['response_time'] or 'N/A':.3f}s")
            
            if result['error_message']:
                print(f"   错误信息: {result['error_message']}")
            
            time.sleep(interval)
            
    except KeyboardInterrupt:
        print("\n监控已停止")
    finally:
        conn.close()

# 使用示例
if __name__ == "__main__":
    # 监控单个网站
    monitor_website("https://www.example.com", interval=300)  # 每5分钟检查一次
    
    # 或者批量检查多个网站
    def batch_check(urls):
        conn = init_database()
        for url in urls:
            result = check_website_status(url)
            save_to_database(conn, result)
            print(f"{url}: {'正常' if result['is_success'] else '异常'}")
        conn.close()
    
    # batch_check(["https://www.google.com", "https://www.github.com"])

代码说明:

  1. check_website_status() 函数用 requests 库检测网站状态,处理超时和网络异常
  2. init_database() 创建 SQLite 数据库表存储检查结果
  3. save_to_database() 将每次检查结果存入数据库
  4. monitor_website() 实现持续监控功能,可设置检查间隔
  5. 支持批量检查和单站点持续监控两种模式

数据库字段说明:

  • url: 被检查的网址
  • status_code: HTTP状态码(200、404等)
  • is_success: 是否成功(2xx/3xx状态码)
  • response_time: 响应时间(秒)
  • timestamp: 检查时间戳
  • error_message: 错误信息(如果有)

建议: 可以考虑添加邮件或短信告警功能。

还在配环境,代码没测试过…

跑一下就知道对不对的事情 还需要发上来看嘛

此楼正解

好图

回到顶部