# docx_processor.py - DOCX/DOC Processing and Content Extraction
# FIXED: Split paragraphs that contain soft returns () into separate blocks
# FIXED: Extract text from hyperlinks in DOCX
import io
import re
import base64
import email
import email.policy
import quopri
from html.parser import HTMLParser
from urllib.parse import unquote, urlparse
# ================================================================
# FORMAT DETECTION
# ================================================================
def detect_doc_format(file_bytes):
if not file_bytes or len(file_bytes) < 4:
return 'unknown'
if file_bytes[:4] == b'PK\x03\x04':
return 'docx'
if file_bytes[:8] == b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1':
return 'ole2'
if file_bytes[:4] == b'\xd0\xcf\x11\xe0':
return 'ole2'
header = file_bytes[:64]
for bom in [b'\xef\xbb\xbf', b'\xff\xfe', b'\xfe\xff']:
if header.startswith(bom):
header = header[len(bom):]
break
if header.lstrip().startswith(b'{\\rtf'):
return 'rtf'
header_str = header.decode('ascii', errors='ignore').strip()
if header_str.upper().startswith('MIME-VERSION'):
return 'mhtml'
sample_512 = file_bytes[:512].decode('ascii', errors='ignore').lower()
if 'mime-version' in sample_512 and 'content-type' in sample_512 and 'boundary' in sample_512:
return 'mhtml'
sample = file_bytes[:4096]
sample_str_lower = ''
for enc in ['utf-8', 'utf-16-le', 'utf-16-be', 'cp1252', 'latin-1']:
try:
sample_str_lower = sample.decode(enc, errors='ignore').lower().strip()
if sample_str_lower:
break
except Exception:
continue
if sample_str_lower:
html_markers = [
'', 'mso-',
]
for marker in html_markers:
if marker in sample_str_lower:
return 'html'
return 'unknown'
# ================================================================
# IMAGE NORMALIZATION HELPERS
# ================================================================
def _normalize_image_key(raw_location):
if not raw_location:
return ''
loc = raw_location.strip()
if loc.lower().startswith('cid:'):
loc = loc[4:]
for _ in range(3):
decoded = unquote(loc)
if decoded == loc:
break
loc = decoded
try:
parsed = urlparse(loc)
path = parsed.path if parsed.path else loc
except Exception:
path = loc
filename = path.replace('\\', '/').rsplit('/', 1)[-1].strip()
return filename.lower()
# ================================================================
# MHTML PROCESSOR
# ================================================================
class MHTMLProcessor:
def __init__(self, file_bytes):
self._file_bytes = file_bytes
self._embedded_images = {}
self._ordered_images = []
def process(self):
html_content = self._extract_html_from_mhtml()
if not html_content:
html_content = self._fallback_extract()
if not html_content:
return {
'metadata': {'title': '', 'author': '', 'subject': ''},
'blocks': [{'type': 'paragraph', 'content': 'โ ๏ธ Could not extract content from MHTML file.'}],
}
print(f" ๐ท MHTML: {len(self._ordered_images)} image parts")
processor = HTMLDocProcessor(
html_content,
embedded_images=self._embedded_images,
ordered_images=self._ordered_images
)
result = processor.process()
img_blocks = sum(1 for b in result.get('blocks', []) if b.get('type') == 'image')
print(f"๐ MHTML processed: {len(result.get('blocks', []))} blocks ({img_blocks} images)")
return result
def _store_image(self, payload_bytes, content_type, content_location, content_id):
fmt = content_type.split('/')[-1].lower()
if fmt in ('x-wmf', 'x-emf', 'wmf', 'emf'):
return
if fmt == 'jpg':
fmt = 'jpeg'
b64_data = base64.b64encode(payload_bytes).decode('ascii')
self._ordered_images.append((b64_data, fmt, content_location or content_id or ''))
if content_location:
self._embedded_images[content_location] = (b64_data, fmt)
norm = _normalize_image_key(content_location)
if norm:
self._embedded_images[norm] = (b64_data, fmt)
if content_id:
self._embedded_images[f'cid:{content_id}'] = (b64_data, fmt)
self._embedded_images[content_id] = (b64_data, fmt)
def _extract_html_from_mhtml(self):
try:
msg = email.message_from_bytes(self._file_bytes, policy=email.policy.default)
html_body = None
if msg.is_multipart():
for part in msg.walk():
ct = part.get_content_type()
cl = part.get('Content-Location', '').strip()
cid = part.get('Content-ID', '').strip('<> ')
if ct == 'text/html':
payload = part.get_payload(decode=True)
if payload:
cs = part.get_content_charset() or 'utf-8'
try: html_body = payload.decode(cs, errors='ignore')
except: html_body = payload.decode('utf-8', errors='ignore')
elif ct and ct.startswith('image/'):
payload = part.get_payload(decode=True)
if payload and len(payload) > 100:
self._store_image(payload, ct, cl, cid)
else:
ct = msg.get_content_type()
if ct in ('text/html', 'multipart/related'):
payload = msg.get_payload(decode=True)
if payload:
cs = msg.get_content_charset() or 'utf-8'
try: html_body = payload.decode(cs, errors='ignore')
except: html_body = payload.decode('utf-8', errors='ignore')
return html_body
except Exception as e:
print(f" โ ๏ธ MIME parsing failed: {e}")
return None
def _fallback_extract(self):
try:
text = self._file_bytes.decode('ascii', errors='ignore')
bm = re.search(r'boundary="?([^\s";\r\n]+)"?', text, re.IGNORECASE)
if not bm: return None
boundary = bm.group(1)
parts = text.split(f'--{boundary}')
html_body = None
for part in parts:
he = part.find('\r\n\r\n')
if he == -1: he = part.find('\n\n')
if he == -1: continue
hs = part[:he]; body = part[he:].strip()
ctm = re.search(r'Content-Type:\s*([^\s;]+)', hs, re.IGNORECASE)
ct = ctm.group(1).lower() if ctm else ''
is_qp = bool(re.search(r'Content-Transfer-Encoding:\s*quoted-printable', hs, re.IGNORECASE))
is_b64 = bool(re.search(r'Content-Transfer-Encoding:\s*base64', hs, re.IGNORECASE))
clm = re.search(r'Content-Location:\s*(.+?)[\r\n]', hs, re.IGNORECASE)
cl = clm.group(1).strip() if clm else ''
cidm = re.search(r'Content-ID:\s*\s\r\n]+)>?', hs, re.IGNORECASE)
cid = cidm.group(1).strip() if cidm else ''
if ct == 'text/html':
if is_qp: body = quopri.decodestring(body.encode('ascii', errors='ignore')).decode('utf-8', errors='ignore')
elif is_b64:
try: body = base64.b64decode(body).decode('utf-8', errors='ignore')
except: pass
if ' 100: self._store_image(pb, ct, cl, cid)
except: pass
return html_body
except Exception as e:
print(f" โ ๏ธ Fallback MHTML failed: {e}")
return None
# ================================================================
# HTML COMMENT CLEANUP
# ================================================================
def _clean_html_comments(html_text):
html_text = re.sub(r'(.*?)', r'\1', html_text, flags=re.DOTALL|re.IGNORECASE)
html_text = re.sub(r'(.*?)', r'\1', html_text, flags=re.DOTALL|re.IGNORECASE)
html_text = re.sub(r'', '', html_text, flags=re.DOTALL|re.IGNORECASE)
html_text = re.sub(r'', '', html_text, flags=re.DOTALL)
return html_text
# ================================================================
# HTML DOC PROCESSOR
# ================================================================
class HTMLDocProcessor:
def __init__(self, file_bytes, embedded_images=None, ordered_images=None):
if isinstance(file_bytes, str):
self._html_text = file_bytes
self._file_bytes = file_bytes.encode('utf-8', errors='ignore')
else:
self._file_bytes = file_bytes
self._html_text = self._decode_html()
self._embedded_images = embedded_images or {}
self._ordered_images = ordered_images or []
self._used_image_indices = set()
def _decode_html(self):
if self._file_bytes[:3] == b'\xef\xbb\xbf': return self._file_bytes[3:].decode('utf-8', errors='ignore')
if self._file_bytes[:2] == b'\xff\xfe': return self._file_bytes[2:].decode('utf-16-le', errors='ignore')
if self._file_bytes[:2] == b'\xfe\xff': return self._file_bytes[2:].decode('utf-16-be', errors='ignore')
sample = self._file_bytes[:4096]
try:
st = sample.decode('ascii', errors='ignore')
cm = re.search(r'charset[="\s]+([a-zA-Z0-9\-]+)', st, re.IGNORECASE)
if cm:
try: return self._file_bytes.decode(cm.group(1).strip().strip('"\''), errors='ignore')
except: pass
except: pass
for enc in ['utf-8', 'cp1252', 'latin-1']:
try: return self._file_bytes.decode(enc, errors='ignore')
except: continue
return self._file_bytes.decode('latin-1', errors='replace')
def process(self):
metadata = {'title': '', 'author': '', 'subject': ''}
tm = re.search(r']*>(.*?)', self._html_text, re.IGNORECASE|re.DOTALL)
if tm: metadata['title'] = self._strip_tags(tm.group(1)).strip()
blocks = self._extract_all_blocks()
blocks = [b for b in blocks if b.get('content', '').strip() or b.get('data')]
if not blocks: blocks = self._simple_extract()
img_count = sum(1 for b in blocks if b.get('type') == 'image')
print(f"๐ HTML-DOC processed: {len(blocks)} blocks ({img_count} images)")
return {'metadata': metadata, 'blocks': blocks}
def _strip_tags(self, html_str):
import html as hm
return hm.unescape(re.sub(r'<[^>]+>', '', html_str))
def _resolve_image_src(self, src):
import html as hm
if not src: return None, None
src = hm.unescape(src).strip()
if src.startswith('data:image'):
dm = re.match(r'data:image/([^;]+);base64,(.+)', src, re.DOTALL)
if dm: return dm.group(2).strip(), dm.group(1)
if src in self._embedded_images:
self._mark_used(self._embedded_images[src][0]); return self._embedded_images[src]
ns = _normalize_image_key(src)
if ns and ns in self._embedded_images:
self._mark_used(self._embedded_images[ns][0]); return self._embedded_images[ns]
if ns and '.' in ns:
nne = ns.rsplit('.', 1)[0]
if nne and nne in self._embedded_images:
self._mark_used(self._embedded_images[nne][0]); return self._embedded_images[nne]
if ns:
for loc, (data, fmt) in self._embedded_images.items():
ln = _normalize_image_key(loc)
if ln and ns and ln == ns: self._mark_used(data); return data, fmt
return self._get_next_unused()
def _mark_used(self, data_prefix):
p = data_prefix[:60]
for i, (b, f, l) in enumerate(self._ordered_images):
if i not in self._used_image_indices and b[:60] == p:
self._used_image_indices.add(i); return
def _get_next_unused(self):
for i, (b, f, l) in enumerate(self._ordered_images):
if i not in self._used_image_indices:
self._used_image_indices.add(i); return b, f
return None, None
def _extract_all_blocks(self):
import html as hm
blocks = []
cleaned = re.sub(r']*>.*?', '', self._html_text, flags=re.DOTALL|re.IGNORECASE)
cleaned = re.sub(r']*>.*?', '', cleaned, flags=re.DOTALL|re.IGNORECASE)
vml_srcs = []
for vm in re.finditer(r'', cleaned, re.DOTALL|re.IGNORECASE):
for im in re.finditer(r']*?\bsrc\s*=\s*["\']([^"\']+)["\']', vm.group(1), re.IGNORECASE|re.DOTALL):
vml_srcs.append((hm.unescape(im.group(1)), vm.start()))
cleaned = _clean_html_comments(cleaned)
cleaned = re.sub(r']+>', '', cleaned, flags=re.IGNORECASE)
bm = re.search(r']*>(.*)', cleaned, re.IGNORECASE|re.DOTALL)
if bm: cleaned = bm.group(1)
img_entries = []
for m in re.finditer(r'