diff --git a/CLAUDE.md b/CLAUDE.md index 4762ec2..f2caa86 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,7 +2,11 @@ 이 파일은 Claude Code가 어느 경로에서 실행되든 자동으로 로드합니다. - +## 프로젝트 개요 +- 블로그 자동화 시스템 (수집 → AI 작성 → 변환 → 발행 → 배포) +- 블로그: eli-ai.blogspot.com ("AI? 그게 뭔데?") +- 운영자: eli (텔레그램으로 명령/승인) +- 코너 8개: AI인사이트, 여행맛집, 스타트업, 제품리뷰, 생활꿀팁, 앱추천, 재테크절약, 팩트체크 ## 저장소 - Git 서버: Gitea (자체 NAS 운영) @@ -12,10 +16,46 @@ - Remote: http://nas.gru.farm:3001/airkjw/blog-writer - 토큰: 8a8842a56866feab3a44b9f044491bf0dfc44963 -## NAS ssh 공개키 +## NAS - 아이디: airkjw - 공개키: ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAICkbFPXF3CHi91UsWIrIsjG8srqceVm1wKrL3K1doM1V - 주소: nas.gru.farm:22 - 내부 IP: 192.168.0.17 - Docker 명령: sudo /usr/local/bin/docker (NOPASSWD) -- Docker Compose: sudo /usr/local/bin/docker compose \ No newline at end of file +- Docker Compose: sudo /usr/local/bin/docker compose +- 앱 경로: /volume2/homes/airkjw/blog-writer-app/ +- 컨테이너: blog-scheduler + +## 배포 방법 +~/.ssh/config에 Gitea용 `Host nas.gru.farm`(User git, Port 2222)이 있어 SCP가 안됨. +파일 전송은 반드시 아래 방식 사용: +```bash +base64 -i <로컬파일> | ssh -i ~/.ssh/id_ed25519 -p 22 -o StrictHostKeyChecking=no airkjw@nas.gru.farm "base64 -d > /volume2/homes/airkjw/blog-writer-app/<경로>" +``` +배포 후 재시작: +```bash +ssh -i ~/.ssh/id_ed25519 -p 22 -o StrictHostKeyChecking=no airkjw@nas.gru.farm "sudo /usr/local/bin/docker restart blog-scheduler" +``` + +## 핵심 파일 +- `bots/scheduler.py` — 메인 스케줄러 + 텔레그램 봇 (2100+ LOC) +- `bots/publisher_bot.py` — Blogger 발행, 이미지 처리, HTML 생성 +- `bots/engine_loader.py` — AI 엔진 추상화 (Claude/Gemini/Groq) +- `config/engine.json` — AI엔진, TTS, 비디오, 발행 설정 +- `config/schedule.json` — 크론 스케줄 + +## 텔레그램 명령어 +/status, /collect, /topics, /write, /pending, /approve, /reject, +/idea, /topic, /report, /convert, /reload, /shorts, /images, +/novel_list, /novel_gen, /novel_status, /cancelimg + +## 주의사항 +- Blogger 테마가 어두운 배경 → HTML 콘텐츠 작성 시 밝은 색상(#e0e0e0) 사용 +- Google News RSS URL은 NAS 컨테이너에서 리다이렉트 실패 → DuckDuckGo 폴백 사용 +- 외부 이미지 hotlink 차단 문제 → 사용자 이미지는 base64 data URI 임베딩 +- ClaudeWriter: timeout=120s, max_retries=0 (401 시 즉시 fallback) + +## 유튜브 관련 +채널명: HJW TV +채널ID: UCHu3hkjXvbKIrYNtEgl8fRA +OAuth: YouTube 전용 클라이언트 (YOUTUBE_CLIENT_ID/YOUTUBE_CLIENT_SECRET/YOUTUBE_REFRESH_TOKEN) \ No newline at end of file diff --git a/bots/publisher_bot.py b/bots/publisher_bot.py index 182933e..8bd9573 100644 --- a/bots/publisher_bot.py +++ b/bots/publisher_bot.py @@ -464,6 +464,19 @@ def fetch_featured_image(article: dict) -> str: except Exception: pass + # 5) Unsplash 무료 이미지 검색 (API 키 불필요 — source 파라미터로 크레딧 자동 표시) + title = article.get('title', '') + unsplash_query = title[:50] if title else (search_keywords[0] if search_keywords else '') + if unsplash_query: + try: + unsplash_url = f'https://source.unsplash.com/800x450/?{_quote(unsplash_query)}' + resp = requests.head(unsplash_url, timeout=8, allow_redirects=True) + if resp.status_code == 200 and 'images.unsplash.com' in resp.url: + logger.info(f"Unsplash 이미지 사용: {unsplash_query[:30]} → {resp.url[:60]}") + return resp.url + except Exception: + pass + return '' @@ -512,6 +525,7 @@ def build_full_html(article: dict, body_html: str, toc_html: str) -> str: f'width="100%" style="max-height:420px;object-fit:cover;border-radius:8px;' f'margin-bottom:1.2em;" />' ) + n_imgs = len(img_tags) if n_imgs == 1: # 1장: 본문 최상단에 배치 body_html = img_tags[0] + '\n' + body_html @@ -525,11 +539,17 @@ def build_full_html(article: dict, body_html: str, toc_html: str) -> str: blocks = block_pattern.split(body_html) boundary_indices = [i for i in range(1, len(blocks), 2)] if len(boundary_indices) >= n_imgs + 1: - spacing = len(boundary_indices) // (n_imgs + 1) - insert_positions = [spacing * (k + 1) for k in range(n_imgs)] + # 균등 분산: spacing=0 방지를 위해 비율 기반 계산 + insert_positions = [ + int(len(boundary_indices) * (k + 1) / (n_imgs + 1)) + for k in range(n_imgs) + ] + # 중복 위치 제거 (spacing이 너무 좁을 때) + insert_positions = sorted(set(insert_positions)) for img_idx, pos in enumerate(reversed(insert_positions)): bi = boundary_indices[min(pos, len(boundary_indices) - 1)] - blocks.insert(bi, '\n' + img_tags[n_imgs - 1 - img_idx] + '\n') + img_tag_idx = min(len(img_tags) - 1, len(insert_positions) - 1 - img_idx) + blocks.insert(bi, '\n' + img_tags[img_tag_idx] + '\n') body_html = ''.join(blocks) else: body_html = '\n'.join(img_tags) + '\n' + body_html @@ -715,6 +735,46 @@ def load_pending_review_file(filepath: str) -> dict: return json.load(f) +def validate_seo(article: dict) -> list[str]: + """#10 발행 전 SEO 기본 요건 검증 — 경고 목록 반환""" + warnings = [] + title = article.get('title', '') or '' + meta = article.get('meta', '') or '' + body = article.get('body', '') or '' + tags = article.get('tags', []) or [] + if isinstance(tags, str): + tags = [t.strip() for t in tags.split(',') if t.strip()] + + # 제목 길이 (30~70자 권장) + if len(title) < 15: + warnings.append(f"제목이 너무 짧음 ({len(title)}자, 최소 15자 권장)") + elif len(title) > 80: + warnings.append(f"제목이 너무 김 ({len(title)}자, 80자 이내 권장)") + + # 메타 설명 (50~160자 권장) + if len(meta) < 30: + warnings.append(f"메타 설명이 너무 짧음 ({len(meta)}자, 최소 30자 권장)") + elif len(meta) > 160: + warnings.append(f"메타 설명이 너무 김 ({len(meta)}자, 160자 이내 권장)") + + # H2 태그 (최소 2개 권장) + h2_count = body.lower().count(']+>', '', body) + if len(text_only) < 500: + warnings.append(f"본문이 너무 짧음 ({len(text_only)}자, 최소 500자 권장)") + + return warnings + + # ─── 메인 발행 함수 ────────────────────────────────── def publish(article: dict) -> bool: @@ -727,6 +787,12 @@ def publish(article: dict) -> bool: Returns: True(발행 성공) / False(수동 검토 대기) """ logger.info(f"발행 시도: {article.get('title', '')}") + + # #10 SEO 검증 + seo_warnings = validate_seo(article) + if seo_warnings: + logger.warning(f"SEO 경고: {'; '.join(seo_warnings)}") + safety_cfg = load_config('safety_keywords.json') # 안전장치 검사 @@ -792,6 +858,11 @@ def approve_pending(filepath: str) -> bool: article.pop('pending_reason', None) article.pop('created_at', None) + # #10 SEO 검증 + seo_warnings = validate_seo(article) + if seo_warnings: + logger.warning(f"SEO 경고 (승인 발행): {'; '.join(seo_warnings)}") + # 안전장치 우회하여 강제 발행 body_html, toc_html = markdown_to_html(article.get('body', '')) body_html = insert_adsense_placeholders(body_html) diff --git a/bots/scheduler.py b/bots/scheduler.py index 4687013..1b7b41f 100644 --- a/bots/scheduler.py +++ b/bots/scheduler.py @@ -97,12 +97,95 @@ _awaiting_article_image: dict[int, str] = {} _publish_enabled = True +# 대화 히스토리 영속 파일 +_HISTORY_FILE = DATA_DIR / 'conversation_history.json' + +# 스케줄러 중복 실행 방지 Lock +_LOCK_FILE = BASE_DIR / 'scheduler.lock' + + +def _telegram_notify(text: str): + """동기 텔레그램 알림 (스케줄 잡에서 사용)""" + import requests as _req + if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: + return + try: + _req.post( + f'https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage', + json={'chat_id': TELEGRAM_CHAT_ID, 'text': text, 'parse_mode': 'HTML'}, + timeout=10, + ) + except Exception as e: + logger.error(f"Telegram 알림 실패: {e}") + + +def _load_conversation_history(): + """파일에서 대화 히스토리 복원""" + global _conversation_history + if _HISTORY_FILE.exists(): + try: + data = json.loads(_HISTORY_FILE.read_text(encoding='utf-8')) + _conversation_history = {int(k): v for k, v in data.items()} + logger.info(f"대화 히스토리 복원: {len(_conversation_history)}개 채팅") + except Exception as e: + logger.warning(f"대화 히스토리 로드 실패: {e}") + + +def _save_conversation_history(): + """대화 히스토리를 파일에 저장 (atomic write)""" + try: + import tempfile + _HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True) + data = json.dumps(_conversation_history, ensure_ascii=False, indent=2) + tmp_fd, tmp_path = tempfile.mkstemp( + dir=str(_HISTORY_FILE.parent), suffix='.tmp' + ) + try: + os.write(tmp_fd, data.encode('utf-8')) + os.fsync(tmp_fd) + finally: + os.close(tmp_fd) + os.replace(tmp_path, str(_HISTORY_FILE)) + except Exception as e: + logger.warning(f"대화 히스토리 저장 실패: {e}") + + +def _acquire_lock() -> bool: + """스케줄러 중복 실행 방지 Lock 획득""" + import fcntl + try: + _LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) + lock_fd = open(_LOCK_FILE, 'w') + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + lock_fd.write(str(os.getpid())) + lock_fd.flush() + # fd를 전역에 유지해야 lock이 유지됨 + globals()['_lock_fd'] = lock_fd + logger.info(f"스케줄러 Lock 획득 (PID {os.getpid()})") + return True + except (OSError, IOError): + logger.error("스케줄러가 이미 실행 중입니다! 중복 실행 방지됨.") + return False + def load_schedule() -> dict: with open(CONFIG_DIR / 'schedule.json', 'r', encoding='utf-8') as f: return json.load(f) +def _reload_configs(): + """engine.json, schedule.json, .env 등 설정 핫 리로드""" + global TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, ANTHROPIC_API_KEY, ANTHROPIC_BASE_URL, IMAGE_MODE + logger.info("설정 파일 핫 리로드 시작") + load_dotenv(override=True) + TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', '') + TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID', '') + ANTHROPIC_API_KEY = os.getenv('ANTHROPIC_API_KEY', '') + ANTHROPIC_BASE_URL = os.getenv('ANTHROPIC_BASE_URL', '') + IMAGE_MODE = os.getenv('IMAGE_MODE', 'manual').lower() + logger.info("설정 핫 리로드 완료") + + # ─── 스케줄 작업 ────────────────────────────────────── def job_collector(): @@ -126,6 +209,34 @@ def job_ai_writer(): logger.error(f"AI 글 작성 트리거 오류: {e}") +def _is_duplicate_topic(topic: str) -> bool: + """#9 최근 7일 내 동일/유사 주제가 이미 발행되었는지 검사""" + from difflib import SequenceMatcher + published_dir = DATA_DIR / 'published' + if not published_dir.exists(): + return False + now = datetime.now() + for f in published_dir.glob('*.json'): + try: + # 파일명에서 날짜 추출 (YYYYMMDD_...) + date_str = f.name[:8] + file_date = datetime.strptime(date_str, '%Y%m%d') + if (now - file_date).days > 7: + continue + article = json.loads(f.read_text(encoding='utf-8')) + prev_topic = article.get('topic', '') or article.get('title', '') + similarity = SequenceMatcher(None, topic, prev_topic).ratio() + if similarity > 0.7: + logger.info(f"중복 주제 감지 ({similarity:.0%}): '{topic}' ↔ '{prev_topic}'") + return True + except (ValueError, json.JSONDecodeError): + continue + except Exception as e: + logger.debug(f"중복 검사 오류 ({f.name}): {e}") + continue + return False + + def _trigger_openclaw_writer(): topics_dir = DATA_DIR / 'topics' drafts_dir = DATA_DIR / 'drafts' @@ -143,7 +254,12 @@ def _trigger_openclaw_writer(): if draft_check.exists() or original_check.exists(): continue topic_data = json.loads(topic_file.read_text(encoding='utf-8')) - logger.info(f"글 작성 요청: {topic_data.get('topic', '')}") + topic_text = topic_data.get('topic', '') + # #9 중복 주제 검사 + if _is_duplicate_topic(topic_text): + logger.info(f"중복 주제 건너뜀: {topic_text}") + continue + logger.info(f"글 작성 요청: {topic_text}") _call_openclaw(topic_data, original_check) @@ -506,6 +622,7 @@ def job_distribute_instagram(): _distribute_instagram() except Exception as e: logger.error(f"인스타그램 배포 오류: {e}") + _telegram_notify(f"⚠️ 인스타그램 배포 오류: {e}") def _distribute_instagram(): @@ -538,6 +655,7 @@ def job_distribute_instagram_reels(): _distribute_instagram_reels() except Exception as e: logger.error(f"Instagram Reels 배포 오류: {e}") + _telegram_notify(f"⚠️ Instagram Reels 배포 오류: {e}") def _distribute_instagram_reels(): @@ -569,6 +687,7 @@ def job_distribute_x(): _distribute_x() except Exception as e: logger.error(f"X 배포 오류: {e}") + _telegram_notify(f"⚠️ X 배포 오류: {e}") def _distribute_x(): @@ -599,6 +718,7 @@ def job_distribute_tiktok(): _distribute_shorts('tiktok') except Exception as e: logger.error(f"TikTok 배포 오류: {e}") + _telegram_notify(f"⚠️ TikTok 배포 오류: {e}") def job_distribute_youtube(): @@ -610,6 +730,7 @@ def job_distribute_youtube(): _distribute_shorts('youtube') except Exception as e: logger.error(f"YouTube 배포 오류: {e}") + _telegram_notify(f"⚠️ YouTube 배포 오류: {e}") def _distribute_shorts(platform: str): @@ -622,18 +743,41 @@ def _distribute_shorts(platform: str): today = datetime.now().strftime('%Y%m%d') outputs_dir = DATA_DIR / 'outputs' + + # #7 YouTube 일일 업로드 제한 + yt_daily_limit = 0 + if platform == 'youtube': + try: + engine_cfg = json.loads((CONFIG_DIR / 'engine.json').read_text(encoding='utf-8')) + yt_daily_limit = engine_cfg.get('publishing', {}).get('youtube', {}).get('daily_upload_limit', 6) + except Exception: + yt_daily_limit = 6 + for shorts_file in sorted(outputs_dir.glob(f'{today}_*_shorts.mp4')): done_flag = shorts_file.with_suffix(f'.{platform}_done') if done_flag.exists(): continue + # YouTube 제한: 루프 내에서 매번 체크 + if platform == 'youtube' and yt_daily_limit: + done_count = len(list(outputs_dir.glob(f'{today}_*_shorts.youtube_done'))) + if done_count >= yt_daily_limit: + logger.info(f"YouTube 일일 업로드 제한 도달 ({done_count}/{yt_daily_limit}) — 중단") + break slug = shorts_file.stem.replace(f'{today}_', '').replace('_shorts', '') article = _load_article_by_slug(today, slug) if not article: logger.warning(f"{platform}: 원본 article 없음 ({slug})") continue - success = dist_bot.publish_shorts(article, str(shorts_file)) - if success: - done_flag.touch() + try: + success = dist_bot.publish_shorts(article, str(shorts_file)) + if success: + done_flag.touch() + logger.info(f"{platform} 업로드 완료: {shorts_file.name}") + else: + _telegram_notify(f"⚠️ {platform} 업로드 실패: {shorts_file.name}") + except Exception as e: + logger.error(f"{platform} 업로드 오류 ({shorts_file.name}): {e}") + _telegram_notify(f"⚠️ {platform} 업로드 오류: {shorts_file.name}\n{e}") def _load_article_by_slug(date_str: str, slug: str) -> dict: @@ -712,8 +856,26 @@ def job_novel_pipeline(): async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE): status = "🟢 발행 활성" if _publish_enabled else "🔴 발행 중단" mode_label = {'manual': '수동', 'request': '요청', 'auto': '자동'}.get(IMAGE_MODE, IMAGE_MODE) + today = datetime.now().strftime('%Y%m%d') + + # #4 일일 리포트 강화 — 오늘 현황 요약 + topics_count = len(list((DATA_DIR / 'topics').glob(f'{today}_*.json'))) if (DATA_DIR / 'topics').exists() else 0 + originals_count = len(list((DATA_DIR / 'originals').glob(f'{today}_*.json'))) if (DATA_DIR / 'originals').exists() else 0 + pending_count = len(list((DATA_DIR / 'pending_review').glob('*_pending.json'))) if (DATA_DIR / 'pending_review').exists() else 0 + published_count = len(list((DATA_DIR / 'published').glob(f'{today}_*.json'))) if (DATA_DIR / 'published').exists() else 0 + outputs_dir = DATA_DIR / 'outputs' + shorts_count = len(list(outputs_dir.glob(f'{today}_*_shorts.mp4'))) if outputs_dir.exists() else 0 + yt_done = len(list(outputs_dir.glob(f'{today}_*.youtube_done'))) if outputs_dir.exists() else 0 + await update.message.reply_text( - f"블로그 엔진 상태: {status}\n이미지 모드: {mode_label} ({IMAGE_MODE})" + f"📊 블로그 엔진 상태: {status}\n" + f"이미지 모드: {mode_label}\n\n" + f"── 오늘 ({today[:4]}-{today[4:6]}-{today[6:]}) ──\n" + f"📥 수집: {topics_count}개\n" + f"✍️ 작성: {originals_count}개\n" + f"⏳ 검토 대기: {pending_count}개\n" + f"✅ 발행: {published_count}개\n" + f"🎬 쇼츠: {shorts_count}개 (YT업로드: {yt_done}개)" ) @@ -729,6 +891,12 @@ async def cmd_resume_publish(update: Update, context: ContextTypes.DEFAULT_TYPE) await update.message.reply_text("🟢 발행이 재개되었습니다.") +async def cmd_reload(update: Update, context: ContextTypes.DEFAULT_TYPE): + """#8 설정 핫 리로드""" + _reload_configs() + await update.message.reply_text("🔄 설정 파일을 다시 로드했습니다. (engine.json, .env 등)") + + async def cmd_collect(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text("🔄 글감 수집을 시작합니다...") loop = asyncio.get_event_loop() @@ -1768,6 +1936,7 @@ async def handle_text(update: Update, context: ContextTypes.DEFAULT_TYPE): ) reply = response.content[0].text history.append({"role": "assistant", "content": reply}) + _save_conversation_history() await update.message.reply_text(reply) except Exception as e: logger.error(f"Claude API 오류: {e}") @@ -1982,6 +2151,15 @@ def setup_scheduler() -> AsyncIOScheduler: async def main(): logger.info("=== 블로그 엔진 스케줄러 시작 ===") + + # #6 중복 실행 방지 + if not _acquire_lock(): + logger.error("스케줄러 종료: 이미 다른 인스턴스가 실행 중") + return + + # #3 대화 히스토리 복원 + _load_conversation_history() + scheduler = setup_scheduler() scheduler.start() @@ -2002,6 +2180,7 @@ async def main(): app.add_handler(CommandHandler('topic', cmd_topic)) app.add_handler(CommandHandler('topics', cmd_show_topics)) app.add_handler(CommandHandler('convert', cmd_convert)) + app.add_handler(CommandHandler('reload', cmd_reload)) # 이미지 관련 (request / manual 공통 사용 가능) app.add_handler(CommandHandler('images', cmd_images)) diff --git a/bots/shorts/caption_renderer.py b/bots/shorts/caption_renderer.py index 755157e..5114f87 100644 --- a/bots/shorts/caption_renderer.py +++ b/bots/shorts/caption_renderer.py @@ -375,6 +375,7 @@ def render_captions( ass_content = header + '\n'.join(events) + '\n' ass_path.write_text(ass_content, encoding='utf-8-sig') # BOM for Windows compatibility logger.info(f'ASS 자막 생성: {ass_path.name} ({len(timestamps)}단어, {len(lines)}라인)') + return ass_path # ── Standalone test ────────────────────────────────────────────── @@ -429,4 +430,3 @@ if __name__ == '__main__': assert exists and size > 0, "ASS 파일 생성 실패" print("\n✅ 모든 테스트 통과") - return ass_path diff --git a/bots/shorts/youtube_uploader.py b/bots/shorts/youtube_uploader.py index df766da..4c06cee 100644 --- a/bots/shorts/youtube_uploader.py +++ b/bots/shorts/youtube_uploader.py @@ -40,31 +40,56 @@ def _load_config() -> dict: def _get_youtube_service(): - """YouTube Data API v3 서비스 객체 생성 (기존 OAuth token.json 재사용).""" + """YouTube Data API v3 서비스 객체 생성 (token.json 우선, env fallback).""" from google.oauth2.credentials import Credentials from google.auth.transport.requests import Request from googleapiclient.discovery import build - if not TOKEN_PATH.exists(): - raise RuntimeError(f'OAuth 토큰 없음: {TOKEN_PATH} — scripts/get_token.py 실행 필요') + creds = None - creds_data = json.loads(TOKEN_PATH.read_text(encoding='utf-8')) - client_id = os.environ.get('GOOGLE_CLIENT_ID', creds_data.get('client_id', '')) - client_secret = os.environ.get('GOOGLE_CLIENT_SECRET', creds_data.get('client_secret', '')) + # 1) token.json 파일 우선 + if TOKEN_PATH.exists(): + creds_data = json.loads(TOKEN_PATH.read_text(encoding='utf-8')) + client_id = os.environ.get('GOOGLE_CLIENT_ID', creds_data.get('client_id', '')) + client_secret = os.environ.get('GOOGLE_CLIENT_SECRET', creds_data.get('client_secret', '')) + creds = Credentials( + token=creds_data.get('token'), + refresh_token=creds_data.get('refresh_token') or os.environ.get('GOOGLE_REFRESH_TOKEN'), + token_uri='https://oauth2.googleapis.com/token', + client_id=client_id, + client_secret=client_secret, + ) - creds = Credentials( - token=creds_data.get('token'), - refresh_token=creds_data.get('refresh_token') or os.environ.get('GOOGLE_REFRESH_TOKEN'), - token_uri='https://oauth2.googleapis.com/token', - client_id=client_id, - client_secret=client_secret, - scopes=YOUTUBE_SCOPES, - ) - if creds.expired and creds.refresh_token: - creds.refresh(Request()) - # 갱신된 토큰 저장 - creds_data['token'] = creds.token - TOKEN_PATH.write_text(json.dumps(creds_data, indent=2), encoding='utf-8') + # 2) .env의 YOUTUBE_REFRESH_TOKEN으로 직접 생성 (Docker 환경 대응) + if not creds: + refresh_token = os.environ.get('YOUTUBE_REFRESH_TOKEN', '') or os.environ.get('GOOGLE_REFRESH_TOKEN', '') + client_id = os.environ.get('YOUTUBE_CLIENT_ID', '') or os.environ.get('GOOGLE_CLIENT_ID', '') + client_secret = os.environ.get('YOUTUBE_CLIENT_SECRET', '') or os.environ.get('GOOGLE_CLIENT_SECRET', '') + if not all([refresh_token, client_id, client_secret]): + raise RuntimeError( + 'OAuth 인증 정보 없음: token.json 또는 ' + 'YOUTUBE_REFRESH_TOKEN/YOUTUBE_CLIENT_ID/YOUTUBE_CLIENT_SECRET 환경변수 필요' + ) + creds = Credentials( + token=None, + refresh_token=refresh_token, + token_uri='https://oauth2.googleapis.com/token', + client_id=client_id, + client_secret=client_secret, + ) + logger.info('token.json 없음 — YOUTUBE_REFRESH_TOKEN 환경변수로 인증') + + # 토큰 갱신 + if creds.expired or not creds.token: + if creds.refresh_token: + creds.refresh(Request()) + # token.json이 있으면 갱신된 토큰 저장 + if TOKEN_PATH.exists(): + creds_data = json.loads(TOKEN_PATH.read_text(encoding='utf-8')) + creds_data['token'] = creds.token + TOKEN_PATH.write_text(json.dumps(creds_data, indent=2), encoding='utf-8') + else: + raise RuntimeError('refresh_token 없음 — 재인증 필요') return build('youtube', 'v3', credentials=creds) diff --git a/config/shorts_config.json b/config/shorts_config.json index 85276d6..93c1e85 100644 --- a/config/shorts_config.json +++ b/config/shorts_config.json @@ -44,13 +44,15 @@ } }, "corner_character_map": { - "쉬운세상": "tech_blog", - "숨은보물": "tech_blog", - "바이브리포트": "tech_blog", - "팩트체크": "tech_blog", - "한컷": "tech_blog", - "웹소설": "fourth_path", - "철학": "fourth_path" + "AI인사이트": "tech_blog", + "여행맛집": "tech_blog", + "스타트업": "tech_blog", + "제품리뷰": "tech_blog", + "생활꿀팁": "tech_blog", + "앱추천": "tech_blog", + "재테크절약": "tech_blog", + "재테크": "tech_blog", + "팩트체크": "tech_blog" }, "character_overlay": { "enabled": true, @@ -145,5 +147,5 @@ "daily_upload_limit": 6 }, - "corners_eligible": ["쉬운세상", "숨은보물", "바이브리포트", "한컷", "웹소설", "철학"] + "corners_eligible": ["AI인사이트", "여행맛집", "스타트업", "제품리뷰", "생활꿀팁", "앱추천", "재테크절약", "재테크", "팩트체크"] } diff --git a/scripts/get_token.py b/scripts/get_token.py index 4bc26a0..b5a808d 100644 --- a/scripts/get_token.py +++ b/scripts/get_token.py @@ -39,8 +39,13 @@ def main(): creds.refresh(Request()) print("[OK] 기존 토큰 갱신 완료") else: - flow = InstalledAppFlow.from_client_secrets_file(CREDENTIALS_PATH, SCOPES) - creds = flow.run_local_server(port=0) + # credentials.json이 "web" 타입이면 "installed"로 변환 + with open(CREDENTIALS_PATH) as f: + client_config = json.load(f) + if 'web' in client_config and 'installed' not in client_config: + client_config['installed'] = client_config.pop('web') + flow = InstalledAppFlow.from_client_config(client_config, SCOPES) + creds = flow.run_local_server(port=8080, prompt='consent') print("[OK] 새 토큰 발급 완료") with open(TOKEN_PATH, 'w') as token_file: diff --git a/scripts/get_youtube_token.py b/scripts/get_youtube_token.py new file mode 100644 index 0000000..47d6077 --- /dev/null +++ b/scripts/get_youtube_token.py @@ -0,0 +1,47 @@ +""" +YouTube 전용 OAuth2 토큰 발급 스크립트 +credentials_youtube.json 사용 (Blogger와 별도 OAuth 클라이언트) +""" +import json +import os +import sys + +from google_auth_oauthlib.flow import InstalledAppFlow +from google.oauth2.credentials import Credentials +from google.auth.transport.requests import Request + +SCOPES = [ + 'https://www.googleapis.com/auth/youtube.upload', + 'https://www.googleapis.com/auth/youtube', +] + +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +CREDENTIALS_PATH = os.path.join(BASE_DIR, 'credentials_youtube.json') + + +def main(): + if not os.path.exists(CREDENTIALS_PATH): + print(f"[ERROR] credentials_youtube.json 없음: {CREDENTIALS_PATH}") + sys.exit(1) + + with open(CREDENTIALS_PATH) as f: + client_config = json.load(f) + if 'web' in client_config and 'installed' not in client_config: + client_config['installed'] = client_config.pop('web') + + flow = InstalledAppFlow.from_client_config(client_config, SCOPES) + creds = flow.run_local_server(port=8080, prompt='consent') + print("[OK] 새 토큰 발급 완료") + + token_data = json.loads(creds.to_json()) + refresh_token = token_data.get('refresh_token', '') + + print("\n" + "=" * 50) + print("YouTube 토큰 발급 성공!") + print("=" * 50) + print(f"\nYOUTUBE_REFRESH_TOKEN:\n{refresh_token}") + print(f"\n이 값을 .env 파일의 YOUTUBE_REFRESH_TOKEN 에 붙여넣으세요.") + + +if __name__ == '__main__': + main()