{disclaimer}
') return '\n'.join(html_parts) # ─── Blogger API ────────────────────────────────────── def publish_to_blogger(article: dict, html_content: str, creds: Credentials) -> dict: """Blogger API v3로 글 발행""" service = build('blogger', 'v3', credentials=creds) blog_id = BLOG_MAIN_ID labels = [article.get('corner', '')] tags = article.get('tags', []) if isinstance(tags, str): tags = [t.strip() for t in tags.split(',')] labels.extend(tags) labels = list(set(filter(None, labels))) body = { 'title': article.get('title', ''), 'content': html_content, 'labels': labels, } result = service.posts().insert( blogId=blog_id, body=body, isDraft=False, ).execute() return result def submit_to_search_console(url: str, creds: Credentials): """Google Search Console URL 색인 요청""" try: service = build('searchconsole', 'v1', credentials=creds) # URL Inspection API (실제 indexing 요청) # 참고: 일반적으로 Blogger sitemap이 자동 제출되므로 보조 수단 logger.info(f"Search Console 제출: {url}") # indexing API는 별도 서비스 계정 필요. 여기서는 로그만 남김. # 실제 색인 촉진은 Blogger 내장 sitemap에 의존 except Exception as e: logger.warning(f"Search Console 제출 실패: {e}") # ─── Telegram ──────────────────────────────────────── def send_telegram(text: str, parse_mode: str = 'HTML'): """Telegram 메시지 전송""" if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: logger.warning("Telegram 설정 없음 — 알림 건너뜀") return url = f'https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage' payload = { 'chat_id': TELEGRAM_CHAT_ID, 'text': text, 'parse_mode': parse_mode, } try: resp = requests.post(url, json=payload, timeout=10) resp.raise_for_status() except Exception as e: logger.error(f"Telegram 전송 실패: {e}") def send_pending_review_alert(article: dict, reason: str): """수동 검토 대기 알림 (Telegram)""" title = article.get('title', '(제목 없음)') corner = article.get('corner', '') preview = article.get('body', '')[:300].replace('<', '<').replace('>', '>') msg = ( f"🔍 [수동 검토 필요]\n\n" f"📌 {title}\n" f"코너: {corner}\n" f"사유: {reason}\n\n" f"미리보기:\n{preview}...\n\n" f"명령:승인 또는 거부"
)
send_telegram(msg)
# ─── 발행 이력 ───────────────────────────────────────
def log_published(article: dict, post_result: dict):
"""발행 이력 저장"""
published_dir = DATA_DIR / 'published'
published_dir.mkdir(exist_ok=True)
record = {
'title': article.get('title', ''),
'corner': article.get('corner', ''),
'url': post_result.get('url', ''),
'post_id': post_result.get('id', ''),
'published_at': datetime.now(timezone.utc).isoformat(),
'quality_score': article.get('quality_score', 0),
'tags': article.get('tags', []),
'sources': article.get('sources', []),
}
filename = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{record['post_id']}.json"
with open(published_dir / filename, 'w', encoding='utf-8') as f:
json.dump(record, f, ensure_ascii=False, indent=2)
return record
def save_pending_review(article: dict, reason: str):
"""수동 검토 대기 글 저장"""
pending_dir = DATA_DIR / 'pending_review'
pending_dir.mkdir(exist_ok=True)
record = {**article, 'pending_reason': reason, 'created_at': datetime.now().isoformat()}
filename = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_pending.json"
with open(pending_dir / filename, 'w', encoding='utf-8') as f:
json.dump(record, f, ensure_ascii=False, indent=2)
return pending_dir / filename
def load_pending_review_file(filepath: str) -> dict:
with open(filepath, 'r', encoding='utf-8') as f:
return json.load(f)
# ─── 메인 발행 함수 ──────────────────────────────────
def publish(article: dict) -> bool:
"""
article: OpenClaw blog-writer가 출력한 파싱된 글 dict
{
title, meta, slug, tags, corner, body (markdown),
coupang_keywords, sources, disclaimer, quality_score
}
Returns: True(발행 성공) / False(수동 검토 대기)
"""
logger.info(f"발행 시도: {article.get('title', '')}")
safety_cfg = load_config('safety_keywords.json')
# 안전장치 검사
needs_review, review_reason = check_safety(article, safety_cfg)
if needs_review:
logger.warning(f"수동 검토 대기: {review_reason}")
save_pending_review(article, review_reason)
send_pending_review_alert(article, review_reason)
return False
# 마크다운 → HTML
body_html, toc_html = markdown_to_html(article.get('body', ''))
# AdSense 플레이스홀더
body_html = insert_adsense_placeholders(body_html)
# 최종 HTML 조합
full_html = build_full_html(article, body_html, toc_html)
# Google 인증
try:
creds = get_google_credentials()
except RuntimeError as e:
logger.error(str(e))
return False
# Blogger 발행
try:
post_result = publish_to_blogger(article, full_html, creds)
post_url = post_result.get('url', '')
logger.info(f"발행 완료: {post_url}")
except Exception as e:
logger.error(f"Blogger 발행 실패: {e}")
return False
# Search Console 제출
if post_url:
submit_to_search_console(post_url, creds)
# 발행 이력 저장
log_published(article, post_result)
# Telegram 알림
title = article.get('title', '')
corner = article.get('corner', '')
send_telegram(
f"✅ 발행 완료!\n\n"
f"📌 {title}\n"
f"코너: {corner}\n"
f"URL: {post_url}"
)
return True
def approve_pending(filepath: str) -> bool:
"""수동 검토 대기 글 승인 후 발행"""
try:
article = load_pending_review_file(filepath)
article.pop('pending_reason', None)
article.pop('created_at', None)
# 안전장치 우회하여 강제 발행
body_html, toc_html = markdown_to_html(article.get('body', ''))
body_html = insert_adsense_placeholders(body_html)
full_html = build_full_html(article, body_html, toc_html)
creds = get_google_credentials()
post_result = publish_to_blogger(article, full_html, creds)
post_url = post_result.get('url', '')
log_published(article, post_result)
# 대기 파일 삭제
Path(filepath).unlink(missing_ok=True)
send_telegram(
f"✅ [수동 승인] 발행 완료!\n\n"
f"📌 {article.get('title', '')}\n"
f"URL: {post_url}"
)
logger.info(f"수동 승인 발행 완료: {post_url}")
return True
except Exception as e:
logger.error(f"승인 발행 실패: {e}")
return False
def reject_pending(filepath: str):
"""수동 검토 대기 글 거부 (파일 삭제)"""
try:
article = load_pending_review_file(filepath)
Path(filepath).unlink(missing_ok=True)
send_telegram(f"🗑 [거부] {article.get('title', '')} — 폐기됨")
logger.info(f"수동 검토 거부: {filepath}")
except Exception as e:
logger.error(f"거부 처리 실패: {e}")
def get_pending_list() -> list[dict]:
"""수동 검토 대기 목록 반환"""
pending_dir = DATA_DIR / 'pending_review'
pending_dir.mkdir(exist_ok=True)
result = []
for f in sorted(pending_dir.glob('*_pending.json')):
try:
data = json.loads(f.read_text(encoding='utf-8'))
data['_filepath'] = str(f)
result.append(data)
except Exception:
pass
return result
if __name__ == '__main__':
# 테스트용: 샘플 아티클 발행 시도
sample = {
'title': '테스트 글',
'meta': '테스트 메타 설명',
'slug': 'test-article',
'tags': ['테스트', 'AI'],
'corner': '쉬운세상',
'body': '## 제목\n\n본문 내용입니다.\n\n## 결론\n\n마무리입니다.',
'coupang_keywords': ['키보드'],
'sources': [
{'url': 'https://example.com/1', 'title': '출처1', 'date': '2026-03-24'},
{'url': 'https://example.com/2', 'title': '출처2', 'date': '2026-03-24'},
],
'disclaimer': '',
'quality_score': 80,
}
result = publish(sample)
print('발행 결과:', result)