# -*- 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, ImageEnhance, ImageStat
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")
HEADER_PROPERTY_TYPE_MAP = {
"apt": "apt",
"apartment": "apt",
"아파트": "apt",
"officetel": "officetel",
"오피스텔": "officetel",
"villa": "villa",
"빌라": "villa",
"store": "store",
"상가": "store",
# LAND 계열: 네이버 부동산에서는 토지 매물이 "대", "전", "답"처럼
# 짧은 유형명으로 내려올 수 있으므로 모두 land로 정규화한다.
"land": "land",
"토지": "land",
"토지/임야": "land",
"대": "land",
"대지": "land",
"전": "land",
"답": "land",
"임야": "land",
"잡종지": "land",
"공장용지": "land",
"창고용지": "land",
"TJ": "land",
"tj": "land",
"E03": "land",
"e03": "land",
"GRND": "land",
"grnd": "land",
"대": "land",
"대지": "land",
"전": "land",
"답": "land",
"임야": "land",
"잡종지": "land",
"공장용지": "land",
"창고용지": "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. 250~350자 이내로 작성해.
8. 문장은 짧게 작성해.
절대 쓰지 마:
- 별표 강조 표시: **, *
- 괄호 설명: (후킹 문장), (배경음악), (장면 설명)
- 대괄호 설명: [자막], [나레이션]
- 시간 표시: 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 "").strip()
# STEP272 HOTFIX:
# 클라이언트가 문자열 "null"/"None" 등을 보낸 경우 실제 제목으로 그리지 않는다.
if text.lower() in {"", "null", "none", "undefined", "nan", "nil", "-"}:
return ""
if re.fullmatch(
r"(?i)(?:null|none|undefined|nan|nil)(?:\s+\d{1,4}동)?",
text,
):
return ""
text = re.sub(
r"(?i)^(?:null|none|undefined|nan|nil)\s+",
"",
text,
).strip()
if re.fullmatch(r"\d{1,4}동", text):
return ""
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)
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 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 ["AI 부동산 숏츠"]
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
)
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 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)
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), "AI 부동산 숏츠", 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
})
@app.route("/generate-script", methods=["POST"])
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
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())
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)
@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("project_id", "0"))
title = data.get("title", "AI 부동산 숏츠")
script = clean_script_text(data.get("script", ""))
post_url = data.get("post_url", "")
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))
blog_image_urls = extract_blog_images(post_url, limit=60)
blog_image_paths = download_blog_images(blog_image_urls, project_id)
if not blog_image_paths:
default_image = make_realestate_default_image(project_id)
blog_image_paths = [default_image]
represent_image_path = blog_image_paths[0]
slide_paths = []
for i, chunk in enumerate(chunks):
image_path = blog_image_paths[i % len(blog_image_paths)]
slide_paths.append(
make_slide_image(
chunk,
i,
len(chunks),
title,
image_path,
represent_image_path
)
)
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
]
selected_bgm = get_random_bgm()
try:
subprocess.run(make_video_cmd, check=True, capture_output=True, text=True)
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)
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)
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)
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,
"image_count": len(blog_image_paths),
"raw_image_url_count": len(blog_image_urls),
"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):
raw_value = str(value or "").strip()
value_l = raw_value.lower()
if not raw_value:
return "etc"
if raw_value in HEADER_PROPERTY_TYPE_MAP:
return HEADER_PROPERTY_TYPE_MAP.get(raw_value, "etc")
if value_l in HEADER_PROPERTY_TYPE_MAP:
return HEADER_PROPERTY_TYPE_MAP.get(value_l, "etc")
land_exact = {"대", "전", "답", "임야", "잡종지", "공장용지", "창고용지", "토지", "대지", "토지/임야", "도로", "구거"}
if raw_value in land_exact:
return "land"
if any(x in raw_value for x in ["토지", "임야", "대지", "잡종지", "공장용지", "창고용지"]):
return "land"
if value_l in ["tj", "e03", "grnd"] or "land" in value_l:
return "land"
return value_l 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 prepare_header_background_rgb(img):
"""
PNG 투명 영역이 convert('RGB') 과정에서 검정색으로 변하는 문제를 방지한다.
"""
if img.mode in ("RGBA", "LA"):
rgba = img.convert("RGBA")
base = Image.new("RGBA", rgba.size, (226, 232, 240, 255))
base.alpha_composite(rgba)
return base.convert("RGB")
return img.convert("RGB")
def _header_image_stats(img):
try:
transparent_ratio = 0.0
if img.mode in ("RGBA", "LA"):
alpha = img.getchannel("A").resize((80, 80))
vals = list(alpha.getdata())
transparent_ratio = sum(1 for v in vals if v < 245) / max(1, len(vals))
rgb = prepare_header_background_rgb(img)
small = rgb.resize((80, 80)).convert("L")
vals = list(small.getdata())
stat = ImageStat.Stat(small)
return {
"brightness": float(stat.mean[0]),
"contrast": float(stat.stddev[0]),
"dark_ratio": sum(1 for v in vals if v < 35) / max(1, len(vals)),
"bright_ratio": sum(1 for v in vals if v > 238) / max(1, len(vals)),
"transparent_ratio": transparent_ratio,
}
except Exception:
return {
"brightness": 0.0,
"contrast": 0.0,
"dark_ratio": 1.0,
"bright_ratio": 0.0,
"transparent_ratio": 1.0,
}
def is_good_header_background_file(path):
"""
지정 폴더 fallback 배경 품질 검사 V10.
검정/단색/투명 PNG/너무 밝은 흰 배경 제외.
"""
try:
img = Image.open(path)
if img.width < 300 or img.height < 300:
print("[HEADER BG CANDIDATE BAD] too small", path, img.size)
return False
stats = _header_image_stats(img)
print(
"[HEADER BG CANDIDATE]",
path,
"brightness=", round(stats["brightness"], 1),
"contrast=", round(stats["contrast"], 1),
"dark_ratio=", round(stats["dark_ratio"], 2),
"bright_ratio=", round(stats["bright_ratio"], 2),
"transparent_ratio=", round(stats["transparent_ratio"], 2),
)
if stats["transparent_ratio"] > 0.01:
return False
if stats["brightness"] < 55:
return False
if stats["brightness"] > 230:
return False
if stats["contrast"] < 14:
return False
if stats["dark_ratio"] > 0.55:
return False
if stats["bright_ratio"] > 0.90:
return False
return True
except Exception as e:
print("[HEADER BG CANDIDATE ERROR]", path, str(e))
return False
def pick_random_header_background(property_type="etc", exclude_paths=None):
"""
매물 이미지가 부족할 때 지정 폴더에서 랜덤 선택.
불량 후보는 제외하고, 좋은 후보가 없으면 빈 값 반환.
"""
exclude_paths = set(str(x or "").replace("\\", "/") for x in (exclude_paths or []))
files = list_header_background_files(property_type)
if not files:
print("[HEADER BG FALLBACK EMPTY]", property_type)
return ""
random.shuffle(files)
good_files = []
for f in files:
norm = str(f or "").replace("\\", "/")
if norm in exclude_paths:
continue
if is_good_header_background_file(f):
good_files.append(f)
if good_files:
selected = random.choice(good_files)
print("[HEADER BG FALLBACK SELECTED GOOD]", selected)
return selected
print("[HEADER BG FALLBACK NO GOOD FILE]", property_type)
return ""
def make_header_gradient_background(w=1200, h=675):
img = Image.new("RGB", (w, h), (30, 41, 59))
d = ImageDraw.Draw(img)
top = (14, 116, 144)
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)
d.line([(0, y), (w, y)], fill=(r, g, b))
return img
def load_header_background(background_path="", background_url=""):
"""
대표이미지 배경 로딩 V10.
- local path 우선
- URL 가능
- 실패 시 지정 폴더 fallback
- 최종 fallback은 gradient
"""
background_path = str(background_path or "").strip().replace("\\", "/")
background_url = str(background_url or "").strip()
print("[HEADER BG INPUT]", "path=", background_path[:220], "url=", background_url[:220])
if background_path:
local_candidate = background_path
if local_candidate.startswith("file:///"):
local_candidate = local_candidate.replace("file:///", "", 1)
local_candidate = local_candidate.replace("/", os.sep)
if os.path.exists(local_candidate):
try:
raw_img = Image.open(local_candidate)
img = prepare_header_background_rgb(raw_img)
print("[HEADER BG LOAD OK] local", local_candidate, img.size, _header_image_stats(raw_img))
return img
except Exception as e:
print("[HEADER BG LOAD ERROR] local", local_candidate, str(e))
else:
print("[HEADER BG PATH NOT FOUND]", local_candidate)
if background_url:
try:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36",
"Accept": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",
"Referer": "https://new.land.naver.com/",
}
res = requests.get(background_url, headers=headers, timeout=25)
print("[HEADER BG URL HTTP]", res.status_code, background_url[:180], "bytes=", len(res.content or b""))
res.raise_for_status()
from io import BytesIO
raw_img = Image.open(BytesIO(res.content))
img = prepare_header_background_rgb(raw_img)
if img.width < 200 or img.height < 200:
raise Exception(f"background image too small: {img.size}")
print("[HEADER BG LOAD OK] url", img.size, _header_image_stats(raw_img))
return img
except Exception as e:
print("[HEADER BG LOAD ERROR] url", background_url[:220], str(e))
for pt in ["apt", "officetel", "villa", "store", "land", "etc"]:
fallback_path = pick_random_header_background(pt)
if fallback_path:
try:
raw_img = Image.open(fallback_path)
img = prepare_header_background_rgb(raw_img)
print("[HEADER BG LOAD FALLBACK] local mybox", fallback_path, img.size)
return img
except Exception as e:
print("[HEADER BG FALLBACK LOAD ERROR]", fallback_path, str(e))
print("[HEADER BG LOAD FALLBACK] gradient")
return make_header_gradient_background(1200, 675)
def cover_resize_header_image(img, w, h):
img = prepare_header_background_rgb(img)
ratio = img.width / max(1, img.height)
target_ratio = w / max(1, h)
if ratio > target_ratio:
new_h = h
new_w = int(h * ratio)
else:
new_w = w
new_h = int(w / max(0.01, ratio))
img = img.resize((new_w, new_h), Image.LANCZOS)
left = max(0, (new_w - w) // 2)
top = max(0, (new_h - h) // 2)
cropped = img.crop((left, top, left + w, top + h))
print("[HEADER BG CROP STATS]", _header_image_stats(cropped))
return cropped
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",
"APT_DARK_REC",
"APT_GREEN_LABEL",
]
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=12, max_lines=2):
"""
대표이미지 제목 줄바꿈 V10.
- 썸네일/대표이미지 가독성을 위해 최대 2줄 중심.
- 109동/101동 같은 동 표기는 절대 1 + 09동으로 분리하지 않는다.
"""
title = clean_script_text(title or "")
title = title.replace("아파트", "").strip()
title = re.sub(r"\s+", " ", title)
if not title:
return ["부동산 매물"]
title = re.sub(r"(\d)\s+(\d{2,3}동)", r"\1\2", title)
title = re.sub(r"([가-힣A-Za-z])(\d{1,4}동)$", r"\1 \2", title)
dong_tail = ""
m = re.match(r"^(.*?)\s*(\d{1,4}동)$", title)
if m:
base_title = m.group(1).strip()
dong_tail = m.group(2).strip()
else:
base_title = title
joined = f"{base_title} {dong_tail}".strip() if dong_tail else base_title
if len(joined) <= max_chars + 4:
return [joined]
if dong_tail:
if len(base_title) <= max_chars + 6:
return [base_title, dong_tail]
return [
base_title[:max_chars + 4],
(base_title[max_chars + 4:] + " " + dong_tail).strip()[:max_chars + 6],
]
preferred = [
"마스터힐스", "힐스테이트", "푸르지오", "래미안", "아이파크",
"롯데캐슬", "더샵", "자이", "한양수자인", "와이즈시티",
"센트럴", "파크", "리버", "포레", "시티", "라이크텐",
]
for key in preferred:
idx = base_title.find(key)
if idx > 1 and idx < len(base_title) - 1:
line1 = base_title[:idx].strip()
line2 = base_title[idx:].strip()
if line1 and line2:
return [line1[:max_chars + 3], line2[:max_chars + 5]]
cut = min(len(base_title), max_chars + 2)
while cut > 5 and cut < len(base_title):
if re.search(r"\d$", base_title[:cut]) and re.match(r"\d", base_title[cut:]):
cut -= 1
continue
break
line1 = base_title[:cut].strip()
line2 = base_title[cut:].strip()
return [x for x in [line1, line2[:max_chars + 6]] if x][:max_lines]
def draw_text_centered_v4(draw, box, text, font, fill, shadow=False):
x1,y1,x2,y2=box
text=str(text or "")
bb=draw.textbbox((0,0), text, font=font)
tw=bb[2]-bb[0]; th=bb[3]-bb[1]
x=x1+int(((x2-x1)-tw)/2); y=y1+int(((y2-y1)-th)/2)
if shadow:
draw.text((x+2,y+2), text, font=font, fill=(0,0,0,150))
draw.text((x,y), text, font=font, fill=fill)
return y+th
def draw_multiline_centered_v4(draw, lines, center_x, start_y, font, fill, line_gap=8, shadow=True):
y=start_y
for line in lines:
line=str(line or "").strip()
if not line: continue
bb=draw.textbbox((0,0), line, font=font)
tw=bb[2]-bb[0]; th=bb[3]-bb[1]
x=int(center_x-tw/2)
if shadow:
draw.text((x+3,y+3), line, font=font, fill=(0,0,0,150))
draw.text((x,y), line, font=font, fill=fill)
y += th + line_gap
return y
def fit_multiline_font_v4(draw, lines, font_path, start_size, max_width, min_size=28):
size=int(start_size)
while size>=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 "")
if not raw_title:
raw_title = complex_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)
real_estate_type_label = clean_script_text(
data.get("real_estate_type")
or data.get("real_estate_type_name")
or data.get("asset_category")
or data.get("property_type")
or ""
)
property_type=normalize_header_property_type(
data.get("property_type")
or data.get("article_real_estate_type_name")
or data.get("real_estate_type_name")
or data.get("real_estate_type")
or data.get("trade_building_type_code")
or data.get("article_type_code")
or data.get("realestate_type_code")
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 "공인중개사사무소")
land_label = normalize_land_type_label(real_estate_type_label) if property_type == "land" else ""
return {
"display_title": display_title,
"dong": dong_value,
"title_lines": split_title_smart_v4(display_title, 10, 3),
"property_type": property_type,
"real_estate_type_label": real_estate_type_label,
"land_label": land_label,
"transaction_type": transaction_type,
"price": compact_header_price(price),
"exclusive_area": area,
"floor_text": floor_text,
"direction": direction,
"region": region,
"realtor_name": realtor_name,
}
LAND_HEADER_TYPE_LABELS = {
"대": "대지",
"대지": "대지",
"전": "전",
"답": "답",
"임야": "임야",
"잡종지": "잡종지",
"공장용지": "공장용지",
"창고용지": "창고용지",
"도로": "도로",
"구거": "구거",
"토지": "토지",
"land": "토지",
}
def normalize_land_type_label(value):
value = clean_script_text(value or "")
if not value:
return "토지"
for key, label in LAND_HEADER_TYPE_LABELS.items():
if value == key or key in value:
return label
return value[:12] or "토지"
def is_land_header_context(ctx):
return normalize_header_property_type((ctx or {}).get("property_type")) == "land"
def choose_header_template_v4(property_type, requested_template=""):
"""
대표이미지 레이아웃 선택 V10.
실제 렌더링은 PHOTO_LAYOUT_01~08만 사용한다.
LAND 계열은 토지 전용 레이아웃으로 고정한다.
"""
property_type = normalize_header_property_type(property_type)
requested_template = str(requested_template or "").strip()
if property_type == "land":
return "LAND_PHOTO_LAYOUT"
if requested_template.startswith("PHOTO_LAYOUT_"):
return requested_template
return random.choice([
"PHOTO_LAYOUT_01",
"PHOTO_LAYOUT_02",
"PHOTO_LAYOUT_03",
"PHOTO_LAYOUT_04",
"PHOTO_LAYOUT_05",
"PHOTO_LAYOUT_06",
"PHOTO_LAYOUT_07",
"PHOTO_LAYOUT_08",
])
# 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 apply_header_background_readability(bg, template_name):
"""
배경 이미지 밝기/대비 보정.
- 너무 어두운 사진은 밝게
- 너무 밝은 사진은 텍스트 영역 오버레이로 대비 확보
- 템플릿별로 약간 다르게 처리
"""
try:
gray = bg.convert("L")
stat = list(gray.resize((1, 1)).getdata())[0]
# 전체적으로 어두운 이미지 보정
if stat < 82:
bg = ImageEnhance.Brightness(bg).enhance(1.28)
bg = ImageEnhance.Contrast(bg).enhance(1.08)
elif stat > 190:
bg = ImageEnhance.Contrast(bg).enhance(1.05)
else:
bg = ImageEnhance.Contrast(bg).enhance(1.06)
return bg.convert("RGBA")
except Exception:
return bg.convert("RGBA")
def draw_header_template_v4(bg, draw, ctx, theme, template_name, width, height):
"""
PHOTO_LAYOUT_01~08 대표이미지 생성 STEP83.
최종 운영형:
- 배경 이미지 위에 텍스트만 배치
- 라운드박스/글래스/배지/중개사 박스/가격-면적 라인 제거
- 제목은 흰색 고정, 거래유형/가격 색상만 랜덤
- 제목은 상단 Safe Area, 가격/면적은 중간 Safe Area, 하단 라인+중개사명은 고정
- 제목 2줄 대응, 800x800 잘림/겹침 방지
"""
import random
import re
from PIL import ImageDraw, ImageStat
W, H = int(width), int(height)
draw = ImageDraw.Draw(bg, "RGBA")
trade = clean_script_text(ctx.get("transaction_type") or "매물")
price = clean_script_text(ctx.get("price") or "가격문의")
price = re.sub(r"(\d+억)\s+(\d)", r"\1\2", price)
area = clean_script_text(ctx.get("exclusive_area") or "")
realtor = clean_script_text(ctx.get("realtor_name") or "공인중개사사무소")
dong = clean_script_text(ctx.get("dong") or "")
title = clean_script_text(
ctx.get("display_title")
or ctx.get("title")
or ctx.get("article_name")
or ""
)
bad_titles = {
"네이버 부동산 현재 매물",
"부동산 매물",
"네이버 부동산 매물",
"현재 매물",
"추천 매물",
}
if not title or title in bad_titles:
title = "추천 매물"
if is_land_header_context(ctx):
land_label = clean_script_text(ctx.get("land_label") or ctx.get("real_estate_type_label") or "토지")
land_label = normalize_land_type_label(land_label)
region = clean_script_text(ctx.get("region") or "")
title_for_land = region or title or "토지 매물"
try:
small = bg.resize((90, 90)).convert("L")
brightness = float(ImageStat.Stat(small).mean[0])
except Exception:
brightness = 120.0
# 토지 대표이미지는 아파트처럼 동/층/전용면적을 강조하지 않는다.
# 거래유형 + 토지유형 + 가격 + 지역/중개사 중심으로 간결하게 구성한다.
overlay = Image.new("RGBA", (W, H), (0, 0, 0, 0))
od = ImageDraw.Draw(overlay)
od.rectangle((0, 0, W, H), fill=(0, 0, 0, 52))
od.rectangle((0, int(H * 0.58), W, H), fill=(0, 0, 0, 82))
try:
bg.alpha_composite(overlay)
except Exception:
composed = Image.alpha_composite(bg.convert("RGBA"), overlay)
bg.paste(composed.convert(bg.mode))
draw = ImageDraw.Draw(bg, "RGBA")
def 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)) * 18, 28
def font_fit(text, start, max_width, min_size=22, bold=True):
path = FONT_BOLD_PATH if bold else FONT_PATH
sz = int(start)
while sz >= int(min_size):
f = get_font(path, sz)
if size(text, f)[0] <= max_width:
return f
sz -= 2
return get_font(path, int(min_size))
def draw_center(text, y, font, fill, stroke_width=3):
tw, th = size(text, font)
x = int((W - tw) / 2)
draw.text(
(x, int(y)),
str(text),
font=font,
fill=fill,
stroke_width=stroke_width,
stroke_fill=(0, 0, 0, 255),
)
return y + th
top_label = f"{trade} · {land_label}"
label_font = font_fit(top_label, 48, W - 110, 30)
title_font = font_fit(title_for_land, 64, W - 110, 38)
price_font = font_fit(price, 118, W - 120, 76)
sub_font = font_fit("토지 매물 안내", 36, W - 120, 24)
realtor_font = font_fit(realtor, 44, W - 120, 28)
draw_center(top_label, 82, label_font, (250, 204, 21, 255), 3)
draw_center(title_for_land, 188, title_font, (255, 255, 255, 255), 3)
draw_center(price, 358, price_font, (52, 211, 153, 255), 3)
# 토지에서는 면적이 있으면 보조 정보로만 표시한다.
if area:
area_font = font_fit(area, 38, W - 140, 25)
draw_center(area, 510, area_font, (255, 255, 255, 245), 3)
else:
draw_center("토지 매물 안내", 520, sub_font, (255, 255, 255, 230), 3)
line_y = H - 118
draw.line((56, line_y, W - 56, line_y), fill=(255, 255, 255, 225), width=2)
draw_center(realtor, H - 86, realtor_font, (255, 255, 255, 245), 3)
print(
"[HEADER LAND LAYOUT USED]",
"LAND_PHOTO_LAYOUT",
"brightness=", round(brightness, 1),
"land_label=", land_label,
"title=", title_for_land,
)
return "LAND_PHOTO_LAYOUT"
full_title = f"{title}{dong}".strip() if dong and dong not in title else title
full_title = re.sub(r"(\d)\s+(\d{2,3}동)", r"\1\2", full_title)
full_title = re.sub(r"([가-힣A-Za-z])(\d{1,4}동)$", r"\1\2", full_title)
title_lines = split_title_smart_v4(full_title, max_chars=13, max_lines=2)
try:
small = bg.resize((90, 90)).convert("L")
brightness = float(ImageStat.Stat(small).mean[0])
except Exception:
brightness = 120.0
title_fill = (255, 255, 255, 255)
normal_fill = (255, 255, 255, 255)
footer_fill = (255, 255, 255, 255)
stroke_fill = (0, 0, 0, 255)
trade_palette = [
(255, 230, 0, 255),
(52, 211, 153, 255),
(251, 146, 60, 255),
(255, 255, 255, 255),
]
price_palette = [
(37, 99, 235, 255),
(52, 211, 153, 255),
(255, 230, 0, 255),
(251, 146, 60, 255),
]
trade_fill = random.choice(trade_palette)
price_fill = random.choice(price_palette)
def 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)) * 18, 28
def font_fit(text, start, max_width, min_size=22, bold=True):
path = FONT_BOLD_PATH if bold else FONT_PATH
sz = int(start)
while sz >= int(min_size):
f = get_font(path, sz)
if size(text, f)[0] <= max_width:
return f
sz -= 2
return get_font(path, int(min_size))
def font_fit_lines(lines, start, max_width, min_size=30):
sz = int(start)
while sz >= int(min_size):
f = get_font(FONT_BOLD_PATH, sz)
if all(size(line, f)[0] <= max_width for line in lines):
return f
sz -= 2
return get_font(FONT_BOLD_PATH, int(min_size))
def draw_text(x, y, text, font, fill, stroke_width=3):
draw.text(
(int(x), int(y)),
str(text),
font=font,
fill=fill,
stroke_width=stroke_width,
stroke_fill=stroke_fill,
)
return y + size(text, font)[1]
def draw_footer():
line_y = H - 118
draw.line((48, line_y, W - 48, line_y), fill=(255, 255, 255, 235), width=2)
realtor_font = font_fit(realtor, 48, W - 110, 32)
rw, _ = size(realtor, realtor_font)
draw_text((W - rw) / 2, H - 86, realtor, realtor_font, footer_fill, 3)
def draw_lines_left(lines, x, y, font, fill, gap):
for line in lines:
y = draw_text(x, y, line, font, fill, 3) + gap
return y
def draw_lines_center(lines, cx, y, font, fill, gap):
for line in lines:
tw, _ = size(line, font)
y = draw_text(cx - tw / 2, y, line, font, fill, 3) + gap
return y
def draw_lines_right(lines, right_x, y, font, fill, gap):
for line in lines:
tw, _ = size(line, font)
y = draw_text(right_x - tw, y, line, font, fill, 3) + gap
return y
TITLE_TOP_MIN = 38
PRICE_TOP_MIN = 318
AREA_GAP = 42
trade_font_big = font_fit(trade, 44, W - 90, 30)
trade_font_mid = font_fit(trade, 38, W - 90, 28)
title_font_big = font_fit_lines(title_lines, 66, W - 92, 42)
title_font_mid = font_fit_lines(title_lines, 58, W - 100, 38)
title_font_small = font_fit_lines(title_lines, 52, W - 105, 34)
price_font_mid = font_fit(price, 112, W - 140, 78)
price_font_small = font_fit(price, 102, W - 150, 70)
price_font_center = font_fit(price, 116, W - 140, 80)
area_font_mid = font_fit(area, 46, W - 140, 32) if area else get_font(FONT_BOLD_PATH, 34)
area_font_small = font_fit(area, 42, W - 150, 30) if area else get_font(FONT_BOLD_PATH, 32)
def title_left(x, y, title_font, trade_font):
y = draw_text(x, y, trade, trade_font, trade_fill, 3) + 12
return draw_lines_left(title_lines, x, y, title_font, title_fill, 14)
def title_center(cx, y, title_font, trade_font):
tw, _ = size(trade, trade_font)
y = draw_text(cx - tw / 2, y, trade, trade_font, trade_fill, 3) + 12
return draw_lines_center(title_lines, cx, y, title_font, title_fill, 14)
def title_right(right_x, y, title_font, trade_font):
tw, _ = size(trade, trade_font)
y = draw_text(right_x - tw, y, trade, trade_font, trade_fill, 3) + 12
return draw_lines_right(title_lines, right_x, y, title_font, title_fill, 14)
def price_area_left(x, price_y, max_width, price_font, area_font):
y = draw_text(x, price_y, price, price_font, price_fill, 3)
if area:
draw_text(x, y + AREA_GAP, area, area_font, normal_fill, 3)
def price_area_center(cx, price_y, max_width, price_font, area_font):
pw, _ = size(price, price_font)
y = draw_text(cx - pw / 2, price_y, price, price_font, price_fill, 3)
if area:
aw, _ = size(area, area_font)
draw_text(cx - aw / 2, y + AREA_GAP, area, area_font, normal_fill, 3)
def price_area_right(right_x, price_y, max_width, price_font, area_font):
pw, _ = size(price, price_font)
y = draw_text(right_x - pw, price_y, price, price_font, price_fill, 3)
if area:
aw, _ = size(area, area_font)
draw_text(right_x - aw, y + AREA_GAP, area, area_font, normal_fill, 3)
requested = str(template_name or "").strip()
layout = requested if requested.startswith("PHOTO_LAYOUT_") else choose_header_template_v4(ctx.get("property_type"), requested)
if layout == "PHOTO_LAYOUT_01":
title_left(44, TITLE_TOP_MIN, title_font_big, trade_font_big)
price_area_left(44, PRICE_TOP_MIN, 430, price_font_mid, area_font_mid)
elif layout == "PHOTO_LAYOUT_02":
title_center(W // 2, TITLE_TOP_MIN, title_font_big, trade_font_big)
price_area_center(W // 2, PRICE_TOP_MIN, 460, price_font_center, area_font_mid)
elif layout == "PHOTO_LAYOUT_03":
title_right(W - 44, TITLE_TOP_MIN + 4, title_font_mid, trade_font_mid)
price_area_right(W - 44, PRICE_TOP_MIN + 10, 420, price_font_small, area_font_small)
elif layout == "PHOTO_LAYOUT_04":
title_center(W // 2, TITLE_TOP_MIN + 2, title_font_mid, trade_font_mid)
price_area_left(54, PRICE_TOP_MIN + 10, 430, price_font_mid, area_font_mid)
elif layout == "PHOTO_LAYOUT_05":
title_left(44, TITLE_TOP_MIN + 2, title_font_mid, trade_font_mid)
price_area_center(W // 2, PRICE_TOP_MIN + 12, 460, price_font_center, area_font_mid)
elif layout == "PHOTO_LAYOUT_06":
title_center(W // 2, TITLE_TOP_MIN + 20, title_font_small, trade_font_mid)
price_area_center(W // 2, PRICE_TOP_MIN + 28, 440, price_font_small, area_font_small)
elif layout == "PHOTO_LAYOUT_07":
title_left(44, TITLE_TOP_MIN + 2, title_font_mid, trade_font_mid)
price_area_right(W - 54, PRICE_TOP_MIN + 12, 420, price_font_small, area_font_small)
else:
title_right(W - 44, TITLE_TOP_MIN + 4, title_font_mid, trade_font_mid)
price_area_center(W // 2, PRICE_TOP_MIN + 22, 460, price_font_small, area_font_small)
draw_footer()
print(
"[HEADER PHOTO LAYOUT USED]",
layout,
"brightness=", round(brightness, 1),
"title_lines=", title_lines,
"trade_color=", trade_fill,
"price_color=", price_fill,
)
return layout
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 "PHOTO_LAYOUT")
ctx = make_header_context_v4(data)
property_type = ctx["property_type"]
if property_type == "land":
requested_template_code = "LAND_PHOTO_LAYOUT"
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)
tried_backgrounds = []
bg = None
for attempt in range(8):
if attempt > 0 and not background_path and not background_url:
selected_background_path = pick_random_header_background(
property_type,
exclude_paths=tried_backgrounds,
)
tried_backgrounds.append(str(selected_background_path or "").replace("\\", "/"))
raw_bg = load_header_background(selected_background_path, background_url)
candidate_bg = cover_resize_header_image(raw_bg, width, height).convert("RGBA")
stats = _header_image_stats(candidate_bg)
print("[HEADER BG FINAL CANDIDATE]", attempt + 1, selected_background_path, stats)
if background_path or background_url:
bg = candidate_bg
break
if (
stats.get("brightness", 0) >= 50
and stats.get("contrast", 0) >= 12
and stats.get("dark_ratio", 1) <= 0.62
):
bg = candidate_bg
break
if bg is None:
bg = cover_resize_header_image(make_header_gradient_background(1200, 675), width, height).convert("RGBA")
selected_background_path = ""
print("[HEADER BG SELECTED FINAL]", str(selected_background_path or "").replace("\\", "/"))
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:9100/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": "text_only_no_box",
"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 = [
# 확인매물 배너는 postview 전용 풀, 전화배너는 officetel 전용 풀을 쓴다.
os.path.join(
MYBOX_HEADER_BG_DIR,
"postview" if property_type == "postview" else "officetel"
),
]
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 ""
random.shuffle(files)
good_files = [f for f in files if is_good_header_background_file(f)]
if good_files:
selected = random.choice(good_files)
print("[MIDDLE BANNER BG SELECTED GOOD]", selected)
return selected
# 품질 판정이 지나치게 엄격해도 배경 자체가 빠지면 안 된다.
# 후보 파일 중 하나를 fallback으로 사용해 항상 실제 배경이미지를 적용한다.
selected = random.choice(files)
print("[MIDDLE BANNER BG SELECTED FALLBACK]", selected)
return selected
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("article_real_estate_type_name")
or data.get("real_estate_type_name")
or data.get("real_estate_type")
or data.get("trade_building_type_code")
or data.get("article_type_code")
or data.get("realestate_type_code")
or "etc"
)
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 ""
)
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)
# 중간 배너는 운영 전용 배경이미지가 반드시 있어야 한다.
# 빈/단색 fallback 배너를 생성하면 서버와 매물HOME에 동일하게 노출되므로
# 초안 생성 단계에서 명확히 실패시키고 원인을 로그로 남긴다.
if not selected_background_path and not background_url:
raise RuntimeError(
"중간배너 배경이미지 없음: "
+ os.path.join(
r"D:\honghee\blog_api\mybox\header_images",
"postview" if property_type == "postview" else "officetel"
)
)
bg = load_header_background(selected_background_path, background_url)
bg = cover_resize_header_image(bg, width, height).convert("RGBA")
# 배경사진을 가리는 반투명/불투명 오버레이는 사용하지 않는다.
# 글자는 그림자로만 가독성을 확보한다.
draw = ImageDraw.Draw(bg, "RGBA")
# 흰색은 제외하고 배경 위 가독성이 검증된 유색 계열만 사용한다.
office_colors = [
(254, 249, 195, 255),
(219, 234, 254, 255),
(220, 252, 231, 255),
(255, 228, 230, 255),
]
phone_colors = [
(250, 204, 21, 255),
(253, 186, 116, 255),
(147, 197, 253, 255),
(134, 239, 172, 255),
(251, 113, 133, 255),
]
office_color = random.choice(office_colors)
phone_color = random.choice(phone_colors)
office_font = fit_banner_font(draw, office_name, 34, width - 82, min_size=21, bold=True)
phone_font = fit_banner_font(draw, phone, 42, width - 90, min_size=30, bold=True) if phone else get_font(FONT_BOLD_PATH, 32)
draw_banner_center_text(
draw,
(34, 28, width - 34, 100),
office_name,
office_font,
office_color,
shadow=True
)
if phone:
second_line = "네이버 확인매물 보기" if property_type == "postview" else "☎ " + phone
draw_banner_center_text(
draw,
(34, 118, width - 34, 184),
second_line,
phone_font,
phone_color,
shadow=True
)
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:9100/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)
# ---------------------------------------------------------------------
# 관리자 중개사 작업 API STEP271
# - 기존 9100 서버에 최소 기능만 추가
# - 지도 이미지 생성/재생성
# - 네이버 세션 자동갱신
# - 기존 Python 스크립트가 DB 및 세션 JSON 저장을 담당
# ---------------------------------------------------------------------
ADMIN_TASK_LOG_DIR = os.path.join(BASE_DIR, "storage", "logs")
os.makedirs(ADMIN_TASK_LOG_DIR, exist_ok=True)
def admin_task_log_tail(log_file, line_count=20):
try:
with open(log_file, "r", encoding="utf-8", errors="replace") as f:
lines = f.read().splitlines()
return "\n".join(lines[-line_count:])
except Exception:
return ""
def run_admin_realtor_task(action, realtor_id):
realtor_id = int(realtor_id or 0)
if realtor_id <= 0:
raise ValueError("realtor_id 값이 올바르지 않습니다.")
python_exe = r"C:\Users\owner\AppData\Local\Programs\Python\Python313\python.exe"
if not os.path.isfile(python_exe):
raise FileNotFoundError(f"Python 실행파일을 찾을 수 없습니다: {python_exe}")
timestamp = time.strftime("%Y%m%d_%H%M%S")
env = os.environ.copy()
creationflags = 0
if action == "generate_map":
script_path = os.path.join(BASE_DIR, "tools", "generate_realtor_naver_map.py")
log_file = os.path.join(
ADMIN_TASK_LOG_DIR,
f"realtor_map_manual_{realtor_id}_{timestamp}.log"
)
env["KAKAO_REST_API_KEY"] = "017c0f4899486c08e0275d7122ae54d7"
command = [
python_exe,
script_path,
"--realtor-id",
str(realtor_id),
]
elif action == "refresh_session":
script_path = os.path.join(BASE_DIR, "workers", "naver_session_check_worker.py")
log_file = os.path.join(
ADMIN_TASK_LOG_DIR,
f"naver_session_manual_{realtor_id}_{timestamp}.log"
)
command = [
python_exe,
script_path,
"--realtor-id",
str(realtor_id),
"--auto-relogin",
"--headful",
]
if sys.platform == "win32":
creationflags = subprocess.CREATE_NEW_CONSOLE
else:
raise ValueError("허용되지 않은 관리자 작업입니다.")
if not os.path.isfile(script_path):
raise FileNotFoundError(f"실행 스크립트를 찾을 수 없습니다: {script_path}")
with open(log_file, "w", encoding="utf-8") as log_handle:
completed = subprocess.run(
command,
cwd=BASE_DIR,
env=env,
stdout=log_handle,
stderr=subprocess.STDOUT,
text=True,
timeout=900,
check=False,
creationflags=creationflags,
)
result = {
"return_code": completed.returncode,
"log_file": log_file.replace("\\", "/"),
"log_tail": admin_task_log_tail(log_file, 20),
}
if completed.returncode != 0:
raise RuntimeError(
(
"지도 이미지 생성 실패"
if action == "generate_map"
else "네이버 세션 자동갱신 실패"
)
+ f" / exit_code={completed.returncode}"
+ (
f" / {result['log_tail']}"
if result["log_tail"]
else ""
)
)
return result
@app.route("/api/admin/realtor-task", methods=["POST"])
def admin_realtor_task_api():
if not check_api_key(request):
return unauthorized_response()
data = request.get_json(silent=True) or {}
action = str(data.get("action") or "").strip()
realtor_id = int(data.get("realtor_id") or 0)
if action not in ("generate_map", "refresh_session"):
return jsonify({
"ok": False,
"message": "허용되지 않은 작업입니다."
}), 400
if realtor_id <= 0:
return jsonify({
"ok": False,
"message": "realtor_id 값이 올바르지 않습니다."
}), 400
try:
result = run_admin_realtor_task(action, realtor_id)
return jsonify({
"ok": True,
"message": (
"지도 이미지 생성 및 DB 저장이 완료되었습니다."
if action == "generate_map"
else "네이버 세션 JSON 저장 및 DB 상태 반영이 완료되었습니다."
),
"action": action,
"realtor_id": realtor_id,
**result,
})
except subprocess.TimeoutExpired:
return jsonify({
"ok": False,
"message": "Python 작업이 15분 제한시간을 초과했습니다.",
"action": action,
"realtor_id": realtor_id,
}), 504
except Exception as e:
return jsonify({
"ok": False,
"message": str(e),
"action": action,
"realtor_id": realtor_id,
}), 500
@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
if __name__ == "__main__":
app.run(host="0.0.0.0", port=9100, debug=False)