import json
import datetime
from kafka import KafkaClient
from kafka import SimpleConsumer

from . import models
from . import sql
import time
import sqlalchemy.exc

from couchpulse import settings

kafka_client = KafkaClient(settings.KAFKA_HOST, settings.KAFKA_PORT)
kafka_consumer = SimpleConsumer(kafka_client, 'couchpulse.logreader', settings.KAFKA_TOPIC, auto_commit=False)


def consume():
    session = sql.Session()
    for message in kafka_consumer:
        message = json.loads(message.message.value)
        if not message.get('id'):
            print ('skipping')
            continue
        type = message['type']
        if type == 'request':
            r = models.RequestLog(message)
            session.add(sql.RequestLog(
                id=r.id,
                method=r.method,
                path=r.path,
                params=r.params,
                size=r.size,
                time=r.time,
                timestamp=datetime.datetime.utcfromtimestamp(r.timestamp),
            ))
            print 'saving request %s (%s)' % (r.timestamp, r.id)
        elif type == 'response':
            r = models.ResponseLog(message)
            session.add(sql.ResponseLog(
                id=r.id,
                size=r.size,
                time=r.time,
            ))
            print 'saving response %s' % r.id
        try:
            session.commit()
        except sqlalchemy.exc.IntegrityError:
            session.rollback()
    session.close()


def consume_forever():
    while True:
        consume()
        time.sleep(1)

@@ -265,47 +265,29 @@
 RT)%0A
-kafka_consumer = SimpleC
+%0A%0Adef c
 onsume
-r
 (kafka_c
 lien
@@ -286,94 +286,15 @@
 ka_c
-lient, 'couchpulse.logreader', settings.KAFKA_TOPIC, auto_commit=False)%0A%0A%0Adef consume(
+onsumer
 ):%0A 
@@ -328,16 +328,22 @@
     for 
+kafka_
 message 
@@ -390,16 +390,22 @@
 n.loads(
+kafka_
 message.
@@ -419,16 +419,52 @@
 .value)%0A
+        print kafka_message.offset,%0A
         
@@ -1419,18 +1419,52 @@
 close()%0A
+    return kafka_consumer.offsets
 %0A%0A
+%0A
 def cons
@@ -1479,43 +1479,523 @@
 ver(
-):%0A    while True:%0A        consume(
+offsets=None, cache=None):%0A    kafka_consumer = SimpleConsumer(kafka_client, 'couchpulse.logreader', settings.KAFKA_TOPIC, auto_commit=False)%0A    if cache and offsets is None:%0A        offsets = cache.get('couchpulse.offsets')%0A    if offsets:%0A        print 'starting at offsets %7B0%7D'.format(offsets)%0A        kafka_consumer.offsets = offsets%0A        print kafka_consumer.offsets%0A    while True:%0A        offsets = consume(kafka_consumer)%0A        if cache and offsets:%0A            cache.set('couchpulse.offsets', offsets
 )%0A  

improve logreader to manage the offsets