Python爬取FAERS数据报错

Question

1. 问题描述

• 使用luigi框架爬取faers数据报错，IDE为pycharm
• 错误信息为

No task specified

Process finished with exit code 1

2.源代码

import os
import re
import shutil
import requests
from io import BytesIO
from zipfile import ZipFile
from urllib.request import urlretrieve
from urllib.request import urlopen
from bs4 import BeautifulSoup
import pandas as pd
import numpy as np
import warnings
import luigi
import sys
import logging


def extractZip(url, source_dir, data_dir):
    logging.debug('In the Task : extractZip')
    r = requests.get(url)
    z = ZipFile(BytesIO(r.content))
    z.extractall(source_dir)
    deletePDF(source_dir)
    copyFile(source_dir, data_dir)


def deletePDF(path):
    logging.debug('In the Task : deletePDF')
    for parent, dirnames, filenames in os.walk(source_dir):
        for fn in filenames:
            if fn.lower().endswith('.pdf'):
                print("Deleteting" + fn)
                os.remove(os.path.join(parent, fn))
            if fn.lower().endswith('.doc'):
                print("Deleteting" + fn)
                os.remove(os.path.join(parent, fn))
            if fn.startswith("RPSR"):
                print("Deleteting" + fn)
                os.remove(os.path.join(parent, fn))
            if fn.startswith("INDI"):
                print("Deleteting" + fn)
                os.remove(os.path.join(parent, fn))
            if fn.startswith("THER"):
                print("Deleteting" + fn)
                os.remove(os.path.join(parent, fn))


def copyFile(source_dir, data_dir):
    logging.debug('In the Task : copyFiles')
    RootDir1 = os.getcwd() + '/' + source_dir
    TargetFolder = os.getcwd() + '/' + data_dir
    for root, dirs, files in os.walk((os.path.normpath(RootDir1)), topdown=False):
        for name in files:
            if name.endswith('.txt'):
                SourceFolder = os.path.join(root, name)
                shutil.move(SourceFolder, TargetFolder)


class get_files_url(luigi.Task):
    logging.debug('In the Task : getWebUrls')

    def requires(self):
        return []

    def run(self):
        source_dir = "FAERSsrc"
        data_dir = "FAERSdata"
        files = {}
        url = {}
        host_url = "http://www.fda.gov"
        target_page = [
            "http://www.fda.gov/Drugs/GuidanceComplianceRegulatoryInformation/Surveillance/AdverseDrugEffects/ucm082193.htm"]
        for page_url in target_page:
            try:
                page_bs = BeautifulSoup(urlopen(page_url), "lxml")
            except:
                page_bs = BeautifulSoup(urlopen(page_url))
            for url in page_bs.find_all("a"):
                a_string = str(url.string)
                if "ASCII" in a_string.upper():
                    files[a_string.encode("utf-8")] = host_url + url["href"]
                    url = host_url + url["href"]
                    extractZip(url, source_dir, data_dir)
            for url in page_bs.find_all("linktitle"):
                a_string = str(url.string)
                if "ASCII" in a_string.upper():
                    files[a_string.encode("utf-8")] = host_url + url.parent["href"]
                    url = host_url + url.parent["href"]
                    extractZip(url, source_dir, data_dir)
        with self.output().open('w') as f:
            f.write("hello")

    def output(self):
        return luigi.LocalTarget('url.txt')


class mergeData(luigi.Task):

    def requires(self):
        return [get_files_url()]

    def run(self):

        directoryPath = os.getcwd() + "/FAERSdata"
        demo = pd.DataFrame(
            columns=['primaryid', 'caseid', 'mfr_dt', 'init_fda_dt', 'rept_cod', 'mfr_num', 'mfr_sndr', 'age',
                     'sex', 'wt', 'wt_cod', 'occp_cod', 'occr_country'])
        drug = pd.DataFrame(columns=['primaryid', 'caseid', 'role_cod', 'drugname', 'route', 'dose_amt', 'dose_unit',
                                     'dose_form', 'dose_freq'])
        reaction = pd.DataFrame(columns=['primaryid', 'caseid', 'pt'])
        outcome = pd.DataFrame(columns=['primaryid', 'caseid', 'outc_cod'])
        print("in run")
        for filename in os.listdir(directoryPath):
            if "DEMO" in filename:
                demo_df = pd.read_csv(directoryPath + "/" + filename, low_memory=False, sep="$", error_bad_lines=False)
                demo_df.drop(
                    ['caseversion', 'i_f_code', 'lit_ref', 'event_dt', 'auth_num', 'fda_dt', 'age_cod', 'age_grp',
                     'e_sub', 'rept_dt', 'to_mfr', 'reporter_country'], inplace=True, axis=1, errors='ignore')
                demo_df = demo_df.loc[(demo_df['wt_cod'] == 'KG')]
                demo_df = demo_df[pd.notnull(demo_df['age'])]
                demo_df = demo_df[1:]
                demo = demo.append(demo_df, ignore_index=True)
            if "DRUG" in filename:
                durg_df = pd.read_csv(directoryPath + "/" + filename, low_memory=False, sep="$", error_bad_lines=False)
                durg_df.drop(['drug_seq', 'val_vbm', 'dose_vbm', 'cum_dose_chr', 'prod_ai', 'cum_dose_unit', 'dechal',
                              'rechal', 'lot_num', 'exp_dt', 'nda_num'], inplace=True, axis=1, errors='ignore')
                durg_df = durg_df[pd.notnull(durg_df['dose_amt'])]
                durg_df = durg_df[pd.notnull(durg_df['dose_unit'])]
                durg_df = durg_df.loc[(durg_df['role_cod'] == 'PS')]
                durg_df = durg_df[1:]
                drug = drug.append(durg_df, ignore_index=True)
            if "REAC" in filename:
                reac_df = pd.read_csv(directoryPath + "/" + filename, low_memory=False, sep="$", error_bad_lines=False)
                reac_df = reac_df.groupby('primaryid')
                reac_df = reac_df.filter(lambda x: len(x) == 1)
                reac_df = reac_df[1:]
                reaction = reaction.append(reac_df, ignore_index=True)
            if "OUTC" in filename:
                out_df = pd.read_csv(directoryPath + "/" + filename, low_memory=False, sep="$", error_bad_lines=False)
                out_df = out_df.groupby('primaryid')
                out_df = out_df.filter(lambda x: len(x) == 1)
                out_df = out_df[1:]
                outcome = outcome.append(out_df, ignore_index=True)

        demo['sex'] = np.where(pd.isnull(demo['sex']), demo['gndr_cod'], demo['sex'])
        demo.drop(['gndr_cod'], inplace=True, axis=1, errors='ignore')
        demo_durg_df = pd.merge(drug, demo, on=('primaryid', 'caseid'), how='left')
        demodurgreact_df = pd.merge(demo_durg_df, reaction, on=('primaryid', 'caseid'), how='inner')
        demodrugreactout_df = pd.merge(demodurgreact_df, outcome, on=('primaryid', 'caseid'), how='inner')
        demodrugreactout_df.drop(['drug_rec_act'], inplace=True, axis=1, errors='ignore')
        demodrugreactout_df['occp_cod'] = demodrugreactout_df['occp_cod'].fillna('OT')
        demodrugreactout_df['rept_cod'] = demodrugreactout_df['rept_cod'].fillna('EXP')
        demodrugreactout_df['mfr_sndr'] = demodrugreactout_df['mfr_sndr'].fillna('Others')
        demodrugreactout_df['route'] = demodrugreactout_df['route'].fillna('Unknown')
        demodrugreactout_df['dose_form'] = demodrugreactout_df['dose_form'].fillna('Others')
        demodrugreactout_df['dose_freq'] = demodrugreactout_df['dose_freq'].fillna('Others')
        demodrugreactout_df.to_csv(self.output().path, header=True, index=False);

    def output(self):
        return luigi.LocalTarget('MergedFile.csv')


if __name__ == '__main__':
    source_dir = "FAERSsrc"
    data_dir = "FAERSdata"
    if not os.path.isdir(source_dir):
        os.makedirs(source_dir)
    if not os.path.isdir(data_dir):
        os.makedirs(data_dir)
    luigi.run()

Answer 1

问题已解决，完整代码见GitHub：https://github.com/Judenpech/...