Inheritance diagram for dc_db.URLFetchTask.URLFetchTask:

Collaboration diagram for dc_db.URLFetchTask.URLFetchTask:

Public Member Functions
def	__init__ (self, keyValueStorageDir, rawDataDir, dBDataTask, dcSiteTemplate, keyValueDefaultFile, dcStatTemplate, dcLogTemplate, mutexLockTTL=Constants.DEFAULT_LOCK_TTL)

def	process (self, urlFetches, queryCallback)

def	processProportial (self, urlFetch, uRLUpdateTask, queryCallback)

def	fillCriterionLimits (self, urlFetch, offset)

def	processSimple (self, urlFetch, uRLUpdateTask, queryCallback)

def	updateUrl (self, urls, urlFetch, uRLUpdateTask, queryCallback)

def	fillSiteListSQLGenerate (self, sitesCriterions, userId=None)

def	fillSiteList (self, incomeSiteList, queryCallback, sitesCriterions, userId=None)

def	getURLFromURLTable (self, urlFetch, globalLen, queryCallback)

def	execAdditionalSQLs (self, sqls, siteId, queryCallback)

Public Member Functions inherited from dc_db.BaseTask.BaseTask
def	isSiteExist (self, siteId, queryCallback, userId=None)

def	generateCriterionSQL (self, criterions, additionWhere=None, siteId=None)

def	fetchByCriterions (self, criterions, queryCallback)

def	dbLock (self, mutexName, queryCallback, sleepTime=1, mutexLockTTL=Constants.DEFAULT_LOCK_TTL)

def	dbUnlock (self, mutexName, queryCallback)

def	createUrlsInsertQuery (self, siteId, localKeys, localValues)

def	copyUrlsToDcUrls (self, siteId, queryCallback)

def	statisticLogUpdate (self, localObj, urlMd5, siteId, status, queryCallback, isInsert=False)

def	calculateMd5FormUrl (self, url, urlType, useNormilize=False)

Static Public Member Functions
def	fillUrlObj (row)

Static Public Member Functions inherited from dc_db.BaseTask.BaseTask
def	readValueFromSiteProp (siteId, propName, queryCallback, urlMd5=None)

Public Attributes
	quantMaxUrls

	localSiteList

	mutexLockTTL

	uRLUpdateTask

	siteUpdateTask

Detailed Description

Definition at line 24 of file URLFetchTask.py.

Constructor & Destructor Documentation

◆ init()

def dc_db.URLFetchTask.URLFetchTask.__init__	(	self,
		keyValueStorageDir,
		rawDataDir,
		dBDataTask,
		dcSiteTemplate,
		keyValueDefaultFile,
		dcStatTemplate,
		dcLogTemplate,
		mutexLockTTL = `Constants.DEFAULT_LOCK_TTL`
	)

Definition at line 30 of file URLFetchTask.py.

                dcStatTemplate, dcLogTemplate, mutexLockTTL=Constants.DEFAULT_LOCK_TTL):
     super(URLFetchTask, self).__init__()
     self.quantMaxUrls = 0
     self.localSiteList = []
     self.mutexLockTTL = mutexLockTTL
     self.uRLUpdateTask = URLUpdateTask(keyValueStorageDir, rawDataDir, dBDataTask)
     self.siteUpdateTask = SiteUpdateTask(dcSiteTemplate, keyValueDefaultFile, keyValueStorageDir, rawDataDir,
                                          dBDataTask, dcStatTemplate, dcLogTemplate, None)
 
 

Member Function Documentation

◆ execAdditionalSQLs()

def dc_db.URLFetchTask.URLFetchTask.execAdditionalSQLs	(	self,
		sqls,
		siteId,
		queryCallback
	)

Definition at line 318 of file URLFetchTask.py.

   def execAdditionalSQLs(self, sqls, siteId, queryCallback):
     ret = {}
     for key in sqls:
       q = sqls[key].replace("%" + Constants.SITE_ID_NAME + "%", str(siteId))
       r = queryCallback(q, Constants.PRIMARY_DB_ID)
       if hasattr(r, '__iter__') and len(r) > 0 and r[0][0] != None:
         ret[key] = r[0][0]
       else:
         ret[key] = "NULL"
 
     logger.debug("Additional SQLs: %s", str(ret))
 
     return ret
 

Here is the caller graph for this function:

◆ fillCriterionLimits()

def dc_db.URLFetchTask.URLFetchTask.fillCriterionLimits	(	self,
		urlFetch,
		offset
	)

Definition at line 126 of file URLFetchTask.py.

   def fillCriterionLimits(self, urlFetch, offset):
     if urlFetch.urlsCriterions is None:
       urlFetch.urlsCriterions = {}
     urlFetch.urlsCriterions[dc.EventObjects.URLFetch.CRITERION_LIMIT] = [offset, self.quantMaxUrls]
 
 

Here is the caller graph for this function:

◆ fillSiteList()

def dc_db.URLFetchTask.URLFetchTask.fillSiteList	(	self,
		incomeSiteList,
		queryCallback,
		sitesCriterions,
		userId = `None`
	)

Definition at line 206 of file URLFetchTask.py.

   def fillSiteList(self, incomeSiteList, queryCallback, sitesCriterions, userId=None):
     ret = []
     newSiteList = []
     localSiteDict = {}
     query = self.fillSiteListSQLGenerate(sitesCriterions, userId)
     logger.debug("!!! query: %s", str(query))
     res = None
     try:
       res = queryCallback(query, Constants.PRIMARY_DB_ID)
       logger.debug("!!! res: %s", str(res))
     except mdb.Error, err:  # pylint: disable=E1101
       logger.error("Error: %s", str(err))
 
     if res is not None and hasattr(res, '__iter__'):
       for row in res:
         localSiteDict[row[0]] = row[0]
         newSiteList.append(row[0])
 
       if len(incomeSiteList) == 0:
         ret = newSiteList
       else:
         newSiteList = []
         for site in incomeSiteList:
           if site in localSiteDict:
             newSiteList.append(site)
         ret = newSiteList
 
     return ret
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ fillSiteListSQLGenerate()

def dc_db.URLFetchTask.URLFetchTask.fillSiteListSQLGenerate	(	self,
		sitesCriterions,
		userId = `None`
	)

Definition at line 174 of file URLFetchTask.py.

   def fillSiteListSQLGenerate(self, sitesCriterions, userId=None):
     query = Constants.SELECT_SQL_TEMPLATE_SIMPLE % ("`Id`", "sites")
     additionQueryStr = ""
     additionWhereCause = ""
     if userId is not None:
       additionWhereCause += (Constants.CHECK_TABLE_SQL_ADDITION % str(userId))
 
     if sitesCriterions is None:
       if additionWhereCause != "":
         additionWhereCause += " AND "
       additionWhereCause += ("`state`=%d" % dc.EventObjects.Site.STATE_ACTIVE)
 
     if sitesCriterions is None:
       if additionWhereCause != "":
         additionQueryStr = " WHERE " + additionWhereCause
     else:
       if additionWhereCause == "":
         additionQueryStr = self.generateCriterionSQL(sitesCriterions)
       else:
         additionQueryStr = self.generateCriterionSQL(sitesCriterions, additionWhereCause)
     query += additionQueryStr
 
     return query
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ fillUrlObj()

def dc_db.URLFetchTask.URLFetchTask.fillUrlObj ( row )

static

Definition at line 241 of file URLFetchTask.py.

   def fillUrlObj(row):
     url = dc.EventObjects.URL(row["Site_Id"], row["URL"], normalizeMask=UrlNormalizator.NORM_NONE)
     for field in Constants.URLTableDict.keys():
       if hasattr(url, field) and Constants.URLTableDict[field] in row:
         setattr(url, field, row[Constants.URLTableDict[field]])
     url.UDate = Constants.readDataTimeField("UDate", row)
     url.CDate = Constants.readDataTimeField("CDate", row)
     url.lastModified = Constants.readDataTimeField("LastModified", row)
     url.tcDate = Constants.readDataTimeField("TcDate", row)
     url.pDate = Constants.readDataTimeField("PDate", row)
     return url
 
 

◆ getURLFromURLTable()

def dc_db.URLFetchTask.URLFetchTask.getURLFromURLTable	(	self,
		urlFetch,
		globalLen,
		queryCallback
	)

Definition at line 259 of file URLFetchTask.py.

   def getURLFromURLTable(self, urlFetch, globalLen, queryCallback):
     urls = []
 
     for siteId in urlFetch.sitesList:
       # Update site TcDate to push rotate sites list
       if urlFetch.siteUpdate is not None:
         urlFetch.siteUpdate.id = siteId
         self.siteUpdateTask.process(urlFetch.siteUpdate, queryCallback)
         logger.debug("Site %s updated", str(siteId))
 
       tableName = Constants.DC_URLS_TABLE_NAME_TEMPLATE % siteId
       # Make criterion
       additionQueryStr = self.generateCriterionSQL(urlFetch.urlsCriterions, None, siteId)
       # Execute additional SQLs for criterion
       if urlFetch.CRITERION_SQL in urlFetch.urlsCriterions and len(urlFetch.urlsCriterions[urlFetch.CRITERION_SQL]) > 0:
         additionalSQLs = self.execAdditionalSQLs(urlFetch.urlsCriterions[urlFetch.CRITERION_SQL], siteId, queryCallback)
       else:
         additionalSQLs = {}
       # Substitute values from additional SQLs results in criterion
       for sql in additionalSQLs:
         additionQueryStr = additionQueryStr.replace("%" + sql + "%", additionalSQLs[sql])
       # Finally add criterion to query
       query = Constants.SELECT_SQL_TEMPLATE_SIMPLE % ("*", tableName)
       if len(additionQueryStr) > 0:
         query += " "
         query += additionQueryStr
 
       logger.debug('>>> getURLFromURLTable UrlFetchQuery: ' + str(query))
       # Execute query
       res = queryCallback(query, Constants.SECONDARY_DB_ID, Constants.EXEC_NAME)
 
       if not hasattr(res, '__iter__'):
         res = []
 
       logger.debug('>>> getURLFromURLTable len(res): ' + str(len(res)))
 
       # Add urlList truncating
       # res = self.limitsURLs(siteId, res, queryCallback):
       if self.quantMaxUrls > 0 and len(res) == self.quantMaxUrls:
         self.localSiteList.append(siteId)
 
       for row in res:
         if urlFetch.maxURLs is not None and urlFetch.maxURLs != 0 and urlFetch.maxURLs == globalLen:
           break
         if "Site_Id" in row and "URL" in row:
           urls.append(URLFetchTask.fillUrlObj(row))
           globalLen += 1
       if urlFetch.maxURLs is not None and urlFetch.maxURLs != 0 and urlFetch.maxURLs == globalLen:
         break
 
     logger.debug("Return urlsFetch count = %s", str(len(urls)))
     return urls
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ process()

def dc_db.URLFetchTask.URLFetchTask.process	(	self,
		urlFetches,
		queryCallback
	)

Definition at line 46 of file URLFetchTask.py.

   def process(self, urlFetches, queryCallback):
     ret = []
     needToLock = False
     try:
       for urlFetch in urlFetches:
         needToLock = urlFetch.isLocking
         if needToLock:
           self.dbLock(Constants.FETCH_LOCK_NAME, queryCallback, urlFetch.lockIterationTimeout, self.mutexLockTTL)
         # @todo add more complex case
         if urlFetch.algorithm is None or urlFetch.algorithm == dc.EventObjects.URLFetch.DEFAULT_ALGORITHM:
           ret.extend(self.processSimple(urlFetch, self.uRLUpdateTask, queryCallback))
         elif urlFetch.algorithm == dc.EventObjects.URLFetch.PROPORTIONAL_ALGORITHM:
           ret.extend(self.processProportial(urlFetch, self.uRLUpdateTask, queryCallback))
 
         for url in ret:
           attributeNames = urlFetch.attributeNames if hasattr(urlFetch, 'attributeNames') else ['*']
           url.attributes = AttrFetchTask.fetchUrlsAttributesByNames(url.siteId,
                                                                     url.urlMd5,
                                                                     queryCallback,
                                                                     attributeNames)
 
         if needToLock:
           self.dbUnlock(Constants.FETCH_LOCK_NAME, queryCallback)
     except Exception, err:
       logger.error('Exception: %s', str(err))
       if needToLock:
         self.dbUnlock(Constants.FETCH_LOCK_NAME, queryCallback)
       raise
     except mdb.Error, err:  # pylint: disable=E1101
       logger.error('mdb.Error: %s', str(err))
       if needToLock:
         self.dbUnlock(Constants.FETCH_LOCK_NAME, queryCallback)
       raise
     except:
       logger.error('Unknown type error')
       if needToLock:
         self.dbUnlock(Constants.FETCH_LOCK_NAME, queryCallback)
       raise
     return ret
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ processProportial()

def dc_db.URLFetchTask.URLFetchTask.processProportial	(	self,
		urlFetch,
		uRLUpdateTask,
		queryCallback
	)

Definition at line 93 of file URLFetchTask.py.

   def processProportial(self, urlFetch, uRLUpdateTask, queryCallback):
     ret = []
     offset = 0
     if urlFetch.maxURLs is not None and urlFetch.maxURLs > 0:
       urlFetch.sitesList = self.fillSiteList(urlFetch.sitesList, queryCallback, urlFetch.sitesCriterions)
 #     look sites from SiteList for clearing (month ... mins) sites limits
 #     self.limitsClearing(urlFetch.sitesList, queryCallback):
       if len(urlFetch.sitesList) > 0:
         self.quantMaxUrls = urlFetch.maxURLs / len(urlFetch.sitesList)
         if self.quantMaxUrls < 1:
           self.quantMaxUrls = 1
 
         while len(ret) < urlFetch.maxURLs and len(urlFetch.sitesList) > 0:
           self.fillCriterionLimits(urlFetch, offset)
           self.localSiteList = []
           logger.debug(">>> Debug-1 = %s %s %s %s",
                        str(urlFetch.urlsCriterions[dc.EventObjects.URLFetch.CRITERION_LIMIT][0]),
                        str(urlFetch.urlsCriterions[dc.EventObjects.URLFetch.CRITERION_LIMIT][1]),
                        str(urlFetch.maxURLs),
                        str(self.quantMaxUrls))
           ret.extend(self.getURLFromURLTable(urlFetch, len(ret), queryCallback))
           urlFetch.sitesList = self.localSiteList
           offset += self.quantMaxUrls
       else:
         logger.debug(">>> UrlFetch proportional >>> Empty SiteList")
     self.updateUrl(ret, urlFetch, uRLUpdateTask, queryCallback)
     return ret
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ processSimple()

def dc_db.URLFetchTask.URLFetchTask.processSimple	(	self,
		urlFetch,
		uRLUpdateTask,
		queryCallback
	)

Definition at line 138 of file URLFetchTask.py.

   def processSimple(self, urlFetch, uRLUpdateTask, queryCallback):
     urlFetch.sitesList = self.fillSiteList(urlFetch.sitesList, queryCallback, urlFetch.sitesCriterions)
 #   look sites from SiteList for clearing (month ... mins) sites limits
 #   self.limitsClearing(urlFetch.sitesList, queryCallback):
     ret = self.getURLFromURLTable(urlFetch, 0, queryCallback)
 
     self.updateUrl(ret, urlFetch, uRLUpdateTask, queryCallback)
     return ret
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ updateUrl()

def dc_db.URLFetchTask.URLFetchTask.updateUrl	(	self,
		urls,
		urlFetch,
		uRLUpdateTask,
		queryCallback
	)

Definition at line 155 of file URLFetchTask.py.

   def updateUrl(self, urls, urlFetch, uRLUpdateTask, queryCallback):
     if len(urls) > 0:
       if urlFetch.urlUpdate is not None:
         for localURL in urls:
           if localURL.urlMd5 is not None:
             urlUpdateList = []
             urlUpdateList.append(urlFetch.urlUpdate)
             urlUpdateList[0].urlMd5 = localURL.urlMd5
             urlUpdateList[0].siteId = localURL.siteId
             uRLUpdateTask.process(urlUpdateList, queryCallback)
           if urlFetch.urlUpdate.status is not None:
             localURL.status = urlFetch.urlUpdate.status
 
 

Here is the caller graph for this function:

Member Data Documentation

◆ localSiteList

dc_db.URLFetchTask.URLFetchTask.localSiteList

Definition at line 33 of file URLFetchTask.py.

◆ mutexLockTTL

dc_db.URLFetchTask.URLFetchTask.mutexLockTTL

Definition at line 34 of file URLFetchTask.py.

◆ quantMaxUrls

dc_db.URLFetchTask.URLFetchTask.quantMaxUrls

Definition at line 32 of file URLFetchTask.py.

◆ siteUpdateTask

dc_db.URLFetchTask.URLFetchTask.siteUpdateTask

Definition at line 36 of file URLFetchTask.py.

◆ uRLUpdateTask

dc_db.URLFetchTask.URLFetchTask.uRLUpdateTask

Definition at line 35 of file URLFetchTask.py.

The documentation for this class was generated from the following file:

sources/hce/dc_db/URLFetchTask.py

Public Member Functions

Static Public Member Functions

Public Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ execAdditionalSQLs()

◆ fillCriterionLimits()

◆ fillSiteList()

◆ fillSiteListSQLGenerate()

◆ fillUrlObj()

◆ getURLFromURLTable()

◆ process()

◆ processProportial()

◆ processSimple()

◆ updateUrl()

Member Data Documentation

◆ localSiteList

◆ mutexLockTTL

◆ quantMaxUrls

◆ siteUpdateTask

◆ uRLUpdateTask

◆ init()