collector.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Comprehensive functions used for collecting the dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Codes below are used for collecting information of Tumblr users and their posts\n",
    "\n",
    "First import all necessary libraries and get signed in and connect to API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# imports\n",
    "# !pip install pytumblr\n",
    "# !pip install pyyaml\n",
    "# !pip install pip-tools\n",
    "\n",
    "from collector import new_oauth\n",
    "from future import standard_library\n",
    "standard_library.install_aliases()\n",
    "from builtins import input\n",
    "import pytumblr\n",
    "import yaml\n",
    "import os\n",
    "from requests_oauthlib import OAuth1Session\n",
    "import csv\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "set up the Tumblr API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "yaml_path = os.path.expanduser('~') + '/.tumblr'\n",
    "alternate_path = os.path.expanduser('~') + '/.pytumblr'\n",
    "\n",
    "if os.path.isdir(yaml_path) or os.path.exists(alternate_path):\n",
    "    yaml_path = alternate_path\n",
    "\n",
    "if not os.path.exists(yaml_path):\n",
    "    tokens = new_oauth(yaml_path)\n",
    "else:\n",
    "    yaml_file = open(yaml_path, \"r\")\n",
    "    tokens = yaml.safe_load(yaml_file)\n",
    "    yaml_file.close()\n",
    "\n",
    "client = pytumblr.TumblrRestClient(\n",
    "    tokens['consumer_key'],\n",
    "    tokens['consumer_secret'],\n",
    "    tokens['oauth_token'],\n",
    "    tokens['oauth_token_secret']\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "define functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# takes a list of user names\n",
    "# output fields to be recorded\n",
    "def write_each_row(user_name, writer):\n",
    "    current_user = client.blog_info(user_name)['blog']\n",
    "    if 'errors' in current_user.keys():\n",
    "        return\n",
    "    writer.writerow({'user_name': user_name, # username\n",
    "                     'name': current_user['title'], # user's customized name\n",
    "                     'avatar': current_user['avatar'][1]['url'], # avatar url for downloading\n",
    "                     'total_posts': current_user['total_posts'], # number of posts\n",
    "                     'description': current_user['description'], # user self description\n",
    "                     'allow_question': current_user['ask'], # allow unfollowing to ask questions\n",
    "                     'allow_question_anon': current_user['ask_anon'] # allow unfollowing to ask questions anonymously\n",
    "                     })\n",
    "\n",
    "# parsing names\n",
    "# choose one post\n",
    "def parsing_names(blog, post_id, accessed):\n",
    "    one_note = client.notes(blog,post_id)['notes']\n",
    "    # get all note info\n",
    "    name_list=[blog]\n",
    "    for each in one_note:\n",
    "        # make sure no duplicate\n",
    "        if each['blog_name'] not in name_list and each['blog_name'] not in accessed:\n",
    "            name_list.append(each['blog_name'])\n",
    "    return name_list\n",
    "\n",
    "# add user info to the given file\n",
    "def add_to_file(csv_file, name_list):\n",
    "    accessed = []\n",
    "    with open(csv_file, 'a', newline='') as csvfile:\n",
    "        fieldnames = ['user_name', 'name', 'avatar', 'total_posts', 'description', 'allow_question', 'allow_question_anon']\n",
    "        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)\n",
    "        added_name = name_list\n",
    "        all_names = added_name\n",
    "        n = 0\n",
    "        while n<1000 and len(added_name)>2:\n",
    "            for user in added_name:\n",
    "                if user not in accessed and user not in all_names:\n",
    "                    write_each_row(user, writer)\n",
    "                    accessed.append(user)\n",
    "            blog = added_name[len(added_name)-1]\n",
    "            post_id = client.posts(blog)['posts'][1]['id']\n",
    "            added_name = parsing_names(blog, post_id, accessed)\n",
    "            all_names.append(added_name)\n",
    "            n = n+1\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "fill in the customized file name and user names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create file and header\n",
    "csv_file = \"\"\n",
    "names = []\n",
    "with open(csv_file, 'w', newline='') as csvfile:\n",
    "    fieldnames = ['user_name', 'name', 'avatar', 'total_posts', 'description', 'allow_question', 'allow_question_anon']\n",
    "    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)\n",
    "    writer.writeheader()\n",
    "    for user in names:\n",
    "        write_each_row(user, writer)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# enter the initial users to collect information. size > 2\n",
    "add_to_file(csv_file, []) # replace with customized input"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Below are codes used for generating posts info files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# this method is for adding posts info\n",
    "# takes a user name, collect all its posts\n",
    "def write_each_row_posts(client_name, writer):\n",
    "    user = client.posts(client_name)\n",
    "    if 'errors' in user.keys():\n",
    "        return\n",
    "    for single_post in user['posts']:\n",
    "\n",
    "        notes_likes = client.notes(client_name,single_post['id'], mode = 'likes')['total_notes']\n",
    "        \n",
    "        writer.writerow({'blog_name': client_name, # user who posts\n",
    "                     'id': single_post['id_string'], # unique id for reference\n",
    "                     'date': single_post['date'], # date posted\n",
    "                     'type': single_post['type'], # blog type\n",
    "                     'is_blocks_post_format': single_post['is_blocks_post_format'],\n",
    "                     'tags': single_post['tags'], # tags of this post\n",
    "                     'note_count': single_post['note_count'], # number of notes\n",
    "                     'total_likes': notes_likes, # number of likes\n",
    "                     'total_reblogs': single_post['note_count']-notes_likes, # number of reblogs\n",
    "                     'summary': single_post['summary'], # summary of the post\n",
    "                     'slug': single_post['slug'],\n",
    "                     'interactability_reblog': single_post['interactability_reblog'],\n",
    "                     'interactability_blaze': single_post['interactability_blaze'],\n",
    "                })\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define list of names to be collected\n",
    "names_for_posts = []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create csv file for demo purpose\n",
    "csv_file = 'post_output.csv'\n",
    "with open(csv_file, 'w', newline='') as csvfile:\n",
    "    fieldnames = ['blog_name', 'id', 'date', 'type', 'is_blocks_post_format', 'tags', 'note_count', 'total_likes', 'total_reblogs', 'summary',\n",
    "            'slug', 'interactability_reblog', 'interactability_blaze']\n",
    "    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)\n",
    "    writer.writeheader()\n",
    "    for user in names_for_posts:\n",
    "        write_each_row_posts(user, writer)\n",
    "        names_for_posts.remove(user)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# optionally get original authors for a list of blogs.\n",
    "# input posts as a list of [blog, id]\n",
    "def get_original_author(posts):\n",
    "    oriauthor = [] #output list\n",
    "    for i in range(len(posts)):\n",
    "        blog = posts[0]\n",
    "        id = posts[1]\n",
    "        single_post=client.posts(blog, id)\n",
    "        if 'errors' in single_post.keys():\n",
    "            if single_post['meta']['status'] == 404:\n",
    "                author = ''\n",
    "            else:\n",
    "                print('error at index: ',i)\n",
    "                pass\n",
    "        else: \n",
    "            trail = single_post['posts'][0]['trail']\n",
    "            if len(trail) < 1:\n",
    "                author = ''\n",
    "            else:\n",
    "                author = trail[0]['blog']['name']\n",
    "            print(author)\n",
    "\n",
    "        oriauthor.append(author)\n",
    "    return oriauthor"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "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.10.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}