Messenger_1Million_Stats/million/analyze/find_holes.py

from typing import List
from million.model.message import Message
from million.model.sequence import Sequence
import million.analyze.message_evaluation as msg_val


def compute_sequences(messages: List[Message], accepted_max: int = 1_000_000) -> List[Sequence]:
    sequences: List[Sequence] = []
    current = Sequence(start_message=messages[0])
    
    for message in messages[1:]:
        if msg_val.get(message) > accepted_max: continue

        if msg_val.get(message) == current.end() + 1:
            current.end_message = message
        else:
            sequences.append(current)
            current = Sequence(start_message=message)

    sequences.sort(key=lambda s: s.start())

    merged_sequences: List[Sequence] = []
    previous = sequences[0]
    
    for sequence in sequences[1:]:
        if previous.overlaps(sequence):
            previous.merge(sequence)
        else:
            merged_sequences.append(previous)
            previous = sequence

    return [s for s in merged_sequences if s.length() > 1]


def find_holes(messages: List[Message], accepted_max: int = 1_000_000) -> List[Sequence]:
    """
    Find the holes in the conversation
    """
    sequences = compute_sequences(messages, accepted_max)
    holes = []
    
    for previous, current in zip(sequences[:-1],sequences[1:]):
        holes.append(Sequence(
            start_message=previous.end_message,
            end_message=current.start_message
        ))

    return holes