# -*- coding: utf-8 -*-
# vim: sw=4:ts=4:expandtab
"""
    pipe2py.modules.pipefetch
    ~~~~~~~~~~~~~~~~~~~~~~~~~
    Provides methods for fetching RSS feeds.

    http://pipes.yahoo.com/pipes/docs?doc=sources#FetchFeed
"""


import speedparser

from functools import partial
from itertools import repeat, imap, ifilter
from urllib2 import urlopen
from twisted.web.client import getPage
from twisted.internet.defer import inlineCallbacks, maybeDeferred, returnValue
from pipe2py.lib import utils
from pipe2py.lib.dotdict import DotDict
from pipe2py.twisted.utils import asyncImap


# Common functions
def get_abs_urls(_INPUT, conf, **kwargs):
    url_defs = map(DotDict, utils.listize(conf['URL']))
    get_value = partial(utils.get_value, **kwargs)
    get_urls = lambda i: imap(get_value, url_defs, repeat(i))

    finite = utils.make_finite(_INPUT)
    inputs = imap(DotDict, finite)
    urls = imap(get_urls, inputs)
    flat_urls = utils.multiplex(urls)
    true_urls = ifilter(None, flat_urls)
    abs_urls = imap(utils.get_abspath, true_urls)
    return abs_urls


# Async functions
# from http://blog.mekk.waw.pl/archives/
# 14-Twisted-inlineCallbacks-and-deferredGenerator.html
# http://code.activestate.com/recipes/277099/
@inlineCallbacks
def asyncParse(url, context=None):
    if context and context.verbose:
        print "pipe_fetch loading:", url

    content = yield getPage(url)
    parsed = yield maybeDeferred(speedparser.parse, content)
    results = utils.gen_entries(parsed)
    returnValue(results)


@inlineCallbacks
def asyncPipeFetch(context=None, _INPUT=None, conf=None, **kwargs):
    """A source that asynchronously fetches and parses one or more feeds to
    return the feed entries. Loopable.

    Parameters
    ----------
    context : pipe2py.Context object
    _INPUT : asyncPipe like object (twisted Deferred iterable of items)
    conf : {'URL': [{'value': <url>, 'type': 'url'}]}

    Returns
    -------
    _OUTPUT : twisted.internet.defer.Deferred generator of items
    """
    _input = yield _INPUT
    abs_urls = get_abs_urls(_input, conf, **kwargs)
    items = yield asyncImap(asyncParse, abs_urls, repeat(context))
    _OUTPUT = utils.multiplex(items)
    returnValue(_OUTPUT)


# Synchronous functions
def parse(url, context=None):
    if context and context.verbose:
        print "pipe_fetch loading:", url

    content = urlopen(url).read()
    parsed = speedparser.parse(content)
    results = utils.gen_entries(parsed)
    return results


def pipe_fetch(context=None, _INPUT=None, conf=None, **kwargs):
    """A source that fetches and parses one or more feeds to return the
    entries. Loopable.

    Parameters
    ----------
    context : pipe2py.Context object
    _INPUT : pipeforever pipe or an iterable of items or fields
    conf : {'URL': [{'value': <url>, 'type': 'url'}]}

    Returns
    -------
    _OUTPUT : generator of items
    """
    abs_urls = get_abs_urls(_INPUT, conf, **kwargs)
    items = imap(parse, abs_urls, repeat(context))
    _OUTPUT = utils.multiplex(items)
    return _OUTPUT

@@ -599,28 +599,24 @@
 ons%0Adef get_
-abs_
 urls(_INPUT,
@@ -629,24 +629,98 @@
  **kwargs):%0A
+    finite = utils.make_finite(_INPUT)%0A    inputs = imap(DotDict, finite)%0A
     url_defs
@@ -879,83 +879,8 @@
 (i))
-%0A%0A    finite = utils.make_finite(_INPUT)%0A    inputs = imap(DotDict, finite)
 %0A   
@@ -956,19 +956,14 @@
     
-true_urls =
+return
  ifi
@@ -988,79 +988,8 @@
 ls)%0A
-    abs_urls = imap(utils.get_abspath, true_urls)%0A    return abs_urls%0A%0A
 %0A# A
@@ -1150,298 +1150,8 @@
 99/%0A
-@inlineCallbacks%0Adef asyncParse(url, context=None):%0A    if context and context.verbose:%0A        print %22pipe_fetch loading:%22, url%0A%0A    content = yield getPage(url)%0A    parsed = yield maybeDeferred(speedparser.parse, content)%0A    results = utils.gen_entries(parsed)%0A    returnValue(results)%0A%0A
 %0A@in
@@ -1665,28 +1665,24 @@
  _INPUT%0A    
-abs_
 urls = get_a
@@ -1676,28 +1676,24 @@
  urls = get_
-abs_
 urls(_input,
@@ -1717,266 +1717,107 @@
     
-items = yield asyncImap(asyncParse, abs_urls, repeat(context))%0A    _OUTPUT = utils.multiplex(items)%0A    returnValue(_OUTPUT)%0A%0A%0A# Synchronous functions%0Adef parse(url, context=Non
+asyncParse = partial(maybeDeferred, speedparser.pars
 e)
-:
 %0A    
-if context and context.verbose:%0A        print %22pipe_fetch loading:%22
+abs_urls = imap(utils.get_abspath
 , url
-%0A
+s)
 %0A   
@@ -1828,85 +1828,115 @@
 tent
+s
  = 
-urlopen(url).read()%0A    parsed = speedparser.p
+yield asyncImap(getPage, abs_urls)%0A    parsed = yield asyncImap(asyncP
 arse
-(
+, 
 content
+s
 )%0A    
-results = 
+items = imap(
 util
@@ -1948,17 +1948,18 @@
 _entries
-(
+, 
 parsed)%0A
@@ -1966,24 +1966,91 @@
     
-return results%0A%0A
+_OUTPUT = utils.multiplex(items)%0A    returnValue(_OUTPUT)%0A%0A%0A# Synchronous functions
 %0Adef
@@ -2457,28 +2457,24 @@
     %22%22%22%0A    
-abs_
 urls = get_a
@@ -2472,20 +2472,16 @@
 s = get_
-abs_
 urls(_IN
@@ -2509,53 +2509,201 @@
     
-items = imap(parse, abs_urls, repeat(context)
+abs_urls = imap(utils.get_abspath, urls)%0A    contents = imap(lambda url: urlopen(url).read(), abs_urls)%0A    parsed = imap(speedparser.parse, contents)%0A    items = imap(utils.gen_entries, parsed
 )%0A  

Optimize to fetch/parse urls cooperatively