Salmon Run: feeds

Showing posts with label feeds. Show all posts

Sunday, January 09, 2011

Feed Aggregation using ROME Fetcher and Nutch 2.0 - Part I

With libraries such as ROME, it is possible for an application to parse RSS and Atom feeds without having to worry about the actual feed format itself. The ROME Fetcher subproject provides a FeedFetcher class that can convert a remote URL into a ROME SyndFeed object. David Herron has built a (the only so far?) ROME based Feed Aggregator using Groovy, inspired by the PlanetPlanet feed reader.

My goals are much less ambitious, all I need is a way to aggregate news from various newspaper feeds, and make that aggregate searchable. My aggregator is written in Java using ROME and ROME Fetcher. It reads the list of sites from an OPML file (an idea I borrowed from David Herron's aggregator), downloads and parses each page, and writes the feed entry information (the link URL and its metadata such as title and description) into a flat file. This flat file is then injected into HBase using Nutch 2.0, and we go through a single cycle of generate, fetch and parse. At this point, we are ready to index - unfortunately, Nutch 2.0 only supports indexing to Solr, so I need a running Solr installation, which I don't have as yet.

The list of sites comes from the NewsRack project - they are doing interesting things with a combined concept/rule based approach to classification. They were kind enough to post a list of RSS feeds from various Indian newspapers that they use as part of their work. Here is what my sites.xml file looks like:

<?xml version="1.0"?>
<opml version="1.1">
  <head>
    <title>Feed Aggregator Test</title>
    <ownerName>Sujit Pal</ownerName>
    <description>India Newspapers Feed (Source:newsrack.in)</description>
  </head>
  </meta>
  <body>
    <outline text="Business Today" 
      siteurl="http://businesstoday.digitaltoday.in" 
      xmlUrl="http://businesstoday.digitaltoday.in/rssforallsec.php?issueid=14"/>
    <outline text="Business World" 
      siteurl="http://www.businessworld.in" 
      xmlUrl="http://www.businessworld.in/component/option,com_rss/feed,RSS2.0/no_html,1/"/>
    <!-- more feed sites ... -->
  </body>
</opml>

The code for the aggregator, as expected, is quite simple. Parse out the configuration file above, and hit each of the URLs in it, then parse the returned page to retrieve the set of (link, title, description, pubDate) tuples from it, and write them out to a flat file in the structured format described below.

// Source: src/main/java/com/mycompany/myagg/fetcher/FeedAggregator.java
package com.mycompany.myagg.fetcher;

import java.io.File;
import java.io.FileWriter;
import java.io.PrintWriter;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.httpclient.ConnectTimeoutException;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jdom.Document;
import org.jdom.Element;
import org.jdom.input.SAXBuilder;

import com.sun.syndication.feed.synd.SyndEntry;
import com.sun.syndication.feed.synd.SyndFeed;
import com.sun.syndication.fetcher.FetcherException;
import com.sun.syndication.fetcher.impl.HashMapFeedInfoCache;
import com.sun.syndication.fetcher.impl.HttpClientFeedFetcher;
import com.sun.syndication.io.FeedException;

/**
 * Standalone Feed crawler that can be launched as a standalone command
 * (usually from cron). Configured using an OPML file that lists the feed
 * URLs to be crawled. Links on each feed along with their metadata are
 * written to a structured flat file to inject into Nutch.
 */
public class FeedAggregator {

  private final Log log = LogFactory.getLog(getClass());
  
  private static final int DEFAULT_CONNECT_TIMEOUT = 5000;
  private static final int DEFAULT_READ_TIMEOUT = 5000;
  private static final SimpleDateFormat ISO_DATE_FORMATTER =
    new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS");
  
  private String configFile;
  private String seedFile;
  private int connectTimeout;
  private int readTimeout;
  
  public void setConfigFile(String configFile) {
    this.configFile = configFile;
  }
  
  public void setSeedFile(String seedFile) {
    this.seedFile = seedFile;
  }
  
  public void setConnectTimeout(int connectTimeout) {
    this.connectTimeout = connectTimeout;
  }

  public void setReadTimeout(int readTimeout) {
    this.readTimeout = readTimeout;
  }

  @SuppressWarnings("unchecked")
  public void aggregate() throws Exception {
    AggregatorConfig config = parseConfig();
    HttpClientFeedFetcher feedFetcher = buildFeedFetcher(config);
    PrintWriter seedWriter = new PrintWriter(new FileWriter(seedFile), true);
    for (SiteInfo site : config.sites) {
      try {
        SyndFeed feed = feedFetcher.retrieveFeed(new URL(site.feedUrl));
        if (feed == null) {
          continue;
        }
        List<SyndEntry> entries = feed.getEntries();
        for (SyndEntry entry : entries) {
          String title = entry.getTitle();
          String link = entry.getLink();
          String description = entry.getDescription() != null ?
            entry.getDescription().getValue() : "";
          Date pubDate = entry.getPublishedDate();
          store(seedWriter, title, link, description, pubDate);
        }
      } catch (FetcherException e) {
        log.info("Fetcher Exception for " + site.publisher, e);
        continue;
      } catch (FeedException e) {
        log.info("Feed Exception for " + site.publisher, e);
        continue;
      } catch (ConnectTimeoutException e) {
        log.info("Connect Timeout Exception for " + site.publisher, e);
        continue;
      }
    }
    seedWriter.flush();
    seedWriter.close();
  }
  
  private HttpClientFeedFetcher buildFeedFetcher(AggregatorConfig config) {
    HttpClientFeedFetcher feedFetcher = new HttpClientFeedFetcher();
    feedFetcher.setUserAgent(config.title + "(" + config.owner + ")");
    feedFetcher.setConnectTimeout(connectTimeout == 0 ? 
      DEFAULT_CONNECT_TIMEOUT : connectTimeout);
    feedFetcher.setReadTimeout(readTimeout == 0 ? 
      DEFAULT_READ_TIMEOUT : readTimeout);
    feedFetcher.setFeedInfoCache(HashMapFeedInfoCache.getInstance());
    return feedFetcher;
  }

  private void store(PrintWriter seedWriter, String title, String link, 
      String description, Date pubDate) throws Exception {
    if (pubDate == null) {
      pubDate = new Date();
    }
    seedWriter.println(StringUtils.join(new String[] {
      link,
      StringUtils.join(new String[] {"u_idx", "news"}, "="),
      StringUtils.join(new String[] {
        "u_title",
        clean(title)
      }, "="),
      StringUtils.join(new String[] {
        "u_summary",
        clean(description)
      }, "="),
      StringUtils.join(new String[] {
        "u_pubdate", 
        ISO_DATE_FORMATTER.format(pubDate)
      }, "=")
    }, "\t"));
  }
  
  private String clean(String str) {
    if (StringUtils.isEmpty(str)) {
      return "";
    }
    return str.replaceAll("\\s+", " ");
  }

  @SuppressWarnings("unchecked")
  protected AggregatorConfig parseConfig() throws Exception {
    AggregatorConfig config = new AggregatorConfig();
    SAXBuilder builder = new SAXBuilder();
    Document doc = builder.build(new File(configFile));
    Element opml = doc.getRootElement();
    config.title = opml.getChild("head").getChildTextTrim("title");
    config.owner = opml.getChild("head").getChildTextTrim("ownerName");
    config.description = opml.getChild("head").getChildTextTrim("description");
    List<Element> outlines = opml.getChild("body").getChildren("outline");
    for (Element outline : outlines) {
      SiteInfo siteInfo = new SiteInfo();
      siteInfo.publisher = outline.getAttributeValue("text");
      siteInfo.siteUrl = outline.getAttributeValue("siteurl");
      siteInfo.feedUrl = outline.getAttributeValue("xmlUrl");
      config.sites.add(siteInfo);
    }
    return config;
  }
  
  protected class AggregatorConfig {
    public String title;
    public String owner;
    public String description;
    public List<SiteInfo> sites = new ArrayList<SiteInfo>();
  }
  
  protected class SiteInfo {
    public String publisher;
    public String siteUrl;
    public String feedUrl;
  }
  
  public static void main(String[] args) {
    if (args.length != 1) {
      System.out.println(
        "Usage: FeedAggregator config_file.xml connect_timeout " + 
        "read_timeout seed_file.txt");
      System.exit(-1);
    }
    try {
      FeedAggregator aggregator = new FeedAggregator();
      aggregator.setConfigFile(args[0]);
      aggregator.setConnectTimeout(Integer.valueOf(args[1]));
      aggregator.setReadTimeout(Integer.valueOf(args[2]));
      aggregator.setSeedFile(args[3]);
      aggregator.aggregate();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}

Running this with the appropriate parameters produces a flat file, a line of which is shown below. This is a tab-separated list of name-value pairs which has been broken into multiple lines below for readability. The first part is the familiar seed URL. The rest of the fields are seed URL metadata name-value pairs. Here we choose to add the index name (assuming we will be storing multiple content types in our HBase table), the title, summary and publication date, which should be used to populate the corresponding index column instead of trying to parse these values from the HTML.

http://www.indiatogether.org/2011/jan/env-brahm.htm\
 u_idx=news\
 u_title=Mine-ing the Brahmaputra's waters\
 u_summary=India and China make competing plans for the river's \
precious waters, ignoring the functions it already performs - in sustaining \
rich ecosystems, flora and fauna, cultures and a wide range of livelihoods. \
Shripad Dharmadhikary reports.\
 u_pubdate=2011-01-06T16:42:56.623
...

Note that this is new-ish functionality - the Nutch Injector supports adding custom metadata in this way since version 1.1, so this feature is usable with a file based crawling system also. In our HBase backed Nutch, the metadata live inside the "mtdt" column family after injection.

We then go through one cycle of generate, fetch and index, since we want to crawl only the pages referenced in the feed XMLs (ie depth 0). The end product of this exercise is, of course, an index with the metadata in it. Nutch 2.0 has made the decision to go with Solr as its search front end, so the only implementation of an indexer job is the SolrIndexerJob, which needs to be pointed at a running Solr installation. I hope to set up Solr and try this part out (and write about it) next week.

Saturday, October 13, 2007

Custom Modules with ROME

I have been looking at ROME lately. ROME is a very popular open-source Java based RSS/Atom Feed parsing and generation library, originally developed by a group of Sun developers. I originally looked at ROME as a way to parse external feeds. Although the feeds were either in RSS or Atom, there are various versions of both RSS and Atom, which are quite different from each other in subtle ways. ROME abstracts away all the differences, allowing you to treat them as high level Java objects. ROME uses JDOM, my favorite XML parsing library, to do the parsing and building of XML behind the scenes.

My first application using ROME was to parse and aggregate a bunch of external feeds, and was ridiculously simple, about half a page of Java code. As Mark Woodman says in his article "Rome in a Day: Parse and Publish Feeds in Java" on XML.com, the sheer variety of RSS and Atom flavors are enough to make a grown programmer cry. However, the way in which ROME abstracts away all these variations and the simplicity of my resulting code almost made me cry... tears of joy.

However, given that most of the "smarts" of the application would be in the selection of the feeds themselves, and since that required domain expertise I did not have, I decided to put that project aside for a while and explore the other part of ROME, trying to use it to build a feed instead. In any case, in retrospect, I would probably be looking at using the rome-fetcher subproject instead to build the aggregator, since that already provides code to build "well-behaved" feed fetchers.

The feed I choose to rebuild was an existing RSS 2.0 feed. It was generated using JSP XML templates powered by Java services. As such, there were many custom extensions built in, which were not part of any of the standard modules that ROME uses. So I basically had to build a custom extension module for ROME to support the custom tags that this feed used. This post describes the (pretty simple) process.

I could not find any place where this process was described in sufficient newbie detail for me, so after quickly looking through the RSS/Atom/ROME books on Amazon, I settled on "RSS and Atom in Action" (RAIA) by Dave Johnson. One reason for my choice was that it was published by Manning and they provide downloadable PDF versions free with their printed book, so I got my book about 15 minutes after I ordered from my home computer. As it turned out, I was done reading the PDF version by the time the print edition arrived about 5 days later. The print version sits unopened on my desk for now, but I am sure I will need it some day.

But enough idle chatter. Lets get right down to building a custom module that supports three of my custom tags, called my:tag, my:tagDate and my:isTagged respectively. The first is a String, the second a Date and the third a Boolean.

Each module has four components that needs to be built and hooked up with ROME. An interface that describes the URI for the namespace in which the custom tags will live and the getters and setters for each custom tag supported, an implementation of that interface, a parser and a generator. They are shown below:

// MyModule.java
package com.mycompany.feeds.modules.mymodule;

import java.util.Date;

import com.sun.syndication.feed.module.Module;

public interface MyModule extends Module {
  
  public static final String URI = "http://www.my.com/spec";

  public String getTag();
  public void setTag(String tag);
  public Date getTagDate();
  public void setTagDate(Date tagDate);
  public Boolean getIsTagged();
  public void setIsTagged(Boolean isTagged);
}

// MyModuleImpl.java
package com.mycompany.feeds.modules.mymodule;

import java.util.Date;

import com.sun.syndication.feed.module.ModuleImpl;

public class MyModuleImpl extends ModuleImpl implements MyModule {

  private static final long serialVersionUID = -8275118704842545845L;

  private Boolean isTagged;
  private Date tagDate;
  private String tag;
  
  // boilerplate code. Eclipse will generate all but the constructor but
  // will keep reporting an error until you do it.
  public MyModuleImpl() {
    super(MyModule.class, MyModule.URI);
  }

  public void copyFrom(Object obj) {
    MyModule module = (MyModule) obj;
    setTag(module.getTag());
    setTagDate(module.getTagDate());
    setIsTagged(module.getIsTagged());
  }

  public Class getInterface() {
    return MyModule.class;
  }

  // getter and setter impls for MyModule interface
  public Boolean getIsTagged() {
    return isTagged;
  }

  public String getTag() {
    return tag;
  }

  public Date getTagDate() {
    return tagDate;
  }

  public void setIsTagged(Boolean isTagged) {
    this.isTagged = isTagged;
  }

  public void setTag(String tag) {
    this.tag = tag;
  }

  public void setTagDate(Date tagDate) {
    this.tagDate = tagDate;
  }
}

// MyModuleGenerator.java
package com.mycompany.feeds.modules.mymodule;

import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

import org.jdom.Element;
import org.jdom.Namespace;

import com.sun.syndication.feed.module.Module;
import com.sun.syndication.io.ModuleGenerator;

public class MyModuleGenerator implements ModuleGenerator {

  // boilerplate code
  private static final Namespace NAMESPACE = Namespace.getNamespace("my", MyModule.URI);
  private static final Set NAMESPACES;
  static {
    Set<Namespace> namespaces = new HashSet<Namespace>();
    namespaces.add(NAMESPACE);
    NAMESPACES = Collections.unmodifiableSet(namespaces);
  }

  public String getNamespaceUri() {
    return MyModule.URI;
  }

  public Set getNamespaces() {
    return NAMESPACES;
  }

  // Implements the module generation logic
  private final SimpleDateFormat dateFormat = new SimpleDateFormat("MM/dd/yyyy");

  public void generate(Module module, Element element) {
    MyModule myModule = (MyModule) module;
    if (myModule.getTag() != null) {
      Element myElement = new Element("tag", NAMESPACE);
      myElement.setText(myModule.getTag());
      element.addContent(myElement);
    }
    if (myModule.getTagDate() != null) {
      Element myElement = new Element("tagDate", NAMESPACE);
      myElement.setText(dateFormat.format(myModule.getTagDate()));
      element.addContent(myElement);
    }
    if (myModule.getIsTagged() != null) {
      Element myElement = new Element("isTagged", NAMESPACE);
      myElement.setText(String.valueOf(myModule.getIsTagged()));
      element.addContent(myElement);
    }
  }
}

// MyModuleParser.java
package com.mycompany.feeds.modules.mymodule;

import java.text.ParseException;
import java.text.SimpleDateFormat;

import org.jdom.Element;
import org.jdom.Namespace;

import com.sun.syndication.feed.module.Module;
import com.sun.syndication.io.ModuleParser;

public class MyModuleParser implements ModuleParser {

  // boilerplate
  public String getNamespaceUri() {
    return MyModule.URI;
  }

  // implements the parsing for MyModule
  private final SimpleDateFormat dateFormat = new SimpleDateFormat("MM/dd/yyyy");

  public Module parse(Element element) {
    Namespace myNamespace = Namespace.getNamespace(MyModule.URI);
    MyModule myModule = new MyModuleImpl();
    if (element.getNamespace().equals(myNamespace)) {
      if (element.getName().equals("tag")) {
        myModule.setTag(element.getTextTrim());
      }
      if (element.getName().equals("tagDate")) {
        try {
          myModule.setTagDate(dateFormat.parse(element.getTextTrim()));
        } catch (ParseException e) {
          // don't set it if bad date format
        }
      }
      if (element.getName().equals("isTagged")) {
        myModule.setIsTagged(Boolean.valueOf(element.getTextTrim()));
      }
    }
    return myModule;
  }
}

To let ROME know that these modules should be used, we need to create a rome.properties file in our classpath. The ROME JAR file already has rome.properties files within it that controls its default configuration, and it will read our rome.properties in addition to its own configuration. Our three tags are all item level tags, so we will need to only map the parsers to the item level. The RAIA book shows you how to map feed level modules as well, but the process is quite similar. Here is my rome.properties file (it would be located in src/main/resources in a Maven2 project). I am building an RSS 2.0 feed, so I map it to that dialect here.

# rome.properties
rss_2.0.item.ModuleParser.classes=\
com.mycompany.feeds.modules.mymodule.MyModuleParser

rss_2.0.item.ModuleGenerator.classes=\
com.mycompany.feeds.modules.mymodule.MyModuleGenerator

Finally, here is how I would call it from within my code:

  @Test
  public void testFeedWithMyModule() throws Exception {
    // create the feed object
    SyndFeed feed = new SyndFeedImpl();
    feed.setFeedType("rss_2.0");
    feed.setTitle("My test feed");
    ...
    // add the MyModule namespace to the feed
    feed.getModules().add(new MyModuleImpl());

    // create the item
    SyndEntry entry = new SyndEntryImpl();
    ...
    // create the module, populate and add to the item
    MyModule myModule = new MyModuleImpl();
    myModule.setTag("tagValue");
    myModule.setTagDate(new Date());
    myModule.setIsTagged(true);
    entry.getModules().add(myModule);
    ...
    // add entry(s) to the feed
    feed.getEntries().add(entry);
    // print it out
    WireFeedOutput output = new WireFeedOutput();
    WireFeed wireFeed = feed.createWireFeed("rss_2.0");
    log.debug(output.outputString(wireFeed));
  }

As you can see, adding custom modules to ROME is a bit involved, but its really quite simple. Before I started on these two projects, I did not know much about all the various flavors of ROME and about these custom extensions. In fact, this is the first time I have used Namespaces in JDOM. However, Dave Johnson's book provides a lot of background information and a lot of nice examples in Java and C#. I would highly recommend it to anyone who needs to get up to speed quickly with ROME and RSS/Atom. Another very informative article is the article "ROME: Escape syndication hell, a Java developer's perspective" written by two of the original developers of ROME.

Salmon Run

Sunday, January 09, 2011

Feed Aggregation using ROME Fetcher and Nutch 2.0 - Part I

Saturday, October 13, 2007

Custom Modules with ROME

Posts

Labels

Blogs I Read

About me

My Nerd Rating

Visitor Map

Contact Me