import base64 import logging import re import requests from odoo import api, fields, models, _ from odoo.exceptions import UserError, ValidationError from odoo.tools import html2plaintext _logger = logging.getLogger(__name__) MAX_REFERENCE_CONTENT_LENGTH = 50000 URL_FETCH_TIMEOUT = 30 # Optional dependencies for better HTML parsing try: from bs4 import BeautifulSoup HAS_BEAUTIFULSOUP = True except ImportError: HAS_BEAUTIFULSOUP = False _logger.info("BeautifulSoup not installed. Basic HTML parsing will be used for URL fetching.") class OtkSeoReferenceTag(models.Model): _name = 'otk.seo.reference.tag' _description = 'SEO Reference Tag' _order = 'name' name = fields.Char('Tag Name', required=True, translate=True) color = fields.Integer('Color Index', default=0) @api.constrains('name') def _check_name_unique(self): for record in self: if record.name: duplicate = self.search([ ('name', '=', record.name), ('id', '!=', record.id), ], limit=1) if duplicate: raise ValidationError(_('Tag name must be unique!')) class OtkSeoReference(models.Model): _name = 'otk.seo.reference' _description = 'SEO Reference Library' _order = 'sequence, name' # === Identification === name = fields.Char('Reference Name', required=True, translate=True, help='A descriptive name for this reference.') reference_type = fields.Selection([ ('internal_content', 'Internal Content'), ('url', 'URL'), ('text', 'Text'), ('file', 'File'), ], string='Type', required=True, default='text', help='The type of reference material.') # === Type-Specific Fields === internal_content_id = fields.Many2one('otk.seo.content', 'Internal Content', domain="[('state', 'in', ['review', 'approved', 'published'])]", help='Link to existing SEO content to use as reference.') url = fields.Char('URL', help='External URL to use as reference. Content will be fetched and stored locally.') # === URL Fetching === url_fetched_content = fields.Text('Fetched Content', help='Content extracted from the URL. Auto-fetched when URL is set.') url_fetch_date = fields.Datetime('Last Fetched', help='When the URL content was last fetched.') url_fetch_status = fields.Selection([ ('pending', 'Pending'), ('fetching', 'Fetching'), ('success', 'Success'), ('error', 'Error'), ], string='Fetch Status', default='pending', help='Status of the URL content fetch operation.') url_fetch_error = fields.Char('Fetch Error', help='Error message if URL fetch failed.') url_content_type = fields.Char('Content Type', help='MIME type of the fetched content.') text_content = fields.Text('Text Content', help='Raw text content to use as reference material.') file = fields.Binary('File', attachment=True, help='Upload a document (PDF, Word, or text file) to use as reference.') file_name = fields.Char('File Name') # === Organization === content_type = fields.Selection([ ('blog_post', 'Blog Post'), ('product_desc', 'Product Description'), ('category_desc', 'Category Description'), ('landing_page', 'Landing Page Content'), ('all', 'All Types'), ], string='For Content Type', default='all', help='Limit this reference to specific content types, or make available for all.') tag_ids = fields.Many2many('otk.seo.reference.tag', 'otk_seo_reference_tag_rel', 'reference_id', 'tag_id', string='Tags', help='Tags for organizing and filtering references.') # === Metadata === description = fields.Text('Description', translate=True, help='Brief description of what this reference contains and when to use it.') active = fields.Boolean('Active', default=True) sequence = fields.Integer('Sequence', default=10) # === Computed === content_preview = fields.Text('Content Preview', compute='_compute_content_preview', help='Preview of the extracted content.') content_length = fields.Integer('Content Length', compute='_compute_content_preview', help='Length of extractable content in characters.') @api.depends('reference_type', 'internal_content_id', 'url', 'text_content', 'file', 'url_fetched_content', 'url_fetch_status') def _compute_content_preview(self): for record in self: content = record._extract_content() record.content_length = len(content) if len(content) > 500: record.content_preview = content[:500] + '...' else: record.content_preview = content @api.constrains('reference_type', 'internal_content_id', 'url', 'text_content', 'file') def _check_required_fields(self): """Ensure required fields are filled based on reference type.""" for record in self: if record.reference_type == 'internal_content' and not record.internal_content_id: raise UserError(_("Please select an internal content for this reference.")) elif record.reference_type == 'url' and not record.url: raise UserError(_("Please provide a URL for this reference.")) elif record.reference_type == 'text' and not record.text_content: raise UserError(_("Please provide text content for this reference.")) elif record.reference_type == 'file' and not record.file: raise UserError(_("Please upload a file for this reference.")) @api.onchange('reference_type') def _onchange_reference_type(self): """Clear irrelevant fields when type changes.""" if self.reference_type != 'internal_content': self.internal_content_id = False if self.reference_type != 'url': self.url = False self.url_fetched_content = False self.url_fetch_date = False self.url_fetch_status = 'pending' self.url_fetch_error = False self.url_content_type = False if self.reference_type != 'text': self.text_content = False if self.reference_type != 'file': self.file = False self.file_name = False def _extract_content(self): """Extract text content from reference based on type.""" self.ensure_one() if self.reference_type == 'url': # Return fetched content if available, otherwise indicate pending if self.url_fetched_content and self.url_fetch_status == 'success': return self.url_fetched_content elif self.url_fetch_status == 'error': return f"[URL fetch error: {self.url_fetch_error or 'Unknown error'}]" elif self.url_fetch_status == 'fetching': return f"[Fetching content from: {self.url}...]" else: return f"[Content pending fetch from: {self.url}]" elif self.reference_type == 'text': return self.text_content or '' elif self.reference_type == 'internal_content': if self.internal_content_id and self.internal_content_id.generated_content: return html2plaintext(self.internal_content_id.generated_content) return '' elif self.reference_type == 'file': return self._extract_file_content() return '' def _extract_file_content(self): """Extract text content from uploaded file.""" self.ensure_one() if not self.file: return '' try: file_data = base64.b64decode(self.file) file_name_lower = (self.file_name or '').lower() # Plain text if file_name_lower.endswith('.txt'): return file_data.decode('utf-8', errors='ignore') # PDF elif file_name_lower.endswith('.pdf'): try: import PyPDF2 from io import BytesIO reader = PyPDF2.PdfReader(BytesIO(file_data)) text_parts = [] for page in reader.pages: text = page.extract_text() if text: text_parts.append(text) return '\n'.join(text_parts) except ImportError: _logger.warning("PyPDF2 not installed for PDF extraction") return _('[PDF content - install PyPDF2 for extraction]') except Exception as e: _logger.error(f"PDF extraction error: {e}") return _('[PDF extraction failed]') # Word documents elif file_name_lower.endswith(('.docx', '.doc')): try: from docx import Document from io import BytesIO doc = Document(BytesIO(file_data)) return '\n'.join(para.text for para in doc.paragraphs if para.text) except ImportError: _logger.warning("python-docx not installed for Word extraction") return _('[Word content - install python-docx for extraction]') except Exception as e: _logger.error(f"Word extraction error: {e}") return _('[Word extraction failed]') else: # Try as plain text return file_data.decode('utf-8', errors='ignore') except Exception as e: _logger.error(f"File content extraction error: {e}") return '' def get_api_payload(self): """Build API payload for this reference.""" self.ensure_one() content = self._extract_content() # Truncate if too long if len(content) > MAX_REFERENCE_CONTENT_LENGTH: content = content[:MAX_REFERENCE_CONTENT_LENGTH] _logger.info(f"Reference {self.id} content truncated to {MAX_REFERENCE_CONTENT_LENGTH} chars") return { 'type': self.reference_type, 'title': self.name, 'content': content, 'source': 'internal' if self.reference_type == 'internal_content' else self.reference_type, } # === URL Fetching Methods === def action_fetch_url_content(self): """Fetch content from URL and store it.""" self.ensure_one() if self.reference_type != 'url' or not self.url: raise UserError(_("This reference is not a URL type or URL is empty.")) self.write({'url_fetch_status': 'fetching', 'url_fetch_error': False}) try: headers = { 'User-Agent': 'Mozilla/5.0 (compatible; OToolKit/1.0; +https://otoolkit.app)', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', } response = requests.get( self.url, headers=headers, timeout=URL_FETCH_TIMEOUT, allow_redirects=True ) response.raise_for_status() content_type = response.headers.get('Content-Type', '') if 'text/html' in content_type: text_content = self._parse_html_content(response.text) elif 'text/plain' in content_type: text_content = response.text elif 'application/json' in content_type: text_content = response.text else: # Try to decode as text anyway try: text_content = response.text except Exception: text_content = f"[Unsupported content type: {content_type}]" # Clean and truncate if needed text_content = self._clean_text_content(text_content) if len(text_content) > MAX_REFERENCE_CONTENT_LENGTH: text_content = text_content[:MAX_REFERENCE_CONTENT_LENGTH] _logger.info(f"URL content truncated to {MAX_REFERENCE_CONTENT_LENGTH} chars for reference {self.id}") self.write({ 'url_fetched_content': text_content, 'url_fetch_date': fields.Datetime.now(), 'url_fetch_status': 'success', 'url_fetch_error': False, 'url_content_type': content_type[:100] if content_type else False, }) _logger.info(f"Successfully fetched {len(text_content)} chars from {self.url} for reference {self.id}") except requests.exceptions.Timeout: self._set_url_fetch_error(_("Request timed out after %d seconds.") % URL_FETCH_TIMEOUT) except requests.exceptions.TooManyRedirects: self._set_url_fetch_error(_("Too many redirects.")) except requests.exceptions.SSLError as e: self._set_url_fetch_error(_("SSL Error: %s") % str(e)[:100]) except requests.exceptions.ConnectionError: self._set_url_fetch_error(_("Could not connect to URL.")) except requests.exceptions.HTTPError as e: self._set_url_fetch_error(_("HTTP Error %s") % e.response.status_code if e.response else str(e)[:100]) except Exception as e: _logger.error(f"URL fetch error for reference {self.id}: {e}") self._set_url_fetch_error(str(e)[:200]) def _set_url_fetch_error(self, error_message): """Set URL fetch error state.""" self.write({ 'url_fetch_status': 'error', 'url_fetch_error': error_message[:255] if error_message else 'Unknown error', 'url_fetch_date': fields.Datetime.now(), }) def _parse_html_content(self, html_text): """Parse HTML and extract readable text content.""" if HAS_BEAUTIFULSOUP: return self._parse_html_with_beautifulsoup(html_text) else: return self._parse_html_basic(html_text) def _parse_html_with_beautifulsoup(self, html_text): """Parse HTML using BeautifulSoup for better extraction.""" soup = BeautifulSoup(html_text, 'html.parser') # Remove script, style, nav, footer, header, aside elements for element in soup(['script', 'style', 'nav', 'footer', 'header', 'aside', 'noscript', 'iframe']): element.decompose() # Try to find main content area main_content = ( soup.find('main') or soup.find('article') or soup.find('div', {'class': re.compile(r'content|main|post|article', re.I)}) or soup.find('div', {'id': re.compile(r'content|main|post|article', re.I)}) or soup.find('body') ) if main_content: # Get text with some structure preserved text_parts = [] for element in main_content.find_all(['h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'p', 'li', 'td', 'th', 'blockquote']): text = element.get_text(strip=True) if text: # Add markers for headings if element.name in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']: text_parts.append(f"\n## {text}\n") elif element.name == 'li': text_parts.append(f"• {text}") else: text_parts.append(text) return '\n'.join(text_parts) else: return soup.get_text(separator='\n', strip=True) def _parse_html_basic(self, html_text): """Basic HTML parsing without BeautifulSoup.""" # Remove script and style content html_text = re.sub(r']*>.*?', '', html_text, flags=re.DOTALL | re.IGNORECASE) html_text = re.sub(r']*>.*?', '', html_text, flags=re.DOTALL | re.IGNORECASE) # Convert some tags to text markers html_text = re.sub(r']*>(.*?)', r'\n## \1\n', html_text, flags=re.DOTALL | re.IGNORECASE) html_text = re.sub(r']*>(.*?)', r'• \1\n', html_text, flags=re.DOTALL | re.IGNORECASE) html_text = re.sub(r']*>(.*?)

', r'\1\n\n', html_text, flags=re.DOTALL | re.IGNORECASE) html_text = re.sub(r'', '\n', html_text, flags=re.IGNORECASE) # Remove remaining tags html_text = re.sub(r'<[^>]+>', '', html_text) # Decode HTML entities html_text = html_text.replace(' ', ' ') html_text = html_text.replace('&', '&') html_text = html_text.replace('<', '<') html_text = html_text.replace('>', '>') html_text = html_text.replace('"', '"') return html_text def _clean_text_content(self, text): """Clean extracted text content.""" if not text: return '' # Normalize whitespace text = re.sub(r'[ \t]+', ' ', text) # Remove excessive newlines text = re.sub(r'\n{3,}', '\n\n', text) # Strip each line lines = [line.strip() for line in text.split('\n')] # Remove empty lines at start/end text = '\n'.join(lines).strip() return text # === Auto-fetch on Create/Write === @api.model_create_multi def create(self, vals_list): """Auto-fetch URL content for new URL references.""" records = super().create(vals_list) for record in records: if record.reference_type == 'url' and record.url: try: record.action_fetch_url_content() except Exception as e: _logger.warning(f"Auto-fetch failed for new reference {record.id}: {e}") return records def write(self, vals): """Auto-fetch URL content when URL changes.""" result = super().write(vals) # Re-fetch if URL changed for URL type references if 'url' in vals and vals.get('url'): for record in self: if record.reference_type == 'url': try: record.action_fetch_url_content() except Exception as e: _logger.warning(f"Auto-fetch failed for reference {record.id}: {e}") return result