lnotify.py

#!/usr/bin/python

import threading
import requests
import logging
import json
import time
import os
import sys
import subprocess

def LorURL(path="/"):
  return "https://www.linux.org.ru%s" % (path,)

def ReadConfig(path="~/.lnotify"):
  path = os.path.expanduser(path)
  with open(path) as f:
    return json.load(f)

class TimeoutManager(object):
    def __init__(self, config):
        self.config = config
        self.active_timeout = int(self.config.get("active_timeout_sec", 10))
        self.passive_timeout = int(self.config.get("passive_timeout_sec", 300))
        self.passivation_cycles_count = int(self.config.get("passivation_cycles_count", 18))
        self.active_timeouts_passed = 0
        self.Wakeup()

    def Wakeup(self):
        self.active_timeouts_passed = self.passivation_cycles_count

    def Wait(self):
        if self.active_timeouts_passed>0:
            self.active_timeouts_passed = self.active_timeouts_passed - 1
            time.sleep(self.active_timeout)
        else:
            time.sleep(self.passive_timeout)


class Py3status(object):
    
    def __init__(self):
        self.session = self._BuildSessionObject()

        self.config = ReadConfig()

        self.timeout_manager = TimeoutManager(self.config)

        self.format_str = self.config.get("format","LOR: %s")
        self.nomsg_format_str = self.config.get("nomsg_format", "LOR: no messages")
        
        self.completed = False
        self.message_count = 0
        self.loggedIn = False
        self.broken = False
        self.extra_message = ""

        self.thread = threading.Thread(target=lambda: self._ThreadProc())
        self.thread.daemon = True
        self.thread.start()
  
    def _BuildSessionObject(self):
        session = requests.Session()
        session.mount(LorURL(), requests.adapters.HTTPAdapter(max_retries=10))
        return session 

    def _BuildConnectedSession(self):
        self.session = self._BuildSessionObject()
        front_page = self.session.get(LorURL())
        csrf_token = front_page.cookies["CSRF_TOKEN"][1:-1]
        login_data = {
            "nick":self.config["username"],
            "passwd":self.config["password"],
            "csrf":csrf_token
        }
        headers={"Accept":"application/json"}
        
        response_text = self.session.post(LorURL("/ajax_login_process"), data=login_data, headers=headers).text
        response = json.loads(response_text)
        return bool(response["loggedIn"])

    def _ThreadProc(self):
        try:
         while not self.completed:
            notifications_response = self.session.get(LorURL("/notifications-count"))
            if notifications_response.status_code==200:
                new_message_count = int(notifications_response.text)
                if new_message_count != self.message_count:
                    self.timeout_manager.Wakeup()
                self.message_count = new_message_count
                self.timeout_manager.Wait()
            else:
                if not self._BuildConnectedSession():
                    self.broken = True
                    self.completed = True
                    self.extra_message = "Could not connect"
        except Exception as e:
            self.extra_message = str(e)

    def EmitMessage(self, i3status_output_json, i3status_config):
        response = {
            "name" : "lnotify",
            "cached_until": time.time()+5
        }
        if self.broken:
            response["full_text"] = "LOG: login failed"
            response["color"] = "#FF0000"
        elif self.message_count != 0:
            response["full_text"] = self.format_str % self.message_count
            response["color"] = "#00FF00"
        else:
            response["full_text"] = self.nomsg_format_str
        response["full_text"] = response["full_text"]+self.extra_message
        return (0,response)
    
    def on_click(self, i3status_output_json, i3status_config, evt):
        subprocess.call(["xdg-open","https://www.linux.org.ru/notifications"],
                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        browser_workspace = self.config.get("browser_workspace",None)
        if browser_workspace:
            subprocess.call(["i3-msg","workspace "+browser_workspace],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    def kill(self):
        self.completed = True