{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!pip install beautifulsoup4","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import urllib.request as req\nimport json\nimport re\nfrom tqdm.auto import tqdm\nfrom bs4 import BeautifulSoup\nMAX_NUM = json.loads(req.urlopen('https://catalog.data.gov/api/3/action/package_list', timeout=60).read().decode())['result']['count']\nMAX_PAGE = (MAX_NUM//20)+1\nPAGE_URL = 'https://catalog.data.gov/dataset?page='\ntitle_pattern = re.compile(r'(?<=<h3 class=\\\"dataset-heading\\\">)(?:[.\\s]*?)(?:<a href=.*?>)(.*?)(?=</a>\\s*</h3>)')","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2021-06-05T05:54:56.922208Z","iopub.execute_input":"2021-06-05T05:54:56.922565Z","iopub.status.idle":"2021-06-05T05:54:58.170996Z","shell.execute_reply.started":"2021-06-05T05:54:56.922536Z","shell.execute_reply":"2021-06-05T05:54:58.169810Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import multiprocess as mp\nfrom multiprocessing import Pool\nimport threading\nimport queue\nfrom concurrent.futures import ThreadPoolExecutor, as_completed, ProcessPoolExecutor\nnum_cores = mp.cpu_count()\nnum_threads = 4","metadata":{"execution":{"iopub.status.busy":"2021-06-05T05:55:02.661355Z","iopub.execute_input":"2021-06-05T05:55:02.661700Z","iopub.status.idle":"2021-06-05T05:55:02.702393Z","shell.execute_reply.started":"2021-06-05T05:55:02.661672Z","shell.execute_reply":"2021-06-05T05:55:02.701332Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class SENTINEL:\n    def __init__(self):\n        pass\n    def __eq__(a, c):\n        return type(a) == type(c)\ndef convert_queue(label):\n    title = []\n    sentinel = SENTINEL()\n    label.put(sentinel)\n    for n in iter(label.get, sentinel):\n        if sentinel != n:\n            title.extend(n)\n    return title","metadata":{"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2021-06-05T05:55:11.770637Z","iopub.execute_input":"2021-06-05T05:55:11.771214Z","iopub.status.idle":"2021-06-05T05:55:11.776703Z","shell.execute_reply.started":"2021-06-05T05:55:11.771147Z","shell.execute_reply":"2021-06-05T05:55:11.776002Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"API_URL = 'https://catalog.data.gov/api/3/action/package_search?rows=1000&start='\ndef api_get(url, n, retry, label ):\n    check_num = n > MAX_NUM-1000\n    try:\n        page = req.urlopen(url, timeout=60)\n        page = json.loads(page.read().decode())\n        if n%10000==0:print('processor '+str(mp.current_process().name)+' now processing batch start at ' + str(n))\n        label_parsed = [re.sub('\\s+', ' ',n['title']) for n in page['result']['results']]\n        if label_parsed and (len(label_parsed)==1000 or check_num):\n            label.put(label_parsed)\n        else:\n            print('processor '+str(mp.current_process().name)+' only parsed ' + str(len(label_parsed)))\n            retry.put(url)\n    except Exception as e:\n        print('processor '+str(mp.current_process().name)+': ', e, ' @ ' ,url, '\\n')\n        retry.put(url)\n\ndef parser_api(parse_range):\n    print('processor '+mp.current_process().name+' is to process from '+ str(parse_range))\n    retry = queue.Queue()\n    label = queue.Queue()\n    with ThreadPoolExecutor(max_workers=4) as executor:\n        result = as_completed([executor.submit(api_get, API_URL+str(n), n, retry, label) for n in range(parse_range[0],parse_range[-1],1000)])\n    with ThreadPoolExecutor(max_workers=4) as executor:\n        while not retry.empty():\n            url = retry.get()\n            print('processor '+mp.current_process().name + ' Now retring @ ' + url)\n            retry_result = as_completed([executor.submit(api_get, url, -1, retry, label)])\n#     labels.put(convert_queue(label))\n    print('processor '+str(mp.current_process().name)+' finished')\n    return label","metadata":{"execution":{"iopub.status.busy":"2021-06-05T06:42:58.306691Z","iopub.execute_input":"2021-06-05T06:42:58.307070Z","iopub.status.idle":"2021-06-05T06:42:58.322541Z","shell.execute_reply.started":"2021-06-05T06:42:58.307035Z","shell.execute_reply":"2021-06-05T06:42:58.321765Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_page_content(url, n, max_len, retry, label):\n    try:\n        check_num = int(url.split('=')[-1]) == MAX_PAGE\n        page = req.urlopen(url, timeout=60)\n        page = page.read().decode()\n        page = BeautifulSoup(page)\n        if n%100 == 0:print('processor '+str(mp.current_process().name)+' now processing page ' + str(n))\n        label_parsed = [re.sub('\\s+', ' ',n.a.text) for n in page.find_all('h3', class_='dataset-heading')]\n        if label_parsed and (len(label_parsed)==20 or check_num):\n            label.put(label_parsed)\n        else:\n            print('processor '+str(mp.current_process().name)+' only parsed ' + str(len(label_parsed)),\n                  ', received ', str(len(re.findall(r'class=\\\"dataset-heading\\\"', page))), ' @', url)\n            retry.put(url)\n    except Exception as e:\n        print('processor '+str(mp.current_process().name)+': ', e, ' @ ' ,url, '\\n')\n        retry.put(url)\n    \ndef parser(parse_range):\n    print('processor '+mp.current_process().name+' is to process (' + str(parse_range[0])+','+str(parse_range[-1])+')\\n')\n\n    label = queue.Queue()\n    retry = queue.Queue()\n    with ThreadPoolExecutor(max_workers=10) as executor:\n        result = as_completed([executor.submit(get_page_content, PAGE_URL+str(n+1),\n                                               n, parse_range[-1], retry, label) for n in parse_range])\n    with ThreadPoolExecutor(max_workers=10) as executor:\n        while not retry.empty():\n            url = retry.get()\n            print('processor '+mp.current_process().name + ' Now retring @ ' + url)\n            retry_result = as_completed([executor.submit(get_page_content, url, -1, parse_range[-1], retry, label)])\n#     labels.put(convert_queue(label))\n    print('processor '+str(mp.current_process().name)+' finished')\n    return label","metadata":{"execution":{"iopub.status.busy":"2021-06-05T05:55:14.550396Z","iopub.execute_input":"2021-06-05T05:55:14.550971Z","iopub.status.idle":"2021-06-05T05:55:14.564692Z","shell.execute_reply.started":"2021-06-05T05:55:14.550935Z","shell.execute_reply":"2021-06-05T05:55:14.563850Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def api():\n    labels = mp.Queue()\n    parse_ranges = [(0, 76000),(76000, 146000),(146000, 219000),(219000, MAX_NUM)]\n    pool = mp.Pool(processes=num_cores)\n    results = pool.map(parser_api, parse_ranges)\n    pool.close()\n    pool.join()\n    labels = []\n    for res in results:\n        labels.extend(convert_queue(res))\n    return labels","metadata":{"execution":{"iopub.status.busy":"2021-06-05T05:55:15.999439Z","iopub.execute_input":"2021-06-05T05:55:16.000013Z","iopub.status.idle":"2021-06-05T05:55:16.005843Z","shell.execute_reply.started":"2021-06-05T05:55:15.999978Z","shell.execute_reply":"2021-06-05T05:55:16.005043Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def webpage():\n    labels = mp.Queue()\n    parse_ranges = [range(n, n+(MAX_PAGE//num_cores)+1) for n in range(0, MAX_PAGE, (MAX_PAGE//num_cores)+1)]\n    pool = mp.Pool(processes=num_cores)\n    results = pool.map(parser, parse_ranges)\n    pool.close()\n    pool.join()\n    labels = []\n    for res in results:\n        labels.extend(convert_queue(res))\n    return labels","metadata":{"execution":{"iopub.status.busy":"2021-06-05T05:55:16.787757Z","iopub.execute_input":"2021-06-05T05:55:16.788122Z","iopub.status.idle":"2021-06-05T05:55:16.794731Z","shell.execute_reply.started":"2021-06-05T05:55:16.788091Z","shell.execute_reply":"2021-06-05T05:55:16.793716Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"labels = api()","metadata":{"execution":{"iopub.status.busy":"2021-06-05T06:43:02.500633Z","iopub.execute_input":"2021-06-05T06:43:02.501011Z","iopub.status.idle":"2021-06-05T06:47:09.781876Z","shell.execute_reply.started":"2021-06-05T06:43:02.500976Z","shell.execute_reply":"2021-06-05T06:47:09.780599Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"len(labels), len(set(labels)), MAX_NUM","metadata":{"execution":{"iopub.status.busy":"2021-06-05T06:55:24.170128Z","iopub.execute_input":"2021-06-05T06:55:24.170546Z","iopub.status.idle":"2021-06-05T06:55:24.214319Z","shell.execute_reply.started":"2021-06-05T06:55:24.170512Z","shell.execute_reply":"2021-06-05T06:55:24.213113Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"with open('external_labels', 'w') as e:\n    json.dump(labels, e)","metadata":{"execution":{"iopub.status.busy":"2021-06-05T05:53:28.922167Z","iopub.status.idle":"2021-06-05T05:53:28.922684Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}