"""ai_patch.py — Helper functions for rewrite, summary, short frame creation.
FIX v2: Vietnamese font auto-detection (not hardcoded DejaVu paths).
"""
import os
import re
import time
import random
import json
import html as html_lib
import subprocess
import requests
import hashlib
from datetime import datetime, timezone
import ai_ext as base
from ai_ext import app
from fastapi import Request
from fastapi.responses import JSONResponse, HTMLResponse, FileResponse
from bs4 import BeautifulSoup
from urllib.parse import quote_plus
try:
from PIL import Image, ImageDraw, ImageFont
except Exception:
Image = ImageDraw = ImageFont = None
_STOP = set('và của các những một được trong với cho tại sau trước khi không người vietnam hôm nay mới nhất nóng tin tức cập nhật theo từ đến là có thì này đã để'.split())
# ===== VIETNAMESE FONT DETECTION =====
_VN_FONT_REG = None
_VN_FONT_BOLD = None
def _get_vn_fonts():
"""Auto-detect fonts that support Vietnamese diacritics."""
global _VN_FONT_REG, _VN_FONT_BOLD
if _VN_FONT_REG is not None:
return _VN_FONT_REG, _VN_FONT_BOLD
# Try system fonts in priority order (Noto first for full Vietnamese)
reg_paths = [
"/usr/share/fonts/truetype/noto/NotoSans-Regular.ttf",
"/usr/share/fonts/truetype/noto/NotoSans-Variable.ttf",
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
"/usr/share/fonts/truetype/freefont/FreeSans.ttf",
"/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf",
]
bold_paths = [
"/usr/share/fonts/truetype/noto/NotoSans-Bold.ttf",
"/usr/share/fonts/truetype/noto/NotoSans-Variable.ttf",
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
"/usr/share/fonts/truetype/freefont/FreeSans.ttf",
"/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf",
]
# Also search broadly
try:
result = subprocess.run(
['find', '/usr/share/fonts', '-name', '*.ttf', '-type', 'f'],
capture_output=True, text=True, timeout=5
)
all_ttf = result.stdout.strip().split('\n')
for key in ['noto', 'dejavu', 'liberation', 'freefont', 'vn']:
for path in all_ttf:
pl = path.lower()
if key in pl:
if ('bold' in pl or 'variable' in pl) and _VN_FONT_BOLD is None:
try:
_VN_FONT_BOLD = ImageFont.truetype(path, 56)
except:
pass
elif ('regular' in pl or path == '/usr/share/fonts/truetype/vn/VNFont.ttf') and _VN_FONT_REG is None:
try:
_VN_FONT_REG = ImageFont.truetype(path, 48)
except:
pass
except:
pass
# Try regular paths
for path in reg_paths:
if os.path.exists(path) and _VN_FONT_REG is None:
try:
_VN_FONT_REG = ImageFont.truetype(path, 48)
except:
continue
for path in bold_paths:
if os.path.exists(path) and _VN_FONT_BOLD is None:
try:
_VN_FONT_BOLD = ImageFont.truetype(path, 56)
except:
continue
if _VN_FONT_REG is None:
_VN_FONT_REG = ImageFont.load_default()
if _VN_FONT_BOLD is None:
_VN_FONT_BOLD = _VN_FONT_REG
return _VN_FONT_REG, _VN_FONT_BOLD
def _clean(s):
"""Clean text - replace None with empty string, collapse whitespace."""
s = s or ''
s = s.strip()
s = re.sub(r'\s+', ' ', s)
return s
def _postprocess_ai_text(text, max_units=20):
"""Clean AI-generated text: remove HTML artifacts, deduplicate, return string with bullet points."""
text = _clean(text)
if not text:
return ""
# Remove malformed HTML with both href and src attributes (e.g., )
text = re.sub(r']*href\s*=\s*"[^"]*"[^>]*\s*src\s*=\s*"[^"]*"[^>]*>', '', text, flags=re.I)
# Remove any remaining HTML artifacts
text = re.sub(r']*>', '', text, flags=re.I)
text = re.sub(r'
]*>', '', text, flags=re.I)
text = re.sub(r'<[^>]+>', '', text)
text = re.sub(r'', '', text, flags=re.I)
# Fix truncated HTML tags (missing closing >)
text = text.replace('= max_units:
break
return "\n".join(out) if out else ""
def _scrape_article_images(url):
"""Use the comprehensive scraper from ai_ext for images + paragraphs."""
sents = []; imgs = []; og = ''
try:
if url:
# Use the comprehensive scraper first
try:
data = base.scrape_any_url(url)
if data and data.get('text'):
sents = [p.strip() for p in data['text'].split('\n') if len(p.strip()) > 40]
imgs = data.get('images', [])
og = data.get('og_image', '')
if not imgs and og:
imgs = [og]
if sents or imgs:
return {'paragraphs': sents, 'images': imgs, 'og_img': og}
except:
pass
# Fallback: basic scrape
r = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=8)
r.encoding = 'utf-8'
soup = BeautifulSoup(r.text, 'lxml')
og = soup.find('meta', property='og:image')
og = og.get('content', '') if og else ''
if og and og.startswith('//'):
og = 'https:' + og
for p in soup.find_all('p'):
t = _clean(p.get_text())
if len(t) > 30:
sents.append(t)
seen = set()
for im in soup.find_all('img'):
src = im.get('data-src') or im.get('src') or im.get('data-lazy') or ''
if src and 'base64' not in src and src not in seen:
seen.add(src)
if src.startswith('//'):
src = 'https:' + src
imgs.append(src)
# Filter: prefer large images
imgs = [im for im in imgs if not any(x in im.lower() for x in ['logo', 'icon', 'avatar', '.svg', 'banner'])]
return {'paragraphs': sents, 'images': imgs, 'og_img': og}
except Exception as e:
return {'paragraphs': [], 'images': [], 'og_img': ''}
def _extract_key_points_for_slides(paragraphs, max_points=8):
"""Extract key points from paragraphs, splitting long ones into sentence-level points."""
pts = []
for p in paragraphs:
if len(p) < 40:
continue
s = _clean(p)
# Split long paragraphs into sentence-level points
if len(s) > 200:
sentences = re.split(r'(?<=[.!?])\s+(?=[A-Z0-9À-Ỹ])', s)
for sent in sentences:
sent = _clean(sent)
if len(sent) > 60 and sent not in pts:
pts.append(sent)
if len(pts) >= max_points:
break
else:
if s and s not in pts:
pts.append(s)
if len(pts) >= max_points:
break
return pts
def _make_summary_prompt(title, body, source):
topic = source or "báo chính"
return f"""Tóm tắt tin tức: {title}
Nguồn: {topic}
---
{body[:1200] if body else ''}
---
Hãy viết tóm tắt ngắn gọn, súc tích bằng tiếng Việt. Dùng bullet point (•) để liệt kê điểm chính. Không dùng định dạng HTML. Không lặp lại tiêu đề. Không viết "Bản tin VNEWS" hay tiêu đề phụ."""
def _source_line(sources=None):
if not sources:
return "Nguồn tham khảo: VNEWS"
src = sources[0] if sources else {}
title = src.get('title', '')
url = src.get('url', '')
via = src.get('via', '')
if url and via:
return f"Nguồn: [{via}]({url})"
if url:
return f"Nguồn: {url}"
if via:
return f"Nguồn: {via}"
return "Nguồn tham khảo: VNEWS"
def _summary_segments_from_post(post, max_segments=15):
"""Split post text into segments for short video. Uses multi-strategy like main generator."""
text = post.get('text', '')
segs = []
if not text:
return segs
# Strategy 1: split by newlines (bullet points)
for line in text.split('\n'):
line = _clean(line)
line_bare = re.sub(r'^[•\-\*\d\.\)\s]+', '', line).strip()
if len(line_bare) > 25:
segs.append(line_bare)
# Strategy 2: split by sentences
if len(segs) < 3:
segs = []
for sent in re.split(r'(?<=[.!?])\s+(?=[A-Z0-9À-ỸĐ])', text):
sent = _clean(sent)
if len(sent) > 25:
segs.append(sent)
# Strategy 3: chunk by size
if len(segs) < 2:
segs = []
words = text.split()
chunk = []
ccount = 0
for w in words:
chunk.append(w)
ccount += len(w) + 1
if ccount > 150:
segs.append(' '.join(chunk))
chunk = []
ccount = 0
if chunk:
segs.append(' '.join(chunk))
return segs[:max_segments]
def _make_scene_frame(post, segment, idx, total, img_path, out_path, emotion='neutral'):
"""Create a scene frame with Vietnamese font support."""
from PIL import Image, ImageDraw, ImageFont
f, fb = _get_vn_fonts()
sz = (1080, 1920)
bg = Image.new('RGB', sz, (15, 23, 38))
d = ImageDraw.Draw(bg)
# Draw title bar
d.rectangle([0, 0, sz[0], 120], fill=(25, 118, 210))
title = post.get('title', '')[:60]
d.text((sz[0]//2, 60), title, fill='white', font=fb, anchor='mm')
# Draw segment text with word wrapping
max_width = 920
words = (segment or '')[:200].split()
lines = []
cur = []
for w in words:
test = ' '.join(cur + [w])
try:
tw = d.textbbox((0, 0), test, font=f)[2]
except:
tw = len(test) * 22
if tw <= max_width:
cur.append(w)
else:
if cur:
lines.append(' '.join(cur))
cur = [w]
if len(lines) >= 10:
break
if cur and len(lines) < 10:
lines.append(' '.join(cur))
y = 200
for line in lines:
d.text((80, y), line, fill='white', font=f)
y += 56
if y > sz[1] - 80:
break
bg.save(out_path, quality=85)
@app.get('/health')
async def health():
return JSONResponse({'status': 'ok'})