Merging pages together into an append log

Question

Situation

I'm scraping data from a paginated list on a website to be sent out as notifications. For the most part we'll only need to scape the first page and handle additions, deletions and duplicates.

When adding a page with items we've never seen before, we just store the page in memory.

local:  
scrape: A B C C D E

Output

A B C C D E

When adding pages with items we've seen before.

Additions:
All additions will be logged to an append only log. I also want notifications to be sent to me, and others. As such, I'm also returning the new data so the calling code can integrate with the other systems.
Deletions:
Whilst deletions are very rare, I'd like the code to be prepared here. Deletions just mean we assume the side with more data is correct. So, programmatically, are handled in a similar way to additions - we assume the side with more data is correct.
Duplicates: Duplicate entries between the log we have and the page we've freshly scraped are very common. We don't want duplicates to be added to the log, unless both side have a duplicate.

For example, say we have the following items in our log and in a fresh scrape:

local:  C C D E
scrape: A B C C E

We want to have the output to be:

A B C C D E

Notice how on the new scrape the data has changed. A and B were added, D was deleted and we normalize to 2, not 4, Cs and 1, not 2, Es. And how all the data is in the same order.

Additionally the list has a couple of additional properties:

New items are always prepended to the start of the list.
The code I've written can work well with both prepended and appended lists. By travelling from 'head' to 'tail' any additions to the data will mean we get a duplicate value on the next page.

For example. If the list has two pages, and after we scrape the first page a new item is added:
```
page 1  | page 2
F E D C | B A
```
We scrape FEDC, and a new item G is added to the list.
```
page 1  | page 2
G F E D | C B A
```
You can see we scrape C again. However if we went against the list, we could have missed out on C.
Items are highly unlikely to be deleted.
Deleting items causes the list to shift in the opposite direction. So we run into the issue we had before by travelling against the list.
```
page 1  | page 2
F E D C | B A
```
We scrape FEDC, and the item E is removed.
```
page 1  | page 2
F D C B | A
```
As you can see we can lose out on B.

Say we scrape the entire paginated list (not what the code does by default). Each page with no overlaps to any other pages will be stored as is in NotificationLog.values. Whilst we can assume each page carries on from the next, the code doesn't by default. Because additions or deletions to the paginated list can remove items from the list, depending on how we're travelling through the data. I decided erring on forcing the consumer to explicitly assume each page is connected - by adding an item on the previous page. As such, we can scrape the list in any direction without much of a care for how pages have been mutated. If we wait until an addition or deletion happens, then we can scape every page a second time ensuring we have any and all values between two pages. And so everything will connect together through the overlap between the pages.

However appends are very common, and deletes are very, very rare. So I've decided to (outside of the class) to lie and say I got one value of the previous page as part of the scrape. So we only need one parse of each page.

Code

I've written the following code to store the pages in memory and then log to a file. I think the code is really complicated, and could be cleaned up.

Note: The code is like 99.9% stable, but isn't 100% stable. In the test test_zip_values__same_tail, provided below, we can see the output is tuplize([1, 2, 4, 5, 6, 3, 7], suffix=(0,)) rather than a stable tuplize([1, 2, 3, 4, 5, 6, 7], suffix=(0,)). The situation is very, very rare so I've skipped implementing a secondary bi-stable merge for now. Please skip over.

from __future__ import annotations

import os
from typing import IO, Iterator, Optional, TypeVar

from .log import LogValues

TTuple = TypeVar("TTuple", bound=tuple)  # type: ignore
AValues = list[list[TTuple]]


class NotificationLog(LogValues[TTuple]):
    @staticmethod
    def _read_last_line(file: IO[bytes]) -> bytes:
        file.seek(-2, os.SEEK_END)
        while (
            file.read(1) != b"\n"
            and file.seek(-2, os.SEEK_CUR)
        ):
            pass
        return file.readline()

    @classmethod
    def _read_values(cls, bytes: IO[bytes], encoding: str = "utf-8") -> AValues[TTuple]:
        ret = cls._read_last_line(bytes).decode(encoding=encoding).rstrip("\n")
        if not ret:
            return []
        else:
            return [
                [
                    tuple(ret.split(","))  # type: ignore
                ]
            ]

    def add_values(self, values: list[TTuple], index: int = -1) -> Optional[list[TTuple]]:
        """
        Add values into the log.

        The values should be a page from a paginate list.
        Any values added to the log we will be returned,
        otherwise ``None`` will be returned.

        ..  note::

            For best performance the paginated list should:

            -   prepend new items to the start of the list, and
            -   have low volatility - few/no deletes.

            And you should add one value from the previous page in ``values``.
        """
        values = list(self.zip_values(iter(values), iter(self.values), index))
        had_values = bool(self.values)
        self.values = list(self._add_values(values, iter(self.values)))
        if values is not self.values[-1]:
            return None
        if had_values:
            values = values[-2::-1]
        for line in values:
            print(*line, sep=",", file=self.file)
        return values

    def _add_values(self, adds: list[TTuple], memories: Iterator[list[TTuple]], index: int = -1) -> Iterator[list[TTuple]]:
        """
        Add new memory into memories.

        ``adds`` will be run through :meth:`zip_values` beforehand.
        As such, all existing memories which overlapped with ``adds`` will be included in ``adds``.
        We need to; remove all memories which overlap ``add``,
        and insert ``adds`` in the correct position.
        """
        add = adds[0]
        try:
            mems = next(memories)
            while add[index] < mems[0][index]:
                yield mems
                mems = next(memories)
            while mems[0] in adds:
                mems = next(memories)
        except StopIteration:
            yield adds
        else:
            yield adds
            yield mems
            yield from memories

    @staticmethod
    def zip_values(adds: Iterator[TTuple], memories: Iterator[list[TTuple]], index: int = -1) -> Iterator[TTuple]:
        """
        Merge new values and pages together.

        Memories are descendingly ordered pages of values.
        The ``index``, defaulting to the last, is used to determine the order.
        The pages in memories are pages returned from a paginated list.
        Many sites use paginated lists, such as Google.
        Sometimes sites expose an 'infinate scroll',
        where the loads the next page automatically through JavaScript.

        Adds should be an ordered page to add to the log.
        If there are no overlaps between the ``adds`` and ``memories``
        then a new memory is added.
        Effectively adding the page as is into ``memories`` in the correct place.

        Once ``add`` reaches the final memory (the last value in the log),
        any additional data is ignored.
        The log should be opened in append mode and so we can only write to the end.
        :class:`NotificationLog` is not suitable for lists with random insertions.

        In prepended and low volatility lits, to merge pages into the log ASAP;
        always include one value from the previous page.
        The code is designed to correctly handle deleted data,
        from both ``add`` and ``memories``.
        As such the previous value can be the first, second, ..., or last value.

        ..  warn::

            If:
            
            -   new items to the list are **appended** to the **end** of the list, or
            -   if the list is highly susceptible to item deletion.

            **Do not** include a value from the previous page.
            Doing so can cause _data loss_ and will require an additional tool to recover from.

        """
        def take_same_index(start: TTuple, it: Iterator[TTuple]) -> tuple[Optional[TTuple], list[TTuple]]:
            values = [start]
            for value in it:
                if value[index] != start[index]:
                    return value, values
                else:
                    values.append(value)
            return None, values

        add: Optional[TTuple]
        mem: Optional[TTuple]
        try:
            add = next(adds)
        except StopIteration:
            return
        while True:
            try:
                mems_ = next(memories)
            except StopIteration:
                yield add
                yield from adds
                return
            if mems_[-1][index] <= add[index]:
                break
        mems = iter(mems_)
        while True:
            try:
                mem = next(mems)
                while mem[index] < add[index]:
                    yield add
                    add = next(adds)
            except StopIteration:
                return
            try:
                while True:
                    if mem[index] == add[index]:
                        if mem == add:
                            yield add
                            try:
                                add = next(adds)
                            except StopIteration:
                                add = None
                                raise
                            mem = next(mems)
                        else:
                            add, _adds = take_same_index(add, adds)
                            mem, _mems = take_same_index(mem, mems)
                            # TODO: Replace with a bi-stable algorithm
                            # Probably never going to need a bi-stable algorithm
                            yield from _mems
                            for _add in _adds:
                                if _add not in _mems:
                                    yield _add
                            if mem is None:
                                return
                            if add is None:
                                yield mem
                                raise StopIteration()
                    elif mem[index] < add[index]:
                        yield add
                        try:
                            add = next(adds)
                        except StopIteration:
                            add = None
                            yield mem
                            raise
                    else:
                        yield mem
                        mem = next(mems)
            except StopIteration:
                yield from mems
                if add is None:
                    return
            try:
                mems = iter(next(memories))
            except StopIteration:
                return

Tests

I have a fairly comprehensive test suite.

import io
from typing import Any

import pytest

from loglet import NotificationLog

VALUES = [
    ("1", "081041"),
    ("5", "081324"),
    ("5", "081327"),
]
STRING = "".join(
    ",".join(values) + "\n"
    for values in VALUES
)
BYTES = STRING.encode('utf-8')


@pytest.fixture
def log():
    return NotificationLog.from_streams(
        io.StringIO(STRING),
        io.BytesIO(BYTES),
    )


@pytest.fixture
def log_empty():
    return NotificationLog(
        io.StringIO(""),
        [],
    )


def test_read_last_line__empty_log():
    output = NotificationLog._read_last_line(io.BytesIO(b""))
    assert output == b""


def test_read_last_line__one_no_newline():
    output = NotificationLog._read_last_line(io.BytesIO(b"foo"))
    assert output == b"foo"


def test_read_last_line__one_log():
    output = NotificationLog._read_last_line(io.BytesIO(b"foo\n"))
    assert output == b"foo\n"


def test_read_last_line__two_log():
    output = NotificationLog._read_last_line(io.BytesIO(b"foo\nbar\n"))
    assert output == b"bar\n"


def test_read_values__empty_log():
    output = NotificationLog._read_values(io.BytesIO(b""))
    assert output == []


def test_read_values__empty_newline_log():
    output = NotificationLog._read_values(io.BytesIO(b"\n"))
    assert output == []


def test_read_values__one_log():
    output = NotificationLog._read_values(io.BytesIO(b"foo\n"))
    assert output == [[("foo",)]]


def test_read_values__two_log():
    output = NotificationLog._read_values(io.BytesIO(b"foo\nbar\n"))
    assert output == [[("bar",)]]


def tuplize(values, *, prefix: tuple[Any, ...] = (), suffix: tuple[Any, ...] = ()) -> list[tuple[Any, ...]]:
    try:
        return [
            prefix
            + (
                value
                if isinstance(value, tuple) else
                (value,)
            )
            + suffix
            for value in values
        ]
    except TypeError:
        return [prefix + (values,)]


def test_zip_values__empty_adds():
    output = list(NotificationLog.zip_values(
        iter(tuplize([])),
        iter([
            tuplize(3),
        ]),
    ))
    assert output == tuplize([])


def test_zip_values__empty_mems():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([]),
    ))
    assert output == tuplize(range(9, -1, -2))


def test_zip_values__basic_final_extra_add_eq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(3),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 3])


def test_zip_values__basic_final_extra_add_neq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(4),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 4])


def test_zip_values__basic_final_extra_mems_eq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, 4, -2))),
        iter([
            tuplize([5, 3, 1]),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 3, 1])


def test_zip_values__basic_final_extra_mems_neq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, 4, -2))),
        iter([
            tuplize([6, 4, 2, 0]),
        ]),
    ))
    assert output == tuplize([9, 7, 6, 5, 4, 2, 0])


def test_zip_values__basic_stem_extra_add_eq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(3),
            tuplize(-10),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 3, 1])


def test_zip_values__basic_stem_extra_add_neq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(4),
            tuplize(-10),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 4, 3, 1])


def test_zip_values__basic_stem_extra_mems_eq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, 4, -2))),
        iter([
            tuplize([5, 3, 1]),
            tuplize(-10),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 3, 1])


def test_zip_values__basic_stem_extra_mems_neq():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, 4, -2))),
        iter([
            tuplize([6, 4, 2, 0]),
            tuplize(-10),
        ]),
    ))
    assert output == tuplize([9, 7, 6, 5, 4, 2, 0])


def test_zip_values__basic_multiple_equal():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(7),
            tuplize(5),
            tuplize(3),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 3])


def test_zip_values__basic_multiple_unequal():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(8),
            tuplize(6),
            tuplize(4),
        ]),
    ))
    assert output == tuplize([9, 8, 7, 6, 5, 4])


def test_zip_values__basic_multiple_chunk_aligned():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize([7, 6, 5]),
            tuplize(4),
        ]),
    ))
    assert output == tuplize([9, 7, 6, 5, 4])


def test_zip_values__basic_multiple_chunk_misaligned():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize([8, 7, 6]),
            tuplize(4),
        ]),
    ))
    assert output == tuplize([9, 8, 7, 6, 5, 4])


def test_zip_values__basic_multiple_extra():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(-10),
            tuplize(-11),
        ]),
    ))
    assert output == tuplize(range(9, -1, -2))


def test_zip_values__basic_head_overlap():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize([10, 9, 8]),
            tuplize(-11),
        ]),
    ))
    assert output == tuplize([10, 9, 8, 7, 5, 3, 1])


def test_zip_values__basic_head_skip():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(20),
            tuplize([10, 9, 8]),
            tuplize(-11),
        ]),
    ))
    assert output == tuplize([10, 9, 8, 7, 5, 3, 1])


def test_zip_values__basic_head_skip():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(-11),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 3, 1])


def test_zip_values__basic_head_skip():
    output = list(NotificationLog.zip_values(
        iter(tuplize(range(9, -1, -2))),
        iter([
            tuplize(-11),
        ]),
    ))
    assert output == tuplize([9, 7, 5, 3, 1])


def test_zip_values__same_tail():
    output = list(NotificationLog.zip_values(
        iter(
            tuplize([1, 3, 4, 5, 7], suffix=(0,)),
        ),
        iter([
            tuplize([1, 2, 4, 5, 6], suffix=(0,)),
        ]),
    ))
    assert output == tuplize([1, 2, 4, 5, 6, 3, 7], suffix=(0,))


def test_zip_values__same_tail_add():
    output = list(NotificationLog.zip_values(
        iter(
            tuplize([1, 3, 4, 5, 7], suffix=(1,)),
        ),
        iter([
            tuplize([1, 2, 4, 5, 6], suffix=(1,))
            + tuplize([0], suffix=(0,)),
        ]),
    ))
    assert output == (
        tuplize([1, 2, 4, 5, 6, 3, 7], suffix=(1,))
        + tuplize([0], suffix=(0,))
    )


def test_zip_values__same_mid():
    output = list(NotificationLog.zip_values(
        iter(
            tuplize([1, 3, 4, 5, 7], suffix=(1,))
            + tuplize([0], suffix=(0,)),
        ),
        iter([
            tuplize([1, 2, 4, 5, 6], suffix=(1,))
            + tuplize([0], suffix=(0,)),
        ]),
    ))
    assert output == (
        tuplize([1, 2, 4, 5, 6, 3, 7], suffix=(1,))
        + tuplize([0], suffix=(0,))
    )


def test_zip_values__same_mid():
    output = list(NotificationLog.zip_values(
        iter(
            tuplize([1, 3, 4, 5, 7], suffix=(1,))
            + tuplize([0], suffix=(0,)),
        ),
        iter([
            tuplize([1, 2, 4, 5, 6], suffix=(1,))
            + tuplize([1], suffix=(0,)),
        ]),
    ))
    assert output == (
        tuplize([1, 2, 4, 5, 6, 3, 7], suffix=(1,))
        + tuplize([1, 0], suffix=(0,))
    )


def test__add_values__head_mem(log: NotificationLog):
    adds = tuplize([3, 2, 1])
    mems = list(log._add_values(
        adds,
        iter([
            tuplize([9, 8, 7]),
            tuplize([6, 5, 4]),
        ]),
    ))
    assert (
        mems
        == [
            tuplize([9, 8, 7]),
            tuplize([6, 5, 4]),
            tuplize([3, 2, 1]),
        ]
    )
    assert mems[-1] is adds


def test__add_values__overwrite_mem(log: NotificationLog):
    adds = tuplize([3, 2, 1])
    mems = list(log._add_values(adds, iter([adds])))
    assert mems == [adds]
    assert mems[-1] is adds


def test__add_values__overwrite_mems(log: NotificationLog):
    adds = tuplize([6, 5, 4])
    mems = list(log._add_values(
        adds,
        iter([
            tuplize([9, 8, 7]),
            tuplize([6, 5, 4]),
            tuplize([3, 2, 1]),
        ]),
    ))
    assert (
        mems
        == [
            tuplize([9, 8, 7]),
            tuplize([6, 5, 4]),
            tuplize([3, 2, 1]),
        ]
    )


def test__add_values__tail_mem(log: NotificationLog):
    adds = tuplize([9, 8, 7])
    mems = list(log._add_values(
        adds,
        iter([
            tuplize([6, 5, 4]),
            tuplize([3, 2, 1]),
        ]),
    ))
    assert (
        mems
        == [
            tuplize([9, 8, 7]),
            tuplize([6, 5, 4]),
            tuplize([3, 2, 1]),
        ]
    )


HEAD_BUG_VALUES = [
    ("2", "181243"),
    ("2", "181241"),
    ("5", "181215"),
    ("5", "181201"),
    ("4", "181141"),
    ("4", "181141"),
    ("1", "180951"),
    ("1", "180948"),
    ("5", "172330"),
    ("7", "171721"),
    ("5", "171437"),
    ("5", "171211"),
    ("4", "170758"),
    ("4", "170757"),
    ("5", "162042"),
    ("6", "161510"),
    ("6", "161509"),
    ("6", "161507"),
    ("6", "161506"),
    ("6", "161505"),
    ("6", "161505"),
    ("5", "161347"),
    ("5", "161346"),
    ("4", "161155"),
    ("4", "161155"),
    ("1", "160744"),
    ("5", "151239"),
    ("7", "150418"),
    ("4", "141945"),
    ("4", "141944"),
    ("5", "141632"),
    ("9", "141532"),
    ("4", "141219"),
    ("5", "141149"),
    ("8", "141115"),
    ("8", "141114"),
    ("8", "141114"),
    ("8", "141113"),
    ("8", "141113"),
    ("8", "141113"),
    ("8", "141112"),
    ("8", "141112"),
    ("5", "132307"),
    ("5", "132303"),
    ("3", "132039"),
    ("3", "132039"),
    ("3", "132039"),
    ("4", "131502"),
    ("4", "131501"),
]


def test_add_values__head_bug(log: NotificationLog):
    assert log.add_values(HEAD_BUG_VALUES) is None


_CONNECTED_VALUES = VALUES + [
    ("5", "082047"),
    ("5", "082312"),
]


def test_add_values__connected_values(log: NotificationLog):
    assert (
        log.add_values(_CONNECTED_VALUES)
        == [
            ('5', '082047'),
            ('5', '081327'),
            ('5', '081324'),
            ('1', '081041'),
        ]
    )


def test_zip_values__head_bug(log: NotificationLog):
    assert list(log.zip_values(iter(HEAD_BUG_VALUES), iter([VALUES]))) == HEAD_BUG_VALUES

If a duplicate only gets into the log if the log has a duplicate, how did that duplicate get into the log in the first place? Also in your first example you state "Notice how A and B were added, D was deleted". However D was not deleted from you expected output. — JonSG, Jun 15, 2021 at 13:27
@JonSG Thank you, some very good questions. I have updated the question. When we get a page we've never seen before we store the page as is in memory for later. Later when we scrape a page with duplicate data we have to handle additions, deletes and duplicates. Sorry, D was deleted from the list (not our code) in-between the first and second scrape. however I want all the data, even if the data has later been removed, so D is included in the output. If anything is still unclear please let me know :) — Phroggie, Jun 15, 2021 at 15:56

Reinderien · Accepted Answer · 2021-06-14 20:16:50Z

_read_last_line is clever but is going to be slow. You're better off reading in buffer chunks of say 1k and doing string operations like .rindex. This will almost certainly contain a newline on the first try for most kinds of log files. Once you have found the newline, you would not need to issue another readline(); simply slice your buffer string.

Your type hinting puts a lot of effort in, but is dubious for a few reasons. You sprinkle # type: ignore all over, which suggests that something (mypy?) doesn't want to swallow your hints. And the hints themselves are weird - _read_values returns a list of lists of tuples of strings, but your hint is for a list of lists of tuples of a generic-parameter type. Why, if they'll always be strings?

Can

        mems = next(memories)
        while add[index] < mems[0][index]:
            yield mems
            mems = next(memories)

be

for mems in memories:
    if add[index] >= mems[0][index]:
        break
    yield mems

? Avoiding manual next is typically preferable. This will avoid hitting a StopIteration error. You may want to add a for/else to catch a missing-break.

Your answer doesn't seem too helpful. 1. Yes, possibly faster, but your suggestion is prone to breaking and fixing the breaking would requires more LOC. 2. Two type: ignores is not "all over". 3. The type won't always be strings, look at the tests - the index will always be comparable with the read string from the log. 4. Are you sure next is preferable here, your probably going to make the code have a performance hit for no readability improvement - or possibly a hit to readability. — Phroggie, Jun 14, 2021 at 20:39

Stack Exchange Network

Merging pages together into an append log

Situation

Code

Tests

1 Answer 1

Your Answer

Not the answer you're looking for? Browse other questions tagged
python
python-3.x
queue
or ask your own question.

Hot Network Questions

Merging pages together into an append log

Situation

Code

Tests

1 Answer 1

Your Answer

Sign up or log in

Post as a guest

Not the answer you're looking for? Browse other questions tagged pythonpython-3.xqueue or ask your own question.

Related

Hot Network Questions

Not the answer you're looking for? Browse other questions tagged
python
python-3.x
queue
or ask your own question.