Merge branch 'DEV-Mael' of DemiSel/Messenger_1Million_Stats into master

.gitignore

 *.pyc
 
+output/*
+!output/.gitkeep
+
 # Packages
 *.egg
 !/tests/**/*.egg

million/analyze/find_holes.py

 
 
 from typing import List
-from million.model.hole import Hole
 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 = Sequence(
-        start=messages[0].get_counted_value(),
-        start_message=messages[0],
-        end=messages[0].get_counted_value(),
-        end_message=messages[0]
-    )
-    for i in range(1, len(messages)):
-        message = messages[i]
-        message_value = message.get_counted_value()
-        if message_value > accepted_max:
-            continue
-        if message_value - current_sequence.end == 1:
-            current_sequence.end = message_value
-            current_sequence.end_message = message
+    """ 
+    Takes a list of messages as input and returns a list of sequences
+    for every following messages with following 'counted values'
+    """
+    sequences: List[Sequence] = [Sequence(start_message=messages[0])]
+    
+    for message in messages[1:]:
+        if msg_val.get(message) > accepted_max: continue
+
+        if msg_val.get(message) == sequences[-1].end() + 1:
+            sequences[-1].end_message = message
         else:
-            sequences.append(current_sequence)
-            current_sequence = Sequence(
-                start=message_value,
-                start_message=message,
-                end=message_value,
-                end_message=message
-            )
-
-    # order the sequences by start
-    sequences.sort(key=lambda s: s.start)
-
-    merged_sequences: List[Sequence] = []
-    current_sequence = sequences[0]
-    for i in range(1, len(sequences)):
-        sequence = sequences[i]
-        sequence_start_is_in_current_sequence = current_sequence.start <= sequence.start and current_sequence.end >= sequence.start
-        sequence_end_is_further = sequence.end > current_sequence.end
-        sequence_start_is_current_end_or_next = sequence.start == current_sequence.end + 1
-
-        if sequence_start_is_in_current_sequence or sequence_start_is_current_end_or_next:
-            if sequence_end_is_further:
-                current_sequence.end = sequence.end
-                current_sequence.end_message = sequence.end_message
+            sequences.append(Sequence(start_message=message))
+
+    return sequences            
+
+
+def merge_duplicates(sequences: List[Sequence]) -> List[Sequence]:
+    """ 
+    Take sequences as an input and returns a list with every
+    overlapping input sequences merged in one
+    """
+    o_sequences = sorted(sequences, key= lambda s : s.start())
+    current = o_sequences[0]
+
+    result = []
+        
+    for sequence in o_sequences[1:]:
+        if current.overlaps(sequence):
+            current.merge(sequence)
         else:
-            merged_sequences.append(current_sequence)
-            current_sequence = sequence
+            result.append(current)
+            current = sequence
 
-    # Having merged the sequences once, any sequence having start = end can be removed
-    return [s for s in merged_sequences if s.start != s.end]
+    return result
 
 
-def find_holes(messages: List[Message], accepted_max: int = 1_000_000) -> List[Hole]:
+
+def invert_sequences(sequences: List[Sequence]) -> List[Sequence]:
+    """ 
+    Returns the sequences representing the spaces between
+    the ones given as input
+    """
+    result = []
+
+    for previous, current in zip(sequences[:-1],sequences[1:]):
+        result.append(Sequence(
+            start_message=previous.end_message,
+            end_message=current.start_message
+        ))
+
+    return result
+
+def find_holes(messages: List[Message], accepted_max: int = 1_000_000) -> List[Sequence]:
     """
     Find the holes in the conversation
+    TODO might need to be moved inside scripts/find_holes
     """
-    merged_sequences = compute_sequences(messages, accepted_max)
-    holes = []
-    for i in range(1, len(merged_sequences)):
-        previous_sequence = merged_sequences[i - 1]
-        sequence = merged_sequences[i]
-        if sequence.start - previous_sequence.end > 1:
-            holes.append(Hole(
-                start=previous_sequence.end,
-                end=sequence.start,
-                start_message=previous_sequence.end_message,
-                end_message=sequence.start_message
-            ))
-    return holes
+    sequences = compute_sequences(messages, accepted_max)
+    merged = merge_duplicates(sequences)
+    merged = [s for s in merged if s.length() > 1]
+
+    return invert_sequences(merged)

million/analyze/message_evaluation.py

+from math import floor
+import re
+from typing import Dict
+from million.model.message import Message
+
+memoization: Dict[Message, int] = {}
+
+# TODO WIP
+# - DNS to resolve audio, gif, pictures with counts
+def __compute__(msg: Message) -> int:
+    value = __computeContent(msg)
+
+    memoization[msg] = value
+    return value
+
+def __computeContent(msg: Message) -> int:
+    # TODO parse potential math expressions in content
+    match = re.search(r"\d+", msg.content)
+    
+    if match:
+        value = int(match[0])
+    else:
+        value = None
+    
+    return value
+
+def reset(msg: Message) -> None:
+    if msg in memoization:
+        memoization.pop(msg)
+
+def reset() -> None:
+    memoization.clear()
+
+def get(msg: Message) -> int:
+    """
+    Returns the estimated value counted in this message
+    """
+    return memoization.get(msg, __compute__(msg))

million/analyze/retain_counts.py

 
 def retain_counts(messages : List[Message])-> List[Message]:
     """
-    Retain only the messages that have a content and a sender_name
+    Retain only the messages that have a content
     """
     return [
         m for m in messages 
-        if m.content and 
-        m.sender_name and
+        if m.content and
         re.search('(\d{2,}|^\d$)', m.content)
         ]

million/model/fb_export.py

+from __future__ import annotations
 
-from typing import List
+from typing import Any, List, Set
 from pydantic import BaseModel
-from million.model.message import Message
 
+from million.model.message import Message
 from million.model.participant import Participant
 
+class Image(BaseModel):
+    creation_timestamp: int
+    uri: str
+
+    def __eq__(self, other: Image) -> bool:
+        return self.creation_timestamp == other.creation_timestamp \
+            and self.uri == other.uri
+
+class JoinableMode(BaseModel):
+    mode: int
+    link: str
 
 class FacebookExport(BaseModel):
     messages: List[Message]
-    participants: List[Participant]
+    participants: Set[Participant]
+    title: str
+    is_still_participant: bool
+    thread_path: str
+    magic_words: Set[Any]
+    image: Image
+    joinable_mode: JoinableMode
+
+    def merge(self, other: FacebookExport) -> None:
+        if self == other:
+            self.messages.extend(other.messages)
+            self.participants.update(other.participants)
+            self.magic_words.update(other.magic_words)
+
+    def sort(self) -> None:
+        self.messages.sort(key = lambda m: m.timestamp_ms)
+
+    # NOTE Toughen equality conditions ?
+    def __eq__(self, other: FacebookExport) -> bool:
+        return self.title == other.title \
+            and self.image == other.image

million/model/hole.py

-
-from pydantic import BaseModel
-
-from million.model.message import Message
-
-
-class Hole(BaseModel):
-    start: int
-    end: int
-    start_message: Message
-    end_message: Message

million/model/message.py

-
+from datetime import datetime
 from math import floor
-from typing import Optional
+from typing import Any, List
 from pydantic import BaseModel
 
+class Reaction(BaseModel):
+    reaction: str
+    actor: str
+
+class AudioFile(BaseModel):
+    uri: str
+    creation_timestamp: int
+
+class Video(BaseModel):
+    uri: str
+    creation_timestamp: int
+
+class Photo(BaseModel):
+    uri: str
+    creation_timestamp: int
+
+class Gif(BaseModel):
+    uri: str
+
+class Share(BaseModel):
+    link: str
+    share_text: str
+
+class Sticker(BaseModel):
+    uri: str
+    ai_stickers: List[Any]
 
 class Message(BaseModel):
     sender_name: str
     timestamp_ms: int
-    content: Optional[str] = None
-    is_geoblocked_for_viewer: Optional[bool] = None
-
-    def get_counted_value(self):
-        """
-        The content of the message should be (or contain) a number
-        """
-        value = None
-        # Remove any number that is not a digit
-        # TODO parse potential math expressions in content
-        cleaned_content = ''.join(
-            [c for c in self.content if c.isdigit() or c in ['.', ',']]).replace(',', '.')
-        try:
-            value = floor(float(cleaned_content))
-        except Exception as e:
-            raise ValueError(
-                f"Message {cleaned_content} does not contain a number ({e})")
-        return value
+    content: str | None = None
+    sticker: Sticker | None = None
+    share: Share | None = None
+    photos: List[Photo] | None = None
+    videos: List[Video] | None = None
+    gifs: List[Gif] | None = None
+    audio_files: List[AudioFile] | None = None
+    call_duration: int | None = None
+    reactions: List[Reaction] | None = None
+    is_unsent: bool | None = None
+    is_geoblocked_for_viewer: bool
+
+    def __str__(self) -> str:
+        dt = datetime.fromtimestamp(self.timestamp_ms / 1000)
+        dt_str = dt.strftime("%d/%m/%Y, %H:%M:%S")
+        return f"{self.sender_name}({dt_str}) : {self.content}"
+
+    def __hash__(self) -> int:
+        return hash(self.sender_name + str(self.timestamp_ms))

million/model/participant.py

 
 
 class Participant(BaseModel):
-    name: str
+    name: str
+
+    def __hash__(self):
+        return hash(self.name)

million/model/sequence.py

+from __future__ import annotations
 
-from typing import Optional
 from pydantic import BaseModel
+import pydantic
 
 from million.model.message import Message
+import million.analyze.message_evaluation as msg_val
 
 
 class Sequence(BaseModel):
-    start: int
     start_message: Message
-    end: int
-    end_message: Message
+    end_message: Message | None = None
+
+    @pydantic.validator('end_message', pre=True, always=True)
+    def default_end_message(cls, v, *, values):
+        return v or values['start_message'] 
+
+    def start(self) -> int:
+        return msg_val.get(self.start_message)
+    
+    def end(self) -> int:
+        return msg_val.get(self.end_message)
+    
+    def length(self) -> int:
+        return self.end() - self.start() + 1
+    
+    def merge(self, other: Sequence) -> None:
+        if other.start() < self.start(): self.start_message = other.start_message
+        if other.end() > self.end(): self.end_message = other.end_message
+    
+    def overlaps(self, other: Sequence) -> bool:
+        return self.start() <= other.end() + 1 and \
+            other.start() <= self.end() + 1

million/parse/fb_exports.py

 
-import json
-import os
-import re
+import os, re
 from typing import List
 
 from million.model.fb_export import FacebookExport
-from million.model.message import Message
-
-
-class FacebookExportParser:
-
-    def __init__(self):
-        pass
-
-    def parse(self, file_dir) -> FacebookExport:
-        files = [file_dir +
-                 f for f in os.listdir(file_dir) if f.endswith('.json')]
-        messages = []
-        participants = []
-        for file in files:
-            print(file)
-            with open(file, 'rb') as f:
-                json_data = self.__read_broken_fb_json(f.read())
-                messages += [Message(**m) for m in json_data['messages']]
-                participants += json_data['participants']
-
-        messages.sort(key=lambda m: m.timestamp_ms)
-        return FacebookExport(messages=messages, participants=participants)
-
-    def __read_broken_fb_json(self, binary_data):
-        repaired = re.sub(
-            rb'\\u00([\da-f]{2})',
-            lambda m: bytes.fromhex(m.group(1).decode()),
-            binary_data
-        )
-        return json.loads(repaired.decode('utf8'))
+
+
+def is_file_valid(file_name: str) -> bool:
+    """ 
+    Check if this file can be parsed into a FacebookExport
+    (Actually only check if its a json file atm)
+    """
+    # NOTE is there a way to peek inside a json file to
+    # check its internal structure ?
+    return os.path.splitext(file_name)[-1].lower() == '.json'
+
+def valid_dirfiles(file_dir: str) -> List[str]:
+    """ 
+    Returns a list of parsable files contained
+    in this directory
+    """
+    return [os.path.join(file_dir, file_name)
+            for file_name in os.listdir(file_dir)
+            if is_file_valid(file_name)]
+
+def parse_file(file_name: str) -> FacebookExport:
+    """ 
+    Parses a single parsable file into a FacebookExport Object
+    """
+    if not is_file_valid(file_name): return None
+
+    with open(file_name, 'rb') as f:
+        json_data = __read_broken_fb_json(f.read())
+        return FacebookExport.model_validate_json(json_data)
+
+def parse_dirfiles(file_dir: str) -> FacebookExport:
+    """ 
+    Parses every parsable files inside this directory
+    into a single FacebookExport Object
+    """
+    exports = [parse_file(f) for f in valid_dirfiles(file_dir)]
+    
+    result = exports[0]
+
+    for ex in exports[1:]: 
+        result.merge(ex)
+
+    result.sort()
+    return result
+
+def __read_broken_fb_json(binary_data):
+    # https://stackoverflow.com/questions/50008296/facebook-json-badly-encoded
+    repaired = re.sub(
+        rb'\\u00([\da-f]{2})',
+        lambda m: bytes.fromhex(m.group(1).decode()),
+        binary_data
+    )
+
+    return repaired.decode('utf8')

scripts/find_holes.py

 from datetime import datetime
 from million.analyze.find_holes import compute_sequences, find_holes
-from million.view.bar_chart import plot as bar_chart
-from million.analyze.count_participations import count_participations
 from million.analyze.retain_counts import retain_counts
-from million.parse.fb_exports import FacebookExportParser
+import million.parse.fb_exports as fb
 
 
 DATA_PATH = './data/'
 
-parser = FacebookExportParser()
-
-export = parser.parse(DATA_PATH)
+export = fb.parse_dirfiles(DATA_PATH)
 
 filtered = retain_counts(export.messages)
 
 sequences = compute_sequences(filtered)
 
-actual_counted = sum([s.end - s.start for s in sequences])
+actual_counted = sum([s.length() for s in sequences])
 
 print(f"Actual counted: {actual_counted}")
 
 print(len(holes))
 
 for hole in holes:
-    print(f"{hole.start} - {hole.end} ({hole.end - hole.start})")
+    print(f"{hole.start() + 1} -> {hole.end() - 1} ({hole.length() - 2})")
 
 
 # lets export a csv file of the holes and the people responsible for them
-with open('holes.csv', 'w') as f:
+with open('output/holes.csv', 'w') as f:
     f.write('début,fin,taille,responsable1,responsable2,date1,date2\n')
     for hole in holes:
         date_start = datetime.utcfromtimestamp(
         date_end = datetime.utcfromtimestamp(
             hole.end_message.timestamp_ms / 1000.0).strftime('%Y-%m-%d %H:%M:%S')
         f.write(
-            f"{hole.start},"
-            f"{hole.end},"
-            f"{hole.end - hole.start},"
+            f"{hole.start()},"
+            f"{hole.end()},"
+            f"{hole.length()},"
             f"{hole.start_message.sender_name},"
             f"{hole.end_message.sender_name},"
             f"{date_start},{date_end}\n"

scripts/read_top.py

 from million.view.bar_chart import plot as bar_chart
 from million.analyze.count_participations import count_participations
 from million.analyze.retain_counts import retain_counts
-from million.parse.fb_exports import FacebookExportParser
+import million.parse.fb_exports as fb
 
 
 DATA_PATH = './data/'
 
-parser = FacebookExportParser()
-
-export = parser.parse(DATA_PATH)
+export = fb.parse_dirfiles(DATA_PATH)
 
 filtered = retain_counts(export.messages)
 

test/TestCase.py

 
 class TestCase(unittest.TestCase):
     def _message_with_text(self, text: str):
-        return Message(content=text, sender_name="test", timestamp_ms=0)
+        return Message(content=text, sender_name="test", timestamp_ms=0, is_geoblocked_for_viewer=True)

test/model/message_test.py

 
 
 from million.model.message import Message
+import million.analyze.message_evaluation as msg_val
 from test.TestCase import TestCase
 
 
     def test_message_nominal(self, overrides=None, exclude=None):
         message = self._message_with_text("1")
 
-        assert 1 == message.get_counted_value()
+        assert 1 == msg_val.get(message)
 
     def test_message_with_text(self, overrides=None, exclude=None):
         message = self._message_with_text("1 text")
 
-        assert 1 == message.get_counted_value()
+        assert 1 == msg_val.get(message)
 
     def test_message_floored_dot(self, overrides=None, exclude=None):
         message = self._message_with_text("1.5")
 
-        assert 1 == message.get_counted_value()
+        assert 1 == msg_val.get(message)
 
     def test_message_floored_comma(self, overrides=None, exclude=None):
         message = self._message_with_text("1,5")
 
-        assert 1 == message.get_counted_value()
+        assert 1 == msg_val.get(message)