如何编写bot_detection函数检测满足特定条件的机器人用户？

Alright, let's tackle this bot detection problem step by step. I'll walk you through the core approach and provide a Python implementation with detailed explanations tailored to your needs.

First, let's break down the key requirements to make sure we cover all bases:

• We need to analyze behavior per user since bot detection is tied to individual activity patterns
• The time window is any 4-minute (240-second) period, so we must sort each user's records by timestamp to efficiently check sliding windows
• Two conditions must be satisfied simultaneously: the window has at least 10 total operations, and at least 5 of those are from the specified action types

Here's the step-by-step plan:

1. Read and parse data: Load the input file, parse each JSON record, and group records by user
2. Sort records per user: Sort each user's records by timestamp—this is essential for the sliding window technique to work
3. Sliding window detection: Use a two-pointer approach to check every possible 4-minute window for each user, counting total actions and specified actions
4. Collect results: If a user has any window that meets the bot criteria, collect all their access records (or just the triggering window, depending on your needs)

import json
from collections import defaultdict

def bot_detection(input_file_path, specified_actions={"navigate", "click"}):
    """
    Detects bot users and collects their access records based on behavior patterns.
    
    Args:
        input_file_path (str): Path to the input data file (each line is a JSON object)
        specified_actions (set): Set of action types to count for bot detection
        
    Returns:
        list: All access records belonging to bot users
    """
    # Step 1: Read input file and group records by user
    user_records = defaultdict(list)
    with open(input_file_path, 'r') as f:
        for line in f:
            try:
                record = json.loads(line.strip())
                # Validate required fields exist to avoid errors later
                if all(key in record for key in ["timestamp", "user", "action"]):
                    user_records[record["user"]].append(record)
            except json.JSONDecodeError:
                # Skip any malformed lines to keep the function running
                continue
    
    bot_records = []
    
    # Step 2: Process each user's sorted records
    for user, records in user_records.items():
        # Sort records by timestamp to enable valid sliding window checks
        sorted_records = sorted(records, key=lambda x: x["timestamp"])
        total_records = len(sorted_records)
        left_ptr = 0
        specified_action_count = 0
        user_is_bot = False
        
        # Step 3: Use two pointers to check sliding windows
        for right_ptr in range(total_records):
            # Update count if current action is in our specified set
            if sorted_records[right_ptr]["action"] in specified_actions:
                specified_action_count += 1
            
            # Shrink the window from the left if it exceeds 4 minutes (240 seconds)
            while sorted_records[right_ptr]["timestamp"] - sorted_records[left_ptr]["timestamp"] > 240:
                if sorted_records[left_ptr]["action"] in specified_actions:
                    specified_action_count -= 1
                left_ptr += 1
            
            # Check if current window meets bot criteria
            window_size = right_ptr - left_ptr + 1
            if window_size >= 10 and specified_action_count >= 5:
                user_is_bot = True
                break  # No need to check further windows for this user
        
        # Step 4: Add all records of bot users to the result
        if user_is_bot:
            bot_records.extend(records)
    
    return bot_records

Let's unpack each part of the code to understand how it works:

1. Data Parsing & Grouping #

• We use defaultdict(list) to automatically group records by the user field, which lets us isolate each user's behavior for analysis.
• We skip malformed JSON lines and validate required fields to prevent runtime errors later.

2. Sorting Records #

• Sorting each user's records by timestamp is non-negotiable here. Without sorted times, we can't efficiently maintain a valid 4-minute window as we iterate through records.

3. Sliding Window Detection #

• The two-pointer approach (left_ptr and right_ptr) lets us maintain a dynamic window of records:
  • The right_ptr expands the window by moving to the next record.
  • The left_ptr contracts the window whenever the time difference between the rightmost and leftmost record exceeds 240 seconds.
• We track specified_action_count incrementally, which avoids recalculating the count from scratch for every window—this keeps the solution efficient (O(n log n) time per user, dominated by the sorting step).
• As soon as we find any window that meets both bot criteria, we mark the user as a bot and stop checking further windows to save time.

4. Collecting Results #

• By default, we add all of a bot user's records to the result. If you only want the specific window(s) that triggered the detection, you can modify this part to collect the slice sorted_records[left_ptr:right_ptr+1] instead of extending with all user records.

Customization Tips #

• The specified_actions parameter lets you define which actions count towards the bot criteria. For example, you could call bot_detection("data.txt", {"login", "submit"}) to target different actions.
• If your input file stores JSON as a single array (not one object per line), adjust the parsing step to load the entire file with json.load(f) instead of iterating line by line.

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