my.instagram.android: use hpikit, kloggy and koncurrent

pyproject.toml

@@ -458,6 +458,7 @@ commands = [
         "my.github.ghexport",
         "my.hackernews.harmonic",
         "my.hackernews.materialistic",
+        "my.instagram.android",
         "my.reddit.rexport",
         "my.twitter.android",
         "my.twitter.archive",
@@ -516,6 +517,7 @@ hpi_modules = [
     "my.hackernews.harmonic",
     "my.hackernews.materialistic",
     "my.hypothesis",
+    "my.instagram.android",
     "my.instapaper",
     "my.ip.all",
     "my.kobo",

src/my/instagram/android.py

@@ -1,33 +1,34 @@
 """
-Bumble data from Android app database (in =/data/data/com.instagram.android/databases/direct.db=)
+Instagram data from Android app database (in =/data/data/com.instagram.android/databases/direct.db=)
 """
 
 from __future__ import annotations
 
+REQUIRES = ['hpikit', 'kloggy', 'koncurrent']
+
 import sqlite3
 from collections.abc import Iterator, Sequence
 from dataclasses import dataclass
 from datetime import datetime
+from functools import partial
 from pathlib import Path
 from typing import assert_never
 
-from my.core import (
-    Json,
-    Paths,
-    Res,
-    datetime_naive,
-    get_files,
-    make_config,
-    make_logger,
-)
+import koncurrent
+from hpikit.datetime import NaiveDateTime, naive
+from hpikit.json import Json, loads
+from hpikit.paths import Paths, get_files
+from kloggy import get_logger
+from kloggy.progress import progress
+
+from my.core import Res, make_config
 from my.core.cachew import mcachew
 from my.core.common import unique_everseen
-from my.core.json import json_loads
-from my.core.sqlite import select, sqlite_connect_immutable
+from my.core.sqlite import select, sqlite_connection
 
 from my.config import instagram as user_config  # isort: skip
 
-logger = make_logger(__name__)
+logger = get_logger(__name__)
 
 
 @dataclass
@@ -60,7 +61,7 @@ class User:
 class _BaseMessage:
     id: str
     # NOTE: ffs, looks like they keep naive timestamps in the db (checked some random messages)
-    created: datetime_naive
+    created: NaiveDateTime
     text: str
     thread_id: str
 
@@ -100,7 +101,7 @@ def __eq__(self, other) -> bool:
         return self.rest == other.rest
 
 
-def _parse_message(j: Json, tid_map: dict[str, str]) -> _Message | None:
+def _parse_message(j: Json, *, tid_map: dict[str, str]) -> _Message | None:
     id = j['item_id']  # noqa: A001
     t = j['item_type']
 
@@ -122,7 +123,7 @@ def _parse_message(j: Json, tid_map: dict[str, str]) -> _Message | None:
         return None
 
     uid = j['user_id']
-    created: datetime_naive = datetime.fromtimestamp(int(j['timestamp']) / 1_000_000)
+    created = naive(datetime.fromtimestamp(int(j['timestamp']) / 1_000_000))
     text: str | None = None
     if t == 'text':
         text = j['text']
@@ -149,22 +150,22 @@ def _parse_message(j: Json, tid_map: dict[str, str]) -> _Message | None:
     )
 
 
-def _process_db(db: sqlite3.Connection) -> Iterator[Res[User | _Message]]:
+def _process_db(db: sqlite3.Connection, *, full_name: str | None) -> Iterator[Res[User | _Message]]:
     # TODO ugh. seems like no way to extract username?
     # sometimes messages (e.g. media_share) contain it in message field
     # but generally it's not present. ugh
     for (self_uid,) in select(('user_id',), 'FROM session', db=db):
         yield User(
             id=str(self_uid),
-            full_name=config.full_name or 'USERS_OWN_FULL_NAME',
-            username=config.full_name or 'USERS_OWN_USERNAME',
+            full_name='USERS_OWN_FULL_NAME' if full_name is None or full_name == '' else full_name,
+            username='USERS_OWN_USERNAME' if full_name is None or full_name == '' else full_name,
         )
 
     # maps local tid to "server tid" (thread_v2_id)
     tid_map: dict[str, str] = {}
 
     for (thread_json,) in select(('thread_info',), 'FROM threads', db=db):
-        j = json_loads(thread_json)
+        j = loads(thread_json)
         thread_v2_id = j.get('thread_v2_id')
         if thread_v2_id is not None:
             # sometimes not present...
@@ -189,7 +190,7 @@ def _process_db(db: sqlite3.Connection) -> Iterator[Res[User | _Message]]:
 
     for (msg_json,) in select(('message',), 'FROM messages ORDER BY timestamp', db=db):
         # eh, seems to contain everything in json?
-        j = json_loads(msg_json)
+        j = loads(msg_json)
         try:
             m = _parse_message(j, tid_map=tid_map)
             if m is not None:
@@ -199,25 +200,33 @@ def _process_db(db: sqlite3.Connection) -> Iterator[Res[User | _Message]]:
             yield e
 
 
+def _process_path(path: Path, *, full_name: str | None) -> list[Res[User | _Message]]:
+    result: list[Res[User | _Message]] = []
+    with sqlite_connection(path, immutable=True) as db:
+        try:
+            # Keep entities already parsed if a later row or table fails.
+            result.extend(_process_db(db=db, full_name=full_name))
+        except Exception as e:
+            result.append(e)
+    return result
+
+
 def _entities() -> Iterator[Res[User | _Message]]:
     # NOTE: definitely need to merge multiple, app seems to recycle old messages
     # TODO: hmm hard to guarantee timestamp ordering when we use synthetic input data...
     # todo use TypedDict?
     paths = inputs()
-    total = len(paths)
-    width = len(str(total))
-    for idx, path in enumerate(paths):
-        logger.info(f'processing [{idx:>{width}}/{total:>{width}}] {path}')
-        with sqlite_connect_immutable(path) as db:
-            try:
-                for m in _process_db(db=db):
-                    if isinstance(m, Exception):
-                        m.add_note(f'^ while processing {path}')
-                    yield m
-            except Exception as e:
-                e.add_note(f'^ while processing {path}')
-                yield e
-                # todo use error policy here
+    parse = partial(_process_path, full_name=config.full_name)
+    with koncurrent.map_results(parse, paths) as results:
+        files = progress(paths, logger=logger, unit='files')
+        for path, result in zip(files, results, strict=True):
+            if isinstance(result, Exception):
+                result.add_note(f'^ while processing {path}')
+                raise result
+            for m in result:
+                if isinstance(m, Exception):
+                    m.add_note(f'^ while processing {path}')
+                yield m
 
 
 @mcachew(depends_on=inputs)

src/my/instagram/common.py

@@ -14,7 +14,9 @@ class User(Protocol):
 
 
 class Message(Protocol):
-    created: datetime
+    @property
+    def created(self) -> datetime: ...
+
     text: str
     thread_id: str
 

src/my/tests/instagram/android.py

@@ -0,0 +1,169 @@
+import json
+from collections.abc import Iterator, Sequence
+from datetime import datetime
+from pathlib import Path
+from sqlite3 import OperationalError
+from typing import assert_type
+
+import pytest
+from hpikit.datetime import NaiveDateTime
+from hpikit.json import Json
+
+from my.core.cfg import tmp_config
+from my.core.sqlite import sqlite_connection
+
+
+@pytest.fixture(autouse=True)
+def _config(tmp_path: Path) -> Iterator[None]:
+    class user_config:
+        class instagram:
+            class android:
+                export_path = tmp_path
+                username = 'owner'
+                full_name = 'Owner Name'
+
+    with tmp_config(modules='my.instagram.android', config=user_config):
+        yield
+
+
+def _make_db(path: Path, *, threads: Sequence[Json], messages: Sequence[Json]) -> None:
+    with sqlite_connection(path) as db:
+        db.executescript('''
+            CREATE TABLE session (user_id INTEGER);
+            INSERT INTO session VALUES (1);
+            CREATE TABLE threads (thread_info TEXT);
+            CREATE TABLE messages (message TEXT, timestamp INTEGER);
+        ''')
+        db.executemany('INSERT INTO threads VALUES (?)', [(json.dumps(thread),) for thread in threads])
+        db.executemany(
+            'INSERT INTO messages VALUES (?, ?)',
+            [(json.dumps(message), int(message['timestamp'])) for message in messages],
+        )
+
+
+def _message(*, id_: str, timestamp: int, item_type: str = 'text', **payload: object) -> Json:
+    return {
+        'item_id': id_,
+        'item_type': item_type,
+        'timestamp': str(timestamp),
+        'thread_key': {'thread_id': 'local'},
+        'user_id': '2',
+        'text': 'A message',
+        **payload,
+    }
+
+
+def test_messages_merge_snapshots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    from my.instagram import android
+
+    first = tmp_path / 'z.db'
+    second = tmp_path / 'a.db'
+    thread = {
+        'thread_id': 'local',
+        'thread_v2_id': 'server',
+        'recipients': [{'id': 2, 'username': 'alice', 'full_name': 'Alice'}],
+    }
+    one = _message(id_='1', timestamp=1_000_123, text='Hello')
+    _make_db(
+        first,
+        threads=[thread],
+        messages=[
+            _message(
+                id_='2', timestamp=2_000_000, item_type='action_log', action_log={'description': 'Liked a message'}
+            ),
+            one,
+            _message(id_='missing-thread', timestamp=3_000_000, thread_key={'thread_id': 'unknown'}),
+        ],
+    )
+    updated_thread = {
+        **thread,
+        'recipients': [{'pk_id': 2, 'username': 'alice_new', 'full_name': 'Alice Updated'}],
+    }
+    _make_db(
+        second,
+        threads=[updated_thread],
+        messages=[
+            one,
+            _message(id_='3', timestamp=3_000_000, item_type='reel_share', reel_share={'text': 'Shared a reel'}),
+            _message(id_='4', timestamp=4_000_000, user_id='1', text='My reply'),
+        ],
+    )
+    monkeypatch.setattr(android.config, 'export_path', [first, second])
+
+    messages = []
+    for item in android.messages():
+        assert isinstance(item, android.Message), item
+        assert_type(item.created, NaiveDateTime)
+        assert item.created.tzinfo is None
+        assert item.thread_id == 'server'
+        messages.append(item)
+    assert [message.id for message in messages] == ['1', '2', '3', '4']
+    assert [message.text for message in messages] == ['Hello', 'Liked a message', 'Shared a reel', 'My reply']
+    assert [message.user.username for message in messages[:3]] == ['alice', 'alice', 'alice_new']
+    assert messages[2].user.full_name == 'Alice Updated'
+    assert messages[3].user.full_name == 'Owner Name'
+    assert messages[0].created == datetime.fromtimestamp(1.000123)
+
+
+def test_messages_keep_partial_results_and_deduplicate_errors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    from my.instagram import android
+
+    first = tmp_path / 'first.db'
+    second = tmp_path / 'second.db'
+    thread = {
+        'thread_id': 'local',
+        'thread_v2_id': 'server',
+        'recipients': [{'id': 2, 'username': 'alice', 'full_name': 'Alice'}],
+    }
+    _make_db(
+        first,
+        threads=[thread],
+        messages=[
+            _message(id_='1', timestamp=1_000_000),
+            _message(id_='photo-1', timestamp=2_000_000, item_type='photo'),
+            _message(id_='photo-2', timestamp=3_000_000, item_type='photo'),
+        ],
+    )
+    with sqlite_connection(first) as db:
+        db.execute('INSERT INTO messages VALUES (?, ?)', ('{', 4_000_000))
+    # The later database needs user information emitted before the first database's JSON error.
+    _make_db(
+        second,
+        threads=[{**thread, 'recipients': []}],
+        messages=[
+            _message(id_='2', timestamp=5_000_000),
+            _message(id_='photo-3', timestamp=6_000_000, item_type='photo'),
+        ],
+    )
+    monkeypatch.setattr(android.config, 'export_path', [first, second])
+
+    one, unsupported, malformed, two = list(android.messages())
+    assert isinstance(one, android.Message)
+    assert isinstance(two, android.Message)
+    assert [one.id, two.id] == ['1', '2']
+    assert one.user == two.user
+    assert isinstance(unsupported, android.MessageError)
+    assert unsupported.args == ('photo-1', "photo isn't handled yet")
+    assert isinstance(malformed, ValueError)
+    for error in (unsupported, malformed):
+        assert f'^ while processing {first}' in error.__notes__
+    assert any(note.startswith('^ while parsing ') for note in unsupported.__notes__)
+
+
+def test_messages_continue_after_database_error(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    from my.instagram import android
+
+    broken = tmp_path / 'broken.db'
+    later = tmp_path / 'later.db'
+    with sqlite_connection(broken) as db:
+        db.execute('CREATE TABLE unrelated (id INTEGER)')
+    thread = {'thread_id': 'local', 'thread_v2_id': 'server', 'recipients': []}
+    _make_db(later, threads=[thread], messages=[_message(id_='1', timestamp=1_000_000, user_id='1')])
+    monkeypatch.setattr(android.config, 'export_path', [broken, later])
+
+    error, message = list(android.messages())
+    assert isinstance(error, OperationalError)
+    assert 'no such table: session' in str(error)
+    assert f'^ while processing {broken}' in error.__notes__
+    assert isinstance(message, android.Message)
+    assert message.id == '1'