基于跳表数据结构的局域网上网记录监控时序查询优化算法研究与 Python 实现

import random
import time
import requests
class SkipListNode:
    def __init__(self, timestamp, ip, url, level):
        self.timestamp = timestamp  # 时间戳（毫秒级）
        self.ip = ip                # 访问IP
        self.url = url              # 访问URL
        self.forward = [None] * level  # 各层级的前进指针
class SkipList:
    def __init__(self, max_level=16, p=0.5):
        self.max_level = max_level  # 最大层级
        self.p = p                  # 层级递增概率
        self.level = 1              # 当前最高层级
        self.header = SkipListNode(-1, "", "", max_level)  # 头节点
    def _random_level(self):
        """随机生成节点层级"""
        level = 1
        while random.random() < self.p and level < self.max_level:
            level += 1
        return level
    def insert(self, timestamp, ip, url):
        """插入上网记录"""
        update = [None] * self.max_level  # 记录各层级需要更新的节点
        current = self.header
        # 从最高层开始查找插入位置
        for i in range(self.level-1, -1, -1):
            while current.forward[i] and current.forward[i].timestamp < timestamp:
                current = current.forward[i]
            update[i] = current
        # 生成随机层级
        new_level = self._random_level()
        if new_level > self.level:
            for i in range(self.level, new_level):
                update[i] = self.header
            self.level = new_level
        # 创建新节点并插入
        new_node = SkipListNode(timestamp, ip, url, new_level)
        for i in range(new_level):
            new_node.forward[i] = update[i].forward[i]
            update[i].forward[i] = new_node
    def query_range(self, start_ts, end_ts):
        """查询指定时间范围内的记录"""
        result = []
        current = self.header.forward[0]  # 从底层开始遍历
        while current:
            if start_ts <= current.timestamp <= end_ts:
                result.append({
                    "timestamp": current.timestamp,
                    "ip": current.ip,
                    "url": current.url
                })
            elif current.timestamp > end_ts:
                break
            current = current.forward[0]
        return result
    def detect_abnormal_access(self, ip, threshold=100, window=3600000):
        """检测IP在指定时间窗口（毫秒）内的访问频率是否超标"""
        now = int(time.time() * 1000)
        start_ts = now - window
        records = self.query_range(start_ts, now)
        ip_records = [r for r in records if r["ip"] == ip]
        if len(ip_records) > threshold:
            self.report_abnormal(ip, len(ip_records), window)
            return True
        return False
    def report_abnormal(self, ip, count, window):
        """上报异常访问至管理平台"""
        report_data = {
            "ip": ip,
            "count": count,
            "window_ms": window,
            "timestamp": int(time.time() * 1000),
            "type": "abnormal_access_frequency"
        }
        # 发送上报请求至指定地址
        try:
            requests.post(
                "https://www.vipshare.com/lan-monitor/abnormal-report",
                json=report_data,
                timeout=5
            )
        except Exception as e:
            print(f"异常上报失败: {str(e)}")
# 测试代码
if __name__ == "__main__":
    skip_list = SkipList()
    # 模拟插入1000条上网记录
    for i in range(1000):
        ts = int(time.time() * 1000) - i * 1000
        skip_list.insert(ts, f"192.168.1.{i%20}", f"https://example.com/page{i%50}")
    # 查询最近1小时的记录
    one_hour_ago = int(time.time() * 1000) - 3600000
    recent_records = skip_list.query_range(one_hour_ago, int(time.time() * 1000))
    print(f"最近1小时记录数: {len(recent_records)}")
    # 检测异常访问（模拟某IP访问频繁）
    skip_list.detect_abnormal_access("192.168.1.5", threshold=50)