<commit_before>import os
import sys
import time
import threading
import urllib2
import bson
import csv
from StringIO import StringIO

from datetime import timedelta, datetime

from multiprocessing import Process
from lxml import etree
import requests

from Malcom.auxiliary.toolbox import debug_output
from Malcom.config.malconf import MalcomSetup
from Malcom.model.model import Model
from Malcom.feeds.messenger import FeedsMessenger



class Feed(object):
	"""This is a feed base class. All other feeds must inherit from this class"""
	def __init__(self, name, run_every="24h"):
		self.name = name

		# parse timedelta
		num = int(run_every[:-1])
		if run_every.endswith('s'):
			self.run_every = timedelta(seconds=num)
		if run_every.endswith('m'):
			self.run_every = timedelta(minutes=num)
		if run_every.endswith('h'):
			self.run_every = timedelta(hours=num)
		if run_every.endswith('d'):
			self.run_every = timedelta(days=num)

		self.last_run = None
		self.next_run = datetime.utcnow()
		self.running = False
		self.elements_fetched = 0
		self.status = "OK"
		self.enabled = False
		self.model = None
		self.testing = False

	def get_dict(self):
		return { 'name': self.name,
				 'last_run': self.last_run,
				 'next_run': self.next_run,
				 'running': self.running,
				 'elements_fetched': self.elements_fetched,
				 'status': self.status,
				 'enabled': self.enabled,
				}

	def update_xml(self, main_node, children, headers={}, auth=None):
		assert self.source != None

		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)
		
		self.status = "OK"

		return self.parse_xml(r.content, main_node, children)

	def parse_xml(self, data, main_node, children):

		tree = etree.parse(StringIO(data))

		for item in tree.findall("//%s"%main_node):
			evil = {}
			for field in children:
				evil[field] = item.findtext(field)

			evil['source'] = self.name

			yield evil


	def update_lines(self, headers={}, auth=None):
		assert self.source != None
		# request = urllib2.Request(self.source, headers=headers)
		# feed = urllib2.urlopen(request).readlines()

		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)

		feed = r.text.split('\n')
		
		self.status = "OK"

		for line in feed:
			yield line

	def update_csv(self, delimiter=';', quotechar="'", headers={}, auth=None):
		assert self.source != None
		# request = urllib2.Request(self.source, headers=headers)
		# feed = urllib2.urlopen(request).readlines()
		print "requesting", self.source
		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)

		feed = r.text.split('\n')
		reader = csv.reader(feed, delimiter=delimiter, quotechar=quotechar)
		
		self.status = "OK"

		for line in reader:
			yield line

	def update_json(self, headers={}, auth=None):
		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)

		return r.json()
		
	def update(self):
		"""
		The update() function has to be implemented in each of your feeds.
		Its role is to:
		 - Fetch data from wherever it needs to
		 - Translate this data into elements understood by Malcom (as defined in malcom.datatypes.element)
		 - Save these newly created elements to the database using the self.model attribute
		"""
		raise NotImplementedError("update: This method must be implemented in your feed class")

	def analyze(self):
		raise NotImplementedError("analyze: This method must be implemented in your feed class")

	def commit_to_db(self, element, testing=False):
		if self.testing:
			self.elements_fetched +=1
			return

		# add an 'evil' tag if it was not specified in the feed
		if 'evil' not in element['tags']: element['tags'] += ['evil']

		element, new = self.model.save(element, with_status=True)
		if new:
			self.elements_fetched += 1

		return element


	def run(self):

		self.running = True
		self.last_run = datetime.utcnow()
		self.next_run = self.last_run + self.run_every
		self.elements_fetched = 0

		# REDIS send messages to webserver
		# self.analytics.notify_progress("Feeding")
		try:
			t0 = datetime.now()
			self.update()
			t1 = datetime.now()
			debug_output("Feed {} added in {}".format(self.name, str(t1-t0)))
			# save time for record in db
			self.model.feed_last_run(self.name)
		except Exception, e:
			debug_output("Error adding feed {}: {}".format(self.name, e))
	 		self.status = "ERROR: {}".format(e)

		self.running = False



class FeedEngine(Process):
	"""Feed engine. This object will load and update feeds"""
	def __init__(self, configuration):
		Process.__init__(self)
		self.configuration = configuration
		self.model = Model(self.configuration)
		self.feeds = {}
		self.threads = {}
		self.global_thread = None
		# self.messenger = FeedsMessenger(self)

	def run_feed(self, feed_name):
		# Check if feed exists in list
		if not self.feeds.get(feed_name):
			return False
<commit_msg>Print stacktrace in logs if feed fails<commit_after>import os
import sys
import time
import threading
import urllib2
import bson
import csv
from StringIO import StringIO

from datetime import timedelta, datetime

from multiprocessing import Process
from lxml import etree
import requests

from Malcom.auxiliary.toolbox import debug_output
from Malcom.config.malconf import MalcomSetup
from Malcom.model.model import Model
from Malcom.feeds.messenger import FeedsMessenger




class Feed(object):
	"""This is a feed base class. All other feeds must inherit from this class"""
	def __init__(self, name, run_every="24h"):
		self.name = name

		# parse timedelta
		num = int(run_every[:-1])
		if run_every.endswith('s'):
			self.run_every = timedelta(seconds=num)
		if run_every.endswith('m'):
			self.run_every = timedelta(minutes=num)
		if run_every.endswith('h'):
			self.run_every = timedelta(hours=num)
		if run_every.endswith('d'):
			self.run_every = timedelta(days=num)

		self.last_run = None
		self.next_run = datetime.utcnow()
		self.running = False
		self.elements_fetched = 0
		self.status = "OK"
		self.enabled = False
		self.model = None
		self.testing = False

	def get_dict(self):
		return { 'name': self.name,
				 'last_run': self.last_run,
				 'next_run': self.next_run,
				 'running': self.running,
				 'elements_fetched': self.elements_fetched,
				 'status': self.status,
				 'enabled': self.enabled,
				}

	def update_xml(self, main_node, children, headers={}, auth=None):
		assert self.source != None

		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)
		
		self.status = "OK"

		return self.parse_xml(r.content, main_node, children)

	def parse_xml(self, data, main_node, children):

		tree = etree.parse(StringIO(data))

		for item in tree.findall("//%s"%main_node):
			evil = {}
			for field in children:
				evil[field] = item.findtext(field)

			evil['source'] = self.name

			yield evil



	def update_lines(self, headers={}, auth=None):
		assert self.source != None
		# request = urllib2.Request(self.source, headers=headers)
		# feed = urllib2.urlopen(request).readlines()

		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)

		feed = r.text.split('\n')
		
		self.status = "OK"

		for line in feed:
			yield line

	def update_csv(self, delimiter=';', quotechar="'", headers={}, auth=None):
		assert self.source != None
		# request = urllib2.Request(self.source, headers=headers)
		# feed = urllib2.urlopen(request).readlines()
		print "requesting", self.source
		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)

		feed = r.text.split('\n')
		reader = csv.reader(feed, delimiter=delimiter, quotechar=quotechar)
		
		self.status = "OK"

		for line in reader:
			yield line

	def update_json(self, headers={}, auth=None):
		if auth:
			r = requests.get(self.source, headers=headers, auth=auth)
		else:
			r = requests.get(self.source, headers=headers)

		return r.json()
		
	def update(self):
		"""
		The update() function has to be implemented in each of your feeds.
		Its role is to:
		 - Fetch data from wherever it needs to
		 - Translate this data into elements understood by Malcom (as defined in malcom.datatypes.element)
		 - Save these newly created elements to the database using the self.model attribute
		"""
		raise NotImplementedError("update: This method must be implemented in your feed class")

	def analyze(self):
		raise NotImplementedError("analyze: This method must be implemented in your feed class")

	def commit_to_db(self, element, testing=False):
		if self.testing:
			self.elements_fetched +=1
			return

		# add an 'evil' tag if it was not specified in the feed
		if 'evil' not in element['tags']: element['tags'] += ['evil']

		element, new = self.model.save(element, with_status=True)
		if new:
			self.elements_fetched += 1

		return element


	def run(self):

		self.running = True
		self.last_run = datetime.utcnow()
		self.next_run = self.last_run + self.run_every
		self.elements_fetched = 0

		# REDIS send messages to webserver
		# self.analytics.notify_progress("Feeding")
		try:
			t0 = datetime.now()
			self.update()
			t1 = datetime.now()
			debug_output("Feed {} added in {}".format(self.name, str(t1-t0)))
			# save time for record in db
			self.model.feed_last_run(self.name)
		except Exception, e:
			debug_output("Error adding feed {}: {}".format(self.name, e))
	 		self.status = "ERROR: {}".format(e)
	 		import traceback
	 		traceback.print_exc()

		self.running = False



class FeedEngine(Process):
	"""Feed engine. This object will load and update feeds"""
	def __init__(self, configuration):
		Process.__init__(self)
		self.configuration = configuration
		self.model = Model(self.configuration)
		self.feeds = {}
		self.threads = {}
		self.global_thread = None
		# self.messenger = FeedsMessenger(self)

	def run_feed(self, feed_name):
		# Check if feed exists in list
		if not self.feeds.get(feed_name):
			return False

{'fraction_non_alphanumeric': 0.11936936936936937, 'fraction_numerical': 0.003133568350959655, 'mean_word_length': 3.95536147501213, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 8}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1263083', 'n_tokens_mistral': 3820, 'n_tokens_neox': 3301, 'n_words': 925}