# -*- coding: utf-8 -*-
import requests
import os
import subprocess
import re
import asyncio
import random
import json
import time
import base64
import sys
try:
from Crypto.Cipher import AES
CRYPTO_AVAILABLE = True
except ImportError:
AES = None
CRYPTO_AVAILABLE = False
try:
from config import APP_CRYPTO_KEY
except Exception:
APP_CRYPTO_KEY = ""
from playwright.sync_api import sync_playwright
from db import get_conn
from config import STORAGE_DIR
from flask import Flask, request, jsonify, send_from_directory, redirect
from bs4 import BeautifulSoup
from PIL import Image, ImageDraw, ImageFont, ImageFilter
os.makedirs(STORAGE_DIR, exist_ok=True)
try:
import edge_tts
EDGE_TTS_AVAILABLE = True
except ImportError:
EDGE_TTS_AVAILABLE = False
app = Flask(__name__)
def normalize_aes_key(key_text):
key = str(key_text or "").encode("utf-8")
if len(key) >= 32:
return key[:32]
return key.ljust(32, b"\0")
def php_openssl_decrypt(encrypted_text):
if not encrypted_text:
return ""
if not CRYPTO_AVAILABLE:
raise Exception("pycryptodome 설치 필요: pip install pycryptodome")
raw = base64.b64decode(str(encrypted_text), validate=True)
iv = raw[:16]
cipher_text = raw[16:]
cipher = AES.new(
normalize_aes_key(APP_CRYPTO_KEY),
AES.MODE_CBC,
iv
)
decrypted = cipher.decrypt(cipher_text)
pad_len = decrypted[-1]
if pad_len < 1 or pad_len > 16:
return ""
return decrypted[:-pad_len].decode("utf-8", errors="ignore")
def naver_session_storage_path(realtor_id):
session_dir = os.path.join(STORAGE_DIR, "naver_sessions")
os.makedirs(session_dir, exist_ok=True)
return os.path.join(
session_dir,
f"naver_session_{realtor_id}.json"
)
def mybox_admin_session_storage_path():
session_dir = os.path.join(STORAGE_DIR, "naver_sessions")
os.makedirs(session_dir, exist_ok=True)
return os.path.join(
session_dir,
"naver_session_admin_mybox.json"
)
API_KEY = "honghee-shorts-secret-2026"
OLLAMA_URL = "http://127.0.0.1:11434/api/generate"
MODEL_NAME = "gemma3"
BASE_DIR = "D:/honghee/blog_api"
OUTPUT_DIR = os.path.join(BASE_DIR, "outputs")
TEMP_DIR = os.path.join(BASE_DIR, "temp")
BGM_DIR = os.path.join(BASE_DIR, "bgm")
FONT_PATH = "C:/Windows/Fonts/malgun.ttf"
FONT_BOLD_PATH = "C:/Windows/Fonts/malgunbd.ttf"
FFMPEG_PATH = "D:/ffmpeg/bin/ffmpeg.exe"
FFPROBE_PATH = "D:/ffmpeg/bin/ffprobe.exe"
BGM_VOLUME = 0.18
# 파이썬 서버 로컬 MYBOX 템플릿/생성 폴더
MYBOX_LOCAL_ROOT = "D:/honghee/blog_api/mybox"
MYBOX_HEADER_BG_DIR = os.path.join(MYBOX_LOCAL_ROOT, "header_images")
MYBOX_GENERATED_HEADER_DIR = os.path.join(MYBOX_LOCAL_ROOT, "generated", "header_images")
MYBOX_V1_IMAGES_DIR = os.path.join(MYBOX_LOCAL_ROOT, "v1_images")
SHORTS_FALLBACK_FOLDER_MAP = {
"아파트": ["아파트", "apt", "apartment"],
"아파트분양권": ["아파트", "apt", "apartment"],
"오피스텔": ["오피스텔", "officetel"],
"빌라": ["빌라", "villa", "주택"],
"단독주택": ["단독다가구", "주택", "house", "villa"],
"다가구": ["단독다가구", "주택", "house", "villa"],
"상가": ["상가", "store", "retail"],
"사무실": ["사무실", "office"],
"토지": ["토지", "land"],
"공장": ["공장", "factory"],
"창고": ["창고", "warehouse", "공장"],
"건물": ["건물", "building"],
"기타": ["기타", "etc"],
}
HEADER_PROPERTY_TYPE_MAP = {
"apt": "apt",
"apartment": "apt",
"아파트": "apt",
"officetel": "officetel",
"오피스텔": "officetel",
"villa": "villa",
"빌라": "villa",
"store": "store",
"상가": "store",
"land": "land",
"토지": "land",
"etc": "etc",
"기타": "etc",
}
TTS_VOICES = [
"ko-KR-SunHiNeural",
"ko-KR-InJoonNeural"
]
os.makedirs(OUTPUT_DIR, exist_ok=True)
os.makedirs(TEMP_DIR, exist_ok=True)
os.makedirs(BGM_DIR, exist_ok=True)
def check_api_key(req):
return req.headers.get("X-API-KEY") == API_KEY
def unauthorized_response():
return jsonify({
"ok": False,
"error": "Unauthorized"
}), 401
def get_font(path, size):
try:
return ImageFont.truetype(path, size)
except Exception:
return ImageFont.truetype(FONT_PATH, size)
def make_prompt(data):
region = data.get("region", "지역 정보 없음")
property_type = data.get("property_type", "부동산 매물")
keywords = data.get("keywords", [])
purpose = data.get("purpose", "블로그 유입 및 상담 문의 유도")
post_title = data.get("post_title", "")
post_summary = data.get("post_summary", "")
post_url = data.get("post_url", "")
keyword_text = ", ".join(keywords) if isinstance(keywords, list) else str(keywords)
return f"""
너는 네이버 부동산 블로그 글을 유튜브 쇼츠/인스타 릴스용 대본으로 바꾸는 전문 작가다.
아래 입력 정보만 기준으로 30초 세로형 숏츠 대본을 작성해줘.
가장 중요한 원칙:
1. 입력 정보에 없는 내용은 절대 추가하지 마.
2. 매물 유형을 바꾸지 마. 토지는 토지로, 아파트는 아파트로, 상가는 상가로 표현해.
3. 토지/대지/임야 매물은 단독주택, 전원주택, 주택, 집, 거주공간, 내 집 마련으로 표현하지 마.
4. 토지/대지/임야 매물은 "대지", "토지", "현장 확인", "입지 조건", "면적 조건" 중심으로만 설명해.
5. "지을 수 있다", "건축 가능", "개발 가능", "투자 가치 확실" 같은 표현은 입력 정보에 명확히 있을 때만 사용해.
6. 가격, 면적, 위치, 학교, 상권, 방향, 층수는 입력값 그대로만 사용해.
7. 없는 교통, 학군, 리모델링, 내부 상태, 입주 가능 여부를 상상해서 쓰지 마.
8. 확정 수익, 급매 보장, 안전한 투자, 무조건 오른다는 표현은 금지해.
9. 과장 광고, 허위 매물처럼 보이는 표현은 금지해.
작성 원칙:
1. 실제 사람이 말하는 구어체로 작성해.
2. 부동산 상담사가 고객에게 자연스럽게 설명하듯이 말해.
3. 첫 문장은 시청자가 멈춰 보게 만드는 후킹 문장으로 시작해.
4. 게시글 제목과 요약 내용을 중심으로 작성해.
5. 마지막은 등록 중개사무소 상담 문의로 자연스럽게 연결해.
6. 220~320자 이내로 작성해.
7. 문장은 짧게 작성해.
8. 최종 대본 1개만 작성해.
절대 쓰지 마:
- 예시 1, 예시 2, 후보 1, 후보 2
- 별표 강조 표시: **, *
- 괄호 설명: (후킹 문장), (배경음악), (장면 설명)
- 대괄호 설명: [자막], [나레이션]
- 시간 표시: 5초, 5~10초
- 제작 지시어: 장면, 자막, 나레이션, 컷, 화면, BGM
- 마크다운 제목: ##, ###
- 해시태그: #부동산 #매물
- 제목:, 설명:, 대본:, 해시태그: 같은 라벨
- 따옴표로 감싼 대화체 예시 여러 개
- 자세한 정보는 블로그에서 확인하세요
- 블로그 기반 실제 매물 안내
- 매물 체크
게시글 분석 정보:
- 게시글 제목: {post_title}
- 게시글 요약 및 원문 정보: {post_summary}
- 게시글 URL: {post_url}
- 지역/주제: {region}
- 매물 유형: {property_type}
- 핵심 키워드: {keyword_text}
- 목적: {purpose}
출력 형식:
대본 문장만 출력해줘.
"""
def clean_script_text(text):
text = str(text or "")
text = text.replace("\r", "\n")
text = text.replace("**", "")
text = text.replace("*", "")
text = re.sub(r"\([^)]*\)", "", text)
text = re.sub(r"\[[^\]]*\]", "", text)
text = re.sub(r"\{[^}]*\}", "", text)
text = re.sub(r"\d+\s*[~\-]\s*\d+\s*초", "", text)
text = re.sub(r"\d+\s*초", "", text)
text = re.sub(r"(후킹 문장|배경음악|장면|자막|나레이션|컷|화면|BGM)\s*[:：]?", "", text)
text = re.sub(r"(제목|설명|해시태그|대본)\s*[:：]", "", text)
for unwanted in [
"자세한 정보는 블로그에서 확인하세요",
"블로그에서 자세히 확인해보세요",
"블로그 기반 실제 매물 안내",
"매물 체크",
]:
text = text.replace(unwanted, "")
lines = []
for line in text.split("\n"):
line = line.strip()
if not line:
continue
if line.startswith("#") or "#" in line:
continue
line = re.sub(r"^\s*\d+\.\s*", "", line)
line = re.sub(r"^\s*[-•]\s*", "", line)
if line in ["후킹", "본문", "마무리", "CTA"]:
continue
lines.append(line)
text = "\n".join(lines)
text = re.sub(r"\n{2,}", "\n", text)
text = re.sub(r"[ \t]+", " ", text)
return text.strip()
def sanitize_realestate_script_for_type(script, property_type="", title=""):
script = clean_script_text(script)
source = f"{property_type} {title} {script}"
land_words = ["토지", "대지", "임야", "전", "답", "잡종지", "공장용지", "창고용지"]
is_land = any(w in source for w in land_words)
if is_land:
replacements = {
"단독 주택": "토지",
"단독주택": "토지",
"전원주택": "토지",
"주택까지": "토지 조건까지",
"주택": "토지",
"내 집 마련": "토지 검토",
"집 마련": "토지 검토",
"당신의 집에서": "해당 입지에서",
"꿈꿔왔던 공간": "관심 있게 볼 수 있는 조건",
"직접 만져볼 수 있어": "현장에서 확인해볼 수 있습니다",
"생활도 즐겁고": "생활권도 함께 확인할 수 있고",
}
for a, b in replacements.items():
script = script.replace(a, b)
banned_patterns = [
r"입주\s*즉시[^.!?\n]*",
r"살기\s*편[^.!?\n]*",
r"거주\s*공간[^.!?\n]*",
]
for pattern in banned_patterns:
script = re.sub(pattern, "", script)
script = re.sub(r"\s+", " ", script).strip()
return script
def clean_tts_text(text):
text = clean_script_text(text)
text = text.replace("#", "")
text = text.replace("@", "")
text = text.replace("~", "에서 ")
text = text.replace("/", " ")
text = text.replace("|", " ")
text = re.sub(r"[\"'`]", "", text)
text = re.sub(r"\s+", " ", text)
return text.strip()
def wrap_text_by_pixel(draw, text, font, max_width):
chars = list(text.strip())
lines = []
current = ""
for ch in chars:
test_line = current + ch
bbox = draw.textbbox((0, 0), test_line, font=font)
test_width = bbox[2] - bbox[0]
if test_width <= max_width:
current = test_line
else:
if current:
lines.append(current)
current = ch
if current:
lines.append(current)
return lines
def split_script(script):
script = clean_script_text(script)
sentences = re.split(r"(?<=[.!?。？！])\s+|\n+", script)
chunks = []
current = ""
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
if len(current) + len(sentence) <= 48:
current = (current + " " + sentence).strip()
else:
if current:
chunks.append(current)
current = sentence
if current:
chunks.append(current)
return chunks[:10] if chunks else ["매물 정보를 안내해드립니다."]
def get_mobile_naver_blog_url(post_url):
post_url = str(post_url or "").strip()
if not post_url:
return ""
match = re.search(r"blog\.naver\.com/([^/]+)/(\d+)", post_url)
if match:
blog_id = match.group(1)
log_no = match.group(2)
return f"https://m.blog.naver.com/PostView.naver?blogId={blog_id}&logNo={log_no}"
return post_url
def is_probably_real_blog_photo_url(url):
if not url:
return False
lower = url.lower()
skip_words = [
"profile", "sticker", "emoticon", "icon", "blank", "spacer",
"logo", "button", "btn", "common", "gnb", "spi", "comment",
"default", "theme", "banner", "adpost", "w80", "w40", "w48",
"w56", "thumbtype", "userprofile"
]
if any(word in lower for word in skip_words):
return False
return (
"blogfiles.pstatic.net" in lower
or "postfiles.pstatic.net" in lower
or "mblogthumb-phinf.pstatic.net" in lower
or "phinf.pstatic.net" in lower
or "landthumb-phinf.pstatic.net" in lower
or "akcdn-pjjh0305cdn.cafe24img.com" in lower
)
def extract_blog_images(post_url, limit=60):
mobile_url = get_mobile_naver_blog_url(post_url)
if not mobile_url:
return []
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
"Referer": "https://m.blog.naver.com/"
}
image_urls = []
try:
res = requests.get(mobile_url, headers=headers, timeout=20)
res.raise_for_status()
html = res.text
soup = BeautifulSoup(html, "html.parser")
for meta_selector in [
("property", "og:image"),
("name", "twitter:image")
]:
meta = soup.find("meta", attrs={meta_selector[0]: meta_selector[1]})
if meta and meta.get("content"):
image_urls.append(meta.get("content"))
for img in soup.find_all("img"):
for attr in ["src", "data-src", "data-lazy-src", "data-original", "data-url", "data-image-url"]:
src = img.get(attr)
if not src:
continue
if src.startswith("//"):
src = "https:" + src
if is_probably_real_blog_photo_url(src):
image_urls.append(src)
patterns = [
r'https?:\\?/\\?/postfiles\.pstatic\.net[^"\']+',
r'https?:\\?/\\?/blogfiles\.pstatic\.net[^"\']+',
r'https?:\\?/\\?/mblogthumb-phinf\.pstatic\.net[^"\']+',
r'https?:\\?/\\?/phinf\.pstatic\.net[^"\']+'
]
for pattern in patterns:
for url in re.findall(pattern, html):
url = url.replace("\\/", "/")
url = url.replace("\\u0026", "&")
url = url.replace("&", "&")
if is_probably_real_blog_photo_url(url):
image_urls.append(url)
filtered = []
for url in image_urls:
if url and url.startswith("http"):
url = url.replace("&", "&")
if url not in filtered:
filtered.append(url)
return filtered[:limit]
except Exception as e:
print("extract_blog_images error:", e)
return []
def extract_image_urls_from_text(text, limit=80):
text = str(text or "")
if not text:
return []
patterns = [
r'https?://[^"\'>\s]+?\.(?:jpg|jpeg|png|webp)(?:\?[^"\'>\s]*)?',
r'https?:\\?/\\?/[^"\'>\s]+?\.(?:jpg|jpeg|png|webp)(?:\\?[^"\'>\s]*)?',
]
urls = []
for pattern in patterns:
for url in re.findall(pattern, text, flags=re.IGNORECASE):
url = url.replace("\\/", "/")
url = url.replace("\\u0026", "&")
url = url.replace("&", "&")
if is_probably_real_blog_photo_url(url) and url not in urls:
urls.append(url)
# 중개사 프로필보다 매물/대표/본문 이미지를 우선
def score(u):
lu = u.lower()
s = 0
if "landthumb-phinf" in lu:
s += 50
if "akcdn-pjjh0305cdn" in lu:
s += 45
if "header_images" in lu or "header_" in lu:
s += 35
if "middle_banner" in lu:
s -= 20
if "cropimage" in lu or "profile" in lu:
s -= 60
if "263" in lu or "262" in lu:
s += 10
return -s
urls = sorted(urls, key=score)
return urls[:limit]
def extract_project_image_urls_from_db(article_no, limit=80):
article_no = str(article_no or "").strip()
if not article_no:
return []
urls = []
conn = None
try:
conn = get_conn()
with conn.cursor() as cur:
# 블로그 초안 본문에 있는 매물 이미지/대표이미지/중간배너 이미지
try:
cur.execute("""
SELECT
draft_html,
clipboard_html,
plain_text,
source_json,
header_image_url,
representative_image_url
FROM blog_article_drafts
WHERE article_no = %s
ORDER BY id DESC
LIMIT 1
""", (article_no,))
draft = cur.fetchone() or {}
for key in [
"representative_image_url",
"header_image_url",
"clipboard_html",
"draft_html",
"source_json",
"plain_text",
]:
for url in extract_image_urls_from_text(draft.get(key, ""), limit=limit):
if url not in urls:
urls.append(url)
except Exception as e:
print("extract_project_image_urls_from_db draft error:", e)
# 수집 원본 raw_json/매물설명 등에 남아 있는 네이버 landthumb 이미지
try:
cur.execute("""
SELECT raw_json, article_feature_desc
FROM blog_realtor_articles
WHERE article_no = %s
LIMIT 1
""", (article_no,))
article = cur.fetchone() or {}
for key in ["raw_json", "article_feature_desc"]:
for url in extract_image_urls_from_text(article.get(key, ""), limit=limit):
if url not in urls:
urls.append(url)
except Exception as e:
print("extract_project_image_urls_from_db article error:", e)
except Exception as e:
print("extract_project_image_urls_from_db error:", e)
finally:
try:
if conn:
conn.close()
except Exception:
pass
# 프로필/배너 제거, 매물 이미지 우선
filtered = []
for url in urls:
lower = url.lower()
if "cropimage" in lower or "profile" in lower:
continue
if "middle_banner" in lower:
continue
if url not in filtered:
filtered.append(url)
return filtered[:limit]
def merge_image_urls(primary, fallback, limit=80):
merged = []
for group in [primary or [], fallback or []]:
for url in group:
url = str(url or "").strip().replace("&", "&")
if url and url.startswith("http") and url not in merged:
merged.append(url)
return merged[:limit]
def download_blog_images(image_urls, project_id):
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
"Referer": "https://m.blog.naver.com/"
}
image_paths = []
for idx, url in enumerate(image_urls):
try:
res = requests.get(url, headers=headers, timeout=20)
res.raise_for_status()
if len(res.content) < 25 * 1024:
continue
path = os.path.join(TEMP_DIR, f"blog_img_{project_id}_{idx}.jpg")
with open(path, "wb") as f:
f.write(res.content)
try:
img = Image.open(path)
w, h = img.size
if w < 400 or h < 300:
os.remove(path)
continue
if w * h < 250000:
os.remove(path)
continue
ratio = w / h
if ratio > 4.2 or ratio < 0.18:
os.remove(path)
continue
image_paths.append(path)
except Exception:
if os.path.exists(path):
os.remove(path)
except Exception:
continue
return image_paths
def make_realestate_default_image(project_id):
w, h = 1080, 1035
img = Image.new("RGB", (w, h), (30, 41, 59))
draw = ImageDraw.Draw(img)
title_font = get_font(FONT_BOLD_PATH, 72)
sub_font = get_font(FONT_PATH, 40)
for y in range(h):
ratio = y / h
r = int(15 * ratio + 30 * (1 - ratio))
g = int(23 * ratio + 41 * (1 - ratio))
b = int(42 * ratio + 59 * (1 - ratio))
draw.line([(0, y), (w, y)], fill=(r, g, b))
draw.rectangle((250, 330, 500, 850), fill=(71, 85, 105))
draw.rectangle((560, 240, 830, 850), fill=(51, 65, 85))
for bx in [290, 370, 600, 690, 780]:
for by in [380, 480, 580, 680]:
draw.rectangle((bx, by, bx + 38, by + 48), fill=(226, 232, 240))
text = "부동산 매물 정보"
bbox = draw.textbbox((0, 0), text, font=title_font)
draw.text(((w - (bbox[2] - bbox[0])) / 2, 120), text, font=title_font, fill=(248, 250, 252))
sub = "블로그 기반 AI 숏츠"
bbox = draw.textbbox((0, 0), sub, font=sub_font)
draw.text(((w - (bbox[2] - bbox[0])) / 2, 220), sub, font=sub_font, fill=(203, 213, 225))
path = os.path.join(TEMP_DIR, f"default_realestate_{project_id}.jpg")
img.save(path)
return path
def make_fallback_background(w, h):
img = Image.new("RGB", (w, h), (15, 23, 42))
draw = ImageDraw.Draw(img)
top = (30, 41, 59)
bottom = (15, 23, 42)
for y in range(h):
ratio = y / h
r = int(top[0] * (1 - ratio) + bottom[0] * ratio)
g = int(top[1] * (1 - ratio) + bottom[1] * ratio)
b = int(top[2] * (1 - ratio) + bottom[2] * ratio)
draw.line([(0, y), (w, y)], fill=(r, g, b))
return img
def make_blurred_background(image_path, w, h):
try:
img = Image.open(image_path).convert("RGB")
img_ratio = img.width / img.height
target_ratio = w / h
if img_ratio > target_ratio:
new_h = h
new_w = int(h * img_ratio)
else:
new_w = w
new_h = int(w / img_ratio)
img = img.resize((new_w, new_h), Image.LANCZOS)
left = (new_w - w) // 2
top = (new_h - h) // 2
img = img.crop((left, top, left + w, top + h))
img = img.filter(ImageFilter.GaussianBlur(22))
overlay = Image.new("RGBA", (w, h), (2, 6, 23, 120))
img = img.convert("RGBA")
img.alpha_composite(overlay)
return img.convert("RGB")
except Exception:
return make_fallback_background(w, h)
def fit_image_to_box(image_path, box_w, box_h):
img = Image.open(image_path).convert("RGB")
img_ratio = img.width / img.height
box_ratio = box_w / box_h
if img_ratio > box_ratio:
new_w = box_w
new_h = int(box_w / img_ratio)
else:
new_h = box_h
new_w = int(box_h * img_ratio)
return img.resize((new_w, new_h), Image.LANCZOS)
def draw_rounded_rectangle(draw, xy, radius, fill, outline=None, width=1):
draw.rounded_rectangle(xy, radius=radius, fill=fill, outline=outline, width=width)
SHORTS_CAPTION_STYLES = ["bottom_card", "center_card", "headline_top", "split_card"]
SHORTS_LAYOUT_STYLES = ["photo_card", "photo_full", "blur_card", "magazine"]
SHORTS_MOTION_STYLES = ["fade", "zoom_in", "zoom_out"]
def pick_shorts_visual_style(index=0):
return {
"layout": random.choice(SHORTS_LAYOUT_STYLES),
"caption": random.choice(SHORTS_CAPTION_STYLES),
"motion": random.choice(SHORTS_MOTION_STYLES),
}
def cover_image_to_canvas(image_path, w, h):
try:
img = Image.open(image_path).convert("RGB")
img_ratio = img.width / img.height
target_ratio = w / h
if img_ratio > target_ratio:
new_h = h
new_w = int(h * img_ratio)
else:
new_w = w
new_h = int(w / img_ratio)
img = img.resize((new_w, new_h), Image.LANCZOS)
left = (new_w - w) // 2
top = (new_h - h) // 2
return img.crop((left, top, left + w, top + h))
except Exception:
return make_fallback_background(w, h)
def paste_image_contained(canvas, image_path, box, radius=34, border=True):
draw = ImageDraw.Draw(canvas, "RGBA")
x1, y1, x2, y2 = box
box_w, box_h = x2 - x1, y2 - y1
draw_rounded_rectangle(
draw,
(x1, y1, x2, y2),
radius,
fill=(15, 23, 42, 218),
outline=(255, 255, 255, 60) if border else None,
width=2
)
if not image_path:
return
try:
photo = fit_image_to_box(image_path, box_w, box_h)
px = x1 + (box_w - photo.width) // 2
py = y1 + (box_h - photo.height) // 2
canvas.paste(photo, (px, py))
except Exception:
pass
def draw_caption_block(draw, text, style, w, h):
text = clean_script_text(text)
body_font = get_font(FONT_BOLD_PATH, 46)
body_font_small = get_font(FONT_BOLD_PATH, 38)
label_font = get_font(FONT_BOLD_PATH, 30)
if style == "center_card":
box = (90, 700, w - 90, 1135)
fill = (0, 0, 0, 205)
text_width = w - 240
max_lines = 6
elif style == "headline_top":
box = (70, 190, w - 70, 510)
fill = (15, 23, 42, 210)
text_width = w - 190
max_lines = 5
elif style == "split_card":
box = (70, 1250, w - 70, 1645)
fill = (255, 255, 255, 235)
text_width = w - 200
max_lines = 5
else:
box = (50, 1365, w - 50, 1690)
fill = (0, 0, 0, 225)
text_width = w - 200
max_lines = 5
x1, y1, x2, y2 = box
text_fill = (15, 23, 42, 255) if style == "split_card" else (255, 255, 255, 255)
accent_fill = (245, 158, 11, 255) if style == "split_card" else (250, 204, 21, 255)
draw_rounded_rectangle(
draw,
box,
38,
fill=fill,
outline=(255, 255, 255, 70),
width=2
)
# 작은 라벨을 랜덤 느낌으로 배치
if style in ["center_card", "bottom_card"]:
draw_rounded_rectangle(
draw,
(x1 + 28, y1 + 26, x1 + 210, y1 + 75),
22,
fill=(244, 63, 94, 235),
)
draw.text((x1 + 50, y1 + 35), "핵심 정보", font=label_font, fill=(255, 255, 255, 255))
paragraphs = [p.strip() for p in str(text).split("\n") if p.strip()]
lines = []
for paragraph in paragraphs:
lines.extend(wrap_text_by_pixel(draw, paragraph, body_font, text_width))
font = body_font
line_height = 68
if len(lines) > max_lines:
font = body_font_small
line_height = 58
lines = []
for paragraph in paragraphs:
lines.extend(wrap_text_by_pixel(draw, paragraph, font, text_width))
visible = lines[:max_lines]
total_h = len(visible) * line_height
y = y1 + int(((y2 - y1) - total_h) / 2)
if style in ["center_card", "bottom_card"]:
y += 22
for i, line in enumerate(visible):
bb = draw.textbbox((0, 0), line, font=font)
tw = bb[2] - bb[0]
x = int((w - tw) / 2)
# 첫 줄은 살짝 강조
fill_color = accent_fill if i == 0 and len(line) <= 18 else text_fill
shadow = (0, 0, 0, 160) if style == "split_card" else (0, 0, 0, 120)
if style != "split_card":
draw.text((x + 2, y + 2), line, font=font, fill=shadow)
draw.text((x, y), line, font=font, fill=fill_color)
y += line_height
def make_info_card_slide(text, index, total, title, card_type="info", brand='AI 부동산 숏츠'):
w, h = 1080, 1920
text = clean_script_text(text)
title = clean_script_text(title or "부동산 매물 안내")
palettes = [
{"top": (15, 23, 42), "bottom": (30, 64, 175), "accent": (250, 204, 21), "chip": (37, 99, 235)},
{"top": (6, 78, 59), "bottom": (15, 23, 42), "accent": (167, 243, 208), "chip": (16, 185, 129)},
{"top": (69, 10, 10), "bottom": (15, 23, 42), "accent": (252, 165, 165), "chip": (220, 38, 38)},
{"top": (76, 29, 149), "bottom": (15, 23, 42), "accent": (221, 214, 254), "chip": (124, 58, 237)},
]
pal = palettes[index % len(palettes)]
img = Image.new("RGB", (w, h), pal["bottom"])
draw = ImageDraw.Draw(img, "RGBA")
for y in range(h):
ratio = y / h
r = int(pal["top"][0] * (1 - ratio) + pal["bottom"][0] * ratio)
g = int(pal["top"][1] * (1 - ratio) + pal["bottom"][1] * ratio)
b = int(pal["top"][2] * (1 - ratio) + pal["bottom"][2] * ratio)
draw.line([(0, y), (w, y)], fill=(r, g, b))
# 배경 장식
for _ in range(10):
cx = random.randint(-100, w + 100)
cy = random.randint(120, h - 220)
radius = random.randint(55, 155)
alpha = random.randint(16, 38)
draw.ellipse((cx - radius, cy - radius, cx + radius, cy + radius), fill=(255, 255, 255, alpha))
badge_font = get_font(FONT_BOLD_PATH, 32)
title_font = get_font(FONT_BOLD_PATH, 58)
body_font = get_font(FONT_BOLD_PATH, 58)
body_small = get_font(FONT_BOLD_PATH, 46)
sub_font = get_font(FONT_PATH, 31)
page_font = get_font(FONT_PATH, 28)
badge_text = "사진 없는 매물도 한눈에"
if card_type == "cta":
badge_text = "상담 문의 안내"
elif card_type == "point":
badge_text = "핵심 포인트"
draw.rounded_rectangle((70, 90, 70 + 420, 152), radius=28, fill=pal["chip"] + (235,))
draw.text((100, 108), badge_text, font=badge_font, fill=(255, 255, 255, 255))
brand_font = get_font(FONT_BOLD_PATH, 28)
brand_text = clean_script_text(brand or "AI 부동산 숏츠")[:18]
bb_brand = draw.textbbox((0, 0), brand_text, font=brand_font)
draw.rounded_rectangle((w - 90 - (bb_brand[2]-bb_brand[0]) - 42, 92, w - 70, 150), radius=25, fill=(0, 0, 0, 92))
draw.text((w - 90 - (bb_brand[2]-bb_brand[0]) - 20, 108), brand_text, font=brand_font, fill=(255, 255, 255, 255))
# 메인 카드
card = (70, 250, w - 70, 1450)
draw.rounded_rectangle(card, radius=52, fill=(255, 255, 255, 236), outline=(255, 255, 255, 120), width=2)
# 제목
title_lines = wrap_text_by_pixel(draw, title[:54], title_font, 850)
ty = 340
for line in title_lines[:2]:
bb = draw.textbbox((0, 0), line, font=title_font)
tw = bb[2] - bb[0]
draw.text(((w - tw) / 2, ty), line, font=title_font, fill=(15, 23, 42, 255))
ty += 76
draw.line((150, ty + 25, w - 150, ty + 25), fill=(203, 213, 225, 255), width=2)
# 본문
lines = []
for paragraph in text.split("\n"):
paragraph = paragraph.strip()
if paragraph:
lines.extend(wrap_text_by_pixel(draw, paragraph, body_font, 820))
font = body_font
line_height = 78
if len(lines) > 6:
font = body_small
line_height = 64
lines = []
for paragraph in text.split("\n"):
paragraph = paragraph.strip()
if paragraph:
lines.extend(wrap_text_by_pixel(draw, paragraph, font, 820))
visible = lines[:7]
total_h = len(visible) * line_height
y = max(620, 870 - total_h // 2)
for i, line in enumerate(visible):
bb = draw.textbbox((0, 0), line, font=font)
tw = bb[2] - bb[0]
color = pal["chip"] + (255,) if i == 0 else (15, 23, 42, 255)
draw.text(((w - tw) / 2, y), line, font=font, fill=color)
y += line_height
# 하단 CTA
draw.rounded_rectangle((90, 1580, w - 90, 1690), radius=34, fill=(0, 0, 0, 165))
cta = "등록 중개사무소로 문의해 주세요"
if card_type == "cta":
cta = "현재 매물 여부는 상담 전 꼭 확인하세요"
bb = draw.textbbox((0, 0), cta, font=sub_font)
draw.text(((w - (bb[2] - bb[0])) / 2, 1618), cta, font=sub_font, fill=(255, 255, 255, 255))
draw.text((900, 1770), f"{index + 1}/{total}", font=page_font, fill=(226, 232, 240, 255))
path = os.path.join(TEMP_DIR, f"info_card_{index}.png")
img.save(path)
return path
def build_card_chunks(chunks, min_count=5):
chunks = [clean_script_text(x) for x in (chunks or []) if clean_script_text(x)]
if not chunks:
chunks = ["매물의 핵심 조건을 카드형으로 안내드립니다."]
result = chunks[:]
fallback_cards = [
"사진이 없는 매물은 가격, 면적, 위치 조건을 중심으로 확인해보시면 좋습니다.",
"실제 조건은 변동될 수 있으니 상담 전 현재 매물 여부를 먼저 확인하세요.",
"자세한 내용은 블로그 본문과 중개사 상담을 통해 확인하실 수 있습니다.",
]
for card in fallback_cards:
if len(result) >= min_count:
break
result.append(card)
return result[:max(min_count, len(chunks))]
def make_slide_image_v2(text, index, total, title, image_path=None, represent_image_path=None, brand='AI 부동산 숏츠'):
w, h = 1080, 1920
text = clean_script_text(text)
title = clean_script_text(title)
style = pick_shorts_visual_style(index)
bg_source = represent_image_path or image_path
if style["layout"] == "photo_full" and image_path:
canvas = cover_image_to_canvas(image_path, w, h).convert("RGB")
overlay = Image.new("RGBA", (w, h), (0, 0, 0, 70))
canvas = canvas.convert("RGBA")
canvas.alpha_composite(overlay)
canvas = canvas.convert("RGB")
elif bg_source:
canvas = make_blurred_background(bg_source, w, h)
else:
canvas = make_fallback_background(w, h)
draw = ImageDraw.Draw(canvas, "RGBA")
badge_font = get_font(FONT_BOLD_PATH, 30)
title_font = get_font(FONT_BOLD_PATH, 43)
small_font = get_font(FONT_PATH, 27)
# 상단 브랜드
draw_brand_badge(draw, brand, 48, 48)
# 제목
title_lines = wrap_text_by_pixel(draw, title[:54], title_font, 960)
y_title = 130
for line in title_lines[:2]:
draw.text((52, y_title + 2), line, font=title_font, fill=(0, 0, 0, 170))
draw.text((50, y_title), line, font=title_font, fill=(255, 255, 255, 255))
y_title += 56
# 레이아웃별 사진 배치
if style["layout"] == "photo_card":
paste_image_contained(canvas, image_path, (50, 280, 1030, 1290), radius=36)
elif style["layout"] == "blur_card":
paste_image_contained(canvas, image_path, (115, 330, 965, 1175), radius=42)
draw_rounded_rectangle(draw, (150, 1210, 930, 1278), 28, fill=(255, 255, 255, 225))
draw.text((190, 1227), "등록 매물 안내", font=small_font, fill=(15, 23, 42, 255))
elif style["layout"] == "magazine":
paste_image_contained(canvas, image_path, (70, 310, 1010, 980), radius=24)
draw_rounded_rectangle(draw, (95, 1015, 985, 1175), 30, fill=(255, 255, 255, 230))
draw.text((135, 1048), "핵심 조건과 위치를 함께 확인해보세요", font=get_font(FONT_BOLD_PATH, 34), fill=(15, 23, 42, 255))
# photo_full은 배경이 사진이므로 별도 카드 없음
draw_caption_block(draw, text, style["caption"], w, h)
# CTA / 페이지
draw_rounded_rectangle(draw, (50, 1740, 780, 1810), 30, fill=(255, 255, 255, 235))
draw.text((80, 1758), "등록 중개사무소로 문의해 주세요", font=small_font, fill=(15, 23, 42, 255))
draw.text((914, 1755), f"{index + 1}/{total}", font=small_font, fill=(226, 232, 240, 255))
path = os.path.join(TEMP_DIR, f"slide_{index}.png")
canvas.save(path)
return path
def make_video_segment_from_slide(slide_path, segment_path, duration, motion_style="fade"):
duration = max(1.5, float(duration or 3.0))
fps = 30
frames = max(1, int(duration * fps))
if motion_style == "zoom_in":
vf = (
f"scale=1160:2062,"
f"zoompan=z='min(zoom+0.0015,1.12)':d={frames}:s=1080x1920:fps={fps},"
f"fade=t=in:st=0:d=0.25,fade=t=out:st={max(0.1, duration-0.35):.2f}:d=0.25,"
f"format=yuv420p"
)
elif motion_style == "zoom_out":
vf = (
f"scale=1160:2062,"
f"zoompan=z='if(eq(on,1),1.12,max(zoom-0.0015,1.0))':d={frames}:s=1080x1920:fps={fps},"
f"fade=t=in:st=0:d=0.25,fade=t=out:st={max(0.1, duration-0.35):.2f}:d=0.25,"
f"format=yuv420p"
)
else:
vf = (
f"scale=1080:1920,"
f"fade=t=in:st=0:d=0.25,fade=t=out:st={max(0.1, duration-0.35):.2f}:d=0.25,"
f"format=yuv420p"
)
cmd = [
FFMPEG_PATH,
"-y",
"-loop", "1",
"-t", f"{duration:.2f}",
"-i", slide_path,
"-vf", vf,
"-r", str(fps),
"-an",
"-c:v", "libx264",
"-pix_fmt", "yuv420p",
segment_path
]
subprocess.run(cmd, check=True, capture_output=True, text=True)
return segment_path
def make_video_from_slides_with_motion(slide_paths, project_id, per_slide_duration, output_path):
segment_paths = []
motion_used = []
for i, slide_path in enumerate(slide_paths):
motion_style = random.choice(SHORTS_MOTION_STYLES)
segment_path = os.path.join(TEMP_DIR, f"segment_{project_id}_{i}.mp4")
make_video_segment_from_slide(slide_path, segment_path, per_slide_duration, motion_style)
segment_paths.append(segment_path)
motion_used.append(motion_style)
list_path = os.path.join(TEMP_DIR, f"segments_{project_id}.txt")
with open(list_path, "w", encoding="utf-8") as f:
for p in segment_paths:
f.write(f"file '{p.replace(os.sep, '/')}'\n")
cmd = [
FFMPEG_PATH,
"-y",
"-f", "concat",
"-safe", "0",
"-i", list_path,
"-c", "copy",
output_path
]
subprocess.run(cmd, check=True, capture_output=True, text=True)
return motion_used
def make_slide_image(text, index, total, title, image_path=None, represent_image_path=None):
w, h = 1080, 1920
text = clean_script_text(text)
title = clean_script_text(title)
bg_source = represent_image_path or image_path
if bg_source:
canvas = make_blurred_background(bg_source, w, h)
else:
canvas = make_fallback_background(w, h)
draw = ImageDraw.Draw(canvas, "RGBA")
title_font = get_font(FONT_BOLD_PATH, 44)
badge_font = get_font(FONT_BOLD_PATH, 30)
subtitle_font = get_font(FONT_PATH, 28)
body_font = get_font(FONT_BOLD_PATH, 50)
small_font = get_font(FONT_PATH, 28)
draw_rounded_rectangle(draw, (50, 50, 315, 108), 25, fill=(244, 63, 94, 235))
draw.text((75, 65), clean_script_text(title)[:24], font=badge_font, fill=(255, 255, 255, 255))
title_lines = wrap_text_by_pixel(draw, title[:46], title_font, 960)
y_title = 130
for line in title_lines[:2]:
draw.text((52, y_title + 2), line, font=title_font, fill=(0, 0, 0, 180))
draw.text((50, y_title), line, font=title_font, fill=(255, 255, 255, 255))
y_title += 58
photo_x1, photo_y1 = 50, 275
photo_x2, photo_y2 = 1030, 1310
photo_w = photo_x2 - photo_x1
photo_h = photo_y2 - photo_y1
draw_rounded_rectangle(
draw,
(photo_x1, photo_y1, photo_x2, photo_y2),
36,
fill=(15, 23, 42, 215),
outline=(255, 255, 255, 55),
width=2
)
if image_path:
try:
photo = fit_image_to_box(image_path, photo_w, photo_h)
px = photo_x1 + (photo_w - photo.width) // 2
py = photo_y1 + (photo_h - photo.height) // 2
panel = Image.new("RGBA", (photo_w, photo_h), (255, 255, 255, 16))
canvas.paste(panel, (photo_x1, photo_y1), panel)
canvas.paste(photo, (px, py))
except Exception:
pass
caption_x1, caption_y1 = 50, 1365
caption_x2, caption_y2 = 1030, 1690
draw_rounded_rectangle(
draw,
(caption_x1, caption_y1, caption_x2, caption_y2),
38,
fill=(0, 0, 0, 225),
outline=(255, 255, 255, 65),
width=2
)
max_text_width = 880
body_lines = []
for paragraph in str(text).split("\n"):
paragraph = paragraph.strip()
if paragraph:
body_lines.extend(wrap_text_by_pixel(draw, paragraph, body_font, max_text_width))
line_height = 66
max_lines = 4
if len(body_lines) > 4:
body_font = get_font(FONT_BOLD_PATH, 44)
line_height = 58
body_lines = []
for paragraph in str(text).split("\n"):
paragraph = paragraph.strip()
if paragraph:
body_lines.extend(wrap_text_by_pixel(draw, paragraph, body_font, max_text_width))
visible_lines = body_lines[:max_lines]
total_text_height = len(visible_lines) * line_height
y = caption_y1 + int(((caption_y2 - caption_y1) - total_text_height) / 2)
for line in visible_lines:
bbox = draw.textbbox((0, 0), line, font=body_font)
tw = bbox[2] - bbox[0]
x = int((w - tw) / 2)
draw.text((x, y), line, font=body_font, fill=(255, 255, 255, 255))
y += line_height
draw_rounded_rectangle(draw, (50, 1740, 760, 1810), 30, fill=(255, 255, 255, 235))
draw.text((80, 1758), "블로그에서 자세히 확인해보세요", font=subtitle_font, fill=(15, 23, 42, 255))
draw.text((920, 1755), f"{index + 1}/{total}", font=small_font, fill=(226, 232, 240, 255))
path = os.path.join(TEMP_DIR, f"slide_{index}.png")
canvas.save(path)
return path
async def make_tts_audio_async(text, output_path, voice):
communicate = edge_tts.Communicate(text, voice)
await communicate.save(output_path)
def make_tts_audio(text, output_path):
if not EDGE_TTS_AVAILABLE:
return False, None
text = clean_tts_text(text)
if not text:
return False, None
voice = random.choice(TTS_VOICES)
asyncio.run(make_tts_audio_async(text, output_path, voice))
return os.path.exists(output_path), voice
def get_audio_duration(audio_path):
try:
cmd = [
FFPROBE_PATH,
"-v", "error",
"-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1",
audio_path
]
result = subprocess.run(cmd, check=True, capture_output=True, text=True)
return float(result.stdout.strip())
except Exception:
return 0
def get_random_bgm():
if not os.path.isdir(BGM_DIR):
return None
files = [
os.path.join(BGM_DIR, f)
for f in os.listdir(BGM_DIR)
if f.lower().endswith((".mp3", ".wav", ".m4a", ".aac"))
]
if not files:
return None
return random.choice(files)
# ---------------------------------------------------------------------
# 관리자 MYBOX 세션/업로드 API Step1
# ---------------------------------------------------------------------
def normalize_mybox_folder_path(folder_path):
folder_path = str(folder_path or "").strip().replace("\\", "/")
folder_path = re.sub(r"/+", "/", folder_path)
folder_path = folder_path.strip("/")
return folder_path
def safe_mybox_folder_part(value):
value = str(value or "").strip()
value = re.sub(r"[^0-9a-zA-Z가-힣._-]+", "_", value)
value = re.sub(r"_+", "_", value).strip("_")
return value[:80] or "asset"
def mybox_default_folder_for_asset(asset_type="header_image", article_no=""):
year = time.strftime("%Y")
month = time.strftime("%m")
article_no = safe_mybox_folder_part(article_no or "common")
if asset_type == "header_image":
return f"REAL_AUTO/header_images/{year}/{month}/{article_no}"
if asset_type == "realtor_banner":
return f"REAL_AUTO/realtor_banners/{year}/{month}/{article_no}"
if asset_type == "location_guide":
return f"REAL_AUTO/location_guides/{year}/{month}/{article_no}"
return f"REAL_AUTO/assets/{year}/{month}/{article_no}"
@app.route("/mybox-session-check", methods=["POST"])
def mybox_session_check():
if not check_api_key(request):
return unauthorized_response()
storage_file = mybox_admin_session_storage_path()
if not os.path.exists(storage_file):
return jsonify({
"ok": True,
"linked": False,
"session_status": "not_linked",
"message": "관리자 MYBOX 세션 파일이 없습니다.",
"storage_file": storage_file.replace("\\", "/")
})
try:
file_size = os.path.getsize(storage_file)
modified_at = time.strftime(
"%Y-%m-%d %H:%M:%S",
time.localtime(os.path.getmtime(storage_file))
)
with open(storage_file, "r", encoding="utf-8") as f:
data = json.load(f)
valid_json = isinstance(data, dict)
cookie_count = len(data.get("cookies", [])) if valid_json else 0
return jsonify({
"ok": True,
"linked": valid_json and cookie_count > 0,
"session_status": "linked" if valid_json and cookie_count > 0 else "error",
"message": "관리자 MYBOX 세션 파일 확인 완료",
"storage_file": storage_file.replace("\\", "/"),
"file_size": file_size,
"modified_at": modified_at,
"cookie_count": cookie_count
})
except Exception as e:
return jsonify({
"ok": False,
"linked": False,
"session_status": "error",
"message": str(e),
"storage_file": storage_file.replace("\\", "/")
}), 500
@app.route("/mybox-login", methods=["POST"])
def mybox_login():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
admin_naver_id = str(data.get("admin_naver_id") or "").strip()
storage_file = mybox_admin_session_storage_path()
try:
with sync_playwright() as p:
browser = p.chromium.launch(
headless=False,
args=[
"--disable-blink-features=AutomationControlled",
"--no-sandbox",
]
)
context = browser.new_context(
locale="ko-KR",
viewport={"width": 1280, "height": 900}
)
page = context.new_page()
page.goto("https://mybox.naver.com/", wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(1500)
login_success = False
last_url = ""
for _ in range(600):
page.wait_for_timeout(1000)
last_url = page.url
try:
cookies = context.cookies()
naver_cookie_names = [
c.get("name", "")
for c in cookies
if "naver.com" in c.get("domain", "")
]
if "NID_AUT" in naver_cookie_names or "NID_SES" in naver_cookie_names:
login_success = True
break
except Exception:
pass
if not login_success:
browser.close()
return jsonify({
"ok": False,
"message": "관리자 네이버 로그인 시간 초과",
"last_url": last_url
}), 500
try:
page.goto("https://mybox.naver.com/", wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(3000)
except Exception:
pass
context.storage_state(path=storage_file)
browser.close()
return jsonify({
"ok": True,
"message": "관리자 MYBOX 세션 저장 완료",
"admin_naver_id": admin_naver_id,
"storage_file": storage_file.replace("\\", "/")
})
except Exception as e:
return jsonify({
"ok": False,
"message": str(e)
}), 500
def mybox_is_introduce_or_not_ready(page):
url = str(page.url or "").lower()
if "/about/introduce" in url:
return True
try:
body = page.locator("body").inner_text(timeout=5000)
except Exception:
body = ""
ready_words = ["올리기", "업로드", "새 폴더", "내 파일", "최근"]
introduce_words = ["시작하기", "MYBOX 소개", "마이박스 소개", "about"]
if any(w in body for w in ready_words):
return False
if any(w in body for w in introduce_words):
return True
return False
def mybox_wait_file_home(page):
"""
MYBOX 파일함 진입 확인.
introduce 페이지면 업로드하지 않고 명확히 실패시킨다.
"""
candidate_urls = [
"https://mybox.naver.com/my",
"https://mybox.naver.com/",
]
last_url = ""
last_text = ""
for url in candidate_urls:
try:
page.goto(url, wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(5000)
last_url = page.url
try:
last_text = page.locator("body").inner_text(timeout=5000)
except Exception:
last_text = ""
if "/about/introduce" not in str(page.url).lower():
if any(w in last_text for w in ["올리기", "업로드", "새 폴더", "내 파일", "최근"]):
return True
except Exception:
continue
raise Exception(
"MYBOX 파일함 화면에 진입하지 못했습니다. "
"관리자 계정으로 MYBOX 시작/동의/초기 설정을 완료한 뒤 세션을 다시 생성하세요. "
f"last_url={last_url}, text={last_text[:200]}"
)
def mybox_find_click_text(page, text_value, timeout=3000):
selectors = [
f"text={text_value}",
f"button:has-text('{text_value}')",
f"a:has-text('{text_value}')",
f"div:has-text('{text_value}')",
f"span:has-text('{text_value}')",
f"[title='{text_value}']",
f"[aria-label='{text_value}']",
]
for selector in selectors:
try:
loc = page.locator(selector).first
if loc.count() > 0:
loc.click(timeout=timeout, force=True)
page.wait_for_timeout(1500)
return True
except Exception:
continue
return False
def mybox_create_folder_if_needed(page, folder_name):
"""
현재 MYBOX 위치에서 folder_name 폴더가 보이면 진입,
없으면 새 폴더를 만들어 진입한다.
"""
folder_name = safe_mybox_folder_part(folder_name)
if not folder_name:
return False
# 이미 보이는 폴더면 클릭 진입
if mybox_find_click_text(page, folder_name, timeout=2500):
page.wait_for_timeout(2500)
return True
# 새 폴더 버튼 찾기
create_buttons = [
"button:has-text('새 폴더')",
"a:has-text('새 폴더')",
"span:has-text('새 폴더')",
"div:has-text('새 폴더')",
"[aria-label*='새 폴더']",
"[title*='새 폴더']",
"button:has-text('폴더')",
"a:has-text('폴더')",
]
clicked = False
last_error = ""
for selector in create_buttons:
try:
loc = page.locator(selector).first
if loc.count() > 0:
loc.click(timeout=4000, force=True)
page.wait_for_timeout(1500)
clicked = True
break
except Exception as e:
last_error = str(e)
if not clicked:
raise Exception(f"MYBOX 새 폴더 버튼을 찾지 못했습니다. folder={folder_name}, error={last_error}")
# 폴더명 입력창 처리
input_selectors = [
"input[type='text']",
"input[placeholder*='폴더']",
"input[aria-label*='폴더']",
"input",
]
typed = False
for selector in input_selectors:
try:
loc = page.locator(selector).last
if loc.count() > 0:
loc.fill(folder_name, timeout=4000)
typed = True
break
except Exception as e:
last_error = str(e)
if not typed:
raise Exception(f"MYBOX 새 폴더명 입력창을 찾지 못했습니다. folder={folder_name}, error={last_error}")
# 확인/만들기 버튼 또는 Enter
confirmed = False
for selector in [
"button:has-text('확인')",
"button:has-text('만들기')",
"button:has-text('생성')",
"a:has-text('확인')",
"a:has-text('만들기')",
]:
try:
loc = page.locator(selector).first
if loc.count() > 0:
loc.click(timeout=4000, force=True)
confirmed = True
break
except Exception:
continue
if not confirmed:
try:
page.keyboard.press("Enter")
confirmed = True
except Exception:
pass
page.wait_for_timeout(3000)
# 생성 후 폴더 진입
if mybox_find_click_text(page, folder_name, timeout=5000):
page.wait_for_timeout(2500)
return True
raise Exception(f"MYBOX 폴더 생성 후 진입 실패: {folder_name}")
def mybox_ensure_target_folder(page, target_folder):
"""
REAL_AUTO/header_images/YYYY/MM/article_no 순서로 폴더 생성/진입.
실패하면 엉뚱한 위치 업로드 방지를 위해 예외를 발생시킨다.
"""
target_folder = normalize_mybox_folder_path(target_folder)
if not target_folder:
return ""
parts = [p for p in target_folder.split("/") if p.strip()]
for part in parts:
mybox_create_folder_if_needed(page, part)
return target_folder
def mybox_try_upload_with_playwright(local_path, target_folder, headless=True):
"""
MYBOX 업로드 자동화 V1.1
- input[type=file] 직접 탐색 방식
- file chooser 이벤트 방식
- 여러 업로드 버튼 셀렉터
를 순차적으로 시도한다.
현재 단계는 업로드 성공 확인까지이며,
공유/CDN URL 추출은 다음 단계에서 구현한다.
"""
storage_file = mybox_admin_session_storage_path()
if not os.path.exists(storage_file):
raise Exception("관리자 MYBOX 세션 파일이 없습니다. /mybox-login 먼저 실행하세요.")
if not os.path.exists(local_path):
raise Exception(f"업로드할 파일이 없습니다: {local_path}")
target_folder = normalize_mybox_folder_path(target_folder)
with sync_playwright() as p:
browser = p.chromium.launch(
headless=headless,
args=[
"--disable-blink-features=AutomationControlled",
"--no-sandbox",
]
)
context = browser.new_context(
storage_state=storage_file,
locale="ko-KR",
viewport={"width": 1360, "height": 920}
)
page = context.new_page()
# MYBOX 파일함 진입 검증
mybox_wait_file_home(page)
page_text = ""
try:
page_text = page.locator("body").inner_text(timeout=7000)
except Exception:
pass
if (
"로그인" in page_text
and "올리기" not in page_text
and "MYBOX" not in page_text
):
browser.close()
raise Exception("MYBOX 세션이 만료된 것으로 보입니다. 관리자 MYBOX 재로그인이 필요합니다.")
# 지정 폴더 생성/진입
# 실패하면 잘못된 현재 디렉토리에 업로드하지 않도록 중단한다.
mybox_ensure_target_folder(page, target_folder)
debug_info = {
"last_url": page.url,
"button_try": [],
"input_count_before": 0,
"input_count_after": 0,
"filechooser": False,
"direct_input": False,
}
try:
debug_info["input_count_before"] = page.locator("input[type='file']").count()
except Exception:
pass
upload_button_selectors = [
"button:has-text('올리기')",
"button:has-text('업로드')",
"a:has-text('올리기')",
"a:has-text('업로드')",
"span:has-text('올리기')",
"span:has-text('업로드')",
"div:has-text('올리기')",
"div:has-text('업로드')",
"[aria-label*='올리기']",
"[aria-label*='업로드']",
"[title*='올리기']",
"[title*='업로드']",
]
input_set = False
last_error = ""
# 1) 먼저 기존 input[type=file]이 이미 있으면 직접 주입
try:
file_inputs = page.locator("input[type='file']")
count = file_inputs.count()
debug_info["input_count_before"] = count
for i in range(count):
try:
file_inputs.nth(i).set_input_files(local_path, timeout=5000)
input_set = True
debug_info["direct_input"] = True
break
except Exception as e:
last_error = str(e)
except Exception as e:
last_error = str(e)
# 2) 업로드 버튼 클릭 + file chooser 이벤트 방식
if not input_set:
for selector in upload_button_selectors:
try:
loc = page.locator(selector).first
if loc.count() <= 0:
continue
debug_info["button_try"].append(selector)
with page.expect_file_chooser(timeout=7000) as fc_info:
loc.click(timeout=5000, force=True)
file_chooser = fc_info.value
file_chooser.set_files(local_path)
input_set = True
debug_info["filechooser"] = True
page.wait_for_timeout(8000)
break
except Exception as e:
last_error = f"{selector} => {e}"
continue
# 3) 버튼 클릭 후 input[type=file] 생성되는 경우 처리
if not input_set:
for selector in upload_button_selectors:
try:
loc = page.locator(selector).first
if loc.count() <= 0:
continue
debug_info["button_try"].append(selector)
loc.click(timeout=5000, force=True)
page.wait_for_timeout(1500)
file_inputs = page.locator("input[type='file']")
count = file_inputs.count()
debug_info["input_count_after"] = count
for i in range(count):
try:
file_inputs.nth(i).set_input_files(local_path, timeout=5000)
input_set = True
debug_info["direct_input"] = True
break
except Exception as e:
last_error = str(e)
if input_set:
page.wait_for_timeout(8000)
break
except Exception as e:
last_error = f"{selector} => {e}"
continue
if not input_set:
# 디버그용 스크린샷 저장
try:
debug_dir = os.path.join(STORAGE_DIR, "debug")
os.makedirs(debug_dir, exist_ok=True)
screenshot_path = os.path.join(debug_dir, f"mybox_upload_fail_{int(time.time())}.png")
page.screenshot(path=screenshot_path, full_page=True)
debug_info["screenshot"] = screenshot_path.replace("\\", "/")
except Exception:
pass
browser.close()
raise Exception(
"MYBOX 업로드 파일 선택창을 열지 못했습니다. "
f"debug={json.dumps(debug_info, ensure_ascii=False)} / last_error={last_error}"
)
# 업로드 완료 대기. MYBOX UI는 상태 문구가 상황마다 달라서 넉넉히 대기한다.
page.wait_for_timeout(12000)
# 업로드 후 세션 갱신
try:
context.storage_state(path=storage_file)
except Exception:
pass
current_url = page.url
try:
page_text_after = page.locator("body").inner_text(timeout=5000)
except Exception:
page_text_after = ""
browser.close()
return {
"uploaded": True,
"current_url": current_url,
"target_folder": target_folder,
"debug": debug_info,
"page_text_sample": page_text_after[:500],
}
@app.route("/mybox-upload-file", methods=["POST"])
def mybox_upload_file():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
local_path = str(data.get("local_path") or data.get("image_path") or "").strip()
article_no = str(data.get("article_no") or "").strip()
asset_type = str(data.get("asset_type") or "header_image").strip()
target_folder = str(data.get("target_folder") or "").strip()
headless = bool(data.get("headless", True))
if not local_path:
return jsonify({
"ok": False,
"message": "local_path required"
}), 400
local_path = local_path.replace("\\", "/")
if not target_folder:
target_folder = mybox_default_folder_for_asset(asset_type, article_no)
try:
result = mybox_try_upload_with_playwright(
local_path=local_path,
target_folder=target_folder,
headless=headless
)
# Step1: 업로드 성공 검증까지만 한다.
# MYBOX의 실제 공유/CDN URL 추출은 Step2에서 별도 구현한다.
return jsonify({
"ok": True,
"message": "MYBOX 업로드 시도 완료",
"uploaded": True,
"asset_type": asset_type,
"article_no": article_no,
"local_path": local_path,
"target_folder": target_folder,
"mybox_uploaded": True,
"mybox_url": "",
"mybox_page_url": result.get("current_url", ""),
"note": "Step1은 MYBOX 업로드 성공 확인 단계입니다. 공유/CDN URL 추출은 다음 단계에서 보강합니다."
})
except Exception as e:
return jsonify({
"ok": False,
"message": str(e),
"uploaded": False,
"asset_type": asset_type,
"article_no": article_no,
"local_path": local_path,
"target_folder": target_folder,
}), 500
@app.route("/naver-auto-login", methods=["POST"])
def naver_auto_login():
data = request.get_json(silent=True) or {}
realtor_id = int(data.get("realtor_id") or 0)
if realtor_id <= 0:
return jsonify({"success": False, "message": "realtor_id required"}), 400
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT a.*, r.office_name
FROM blog_naver_accounts a
INNER JOIN blog_realtors r ON a.realtor_id = r.id
WHERE a.realtor_id = %s
LIMIT 1
""", (realtor_id,))
account = cur.fetchone()
if not account:
return jsonify({"success": False, "message": "등록된 네이버 계정이 없습니다."}), 404
if account.get("consent_status") != "agreed":
return jsonify({"success": False, "message": "자동발행 동의 상태가 아닙니다."}), 400
naver_id = str(account.get("naver_id") or "").strip()
naver_pw = php_openssl_decrypt(account.get("naver_password_enc") or "")
if not naver_id or not naver_pw:
return jsonify({"success": False, "message": "네이버 ID 또는 비밀번호가 없습니다."}), 400
storage_file = naver_session_storage_path(realtor_id)
with sync_playwright() as p:
browser = p.chromium.launch(
headless=False,
args=[
"--disable-blink-features=AutomationControlled",
"--no-sandbox",
]
)
context = browser.new_context(locale="ko-KR", viewport={"width": 1280, "height": 900})
page = context.new_page()
page.goto("https://nid.naver.com/nidlogin.login", wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(1500)
page.evaluate("""
([naverId, naverPw]) => {
const id = document.querySelector('#id');
const pw = document.querySelector('#pw');
if (id) {
id.value = naverId;
id.dispatchEvent(new Event('input', { bubbles: true }));
id.dispatchEvent(new Event('change', { bubbles: true }));
}
if (pw) {
pw.value = naverPw;
pw.dispatchEvent(new Event('input', { bubbles: true }));
pw.dispatchEvent(new Event('change', { bubbles: true }));
}
}
""", [naver_id, naver_pw])
page.wait_for_timeout(500)
page.click("#log\\.login")
login_success = False
last_url = ""
for _ in range(180):
page.wait_for_timeout(1000)
last_url = page.url.lower()
cookies = context.cookies()
cookie_names = [
c.get("name", "")
for c in cookies
if "naver.com" in c.get("domain", "")
]
if "NID_AUT" in cookie_names or "NID_SES" in cookie_names:
login_success = True
break
if not login_success:
browser.close()
with conn.cursor() as cur:
cur.execute("""
UPDATE blog_naver_accounts
SET login_status = 'failed',
last_login_error = %s,
updated_at = NOW()
WHERE realtor_id = %s
""", (f"login timeout / last_url={last_url}", realtor_id))
conn.commit()
return jsonify({
"success": False,
"message": "네이버 자동 로그인 실패 또는 시간 초과"
}), 500
context.storage_state(path=storage_file)
with open(storage_file, "r", encoding="utf-8") as f:
session_json = f.read()
browser.close()
with conn.cursor() as cur:
cur.execute("""
INSERT INTO blog_naver_sessions
(
realtor_id,
naver_id,
session_status,
session_data,
issued_at,
last_verified_at,
created_at,
updated_at
)
VALUES
(
%s,
%s,
'linked',
%s,
NOW(),
NOW(),
NOW(),
NOW()
)
""", (realtor_id, naver_id, session_json))
cur.execute("""
UPDATE blog_naver_accounts
SET login_status = 'linked',
last_login_at = NOW(),
last_login_error = NULL,
updated_at = NOW()
WHERE realtor_id = %s
""", (realtor_id,))
cur.execute("""
UPDATE blog_publish_settings
SET auto_publish_enabled = 1,
session_status = 'linked',
session_last_checked_at = NOW(),
updated_at = NOW()
WHERE realtor_id = %s
""", (realtor_id,))
conn.commit()
return jsonify({
"success": True,
"message": "네이버 자동 로그인 및 세션 저장 완료",
"realtor_id": realtor_id,
"naver_id": naver_id,
"storage_file": storage_file.replace("\\", "/")
})
except Exception as e:
conn.rollback()
try:
with conn.cursor() as cur:
cur.execute("""
UPDATE blog_naver_accounts
SET login_status = 'failed',
last_login_error = %s,
updated_at = NOW()
WHERE realtor_id = %s
""", (str(e)[:1000], realtor_id))
conn.commit()
except Exception:
pass
return jsonify({"success": False, "message": str(e)}), 500
finally:
conn.close()
@app.route("/naver-login", methods=["GET"])
def naver_login():
token = str(request.args.get("token", "")).strip()
if not token:
return "invalid token", 400
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT *
FROM blog_naver_connect_requests
WHERE connect_token = %s
LIMIT 1
""", (token,))
req = cur.fetchone()
if not req:
return "invalid request", 404
realtor_id = req["realtor_id"]
storage_file = os.path.join(
STORAGE_DIR,
f"naver_session_{realtor_id}.json"
)
with sync_playwright() as p:
browser = p.chromium.launch(
headless=False
)
context = browser.new_context()
page = context.new_page()
page.goto(
"https://nid.naver.com/nidlogin.login",
wait_until="domcontentloaded"
)
login_success = False
for _ in range(600):
current_url = page.url.lower()
print("CURRENT URL:", current_url)
try:
cookies = context.cookies()
naver_cookies = [
c for c in cookies
if "naver.com" in c.get("domain", "")
]
if len(naver_cookies) >= 3:
login_success = True
print("NAVER LOGIN SUCCESS")
break
except Exception as e:
print("COOKIE CHECK ERROR:", e)
time.sleep(1)
if not login_success:
page_text = ""
try:
page_text = page.locator("body").inner_text(timeout=5000)
except Exception:
pass
page_text_lower = page_text.lower()
fail_reason = "네이버 로그인 실패"
#--------------------------------------------------------------------------
# 실패 원인 분석
#--------------------------------------------------------------------------
if (
"2단계 인증" in page_text
or "2차 인증" in page_text
or "otp" in page_text_lower
or "인증번호" in page_text
):
fail_reason = "2차 인증 필요"
elif (
"새로운 환경에서 로그인" in page_text
or "본인 확인" in page_text
or "새 기기" in page_text
):
fail_reason = "새 기기 로그인 인증 필요"
elif (
"자동입력 방지문자" in page_text
or "captcha" in page_text_lower
or "보안문자" in page_text
):
fail_reason = "캡차 발생"
elif (
"비밀번호가 일치하지 않습니다" in page_text
or "아이디 또는 비밀번호" in page_text
):
fail_reason = "아이디 또는 비밀번호 오류"
elif (
"해외" in page_text
or "지역" in page_text
or "로그인 차단" in page_text
):
fail_reason = "타지역 로그인 제한"
elif (
"보안" in page_text
or "의심" in page_text
):
fail_reason = "네이버 보안 감지"
fail_detail = (
f"{fail_reason} / "
f"last_url={last_url}"
)
browser.close()
with conn.cursor() as cur:
cur.execute("""
UPDATE blog_naver_accounts
SET
login_status = 'failed',
last_login_error = %s,
updated_at = NOW()
WHERE realtor_id = %s
""", (
fail_detail[:1000],
realtor_id,
))
conn.commit()
return jsonify({
"success": False,
"message": fail_reason
}), 500
context.storage_state(
path=storage_file
)
print("STORAGE FILE:", storage_file)
print("FILE EXISTS:", os.path.exists(storage_file))
session_json = ""
try:
with open(storage_file, "r", encoding="utf-8") as f:
session_json = f.read()
except Exception:
pass
print("SESSION JSON LENGTH:", len(session_json))
with conn.cursor() as cur:
cur.execute("""
INSERT INTO blog_naver_sessions
(
realtor_id,
session_status,
session_data,
issued_at,
last_verified_at,
created_at,
updated_at
)
VALUES
(
%s,
'linked',
%s,
NOW(),
NOW(),
NOW(),
NOW()
)
""", (
realtor_id,
session_json
))
cur.execute("""
UPDATE blog_naver_connect_requests
SET
request_status = 'connected',
connected_at = NOW(),
updated_at = NOW()
WHERE id = %s
""", (
req["id"],
))
cur.execute("""
UPDATE blog_publish_settings
SET
auto_publish_enabled = 1,
session_status = 'linked',
session_last_checked_at = NOW(),
updated_at = NOW()
WHERE realtor_id = %s
""", (
realtor_id,
))
conn.commit()
browser.close()
return f"""
네이버 연동 완료
자동발행 세션 저장이 완료되었습니다.
이제 자동발행이 활성화됩니다.
"""
except Exception as e:
return str(e), 500
finally:
conn.close()
@app.route("/", methods=["GET"])
def index():
return jsonify({
"ok": True,
"message": "AI Shorts API is running"
})
@app.route("/health", methods=["GET"])
def health():
return jsonify({
"ok": True,
"message": "shorts ai api running",
"model": MODEL_NAME,
"tts": EDGE_TTS_AVAILABLE,
"bgm_dir": BGM_DIR
})
def make_realestate_blog_text_prompt(data):
detail = data.get("detail", {}) or {}
prices = data.get("prices", []) or []
schools = data.get("schools", []) or []
return f"""
너는 네이버 부동산 블로그 글을 자연스럽게 작성하는 전문 작가다.
아래 매물 정보를 바탕으로 사람이 직접 쓴 것처럼 자연스러운 설명문을 작성해줘.
중요 원칙:
1. 과장 금지
2. 확정 수익 표현 금지
3. 허위 매물처럼 보이는 표현 금지
4. 상담 유도는 자연스럽게
5. 문장은 부드럽고 실제 중개사가 설명하는 느낌으로
6. 마크다운 사용 금지
7. 제목, 설명, 대본 같은 라벨 출력 금지
8. HTML 태그 출력 금지
출력은 JSON 형식으로만 해줘.
JSON 구조:
{{
"market_comment": "지역과 시세 관점 설명",
"living_comment": "실거주 관점 설명",
"investment_comment": "투자/자산 관점 설명",
"closing_comment": "마무리 상담 유도 문장"
}}
매물 정보:
- 매물명: {detail.get("building_name", "")}
- 거래유형: {detail.get("trade_type", "")}
- 부동산유형: {detail.get("real_estate_type", "")}
- 가격: {detail.get("price_text", "")}
- 주소: {detail.get("address", "")}
- 도로명주소: {detail.get("road_address", "")}
- 층수: {detail.get("floor_info", "")}
- 면적: {detail.get("area_info", "")}
- 방수: {detail.get("room_count", "")}
- 욕실수: {detail.get("bathroom_count", "")}
- 방향: {detail.get("direction", "")}
- 주차정보: {detail.get("parking_info", "")}
- 매물특징: {detail.get("article_desc", "")}
시세 참고 데이터:
{prices}
학군 참고 데이터:
{schools}
"""
def extract_json_from_text(text):
text = str(text or "").strip()
try:
return json.loads(text)
except Exception:
pass
match = re.search(r"\{.*\}", text, flags=re.DOTALL)
if not match:
return {}
try:
return json.loads(match.group(0))
except Exception:
return {}
@app.route("/generate-realestate-blog-text", methods=["POST"])
def generate_realestate_blog_text():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
prompt = make_realestate_blog_text_prompt(data)
payload = {
"model": MODEL_NAME,
"prompt": prompt,
"stream": False
}
try:
response = requests.post(OLLAMA_URL, json=payload, timeout=240)
response.raise_for_status()
result = response.json()
raw_text = result.get("response", "").strip()
parsed = extract_json_from_text(raw_text)
return jsonify({
"ok": True,
"model": MODEL_NAME,
"raw_text": raw_text,
"ai_text": {
"market_comment": clean_script_text(parsed.get("market_comment", "")),
"living_comment": clean_script_text(parsed.get("living_comment", "")),
"investment_comment": clean_script_text(parsed.get("investment_comment", "")),
"closing_comment": clean_script_text(parsed.get("closing_comment", "")),
}
})
except requests.exceptions.ConnectionError:
return jsonify({
"ok": False,
"error": "Ollama 서버에 연결할 수 없습니다."
}), 500
except requests.exceptions.Timeout:
return jsonify({
"ok": False,
"error": "Ollama 응답 시간이 초과되었습니다."
}), 500
except Exception as e:
return jsonify({
"ok": False,
"error": str(e)
}), 500
@app.route("/generate-script", methods=["POST"])
def generate_script():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
prompt = make_prompt(data)
payload = {
"model": MODEL_NAME,
"prompt": prompt,
"stream": False
}
try:
response = requests.post(OLLAMA_URL, json=payload, timeout=180)
response.raise_for_status()
result = response.json()
script = clean_script_text(result.get("response", "").strip())
script = sanitize_realestate_script_for_type(
script,
property_type=data.get("property_type", ""),
title=data.get("post_title", "") or data.get("title", "")
)
return jsonify({
"ok": True,
"script": script,
"model": MODEL_NAME
})
except requests.exceptions.ConnectionError:
return jsonify({
"ok": False,
"error": "Ollama 서버에 연결할 수 없습니다."
}), 500
except requests.exceptions.Timeout:
return jsonify({
"ok": False,
"error": "Ollama 응답 시간이 초과되었습니다."
}), 500
except Exception as e:
return jsonify({
"ok": False,
"error": str(e)
}), 500
@app.route("/debug-images", methods=["POST"])
def debug_images():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
post_url = data.get("post_url", "")
project_id = str(data.get("project_id", "debug"))
image_urls = extract_blog_images(post_url, limit=60)
image_paths = download_blog_images(image_urls, f"debug_{project_id}")
image_files = []
for path in image_paths:
filename = os.path.basename(path)
image_files.append({
"filename": filename,
"url": f"http://61.32.69.107:9000/temp/{filename}",
"path": path.replace("\\", "/")
})
return jsonify({
"ok": True,
"post_url": post_url,
"raw_image_url_count": len(image_urls),
"download_image_count": len(image_paths),
"raw_image_urls": image_urls,
"images": image_files
})
@app.route("/temp/", methods=["GET"])
def serve_temp_file(filename):
return send_from_directory(TEMP_DIR, filename, as_attachment=False)
@app.route("/outputs/", methods=["GET"])
def serve_output_file(filename):
return send_from_directory(OUTPUT_DIR, filename, as_attachment=False)
def get_project_meta_from_db(article_no):
article_no = str(article_no or "").strip()
meta = {}
if not article_no:
return meta
conn = None
try:
conn = get_conn()
with conn.cursor() as cur:
cur.execute("""
SELECT
a.article_name,
a.real_estate_type,
a.trade_type,
a.price_text,
a.area_info,
a.floor_info,
a.building_name,
a.matched_office_name,
a.matched_phone,
a.realtor_id,
r.office_name,
r.representative_name,
r.phone,
r.mobile
FROM blog_realtor_articles a
LEFT JOIN blog_realtors r
ON r.id = a.realtor_id
WHERE a.article_no = %s
LIMIT 1
""", (article_no,))
row = cur.fetchone() or {}
meta.update(row)
except Exception as e:
print("get_project_meta_from_db error:", e)
finally:
try:
if conn:
conn.close()
except Exception:
pass
return meta
def build_shorts_title(title, article_no="", meta=None):
meta = meta or {}
title = clean_script_text(title or "")
if title and title not in ["부동산 숏츠", "AI 부동산 숏츠", "부동산 매물", "매물 안내"]:
return title[:54]
name = clean_script_text(meta.get("building_name") or meta.get("article_name") or "")
property_type = clean_script_text(meta.get("real_estate_type") or "")
trade_type = clean_script_text(meta.get("trade_type") or "")
price = clean_script_text(meta.get("price_text") or "")
parts = []
if name:
parts.append(name)
if property_type and property_type not in name:
parts.append(property_type)
if trade_type:
parts.append(trade_type)
if price:
parts.append(price)
result = " ".join(parts).strip()
if result:
return result[:54]
return f"부동산 매물 안내 {article_no}".strip()[:54]
def is_actual_listing_image_url(url):
lower = str(url or "").lower()
bad_words = [
"cropimage", "profile", "middle_banner", "realtor", "logo",
"button", "icon", "sticker", "emoticon", "blank", "spacer",
"adpost", "userprofile"
]
if any(w in lower for w in bad_words):
return False
# 블로그 대표이미지는 실사진이 아닐 수 있어 실매물 사진 점수에서 제외
if "akcdn-pjjh0305cdn.cafe24img.com" in lower and ("header_" in lower or "header_images" in lower):
return False
return (
"landthumb-phinf.pstatic.net" in lower
or "postfiles.pstatic.net" in lower
or "blogfiles.pstatic.net" in lower
or "mblogthumb-phinf.pstatic.net" in lower
)
def filter_actual_listing_image_urls(urls, limit=80):
filtered = []
for url in urls or []:
if is_actual_listing_image_url(url) and url not in filtered:
filtered.append(url)
return filtered[:limit]
def build_shorts_brand(meta=None):
meta = meta or {}
office = clean_script_text(
meta.get("matched_office_name")
or meta.get("office_name")
or ""
)
representative = clean_script_text(meta.get("representative_name") or "")
if office:
short_office = office.replace("공인중개사사무소", "공인중개사")
return short_office[:22]
if representative:
return f"{representative} 공인중개사"[:22]
return "공인중개사 매물 안내"
def build_shorts_phone(meta=None):
meta = meta or {}
return clean_script_text(
meta.get("matched_phone")
or meta.get("mobile")
or meta.get("phone")
or ""
)
def build_shorts_cta_brand(meta=None):
meta = meta or {}
office = clean_script_text(
meta.get("matched_office_name")
or meta.get("office_name")
or ""
)
representative = clean_script_text(meta.get("representative_name") or "")
if office:
return office[:28]
if representative:
return f"{representative} 공인중개사"[:28]
return "공인중개사 상담 안내"
def draw_brand_badge(draw, brand, x=48, y=48):
brand = clean_script_text(brand or "AI 부동산 숏츠")
font = get_font(FONT_BOLD_PATH, 29)
bbox = draw.textbbox((0, 0), brand, font=font)
w = min(540, max(270, bbox[2] - bbox[0] + 56))
draw_rounded_rectangle(draw, (x, y, x + w, y + 60), 25, fill=(14, 165, 233, 235))
draw.text((x + 26, y + 15), brand[:20], font=font, fill=(255, 255, 255, 255))
def make_final_cta_slide(title, brand="", phone="", article_no="", total=1):
w, h = 1080, 1920
title = clean_script_text(title or "부동산 매물 안내")
brand = clean_script_text(brand or "공인중개사")
phone = clean_script_text(phone or "")
article_no = clean_script_text(article_no or "")
palettes = [
((15, 23, 42), (30, 64, 175), (250, 204, 21), (37, 99, 235)),
((6, 78, 59), (15, 23, 42), (167, 243, 208), (16, 185, 129)),
((69, 10, 10), (15, 23, 42), (252, 165, 165), (220, 38, 38)),
]
top, bottom, accent, chip = random.choice(palettes)
img = Image.new("RGB", (w, h), bottom)
draw = ImageDraw.Draw(img, "RGBA")
for y in range(h):
ratio = y / h
r = int(top[0] * (1 - ratio) + bottom[0] * ratio)
g = int(top[1] * (1 - ratio) + bottom[1] * ratio)
b = int(top[2] * (1 - ratio) + bottom[2] * ratio)
draw.line([(0, y), (w, y)], fill=(r, g, b))
for _ in range(14):
cx = random.randint(-100, w + 100)
cy = random.randint(100, h - 100)
rr = random.randint(60, 180)
draw.ellipse((cx - rr, cy - rr, cx + rr, cy + rr), fill=(255, 255, 255, random.randint(10, 34)))
label_font = get_font(FONT_BOLD_PATH, 34)
title_font = get_font(FONT_BOLD_PATH, 62)
brand_font = get_font(FONT_BOLD_PATH, 58)
body_font = get_font(FONT_BOLD_PATH, 40)
small_font = get_font(FONT_PATH, 30)
draw_rounded_rectangle(draw, (90, 150, w - 90, 260), 42, fill=chip + (235,))
text = "상담 전 현재 매물 확인"
bb = draw.textbbox((0, 0), text, font=label_font)
draw.text(((w - (bb[2]-bb[0]))/2, 186), text, font=label_font, fill=(255, 255, 255, 255))
card = (70, 380, w - 70, 1410)
draw_rounded_rectangle(draw, card, 58, fill=(255, 255, 255, 238), outline=(255, 255, 255, 120), width=2)
title_lines = wrap_text_by_pixel(draw, title[:58], title_font, 820)
y = 500
for line in title_lines[:2]:
bb = draw.textbbox((0, 0), line, font=title_font)
draw.text(((w - (bb[2]-bb[0]))/2, y), line, font=title_font, fill=(15, 23, 42, 255))
y += 82
draw.line((150, y + 35, w - 150, y + 35), fill=(203, 213, 225, 255), width=2)
y += 110
for line, font, color in [
(brand, brand_font, chip + (255,)),
(phone if phone else "블로그에서 상세 정보를 확인하세요", body_font, (15, 23, 42, 255)),
(f"매물번호 {article_no}" if article_no else "상담 문의 환영합니다", body_font, (71, 85, 105, 255)),
]:
if not line:
continue
lines = wrap_text_by_pixel(draw, line, font, 800)
for l in lines[:2]:
bb = draw.textbbox((0, 0), l, font=font)
draw.text(((w - (bb[2]-bb[0]))/2, y), l, font=font, fill=color)
y += 70
cta = "방문 전 전화로 현재 조건을 확인해 주세요"
draw_rounded_rectangle(draw, (90, 1540, w - 90, 1665), 38, fill=(0, 0, 0, 172))
bb = draw.textbbox((0, 0), cta, font=small_font)
draw.text(((w - (bb[2]-bb[0]))/2, 1585), cta, font=small_font, fill=(255, 255, 255, 255))
draw.text((900, 1770), f"{total}/{total}", font=small_font, fill=(226, 232, 240, 255))
path = os.path.join(TEMP_DIR, f"cta_slide_{article_no or int(time.time())}.png")
img.save(path)
return path
def normalize_shorts_property_type(value):
text = clean_script_text(value or "")
for key in SHORTS_FALLBACK_FOLDER_MAP:
if key in text:
return key
if any(word in text for word in ["단독", "다가구", "다세대", "연립", "주택"]):
return "단독주택"
return "기타"
def list_shorts_fallback_images(property_type, limit=6):
"""실제 매물 사진이 0장일 때만 사용할 v1_images 매물종류 이미지 목록."""
normalized = normalize_shorts_property_type(property_type)
folder_names = SHORTS_FALLBACK_FOLDER_MAP.get(normalized, SHORTS_FALLBACK_FOLDER_MAP["기타"])
allowed_ext = {".jpg", ".jpeg", ".png", ".webp", ".bmp"}
files = []
for folder_name in folder_names:
folder = os.path.join(MYBOX_V1_IMAGES_DIR, folder_name)
if not os.path.isdir(folder):
continue
for root, _, names in os.walk(folder):
for name in names:
path = os.path.join(root, name)
if os.path.splitext(name)[1].lower() in allowed_ext and path not in files:
files.append(path)
if not files and os.path.isdir(MYBOX_V1_IMAGES_DIR):
for folder_name in SHORTS_FALLBACK_FOLDER_MAP["기타"]:
folder = os.path.join(MYBOX_V1_IMAGES_DIR, folder_name)
if not os.path.isdir(folder):
continue
for root, _, names in os.walk(folder):
for name in names:
path = os.path.join(root, name)
if os.path.splitext(name)[1].lower() in allowed_ext and path not in files:
files.append(path)
random.shuffle(files)
return files[:max(1, int(limit or 6))]
def pick_v1_shorts_theme():
palettes = [
{"name": "navy_gold", "overlay": (7, 18, 48, 150), "panel": (8, 20, 48, 225), "accent": (250, 204, 21), "text": (255, 255, 255)},
{"name": "emerald", "overlay": (3, 52, 44, 155), "panel": (4, 63, 54, 228), "accent": (110, 231, 183), "text": (255, 255, 255)},
{"name": "wine", "overlay": (63, 13, 30, 150), "panel": (74, 16, 36, 228), "accent": (253, 164, 175), "text": (255, 255, 255)},
{"name": "charcoal_mint", "overlay": (17, 24, 39, 158), "panel": (17, 24, 39, 230), "accent": (94, 234, 212), "text": (255, 255, 255)},
{"name": "blue_orange", "overlay": (12, 42, 86, 150), "panel": (15, 52, 96, 228), "accent": (251, 146, 60), "text": (255, 255, 255)},
]
font_pairs = [
("C:/Windows/Fonts/malgunbd.ttf", "C:/Windows/Fonts/malgun.ttf"),
("C:/Windows/Fonts/NanumGothicBold.ttf", "C:/Windows/Fonts/NanumGothic.ttf"),
("C:/Windows/Fonts/NanumSquareB.ttf", "C:/Windows/Fonts/NanumSquareR.ttf"),
]
available_pairs = [pair for pair in font_pairs if os.path.exists(pair[0]) and os.path.exists(pair[1])]
theme = dict(random.choice(palettes))
theme["font_bold"], theme["font_regular"] = random.choice(available_pairs or [(FONT_BOLD_PATH, FONT_PATH)])
theme["photo_box"] = random.choice([
(60, 300, 1020, 1190),
(95, 310, 985, 1190),
(50, 285, 1030, 1205),
])
return theme
def make_v1_fixed_background(background_path, theme, w=1080, h=1920):
canvas = make_blurred_background(background_path, w, h) if background_path else make_fallback_background(w, h)
canvas = canvas.convert("RGBA")
canvas.alpha_composite(Image.new("RGBA", (w, h), theme["overlay"]))
return canvas.convert("RGB")
def draw_centered_wrapped_text(draw, text, font, max_width, center_x, start_y, fill, max_lines=2, line_height=None):
lines = wrap_text_by_pixel(draw, clean_script_text(text), font, max_width)[:max_lines]
line_height = line_height or int(getattr(font, "size", 40) * 1.35)
y = start_y
for line in lines:
bbox = draw.textbbox((0, 0), line, font=font)
draw.text((center_x - (bbox[2] - bbox[0]) / 2, y), line, font=font, fill=fill)
y += line_height
return y
def make_v1_shorts_slide(text, index, total, title, image_path, fixed_background_path, brand, phone, theme, reference_image=False):
"""배경/영역은 영상 전체에서 고정하고 중앙 매물 사진만 교체하는 v1 슬라이드."""
w, h = 1080, 1920
canvas = make_v1_fixed_background(fixed_background_path, theme, w, h)
draw = ImageDraw.Draw(canvas, "RGBA")
title_font = get_font(theme["font_bold"], 46)
caption_font = get_font(theme["font_bold"], 43)
caption_small = get_font(theme["font_bold"], 36)
footer_font = get_font(theme["font_bold"], 31)
small_font = get_font(theme["font_regular"], 25)
accent = theme["accent"] + (255,)
white = theme["text"] + (255,)
# 상단 제목 전용 영역: 기존 'AI 부동산 숏츠' 문구를 사용하지 않는다.
draw_rounded_rectangle(draw, (40, 35, 1040, 255), 34, fill=theme["panel"], outline=theme["accent"] + (150,), width=2)
draw_centered_wrapped_text(draw, title[:70], title_font, 900, 540, 78, white, max_lines=2, line_height=61)
# 중앙 사진 영역. 모든 슬라이드에서 위치/크기가 동일하다.
photo_box = theme["photo_box"]
paste_image_contained(canvas, image_path, photo_box, radius=34, border=True)
if reference_image:
x1, y1, _, _ = photo_box
label = "참고 이미지"
bbox = draw.textbbox((0, 0), label, font=small_font)
draw_rounded_rectangle(draw, (x1 + 18, y1 + 18, x1 + 54 + bbox[2] - bbox[0], y1 + 62), 18, fill=(0, 0, 0, 175))
draw.text((x1 + 35, y1 + 27), label, font=small_font, fill=(255, 255, 255, 255))
# 자막 전용 영역: 사진과 절대 겹치지 않는다.
caption_box = (50, 1240, 1030, 1645)
draw_rounded_rectangle(draw, caption_box, 38, fill=theme["panel"], outline=(255, 255, 255, 65), width=2)
cleaned = clean_script_text(text)
lines = wrap_text_by_pixel(draw, cleaned, caption_font, 850)
font = caption_font
line_height = 60
if len(lines) > 5:
font = caption_small
line_height = 52
lines = wrap_text_by_pixel(draw, cleaned, font, 850)
visible = lines[:6]
y = 1240 + max(35, (405 - len(visible) * line_height) // 2)
for line_no, line in enumerate(visible):
bbox = draw.textbbox((0, 0), line, font=font)
color = accent if line_no == 0 and len(line) <= 20 else white
draw.text(((w - (bbox[2] - bbox[0])) / 2, y), line, font=font, fill=color)
y += line_height
# 하단 고정 중개사 정보 영역.
draw_rounded_rectangle(draw, (50, 1690, 1030, 1845), 34, fill=(255, 255, 255, 238), outline=theme["accent"] + (160,), width=2)
footer_text = clean_script_text(brand or "공인중개사")
if phone:
footer_text = f"{footer_text} · {clean_script_text(phone)}"
draw_centered_wrapped_text(draw, footer_text, footer_font, 870, 540, 1735, (15, 23, 42, 255), max_lines=2, line_height=43)
draw.text((930, 1870), f"{index + 1}/{total}", font=small_font, fill=(226, 232, 240, 255))
path = os.path.join(TEMP_DIR, f"v1_slide_{index}_{int(time.time() * 1000)}.png")
canvas.save(path)
return path
def make_video_from_slides_with_xfade_lite(slide_paths, project_id, per_slide_duration, output_path):
"""가벼운 S-11 장면전환 엔진.
성공하면 xfade 전환을 쓰고, 실패하면 호출부에서 fast concat으로 fallback한다.
"""
if len(slide_paths) < 2:
raise Exception("xfade requires at least 2 slides")
duration = max(2.6, float(per_slide_duration or 3.0))
transition_duration = min(0.45, max(0.25, duration * 0.16))
transitions = ["fade", "wipeleft", "wiperight", "slideleft", "slideright", "circleopen", "circleclose"]
inputs = []
for p in slide_paths:
inputs.extend(["-loop", "1", "-t", f"{duration:.2f}", "-i", p])
filters = []
for i in range(len(slide_paths)):
filters.append(f"[{i}:v]scale=1080:1920,format=rgba,setsar=1[v{i}]")
motion_used = []
current = "v0"
for i in range(1, len(slide_paths)):
trans = random.choice(transitions)
motion_used.append(trans)
offset = max(0.1, (duration - transition_duration) * i)
out_label = f"x{i}"
filters.append(
f"[{current}][v{i}]xfade=transition={trans}:duration={transition_duration:.2f}:offset={offset:.2f},format=yuv420p[{out_label}]"
)
current = out_label
cmd = [
FFMPEG_PATH,
"-y",
*inputs,
"-filter_complex",
";".join(filters),
"-map",
f"[{current}]",
"-r",
"30",
"-an",
"-c:v",
"libx264",
"-pix_fmt",
"yuv420p",
output_path
]
subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=240)
return motion_used
@app.route("/render-video", methods=["POST"])
def render_video():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
project_id = str(data.get("render_key") or data.get("project_id", "0"))
raw_title = data.get("title", "AI 부동산 숏츠")
post_url = data.get("post_url", "")
article_no = str(data.get("article_no") or "").strip()
project_meta = get_project_meta_from_db(article_no)
title = build_shorts_title(raw_title, article_no=article_no, meta=project_meta)
brand = build_shorts_brand(project_meta)
cta_brand = build_shorts_cta_brand(project_meta)
contact_phone = build_shorts_phone(project_meta)
script = sanitize_realestate_script_for_type(
data.get("script", ""),
property_type=project_meta.get("real_estate_type", ""),
title=title
)
chunks = split_script(script)
audio_path = os.path.join(TEMP_DIR, f"tts_{project_id}.mp3")
temp_video_path = os.path.join(TEMP_DIR, f"shorts_no_audio_{project_id}.mp4")
output_path = os.path.join(OUTPUT_DIR, f"shorts_{project_id}.mp4")
tts_success, selected_voice = make_tts_audio(script, audio_path)
audio_duration = get_audio_duration(audio_path) if tts_success else 0
per_slide_duration = 3
if audio_duration > 0 and len(chunks) > 0:
per_slide_duration = max(2.4, audio_duration / len(chunks))
# v1 전용 Worker는 검증한 매물 이미지 목록을 직접 전달한다.
# 기존 요청(render_key 없음)은 종전처럼 블로그/DB에서 이미지를 찾는다.
supplied_image_urls = data.get("image_urls") if data.get("render_key") else []
if not isinstance(supplied_image_urls, list):
supplied_image_urls = []
supplied_image_urls = [
str(url).strip() for url in supplied_image_urls
if str(url).strip().startswith(("http://", "https://"))
]
blog_image_urls = (
merge_image_urls(supplied_image_urls, [], limit=80)
if supplied_image_urls
else extract_blog_images(post_url, limit=60)
)
db_image_urls = extract_project_image_urls_from_db(article_no, limit=80)
# 네이버 블로그에서 1장만 잡히는 경우가 많아 DB 초안/수집 원본 이미지로 보강한다.
if len(blog_image_urls) < 3:
blog_image_urls = merge_image_urls(blog_image_urls, db_image_urls, limit=80)
actual_image_urls = filter_actual_listing_image_urls(blog_image_urls, limit=80)
# STEP-S11:
# 대표이미지/header/배너가 실사진처럼 다운로드되는 문제를 막기 위해
# 실제 매물 사진으로 판정된 URL만 다운로드한다.
# 실제 사진이 0장이면 CARD_MODE로 분기한다.
blog_image_urls = actual_image_urls
blog_image_paths = download_blog_images(blog_image_urls, project_id)
# 다운로드 후에도 2장 미만이면 DB 이미지 중 실제 매물 사진만 다시 시도한다.
if len(blog_image_paths) < 2 and db_image_urls:
retry_urls = filter_actual_listing_image_urls(db_image_urls, limit=80)
retry_paths = download_blog_images(retry_urls, f"{project_id}_db")
for p in retry_paths:
if p not in blog_image_paths:
blog_image_paths.append(p)
real_image_count = len(blog_image_paths)
fallback_paths = []
if real_image_count == 0:
fallback_paths = list_shorts_fallback_images(
project_meta.get("real_estate_type", ""),
limit=max(6, len(chunks) or 1)
)
blog_image_paths.extend(fallback_paths)
fallback_image_count = len(fallback_paths)
visual_mode = "photo" if real_image_count else ("fallback" if fallback_image_count else "graphic")
reference_image = fallback_image_count > 0
fixed_background_path = blog_image_paths[0] if blog_image_paths else None
visual_theme = pick_v1_shorts_theme()
# 영상 전체의 배경/텍스트 영역은 고정하고 중앙 사진만 순서대로 교체한다.
# 사진이 전혀 없고 v1_images에도 대체 사진이 없으면 그래픽 배경으로 정상 렌더링한다.
render_chunks = chunks or ["매물 정보를 안내해드립니다."]
slide_paths = []
for i, chunk in enumerate(render_chunks):
image_path = blog_image_paths[i % len(blog_image_paths)] if blog_image_paths else None
slide_paths.append(
make_v1_shorts_slide(
text=chunk,
index=i,
total=len(render_chunks),
title=title,
image_path=image_path,
fixed_background_path=fixed_background_path,
brand=cta_brand or brand,
phone=contact_phone,
theme=visual_theme,
reference_image=reference_image
)
)
# STEP-S11:
# 우선 xfade 랜덤 장면전환을 시도하고, 실패하면 기존 fast concat으로 자동 fallback한다.
transition_engine = "xfade_lite"
xfade_error = ""
selected_bgm = get_random_bgm()
motion_used = []
try:
motion_used = make_video_from_slides_with_xfade_lite(
slide_paths=slide_paths,
project_id=project_id,
per_slide_duration=per_slide_duration,
output_path=temp_video_path
)
except Exception as e:
transition_engine = "lite_fast_concat"
xfade_error = str(e)[:500]
list_path = os.path.join(TEMP_DIR, f"list_{project_id}.txt")
with open(list_path, "w", encoding="utf-8") as f:
for p in slide_paths:
f.write(f"file '{p.replace(os.sep, '/')}'\n")
f.write(f"duration {per_slide_duration:.2f}\n")
f.write(f"file '{slide_paths[-1].replace(os.sep, '/')}'\n")
make_video_cmd = [
FFMPEG_PATH,
"-y",
"-f", "concat",
"-safe", "0",
"-i", list_path,
"-vf", "scale=1080:1920,format=yuv420p",
"-r", "30",
temp_video_path
]
transition_pool = ["card_cut", "soft_cut", "layout_switch"]
motion_used = [random.choice(transition_pool) for _ in slide_paths]
subprocess.run(make_video_cmd, check=True, capture_output=True, text=True, timeout=180)
try:
if tts_success and os.path.exists(audio_path) and selected_bgm:
combine_cmd = [
FFMPEG_PATH,
"-y",
"-i", temp_video_path,
"-i", audio_path,
"-stream_loop", "-1",
"-i", selected_bgm,
"-filter_complex",
f"[2:a]volume={BGM_VOLUME}[bgm];[1:a][bgm]amix=inputs=2:duration=first:dropout_transition=2[aout]",
"-map", "0:v",
"-map", "[aout]",
"-c:v", "copy",
"-c:a", "aac",
"-shortest",
output_path
]
subprocess.run(combine_cmd, check=True, capture_output=True, text=True, timeout=180)
elif tts_success and os.path.exists(audio_path):
combine_cmd = [
FFMPEG_PATH,
"-y",
"-i", temp_video_path,
"-i", audio_path,
"-c:v", "copy",
"-c:a", "aac",
"-shortest",
output_path
]
subprocess.run(combine_cmd, check=True, capture_output=True, text=True, timeout=180)
elif selected_bgm:
combine_cmd = [
FFMPEG_PATH,
"-y",
"-i", temp_video_path,
"-stream_loop", "-1",
"-i", selected_bgm,
"-filter_complex",
f"[1:a]volume={BGM_VOLUME}[bgm]",
"-map", "0:v",
"-map", "[bgm]",
"-c:v", "copy",
"-c:a", "aac",
"-shortest",
output_path
]
subprocess.run(combine_cmd, check=True, capture_output=True, text=True, timeout=180)
else:
os.replace(temp_video_path, output_path)
video_filename = os.path.basename(output_path)
video_url = f"http://61.32.69.107:9000/outputs/{video_filename}"
return jsonify({
"ok": True,
"video_path": output_path.replace("\\", "/"),
"video_url": video_url,
"tts": tts_success,
"voice": selected_voice,
"bgm": os.path.basename(selected_bgm) if selected_bgm else None,
"render_title": title,
"brand": brand,
"cta_brand": cta_brand,
"contact_phone": contact_phone,
"transition_engine": transition_engine,
"xfade_error": xfade_error,
"image_count": len(blog_image_paths),
"real_image_count": real_image_count,
"fallback_image_count": fallback_image_count,
"raw_image_url_count": len(blog_image_urls),
"db_image_url_count": len(db_image_urls),
"actual_image_url_count": len(actual_image_urls),
"motion_styles": motion_used,
"visual_upgrade": True,
"visual_mode": visual_mode,
"visual_theme": visual_theme.get("name"),
"fixed_background": True,
"reference_image": reference_image,
"render_engine": transition_engine,
"post_url": post_url
})
except subprocess.CalledProcessError as e:
return jsonify({
"ok": False,
"error": e.stderr
}), 500
except FileNotFoundError:
return jsonify({
"ok": False,
"error": "FFmpeg 또는 FFprobe 실행 파일을 찾을 수 없습니다."
}), 500
except Exception as e:
return jsonify({
"ok": False,
"error": str(e)
}), 500
# ---------------------------------------------------------------------
# 대표이미지 생성 엔진 V4
# - 파이썬 서버 로컬 배경 폴더에서 랜덤 배경 선택
# - 동/호수 선택 표시
# - 거래유형/가격 중복 제거
# - 하단 고지 문구 제거
# - 반투명 패널 색상 랜덤 적용
# - 네이버 블로그 대표이미지 연결을 위한 is_representative 반환
# ---------------------------------------------------------------------
HEADER_IMAGE_DIR = MYBOX_GENERATED_HEADER_DIR
os.makedirs(HEADER_IMAGE_DIR, exist_ok=True)
os.makedirs(MYBOX_HEADER_BG_DIR, exist_ok=True)
HEADER_PANEL_THEMES = [
{
"name": "deep_navy_gold",
"panel": (15, 23, 42, 172),
"outline": (245, 207, 132, 178),
"accent": (245, 207, 132, 235),
"trade": (245, 158, 11, 238),
"price_text": (15, 23, 42, 255),
"price_left": (250, 225, 168, 240),
},
{
"name": "emerald_dark",
"panel": (6, 78, 59, 166),
"outline": (167, 243, 208, 145),
"accent": (167, 243, 208, 235),
"trade": (16, 185, 129, 238),
"price_text": (6, 78, 59, 255),
"price_left": (209, 250, 229, 240),
},
{
"name": "slate_blue",
"panel": (30, 41, 59, 170),
"outline": (147, 197, 253, 145),
"accent": (147, 197, 253, 235),
"trade": (37, 99, 235, 238),
"price_text": (15, 23, 42, 255),
"price_left": (219, 234, 254, 240),
},
{
"name": "wine_black",
"panel": (69, 10, 10, 158),
"outline": (252, 165, 165, 138),
"accent": (252, 165, 165, 235),
"trade": (220, 38, 38, 238),
"price_text": (69, 10, 10, 255),
"price_left": (254, 226, 226, 240),
},
{
"name": "teal_black",
"panel": (19, 78, 74, 164),
"outline": (153, 246, 228, 145),
"accent": (153, 246, 228, 235),
"trade": (20, 184, 166, 238),
"price_text": (19, 78, 74, 255),
"price_left": (204, 251, 241, 240),
},
]
def get_header_image_output_dir():
year = time.strftime("%Y")
month = time.strftime("%m")
out_dir = os.path.join(MYBOX_GENERATED_HEADER_DIR, year, month)
os.makedirs(out_dir, exist_ok=True)
return out_dir
def safe_filename_part(value):
value = str(value or "").strip()
value = re.sub(r"[^0-9a-zA-Z가-힣_-]+", "_", value)
value = re.sub(r"_+", "_", value).strip("_")
return value[:80] or "header"
def normalize_header_property_type(value):
value = str(value or "").strip().lower()
return HEADER_PROPERTY_TYPE_MAP.get(value, value or "etc")
def list_header_background_files(property_type="etc"):
property_type = normalize_header_property_type(property_type)
search_dirs = [
os.path.join(MYBOX_HEADER_BG_DIR, property_type),
os.path.join(MYBOX_HEADER_BG_DIR, "etc"),
MYBOX_HEADER_BG_DIR,
]
exts = (".jpg", ".jpeg", ".png", ".webp", ".bmp")
files = []
for folder in search_dirs:
if not os.path.isdir(folder):
continue
for name in os.listdir(folder):
path = os.path.join(folder, name)
if os.path.isfile(path) and name.lower().endswith(exts):
files.append(path)
if files:
break
return files
def pick_random_header_background(property_type="etc"):
files = list_header_background_files(property_type)
if not files:
return ""
return random.choice(files)
def load_header_background(background_path="", background_url=""):
if background_path:
path = str(background_path).strip().replace("\\", "/")
if os.path.exists(path):
try:
return Image.open(path).convert("RGB")
except Exception as e:
print("[HEADER BACKGROUND LOCAL LOAD ERROR]", str(e))
if background_url:
try:
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
res = requests.get(background_url, headers=headers, timeout=20)
res.raise_for_status()
from io import BytesIO
return Image.open(BytesIO(res.content)).convert("RGB")
except Exception as e:
print("[HEADER BACKGROUND URL LOAD ERROR]", str(e))
w, h = 1200, 675
img = Image.new("RGB", (w, h), (15, 23, 42))
draw = ImageDraw.Draw(img)
top = (15, 118, 110)
bottom = (15, 23, 42)
for y in range(h):
ratio = y / h
r = int(top[0] * (1 - ratio) + bottom[0] * ratio)
g = int(top[1] * (1 - ratio) + bottom[1] * ratio)
b = int(top[2] * (1 - ratio) + bottom[2] * ratio)
draw.line([(0, y), (w, y)], fill=(r, g, b))
return img
def cover_resize_header_image(img, w, h):
img = img.convert("RGB")
ratio = img.width / img.height
target_ratio = w / h
if ratio > target_ratio:
new_h = h
new_w = int(h * ratio)
else:
new_w = w
new_h = int(w / ratio)
img = img.resize((new_w, new_h), Image.LANCZOS)
left = (new_w - w) // 2
top = (new_h - h) // 2
return img.crop((left, top, left + w, top + h))
def draw_text_with_shadow(draw, xy, text, font, fill, shadow_fill=(0, 0, 0, 115), shadow_offset=(3, 3)):
x, y = xy
sx, sy = shadow_offset
draw.text((x + sx, y + sy), text, font=font, fill=shadow_fill)
draw.text((x, y), text, font=font, fill=fill)
def fit_text_font(draw, text, font_path, start_size, max_width, min_size=28):
size = int(start_size)
while size >= int(min_size):
font = get_font(font_path, size)
bbox = draw.textbbox((0, 0), str(text), font=font)
width = bbox[2] - bbox[0]
if width <= max_width:
return font
size -= 2
return get_font(font_path, min_size)
def remove_dong_from_title_for_display(title, dong_value=""):
"""
대표이미지 제목에서는 동 정보를 제거한다.
예:
호려울9단지한양수자인와이즈시티 911동
→ 호려울9단지한양수자인와이즈시티
"""
title = clean_script_text(title or "")
dong_value = clean_script_text(dong_value or "")
if not title:
return ""
if dong_value and title.endswith(dong_value):
title = title[: -len(dong_value)].strip()
title = re.sub(r"\s+\d{1,4}동\s*$", "", title).strip()
title = re.sub(r"\s+\d{1,5}호\s*$", "", title).strip()
return title
def split_display_title_lines(title):
"""
대표이미지 제목 표시용 줄바꿈.
너무 긴 단지명은 2줄로 분리한다.
"""
title = clean_script_text(title or "")
if not title:
return ["부동산 매물 안내"]
if len(title) <= 18:
return [title]
# 숫자단지명+브랜드명이 붙은 경우 적당히 2줄 분리
# 예: 호려울9단지한양수자인와이즈시티
candidates = [
"한양수자인",
"와이즈시티",
"힐스테이트",
"푸르지오",
"아이파크",
"래미안",
"자이",
"더샵",
"롯데캐슬",
]
for key in candidates:
idx = title.find(key)
if idx > 3:
line1 = title[:idx].strip()
line2 = title[idx:].strip()
if line1 and line2:
return [line1[:22], line2[:22]]
mid = len(title) // 2
# 중간 주변에서 자를 위치 탐색
cut = mid
for offset in range(0, 6):
for pos in [mid + offset, mid - offset]:
if 8 <= pos <= len(title) - 6:
cut = pos
break
else:
continue
break
return [title[:cut].strip(), title[cut:].strip()]
def split_korean_title_v3(title, complex_name=""):
title = clean_script_text(title or "")
complex_name = clean_script_text(complex_name or "")
if complex_name:
return [complex_name[:24]]
title = re.sub(r"\s+", " ", title).strip()
title = title.replace(" 매물 안내", "").replace("매물 안내", "")
title = title.replace(" 매물", "").replace("매물", "")
if len(title) <= 14:
return [title]
words = title.split(" ")
if len(words) >= 2:
mid = max(1, len(words) // 2)
line1 = " ".join(words[:mid]).strip()
line2 = " ".join(words[mid:]).strip()
if len(line1) < 6 and len(words) >= 3:
line1 = " ".join(words[:mid + 1]).strip()
line2 = " ".join(words[mid + 1:]).strip()
return [line1[:22], line2[:22]] if line2 else [line1[:24]]
cut = 14
return [title[:cut], title[cut:cut + 16]]
def parse_trade_and_price_v3(transaction_type, price):
transaction_type = clean_script_text(transaction_type or "")
price = clean_script_text(price or "")
known = ["매매", "전세", "월세", "단기", "분양", "임대"]
if not transaction_type:
for k in known:
if price.startswith(k):
transaction_type = k
price = price[len(k):].strip()
break
else:
# 가격 앞에 거래유형이 중복되어 있으면 제거
if price.startswith(transaction_type):
price = price[len(transaction_type):].strip()
for k in known:
# "매매 3억"처럼 다른 거래명이 앞에 붙어도 제거
if price.startswith(k):
if k == transaction_type:
price = price[len(k):].strip()
break
if not transaction_type:
transaction_type = "매물"
if not price:
price = "상담문의"
return transaction_type, price
def is_header_dong_like(value):
value = clean_script_text(value or "")
return bool(re.fullmatch(r"\d{1,4}동", value))
def infer_complex_and_dong_from_title(title, complex_name="", dong=""):
"""
API 방어 로직.
complex_name에 911동만 들어와도 title이
'호려울9단지한양수자인와이즈시티 911동'이면 단지명을 복구한다.
"""
title = clean_script_text(title or "")
complex_name = clean_script_text(complex_name or "")
dong = clean_script_text(dong or "")
m = re.search(r"(.+?)\s+(\d{1,4}동)\s*$", title)
if m:
title_complex = clean_script_text(m.group(1))
title_dong = clean_script_text(m.group(2))
if (not complex_name) or is_header_dong_like(complex_name):
complex_name = title_complex
if not dong:
dong = title_dong
return complex_name, dong
def normalize_dong_ho(dong="", ho="", dong_ho=""):
dong = clean_script_text(dong or "")
ho = clean_script_text(ho or "")
dong_ho = clean_script_text(dong_ho or "")
if dong_ho:
return dong_ho
parts = []
if dong:
parts.append(dong if dong.endswith("동") else f"{dong}동")
if ho:
parts.append(ho if ho.endswith("호") else f"{ho}호")
return " ".join(parts).strip()
# ---------------------------------------------------------------------
# 대표이미지 생성 엔진 V4 유틸
# - 800x800 정사각형
# - 아파트 대표이미지 구성 12종 랜덤
# - 글자 잘림 방지
# - 허위/과장 문구 제외, 실제 수집 데이터만 사용
# ---------------------------------------------------------------------
APT_HEADER_TEMPLATE_POOL_V4 = [
"APT_REAL_PHOTO_GLASS",
]
DEFAULT_HEADER_TEMPLATE_POOL_V4 = [
"DEFAULT_PRICE_BIG",
"DEFAULT_CLEAN_CARD",
"DEFAULT_PHOTO_FULL",
]
def compact_header_price(value):
value = clean_script_text(value or "")
value = re.sub(r"(\d+억)\s+(\d)", r"\1\2", value)
return value
def extract_exclusive_area_for_header(value):
value = clean_script_text(value or "")
if not value:
return ""
m = re.search(r"전용\s*([0-9.]+\s*(?:㎡|m²|m2))", value, flags=re.IGNORECASE)
if m:
return "전용 " + m.group(1).replace("m2", "㎡").replace("m²", "㎡").strip()
m = re.search(r"/\s*([0-9.]+\s*(?:㎡|m²|m2))", value, flags=re.IGNORECASE)
if m:
return "전용 " + m.group(1).replace("m2", "㎡").replace("m²", "㎡").strip()
return value
def normalize_floor_for_header(value):
value = clean_script_text(value or "")
if not value:
return ""
m = re.match(r"^\s*(\d+)\s*/\s*(\d+)\s*$", value)
if m:
return f"{m.group(1)}층 / 총 {m.group(2)}층"
m = re.match(r"^\s*(\d+)층\s*/\s*총?\s*(\d+)층\s*$", value)
if m:
return f"{m.group(1)}층 / 총 {m.group(2)}층"
return value.replace("층층", "층")
def split_title_smart_v4(title, max_chars=10, max_lines=3):
title = clean_script_text(title or "")
title = title.replace("아파트", "").strip()
if not title:
return ["부동산", "매물안내"]
preferred = [
"마스터힐스", "힐스테이트", "푸르지오", "래미안", "아이파크",
"롯데캐슬", "더샵", "자이", "한양수자인", "와이즈시티",
"센트럴", "파크", "리버", "포레", "시티",
]
for key in preferred:
idx = title.find(key)
if idx > 1 and idx < len(title) - 1:
lines = [title[:idx].strip(), title[idx:].strip()]
return [x[:max_chars + 4] for x in lines if x][:max_lines]
if len(title) <= max_chars:
return [title]
lines=[]
remain=title
while remain and len(lines)=int(min_size):
font=get_font(font_path, size)
ok=True
for line in lines:
bb=draw.textbbox((0,0), str(line), font=font)
if bb[2]-bb[0] > max_width:
ok=False; break
if ok:
return font
size -= 2
return get_font(font_path, min_size)
def draw_info_chip_v4(draw, x, y, text, font, fill=(255,255,255,230), text_fill=(15,23,42,255)):
text=clean_script_text(text or "")
if not text: return x
pad_x,pad_y=14,8
bb=draw.textbbox((0,0), text, font=font)
w=bb[2]-bb[0]+pad_x*2; h=bb[3]-bb[1]+pad_y*2
draw.rounded_rectangle((x,y,x+w,y+h), radius=16, fill=fill)
draw.text((x+pad_x,y+pad_y-2), text, font=font, fill=text_fill)
return x+w+10
def make_header_context_v4(data):
raw_title=clean_script_text(data.get("title") or "부동산 매물 안내")
complex_name=clean_script_text(data.get("complex_name") or data.get("building_name") or "")
dong_value=clean_script_text(data.get("dong") or data.get("building_dong") or "")
complex_name, dong_value = infer_complex_and_dong_from_title(raw_title, complex_name=complex_name, dong=dong_value)
display_title = complex_name or remove_dong_from_title_for_display(raw_title, dong_value)
display_title = remove_dong_from_title_for_display(display_title, dong_value)
property_type=normalize_header_property_type(data.get("property_type") or data.get("real_estate_type") or data.get("asset_category") or "etc")
transaction_type_raw=clean_script_text(data.get("transaction_type") or data.get("trade_type") or "")
price_raw=clean_script_text(data.get("price") or data.get("price_text") or "")
transaction_type, price = parse_trade_and_price_v3(transaction_type_raw, price_raw)
area=extract_exclusive_area_for_header(data.get("exclusive_area") or data.get("area_info") or "")
floor_text=normalize_floor_for_header(data.get("floor_info") or data.get("floor") or "")
direction=clean_script_text(data.get("direction") or "")
region=clean_script_text(data.get("region_name") or data.get("region") or "")
realtor_name=clean_script_text(data.get("realtor_name") or data.get("office_name") or "공인중개사사무소")
return {
"display_title": display_title,
"dong": dong_value,
"title_lines": split_title_smart_v4(display_title, 10, 3),
"property_type": property_type,
"transaction_type": transaction_type,
"price": compact_header_price(price),
"exclusive_area": area,
"floor_text": floor_text,
"direction": direction,
"region": region,
"realtor_name": realtor_name,
}
def choose_header_template_v4(property_type, requested_template=""):
"""
대표이미지 템플릿 선택.
- generate_blog_drafts.py에서 APT_BASIC_SQUARE_V2를 넘겨도
아파트 대표이미지는 APT_REAL_PHOTO_GLASS로 고정한다.
"""
property_type = normalize_header_property_type(property_type)
if property_type == "apt":
return "APT_REAL_PHOTO_GLASS"
requested_template = str(requested_template or "").strip()
if requested_template and requested_template not in [
"APT_BASIC_V3",
"APT_BASIC_SQUARE_V1",
"APT_BASIC_SQUARE_V2",
"APT_REAL_PHOTO_GLASS",
]:
return requested_template
if DEFAULT_HEADER_TEMPLATE_POOL_V4:
return random.choice(DEFAULT_HEADER_TEMPLATE_POOL_V4)
return "DEFAULT_PRICE_BIG"
# Backward-compatible aliases.
# 대표이미지 템플릿에서 실수로 구 함수명을 호출해도 생성 실패하지 않도록 유지한다.
draw_text_centered = draw_text_centered_v4
draw_multiline_centered = draw_multiline_centered_v4
def fit_text_font(draw, text, font_path, start_size, max_width, min_size=18):
"""
한 줄 텍스트가 max_width 안에 들어가도록 폰트 크기를 줄인다.
대표이미지 가격/면적/중개사명 잘림 방지용.
"""
text = str(text or "")
size = int(start_size)
while size >= int(min_size):
font = get_font(font_path, size)
try:
bb = draw.textbbox((0, 0), text, font=font)
width = bb[2] - bb[0]
except Exception:
width = len(text) * size
if width <= max_width:
return font
size -= 2
return get_font(font_path, int(min_size))
def fit_multiline_font(draw, lines, font_path, start_size, max_width, min_size=24):
"""
여러 줄 제목이 max_width 안에 들어가도록 폰트 크기를 줄인다.
대표이미지 단지명/동 정보 잘림 방지용.
"""
if not isinstance(lines, (list, tuple)):
lines = [str(lines or "")]
lines = [str(x or "").strip() for x in lines if str(x or "").strip()]
if not lines:
return get_font(font_path, int(min_size))
size = int(start_size)
while size >= int(min_size):
font = get_font(font_path, size)
ok = True
for line in lines:
try:
bb = draw.textbbox((0, 0), line, font=font)
width = bb[2] - bb[0]
except Exception:
width = len(line) * size
if width > max_width:
ok = False
break
if ok:
return font
size -= 2
return get_font(font_path, int(min_size))
def split_title_smart(title, max_chars=11, max_lines=2):
"""
대표이미지 제목 줄바꿈 안전 함수.
- 강제 말줄임표 금지
- 잘린 느낌 방지
- 긴 단지명은 2줄 중심으로 분리
"""
title = clean_script_text(title or "")
title = title.replace("아파트", "").strip()
if not title:
return ["부동산", "매물안내"]
if len(title) <= max_chars:
return [title]
preferred = [
"마스터힐스", "힐스테이트", "푸르지오", "래미안", "아이파크",
"롯데캐슬", "더샵", "자이", "한양수자인", "와이즈시티",
"센트럴", "파크", "리버", "포레", "시티", "마을",
]
for key in preferred:
idx = title.find(key)
if idx > 1 and idx < len(title) - 1:
lines = [title[:idx].strip(), title[idx:].strip()]
lines = [x for x in lines if x]
if len(lines) <= max_lines:
return lines
lines = []
remaining = title
while remaining and len(lines) < max_lines:
if len(remaining) <= max_chars:
lines.append(remaining)
remaining = ""
break
cut = max_chars
# 숫자/브랜드 중간 절단 느낌을 줄이기 위해 조금 뒤까지 허용
if len(remaining) > max_chars + 2:
cut = max_chars + 2
lines.append(remaining[:cut].strip())
remaining = remaining[cut:].strip()
if remaining:
if lines:
lines[-1] = (lines[-1] + remaining).strip()
else:
lines.append(remaining)
return [x for x in lines if x]
def fit_text_font(draw, text, font_path, start_size, max_width, min_size=18):
"""
한 줄 텍스트가 max_width 안에 들어가도록 폰트 크기를 줄인다.
"""
text = str(text or "")
size = int(start_size)
while size >= int(min_size):
font = get_font(font_path, size)
try:
bb = draw.textbbox((0, 0), text, font=font)
width = bb[2] - bb[0]
except Exception:
width = len(text) * size
if width <= max_width:
return font
size -= 2
return get_font(font_path, int(min_size))
def fit_multiline_font(draw, lines, font_path, start_size, max_width, min_size=24):
"""
여러 줄 제목이 max_width 안에 들어가도록 폰트 크기를 줄인다.
"""
if not isinstance(lines, (list, tuple)):
lines = [str(lines or "")]
lines = [str(x or "").strip() for x in lines if str(x or "").strip()]
if not lines:
return get_font(font_path, int(min_size))
size = int(start_size)
while size >= int(min_size):
font = get_font(font_path, size)
ok = True
for line in lines:
try:
bb = draw.textbbox((0, 0), line, font=font)
width = bb[2] - bb[0]
except Exception:
width = len(line) * size
if width > max_width:
ok = False
break
if ok:
return font
size -= 2
return get_font(font_path, int(min_size))
def draw_header_template_v4(bg, draw, ctx, theme, template_name, width, height):
"""
STEP20 대표이미지 - 실제 단지사진 우선 + 실중개사형 Glass 카드.
외부 보조 함수 의존 없이 이 함수 내부에서 처리한다.
"""
W, H = int(width), int(height)
def _clean(v):
return str(v or "").strip()
def _text_size(text, font):
try:
bb = draw.textbbox((0, 0), str(text), font=font)
return bb[2] - bb[0], bb[3] - bb[1]
except Exception:
return len(str(text)) * 20, 30
def _fit_font(text, start_size, max_width, min_size=20, bold=True):
text = str(text or "")
size = int(start_size)
path = FONT_BOLD_PATH if bold else FONT_PATH
while size >= int(min_size):
font = get_font(path, size)
tw, _ = _text_size(text, font)
if tw <= max_width:
return font
size -= 2
return get_font(path, int(min_size))
def _draw_center(text, box, font, fill, shadow=True, shadow_alpha=85):
x1, y1, x2, y2 = box
text = str(text or "")
tw, th = _text_size(text, font)
x = int(x1 + ((x2 - x1) - tw) / 2)
y = int(y1 + ((y2 - y1) - th) / 2)
if shadow:
draw.text((x + 2, y + 2), text, font=font, fill=(0, 0, 0, shadow_alpha))
draw.text((x, y), text, font=font, fill=fill)
return y + th
def _split_title(full_title):
full_title = _clean(full_title)
if not full_title:
return ["부동산 매물"]
if len(full_title) <= 17:
return [full_title]
preferred = [
"마스터힐스", "힐스테이트", "푸르지오", "래미안", "아이파크",
"롯데캐슬", "더샵", "자이", "한양수자인", "와이즈시티",
"센트럴", "파크", "리버", "포레", "시티",
]
for key in preferred:
idx = full_title.find(key)
if idx > 1:
a = full_title[:idx].strip()
b = full_title[idx:].strip()
if a and b:
return [a, b]
return [full_title[:17].strip(), full_title[17:].strip()]
def _price_clean(value):
value = _clean(value)
value = re.sub(r"(\d+억)\s+(\d)", r"\1 \2", value)
value = re.sub(r"(\d+억)(\d)", r"\1 \2", value)
return value
trade = _clean(ctx.get("transaction_type") or "매물")
price = _price_clean(ctx.get("price") or "가격문의")
area = _clean(ctx.get("exclusive_area") or "")
realtor = _clean(ctx.get("realtor_name") or "공인중개사사무소")
dong = _clean(ctx.get("dong") or "")
title = _clean(ctx.get("display_title") or ctx.get("raw_title") or "부동산 매물")
if dong and dong not in title:
full_title = f"{title} {dong}".strip()
else:
full_title = title
title_lines = _split_title(full_title)
overlay = Image.new("RGBA", (W, H), (0, 0, 0, 0))
od = ImageDraw.Draw(overlay)
# 실제 사진을 살리되 텍스트 영역은 어둡게
od.rectangle((0, 0, W, H), fill=(0, 0, 0, 18))
panel = (70, 88, W - 70, H - 78)
od.rounded_rectangle(
panel,
radius=28,
fill=(8, 22, 42, 150),
outline=(255, 255, 255, 140),
width=2
)
od.line((120, 390, W - 120, 390), fill=(255, 255, 255, 85), width=1)
od.line((120, 642, W - 120, 642), fill=(255, 255, 255, 85), width=1)
try:
bg.alpha_composite(overlay)
except Exception:
composed = Image.alpha_composite(bg.convert("RGBA"), overlay)
bg.paste(composed.convert(bg.mode))
white = (255, 255, 255, 255)
mint = (190, 255, 230, 255)
yellow = (250, 204, 21, 255)
green = (16, 185, 129, 255)
badge_font = get_font(FONT_BOLD_PATH, 31)
bw, _ = _text_size(trade, badge_font)
badge_box = (112, 116, 112 + max(115, bw + 48), 170)
draw.rounded_rectangle(
badge_box,
radius=26,
fill=green,
outline=(255, 255, 255),
width=1
)
_draw_center(trade, badge_box, badge_font, white, shadow=False)
max_line = max(title_lines, key=len) if title_lines else full_title
title_font = _fit_font(max_line, start_size=58, max_width=W - 170, min_size=36, bold=True)
title_y = 230 if len(title_lines) == 1 else 205
y = title_y
for line in title_lines:
_, th = _text_size(line, title_font)
_draw_center(
line,
(90, y, W - 90, y + th + 8),
title_font,
white,
shadow=True,
shadow_alpha=80
)
y += th + 12
price_font = _fit_font(price, start_size=88, max_width=W - 170, min_size=52, bold=True)
_draw_center(
price,
(90, 435, W - 90, 540),
price_font,
yellow,
shadow=True,
shadow_alpha=80
)
if area:
area_font = _fit_font(area, start_size=34, max_width=W - 190, min_size=22, bold=True)
_draw_center(
area,
(100, 555, W - 100, 602),
area_font,
mint,
shadow=True,
shadow_alpha=65
)
realtor_font = _fit_font(realtor, start_size=31, max_width=W - 150, min_size=20, bold=True)
_draw_center(
realtor,
(90, H - 125, W - 90, H - 82),
realtor_font,
white,
shadow=True,
shadow_alpha=75
)
return "APT_REAL_PHOTO_GLASS"
def make_header_image(data):
width = int(data.get("width") or 800)
height = int(data.get("height") or 800)
if width < 600:
width = 800
if height < 600:
height = 800
article_no = safe_filename_part(data.get("article_no") or data.get("project_id") or str(int(time.time())))
requested_template_code = safe_filename_part(data.get("template_code") or "APT_BASIC_SQUARE_V2")
ctx = make_header_context_v4(data)
property_type = ctx["property_type"]
background_path = data.get("background_path") or ""
background_url = (
data.get("background_url")
or data.get("background_image_url")
or data.get("header_background_url")
or ""
)
selected_background_path = background_path
if not selected_background_path and not background_url:
selected_background_path = pick_random_header_background(property_type)
theme = random.choice(HEADER_PANEL_THEMES)
bg = load_header_background(selected_background_path, background_url)
bg = cover_resize_header_image(bg, width, height).convert("RGBA")
grad = Image.new("RGBA", (width, height), (0, 0, 0, 0))
grad_draw = ImageDraw.Draw(grad)
for y in range(height):
alpha = int(max(0, (y - height * 0.15) / (height * 0.85)) * 95)
grad_draw.line([(0, y), (width, y)], fill=(0, 0, 0, alpha))
bg.alpha_composite(grad)
draw = ImageDraw.Draw(bg, "RGBA")
used_template = choose_header_template_v4(
property_type=property_type,
requested_template=requested_template_code,
)
used_template = draw_header_template_v4(
bg=bg,
draw=draw,
ctx=ctx,
theme=theme,
template_name=used_template,
width=width,
height=height,
)
out_dir = get_header_image_output_dir()
filename = f"header_{requested_template_code}_{used_template}_{article_no}_{int(time.time())}.png"
output_path = os.path.join(out_dir, filename)
bg.convert("RGB").save(output_path, "PNG", optimize=True)
relative = os.path.relpath(output_path, MYBOX_LOCAL_ROOT).replace("\\", "/")
image_url = f"http://61.32.69.107:9000/mybox/{relative}"
return {
"image_path": output_path.replace("\\", "/"),
"image_url": image_url,
"filename": filename,
"relative_path": relative,
"width": width,
"height": height,
"property_type": property_type,
"transaction_type": ctx["transaction_type"],
"price": ctx["price"],
"exclusive_area": ctx["exclusive_area"],
"floor_text": ctx["floor_text"],
"direction": ctx["direction"],
"display_title": ctx["display_title"],
"template_code": requested_template_code,
"used_template": used_template,
"theme": theme["name"],
"background_path": str(selected_background_path or "").replace("\\", "/"),
"background_url": background_url,
"is_representative": True,
"representative_role": "blog_header",
}
@app.route("/generate-header-image", methods=["POST"])
def generate_header_image():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
try:
result = make_header_image(data)
return jsonify({
"ok": True,
"message": "대표이미지 생성 완료",
**result
})
except Exception as e:
return jsonify({
"ok": False,
"error": str(e)
}), 500
# ---------------------------------------------------------------------
# 중간배너 이미지 생성 엔진
# - 600x200 가로형 CTA 배너
# - 대표이미지와 동일하게 mybox/generated 아래 PNG 저장
# - generate_blog_drafts.py에서 Cafe24 CDN 업로드 후 본문 ![]()
로 사용
# ---------------------------------------------------------------------
MIDDLE_BANNER_GENERATED_DIR = os.path.join(MYBOX_LOCAL_ROOT, "generated", "middle_banners")
os.makedirs(MIDDLE_BANNER_GENERATED_DIR, exist_ok=True)
def get_middle_banner_output_dir():
year = time.strftime("%Y")
month = time.strftime("%m")
out_dir = os.path.join(MIDDLE_BANNER_GENERATED_DIR, year, month)
os.makedirs(out_dir, exist_ok=True)
return out_dir
def list_middle_banner_background_files(property_type="etc"):
property_type = normalize_header_property_type(property_type)
search_dirs = [
os.path.join(MYBOX_HEADER_BG_DIR, property_type),
os.path.join(MYBOX_HEADER_BG_DIR, "officetel"),
os.path.join(MYBOX_HEADER_BG_DIR, "apt"),
os.path.join(MYBOX_HEADER_BG_DIR, "etc"),
MYBOX_HEADER_BG_DIR,
]
exts = (".jpg", ".jpeg", ".png", ".webp", ".bmp")
files = []
for folder in search_dirs:
if not os.path.isdir(folder):
continue
for name in os.listdir(folder):
path = os.path.join(folder, name)
if os.path.isfile(path) and name.lower().endswith(exts):
files.append(path)
if files:
break
return files
def pick_random_middle_banner_background(property_type="etc"):
files = list_middle_banner_background_files(property_type)
if not files:
return ""
return random.choice(files)
def text_size_safe(draw, text, font):
try:
bbox = draw.textbbox((0, 0), str(text), font=font)
return bbox[2] - bbox[0], bbox[3] - bbox[1]
except Exception:
return len(str(text)) * 20, 30
def fit_banner_font(draw, text, start_size, max_width, min_size=16, bold=True):
text = str(text or "")
size = int(start_size)
font_path = FONT_BOLD_PATH if bold else FONT_PATH
while size >= int(min_size):
font = get_font(font_path, size)
tw, _ = text_size_safe(draw, text, font)
if tw <= max_width:
return font
size -= 2
return get_font(font_path, int(min_size))
def draw_banner_center_text(draw, box, text, font, fill, shadow=True):
x1, y1, x2, y2 = box
text = str(text or "")
tw, th = text_size_safe(draw, text, font)
x = int(x1 + ((x2 - x1) - tw) / 2)
y = int(y1 + ((y2 - y1) - th) / 2)
if shadow:
draw.text((x + 2, y + 2), text, font=font, fill=(0, 0, 0, 120))
draw.text((x, y), text, font=font, fill=fill)
def make_middle_banner_image(data):
width = int(data.get("width") or 600)
height = int(data.get("height") or 200)
if width < 500:
width = 600
if height < 160:
height = 200
article_no = safe_filename_part(
data.get("article_no")
or data.get("project_id")
or str(int(time.time()))
)
property_type = normalize_header_property_type(
data.get("property_type")
or data.get("real_estate_type")
or "apt"
)
office_name = clean_script_text(
data.get("office_name")
or data.get("realtor_name")
or "공인중개사사무소"
)
phone = clean_script_text(
data.get("phone")
or data.get("office_phone")
or data.get("mobile_phone")
or ""
)
button_text = clean_script_text(
data.get("button_text")
or "전체매물 보기"
)
background_path = str(data.get("background_path") or "").strip()
background_url = str(data.get("background_url") or "").strip()
selected_background_path = background_path
if not selected_background_path and not background_url:
selected_background_path = pick_random_middle_banner_background(property_type)
bg = load_header_background(selected_background_path, background_url)
bg = cover_resize_header_image(bg, width, height).convert("RGBA")
overlay = Image.new("RGBA", (width, height), (0, 0, 0, 0))
od = ImageDraw.Draw(overlay)
od.rectangle((0, 0, width, height), fill=(0, 0, 0, 58))
od.rounded_rectangle(
(18, 18, width - 18, height - 18),
radius=22,
fill=(8, 22, 42, 158),
outline=(255, 255, 255, 138),
width=2
)
bg.alpha_composite(overlay)
draw = ImageDraw.Draw(bg, "RGBA")
white = (255, 255, 255, 255)
blue_soft = (191, 219, 254, 255)
yellow = (250, 204, 21, 255)
orange = (249, 115, 22, 255)
button_bg = (37, 99, 235, 245)
label_font = get_font(FONT_BOLD_PATH, 18)
office_font = fit_banner_font(draw, office_name, 34, width - 82, min_size=21, bold=True)
phone_font = fit_banner_font(draw, phone, 30, width - 150, min_size=20, bold=True) if phone else get_font(FONT_BOLD_PATH, 24)
button_font = get_font(FONT_BOLD_PATH, 18)
draw.text((42, 36), "네이버 등록 매물 안내", font=label_font, fill=blue_soft)
draw_banner_center_text(
draw,
(34, 58, width - 34, 106),
office_name,
office_font,
white,
shadow=True
)
if phone:
draw.text((46, 118), "☎", font=phone_font, fill=orange)
draw_banner_center_text(
draw,
(78, 112, width - 220, 154),
phone,
phone_font,
yellow,
shadow=True
)
button_box = (width - 196, height - 60, width - 38, height - 26)
draw.rounded_rectangle(button_box, radius=10, fill=button_bg)
draw_banner_center_text(
draw,
button_box,
button_text + " ↗",
button_font,
white,
shadow=False
)
out_dir = get_middle_banner_output_dir()
filename = f"middle_banner_{property_type}_{article_no}_{int(time.time())}.png"
output_path = os.path.join(out_dir, filename)
bg.convert("RGB").save(output_path, "PNG", optimize=True)
relative = os.path.relpath(output_path, MYBOX_LOCAL_ROOT).replace("\\", "/")
image_url = f"http://61.32.69.107:9000/mybox/{relative}"
return {
"image_path": output_path.replace("\\", "/"),
"image_url": image_url,
"filename": filename,
"relative_path": relative,
"width": width,
"height": height,
"property_type": property_type,
"office_name": office_name,
"phone": phone,
"background_path": str(selected_background_path or "").replace("\\", "/"),
"background_url": background_url,
"theme": "middle_banner_dark",
"asset_type": "middle_banner",
}
@app.route("/generate-middle-banner-image", methods=["POST"])
def generate_middle_banner_image():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
try:
result = make_middle_banner_image(data)
return jsonify({
"ok": True,
"message": "중간배너 이미지 생성 완료",
**result
})
except Exception as e:
return jsonify({
"ok": False,
"error": str(e)
}), 500
@app.route("/mybox/", methods=["GET"])
def serve_mybox_file(filename):
return send_from_directory(MYBOX_LOCAL_ROOT, filename, as_attachment=False)
@app.route("/storage", methods=["GET"])
@app.route("/storage/", methods=["GET"])
def storage_index():
return jsonify({
"ok": True,
"message": "storage route is running",
"storage_dir": STORAGE_DIR
})
@app.route("/storage/", methods=["GET"])
def serve_storage_file(filename):
return send_from_directory(STORAGE_DIR, filename, as_attachment=False)
@app.route("/collect-realtor-articles", methods=["POST"])
def collect_realtor_articles_api():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
realtor_id = data.get("realtor_id")
office_name = str(data.get("office_name", "")).strip()
representative_name = str(data.get("representative_name", "")).strip()
office_phone = str(data.get("office_phone", "")).strip()
mobile_phone = str(data.get("mobile_phone", "")).strip()
license_number = str(data.get("license_number", "")).strip()
business_number = str(data.get("business_number", "")).strip()
if not realtor_id:
return jsonify({
"ok": False,
"error": "realtor_id가 없습니다."
}), 400
if not office_name:
return jsonify({
"ok": False,
"error": "office_name이 없습니다."
}), 400
try:
from db import get_conn
search_keyword_parts = [
office_name,
representative_name,
office_phone,
mobile_phone,
license_number,
business_number,
]
search_keyword = " ".join([
str(x).strip()
for x in search_keyword_parts
if str(x).strip()
])
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO blog_realtor_search_runs
(
realtor_id,
search_keyword,
search_region,
total_found,
matched_count,
status,
started_at,
created_at,
updated_at
)
VALUES
(
%s,
%s,
'',
0,
0,
'pending',
NOW(),
NOW(),
NOW()
)
""", (
realtor_id,
search_keyword
))
search_run_id = cur.lastrowid
cur.execute("""
UPDATE blog_realtors
SET
collect_status = 'running',
collect_last_message = %s,
updated_at = NOW()
WHERE id = %s
""", (
f"수집 작업 생성 완료. run_id={search_run_id}",
realtor_id
))
conn.commit()
finally:
conn.close()
return jsonify({
"ok": True,
"message": "중개업소 기준 수집 작업이 생성되었습니다.",
"realtor_id": realtor_id,
"search_run_id": search_run_id,
"search_keyword": search_keyword,
"matched_count": 0,
"articles": []
})
except Exception as e:
return jsonify({
"ok": False,
"error": str(e)
}), 500
try:
from v1.shorts_v1_routes import register_v1_routes
register_v1_routes(app, sys.modules[__name__])
print("[V1 SHORTS ROUTES] registered: /v1/generate-script, /v1/render-video")
except Exception as v1_route_error:
print("[V1 SHORTS ROUTES] not registered:", v1_route_error)
if __name__ == "__main__":
app.run(
host="0.0.0.0",
port=9000,
debug=False,
use_reloader=False,
)