# pipe2py modules package
# Author: Greg Gaughan

# Note: each module name must match the name used internally by Yahoo, preceded
# by pipe

from functools import partial
from itertools import imap, repeat
from twisted.internet.defer import maybeDeferred, inlineCallbacks, returnValue
from pipe2py.lib import utils
from pipe2py.lib.dotdict import DotDict
from pipe2py.twisted.utils import asyncReturn, asyncNone, asyncBroadcast

__all__ = [
    # Source Modules
    'pipecsv',
    'pipefeedautodiscovery',
    'pipefetch',
    'pipefetchdata',
    'pipefetchpage',
    'pipefetchsitefeed',
    'pipeforever',
    'pipeitembuilder',
    'piperssitembuilder',
    'pipexpathfetchpage',
    'pipeyql',

    # User Input Modules
    'pipenumberinput',
    'pipeprivateinput',
    'pipetextinput',
    'pipeurlinput',
    # 'pipedateinput',
    # 'pipelocationinput',
    # 'pipeprivateinput',

    # Operator Modules
    'pipecount',
    'pipecreaterss',
    'pipefilter',
    'pipeloop',
    'piperegex',
    'piperename',
    'pipereverse',
    'pipesort',
    'pipesplit',
    'pipesubelement',
    'pipetail',
    'pipetruncate',
    'pipeunion',
    'pipeuniq',
    # 'pipewebservice',
    # 'pipelocationextractor',

    # URL Modules
    'pipeurlbuilder',

    # String Modules
    'pipeexchangerate',
    'pipehash',
    'pipestrconcat',
    'pipestrregex',
    'pipestrreplace',
    'pipestringtokenizer',
    'pipestrtransform',
    'pipesubstr',
    # 'pipetermextractor',
    # 'pipetranslate',
    # 'pipeyahooshortcuts',
    # 'pipestrprivate',

    # Date Modules
    'pipedatebuilder',
    'pipedateformat',

    # Location Modules
    # 'pipelocationbuilder',

    # Number Modules
    'pipesimplemath',
    'pipecurrencyformat',

    # Output Modules
    'pipeoutput',
    # 'pipeoutputjson',
    # 'pipeoutputical',
    # 'pipeoutputkml',
    # 'pipeoutputcsv',
]


def _get_broadcast_funcs(pieces, ftype='with', **kwargs):
    test = kwargs.pop('pass_if', None)
    listize = kwargs.pop('listize', True)
    parse = kwargs.pop('parse', True)
    get_value = partial(utils.get_value, **kwargs)
    get_pass = partial(utils.get_pass, test=test)
    get_with = partial(utils.get_with, **kwargs)
    parse_conf = partial(utils.parse_conf, parse_func=get_value, **kwargs)
    func = parse_conf if parse else get_value

    if listize:
        piece_defs = map(DotDict, utils.listize(pieces))
        get_pieces = lambda item: imap(func, piece_defs, repeat(item))
    else:
        piece_defs = DotDict(pieces)
        get_pieces = partial(func, piece_defs)

    return (get_pieces, get_with, get_pass)


def get_async_broadcast_funcs(pieces, ftype='with', **kwargs):
    funcs = _get_broadcast_funcs(pieces, ftype, **kwargs)
    get_pieces, get_with, get_pass = funcs

    f = {
        'with': partial(maybeDeferred, get_with),
        'pass': asyncReturn,
        None: lambda item: asyncNone
    }

    asyncGetPieces = partial(maybeDeferred, get_pieces)
    asyncGetPass = partial(maybeDeferred, get_pass)
    return [asyncGetPieces, f[ftype], asyncGetPass]


def get_broadcast_funcs(pieces, ftype='with', **kwargs):
    funcs = _get_broadcast_funcs(pieces, ftype, **kwargs)
    get_pieces, get_with, get_pass = funcs
    f = {'with': get_with, 'pass': utils.passthrough, None: utils.passnone}
    return [get_pieces, f[ftype], get_pass]


def get_dispatch_funcs(ftype='word', first=utils.passthrough):
    f = {
        'word': utils.get_word,
        'num': utils.get_num,
        'pass': utils.passthrough,
        None: utils.passnone,
    }

    return [first, f[ftype], utils.passthrough]


def get_async_dispatch_funcs(ftype='word', first=asyncReturn):
    f = {
        'word': partial(maybeDeferred, utils.get_word),
        'num': partial(maybeDeferred, utils.get_num),
        'pass': asyncReturn,
        None: lambda item: asyncNone,
    }

    return [first, f[ftype], asyncReturn]


def get_splits(_INPUT, pieces=None, finitize=False, funcs=None, **kwargs):
    finite = utils.finitize(_INPUT) if finitize and _INPUT else _INPUT
    inputs = imap(DotDict, finite) if finite else None
    funcs = funcs or get_broadcast_funcs(pieces, **kwargs)
    return utils.broadcast(inputs, *funcs) if inputs else funcs


@inlineCallbacks
def asyncGetSplits(_INPUT, pieces, finitize=False, **kwargs):
    _input = yield _INPUT
    # asyncDict = partial(maybeDeferred, DotDict)
    # inputs = yield asyncCmap(asyncDict, _input)
    finite = utils.finitize(_input) if finitize and _input else _input
    inputs = imap(DotDict, finite) if finite else None
    funcs = get_async_broadcast_funcs(pieces, **kwargs)

    if inputs:
        result = yield asyncBroadcast(inputs, *funcs)
    else:
        result = yield asyncReturn(funcs)

    returnValue(result)

@@ -2059,19 +2059,89 @@
     
-get_value =
+cust_func = kwargs.pop('cust_func', False)%0A    get_value = None if cust_func else
  par
@@ -2275,22 +2275,88 @@
 gs)%0A
+%0A
     
-parse_conf
+if cust_func:%0A        get_func = cust_func%0A    elif parse:%0A        get_func
  = p
@@ -2420,39 +2420,32 @@
     
-func = parse_conf if parse else
+else:%0A        get_func =
  get
@@ -2564,16 +2564,20 @@
 m: imap(
+get_
 func, pi
@@ -2676,16 +2676,20 @@
 partial(
+get_
 func, pi

Add custom function support