import json from pathlib import Path from django.conf import settings from django.contrib import admin, messages from django.shortcuts import redirect, render from django.urls import path, reverse from django.utils.html import format_html from blogs.models import BlogArticle, BlogArticleImage, BlogGenerationLog, BlogSettings from blogs.services.automation_service import BlogAutomationService from blogs.services.shopify_blog_service import ShopifyBlogService class BlogArticleImageInline(admin.TabularInline): model = BlogArticleImage extra = 0 readonly_fields = ( "image_url", "local_file", "alt_text", "position", "generation_status", "prompt_used", ) def has_add_permission(self, request, obj=None): return False class BlogGenerationLogInline(admin.TabularInline): model = BlogGenerationLog extra = 0 can_delete = False classes = ("collapse",) ordering = ("-created_at",) readonly_fields = ("step", "status", "details_display", "created_at") fields = ("step", "status", "details_display", "created_at") verbose_name_plural = "Generation log" @admin.display(description="Details") def details_display(self, obj): if not obj.details: return "—" text = json.dumps(obj.details, indent=2, ensure_ascii=False) return format_html( '
{}
', text, ) def has_add_permission(self, request, obj=None): return False @admin.register(BlogArticle) class BlogArticleAdmin(admin.ModelAdmin): list_display = ( "title", "target_keyword", "status", "meta_title", "scheduled_date", "shopify_article_id", "posted_at", ) list_filter = ("status", "calendar_week") search_fields = ("title", "target_keyword", "slug", "shopify_article_id") readonly_fields = ( "content", "meta_title", "meta_description", "tags", "slug", "shopify_article_url", "shopify_metafields", "word_count", "keyword_density", "generation_error", "posted_at", "created_at", "updated_at", ) fieldsets = ( ( None, { "fields": ( "title", "status", "target_keyword", "secondary_keywords", "scheduled_date", "calendar_week", "locale", "generation_error", ) }, ), ( "Content & SEO", { "fields": ( "slug", "meta_title", "meta_description", "tags", "content", "word_count", "keyword_density", ) }, ), ( "Shopify", { "fields": ( "shopify_article_id", "shopify_article_url", "shopify_metafields", "featured_image_url", "posted_at", ) }, ), ( "Timestamps", {"fields": ("created_at", "updated_at"), "classes": ("collapse",)}, ), ) inlines = [BlogArticleImageInline, BlogGenerationLogInline] def has_add_permission(self, request): return False def get_inline_instances(self, request, obj=None): instances = super().get_inline_instances(request, obj) if obj and (obj.generation_error or obj.status == BlogArticle.STATUS_FAILED): for inline in instances: if isinstance(inline, BlogGenerationLogInline): inline.classes = () return instances def delete_model(self, request, obj): shopify_id = (obj.shopify_article_id or "").strip() super().delete_model(request, obj) if shopify_id: messages.success( request, f"Deleted from Django and Shopify (article id {shopify_id}).", ) def delete_queryset(self, request, queryset): shopify_ids = [ (obj.shopify_article_id or "").strip() for obj in queryset if (obj.shopify_article_id or "").strip() ] super().delete_queryset(request, queryset) if shopify_ids: messages.success( request, f"Deleted {len(shopify_ids)} article(s) from Django and Shopify.", ) @admin.register(BlogSettings) class BlogSettingsAdmin(admin.ModelAdmin): """Singleton settings — separate admin page for blog automation.""" change_form_template = "admin/blogs/blogsettings/change_form.html" readonly_fields = ( "external_blog_inventory_display", "external_blog_inventory_updated_at", "external_blog_inventory_error", "trending_topics_display", "trending_topics_updated_at", "trending_topics_error", "last_run_at", "last_run_status", "last_run_error", "updated_at", ) fieldsets = ( ( "Automation", { "fields": ( "enabled", "posts_per_week", "include_images", "last_run_at", "last_run_status", "last_run_error", ) }, ), ( "Shopify", { "fields": ( "shopify_blog_id", "store_name", "store_domain", "blog_author_name", ) }, ), ( "Existing blog (live site)", {"fields": ("blog_list_url",)}, ), ( "Store context", { "fields": ( "description", "language", "tone_description", "target_audience", ) }, ), ( "Existing blogs (cached by weekly cron)", { "classes": ("collapse",), "fields": ( "external_blog_inventory_display", "external_blog_inventory_updated_at", "external_blog_inventory_error", ), }, ), ( "Trending topics (cached by weekly cron)", { "classes": ("collapse",), "fields": ( "trending_topics_display", "trending_topics_updated_at", "trending_topics_error", ), }, ), ) @admin.display(description="External blog inventory") def external_blog_inventory_display(self, obj): data = obj.external_blog_inventory or [] return format_html( '
{}
', json.dumps(data, indent=2, ensure_ascii=False)[:12000] or "[]", ) @admin.display(description="Trending topics") def trending_topics_display(self, obj): data = obj.trending_topics or [] return format_html( '
{}
', json.dumps(data, indent=2, ensure_ascii=False)[:12000] or "[]", ) def has_add_permission(self, request): return False def has_delete_permission(self, request, obj=None): return False def changelist_view(self, request, extra_context=None): return redirect(reverse("admin:blogs_blogsettings_change", args=[1])) def get_urls(self): urls = super().get_urls() custom = [ path( "actions/", self.admin_site.admin_view(self.actions_view), name="blogs_blogsettings_actions", ), ] return custom + urls def changeform_view(self, request, object_id, form_url="", extra_context=None): extra_context = extra_context or {} extra_context["actions_url"] = reverse("admin:blogs_blogsettings_actions") try: extra_context["shopify_blogs"] = ShopifyBlogService.list_blogs() except Exception: extra_context["shopify_blogs"] = [] return super().changeform_view(request, object_id, form_url, extra_context) def actions_view(self, request): settings_obj = BlogSettings.load() context = { **self.admin_site.each_context(request), "opts": self.model._meta, "settings": settings_obj, "settings_url": reverse("admin:blogs_blogsettings_change", args=[1]), "live_count": len(settings_obj.external_blog_inventory or []), "trending_count": len(settings_obj.trending_topics or []), } if request.method == "POST": action = request.POST.get("action") if action == "test_shopify": try: ok = ShopifyBlogService.test_connection(settings_obj.shopify_blog_id) if ok: messages.success(request, "Shopify blog connection OK.") else: messages.error(request, "Shopify connection failed.") except Exception as exc: messages.error(request, f"Connection error: {exc}") elif action == "generate_and_publish_article": background = request.POST.get("background") == "1" try: result = BlogAutomationService.generate_and_publish_article( background=background ) article_url = reverse( "admin:blogs_blogarticle_change", args=[result["article_id"]] ) if background: messages.success( request, format_html( "Planned “{}”. Background job started (pid {}). " 'Log: {}. View article', result.get("title", ""), result.get("worker_pid", ""), result.get("log_path", ""), article_url, ), ) else: messages.success( request, format_html( 'Published “{}”. Shopify ' '| View article', result.get("title", ""), result.get("shopify_article_url") or "#", article_url, ), ) except Exception as exc: messages.error(request, f"Generate and publish failed: {exc}") elif action == "run_now": limit = int(request.POST.get("limit") or settings_obj.posts_per_week) sync = request.POST.get("sync") == "1" dry_run = request.POST.get("dry_run") == "1" try: result = BlogAutomationService.run_weekly( limit=limit, sync=sync, dry_run=dry_run ) if dry_run: messages.success(request, f"Dry run: {result}") elif sync: messages.success(request, f"Run complete (sync): {result}") else: messages.success( request, "Weekly plan saved (research + trends + topics). " f"Scheduled dates: {result.get('scheduled_dates', [])}. " "Generation runs on each scheduled day.", ) except Exception as exc: messages.error(request, f"Run failed: {exc}") return redirect("admin:blogs_blogsettings_actions") return render(request, "admin/blogs/settings_actions.html", context)