Start working on record analysis

venv/

__pycache__

instance/

 No newline at end of file

"""

Kowalski is the package responsible for view count analysis.

"""

from dataclasses import dataclass

from datetime import datetime

from typing import Literal, Self, Any

type EntityId = int

type EntityType = Literal["post", "page", "attachment"]

@dataclass(eq=True, frozen=True)

class Record:

    """Record of the views of a given url at a given timestamp.

    It also stores a bit more details about the entity at the given URL (namely id, title and type).

    """

    url: str

    view_count: int

    timestamp: datetime

    entity_id: EntityId

    entity_title: str

    entity_type: EntityType

    @classmethod

    def from_dict_and_timestamp(cls, d: dict[str, Any], timestamp: datetime) -> Self:

        return cls(

            url=d["url"],

            view_count=d["view_cnt"],

            timestamp=timestamp,

            entity_id=d["ID"],

            entity_title=d["post_title"],

            entity_type=d["post_type"],

        )

from dataclasses import dataclass

from datetime import datetime

from functools import reduce

from pathlib import Path

from typing import Self

import json

from kowalski.record import Record

@dataclass(eq=True, frozen=True)

class RecordCollection:

    records: tuple[Record, ...]

    def __post_init__(self) -> None:

        # Validate the records are actually compatible with each other.

        # First, we calculate all the unique (timestamp, id) pairs from the records.

        unique_timestamps_and_ids = set((r.timestamp, r.entity_id) for r in self.records)

        # Then, we calculate all the unique records.

        unique_records = set(self.records)

        if len(unique_timestamps_and_ids) != len(unique_records):

            # Because the unique timestamps and ids are generated from the records, this means

            # that in this case, there must be fewer unique timestamps and ids than unique records.

            assert len(unique_timestamps_and_ids) < len(unique_records)

            # This is a problem, because it means we have at least two records for the same (timestamp, id)

            # with different other fields. That is an inconsistency and can't be.

            raise ValueError(

                "Collections contained at least two incompatible records (same id, same timestamp, but different other fields)."

            )

    def merge(self, other: Self) -> Self:

        """Merge two record collections, creating a new one, containing both records."""

        all_records = self.records + other.records

        return self.__class__(all_records)

    @classmethod

    def from_file(cls, json_path: Path) -> Self:

        timestamp_iso_str = json_path.stem.replace("_", ":")

        timestamp = datetime.fromisoformat(timestamp_iso_str)

        with open(json_path, "r") as f:

            record_dicts = json.load(f)

        return cls(

            tuple(Record.from_dict_and_timestamp(d, timestamp) for d in record_dicts)

        )

    @classmethod

    def from_dir(cls, dir_path: Path) -> Self:

        collections = (

            cls.from_file(json_path)

            for json_path in dir_path.glob("*.json")

        )

        return reduce(lambda one, other: one.merge(other), collections)