https://github.com/grangier/python-goose
Revision 383eb8e5588f1b913b32b83a6a903af8380494e9 authored by Xavier Grangier on 16 June 2013, 08:23:01 UTC, committed by Xavier Grangier on 16 June 2013, 08:23:01 UTC
Improved article tag extraction
Tip revision: 383eb8e5588f1b913b32b83a6a903af8380494e9 authored by Xavier Grangier on 16 June 2013, 08:23:01 UTC
Merge pull request #20 from litso/improved_article_tags
Merge pull request #20 from litso/improved_article_tags
Tip revision: 383eb8e
extractors.py
# -*- coding: utf-8 -*-
"""\
This is a python port of "Goose" orignialy licensed to Gravity.com
under one or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.
Python port was written by Xavier Grangier for Recrutae
Gravity.com licenses this file
to you under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import re
from copy import deepcopy
from urlparse import urlparse, urljoin
from goose.utils import StringSplitter
from goose.utils import StringReplacement
from goose.utils import ReplaceSequence
from goose.parsers import Parser
MOTLEY_REPLACEMENT = StringReplacement("�", "")
ESCAPED_FRAGMENT_REPLACEMENT = StringReplacement(u"#!", u"?_escaped_fragment_=")
TITLE_REPLACEMENTS = ReplaceSequence().create(u"»").append(u"»")
PIPE_SPLITTER = StringSplitter("\\|")
DASH_SPLITTER = StringSplitter(" - ")
ARROWS_SPLITTER = StringSplitter("»")
COLON_SPLITTER = StringSplitter(":")
SPACE_SPLITTER = StringSplitter(' ')
NO_STRINGS = set()
A_REL_TAG_SELECTOR = "a[rel=tag]"
A_HREF_TAG_SELECTOR = "a[href*='/tag/'], a[href*='/tags/']"
RE_LANG = r'^[A-Za-z]{2}$'
class ContentExtractor(object):
def __init__(self, config):
self.config = config
self.language = config.target_language
self.stopwords_class = config.stopwords_class
def get_language(self, article):
"""\
Returns the language is by the article or
the configuration language
"""
# we don't want to force the target laguage
# so we use the article.meta_lang
if self.config.use_meta_language == True:
if article.meta_lang:
self.language = article.meta_lang[:2]
self.language = self.config.target_language
def get_title(self, article):
"""\
Fetch the article title and analyze it
"""
title = ''
doc = article.doc
title_element = Parser.getElementsByTag(doc, tag='title')
# no title found
if title_element is None or len(title_element) == 0:
return title
# title elem found
title_text = Parser.getText(title_element[0])
used_delimeter = False
# split title with |
if '|' in title_text:
title_text = self.split_title(title_text, PIPE_SPLITTER)
used_delimeter = True
# split title with -
if not used_delimeter and '-' in title_text:
title_text = self.split_title(title_text, DASH_SPLITTER)
used_delimeter = True
# split title with »
if not used_delimeter and u'»' in title_text:
title_text = self.split_title(title_text, ARROWS_SPLITTER)
used_delimeter = True
# split title with :
if not used_delimeter and ':' in title_text:
title_text = self.split_title(title_text, COLON_SPLITTER)
used_delimeter = True
title = MOTLEY_REPLACEMENT.replaceAll(title_text)
return title
def split_title(self, title, splitter):
"""\
Split the title to best part possible
"""
large_text_length = 0
large_text_index = 0
title_pieces = splitter.split(title)
# find the largest title piece
for i in range(len(title_pieces)):
current = title_pieces[i]
if len(current) > large_text_length:
large_text_length = len(current)
large_text_index = i
# replace content
title = title_pieces[large_text_index]
return TITLE_REPLACEMENTS.replaceAll(title).strip()
def get_favicon(self, article):
"""\
Extract the favicon from a website
http://en.wikipedia.org/wiki/Favicon
<link rel="shortcut icon" type="image/png" href="favicon.png" />
<link rel="icon" type="image/png" href="favicon.png" />
"""
kwargs = {'tag': 'link', 'attr': 'rel', 'value': 'icon'}
meta = Parser.getElementsByTag(article.doc, **kwargs)
if meta:
favicon = meta[0].attrib.get('href')
return favicon
return ''
def get_meta_lang(self, article):
"""\
Extract content language from meta
"""
# we have a lang attribute in html
attr = Parser.getAttribute(article.doc, attr='lang')
if attr is None:
# look up for a Content-Language in meta
items = [
{'tag': 'meta', 'attr': 'http-equiv', 'value': 'content-language'},
{'tag': 'meta', 'attr': 'name', 'value': 'lang'}
]
for item in items:
meta = Parser.getElementsByTag(article.doc, **item)
if meta:
attr = Parser.getAttribute(meta[0], attr='content')
break
if attr:
value = attr[:2]
if re.search(RE_LANG, value):
self.language = value.lower()
return value.lower()
return None
def get_meta_content(self, doc, metaName):
"""\
Extract a given meta content form document
"""
meta = Parser.css_select(doc, metaName)
content = None
if meta is not None and len(meta) > 0:
content = meta[0].attrib.get('content')
if content:
return content.strip()
return ''
def get_meta_description(self, article):
"""\
if the article has meta description set in the source, use that
"""
return self.get_meta_content(article.doc, "meta[name=description]")
def get_meta_keywords(self, article):
"""\
if the article has meta keywords set in the source, use that
"""
return self.get_meta_content(article.doc, "meta[name=keywords]")
def get_canonical_link(self, article):
"""\
if the article has meta canonical link set in the url
"""
if article.final_url:
kwargs = {'tag': 'link', 'attr': 'rel', 'value': 'canonical'}
meta = Parser.getElementsByTag(article.doc, **kwargs)
if meta is not None and len(meta) > 0:
href = meta[0].attrib.get('href')
if href:
href = href.strip()
o = urlparse(href)
if not o.hostname:
z = urlparse(article.final_url)
domain = '%s://%s' % (z.scheme, z.hostname)
href = urljoin(domain, href)
return href
return article.final_url
def get_domain(self, url):
if url:
o = urlparse(url)
return o.hostname
return None
def extract_tags(self, article):
node = article.doc
# node doesn't have chidren
if len(list(node)) == 0:
return NO_STRINGS
elements = Parser.css_select(node, A_REL_TAG_SELECTOR)
if not elements:
elements = Parser.css_select(node, A_HREF_TAG_SELECTOR)
if not elements:
return NO_STRINGS
tags = []
for el in elements:
tag = Parser.getText(el)
if tag:
tags.append(tag)
return set(tags)
def calculate_best_node(self, article):
doc = article.doc
top_node = None
nodes_to_check = self.nodes_to_check(doc)
starting_boost = float(1.0)
cnt = 0
i = 0
parent_nodes = []
nodes_with_text = []
for node in nodes_to_check:
text_node = Parser.getText(node)
word_stats = self.stopwords_class(language=self.language).get_stopword_count(text_node)
high_link_density = self.is_highlink_density(node)
if word_stats.get_stopword_count() > 2 and not high_link_density:
nodes_with_text.append(node)
nodes_number = len(nodes_with_text)
negative_scoring = 0
bottom_negativescore_nodes = float(nodes_number) * 0.25
for node in nodes_with_text:
boost_score = float(0)
# boost
if(self.is_boostable(node)):
if cnt >= 0:
boost_score = float((1.0 / starting_boost) * 50)
starting_boost += 1
# nodes_number
if nodes_number > 15:
if (nodes_number - i) <= bottom_negativescore_nodes:
booster = float(bottom_negativescore_nodes - (nodes_number - i))
boost_score = float(-pow(booster, float(2)))
negscore = -abs(boost_score) + negative_scoring
if negscore > 40:
boost_score = float(5)
text_node = Parser.getText(node)
word_stats = self.stopwords_class(language=self.language).get_stopword_count(text_node)
upscore = int(word_stats.get_stopword_count() + boost_score)
# parent node
parent_node = Parser.getParent(node)
self.update_score(parent_node, upscore)
self.update_node_count(node.getparent(), 1)
if node.getparent() not in parent_nodes:
parent_nodes.append(node.getparent())
# parentparent node
parent_parent_node = Parser.getParent(parent_node)
if parent_parent_node is not None:
self.update_node_count(parent_parent_node, 1)
self.update_score(parent_parent_node, upscore / 2)
if parent_parent_node not in parent_nodes:
parent_nodes.append(parent_parent_node)
cnt += 1
i += 1
top_node_score = 0
for e in parent_nodes:
score = self.get_score(e)
if score > top_node_score:
top_node = e
top_node_score = score
if top_node is None:
top_node = e
return top_node
def is_boostable(self, node):
"""\
alot of times the first paragraph might be the caption under an image
so we'll want to make sure if we're going to boost a parent node that
it should be connected to other paragraphs,
at least for the first n paragraphs so we'll want to make sure that
the next sibling is a paragraph and has at
least some substatial weight to it
"""
para = "p"
steps_away = 0
minimum_stopword_count = 5
max_stepsaway_from_node = 3
nodes = self.walk_siblings(node)
for current_node in nodes:
# p
if current_node.tag == para:
if steps_away >= max_stepsaway_from_node:
return False
paraText = Parser.getText(current_node)
word_stats = self.stopwords_class(language=self.language).get_stopword_count(paraText)
if word_stats.get_stopword_count() > minimum_stopword_count:
return True
steps_away += 1
return False
def walk_siblings(self, node):
current_sibling = Parser.previousSibling(node)
b = []
while current_sibling is not None:
b.append(current_sibling)
previousSibling = Parser.previousSibling(current_sibling)
current_sibling = None if previousSibling is None else previousSibling
return b
def add_siblings(self, top_node):
baselinescore_siblings_para = self.get_siblings_score(top_node)
results = self.walk_siblings(top_node)
for current_node in results:
ps = self.get_siblings_content(current_node, baselinescore_siblings_para)
for p in ps:
top_node.insert(0, p)
return top_node
def get_siblings_content(self, current_sibling, baselinescore_siblings_para):
"""\
adds any siblings that may have a decent score to this node
"""
if current_sibling.tag == 'p' and len(Parser.getText(current_sibling)) > 0:
e0 = current_sibling
if e0.tail:
e0 = deepcopy(e0)
e0.tail = ''
return [e0]
else:
potential_paragraphs = Parser.getElementsByTag(current_sibling, tag='p')
if potential_paragraphs is None:
return None
else:
ps = []
for first_paragraph in potential_paragraphs:
text = Parser.getText(first_paragraph)
if len(text) > 0:
word_stats = self.stopwords_class(language=self.language).get_stopword_count(text)
paragraph_score = word_stats.get_stopword_count()
sibling_baseline_score = float(.30)
high_link_density = self.is_highlink_density(first_paragraph)
score = float(baselinescore_siblings_para * sibling_baseline_score)
if score < paragraph_score and not high_link_density:
p = Parser.createElement(tag='p', text=text, tail=None)
ps.append(p)
return ps
def get_siblings_score(self, top_node):
"""\
we could have long articles that have tons of paragraphs
so if we tried to calculate the base score against
the total text score of those paragraphs it would be unfair.
So we need to normalize the score based on the average scoring
of the paragraphs within the top node.
For example if our total score of 10 paragraphs was 1000
but each had an average value of 100 then 100 should be our base.
"""
base = 100000
paragraphs_number = 0
paragraphs_score = 0
nodes_to_check = Parser.getElementsByTag(top_node, tag='p')
for node in nodes_to_check:
text_node = Parser.getText(node)
word_stats = self.stopwords_class(language=self.language).get_stopword_count(text_node)
high_link_density = self.is_highlink_density(node)
if word_stats.get_stopword_count() > 2 and not high_link_density:
paragraphs_number += 1
paragraphs_score += word_stats.get_stopword_count()
if paragraphs_number > 0:
base = paragraphs_score / paragraphs_number
return base
def update_score(self, node, addToScore):
"""\
adds a score to the gravityScore Attribute we put on divs
we'll get the current score then add the score
we're passing in to the current
"""
current_score = 0
score_string = node.attrib.get('gravityScore')
if score_string:
current_score = int(score_string)
new_score = current_score + addToScore
node.set("gravityScore", str(new_score))
def update_node_count(self, node, add_to_count):
"""\
stores how many decent nodes are under a parent node
"""
current_score = 0
count_string = node.attrib.get('gravityNodes')
if count_string:
current_score = int(count_string)
new_score = current_score + add_to_count
node.set("gravityNodes", str(new_score))
def is_highlink_density(self, e):
"""\
checks the density of links within a node,
is there not much text and most of it contains linky shit?
if so it's no good
"""
links = Parser.getElementsByTag(e, tag='a')
if links is None or len(links) == 0:
return False
text = Parser.getText(e)
words = text.split(' ')
words_number = float(len(words))
sb = []
for link in links:
sb.append(Parser.getText(link))
linkText = ''.join(sb)
linkWords = linkText.split(' ')
numberOfLinkWords = float(len(linkWords))
numberOfLinks = float(len(links))
linkDivisor = float(numberOfLinkWords / words_number)
score = float(linkDivisor * numberOfLinks)
if score >= 1.0:
return True
return False
# return True if score > 1.0 else False
def get_score(self, node):
"""\
returns the gravityScore as an integer from this node
"""
return self.get_node_gravity_score(node) or 0
def get_node_gravity_score(self, node):
grvScoreString = node.attrib.get('gravityScore')
if not grvScoreString:
return None
return int(grvScoreString)
def nodes_to_check(self, doc):
"""\
returns a list of nodes we want to search
on like paragraphs and tables
"""
nodes_to_check = []
for tag in ['p', 'pre', 'td']:
items = Parser.getElementsByTag(doc, tag=tag)
nodes_to_check += items
return nodes_to_check
def is_table_and_no_para_exist(self, e):
subParagraphs = Parser.getElementsByTag(e, tag='p')
for p in subParagraphs:
txt = Parser.getText(p)
if len(txt) < 25:
Parser.remove(p)
subParagraphs2 = Parser.getElementsByTag(e, tag='p')
if len(subParagraphs2) == 0 and e.tag is not "td":
return True
return False
def is_nodescore_threshold_met(self, node, e):
top_node_score = self.get_score(node)
current_nodeScore = self.get_score(e)
thresholdScore = float(top_node_score * .08)
if (current_nodeScore < thresholdScore) and e.tag != 'td':
return False
return True
def post_cleanup(self, targetNode):
"""\
remove any divs that looks like non-content,
clusters of links, or paras with no gusto
"""
node = self.add_siblings(targetNode)
for e in node.getchildren():
if e.tag != 'p':
if self.is_highlink_density(e) \
or self.is_table_and_no_para_exist(e) \
or not self.is_nodescore_threshold_met(node, e):
Parser.remove(e)
return node
class StandardContentExtractor(ContentExtractor):
pass
Computing file changes ...