Add webhook feature (ThePornDatabase#272)

* add webhook feature

* refactor

---------

Co-authored-by: sigitosh <¨sigitosh@proton.me¨>
Co-authored-by: DirtyRacer <DirtyRacer@ya.ru>

Author: 3 people

namer/configuration.py

@@ -460,6 +460,16 @@ class NamerConfig:
     Add creation date from failed log to table in web interface
     """
 
+    webhook_enabled: bool = False
+    """
+    Enable webhook notification when a file is successfully renamed
+    """
+
+    webhook_url: str = ''
+    """
+    URL to send HTTP POST notification to when a file is successfully renamed
+    """
+
     cache_session: Optional[CachedSession] = None
     """
     If use_requests_cache is true this http.session will be constructed and used for requests to TPDB.

namer/configuration_utils.py

@@ -308,6 +308,8 @@ def set_boolean(updater: ConfigUpdater, section: str, key: str, value: bool) ->
     'allow_delete_files': ('watchdog', to_bool, from_bool),
     'add_columns_from_log': ('watchdog', to_bool, from_bool),
     'add_complete_column': ('watchdog', to_bool, from_bool),
+    'webhook_enabled': ('webhook', to_bool, from_bool),
+    'webhook_url': ('webhook', None, None),
     'debug': ('watchdog', to_bool, from_bool),
     'console_format': ('watchdog', None, None),
     'manual_mode': ('watchdog', to_bool, from_bool),

namer/namer.cfg.default

@@ -309,3 +309,10 @@ manual_mode = False
 # values in the stack trace, potentially including the porndb token, this setting should only be turned on
 # if you are going to check an logs you share for your token.
 diagnose_errors = False
+
+[webhook]
+# Enable webhook notification when a file is successfully renamed
+webhook_enabled = False
+
+# URL to send HTTP POST notification to when a file is successfully renamed
+webhook_url =

namer/namer.py

@@ -23,6 +23,7 @@
 from namer.database import search_file_in_database, write_file_to_database
 from namer.ffmpeg import FFProbeResults, FFMpeg
 from namer.fileinfo import FileInfo
+from namer.http import Http
 from namer.metadataapi import get_complete_metadataapi_net_fileinfo, get_image, get_trailer, match, share_hash
 from namer.moviexml import parse_movie_xml_file, write_nfo
 from namer.name_formatter import PartialFormatter
@@ -239,7 +240,7 @@ def process_file(command: Command) -> Optional[Command]:
                 target = move_to_final_location(command, new_metadata)
                 tag_in_place(target.target_movie_file, command.config, new_metadata, ffprobe_results)
                 add_extra_artifacts(target.target_movie_file, new_metadata, search_results, phash, command.config)
-
+                send_webhook_notification(target.target_movie_file, command.config)
                 logger.success('Done processing file: {}, moved to {}', command.target_movie_file, target.target_movie_file)
                 return target
         elif command.inplace is False:
@@ -272,6 +273,27 @@ def add_extra_artifacts(video_file: Path, new_metadata: LookedUpFileInfo, search
             write_nfo(video_file, new_metadata, config, trailer, poster, background, phash)
 
 
+def send_webhook_notification(video_file: Path, config: NamerConfig):
+    """
+    Send a webhook notification to the configured URL when a file is successfully renamed.
+    """
+    if not config.webhook_enabled or not config.webhook_url:
+        return
+
+    payload = {
+        'target_movie_file': str(video_file)
+    }
+
+    response = None
+    try:
+        response = Http.post(config.webhook_url, json=payload)
+    except Exception as e:
+        logger.error(f'Failed to send webhook notification: {str(e)}')
+
+    if response and response.ok:
+        logger.info(f'Webhook notification sent successfully to {config.webhook_url}')
+
+
 def check_arguments(file_to_process: Path, dir_to_process: Path, config_override: Path):
     """
     check arguments.

test/namer_webhook_test.py

@@ -0,0 +1,101 @@
+"""
+Test webhook notification functionality in namer.py
+"""
+import logging
+import unittest
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+from namer.namer import send_webhook_notification
+from test.utils import sample_config, environment, new_ea
+
+
+class WebhookTest(unittest.TestCase):
+    """
+    Test the webhook notification functionality.
+    """
+
+    def test_webhook_disabled(self):
+        """
+        Test that no webhook is sent when the feature is disabled.
+        """
+        config = sample_config()
+        config.webhook_enabled = False
+        config.webhook_url = 'http://example.com/webhook'
+        
+        with patch('requests.request') as mock_post:
+            send_webhook_notification(Path('/some/path/movie.mp4'), config)
+            mock_post.assert_not_called()
+
+    def test_webhook_no_url(self):
+        """
+        Test that no webhook is sent when the URL is not configured.
+        """
+        config = sample_config()
+        config.webhook_enabled = True
+        config.webhook_url = ''
+        
+        with patch('requests.request') as mock_post:
+            send_webhook_notification(Path('/some/path/movie.mp4'), config)
+            mock_post.assert_not_called()
+
+    def test_webhook_success(self):
+        """
+        Test that a webhook is successfully sent when enabled and URL is configured.
+        """
+        config = sample_config()
+        config.webhook_enabled = True
+        config.webhook_url = 'http://example.com/webhook'
+        
+        with patch('requests.request') as mock_post:
+            mock_response = MagicMock()
+            mock_response.raise_for_status.return_value = None
+            mock_post.return_value = mock_response
+            
+            send_webhook_notification(Path('/some/path/movie.mp4'), config)
+            
+            mock_post.assert_called_once()
+            # Verify payload structure
+            args, kwargs = mock_post.call_args
+            self.assertEqual(args[0], 'POST')
+            self.assertEqual(args[1], 'http://example.com/webhook')
+            self.assertEqual(kwargs['json'], {'target_movie_file': str(Path('/some/path/movie.mp4'))})
+
+    def test_webhook_failure(self):
+        """
+        Test that webhook errors are properly handled.
+        """
+        config = sample_config()
+        config.webhook_enabled = True
+        config.webhook_url = 'http://example.com/webhook'
+        
+        with patch('requests.request') as mock_post:
+            mock_post.side_effect = Exception('Connection error')
+            
+            # Should not raise an exception
+            send_webhook_notification(Path('/some/path/movie.mp4'), config)
+            
+            mock_post.assert_called_once()
+
+    def test_integration_with_file_processing(self):
+        """
+        Test that webhook is triggered when a file is successfully processed.
+        """
+        with environment() as (tempdir, _fakeTPDB, config):
+            config.webhook_enabled = True
+            config.webhook_url = 'http://example.com/webhook'
+            
+            with patch('namer.namer.send_webhook_notification') as mock_webhook:
+                targets = [new_ea(tempdir, use_dir=False)]
+                
+                # Process the file
+                from namer.namer import main
+                main(['-f', str(targets[0].file), '-c', str(config.config_file)])
+                
+                # Verify webhook was called
+                mock_webhook.assert_called()
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.INFO)
+    unittest.main()