import base64 import csv import io import logging from odoo import api, fields, models, _ from odoo.exceptions import UserError, ValidationError _logger = logging.getLogger(__name__) class SeoContentBatch(models.Model): _name = 'otk.seo.content.batch' _description = 'SEO Content Batch Job' _inherit = ['mail.thread', 'mail.activity.mixin'] _order = 'create_date desc' name = fields.Char('Batch Name', required=True, default=lambda self: _('New Batch')) state = fields.Selection([ ('draft', 'Draft'), ('queued', 'Queued'), ('processing', 'Processing'), ('paused', 'Paused'), ('done', 'Completed'), ('failed', 'Failed'), ('cancelled', 'Cancelled'), ], string='Status', default='draft', tracking=True) # === Concurrency Settings === concurrency = fields.Integer('Concurrency', default=5, help='Number of items to process per cron run. Higher values process faster but use more resources.') @api.constrains('concurrency') def _check_concurrency_range(self): for record in self: if not (1 <= record.concurrency <= 10): raise ValidationError(_('Concurrency must be between 1 and 10.')) # === Source Configuration === source_type = fields.Selection([ ('products', 'From Products'), ('csv', 'From CSV'), ('keywords', 'From Keywords List'), ], string='Source Type', default='products', required=True) product_ids = fields.Many2many('product.template', string='Products', help='Products to generate content for') csv_file = fields.Binary('CSV File', attachment=True) csv_filename = fields.Char('CSV Filename') keywords_list = fields.Text('Keywords List', help='One keyword/topic per line') # === Batch Items === item_ids = fields.One2many('otk.seo.content.batch.item', 'batch_id', string='Batch Items') # === Shared Settings === content_type = fields.Selection([ ('blog_post', 'Blog Post'), ('product_desc', 'Product Description'), ('category_desc', 'Category Description'), ('landing_page', 'Landing Page Content'), ], string='Content Type', default='product_desc', required=True) template_id = fields.Many2one('otk.seo.template', 'Content Template', domain="[('content_type', '=', content_type)]") brand_voice_id = fields.Many2one('otk.seo.brand.voice', 'Brand Voice', default=lambda self: self.env['otk.seo.brand.voice'].search([('is_default', '=', True)], limit=1)) tone = fields.Selection([ ('professional', 'Professional'), ('casual', 'Casual'), ('technical', 'Technical'), ('persuasive', 'Persuasive'), ('conversational', 'Conversational'), ], string='Tone', default='professional') 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') language_id = fields.Many2one('res.lang', 'Language', default=lambda self: self.env['res.lang']._lang_get(self.env.lang or 'en_US')) # === Image Settings === include_images = fields.Boolean('Generate Images', default=True) image_count = fields.Integer('Images per Content', default=1) 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') # === Blog Settings === blog_id = fields.Many2one('blog.blog', 'Target Blog', help='The blog where posts will be published. Only used for Blog Post content type.') # === Progress Tracking === total_items = fields.Integer('Total Items', compute='_compute_progress', store=True) completed_items = fields.Integer('Completed', compute='_compute_progress', store=True) failed_items = fields.Integer('Failed', compute='_compute_progress', store=True) progress_percent = fields.Float('Progress %', compute='_compute_progress', store=True) # === Generated Content === content_ids = fields.One2many('otk.seo.content', 'batch_id', string='Generated Content') content_count = fields.Integer('Content Count', compute='_compute_content_count') # === Timing === started_at = fields.Datetime('Started At') completed_at = fields.Datetime('Completed At') duration = fields.Float('Duration (minutes)', compute='_compute_duration') @api.depends('item_ids', 'item_ids.state') def _compute_progress(self): for batch in self: items = batch.item_ids batch.total_items = len(items) batch.completed_items = len(items.filtered(lambda i: i.state == 'done')) batch.failed_items = len(items.filtered(lambda i: i.state == 'failed')) batch.progress_percent = ( (batch.completed_items + batch.failed_items) / batch.total_items * 100 if batch.total_items else 0 ) @api.depends('content_ids') def _compute_content_count(self): for batch in self: batch.content_count = len(batch.content_ids) # === ETA === eta_minutes = fields.Float('ETA (minutes)', compute='_compute_eta', help='Estimated time remaining for batch completion.') avg_processing_time = fields.Float('Avg Time (s)', compute='_compute_eta', help='Average processing time per item in seconds.') @api.depends('started_at', 'completed_at') def _compute_duration(self): for batch in self: if batch.started_at and batch.completed_at: delta = batch.completed_at - batch.started_at batch.duration = delta.total_seconds() / 60 else: batch.duration = 0 @api.depends('started_at', 'completed_items', 'failed_items', 'total_items') def _compute_eta(self): for batch in self: done_count = batch.completed_items + batch.failed_items if batch.started_at and done_count > 0: now = fields.Datetime.now() elapsed = (now - batch.started_at).total_seconds() avg_time = elapsed / done_count remaining = batch.total_items - done_count batch.avg_processing_time = round(avg_time, 1) batch.eta_minutes = round((avg_time * remaining) / 60, 1) else: batch.avg_processing_time = 0 batch.eta_minutes = 0 @api.onchange('source_type') def _onchange_source_type(self): """Update content type based on source selection.""" if self.source_type == 'products': self.content_type = 'product_desc' elif self.source_type == 'keywords': # Keywords can be for any content type, default to blog if self.content_type == 'product_desc': self.content_type = 'blog_post' @api.onchange('content_type') def _onchange_content_type(self): """Reset template when content type changes and update source type.""" self.template_id = False # If switching to product_desc, suggest products source if self.content_type == 'product_desc' and self.source_type == 'keywords': self.source_type = 'products' # If switching away from product_desc with products source, switch to keywords elif self.content_type != 'product_desc' and self.source_type == 'products': self.source_type = 'keywords' @api.onchange('template_id') def _onchange_template_id(self): """Apply template settings to batch.""" if self.template_id: self.tone = self.template_id.default_tone self.target_word_count = self.template_id.default_word_count self.include_images = self.template_id.include_images self.image_count = self.template_id.default_image_count self.image_style = self.template_id.default_image_style def action_prepare_items(self): """Parse source and ADD items to batch (preserves existing items).""" self.ensure_one() # Collect existing items to avoid duplicates existing_keys = set() for item in self.item_ids: if item.product_id: existing_keys.add(f"product_{item.product_id.id}") elif item.keywords: existing_keys.add(f"keywords_{item.keywords.lower().strip()}") items_data = [] if self.source_type == 'products': if not self.product_ids: raise UserError(_("Please select at least one product.")) for product in self.product_ids: key = f"product_{product.id}" if key not in existing_keys: items_data.append({ 'batch_id': self.id, 'name': product.name, 'product_id': product.id, 'keywords': product.name, 'topic': self._prepare_product_brief(product), }) # Clear product selection after adding self.product_ids = [(5, 0, 0)] elif self.source_type == 'csv': if not self.csv_file: raise UserError(_("Please upload a CSV file.")) csv_items = self._parse_csv() for item in csv_items: key = f"keywords_{item.get('keywords', '').lower().strip()}" if key not in existing_keys: item['batch_id'] = self.id items_data.append(item) # Clear CSV after adding self.csv_file = False self.csv_filename = False elif self.source_type == 'keywords': if not self.keywords_list: raise UserError(_("Please enter keywords or topics.")) for line in self.keywords_list.strip().split('\n'): line = line.strip() if line: key = f"keywords_{line.lower()}" if key not in existing_keys: items_data.append({ 'batch_id': self.id, 'name': line[:100], 'keywords': line, }) # Clear keywords list after adding self.keywords_list = False if not items_data: raise UserError(_("No new items to add. Items may already exist in the batch.")) # Create new batch items self.env['otk.seo.content.batch.item'].create(items_data) def action_clear_items(self): """Remove all items from the batch.""" self.ensure_one() if self.state not in ('draft',): raise UserError(_("Can only clear items when batch is in Draft state.")) self.item_ids.unlink() def _prepare_product_brief(self, product): """Prepare product information for AI.""" brief = f"Product: {product.name}\n" if product.description: brief += f"Description: {product.description}\n" if product.list_price: brief += f"Price: {product.list_price}\n" if product.categ_id: brief += f"Category: {product.categ_id.complete_name}\n" return brief def _parse_csv(self): """Parse CSV file and return items data (without batch_id).""" items_data = [] try: csv_data = base64.b64decode(self.csv_file).decode('utf-8') reader = csv.DictReader(io.StringIO(csv_data)) for row in reader: # Expected columns: keywords, topic (optional), name (optional) keywords = row.get('keywords', row.get('keyword', '')) topic = row.get('topic', row.get('brief', '')) name = row.get('name', keywords[:100] if keywords else 'Untitled') if keywords or topic: items_data.append({ 'name': name, 'keywords': keywords, 'topic': topic, }) except Exception as e: raise UserError(_("Error parsing CSV file: %s") % str(e)) if not items_data: raise UserError(_("No valid items found in CSV. Expected columns: keywords, topic (optional)")) return items_data def action_start_batch(self): """Queue the batch for async processing via cron.""" self.ensure_one() if not self.item_ids: self.action_prepare_items() if not self.item_ids: raise UserError(_("No items to process.")) self.write({ 'state': 'queued', 'started_at': fields.Datetime.now(), }) # Mark all items as pending - they will be picked up by the cron job self.item_ids.write({'state': 'pending'}) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _('Batch Queued'), 'message': _('Batch has been queued for processing. Items will be processed asynchronously.'), 'type': 'success', 'sticky': False, } } def action_pause(self): """Pause the batch processing.""" self.ensure_one() if self.state in ('queued', 'processing'): self.write({'state': 'paused'}) def action_resume(self): """Resume paused batch processing.""" self.ensure_one() if self.state == 'paused': self.write({'state': 'processing'}) def action_cancel(self): """Cancel the batch.""" self.ensure_one() if self.state in ('queued', 'processing', 'paused'): self.write({'state': 'cancelled'}) self.item_ids.filtered(lambda i: i.state == 'pending').write({'state': 'cancelled'}) return True @api.model def cron_process_batch_items(self): """Cron job to process batch items asynchronously. Processes up to `concurrency` items per active batch per cron run. Skips paused and cancelled batches. """ # Find all batches that are queued or processing (not paused) active_batches = self.search([ ('state', 'in', ['queued', 'processing']), ]) for batch in active_batches: # Skip if cancelled or paused if batch.state in ('cancelled', 'paused'): continue # Get pending items up to concurrency limit items_limit = min(batch.concurrency or 5, 10) pending_items = batch.item_ids.filtered(lambda i: i.state == 'pending')[:items_limit] if not pending_items: # All items processed, complete the batch batch._complete_batch() continue # Update batch state to processing if it was queued if batch.state == 'queued': batch.state = 'processing' # Process items (with commit after each to preserve state) for item in pending_items: try: item._process_item() self.env.cr.commit() except Exception as e: _logger.error("Error processing batch item %s: %s", item.id, e) item.write({ 'state': 'failed', 'error_message': str(e), }) self.env.cr.commit() return True def _complete_batch(self): """Mark batch as completed.""" self.ensure_one() failed = self.item_ids.filtered(lambda i: i.state == 'failed') total = len(self.item_ids) if total == 0: state = 'done' elif len(failed) == total: state = 'failed' else: state = 'done' self.write({ 'state': state, 'completed_at': fields.Datetime.now(), }) # Send bus notification to batch creator if self.create_uid and self.create_uid.partner_id: succeeded = len(self.item_ids) - len(failed) self.env['bus.bus']._sendone( self.create_uid.partner_id, 'simple_notification', { 'title': _("Batch Complete!"), 'message': _("Batch '%s' complete: %d/%d items succeeded.") % ( self.name, succeeded, len(self.item_ids)), 'type': 'success' if not failed else 'warning', 'sticky': False, } ) def action_view_content(self): """View generated content.""" self.ensure_one() return { 'type': 'ir.actions.act_window', 'name': _('Generated Content'), 'res_model': 'otk.seo.content', 'view_mode': 'list,form', 'domain': [('batch_id', '=', self.id)], 'context': {'default_batch_id': self.id}, } def action_retry_failed(self): """Retry failed items - re-queue them for async processing.""" self.ensure_one() failed_items = self.item_ids.filtered(lambda i: i.state == 'failed') if not failed_items: raise UserError(_("No failed items to retry.")) failed_items.write({'state': 'pending', 'error_message': False}) self.write({'state': 'queued'}) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _('Items Re-queued'), 'message': _('%d failed items have been re-queued for processing.') % len(failed_items), 'type': 'success', 'sticky': False, } } class SeoContentBatchItem(models.Model): _name = 'otk.seo.content.batch.item' _description = 'SEO Content Batch Item' _order = 'sequence, id' batch_id = fields.Many2one('otk.seo.content.batch', string='Batch', required=True, ondelete='cascade') sequence = fields.Integer('Sequence', default=10) name = fields.Char('Name', required=True) state = fields.Selection([ ('draft', 'Draft'), ('pending', 'Pending'), ('processing', 'Processing'), ('done', 'Completed'), ('failed', 'Failed'), ('cancelled', 'Cancelled'), ], string='Status', default='draft') # === Source Data === product_id = fields.Many2one('product.template', 'Product') keywords = fields.Char('Keywords') topic = fields.Text('Topic/Brief') # === Result === content_id = fields.Many2one('otk.seo.content', 'Generated Content', ondelete='set null') error_message = fields.Text('Error Message') def _process_item(self): """Process this batch item - generate content.""" self.ensure_one() self.state = 'processing' try: batch = self.batch_id # Create content record content_vals = { 'name': self.name, 'content_type': batch.content_type, 'source_keywords': self.keywords, 'source_topic': self.topic, 'source_product_id': self.product_id.id if self.product_id else False, 'template_id': batch.template_id.id if batch.template_id else False, 'brand_voice_id': batch.brand_voice_id.id if batch.brand_voice_id else False, 'tone': batch.tone, 'target_word_count': batch.target_word_count, 'language_id': batch.language_id.id if batch.language_id else False, 'requested_image_count': batch.image_count if batch.include_images else 0, 'image_style': batch.image_style if batch.include_images else False, 'blog_id': batch.blog_id.id if batch.blog_id else False, 'batch_id': batch.id, 'state': 'draft', } content = self.env['otk.seo.content'].create(content_vals) self.content_id = content.id # Generate content content.action_generate() self.state = 'done' _logger.info(f"Batch item {self.id} processed successfully: content {content.id}") except Exception as e: self.state = 'failed' self.error_message = str(e) _logger.error(f"Batch item {self.id} failed: {e}") return True