Meetup Import Neo4j.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "import requests\n",
    "import time\n",
    "from neo4j.v1 import GraphDatabase, basic_auth\n",
    "\n",
    "neo4jUrl = os.environ.get('NEO4J_URL',\"bolt://localhost\")\n",
    "neo4jUser = os.environ.get('NEO4J_USER',\"neo4j\")\n",
    "neo4jPass = os.environ.get('NEO4J_PASSWORD',\"test\")\n",
    "meetupKey = os.environ.get('MEETUP_API_KEY',\"\")\n",
    "\n",
    "if len(meetupKey) == 0 : \n",
    "    raise(Exception(\"No Meetup API Key configured\"))\n",
    "\n",
    "driver = GraphDatabase.driver(neo4jUrl, auth=basic_auth(neo4jUser, neo4jPass))\n",
    "\n",
    "session = driver.session()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(session.run( \"MATCH (n:Meetup) RETURN count(*)\").single())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "importGroups = \"\"\"\n",
    "UNWIND {json} as g\n",
    "MERGE (group:Group:Meetup:Container {id:g.id}) \n",
    "  ON CREATE SET group.title=g.name,group.text=g.description,group.key=g.urlname,group.country=g.country,group.city=g.city,group.created=g.created,group.link=g.link,group.longitude=g.lon,group.latitude=g.lat,group.members=g.members\n",
    "SET group.rating=g.rating\n",
    "FOREACH (organizer IN [o in [g.organizer] WHERE g.organizer IS NOT NULL] |\n",
    "  MERGE (owner:User:Meetup {id:organizer.member_id}) ON CREATE SET owner.name = organizer.name\n",
    "  MERGE (owner)-[:CREATED]->(group)\n",
    ")\n",
    "\n",
    "FOREACH (t IN g.topics | MERGE (tag:Tag:Meetup {name:t.urlkey}) ON CREATE SET tag.id = t.id, tag.description=t.name MERGE (group)-[:TAGGED]->(tag))\n",
    "WITH group WHERE (group.title + group.text) =~ \"(?is).*(graph|neo4j).*\"\n",
    "SET group:Graph\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "importMembers = \"\"\"\n",
    "MATCH (g:Group {id:{group}})\n",
    "UNWIND {json} as m\n",
    "MERGE (user:User:Meetup {id:m.id}) \n",
    "  ON CREATE SET user.name=m.name,user.text=m.bio,user.country=m.country,user.city=m.city,user.state=m.state,user.hometown=m.hometown,\n",
    "  user.created=m.joined,user.latitude=m.lat,user.longitude=m.lon,user.link=m.link,user.groups=m.membership_count,user.picture=m.photo_url\n",
    "SET user.updated=m.visited\n",
    "MERGE (user)-[:JOINED]->(g)\n",
    "FOREACH (t IN m.topics | MERGE (tag:Tag:Meetup {name:t.urlkey}) ON CREATE SET tag.id = t.id, tag.description=t.name MERGE (user)-[:TAGGED]->(tag))\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "importEvents = \"\"\"\n",
    "UNWIND {json} as e\n",
    "MATCH (g:Group {id:e.group.id})\n",
    "WITH e,g WHERE (g:Graph OR (e.name + e.description) =~ \"(?is).*(graph|neo4j).*\")\n",
    "MERGE (event:Event:Meetup {id:e.id}) \n",
    "ON CREATE SET event.title=e.name,event.text=e.description,event.created=e.created,\n",
    "event.link=e.event_url,event.time=e.time,\n",
    "event.utc_offset=e.utc_offset,event.duration=e.duration\n",
    "SET event.updated=e.updated,event.headcount=e.headcount,event.waitlist_count=e.waitlist_count,\n",
    "event.maybe_rsvp_count=e.maybe_rsvp_count,event.yes_rsvp_count=e.yes_rsvp_count,event.rsvp_limit=e.rsvp_limit,\n",
    "event.announced=e.announced,event.comment_count=e.comment_count,event.status=e.status,event.rating=e.rating.average,event.ratings=e.rating.count\n",
    "MERGE (g)-[:CONTAINED]->(event)\n",
    "FOREACH (o in coalesce(e.event_hosts,[]) |\n",
    "  MERGE (host:User:Meetup {id:o.member_id}) ON CREATE SET host.name = o.member_name\n",
    "  MERGE (host)-[:CREATED]->(event)\n",
    ")\n",
    "\n",
    "WITH event, e.venue as v\n",
    "WHERE v IS NOT NULL AND v.id IS NOT NULL\n",
    "MERGE (venue:Venue:Meetup {id:v.id}) ON CREATE SET venue.name=v.name, venue.longitude=v.lon,venue.latitude=v.lat,venue.country = v.country, venue.city=v.city,venue.address=v.address_1,venue.country_name = v.localized_country_name\n",
    "\n",
    "MERGE (venue)-[:HOSTED]->(event)\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "importRsvps = \"\"\"\n",
    "UNWIND {json} as r\n",
    "MATCH (user:User:Meetup {id:r.member.member_id}) \n",
    "MATCH (event:Event:Meetup {id:r.event.id})\n",
    "MERGE (user)-[rsvp:ATTENDED]->(event) \n",
    "ON CREATE SET rsvp.comments=r.comments,rsvp.created=r.created\n",
    "SET rsvp.updated=r.mtime,rsvp.guests=r.guests,rsvp.host=r.host,rsvp.response=r.response\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def run_import(type, url, importQuery, params):\n",
    "    page=0\n",
    "    hasMore=True\n",
    "    items=100\n",
    "\n",
    "    while hasMore == True:\n",
    "        apiUrl = url + \"&key={key}&offset={offset}&page={items}\".format(key=meetupKey,offset=page,items=items)\n",
    "\n",
    "        response = requests.get(apiUrl, headers = {\"accept\":\"application/json\"})\n",
    "        if response.status_code != 200:\n",
    "            print(response.text)\n",
    "\n",
    "        rate_remain=int(response.headers['X-RateLimit-Remaining'])\n",
    "        rate_reset=int(response.headers['X-RateLimit-Reset'])\n",
    "\n",
    "        json = response.json()\n",
    "        meta = json['meta']\n",
    "        results = json.get(\"results\",[])\n",
    "        hasMore = len(meta.get(\"next\",\"\")) > 0\n",
    "        if  len(results) > 0:\n",
    "            p = {\"json\":results}\n",
    "            p.update(params)\n",
    "            result = session.run(importQuery,p)\n",
    "            print(result.consume().counters)\n",
    "            page = page + 1\n",
    "\n",
    "        print(type,\"results\",len(results),\"has_more\",hasMore,\"quota\",rate_remain,\"reset (s)\",rate_reset,\"page\",page)\n",
    "        time.sleep(1)\n",
    "        if rate_remain <= 0:\n",
    "            time.sleep(rate_reset) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "tag=\"neo4j\"\n",
    "groupUrl=\"https://api.meetup.com/2/groups?topic={tag}&radius=40000&text_format=plain&order=id&omit=contributions,group_photo,approved,join_info,membership_dues,self,similar_groups,sponsors,simple_html_description,welcome_message\".format(tag=tag)\n",
    "run_import(\"groups\",groupUrl,importGroups,{})\n",
    "\n",
    "result = session.run(\"MATCH (g:Group:Meetup) RETURN g.id as id, g.key as key\")\n",
    "groups = []\n",
    "for record in result:\n",
    "    if record[\"id\"] != None:\n",
    "        group = record[\"id\"]\n",
    "        memberUrl=\"https://api.meetup.com/2/members?group_id={group}&text_format=plain&order=visited&omit=photo,photos\".format(group=group)\n",
    "        run_import(\"members \"+record[\"key\"],memberUrl,importMembers,{\"group\":group})\n",
    "        groups += [str(group)]\n",
    "        \n",
    "eventUrl=\"https://api.meetup.com/2/events?group_id={groups}&status=upcoming,past&text_format=plain&order=time&omit=fee,photo_sample,rsvp_rules,rsvp_sample&fields=event_hosts\".format(groups=\",\".join(groups))\n",
    "run_import(\"events\",eventUrl,importEvents,{})\n",
    "\n",
    "result = session.run(\"MATCH (g:Group:Meetup)-[:CONTAINED]->(e:Event) WHERE NOT e:Past RETURN g.key as group, collect(toString(e.id)) as events\")\n",
    "for record in result:\n",
    "    events = record[\"events\"]\n",
    "    while len(events) > 0:\n",
    "        part = events[:100]\n",
    "        events = events[100:]\n",
    "        rsvpUrl=\"https://api.meetup.com/2/rsvps?event_id={events}&omit=member_photo,pay_status,venue,group,answers\".format(events=\",\".join(part))\n",
    "        run_import(\"rsvps \"+record[\"group\"],rsvpUrl,importRsvps,{})\n",
    "\n",
    "result = session.run(\"MATCH (event:Event:Meetup) WHERE event.time < timestamp() SET event:Past\")\n",
    "print(\"set past event label\",result.consume().counters)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "session.close()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}