Telegram の Saved Messages を Python で取り込んで Obsidian に流し込む

きっかけ

そろそろちゃんと情報を蓄積しなきゃいけないなと思った。

ブックマークはある。後で見返すつもりで溜めたメモもある。ところが後で見返したためしがない。「あの記事どこだっけ」となっても探せないし、そもそもブックマークしたこと自体を忘れている。情報を保存することと、情報を活用することは全然別の話なのだと改めて気がついた。

とはいえ、まずは「溜める」習慣から作らないことには話にならない。 Claude につたなく相談したところ、 Telegram が使えると言われた。

Telegram を使う

そもそも、現代のメッセンジャーアプリはその管理者が中身を読もうと思えば読める構造になっている。検閲の禁止はどうなったのか。通信の秘密はどうなったのか。 E2E 暗号化での通信は最早エンジニアとして、いや現代人としての「義務」だ。

そういうわけで Telegram を使う。

Telegram には Saved Messages という機能がある。 LINE で言うところの Keep みたいなもので、自分自身にメッセージを送る専用のチャットだ。 iOS ならショートカットアプリからこのチャットを直接開けるようにしておくと、ブラウザの「共有」ボタンからワンタップで放り込める。

iOS ショートカットの作り方

1. ショートカットアプリを開き、「＋」で新規作成
2. アクションに「URL を開く」を追加
3. URL に tg://resolve?domain=<自分のユーザー名> を入力
4. ショートカットを「ホーム画面に追加」または「共有シートに追加」

これで共有ボタンから直接 Saved Messages へ送れるようになる。

これで「溜める」入り口はできた。

Python で取り込む

Telegram はサードパーティライブラリ Telethon を通じて Python からメッセージを取得できる。まず Telegram のデベロッパーページでアプリを登録して API キーを発行する。

アプリ登録の手順

1. https://my.telegram.org にアクセスしてログイン
2. 「API development tools」を開く
3. アプリ名・プラットフォーム（Desktop を選べばよい）を入力して作成
4. api_id と api_hash が発行される

取得した認証情報は .env とか、どこかに書いておく。

TELEGRAM_API_ID=123456
TELEGRAM_API_HASH=abcdef1234567890abcdef1234567890

あとは Telethon を使って Saved Messages を取得するだけだ。初回実行時は QR コードか電話番号でのログインが必要だが、セッションファイルが生成されて以降は不要になる。

from telethon import TelegramClient
from telethon.tl.types import Message
import os
from dotenv import load_dotenv

load_dotenv()
client = TelegramClient(
    "session",
    int(os.environ["TELEGRAM_API_ID"]),
    os.environ["TELEGRAM_API_HASH"],
)

async def fetch():
    async with client:
        async for msg in client.iter_messages("me", reverse=True):
            if isinstance(msg, Message):
                print(msg.date, msg.text)

取り込んだメッセージはひとまず Obsidian の inbox に流し込む設計にした。

ブラウザや Twitter から「共有」すると、 Saved Messages に届くのは URL だけ だが、
まぁそれは後から取得しに行けばよかろう。と思っていたが、そんな事はなかった。

ツイートは後から取得できない

ブログ記事なら URL を fetch して本文を取れるが、 Twitter（現 X ）は API が事実上封じられている。スクレイピングもログインが必要になっており、素直には取れない。

しかし先人というのは偉大な物で、 fxTwitter というものがあった。 api.fxtwitter.com というエンドポイントにツイート ID を投げると、本文や画像の情報が JSON で返ってくる。非公式だが安定して動いている。

import httpx, re

FXTWITTER_BASE = "https://api.fxtwitter.com"
TWEET_URL_RE = re.compile(
    r"https?://(?:www\.)?(?:twitter\.com|x\.com)/[A-Za-z0-9_]+/status/(\d+)"
)

async def fetch_tweet(client: httpx.AsyncClient, tweet_id: str) -> dict:
    resp = await client.get(f"{FXTWITTER_BASE}/status/{tweet_id}")
    tweet = resp.json().get("tweet", {})
    article = tweet.get("article")
    return {
        "text": tweet.get("text"),                          # 通常ツイート
        "article_title":   article.get("title")        if article else None,
        "article_preview": article.get("preview_text") if article else None,
    }

ツイートには通常の投稿と「記事」投稿の2種類があり、 fxTwitter はそれを別フィールドで返してくれる。これで Twitter の問題はひとまず解決した。

Zenn もしかり

一方で Zenn はどうにもならなかった。 og:description は存在しない。 __NEXT_DATA__ にも本文は入っていない。 Zenn 非公式 API は記事一覧の取得はできるが、本文取得のエンドポイントは存在しない。

結局 Zenn はタイトルだけ取れればよしとした。タイトルだけでも、タグ付けと検索には十分使える。

今どこにいるか

inbox には 190 件ほど流し込めた。タグ付けはキーワードベースのルールで自動で行っている。

この先どう活用していくかはまだ模索中だ。溜めたものを眺めながら記事にする、というのが今回のこの記事の出自でもある。うまく回せるようになったらまた書くかもしれない。