.gitea/workflows/test.yml

@@ -0,0 +1,57 @@
+name: Tests
+
+on:
+  pull_request:
+    branches: [main]
+  push:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:17
+        env:
+          POSTGRES_DB: matitos
+          POSTGRES_USER: supermatitos
+          POSTGRES_PASSWORD: supermatitos
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    env:
+      DJANGO_SECRET_KEY: ci-test-secret
+      DJANGO_ALLOWED_HOSTS: '*'
+      DJANGO_ALLOWED_ORIGINS: 'http://localhost'
+      DB_NAME: matitos
+      DB_USER: supermatitos
+      DB_PASSWORD: supermatitos
+      DB_HOST: localhost
+      DB_PORT: '5432'
+      REDIS_CACHE_HOST: localhost
+      REDIS_CACHE_PORT: '6379'
+      REDIS_CELERY_HOST: localhost
+      REDIS_CELERY_PORT: '6379'
+      PATH_LOGS_DIRECTORY: /tmp/logs
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+          cache: 'pip'
+          cache-dependency-path: app_urls/requirements.txt
+
+      - name: Install dependencies
+        run: grep -v '^git+' app_urls/requirements.txt | pip install -r /dev/stdin
+
+      - name: Run tests
+        working-directory: app_urls
+        run: python manage.py test fetcher.tests -v 2

app_urls/core/settings.py

@@ -123,7 +123,8 @@ CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers.DatabaseScheduler'
 
 CELERY_TASK_QUEUES = (
     Queue('default'),
-    Queue('low'),
+    Queue('light'),
+    Queue('heavy'),
 )
 
 

app_urls/fetcher/src/db_utils.py

@@ -212,7 +212,7 @@ class DB_Handler():
             # Sort pattern tuples by priority. (pattern, priority, status)
             for regex_pattern, regex_priority, status_if_match in sorted(list_pattern_status_tuple, key=lambda tup: tup[1], reverse=True):
                 # Regular expression pattern matching: https://regexr.com/
-                if bool(re.match(regex_pattern, obj_url.url)):
+                if bool(re.match(regex_pattern, url)):
                     # logger.debug("Regex pattern found, status '{}' for URL: {}".format(status_if_match, url))
                     return status_if_match
             return None

app_urls/fetcher/src/fetch_search_instances.py

@@ -8,10 +8,26 @@ from .logger import get_logger
 logger = get_logger()
 
 from furl import furl
-from gnews import GNews
 from ddgs import DDGS
-from GoogleNews import GoogleNews
+
-from search_engines import Yahoo, Aol
+try:
+    from gnews import GNews
+except ImportError:
+    GNews = None
+    logger.warning("gnews not installed — SearchGNews disabled")
+
+try:
+    from GoogleNews import GoogleNews
+except ImportError:
+    GoogleNews = None
+    logger.warning("GoogleNews not installed — SearchGoogleNews/SearchGoogleGeneral/SearchGoogleNewsRSS disabled")
+
+try:
+    from search_engines import Yahoo, Aol
+except ImportError:
+    Yahoo = None
+    Aol = None
+    logger.warning("search_engines not installed — SearchYahooGeneral/SearchAOLGeneral disabled")
 
 ###########################################################################
 ###########################################################################

app_urls/fetcher/templates/url_detail.html

@@ -246,7 +246,7 @@
             </tr>
             <tr>
                 <th>Image URLs</th>
-                <td>{{ url_content.image_urls|default:"" }}</td>
+                <td>{{ url_content.images_url|default:"" }}</td>
             </tr>
             <tr>
                 <th>Video URLs</th>

app_urls/fetcher/tests.py

@@ -1,3 +0,0 @@
-from django.test import TestCase
-
-# Create your tests here.

app_urls/fetcher/tests/test_config.py

@@ -0,0 +1,68 @@
+import json
+import os
+
+from django.conf import settings
+from django.test import SimpleTestCase
+
+SCHEDULED_TASKS_PATH = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
+    "scheduled_tasks.json",
+)
+
+
+class ScheduledTasksFixtureTests(SimpleTestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        with open(SCHEDULED_TASKS_PATH) as f:
+            cls.data = json.load(f)
+        cls.periodic_tasks = [
+            e
+            for e in cls.data
+            if e.get("model") == "django_celery_beat.periodictask"
+        ]
+
+    def test_json_is_valid_and_loads(self):
+        self.assertIsInstance(self.data, list)
+        self.assertTrue(len(self.data) > 0)
+
+    def test_periodic_task_primary_keys_are_unique(self):
+        pks = [e["pk"] for e in self.periodic_tasks]
+        self.assertEqual(len(pks), len(set(pks)))
+
+    def test_notify_status_has_unique_pk(self):
+        notify = [
+            e
+            for e in self.periodic_tasks
+            if e["fields"].get("task") == "fetcher.tasks.notify_status"
+        ]
+        self.assertEqual(len(notify), 1)
+        other_pks = {
+            e["pk"]
+            for e in self.periodic_tasks
+            if e["fields"].get("task") != "fetcher.tasks.notify_status"
+        }
+        self.assertNotIn(notify[0]["pk"], other_pks)
+
+    def test_interval_schedule_primary_keys_are_unique(self):
+        schedules = [e for e in self.data if e["model"] == "django_celery_beat.intervalschedule"]
+        pks = [e["pk"] for e in schedules]
+        self.assertEqual(len(pks), len(set(pks)))
+
+    def test_expected_critical_tasks_present(self):
+        tasks = {e["fields"].get("task") for e in self.periodic_tasks}
+        self.assertIn("fetcher.tasks.process_raw_urls", tasks)
+        self.assertIn("fetcher.tasks.fetch_feeds", tasks)
+        self.assertIn("fetcher.tasks.notify_status", tasks)
+
+
+class CeleryQueueSettingsTests(SimpleTestCase):
+    def test_contains_worker_queues(self):
+        queues = [q.name for q in settings.CELERY_TASK_QUEUES]
+        self.assertIn("default", queues)
+        self.assertIn("light", queues)
+        self.assertIn("heavy", queues)
+
+    def test_does_not_contain_deprecated_low_queue(self):
+        queues = [q.name for q in settings.CELERY_TASK_QUEUES]
+        self.assertNotIn("low", queues)

app_urls/fetcher/tests/test_publish.py

@@ -0,0 +1,44 @@
+from django.test import RequestFactory, SimpleTestCase
+from django.urls import resolve, reverse
+from unittest.mock import patch
+
+from .. import views
+
+
+class PublishRouteTests(SimpleTestCase):
+    def setUp(self):
+        self.factory = RequestFactory()
+
+    def test_publish_url_resolves_to_publish_view(self):
+        match = resolve("/task/publish_42/")
+        self.assertEqual(match.func, views.publish)
+        self.assertEqual(match.kwargs, {"id": 42})
+
+    def test_publish_url_reverse(self):
+        url = reverse("publish", args=[7])
+        self.assertEqual(url, "/task/publish_7/")
+
+    @patch("fetcher.views.Publisher")
+    def test_publish_invokes_publisher(self, mock_publisher_cls):
+        mock_instance = mock_publisher_cls.return_value
+        mock_instance.publish.return_value = None
+
+        request = self.factory.get("/task/publish_9/")
+        response = views.publish(request, 9)
+
+        self.assertEqual(response.status_code, 200)
+        mock_publisher_cls.assert_called_once_with()
+        mock_instance.publish.assert_called_once_with(9)
+        self.assertIn(b"9", response.content)
+
+    @patch("fetcher.views.Publisher")
+    def test_publish_reports_error_gracefully(self, mock_publisher_cls):
+        mock_instance = mock_publisher_cls.return_value
+        mock_instance.publish.side_effect = RuntimeError("boom")
+
+        request = self.factory.get("/task/publish_3/")
+        response = views.publish(request, 3)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"Error publishing", response.content)
+        mock_instance.publish.assert_called_once_with(3)

app_urls/fetcher/tests/test_search_postprocess.py

@@ -0,0 +1,45 @@
+from django.test import SimpleTestCase
+
+from ..models import Search
+from ..src.fetch_search_instances import FetcherAbstract
+
+
+class _ConcreteFetcher(FetcherAbstract):
+    def _fetch_raw_urls(self, keyword_search):
+        raise NotImplementedError
+
+    def _get_name(self):
+        return "test-fetcher"
+
+
+class PostProcessUrlsTests(SimpleTestCase):
+    def setUp(self):
+        self.fetcher = _ConcreteFetcher()
+
+    def test_url_host_filters_to_host(self):
+        obj_search = Search(search="cnbc.com", type=Search.TYPE_ENUM.URL_HOST)
+        raw = [
+            "https://www.cnbc.com/article/1",
+            "https://www.foxnews.com/article/2",
+            "cnbc.com/article/3",
+        ]
+        result = self.fetcher._post_process_urls(raw, obj_search)
+        self.assertEqual(
+            result,
+            ["https://www.cnbc.com/article/1", "cnbc.com/article/3"],
+        )
+
+    def test_keyword_search_keeps_all_urls(self):
+        obj_search = Search(search="child abuse", type=Search.TYPE_ENUM.KEYWORD_SEARCH)
+        raw = [
+            "https://www.cnbc.com/article/1",
+            "https://www.foxnews.com/article/2",
+        ]
+        result = self.fetcher._post_process_urls(raw, obj_search)
+        self.assertEqual(result, raw)
+
+    def test_query_parameters_are_removed(self):
+        obj_search = Search(search="cnbc.com", type=Search.TYPE_ENUM.URL_HOST)
+        raw = ["https://www.cnbc.com/article/1?param=1234&h=yes"]
+        result = self.fetcher._post_process_urls(raw, obj_search)
+        self.assertEqual(result, ["https://www.cnbc.com/article/1"])

app_urls/fetcher/tests/test_template_field.py

@@ -0,0 +1,62 @@
+from datetime import datetime
+
+from django.test import SimpleTestCase
+from django.template.loader import get_template
+from types import SimpleNamespace
+
+
+def _build_url_item():
+    return SimpleNamespace(
+        id=1,
+        url="https://example.com/article",
+        ts_fetch=datetime(2025, 7, 1, 12, 0, 0),
+        status="valid",
+    )
+
+
+def _build_url_content(images_url=None):
+    return SimpleNamespace(
+        date_published=datetime(2025, 7, 1, 11, 0, 0),
+        title="Example title",
+        description="Example description",
+        content="Example body",
+        valid_content=True,
+        language="en",
+        tags=["a", "b"],
+        authors=["Author"],
+        keywords=["kw"],
+        image_main_url="https://example.com/img.jpg",
+        images_url=images_url,
+        videos_url=["https://example.com/v.mp4"],
+        url_host="https://example.com",
+        site_name="Example",
+    )
+
+
+class UrlDetailTemplateTests(SimpleTestCase):
+    def _render(self, url_content):
+        template = get_template("url_detail.html")
+        context = {
+            "url_item": _build_url_item(),
+            "url_content": url_content,
+            "sources": ["feeds"],
+            "searches": ["child abuse"],
+            "models": ["llama3.2:3b"],
+            "prompt": "Summarize",
+            "url_canonical": None,
+        }
+        return template.render(context)
+
+    def test_image_urls_field_renders(self):
+        html = self._render(_build_url_content(images_url=["https://example.com/1.jpg"]))
+        self.assertIn("https://example.com/1.jpg", html)
+
+    def test_legacy_image_urls_attribute_is_not_referenced(self):
+        # Model exposes `images_url`, not `image_urls`. Rendering the template
+        # must not fail nor silently rely on a non-existent field.
+        with open(
+            "fetcher/templates/url_detail.html"
+        ) as f:
+            source = f.read()
+        self.assertNotIn("url_content.image_urls", source)
+        self.assertIn("url_content.images_url", source)

app_urls/fetcher/tests/test_utils.py

@@ -0,0 +1,22 @@
+from django.test import SimpleTestCase
+from ..src.utils import get_with_protocol
+
+
+class GetWithProtocolTests(SimpleTestCase):
+    def test_missing_protocol_prepends_https(self):
+        self.assertEqual(get_with_protocol("example.com"), "https://example.com")
+
+    def test_missing_protocol_with_path(self):
+        self.assertEqual(
+            get_with_protocol("example.com/article"),
+            "https://example.com/article",
+        )
+
+    def test_upgrades_http_to_https(self):
+        self.assertEqual(get_with_protocol("http://example.com"), "https://example.com")
+
+    def test_preserves_https(self):
+        self.assertEqual(get_with_protocol("https://example.com"), "https://example.com")
+
+    def test_empty_string_becomes_https(self):
+        self.assertEqual(get_with_protocol(""), "https://")

app_urls/fetcher/urls.py

@@ -17,4 +17,5 @@ urlpatterns = [
     #
     path('urls/', views.filtered_urls, name='filtered_urls'),
     path('urls/<int:id>/', views.url_detail_view, name='url_detail'),
+    path('task/publish_<int:id>/', views.publish, name='publish'),
 ]

app_urls/fetcher/views.py

@@ -2,12 +2,13 @@ from .views_base import link_list, logs, log_db #, trigger_task,
 
 from django.core.paginator import Paginator
 from django.shortcuts import render, get_object_or_404
-from django.http import StreamingHttpResponse, JsonResponse
+from django.http import StreamingHttpResponse, JsonResponse, HttpResponse
 from django.db.models import Q, Count
 from django.utils import timezone
 from django.utils.timezone import now, timedelta
 from .models import Urls, Source, Search, UrlContent, UrlsSourceSearch, UrlsDuplicate
 from .src.llm import OllamaClient
+from .src.publisher import Publisher
 import json
 
 
@@ -62,6 +63,15 @@ def url_detail_view(request, id):
     }
     return render(request, 'url_detail.html', context)
 
+def publish(request, id):
+    # Publish URL content to Ghost
+    try:
+        Publisher().publish(id)
+        message = "URL ID {} published".format(id)
+    except Exception as e:
+        message = "Error publishing URL ID {}: {}".format(id, str(e))
+    return HttpResponse(message)
+
 ####################################################################################################
 def charts(request):
     return render(request, 'charts.html')

app_urls/init_db.py

@@ -210,10 +210,9 @@ def initialize_data():
             for list_pattern_status_priority in data_json.get("REGEX_PATTERN_STATUS_PRIORITY", []):
                 # Decode
                 pattern, status, priority = list_pattern_status_priority
-                # Query
+                # Query (parameterized to avoid SQL injection)
-                query = "INSERT INTO STATUS_PATTERN_MATCHING (pattern, priority, status) VALUES ('{}', {}, '{}');".format(pattern, priority, status)
+                query = "INSERT INTO STATUS_PATTERN_MATCHING (pattern, priority, status) VALUES (%s, %s, %s);"
-                print(query)
+                cur.execute(query, (pattern, priority, status))
-                cur.execute(query)
 
     # Connect to an existing database
     with psycopg.connect(connection_info) as conn:
@@ -222,9 +221,8 @@ def initialize_data():
             # Feeds, URL host, keyword search
             for search_type, list_searches in data_json.get("SEARCH", {}).items():
                 for search in list_searches:
-                    query = "INSERT INTO SEARCH (search, type) VALUES ('{}', '{}');".format(search, search_type)
+                    insert_search_sql = "INSERT INTO SEARCH (search, type) VALUES (%s, %s);"
-                    print(query)
+                    cur.execute(insert_search_sql, (search, search_type))
-                    cur.execute(query)
 
                     # Try finding RSS feed
                     if (search_type == "url_host"):
@@ -232,9 +230,8 @@ def initialize_data():
                         list_feeds = find_feeds(url_host)
                         # If not exists, insert feed
                         for feed in list_feeds:
-                            query = "INSERT INTO SEARCH (search, type) VALUES ('{}', '{}') ON CONFLICT DO NOTHING;".format(feed, "rss_feed")
+                            insert_feed_sql = "INSERT INTO SEARCH (search, type) VALUES (%s, %s) ON CONFLICT DO NOTHING;"
-                            print(query)
+                            cur.execute(insert_feed_sql, (feed, "rss_feed"))
-                            cur.execute(query)
 
 
 if __name__ == '__main__':

app_urls/scheduled_tasks.json

@@ -393,7 +393,7 @@
   },
   {
     "model": "django_celery_beat.periodictask",
-    "pk": 4,
+    "pk": 15,
     "fields": {
       "name": "Notify status",
       "task": "fetcher.tasks.notify_status",