-- ----------------------------------------------------------------------------

{- |
  Module     : Holumbus.Build.Crawl
  Copyright  : Copyright (C) 2008 Sebastian M. Schlatt
  License    : MIT
  
  Maintainer : Sebastian M. Schlatt (sms@holumbus.org)
  Stability  : experimental
  Portability: portable
  Version    : 0.1
  
  Indexer functions

-}

-- -----------------------------------------------------------------------------
{-# OPTIONS -fglasgow-exts #-}
-- -----------------------------------------------------------------------------

module Examples.MapReduce.Crawler.Index 
  (
    indexerOccurrencesAction
  , indexerBuildIndexAction
  -- * Building indexes
  , buildIndex
  -- , buildIndexM
  
  -- * Indexer Configuration
  , IndexerConfig (..)
  -- , ContextConfig (..)
  -- , mergeIndexerConfigs
  
  -- , getTexts
  )

where

import           Data.List
import           Data.Binary hiding (Word)
--import           Holumbus.Common.MRBinary
import qualified Data.Map    as M
import qualified Data.IntMap as IM
import qualified Data.IntSet as IS
import           Data.Maybe

-- import           Control.Exception
import           Control.Monad
import           Control.Parallel.Strategies

import           Holumbus.Build.Config
-- import           Holumbus.Control.MapReduce.ParallelWithClass 
import           Holumbus.Index.Common
import           Holumbus.Index.Inverted.Memory
import           Holumbus.Utility

import           System.Time

import           Text.XML.HXT.Arrow hiding (getXPathTrees)     -- import all stuff for parsing, validating, and transforming XML

import qualified Holumbus.FileSystem.FileSystem as FS
import           Holumbus.MapReduce.Types
import           Holumbus.MapReduce.MapReduce

import           Examples.MapReduce.Crawler.Config



indexerOccurrencesAction
  :: MRCrawlerConfig d a
  -> ActionConfiguration 
       ()                         -- state
       DocId String               -- k1, v1
       (Context,Word) Occurrences -- k2, v2
       Occurrences                -- v3 == v2
       Occurrences                -- v4
indexerOccurrencesAction cc
  = (defaultActionConfiguration "INDEX_OCCURRENCES")
        { ac_Map     = Just mapAction
        , ac_Combine = Nothing
        , ac_Reduce  = Just reduceAction
        }
    where
      mapAction 
        = (defaultMapConfiguration (mapIndex cc))
      reduceAction
        = (defaultReduceConfiguration (reduceOccurrences))


mapIndex 
  :: MRCrawlerConfig d a
  -> ActionEnvironment -> ()
  -> DocId -> String
  -> IO [((Context, Word), Occurrences)]
mapIndex cc env _ docId theUri
  = do
    let idxConfig      = cc_IndexerConfig cc
        traceLevel     = cc_TraceLevel cc
        fromTmp        = (isJust $ ic_tempPath idxConfig)
        contextConfigs = (ic_contextConfigs idxConfig)
        attrs          = (ic_readAttributes idxConfig)
        fileSystem     = ae_FileSystem env
    computeOccurrences traceLevel fileSystem fromTmp contextConfigs attrs {- cache -} docId theUri


reduceOccurrences 
  :: ActionEnvironment -> ()
  -> (Context, Word) -> [Occurrences]
  -> IO (Maybe Occurrences)
reduceOccurrences _ _ _ os
 = do
   let os' = (IM.unionsWith IS.union os)
   return $ Just $ os'


indexerBuildIndexAction
  :: ActionConfiguration 
       ()                                -- state
       () ((Context, Word),Occurrences)  -- k1, v1
       () ((Context, Word),Occurrences)  -- k2, v2
       ((Context, Word),Occurrences)     -- v3 == v2
       Inverted                          -- v4
indexerBuildIndexAction
  = (defaultActionConfiguration "INDEX_BUILD")
        { ac_Map     = Nothing
        , ac_Combine = Nothing
        , ac_Reduce  = Just reduceAction
        }
    where
      reduceAction
        = (defaultReduceConfiguration (reduceBuildIndex))


reduceBuildIndex 
  :: ActionEnvironment -> ()
  -> () -> [((Context, Word),Occurrences)]
  -> IO (Maybe Inverted)
reduceBuildIndex _ _ _ os
  = do
    let i   = emptyInverted
    let idx = foldl (\i' ((c,w),o) -> insertOccurrences c w o i') i os
        _   = rnf idx
    return $ Just $ idx 


-- -----------------------------------------------------------------------------

buildIndex :: (HolDocuments d a, {- HolIndex i -} MapReduce mr {-, HolCache c -}) 
           => MRCrawlerConfig d a
           --   Int                -- ^ Number of parallel threads for MapReduce
           --   Int                -- ^ TraceLevel for Arrows
           -> d a                -- ^ List of input Data
           -- -> IndexerConfig      -- ^ Configuration for the Indexing process
           -- -> Inverted           -- ^ An empty HolIndex. This is used to determine which kind of index to use.
           -> mr
           -> IO [Inverted]        -- ^ returns a HolIndex
buildIndex config {- workerThreads traceLevel -} docs {- idxConfig emptyIndex -} mr
  = do
    let docs' =  (map (\(i,d) -> (i, uri d)) (IM.toList $ toMap docs))
    
    runX (traceMsg 1 (" run indexer phase 1: "))
                     
    (res,_) <- doMapReduce (indexerOccurrencesAction config) () docs' [] 1 5 1 1 TOTRawTuple mr 
       
    runX (traceMsg 1 (" result of phase 1: "))       
    
    runX (traceMsg 1 (" num of tuples: "))
    runX (traceMsg 1 (show $ length res))

    let os' = map (\t -> ((),t)) res

    runX (traceMsg 1 (" run indexer phase 2: "))
                      
    (res',_) <- doMapReduce (indexerBuildIndexAction) () os' [] 1 1 1 1 TOTRawTuple mr
    
    runX (traceMsg 1 (" result of the indexer: "))       
    
    runX (traceMsg 1 (" num of indexes: "))
    runX (traceMsg 1 (show $ length res'))
    
    let idx = map (snd) res'
    
    return idx


{-
buildIndexM :: (HolDocuments d a, HolIndexM m i, HolCache c) => 
              Int                -- ^ Number of parallel threads for MapReduce
           -> Int                -- ^ TraceLevel for Arrows
           -> d a                -- ^ List of input Data
           -> IndexerConfig      -- ^ Configuration for the Indexing process
           -> i                  -- ^ An empty HolIndexM. This is used to determine which kind of index to use.
           -> Maybe c            -- ^ Just HolCache switches cache construction on
           -> IO i               -- ^ returns a HolIndexM
buildIndexM workerThreads traceLevel docs idxConfig emptyIndex cache
  = let docs' =  (map (\(i,d) -> (i, uri d)) (IM.toList $ toMap docs)) in
       -- assert ((sizeWords emptyIndex) == 0) 
                 (mapReduce 
                    workerThreads
--                    (ic_indexerTimeOut idxConfig)
--                    (( fromMaybe "/tmp/" (ic_tempPath idxConfig)) ++ "MapReduce.db")
                    emptyIndex
                    (computeOccurrences traceLevel 
                              (isJust $ ic_tempPath idxConfig)
                              (ic_contextConfigs idxConfig) 
                              (ic_readAttributes idxConfig)
                              cache
                    )
                    docs'
                 )
-}


-- | The MAP function in a MapReduce computation for building indexes.
--   The first three parameters have to be passed to the function to receive
--   a function with a valid MapReduce-map signature.
--
--   The function optionally outputs some debug information and then starts
--   the processing of a file by passing it together with the configuration
--   for different contexts to the @processDocument@ function where the file
--   is read and then the interesting parts configured in the
--   context configurations are extracted.

computeOccurrences :: -- HolCache c =>
               Int -> FS.FileSystem -> Bool -> [ContextConfig] -> Attributes -- -> Maybe c  
            -> DocId -> String -> IO [((Context, Word), Occurrences)]
computeOccurrences traceLevel fileSystem fromTmp contextConfigs attrs {- cache -} docId theUri
    = do
      clt <- getClockTime
      cat <- toCalendarTime clt
      res <- runX (     setTraceLevel traceLevel
                    >>> traceMsg 1 ((calendarTimeToString cat) ++ " - indexing document: " 
                                                               ++ show docId ++ " -> "
                                                               ++ show theUri)
                    >>> processDocument traceLevel fileSystem attrs' contextConfigs {- cache -} docId theUri
--                    >>> arr (\ (c, w, d, p) -> (c, (w, d, p)))
                    >>> strictA
             )
      return $ buildPositions res 
    where
    attrs' = if fromTmp then addEntries standardReadTmpDocumentAttributes attrs else attrs
    buildPositions :: [(Context, Word, DocId, Position)] -> [((Context, Word),  Occurrences)]
    buildPositions l = M.foldWithKey (\(c,w,d) ps acc -> ((c,w),IM.singleton d ps) : acc) [] $
      foldl (\m (c,w,d,p) -> M.insertWith IS.union (c,w,d) (IS.singleton p) m) M.empty l 
                             

-- -----------------------------------------------------------------------------
    
-- | Downloads a document and calls the function to process the data for the
--   different contexts of the index
processDocument :: -- HolCache c =>  
     Int
  -> FS.FileSystem
  -> Attributes
  -> [ContextConfig]
--  -> Maybe c
  -> DocId 
  -> URI
  -> IOSLA (XIOState s) b (Context, Word, DocId, Position)
processDocument traceLevel fileSystem attrs ccs {- cache -} docId theUri =
        withTraceLevel (traceLevel - traceOffset) 
        -- (readDocument attrs theUri)        
        (arrIO (\_ ->  do
          c <- FS.getFileContent theUri fileSystem
          case c of
            (Just cs)   -> return $ decode cs
            (Nothing) -> return "" -- TODO mark error here
          )
        )        
    >>> (readFromString attrs)
    >>> (catA $ map (processContext {- cache -} docId) ccs )   -- process all context configurations  
    


-- | Process a Context. Applies the given context configuration to extract information from
--   the XmlTree that is passed in the arrow.
processContext :: 
--  ( HolCache c) => 
--     Maybe c
     DocId
  -> ContextConfig
  -> IOSLA (XIOState s) XmlTree (Context, Word, DocId, Position)

processContext {-cache-} docId cc
    = cc_preFilter cc                                         -- convert XmlTree
      >>>
      fromLA extractWords
      >>>
{-      ( if ( isJust cache
       &&
       cc_addToCache cc
     )
  then perform (arrIO storeInCache)
  else this
      )
      >>>
-}    arrL genWordList
      >>>
      strictA
    where
    extractWords  :: LA XmlTree [String]
    extractWords
      = listA
        ( xshow ( (cc_fExtract cc)                           -- extract interesting parts
            >>>
            getTexts
          )
          >>>
          arrL ( filter (not . null) . cc_fTokenize cc )
        )

    genWordList   :: [String] -> [(Context, Word, DocId, Position)]
    genWordList
      = zip [1..]                                           -- number words
        >>>                                                 -- arrow for pure functions
        filter (not . (cc_fIsStopWord cc) . snd)            -- delete boring words
        >>>
        map ( \ (p, w) -> (cc_name cc, w, docId, p) )       -- attach context and docId
{-
    storeInCache s
      = let t = unwords s in 
            if t /= "" then putDocText (fromJust cache) (cc_name cc) docId t
                       else return()
-}

getTexts  :: LA XmlTree XmlTree
getTexts                                                      -- select all text nodes
    =  choiceA
       [ isElem :-> ( space                                   -- substitute tags by a single space
          <+>                                     -- so tags act as word delimiter
          (getChildren >>> getTexts)
          <+>
          space
        )                                       -- tags are interpreted as word delimiter
       , isText :-> this              -- take the text nodes
       , this   :-> none              -- ignore anything else
       ]
    where
    space = txt " "