Build_wiki_weeks.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# conf.getAll()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import timezone\n",
    "from datetime import datetime\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import pyspark.sql.functions as f\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import Window\n",
    "from functools import reduce\n",
    "from pyspark.sql.functions import udf\n",
    "from dateutil import parser\n",
    "import pickle"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sparkContext.addPyFile(\"../.local/lib/python3.5/site-packages/mwcomments-0.3.3-py3.5.egg\")\n",
    "spark.sparkContext.addPyFile(\"../.local/lib/python3.5/site-packages/sortedcontainers-2.1.0-py3.5.egg\")\n",
    "spark.sparkContext.addPyFile(\"../.local/lib/python3.5/site-packages/python_dateutil-2.8.0-py3.5.egg\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import mwcomments"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "wtm = mwcomments.WikiToolMap.load_WikiToolMap()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "broad_wtm = sc.broadcast(wtm)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "treated_wikis = pd.read_csv(\"data/ores_rcfilters_cutoffs.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# select canonical cutoffs\n",
    "cutoffs = treated_wikis.groupby(treated_wikis.wiki_db).min()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "cutoffs.timestamp = pd.to_datetime(cutoffs.deploy_dt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "cutoffs = cutoffs.reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# ok we're ready to fire up spark and make a stratified sample\n",
    "wmhist = spark.read.table(\"wmf.mediawiki_history\")\n",
    "# we only need the latest snapshot\n",
    "wmhist = wmhist.filter(f.col(\"snapshot\") == \"2019-04\")\n",
    "wmhist = wmhist.filter((f.col(\"event_entity\") == \"revision\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[wiki_db: string, event_entity: string, event_type: string, event_timestamp: string, event_comment: string, event_user_id: bigint, event_user_text_historical: string, event_user_text: string, event_user_blocks_historical: array<string>, event_user_blocks: array<string>, event_user_groups_historical: array<string>, event_user_groups: array<string>, event_user_is_bot_by_historical: array<string>, event_user_is_bot_by: array<string>, event_user_is_created_by_self: boolean, event_user_is_created_by_system: boolean, event_user_is_created_by_peer: boolean, event_user_is_anonymous: boolean, event_user_registration_timestamp: string, event_user_creation_timestamp: string, event_user_first_edit_timestamp: string, event_user_revision_count: bigint, event_user_seconds_since_previous_revision: bigint, page_id: bigint, page_title_historical: string, page_title: string, page_namespace_historical: int, page_namespace_is_content_historical: boolean, page_namespace: int, page_namespace_is_content: boolean, page_is_redirect: boolean, page_is_deleted: boolean, page_creation_timestamp: string, page_first_edit_timestamp: string, page_revision_count: bigint, page_seconds_since_previous_revision: bigint, user_id: bigint, user_text_historical: string, user_text: string, user_blocks_historical: array<string>, user_blocks: array<string>, user_groups_historical: array<string>, user_groups: array<string>, user_is_bot_by_historical: array<string>, user_is_bot_by: array<string>, user_is_created_by_self: boolean, user_is_created_by_system: boolean, user_is_created_by_peer: boolean, user_is_anonymous: boolean, user_registration_timestamp: string, user_creation_timestamp: string, user_first_edit_timestamp: string, revision_id: bigint, revision_parent_id: bigint, revision_minor_edit: boolean, revision_deleted_parts: array<string>, revision_deleted_parts_are_suppressed: boolean, revision_text_bytes: bigint, revision_text_bytes_diff: bigint, revision_text_sha1: string, revision_content_model: string, revision_content_format: string, revision_is_deleted_by_page_deletion: boolean, revision_deleted_by_page_deletion_timestamp: string, revision_is_identity_reverted: boolean, revision_first_identity_reverting_revision_id: bigint, revision_seconds_to_identity_revert: bigint, revision_is_identity_revert: boolean, revision_is_from_before_page_creation: boolean, revision_tags: array<string>, snapshot: string]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "wmhist"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.select(['wiki_db','event_timestamp','event_comment','revision_id','revision_parent_id','revision_text_bytes','revision_text_bytes_diff','revision_text_sha1','revision_is_identity_reverted','revision_first_identity_reverting_revision_id','revision_is_identity_revert','revision_tags','event_user_id','event_user_text','event_user_is_anonymous','event_user_creation_timestamp','event_user_first_edit_timestamp','event_user_revision_count','event_user_seconds_since_previous_revision','page_id','page_title_historical','page_title',                'page_namespace','page_is_redirect','page_is_deleted','page_revision_count','page_seconds_since_previous_revision',\"event_user_groups\",\"event_user_is_bot_by\",\"revision_deleted_parts\",'revision_deleted_parts_are_suppressed','revision_tags'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(wiki_db='ptwiki', event_timestamp='2014-10-28 04:32:30.0', event_comment='Aviso sobre a eliminação da página \"[[Cur.to]]\", usando [[WP:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='cawiki', event_timestamp='2017-11-30 11:25:43.0', event_comment='Avís a usuari, usant [[:es:Wikipedia:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='ptwiki', event_timestamp='2018-08-08 05:55:33.0', event_comment='Página marcada que existem referências sem formatação, usando [[WP:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='ptwiki', event_timestamp='2015-08-25 16:28:07.0', event_comment='Página marcada como sem fontes, usando [[WP:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='ptwiki', event_timestamp='2015-09-15 12:07:45.0', event_comment='Aviso sobre a eliminação da página \"[[Sistema de alta pressão]]\", usando [[WP:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='ptwiki', event_timestamp='2019-01-30 14:36:29.0', event_comment='Adicionando novo pedido, usando [[WP:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='ptwiki', event_timestamp='2018-11-08 12:32:16.0', event_comment='Adicionando novo pedido, usando [[WP:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='ptwiki', event_timestamp='2016-09-20 01:34:59.0', event_comment='Adicionando mensagem com a predefinição \"[[Predefinição:bem-vindo IP|bem-vindo IP]]\", usando [[WP:FastButtons|FastButtons]]', revision_tags=None),\n",
       " Row(wiki_db='eswiki', event_timestamp='2013-05-18 03:57:54.0', event_comment='Aviso al usuario, usando [[Wikipedia:FastButtons|Fastbuttons]]', revision_tags=None),\n",
       " Row(wiki_db='ptwiki', event_timestamp='2016-06-02 11:45:11.0', event_comment='Adicionando mensagem com a predefinição \"[[Predefinição:aviso-vandalismo|aviso-vandalismo]]\", usando [[WP:FastButtons|FastButtons]]', revision_tags=None)]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "wmhist.filter(f.col(\"event_comment\").like(\"%FastButtons%\")).select(['wiki_db','event_timestamp','event_comment','revision_tags']).limit(10).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "treated_wikis = set(treated_wikis.wiki_db)\n",
    "\n",
    "wmhist = wmhist.withColumn(\"treated\",f.col(\"wiki_db\").isin(treated_wikis))\n",
    "\n",
    "wmhist = wmhist.filter(f.col(\"treated\") == True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "if not os.path.exists(\"deleted_config_revisions_treated.csv\"):\n",
    "\n",
    "    missing_configs = wmhist.filter( (f.col(\"page_title\").like(\"%Undo-summary%\")) | (f.col(\"page_title\").like(\"%Revertpage%\"))).filter(f.col(\"page_is_deleted\")).select('wiki_db','revision_id','page_id','page_title').toPandas()\n",
    "\n",
    "    missing_configs.to_csv(\"deleted_config_revisions_treated.csv\")\n",
    "else:\n",
    "    missing_configs = pd.read_csv(\"deleted_config_revisions_treated.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from spark_functions import build_wmhist_step1\n",
    "wmhist = build_wmhist_step1(wmhist)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "from spark_functions import process_reverts, add_revert_types\n",
    "reverts = process_reverts(wmhist, spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from spark_functions import add_revert_types, broadcast_match_comment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<function spark_functions.broadcast_match_comment.<locals>.my_match_comment>"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "broadcast_match_comment(spark.sparkContext)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = add_revert_types(wmhist, comment_column='event_comment')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = reverts.select(['wiki_db_l','revert_timestamp','reverted_revision_id',f.col('role_type').alias(\"revert_role_type\"),f.col('anon_new_established').alias('reverted_anon_new_established'),'is_damage','time_to_revert','revert_comment','revert_user_Nreverts_past_month','revert_user_text','revert_user_id'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.join(reverts, on =[wmhist.wiki_db == reverts.wiki_db_l, wmhist.revision_id == reverts.reverted_revision_id],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# testpd = wmhist.filter( (f.col(\"wiki_db\") == 'enwiki') & \n",
    "#               (f.array_contains(col=\"revision_tags\",value=\"mw-undo\")) & \n",
    "#               (f.array_contains(col=\"revert_tools_match\",value=\"undo\"))).select(['event_comment',\n",
    "#                                                                             'revert_tools_match',\n",
    "#                                                                              'revision_tags',\n",
    "#                                                                             'revision_id',\n",
    "#                                                                             'event_timestamp']).limit(30).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.filter(f.col(\"wiki_db\").isin(list(treated_wikis)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "# for convenience we'll use the start of the week as the time of intervention and ignore the fact that it was actually mid-week in some cases\n",
    "wmhist = wmhist.withColumn(\"week\",f.date_trunc(\"week\",wmhist.event_timestamp))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "min_time = f.lit(\"2007-01-01\")\n",
    "wmhist = wmhist.filter(f.col('week') > min_time)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits_by_user_week = wmhist.filter(f.col(\"page_namespace\")==0).groupBy(['wiki_db','week','event_user_id']).agg(f.count(wmhist.revision_id).alias(\"wiki_week_counts\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "wikis = wmhist.select(\"wiki_db\").distinct().toPandas()\n",
    "weeks = wmhist.select(\"week\").distinct().toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "wikis.wiki_db.to_csv(\"all_wikis.csv\",index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "wikis = wikis.assign(key=1)\n",
    "weeks = weeks.assign(key=1)\n",
    "wiki_weeks = wikis.merge(weeks,on='key')\n",
    "wiki_weeks = wiki_weeks.drop('key',axis=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = spark.createDataFrame(wiki_weeks)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "# spot check\n",
    "#wmhist.filter((wmhist.event_user_text == \"AaronSw\") & (wmhist.wiki_db == \"enwiki\")).select(['event_user_id','week','wiki_db','event_user_text']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "# checks out\n",
    "###edits_by_user_week.filter((edits_by_user_week.event_user_id==20842) & (edits_by_user_week.wiki_db==\"enwiki\")).orderBy(edits_by_user_week.week).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits_by_user_week = edits_by_user_week.withColumn(\"week_unix\", f.unix_timestamp(edits_by_user_week.week))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "tmin_d28_in_s = (28+7)*24*60*60\n",
    "tmin_d7_in_s =  (7)*24*60*60\n",
    "\n",
    "active_window = Window.partitionBy(['wiki_db','event_user_id']).orderBy(['week_unix']).rangeBetween(-1 * tmin_d28_in_s,-1*tmin_d7_in_s)\n",
    "\n",
    "edits_by_user_week = edits_by_user_week.withColumn(\"user_edits_last_month\",f.sum(\"wiki_week_counts\").over(active_window))\n",
    "edits_by_user_week = edits_by_user_week.withColumn(\"is_active\", (~ f.isnull(edits_by_user_week.user_edits_last_month)) &   (edits_by_user_week.user_edits_last_month >= 5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "# spot check\n",
    "#edits_by_user_week.filter((edits_by_user_week.event_user_id==20842) & (edits_by_user_week.wiki_db=='enwiki')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits_by_week = edits_by_user_week.groupBy(['wiki_db','week']).agg(\n",
    "    f.sum(\"wiki_week_counts\").alias(\"total_edits\"),\n",
    "    f.sum(edits_by_user_week.is_active.cast(IntegerType())).alias(\"active_editors\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "# break reverts out into: undo, huggle, bot "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist.filter((f.col(\"wiki_db\")==\"aawiki\") & (f.col(\"week\") == \"2007-11-19 00:00:00\") &( wmhist.revision_is_identity_revert == True)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.fillna(0,[\"mean_user_reverts\",\"n_reverts\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wiki_weeks.filter(f.isnull(f.col(\"user_week_sd\"))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = reverts.withColumnRenamed(\"wiki_db_l\",\"wiki_db\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = reverts.withColumn(\"week\",f.date_trunc(\"week\",f.col('revert_timestamp')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "# we only want to look at the damaging reverts for measuring our vandalism fighting outcomes\n",
    "\n",
    "reverts = reverts.filter(f.col(\"is_damage\") == True)      \n",
    "# # exclude reverts with ttr > 30 days = 60 seconds * 60 minutes / second * 24hours / day * 30 days           \n",
    "reverts = reverts.filter(f.col(\"time_to_revert\") <= 30*24*60*60)                                                                                                                                               \n",
    "reverts = reverts.withColumn(\"med_ttr\", f.expr('percentile_approx(time_to_revert, 0.5,1)').over(Window.partitionBy(['wiki_db','week'])))                                                                       "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "# crap the tags only exist starting in 2018"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist.filter( (f.col(\"revert_tool\")==\"undo\") & f.col(\"wiki_db\") == \"enwiki\").select(f.min(f.col(\"week\"))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "# the spike in reverts is real"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "# for the distribution of reverting activity, we need the counts of reverts by user, wiki, and week\n",
    "reverts_by_user_week = reverts.groupby(['wiki_db',\"revert_user_text\",\"week\"]).agg(f.sum(f.when( (f.col(\"is_damage\") == True),1).otherwise(0)).alias(\"user_week_reverts\"))\n",
    "                                                               "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "gb = reverts_by_user_week.groupby(['wiki_db','week']).agg(f.sum('user_week_reverts').alias(\"wiki_week_reverts\"),\n",
    "                                                         f.mean('user_week_reverts').alias(\"mean_user_reverts\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts_by_user_week = reverts_by_user_week.join(gb,on=[\"wiki_db\",\"week\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts_by_user_week = reverts_by_user_week.withColumn(\"p_reverts\",f.col(\"user_week_reverts\")/f.col(\"wiki_week_reverts\"))\n",
    "\n",
    "reverts_by_user_week = reverts_by_user_week.withColumn(\"p_reverts_sq\",f.col(\"p_reverts\") * f.col(\"p_reverts\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts_by_user_week = reverts_by_user_week.withColumn(\"user_week_deviance\", reverts_by_user_week.user_week_reverts - reverts_by_user_week.mean_user_reverts)\n",
    "\n",
    "reverts_by_user_week = reverts_by_user_week.withColumn(\"user_week_var_part\", reverts_by_user_week.user_week_deviance * reverts_by_user_week.user_week_deviance)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts_by_week = reverts_by_user_week.groupby([\"wiki_db\",\"week\"]).agg(f.mean(\"user_week_reverts\"). \\\n",
    "                                                                       alias(\"mean_user_reverts\"),\n",
    "                                                                       f.sum(\"user_week_reverts\"). \\\n",
    "                                                                       alias(\"n_reverts\"),\n",
    "                                                                       f.mean(\"user_week_var_part\"). \\\n",
    "                                                                       alias(\"user_week_var\"),\n",
    "                                                                       f.sum(\"p_reverts_sq\"). \\\n",
    "                                                                       alias(\"revert_hhi\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts_by_week = reverts_by_week.withColumn(\"user_week_sd\",f.sqrt(reverts_by_week.user_week_var))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts_by_week = reverts_by_week.withColumn(\"user_week_revert_cv\",reverts_by_week.user_week_sd / reverts_by_week.mean_user_reverts)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(reverts_by_week,on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(edits_by_week,on=['wiki_db','week'], how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# spot check\n",
    "#reverts.filter(f.col(\"is_damage\") == False).filter(f.col(\"wiki_db\")==\"enwiki\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "time_to_revert_gb = reverts.groupby(['wiki_db','week'])\n",
    "time_to_revert = time_to_revert_gb.agg(\n",
    "    f.mean(f.col(\"time_to_revert\")).alias(\"mean_ttr\"),\n",
    "    f.stddev(f.col(\"time_to_revert\")).alias(\"sd_ttr\"),\n",
    "    f.exp(f.mean(f.log(f.col(\"time_to_revert\")))).alias(\"geom_mean_ttr\"),\n",
    "    f.first(f.col(\"med_ttr\")).alias(\"med_ttr\"),\n",
    "    f.count(f.col(\"week\")).alias(\"N_revert\"))\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(time_to_revert,on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [],
   "source": [
    "# so we see that the count in the main table can be wrong. Why?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.fillna(0,[\"N_reverteds\",\"active_editors\",\"total_edits\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wiki_weeks.filter(f.isnull(f.col('geom_mean_ttr'))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = reverts.withColumnRenamed(\"role_type\",\"revert_role_type\")\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = reverts.withColumnRenamed(\"wiki_db_l\",\"wiki_db\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = reverts.withColumn(\"event_timestamp\",f.col(\"revert_timestamp\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = add_revert_types(reverts, comment_column='revert_comment')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_week_tools = reverts.groupby([\"wiki_db\",\"week\"]).pivot(\"revert_tool\",['undo','rollback','huggle','twinkle','otherTool','LiveRC','fastbuttons']).agg(f.mean(f.col(\"time_to_revert\")).alias(\"mean_ttr\"),\n",
    "    f.stddev(f.col(\"time_to_revert\")).alias(\"sd_ttr\"),\n",
    "    f.exp(f.mean(f.log(f.col(\"time_to_revert\")))).alias(\"geom_mean_ttr\"),\n",
    "    f.first(f.col(\"med_ttr\")).alias(\"med_ttr\"),\n",
    "    f.count(\"reverted_revision_id\").alias(\"N_reverts\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "wiki_week_roles = reverts.groupby([\"wiki_db\",\"week\"]).pivot(\"revert_role_type\",['admin','bot','other','patroller']).agg(f.mean(f.col(\"time_to_revert\")).alias(\"mean_ttr\"),\n",
    "    f.stddev(f.col(\"time_to_revert\")).alias(\"sd_ttr\"),\n",
    "    f.exp(f.mean(f.log(f.col(\"time_to_revert\")))).alias(\"geom_mean_ttr\"),\n",
    "    f.first(f.col(\"med_ttr\")).alias(\"med_ttr\"),\n",
    "    f.count(\"reverted_revision_id\").alias(\"N_reverts\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_week_editortypes = reverts.groupby([\"wiki_db\",\"week\"]).pivot(\"reverted_anon_new_established\",['anonymous','newcomer','established']).agg(f.mean(f.col(\"time_to_revert\")).alias(\"mean_ttr\"),\n",
    "    f.stddev(f.col(\"time_to_revert\")).alias(\"sd_ttr\"),\n",
    "    f.exp(f.mean(f.log(f.col(\"time_to_revert\")))).alias(\"geom_mean_ttr\"),\n",
    "    f.first(f.col(\"med_ttr\")).alias(\"med_ttr\"),\n",
    "    f.count(\"reverted_revision_id\").alias(\"N_reverts\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(wiki_week_tools, on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(wiki_week_roles, on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(wiki_week_editortypes, on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = reverts.withColumn('editortype_x_reverttool', f.concat_ws('_',f.col('reverted_anon_new_established'),f.col(\"revert_tool\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "pivot_values = [ed+'_'+tool for ed in ['anonymous','newcomer','established'] for tool in ['undo','rollback','huggle','twinkle','otherTool','fastbuttons','LiveRC']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_week_editortypesxtools = reverts.groupby([\"wiki_db\",\"week\"]).pivot(\"editortype_x_reverttool\",pivot_values).agg(\n",
    "    f.exp(f.mean(f.log(f.col(\"time_to_revert\")))).alias(\"geom_mean_ttr\"),\n",
    "    f.first(f.col(\"med_ttr\")).alias(\"med_ttr\"),\n",
    "    f.count(\"reverted_revision_id\").alias(\"N_reverts\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(wiki_week_editortypesxtools, on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "interesting_namespaces = [0,1,2,3,4]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tm_vars = wmhist.withColumn(\"user_new_anon\",f.when(wmhist.event_user_is_anonymous,\"anon\").otherwise(\n",
    "    f.when(wmhist.event_user_is_newcomer,\"newcomer\").otherwise(\"non_anon_newcomer\")))\n",
    "\n",
    "tm_vars = tm_vars.filter(tm_vars.page_namespace.isin(interesting_namespaces)).withColumn(\"user_new_anon_namespace\",f.concat_ws('_',tm_vars.page_namespace,tm_vars.user_new_anon))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tm_vars = tm_vars.groupby(['wiki_db','week']) \\\n",
    "                .pivot(\"user_new_anon_namespace\") \\\n",
    "                .agg(f.count(wmhist.revision_id).alias(\"N_edits\"),\n",
    "                     f.countDistinct(wmhist.event_user_text).alias(\"N_editors\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#tm_vars.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(tm_vars,on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.withColumn(\"year\",f.year(f.col('week')))\n",
    "wiki_weeks = wiki_weeks.withColumn(\"month\",f.month(f.col('week')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wiki_weeks.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wiki_weeks.write.parquet(\"/user/nathante/ores_bias/wiki_weeks.parquet\",partitionBy=[\"wiki_db\",'year','month'],mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "import requests\n",
    "from itertools import chain\n",
    "response = requests.api.get(\"https://meta.wikimedia.org/w/api.php\",{\"action\":\"sitematrix\",\"formatversion\":2,\"format\":\"json\",\"maxage\":3600,\"smaxage\":3600})\n",
    "sitematrix = response.json()[\"sitematrix\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "projname_re = re.compile(r\"https?://(.*).org\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mapping = {}\n",
    "rev_mapping = {}\n",
    "for _,v in sitematrix.items():\n",
    "    try: \n",
    "        site = v['site']\n",
    "        dbname = site[0]['dbname']\n",
    "        url = site[0]['url']\n",
    "        projname = projname_re.findall(url)[0]\n",
    "        mapping[projname]=dbname\n",
    "        rev_mapping[dbname]=projname\n",
    "    except Exception as e:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark_mapping = f.create_map([f.lit(x) for x in chain(*mapping.items())])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pageviews = spark.read.table(\"wmf.projectview_hourly\")\n",
    "\n",
    "pageviews = pageviews.filter(pageviews.agent_type == 'user').select(['project','year','month','day','hour','view_count'])\n",
    "pageviews = pageviews.withColumn(\"date\", f.concat_ws(\"-\",pageviews.year,pageviews.month,pageviews.day))\n",
    "pageviews = pageviews.groupBy(['project',\"date\"]).agg(f.sum('view_count').alias(\"view_count\"))\n",
    "pageviews = pageviews.withColumn(\"week\", f.date_trunc('week',pageviews.date))\n",
    "pageviews = pageviews.withColumn(\"wiki_db\",spark_mapping.getItem(pageviews.project))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#missing_in_mapping = pageviews.filter(f.isnull(f.col(\"wiki_db\"))).select(['project','wiki_db']).distinct().collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(pageviews,on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wiki_weeks.filter(f.col(\"wiki_db\") == \"enwiki\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "missing_views = wiki_weeks.groupBy('wiki_db').agg(f.mean(f.isnull(f.col(\"view_count\")).cast(IntegerType())).alias(\"p_no_views\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# missing_views = missing_views.filter(f.col(\"p_no_views\") == 1).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "week0 = wmhist.agg(f.min(\"week\")).collect()[0]['min(week)']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "page_history = spark.read.table(\"wmf.mediawiki_page_history\")\n",
    "page_history = page_history.filter(f.col(\"snapshot\") == \"2019-04\")\n",
    "# we only need revisions during our time period\n",
    "# let's assume that pages don't get deleted for now\n",
    "page_history = page_history.withColumn(\"page_creation_week\",f.date_trunc(\"week\",\"page_creation_timestamp\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_pages_baseline = page_history.filter(\n",
    "    (page_history.page_creation_week < min_time)) \\\n",
    ".groupBy(['wiki_db']) \\\n",
    ".pivot(\"page_namespace\",interesting_namespaces) \\\n",
    ".count().alias(\"n_pages_baseline\")\n",
    "\n",
    "cols = [\"wiki_db\"]\n",
    "cols.extend(\"n_pages_baseline_ns_{0}\".format(i) for i in n_pages_baseline.columns[1:])\n",
    "n_pages_baseline = n_pages_baseline.toDF(* cols)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_new_pages_by_week = page_history \\\n",
    ".filter((page_history.page_creation_week >= min_time)) \\\n",
    ".groupBy([\"wiki_db\",\"page_creation_week\"]) \\\n",
    ".pivot(\"page_namespace\",interesting_namespaces)\\\n",
    ".agg(f.count(\"page_id\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cols = [\"wiki_db\",'page_creation_week']\n",
    "cols.extend(\"n_pages_created_ns_{0}\".format(i) for i in n_new_pages_by_week.columns[2:])\n",
    "n_new_pages_by_week = n_new_pages_by_week.toDF(* cols)\n",
    "\n",
    "n_pages = n_new_pages_by_week.join(n_pages_baseline, on=['wiki_db'], how='full_outer')\n",
    "\n",
    "#n_pages = n_pages.withColumn(\"n_pages_ns\",n_pages.n_pages_baseline + n_pages.n_created_pages)\n",
    "n_pages = n_pages.withColumnRenamed('page_creation_week','week')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(n_pages,on=['wiki_db','week'],how='full_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# newcomer survival\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.withColumn(\"year\",f.year(f.col(\"week\")))\n",
    "wiki_weeks = wiki_weeks.withColumn(\"month\",f.month(f.col(\"week\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Change below to use new format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks_out = wiki_weeks.repartition(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks_out.write.csv(\"/user/nathante/ores_bias_data/wiki_weeks.csv\",mode='overwrite',compression=\"none\",header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [],
   "source": [
    "?wiki_weeks_out.write.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks.select(\"wiki_db\").distinct().collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks.filter(f.col(\"wiki_db\")=='enwiki').select(['other_huggle_N_reverts','week']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks.write.parquet(\"/user/nathante/ores_bias/wiki_weeks.parquet\",partitionBy=[\"wiki_db\",\"year\",\"month\"],mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.catalog.setCurrentDatabase(\"nathante\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.catalog.listTables()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = spark.read.parquet(\"/user/nathante/ores_bias/wiki_weeks.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyarrow as pa\n",
    "import pyarrow.parquet as pq"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "import os \n",
    "os.environ['JAVA_HOME'] = '/usr/lib/jvm/java-8-openjdk-amd64'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hdfs = pa.hdfs.connect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# hdfs_file = pa.HdfsFile(\"/user/nathante/ores_bias/wiki_weeks.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks_out"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset = pq.ParquetDataset(\"/user/nathante/ores_bias/wiki_weeks.parquet\",filesystem=hdfs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "?pd.read_parquet(dataset)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# try pulling just the columns you really need. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ww_pdf = dataset.read_pandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file = hdfs.read_parquet(\"/user/nathante/ores_bias/wiki_weeks.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset = pq.ParquetDataset(\"/user/nathante/ores_bias/wiki_weeks.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "ww_pddf = wiki_weeks_out.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ww_pdf.to_csv(\"ores_bias_data/wiki_weeks2.csv\",index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark - YARN - 12g",
   "language": "python",
   "name": "spark_yarn_pyspak_nathante_12g"
  },
  "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.5.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}