Holumbus-MapReduce-0.1.1: Examples/MapReduce/Crawler/Crawl.hs
-- ----------------------------------------------------------------------------
{- |
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