当前位置：网站首页>使用NLP和ML来提取和构造Web数据

使用NLP和ML来提取和构造Web数据

2020-11-06 01:28:00 【人工智能遇见磐创】

flutter + 小程序 + 后端，3个人就可开启你的APP之路 | 注册即送大疆云台、华为手表！>>>

作者|Conner Brew 编译|VK 来源|Towards Data Science

介绍

在本文中，我们将创建一个基于战争研究所（ISW）的结构化文档数据库。ISW为外交和情报专业人员提供信息产品，以加深对世界各地发生的冲突的了解。

要查看与本文相关联的原始代码和Notebook，请访问以下链接:https://colab.research.google.com/drive/1pTrOXW3k5VQo1lEaahCo79AHpyp5ZdfQ?usp=sharing

要访问Kaggle上托管的最终结构化数据集，请访问以下链接:https://www.kaggle.com/connerbrew2/isw-web-scrape-and-nlp-enrichment

本文将是一个关于web抽取、自然语言处理（NLP）和命名实体识别（NER）的练习。对于NLP，我们将主要使用开源Python库NLTK和Spacy。

本文旨在演示web提取和NLP的一个用例，而不是关于这两种技术使用的全面初学者教程。如果你是NLP或web提取的新手，我建议你遵循不同的教程，或者浏览Spacy、BeautifulSoup和NLTK文档页面。

# 导入库

import requests
import nltk
import math
import re
import spacy
import regex as re
import pandas as pd
import numpy as np
import statistics as stats
import matplotlib.pyplot as plt
import matplotlib.cm as cm
import json

# 你需要从NLTK下载一些包。

from bs4 import BeautifulSoup 
from nltk import *
nltk.download('stopwords')
nltk.download('punkt')
from nltk.corpus import stopwords

# #在大多数环境中，你需要安装NER-D。

!pip install ner-d
from nerd import ner

from sklearn.feature_extraction.text import CountVectorizer
from sklearn.feature_extraction.text import TfidfTransformer
from sklearn.cluster import MiniBatchKMeans
from sklearn.feature_extraction.text import TfidfVectorizer

初始化变量

首先，我们将初始化最终结构化数据中需要的数据字段。对于每个文档，我要提取标题、发布日期、人名、地名和其他各种信息。我们还将增强文档中已经存在的信息—例如，我们将使用文档中的地名来获取相关的坐标，这对于以后可视化数据非常有用。

# 初始化最终数据集的数据字段

dates=[]
titles=[]
locations=[]
people=[]
key_countries=[]
content_text=[]
links=[]
coord_list=[]
mentioned_countries=[]
keywords=[]
topic_categories=[]

# 为后面的主题模型初始化簇变量

cluster_keywords=[]
cluster_number=[]

# 使用SPACY库初始化NLP对象
nlp = spacy.load("en_core_web_sm")

提取href

我们将从ISW的生产库中提取文档。首先，我们将抓取“浏览”页面以获取每个产品的单独href链接。然后我们将这些链接存储在一个列表中，供提取函数稍后访问。

# #从ISW浏览页面获取产品链接

urls=['http://www.understandingwar.org/publications?page={}'.format(i) for i in range(179)]
hrefs=[]

def get_hrefs(page,class_name):
  page=requests.get(page)
  soup=BeautifulSoup(page.text,'html.parser')
  container=soup.find_all('div',{'class':class_name})
  container_a=container[0].find_all('a')
  links=[container_a[i].get('href') for i in range(len(container_a))]
  for link in links:
    if link[0]=='/':
      hrefs.append('http://www.understandingwar.org'+link)

for url in urls:
  get_hrefs(url,'view-content')

Web爬取

我们将要编写的前几个函数是相当简单的文本提取。本教程不是关于BeautifulSoup用法的教程，要了解Python中的web爬取，请查看这里的文档：https://www.crummy.com/software/BeautifulSoup/bs4/doc/

获得日期

对于我们的第一个函数，我们将提取发布日期。它扫描从产品网页中提取的html文档，并找到一个类为“submitted”的字段。这是我们的生产日期。

获得标题

接下来，我们需要产品名称。同样，这个字段被方便地标记为“title”类。

获取所有文本

最后，我们将提取文档的全文。当我提取文本时，我通常遵循“先提取，后过滤”的web提取方式。这意味着，在最初的文本提取中，我对文本执行最少的过滤和处理。我更愿意在以后的分析中进行处理，因为这是必要的。但是，如果你想更进一步，你可能希望对提取的文本进行比下面函数演示的更多的预处理。

对于我的get_contents函数，我坚持最基本的原则——我在黑名单中列出了一些不想被提取的文本。然后从页面中提取所有文本并将其附加到一个临时字符串中，该字符串又被附加到列表content_text中。

# 提取发布数据

def get_date(soup):
  try:
    data=soup.find('span',{'class':'submitted'})
    content=data.find('span')
    date=content.get('content')
    dates.append(date)
  except Exception:
    dates.append('')
    pass

# 提取产品标题
def get_title(soup):
  try:
    title=soup.find('h1',{'class':'title'}).contents
    titles.append(title[0])
  except Exception:
    titles.append('')
    pass

# 提取产品的文本内容
def get_contents(soup):
  try:
    parents_blacklist=['[document]','html','head',
                       'style','script','body',
                       'div','a','section','tr',
                       'td','label','ul','header',
                       'aside',]
    content=''
    text=soup.find_all(text=True)
  
    for t in text:
      if t.parent.name not in parents_blacklist and len(t) > 10:
        content=content+t+' '

    content_text.append(content)
  except Exception:
    content_text.append('')
    pass

自然语言处理

接下来，我们将找出产品中引用了哪些国家。有很多API可以用于检查国家的文本内容，但这里我们将使用一个简单的方法：列出世界上所有国家的列表。这个列表来自维基百科：https://en.wikipedia.org/wiki/Lists_of_countries_and_territories

在函数得到all_mentioned_countries 后，它使用基本统计分析来确定哪些国家最突出——这些国家最有可能成为文件叙述的焦点。为此，该函数计算整个文档中提到一个国家的次数，然后查找比平均值提到次数多的国家。然后将这些国家追加到key_countries列表中。

# 在正文中引用所有国家的列表。
# 如果文本中的一个单词与列表中的一个国家匹配，那么它将被添加到国家列表中。 

def get_countries(content_list):
  iteration=1
  for i in range(len(content_list)):
    print('Getting countries',iteration,'/',len(content_list))
    temp_list=[]
    for word in word_tokenize(content_list[i]):
      for country in country_list:
        if word.lower().strip() == country.lower().strip():
          temp_list.append(country)

    counted_countries=dict(Counter(temp_list))
    temp_dict=dict.fromkeys(temp_list,0)
    temp_list=list(temp_dict)
    if len(temp_list)==0:
      temp_list.append('Worldwide')
    mentioned_countries.append(temp_list)

    # 计算每个国家被提及的次数，然后对照平均值检查每次计数。 
    # 如果一个国家被提及的次数超过了平均次数，它就会作为一个关键字被记录。

    keywords=[]
    for key in counted_countries.keys():
      if counted_countries[key] > np.mean(list(counted_countries.values())):
        keywords.append(key)
    if len(keywords) != 0:
      key_countries.append(keywords)
    else:
      key_countries.append(temp_list)
    iteration+=1

命名实体识别：地点

接下来，我们要丰富我们的数据。最终，结构化数据的目标通常是执行某种分析或可视化——在这种国际冲突信息的情况下，将信息按地理位置绘制出来是很有价值的。为此，我们需要与文档对应的坐标。

找到地名

首先，我们将使用自然语言处理（NLP）和命名实体识别（NER）从文本中提取地名。

NLP是机器学习的一种形式，计算机算法使用语法和语法规则来学习文本中单词之间的关系。通过这种学习，NER能够理解某些单词在句子或段落中所起的作用。本教程并不打算全面介绍NLP—对于这样的资源，请查看：https://medium.com/@ODSC/an-introduction-to-natural-language-processing-nlp-8e476d9f5f59

从外部API获取坐标

为了找到地名的坐标，我们将使用 Open Cage API查询坐标；你可以在这里创建一个免费帐户并接收API密钥。还有许多其他流行的地理api可供选择，但通过反复试验，我发现Open-Cage在中东地区有着最好的性能。

首先，我们迭代从文档中检索到的每个地名，并在Open Cage中查询它。一旦完成这项工作，我们将对比Open Cage与先前创建的mentioned_countries 列表。这将确保我们检索的查询结果位于正确的位置。

# 使用NLP提取地名，然后查询open-cage API以获得绘图所需的坐标

# 插入你自己的OpenCage API key:
geo_api_key='Insert Your API Key Here'

def get_coords(content_list):
  iteration=1
  for i in range(len(content_list)):
    print('Getting coordinates',iteration,'/',len(content_list))
    temp_list=[]
    text=content_list[i]

    # 应用一个NER算法，从python库'ner-d'中查找地名。

    doc=nlp(text)
    location=[X.text for X in doc.ents if X.label_ == 'GPE']
    location_dict=dict.fromkeys(location,0)
    location=list(location_dict)


    # 查询位置。
    
    for l in location:
      try: 
        request_url='https://api.opencagedata.com/geocode/v1/json?q={}&key={}'.format(l,geo_api_key)
        page=requests.get(request_url)
        data=page.json()
        for n in range(len(data)):

          # 这行代码检查查询结果中的国家是否与mentioned_countries之一相匹配。如果不是，那么查询结果很可能是假正例。

          if data['results'][n]['components']['country'] in mentioned_countries[i]:
            lat=data['results'][n]['geometry']['lat']
            lng=data['results'][n]['geometry']['lng']
            coordinates={'Location': l,
                          'Lat': lat,
                          'Lon': lng}
            temp_list.append(coordinates)
            break
          else:
            continue    
      except Exception: 
        continue
    coord_list.append(temp_list)
    iteration+=1

命名实体识别：人

接下来，我们将提取文档中提到的人的姓名。为此，我们将再次使用NER-d python库中的NER算法。

获取全名

在最终的结构化数据中，我只想要全名。只找到“Jack”或“John”的数据，会不会令人困惑？为此，我们将再次使用一些基本的统计数据。当提到全名时，函数将跟踪全名，通常是在文本的开头。

当后面提到部分名称时，它将引用全名列表，以标识部分名称引用的是谁。例如，如果一篇新闻文章是这样写的：“乔·拜登正在竞选总统。乔是前总统奥巴马的副总统，我们知道乔指的是拜登，因为他的全名在文中早些时候已经给出。此函数将以相同的方式运行。

重复的名字

如果出现了重复的情况，该函数将使用前面用于国家/地区函数的相同统计数据。它将测量一个名字被提及的次数，并将其作为最有可能的名字。例如：乔·拜登和他的儿子亨特·拜登都是受欢迎的美国政治家。乔·拜登是前副总统。拜登现在正在与现任总统唐纳德·特朗普竞选总统”。根据文本的统计重点，这篇文章显然是关于乔·拜登，而不是亨特·拜登。

验证名字

一旦函数计算出所有提到的全名，它将把它们添加到一个列表中。然后，它将查询维基百科中的每个名字，以验证它是否是值得包含在结构化数据中的有影响力的人的名字。

def get_people(content_list):
  iteration=1

  # 使用NER在文本中查找人名。
  
  for i in range(len(content_list)):
    print('Getting people',iteration,'/',len(content_list))
    temp_list=[]
    text=content_list[i]
    doc=nlp(text)
    persons=[X.text for X in doc.ents if X.label_ == 'PERSON']
    persons_dict=dict.fromkeys(persons,0)
    persons=list(persons_dict)

    full_names=[]
    for person in persons: 
      if len(word_tokenize(person)) >= 2:
        string_name=re.sub(r"[^a-zA-Z0-9]+", ' ', person).strip()
        full_names.append(string_name)
  
    final_names=[]
    for person in persons:
      for name in full_names:
        tokens=word_tokenize(name)
        for n in range(len(tokens)):
          if person==tokens[n]:
            final_names.append(name)

    for name in full_names:
      final_names.append(name)

    name_dict=dict.fromkeys(final_names,0)
    final_names=list(name_dict)
    valid_names=[]

    for name in final_names:
      page=requests.get('https://en.wikipedia.org/wiki/'+name)
      if page.status_code==200:
        valid_names.append(name)

    people.append(valid_names)
    iteration+=1

关键词提取：TF-IDF

我们的下一个任务是从文本中提取关键字。最常见的方法是使用一种称为TF-IDF的方法。TF-IDF模型测量单个文档中单词的使用频率，然后将其与整个文档语料库中的平均使用率进行比较。

如果一个术语在单个文档中频繁使用，并且很少在整个文档语料库中使用，那么该术语很可能表示该特定文档特有的关键字。这篇文章并不是一篇关于TF-IDF模型的全面概述。要了解更多信息，请查看这篇关于Medium的文章：https://medium.com/datadriveninvestor/tf-idf-in-natural-language-processing-8db8ef4a7736

首先，我们的函数将创建通常所说的“词袋”。这将跟踪每个文档中使用的每个单词。然后，它将计算每个文档中每个单词的每次使用次数—单词频率(TF)。然后，它计算逆文档频率(IDF)。然后将这些值写入矩阵中的坐标，然后对矩阵进行排序，以帮助我们找到最有可能表示文档的单词。

# 第一个函数通过降低字符大小写和删除特殊字符对文本进行预处理。

def pre_process(text):
    text=text.lower()
    text=re.sub("</?.*?>"," <> ",text)
    text=re.sub("(\\d|\\W)+"," ",text)
    return text

# 这个函数将矩阵映射到坐标。TF-IDF函数将频率分数映射到矩阵，然后需要对这些矩阵进行排序，以帮助我们找到关键字。 

def sort_coo(coo_matrix):
    tuples = zip(coo_matrix.col, coo_matrix.data)
    return sorted(tuples, key=lambda x: (x[1], x[0]), reverse=True)

# 与上面一样，这是一个帮助函数，一旦频率映射到矩阵，它将帮助排序和选择关键字。 
# 这个函数专门帮助我们根据TF-IDF统计数据选择最相关的关键字

def extract_topn_from_vector(feature_names, sorted_items, topn=10):
    sorted_items = sorted_items[:topn]
    score_vals = []
    feature_vals = []

    for idx, score in sorted_items:
        fname = feature_names[idx]
        score_vals.append(round(score, 3))
        feature_vals.append(feature_names[idx])

    results= {}
    for idx in range(len(feature_vals)):
        results[feature_vals[idx]]=score_vals[idx]
    return results

#最后一个函数包含了上述helper函数，它对正文应用TF-IDF算法，根据使用频率查找关键字。

def get_keywords(content_list):
  iteration=1
  processed_text=[pre_process(text) for text in content_list]
  stop_words=set(stopwords.words('english'))
  cv=CountVectorizer(max_df=0.85,stop_words=stop_words)
  word_count_vector=cv.fit_transform(processed_text)

  tfidf_transformer=TfidfTransformer(smooth_idf=True,use_idf=True)
  tfidf_transformer.fit(word_count_vector)

  feature_names=cv.get_feature_names()

  for i in range(len(processed_text)):
    print('Getting Keywords',iteration,'/',len(content_list))
    doc=processed_text[i]
    tf_idf_vector=tfidf_transformer.transform(cv.transform([doc]))
    sorted_items=sort_coo(tf_idf_vector.tocoo())
    keys=extract_topn_from_vector(feature_names,sorted_items,10)
    keywords.append(list(keys.keys()))
    iteration+=1

主题模型

NLP中最常见的任务之一就是主题模型。这是一种聚类形式，它尝试根据文档的文本内容自动对文档进行分类。在这个具体的例子中，我想一眼就知道ISW涉及哪些主题。通过根据文本内容对文档进行分类，我可以轻松地对文档的主要思想有一个大致的了解。

向量化

对于这个例子，我将使用k-means聚类算法来进行主题建模。首先，我将再次使用TF-IDF算法对每个文档进行向量化。向量化是一个机器学习术语，指的是将非数字数据转换成计算机可以用来执行机器学习任务的数字空间数据。

优化

一旦文档被向量化，helper函数就会检查簇的最佳数量。（k表示k-means的k）。在本例中，最佳数目是50。一旦我找到了最佳的数字，在这个例子中，我注释掉了这行代码，并手动将参数调整为等于50。这是因为我正在分析的数据集不会经常更改，所以我可以期望随着时间的推移，最佳簇的数量会保持不变。对于变化更频繁的数据，你应该返回最佳的簇数量作为变量-这将帮助你的聚类算法自动设置其最佳参数。我在我的时间序列分析文章中展示了一个例子。

聚类

每个簇完成后，我将每个簇的编号（1–50）保存到簇编号的列表中，而组成每个簇的关键字保存到cluster_keywords的列表中。这些簇关键字稍后将用于向每个主题簇添加标题。

# 该函数根据各种“k”参数检查聚类算法，以找到“k”的最优值。

def find_optimal_clusters(data, max_k):
    iters = range(2, max_k+1, 2)
    
    sse = []
    for k in iters:
        sse.append(MiniBatchKMeans(n_clusters=k, 
                                   init_size=1024, 
                                   batch_size=2048,
                                   random_state=20).fit(data).inertia_)
        
        print('Fit {} clusters'.format(k))
        
    f, ax = plt.subplots(1, 1)
    ax.plot(iters, sse, marker='o')
    ax.set_xlabel('Cluster Centers')
    ax.set_xticks(iters)
    ax.set_xticklabels(iters)
    ax.set_ylabel('SSE')
    ax.set_title('SSE by Cluster Center Plot')

# 从内容列表中获取关键词来帮助对主题模型的分类

def get_top_keywords(data, clusters, labels, n_terms):
  df = pd.DataFrame(data.todense()).groupby(clusters).mean()
  for i,r in df.iterrows():
    cluster_keywords.append(','.join([labels[t] for t in np.argsort(r)[-n_terms:]]))
    
# 应用于主题建模的内容列表

def get_topics(content_list):
  processed_text=[pre_process(text) for text in content_list]
  stop_words=set(stopwords.words('english'))
  cv=CountVectorizer(max_df=0.85,stop_words=stop_words)
  word_count_vector=cv.fit_transform(processed_text)

  tfidf_transformer=TfidfTransformer(smooth_idf=True,use_idf=True)
  tfidf_transformer.fit(word_count_vector)

  feature_names=cv.get_feature_names()
  vector=tfidf_transformer.transform(cv.transform(processed_text))

  #find_optimal_clusters(vector,50)

  clusters = MiniBatchKMeans(n_clusters=50, init_size=1024, batch_size=2048, random_state=20).fit_predict(vector)
  for cluster in clusters:
    cluster_number.append(int(cluster))
  
  get_top_keywords(vector, clusters, cv.get_feature_names(), 20)

放在一起

最后，我们将提取我们的数据。使用我们之前得到的href列表，现在是将所有提取函数应用于web内容的时候了。

# 遍历从“browse”中提取的href，提取相关内容

iteration=1

# 前几个函数依赖于原始提取的web内容作为参数。这些都是基本的web抓取技术。

for href in hrefs:
  print('Web scraping: iteration',iteration,'/',len(hrefs))
  page=requests.get(href)
  soup=BeautifulSoup(page.text,'html.parser')
  links.append(href)
  get_date(soup)
  get_title(soup)
  get_contents(soup)
  iteration+=1

# 下面这些函数依赖于文本主体作为参数。
# 这些是基于nlp的函数。

# 注意:由于查询外部API，
# 需要一个超时来阻止服务器过载。
# 这部分代码的运行时间很长。

get_countries(content_text)
get_coords(content_text)
get_people(content_text)
get_keywords(content_text)
get_topics(content_text)

丰富主题模型

我们的下一个问题是：我们的簇为我们提供了一个与每个簇相关联的单词列表，但是簇的名称仅仅是数字。这使我们有机会绘制一个词云或其他有趣的可视化图，可以帮助我们理解每个簇，但对于结构化数据集中的一目了然的理解来说，它并没有那么有用。另外，我认为有些文档可能属于多个主题类别。k-means不支持多重聚类，因此我必须手动识别这些文档。首先，我将打印前几行关键字，以了解我正在处理的数据。

与每个主题相关联的一些关键字。我们将使用这些关键字将簇分类到预定义的类别中。

在对各种技术进行了大量实验之后，我决定采用一种非常简单的方法。我扫描了与每个簇相关的每个关键字列表，并在每个与特定主题相关的关键字中记录了重要的关键字。在这个阶段，领域知识是关键。例如，我知道，ISW文件中的阿勒颇几乎肯定提到了叙利亚内战。对于你的数据，如果你缺乏适当的领域知识，你可能需要做进一步的研究，咨询你团队中的其他人，或者定义一个更高级的编程方法来命名簇。

然而，对于这个例子，简单的方法很有效。在记录了簇列表中存在的几个重要关键字之后，我自己制作了几个列表，其中包含了与结构化数据中我想要的最终主题类别相关联的关键字。该函数简单地将每个簇的关键字列表与我创建的列表进行比较，然后根据列表中的匹配项分配主题名称。然后将这些最后的主题附加到主题类别列表中。

#搜索与主题对应的关键词列表，与聚类词库交叉引用，为每篇文章分配一个主题类别。

oir=['OIR Iraq','yezidis','mosul','peshmerga','isis','iraq','sinjar','baghdad','maliki',
     'daquq','anbar','isf','abadi','malaki','ramadi','iraqi','fallujah','dabiq']

terrorism=['Terrorism','jihadi','islamic','salafi','qaeda',
           'caliphate','isis','terrorist','terrorism']

syrian_conflict=['Syrian Conflict','sana','syria','assad',
                 'idlib','afrin','aleppo']

russia=['Russia','russia','belarus','slavic','kremlin','russian',
        'minsk','ukraine','putin']

iran=['Iran','iran','iranian','proxy','militias','militia','marjah']

turkey=['Turkey','erdogan','turkish','turkey']

ors=['ORS','kabul','ghani','pakistan','afghan','afghanistan',
     'taliban','ansf','karzai','helmand']

africa=['Africa','libya','libyan','egypt','egyptian','africa','african']

cat_list=[oir,terrorism,syrian_conflict,russia,iran,turkey,ors,africa]

topic_dict={}

for i in range(len(cluster_keywords)):
  temp_list=[]
  for n in nltk.word_tokenize(cluster_keywords[i]):
    for item in cat_list:
      if n in item:
        temp_list.append(item[0])
  
  temp_dict=dict.fromkeys(temp_list,0)
  temp_list=list(temp_dict)

  topic_dict[i] = temp_list

for num in cluster_number:
  topic_categories.append(topic_dict[num])

数据库创建

最后一步是将我们提取的所有数据集中起来。对于这些数据，我更喜欢JSON格式。这是因为我想以不同的方式组织某些类型的数据—例如，locations字段将包含地名、纬度和经度的字典列表。在我看来，JSON格式是将这种格式化的数据存储到本地磁盘的最有效的方法。我还在文档数据库MongoDB中备份了这个数据库的副本，但这不是本文的重点。

#将一个空列表初始化

db=[]

for i in range(len(hrefs)):
  countries={
      'focus area': key_countries[i],
      'all mentioned countries': mentioned_countries[i]
  }

  # 将函数中定义的所有列表添加到新的存储列表中

  doc={ 
      '_id': len(hrefs) - i,
       'title': titles[i],
       'date': dates[i],
       'places': coord_list[i],
       'people': people[i],
       'keywords': keywords[i],
       'countries': countries,
       'full text': content_text[i],
       'url': links[i],
       'topic cluster': cluster_number[i],
       'categories': topic_categories[i]
  }

  db.append(doc)

# 将列表保存为谷歌驱动器内的.JSON数据存储文件(用于演示目的)

with open ('/content/drive/My Drive/Colab Notebooks/isw_products.json', 'w') as fout:
  json.dump(db, fout)