content: -- ----------------------------------------------------------------------------
  
  {- |
    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
    
    This module provides some functions to crawl Web Pages recursively
  
  -}
  
  -- ----------------------------------------------------------------------------
  {-# OPTIONS -fglasgow-exts -fno-warn-orphans #-}
  -- -----------------------------------------------------------------------------
  
  module Examples.MapReduce.Crawler.Crawl
    (
    -- * CrawlerState type
      CrawlerState (..)
  
    , crawlerAction  
  
    -- * Crawling
    , crawl
    
    -- * Initializing
    , initialCrawlerState
  --  , getRefs
    
    -- * Persistence
    , loadCrawlerState
    , saveCrawlerState
    )
  where
  
  import           Control.Monad hiding (when)
  
  import           Data.Binary
  --import           Holumbus.Common.MRBinary
  import           Data.Char
  import           Data.List
  import           Data.Maybe
  import qualified Data.Map    as M
  import qualified Data.Set    as S
  -- import qualified Data.IntMap as IM
  import           Data.Digest.MD5
  import           Data.ByteString.Lazy.Char8(pack)
  
  import           Holumbus.Build.Config
  -- import           Holumbus.Control.MapReduce.ParallelWithClass -- Persistent
  -- import           Holumbus.Control.MapReduce.MapReducible
  
  import           Holumbus.Index.Common
  import           Holumbus.Utility
  
  import           Text.XML.HXT.Arrow hiding (getXPathTrees)
  import           Text.XML.HXT.Arrow.XPathSimple
  -- import           System.Directory
  import           System.Time
  
  import qualified Holumbus.FileSystem.FileSystem as FS
  import           Holumbus.MapReduce.Types
  import           Holumbus.MapReduce.MapReduce
  
  import           Examples.MapReduce.Crawler.Config
  
  -- import Control.DeepSeq
  
  type MD5Hash = String
  
  
  -- because of IO(Maybe v3) --> has to instanciate the typeclass show
  instance Show (CrawlerState d a) where
    show _ = "CrawlerState"
  
  -- ----------------------------------------------------------------------------
  -- Map-Action
  -- ----------------------------------------------------------------------------
  
  
  crawlerAction
    :: (Binary a, Show a, HolDocuments d a)
    => MRCrawlerConfig d a                           -- whitness and configuration
    -> ActionConfiguration 
         (Int, CrawlerState d a)                     -- state
         DocId String                                -- k1, v1
         () (MD5Hash, Maybe (Document a), S.Set URI) -- k2, v2
         (MD5Hash, Maybe (Document a), S.Set URI)    -- v3 == v2
         (CrawlerState d a)                          -- v4
  crawlerAction cc
    = (defaultActionConfiguration "CRAWL")
          { ac_Map     = Just mapAction
          , ac_Combine = Nothing
          , ac_Reduce  = Just reduceAction
          }
      where
        mapAction 
          = (defaultMapConfiguration (mapCrawl cc))           
        reduceAction
          = (defaultReduceConfiguration (reduceCrawl cc))
              { rc_Merge     = mergeCrawl }
  
  
  mapCrawl
    :: (Binary a, Show a, HolDocuments d a)
    => MRCrawlerConfig d a                 -- ^ static configuration
    -> ActionEnvironment
    -> (Int, CrawlerState d a)             -- ^ state
    -> DocId -> String                     -- ^ key - value
    -> IO [((), (MD5Hash, Maybe (Document a), S.Set URI))]
  mapCrawl cc env (traceLevel,_) docId theUri
    = do
      let cs         = cc_CrawlerState cc
          attrs      = cs_readAttributes cs 
          tmpPath    = cs_tempPath cs
          getRefs    = cs_fGetReferences cs
          getCustom  = cs_fGetCustom cs
          fileSystem = ae_FileSystem env
  
      runX (traceMsg 0 ("DocId: " ++ show docId ++ " - Uri: " ++ show theUri))
      -- putStrLn "################"
      -- putStrLn $ show attrs
      -- putStrLn "################"
      res <- crawlDoc traceLevel fileSystem attrs tmpPath getRefs getCustom docId theUri
      runX (traceMsg 0 ("Result: "))
      runX (traceMsg 0 (show res))
      return res
  
  
  -- ----------------------------------------------------------------------------
  -- Reduce-Action
  -- ----------------------------------------------------------------------------
  
  mergeCrawl
    :: (Binary a, Show a, HolDocuments d a)
    -- => CrawlerState d a         -- ^ type Whitness
    => ActionEnvironment
    -> (Int,CrawlerState d a)   -- ^ state
    -> [((), (MD5Hash, Maybe (Document a), S.Set URI))]
    -> IO [((), [(MD5Hash, Maybe (Document a), S.Set URI)])]
  mergeCrawl _ _ ls = return [((), ls')]
    where
    ls' = foldr (\(_,v) vs -> v:vs) [] ls
  
  
  reduceCrawl
    :: (Binary a, Show a, HolDocuments d a)
    => MRCrawlerConfig d a          -- ^ type Whitness 
    -> ActionEnvironment
    -> (Int,CrawlerState d a)    -- ^ state
    -> () -> [(MD5Hash, Maybe (Document a), S.Set URI)]
    -> IO (Maybe (CrawlerState d a))
  reduceCrawl cc _ (_,cs) k ls = processCrawlResults cs' cs k ls
    where
    cs' = cc_CrawlerState cc
  
  
  
  -- | The crawl function. MapReduce is used to scan the documents for references to other documents
  --   that have to be added to the 'Documents', too.
  crawl
   :: (Show a, HolDocuments d a, Binary a, Binary (d a), XmlPickler (d a)
      , MapReduce mr) 
   => MRCrawlerConfig d a                           -- whitness and configuration
   -> Int -> Int
   -> CrawlerState d a
   -> mr
   -> IO (d a)
  crawl config traceLevel maxDocs cs mr = 
    if S.null ( cs_toBeProcessed cs ) -- if no more documents have to be processed, 
      then do
        runX (traceMsg 1 (" no more documents to processed"))
        return (cs_docs cs)        -- the Documents are returned
      else do                         -- otherwise, the crawling is continued
         runX (traceMsg 1 (" new crawler-cycle"))
         -- get the Docs that have to be processed and add docIds
         -- for the MapReduce Computation
         d    <- return $ if maxDocs == 0 
                            then cs_toBeProcessed cs
                            else (S.fromList . (take maxDocs) . S.toList) (cs_toBeProcessed cs)
         d'   <- return $ zip [(cs_nextDocId cs)..] (S.toList d) -- (S.toList $ cs_toBeProcessed cs)
         runX (traceMsg 1 (" next docs to be pulled:"))
         runX (traceMsg 1 (show d'))
         
         -- saveCrawlerState ( (fromMaybe "/tmp/" (cs_tempPath cs) ) ++ "CrawlerState.bin") cs
         -- writeToXmlFile   ( (fromMaybe "/tmp/" (cs_tempPath cs) ) ++ "Docs.xml") (cs_docs cs)
                                      
         runX (traceMsg 1 ("          Status: already processed: " ++ show (S.size $ cs_wereProcessed cs) ++ 
                           ", to be processed: "   ++ show (S.size $ cs_toBeProcessed cs)))
         
         cs'  <- return $ cs { cs_nextDocId = cs_nextDocId cs + length d'
                             , cs_wereProcessed = S.union d (cs_wereProcessed cs)
                             , cs_toBeProcessed = S.difference (cs_toBeProcessed cs) d
                             }
  
         (res,_) <- doMapReduce (crawlerAction config) (traceLevel,cs') d' [] 1 5 1 1 TOTRawTuple mr
         
         runX (traceMsg 1 (" result of this cycle: "))       
         
         let (_,csnew) = head res
         
         runX (traceMsg 1 (show $ cs_toBeProcessed csnew))
                               
         crawl config traceLevel maxDocs csnew mr
  
  
  -- | The REDUCE function for the crawling MapReduce computation. The 'Documents' and their contained
  --   links are used to modify the 'CrawlerState'. New documents are added to the list of 
  --   unprocessed docs and the data of the already crawled documents are added to the 'Documents'. 
  processCrawlResults :: (HolDocuments d a, Binary a) => 
                         CrawlerState d a
                      -> CrawlerState d a                           -- ^ state before last MapReduce job
                      -> ()                                         -- ^
                      -> [(MD5Hash, Maybe (Document a), S.Set URI)] -- ^ data produced in the crawl phase
                      -> IO (Maybe (CrawlerState d a))
  processCrawlResults cc oldCs _ l = 
    do 
    cs' <- foldM process oldCs l
    runX (traceMsg 1 (" ----------------------"))
    runX (traceMsg 1 (" old cs: "))
    runX (traceMsg 1 (show $ cs_toBeProcessed oldCs))
    runX (traceMsg 1 (" ----------------------"))
    runX (traceMsg 1 (" new cs: "))
    runX (traceMsg 1 (show $ cs_toBeProcessed cs'))
    return $ Just cs'
    where 
      process :: (HolDocuments d a, Binary a) => 
                 CrawlerState d a -> (String, Maybe (Document a), S.Set URI) -> IO (CrawlerState d a)
      process cs (theMD5, mdoc, refs) 
        = if isJust mdoc
            then 
                 do
                 if isJust (cs_docHashes cs) && M.member theMD5 (fromJust $ cs_docHashes cs)
                   then do
                        let hashes = fromJust $ cs_docHashes cs
                        old  <- M.lookup theMD5 hashes
                        new  <- return $ uri $ fromJust mdoc
                        (newDocs, newHashes) <- update (cs_docs cs) hashes theMD5 old new
                        return cs { cs_docs = newDocs 
                                  , cs_toBeProcessed = S.union 
                                         (cs_toBeProcessed cs) 
                                         (S.difference (S.filter (cs_fCrawlFilter cc) refs) 
                                                       (cs_wereProcessed cs)
                                         )
                                  , cs_docHashes = Just $ newHashes
                                  }
                   else do
                        return $ 
                          cs { cs_toBeProcessed = S.union (cs_toBeProcessed cs) 
                                 (S.difference (S.filter (cs_fCrawlFilter cc) refs) 
                                               (cs_wereProcessed cs)
                                 )
                             , cs_docs      = snd (insertDoc (cs_docs cs) (fromJust mdoc))                                                
                             , cs_docHashes = if isJust (cs_docHashes cs)
                                                then Just $ M.insert theMD5 (uri $ fromJust mdoc) (fromJust $ cs_docHashes cs)
                                                else Nothing
                             }
            else return cs
      update :: (HolDocuments d a, Binary a) => 
                d a -> M.Map String URI -> String -> URI -> URI -> IO (d a, M.Map String URI)
      update docs hashes md5String oldUri newUri 
        = if length oldUri <= length newUri 
            then return $ (docs, hashes)
            else do 
                 docId  <- lookupByURI docs oldUri
                 oldDoc <- lookupById  docs docId
                 newDoc <- return oldDoc {uri = newUri}
                 return (updateDoc docs docId newDoc, M.insert md5String newUri hashes)
  
  
  
  -- | Wrapper function for the "real" crawlDoc functions. The current time is
  --   traced (to identify documents that take a lot of time to be processed while
  --   testing) and the crawlDoc'-arrow is run
  crawlDoc :: (Binary a, Show a) => 
              Int 
           -> FS.FileSystem
           -> Attributes
           -> Maybe String
           -> IOSArrow XmlTree [URI]
           -> Custom a
           -> DocId 
           -> String 
           -> IO [((), (MD5Hash, Maybe (Document a), S.Set URI))]
  crawlDoc traceLevel fileSystem attrs tmpPath getRefs getCustom docId theUri 
    = do
      clt <- getClockTime               -- get Clock Time for debugging
      cat <- toCalendarTime clt         -- and convert it to "real" date and time
      r <- runX (     setTraceLevel traceLevel
                 >>> traceMsg 1 (calendarTimeToString cat)  
                 >>> (     constA ()        -- this is needed to fit the MapReduce abstraction
                       &&& crawlDoc' traceLevel fileSystem attrs tmpPath getRefs getCustom (docId, theUri) 
                     )
                 >>^ (\(i, (h, d, u)) -> (i, (h, d, S.fromList u)))       
                )
  --    putStrLn (show r)
      return $ r 
  
  -- | Download & read document and compute document title and included refs to
  --   other documents 
  crawlDoc' :: (Binary a) =>
       Int
    -> FS.FileSystem
    -> Attributes    -- ^ options for readDocument
    -> Maybe String  -- ^ path for serialized tempfiles
    -> IOSArrow XmlTree [URI]
  --  -> [String]
    -> Custom a
    -> (DocId, URI)   -- ^ DocId, URI
    -> IOSArrow c (String, Maybe (Document a), [URI])
  crawlDoc' traceLevel fileSystem attrs tmpPath getRefs getCustom (docId, theUri) =
          traceMsg 1 ("  crawling document: " ++ show docId ++ " -> " ++ show theUri )
      >>> withTraceLevel (traceLevel - traceOffset) (readDocument attrs theUri)
      >>> (
            documentStatusOk `guards`   -- make sure that the document could be accessed
                  -- if it is set in the crawler options, write a temporary copy of the document to
                  -- the hard disk
            (     ( if isJust tmpPath
            then withTraceLevel (traceLevel - traceOffset) $
                          (replaceBaseElement $< computeDocBase)    -- make sure that there ist a <base> tag
                          -- a b c >>> a c d -> a b d
                          >>> -- writeDocument :: Attributes -> String -> IOStateArrow s XmlTree XmlTree
                              -- writeDocument standardWriteTmpDocumentAttributes
                              --         (fromJust tmpPath ++ tmpFile docId theUri)
                          
                              -- writeDocumentToString :: Attributes -> IOStateArrow s XmlTree String      
                              (writeDocumentToString standardWriteTmpDocumentAttributes
                               >>> 
                               --   arrIO :: (b -> IO c) -> a b c
                               arrIO (\s -> FS.createFile 
                                             ({- fromJust tmpPath ++ -} tmpFile docId theUri) 
                                             (encode s)
                                             fileSystem))
                               &&& (returnA)
                            >>> arr (\(_,t) -> t)
                          --IOStateArrow s XmlTree XmlTree
                          -- arr           b       c          arr          c      d      arr          b       d
                          --IOStateArrow s XmlTree String >>> IOStateArrow String () >>> IOSTateArrow XmlTree () 
                          --arr b c >>> arr c d >> arr b c
            else this
                  ) 
                  -- compute a pair of Document b (Holumbus datatype) and a list of contained links
              >>> 
                  (     (  xshow (this) >>^ (show . md5 . pack) )
                    &&& getDocument theUri getCustom
                    &&& ( getRefs >>> strictA >>> perform (theTrace $< arr length) )
                  )   >>^ (\(a,(b,c)) -> (a,b,c))
            )
            `orElse` (                  -- if an error occurs with the current document, the global 
                  traceMsg 0 (  "something went wrong with doc: \"" ++ theUri ++"\"")    
              >>> clearErrStatus        -- error status has to be reset, else the crawler would stop
              >>> constA (show ( md5 (pack "foo")) , Nothing, [])  -- Nothing indicates the error, the empty list shows
            )                           -- that - caused by the error - no new links were found
          )
          where 
            theTrace i = traceMsg 2 ("found " ++ (show i) ++ " references")
  
  
  -- | Replaces the <base> tag in the Document. If there is no <base> tag, a new one is inserted.
  -- This is especially usefull when saving the document to a local file. With a <base> tag
  -- links in the document can be properly resolved.
  replaceBaseElement :: ArrowXml a => String -> a XmlTree XmlTree
  replaceBaseElement base = processTopDownUntil ( (isElem >>> hasName "html") `guards` processHtml )
      where
      processHtml = processTopDownUntil ( (isElem >>> hasName "head") `guards` processHead)
      processHead = ifA (getChildren >>> isElem >>> hasName "base") 
                          (processTopDownUntil ( (isElem >>> hasName "base") `guards` processAttr) )
                          addBaseElem
      processAttr = processAttrl ( changeAttrValue (\_ -> base ) `when` hasName "href")
      addBaseElem = replaceChildren (getChildren <+> aelem "base" [sattr "href" base]) 
            
  
  -- | extract the Title of a Document (for web pages the <title> tag) and combine
  --   it with the document uri
  getDocument :: (Binary b) => 
                 URI        -- ^ URI of the 'Document' 
              -> Custom b   -- ^ Function to extract the custom Data from the 'Document'
              -> IOSArrow XmlTree (Maybe (Document b))
  getDocument theUri getCustom 
    = mkDoc $<<< (      -- get the document title. to avoid failure of the arrow if the title tag is
                        -- absent, a default value is supplied
                        ( withDefault (getXPathTrees "/html/head/title/text()" >>> getText) "")  
                   &&& constA theUri
                   &&& getCustom
                 )
      where
      mkDoc t u c = constA $ Just $ Document t u c          
  
  
           
  