数据库文件名

要将节点订阅导入到项目中,可以按照以下步骤进行：

步骤1：安装必要的库

确保你已经安装了所需的库,以下是一些常用的库：

pip install requests
pip install json
pip install datetime
pip install sqlite3

这些库将帮助你处理HTTP请求、解析JSON数据以及存储数据。

步骤2：导入所需的模块

在你的Python脚本中导入这些模块：

import requests
import json
import datetime
import sqlite3

步骤3：定义数据库连接

创建一个数据库连接对象,并准备一个数据库文件：

# 连接数据库
db = sqlite3.connect(DB_FILE)
cursor = db.cursor()

步骤4：定义API获取函数

编写一个函数来从指定的API获取数据,并返回解析后的JSON对象：

def get_api_data(url):
    try:
        response = requests.get(url, timeout=5)
        if response.status_code == 200:
            return json.loads(response.text)
        else:
            print(f"API请求失败，状态码：{response.status_code}")
            return None
    except requests.exceptions.RequestException:
        print("请求API时发生了错误")
        return None

步骤5：定期调用API

设置一个定时任务,定期调用get_api_data函数，获取最新的数据：

from threading import Thread
from time import sleep
import schedule
# 定期调用API获取数据
def periodic_data_fetch():
    while True:
        now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"开始获取数据，时间：{now}")
        data = get_api_data("https://example.com/data")
        if data is not None:
            save_to_database(data)
        print(f"获取数据完成，时间：{now}")
        sleep(60 * 60)  # 每小时获取一次数据
# 启动定时任务
def start_periodic():
    periodic_thread = Thread(target=periodic_data_fetch)
    periodic_thread.start()
# 开始定时任务
start_periodic()

步骤6：保存数据到数据库

编写一个函数,将解析后的数据保存到数据库中：

def save_to_database(data):
    try:
        # 假设数据表名为'nodes'，字段为'id'和'timestamp'
        SQL = "INSERT INTO nodes (id, timestamp) VALUES (?, ?)"
        cursor.execute(SQL, (data.get('id'), datetime.datetime.now()))
        db.commit()
        print(f"新数据已保存，ID：{data.get('id')}")
    except sqlite3.Error as e:
        print(f"保存数据时发生错误：{e}")
        db.rollback()

步骤7：初始化数据库

创建数据库表结构：

# 创建数据表
SQL_CREATE_TABLE = """CREATE TABLE IF NOT EXISTS nodes
                     (id TEXT PRIMARY KEY,
                      timestamp DATETIME NOT NULL)"""
cursor.execute(SQL_CREATE_TABLE)
db.commit()
print("数据库表已创建")

步骤8：运行脚本

将以上代码整合成一个脚本,并运行：

if __name__ == "__main__":
    # 初始化数据库
    db = sqlite3.connect(DB_FILE)
    cursor = db.cursor()
    # 创建数据表
    SQL_CREATE_TABLE = """CREATE TABLE IF NOT EXISTS nodes
                         (id TEXT PRIMARY KEY,
                          timestamp DATETIME NOT NULL)"""
    cursor.execute(SQL_CREATE_TABLE)
    db.commit()
    print("数据库表已创建")
    # 定期调用API
    from threading import Thread
    from time import sleep
    import schedule
    def periodic_data_fetch():
        while True:
            now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            print(f"开始获取数据，时间：{now}")
            data = get_api_data("https://example.com/data")
            if data is not None:
                save_to_database(data)
            print(f"获取数据完成，时间：{now}")
            sleep(60 * 60)  # 每小时获取一次数据
    def start_periodic():
        periodic_thread = Thread(target=periodic_data_fetch)
        periodic_thread.start()
    start_periodic()
    try:
        while True:
            sleep(1)
    except KeyboardInterrupt:
        print("被终止，退出")
        db.close()

步骤9：处理异常和终止

确保在程序中处理异常,特别是当使用Ctrl+C终止脚本时，确保数据库连接正确关闭：

try:
    while True:
        sleep(1)
except KeyboardInterrupt:
    print("程序被终止，正在退出...")
    db.close()
    print("数据库已经关闭")

整合以上步骤,得到一个完整的Python脚本：

import requests
import json
import datetime
import sqlite3
import threading
import schedule
DB_FILE = "data.db"
def get_api_data(url):
    try:
        response = requests.get(url, timeout=5)
        if response.status_code == 200:
            return json.loads(response.text)
        else:
            print(f"API请求失败，状态码：{response.status_code}")
            return None
    except requests.exceptions.RequestException:
        print("请求API时发生了错误")
        return None
def save_to_database(data):
    try:
        SQL = "INSERT INTO nodes (id, timestamp) VALUES (?, ?)"
        cursor.execute(SQL, (data.get('id'), datetime.datetime.now()))
        db.commit()
        print(f"新数据已保存，ID：{data.get('id')}")
    except sqlite3.Error as e:
        print(f"保存数据时发生错误：{e}")
        db.rollback()
def periodic_data_fetch():
    while True:
        now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"开始获取数据，时间：{now}")
        data = get_api_data("https://example.com/data")
        if data is not None:
            save_to_database(data)
        print(f"获取数据完成，时间：{now}")
        sleep(60 * 60)  # 每小时获取一次数据
def start_periodic():
    periodic_thread = threading.Thread(target=periodic_data_fetch)
    periodic_thread.start()
if __name__ == "__main__":
    db = sqlite3.connect(DB_FILE)
    cursor = db.cursor()
    # 创建数据表
    SQL_CREATE_TABLE = """CREATE TABLE IF NOT EXISTS nodes
                         (id TEXT PRIMARY KEY,
                          timestamp DATETIME NOT NULL)"""
    cursor.execute(SQL_CREATE_TABLE)
    db.commit()
    print("数据库表已创建")
    # 开始定时任务
    start_periodic()
    try:
        while True:
            sleep(1)
    except KeyboardInterrupt:
        print("程序被终止，正在退出...")
        db.close()
        print("数据库已经关闭")

使用说明

1. 安装依赖：运行pip install requests json datetime sqlite3安装所需的库。
2. 配置数据库文件：将DB_FILE变量替换为实际的数据库文件路径。
3. 修改API URL：将"https://example.com/data"替换为你实际使用的API URL。
4. 运行脚本：将上述代码保存为node_subscription_import.py，然后运行它。

注意事项

• API速率限制：确保你的API允许频繁的请求，或者在代码中添加Rate Limiting逻辑。
• 数据处理：根据实际需要调整数据存储方式，可以选择 SQLite、MySQL 或其他数据库。
• 线程管理：如果需要更高的并发处理，可以增加线程池或使用更复杂的调度器。

通过以上步骤,你可以成功地将节点订阅数据导入到你的项目中，并定期更新数据。