454 lines
18 KiB
Python
454 lines
18 KiB
Python
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'<script[^>]*>.*?</script>', '', html_text, flags=re.DOTALL | re.IGNORECASE)
|
|
html_text = re.sub(r'<style[^>]*>.*?</style>', '', html_text, flags=re.DOTALL | re.IGNORECASE)
|
|
|
|
# Convert some tags to text markers
|
|
html_text = re.sub(r'<h[1-6][^>]*>(.*?)</h[1-6]>', r'\n## \1\n', html_text, flags=re.DOTALL | re.IGNORECASE)
|
|
html_text = re.sub(r'<li[^>]*>(.*?)</li>', r'• \1\n', html_text, flags=re.DOTALL | re.IGNORECASE)
|
|
html_text = re.sub(r'<p[^>]*>(.*?)</p>', r'\1\n\n', html_text, flags=re.DOTALL | re.IGNORECASE)
|
|
html_text = re.sub(r'<br\s*/?>', '\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
|