parse-comments.py

#!/usr/bin/env python3
import json
import logging
import subprocess
import threading
import time
from pprint import pformat
from urllib.parse import urlencode
from urllib.request import urlopen

import click

import clang.cindex

#  logging.basicConfig(level=logging.DEBUG)
logging.basicConfig(level=logging.WARNING)


class LanguageToolServerThread(threading.Thread):

    def __init__(self, port, log_path, jar_path):
        self._port = port
        self._log_path = log_path
        self._jar_path = jar_path
        self._process = None
        threading.Thread.__init__(self)

    def run(self):
        with open(self._log_path, 'a') as out:
            self._process = subprocess.Popen(["java", "-cp", self._jar_path,
                                              "org.languagetool.server.HTTPServer",
                                              "--port", str(self._port),
                                              "--allow-origin", '*'],
                                             stdout=out,
                                             stderr=out,
                                             close_fds=True)

    def terminate(self):
        self._process.terminate()


class LanguageToolServer:

    _server_port = 8081
    _log_path = "./lt_server.log"
    _jar_path = "/home/jubnzv/.local/LanguageTool-4.6/languagetool-server.jar"
    _disabled_rules = ["UPPERCASE_SENTENCE_START"]

    def __init__(self):
        self._server_url = "http://localhost:%d/v2/check" % self._server_port
        self._server_thread = LanguageToolServerThread(
            self._server_port, self._log_path, self._jar_path)

    def start(self):
        self._server_thread.start()
        limit = 30
        with open(self._log_path, 'r') as log:
            while True:
                line = log.readline()
                if "Server started" in line:
                    logging.info("Server started")
                    break
                time.sleep(1)
                limit -= 1
                if limit == 0:
                    raise TimeoutError

    def stop(self):
        logging.info("Stopping server...")
        self._server_thread.terminate()
        self._server_thread.join()
        logging.info("Stopping stopped")

    def check(self, text):
        payload = {
            'language': "en-US",
            'text': text.encode('utf8'),
            'disabledRules': ','.join(self._disabled_rules)
        }
        content = self._post(payload)
        if content:
            j = json.loads(content.decode('utf-8'))
            logging.debug(pformat(j['matches']))
            return j['matches']
        else:
            return None

    def _post(self, payload):
        data = urlencode(payload).encode('utf8')
        try:
            content = urlopen(self._server_url, data).read()
            return content
        except IOError:
            return None


class SourceFile:

    def __init__(self, file_path):
        self._file_path = file_path
        self._comments = self._clang_get_comments()

    def __str__(self):
        return self._file_path

    @property
    def comments(self):
        return self._comments

    def _clang_get_comments(self):
        comments = []
        index = clang.cindex.Index.create()
        tu = index.parse(self._file_path, args=['-x', 'c++'])
        for x in tu.cursor.get_tokens():
            if (x.kind.name == 'COMMENT'):
                comments.append({"text": str(x.spelling),
                                 "start_line": x.extent.start.line,
                                 "end_line": x.extent.end.line,
                                 "start_column": x.extent.start.column,
                                 "end_column": x.extent.end.column})
        return comments


@click.command()
@click.argument("files", nargs=-1, type=click.Path(exists=True))
def main(files):
    lt = LanguageToolServer()
    source_files = [SourceFile(f) for f in files]
    try:
        lt.start()
        for f in source_files:
            logging.debug("Checking %s (%d) comments" % (f, len(f.comments)))
            for comment in f.comments:
                errors = lt.check(comment["text"])
                if len(errors) == 0:
                    logging.debug("%s: No errors found", f)
                else:
                    print("%s: Found errors at %d:%d: %s" % (f,
                                                             comment["start_line"],
                                                             comment["end_line"],
                                                             comment["text"]))
                    for error in errors:
                        print(error["message"])
                    print()
    finally:
        lt.stop()


if __name__ == '__main__':
    main()