feat: added aw-cli with some utils (#110)

Author: ErikBjare

aw_cli/__main__.py

@@ -0,0 +1,79 @@
+"""
+The idea behind this `aw` or `aw-cli` wrapper script is to act as a collection of helper tools,
+and perhaps even as a way to list and run ActivityWatch modules on a system (a bit like aw-qt, but without the GUI).
+"""
+
+from pathlib import Path
+from datetime import datetime
+import subprocess
+
+import click
+
+from aw_cli.log import find_oldest_log, print_log, LOGLEVELS
+
+
+@click.group()
+@click.option("--testing", is_flag=True)
+def main(testing: bool = False):
+    pass
+
+
+@main.command()
+@click.pass_context
+def qt(ctx):
+    return subprocess.call(
+        ["aw-qt"] + (["--testing"] if ctx.parent.params["testing"] else [])
+    )
+
+
+@main.command()
+def directories():
+    # Print all directories
+    from aw_core.dirs import get_data_dir, get_config_dir, get_cache_dir, get_log_dir
+
+    print("Directory paths used")
+    print(" - config: ", get_config_dir(None))
+    print(" - data:   ", get_data_dir(None))
+    print(" - logs:   ", get_log_dir(None))
+    print(" - cache:  ", get_cache_dir(None))
+
+
+@main.command()
+@click.pass_context
+@click.argument("module_name", type=str, required=False)
+@click.option(
+    "--since",
+    type=click.DateTime(formats=["%Y-%m-%d"]),
+    help="Only show logs since this date",
+)
+@click.option(
+    "--level",
+    type=click.Choice(LOGLEVELS),
+    help="Only show logs of this level, or higher.",
+)
+def logs(ctx, module_name: str = None, since: datetime = None, level: str = None):
+    from aw_core.dirs import get_log_dir
+
+    testing = ctx.parent.params["testing"]
+    logdir: Path = Path(get_log_dir(None))
+
+    # find the oldest logfile in each of the subdirectories in the logging directory, and print the last lines in each one.
+
+    if module_name:
+        print_oldest_log(logdir / module_name, testing, since, level)
+    else:
+        for subdir in sorted(logdir.iterdir()):
+            if subdir.is_dir():
+                print_oldest_log(subdir, testing, since, level)
+
+
+def print_oldest_log(path, testing, since, level):
+    path = find_oldest_log(path, testing)
+    if path:
+        print_log(path, since, level)
+    else:
+        print(f"No logfile found in {path}")
+
+
+if __name__ == "__main__":
+    main()

aw_cli/log.py

@@ -0,0 +1,55 @@
+from pathlib import Path
+from datetime import datetime
+
+
+LOGLEVELS = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
+
+
+def print_log(path: Path, since: datetime = None, level: str = None):
+    if not path.is_file():
+        return
+
+    show_levels = LOGLEVELS[LOGLEVELS.index(level) :] if level else None
+
+    lines_printed = 0
+    with path.open("r") as f:
+        lines = f.readlines()
+        print(f"Logs for module {path.parent.name} ({path.name}, {len(lines)} lines)")
+        for line in lines:
+            if since:
+                try:
+                    linedate = datetime.strptime(line.split(" ")[0], "%Y-%m-%d")
+                except ValueError:
+                    # Could not parse the date, so skip this line
+                    # NOTE: Just because the date could not be parsed, doesn't mean there isn't meaningful info there.
+                    #       Would be better to find the first line after the cutoff, and then just print everything past that.
+                    continue
+                # Skip lines before the date
+                if linedate < since:
+                    continue
+            if level:
+                if not any(level in line for level in show_levels):
+                    continue
+            print(line, end="")
+            lines_printed += 1
+
+    print(f"  (Filtered {lines_printed}/{len(lines)} lines)")
+
+
+def find_oldest_log(path: Path, testing=False) -> Path:
+    if not path.is_dir():
+        return
+
+    logfiles = [
+        f
+        for f in path.iterdir()
+        if f.is_file()
+        and f.name.endswith(".log")
+        and ("testing" in f.name if testing else "testing" not in f.name)
+    ]
+    if not logfiles:
+        return
+
+    logfiles.sort(key=lambda f: f.stat().st_mtime)
+    logfile = logfiles[-1]
+    return logfile

aw_core/dirs.py

@@ -14,7 +14,7 @@ def ensure_path_exists(path: str) -> None:
 
 def _ensure_returned_path_exists(f: GetDirFunc) -> GetDirFunc:
     @wraps(f)
-    def wrapper(subpath: Optional[str]) -> str:
+    def wrapper(subpath: Optional[str] = None) -> str:
         path = f(subpath)
         ensure_path_exists(path)
         return path
@@ -23,24 +23,24 @@ def wrapper(subpath: Optional[str]) -> str:
 
 
 @_ensure_returned_path_exists
-def get_data_dir(module_name: Optional[str]) -> str:
+def get_data_dir(module_name: Optional[str] = None) -> str:
     data_dir = appdirs.user_data_dir("activitywatch")
     return os.path.join(data_dir, module_name) if module_name else data_dir
 
 
 @_ensure_returned_path_exists
-def get_cache_dir(module_name: Optional[str]) -> str:
+def get_cache_dir(module_name: Optional[str] = None) -> str:
     cache_dir = appdirs.user_cache_dir("activitywatch")
     return os.path.join(cache_dir, module_name) if module_name else cache_dir
 
 
 @_ensure_returned_path_exists
-def get_config_dir(module_name: Optional[str]) -> str:
+def get_config_dir(module_name: Optional[str] = None) -> str:
     config_dir = appdirs.user_config_dir("activitywatch")
     return os.path.join(config_dir, module_name) if module_name else config_dir
 
 
 @_ensure_returned_path_exists
-def get_log_dir(module_name: Optional[str]) -> str:  # pragma: no cover
+def get_log_dir(module_name: Optional[str] = None) -> str:  # pragma: no cover
     log_dir = appdirs.user_log_dir("activitywatch")
     return os.path.join(log_dir, module_name) if module_name else log_dir

pyproject.toml

@@ -15,6 +15,9 @@ packages = [
     { include = "aw_query" },
 ]
 
+[tool.poetry.scripts]
+aw-cli = "aw_cli.__main__:main"
+
 [tool.poetry.dependencies]
 python = "^3.7"
 jsonschema = "^4.3"