tweet_live: comparison script/lib/iri

equal deleted inserted replaced

-:cdd7d3c0549c
+:9213a63fa34a
 else:
 return value
 return dict([(str(k),adapt_one_field(k,v)) for k,v in fields_dict.items()])
+class ObjectBufferProxy(object):
+def __init__(self, klass, args, kwargs, must_flush):
+self.klass= klass
+self.args = args
+self.kwargs = kwargs
+self.must_flush = must_flush
+self.instance = None
+def persists(self, session):
+new_args = [arg() if callable(arg) else arg for arg in self.args] if self.args is not None else []
+new_kwargs = dict([(k,v()) if callable(v) else (k,v) for k,v in self.kwargs.items()]) if self.kwargs is not None else {}
+self.instance = self.klass(*new_args, **new_kwargs)
+session.add(self.instance)
+if self.must_flush:
+session.flush()
+def __getattr__(self, name):
+return lambda : getattr(self.instance, name) if self.instance else None
+class ObjectsBuffer(object):
+def __init__(self):
+self.__bufferlist = []
+def persists(self, session):
+for object_proxy in self.__bufferlist:
+object_proxy.persists(session)
+def add_object(self, klass, args, kwargs, must_flush):
+new_proxy = ObjectBufferProxy(klass, args, kwargs, must_flush)
+self.__bufferlist.append(new_proxy)
+return new_proxy
 class TwitterProcessorException(Exception):
 pass
 class TwitterProcessor(object):
 raise TwitterProcessorException("No id in json")
 self.source_id = source_id
 self.session = session
 self.token_filename = token_filename
+self.obj_buffer = ObjectsBuffer()
 def __get_user(self, user_dict):
-logging.debug("Get user : " + repr(user_dict)) #@UndefinedVariable
+logger.debug("Get user : " + repr(user_dict)) #@UndefinedVariable
 user_id = user_dict.get("id",None)
 user_name = user_dict.get("screen_name", user_dict.get("name", None))
 if user_id is None and user_name is None:
 if user_id:
 user_dict = t.users.show(user_id=user_id)
 else:
 user_dict = t.users.show(screen_name=user_name)
 except Exception as e:
-logging.info("get_user : TWITTER ERROR : " + repr(e)) #@UndefinedVariable
+logger.info("get_user : TWITTER ERROR : " + repr(e)) #@UndefinedVariable
-logging.info("get_user : TWITTER ERROR : " + str(e)) #@UndefinedVariable
+logger.info("get_user : TWITTER ERROR : " + str(e)) #@UndefinedVariable
 user_dict = adapt_fields(user_dict, fields_adapter["stream"]["user"])
 if "id" not in user_dict:
 return None
 self.session.flush()
 return user
 def __process_entity(self, ind, ind_type):
-logging.debug("Process_entity : " + repr(ind) + " : " + repr(ind_type)) #@UndefinedVariable
+logger.debug("Process_entity : " + repr(ind) + " : " + repr(ind_type)) #@UndefinedVariable
 ind = clean_keys(ind)
 entity_dict = {
 "indice_start": ind["indices"][0],
 'hashtags': process_hashtags,
 'user_mentions' : process_user_mentions,
 'urls' : process_urls
 }[ind_type]()
-logging.debug("Process_entity entity_dict: " + repr(entity_dict)) #@UndefinedVariable
+logger.debug("Process_entity entity_dict: " + repr(entity_dict)) #@UndefinedVariable
 if entity:
 self.session.add(entity)
 self.session.flush()
 ts_copy = adapt_fields(self.json_dict, fields_adapter["stream"]["tweet"])
 # get or create user
 user = self.__get_user(self.json_dict["user"])
 if user is None:
-logging.warning("USER not found " + repr(self.json_dict["user"])) #@UndefinedVariable
+logger.warning("USER not found " + repr(self.json_dict["user"])) #@UndefinedVariable
 ts_copy["user"] = None
 ts_copy["user_id"] = None
 else:
 ts_copy["user"] = user
 ts_copy["user_id"] = ts_copy["user"].id
 'screen_name' : self.json_dict["from_user"],
 }
 user = self.__get_user(user_fields)
 if user is None:
-logging.warning("USER not found " + repr(user_fields)) #@UndefinedVariable
+logger.warning("USER not found " + repr(user_fields)) #@UndefinedVariable
 tweet_fields["user"] = None
 tweet_fields["user_id"] = None
 else:
 tweet_fields["user"] = user
 tweet_fields["user_id"] = user.id
 extractor = twitter_text.Extractor(text)
 for ind in extractor.extract_hashtags_with_indices():
 self.__process_entity(ind, "hashtags")
+for ind in extractor.extract_urls_with_indices():
+self.__process_entity(ind, "urls")
 for ind in extractor.extract_mentioned_screen_names_with_indices():
 self.__process_entity(ind, "user_mentions")
-for ind in extractor.extract_urls_with_indices():
-self.__process_entity(ind, "urls")
-self.session.flush()
 def process(self):
 if self.source_id is None:
 tweet_source = TweetSource(original_json=self.json_txt);
 self.session.add(tweet_source)
 self.session.flush()
 self.source_id = tweet_source.id
-try:
+if "metadata" in self.json_dict:
-if "metadata" in self.json_dict:
+self.__process_twitter_rest()
-self.__process_twitter_rest()
+else:
-else:
+self.__process_twitter_stream()
-self.__process_twitter_stream()
+tweet_log = TweetLog(tweet_source_id = self.source_id, status = TweetLog.TWEET_STATUS['OK'])
-tweet_log = TweetLog(tweet_source_id = self.source_id, status = TweetLog.TWEET_STATUS['OK'])
+self.session.add(tweet_log)
-except:
-raise
+def set_logging(options, plogger=None):
+logging_config = {
-def set_logging(options):
+"format" : '%(asctime)s %(levelname)s:%(name)s:%(message)s',
+"level" : max(logging.NOTSET, min(logging.CRITICAL, logging.WARNING - 10 * options.verbose + 10 * options.quiet)), #@UndefinedVariable
-logging_config = {}
+}
 if options.logfile == "stdout":
 logging_config["stream"] = sys.stdout
 elif options.logfile == "stderr":
 logging_config["stream"] = sys.stderr
 else:
 logging_config["filename"] = options.logfile
-logging_config["level"] = max(logging.NOTSET, min(logging.CRITICAL, logging.WARNING - 10 * options.verbose + 10 * options.quiet)) #@UndefinedVariable
+logger = plogger
-logging.basicConfig(**logging_config) #@UndefinedVariable
+if logger is None:
+logger = logging.getLogger() #@UndefinedVariable
+if len(logger.handlers) == 0:
+filename = logging_config.get("filename")
+if filename:
+mode = logging_config.get("filemode", 'a')
+hdlr = logging.FileHandler(filename, mode) #@UndefinedVariable
+else:
+stream = logging_config.get("stream")
+hdlr = logging.StreamHandler(stream) #@UndefinedVariable
+fs = logging_config.get("format", logging.BASIC_FORMAT) #@UndefinedVariable
+dfs = logging_config.get("datefmt", None)
+fmt = logging.Formatter(fs, dfs) #@UndefinedVariable
+hdlr.setFormatter(fmt)
+logger.addHandler(hdlr)
+level = logging_config.get("level")
+if level is not None:
+logger.setLevel(level)
 options.debug = (options.verbose-options.quiet > 0)
 def set_logging_options(parser):
 parser.add_option("-l", "--log", dest="logfile",
 query = query.filter(or_(*map(lambda h: Hashtag.text.contains(h), htags))) #@UndefinedVariable
 return query.distinct()
+logger = logging.getLogger() #@UndefinedVariable

changeset 243	9213a63fa34a
parent 242	cdd7d3c0549c
child 244	d4b7d6e2633f