Auto-parsing timestamp_ms in JSON Message

from int to datetime object

million/model/fb_export.py

             self.magic_words.update(other.magic_words)
 
     def sort(self) -> None:
-        self.messages.sort(key = lambda m: m.timestamp_ms)
+        self.messages.sort(key = lambda m: m.date_time)
 
 
     def __eq__(self, other: FacebookExport) -> bool:

million/model/message.py

 from datetime import datetime
 from typing import Any, List
 from uuid import uuid4
-from pydantic import BaseModel, PrivateAttr, computed_field
+from pydantic import BaseModel, Field, PrivateAttr, computed_field, validator
+
 
 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
+    date_time: datetime = Field(alias="timestamp_ms")
     content: str | None = None
     sticker: Sticker | None = None
     share: Share | None = None
 
     _id: str = PrivateAttr(default_factory=lambda: str(uuid4()))
 
-
     def __str__(self) -> str:
-        dt = datetime.fromtimestamp(self.timestamp_ms / 1000)
-        dt_str = dt.strftime("%d/%m/%Y, %H:%M:%S")
+        dt_str = self.date_time.strftime("%d/%m/%Y, %H:%M:%S")
         return f"{self.sender_name}({dt_str}) : {self.content}"
 
     def __hash__(self) -> int:
         return hash(self.item_id)
-    
+
     @computed_field
     @property
     def item_id(self) -> str:
-        return self._id
+        return self._id
+
+    @validator("date_time", pre=True, always=True)
+    def parse_timestamp(cls, v):
+        if isinstance(v, int):
+            return datetime.fromtimestamp(v / 1000)
+        return v

scripts/find_gromots.py

 from datetime import datetime
-from million.analyze.word_finder import findWords
+from million.analyze.word_finder import find_words
 import million.parse.fb_exports as fb
 
 
     ]
 
 export = fb.parse_dirfiles(DATA_PATH)
-msg_gros_mots = findWords(export.messages, gros_mots)
+msg_gros_mots = find_words(export.messages, gros_mots)
 
 msg_gros_mots_grp = {}
 
     print(name)
 
     for msg in msg_gros_mots_grp[name]:
-        time = datetime.fromtimestamp(msg.timestamp_ms / 1000)
-        time_str = time.strftime("%d/%m/%Y %H:%M:%S")
+        time_str = msg.date_time.strftime("%d/%m/%Y %H:%M:%S")
         print(f"\t{time_str} : {msg.content}")

scripts/find_holes.py

 import million.parse.fb_exports as fb
 
 
-DATA_PATH = './data/'
+DATA_PATH = "./data/"
 
 export = fb.parse_dirfiles(DATA_PATH)
 
 
 
 # lets export a csv file of the holes and the people responsible for them
-with open('output/holes.csv', 'w') as f:
-    f.write('début,fin,taille,responsable1,responsable2,date1,date2\n')
+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(
-            hole.start_message.timestamp_ms / 1000.0).strftime('%Y-%m-%d %H:%M:%S')
-        date_end = datetime.utcfromtimestamp(
-            hole.end_message.timestamp_ms / 1000.0).strftime('%Y-%m-%d %H:%M:%S')
+        date_start = hole.start_message.date_time.strftime("%Y-%m-%d %H:%M:%S")
+        date_end = hole.end_message.date_time.strftime("%Y-%m-%d %H:%M:%S")
         f.write(
             f"{hole.start()},"
             f"{hole.end()},"