1923 lines
81 KiB
Python
1923 lines
81 KiB
Python
import ast
|
|
import difflib
|
|
import json
|
|
import logging
|
|
import re
|
|
import requests
|
|
import time
|
|
import jwt
|
|
from bs4 import BeautifulSoup
|
|
|
|
from markupsafe import Markup
|
|
|
|
from odoo import api, fields, models, _
|
|
from odoo.exceptions import UserError, ValidationError
|
|
from odoo.tools import html2plaintext
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# === Input Validation Constants ===
|
|
MAX_KEYWORDS_LENGTH = 500
|
|
MAX_TOPIC_LENGTH = 5000
|
|
MAX_SOURCE_TEXT_LENGTH = 50000
|
|
MAX_CUSTOM_INSTRUCTIONS_LENGTH = 2000
|
|
MAX_REFERENCES = 5
|
|
|
|
|
|
class OtkSeoContent(models.Model):
|
|
_name = 'otk.seo.content'
|
|
_description = 'SEO Content'
|
|
_inherit = ['mail.thread', 'website.seo.metadata']
|
|
_order = 'create_date desc'
|
|
|
|
# === Identification ===
|
|
name = fields.Char('Title', required=True, tracking=True,
|
|
help='The main title for your content. This will be used as the default title if no AI-generated title is created.')
|
|
|
|
content_type = fields.Selection([
|
|
('blog_post', 'Blog Post'),
|
|
('product_desc', 'Product Description'),
|
|
('category_desc', 'Category Description'),
|
|
('landing_page', 'Landing Page Content'),
|
|
('social_post', 'Social Media Post'),
|
|
], string='Content Type', required=True, default='blog_post', tracking=True,
|
|
help='The type of content to generate. Each type has different formatting and structure optimized for its purpose.')
|
|
|
|
# === Source Information ===
|
|
source_keywords = fields.Char('Target Keywords',
|
|
help='Comma-separated keywords to target in the content')
|
|
source_topic = fields.Text('Topic/Brief',
|
|
help='Description of what the content should be about')
|
|
source_product_id = fields.Many2one('product.template', 'Source Product',
|
|
help='Generate content based on this product')
|
|
source_text = fields.Text('Source Content',
|
|
help='Existing content to optimize or use as reference')
|
|
|
|
# === Generation Settings ===
|
|
tone = fields.Selection([
|
|
('professional', 'Professional'),
|
|
('casual', 'Casual'),
|
|
('technical', 'Technical'),
|
|
('persuasive', 'Persuasive'),
|
|
('conversational', 'Conversational'),
|
|
], string='Tone', default='professional',
|
|
help='The writing style and voice for the generated content. Professional is formal and business-like, Casual is friendly and relaxed, Technical is detailed and precise, Persuasive is sales-oriented, and Conversational is engaging and personal.')
|
|
|
|
target_word_count = fields.Selection([
|
|
('short', 'Short (300-500 words)'),
|
|
('medium', 'Medium (500-1000 words)'),
|
|
('long', 'Long (1000-2000 words)'),
|
|
], string='Target Length', default='medium',
|
|
help='The approximate length of the generated content. Longer content typically ranks better for SEO but requires more tokens.')
|
|
|
|
language_id = fields.Many2one('res.lang', 'Language',
|
|
default=lambda self: self._default_language(),
|
|
help='The language in which the content will be generated. The AI will write in this language.')
|
|
|
|
template_id = fields.Many2one('otk.seo.template', 'Content Template',
|
|
domain="[('content_type', '=', content_type)]",
|
|
help='A predefined template that includes system prompts, formatting rules, and default settings. Templates help ensure consistent content structure and style.')
|
|
|
|
brand_voice_id = fields.Many2one('otk.seo.brand.voice', 'Brand Voice',
|
|
default=lambda self: self.env['otk.seo.brand.voice'].get_default_voice(),
|
|
help='Brand voice guidelines to apply to generated content. Defines tone, style, vocabulary, and other brand-specific rules.')
|
|
|
|
# === References ===
|
|
reference_ids = fields.Many2many('otk.seo.reference',
|
|
'otk_seo_content_reference_rel', 'content_id', 'reference_id',
|
|
string='References',
|
|
domain="['|', ('content_type', '=', content_type), ('content_type', '=', 'all')]",
|
|
help='Reference materials to provide context and inspiration for content generation. Max 5 references.')
|
|
|
|
# === Generated Content ===
|
|
generated_title = fields.Char('Generated Title', translate=True,
|
|
help='The AI-generated title optimized for SEO. You can edit this before publishing.')
|
|
generated_subtitle = fields.Char('Generated Subtitle', translate=True,
|
|
help='A subtitle or tagline generated by the AI to complement the main title.')
|
|
generated_content = fields.Html('Generated Content', sanitize=True, translate=True,
|
|
help='The main body of the AI-generated content in HTML format. This includes headings, paragraphs, lists, and other formatting.')
|
|
generated_teaser = fields.Text('Generated Teaser', translate=True,
|
|
help='A short excerpt or summary of the content, typically used for blog post previews and social sharing.')
|
|
generated_keywords = fields.Char('Suggested Keywords', translate=True,
|
|
help='Additional keywords suggested by the AI based on the generated content. Use these to improve your SEO strategy.')
|
|
|
|
# Note: SEO fields come from website.seo.metadata:
|
|
# - website_meta_title
|
|
# - website_meta_description
|
|
# - website_meta_keywords
|
|
# - website_meta_og_img
|
|
# - seo_name
|
|
|
|
# === SEO Analysis ===
|
|
seo_score = fields.Integer('SEO Score', compute='_compute_seo_score', store=True,
|
|
help='Overall SEO score from 0-100 based on title length, meta description, keyword density, content length, header structure, and image optimization.')
|
|
keyword_density = fields.Float('Keyword Density %', digits=(5, 2),
|
|
help='Percentage of target keywords in the content. Optimal range is 1-2.5%. Too low means keywords are underused; too high may be seen as keyword stuffing.')
|
|
readability_score = fields.Char('Readability',
|
|
help='Assessment of how easy the content is to read, based on sentence length, vocabulary complexity, and structure.')
|
|
seo_analysis = fields.Text('SEO Analysis Details',
|
|
help='Detailed JSON breakdown of SEO scoring for each factor: title, meta description, keywords, content length, headers, and images.')
|
|
seo_analysis_html = fields.Html('SEO Analysis Display', compute='_compute_seo_analysis_html',
|
|
sanitize=False, help='User-friendly display of SEO analysis with actionable advice.')
|
|
seo_improvements = fields.Text('Improvement Suggestions', compute='_compute_seo_analysis_html',
|
|
help='List of actionable improvements to boost SEO score.')
|
|
word_count = fields.Integer('Word Count', compute='_compute_word_count', store=True,
|
|
help='Total number of words in the generated content. Longer content (1000+ words) typically performs better in search rankings.')
|
|
|
|
# === Meta Length Indicators ===
|
|
meta_title_length = fields.Integer('Title Length', compute='_compute_meta_lengths',
|
|
help='Character count of the meta title.')
|
|
meta_desc_length = fields.Integer('Description Length', compute='_compute_meta_lengths',
|
|
help='Character count of the meta description.')
|
|
meta_title_status = fields.Selection([
|
|
('short', 'Too Short'),
|
|
('optimal', 'Optimal'),
|
|
('long', 'Too Long'),
|
|
], string='Title Status', compute='_compute_meta_lengths',
|
|
help='Whether the meta title length is optimal for SEO.')
|
|
meta_desc_status = fields.Selection([
|
|
('short', 'Too Short'),
|
|
('optimal', 'Optimal'),
|
|
('long', 'Too Long'),
|
|
], string='Description Status', compute='_compute_meta_lengths',
|
|
help='Whether the meta description length is optimal for SEO.')
|
|
|
|
# === Images ===
|
|
image_ids = fields.One2many('otk.seo.content.image', 'content_id', 'Generated Images',
|
|
help='AI-generated images for this content. Images are generated asynchronously and will appear here when ready.')
|
|
image_count = fields.Integer('Image Count', compute='_compute_image_count',
|
|
help='Total number of images associated with this content.')
|
|
images_pending = fields.Boolean('Images Pending', compute='_compute_images_pending',
|
|
help='Indicates if there are images still being generated or waiting to be processed.')
|
|
has_insertable_images = fields.Boolean('Has Insertable Images', compute='_compute_has_insertable_images',
|
|
help='True if there are completed non-cover images that have not yet been inserted into content.')
|
|
requested_image_count = fields.Integer('Requested Images', default=1,
|
|
help='Number of images to generate for this content. The first image will be used as the cover image.')
|
|
image_style = fields.Selection([
|
|
('photorealistic', 'Photorealistic'),
|
|
('illustration', 'Illustration'),
|
|
('digital_art', 'Digital Art'),
|
|
('watercolor', 'Watercolor'),
|
|
('sketch', 'Sketch'),
|
|
('3d_render', '3D Render'),
|
|
], string='Image Style', default='photorealistic',
|
|
help='Visual style for generated images. Applied to all images created for this content.')
|
|
|
|
# === Workflow ===
|
|
state = fields.Selection([
|
|
('draft', 'Draft'),
|
|
('generating', 'Generating'),
|
|
('images_pending', 'Waiting for Images'),
|
|
('review', 'Review'),
|
|
('approved', 'Approved'),
|
|
('published', 'Published'),
|
|
('archived', 'Archived'),
|
|
], string='Status', default='draft', tracking=True, copy=False, index=True,
|
|
help='Current status in the content workflow. Draft: initial state; Generating: AI is creating content; Waiting for Images: content done, images processing; Review: ready for review; Approved: approved for publishing; Published: blog post created; Archived: no longer active.')
|
|
|
|
# === Output ===
|
|
blog_post_id = fields.Many2one('blog.post', 'Created Blog Post', readonly=True, copy=False,
|
|
help='The blog post created from this content. Click to view or edit the published post.')
|
|
blog_id = fields.Many2one('blog.blog', 'Target Blog',
|
|
help='The blog where the post will be published. If not set, the default blog will be used.')
|
|
|
|
# === Batch Processing ===
|
|
batch_id = fields.Many2one('otk.seo.content.batch', 'Batch Job', ondelete='set null',
|
|
help='The batch job that created this content, if generated as part of a bulk operation.')
|
|
|
|
# === Tracking ===
|
|
token_cost = fields.Integer('Tokens Used', readonly=True, store=True,
|
|
help='Number of API tokens consumed for generating this content. This affects your O\'Toolkit credit balance.')
|
|
generation_time = fields.Float('Generation Time (s)', readonly=True, store=True,
|
|
help='Time taken to generate the content in seconds. Useful for performance monitoring.')
|
|
error_message = fields.Text('Error Message',
|
|
help='If generation failed, this contains the error details. Check this to troubleshoot issues.')
|
|
api_task_id = fields.Char('API Task ID',
|
|
help='Internal identifier for tracking the generation task in the O\'Toolkit API.')
|
|
|
|
# === Retry Logic ===
|
|
retry_count = fields.Integer('Retry Count', default=0,
|
|
help='Number of retry attempts made for this content generation.')
|
|
max_retries = fields.Integer('Max Retries', default=3,
|
|
help='Maximum number of retry attempts before permanently failing.')
|
|
next_retry_at = fields.Datetime('Next Retry At',
|
|
help='Scheduled time for next retry attempt.')
|
|
generating_since = fields.Datetime('Generating Since',
|
|
help='Timestamp of when generation started, for stuck task detection.')
|
|
|
|
# === Content Freshness ===
|
|
last_refreshed_date = fields.Date('Last Refreshed',
|
|
default=fields.Date.today,
|
|
help='The date when this content was last reviewed or updated. Used to track content freshness.')
|
|
refresh_threshold_days = fields.Integer('Refresh Threshold (days)',
|
|
default=180,
|
|
help='Number of days after which content should be reviewed for freshness. Default is 180 days (6 months).')
|
|
is_stale = fields.Boolean('Needs Refresh', compute='_compute_freshness', store=True,
|
|
help='Indicates if the content has exceeded its refresh threshold and should be reviewed.')
|
|
days_since_refresh = fields.Integer('Days Since Refresh', compute='_compute_freshness', store=True,
|
|
help='Number of days since the content was last refreshed.')
|
|
refresh_alert_sent = fields.Boolean('Refresh Alert Sent', default=False,
|
|
help='Indicates if a refresh alert has already been sent for this content.')
|
|
|
|
# === Version History ===
|
|
version_ids = fields.One2many('otk.seo.content.version', 'content_id', 'Versions', copy=False,
|
|
help='Previous versions of this content. You can restore any version if needed.')
|
|
current_version = fields.Integer('Current Version', default=1, copy=False,
|
|
help='The current version number. Incremented each time content is regenerated or restored.')
|
|
|
|
@api.constrains('requested_image_count')
|
|
def _check_image_count_range(self):
|
|
for record in self:
|
|
if not (0 <= record.requested_image_count <= 10):
|
|
raise ValidationError(_('Image count must be between 0 and 10.'))
|
|
|
|
def _default_language(self):
|
|
"""Get default language for content generation."""
|
|
lang_code = self.env.lang or 'en_US'
|
|
lang = self.env['res.lang']._lang_get(lang_code)
|
|
return lang.id if lang else False
|
|
|
|
@api.onchange('template_id')
|
|
def _onchange_template_id_references(self):
|
|
"""Inherit references from template when selected."""
|
|
if self.template_id and self.template_id.reference_ids:
|
|
# Add template references (user can still modify)
|
|
self.reference_ids = [(6, 0, self.template_id.reference_ids.ids)]
|
|
|
|
@api.depends('last_refreshed_date', 'refresh_threshold_days')
|
|
def _compute_freshness(self):
|
|
"""Compute content freshness based on last refresh date."""
|
|
today = fields.Date.today()
|
|
for record in self:
|
|
if record.last_refreshed_date:
|
|
delta = today - record.last_refreshed_date
|
|
record.days_since_refresh = delta.days
|
|
record.is_stale = delta.days >= (record.refresh_threshold_days or 180)
|
|
else:
|
|
record.days_since_refresh = 0
|
|
record.is_stale = False
|
|
|
|
@api.depends('generated_content')
|
|
def _compute_word_count(self):
|
|
for record in self:
|
|
if record.generated_content:
|
|
text = html2plaintext(record.generated_content)
|
|
record.word_count = len(text.split())
|
|
else:
|
|
record.word_count = 0
|
|
|
|
@api.depends('website_meta_title', 'website_meta_description')
|
|
def _compute_meta_lengths(self):
|
|
for record in self:
|
|
# Title length and status
|
|
title_len = len(record.website_meta_title or '')
|
|
record.meta_title_length = title_len
|
|
if title_len == 0:
|
|
record.meta_title_status = 'short'
|
|
elif 30 <= title_len <= 60:
|
|
record.meta_title_status = 'optimal'
|
|
elif title_len < 30:
|
|
record.meta_title_status = 'short'
|
|
else:
|
|
record.meta_title_status = 'long'
|
|
|
|
# Description length and status
|
|
desc_len = len(record.website_meta_description or '')
|
|
record.meta_desc_length = desc_len
|
|
if desc_len == 0:
|
|
record.meta_desc_status = 'short'
|
|
elif 120 <= desc_len <= 155:
|
|
record.meta_desc_status = 'optimal'
|
|
elif desc_len < 120:
|
|
record.meta_desc_status = 'short'
|
|
else:
|
|
record.meta_desc_status = 'long'
|
|
|
|
@api.depends('image_ids')
|
|
def _compute_image_count(self):
|
|
for record in self:
|
|
record.image_count = len(record.image_ids)
|
|
|
|
@api.depends('image_ids.state')
|
|
def _compute_images_pending(self):
|
|
for record in self:
|
|
record.images_pending = any(
|
|
img.state in ('pending', 'processing')
|
|
for img in record.image_ids
|
|
)
|
|
|
|
@api.depends('image_ids', 'image_ids.state', 'image_ids.is_cover', 'image_ids.inserted_in_content')
|
|
def _compute_has_insertable_images(self):
|
|
for record in self:
|
|
record.has_insertable_images = any(
|
|
img.state == 'done' and not img.is_cover and not img.inserted_in_content
|
|
for img in record.image_ids
|
|
)
|
|
|
|
@api.depends('generated_content', 'website_meta_title', 'website_meta_description',
|
|
'website_meta_keywords', 'source_keywords', 'generated_title',
|
|
'image_ids.state', 'image_ids.alt_text')
|
|
def _compute_seo_score(self):
|
|
for record in self:
|
|
score = 0
|
|
analysis = {}
|
|
|
|
# Title optimization (15 points)
|
|
if record.website_meta_title:
|
|
title_len = len(record.website_meta_title)
|
|
if 30 <= title_len <= 60:
|
|
score += 15
|
|
analysis['title'] = {'score': 15, 'status': 'optimal', 'length': title_len}
|
|
elif 20 <= title_len < 30 or 60 < title_len <= 70:
|
|
score += 8
|
|
analysis['title'] = {'score': 8, 'status': 'acceptable', 'length': title_len}
|
|
else:
|
|
analysis['title'] = {'score': 0, 'status': 'needs_work', 'length': title_len}
|
|
else:
|
|
analysis['title'] = {'score': 0, 'status': 'missing'}
|
|
|
|
# Meta description (15 points)
|
|
if record.website_meta_description:
|
|
desc_len = len(record.website_meta_description)
|
|
if 120 <= desc_len <= 155:
|
|
score += 15
|
|
analysis['meta_desc'] = {'score': 15, 'status': 'optimal', 'length': desc_len}
|
|
elif 70 <= desc_len < 120 or 155 < desc_len <= 160:
|
|
score += 8
|
|
analysis['meta_desc'] = {'score': 8, 'status': 'acceptable', 'length': desc_len}
|
|
else:
|
|
analysis['meta_desc'] = {'score': 0, 'status': 'needs_work', 'length': desc_len}
|
|
else:
|
|
analysis['meta_desc'] = {'score': 0, 'status': 'missing'}
|
|
|
|
# Meta keywords (10 points)
|
|
if record.website_meta_keywords:
|
|
keywords = [k.strip() for k in record.website_meta_keywords.split(',') if k.strip()]
|
|
keyword_count = len(keywords)
|
|
if 3 <= keyword_count <= 10:
|
|
score += 10
|
|
analysis['meta_keywords'] = {'score': 10, 'status': 'optimal', 'count': keyword_count}
|
|
elif 1 <= keyword_count < 3 or 10 < keyword_count <= 15:
|
|
score += 5
|
|
analysis['meta_keywords'] = {'score': 5, 'status': 'acceptable', 'count': keyword_count}
|
|
else:
|
|
analysis['meta_keywords'] = {'score': 0, 'status': 'too_many', 'count': keyword_count}
|
|
else:
|
|
analysis['meta_keywords'] = {'score': 0, 'status': 'missing'}
|
|
|
|
# Cache plaintext conversion (used by multiple checks below)
|
|
_plain_text = html2plaintext(record.generated_content) if record.generated_content else ''
|
|
_plain_text_lower = _plain_text.lower() if _plain_text else ''
|
|
_plain_word_count = len(_plain_text.split()) if _plain_text else 0
|
|
|
|
# Keyword usage in content (20 points)
|
|
if record.source_keywords and _plain_text:
|
|
keywords = [k.strip().lower() for k in record.source_keywords.split(',')]
|
|
keyword_count = sum(_plain_text_lower.count(kw) for kw in keywords if kw)
|
|
|
|
if _plain_word_count > 0:
|
|
density = (keyword_count / _plain_word_count) * 100
|
|
record.keyword_density = round(density, 2)
|
|
|
|
if 1 <= density <= 2.5:
|
|
score += 20
|
|
analysis['keywords'] = {'score': 20, 'status': 'optimal', 'density': density}
|
|
elif 0.5 <= density < 1 or 2.5 < density <= 3:
|
|
score += 10
|
|
analysis['keywords'] = {'score': 10, 'status': 'acceptable', 'density': density}
|
|
else:
|
|
analysis['keywords'] = {'score': 0, 'status': 'needs_work', 'density': density}
|
|
else:
|
|
analysis['keywords'] = {'score': 0, 'status': 'no_content'}
|
|
else:
|
|
analysis['keywords'] = {'score': 0, 'status': 'missing'}
|
|
|
|
# Content length (20 points)
|
|
if _plain_text:
|
|
wc = _plain_word_count
|
|
|
|
if wc >= 1000:
|
|
score += 20
|
|
analysis['content_length'] = {'score': 20, 'status': 'optimal', 'words': wc}
|
|
elif wc >= 500:
|
|
score += 15
|
|
analysis['content_length'] = {'score': 15, 'status': 'good', 'words': wc}
|
|
elif wc >= 300:
|
|
score += 10
|
|
analysis['content_length'] = {'score': 10, 'status': 'acceptable', 'words': wc}
|
|
else:
|
|
analysis['content_length'] = {'score': 0, 'status': 'too_short', 'words': wc}
|
|
else:
|
|
analysis['content_length'] = {'score': 0, 'status': 'no_content'}
|
|
|
|
# Header structure (10 points)
|
|
if record.generated_content:
|
|
content_lower = record.generated_content.lower()
|
|
has_h2 = '<h2' in content_lower
|
|
has_h3 = '<h3' in content_lower
|
|
|
|
if has_h2 and has_h3:
|
|
score += 10
|
|
analysis['headers'] = {'score': 10, 'status': 'optimal', 'h2': has_h2, 'h3': has_h3}
|
|
elif has_h2:
|
|
score += 5
|
|
analysis['headers'] = {'score': 5, 'status': 'acceptable', 'h2': has_h2, 'h3': has_h3}
|
|
else:
|
|
analysis['headers'] = {'score': 0, 'status': 'missing', 'h2': has_h2, 'h3': has_h3}
|
|
else:
|
|
analysis['headers'] = {'score': 0, 'status': 'no_content'}
|
|
|
|
# Images (10 points)
|
|
done_images = record.image_ids.filtered(lambda i: i.state == 'done')
|
|
images_with_alt = done_images.filtered(lambda i: i.alt_text)
|
|
|
|
if images_with_alt:
|
|
score += 10
|
|
analysis['images'] = {'score': 10, 'status': 'optimal', 'count': len(done_images), 'with_alt': len(images_with_alt)}
|
|
elif done_images:
|
|
score += 5
|
|
analysis['images'] = {'score': 5, 'status': 'missing_alt', 'count': len(done_images), 'with_alt': 0}
|
|
else:
|
|
analysis['images'] = {'score': 0, 'status': 'no_images', 'count': 0}
|
|
|
|
record.seo_score = score
|
|
record.seo_analysis = analysis # fields.Json expects a dict, not a string
|
|
|
|
@api.depends('seo_analysis', 'seo_score')
|
|
def _compute_seo_analysis_html(self):
|
|
"""Generate user-friendly HTML display of SEO analysis."""
|
|
for record in self:
|
|
if not record.seo_analysis:
|
|
record.seo_analysis_html = '''
|
|
<div class="text-muted text-center p-4">
|
|
<i class="fa fa-info-circle fa-2x mb-2"></i>
|
|
<p>Generate content to see SEO analysis.</p>
|
|
</div>
|
|
'''
|
|
record.seo_improvements = ''
|
|
continue
|
|
|
|
# Parse analysis
|
|
analysis = ast.literal_eval(record.seo_analysis or "{}")
|
|
|
|
score = record.seo_score or 0
|
|
improvements = []
|
|
|
|
# Build HTML
|
|
html_parts = []
|
|
|
|
# Score overview card
|
|
score_color = 'success' if score >= 70 else ('warning' if score >= 40 else 'danger')
|
|
score_icon = 'check-circle' if score >= 70 else ('exclamation-triangle' if score >= 40 else 'times-circle')
|
|
score_desc = record._get_score_description(score)
|
|
|
|
html_parts.append(f'''
|
|
<div class="alert alert-{score_color} d-flex align-items-center mb-4" role="alert">
|
|
<div class="me-3" style="font-size: 2.5rem; font-weight: bold; min-width: 80px; text-align: center;">
|
|
{score}<small style="font-size: 0.5em">/100</small>
|
|
</div>
|
|
<div>
|
|
<strong><i class="fa fa-{score_icon}"></i> {score_desc['title']}</strong>
|
|
<p class="mb-0 small">{score_desc['message']}</p>
|
|
</div>
|
|
</div>
|
|
''')
|
|
|
|
# Analysis sections
|
|
sections = [
|
|
('title', 'Meta Title', 'fa-heading', 15),
|
|
('meta_desc', 'Meta Description', 'fa-align-left', 15),
|
|
('meta_keywords', 'Meta Keywords', 'fa-tags', 10),
|
|
('keywords', 'Keyword Density', 'fa-key', 20),
|
|
('content_length', 'Content Length', 'fa-file-text-o', 20),
|
|
('headers', 'Header Structure', 'fa-list-ol', 10),
|
|
('images', 'Images', 'fa-image', 10),
|
|
]
|
|
|
|
html_parts.append('<div class="row">')
|
|
|
|
for key, label, icon, max_score in sections:
|
|
section_data = analysis.get(key, {})
|
|
section_score = section_data.get('score', 0)
|
|
|
|
# Determine styling
|
|
if section_score == max_score:
|
|
bg_class = 'bg-success-subtle border-success'
|
|
text_class = 'text-success'
|
|
status_icon = 'fa-check-circle'
|
|
elif section_score > 0:
|
|
bg_class = 'bg-warning-subtle border-warning'
|
|
text_class = 'text-warning'
|
|
status_icon = 'fa-exclamation-circle'
|
|
else:
|
|
bg_class = 'bg-danger-subtle border-danger'
|
|
text_class = 'text-danger'
|
|
status_icon = 'fa-times-circle'
|
|
|
|
# Get advice
|
|
advice = record._get_seo_advice(key, section_data, max_score)
|
|
if advice and section_score < max_score:
|
|
improvements.append(advice)
|
|
|
|
# Get details
|
|
details = record._get_section_details(key, section_data)
|
|
|
|
html_parts.append(f'''
|
|
<div class="col-md-6 col-lg-4 mb-3">
|
|
<div class="card h-100 border {bg_class}">
|
|
<div class="card-body p-3">
|
|
<div class="d-flex justify-content-between align-items-start mb-2">
|
|
<span class="fw-bold">
|
|
<i class="fa {icon} me-1"></i> {label}
|
|
</span>
|
|
<span class="badge bg-{'success' if section_score == max_score else ('warning' if section_score > 0 else 'danger')}">
|
|
{section_score}/{max_score}
|
|
</span>
|
|
</div>
|
|
<div class="{text_class} small">
|
|
<i class="fa {status_icon}"></i>
|
|
{advice if advice else '<span class="text-success">Excellent!</span>'}
|
|
</div>
|
|
{f'<div class="text-muted small mt-2 border-top pt-2">{details}</div>' if details else ''}
|
|
</div>
|
|
</div>
|
|
</div>
|
|
''')
|
|
|
|
html_parts.append('</div>')
|
|
|
|
record.seo_analysis_html = ''.join(html_parts)
|
|
record.seo_improvements = '\n'.join(f"• {imp}" for imp in improvements) if improvements else ''
|
|
|
|
def _get_score_description(self, score):
|
|
"""Get title and message based on score."""
|
|
if score >= 80:
|
|
return {
|
|
'title': 'Excellent SEO',
|
|
'message': 'Your content is well-optimized for search engines. Great job!'
|
|
}
|
|
elif score >= 60:
|
|
return {
|
|
'title': 'Good SEO',
|
|
'message': 'Your content is fairly optimized but has room for improvement.'
|
|
}
|
|
elif score >= 40:
|
|
return {
|
|
'title': 'Needs Improvement',
|
|
'message': 'Your content needs work to rank well. Follow the suggestions below.'
|
|
}
|
|
else:
|
|
return {
|
|
'title': 'Poor SEO',
|
|
'message': 'Significant improvements needed. Address the issues highlighted below.'
|
|
}
|
|
|
|
def _get_seo_advice(self, key, data, max_score):
|
|
"""Get actionable advice for each SEO factor."""
|
|
status = data.get('status', 'missing')
|
|
current_score = data.get('score', 0)
|
|
|
|
if current_score == max_score:
|
|
return None # No advice needed
|
|
|
|
advice_map = {
|
|
'title': {
|
|
'missing': 'Add a meta title (30-60 characters recommended)',
|
|
'needs_work': f"Optimize title length ({data.get('length', 0)} chars) - aim for 30-60 characters",
|
|
'acceptable': f"Title is {data.get('length', 0)} chars - optimal is 30-60 characters",
|
|
},
|
|
'meta_desc': {
|
|
'missing': 'Add a meta description (120-155 characters recommended)',
|
|
'needs_work': f"Adjust description length ({data.get('length', 0)} chars) - aim for 120-155 characters",
|
|
'acceptable': f"Description is {data.get('length', 0)} chars - optimal is 120-155 characters",
|
|
},
|
|
'meta_keywords': {
|
|
'missing': 'Add 3-10 meta keywords separated by commas',
|
|
'too_many': f"Reduce keywords from {data.get('count', 0)} to 3-10 for best results",
|
|
'acceptable': f"Add more keywords (currently {data.get('count', 0)}) - aim for 3-10",
|
|
},
|
|
'keywords': {
|
|
'missing': 'Add target keywords to analyze density',
|
|
'no_content': 'Generate content to analyze keyword usage',
|
|
'needs_work': f"Adjust keyword density ({data.get('density', 0):.1f}%) - optimal is 1-2.5%",
|
|
'acceptable': f"Keyword density ({data.get('density', 0):.1f}%) is acceptable - optimal is 1-2.5%",
|
|
},
|
|
'content_length': {
|
|
'no_content': 'Generate content first',
|
|
'too_short': f"Expand content from {data.get('words', 0)} words - aim for 1000+ words",
|
|
'acceptable': f"Content has {data.get('words', 0)} words - 1000+ is optimal for SEO",
|
|
'good': f"Good length ({data.get('words', 0)} words) - 1000+ is optimal",
|
|
},
|
|
'headers': {
|
|
'no_content': 'Generate content first',
|
|
'missing': 'Add H2 and H3 headers to structure your content',
|
|
'acceptable': 'Add H3 subheaders for better content structure',
|
|
},
|
|
'images': {
|
|
'no_images': 'Add images to improve engagement and SEO',
|
|
'missing_alt': f"Add alt text to your {data.get('count', 0)} image(s) for accessibility and SEO",
|
|
},
|
|
}
|
|
|
|
return advice_map.get(key, {}).get(status)
|
|
|
|
def _get_section_details(self, key, data):
|
|
"""Get additional details for display."""
|
|
parts = []
|
|
|
|
if key == 'title' and 'length' in data:
|
|
parts.append(f"<strong>Length:</strong> {data['length']} characters")
|
|
elif key == 'meta_desc' and 'length' in data:
|
|
parts.append(f"<strong>Length:</strong> {data['length']} characters")
|
|
elif key == 'meta_keywords' and 'count' in data:
|
|
parts.append(f"<strong>Count:</strong> {data['count']} keywords")
|
|
elif key == 'keywords' and 'density' in data:
|
|
parts.append(f"<strong>Density:</strong> {data['density']:.2f}%")
|
|
elif key == 'content_length' and 'words' in data:
|
|
parts.append(f"<strong>Word count:</strong> {data['words']}")
|
|
elif key == 'headers':
|
|
h2 = '✓' if data.get('h2') else '✗'
|
|
h3 = '✓' if data.get('h3') else '✗'
|
|
parts.append(f"<strong>H2:</strong> {h2} <strong>H3:</strong> {h3}")
|
|
elif key == 'images':
|
|
count = data.get('count', 0)
|
|
with_alt = data.get('with_alt', 0)
|
|
parts.append(f"<strong>Images:</strong> {count} ({with_alt} with alt text)")
|
|
|
|
return ' '.join(parts)
|
|
|
|
# === API Methods ===
|
|
def _get_api_config(self):
|
|
"""Get API configuration."""
|
|
ICP = self.env['ir.config_parameter'].sudo()
|
|
base_url = ICP.get_param('otoolkit.api.endpoint', '')
|
|
api_key = ICP.get_param('otoolkit_api_key', '')
|
|
|
|
if not api_key:
|
|
raise UserError(_("Please configure your O'Toolkit API key in Settings > O'Toolkit."))
|
|
if not base_url:
|
|
raise UserError(_("O'Toolkit API endpoint is not configured."))
|
|
|
|
return base_url, api_key
|
|
|
|
@api.model
|
|
def _fetch_pricing_config(self, force=False):
|
|
"""Fetch pricing config from O'Toolkit API and cache in ir.config_parameter.
|
|
|
|
Cached for 1 hour. Returns dict with 'seo_content' and 'seo_content_image' keys,
|
|
each containing 'rate' and 'min_cost'.
|
|
"""
|
|
ICP = self.env['ir.config_parameter'].sudo()
|
|
cache_key = 'otoolkit_seo_content.pricing_cache'
|
|
cache_ts_key = 'otoolkit_seo_content.pricing_cache_ts'
|
|
|
|
# Check cache (1 hour TTL)
|
|
if not force:
|
|
cached = ICP.get_param(cache_key, '')
|
|
cache_ts = float(ICP.get_param(cache_ts_key, '0'))
|
|
if cached and (time.time() - cache_ts) < 3600:
|
|
try:
|
|
return json.loads(cached)
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass
|
|
|
|
# Fetch from API
|
|
try:
|
|
base_url = ICP.get_param('otoolkit.api.endpoint', '')
|
|
if not base_url:
|
|
return {}
|
|
url = f"{base_url.rstrip('/')}/api/seo-content/pricing/"
|
|
response = requests.get(url, timeout=10)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
ICP.set_param(cache_key, json.dumps(data))
|
|
ICP.set_param(cache_ts_key, str(time.time()))
|
|
_logger.info("SEO pricing config refreshed from API: %s", data)
|
|
return data
|
|
except Exception as e:
|
|
_logger.warning("Failed to fetch pricing config: %s", e)
|
|
|
|
return {}
|
|
|
|
@api.model
|
|
def get_pricing_estimates(self):
|
|
"""Return estimated O'Token costs for content and images.
|
|
|
|
Uses pre-computed estimates from the API pricing endpoint,
|
|
so all pricing logic lives server-side and can be tuned
|
|
without updating the Odoo module.
|
|
"""
|
|
pricing = self._fetch_pricing_config()
|
|
estimates = pricing.get('estimates')
|
|
|
|
if estimates:
|
|
return {
|
|
'text': estimates.get('text', {}),
|
|
'image': estimates.get('image', 0),
|
|
}
|
|
|
|
# Fallback if API doesn't return estimates (old API version)
|
|
return {
|
|
'text': {'short': 200, 'medium': 440, 'long': 850},
|
|
'image': 250,
|
|
}
|
|
|
|
def _call_api(self, endpoint, payload, timeout=180):
|
|
"""Make API call to O'Toolkit backend."""
|
|
base_url, api_key = self._get_api_config()
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Odoo-Api-Key': api_key,
|
|
}
|
|
|
|
url = f"{base_url.rstrip('/')}{endpoint}"
|
|
|
|
try:
|
|
response = requests.post(
|
|
url,
|
|
headers=headers,
|
|
json=payload,
|
|
timeout=timeout
|
|
)
|
|
|
|
_logger.info(f"API Response: {response.status_code} - {response.text[:500] if response.text else 'empty'}")
|
|
|
|
if response.status_code == 400:
|
|
try:
|
|
error_data = response.json()
|
|
except (ValueError, json.JSONDecodeError):
|
|
raise UserError(_("API Error (400): %s") % response.text[:500])
|
|
|
|
if error_data.get('type') == 'insufficient_funds':
|
|
raise UserError(_("Insufficient tokens. Please add more credits to your O'Toolkit account."))
|
|
# Handle both 'error' and 'detail' keys
|
|
error_msg = error_data.get('error') or error_data.get('detail') or 'Unknown error'
|
|
raise UserError(_("API Error: %s") % error_msg)
|
|
|
|
if response.status_code == 401:
|
|
raise UserError(_("Invalid API key. Please check your O'Toolkit configuration."))
|
|
|
|
if response.status_code >= 400:
|
|
raise UserError(_("API Error (%s): %s") % (response.status_code, response.text[:500]))
|
|
|
|
response.raise_for_status()
|
|
|
|
try:
|
|
return response.json()
|
|
except (ValueError, json.JSONDecodeError):
|
|
raise UserError(_("Invalid response from API: %s") % response.text[:200])
|
|
|
|
except requests.exceptions.Timeout:
|
|
raise UserError(_("Request timed out. Please try again."))
|
|
except requests.exceptions.ConnectionError:
|
|
raise UserError(_("Could not connect to O'Toolkit API. Please check your internet connection."))
|
|
except requests.exceptions.RequestException as e:
|
|
_logger.error(f"API Error: {e}")
|
|
raise UserError(_("API request failed: %s") % str(e))
|
|
|
|
# === Input Validation ===
|
|
def _validate_generation_inputs(self):
|
|
"""Validate inputs before sending to API to prevent abuse and errors."""
|
|
self.ensure_one()
|
|
errors = []
|
|
|
|
# Validate keywords length
|
|
if self.source_keywords and len(self.source_keywords) > MAX_KEYWORDS_LENGTH:
|
|
errors.append(_("Keywords are too long (max %d characters, got %d).") % (
|
|
MAX_KEYWORDS_LENGTH, len(self.source_keywords)))
|
|
|
|
# Validate topic length
|
|
if self.source_topic and len(self.source_topic) > MAX_TOPIC_LENGTH:
|
|
errors.append(_("Topic/brief is too long (max %d characters, got %d).") % (
|
|
MAX_TOPIC_LENGTH, len(self.source_topic)))
|
|
|
|
# Validate source text length
|
|
if self.source_text and len(self.source_text) > MAX_SOURCE_TEXT_LENGTH:
|
|
errors.append(_("Source text is too long (max %d characters, got %d).") % (
|
|
MAX_SOURCE_TEXT_LENGTH, len(self.source_text)))
|
|
|
|
# Validate required fields - accept any valid source input
|
|
has_valid_source = bool(
|
|
self.source_keywords or self.source_topic
|
|
or self.source_text or self.source_product_id
|
|
)
|
|
if not has_valid_source:
|
|
errors.append(_("Please provide keywords, a topic, existing content, or a product."))
|
|
|
|
# Validate requested image count
|
|
if self.requested_image_count < 0 or self.requested_image_count > 10:
|
|
errors.append(_("Image count must be between 0 and 10."))
|
|
|
|
# Validate references count
|
|
if len(self.reference_ids) > MAX_REFERENCES:
|
|
errors.append(_("Maximum %d references allowed (got %d).") % (
|
|
MAX_REFERENCES, len(self.reference_ids)))
|
|
|
|
if errors:
|
|
raise UserError('\n'.join(errors))
|
|
|
|
# === Actions ===
|
|
def action_generate(self):
|
|
"""Submit content generation request to API (async).
|
|
|
|
The content is generated asynchronously. A cron job polls for completion
|
|
and updates the record when ready. Images are generated AFTER content
|
|
is complete to enable content-aware prompts.
|
|
"""
|
|
self.ensure_one()
|
|
|
|
# Validate inputs before proceeding
|
|
self._validate_generation_inputs()
|
|
|
|
self.write({
|
|
'state': 'generating',
|
|
'error_message': False,
|
|
'api_task_id': False,
|
|
'generating_since': fields.Datetime.now(),
|
|
})
|
|
|
|
# Prepare payload
|
|
payload = {
|
|
'keywords': self.source_keywords or '',
|
|
'topic': self.source_topic or '',
|
|
'content_type': self.content_type,
|
|
'tone': self.tone,
|
|
'word_count': self.target_word_count,
|
|
'language': self.language_id.code if self.language_id else 'en_US',
|
|
'odoo_user_id': self.env.uid,
|
|
'options': {
|
|
'include_toc': self.template_id.include_toc if self.template_id else False,
|
|
'include_faq': self.template_id.include_faq if self.template_id else False,
|
|
'header_structure': self.template_id.header_structure if self.template_id else 'h2_h3',
|
|
}
|
|
}
|
|
|
|
# Add source text if available (optimize existing content)
|
|
if self.source_text:
|
|
payload['source_text'] = self.source_text
|
|
|
|
# Add product data if available
|
|
if self.source_product_id:
|
|
payload['product'] = {
|
|
'name': self.source_product_id.name,
|
|
'description': self.source_product_id.description or '',
|
|
'price': self.source_product_id.list_price,
|
|
'category': self.source_product_id.categ_id.complete_name if self.source_product_id.categ_id else '',
|
|
}
|
|
|
|
# Add template prompts if available
|
|
if self.template_id:
|
|
payload['template'] = {
|
|
'system_prompt': self.template_id.system_prompt,
|
|
'content_prompt': self.template_id.content_prompt_template,
|
|
}
|
|
|
|
# Add brand voice guidelines if available
|
|
if self.brand_voice_id:
|
|
payload['brand_voice'] = self.brand_voice_id.get_prompt_instructions()
|
|
|
|
# Add references if available
|
|
if self.reference_ids:
|
|
references_payload = []
|
|
for ref in self.reference_ids[:MAX_REFERENCES]:
|
|
ref_data = ref.get_api_payload()
|
|
if ref_data.get('content'): # Only include non-empty references
|
|
references_payload.append(ref_data)
|
|
if references_payload:
|
|
payload['references'] = references_payload
|
|
|
|
try:
|
|
result = self._call_api('/api/seo-content/generate/', payload)
|
|
|
|
if result.get('success') and result.get('task_id'):
|
|
# Store task ID for polling
|
|
self.write({
|
|
'api_task_id': str(result['task_id']),
|
|
})
|
|
_logger.info(f"Content generation task submitted for {self.id}: task_id={result['task_id']}")
|
|
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('Generation Started'),
|
|
'message': _('Content is being generated. You will be notified when ready.'),
|
|
'type': 'info',
|
|
'sticky': False,
|
|
}
|
|
}
|
|
else:
|
|
self.write({
|
|
'state': 'draft',
|
|
'error_message': result.get('error', 'Unknown error occurred'),
|
|
})
|
|
raise UserError(_("Content generation failed: %s") % result.get('error', 'Unknown error'))
|
|
|
|
except UserError:
|
|
raise
|
|
except Exception as e:
|
|
_logger.exception("Content generation error")
|
|
self.write({
|
|
'state': 'draft',
|
|
'error_message': str(e),
|
|
})
|
|
raise UserError(_("Content generation failed: %s") % str(e))
|
|
|
|
def _schedule_retry(self, error_msg):
|
|
"""Schedule a retry with exponential backoff, or permanently fail."""
|
|
self.ensure_one()
|
|
backoff_minutes = [1, 5, 15]
|
|
|
|
if self.retry_count >= self.max_retries:
|
|
self.write({
|
|
'state': 'draft',
|
|
'error_message': _("Permanently failed after %d retries. Last error: %s") % (
|
|
self.retry_count, error_msg),
|
|
'generating_since': False,
|
|
})
|
|
_logger.warning("Content %s permanently failed after %d retries", self.id, self.retry_count)
|
|
# Send failure notification
|
|
if self.create_uid and self.create_uid.partner_id:
|
|
self.env['bus.bus']._sendone(
|
|
self.create_uid.partner_id,
|
|
'simple_notification',
|
|
{
|
|
'title': _("Generation Failed"),
|
|
'message': _("'%s' failed after %d retries.") % (self.name, self.retry_count),
|
|
'type': 'warning',
|
|
'sticky': True,
|
|
}
|
|
)
|
|
return
|
|
|
|
delay = backoff_minutes[min(self.retry_count, len(backoff_minutes) - 1)]
|
|
from datetime import timedelta
|
|
next_retry = fields.Datetime.now() + timedelta(minutes=delay)
|
|
|
|
self.write({
|
|
'state': 'draft',
|
|
'retry_count': self.retry_count + 1,
|
|
'next_retry_at': next_retry,
|
|
'error_message': _("Retry %d/%d scheduled. Error: %s") % (
|
|
self.retry_count + 1, self.max_retries, error_msg),
|
|
'generating_since': False,
|
|
'api_task_id': False,
|
|
})
|
|
_logger.info("Content %s retry %d/%d scheduled at %s",
|
|
self.id, self.retry_count, self.max_retries, next_retry)
|
|
|
|
@api.model
|
|
def cron_poll_content_status(self):
|
|
"""Cron job to poll status of generating content.
|
|
|
|
Checks all content records in 'generating' state with an api_task_id,
|
|
polls the API for task status, and processes completed results.
|
|
Also handles stuck tasks and pending retries.
|
|
"""
|
|
from datetime import timedelta
|
|
now = fields.Datetime.now()
|
|
|
|
# --- Detect stuck tasks (generating for > 1 hour) ---
|
|
stuck_cutoff = now - timedelta(hours=1)
|
|
stuck_content = self.search([
|
|
('state', '=', 'generating'),
|
|
('generating_since', '!=', False),
|
|
('generating_since', '<', stuck_cutoff),
|
|
], limit=10)
|
|
for content in stuck_content:
|
|
_logger.warning("Content %s stuck in generating since %s, scheduling retry",
|
|
content.id, content.generating_since)
|
|
content._schedule_retry(_("Task stuck for over 1 hour"))
|
|
|
|
# --- Process pending retries ---
|
|
pending_retries = self.search([
|
|
('state', '=', 'draft'),
|
|
('next_retry_at', '!=', False),
|
|
('next_retry_at', '<=', now),
|
|
('retry_count', '<', 3),
|
|
], limit=5)
|
|
for content in pending_retries:
|
|
try:
|
|
_logger.info("Retrying content %s (attempt %d)", content.id, content.retry_count)
|
|
content.write({'next_retry_at': False})
|
|
content.action_generate()
|
|
except Exception as e:
|
|
_logger.error("Retry failed for content %s: %s", content.id, e)
|
|
content._schedule_retry(str(e))
|
|
|
|
# --- Check content stuck in images_pending with no active images ---
|
|
images_stuck_cutoff = now - timedelta(hours=2)
|
|
stuck_images_content = self.search([
|
|
('state', '=', 'images_pending'),
|
|
('write_date', '<', images_stuck_cutoff),
|
|
], limit=10)
|
|
for content in stuck_images_content:
|
|
pending_imgs = content.image_ids.filtered(
|
|
lambda i: i.state in ('pending', 'processing')
|
|
)
|
|
if not pending_imgs:
|
|
_logger.info("Content %s stuck in images_pending with no active images, moving to review", content.id)
|
|
content.write({'state': 'review'})
|
|
|
|
# --- Poll generating tasks ---
|
|
generating = self.search([
|
|
('state', '=', 'generating'),
|
|
('api_task_id', '!=', False),
|
|
], limit=20)
|
|
|
|
if not generating:
|
|
return
|
|
|
|
# Batch poll all tasks
|
|
task_ids = []
|
|
task_id_map = {} # Maps task_id to content record
|
|
for content in generating:
|
|
if content.api_task_id:
|
|
try:
|
|
task_id_int = int(content.api_task_id)
|
|
task_ids.append(task_id_int)
|
|
task_id_map[str(task_id_int)] = content
|
|
except (ValueError, TypeError):
|
|
_logger.warning(f"Invalid api_task_id for content {content.id}: {content.api_task_id}")
|
|
content.write({
|
|
'state': 'draft',
|
|
'error_message': f'Invalid task ID format: {content.api_task_id}',
|
|
})
|
|
|
|
if not task_ids:
|
|
return
|
|
|
|
try:
|
|
base_url, api_key = generating[0]._get_api_config()
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Odoo-Api-Key': api_key,
|
|
}
|
|
|
|
response = requests.post(
|
|
f"{base_url.rstrip('/')}/api/tasks/",
|
|
headers=headers,
|
|
json={'task_ids': task_ids},
|
|
timeout=30
|
|
)
|
|
|
|
response.raise_for_status()
|
|
|
|
try:
|
|
data = response.json()
|
|
except (ValueError, json.JSONDecodeError):
|
|
_logger.error("Invalid JSON from content tasks API: %s", response.text[:200])
|
|
return
|
|
|
|
tasks = data.get('tasks', []) if isinstance(data, dict) else data
|
|
task_result_map = {str(t.get('id')): t for t in tasks if t.get('id')}
|
|
|
|
for task_id_str, content in task_id_map.items():
|
|
task = task_result_map.get(task_id_str)
|
|
if not task:
|
|
continue
|
|
|
|
status = task.get('status')
|
|
|
|
if status == 'completed':
|
|
result = task.get('result', {})
|
|
content._process_completed_content(result)
|
|
elif status == 'failed':
|
|
error_msg = task.get('result', {}).get('error', 'Generation failed')
|
|
content._schedule_retry(error_msg)
|
|
|
|
except Exception as e:
|
|
_logger.error("Content status poll cron error: %s", e)
|
|
|
|
def _process_completed_content(self, result):
|
|
"""Process completed content generation result.
|
|
|
|
Updates content fields and triggers image generation if requested.
|
|
Images are generated AFTER content to enable content-aware prompts.
|
|
"""
|
|
self.ensure_one()
|
|
|
|
# Create version before updating
|
|
if self.generated_content:
|
|
self._create_version('Before regeneration')
|
|
|
|
# Build keywords string from API response
|
|
keywords_list = result.get('keywords', [])
|
|
keywords_str = ', '.join(keywords_list) if keywords_list else ''
|
|
|
|
self.write({
|
|
'generated_title': result.get('title', ''),
|
|
'generated_subtitle': result.get('subtitle', ''),
|
|
'generated_content': result.get('content', ''),
|
|
'generated_teaser': result.get('teaser', ''),
|
|
'website_meta_title': result.get('meta_title', ''),
|
|
'website_meta_description': result.get('meta_description', ''),
|
|
'generated_keywords': keywords_str,
|
|
'website_meta_keywords': keywords_str,
|
|
'token_cost': result.get('cost', 0),
|
|
'api_task_id': False, # Clear task ID
|
|
'state': 'review' if self.requested_image_count == 0 else 'images_pending',
|
|
})
|
|
|
|
# Reset retry fields on success
|
|
self.write({
|
|
'retry_count': 0,
|
|
'next_retry_at': False,
|
|
'generating_since': False,
|
|
})
|
|
|
|
_logger.info(f"Content {self.id} generation completed: {self.generated_title}")
|
|
|
|
# Send bus notification to content creator
|
|
if self.create_uid and self.create_uid.partner_id:
|
|
self.env['bus.bus']._sendone(
|
|
self.create_uid.partner_id,
|
|
'simple_notification',
|
|
{
|
|
'title': _("Content Ready!"),
|
|
'message': _("'%s' is ready for review.") % (self.generated_title or self.name),
|
|
'type': 'success',
|
|
'sticky': False,
|
|
}
|
|
)
|
|
|
|
# Generate images AFTER content is ready (content-aware prompts)
|
|
if self.requested_image_count > 0:
|
|
self._generate_images()
|
|
|
|
def _generate_images(self):
|
|
"""Create image generation requests."""
|
|
self.ensure_one()
|
|
|
|
# Generate prompts based on content
|
|
prompts = self._generate_image_prompts()
|
|
|
|
# Use content's image_style, fallback to template default, then 'photorealistic'
|
|
image_style = self.image_style or (
|
|
self.template_id.default_image_style if self.template_id else 'photorealistic'
|
|
)
|
|
|
|
for i, prompt in enumerate(prompts[:self.requested_image_count]):
|
|
self.env['otk.seo.content.image'].create({
|
|
'content_id': self.id,
|
|
'name': f"Image {i + 1}",
|
|
'prompt': prompt,
|
|
'style': image_style,
|
|
'aspect_ratio': '1792x1024' if i == 0 else '1024x1024', # First image landscape for cover
|
|
'is_cover': i == 0,
|
|
'state': 'pending',
|
|
})
|
|
|
|
def _generate_image_prompts(self):
|
|
"""Generate image prompts based on content context."""
|
|
prompts = []
|
|
|
|
title = self.generated_title or self.name
|
|
keywords = self.source_keywords or ''
|
|
|
|
# Extract rich context from generated content
|
|
content_context = self._extract_content_context()
|
|
|
|
# Cover image - main theme with full context
|
|
cover_prompt = self._build_cover_prompt(title, keywords, content_context)
|
|
prompts.append(cover_prompt)
|
|
|
|
# Generate section-specific prompts for additional images
|
|
if self.generated_content and self.requested_image_count > 1:
|
|
section_prompts = self._generate_section_prompts(title, content_context)
|
|
prompts.extend(section_prompts)
|
|
|
|
return prompts
|
|
|
|
def _extract_content_context(self):
|
|
"""Extract key context from generated content for image prompts."""
|
|
if not self.generated_content:
|
|
return {
|
|
'sections': [],
|
|
'summary': '',
|
|
'word_count': 0,
|
|
'content_type': self.content_type,
|
|
'tone': self.tone,
|
|
}
|
|
|
|
content_text = html2plaintext(self.generated_content)
|
|
|
|
# Extract H2 headers (main topic sections)
|
|
h2_pattern = r'<h2[^>]*>(.*?)</h2>'
|
|
h2_matches = re.findall(h2_pattern, self.generated_content, re.IGNORECASE | re.DOTALL)
|
|
sections = [html2plaintext(h2).strip() for h2 in h2_matches if h2.strip()]
|
|
|
|
# Get first paragraph as summary
|
|
first_para = ''
|
|
para_match = re.search(r'<p[^>]*>(.*?)</p>', self.generated_content, re.IGNORECASE | re.DOTALL)
|
|
if para_match:
|
|
first_para = html2plaintext(para_match.group(1)).strip()[:200]
|
|
|
|
# Extract any key points from lists
|
|
list_items = []
|
|
li_pattern = r'<li[^>]*>(.*?)</li>'
|
|
li_matches = re.findall(li_pattern, self.generated_content, re.IGNORECASE | re.DOTALL)
|
|
for li in li_matches[:5]: # First 5 list items
|
|
item_text = html2plaintext(li).strip()
|
|
if item_text and len(item_text) < 100:
|
|
list_items.append(item_text)
|
|
|
|
return {
|
|
'sections': sections[:5], # Top 5 sections
|
|
'summary': first_para,
|
|
'key_points': list_items,
|
|
'word_count': len(content_text.split()),
|
|
'content_type': self.content_type,
|
|
'tone': self.tone,
|
|
}
|
|
|
|
def _build_cover_prompt(self, title, keywords, context):
|
|
"""Build a rich, context-aware cover image prompt."""
|
|
prompt_parts = []
|
|
|
|
# Tone-based styling prefix
|
|
tone_styles = {
|
|
'professional': 'Professional, polished',
|
|
'casual': 'Friendly, approachable',
|
|
'technical': 'Technical, precise',
|
|
'persuasive': 'Bold, compelling',
|
|
'conversational': 'Warm, engaging',
|
|
}
|
|
tone_style = tone_styles.get(context.get('tone', 'professional'), 'Professional')
|
|
|
|
# Start with tone and title
|
|
prompt_parts.append(f"{tone_style} hero image for: {title}.")
|
|
|
|
# Add summary context if available
|
|
if context.get('summary'):
|
|
# Truncate summary for prompt
|
|
summary = context['summary'][:150]
|
|
prompt_parts.append(f"Visualizing: {summary}.")
|
|
|
|
# Add keywords as themes
|
|
if keywords:
|
|
prompt_parts.append(f"Key themes: {keywords}.")
|
|
|
|
# Content type specific styling
|
|
type_styles = {
|
|
'blog_post': "Modern blog header style, engaging and shareable, wide cinematic composition.",
|
|
'product_desc': "Product-focused imagery, clean e-commerce aesthetic, professional lighting.",
|
|
'category_desc': "Category showcase, diverse visual elements representing the category.",
|
|
'landing_page': "Hero banner style, impactful and conversion-focused, bold composition.",
|
|
'social_post': "Social media optimized, eye-catching, vibrant colors.",
|
|
}
|
|
style_desc = type_styles.get(context.get('content_type', 'blog_post'),
|
|
"Clean, modern web imagery.")
|
|
prompt_parts.append(style_desc)
|
|
|
|
# Quality and style modifiers
|
|
prompt_parts.append("High quality, professionally composed, no text overlays or watermarks.")
|
|
|
|
return ' '.join(prompt_parts)
|
|
|
|
def _generate_section_prompts(self, main_title, context):
|
|
"""Generate prompts for section-specific supporting images."""
|
|
prompts = []
|
|
sections = context.get('sections', [])
|
|
key_points = context.get('key_points', [])
|
|
tone = context.get('tone', 'professional')
|
|
|
|
# Generate prompts from H2 sections first
|
|
for i, section in enumerate(sections[:self.requested_image_count - 1]):
|
|
prompt_parts = [
|
|
f"Supporting illustration for: '{section}'.",
|
|
f"Related to main topic: {main_title}.",
|
|
]
|
|
|
|
# Add visual style variation based on position
|
|
if i == 0:
|
|
prompt_parts.append("Conceptual illustration style, clear visual metaphor.")
|
|
elif i == 1:
|
|
prompt_parts.append("Informative diagram style, showing process or relationship.")
|
|
else:
|
|
prompt_parts.append("Supporting visual, complementary to main theme.")
|
|
|
|
prompt_parts.append("Clean composition, web-friendly, no text.")
|
|
prompts.append(' '.join(prompt_parts))
|
|
|
|
# If we need more prompts, use key points
|
|
remaining_count = (self.requested_image_count - 1) - len(prompts)
|
|
if remaining_count > 0 and key_points:
|
|
for point in key_points[:remaining_count]:
|
|
prompt = (
|
|
f"Visual representation of: '{point}'. "
|
|
f"Supporting image for {main_title}. "
|
|
f"Clean, focused illustration, no text overlay."
|
|
)
|
|
prompts.append(prompt)
|
|
|
|
# Fallback generic prompts if still need more
|
|
remaining_count = (self.requested_image_count - 1) - len(prompts)
|
|
for i in range(remaining_count):
|
|
fallback_styles = [
|
|
"Infographic-style supporting visual",
|
|
"Abstract conceptual illustration",
|
|
"Detail-focused supporting image",
|
|
]
|
|
style = fallback_styles[i % len(fallback_styles)]
|
|
prompt = (
|
|
f"{style} for: {main_title}. "
|
|
f"Professional quality, web-optimized, no text."
|
|
)
|
|
prompts.append(prompt)
|
|
|
|
return prompts
|
|
|
|
# === Image Integration ===
|
|
def _insert_images_into_content(self):
|
|
"""Insert generated images into content body.
|
|
|
|
First checks for placeholder format, then falls back to section-based insertion.
|
|
"""
|
|
self.ensure_one()
|
|
|
|
if not self.generated_content:
|
|
return False
|
|
|
|
# Check for placeholder format first
|
|
if 'otk-image-placeholder' in self.generated_content:
|
|
return self._replace_image_placeholders()
|
|
|
|
# Fallback: Insert at section breaks
|
|
return self._insert_images_at_sections()
|
|
|
|
def _replace_image_placeholders(self):
|
|
"""Replace image placeholders with actual generated images."""
|
|
self.ensure_one()
|
|
|
|
# Convert to string to work with regex (Html field might be Markup)
|
|
content = str(self.generated_content or '')
|
|
done_images = self.image_ids.filtered(
|
|
lambda i: i.state == 'done' and i.image and not i.is_cover
|
|
).sorted('sequence')
|
|
|
|
if not done_images:
|
|
return False
|
|
|
|
# Find and replace placeholders
|
|
placeholder_pattern = r'<figure[^>]*class="[^"]*otk-image-placeholder[^"]*"[^>]*data-position="(\d+)"[^>]*>.*?</figure>'
|
|
|
|
def replace_placeholder(match):
|
|
position = int(match.group(1))
|
|
image_index = position - 1 # 0-indexed
|
|
if image_index < len(done_images):
|
|
img = done_images[image_index]
|
|
return self._build_image_html(img)
|
|
return '' # Remove placeholder if no image
|
|
|
|
new_content = re.sub(placeholder_pattern, replace_placeholder, content, flags=re.DOTALL)
|
|
|
|
if new_content != content:
|
|
# Mark images as inserted
|
|
done_images.write({'inserted_in_content': True})
|
|
# Use Markup to prevent HTML escaping
|
|
self.generated_content = Markup(new_content)
|
|
return True
|
|
|
|
return False
|
|
|
|
def _insert_images_at_sections(self):
|
|
"""Insert images at section breaks (after H2 headers) when no placeholders exist."""
|
|
self.ensure_one()
|
|
|
|
# Convert to string to work with regex (Html field might be Markup)
|
|
content = str(self.generated_content or '')
|
|
done_images = self.image_ids.filtered(
|
|
lambda i: i.state == 'done' and i.image and not i.is_cover and not i.inserted_in_content
|
|
).sorted('sequence')
|
|
|
|
if not done_images:
|
|
return False
|
|
|
|
# Find H2 closing tags to insert images after
|
|
h2_pattern = r'(</h2>)'
|
|
matches = list(re.finditer(h2_pattern, content, re.IGNORECASE))
|
|
|
|
if not matches:
|
|
# No H2s found, try inserting after first paragraph
|
|
p_match = re.search(r'(</p>)', content, re.IGNORECASE)
|
|
if p_match and done_images:
|
|
img = done_images[0]
|
|
img_html = self._build_image_html(img)
|
|
insert_pos = p_match.end()
|
|
content = content[:insert_pos] + img_html + content[insert_pos:]
|
|
img.inserted_in_content = True
|
|
# Use Markup to prevent HTML escaping
|
|
self.generated_content = Markup(content)
|
|
return True
|
|
return False
|
|
|
|
# Insert images after every 2nd H2 (spread them out)
|
|
# For 1 image: after 2nd H2
|
|
# For 2 images: after 2nd and 4th H2
|
|
# etc.
|
|
insert_points = matches[1::2] # Every other H2, starting from 2nd (index 1)
|
|
|
|
if not insert_points:
|
|
insert_points = matches[:1] # Use first H2 if only one exists
|
|
|
|
offset = 0
|
|
inserted_count = 0
|
|
for i, match in enumerate(insert_points):
|
|
if i >= len(done_images):
|
|
break
|
|
|
|
img = done_images[i]
|
|
img_html = self._build_image_html(img)
|
|
insert_pos = match.end() + offset
|
|
content = content[:insert_pos] + img_html + content[insert_pos:]
|
|
offset += len(img_html)
|
|
img.inserted_in_content = True
|
|
inserted_count += 1
|
|
|
|
if inserted_count > 0:
|
|
# Use Markup to prevent HTML escaping
|
|
self.generated_content = Markup(content)
|
|
return True
|
|
|
|
return False
|
|
|
|
def _build_image_html(self, image):
|
|
"""Build HTML for inserting an image into content."""
|
|
alt_text = image.alt_text or ''
|
|
# Escape HTML special characters in alt text
|
|
alt_text_escaped = alt_text.replace('"', '"').replace('<', '<').replace('>', '>')
|
|
|
|
html = f'''
|
|
<figure class="otk-generated-image text-center my-4">
|
|
<img src="/web/image/otk.seo.content.image/{image.id}/image"
|
|
alt="{alt_text_escaped}"
|
|
class="img-fluid rounded shadow-sm"
|
|
loading="lazy"
|
|
style="max-width: 100%; height: auto;"/>'''
|
|
|
|
if alt_text:
|
|
html += f'''
|
|
<figcaption class="text-muted small mt-2 fst-italic">{alt_text_escaped}</figcaption>'''
|
|
|
|
html += '''
|
|
</figure>
|
|
'''
|
|
return html
|
|
|
|
def action_insert_images(self):
|
|
"""Manually insert/refresh images in content."""
|
|
self.ensure_one()
|
|
|
|
done_images = self.image_ids.filtered(lambda i: i.state == 'done' and i.image)
|
|
if not done_images:
|
|
raise UserError(_("No completed images to insert. Please generate images first."))
|
|
|
|
# Reset inserted flag to allow re-insertion
|
|
done_images.filtered(lambda i: not i.is_cover).write({'inserted_in_content': False})
|
|
|
|
inserted = self._insert_images_into_content()
|
|
|
|
if inserted:
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('Images Inserted'),
|
|
'message': _('Images have been inserted into the content.'),
|
|
'type': 'success',
|
|
'sticky': False,
|
|
'next': {'type': 'ir.actions.client', 'tag': 'soft_reload'},
|
|
}
|
|
}
|
|
else:
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('No Changes'),
|
|
'message': _('No suitable insertion points found in content.'),
|
|
'type': 'warning',
|
|
'sticky': False,
|
|
}
|
|
}
|
|
|
|
def action_regenerate(self):
|
|
"""Regenerate content."""
|
|
self.ensure_one()
|
|
return self.action_generate()
|
|
|
|
def action_approve(self):
|
|
"""Approve content for publishing."""
|
|
self.ensure_one()
|
|
self.write({'state': 'approved'})
|
|
|
|
def action_back_to_draft(self):
|
|
"""Return to draft state for editing."""
|
|
self.ensure_one()
|
|
self.write({'state': 'draft'})
|
|
|
|
def action_create_blog_post(self):
|
|
"""Create a blog.post from generated content."""
|
|
self.ensure_one()
|
|
|
|
# Check if website_blog is installed
|
|
if not self.env['ir.module.module'].search([
|
|
('name', '=', 'website_blog'),
|
|
('state', '=', 'installed')
|
|
]):
|
|
raise UserError(_("Website Blog module is not installed. Please install it first."))
|
|
|
|
if not self.blog_id:
|
|
# Get default blog
|
|
self.blog_id = self.env['blog.blog'].search([], limit=1)
|
|
if not self.blog_id:
|
|
raise UserError(_("No blog found. Please create a blog first or select one."))
|
|
|
|
# Insert non-cover images into content before creating blog post
|
|
non_cover_images = self.image_ids.filtered(
|
|
lambda i: i.state == 'done' and i.image and not i.is_cover and not i.inserted_in_content
|
|
)
|
|
if non_cover_images:
|
|
self._insert_images_into_content()
|
|
|
|
# Prepare cover image
|
|
cover_properties = {}
|
|
cover_image = self.image_ids.filtered(lambda i: i.is_cover and i.state == 'done')[:1]
|
|
if cover_image:
|
|
cover_properties = {
|
|
'background-image': f"url('/web/image/otk.seo.content.image/{cover_image.id}/image')",
|
|
'resize_class': 'o_half_screen_height',
|
|
'opacity': '0.4',
|
|
'background_color_class': 'o_cc3',
|
|
}
|
|
|
|
# Create blog post
|
|
blog_post = self.env['blog.post'].create({
|
|
'name': self.generated_title or self.name,
|
|
'subtitle': self.generated_subtitle,
|
|
'content': self.generated_content,
|
|
'teaser_manual': self.generated_teaser,
|
|
'website_meta_title': self.website_meta_title,
|
|
'website_meta_description': self.website_meta_description,
|
|
'website_meta_keywords': self.website_meta_keywords,
|
|
'cover_properties': json.dumps(cover_properties) if cover_properties else False,
|
|
'author_id': self.env.user.partner_id.id,
|
|
'blog_id': self.blog_id.id,
|
|
'is_published': False,
|
|
})
|
|
|
|
self.write({
|
|
'blog_post_id': blog_post.id,
|
|
'state': 'published',
|
|
})
|
|
|
|
return {
|
|
'type': 'ir.actions.act_window',
|
|
'res_model': 'blog.post',
|
|
'res_id': blog_post.id,
|
|
'view_mode': 'form',
|
|
'target': 'current',
|
|
}
|
|
|
|
def action_view_blog_post(self):
|
|
"""Open the created blog post."""
|
|
self.ensure_one()
|
|
if not self.blog_post_id:
|
|
raise UserError(_("No blog post has been created yet."))
|
|
|
|
return {
|
|
'type': 'ir.actions.act_window',
|
|
'res_model': 'blog.post',
|
|
'res_id': self.blog_post_id.id,
|
|
'view_mode': 'form',
|
|
'target': 'current',
|
|
}
|
|
|
|
def action_view_images(self):
|
|
"""View generated images."""
|
|
self.ensure_one()
|
|
return {
|
|
'type': 'ir.actions.act_window',
|
|
'res_model': 'otk.seo.content.image',
|
|
'view_mode': 'list,form',
|
|
'domain': [('content_id', '=', self.id)],
|
|
'context': {'default_content_id': self.id},
|
|
'name': _('Generated Images'),
|
|
}
|
|
|
|
def action_apply_to_product(self):
|
|
"""Apply generated SEO content to the source product."""
|
|
self.ensure_one()
|
|
|
|
if not self.source_product_id:
|
|
raise UserError(_("No source product is linked to this content."))
|
|
|
|
if self.content_type != 'product_desc':
|
|
raise UserError(_("This action is only available for Product Description content."))
|
|
|
|
# Prepare values to update
|
|
vals = {}
|
|
|
|
if self.website_meta_title:
|
|
vals['website_meta_title'] = self.website_meta_title
|
|
if self.website_meta_description:
|
|
vals['website_meta_description'] = self.website_meta_description
|
|
if self.website_meta_keywords:
|
|
vals['website_meta_keywords'] = self.website_meta_keywords
|
|
if self.generated_content:
|
|
vals['website_description'] = self.generated_content
|
|
|
|
if not vals:
|
|
raise UserError(_("No SEO content to apply. Please generate content first."))
|
|
|
|
# Apply to product
|
|
self.source_product_id.write(vals)
|
|
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('SEO Applied'),
|
|
'message': _('SEO content has been applied to product "%s".') % self.source_product_id.name,
|
|
'type': 'success',
|
|
'sticky': False,
|
|
}
|
|
}
|
|
|
|
# === Dashboard ===
|
|
@api.model
|
|
def get_dashboard_data(self):
|
|
"""Return aggregated dashboard data for the SEO content dashboard."""
|
|
Content = self.env['otk.seo.content']
|
|
|
|
# Total content
|
|
total_content = Content.search_count([])
|
|
|
|
# Average SEO score (using read_group)
|
|
avg_data = Content._read_group(
|
|
[('seo_score', '>', 0)],
|
|
[],
|
|
['seo_score:avg'],
|
|
)
|
|
avg_seo_score = round(avg_data[0][0], 1) if avg_data else 0
|
|
|
|
# Total tokens
|
|
token_data = Content._read_group(
|
|
[],
|
|
[],
|
|
['token_cost:sum'],
|
|
)
|
|
total_tokens = token_data[0][0] or 0 if token_data else 0
|
|
|
|
# Stale content count
|
|
stale_count = Content.search_count([('is_stale', '=', True)])
|
|
|
|
# Content by state
|
|
state_data = Content._read_group(
|
|
[],
|
|
['state'],
|
|
['__count'],
|
|
)
|
|
content_by_state = {state: count for state, count in state_data}
|
|
|
|
# Content by type
|
|
type_data = Content._read_group(
|
|
[],
|
|
['content_type'],
|
|
['__count'],
|
|
)
|
|
content_by_type = {ctype: count for ctype, count in type_data}
|
|
|
|
# Recent content (last 10)
|
|
recent = Content.search_read(
|
|
[],
|
|
['name', 'content_type', 'state', 'seo_score', 'create_date'],
|
|
limit=10,
|
|
order='create_date desc',
|
|
)
|
|
|
|
return {
|
|
'totalContent': total_content,
|
|
'avgSeoScore': avg_seo_score,
|
|
'totalTokens': total_tokens,
|
|
'staleCount': stale_count,
|
|
'contentByState': content_by_state,
|
|
'contentByType': content_by_type,
|
|
'recentContent': recent,
|
|
}
|
|
|
|
# === Version Management ===
|
|
def _create_version(self, note=None):
|
|
"""Create a version snapshot before changes."""
|
|
self.ensure_one()
|
|
|
|
# Limit versions (configurable via settings)
|
|
ICP = self.env['ir.config_parameter'].sudo()
|
|
max_versions = int(ICP.get_param('otoolkit_seo_content.max_versions', '5'))
|
|
if len(self.version_ids) >= max_versions:
|
|
oldest = self.version_ids.sorted('version_number')[0]
|
|
oldest.unlink()
|
|
|
|
self.env['otk.seo.content.version'].create({
|
|
'content_id': self.id,
|
|
'version_number': self.current_version,
|
|
'generated_title': self.generated_title,
|
|
'generated_subtitle': self.generated_subtitle,
|
|
'generated_content': self.generated_content,
|
|
'generated_teaser': self.generated_teaser,
|
|
'website_meta_title': self.website_meta_title,
|
|
'website_meta_description': self.website_meta_description,
|
|
'note': note or f'Version {self.current_version}',
|
|
})
|
|
|
|
self.current_version += 1
|
|
|
|
def action_restore_version(self):
|
|
"""Open wizard to select version to restore."""
|
|
self.ensure_one()
|
|
return {
|
|
'type': 'ir.actions.act_window',
|
|
'res_model': 'otk.seo.content.version',
|
|
'view_mode': 'list',
|
|
'domain': [('content_id', '=', self.id)],
|
|
'name': _('Version History'),
|
|
'target': 'new',
|
|
}
|
|
|
|
# === Content Freshness ===
|
|
def action_mark_refreshed(self):
|
|
"""Mark content as freshly reviewed/updated."""
|
|
self.ensure_one()
|
|
self.write({
|
|
'last_refreshed_date': fields.Date.today(),
|
|
'refresh_alert_sent': False,
|
|
})
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('Content Refreshed'),
|
|
'message': _('Content has been marked as refreshed.'),
|
|
'type': 'success',
|
|
'sticky': False,
|
|
'next': {'type': 'ir.actions.client', 'tag': 'soft_reload'},
|
|
}
|
|
}
|
|
|
|
@api.model
|
|
def cron_check_stale_content(self):
|
|
"""Cron job to check for stale content and send alerts."""
|
|
# Find published/approved content that is stale and hasn't had alert sent
|
|
stale_content = self.search([
|
|
('state', 'in', ['published', 'approved']),
|
|
('is_stale', '=', True),
|
|
('refresh_alert_sent', '=', False),
|
|
])
|
|
|
|
for content in stale_content:
|
|
# Create activity for content creator
|
|
content.activity_schedule(
|
|
'mail.mail_activity_data_todo',
|
|
summary=_('Content needs refresh: %s') % content.name,
|
|
note=_('This content was last refreshed %d days ago and may need updating to maintain SEO rankings.') % content.days_since_refresh,
|
|
user_id=content.create_uid.id,
|
|
)
|
|
content.refresh_alert_sent = True
|
|
|
|
_logger.info(f"Refresh alert sent for content {content.id}: {content.name}")
|
|
|
|
return True
|
|
|
|
@api.model
|
|
def action_view_stale_content(self):
|
|
"""View all stale content that needs refresh."""
|
|
return {
|
|
'type': 'ir.actions.act_window',
|
|
'name': _('Content Needing Refresh'),
|
|
'res_model': 'otk.seo.content',
|
|
'view_mode': 'list,form',
|
|
'domain': [('is_stale', '=', True)],
|
|
'context': {'search_default_filter_stale': True},
|
|
}
|
|
|
|
|
|
class OtkSeoContentVersion(models.Model):
|
|
_name = 'otk.seo.content.version'
|
|
_description = 'SEO Content Version History'
|
|
_order = 'version_number desc'
|
|
|
|
content_id = fields.Many2one('otk.seo.content', required=True, ondelete='cascade',
|
|
help='The parent content record this version belongs to.')
|
|
version_number = fields.Integer('Version', required=True,
|
|
help='Sequential version number for this snapshot.')
|
|
|
|
# Snapshot of content at this version
|
|
generated_title = fields.Char('Title',
|
|
help='The title at the time this version was saved.')
|
|
generated_subtitle = fields.Char('Subtitle',
|
|
help='The subtitle at the time this version was saved.')
|
|
generated_content = fields.Html('HTML Content', sanitize=False,
|
|
help='The full HTML content at the time this version was saved.')
|
|
generated_teaser = fields.Text('Teaser',
|
|
help='The teaser text at the time this version was saved.')
|
|
website_meta_title = fields.Char('Meta Title',
|
|
help='The SEO meta title at the time this version was saved.')
|
|
website_meta_description = fields.Text('Meta Description',
|
|
help='The SEO meta description at the time this version was saved.')
|
|
|
|
# Metadata
|
|
created_by = fields.Many2one('res.users', default=lambda self: self.env.user,
|
|
help='The user who created this version snapshot.')
|
|
created_at = fields.Datetime(default=fields.Datetime.now,
|
|
help='When this version was saved.')
|
|
note = fields.Char('Version Note',
|
|
help='A brief description of why this version was created (e.g., "Before regeneration", "Manual edit").')
|
|
|
|
# Diff preview
|
|
diff_preview = fields.Html('Diff Preview', compute='_compute_diff_preview', sanitize=False,
|
|
help='Shows differences between this version and the current content.')
|
|
|
|
@api.depends('generated_content', 'content_id.generated_content')
|
|
def _compute_diff_preview(self):
|
|
for record in self:
|
|
if not record.generated_content or not record.content_id.generated_content:
|
|
record.diff_preview = False
|
|
continue
|
|
|
|
old_text = html2plaintext(record.generated_content).splitlines()
|
|
new_text = html2plaintext(record.content_id.generated_content).splitlines()
|
|
|
|
diff = difflib.unified_diff(
|
|
old_text, new_text,
|
|
fromfile=_('Version %s') % record.version_number,
|
|
tofile=_('Current'),
|
|
lineterm='',
|
|
)
|
|
|
|
diff_lines = []
|
|
for line in diff:
|
|
if line.startswith('+++') or line.startswith('---'):
|
|
diff_lines.append(f'<div class="o_diff_header">{line}</div>')
|
|
elif line.startswith('@@'):
|
|
diff_lines.append(f'<div class="o_diff_section">{line}</div>')
|
|
elif line.startswith('+'):
|
|
diff_lines.append(f'<div class="o_diff_add">{line}</div>')
|
|
elif line.startswith('-'):
|
|
diff_lines.append(f'<div class="o_diff_remove">{line}</div>')
|
|
else:
|
|
diff_lines.append(f'<div class="o_diff_context">{line}</div>')
|
|
|
|
if diff_lines:
|
|
record.diff_preview = Markup(
|
|
'<div class="o_seo_diff_view">' + ''.join(diff_lines) + '</div>'
|
|
)
|
|
else:
|
|
record.diff_preview = Markup(
|
|
'<div class="text-muted text-center py-3">'
|
|
'<i class="fa fa-check-circle text-success"/> No differences</div>'
|
|
)
|
|
|
|
def action_view_diff(self):
|
|
"""Open a dialog showing the diff between this version and current content."""
|
|
self.ensure_one()
|
|
return {
|
|
'type': 'ir.actions.act_window',
|
|
'name': _('Version %s - Diff') % self.version_number,
|
|
'res_model': 'otk.seo.content.version',
|
|
'res_id': self.id,
|
|
'view_mode': 'form',
|
|
'view_id': self.env.ref('otoolkit_seo_content.otk_seo_content_version_view_diff').id,
|
|
'target': 'new',
|
|
}
|
|
|
|
def action_restore(self):
|
|
"""Restore this version to the content."""
|
|
self.ensure_one()
|
|
|
|
# Create version of current state first
|
|
self.content_id._create_version(note='Before restore')
|
|
|
|
# Restore
|
|
self.content_id.write({
|
|
'generated_title': self.generated_title,
|
|
'generated_subtitle': self.generated_subtitle,
|
|
'generated_content': self.generated_content,
|
|
'generated_teaser': self.generated_teaser,
|
|
'website_meta_title': self.website_meta_title,
|
|
'website_meta_description': self.website_meta_description,
|
|
})
|
|
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('Version Restored'),
|
|
'message': _('Content has been restored to version %s.') % self.version_number,
|
|
'type': 'success',
|
|
'next': {'type': 'ir.actions.client', 'tag': 'soft_reload'},
|
|
}
|
|
}
|