🚚 Move __init__.py to hooks/wakatime.py

Author: Freed-Wu

.github/FUNDING.yml

@@ -1,4 +1,5 @@
 ---
+patreon: user?u=83975719
 custom:
   - "https://user-images.githubusercontent.com/32936898/199681341-1c5cfa61-4411-4b67-b268-7cd87c5867bb.png"
   - "https://user-images.githubusercontent.com/32936898/199681363-1094a0be-85ca-49cf-a410-19b3d7965120.png"

.pre-commit-config.yaml

@@ -38,7 +38,7 @@ repos:
         args:
           - --msg-filename
   - repo: https://github.com/editorconfig-checker/editorconfig-checker.python
-    rev: 2.7.2
+    rev: 2.7.3
     hooks:
       - id: editorconfig-checker
   - repo: https://github.com/jumanjihouse/pre-commit-hooks
@@ -50,7 +50,7 @@ repos:
     hooks:
       - id: actionlint
   - repo: https://github.com/adrienverge/yamllint
-    rev: v1.32.0
+    rev: v1.33.0
     hooks:
       - id: yamllint
   - repo: https://github.com/executablebooks/mdformat
@@ -71,14 +71,14 @@ repos:
     hooks:
       - id: markdownlint-cli2
         additional_dependencies:
-          - markdown-it-texmath@0.9.1
+          - markdown-it-texmath
   - repo: https://github.com/Freed-Wu/pre-commit-hooks
     rev: 0.0.11
     hooks:
       - id: update-CITATION.cff
       - id: update-pyproject.toml
   - repo: https://github.com/psf/black
-    rev: 23.9.1
+    rev: 23.11.0
     hooks:
       - id: black
   - repo: https://github.com/PyCQA/isort
@@ -92,7 +92,7 @@ repos:
         additional_dependencies:
           - tomli
   - repo: https://github.com/kumaraditya303/mirrors-pyright
-    rev: v1.1.329
+    rev: v1.1.335
     hooks:
       - id: pyright
   - repo: https://github.com/PyCQA/bandit

README.md

@@ -94,11 +94,18 @@ at the end of file.
 ```python
 from repl_python_wakatime.python import install_hook
 
-install_hook(f, args, kwargs)
+install_hook(hook_function, args, kwargs)
 ```
 
-will execute `f(*args, **kwargs)` after every output/input. Other REPLs are
-similar.
+will execute `hook_function(*args, **kwargs)` after every output/input. Other
+REPLs are similar. Currently, `hook_function` can be:
+
+- `repl_python_wakatime.hooks.wakatime.wakatime_hook()`: By default.
+- `repl_python_wakatime.hooks.codestats.codestats_hook()`: for [codestats](https://codestats.net/)
+- Create your hooks for other similar projects, such as:
+  - [codetime](https://codetime.dev/)
+  - [rescuetime](https://www.rescuetime.com/)
+  - ...
 
 ## Similar projects
 

pyproject.toml

@@ -48,6 +48,9 @@ file = "requirements/dev.txt"
 [tool.setuptools.dynamic.optional-dependencies.ipython]
 file = "requirements/ipython.txt"
 
+[tool.setuptools.dynamic.optional-dependencies.keyring]
+file = "requirements/keyring.txt"
+
 [tool.setuptools.dynamic.optional-dependencies.ptipython]
 file = "requirements/ptipython.txt"
 
@@ -88,5 +91,5 @@ exclude_lines = [
 [tool.bandit.assert_used]
 skips = [
   "*_test.py",
-  "test_*.py",
+  "*/test_*.py",
 ]

requirements/dev.txt

@@ -1,6 +1,6 @@
 #!/usr/bin/env -S pip install -r
 # For unit test and code coverage rate test.
 
-pre-commit
+keyring
 ptpython[ptipython]
 pytest-cov

requirements/keyring.txt

@@ -0,0 +1,4 @@
+#!/usr/bin/env -S pip install -r
+# Store your codestats API key to `keyring get codestats HOST_NAME`.
+
+keyring

src/repl_python_wakatime/__init__.py

@@ -1,74 +1,10 @@
-"""Refer `create-plugin <https://wakatime.com/help/creating-plugin>`_."""
-import logging
-import os
-from shutil import which
-from subprocess import run  # nosec: B404
-from threading import Thread
-from typing import Any, Callable
-
+r"""Provide ``__version__`` for
+`importlib.metadata.version() <https://docs.python.org/3/library/importlib.metadata.html#distribution-versions>`_.
+"""
 try:
     from ._version import __version__, __version_tuple__  # type: ignore
-except ImportError:
+except ImportError:  # for setuptools-generate
     __version__ = "rolling"
     __version_tuple__ = (0, 0, 0, __version__, "")
 
-logger = logging.getLogger(__name__)
-
-
-def send_wakatime_heartbeat(
-    project: str = "",
-    category: str = "coding",
-    plugin: str = "repl-python-wakatime",
-    filenames: list[str] = [".git"],
-    detect_func: Callable[[str], bool] = os.path.isdir,
-) -> None:
-    """Send wakatime heartbeat.
-
-    If ``project == ""``, detect automatically.
-
-    ``plugin`` must be the format of ``repl-REPL_NAME-wakatime`` to let
-    wakatime detect correctly.
-
-    :param project:
-    :type project: str
-    :param category:
-    :type category: str
-    :param plugin:
-    :type plugin: str
-    :param filenames:
-    :type filenames: list[str]
-    :param detect_func:
-    :type detect_func: Callable[[str], bool]
-    :rtype: None
-    """
-    if project == "":
-        from .utils import get_project
-
-        project = get_project(filenames, detect_func)
-    run(  # nosec: B603 B607
-        [
-            "wakatime-cli",
-            "--write",
-            f"--category={category}",
-            f"--plugin={plugin}",
-            "--entity-type=app",
-            "--entity=python",
-            "--alternate-language=python",
-            f"--project={project}",
-        ],
-        stdout=open(os.devnull, "w"),
-    )
-
-
-def wakatime_hook(*args: Any, **kwargs: Any) -> None:
-    """Wakatime hook.
-
-    :param args:
-    :type args: Any
-    :param kwargs:
-    :type kwargs: Any
-    :rtype: None
-    """
-    task = Thread(target=send_wakatime_heartbeat, args=args, kwargs=kwargs)
-    task.daemon = True
-    task.start()
+__all__ = ["__version__", "__version_tuple__"]

src/repl_python_wakatime/hooks/codestats.py

@@ -0,0 +1,166 @@
+"""Codestats
+============
+
+Refer `code-stats-vim
+<https://gitlab.com/code-stats/code-stats-vim/-/blob/master/pythonx/codestats.py>
+`_.
+"""
+import json
+import logging
+import threading
+import time
+from datetime import datetime
+from http.client import HTTPException
+from socket import gethostname
+from ssl import CertificateError
+from typing import NoReturn
+from urllib.error import URLError
+from urllib.request import Request, urlopen
+
+logger = logging.getLogger(__name__)
+INTERVAL = 10  # interval at which stats are sent
+SLEEP_INTERVAL = 0.1  # sleep interval for timeslicing
+TIMEOUT = 2  # request timeout value (in seconds)
+
+codestats = None
+
+
+def codestats_hook(
+    api_key: str = "",
+    url: str = "https://codestats.net/api/my/pulses",
+    language_type: str = "Terminal (python)",
+    service_name: str = "codestats",
+    user_name: str = gethostname(),
+) -> None:
+    """Codestats hook.
+
+    :param api_key:
+    :type api_key: str
+    :param url:
+    :type url: str
+    :param language_type:
+    :type language_type: str
+    :param service_name:
+    :type service_name: str
+    :param user_name:
+    :type user_name: str
+    :rtype: None
+    """
+    global codestats
+    if codestats is None:
+        if api_key == "":
+            from ..utils.api import get_api_key
+
+            api_key = get_api_key(service_name, user_name)
+        codestats = CodeStats(api_key, url, language_type)
+    codestats.add_xp()
+
+
+class CodeStats:
+    """Codestats."""
+
+    def __init__(
+        self,
+        api_key: str,
+        url: str = "https://codestats.net/api/my/pulses",
+        language_type: str = "Terminal (python)",
+    ) -> None:
+        """Init.
+
+        :param api_key:
+        :type api_key: str
+        :param url:
+        :type url: str
+        :param language_type:
+        :type language_type: str
+        :rtype: None
+        """
+        self.url = url
+        self.api_key = api_key
+        self.language_type = language_type
+        self.xp_dict = {language_type: 0}
+
+        self.sem = threading.Semaphore()
+
+        self.cs_thread = threading.Thread(target=self.main_thread)
+        self.cs_thread.daemon = True
+        self.cs_thread.start()
+
+    def add_xp(self, xp: int = 1) -> None:
+        """Add xp.
+
+        Sem sections are super small so this should be quick if it blocks.
+
+        :param xp:
+        :type xp: int
+        :rtype: None
+        """
+        self.sem.acquire()
+        self.xp_dict[self.language_type] += xp
+        self.sem.release()
+
+    def send_xp(self) -> None:
+        """Send xp.
+
+        Acquire the lock to get the list of xp to send.
+
+        :rtype: None
+        """
+        if len(self.xp_dict) == 0:
+            return
+
+        self.sem.acquire()
+        xp_list = [dict(language=ft, xp=xp) for ft, xp in self.xp_dict.items()]
+        self.xp_dict = {self.language_type: 0}
+        self.sem.release()
+
+        headers = {
+            "Content-Type": "application/json",
+            "User-Agent": "code-stats-python/{0}".format(__version__),
+            "X-API-Token": self.api_key,
+            "Accept": "*/*",
+        }
+
+        # after lock is released we can send the payload
+        utc_now = datetime.now().astimezone().isoformat()
+        pulse_json = json.dumps(
+            {"coded_at": "{0}".format(utc_now), "xps": xp_list}
+        ).encode("utf-8")
+        req = Request(url=self.url, data=pulse_json, headers=headers)
+        error = ""
+        try:
+            response = urlopen(req, timeout=TIMEOUT)  # nosec: B310
+            response.read()
+            # connection might not be closed without .read()
+        except URLError as e:
+            try:
+                # HTTP error
+                error = "{0} {1}".format(
+                    e.code,  # type: ignore
+                    e.read().decode("utf-8"),  # type: ignore
+                )
+            except AttributeError:
+                # non-HTTP error, eg. no network
+                error = e.reason
+        except CertificateError as e:
+            # SSL certificate error (eg. a public wifi redirects traffic)
+            error = e
+        except HTTPException as e:
+            error = "HTTPException on send data. Msg: {0}\nDoc?:{1}".format(
+                e.message, e.__doc__  # type: ignore
+            )
+        if error:
+            logger.error(error)
+
+    def main_thread(self) -> NoReturn:
+        """Run main thread.
+
+        Needs to be able to send XP at an interval.
+        """
+        while True:
+            cur_time = 0
+            while cur_time < INTERVAL:
+                time.sleep(SLEEP_INTERVAL)
+                cur_time += SLEEP_INTERVAL
+
+            self.send_xp()