import os import io import json import numpy as np import cv2 from datetime import date from flask import Flask, render_template, request, jsonify, send_file, redirect, url_for, session from werkzeug.security import generate_password_hash, check_password_hash from deep_translator import GoogleTranslator from google import genai from google.genai import types from docx import Document from PyPDF2 import PdfReader from reportlab.pdfgen import canvas from reportlab.lib.pagesizes import letter from werkzeug.utils import secure_filename from rembg import remove as rembg_remove import socket socket.setdefaulttimeout(15.0) app = Flask(__name__) app.secret_key = 'pums-toolkit-super-secret-key-12345' app.config['UPLOAD_FOLDER'] = 'uploads' app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "AQ.Ab8RN6JxMSl1LlebmW8kq6jjam99Z0qKf2wkwu8WjFKVEp6kGQ") USAGE_FILE = os.path.join(app.config['UPLOAD_FOLDER'], 'gemini_usage.json') GEMINI_MODELS = [ {"id": "gemini-2.5-flash", "name": "Gemini 2.5 Flash", "rpd": 500}, {"id": "gemini-2.5-flash-lite", "name": "Gemini 2.5 Flash Lite", "rpd": 1000}, {"id": "gemini-2.0-flash", "name": "Gemini 2.0 Flash", "rpd": 200}, {"id": "gemini-2.0-flash-lite", "name": "Gemini 2.0 Flash Lite", "rpd": 1000}, {"id": "gemini-1.5-flash", "name": "Gemini 1.5 Flash", "rpd": 1500}, {"id": "gemini-1.5-flash-8b", "name": "Gemini 1.5 Flash 8B", "rpd": 1500}, {"id": "gemini-1.5-pro", "name": "Gemini 1.5 Pro", "rpd": 50}, ] DEFAULT_MODEL = "gemini-2.5-flash" # ── Usage tracking ──────────────────────────────────────────── def load_usage(): if not os.path.exists(USAGE_FILE): return {} with open(USAGE_FILE, 'r') as f: data = json.load(f) if data.get('date') != str(date.today()): return {} return data.get('usage', {}) def save_usage(usage_dict): with open(USAGE_FILE, 'w') as f: json.dump({'date': str(date.today()), 'usage': usage_dict}, f) def increment_usage(model_id): usage = load_usage() usage[model_id] = usage.get(model_id, 0) + 1 save_usage(usage) def get_available_models(): """Trả về danh sách model còn quota, ưu tiên model tốt nhất trước.""" usage = load_usage() available = [] for m in GEMINI_MODELS: used = usage.get(m['id'], 0) if used < m['rpd']: available.append(m['id']) return available def generate_with_fallback(preferred_model_id, contents, config=None): """ Gọi Gemini API với auto-fallback: - Thử model được chọn trước - Nếu lỗi 429/quota, tự động thử model tiếp theo còn quota - Trả về (response, model_id_used) hoặc raise Exception nếu tất cả hết quota """ # Xây dựng thứ tự thử: model được chọn trước, rồi các model còn lại available = get_available_models() # Đưa preferred lên đầu nếu còn quota ordered = ([preferred_model_id] if preferred_model_id in available else []) ordered += [m for m in available if m != preferred_model_id] if not ordered: raise Exception("Tất cả model Gemini đã hết quota hôm nay. Vui lòng thử lại vào ngày mai.") last_err = None for model_id in ordered: try: kwargs = {'model': model_id, 'contents': contents} if config: kwargs['config'] = config resp = client.models.generate_content(**kwargs) increment_usage(model_id) return resp, model_id except Exception as e: err_str = str(e) if '429' in err_str or 'quota' in err_str.lower() or 'RESOURCE_EXHAUSTED' in err_str: last_err = e continue # thử model tiếp theo raise # lỗi khác → throw ngay raise Exception(f"Tất cả model Gemini đã hết quota. Lỗi cuối: {_err_msg(str(last_err))}") try: client = genai.Client(api_key=GEMINI_API_KEY.strip()) except Exception as e: print(f"Lỗi cấu hình Gemini: {e}") # ── Translation helpers ─────────────────────────────────────── def translate_text(text, target_lang, engine, style=None, gemini_model=None): if not text.strip(): return "" if engine == 'google': return GoogleTranslator(source='auto', target=target_lang).translate(text) elif engine == 'gemini': model_id = gemini_model if gemini_model else DEFAULT_MODEL style_prompt = f" in a {style} style" if style and style != 'default' else "" prompt = (f"Translate the following text to the language with code '{target_lang}'{style_prompt}. " f"Maintain the original meaning and formatting. " f"Detect the source language automatically. " f"Return only the translated text:\n\n{text}") response = client.models.generate_content(model=model_id, contents=prompt) increment_usage(model_id) return response.text.strip() return text def process_file(filepath, filename, target_lang, engine, style, gemini_model=None): ext = filename.rsplit('.', 1)[1].lower() output_filename = f"translated_{filename}" output_path = os.path.join(app.config['UPLOAD_FOLDER'], output_filename) if ext == 'txt': with open(filepath, 'r', encoding='utf-8') as f: text = f.read() translated_text = translate_text(text, target_lang, engine, style, gemini_model) with open(output_path, 'w', encoding='utf-8') as f: f.write(translated_text) elif ext == 'docx': doc = Document(filepath) new_doc = Document() for para in doc.paragraphs: if para.text.strip(): new_doc.add_paragraph(translate_text(para.text, target_lang, engine, style, gemini_model)) else: new_doc.add_paragraph("") new_doc.save(output_path) elif ext == 'pdf': reader = PdfReader(filepath) text = "".join((page.extract_text() or '') + "\n" for page in reader.pages) translated_text = translate_text(text, target_lang, engine, style, gemini_model) c = canvas.Canvas(output_path, pagesize=letter) width, height = letter text_obj = c.beginText(40, height - 40) for line in translated_text.split('\n'): if text_obj.getY() < 40: c.drawText(text_obj) c.showPage() text_obj = c.beginText(40, height - 40) text_obj.textLine(line) c.drawText(text_obj) c.save() return output_filename def _err_msg(err): if '429' in err or 'quota' in err.lower(): return 'Gemini API đã hết quota. Vui lòng thử lại sau hoặc chuyển sang Google Translate.' if 'API_KEY' in err or 'api key' in err.lower(): return 'API key không hợp lệ. Vui lòng kiểm tra cấu hình.' return err # ── Routes: General ────────────────────────────────────────── @app.route('/') def index(): return render_template('index.html') @app.route('/download/') def download(filename): return send_file(os.path.join(app.config['UPLOAD_FOLDER'], filename), as_attachment=True) @app.route('/favicon.ico') def favicon(): return '', 204 # ── Routes: API ─────────────────────────────────────────────── @app.route('/api/gemini-models') def gemini_models(): usage = load_usage() result = [] for m in GEMINI_MODELS: used = usage.get(m['id'], 0) result.append({**m, 'used': used, 'remaining': max(0, m['rpd'] - used)}) return jsonify(result) @app.route('/api/gemini-quota') def gemini_quota(): return jsonify(load_usage()) @app.route('/api/languages') def languages(): langs = [ {"code": "auto", "name": "Tự động phát hiện", "source_only": True}, {"code": "af", "name": "Afrikaans"}, {"code": "sq", "name": "Albanian"}, {"code": "am", "name": "Amharic"}, {"code": "ar", "name": "Arabic (العربية)"}, {"code": "hy", "name": "Armenian"}, {"code": "az", "name": "Azerbaijani"}, {"code": "eu", "name": "Basque"}, {"code": "be", "name": "Belarusian"}, {"code": "bn", "name": "Bengali (বাংলা)"}, {"code": "bs", "name": "Bosnian"}, {"code": "bg", "name": "Bulgarian"}, {"code": "ca", "name": "Catalan"}, {"code": "ceb", "name": "Cebuano"}, {"code": "zh-CN", "name": "Chinese Simplified (中文简体)"}, {"code": "zh-TW", "name": "Chinese Traditional (中文繁體)"}, {"code": "co", "name": "Corsican"}, {"code": "hr", "name": "Croatian"}, {"code": "cs", "name": "Czech"}, {"code": "da", "name": "Danish"}, {"code": "nl", "name": "Dutch"}, {"code": "en", "name": "English"}, {"code": "eo", "name": "Esperanto"}, {"code": "et", "name": "Estonian"}, {"code": "fi", "name": "Finnish"}, {"code": "fr", "name": "French"}, {"code": "fy", "name": "Frisian"}, {"code": "gl", "name": "Galician"}, {"code": "ka", "name": "Georgian"}, {"code": "de", "name": "German"}, {"code": "el", "name": "Greek"}, {"code": "gu", "name": "Gujarati"}, {"code": "ht", "name": "Haitian Creole"}, {"code": "ha", "name": "Hausa"}, {"code": "haw", "name": "Hawaiian"}, {"code": "he", "name": "Hebrew"}, {"code": "hi", "name": "Hindi (हिन्दी)"}, {"code": "hmn", "name": "Hmong"}, {"code": "hu", "name": "Hungarian"}, {"code": "is", "name": "Icelandic"}, {"code": "ig", "name": "Igbo"}, {"code": "id", "name": "Indonesian"}, {"code": "ga", "name": "Irish"}, {"code": "it", "name": "Italian"}, {"code": "ja", "name": "Japanese (日本語)"}, {"code": "jv", "name": "Javanese"}, {"code": "kn", "name": "Kannada"}, {"code": "kk", "name": "Kazakh"}, {"code": "km", "name": "Khmer"}, {"code": "rw", "name": "Kinyarwanda"}, {"code": "ko", "name": "Korean (한국어)"}, {"code": "ku", "name": "Kurdish"}, {"code": "ky", "name": "Kyrgyz"}, {"code": "lo", "name": "Lao"}, {"code": "la", "name": "Latin"}, {"code": "lv", "name": "Latvian"}, {"code": "lt", "name": "Lithuanian"}, {"code": "lb", "name": "Luxembourgish"}, {"code": "mk", "name": "Macedonian"}, {"code": "mg", "name": "Malagasy"}, {"code": "ms", "name": "Malay"}, {"code": "ml", "name": "Malayalam"}, {"code": "mt", "name": "Maltese"}, {"code": "mi", "name": "Maori"}, {"code": "mr", "name": "Marathi"}, {"code": "mn", "name": "Mongolian"}, {"code": "my", "name": "Myanmar (Burmese)"}, {"code": "ne", "name": "Nepali"}, {"code": "no", "name": "Norwegian"}, {"code": "ny", "name": "Nyanja (Chichewa)"}, {"code": "or", "name": "Odia (Oriya)"}, {"code": "ps", "name": "Pashto"}, {"code": "fa", "name": "Persian (فارسی)"}, {"code": "pl", "name": "Polish"}, {"code": "pt", "name": "Portuguese"}, {"code": "pa", "name": "Punjabi"}, {"code": "ro", "name": "Romanian"}, {"code": "ru", "name": "Russian (Русский)"}, {"code": "sm", "name": "Samoan"}, {"code": "gd", "name": "Scots Gaelic"}, {"code": "sr", "name": "Serbian"}, {"code": "st", "name": "Sesotho"}, {"code": "sn", "name": "Shona"}, {"code": "sd", "name": "Sindhi"}, {"code": "si", "name": "Sinhala"}, {"code": "sk", "name": "Slovak"}, {"code": "sl", "name": "Slovenian"}, {"code": "so", "name": "Somali"}, {"code": "es", "name": "Spanish"}, {"code": "su", "name": "Sundanese"}, {"code": "sw", "name": "Swahili"}, {"code": "sv", "name": "Swedish"}, {"code": "tl", "name": "Tagalog (Filipino)"}, {"code": "tg", "name": "Tajik"}, {"code": "ta", "name": "Tamil"}, {"code": "tt", "name": "Tatar"}, {"code": "te", "name": "Telugu"}, {"code": "th", "name": "Thai (ไทย)"}, {"code": "tr", "name": "Turkish"}, {"code": "tk", "name": "Turkmen"}, {"code": "uk", "name": "Ukrainian"}, {"code": "ur", "name": "Urdu"}, {"code": "ug", "name": "Uyghur"}, {"code": "uz", "name": "Uzbek"}, {"code": "vi", "name": "Vietnamese (Tiếng Việt)"}, {"code": "cy", "name": "Welsh"}, {"code": "xh", "name": "Xhosa"}, {"code": "yi", "name": "Yiddish"}, {"code": "yo", "name": "Yoruba"}, {"code": "zu", "name": "Zulu"}, ] return jsonify(langs) # ── Routes: Translator ──────────────────────────────────────── @app.route('/translator') def translator_tool(): return redirect(url_for('translator_text')) @app.route('/translator/text') def translator_text(): return render_template('tools/translator/index.html', active_tab='text') @app.route('/translator/file') def translator_file(): return render_template('tools/translator/index.html', active_tab='file') @app.route('/translator/image') def translator_image(): return render_template('tools/translator/image.html') @app.route('/translate', methods=['POST']) def translate(): if 'file' in request.files and request.files['file'].filename != '': file = request.files['file'] target_lang = request.form.get('target_lang', 'vi') engine = request.form.get('engine', 'google') style = request.form.get('style', 'default') gemini_model = request.form.get('gemini_model', DEFAULT_MODEL) filename = secure_filename(file.filename) filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(filepath) try: output_file = process_file(filepath, filename, target_lang, engine, style, gemini_model) return jsonify({'download_url': f'/download/{output_file}'}) except Exception as e: return jsonify({'error': _err_msg(str(e))}), 500 data = request.json if not data or not data.get('text'): return jsonify({'error': 'No text provided'}), 400 try: translated = translate_text( data['text'], data.get('target_lang', 'vi'), data.get('engine', 'google'), data.get('style', 'default'), data.get('gemini_model', DEFAULT_MODEL) ) return jsonify({'translated_text': translated}) except Exception as e: return jsonify({'error': _err_msg(str(e))}), 500 @app.route('/preview-file', methods=['POST']) def preview_file(): if 'file' not in request.files or request.files['file'].filename == '': return jsonify({'error': 'Không có file'}), 400 file = request.files['file'] filename = secure_filename(file.filename) ext = filename.rsplit('.', 1)[-1].lower() preview_chars = 1000 try: if ext == 'txt': text = file.read().decode('utf-8', errors='ignore') elif ext == 'docx': doc = Document(file) text = '\n'.join(p.text for p in doc.paragraphs if p.text.strip()) elif ext == 'pdf': reader = PdfReader(file) text = '' for page in reader.pages: text += (page.extract_text() or '') + '\n' if len(text) >= preview_chars: break else: return jsonify({'error': 'Định dạng không hỗ trợ'}), 400 preview = text[:preview_chars] return jsonify({'preview': preview, 'truncated': len(text) > preview_chars, 'total_chars': len(text)}) except Exception as e: return jsonify({'error': str(e)}), 500 # ── Routes: Image Studio ────────────────────────────────────── @app.route('/image-studio') def image_studio(): return redirect(url_for('image_studio_remove_bg')) @app.route('/image-studio/remove-bg') def image_studio_remove_bg(): return render_template('tools/image_studio/index.html', active_tool='remove-bg') @app.route('/image-studio/upscale') def image_studio_upscale(): return render_template('tools/image_studio/index.html', active_tool='upscale') @app.route('/image/remove-bg', methods=['POST']) def image_remove_bg(): if 'image' not in request.files or request.files['image'].filename == '': return jsonify({'error': 'Không có ảnh'}), 400 file = request.files['image'] filename = secure_filename(file.filename) ext = filename.rsplit('.', 1)[-1].lower() if ext not in ['jpg', 'jpeg', 'png', 'webp']: return jsonify({'error': 'Định dạng không hỗ trợ. Dùng JPG, PNG hoặc WEBP.'}), 400 try: img_bytes = file.read() output_bytes = rembg_remove(img_bytes) out_filename = f"nobg_{os.path.splitext(filename)[0]}.png" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) with open(out_path, 'wb') as f: f.write(output_bytes) return jsonify({'result_url': f'/download/{out_filename}', 'filename': out_filename}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/image/upscale', methods=['POST']) def image_upscale(): if 'image' not in request.files or request.files['image'].filename == '': return jsonify({'error': 'Không có ảnh'}), 400 file = request.files['image'] scale = int(request.form.get('scale', 2)) if scale not in [2, 3, 4]: scale = 2 filename = secure_filename(file.filename) ext = filename.rsplit('.', 1)[-1].lower() if ext not in ['jpg', 'jpeg', 'png', 'webp']: return jsonify({'error': 'Định dạng không hỗ trợ. Dùng JPG, PNG hoặc WEBP.'}), 400 try: img_bytes = np.frombuffer(file.read(), np.uint8) img = cv2.imdecode(img_bytes, cv2.IMREAD_UNCHANGED) h, w = img.shape[:2] new_w, new_h = w * scale, h * scale upscaled = cv2.resize(img, (new_w, new_h), interpolation=cv2.INTER_LANCZOS4) out_filename = f"upscaled_{scale}x_{filename}" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) cv2.imwrite(out_path, upscaled) return jsonify({ 'result_url': f'/download/{out_filename}', 'filename': out_filename, 'original_size': f"{w}x{h}", 'new_size': f"{new_w}x{new_h}" }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/translate-image', methods=['POST']) def translate_image(): if 'image' not in request.files or request.files['image'].filename == '': return jsonify({'error': 'Không có ảnh'}), 400 file = request.files['image'] filename = secure_filename(file.filename) ext = filename.rsplit('.', 1)[-1].lower() if ext not in ['jpg', 'jpeg', 'png', 'webp', 'gif']: return jsonify({'error': 'Định dạng không hỗ trợ. Dùng JPG, PNG, WEBP hoặc GIF.'}), 400 target_lang = request.form.get('target_lang', 'vi') model_id = request.form.get('gemini_model', DEFAULT_MODEL) mode = request.form.get('mode', 'translate') # translate | ocr_only try: import base64 img_bytes = file.read() img_b64 = base64.b64encode(img_bytes).decode('utf-8') mime_map = {'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png', 'webp': 'image/webp', 'gif': 'image/gif'} mime_type = mime_map.get(ext, 'image/jpeg') from google.genai import types as genai_types if mode == 'ocr_only': prompt = "Hãy đọc và trích xuất toàn bộ văn bản trong ảnh này. Chỉ trả về văn bản, không giải thích thêm." else: prompt = (f"Hãy thực hiện 2 bước:\n" f"1. Đọc toàn bộ văn bản trong ảnh\n" f"2. Dịch văn bản đó sang ngôn ngữ có mã '{target_lang}'\n\n" f"Trả về theo định dạng:\n" f"**Văn bản gốc:**\n[nội dung đọc được]\n\n" f"**Bản dịch:**\n[bản dịch]") response = client.models.generate_content( model=model_id, contents=[ genai_types.Part.from_bytes(data=img_bytes, mime_type=mime_type), prompt ] ) increment_usage(model_id) return jsonify({'result': response.text.strip()}) except Exception as e: return jsonify({'error': _err_msg(str(e))}), 500 @app.route('/tts') def tts_index(): return redirect(url_for('tts_vieneu')) @app.route('/tts/vieneu') def tts_vieneu(): return render_template('tools/tts/vieneu.html') @app.route('/tts/gtts') def tts_gtts(): return render_template('tools/tts/gtts.html') @app.route('/tts/edge') def tts_edge(): return render_template('tools/tts/edge.html') @app.route('/api/tts-voices') def tts_voices(): try: from vieneu import Vieneu tts = Vieneu() voices = tts.list_preset_voices() return jsonify([{'label': label, 'id': vid} for label, vid in voices]) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/edge-voices') def edge_voices(): import asyncio, edge_tts try: voices = asyncio.run(edge_tts.list_voices()) result = [] for v in voices: result.append({ 'name': v['ShortName'], 'display': v.get('FriendlyName', v['ShortName']), 'locale': v['Locale'], 'gender': v['Gender'], 'lang': v['Locale'].split('-')[0] }) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tts/edge/generate', methods=['POST']) def tts_edge_generate(): import asyncio, edge_tts text = request.form.get('text', '').strip() voice = request.form.get('voice', 'en-US-JennyNeural') rate = request.form.get('rate', '+0%') pitch = request.form.get('pitch', '+0Hz') if not text: return jsonify({'error': 'Vui lòng nhập văn bản'}), 400 if len(text) > 10000: return jsonify({'error': 'Văn bản quá dài (tối đa 10.000 ký tự)'}), 400 out_filename = f"tts_edge_{os.urandom(4).hex()}.mp3" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) async def _generate(): communicate = edge_tts.Communicate(text, voice, rate=rate, pitch=pitch) await communicate.save(out_path) try: asyncio.run(_generate()) return jsonify({'audio_url': f'/download/{out_filename}', 'filename': out_filename}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/gtts-langs') def gtts_langs(): return jsonify([ {"code": "vi", "name": "Tiếng Việt"}, {"code": "en", "name": "Tiếng Anh"}, {"code": "en-us", "name": "Tiếng Anh (Mỹ)"}, {"code": "en-gb", "name": "Tiếng Anh (Anh)"}, {"code": "en-au", "name": "Tiếng Anh (Úc)"}, {"code": "ja", "name": "Tiếng Nhật"}, {"code": "ko", "name": "Tiếng Hàn"}, {"code": "zh-CN", "name": "Tiếng Trung (Giản thể)"}, {"code": "zh-TW", "name": "Tiếng Trung (Phồn thể)"}, {"code": "fr", "name": "Tiếng Pháp"}, {"code": "de", "name": "Tiếng Đức"}, {"code": "es", "name": "Tiếng Tây Ban Nha"}, {"code": "it", "name": "Tiếng Ý"}, {"code": "pt", "name": "Tiếng Bồ Đào Nha"}, {"code": "ru", "name": "Tiếng Nga"}, {"code": "ar", "name": "Tiếng Ả Rập"}, {"code": "hi", "name": "Tiếng Hindi"}, {"code": "th", "name": "Tiếng Thái"}, {"code": "id", "name": "Tiếng Indonesia"}, {"code": "ms", "name": "Tiếng Mã Lai"}, {"code": "nl", "name": "Tiếng Hà Lan"}, {"code": "pl", "name": "Tiếng Ba Lan"}, {"code": "sv", "name": "Tiếng Thụy Điển"}, {"code": "tr", "name": "Tiếng Thổ Nhĩ Kỳ"}, {"code": "uk", "name": "Tiếng Ukraina"}, ]) @app.route('/tts/generate', methods=['POST']) def tts_generate(): text = request.form.get('text', '').strip() voice_id = request.form.get('voice', 'Ngọc Lan') ref_audio = request.files.get('ref_audio') engine = request.form.get('engine', 'vieneu') # vieneu | gtts gtts_lang = request.form.get('gtts_lang', 'vi') gtts_slow = request.form.get('gtts_slow', 'false') == 'true' if not text: return jsonify({'error': 'Vui lòng nhập văn bản'}), 400 out_filename = f"tts_{os.urandom(4).hex()}.mp3" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) try: if engine == 'gtts': if len(text) > 5000: return jsonify({'error': 'Văn bản quá dài (tối đa 5000 ký tự với gTTS)'}), 400 from gtts import gTTS tts_obj = gTTS(text=text, lang=gtts_lang, slow=gtts_slow) tts_obj.save(out_path) return jsonify({'audio_url': f'/download/{out_filename}', 'filename': out_filename}) # VieNeu engine if len(text) > 3000: return jsonify({'error': 'Văn bản quá dài (tối đa 3000 ký tự với VieNeu)'}), 400 from vieneu import Vieneu tts_obj = Vieneu() out_filename = f"tts_{os.urandom(4).hex()}.wav" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) if ref_audio and ref_audio.filename: ref_filename = secure_filename(ref_audio.filename) ref_path = os.path.join(app.config['UPLOAD_FOLDER'], f"ref_{ref_filename}") ref_audio.save(ref_path) audio = tts_obj.infer(text=text, ref_audio=ref_path) else: audio = tts_obj.infer(text=text, voice=voice_id) tts_obj.save(audio, out_path) return jsonify({'audio_url': f'/download/{out_filename}', 'filename': out_filename}) except Exception as e: return jsonify({'error': str(e)}), 500 def extract_pdf_text(file): reader = PdfReader(file) return "".join((page.extract_text() or '') + "\n" for page in reader.pages).strip() @app.route('/pdf') def pdf_index(): return redirect(url_for('pdf_summarize')) @app.route('/pdf/summarize') def pdf_summarize(): return render_template('tools/smart_pdf/index.html', active_tool='summarize') @app.route('/pdf/extract') def pdf_extract(): return render_template('tools/smart_pdf/index.html', active_tool='extract') @app.route('/pdf/process', methods=['POST']) def pdf_process(): if 'file' not in request.files or request.files['file'].filename == '': return jsonify({'error': 'Không có file PDF'}), 400 file = request.files['file'] filename = secure_filename(file.filename) if not filename.lower().endswith('.pdf'): return jsonify({'error': 'Chỉ hỗ trợ file PDF'}), 400 tool = request.form.get('tool', 'summarize') model_id = request.form.get('gemini_model', DEFAULT_MODEL) lang = request.form.get('lang', 'vi') extract_type = request.form.get('extract_type', 'table') try: text = extract_pdf_text(file) if not text: return jsonify({'error': 'Không đọc được nội dung PDF. File có thể bị scan (ảnh).'}), 400 if tool == 'summarize': prompt = (f"Hãy tóm tắt tài liệu PDF sau bằng ngôn ngữ có mã '{lang}'. " f"Tóm tắt cần:\n1. Ý chính tổng quan (2-3 câu)\n2. Các điểm quan trọng (bullet points)\n" f"3. Kết luận/khuyến nghị (nếu có)\n\nNội dung tài liệu:\n{text[:15000]}") else: type_map = { 'table': 'bảng dữ liệu (tables) dưới dạng Markdown', 'key_value': 'các cặp thông tin quan trọng (key-value) như tên, ngày, số liệu', 'entity': 'các thực thể: tên người, tổ chức, địa điểm, ngày tháng, số tiền', 'custom': request.form.get('custom_prompt', 'thông tin quan trọng') } extract_desc = type_map.get(extract_type, type_map['key_value']) prompt = (f"Từ tài liệu PDF sau, hãy trích xuất {extract_desc}. " f"Trả lời bằng ngôn ngữ có mã '{lang}'. Định dạng output rõ ràng, dễ đọc.\n\n" f"Nội dung tài liệu:\n{text[:15000]}") response = client.models.generate_content(model=model_id, contents=prompt) increment_usage(model_id) return jsonify({'result': response.text.strip(), 'total_chars': len(text), 'truncated': len(text) > 15000}) except Exception as e: return jsonify({'error': _err_msg(str(e))}), 500 # ── Routes: Video AI ────────────────────────────────────────── import urllib.request import urllib.parse def get_music_file(music_type): if music_type == 'none' or not music_type: return None urls = { 'calm': 'https://www.soundhelix.com/examples/mp3/SoundHelix-Song-8.mp3', 'upbeat': 'https://www.soundhelix.com/examples/mp3/SoundHelix-Song-1.mp3', 'epic': 'https://www.soundhelix.com/examples/mp3/SoundHelix-Song-3.mp3', 'ambient': 'https://www.soundhelix.com/examples/mp3/SoundHelix-Song-4.mp3' } url = urls.get(music_type) if not url: return None music_dir = os.path.join(app.config['UPLOAD_FOLDER'], 'music') os.makedirs(music_dir, exist_ok=True) file_path = os.path.join(music_dir, f"{music_type}.mp3") if not os.path.exists(file_path): try: print(f"Downloading background music {music_type} from {url}...") req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=8) as resp: data = resp.read() with open(file_path, 'wb') as f: f.write(data) print("Download completed.") except Exception as e: print(f"Error downloading music: {e}") return None def draw_pil_subtitles(frame, text, font_size=28): from PIL import Image, ImageDraw, ImageFont import numpy as np pil_img = Image.fromarray(frame) draw = ImageDraw.Draw(pil_img) try: font = ImageFont.truetype("arial.ttf", font_size) except: try: font = ImageFont.truetype("C:\\Windows\\Fonts\\arial.ttf", font_size) except: font = ImageFont.load_default() w, h = pil_img.size words = text.split(' ') lines = [] current_line = "" for word in words: test_line = current_line + " " + word if current_line else word bbox = font.getbbox(test_line) tw = bbox[2] - bbox[0] if tw > w - 100: if current_line: lines.append(current_line) current_line = word else: current_line = test_line if current_line: lines.append(current_line) line_height = font.getbbox("A")[3] - font.getbbox("A")[1] + 10 y_start = h - 90 - (len(lines) - 1) * line_height for i, line in enumerate(lines): y = y_start + i * line_height bbox = font.getbbox(line) tw = bbox[2] - bbox[0] x = (w - tw) // 2 outline_color = (0, 0, 0) for dx, dy in [(-2,-2), (-2,2), (2,-2), (2,2), (-2,0), (2,0), (0,-2), (0,2)]: draw.text((x + dx, y + dy), line, font=font, fill=outline_color) draw.text((x, y), line, font=font, fill=(255, 255, 255)) return np.array(pil_img) def apply_zoom_effect(clip, zoom_ratio=0.07, mode='in'): import cv2 def effect(get_frame, t): frame = get_frame(t) h, w = frame.shape[:2] if mode == 'in': current_zoom = 1.0 + zoom_ratio * (t / clip.duration) else: current_zoom = 1.0 + zoom_ratio * (1.0 - t / clip.duration) new_w = int(w * current_zoom) new_h = int(h * current_zoom) resized_frame = cv2.resize(frame, (new_w, new_h)) dx = (new_w - w) // 2 dy = (new_h - h) // 2 cropped = resized_frame[dy:dy+h, dx:dx+w] return cropped return clip.transform(effect) def apply_subtitles_effect(clip, subtitle_text): def effect(get_frame, t): frame = get_frame(t) frame_copy = frame.copy() return draw_pil_subtitles(frame_copy, subtitle_text) return clip.transform(effect) def generate_gradient_background(w, h, text): import numpy as np import cv2 img = np.zeros((h, w, 3), dtype=np.uint8) for y in range(h): r = int(30 + (50 - 30) * (y / h)) g = int(20 + (30 - 20) * (y / h)) b = int(50 + (100 - 50) * (y / h)) img[y, :] = [b, g, r] font = cv2.FONT_HERSHEY_SIMPLEX font_scale = 1.0 thickness = 2 (text_w, text_h), _ = cv2.getTextSize(text, font, font_scale, thickness) x = (w - text_w) // 2 y = (h + text_h) // 2 cv2.putText(img, text, (x+2, y+2), font, font_scale, (0, 0, 0), thickness+2, cv2.LINE_AA) cv2.putText(img, text, (x, y), font, font_scale, (255, 255, 255), thickness, cv2.LINE_AA) return img def generate_hf_video(image_path, prompt, hf_token=None): import socket # Patch socket for httpx/gradio getaddrinfo issue orig_getaddrinfo = socket.getaddrinfo socket.getaddrinfo = lambda host, port, family=0, type=0, proto=0, flags=0: orig_getaddrinfo(host, port, socket.AF_INET, type, proto, flags) # Patch click.Choice class for Python 3.13 typing issue import click if not hasattr(click.Choice, '__class_getitem__'): click.Choice.__class_getitem__ = classmethod(lambda cls, item: cls) from gradio_client import Client, handle_file import shutil try: print(f"Connecting to zerogpu-aoti/wan2-2-fp8da-aoti-faster (HF Token provided: {bool(hf_token)})...") client = Client('zerogpu-aoti/wan2-2-fp8da-aoti-faster', token=hf_token if hf_token else None) print(f"Running video generation for prompt: {prompt}...") res = client.predict( handle_file(os.path.abspath(image_path)), prompt, 6, # steps "static, ugly, watermark, worst quality", # negative prompt 3.5, # duration_seconds 1.0, # guidance_scale 1.0, # guidance_scale_2 42, # seed True, # randomize_seed api_name="/generate_video" ) # res is a tuple: (temp_mp4_path, seed) temp_mp4 = res[0] out_filename = f"hf_video_{os.urandom(4).hex()}.mp4" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) shutil.copy(temp_mp4, out_path) print(f"HF Video generated successfully: {out_path}") return out_path except Exception as e: print(f"Error in generate_hf_video: {e}") return None REPLICATE_API_TOKEN = os.environ.get("REPLICATE_API_TOKEN", "r8_BTj77jONwojDdNOUUJw6TzA4E1gzajI2KKSJJ") # Đọc từ biến môi trường HF_API_TOKEN = os.environ.get("HF_API_TOKEN", "") # Đọc từ biến môi trường, tránh lộ token trên Git @app.route('/video-ai') def video_ai_index(): return render_template('tools/video_ai/index.html') @app.route('/api/video-ai/models') def video_ai_models(): return jsonify([ {"id": "hf-wan2.2-i2v-free", "name": "Wan 2.2 I2V (Free - Không cần Key)", "desc": "Tạo video AI chuyển động thật, hoàn toàn miễn phí"}, {"id": "wavespeedai/wan-2.1-t2v-480p", "name": "Wan 2.1 T2V 480p (Replicate)", "desc": "Text → Video, 480p, nhanh"}, {"id": "wavespeedai/wan-2.1-t2v-720p", "name": "Wan 2.1 T2V 720p (Replicate)", "desc": "Text → Video, 720p, sắc nét"}, {"id": "minimax/video-01", "name": "MiniMax Video-01 (Replicate)", "desc": "Text → Video, ngắn, nhanh"}, {"id": "stability-ai/stable-video-diffusion", "name": "Stable Video Diffusion (Replicate)", "desc": "Image → Video"}, ]) @app.route('/api/video-ai/generate', methods=['POST']) def api_video_ai_generate(): import asyncio, edge_tts, urllib.request, urllib.parse prompt = request.form.get('prompt', '').strip() gemini_model = request.form.get('gemini_model', DEFAULT_MODEL) lang = request.form.get('lang', 'vi') voice = request.form.get('voice', 'vi-VN-HoaiMyNeural') style = request.form.get('style', 'cinematic') duration = int(request.form.get('duration', '5')) music = request.form.get('music', 'none') ratio = request.form.get('ratio', '16:9') replicate_model = request.form.get('replicate_model', 'wavespeedai/wan-2.1-t2v-480p') api_token = request.form.get('api_token', REPLICATE_API_TOKEN).strip() hf_token = request.form.get('hf_token', '').strip() if not hf_token: hf_token = HF_API_TOKEN narrate = request.form.get('narrate', 'true') == 'true' if not prompt: return jsonify({'error': 'Vui lòng nhập chủ đề video'}), 400 result = {} try: # 1. Gemini viết kịch bản + enhance prompt if api_token: script_prompt = ( f"Viết kịch bản dẫn truyện bằng ngôn ngữ có mã '{lang}' cho video {duration} giây " f"về chủ đề: '{prompt}'. Ngắn gọn, súc tích vừa đủ {duration} giây đọc. " f"Chỉ trả về nội dung dẫn truyện." ) script_resp = client.models.generate_content(model=gemini_model, contents=script_prompt) script_text = script_resp.text.strip() video_prompt_req = ( f"Dịch và mở rộng prompt sau sang tiếng Anh để tạo video AI chất lượng cao: '{prompt}'. " f"Thêm các chi tiết về ánh sáng, góc quay, phong cách '{style}', màu sắc. " f"Chỉ trả về prompt tiếng Anh, không giải thích." ) vp_resp = client.models.generate_content(model=gemini_model, contents=video_prompt_req) video_prompt_en = vp_resp.text.strip() increment_usage(gemini_model) result['script'] = script_text result['video_prompt'] = video_prompt_en # 2. Tạo giọng đọc (Edge TTS) if narrate: narration_file = f"narration_{os.urandom(4).hex()}.mp3" narration_path = os.path.join(app.config['UPLOAD_FOLDER'], narration_file) asyncio.run(edge_tts.Communicate(script_text, voice).save(narration_path)) result['audio_url'] = f'/download/{narration_file}' import replicate os.environ['REPLICATE_API_TOKEN'] = api_token ratio_map = {'16:9': '1280x720', '9:16': '720x1280', '1:1': '720x720'} resolution = ratio_map.get(ratio, '1280x720') style_suffix = { 'cinematic': 'cinematic lighting, film grain, professional camera', 'anime': 'anime style, vibrant colors, Studio Ghibli', 'realistic': 'photorealistic, 8K, HDR', 'tech': 'cyberpunk, neon lights, futuristic', 'nature': 'nature documentary style, golden hour', 'space': 'space, cosmic, nebula, stars', 'watercolor': 'watercolor painting style, soft brush strokes', 'vintage': 'vintage film, retro aesthetic, 1970s' }.get(style, '') full_prompt = f"{video_prompt_en}. {style_suffix}" output = replicate.run( replicate_model, input={ "prompt": full_prompt, "num_frames": duration * 8, "duration": duration, } ) video_url = str(output) if isinstance(output, str) else (output[0] if isinstance(output, list) else str(output)) # Download và lưu local import urllib.request video_file = f"video_{os.urandom(4).hex()}.mp4" video_path = os.path.join(app.config['UPLOAD_FOLDER'], video_file) urllib.request.urlretrieve(video_url, video_path) result['video_url'] = f'/download/{video_file}' else: # FREE MOVIEPY & POLLINATIONS FLOW # Determine number of scenes num_scenes = 1 if duration >= 15: num_scenes = 3 elif duration >= 10: num_scenes = 2 # Try to query JSON structured script script_prompt = ( f"You are an AI video producer. Create a detailed, structured video script about the topic: '{prompt}'. " f"The video has a total target duration of {duration} seconds, so divide it into exactly {num_scenes} scenes.\n\n" f"You MUST respond ONLY with a valid JSON array of objects, containing exactly {num_scenes} objects. " f"Do not write any markdown code blocks, backticks, or text before/after the JSON.\n\n" f"Each scene object must contain:\n" f"- 'narration': A short spoken narration text in the language '{lang}'. Ensure the narration text is short and suitable for the scene duration (about {duration / num_scenes} seconds of speech).\n" f"- 'visual_prompt': An English description of the scene's visual content, formatted as a detailed image generation prompt with style '{style}', lighting, colors, and camera angle.\n\n" f"Example format:\n" f"[\n" f" {{\n" f" \"narration\": \"Lời dẫn phân cảnh 1...\",\n" f" \"visual_prompt\": \"A close-up shot of ...\"\n" f" }}\n" f"]" ) scenes = [] try: response = client.models.generate_content( model=gemini_model, contents=script_prompt, config=types.GenerateContentConfig( response_mime_type="application/json" ) ) increment_usage(gemini_model) text_resp = response.text.strip() if text_resp.startswith("```"): if text_resp.startswith("```json"): text_resp = text_resp[7:] else: text_resp = text_resp[3:] if text_resp.endswith("```"): text_resp = text_resp[:-3] text_resp = text_resp.strip() scenes = json.loads(text_resp) except Exception as e: print(f"Gemini JSON generation failed, trying fallback standard prompt: {e}") try: fallback_prompt = ( f"Viết kịch bản dẫn truyện bằng ngôn ngữ có mã '{lang}' cho video {duration} giây " f"về chủ đề: '{prompt}'. Ngắn gọn, súc tích vừa đủ {duration} giây đọc. " f"Chỉ trả về nội dung dẫn truyện." ) fallback_resp = client.models.generate_content(model=gemini_model, contents=fallback_prompt) increment_usage(gemini_model) script_text = fallback_resp.text.strip() scenes = [{ "narration": script_text, "visual_prompt": prompt }] except Exception as e2: print(f"Gemini fallback prompt also failed: {e2}. Using local fallback.") scenes = [{ "narration": f"Video giới thiệu về: {prompt}.", "visual_prompt": prompt }] # Map style to style suffix style_suffixes = { 'cinematic': 'cinematic lighting, film grain, professional camera, 8k resolution, cinematic composition', 'anime': 'anime style, vibrant colors, detailed, Studio Ghibli aesthetic, hand-drawn vector art', 'realistic': 'photorealistic, highly detailed, 8k resolution, realistic textures, natural lighting', 'nature': 'nature photography, beautiful landscape, golden hour lighting, national geographic style', 'tech': 'cyberpunk aesthetic, neon lights, futuristic technology, detailed sci-fi concept art', 'space': 'space, cosmic, nebula, stars, sci-fi scene, highly detailed', 'watercolor': 'soft watercolor painting, artistic brush strokes, textured paper, pastel colors', 'vintage': 'vintage film, retro color grading, 1970s look, scratches, analog film photography' } suffix = style_suffixes.get(style, '') # Map ratio to resolution ratio_resolutions = { '16:9': (1024, 576), '9:16': (576, 1024), '1:1': (768, 768), '4:3': (800, 600) } w, h = ratio_resolutions.get(ratio, (1024, 576)) from moviepy import VideoClip, ImageClip, AudioFileClip, VideoFileClip, concatenate_videoclips, CompositeAudioClip, concatenate_audioclips clips = [] full_narration_audio_paths = [] combined_script = [] combined_visual_prompts = [] for i, scene in enumerate(scenes): narration_text = scene.get('narration', '').strip() visual_prompt = scene.get('visual_prompt', prompt).strip() combined_script.append(narration_text) combined_visual_prompts.append(visual_prompt) scene_duration = duration / len(scenes) audio_file = None if narrate and narration_text: audio_filename = f"scene_{i}_{os.urandom(4).hex()}.mp3" audio_path = os.path.join(app.config['UPLOAD_FOLDER'], audio_filename) try: asyncio.run(edge_tts.Communicate(narration_text, voice).save(audio_path)) audio_file = audio_path full_narration_audio_paths.append(audio_path) except Exception as tts_err: print(f"TTS error for scene {i}: {tts_err}") audio_file = None audio_clip = None if audio_file: try: audio_clip = AudioFileClip(audio_file) scene_duration = audio_clip.duration if scene_duration < 1.0: scene_duration = 3.0 except Exception as aud_err: print(f"AudioFileClip error: {aud_err}") scene_duration = 4.0 else: scene_duration = duration / len(scenes) # Download image image_filename = f"scene_{i}_{os.urandom(4).hex()}.jpg" image_path = os.path.join(app.config['UPLOAD_FOLDER'], image_filename) scene_prompt = f"{visual_prompt}. {suffix}" encoded_prompt = urllib.parse.quote(scene_prompt) img_url = f"https://image.pollinations.ai/prompt/{encoded_prompt}?width={w}&height={h}&nologo=true&private=true" # Retry image download up to 3 times to prevent transient failures img_downloaded = False import time for attempt in range(3): try: req = urllib.request.Request(img_url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=15) as img_resp: img_data = img_resp.read() with open(image_path, 'wb') as img_f: img_f.write(img_data) img_downloaded = True print(f"Successfully downloaded scene {i+1} image on attempt {attempt+1}") break except Exception as img_err: print(f"Attempt {attempt+1} failed to download image for scene {i+1}: {img_err}") time.sleep(1.5) if not img_downloaded: # Fallback to beautiful colorful gradient background instead of flat black image print(f"All image download attempts failed. Using gradient fallback for scene {i+1}") gradient_img = generate_gradient_background(w, h, f"Scene {i+1}") import cv2 cv2.imwrite(image_path, gradient_img) try: img_clip = None if replicate_model == 'hf-wan2.2-i2v-free': try: # Generate real AI video via Gradio (forwarding hf_token if present) hf_video_file = generate_hf_video(image_path, f"{visual_prompt}. {suffix}", hf_token) if hf_video_file and os.path.exists(hf_video_file): img_clip = VideoFileClip(hf_video_file).with_duration(scene_duration) print("Successfully generated and loaded real HF AI video clip!") except Exception as hf_err: print(f"HuggingFace Space generation failed, falling back to slideshow: {hf_err}") img_clip = None # If not using hf-wan2.2-i2v-free OR if HF generation failed, fall back to slideshow if img_clip is None: # Read image using OpenCV and convert BGR to RGB import cv2 img = cv2.imread(image_path) if img is None: import numpy as np img = np.zeros((h, w, 3), dtype=np.uint8) img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) # Create VideoClip instead of ImageClip to bypass caching and allow dynamic zoom img_clip = VideoClip(lambda t: img_rgb).with_duration(scene_duration) # Apply Ken Burns zoom effect (alternate zoom in/out per scene) zoom_mode = 'in' if i % 2 == 0 else 'out' img_clip = apply_zoom_effect(img_clip, zoom_ratio=0.07, mode=zoom_mode) # Apply burn-in subtitles (Pillow-based, unicode-compatible) if narrate and narration_text: img_clip = apply_subtitles_effect(img_clip, narration_text) if audio_clip: img_clip = img_clip.with_audio(audio_clip) clips.append(img_clip) except Exception as clip_err: print(f"Error creating VideoClip: {clip_err}") if not clips: raise Exception("Không thể tạo các phân cảnh video.") video_clip = concatenate_videoclips(clips, method="compose") # Mix music if music != 'none': music_path = get_music_file(music) if music_path: try: music_clip = AudioFileClip(music_path).with_duration(video_clip.duration) vol = 0.12 if narrate else 0.25 music_clip = music_clip.multiply_volume(vol) if video_clip.audio: combined_audio = CompositeAudioClip([video_clip.audio, music_clip]) video_clip = video_clip.with_audio(combined_audio) else: video_clip = video_clip.with_audio(music_clip) except Exception as mus_err: print(f"Music mix error: {mus_err}") out_video_file = f"video_{os.urandom(4).hex()}.mp4" out_video_path = os.path.join(app.config['UPLOAD_FOLDER'], out_video_file) video_clip.write_videofile( out_video_path, fps=24, codec="libx264", audio_codec="aac" ) # Close clips video_clip.close() for c in clips: c.close() # Merge audio narration for download button final_audio_url = None if narrate and full_narration_audio_paths: try: audio_clips = [AudioFileClip(p) for p in full_narration_audio_paths] combined_audio_clip = concatenate_audioclips(audio_clips) out_audio_file = f"narration_{os.urandom(4).hex()}.mp3" out_audio_path = os.path.join(app.config['UPLOAD_FOLDER'], out_audio_file) combined_audio_clip.write_audiofile(out_audio_path) combined_audio_clip.close() for ac in audio_clips: ac.close() final_audio_url = f'/download/{out_audio_file}' except Exception as concat_aud_err: print(f"Error concatenating narration audio: {concat_aud_err}") if full_narration_audio_paths: final_audio_url = f'/download/{os.path.basename(full_narration_audio_paths[0])}' result['script'] = "\n\n".join(combined_script) result['video_prompt'] = "\n\n".join(combined_visual_prompts) result['video_url'] = f'/download/{out_video_file}' if narrate: result['audio_url'] = final_audio_url return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 # ── Routes: Prompt Generator ────────────────────────────────── @app.route('/prompt-generator') def prompt_generator(): return redirect(url_for('prompt_generator_code')) @app.route('/prompt-generator/code') def prompt_generator_code(): return render_template('tools/prompt_studio/index.html', active_tab='code') @app.route('/prompt-generator/video') def prompt_generator_video(): return render_template('tools/prompt_studio/index.html', active_tab='video') @app.route('/prompt-generator/music') def prompt_generator_music(): return render_template('tools/prompt_studio/index.html', active_tab='music') @app.route('/prompt-generator/story') def prompt_generator_story(): return render_template('tools/prompt_studio/index.html', active_tab='story') @app.route('/prompt-generator/video-script') def prompt_generator_video_script(): return render_template('tools/prompt_studio/index.html', active_tab='video_script') @app.route('/prompt-generator/audio-story') def prompt_generator_audio_story(): return render_template('tools/prompt_studio/index.html', active_tab='audio_story') @app.route('/prompt-generator/marketing') def prompt_generator_marketing(): return render_template('tools/prompt_studio/index.html', active_tab='marketing') @app.route('/prompt-generator/education') def prompt_generator_education(): return render_template('tools/prompt_studio/index.html', active_tab='education') @app.route('/prompt-generator/data') def prompt_generator_data(): return render_template('tools/prompt_studio/index.html', active_tab='data') @app.route('/prompt-generator/office') def prompt_generator_office(): return render_template('tools/prompt_studio/index.html', active_tab='office') @app.route('/prompt-generator/brainstorm') def prompt_generator_brainstorm(): return render_template('tools/prompt_studio/index.html', active_tab='brainstorm') @app.route('/prompt-generator/enhancer') def prompt_generator_enhancer(): return render_template('tools/prompt_studio/index.html', active_tab='enhancer') @app.route('/api/prompt-generator/generate', methods=['POST']) def api_prompt_generate(): data = request.json or {} category = data.get('category', 'general') model_id = data.get('gemini_model', DEFAULT_MODEL) if category == 'code': lang = data.get('code_lang', 'Python') task = data.get('code_task', '') framework = data.get('code_framework', '') style = data.get('code_style', 'Clean Code, documented, PEP8') requirements = data.get('code_requirements', '') system_prompt = ( "You are an expert Prompt Engineer for AI Coding Assistants (like Gemini, Claude, or ChatGPT).\n" "Your task is to write a highly detailed, professional prompt in Vietnamese that the user can copy and paste to get the exact code they want.\n" "The generated prompt MUST guide the AI coder to write high-quality code and should include:\n" "- A clear role definition.\n" "- The specific coding task and language/framework to be used.\n" "- Code quality and style guidelines (e.g., Clean Code, DRY, SOLID, error handling).\n" "- Desired output structure (e.g., directory structure, code files, explanations, tests).\n" "- Constraints and edge cases to handle.\n\n" "Input details provided by user:\n" f"- Ngôn ngữ lập trình: {lang}\n" f"- Công việc/Tính năng cần viết: {task}\n" f"- Thư viện/Framework: {framework}\n" f"- Phong cách/Quy chuẩn: {style}\n" f"- Yêu cầu bổ sung: {requirements}\n\n" "Generate ONLY the optimized prompt itself in a clean Markdown format. Do not include any introductory or concluding text. Do not wrap the output in a markdown block of the prompt itself (just output the raw prompt text which is formatted as Markdown)." ) elif category == 'video': topic = data.get('video_topic', '') style = data.get('video_style', 'Cinematic') shot = data.get('video_shot', 'Medium Shot') lighting = data.get('video_lighting', 'Natural Light') aspect_ratio = data.get('video_ratio', '16:9') details = data.get('video_details', '') system_prompt = ( "You are an expert Prompt Engineer for AI Video and Image generation tools (like Midjourney, Stable Diffusion, Runway Gen-2, Sora, Kling, Luma Dream Machine).\n" "Your task is to write a highly detailed, professional English prompt (since image/video AIs work best with English) based on the user's description. The output itself should be written in English, but you can include brief explanations in Vietnamese if necessary. However, the core prompt MUST be in English.\n" "Structure the generated prompt for high quality:\n" "- Core subject description.\n" "- Visual style, art direction, and details.\n" "- Camera settings, lens type, shot type, and camera movement.\n" "- Lighting, color grading, and atmosphere.\n" "- Negative prompts or aspect ratio parameters if suitable.\n\n" "Input details provided by user:\n" f"- Topic/Concept: {topic}\n" f"- Style: {style}\n" f"- Camera Angle/Shot: {shot}\n" f"- Lighting: {lighting}\n" f"- Aspect Ratio: {aspect_ratio}\n" f"- Extra Details: {details}\n\n" "Generate ONLY the optimized prompt in a clean copyable format. Do not include any introductory or concluding text." ) elif category == 'music': theme = data.get('music_theme', '') genre = data.get('music_genre', 'Pop') mood = data.get('music_mood', 'Energetic') instruments = data.get('music_instruments', '') vocals = data.get('music_vocals', 'Vocals') lyrics_idea = data.get('music_lyrics_idea', '') system_prompt = ( "You are an expert Prompt Engineer for AI Music generation tools (like Suno, Udio).\n" "Your task is to write a highly detailed, professional prompt setup. For Suno/Udio, a prompt usually consists of two parts:\n" "1. Style tags (a comma-separated list of genres, sub-genres, moods, instruments, vocal description in English).\n" "2. Lyrics prompt or song structure prompts.\n" "Write the style tags in English and provide structured lyric suggestions (or structure tags like [Verse 1], [Chorus], etc.) in the user's desired language (Vietnamese by default).\n\n" "Input details provided by user:\n" f"- Chủ đề/Ý tưởng nhạc: {theme}\n" f"- Thể loại nhạc: {genre}\n" f"- Cảm xúc/Mood: {mood}\n" f"- Nhạc cụ sử dụng: {instruments}\n" f"- Giọng hát: {vocals}\n" f"- Ý tưởng lời bài hát: {lyrics_idea}\n\n" "Format the output clearly, showing:\n" "- **Style of Music Tags** (comma separated English keywords, e.g. 'synthwave, energetic, 120 bpm, male vocals, retro electric guitar')\n" "- **Prompt Description / Lyrics Structure** (with structured sections like [Verse 1], [Chorus], [Guitar Solo] and recommended lyrics/pacing in Vietnamese)\n\n" "Generate ONLY this structured response. Do not include introductory text." ) elif category == 'story': title = data.get('story_title', '') genre = data.get('story_genre', 'Fantasy') characters = data.get('story_characters', '') plot = data.get('story_plot', '') tone = data.get('story_tone', 'Dramatic') length = data.get('story_length', 'Short Story') system_prompt = ( "You are an expert Prompt Engineer specializing in Creative Writing and Novel Scripting.\n" "Your task is to write a detailed, professional prompt in Vietnamese for a writing AI (like Claude or ChatGPT) to produce a compelling, high-quality story, novel chapter, or script.\n" "The generated prompt should include:\n" "- A detailed persona for the AI writer.\n" "- Background context and setting details.\n" "- Detailed character sheets and their relationship dynamics.\n" "- Plot guidelines, pacing instructions, and key narrative beats (e.g., the hook, the climax).\n" "- Writing style guidelines (e.g., show don't tell, vocabulary level, pacing, dialogue naturalness, tone).\n\n" "Input details provided by user:\n" f"- Tên truyện/Chủ đề: {title}\n" f"- Thể loại: {genre}\n" f"- Các nhân vật chính: {characters}\n" f"- Tóm tắt cốt truyện: {plot}\n" f"- Giọng văn/Tông màu: {tone}\n" f"- Độ dài/Định dạng: {length}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) elif category == 'video_script': topic = data.get('script_topic', '') platform = data.get('script_platform', 'YouTube') duration = data.get('script_duration', '5 minutes') structure = data.get('script_structure', 'Hook-Body-CTA') tone = data.get('script_tone', 'Engaging') cta = data.get('script_cta', '') system_prompt = ( "You are an expert Prompt Engineer for Video Script Writing and Storyboarding.\n" "Your task is to write a detailed, professional prompt in Vietnamese for an AI scriptwriter (like Claude or ChatGPT) to write a high-converting, engaging script for YouTube, TikTok, or Reels.\n" "The generated prompt should specify:\n" "- Target audience, platform context, and goals.\n" "- Script structure details (e.g., hook timing, body progression, transitions, audio cues, visual prompts for the video, and call-to-action).\n" "- Writing voice, pacing, and vocabulary constraints.\n" "- Format instructions (e.g., a two-column table with 'Visual description' and 'Spoken voiceover').\n\n" "Input details provided by user:\n" f"- Chủ đề video: {topic}\n" f"- Nền tảng: {platform}\n" f"- Thời lượng mong muốn: {duration}\n" f"- Cấu trúc kịch bản: {structure}\n" f"- Tông giọng: {tone}\n" f"- Lời kêu gọi hành động (CTA): {cta}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) elif category == 'audio_story': topic = data.get('audio_topic', '') genre = data.get('audio_genre', 'Kinh dị') pacing = data.get('audio_pacing', 'Chậm rãi') sfx = data.get('audio_sfx', '') duration = data.get('audio_duration', '10 minutes') format_ = data.get('audio_format', 'Narrative prose') system_prompt = ( "You are an expert Prompt Engineer specializing in Audio Books, Podcast Stories, and Radio Dramas (Truyện Audio YouTube/TikTok).\n" "Your task is to write a detailed, professional prompt in Vietnamese for a writing AI (like Claude or ChatGPT) to produce a high-quality audio story script optimized for narration.\n" "The generated prompt should instruct the AI to:\n" "- Write in a style optimized for spoken voiceover (easy to read, natural-sounding sentences, emotive phrasing).\n" "- Include sound effects (SFX) and background music (BGM) cues at appropriate dramatic moments (e.g., [SFX: Tiếng cửa kẽo kẹt], [BGM: Nhạc piano trầm buồn nổi lên]).\n" "- Insert pacing and emotion cues for the voice actor (e.g., (thì thầm), (ngập ngừng), (giọng sợ hãi)).\n" "- Structure the story according to the chosen genre and output format.\n\n" "Input details provided by user:\n" f"- Chủ đề/Ý tưởng: {topic}\n" f"- Thể loại truyện: {genre}\n" f"- Tiết tấu & Giọng đọc: {pacing}\n" f"- Tiếng động (SFX) / Nhạc nền (BGM): {sfx}\n" f"- Thời lượng/Độ dài: {duration}\n" f"- Định dạng đầu ra: {format_}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) elif category == 'marketing': topic = data.get('marketing_topic', '') platform = data.get('marketing_platform', 'Facebook') tone = data.get('marketing_tone', 'Persuasive') cta = data.get('marketing_cta', '') details = data.get('marketing_details', '') system_prompt = ( "You are an expert Prompt Engineer for Marketing and Copywriting.\n" "Your task is to write a highly detailed, professional prompt in Vietnamese for a writing AI (like ChatGPT or Claude) to create an engaging, high-converting marketing copy.\n" "The generated prompt should include:\n" "- A detailed role and target persona for the AI copywriter.\n" "- Target audience details, selling proposition, and marketing angle.\n" "- Structural specifications (e.g., AIDA formula, hook, body, Call to Action, emoji usage, line spacing).\n" "- Writing style guidelines (e.g., tone of voice, forbidden words, word count limits, formatting).\n\n" "Input details provided by user:\n" f"- Chủ đề/Sản phẩm: {topic}\n" f"- Nền tảng: {platform}\n" f"- Tông giọng: {tone}\n" f"- Lời kêu gọi hành động (CTA): {cta}\n" f"- Yêu cầu chi tiết khác: {details}\n\n" "Generate ONLY the optimized prompt in a clean copyable format. Do not include any introductory or concluding text." ) elif category == 'education': subject = data.get('edu_subject', '') level = data.get('edu_level', 'General') method = data.get('edu_method', 'Analogy') goals = data.get('edu_goals', '') system_prompt = ( "You are an expert Prompt Engineer specializing in Education, Pedagogy, and Tutoring.\n" "Your task is to write a detailed, professional prompt in Vietnamese for an AI tutor to explain a topic, design a study guide, or tutor a student.\n" "The generated prompt should specify:\n" "- The AI's role (e.g., patient Socratic tutor, experienced professor, or ELI5 expert).\n" "- The target student's educational level and background knowledge.\n" "- The pedagogical method to use (e.g., step-by-step Socratic questioning, real-world analogies, conceptual breakdown, active recall quizzes).\n" "- Output structure (e.g., summary, analogies, vocabulary list, self-assessment questions).\n\n" "Input details provided by user:\n" f"- Chủ đề học tập: {subject}\n" f"- Trình độ học sinh: {level}\n" f"- Phương pháp sư phạm: {method}\n" f"- Mục tiêu học tập cụ thể: {goals}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) elif category == 'data': source = data.get('data_source', 'SQL Table') goal = data.get('data_goal', '') output = data.get('data_output', 'Executive Summary') details = data.get('data_details', '') system_prompt = ( "You are an expert Prompt Engineer for Data Analysis and Business Intelligence.\n" "Your task is to write a detailed, professional prompt in Vietnamese for an AI Data Analyst (like ChatGPT Advanced Data Analysis or Claude) to analyze data, write queries, or draft reports.\n" "The generated prompt should include:\n" "- The analyst role and technical profile (e.g., senior SQL developer, data scientist).\n" "- Data schema context and data source type.\n" "- The specific business question, analysis goals, and statistical methods (e.g., cohort analysis, trend analysis, correlation).\n" "- Output specifications (e.g., SQL queries, Python/pandas code, visualizations recommendations, executive summaries with action items).\n\n" "Input details provided by user:\n" f"- Nguồn dữ liệu & Cấu trúc: {source}\n" f"- Mục tiêu phân tích/Câu hỏi: {goal}\n" f"- Định dạng đầu ra mong muốn: {output}\n" f"- Chi tiết yêu cầu: {details}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) elif category == 'office': text = data.get('office_text', '') task = data.get('office_task', 'Summarize') tone = data.get('office_tone', 'Professional') instructions = data.get('office_instructions', '') system_prompt = ( "You are an expert Prompt Engineer for Office Administration and Document Processing.\n" "Your task is to write a detailed, professional prompt in Vietnamese to guide an AI assistant in performing administrative tasks like summarizing, proofreading, drafting emails, rewriting reports, or polishing documents.\n" "The generated prompt should specify:\n" "- The assistant's persona (e.g., executive assistant, expert proofreader).\n" "- The task details (e.g., bulleted executive summary, email reply matching a specific context, grammatical correction).\n" "- Writing tone, formatting constraints, length constraints, and context preservation guidelines.\n\n" "Input details provided by user:\n" f"- Văn bản gốc: {text}\n" f"- Công việc cần thực hiện: {task}\n" f"- Tông giọng/Phong cách: {tone}\n" f"- Hướng dẫn thêm: {instructions}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) elif category == 'brainstorm': topic = data.get('brain_topic', '') type_ = data.get('brain_type', 'Product Ideas') scope = data.get('brain_scope', 'Broad') details = data.get('brain_details', '') system_prompt = ( "You are an expert Prompt Engineer specializing in Creative Ideation and Brainstorming.\n" "Your task is to write a detailed, professional prompt in Vietnamese to guide an AI in brainstorming innovative concepts, product features, solution ideas, or design variations.\n" "The generated prompt should specify:\n" "- The creative consultant persona.\n" "- The core problem/opportunity and constraints.\n" "- The ideation framework to use (e.g., SCAMPER, lateral thinking, first principles).\n" "- Output structure (e.g., categorized list of 10 ideas, mindmap format, pros/cons list).\n\n" "Input details provided by user:\n" f"- Chủ đề/Bài toán cần giải: {topic}\n" f"- Loại ý tưởng mong muốn: {type_}\n" f"- Phạm vi/Mức độ sáng tạo: {scope}\n" f"- Chi tiết yêu cầu: {details}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) else: # Custom Enhancer / General raw_prompt = data.get('raw_prompt', '') target_ai = data.get('target_ai', 'General AI') custom_instructions = data.get('custom_instructions', '') system_prompt = ( "You are an expert Prompt Engineer.\n" "Your task is to take the user's raw prompt and rewrite it into a highly effective, optimized, and detailed prompt tailored for the target AI tool. The output should be in Vietnamese by default, unless the user requested otherwise.\n" "Add structure, clear guidelines, persona definition, input-output specifications, constraints, and examples where applicable.\n\n" "Input details provided by user:\n" f"- Raw Prompt: {raw_prompt}\n" f"- Target AI Tool: {target_ai}\n" f"- Custom Guidelines/Instructions: {custom_instructions}\n\n" "Generate ONLY the optimized prompt. Do not include introductory or concluding text." ) try: response = client.models.generate_content( model=model_id, contents=system_prompt ) increment_usage(model_id) return jsonify({ 'success': True, 'generated_prompt': response.text.strip() }) except Exception as e: return jsonify({ 'success': False, 'error': _err_msg(str(e)) }), 500 @app.route('/api/prompt-generator/execute', methods=['POST']) def api_prompt_execute(): data = request.json or {} generated_prompt = data.get('prompt', '').strip() model_id = data.get('gemini_model', DEFAULT_MODEL) if not generated_prompt: return jsonify({'success': False, 'error': 'Không tìm thấy prompt để chạy thử'}), 400 try: response = client.models.generate_content( model=model_id, contents=generated_prompt ) increment_usage(model_id) return jsonify({ 'success': True, 'execution_result': response.text.strip() }) except Exception as e: return jsonify({ 'success': False, 'error': _err_msg(str(e)) }), 500 @app.route('/audio-story-studio') def audio_story_studio_index(): return render_template('tools/audio_story_studio/index.html') # ── Step 1: Generate optimized prompt from user config ── @app.route('/api/audio-story-studio/create-prompt', methods=['POST']) def audio_story_studio_create_prompt(): data = request.json or {} idea = data.get('idea', '').strip() genre = data.get('genre', 'Drama gia đình') style = data.get('style', 'Kịch tính, gay cấn') mood = data.get('mood', 'Căng thẳng') length = data.get('length', '10 phút') audience = data.get('audience', 'Người lớn') extras = data.get('extras', '') model_id = data.get('gemini_model', DEFAULT_MODEL) if not idea: return jsonify({'error': 'Vui lòng nhập ý tưởng truyện'}), 400 system_instruction = ( "Bạn là chuyên gia viết prompt cho AI tạo kịch bản truyện audio.\n" "Nhiệm vụ: Từ các thông số người dùng cung cấp, hãy viết MỘT prompt tối ưu bằng tiếng Việt " "để một AI khác có thể dùng prompt đó tạo ra kịch bản truyện audio hoàn chỉnh.\n\n" "Prompt tối ưu cần:\n" "- Mô tả rõ ràng bối cảnh, nhân vật chính, xung đột trung tâm\n" "- Yêu cầu cấu trúc kịch bản: mở đầu hấp dẫn, diễn biến leo thang, cao trào, kết thúc bất ngờ\n" "- Chỉ định tông giọng kể, nhịp độ, cảm xúc\n" "- Yêu cầu viết tối ưu cho đọc thành tiếng (câu ngắn gọn, dùng dấu ... để tạo nhịp)\n" "- Ghi rõ độ dài mong muốn\n\n" "CHỈ trả về prompt tối ưu, không thêm lời giới thiệu hay giải thích.\n" ) user_input = ( f"Ý tưởng: {idea}\n" f"Thể loại: {genre}\n" f"Phong cách kể: {style}\n" f"Tâm trạng/Mood: {mood}\n" f"Độ dài mong muốn: {length}\n" f"Đối tượng khán giả: {audience}\n" ) if extras: user_input += f"Yêu cầu thêm: {extras}\n" try: response, used_model = generate_with_fallback( model_id, user_input, config=types.GenerateContentConfig( system_instruction=system_instruction ) ) return jsonify({ 'success': True, 'optimized_prompt': response.text.strip() }) except Exception as e: return jsonify({'error': f"Lỗi tạo prompt: {_err_msg(str(e))}"}), 500 # ── Step 2: Generate full script from prompt ── @app.route('/api/audio-story-studio/create-script', methods=['POST']) def audio_story_studio_create_script(): data = request.json or {} prompt = data.get('prompt', '').strip() model_id = data.get('gemini_model', DEFAULT_MODEL) if not prompt: return jsonify({'error': 'Vui lòng cung cấp prompt'}), 400 system_instruction = ( "Bạn là nhà văn chuyên viết kịch bản truyện audio cho YouTube và Podcast.\n" "Nhiệm vụ: Dựa vào prompt được cung cấp, hãy viết kịch bản truyện audio hoàn chỉnh bằng tiếng Việt.\n\n" "LƯU Ý QUAN TRỌNG ĐỂ TRÁNH LỖI HỢP LỆ JSON:\n" "1. Tuyệt đối KHÔNG sử dụng dấu ngoặc kép kép (\") trong nội dung của các giá trị chuỗi (như 'script', 'title', 'reason'). Nếu có lời thoại nhân vật, hãy dùng dấu gạch ngang (-) hoặc dấu ngoặc đơn (') thay thế.\n" "2. Không sử dụng ký tự xuống dòng thực tế trong các giá trị chuỗi JSON. Dùng '\\n' (chuỗi ký tự viết thường) thay thế nếu muốn ngắt dòng.\n" "3. Đảm bảo toàn bộ phản hồi trả về là một đối tượng JSON hợp lệ.\n\n" "Cấu trúc JSON cần trả về:\n" "- 'title': Tiêu đề truyện hấp dẫn\n" "- 'script': Kịch bản truyện audio đầy đủ bằng tiếng Việt, viết dạng văn xuôi dễ đọc, không dùng markup hay action tags.\n" "- 'suggested_voice': 'vi-VN-HoaiMyNeural' (Nữ) hoặc 'vi-VN-NamMinhNeural' (Nam)\n" "- 'rate': Tốc độ đọc khuyến nghị ('-10%', '-5%', '+0%', '+5%')\n" "- 'reason': Giải thích ngắn gọn bằng tiếng Việt vì sao chọn giọng và tốc độ này\n" ) try: response, used_model = generate_with_fallback( model_id, prompt, config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() # Clean code block indicators if model generated them if raw_text.startswith("```"): lines = raw_text.splitlines() if lines[0].startswith("```"): lines = lines[1:] if lines and lines[-1].startswith("```"): lines = lines[:-1] raw_text = "\n".join(lines).strip() # Find outer JSON braces start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, **result}) except Exception as e: print(f"JSON Parse error raw response: {response.text if 'response' in locals() else 'None'}") return jsonify({'error': f"Lỗi tạo kịch bản: {_err_msg(str(e))}"}), 500 # ── Step 3: Synthesize audio from script ── @app.route('/api/audio-story-studio/synthesize', methods=['POST']) def audio_story_studio_synthesize(): data = request.json or {} text = data.get('script', '').strip() voice = data.get('voice', 'vi-VN-HoaiMyNeural') rate = data.get('rate', '+0%') if not text: return jsonify({'error': 'Vui lòng cung cấp nội dung kịch bản'}), 400 # Auto detect engine: Edge-TTS voices start with 'vi-VN-' is_edge = voice.startswith('vi-VN-') if is_edge: import asyncio, edge_tts out_filename = f"audio_story_{os.urandom(4).hex()}.mp3" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) async def _generate(): communicate = edge_tts.Communicate(text, voice, rate=rate) await communicate.save(out_path) try: asyncio.run(_generate()) return jsonify({ 'success': True, 'audio_url': f'/download/{out_filename}', 'filename': out_filename }) except Exception as e: return jsonify({'error': f"Lỗi chuyển giọng đọc Edge-TTS: {str(e)}"}), 500 else: # VieNeu engine if len(text) > 3000: return jsonify({'error': 'Kịch bản quá dài (tối đa 3000 ký tự với VieNeu)'}), 400 try: from vieneu import Vieneu tts_obj = Vieneu() out_filename = f"audio_story_{os.urandom(4).hex()}.wav" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) # Infer using VieNeu preset voice ID audio = tts_obj.infer(text=text, voice=voice) tts_obj.save(audio, out_path) return jsonify({ 'success': True, 'audio_url': f'/download/{out_filename}', 'filename': out_filename }) except Exception as e: return jsonify({'error': f"Lỗi chuyển giọng đọc VieNeu-TTS: {str(e)}"}), 500 # ── Helper for Shorts Subtitles ── def draw_pil_subtitles_shorts(frame, text, font_size=36): from PIL import Image, ImageDraw, ImageFont import numpy as np pil_img = Image.fromarray(frame) draw = ImageDraw.Draw(pil_img) try: font = ImageFont.truetype("arial.ttf", font_size) except: try: font = ImageFont.truetype("C:\\Windows\\Fonts\\arial.ttf", font_size) except: font = ImageFont.load_default() w, h = pil_img.size words = text.split(' ') lines = [] current_line = "" for word in words: test_line = current_line + " " + word if current_line else word bbox = font.getbbox(test_line) tw = bbox[2] - bbox[0] if tw > w - 80: # padding for vertical frame if current_line: lines.append(current_line) current_line = word else: current_line = test_line if current_line: lines.append(current_line) line_height = font.getbbox("A")[3] - font.getbbox("A")[1] + 12 y_start = (h // 2) - (len(lines) * line_height) // 2 for i, line in enumerate(lines): y = y_start + i * line_height bbox = font.getbbox(line) tw = bbox[2] - bbox[0] x = (w - tw) // 2 outline_color = (0, 0, 0) for dx, dy in [(-3,-3), (-3,3), (3,-3), (3,3), (-3,0), (3,0), (0,-3), (0,3)]: draw.text((x + dx, y + dy), line, font=font, fill=outline_color) draw.text((x, y), line, font=font, fill=(255, 223, 0)) return np.array(pil_img) def apply_subtitles_effect_shorts(clip, subtitle_text): def effect(get_frame, t): frame = get_frame(t) frame_copy = frame.copy() return draw_pil_subtitles_shorts(frame_copy, subtitle_text) return clip.transform(effect) # ── 1. AI Podcast Studio ── @app.route('/podcast-studio') def podcast_studio_index(): return render_template('tools/podcast_studio/index.html') @app.route('/api/podcast-studio/generate-script', methods=['POST']) def podcast_studio_generate_script(): data = request.json or {} topic = data.get('topic', '').strip() voice_a = data.get('voice_a', 'vi-VN-NamMinhNeural') voice_b = data.get('voice_b', 'vi-VN-HoaiMyNeural') style = data.get('style', 'Thân thiện, cởi mở') model_id = data.get('gemini_model', DEFAULT_MODEL) if not topic: return jsonify({'error': 'Vui lòng nhập chủ đề Podcast'}), 400 system_instruction = ( "Bạn là nhà biên kịch podcast chuyên nghiệp.\n" "Nhiệm vụ: Viết một kịch bản cuộc trò chuyện đối thoại tự nhiên, lôi cuốn giữa Host A và Host B về chủ đề được yêu cầu.\n\n" "LƯU Ý ĐẶC BIỆT ĐỂ TRÁNH LỖI PHÂN TÍCH JSON:\n" "1. Tuyệt đối KHÔNG sử dụng dấu ngoặc kép kép (\") bên trong các lời thoại. Dùng dấu ngoặc đơn (') thay thế.\n" "2. Không sử dụng phím Enter ngắt dòng vật lý trong văn bản đối thoại, hãy dùng ký tự '\\n'.\n" "3. Trả về đúng kịch bản dạng đối thoại xen kẽ giữa Host A và Host B.\n\n" "Cấu trúc JSON cần trả về:\n" "{\n" " 'title': 'Tiêu đề Podcast ngắn gọn và hấp dẫn',\n" " 'dialogue': [\n" " {'speaker': 'A', 'text': 'Lời thoại của Host A...'},\n" " {'speaker': 'B', 'text': 'Lời thoại của Host B...'}\n" " ]\n" "}" ) user_input = f"Chủ đề: {topic}\nPhong cách: {style}\nHost A: {voice_a}\nHost B: {voice_b}" try: response, used_model = generate_with_fallback( model_id, user_input, config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, **result}) except Exception as e: return jsonify({'error': f"Lỗi tạo kịch bản Podcast: {_err_msg(str(e))}"}), 500 @app.route('/api/podcast-studio/synthesize', methods=['POST']) def podcast_studio_synthesize(): data = request.json or {} dialogue = data.get('dialogue', []) voice_a = data.get('voice_a', 'vi-VN-NamMinhNeural') voice_b = data.get('voice_b', 'vi-VN-HoaiMyNeural') if not dialogue: return jsonify({'error': 'Không có lời thoại nào để thu âm'}), 400 from moviepy.editor import AudioFileClip, concatenate_audioclips clips = [] temp_files = [] try: for idx, turn in enumerate(dialogue): speaker = turn.get('speaker', 'A') text = turn.get('text', '').strip() if not text: continue voice = voice_a if speaker == 'A' else voice_b is_edge = voice.startswith('vi-VN-') temp_filename = f"podcast_temp_{idx}_{os.urandom(4).hex()}.mp3" if is_edge else f"podcast_temp_{idx}_{os.urandom(4).hex()}.wav" temp_path = os.path.join(app.config['UPLOAD_FOLDER'], temp_filename) temp_files.append(temp_path) if is_edge: import asyncio, edge_tts async def _gen(): await edge_tts.Communicate(text, voice).save(temp_path) asyncio.run(_gen()) else: from vieneu import Vieneu tts_obj = Vieneu() audio = tts_obj.infer(text=text, voice=voice) tts_obj.save(audio, temp_path) clips.append(AudioFileClip(temp_path)) if not clips: return jsonify({'error': 'Không tạo được file âm thanh nào'}), 400 final_clip = concatenate_audioclips(clips) out_filename = f"podcast_{os.urandom(4).hex()}.mp3" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) final_clip.write_audiofile(out_path) for clip in clips: clip.close() for path in temp_files: try: os.remove(path) except: pass return jsonify({ 'success': True, 'audio_url': f'/download/{out_filename}', 'filename': out_filename }) except Exception as e: for clip in clips: try: clip.close() except: pass for path in temp_files: try: os.remove(path) except: pass return jsonify({'error': f"Lỗi thu âm Podcast: {str(e)}"}), 500 # ── 2. AI Shorts/TikTok Generator ── @app.route('/shorts-generator') def shorts_generator_index(): return render_template('tools/shorts_generator/index.html') @app.route('/api/shorts-generator/generate-script', methods=['POST']) def shorts_generator_generate_script(): data = request.json or {} topic = data.get('topic', '').strip() style = data.get('style', 'cinematic') model_id = data.get('gemini_model', DEFAULT_MODEL) if not topic: return jsonify({'error': 'Vui lòng nhập chủ đề video ngắn'}), 400 system_instruction = ( "Bạn là đạo diễn chuyên sản xuất video ngắn TikTok và YouTube Shorts (9:16).\n" "Hãy tạo kịch bản phân cảnh chi tiết gồm 3-5 cảnh cho chủ đề được cung cấp.\n\n" "LƯU Ý ĐẶC BIỆT ĐỂ TRÁNH LỖI PHÂN TÍCH JSON:\n" "1. Tuyệt đối KHÔNG sử dụng dấu ngoặc kép kép (\") bên trong các giá trị chuỗi. Dùng dấu ngoặc đơn (') thay thế.\n" "2. Không sử dụng phím Enter ngắt dòng vật lý trong các chuỗi, hãy dùng ký tự '\\n'.\n\n" "Cấu trúc JSON cần trả về:\n" "{\n" " 'title': 'Tiêu đề video Shorts cuốn hút',\n" " 'scenes': [\n" " {\n" " 'scene_num': 1,\n" " 'narration': 'Lời bình dẫn truyện cho cảnh này bằng tiếng Việt...',\n" " 'visual_prompt': 'Mô tả hình ảnh chi tiết bằng tiếng Anh để vẽ tranh minh họa...'\n" " }\n" " ]\n" "}" ) try: response, used_model = generate_with_fallback( model_id, f"Chủ đề: {topic}\nPhong cách: {style}", config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, **result}) except Exception as e: return jsonify({'error': f"Lỗi tạo kịch bản Shorts: {_err_msg(str(e))}"}), 500 @app.route('/api/shorts-generator/build-video', methods=['POST']) def shorts_generator_build_video(): import urllib.parse, urllib.request data = request.json or {} scenes = data.get('scenes', []) voice = data.get('voice', 'vi-VN-HoaiMyNeural') style = data.get('style', 'cinematic') music = data.get('music', 'none') if not scenes: return jsonify({'error': 'Không có phân cảnh nào để tạo video'}), 400 # 9:16 Vertical size w, h = 576, 1024 style_suffixes = { 'cinematic': 'cinematic lighting, film grain, photorealistic, professional camera portrait 9:16 style', 'anime': 'anime portrait style, vibrant colors, detailed, Studio Ghibli vertical composition', 'realistic': 'photorealistic portrait, highly detailed, 8k resolution, realistic lighting 9:16', 'watercolor': 'soft watercolor painting, textured paper, artistic brush strokes, vertical art', 'cyberpunk': 'cyberpunk neon style, futuristic sci-fi city portrait, detailed digital painting' } suffix = style_suffixes.get(style, '') from moviepy import VideoClip, AudioFileClip, concatenate_videoclips, CompositeAudioClip import cv2 import numpy as np clips = [] temp_files = [] try: for i, scene in enumerate(scenes): narration_text = scene.get('narration', '').strip() visual_prompt = scene.get('visual_prompt', '').strip() # 1. Download vertical image image_filename = f"shorts_img_{i}_{os.urandom(4).hex()}.jpg" image_path = os.path.join(app.config['UPLOAD_FOLDER'], image_filename) temp_files.append(image_path) scene_prompt = f"{visual_prompt}. {suffix}" encoded_prompt = urllib.parse.quote(scene_prompt) img_url = f"https://image.pollinations.ai/prompt/{encoded_prompt}?width={w}&height={h}&nologo=true&private=true" img_downloaded = False for attempt in range(3): try: req = urllib.request.Request(img_url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=12) as img_resp: img_data = img_resp.read() with open(image_path, 'wb') as img_f: img_f.write(img_data) img_downloaded = True break except: import time time.sleep(1.0) if not img_downloaded: # Fallback to colorful solid image img_array = np.zeros((h, w, 3), dtype=np.uint8) img_array[:, :, 0] = 50 + i * 30 # Blue img_array[:, :, 1] = 20 + i * 20 # Green img_array[:, :, 2] = 120 # Red cv2.imwrite(image_path, img_array) # 2. TTS Voiceover audio_filename = f"shorts_aud_{i}_{os.urandom(4).hex()}.mp3" audio_path = os.path.join(app.config['UPLOAD_FOLDER'], audio_filename) temp_files.append(audio_path) is_edge = voice.startswith('vi-VN-') if is_edge: import asyncio, edge_tts async def _gen(): await edge_tts.Communicate(narration_text, voice).save(audio_path) asyncio.run(_gen()) else: from vieneu import Vieneu tts_obj = Vieneu() audio = tts_obj.infer(text=narration_text, voice=voice) tts_obj.save(audio, audio_path) # 3. Create Clip audio_clip = AudioFileClip(audio_path) duration = audio_clip.duration if duration < 1.0: duration = 3.0 img = cv2.imread(image_path) if img is None: img = np.zeros((h, w, 3), dtype=np.uint8) img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) # Create video clip with Ken Burns zoom effect img_clip = VideoClip(lambda t: img_rgb).with_duration(duration) zoom_mode = 'in' if i % 2 == 0 else 'out' img_clip = apply_zoom_effect(img_clip, zoom_ratio=0.08, mode=zoom_mode) img_clip = apply_subtitles_effect_shorts(img_clip, narration_text) img_clip = img_clip.with_audio(audio_clip) clips.append(img_clip) if not clips: return jsonify({'error': 'Không tạo được phân cảnh video nào'}), 400 video_clip = concatenate_videoclips(clips, method="compose") # 4. Background music if music != 'none': music_path = get_music_file(music) if music_path: try: music_clip = AudioFileClip(music_path).with_duration(video_clip.duration) music_clip = music_clip.multiply_volume(0.12) if video_clip.audio: combined_audio = CompositeAudioClip([video_clip.audio, music_clip]) video_clip = video_clip.with_audio(combined_audio) else: video_clip = video_clip.with_audio(music_clip) except Exception as mus_err: print(f"Music error: {mus_err}") out_filename = f"shorts_{os.urandom(4).hex()}.mp4" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) video_clip.write_videofile( out_path, fps=24, codec="libx264", audio_codec="aac", temp_audiofile=os.path.join(app.config['UPLOAD_FOLDER'], f"temp_shorts_{os.urandom(2).hex()}.mp3"), remove_temp=True ) # Close all clips video_clip.close() for clip in clips: clip.close() # Clean temp files for path in temp_files: try: os.remove(path) except: pass return jsonify({ 'success': True, 'video_url': f'/download/{out_filename}', 'filename': out_filename }) except Exception as e: # Clean up in case of failure for clip in clips: try: clip.close() except: pass for path in temp_files: try: os.remove(path) except: pass return jsonify({'error': f"Lỗi xuất video: {str(e)}"}), 500 # ── 3. AI Comic & Storyboard Creator ── @app.route('/comic-creator') def comic_creator_index(): return render_template('tools/comic_creator/index.html') @app.route('/api/comic-creator/generate-panels', methods=['POST']) def comic_creator_generate_panels(): data = request.json or {} plot = data.get('plot', '').strip() style = data.get('style', 'manga') model_id = data.get('gemini_model', DEFAULT_MODEL) if not plot: return jsonify({'error': 'Vui lòng nhập cốt truyện'}), 400 system_instruction = ( "Bạn là họa sĩ biên kịch phân cảnh truyện tranh (Storyboard Artist).\n" "Hãy chuyển thể cốt truyện thành một bộ truyện tranh gồm 4-6 khung hình (panels).\n\n" "LƯU Ý ĐẶC BIỆT ĐỂ TRÁNH LỖI PHÂN TÍCH JSON:\n" "1. Tuyệt đối KHÔNG sử dụng dấu ngoặc kép kép (\") bên trong các chuỗi. Dùng dấu ngoặc đơn (') thay thế.\n" "2. Không sử dụng phím Enter ngắt dòng vật lý, dùng ký tự '\\n' để ngắt dòng.\n\n" "Cấu trúc JSON cần trả về:\n" "{\n" " 'title': 'Tên truyện tranh',\n" " 'panels': [\n" " {\n" " 'panel_num': 1,\n" " 'visual_prompt': 'Mô tả chi tiết hình ảnh bằng tiếng Anh để vẽ tranh minh họa...',\n" " 'narration': 'Lời dẫn thoại hoặc lời thuyết minh khung tranh bằng tiếng Việt...'\n" " }\n" " ]\n" "}" ) try: response, used_model = generate_with_fallback( model_id, f"Cốt truyện: {plot}\nPhong cách vẽ: {style}", config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, **result}) except Exception as e: return jsonify({'error': f"Lỗi thiết kế khung hình: {_err_msg(str(e))}"}), 500 @app.route('/api/comic-creator/compile-pdf', methods=['POST']) def comic_creator_compile_pdf(): data = request.json or {} title = data.get('title', 'Truyện Tranh AI').strip() panels = data.get('panels', []) if not panels: return jsonify({'error': 'Không có khung hình nào để tạo PDF'}), 400 out_filename = f"comic_{os.urandom(4).hex()}.pdf" out_path = os.path.join(app.config['UPLOAD_FOLDER'], out_filename) try: from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image as RLImage, PageBreak from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import inch import urllib.request, urllib.parse # Setup PDF document doc = SimpleDocTemplate(out_path, pagesize=letter, rightMargin=36, leftMargin=36, topMargin=36, bottomMargin=36) story = [] styles = getSampleStyleSheet() # Custom styles title_style = ParagraphStyle( 'ComicTitle', parent=styles['Heading1'], fontName='Helvetica-Bold', fontSize=26, leading=30, textColor='#4361ee', alignment=1, # Center spaceAfter=15 ) caption_style = ParagraphStyle( 'ComicCaption', parent=styles['Normal'], fontName='Helvetica', fontSize=11, leading=15, textColor='#1e293b', spaceBefore=10, spaceAfter=15, alignment=1 # Center ) # Cover page title story.append(Spacer(1, 100)) story.append(Paragraph(title.upper(), title_style)) story.append(Spacer(1, 20)) story.append(Paragraph("Sách truyện tranh minh họa tạo tự động bởi AI", caption_style)) story.append(PageBreak()) for idx, panel in enumerate(panels): visual_prompt = panel.get('visual_prompt', '').strip() narration = panel.get('narration', '').strip() # Download panel image temp_img_name = f"temp_pdf_panel_{idx}_{os.urandom(2).hex()}.jpg" temp_img_path = os.path.join(app.config['UPLOAD_FOLDER'], temp_img_name) encoded_prompt = urllib.parse.quote(visual_prompt) img_url = f"https://image.pollinations.ai/prompt/{encoded_prompt}?width=600&height=450&nologo=true&private=true" img_downloaded = False for attempt in range(3): try: req = urllib.request.Request(img_url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=12) as img_resp: img_data = img_resp.read() with open(temp_img_path, 'wb') as img_f: img_f.write(img_data) img_downloaded = True break except: import time time.sleep(1.0) if img_downloaded: # Add title for this panel story.append(Paragraph(f"KHUNG HÌNH {idx+1}", ParagraphStyle('PTitle', parent=styles['Heading2'], textColor='#7c3aed', alignment=1))) story.append(Spacer(1, 10)) # Add image - scaled to fit letter page width safely story.append(RLImage(temp_img_path, width=5.5*inch, height=4.125*inch)) story.append(Spacer(1, 10)) # Add narration subtitle story.append(Paragraph(narration, caption_style)) story.append(PageBreak()) doc.build(story) return jsonify({ 'success': True, 'pdf_url': f'/download/{out_filename}', 'filename': out_filename }) except Exception as e: return jsonify({'error': f"Lỗi tạo sách PDF: {str(e)}"}), 500 # ── 4. AI Omnichannel Content Writer ── @app.route('/omnichannel-writer') def omnichannel_writer_index(): return render_template('tools/omnichannel_writer/index.html') @app.route('/api/omnichannel-writer/generate', methods=['POST']) def omnichannel_writer_generate(): data = request.json or {} topic = data.get('topic', '').strip() model_id = data.get('gemini_model', DEFAULT_MODEL) if not topic: return jsonify({'error': 'Vui lòng nhập ý tưởng bài viết'}), 400 system_instruction = ( "Bạn là chuyên gia sáng tạo nội dung đa kênh (Omnichannel Copywriter).\n" "Nhiệm vụ: Chuyển đổi nội dung/ý tưởng được cung cấp thành 4 bài đăng chuyên biệt cho: Facebook, TikTok (kịch bản thoại ngắn), Threads, và LinkedIn.\n\n" "LƯU Ý ĐẶC BIỆT ĐỂ TRÁNH LỖI PHÂN TÍCH JSON:\n" "1. Tuyệt đối KHÔNG sử dụng dấu ngoặc kép kép (\") bên trong các chuỗi. Dùng dấu ngoặc đơn (') thay thế.\n" "2. Sử dụng ký tự '\\n' để ngắt dòng thay vì phím Enter thực tế.\n\n" "Cấu trúc JSON cần trả về:\n" "{\n" " 'facebook': 'Nội dung bài đăng Facebook (thân thiện, nhiều icon sinh động, hashtags)...',\n" " 'tiktok': 'Kịch bản thoại ngắn TikTok (gồm: Hook giật gân, Nội dung lời bình thuyết minh, Kêu gọi hành động)...',\n" " 'threads': 'Nội dung chuỗi bài viết ngắn Threads (ngắn gọn, xúc tích, khơi gợi bình luận)...',\n" " 'linkedin': 'Nội dung bài viết LinkedIn (chuyên nghiệp, sâu sắc, chia sẻ giá trị, định dạng rõ ràng)...'\n" "}" ) try: response, used_model = generate_with_fallback( model_id, topic, config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, **result}) except Exception as e: return jsonify({'error': f"Lỗi tạo nội dung đa kênh: {_err_msg(str(e))}"}), 500 # ============================================== # TOOL: AI VIDEO CHAPTERS # ============================================== @app.route('/video-chapters') def tool_video_chapters(): return render_template('tools/video_chapters/index.html') @app.route('/api/video-chapters/generate', methods=['POST']) def api_video_chapters_generate(): data = request.json or {} transcript = data.get('transcript', '').strip() duration_min = data.get('duration_min', 5) model_id = data.get('gemini_model', '') if not transcript: return jsonify({'error': 'Vui lòng nhập văn bản phụ đề/transcript'}), 400 system_instruction = ( "Bạn là chuyên gia SEO và biên tập Video YouTube chuyên nghiệp.\n" f"Nhiệm vụ của bạn là phân tích đoạn transcript thô và phân chia thành các mốc chương mục (timestamps/chapters) hợp lý dựa trên tổng thời lượng ước lượng là {duration_min} phút.\n" "Hãy đảm bảo mỗi chương mục có mốc thời gian bắt đầu chính xác tăng dần từ 00:00 và có mô tả ngắn gọn, hấp dẫn bằng tiếng Việt.\n" "Trả về kết quả dưới dạng JSON có khóa 'chapters' là một danh mục các đối tượng chứa 'time' (ví dụ: '01:32') và 'title' (tiêu đề chương mục).\n" "LƯU Ý QUAN TRỌNG ĐỂ TRÁNH LỖI PHÂN TÍCH JSON:\n" "1. Chỉ trả về một đối tượng JSON hợp lệ duy nhất.\n" "2. Không sử dụng ký tự đặc biệt hay phím xuống dòng thực tế trong chuỗi.\n\n" "Cấu trúc JSON mong muốn:\n" "{\n" " 'chapters': [\n" " { 'time': '00:00', 'title': 'Giới thiệu' },\n" " { 'time': '01:30', 'title': 'Tại sao cách cũ không hoạt động' }\n" " ]\n" "}" ) try: response, used_model = generate_with_fallback( model_id, transcript, config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, 'chapters': result.get('chapters', [])}) except Exception as e: return jsonify({'error': f"Lỗi tạo mốc chương mục: {_err_msg(str(e))}"}), 500 # ============================================== # TOOL: AI THUMBNAIL DESIGNER # ============================================== @app.route('/thumbnail-designer') def tool_thumbnail_designer(): return render_template('tools/thumbnail_designer/index.html') @app.route('/api/thumbnail-designer/generate-titles-prompts', methods=['POST']) def api_thumbnail_designer_generate(): data = request.json or {} idea = data.get('idea', '').strip() art_style = data.get('art_style', 'cinematic') model_id = data.get('gemini_model', '') if not idea: return jsonify({'error': 'Vui lòng nhập ý tưởng video'}), 400 system_instruction = ( "Bạn là chuyên gia nghiên cứu thuật toán YouTube và thiết kế hình ảnh Thumbnail có CTR (tỷ lệ nhấp chuột) cao.\n" f"Từ ý tưởng video thô của người dùng, hãy đề xuất:\n" "1. 5 kiểu tiêu đề cuốn hút khác nhau bằng tiếng Việt: Clickbait (giật gân), Curious (tò mò), Question (câu hỏi), Story (kể chuyện), Direct (trực diện).\n" "2. Đề xuất bố cục thiết kế Thumbnail chi tiết (nơi đặt chữ, hình ảnh tiêu điểm, màu sắc phối hợp).\n" f"3. 2 prompt tiếng Anh tối ưu để sinh hình nền bằng AI nghệ thuật theo phong cách '{art_style}'. Không chứa dấu ngoặc kép kép (\").\n" "Trả về kết quả dưới dạng JSON duy nhất chứa: 'titles' (object chứa 5 kiểu tiêu đề), 'visual_layout' (chuỗi mô tả bố cục), và 'prompts' (mảng chứa 2 chuỗi prompt tiếng Anh).\n\n" "Cấu trúc JSON mong muốn:\n" "{\n" " 'titles': {\n" " 'clickbait': '...',\n" " 'curious': '...',\n" " 'question': '...',\n" " 'story': '...',\n" " 'direct': '...'\n" " },\n" " 'visual_layout': 'Mô tả bố cục...',\n" " 'prompts': ['english prompt 1', 'english prompt 2']\n" "}" ) try: response, used_model = generate_with_fallback( model_id, idea, config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, **result}) except Exception as e: return jsonify({'error': f"Lỗi thiết kế tiêu đề/prompt: {_err_msg(str(e))}"}), 500 # ============================================== # TOOL: AI AUTO-CUT SCRIPT & HOOK # ============================================== @app.route('/script-hook-generator') def tool_script_hook_generator(): return render_template('tools/script_hook_generator/index.html') @app.route('/api/script-hook-generator/optimize', methods=['POST']) def api_script_hook_optimize(): data = request.json or {} draft_script = data.get('draft_script', '').strip() pace = data.get('pace', 'vừa phải') model_id = data.get('gemini_model', '') if not draft_script: return jsonify({'error': 'Vui lòng nhập kịch bản nháp'}), 400 system_instruction = ( "Bạn là biên kịch video ngắn TikTok / YouTube Shorts chuyên nghiệp, sở hữu nhiều video triệu view.\n" f"Hãy tối ưu hóa kịch bản nháp của người dùng theo nhịp điệu đọc '{pace}' bằng cách:\n" "1. Tạo ra 3 kiểu Hook giữ chân 3 giây đầu tiên bằng tiếng Việt: Kịch tính (Dramatic), Tò mò (Curious), Đảo ngược thực tế (Reverse).\n" "2. Viết lại kịch bản tối ưu (ngắn gọn, tập trung, dễ hiểu), chèn thêm các hướng dẫn âm thanh hình ảnh như '[B-Roll: mô tả cảnh]', '[SFX: mô tả âm thanh]'.\n" "3. Tính toán thời gian đọc ước lượng (giây).\n" "Trả về kết quả dưới dạng JSON chứa các khóa: 'hooks' (object chứa 3 hook), 'optimized_script' (kịch bản tối ưu), 'duration_seconds' (số nguyên giây).\n\n" "Cấu trúc JSON mong muốn:\n" "{\n" " 'hooks': {\n" " 'dramatic': '...',\n" " 'curious': '...',\n" " 'reverse': '...'\n" " },\n" " 'optimized_script': 'Kịch bản tối ưu...',\n" " 'duration_seconds': 45\n" "}" ) try: response, used_model = generate_with_fallback( model_id, draft_script, config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, **result}) except Exception as e: return jsonify({'error': f"Lỗi tối ưu kịch bản: {_err_msg(str(e))}"}), 500 # ============================================== # TOOL: AI CONTENT PLANNER # ============================================== @app.route('/content-planner') def tool_content_planner(): return render_template('tools/content_planner/index.html') @app.route('/api/content-planner/generate', methods=['POST']) def api_content_planner_generate(): data = request.json or {} niche = data.get('niche', '').strip() platform = data.get('platform', 'YouTube & TikTok') model_id = data.get('gemini_model', '') if not niche: return jsonify({'error': 'Vui lòng nhập chủ đề/lĩnh vực kênh'}), 400 system_instruction = ( f"Bạn là giám đốc chiến lược nội dung trên nền tảng mạng xã hội {platform}.\n" f"Nhiệm vụ của bạn là lập kế hoạch nội dung chi tiết trong vòng 10 ngày cho lĩnh vực/ngách '{niche}'.\n" "Hãy đảm bảo mỗi ngày là một chủ đề cụ thể, không trùng lặp, hướng tới người xem thực tế.\n" "Trả về kết quả dưới dạng JSON chứa khóa 'calendar' là mảng chứa 10 đối tượng. Mỗi đối tượng gồm: 'day' (ngày thứ mấy, từ 1-10), 'title' (tiêu đề đề xuất), 'hook_idea' (ý tưởng câu kéo người xem), 'target_keywords' (các từ khóa tìm kiếm chính), 'angle' (góc tiếp cận độc đáo để chiến thắng đối thủ).\n\n" "Cấu trúc JSON mong muốn:\n" "{\n" " 'calendar': [\n" " {\n" " 'day': 1,\n" " 'title': 'Tiêu đề ngày 1',\n" " 'hook_idea': 'Ý tưởng hook...',\n" " 'target_keywords': 'từ khóa 1, từ khóa 2',\n" " 'angle': 'Góc nhìn...'\n" " }\n" " ]\n" "}" ) try: response, used_model = generate_with_fallback( model_id, niche, config=types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json" ) ) raw_text = response.text.strip() start_idx = raw_text.find('{') end_idx = raw_text.rfind('}') if start_idx != -1 and end_idx != -1: raw_text = raw_text[start_idx:end_idx+1] result = json.loads(raw_text) return jsonify({'success': True, 'calendar': result.get('calendar', [])}) except Exception as e: return jsonify({'error': f"Lỗi lập kế hoạch nội dung: {_err_msg(str(e))}"}), 500 # ============================================== # TOOL: AI BATCH GENERATOR (SQL SERVER CONNECTED) # ============================================== import uuid import threading import pymssql def get_db_connection(): return pymssql.connect( server='db57971.public.databaseasp.net', user='db57971', password='Qn4@3F#p!9fD', database='db57971', charset='utf8' ) def init_db(): try: conn = get_db_connection() cursor = conn.cursor() # Create users table if not exists cursor.execute(""" IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='users' AND xtype='U') CREATE TABLE users ( id INT IDENTITY(1,1) PRIMARY KEY, email VARCHAR(100) NOT NULL UNIQUE, password_hash VARCHAR(255) NOT NULL, role VARCHAR(20) NOT NULL DEFAULT 'user', created_at DATETIME DEFAULT GETDATE() ) """) # Check and alter users table if role column does not exist cursor.execute(""" IF NOT EXISTS (SELECT * FROM sys.columns WHERE object_id = OBJECT_ID('users') AND name = 'role') ALTER TABLE users ADD role VARCHAR(20) NOT NULL DEFAULT 'user' """) # Create user_permissions table if not exists cursor.execute(""" IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='user_permissions' AND xtype='U') CREATE TABLE user_permissions ( id INT IDENTITY(1,1) PRIMARY KEY, user_id INT FOREIGN KEY REFERENCES users(id) ON DELETE CASCADE, tool_path VARCHAR(100) NOT NULL, CONSTRAINT UQ_user_permissions UNIQUE (user_id, tool_path) ) """) # Create batch_jobs table if not exists cursor.execute(""" IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='batch_jobs' AND xtype='U') CREATE TABLE batch_jobs ( id VARCHAR(50) PRIMARY KEY, mode VARCHAR(10) NOT NULL, status VARCHAR(20) NOT NULL, created_at DATETIME DEFAULT GETDATE() ) """) # Create batch_prompts table if not exists cursor.execute(""" IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='batch_prompts' AND xtype='U') CREATE TABLE batch_prompts ( id INT IDENTITY(1,1) PRIMARY KEY, batch_id VARCHAR(50) FOREIGN KEY REFERENCES batch_jobs(id) ON DELETE CASCADE, prompt_text NVARCHAR(MAX) NOT NULL, status VARCHAR(20) NOT NULL, output_url VARCHAR(2083) NULL, error_message NVARCHAR(MAX) NULL, created_at DATETIME DEFAULT GETDATE() ) """) # Seed default admin user 'admin' cursor.execute("SELECT id FROM users WHERE email = 'admin'") if not cursor.fetchone(): hashed_pwd = generate_password_hash('!Abc123') cursor.execute("INSERT INTO users (email, password_hash, role) VALUES (%s, %s, %s)", ('admin', hashed_pwd, 'admin')) print("Default admin user 'admin' seeded successfully.") else: cursor.execute("UPDATE users SET role = 'admin' WHERE email = 'admin'") # Also seed 'admin@gmail.com' for compatibility cursor.execute("SELECT id FROM users WHERE email = 'admin@gmail.com'") if not cursor.fetchone(): hashed_pwd = generate_password_hash('!Abc123') cursor.execute("INSERT INTO users (email, password_hash, role) VALUES (%s, %s, %s)", ('admin@gmail.com', hashed_pwd, 'admin')) print("Default admin user admin@gmail.com seeded successfully.") else: cursor.execute("UPDATE users SET role = 'admin' WHERE email = 'admin@gmail.com'") conn.commit() conn.close() print("SQL Server initialized and tables verified successfully.") except Exception as e: print(f"Error initializing SQL Server database: {e}") # Run database setup on startup init_db() def run_batch_replicate_db(batch_id, mode, replicate_token): # Model configuration if mode == 'video': model_version = "748dae3f714ccf25a72064119d854203f1917b2b8c56fa769f3e498c0d9c491f" else: model_version = "f2030794592a6c15b7ba8d6c70b8f1074e2d3126b8637953267b2d56a2bb2153" headers = { 'Authorization': f'Token {replicate_token}', 'Content-Type': 'application/json' } try: # Fetch all prompts to process conn = get_db_connection() cursor = conn.cursor(as_dict=True) cursor.execute("SELECT id, prompt_text FROM batch_prompts WHERE batch_id = %s", (batch_id,)) prompts = cursor.fetchall() conn.close() except Exception as e: print(f"Error fetching prompts from DB: {e}") return for prompt_item in prompts: prompt_id = prompt_item['id'] prompt_text = prompt_item['prompt_text'] # 1. Update status to 'processing' try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE batch_prompts SET status = 'processing' WHERE id = %d", (prompt_id,)) conn.commit() conn.close() except Exception as e: print(f"Error updating status to processing: {e}") # Prepare payload payload = { "version": model_version, "input": {"prompt": prompt_text} } if mode == 'video': payload["input"]["num_inference_steps"] = 4 payload["input"]["guidance_scale"] = 1.0 try: # Create prediction request req = urllib.request.Request( "https://api.replicate.com/v1/predictions", data=json.dumps(payload).encode('utf-8'), headers=headers, method='POST' ) with urllib.request.urlopen(req, timeout=10) as response: resp_data = json.loads(response.read().decode('utf-8')) prediction_id = resp_data.get('id') if not prediction_id: raise Exception("Không nhận được prediction_id từ Replicate") # Polling loop poll_url = f"https://api.replicate.com/v1/predictions/{prediction_id}" success = False for _ in range(30): # max 60 seconds polling (30 * 2) time.sleep(2) poll_req = urllib.request.Request(poll_url, headers=headers) with urllib.request.urlopen(poll_req, timeout=10) as poll_resp: status_data = json.loads(poll_resp.read().decode('utf-8')) pred_status = status_data.get('status') if pred_status == 'succeeded': output = status_data.get('output') output_url = None if isinstance(output, list) and len(output) > 0: output_url = output[0] elif isinstance(output, str): output_url = output else: output_url = str(output) # Update status to 'done' in DB conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE batch_prompts SET status = 'done', output_url = %s WHERE id = %d", (output_url, prompt_id)) conn.commit() conn.close() success = True break elif pred_status in ['failed', 'canceled']: error_msg = status_data.get('error') or "Đã bị hủy hoặc lỗi" # Update status to 'failed' in DB conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE batch_prompts SET status = 'failed', error_message = %s WHERE id = %d", (error_msg, prompt_id)) conn.commit() conn.close() success = True break if not success: conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE batch_prompts SET status = 'failed', error_message = 'Quá thời gian chờ phản hồi (Timeout)' WHERE id = %d", (prompt_id,)) conn.commit() conn.close() except Exception as e: # Update status to 'failed' in DB try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE batch_prompts SET status = 'failed', error_message = %s WHERE id = %d", (str(e), prompt_id)) conn.commit() conn.close() except Exception as dbe: print(f"Error updating failure in DB: {dbe}") # Mark batch job as completed try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE batch_jobs SET status = 'completed' WHERE id = %s", (batch_id,)) conn.commit() conn.close() except Exception as e: print(f"Error updating batch job completion: {e}") @app.route('/batch-generator') def tool_batch_generator(): return render_template('tools/batch_generator/index.html') @app.route('/api/batch-generator/start', methods=['POST']) def api_batch_generator_start(): data = request.json or {} raw_prompts = data.get('prompts', []) mode = data.get('mode', 'image') # 'image' or 'video' replicate_token = data.get('replicate_token', '').strip() if not replicate_token: return jsonify({'error': 'Vui lòng cung cấp Replicate API Token'}), 400 if not raw_prompts or not isinstance(raw_prompts, list): return jsonify({'error': 'Danh sách prompt không hợp lệ'}), 400 prompts_list = [str(p).strip() for p in raw_prompts if str(p).strip()] if not prompts_list: return jsonify({'error': 'Không tìm thấy prompt hợp lệ nào'}), 400 batch_id = str(uuid.uuid4()) try: # Insert batch job and prompts in database conn = get_db_connection() cursor = conn.cursor() cursor.execute("INSERT INTO batch_jobs (id, mode, status) VALUES (%s, %s, %s)", (batch_id, mode, 'running')) for prompt_text in prompts_list: cursor.execute("INSERT INTO batch_prompts (batch_id, prompt_text, status) VALUES (%s, %s, %s)", (batch_id, prompt_text, 'queued')) conn.commit() conn.close() except Exception as e: return jsonify({'error': f'Lỗi lưu thông tin vào CSDL SQL Server: {e}'}), 500 # Start processing thread thread = threading.Thread( target=run_batch_replicate_db, args=(batch_id, mode, replicate_token) ) thread.daemon = True thread.start() return jsonify({ 'success': True, 'batch_id': batch_id, 'total': len(prompts_list) }) @app.route('/api/batch-generator/status/') def api_batch_generator_status(batch_id): try: conn = get_db_connection() cursor = conn.cursor(as_dict=True) cursor.execute("SELECT status, mode FROM batch_jobs WHERE id = %s", (batch_id,)) job = cursor.fetchone() if not job: conn.close() return jsonify({'error': 'Không tìm thấy mã Batch'}), 404 cursor.execute("SELECT prompt_text AS prompt, status, output_url, error_message AS error FROM batch_prompts WHERE batch_id = %s ORDER BY id ASC", (batch_id,)) prompts = cursor.fetchall() conn.close() return jsonify({ 'success': True, 'status': job['status'], 'mode': job['mode'], 'prompts': prompts }) except Exception as e: return jsonify({'error': f'Lỗi truy vấn CSDL SQL Server: {e}'}), 500 # ============================================== # AUTHENTICATION: LOGIN & LOGOUT SYSTEM # ============================================== AVAILABLE_TOOLS = [ {'path': '/tts', 'name': 'Text to Speech', 'group': 'Âm thanh & Lồng tiếng'}, {'path': '/audio-story-studio', 'name': 'AI Audio Story', 'group': 'Âm thanh & Lồng tiếng'}, {'path': '/podcast-studio', 'name': 'AI Podcast', 'group': 'Âm thanh & Lồng tiếng'}, {'path': '/image-studio', 'name': 'Image Studio', 'group': 'Hình ảnh & Video AI'}, {'path': '/video-ai', 'name': 'Video AI Studio', 'group': 'Hình ảnh & Video AI'}, {'path': '/shorts-generator', 'name': 'AI Shorts', 'group': 'Hình ảnh & Video AI'}, {'path': '/comic-creator', 'name': 'AI Comic Book', 'group': 'Hình ảnh & Video AI'}, {'path': '/video-chapters', 'name': 'Video Chapters', 'group': 'Hình ảnh & Video AI'}, {'path': '/thumbnail-designer', 'name': 'Thumbnail Designer', 'group': 'Hình ảnh & Video AI'}, {'path': '/batch-generator', 'name': 'AI Batch Generator', 'group': 'Hình ảnh & Video AI'}, {'path': '/prompt-generator', 'name': 'Prompt Studio', 'group': 'Trợ lý nội dung'}, {'path': '/omnichannel-writer', 'name': 'Omnichannel Writer', 'group': 'Trợ lý nội dung'}, {'path': '/script-hook-generator', 'name': 'Script Optimizer', 'group': 'Trợ lý nội dung'}, {'path': '/content-planner', 'name': 'Content Planner', 'group': 'Trợ lý nội dung'}, {'path': '/translator', 'name': 'Translator Pro', 'group': 'Dịch thuật & Tài liệu'}, {'path': '/pdf', 'name': 'Smart PDF', 'group': 'Dịch thuật & Tài liệu'} ] @app.before_request def require_login(): # Bypass login check for dashboard, static assets, and login endpoint if request.path == '/login' or request.path == '/' or request.path.startswith('/static/'): return None # Check if user session exists if 'user_id' not in session: if request.path.startswith('/api/'): return jsonify({'error': 'Unauthorized. Please login.'}), 401 return redirect('/login') # Admin has all permissions if session.get('role') == 'admin': return None # Role check for administrative routes if request.path.startswith('/admin/'): return "403 Forbidden: Quyền truy cập bị từ chối.", 403 # For normal users, check tool permissions path = request.path if path.startswith('/api/'): parts = path.split('/') if len(parts) > 2: path = '/' + parts[2] tool_prefixes = [t['path'] for t in AVAILABLE_TOOLS] requested_tool = None for prefix in tool_prefixes: if path.startswith(prefix): requested_tool = prefix break if requested_tool: user_id = session.get('user_id') try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT id FROM user_permissions WHERE user_id = %d AND tool_path = %s", (user_id, requested_tool)) has_permission = cursor.fetchone() conn.close() if not has_permission: if request.path.startswith('/api/'): return jsonify({'error': 'Quyền truy cập bị từ chối. Bạn không có quyền sử dụng công cụ này.'}), 403 return "403 Forbidden: Bạn không có quyền truy cập công cụ này. Vui lòng liên hệ Admin.", 403 except Exception as e: return f"Lỗi xác thực quyền hạn: {e}", 500 @app.route('/login', methods=['GET', 'POST']) def route_login(): if request.method == 'POST': email = request.form.get('email', '').strip() password = request.form.get('password', '') if not email or not password: return render_template('login.html', error='Vui lòng điền đầy đủ email và mật khẩu.') try: conn = get_db_connection() cursor = conn.cursor(as_dict=True) cursor.execute("SELECT id, email, password_hash, role FROM users WHERE email = %s", (email,)) user = cursor.fetchone() conn.close() if user and check_password_hash(user['password_hash'], password): session['user_id'] = user['id'] session['email'] = user['email'] session['role'] = user['role'] return redirect('/') else: return render_template('login.html', error='Email hoặc mật khẩu không chính xác.') except Exception as e: return render_template('login.html', error=f'Lỗi kết nối cơ sở dữ liệu: {e}') return render_template('login.html') @app.route('/logout') def route_logout(): session.clear() return redirect('/login') # ============================================== # ADMIN PORTAL: ACCOUNT MANAGEMENT CRUD # ============================================== @app.route('/admin/users') def admin_users_list(): try: conn = get_db_connection() cursor = conn.cursor(as_dict=True) cursor.execute("SELECT id, email, role, created_at FROM users ORDER BY id ASC") user_list = cursor.fetchall() for u in user_list: # Convert datetime to string so Jinja2 tojson can serialize it if u.get('created_at') and not isinstance(u['created_at'], str): u['created_at'] = str(u['created_at']) cursor.execute("SELECT tool_path FROM user_permissions WHERE user_id = %s", (u['id'],)) u['permissions'] = [row['tool_path'] for row in cursor.fetchall()] conn.close() return render_template('tools/admin_users/index.html', users=user_list, available_tools=AVAILABLE_TOOLS) except Exception as e: return f"Lỗi truy vấn CSDL: {e}", 500 @app.route('/admin/users/create', methods=['POST']) def admin_users_create(): email = request.form.get('email', '').strip() password = request.form.get('password', '') role = request.form.get('role', 'user').strip() if not email or not password: return "Email và mật khẩu là bắt buộc.", 400 try: conn = get_db_connection() cursor = conn.cursor() # Check duplicates cursor.execute("SELECT id FROM users WHERE email = %s", (email,)) if cursor.fetchone(): conn.close() return "Email này đã được sử dụng.", 400 hashed_pwd = generate_password_hash(password) cursor.execute("INSERT INTO users (email, password_hash, role) VALUES (%s, %s, %s)", (email, hashed_pwd, role)) # Get new user id to insert permissions cursor.execute("SELECT id FROM users WHERE email = %s", (email,)) new_user = cursor.fetchone() if new_user and role == 'user': new_user_id = new_user[0] permitted_tools = request.form.getlist('tools') for t in permitted_tools: cursor.execute("INSERT INTO user_permissions (user_id, tool_path) VALUES (%s, %s)", (new_user_id, t)) conn.commit() conn.close() return redirect('/admin/users') except Exception as e: return f"Lỗi cơ sở dữ liệu: {e}", 500 @app.route('/admin/users/update/', methods=['POST']) def admin_users_update(user_id): email = request.form.get('email', '').strip() role = request.form.get('role', 'user').strip() password = request.form.get('password', '') if not email: return "Email là bắt buộc.", 400 try: conn = get_db_connection() cursor = conn.cursor() # Update details if password: hashed_pwd = generate_password_hash(password) cursor.execute("UPDATE users SET email = %s, role = %s, password_hash = %s WHERE id = %d", (email, role, hashed_pwd, user_id)) else: cursor.execute("UPDATE users SET email = %s, role = %s WHERE id = %d", (email, role, user_id)) # Update permissions cursor.execute("DELETE FROM user_permissions WHERE user_id = %d", (user_id,)) if role == 'user': permitted_tools = request.form.getlist('tools') for t in permitted_tools: cursor.execute("INSERT INTO user_permissions (user_id, tool_path) VALUES (%s, %s)", (user_id, t)) # If modifying own role, update session if session.get('user_id') == user_id: session['role'] = role session['email'] = email conn.commit() conn.close() return redirect('/admin/users') except Exception as e: return f"Lỗi cơ sở dữ liệu: {e}", 500 @app.route('/admin/users/delete/', methods=['POST']) def admin_users_delete(user_id): # Prevent self deletion if session.get('user_id') == user_id: return "Không được tự xóa tài khoản chính mình!", 400 try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("DELETE FROM users WHERE id = %d", (user_id,)) conn.commit() conn.close() return redirect('/admin/users') except Exception as e: return f"Lỗi cơ sở dữ liệu: {e}", 500 if __name__ == '__main__': app.run(debug=True, port=5000)