Watch for new weibo and push notifications

weibo_watcher.py

#!/usr/bin/env python3

"""
Copyright (c) 2020 Hao Guan

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""

import json
import random
import time
from datetime import datetime

import requests
from dateutil import parser
from lxml import etree


UA_STRING = (
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_6)'
    ' AppleWebKit/605.1.15 (KHTML, like Gecko)'
    ' Version/14.0.3 Safari/605.1.15'
)


class WeiboWatcher(object):
    """
    A watcher for new weibo published by user with <UID>.
    """
    def __init__(self, uid: int) -> None:
        super().__init__()
        self.uid = uid
        self.headers = WeiboWatcher._xhr_headers(uid)
        self.name = None
        self.weibo_cid = None
        self.latest_weibo = {
            'id': None,
            'name': None,
            'link': None,
            'date': None,
            'text': None,
        }
        self.setup()

    @staticmethod
    def _api_url(value: str, containerid: str) -> str:
        ret = (
            "https://m.weibo.cn/api/container/getIndex?"
            "type=uid"
            f"&value={value}"
        )
        if containerid is not None:
            ret = f"{ret}&containerid={containerid}"
        return ret


    @staticmethod
    def _xhr_headers(uid: str) -> dict:
        return {
            'Accept': 'application/json, text/plain, */*',
            'Referer': f'https://m.weibo.cn/u/{uid}',
            'User-Agent': UA_STRING,
            'X-Requested-With': 'XMLHttpRequest',
            'MWeibo-Pwa': '1'
        }


    def setup(self) -> None:
        """
        Retrieve basic data for a user.
        """
        data = json.loads(
            requests.get(
                WeiboWatcher._api_url(self.uid, None),
                headers=self.headers
            ).text
        )
        self.name = data["data"]["userInfo"]["screen_name"]
        tabs = data["data"]["tabsInfo"]["tabs"]
        for tab in tabs:
            if tab["tab_type"] == "weibo":
                self.weibo_cid = tab["containerid"]
                break
        self.latest_weibo['name'] = self.name
        self.update()


    def update(self) -> bool:
        """
        Check for updates.
        Return True if a new weibo is published.
        """
        ret = False
        try:
            data = json.loads(
                requests.get(
                    WeiboWatcher._api_url(self.uid, self.weibo_cid),
                    headers = self.headers
                ).text
            )

            for card in data['data']['cards']:
                if card['card_type'] == 9:
                    datet = parser.parse(card['mblog']['created_at'])
                    if self.latest_weibo['date'] is None:
                        self.latest_weibo['id'] = card['mblog']['id']
                        self.latest_weibo['date'] = datet
                        self.latest_weibo['link'] = card['scheme']
                        self.latest_weibo['text'] = card['mblog']['text']
                    elif datet > self.latest_weibo['date']:
                        ret = True
                        self.latest_weibo['id'] = card['mblog']['id']
                        self.latest_weibo['date'] = datet
                        self.latest_weibo['link'] = card['scheme']
                        self.latest_weibo['text'] = card['mblog']['text']
        except requests.RequestException as req_err:
            print(req_err)
        except json.JSONDecodeError as json_err:
            print(json_err)
        return ret


    def watch(self, callback) -> None:
        """
        Run as a daemon to watch the user.
        """
        while True:
            wait_sec = random.randint(6, 12)
            if self.update():
                print(f"\nNew from {self.name}: {self.latest_weibo['date']}")
                callback(self.latest_weibo)
            else:
                print(f"[{datetime.now()}]: Waiting...", end='\r')
            time.sleep(wait_sec)


def bark_notify(weibo: dict, api_tokens: list) -> None:
    """
    Notify <weibo> via Bark with tokens in <api_tokens>
    """
    selector = etree.HTML(weibo['text']).xpath("body/*")
    message = "New Weibo"
    for text in selector:
        if text.tail:
            message = text.tail
            break

    for token in api_tokens:
        requests.get(
            f"https://api.day.app/{token}"
            f"/{weibo['name']}"
            f"/{message}"
            f"?url={weibo['link']}"
        )


def tgbot_notify(weibo: dict, bot: str, chats: list) -> None:
    """
    Send contents in <weibo> via Telegram <bot> to <chats>.
    """
    selector = etree.HTML(
       json.loads(
            requests.get(
               f"https://m.weibo.cn/statuses/extend?id={weibo['id']}"
            ).text
        )['data']['longTextContent']
    ).xpath("body/*")

    texts = []
    for text in selector:
        if text.text:
            texts.append(text.text)
        if text.tail:
            texts.append(text.tail)

    texts.append(f"\n{weibo['link']}")

    for chat in chats:
        data = {
            "chat_id": chat,
            "text": "\n".join(texts)
        }
        requests.post(f"https://api.telegram.org/bot{bot}/sendMessage",
                        data=data)


def callbacks(weibo):
    """
    Callback function triggered by new weibo.
    Stuffs such as push notifications, save to file can be put here.
    """
    # bark_notify(weibo, ["token1", "token2"])

    # tgbot_notify(weibo, "bot_http_api", [chat_id1, chat_id2])


if __name__ == "__main__":
    WeiboWatcher("6279793937").watch(callbacks)  # @明日方舟Arknights