怎么用SQL分析登录中断模式_SQL分析登录中断规律方法

WITH LoginAttempts AS (
    SELECT
        user_id,
        login_time,
        login_status,
        -- 使用LAG函数获取前一个登录事件的时间和状态
        LAG(login_time, 1) OVER (PARTITION BY user_id ORDER BY login_time) AS prev_login_time,
        LAG(login_status, 1) OVER (PARTITION BY user_id ORDER BY login_time) AS prev_login_status
    FROM
        login_logs
    WHERE
        login_status = 'failure' -- 只关注失败登录
),
ConsecutiveFailures AS (
    SELECT
        user_id,
        login_time,
        login_status,
        -- 计算当前失败和前一个失败之间的时间间隔（秒）
        UNIX_TIMESTAMP(login_time) - UNIX_TIMESTAMP(prev_login_time) AS time_diff_seconds,
        -- 如果前一个也是失败，则序列号加1，否则从1开始
        CASE
            WHEN prev_login_status = 'failure' AND (UNIX_TIMESTAMP(login_time) - UNIX_TIMESTAMP(prev_login_time)) <= 300 -- 5分钟内
            THEN COALESCE(LAG(failure_sequence_num, 1) OVER (PARTITION BY user_id ORDER BY login_time), 0) + 1
            ELSE 1
        END AS failure_sequence_num
    FROM
        LoginAttempts
)
SELECT
    user_id,
    login_time AS last_failure_time,
    failure_sequence_num AS consecutive_failures
FROM
    ConsecutiveFailures
WHERE
    failure_sequence_num >= 5 -- 连续失败次数达到或超过5次
ORDER BY
    user_id, last_failure_time;

LAG

UNIX_TIMESTAMP

INTERVAL

LAG()

ip_address

user_agent

ip_geo_mapping

WITH UserLoginContext AS (
    SELECT
        ll.user_id,
        ll.login_time,
        ll.ip_address,
        ll.user_agent,
        -- 获取上一次登录的IP和User-Agent
        LAG(ll.ip_address, 1) OVER (PARTITION BY ll.user_id ORDER BY ll.login_time) AS prev_ip_address,
        LAG(ll.user_agent, 1) OVER (PARTITION BY ll.user_id ORDER BY ll.login_time) AS prev_user_agent
    FROM
        login_logs ll
    WHERE
        ll.login_status = 'success' -- 通常我们更关心成功登录后的异地/异设备情况
),
LocationAndDeviceChanges AS (
    SELECT
        ulc.user_id,
        ulc.login_time,
        ulc.ip_address,
        ulc.user_agent,
        ulc.prev_ip_address,
        ulc.prev_user_agent,
        -- 判断IP是否变化
        CASE WHEN ulc.ip_address != ulc.prev_ip_address THEN 'IP_CHANGED' ELSE 'IP_SAME' END AS ip_change_status,
        -- 判断User-Agent是否变化（这里简化处理，实际可能需要更复杂的UA解析）
        CASE WHEN ulc.user_agent != ulc.prev_user_agent THEN 'DEVICE_CHANGED' ELSE 'DEVICE_SAME' END AS device_change_status
    FROM
        UserLoginContext ulc
    WHERE
        ulc.prev_ip_address IS NOT NULL -- 排除第一次登录
)
SELECT
    user_id,
    login_time,
    ip_address,
    prev_ip_address,
    ip_change_status,
    user_agent,
    prev_user_agent,
    device_change_status
FROM
    LocationAndDeviceChanges
WHERE
    ip_change_status = 'IP_CHANGED'
    OR device_change_status = 'DEVICE_CHANGED'
ORDER BY
    user_id, login_time DESC;

WHERE

LocationAndDeviceChanges

SELECT
    ll.user_id,
    MAX(ll.login_time) AS last_login_time,
    DATEDIFF(CURRENT_DATE(), MAX(ll.login_time)) AS days_since_last_login
FROM
    login_logs ll
GROUP BY
    ll.user_id
HAVING
    DATEDIFF(CURRENT_DATE(), MAX(ll.login_time)) > 90 -- 筛选超过90天未登录的用户
ORDER BY
    days_since_last_login DESC;

CURRENT_DATE()

DATEDIFF()

AGE()

DATEDIFF()

WITH UserLoginSummary AS (
    SELECT
        user_id,
        MAX(login_time) AS latest_login,
        MIN(login_time) AS first_login_ever,
        COUNT(DISTINCT DATE(login_time)) AS distinct_login_days,
        -- 计算上次登录和倒数第二次登录的时间间隔
        DATEDIFF(MAX(login_time), LAG(MAX(login_time), 1) OVER (PARTITION BY user_id ORDER BY MAX(login_time))) AS days_between_last_two_logins
    FROM
        login_logs
    GROUP BY
        user_id
),
DormantToActive AS (
    SELECT
        uls.user_id,
        uls.latest_login,
        uls.first_login_ever,
        uls.distinct_login_days,
        uls.days_between_last_two_logins
    FROM
        UserLoginSummary uls
    WHERE
        -- 假设我们定义“休眠”为上次登录距今超过90天
        DATEDIFF(CURRENT_DATE(), uls.latest_login) < 7 -- 最近7天内有登录
        AND DATEDIFF(CURRENT_DATE(), uls.first_login_ever) > 90 -- 且首次登录距今超过90天
        AND uls.days_between_last_two_logins > 90 -- 并且最近两次登录间隔也超过90天 (可选，进一步确认休眠)
)
SELECT * FROM DormantToActive;