edit sources

repub/model.py

@@ -89,6 +89,59 @@ def source_slug_exists(slug: str) -> bool:
         return Source.select().where(Source.slug == slug).exists()
 
 
+def load_source_form(slug: str) -> dict[str, object] | None:
+    with database.connection_context():
+        source = Source.get_or_none(Source.slug == slug)
+        if source is None:
+            return None
+
+        job = Job.get(Job.source == source)
+        form_data: dict[str, object] = {
+            "name": source.name,
+            "slug": source.slug,
+            "source_type": source.source_type,
+            "notes": source.notes,
+            "spider_arguments": job.spider_arguments,
+            "enabled": job.enabled,
+            "cron_minute": job.cron_minute,
+            "cron_hour": job.cron_hour,
+            "cron_day_of_month": job.cron_day_of_month,
+            "cron_day_of_week": job.cron_day_of_week,
+            "cron_month": job.cron_month,
+            "feed_url": "",
+            "pangea_domain": "",
+            "pangea_category": "",
+            "content_format": "MOBILE_3",
+            "content_type": "articles",
+            "max_articles": "10",
+            "oldest_article": "3",
+            "only_newest": True,
+            "include_authors": True,
+            "exclude_media": False,
+            "include_content": True,
+        }
+        if source.source_type == "feed":
+            feed = SourceFeed.get(SourceFeed.source == source)
+            form_data["feed_url"] = feed.feed_url
+        else:
+            pangea = SourcePangea.get(SourcePangea.source == source)
+            form_data.update(
+                {
+                    "pangea_domain": pangea.domain,
+                    "pangea_category": pangea.category_name,
+                    "content_format": pangea.content_format,
+                    "content_type": pangea.content_type,
+                    "max_articles": str(pangea.max_articles),
+                    "oldest_article": str(pangea.oldest_article),
+                    "only_newest": pangea.only_newest,
+                    "include_authors": pangea.include_authors,
+                    "exclude_media": pangea.exclude_media,
+                    "include_content": pangea.include_content,
+                }
+            )
+        return form_data
+
+
 def create_source(
     *,
     name: str,
@@ -154,6 +207,94 @@ def create_source(
             return source
 
 
+def update_source(
+    source_slug: str,
+    *,
+    name: str,
+    slug: str,
+    source_type: str,
+    notes: str,
+    spider_arguments: str,
+    enabled: bool,
+    cron_minute: str,
+    cron_hour: str,
+    cron_day_of_month: str,
+    cron_day_of_week: str,
+    cron_month: str,
+    feed_url: str = "",
+    pangea_domain: str = "",
+    pangea_category: str = "",
+    content_type: str = "",
+    only_newest: bool = True,
+    max_articles: int | None = None,
+    oldest_article: int | None = None,
+    include_authors: bool = True,
+    exclude_media: bool = False,
+    include_content: bool = True,
+    content_format: str = "",
+) -> Source | None:
+    with database.connection_context():
+        with database.atomic():
+            source = Source.get_or_none(Source.slug == source_slug)
+            if source is None:
+                return None
+
+            source.name = name
+            source.notes = notes
+            source.source_type = source_type
+            source.save()
+
+            job = Job.get(Job.source == source)
+            job.enabled = enabled
+            job.spider_arguments = spider_arguments
+            job.cron_minute = cron_minute
+            job.cron_hour = cron_hour
+            job.cron_day_of_month = cron_day_of_month
+            job.cron_day_of_week = cron_day_of_week
+            job.cron_month = cron_month
+            job.save()
+
+            if source_type == "feed":
+                SourcePangea.delete().where(SourcePangea.source == source).execute()
+                feed = SourceFeed.get_or_none(SourceFeed.source == source)
+                if feed is None:
+                    SourceFeed.create(source=source, feed_url=feed_url)
+                else:
+                    feed.feed_url = feed_url
+                    feed.save()
+            else:
+                SourceFeed.delete().where(SourceFeed.source == source).execute()
+                pangea = SourcePangea.get_or_none(SourcePangea.source == source)
+                if pangea is None:
+                    SourcePangea.create(
+                        source=source,
+                        domain=pangea_domain,
+                        category_name=pangea_category,
+                        content_type=content_type,
+                        only_newest=only_newest,
+                        max_articles=max_articles,
+                        oldest_article=oldest_article,
+                        include_authors=include_authors,
+                        exclude_media=exclude_media,
+                        include_content=include_content,
+                        content_format=content_format,
+                    )
+                else:
+                    pangea.domain = pangea_domain
+                    pangea.category_name = pangea_category
+                    pangea.content_type = content_type
+                    pangea.only_newest = only_newest
+                    pangea.max_articles = max_articles
+                    pangea.oldest_article = oldest_article
+                    pangea.include_authors = include_authors
+                    pangea.exclude_media = exclude_media
+                    pangea.include_content = include_content
+                    pangea.content_format = content_format
+                    pangea.save()
+
+            return source
+
+
 def load_sources() -> tuple[dict[str, object], ...]:
     with database.connection_context():
         sources = tuple(Source.select().order_by(Source.created_at.desc()))