"""VNEWS AI Extension - rewrite + auto short video generation.
Imported by app_v2_entry.py to register /api/rewrite_share, /api/topic_post,
/api/ai_wall, /api/wall, /api/ai/short endpoints on the main FastAPI app.
Uses main.py's WALL_FILE (wall_posts.json) for unified data store.
TTS: edge-tts (HoaiMy female, NamMinh male) with speed control + gTTS fallback.
"""
import os, re, json, time, random, html as html_lib, subprocess, asyncio
from urllib.parse import quote_plus, quote, urlparse, urljoin
from typing import Optional, List, Dict
import requests
from bs4 import BeautifulSoup
from fastapi import Request, Query
from fastapi.responses import HTMLResponse, JSONResponse, FileResponse
from main import app
# Import wall store from main.py so we read/write the SAME file
try:
from main import _load_wall, _save_wall, _web_context # noqa: F401
except ImportError:
_data_dir = "/data" if os.path.isdir("/data") else "/app/data"
_wall_file = os.path.join(_data_dir, "wall_posts.json")
def _load_wall():
try:
if os.path.exists(_wall_file):
with open(_wall_file, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return []
def _save_wall(posts):
try:
os.makedirs(os.path.dirname(_wall_file), exist_ok=True)
tmp = _wall_file + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(posts[:100], f, ensure_ascii=False)
os.replace(tmp, _wall_file)
except Exception:
pass
def _web_context(topic):
return ""
try:
from huggingface_hub import AsyncInferenceClient
except Exception:
AsyncInferenceClient = None
try:
from gtts import gTTS
except Exception:
gTTS = None
try:
from PIL import Image, ImageDraw, ImageFont
except Exception:
Image = ImageDraw = ImageFont = None
try:
import edge_tts
except Exception:
edge_tts = None
def _hf_token():
for k in ("HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HF_API_TOKEN"):
v = os.getenv(k, "").strip()
if v:
return v
return ""
HF_TOKEN = _hf_token()
QWEN_VL_MODEL = os.getenv("QWEN_VL_MODEL", "Qwen/Qwen2.5-VL-7B-Instruct")
# Fast TEXT models for summaries that don't need vision (much faster than the VL model).
QWEN_TEXT_MODELS = [m.strip() for m in os.getenv(
"QWEN_TEXT_MODELS",
"Qwen/Qwen2.5-72B-Instruct,meta-llama/Llama-3.3-70B-Instruct,Qwen/Qwen2.5-7B-Instruct"
).split(",") if m.strip()]
_WORKING_MODEL_TEXT = None # cached last-working text model
_WORKING_MODEL_VL = None # cached last-working vision model
DATA_DIR = "/data" if os.path.isdir("/data") else "/app/data"
SHORTS_DIR = os.path.join(DATA_DIR, "ai_shorts")
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Accept-Language": "vi-VN,vi;q=0.9,en;q=0.8"
}
LAST_QWEN_ERROR = ""
# ===== TTS VOICE CONFIG =====
# Multilingual neural voices grouped by country/language
# Format: key -> {id, gender, name, country, lang, flag}
TTS_VOICES = {
# === VIETNAM (edge-tts) ===
"hoaimy": {"id": "vi-VN-HoaiMyNeural", "gender": "female", "name": "Hoài My", "country": "Việt Nam", "lang": "vi", "flag": "🇻🇳", "engine": "edge"},
"namminh": {"id": "vi-VN-NamMinhNeural", "gender": "male", "name": "Nam Minh", "country": "Việt Nam", "lang": "vi", "flag": "🇻🇳", "engine": "edge"},
# === gTTS (Google, tiếng Việt cơ bản) ===
"gtts_vi": {"id": "gtts", "gender": "female", "name": "gTTS Google", "country": "Việt Nam", "lang": "vi", "flag": "🇻🇳", "engine": "gtts"},
# === MULTILINGUAL (đa ngôn ngữ — đọc được tiếng Việt + nhiều thứ tiếng) ===
"en_au_william": {"id": "en-AU-WilliamMultilingualNeural", "gender": "male", "name": "William (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"en_us_andrew": {"id": "en-US-AndrewMultilingualNeural", "gender": "male", "name": "Andrew (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"en_us_ava": {"id": "en-US-AvaMultilingualNeural", "gender": "female", "name": "Ava (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"en_us_brian": {"id": "en-US-BrianMultilingualNeural", "gender": "male", "name": "Brian (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"en_us_emma": {"id": "en-US-EmmaMultilingualNeural", "gender": "female", "name": "Emma (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"fr_vivienne": {"id": "fr-FR-VivienneMultilingualNeural","gender": "female", "name": "Vivienne (Đa NN)","country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"fr_remy": {"id": "fr-FR-RemyMultilingualNeural", "gender": "male", "name": "Rémy (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"de_seraphina": {"id": "de-DE-SeraphinaMultilingualNeural","gender": "female","name": "Seraphina (Đa NN)","country": "Đa ngôn ngữ","lang": "multi", "flag": "🌐", "engine": "edge"},
"de_florian": {"id": "de-DE-FlorianMultilingualNeural", "gender": "male", "name": "Florian (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"it_giuseppe": {"id": "it-IT-GiuseppeMultilingualNeural","gender": "male", "name": "Giuseppe (Đa NN)","country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"ko_hyunsu": {"id": "ko-KR-HyunsuMultilingualNeural", "gender": "male", "name": "Hyunsu (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
"pt_thalita": {"id": "pt-BR-ThalitaMultilingualNeural", "gender": "female", "name": "Thalita (Đa NN)", "country": "Đa ngôn ngữ", "lang": "multi", "flag": "🌐", "engine": "edge"},
}
TTS_DEFAULT_VOICE = "hoaimy"
TTS_DEFAULT_SPEED = 1.2 # 1.2x speed for faster reading
# Topic -> voice mapping (auto-detect based on topic keywords)
TOPIC_VOICE_MAP = {
# Sports -> male voice
"bóng đá": "namminh", "thể thao": "namminh", "world cup": "namminh",
"premier league": "namminh", "champions league": "namminh", "la liga": "namminh",
"serie a": "namminh", "bundesliga": "namminh", "v-league": "namminh",
"tennis": "namminh", "olympic": "namminh", "f1": "namminh", "moto": "namminh",
# Lifestyle/Health/Entertainment -> female voice
"sức khỏe": "hoaimy", "làm đẹp": "hoaimy", "giải trí": "hoaimy",
"âm nhạc": "hoaimy", "phim": "hoaimy", "thời trang": "hoaimy",
"ẩm thực": "hoaimy", "du lịch": "hoaimy", "gia đình": "hoaimy",
"tình yêu": "hoaimy", "hôn nhân": "hoaimy", "mẹ và bé": "hoaimy",
# Tech/Science -> male voice
"công nghệ": "namminh", "ai": "namminh", "robot": "namminh",
"khoa học": "namminh", "vũ trụ": "namminh", "điện thoại": "namminh",
"laptop": "namminh", "game": "namminh",
# News/Politics/Economy -> male voice
"chính trị": "namminh", "kinh tế": "namminh", "tài chính": "namminh",
"chứng khoán": "namminh", "ngân hàng": "namminh", "thị trường": "namminh",
"xã hội": "namminh", "pháp luật": "namminh", "giáo dục": "namminh",
}
def _detect_voice_for_topic(title: str, text: str) -> str:
"""Auto-detect the best voice based on topic keywords."""
combined = (title + " " + text[:500]).lower()
for keyword, voice_id in TOPIC_VOICE_MAP.items():
if keyword in combined:
return voice_id
return TTS_DEFAULT_VOICE
# ===== EMOTION (CẢM XÚC) FOR TTS =====
# edge-tts does NOT support Azure express-as styles, so emotion is simulated
# with pitch + rate(speed multiplier) + volume. Each preset is a delta.
# rate_mul multiplies the user/auto speed; pitch is absolute Hz; volume is percent.
EMOTION_PRESETS = {
"vui": {"label": "Vui tươi", "emoji": "😊", "rate_mul": 1.06, "pitch": "+15Hz", "volume": "+6%"},
"hao_hung": {"label": "Hào hứng", "emoji": "🔥", "rate_mul": 1.12, "pitch": "+24Hz", "volume": "+12%"},
"nghiem": {"label": "Nghiêm túc", "emoji": "📰", "rate_mul": 1.00, "pitch": "-3Hz", "volume": "+0%"},
"tram": {"label": "Trầm ấm", "emoji": "🌙", "rate_mul": 0.94, "pitch": "-10Hz", "volume": "+0%"},
"buon": {"label": "Buồn/Xúc động", "emoji": "💧", "rate_mul": 0.88, "pitch": "-18Hz", "volume": "-4%"},
"trung_tinh":{"label": "Trung tính", "emoji": "🎙️", "rate_mul": 1.00, "pitch": "+0Hz", "volume": "+0%"},
}
EMOTION_DEFAULT = "trung_tinh"
# Topic keyword -> emotion. Checked in order; first match wins.
TOPIC_EMOTION_MAP = {
# Sports / wins -> excited
"chiến thắng": "hao_hung", "vô địch": "hao_hung", "world cup": "hao_hung",
"bóng đá": "hao_hung", "thể thao": "hao_hung", "ghi bàn": "hao_hung",
"champions league": "hao_hung", "premier league": "hao_hung", "chung kết": "hao_hung",
"olympic": "hao_hung", "kỷ lục": "hao_hung",
# Entertainment / lifestyle / good news -> cheerful
"giải trí": "vui", "âm nhạc": "vui", "phim": "vui", "lễ hội": "vui",
"du lịch": "vui", "ẩm thực": "vui", "thời trang": "vui", "ra mắt": "vui",
"khai trương": "vui", "tin vui": "vui", "hạnh phúc": "vui",
# Sad / accidents / loss -> sad
"tai nạn": "buon", "qua đời": "buon", "tử vong": "buon", "thiệt mạng": "buon",
"động đất": "buon", "lũ lụt": "buon", "thiên tai": "buon", "cháy": "buon",
"tang lễ": "buon", "mất tích": "buon", "thương tâm": "buon",
# Health / science / calm -> calm warm
"sức khỏe": "tram", "y tế": "tram", "bệnh": "tram", "dinh dưỡng": "tram",
"tâm lý": "tram", "thiền": "tram", "giấc ngủ": "tram",
# News / politics / economy / law -> serious
"chính trị": "nghiem", "kinh tế": "nghiem", "tài chính": "nghiem",
"chứng khoán": "nghiem", "pháp luật": "nghiem", "tòa án": "nghiem",
"ngân hàng": "nghiem", "thị trường": "nghiem", "lạm phát": "nghiem",
"công nghệ": "nghiem", "ai": "nghiem", "khoa học": "nghiem", "giáo dục": "nghiem",
}
def _detect_emotion_for_topic(title: str, text: str) -> str:
"""Auto-detect emotion (cảm xúc) from topic/content keywords."""
combined = (title + " " + text[:600]).lower()
for keyword, emo in TOPIC_EMOTION_MAP.items():
if keyword in combined:
return emo
return EMOTION_DEFAULT
def _detect_voice_emotion(title: str, text: str) -> tuple:
"""Return (voice_id, emotion_id) auto-chosen for the article's topic."""
return _detect_voice_for_topic(title, text), _detect_emotion_for_topic(title, text)
# ===== TEXT HELPERS =====
def _clean_text(s: str) -> str:
s = html_lib.unescape(s or "")
return re.sub(r"\s+", " ", s).strip()
def _domain(u):
try:
return urlparse(u).netloc.replace("www.", "")
except Exception:
return ""
def _safe_name(s):
return re.sub(r"[^a-zA-Z0-9_-]+", "_", str(s))[:80]
# ===== CLEAN AI OUTPUT =====
def _clean_ai_output(text: str) -> str:
"""Remove markdown artifacts, instruction leakage, and aggressively dedup content."""
if not text:
return ""
# Remove markdown headings, bold, italic, horizontal rules
text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
text = re.sub(r'\*\*([^*]+)\*\*', r'\1', text)
text = re.sub(r'\*([^*]+)\*', r'\1', text)
text = re.sub(r'^---+\s*$', '', text, flags=re.MULTILINE)
text = re.sub(r'^[-*_]{3,}\s*$', '', text, flags=re.MULTILINE)
# Remove common AI instruction leakage phrases (entire line)
leakage = [
r'Dưới đây là', r'Theo yêu cầu', r'Tôi sẽ viết', r'Tôi sẽ tóm tắt',
r'Đây là bài', r'Đây là nội dung', r'Bài viết sau đây',
r'Nội dung (tóm tắt|chính)', r'Nhiệm vụ', r'Vai trò', r'Tôi là',
r'Dựa trên.*tôi sẽ', r'Hãy', r'Bạn cần', r'Đọc bài viết',
r'Tôi xin', r'Xin chào', r'Trân trọng', r'Kính thưa',
r'Dựa trên.*dưới đây', r'Sau đây là', r'Dưới đây là bài',
]
for phrase in leakage:
text = re.sub(r'^' + phrase + r'[^\n]*\n?', '', text, flags=re.MULTILINE | re.IGNORECASE)
text = re.sub(r'\n{3,}', '\n\n', text)
# --- Aggressive dedup: split into sentences, remove any that repeats ---
# Normalize: collapse whitespace, strip
def _norm(s):
return re.sub(r'\s+', ' ', s.strip().lower())
# Split by sentence-ending punctuation (keep delimiters)
raw_parts = re.split(r'(?<=[.!?])\s+', text.strip())
seen_sentences = set()
unique_parts = []
for part in raw_parts:
n = _norm(part)
# Skip near-duplicate: if >70% of an existing seen sentence matches
is_dup = False
if n:
if n in seen_sentences:
is_dup = True
else:
partial = re.sub(r'\W+', '', n)
for seen in seen_sentences:
seen_clean = re.sub(r'\W+', '', seen)
# Check substring match for very similar sentences
if partial and seen_clean and (
partial in seen_clean or seen_clean in partial
):
shorter = min(len(partial), len(seen_clean))
longer = max(len(partial), len(seen_clean))
if shorter > 20 and shorter / longer > 0.75:
is_dup = True
break
if is_dup:
continue
if n:
seen_sentences.add(n)
unique_parts.append(part)
result = ' '.join(unique_parts).strip()
# Final pass: remove any remaining consecutive duplicate lines
lines = result.split('\n')
final_lines = []
prev_line = ""
for line in lines:
stripped = line.strip()
if stripped and stripped == prev_line:
continue
final_lines.append(line)
prev_line = stripped
result = '\n'.join(final_lines).strip()
return result
# ===== EXTRACT ALL IMAGES FROM ARTICLE =====
def _extract_all_images(soup, base_url: str) -> List[Dict]:
"""Extract ALL content images from an article page using multi-strategy approach."""
images = []
seen_urls = set()
skip_patterns = [
"avatar", "icon", "logo", "button", "banner-ad", "tracking",
"beacon", "pixel", "1x1", "spacer", "emoji", "sprite", "placeholder",
"advertisement", "ads", "widget", "sidebar", "footer-logo",
]
def _add_image(src: str, alt: str = "", source_tag: str = "img"):
if not src or src.startswith("data:"):
return
abs_url = urljoin(base_url, src.strip())
if abs_url in seen_urls:
return
# Skip non-content images by URL pattern
if any(p in abs_url.lower() for p in skip_patterns):
return
# Skip very small images (likely icons)
try:
parsed = urlparse(abs_url)
path = parsed.path.lower()
if any(path.endswith(ext) for ext in ['.svg', '.ico', '.gif']):
return
except Exception:
pass
seen_urls.add(abs_url)
images.append({"url": abs_url, "alt": alt, "source": source_tag})
# Strategy 1: Standard ![]()
tags with all lazy-load attributes
for img in soup.find_all("img"):
src = (img.get("src") or img.get("data-src") or img.get("data-lazy-src") or
img.get("data-original") or img.get("data-srcset", "").split(",")[0].strip().split(" ")[0])
_add_image(src, alt=img.get("alt", ""), source_tag="img")
# Strategy 2: srcset on ![]()
for img in soup.find_all("img", srcset=True):
for part in img["srcset"].split(","):
part = part.strip()
if part:
_add_image(part.split(" ")[0], alt=img.get("alt", ""), source_tag="srcset")
# Strategy 3: with
for picture in soup.find_all("picture"):
for source in picture.find_all("source"):
srcset = source.get("srcset", "")
for part in srcset.split(","):
part = part.strip()
if part:
_add_image(part.split(" ")[0], source_tag="picture/srcset")
fallback_img = picture.find("img")
if fallback_img:
_add_image(
fallback_img.get("src") or fallback_img.get("data-src"),
alt=fallback_img.get("alt", ""),
source_tag="picture/img"
)
# Strategy 4: WordPress CMS patterns
for img in soup.find_all("img", class_=re.compile(r"wp-image|size-large|size-full|aligncenter")):
_add_image(img.get("data-src") or img.get("src"),
alt=img.get("alt", ""), source_tag="wp-image")
# Strategy 5: Background images in style attributes
for tag in soup.find_all(style=re.compile(r"background-image")):
for m in re.findall(r'url\(["\']?(.*?)["\']?\)', tag.get("style", "")):
_add_image(m, source_tag="background-style")
# Strategy 6: og:image (featured/hero image)
og_image = soup.find("meta", property="og:image")
if og_image and og_image.get("content"):
_add_image(og_image["content"], source_tag="og:image")
# Strategy 7: twitter:image
tw_image = soup.find("meta", attrs={"name": "twitter:image"})
if tw_image and tw_image.get("content"):
_add_image(tw_image["content"], source_tag="twitter:image")
# Strategy 8: with
for figure in soup.find_all("figure"):
img = figure.find("img")
if img:
src = img.get("data-src") or img.get("src")
figcaption = figure.find("figcaption")
alt = figcaption.get_text(strip=True) if figcaption else img.get("alt", "")
_add_image(src, alt=alt, source_tag="figure")
# Strategy 9: tags linking to images
for a in soup.find_all("a", href=True):
href = a["href"]
if any(href.lower().endswith(ext) for ext in [".jpg", ".jpeg", ".png", ".webp", ".gif"]):
_add_image(href, alt=a.get_text(strip=True)[:80], source_tag="link")
return images
# ===== JINA READER =====
def _reader_url(target_url: str) -> str:
safe = quote(target_url, safe=":/?#[]@!$&'()*+,;=%")
return "https://r.jina.ai/http://" + safe
def jina_reader_markdown(url: str) -> str:
jr = _reader_url(url)
r = requests.get(jr, headers={"Accept": "text/markdown,text/plain,*/*", "X-Return-Format": "markdown", "User-Agent": "Mozilla/5.0"}, timeout=35)
r.raise_for_status()
return r.text or ""
def _parse_jina_markdown(md: str, url: str):
lines = [x.rstrip() for x in (md or "").splitlines()]
title = ""; first_image = ""; all_images = []; content_lines = []; in_content = False
for ln in lines:
if ln.startswith("Title:") and not title:
title = _clean_text(ln.replace("Title:", "", 1)); continue
if ln.startswith("URL Source:"):
continue
if ln.startswith("Markdown Content:"):
in_content = True; continue
# Extract ALL images from markdown 
for mimg in re.finditer(r'!\[[^\]]*\]\((https?://[^)]+)\)', ln):
img_url = mimg.group(1)
if img_url not in all_images:
all_images.append(img_url)
if not first_image:
first_image = img_url
if in_content or (title and not ln.startswith("Title:")):
if ln.strip():
content_lines.append(ln)
text = "\n".join(content_lines)
text = re.sub(r'!\[[^\]]*\]\([^)]+\)', '', text)
paras = []
for part in re.split(r'\n{2,}|\n(?=#{1,3}\s)', text):
t = _clean_text(re.sub(r'^#{1,6}\s*', '', part))
if len(t) >= 40:
paras.append(t)
if len(paras) >= 35:
break
if not title and paras:
title = paras[0][:90]
return {"url": url, "title": title or url, "summary": paras[0] if paras else "",
"text": "\n".join(paras), "image": first_image,
"images": all_images, "via": "jina"}
# ===== WEB SCRAPE (with full image extraction) =====
def _best_content_block(soup):
best, best_score = None, 0
for el in soup.find_all(["article", "main", "section", "div"]):
ps = el.find_all("p")
txt = " ".join(p.get_text(" ", strip=True) for p in ps)
score = len(ps) * 100 + len(txt)
cls = " ".join(el.get("class", []))
if any(k in cls.lower() for k in ["content", "article", "detail", "body", "post", "entry"]):
score += 800
if score > best_score:
best, best_score = el, score
return best
def scrape_any_url_direct(url: str):
r = requests.get(url, headers=HEADERS, timeout=18)
if r.status_code in {401, 403, 406, 409, 429, 451, 503}:
raise RuntimeError(f"blocked status {r.status_code}")
r.encoding = "utf-8"
soup = BeautifulSoup(r.text, "lxml")
for tag in soup.find_all(["script", "style", "nav", "footer", "aside", "form", "noscript"]):
tag.decompose()
# Title
title = soup.find("h1").get_text(" ", strip=True) if soup.find("h1") else ""
if not title:
ogt = soup.find("meta", property="og:title") or soup.find("meta", attrs={"name": "title"})
title = ogt.get("content", "") if ogt else (soup.title.get_text(strip=True) if soup.title else "")
# Summary
desc_tag = soup.find("meta", property="og:description") or soup.find("meta", attrs={"name": "description"})
summary = desc_tag.get("content", "") if desc_tag else ""
# Featured image (og:image)
img_tag = soup.find("meta", property="og:image") or soup.find("meta", attrs={"name": "twitter:image"})
image = img_tag.get("content", "") if img_tag else ""
if image and image.startswith("//"):
image = "https:" + image
# Extract ALL images from the article
all_images = _extract_all_images(soup, url)
image_urls = [img["url"] for img in all_images]
# Ensure featured image is first
if image and image not in image_urls:
image_urls.insert(0, image)
elif image in image_urls:
image_urls.remove(image)
image_urls.insert(0, image)
# Content paragraphs
block = _best_content_block(soup) or soup
paras, seen_p = [], set()
for p in block.find_all("p"):
t = _clean_text(p.get_text(" ", strip=True))
if len(t) >= 40 and t not in seen_p:
seen_p.add(t)
paras.append(t)
if len(paras) >= 35:
break
if not title and paras:
title = paras[0][:90]
return {
"url": url, "title": title or url, "summary": paras[0] if paras else "",
"text": "\n".join(paras), "image": image_urls[0] if image_urls else "",
"images": image_urls, "via": _domain(url)
}
def scrape_any_url(url: str):
"""Try direct scrape first, fall back to Jina Reader."""
data = scrape_any_url_direct(url)
raw_text = (data.get("summary", "") + "\n" + data.get("text", "")).strip()
if len(raw_text) >= 120:
return data
try:
md = jina_reader_markdown(url)
if md:
jr = _parse_jina_markdown(md, url)
if jr.get("text"):
if data.get("title") and data["title"] != url:
jr["title"] = data["title"]
if data.get("image"):
jr["image"] = data["image"]
if data.get("images"):
jr["images"] = data["images"]
jr["via"] = data.get("via", _domain(url)) + " + jina"
return jr
except Exception:
pass
return data
# ===== POLLINATIONS IMAGE =====
def pollinations_image_url(topic: str) -> str:
prompt = "editorial illustration, Vietnamese news, " + topic
return "https://image.pollinations.ai/prompt/" + quote(prompt, safe="") + "?width=1024&height=576&nologo=true"
@app.get("/api/ai/probe")
async def api_ai_probe():
"""Diagnostic: test which chat models actually work on this token + their latency."""
import time as _t
tok = _hf_token()
out = []
extra = os.getenv("PROBE_MODELS", "").split(",")
cand = [m.strip() for m in extra if m.strip()] + [
"Qwen/Qwen2.5-VL-7B-Instruct", "Qwen/Qwen2.5-VL-3B-Instruct", "Qwen/Qwen2-VL-7B-Instruct",
"Qwen/Qwen2.5-VL-72B-Instruct", "Qwen/Qwen3-8B", "Qwen/Qwen3-4B", "Qwen/Qwen3-32B",
"Qwen/Qwen2.5-7B-Instruct-1M", "meta-llama/Llama-3.2-3B-Instruct"]
seen = set()
for m in cand:
if not m or m in seen:
continue
seen.add(m)
t0 = _t.time()
try:
c = AsyncInferenceClient(provider="auto", api_key=tok, timeout=40)
r = await c.chat_completion(model=m, messages=[{"role": "user", "content": "Trả lời đúng 1 từ: xin chào"}], max_tokens=10)
out.append({"model": m, "ok": True, "sec": round(_t.time() - t0, 1), "txt": (r.choices[0].message.content or "")[:30]})
except Exception as e:
out.append({"model": m, "ok": False, "sec": round(_t.time() - t0, 1), "err": (type(e).__name__ + ": " + str(e))[-300:]})
return JSONResponse({"results": out})
# ===== QWEN AI (strict, concise) =====
async def qwen_generate(prompt: str, image_url: Optional[str] = None, max_tokens: int = 500, image_urls: Optional[List[str]] = None):
global LAST_QWEN_ERROR, HF_TOKEN
HF_TOKEN = _hf_token()
if not HF_TOKEN:
LAST_QWEN_ERROR = "Không tìm thấy token"
return None
if not AsyncInferenceClient:
LAST_QWEN_ERROR = "Thiếu huggingface_hub"
return None
errors = []; models = []
has_images = bool(image_urls) or bool(image_url)
if has_images:
# Vision needed -> VL models
candidate = [QWEN_VL_MODEL, "Qwen/Qwen2.5-VL-7B-Instruct", "Qwen/Qwen2.5-VL-3B-Instruct"]
else:
# Text-only summary -> FAST text models first, VL only as last resort.
candidate = QWEN_TEXT_MODELS + [QWEN_VL_MODEL]
# Use the last-known-working model first to avoid wasting time on unavailable models.
global _WORKING_MODEL_TEXT, _WORKING_MODEL_VL
cached_ok = _WORKING_MODEL_VL if has_images else _WORKING_MODEL_TEXT
if cached_ok and cached_ok in candidate:
candidate = [cached_ok] + [m for m in candidate if m != cached_ok]
for m in candidate:
if m and m not in models:
models.append(m)
for model in models:
try:
client = AsyncInferenceClient(provider="auto", api_key=HF_TOKEN, timeout=60)
content = []
# Collect all images: image_urls list takes priority, fall back to single image_url
all_img_urls = []
if image_urls:
all_img_urls = image_urls[:6] # max 6 images to avoid context overflow
elif image_url:
all_img_urls = [image_url]
for img_u in all_img_urls:
if img_u and img_u.startswith("http"):
content.append({"type": "image_url", "image_url": {"url": img_u}})
content.append({"type": "text", "text": prompt})
messages = [
{"role": "system", "content": (
"Bạn là biên tập viên báo điện tử tiếng Việt. "
"NHIỆM VỤ: Chỉ TÓM TẮT nội dung, KHÔNG viết lại bài đầy đủ. "
"QUY TẮC CỨNG: "
"(1) KHÔNG lặp lại bất kỳ nội dung nào — mỗi ý chỉ xuất hiện ĐÚNG 1 LẦN. "
"(2) Nếu 2 câu diễn đạt cùng 1 ý → bỏ cây thứ 2. "
"(3) KHÔNG dùng Markdown (##, **, ---, *). "
"(4) KHÔNG viết 'Dưới đây là', 'Tôi sẽ', 'Theo yêu cầu', 'Nhiệm vụ', 'Vai trò', 'Đây là bài tóm tắt'. "
"(5) KHÔNG bịa thông tin ngoài nguồn. "
"(6) Chỉ viết ĐOẠN VĂN THUẦN, không bullet points. "
"(7) Tối đa 200 từ. Ngắn gọn, súc tích."
)},
{"role": "user", "content": content}
]
resp = await client.chat_completion(model=model, messages=messages, max_tokens=max_tokens, temperature=0.3, top_p=0.8)
txt = (resp.choices[0].message.content or "").strip()
if txt:
LAST_QWEN_ERROR = ""
if has_images: _WORKING_MODEL_VL = model
else: _WORKING_MODEL_TEXT = model
return txt
except Exception as e:
errors.append(f"{model}: {type(e).__name__}: {str(e)[:220]}")
LAST_QWEN_ERROR = " | ".join(errors) or "Qwen không trả nội dung."
print("[qwen errors]", LAST_QWEN_ERROR)
return None
# ===== TTS GENERATION =====
async def _generate_tts_edge(text: str, voice_id: str, speed: float, out_path: str, emotion: str = None):
"""Generate TTS using edge-tts (or gTTS if engine=gtts) with voice, speed & emotion control.
Emotion (cảm xúc) is simulated via pitch + rate + volume (edge-tts has no express-as).
"""
vcfg = TTS_VOICES.get(voice_id, TTS_VOICES[TTS_DEFAULT_VOICE])
# gTTS engine (no voice/speed/emotion control)
if vcfg.get("engine") == "gtts":
_generate_tts_gtts(text, out_path)
return
if edge_tts is None:
raise RuntimeError("edge-tts chưa cài đặt")
voice = vcfg["id"]
emo = EMOTION_PRESETS.get(emotion or EMOTION_DEFAULT, EMOTION_PRESETS[EMOTION_DEFAULT])
# Apply emotion rate multiplier on top of base speed
eff_speed = speed * emo.get("rate_mul", 1.0)
pct = int(round((eff_speed - 1.0) * 100))
rate = f"+{pct}%" if pct >= 0 else f"{pct}%"
pitch = emo.get("pitch", "+0Hz")
volume = emo.get("volume", "+0%")
communicate = edge_tts.Communicate(text, voice, rate=rate, pitch=pitch, volume=volume)
await communicate.save(out_path)
def _generate_tts_gtts(text: str, out_path: str):
"""Fallback TTS using gTTS (no voice/speed control)."""
if gTTS is None:
raise RuntimeError("gTTS chưa cài đặt")
gTTS(text, lang="vi").save(out_path)
# ===== SHORT VIDEO GENERATION (multi-segment: each key point with its own image) =====
def _download_image(url, fallback_topic, out_path):
"""Download an image (un-proxying our own /api/proxy/img). Falls back to generated image."""
if url:
u = url
m = re.search(r'/api/proxy/img\?url=(.+)$', u)
if m:
from urllib.parse import unquote
u = unquote(m.group(1))
try:
r = requests.get(u, headers={**HEADERS, "Referer": "https://dantri.com.vn/"}, timeout=15)
if r.status_code == 200 and len(r.content) > 1000:
with open(out_path, "wb") as f:
f.write(r.content)
if Image:
Image.open(out_path).verify()
return out_path
except Exception:
pass
gen = pollinations_image_url(fallback_topic)
try:
r = requests.get(gen, headers=HEADERS, timeout=25)
if r.status_code == 200 and len(r.content) > 1000:
with open(out_path, "wb") as f:
f.write(r.content)
return out_path
except Exception:
pass
if Image:
Image.new("RGB", (1080, 980), (30, 55, 42)).save(out_path)
return out_path
raise RuntimeError("Không tạo được ảnh")
def _split_keypoint_sentences(text, max_points=6):
"""Split summary text into key points: prefer bullet markers, else sentences."""
text = _clean_text(text)
parts = re.split(r'\s*•\s*', text)
good = [p.strip() for p in parts if len(p.strip()) > 20]
if len(good) >= 2:
# Explicit bullet points: keep each one as-is (never merge).
return good[:max_points]
# Fallback: split into sentences and merge orphan short fragments.
pts = [p.strip() for p in re.split(r'(?<=[.!?])\s+', text) if len(p.strip()) > 20]
out = []
for p in pts:
if out and len(p) < 40:
out[-1] = (out[-1] + " " + p).strip()
else:
out.append(p)
return out[:max_points] if out else ([text] if text else [])
def _build_keypoints(post, max_points=6):
"""Return [{text, image}] pairing each key point with its own image."""
slides = post.get("slides") or []
images = post.get("images") or ([post.get("img")] if post.get("img") else [])
images = [i for i in images if i]
if slides:
kps = []
for i, s in enumerate(slides[:max_points]):
t = _clean_text(s.get("text", ""))
img = s.get("image") or (images[i] if i < len(images) else (images[-1] if images else ""))
if t:
kps.append({"text": t, "image": img})
if kps:
return kps
points = _split_keypoint_sentences(post.get("text", ""), max_points)
kps = []
for i, t in enumerate(points):
img = images[i] if i < len(images) else (images[-1] if images else "")
kps.append({"text": t, "image": img})
if not kps:
kps = [{"text": _clean_text(post.get("title", "")) or "VNEWS", "image": images[0] if images else ""}]
return kps
def _wrap_text(draw, text, font, max_w):
words = text.split()
lines, cur = [], ""
for w in words:
test = (cur + " " + w).strip()
if draw.textlength(test, font=font) <= max_w:
cur = test
else:
if cur:
lines.append(cur)
cur = w
if cur:
lines.append(cur)
return lines
def _make_segment_frame(title, point_text, img_path, idx, total, out_path):
"""Render a 1080x1920 vertical frame: image on top, key point text below."""
if Image is None:
raise RuntimeError("Pillow chưa sẵn sàng")
W, H = 1080, 1920
IMG_H = 980
bg = Image.new("RGB", (W, H), (12, 14, 18))
try:
im = Image.open(img_path).convert("RGB")
tr = W / IMG_H
ir = im.width / im.height
if ir > tr:
nh = IMG_H; nw = int(nh * ir)
else:
nw = W; nh = int(nw / ir)
im = im.resize((nw, nh))
left = (nw - W) // 2; top = (nh - IMG_H) // 2
im = im.crop((left, top, left + W, top + IMG_H))
bg.paste(im, (0, 0))
except Exception:
pass
draw = ImageDraw.Draw(bg)
draw.rectangle((0, IMG_H, W, H), fill=(12, 14, 18))
try:
f_label = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 34)
f_title = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 46)
f_point = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 52)
except Exception:
f_label = f_title = f_point = ImageFont.load_default()
draw.text((54, IMG_H + 24), "VNEWS · Tường AI", fill=(92, 184, 122), font=f_label)
cnt = f"{idx + 1}/{total}"
draw.text((W - 54 - draw.textlength(cnt, font=f_label), IMG_H + 24), cnt, fill=(240, 192, 64), font=f_label)
y = IMG_H + 86
for ln in _wrap_text(draw, _clean_text(title), f_title, W - 108)[:2]:
draw.text((54, y), ln, fill=(255, 255, 255), font=f_title)
y += 56
y += 16
for ln in _wrap_text(draw, _clean_text(point_text), f_point, W - 108)[:11]:
draw.text((54, y), ln, fill=(225, 230, 235), font=f_point)
y += 64
bg.save(out_path, quality=92)
return out_path
def _ffmpeg_bin():
return os.environ.get("FFMPEG_BIN", "ffmpeg")
def _audio_duration(path):
try:
out = subprocess.run([_ffmpeg_bin(), "-i", path], capture_output=True, text=True, timeout=30).stderr
m = re.search(r"Duration:\s*(\d+):(\d+):(\d+\.\d+)", out)
if m:
h, mi, s = m.groups()
return int(h) * 3600 + int(mi) * 60 + float(s)
except Exception:
pass
return 0.0
async def _generate_short_video(post, post_id: str, voice_id: str = None, speed: float = None, emotion: str = None) -> str:
"""Generate a multi-segment MP4 short: each key point shown with its OWN image + narration."""
try:
os.makedirs(SHORTS_DIR, exist_ok=True)
out_mp4 = os.path.join(SHORTS_DIR, _safe_name(post_id) + ".mp4")
if os.path.exists(out_mp4) and voice_id is None and speed is None and emotion is None:
return "/api/ai/short-file/" + post_id
work = os.path.join(SHORTS_DIR, _safe_name(post_id) + "_work")
os.makedirs(work, exist_ok=True)
title = _clean_text(post.get("title", "")) or "VNEWS"
kps = _build_keypoints(post)
# Resolve voice + emotion (auto from topic, or from post, or explicit args)
auto_voice, auto_emotion = _detect_voice_emotion(post.get("title", ""), post.get("text", ""))
if voice_id is None:
voice_id = post.get("voice") or auto_voice
if emotion is None:
emotion = post.get("emotion") or auto_emotion
if speed is None:
speed = TTS_DEFAULT_SPEED
vcfg = TTS_VOICES.get(voice_id, TTS_VOICES[TTS_DEFAULT_VOICE])
seg_files = []
ff = _ffmpeg_bin()
for i, kp in enumerate(kps):
img_path = os.path.join(work, f"img{i}.jpg")
frame_path = os.path.join(work, f"frame{i}.jpg")
audio_path = os.path.join(work, f"voice{i}.mp3")
seg_mp4 = os.path.join(work, f"seg{i}.mp4")
_download_image(kp.get("image", ""), title, img_path)
_make_segment_frame(title, kp["text"], img_path, i, len(kps), frame_path)
narration = (title + ". " + kp["text"]) if i == 0 else kp["text"]
try:
await _generate_tts_edge(narration, voice_id, speed, audio_path, emotion=emotion)
except Exception as e:
print(f"[TTS edge-tts error] {e}, falling back to gTTS")
if gTTS:
_generate_tts_gtts(narration, audio_path)
else:
return ""
dur = _audio_duration(audio_path)
if dur < 1.0:
dur = 2.0
cmd = [ff, "-y", "-loop", "1", "-i", frame_path, "-i", audio_path,
"-c:v", "libx264", "-tune", "stillimage", "-pix_fmt", "yuv420p",
"-t", f"{dur + 0.4:.2f}", "-c:a", "aac", "-b:a", "128k", "-ar", "44100",
"-vf", "scale=1080:1920", "-r", "25", seg_mp4]
subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=180)
seg_files.append(seg_mp4)
if not seg_files:
return ""
if len(seg_files) == 1:
os.replace(seg_files[0], out_mp4)
return "/api/ai/short-file/" + post_id
listfile = os.path.join(work, "concat.txt")
with open(listfile, "w", encoding="utf-8") as f:
f.write("\n".join(f"file '{s}'" for s in seg_files))
cmd = [ff, "-y", "-f", "concat", "-safe", "0", "-i", listfile,
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", out_mp4]
subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=300)
return "/api/ai/short-file/" + post_id
except Exception as e:
print(f"[short video error] {e}")
return ""
import threading as _threading
def _spawn_background_video(post):
"""Generate the short video in a BACKGROUND thread so the rewrite/topic endpoint
can return immediately. When done, persist the video URL onto the wall post.
This is the main fix for 'rewrite tu cac nguon tin qua lau' — the user gets the
text post instantly; the video appears shortly after (or via the 'Tao Video' button)."""
pid = post.get("id")
if not pid:
return
def _run():
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
video_url = loop.run_until_complete(_generate_short_video(post, pid))
loop.close()
if video_url:
posts = _load_wall()
for i, p in enumerate(posts):
if str(p.get("id")) == str(pid):
posts[i]["video"] = video_url
break
_save_wall(posts)
print(f"[bg-video] done {pid} -> {video_url}")
except Exception as e:
print(f"[bg-video] error {pid}: {e}")
_threading.Thread(target=_run, daemon=True).start()
# ===== MAKE POST =====
def make_post(title, text, image, source_url, kind, sources=None, images=None, voice=None, emotion=None):
# Auto-pick voice + emotion from the article topic when not provided
auto_voice, auto_emotion = _detect_voice_emotion(title or "", text or "")
return {
"id": str(int(time.time() * 1000)) + str(random.randint(100, 999)),
"title": title, "text": text, "img": image, "url": source_url,
"kind": kind, "sources": sources or [], "video": "",
"images": images or [], "ts": int(time.time()),
"voice": voice or auto_voice,
"emotion": emotion or auto_emotion,
}
# ===== SHARED PROMPT BUILDER =====
def _build_rewrite_prompt(title: str, raw: str, images: List[str] = None) -> str:
image_info = ""
if images:
num = len(images)
if num == 1:
image_info = "\n\nBài viết có 1 ảnh minh họa. Hãy tham khảo ảnh để hiểu ngữ cảnh (nếu phù hợp)."
else:
image_info = f"\n\nBài viết có {num} ảnh minh họa. Hãy tham khảo tất cả ảnh để hiểu ngữ cảnh và bổ sung thông tin cho bài viết (nếu phù hợp)."
return f"""Tóm tắt bài viết sau thành bài TÓM TẮT đăng Tường AI.
QUY TẮC BẮT BUỘC:
1. Chỉ viết TÓM TẮT các ý chính. KHÔNG sao chép nguyên văn từ bài gốc.
2. KHÔNG lặp lại bất kỳ nội dung nào. Mỗi thông tin chỉ xuất hiện ĐÚNG 1 LẦN.
3. Nếu 2 câu nói cùng 1 ý → chỉ giữ 1 câu, bỏ cây còn lại.
4. KHÔNG dùng Markdown (##, **, ---, *).
5. KHÔNG viết "Dưới đây là", "Tôi sẽ", "Theo yêu cầu", "Nhiệm vụ", "Vai trò", "Đây là bài tóm tắt".
6. Viết thành ĐOẠN VĂN THUẦN, mạch lạc, dễ đọc. Không dùng bullet points.
7. Giữ sự thật, KHÔNG bịa thông tin.
8. Tối đa 200 từ. Ngắn gọn, đủ ý.{image_info}
Tiêu đề gốc: {title}
Nội dung gốc:
{raw[:14000]}"""
def _build_topic_prompt(topic: str, ctx: str) -> str:
return f"""Viết bài TÓM TẮT NGẮN GỌN về chủ đề: "{topic}".
QUY TẮC BẮT BUỘC:
1. Chỉ viết TÓM TẮT các ý chính từ nguồn. KHÔNG sao chép nguyên văn.
2. KHÔNG lặp lại bất kỳ nội dung nào. Mỗi thông tin chỉ xuất hiện ĐÚNG 1 LẦN.
3. Nếu 2 câu nói cùng 1 ý → chỉ giữ 1 câu.
4. KHÔNG dùng Markdown (##, **, ---, *).
5. KHÔNG viết "Dưới đây là", "Tôi sẽ", "Theo yêu cầu", "Nhiệm vụ", "Vai trò".
6. Viết thành ĐOẠN VĂN THUẦN, mạch lạc. Không dùng bullet points.
7. Giữ sự thật, KHÔNG bịa.
8. Tối đa 200 từ. Ngắn gọn, đủ ý.
Nguồn thực tế:
{ctx[:12000]}"""
# ===== WRITE ENDPOINTS =====
@app.post("/api/rewrite_share")
async def api_rewrite_share(request: Request):
body = await request.json()
url = _clean_text(body.get("url", ""))
if not url.startswith("http"):
return JSONResponse({"error": "missing url"}, status_code=400)
try:
data = scrape_any_url(url)
except Exception as e:
return JSONResponse({"error": "Không đọc được bài viết: " + str(e)[:180]}, status_code=422)
raw = (data.get("summary", "") + "\n" + data.get("text", "")).strip()
if len(raw) < 60:
return JSONResponse({"error": "Bài viết quá ngắn để tóm tắt"}, status_code=422)
images = data.get("images", [])
prompt = _build_rewrite_prompt(data.get("title", ""), raw, images)
# Text-only summary for SPEED (images are kept on the post for display + short video).
text = await qwen_generate(prompt, max_tokens=500)
if not text:
return JSONResponse({"error": "Qwen2.5-VL chưa sẵn sàng: " + LAST_QWEN_ERROR}, status_code=503)
text = _clean_ai_output(text)
post = make_post(data.get("title") or "Bài viết", text,
images[0] if images else data.get("image", ""),
url, "rewrite", images=images)
# Save post and return IMMEDIATELY; generate the short video in the background
# (so rewrite is fast). Video appears on the wall when ready / via 'Tao Video' button.
posts = _load_wall()
posts.insert(0, post)
_save_wall(posts)
_spawn_background_video(post)
return JSONResponse({"post": post})
@app.post("/api/url_wall")
async def api_url_wall(request: Request):
body = await request.json()
url = _clean_text(body.get("url", ""))
if not url.startswith("http"):
return JSONResponse({"error": "missing url"}, status_code=400)
try:
data = scrape_any_url(url)
except Exception as e:
return JSONResponse({"error": "Không scrape được URL: " + str(e)[:180]}, status_code=422)
raw = (data.get("summary", "") + "\n" + data.get("text", "")).strip()
if len(raw) < 60:
return JSONResponse({"error": "URL không có đủ nội dung"}, status_code=422)
images = data.get("images", [])
prompt = _build_rewrite_prompt(data.get("title", ""), raw, images)
# Text-only summary for SPEED (images are kept on the post for display + short video).
text = await qwen_generate(prompt, max_tokens=500)
if not text:
return JSONResponse({"error": "Qwen2.5-VL chưa sẵn sàng: " + LAST_QWEN_ERROR}, status_code=503)
text = _clean_ai_output(text)
post = make_post(data.get("title") or "Bài viết", text,
images[0] if images else data.get("image", ""),
url, "url", images=images)
posts = _load_wall()
posts.insert(0, post)
_save_wall(posts)
_spawn_background_video(post)
return JSONResponse({"post": post})
@app.post("/api/topic_post")
async def api_topic_post(request: Request):
body = await request.json()
topic = _clean_text(body.get("topic", ""))
if not topic:
return JSONResponse({"error": "missing topic"}, status_code=400)
ctx = _web_context(topic)
if not ctx:
return JSONResponse({"error": "Không lấy được dữ liệu cho chủ đề này"}, status_code=422)
image = pollinations_image_url(topic)
prompt = _build_topic_prompt(topic, ctx)
# NOTE: do NOT pass the decorative pollinations image to the VL model — feeding an
# image makes Qwen2.5-VL much slower (it must download+process it) with no benefit for
# a text summary. We keep the image only for display on the post. This is a major
# speed-up for 'rewrite tong hop'. (Text-only inference is several times faster.)
text = await qwen_generate(prompt, max_tokens=500)
if not text:
return JSONResponse({"error": "Qwen2.5-VL chưa sẵn sàng: " + LAST_QWEN_ERROR}, status_code=503)
text = _clean_ai_output(text)
post = make_post(topic, text, image, "", "topic")
posts = _load_wall()
posts.insert(0, post)
_save_wall(posts)
_spawn_background_video(post)
return JSONResponse({"post": post})
# ===== WALL ENDPOINTS =====
@app.get("/api/ai_wall")
def api_ai_wall():
return JSONResponse({"posts": _load_wall()[:80]})
@app.get("/api/wall")
def api_wall():
return JSONResponse({"posts": _load_wall()[:80]})
# ===== SHORT VIDEO ENDPOINT (with voice + speed params) =====
@app.post("/api/ai/short/{post_id}")
async def api_ai_short(post_id: str, voice: str = Query(default=None), speed: float = Query(default=None), emotion: str = Query(default=None)):
"""Generate (or retrieve cached) short video for a wall post.
Query params:
- voice: 'hoaimy' (female) | 'namminh' (male) | auto-detect if not specified
- speed: float (default 1.2), e.g. 1.0=normal, 1.2=fast, 0.8=slow
- emotion: 'vui'|'hao_hung'|'nghiem'|'tram'|'buon'|'trung_tinh' | auto by topic
"""
posts = _load_wall()
post = next((p for p in posts if str(p.get("id")) == str(post_id)), None)
if not post:
return JSONResponse({"error": "post not found"}, status_code=404)
os.makedirs(SHORTS_DIR, exist_ok=True)
out_mp4 = os.path.join(SHORTS_DIR, _safe_name(post_id) + ".mp4")
# If cached and no custom voice/speed/emotion requested, return cached
if os.path.exists(out_mp4) and voice is None and speed is None and emotion is None:
video_url = "/api/ai/short-file/" + post_id
for i, p in enumerate(posts):
if str(p.get("id")) == str(post_id):
posts[i]["video"] = video_url
break
_save_wall(posts)
return JSONResponse({"video": video_url})
# Validate params
if voice is not None and voice not in TTS_VOICES:
return JSONResponse({"error": f"voice không hợp lệ. Chọn: {list(TTS_VOICES.keys())}"}, status_code=400)
if emotion is not None and emotion not in EMOTION_PRESETS:
return JSONResponse({"error": f"emotion không hợp lệ. Chọn: {list(EMOTION_PRESETS.keys())}"}, status_code=400)
video_url = await _generate_short_video(post, post_id, voice_id=voice, speed=speed, emotion=emotion)
if video_url:
for i, p in enumerate(posts):
if str(p.get("id")) == str(post_id):
posts[i]["video"] = video_url
if voice:
posts[i]["voice"] = voice
if emotion:
posts[i]["emotion"] = emotion
break
_save_wall(posts)
return JSONResponse({"video": video_url})
return JSONResponse({"error": "Không tạo được shorts"}, status_code=500)
@app.get("/api/ai/short-file/{post_id}")
def api_ai_short_file(post_id: str):
path = os.path.join(SHORTS_DIR, _safe_name(post_id) + ".mp4")
if not os.path.exists(path):
return JSONResponse({"error": "not found"}, status_code=404)
return FileResponse(path, media_type="video/mp4", filename=f"vnews-ai-{post_id}.mp4")
@app.get("/api/ai/status")
def api_ai_status():
return JSONResponse({
"has_token": bool(_hf_token()),
"client_imported": AsyncInferenceClient is not None,
"last_error": LAST_QWEN_ERROR,
"working_text_model": _WORKING_MODEL_TEXT,
"working_vl_model": _WORKING_MODEL_VL,
})