已完成网页爬虫，如何每5分钟将数据带时间戳存入SQLite？

1. 整合SQLite数据库操作 #

Python自带sqlite3库，无需额外安装。直接在你的爬虫代码中添加数据库连接、表创建、数据插入逻辑：

import requests
from bs4 import BeautifulSoup
import sqlite3
from datetime import datetime

# 爬虫部分
URL = "http://ludvikasegel.com/wx/cloudbase.asp"
page = requests.get(URL)
soup = BeautifulSoup(page.content, "html.parser")
molnbas = soup.find_all("div", class_="cloudbase")
cloud_data = molnbas[0].text.strip()
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")  # 格式化时间戳

# 数据库操作部分
# 连接数据库（不存在则自动创建）
conn = sqlite3.connect("cloudbase_data.db")
cursor = conn.cursor()

# 创建数据表（首次运行自动创建，重复执行无影响）
cursor.execute('''
CREATE TABLE IF NOT EXISTS cloud_records (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    cloudbase TEXT NOT NULL,
    record_time TEXT NOT NULL
)
''')

# 插入数据
cursor.execute("INSERT INTO cloud_records (cloudbase, record_time) VALUES (?, ?)", (cloud_data, current_time))

# 提交并关闭连接
conn.commit()
conn.close()

print(f"已保存数据: {cloud_data}，时间: {current_time}")

2. 实现每5分钟定时执行 #

用schedule库实现定时任务，先安装依赖：

pip install schedule

修改代码加入定时逻辑，同时增加异常处理避免爬虫失败导致程序中断：

import requests
from bs4 import BeautifulSoup
import sqlite3
from datetime import datetime
import schedule
import time

def crawl_and_save():
    # 爬虫部分
    URL = "http://ludvikasegel.com/wx/cloudbase.asp"
    try:
        page = requests.get(URL, timeout=10)
        page.raise_for_status()  # 捕获HTTP请求错误
        soup = BeautifulSoup(page.content, "html.parser")
        molnbas = soup.find_all("div", class_="cloudbase")
        if molnbas:
            cloud_data = molnbas[0].text.strip()
            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            # 数据库操作
            conn = sqlite3.connect("cloudbase_data.db")
            cursor = conn.cursor()
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS cloud_records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                cloudbase TEXT NOT NULL,
                record_time TEXT NOT NULL
            )
            ''')
            cursor.execute("INSERT INTO cloud_records (cloudbase, record_time) VALUES (?, ?)", (cloud_data, current_time))
            conn.commit()
            conn.close()
            print(f"已保存数据: {cloud_data}，时间: {current_time}")
        else:
            print("未找到目标数据")
    except Exception as e:
        print(f"执行出错: {str(e)}")

# 设置每5分钟执行一次
schedule.every(5).minutes.do(crawl_and_save)

# 启动定时任务循环
print("定时任务已启动，每5分钟执行一次...")
while True:
    schedule.run_pending()
    time.sleep(1)

3. 验证数据是否存入 #

可以用sqlite3命令行打开数据库文件，执行查询语句验证：

SELECT * FROM cloud_records;

或者在Python中添加以下代码快速查看：

conn = sqlite3.connect("cloudbase_data.db")
cursor = conn.cursor()
cursor.execute("SELECT * FROM cloud_records")
rows = cursor.fetchall()
for row in rows:
    print(row)
conn.close()

内容的提问来源于stack exchange，提问作者Armond