ads-metadata.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import csv\n",
    "from copy import deepcopy\n",
    "from datetime import datetime\n",
    "import json\n",
    "from pathlib import Path\n",
    "\n",
    "import idutils\n",
    "import requests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Get your ADS API token from https://ui.adsabs.harvard.edu/#user/settings/token\n",
    "ADS_TOKEN = 'CHANGEME'\n",
    "ADS_API_URL = 'https://api.adsabs.harvard.edu/v1/search/query'\n",
    "ADS_API_HEADERS = {\"Authorization\": \"Bearer:{}\".format(ADS_TOKEN)}\n",
    "ADS_API_PARAMS = {\n",
    "    \"q\": \"identifier:CHANGEME\",\n",
    "    \"fl\": \"title,author,doi,bibcode,identifier,doctype,pub,year,pubdate\",\n",
    "}\n",
    "\n",
    "ADS_TYPE_TO_ASCLEPIAS = {\n",
    "    'article': 'literature',\n",
    "    'eprint': 'literature',\n",
    "    'inproceedings': 'literature',\n",
    "    'inbook': 'literature',\n",
    "    'abstract': 'unknown',\n",
    "    'book': 'literature',\n",
    "    'bookreview': 'unknown',\n",
    "    'catalog': 'unknown',\n",
    "    'circular': 'unknown',\n",
    "    'erratum': 'unknown',\n",
    "    'mastersthesis': 'literature',\n",
    "    'newsletter': 'unknown',\n",
    "    'obituary': 'unknown',\n",
    "    'phdthesis': 'literature',\n",
    "    'pressrelease': 'unknown',\n",
    "    'proceedings': 'unknown',\n",
    "    'proposal': 'unknown',\n",
    "    'software': 'software',\n",
    "    'talk': 'unknown',\n",
    "    'techreport': 'unknown',\n",
    "    'misc': 'unknown',\n",
    "}\n",
    "\n",
    "\n",
    "OUTPUT_DIR = 'ads_metadata'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def fetch_metadata(bibcode):\n",
    "    params = deepcopy(ADS_API_PARAMS)\n",
    "    params['q'] = 'identifier:{}'.format(bibcode)\n",
    "    res = requests.get(ADS_API_URL, params=params, headers=ADS_API_HEADERS)\n",
    "    data = res.json()\n",
    "    assert data['response']['numFound'] == 1\n",
    "    return data['response']['docs'][0]\n",
    "\n",
    "\n",
    "def build_identifiers(data):\n",
    "    ids = set()\n",
    "    if data.get('bibcode'):\n",
    "        ids.add((data.get('bibcode'), 'ads'))\n",
    "    ids |= {(d, 'doi') for d in data.get('doi', []) if d}\n",
    "    for id_ in data.get('identifier', []):\n",
    "        try:\n",
    "            ids.add((id_, idutils.detect_identifier_schemes(id_)[0]))\n",
    "        except Exception:\n",
    "            pass\n",
    "    return [{'ID': i, 'IDScheme': s} for i, s in ids if i and s]\n",
    "\n",
    "\n",
    "def extract_date(data):\n",
    "    try:\n",
    "        return datetime.strptime(data.get('pubdate'), '%Y-%m-%d').isoformat()\n",
    "    except Exception:\n",
    "        return data.get('year')\n",
    "\n",
    "    \n",
    "def build_document(data):\n",
    "    return {\n",
    "        'Provider': 'SAO/NASA Astrophysics Data System',\n",
    "        'Object': {\n",
    "            'Identifier': build_identifiers(data),\n",
    "            'Publisher': [{'Name': data['pub']}] if data.get('pub') else None,\n",
    "            'Creator': [{'Name': n} for n in data.get('author', []) if n],\n",
    "            'Title': data.get('title', [None])[0],\n",
    "            'PublicationDate': extract_date(data),\n",
    "            'Type': { 'Name': ADS_TYPE_TO_ASCLEPIAS.get(data.get('doctype'))},\n",
    "        }\n",
    "    }\n",
    "\n",
    "\n",
    "def fetch_all_docs(csv_path):\n",
    "    with open(csv_path, 'r') as csvfile:\n",
    "        citations = list(csv.reader(csvfile, delimiter=','))[1:]\n",
    "    bibcodes = {bibcode for bibcode, *_ in citations}\n",
    "    print(len(bibcodes), 'citations read from CSV')\n",
    "    errors = []\n",
    "    docs = []\n",
    "    for i, bibcode in enumerate(bibcodes):\n",
    "        if i % 10 == 0:\n",
    "            print(i)\n",
    "        try:\n",
    "            docs.append((bibcode, build_document(fetch_metadata(bibcode))))\n",
    "        except Exception as ex:\n",
    "            errors.append((bibcode, ex))\n",
    "    return docs, errors\n",
    "\n",
    "def write_documents(documents):\n",
    "    output_dir = Path(OUTPUT_DIR)\n",
    "    output_dir.mkdir(exist_ok=True)\n",
    "    for bibcode, data in documents:\n",
    "        if '/' not in bibcode:\n",
    "            (output_dir / (bibcode + '.json')).write_text(json.dumps(data))\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "204 citations read from CSV\n",
      "0\n",
      "10\n",
      "20\n",
      "30\n",
      "40\n",
      "50\n",
      "60\n",
      "70\n",
      "80\n",
      "90\n",
      "100\n",
      "110\n",
      "120\n",
      "130\n",
      "140\n",
      "150\n",
      "160\n",
      "170\n",
      "180\n",
      "190\n",
      "200\n"
     ]
    }
   ],
   "source": [
    "docs, errors = fetch_all_docs('TriangleCornerCitations.csv')\n",
    "write_documents(docs)"
   ]
  }
 ],
 "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}