import ast import difflib import json import logging import re import requests import time 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 = '

Generate content to see SEO analysis.

''' 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''' ''') # 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('
') 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'''
{label} {section_score}/{max_score}
{advice if advice else 'Excellent!'}
{f'
{details}
' if details else ''}
''') html_parts.append('
') 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"Length: {data['length']} characters") elif key == 'meta_desc' and 'length' in data: parts.append(f"Length: {data['length']} characters") elif key == 'meta_keywords' and 'count' in data: parts.append(f"Count: {data['count']} keywords") elif key == 'keywords' and 'density' in data: parts.append(f"Density: {data['density']:.2f}%") elif key == 'content_length' and 'words' in data: parts.append(f"Word count: {data['words']}") elif key == 'headers': h2 = '✓' if data.get('h2') else '✗' h3 = '✓' if data.get('h3') else '✗' parts.append(f"H2: {h2}   H3: {h3}") elif key == 'images': count = data.get('count', 0) with_alt = data.get('with_alt', 0) parts.append(f"Images: {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_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']*>(.*?)

', 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_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']*class="[^"]*otk-image-placeholder[^"]*"[^>]*data-position="(\d+)"[^>]*>.*?' 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'()' 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'(

)', 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'''
{alt_text_escaped}''' if alt_text: html += f'''
{alt_text_escaped}
''' html += '''
''' 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'
{line}
') elif line.startswith('@@'): diff_lines.append(f'
{line}
') elif line.startswith('+'): diff_lines.append(f'
{line}
') elif line.startswith('-'): diff_lines.append(f'
{line}
') else: diff_lines.append(f'
{line}
') if diff_lines: record.diff_preview = Markup( '
' + ''.join(diff_lines) + '
' ) else: record.diff_preview = Markup( '
' ' No differences
' ) 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'}, } }