social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import json
2import sqlite3
3
4
5def migrate(conn: sqlite3.Connection):
6 columns = conn.execute("PRAGMA table_info(posts)")
7 column_names = [col[1] for col in columns]
8 if "extra_data" not in column_names:
9 _ = conn.execute("""
10 ALTER TABLE posts
11 ADD COLUMN extra_data TEXT NULL
12 """)
13
14 # migrate old bsky identifiers from json to uri as id and cid in extra_data
15 data = conn.execute("SELECT id, identifier FROM posts WHERE service = 'https://bsky.app';").fetchall()
16 rewrites: list[tuple[str, str, int]] = []
17 for row in data:
18 if row[1][0] == '{' and row[1][-1] == '}':
19 data = json.loads(row[1])
20 rewrites.append((data['uri'], json.dumps({'cid': data['cid']}), row[0]))
21 if rewrites:
22 _ = conn.executemany("UPDATE posts SET identifier = ?, extra_data = ? WHERE id = ?;", rewrites)