from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, DateTime, String, Boolean from sqlalchemy.types import TypeDecorator from datetime import datetime from Xlib import X import json Base = declarative_base() MODIFIERS = { X.ShiftMask: "Shift", X.LockMask: "Lock", X.ControlMask: "Ctrl", X.Mod1Mask: "Alt", X.Mod2Mask: "Mod2", X.Mod3Mask: "Mod3", X.Mod4Mask: "Super", X.Mod5Mask: "AltGr" } class Event: id = Column(Integer, primary_key=True) time = Column(DateTime) def __init__(self, **args): if "time" not in args: self.time = datetime.utcnow() for key, value in args.iteritems(): setattr(self, key, value) def to_dict(self): return {col: getattr(self, col) for col in self.__table__.columns.keys()} def __str__(self): r = self.to_dict() del r["id"] r["time"] = r["time"].isoformat("T") + "Z" r["type"] = self.__class__.__name__ return json.dumps(r, ensure_ascii=False) class Modifiers(TypeDecorator): """ Custom type for lists of key modifiers. Lists are serialized/deserialized into/from integers using bitfields. """ impl = Integer def process_bind_param(self, value, dialect): return sum(key for key, v in MODIFIERS.iteritems() if v in value) def process_result_value(self, value, dialect): return [v for key, v in MODIFIERS.iteritems() if value & key] class KeyEvent(Event, Base): __tablename__ = "key_events" key = Column(Integer) key_name = Column(String) modifiers = Column(Modifiers) repeat = Column(Boolean) class ClickEvent(Event, Base): __tablename__ = "click_events" button = Column(Integer) x = Column(Integer) y = Column(Integer) class WindowEvent(Event, Base): __tablename__ = "window_events" window_id = Column(Integer) name = Column(String) class1 = Column(String) class2 = Column(String) class BrowserEvent(Event, Base): __tablename__ = "browser_events" url = Column(String) title = Column(String) if __name__ == "__main__": from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine = create_engine("sqlite:///test.db") Session = sessionmaker(bind=engine) Base.metadata.create_all(engine)