Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _build_log_id_map(self):
from . import logging
if not hasattr(logging.Logger, 'log_id_map'):
log_db = self.db.database.logs
logging.Logger.log_id_map = dict(
(logging.MongoDBLogger.extract_legacy_id(rec['_id']), rec['_id'])
for rec in log_db.find(projection=[])
)
return logging.Logger.log_id_map
def import_(self, message):
# construct a unique objectid with the correct datetime.
dt = message['datetime']
oid_time = storage.bson.objectid.ObjectId.from_datetime(dt)
# store the original sqlite object ID in the
orig_id = message.pop('id')
orig_id_packed = struct.pack('>Q', orig_id)
oid_new = oid_time.binary[:4] + orig_id_packed
oid = storage.bson.objectid.ObjectId(oid_new)
if not hasattr(Logger, 'log_id_map'):
Logger.log_id_map = dict()
Logger.log_id_map[orig_id] = oid
message['_id'] = oid
message['datetime'] = self._fmt_date(dt)
self.db.insert(message)
def where(channel, nick, rest):
"When did pmxbot last see speak?"
onick = rest.strip()
last = Logger.store.last_seen(onick)
if last:
tm, chan = last
tmpl = "I last saw {onick} speak at {tm} in channel #{chan}"
return tmpl.format(tm=tm, chan=chan, onick=onick)
else:
return "Sorry! I don't have any record of %s speaking" % onick
def import_(self, message):
# construct a unique objectid with the correct datetime.
dt = message['datetime']
oid_time = storage.bson.objectid.ObjectId.from_datetime(dt)
# store the original sqlite object ID in the
orig_id = message.pop('id')
orig_id_packed = struct.pack('>Q', orig_id)
oid_new = oid_time.binary[:4] + orig_id_packed
oid = storage.bson.objectid.ObjectId(oid_new)
if not hasattr(Logger, 'log_id_map'):
Logger.log_id_map = dict()
Logger.log_id_map[orig_id] = oid
message['_id'] = oid
message['datetime'] = self._fmt_date(dt)
self.db.insert(message)