from __future__ import annotations
import calendar
from datetime import datetime
from typing import Any, Iterable, Mapping
import tls_client.response
from ..models import Inbox, Message
from ._base import ResourceBase
__all__ = ['MessagesResource']
[docs]
class MessagesResource(ResourceBase):
"""Private messages: inbox, conversations, compose, send, delete.
Maps to ``/private.php`` and (the alias) ``/messages``. Paths used:
- ``/private.php`` — inbox listing.
- ``/private.php?action=send`` — compose form.
- ``/private.php?action=read&convid={id}`` — single conversation.
- ``/private.php?action=tracking`` — tracking (sent / read receipts).
- ``/private.php`` (POST, ``action=do_send``) — send.
- ``/private.php`` (POST, ``action=do_stuff``) — bulk delete.
"""
[docs]
async def get_inbox(self, *, page: int | None = None) -> tls_client.response.Response:
"""Fetch the raw inbox page.
Args:
page: 1-indexed pagination page. ``None`` for the first page.
Returns:
Raw response. Pass ``response.text`` to :meth:`extract_messages`
and :meth:`extract_conversation_ids`, or use :meth:`inbox` to get
an :class:`~ogu_api.Inbox` dataclass directly.
"""
path = '/private.php'
if page is not None:
path = f'/private.php?page={page}'
return await self._http.get(path)
[docs]
async def inbox(self, *, page: int | None = None) -> Inbox:
"""Fetch and parse the inbox page.
Args:
page: 1-indexed pagination page.
Returns:
:class:`~ogu_api.Inbox` with ``messages``, ``conversation_ids``,
and ``my_post_key``. ``my_post_key`` is the form CSRF token used
by every state-changing call across the SDK; you can pass it
through to subsequent ``send`` / ``delete`` calls to skip an
extra round-trip.
Example:
>>> inbox = await client.messages.inbox()
>>> for m in inbox.messages:
... print(m.username, m.message)
"""
response = await self.get_inbox(page = page)
html = response.text
return Inbox(
messages = tuple(self.extract_messages(html)),
conversation_ids = tuple(self.extract_conversation_ids(html)),
my_post_key = self.extract_my_post_key(html),
)
[docs]
async def get_compose_page(self) -> tls_client.response.Response:
"""Fetch the compose form (``/private.php?action=send``)."""
return await self._http.get('/private.php?action=send')
[docs]
async def get_conversation(self, conversation_id: str) -> tls_client.response.Response:
"""Fetch a single conversation thread by id.
Args:
conversation_id: The opaque ``convid`` from
:meth:`extract_conversation_ids` or :attr:`Inbox.conversation_ids`.
"""
return await self._http.get(
f'/private.php?action=read&convid={conversation_id}',
)
[docs]
async def get_tracking(self) -> tls_client.response.Response:
"""Fetch the message-tracking page (sent / read receipts)."""
return await self._http.get('/private.php?action=tracking')
[docs]
async def send(
self,
to: str,
message: str,
*,
my_post_key: str | None = None,
hidden: Mapping[str, Any] | None = None,
) -> tls_client.response.Response:
"""Send a private message.
When ``my_post_key`` and ``hidden`` are omitted, the SDK first GETs
the compose page to recover them, then POSTs the message.
Args:
to: Recipient username.
message: Message body.
my_post_key: Pre-fetched CSRF token. Skips the auto-fetch round-trip.
hidden: Pre-fetched hidden form fields. Skips the auto-fetch round-trip.
Returns:
Raw POST response. A 200 with the inbox layout indicates success.
Example:
>>> await client.messages.send(to = 'recipient', message = 'hello')
>>> # Reuse the key when sending many at once
>>> inbox = await client.messages.inbox()
>>> for r in recipients:
... await client.messages.send(to = r, message = 'hi', my_post_key = inbox.my_post_key)
"""
if my_post_key is None or hidden is None:
page = await self.get_compose_page()
text = page.text
if hidden is None:
hidden = self.extract_compose_hidden(text)
if my_post_key is None:
my_post_key = self.extract_my_post_key(text)
return await self._http.post(
'/private.php',
data = {
**hidden,
'action': 'do_send',
'my_post_key': my_post_key,
'to': to,
'message': message,
},
)
[docs]
async def delete(
self,
message_hashes: Iterable[str],
*,
my_post_key: str | None = None,
) -> tls_client.response.Response:
"""Bulk-delete messages by hash.
Each row in the inbox table has a ``toDelete[<hash>]`` checkbox; the
``hash`` is an opaque per-message id that you'd extract from the
inbox page yourself if you need fine control.
Args:
message_hashes: Iterable of message hashes to delete.
my_post_key: Pre-fetched CSRF token. Auto-fetched from the inbox
page when omitted.
Returns:
Raw POST response.
"""
if my_post_key is None:
page = await self.get_inbox()
my_post_key = self.extract_my_post_key(page.text)
data: dict[str, Any] = {
'action': 'do_stuff',
'my_post_key': my_post_key,
'delete': 'Delete',
}
for hash_value in message_hashes:
data[f'toDelete[{hash_value}]'] = '1'
return await self._http.post('/private.php', data = data)