"""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'})