Mastodon Export to Mastodon.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Fediporter\n",
    "Jupyter notebook to migrate content like Tweets oder Mastodon posts to Mastodon.\n",
    "\n",
    "Only works with instances that are patched to allow backdated posts through the API. More info: https://github.com/lucahammer/fediporter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2022-12-23T10:05:14.110272Z",
     "start_time": "2022-12-23T10:05:14.107618Z"
    }
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "from bs4 import BeautifulSoup\n",
    "from tqdm.notebook import tqdm\n",
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 163,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\"name\":\"Imported from det.social/@Luca\",\"website\":\"https://det.social/@luca\",\"vapid_key\":\"uXL829wxgH9wYavQxkARcWv6fiNUZ94B\"}\n"
     ]
    }
   ],
   "source": [
    "with open('config.json', 'r') as f:\n",
    "    config = json.loads(f.read())\n",
    "\n",
    "API_BASE_URL = config['mastodon_url']\n",
    "MASTODON_BEARER =  \"oyufS1RbwqgIMlLcudOtA33F4N5YZrkDHiHDlw5Y-dA\"  #config['mastodon_bearer']\n",
    "HEADERS = headers={'Authorization': f'Bearer {MASTODON_BEARER}'}\n",
    "\n",
    "# Test Mastodon bearer token\n",
    "url = f\"{API_BASE_URL}/api/v1/apps/verify_credentials\"\n",
    "r = requests.get(url, headers=HEADERS)\n",
    "print(r.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2022-12-23T11:57:44.037582Z",
     "start_time": "2022-12-23T11:57:43.929260Z"
    }
   },
   "outputs": [],
   "source": [
    "def post_status(data):\n",
    "    HEADERS = {'Authorization': f'Bearer {MASTODON_BEARER}'}\n",
    "    url = f\"{API_BASE_URL}/api/v1/statuses\"\n",
    "    r = requests.post(url, \n",
    "                      data=data, \n",
    "                      headers=HEADERS)\n",
    "    return r.json()\n",
    "\n",
    "def load_mastodon_posts(file):\n",
    "    with open(file, 'r') as f:\n",
    "        posts = json.loads(f.read())['orderedItems']\n",
    "    return(posts)\n",
    "\n",
    "def fix_text(text, tags):\n",
    "    text = text.replace('</p><p>', '++breakme++</p><p>') #add placeholder to replace with linebreaks to keep paragraphs (it's weird)\n",
    "    text = BeautifulSoup(text, 'html.parser').get_text()\n",
    "    tags = [tag for tag in tags if tag['type'] == 'Mention']\n",
    "    for tag in tags:\n",
    "        name = f\"@{tag['name'].split('@')[1]} \"\n",
    "        text = text.replace(name,f\"{tag['name']} \")\n",
    "    text = text.replace('++breakme++', '\\n\\n')\n",
    "    return (text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "posts = load_mastodon_posts('detsocial/outbox.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 258,
   "metadata": {},
   "outputs": [],
   "source": [
    "masto_dict = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 262,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "10d15159e0b140d7a6289e8f773fbb89",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(HTML(value=''), FloatProgress(value=0.0, max=574.0), HTML(value='')))"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "for post in tqdm(posts):\n",
    "    if 'contentMap' in post['object'] and post['id'] not in masto_dict and (('inReplyTo' in post and '/luca/' in post['inReplyTo']) or 'inReplyTo' not in post):\n",
    "        if 'en' in post['object']['contentMap']:\n",
    "            toot = {\n",
    "                'status': fix_text(post['object']['contentMap']['en'], post['object']['tag']),\n",
    "                'language': 'en'}\n",
    "        if 'de' in post['object']['contentMap']:\n",
    "            toot = {\n",
    "                'status': fix_text(post['object']['contentMap']['de'], post['object']['tag']),\n",
    "                'language': 'de'}\n",
    "        toot['created_at'] = post['object']['published']\n",
    "        if post['to'][0].endswith('/followers') and len(post['cc']) > 0 and post['cc'][0].endswith('#Public'):\n",
    "            toot['visibility'] = 'unlisted'\n",
    "        elif post['to'][0].endswith('#Public'):\n",
    "            toot['visibility'] = 'public'\n",
    "        elif post['to'][0].endswith('/followers') and len(post['cc']) > 0 and not post['cc'][0].endswith('#Public'):\n",
    "            toot['visibility'] = 'private'\n",
    "        else:\n",
    "            toot['visibility'] = 'direct'\n",
    "        \n",
    "        if 'attachment' in post['object'] and len(post['object']['attachment']) > 0:\n",
    "            #print(post['object']['attachment'])\n",
    "            media_ids = []\n",
    "            for attachment in post['object']['attachment']:           \n",
    "                image_path = f\"detsocial/home/det/live/public/system/{attachment['url']}\"\n",
    "                file = open(image_path, 'rb')\n",
    "                data = file.read()\n",
    "                url = f\"{API_BASE_URL}/api/v2/media\"\n",
    "                files={'file': (image_path, data, 'application/octet-stream')}\n",
    "                if 'name' in attachment:\n",
    "                    values = {'description' : attachment['name']}\n",
    "                    r = requests.post(url, files=files, data=values, headers=HEADERS)\n",
    "                else:\n",
    "                    r = requests.post(url, files=files, headers=HEADERS)\n",
    "                json_data = r.json()\n",
    "                media_ids.append(json_data['id'])\n",
    "            toot['media_ids[]'] = media_ids\n",
    "        \n",
    "        if 'inReplyTo' in post['object'] and post['object']['inReplyTo'] is not None and '/luca/' in post['object']['inReplyTo']:\n",
    "            # if post is part of a thread, get ID of previous post\n",
    "            try:\n",
    "                toot['in_reply_to_id'] = masto_dict.get(post['object']['inReplyTo'])\n",
    "            except:\n",
    "                print(post)\n",
    "\n",
    "        #print(toot)        \n",
    "        posted = post_status(toot)\n",
    "        #print(posted)\n",
    "        \n",
    "        masto_dict[post['object']['id']] = posted['id']\n",
    "    else:\n",
    "        pass\n",
    "        #print('skipped')\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 280,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Manually adding new statuses\n",
    "\n",
    "toot = {}\n",
    "media_ids = []\n",
    "\n",
    "toot = {\n",
    "    'status': '''xxx''',\n",
    "    'language': 'en'}\n",
    "\n",
    "image_path = f\"xxx.png\"\n",
    "file = open(image_path, 'rb')\n",
    "data = file.read()\n",
    "url = f\"{API_BASE_URL}/api/v2/media\"\n",
    "files={'file': (image_path, data, 'application/octet-stream')}\n",
    "r = requests.post(url, files=files, data={'description' : '''xxx'''}, headers=HEADERS)\n",
    "json_data = r.json()\n",
    "media_ids.append(json_data['id'])\n",
    "toot['media_ids[]'] = media_ids\n",
    "toot['in_reply_to_id'] = 'xxx'\n",
    "toot['created_at'] = \"2022-12-25T20:21:00Z\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "post_status(toot)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# todo\n",
    "- content warnings?\n",
    "- URLs?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 257,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a8edcfd337e549e19b1533a1ec4f4a48",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(HTML(value=''), FloatProgress(value=0.0, max=2.0), HTML(value='')))"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "# delete the posted posts because something went wrong\n",
    "\n",
    "def delete_status(status_id):\n",
    "    HEADERS = {'Authorization': f'Bearer {MASTODON_BEARER}'}\n",
    "    url = f\"{API_BASE_URL}/api/v1/statuses/{status_id}\"\n",
    "    r = requests.delete(url, \n",
    "                      headers=HEADERS)\n",
    "    return r.json()\n",
    "\n",
    "for thing in tqdm(masto_dict.values()):\n",
    "    x = (delete_status(thing))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}