Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions .github/workflows/build.yml
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,18 @@ name: Build
on: [push]

jobs:
ruff:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v5
- uses: actions/setup-python@v6
with:
python-version: "3.14"
- uses: astral-sh/setup-uv@v6
- run: uv sync --locked --all-extras --dev
- run: uv run ruff check --output-format=github --target-version=py314
- run: uv run ruff format --diff --target-version=py314

test:
runs-on: ubuntu-latest
strategy:
Expand Down
5 changes: 5 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,8 @@ watch-diff = "watch_diff.__main__:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[dependency-groups]
dev = [
"ruff>=0.14.6",
]
29 changes: 13 additions & 16 deletions tests/test_watch_diff.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,15 @@
"""
"""

import os
import unittest

import watch_diff


smtp_host = os.environ.get('SMTP_HOST')
smtp_port = os.environ.get('SMTP_PORT')
smtp_user = os.environ.get('SMTP_USER')
smtp_pass = os.environ.get('SMTP_PASS')
smtp_host = os.environ.get("SMTP_HOST")
smtp_port = os.environ.get("SMTP_PORT")
smtp_user = os.environ.get("SMTP_USER")
smtp_pass = os.environ.get("SMTP_PASS")


class TestWatchDiff(unittest.TestCase):

def test_api_available(self):
self.assertTrue(watch_diff.Command)
self.assertTrue(watch_diff.Diff)
Expand All @@ -25,16 +20,18 @@ def test_api_available(self):
self.assertTrue(watch_diff.OutputFormatting)

def test_command(self):
c = watch_diff.Command('date')
c = watch_diff.Command("date")
self.assertFalse(c)
d = c.run()
self.assertTrue(c)
self.assertTrue(d)

@unittest.skipIf(not smtp_host, 'SMTP_HOST is not available')
@unittest.skipIf(not smtp_port, 'SMTP_PORT is not available')
@unittest.skipIf(not smtp_user, 'SMTP_USER is not available')
@unittest.skipIf(not smtp_pass, 'SMTP_PASS is not available')
@unittest.skipIf(not smtp_host, "SMTP_HOST is not available")
@unittest.skipIf(not smtp_port, "SMTP_PORT is not available")
@unittest.skipIf(not smtp_user, "SMTP_USER is not available")
@unittest.skipIf(not smtp_pass, "SMTP_PASS is not available")
def test_email(self):
e = watch_diff.Email(smtp_host, smtp_port, smtp_user, smtp_pass, 'watch-diff-tests', smtp_user)
e.send_email('watch diff tests', 'text', 'html')
e = watch_diff.Email(
smtp_host, smtp_port, smtp_user, smtp_pass, "watch-diff-tests", smtp_user
)
e.send_email("watch diff tests", "text", "html")
36 changes: 36 additions & 0 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

27 changes: 18 additions & 9 deletions watch_diff/__init__.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,19 @@
"""
"""
from watch_diff.command import Command
from watch_diff.diff import Diff
from watch_diff.email import Email
from watch_diff.format import (
ConsoleFormatter,
DefaultFormatter,
HTMLFormatter,
OutputFormatting,
)

__version__ = '0.6.0'


from .command import Command
from .diff import Diff
from .email import Email
from .format import DefaultFormatter, ConsoleFormatter, HTMLFormatter, OutputFormatting
__all__ = [
"Command",
"Diff",
"Email",
"ConsoleFormatter",
"DefaultFormatter",
"HTMLFormatter",
"OutputFormatting",
]
90 changes: 57 additions & 33 deletions watch_diff/__main__.py
Original file line number Diff line number Diff line change
@@ -1,32 +1,48 @@
"""
"""

import argparse
import datetime
import getpass
import logging
import os
import time

from email.utils import make_msgid

from . import command
from . import email

from watch_diff import command, email

logger = logging.getLogger(__name__)

parser = argparse.ArgumentParser(description='Watch command output and get notified on changes')
logging_group = parser.add_argument_group('logging level').add_mutually_exclusive_group()
logging_group.add_argument('-v', '--verbose', action='store_const',
const=logging.INFO, default=logging.CRITICAL,
dest='loglevel', help='enable verbose output')
logging_group.add_argument('-d', '--debug', action='store_const',
const=logging.DEBUG, dest='loglevel',
help='show debugging statements')
parser.add_argument('-i', '--interval', type=int, default=5, metavar='SECONDS', help='number of seconds between executions')
parser.add_argument('-r', '--recipient', help='send email to recipient')
parser.add_argument('command', help='the command to watch')
parser = argparse.ArgumentParser(
description="Watch command output and get notified on changes"
)
logging_group = parser.add_argument_group(
"logging level"
).add_mutually_exclusive_group()
logging_group.add_argument(
"-v",
"--verbose",
action="store_const",
const=logging.INFO,
default=logging.CRITICAL,
dest="loglevel",
help="enable verbose output",
)
logging_group.add_argument(
"-d",
"--debug",
action="store_const",
const=logging.DEBUG,
dest="loglevel",
help="show debugging statements",
)
parser.add_argument(
"-i",
"--interval",
type=int,
default=5,
metavar="SECONDS",
help="number of seconds between executions",
)
parser.add_argument("-r", "--recipient", help="send email to recipient")
parser.add_argument("command", help="the command to watch")


def _main():
Expand All @@ -35,43 +51,51 @@ def _main():
e = None

if args.recipient:
smtp_host = os.environ.get('SMTP_HOST') or input('SMTP_HOST: ')
smtp_port = os.environ.get('SMTP_PORT') or input('SMTP_PORT: ')
smtp_user = os.environ.get('SMTP_USER') or input('SMTP_USER: ')
smtp_pass = os.environ.get('SMTP_PASS') or getpass.getpass('SMTP_PASS: ')
e = email.Email(smtp_host, smtp_port, smtp_user, smtp_pass, 'watch-diff', args.recipient)
smtp_host = os.environ.get("SMTP_HOST") or input("SMTP_HOST: ")
smtp_port = os.environ.get("SMTP_PORT") or input("SMTP_PORT: ")
smtp_user = os.environ.get("SMTP_USER") or input("SMTP_USER: ")
smtp_pass = os.environ.get("SMTP_PASS") or getpass.getpass("SMTP_PASS: ")
e = email.Email(
smtp_host, smtp_port, smtp_user, smtp_pass, "watch-diff", args.recipient
)

first_run = True
c = command.Command(args.command)
previous_msg_id = None

while True:
now = str(datetime.datetime.now())
logger.info('executing command with time {}'.format(now))
logger.info("executing command with time {}".format(now))
diff = c.run(now)

if first_run:
print('[{}] first_run:'.format(now))
print("[{}] first_run:".format(now))
print(c.to_console())
subject = 'watch-diff first_run: {}'.format(args.command)
subject = "watch-diff first_run: {}".format(args.command)
if e:
logger.info('sending first_run email to {}'.format(args.recipient))
logger.info("sending first_run email to {}".format(args.recipient))
msg_id = make_msgid()
e.send_email(subject, str(c), c.to_html(full_html=True), msg_id)
previous_msg_id = msg_id
elif diff:
print('[{}] diff:'.format(now))
print("[{}] diff:".format(now))
print(diff.to_console())
subject = 'watch-diff diff: {}'.format(args.command)
subject = "watch-diff diff: {}".format(args.command)
if e:
logger.info('sending diff email to {}'.format(args.recipient))
logger.info("sending diff email to {}".format(args.recipient))
msg_id = make_msgid()
e.send_email(subject, str(diff), diff.to_html(full_html=True), msg_id, previous_msg_id)
e.send_email(
subject,
str(diff),
diff.to_html(full_html=True),
msg_id,
previous_msg_id,
)
previous_msg_id = msg_id
else:
print('[{}] no diff'.format(now))
print("[{}] no diff".format(now))

logger.info('sleeping for {} seconds'.format(args.interval))
logger.info("sleeping for {} seconds".format(args.interval))
time.sleep(args.interval)
first_run = False

Expand Down
22 changes: 11 additions & 11 deletions watch_diff/command.py
Original file line number Diff line number Diff line change
@@ -1,23 +1,18 @@
"""
"""

import datetime
import subprocess

from . import diff
from . import format
from watch_diff import diff, format


class Command(format.OutputFormatting):

def __init__(self, command):
self._command = command

self._previous_datetime = ''
self._previous_result = ''
self._previous_datetime = ""
self._previous_result = ""

self._current_datetime = ''
self._current_result = ''
self._current_datetime = ""
self._current_result = ""

def __bool__(self):
return bool(self._current_result)
Expand All @@ -26,7 +21,12 @@ def _format(self, formatter=format.DefaultFormatter):
return self._current_result

def _diff(self):
return diff.Diff(self._previous_result, self._current_result, self._previous_datetime, self._current_datetime)
return diff.Diff(
self._previous_result,
self._current_result,
self._previous_datetime,
self._current_datetime,
)

def _run(self):
return subprocess.getoutput(self._command)
Expand Down
43 changes: 31 additions & 12 deletions watch_diff/diff.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,26 @@
"""
"""

import difflib

from . import format
from watch_diff import format


class Diff(format.OutputFormatting):

def __init__(self, a, b, previous_datetime, current_datetime):
self._a = a
self._b = b
self._diff = '\n'.join(difflib.unified_diff(a.splitlines(), b.splitlines(), 'Previous', 'Current', previous_datetime, current_datetime, lineterm='')) or None
self._diff = (
"\n".join(
difflib.unified_diff(
a.splitlines(),
b.splitlines(),
"Previous",
"Current",
previous_datetime,
current_datetime,
lineterm="",
)
)
or None
)

def __bool__(self):
return bool(self._diff)
Expand All @@ -20,12 +29,22 @@ def _format(self, formatter=format.DefaultFormatter):
lines = self._diff.splitlines()
output = []
for line in lines[:2]:
output.append('{}{}{}'.format(formatter.header_start, line, formatter.header_end))
output.append(
"{}{}{}".format(formatter.header_start, line, formatter.header_end)
)
for line in lines[2:]:
if line[0] == '+':
output.append('{}{}{}'.format(formatter.addition_start, line, formatter.addition_end))
elif line[0] == '-':
output.append('{}{}{}'.format(formatter.subtraction_start, line, formatter.subtraction_end))
if line[0] == "+":
output.append(
"{}{}{}".format(
formatter.addition_start, line, formatter.addition_end
)
)
elif line[0] == "-":
output.append(
"{}{}{}".format(
formatter.subtraction_start, line, formatter.subtraction_end
)
)
else:
output.append(line)
return '\n'.join(output)
return "\n".join(output)
Loading