From 96e63226cf829dfedc1098990cdb632c10a6aeb6 Mon Sep 17 00:00:00 2001 From: n8n Blog Runner Date: Sat, 4 Jul 2026 12:47:40 +0900 Subject: [PATCH 1/5] feat: add n8n collect-only topic scout --- AGENTS.md | 2 +- CLAUDE.md | 2 +- scripts/collect_topics.py | 88 +++++++++++++++++ scripts/n8n_codex_blog_runner.py | 140 ++++++++++++++++++++++----- scripts/test_humanizer_gate.py | 31 ++++++ scripts/trend_writer.py | 160 +++++++++++++++---------------- 6 files changed, 311 insertions(+), 112 deletions(-) create mode 100644 scripts/collect_topics.py diff --git a/AGENTS.md b/AGENTS.md index 32e2842..6d4d674 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,7 +31,7 @@ python trend_writer.py # 로컬 Codex CLI 인증 필요 ### 이중 파이프라인 1. **배포 파이프라인** (`.github/workflows/deploy.yml`): `main` 푸시 시 Hugo 빌드 → GitHub Pages 배포 -2. **자동 포스트 생성** (Mac mini n8n): `POST /run {"track":"issue"}`는 1시간마다 화제성 이슈 글감 확인, `POST /run {"track":"tech"}`는 6시간마다 넓은 기술 주제 글감 확인 → `scripts/trend_writer.py`가 트랙별 RSS 수집 → Codex CLI로 한국어 포스트 생성 → Hugo 빌드 +2. **자동 포스트 생성** (Mac mini n8n): `POST /run {"track":"issue"}`는 1시간마다 화제성 이슈 후보 수집(collectOnly) + 4시간마다 발행, `POST /run {"track":"tech"}`는 4시간마다 넓은 기술 주제 발행(cron `0 35 2,6,10,14,18,22 * * *`, 하루 6회) + 1시간마다 후보 수집 → `scripts/trend_writer.py`가 트랙별 RSS 수집 → Codex CLI로 한국어 포스트 생성 → Hugo 빌드 ### Trend Writer 처리 흐름 diff --git a/CLAUDE.md b/CLAUDE.md index a1c2c48..0c65b66 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -31,7 +31,7 @@ python trend_writer.py # 로컬 Codex CLI 인증 필요 ### 이중 파이프라인 1. **배포 파이프라인** (`.github/workflows/deploy.yml`): `main` 푸시 시 Hugo 빌드 → GitHub Pages 배포 -2. **자동 포스트 생성** (Mac mini n8n): `POST /run {"track":"issue"}`는 1시간마다 화제성 이슈 글감 확인, `POST /run {"track":"tech"}`는 6시간마다 넓은 기술 주제 글감 확인 → `scripts/trend_writer.py`가 트랙별 RSS 수집 → Codex CLI로 한국어 포스트 생성 → Hugo 빌드 +2. **자동 포스트 생성** (Mac mini n8n): `POST /run {"track":"issue"}`는 1시간마다 화제성 이슈 후보 수집(collectOnly) + 4시간마다 발행, `POST /run {"track":"tech"}`는 4시간마다 넓은 기술 주제 발행(cron `0 35 2,6,10,14,18,22 * * *`, 하루 6회) + 1시간마다 후보 수집 → `scripts/trend_writer.py`가 트랙별 RSS 수집 → Codex CLI로 한국어 포스트 생성 → Hugo 빌드 ### Trend Writer 처리 흐름 diff --git a/scripts/collect_topics.py b/scripts/collect_topics.py new file mode 100644 index 0000000..dc6905b --- /dev/null +++ b/scripts/collect_topics.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +"""Collect topic candidates without generating or publishing posts.""" + +from __future__ import annotations + +import argparse +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import trend_writer + + +SCRIPT_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIR.parent +SNAPSHOT_DIR = REPO_ROOT / "local_drafts" / "topic_snapshots" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Collect blog topic candidates only.") + parser.add_argument("--track", choices=sorted(trend_writer.TRACKS), default=trend_writer.DEFAULT_TRACK) + parser.add_argument("--limit", type=int, default=30) + return parser.parse_args() + + +def article_uid(article: dict[str, Any]) -> str: + if hasattr(trend_writer, "article_uid"): + return trend_writer.article_uid(article) + return str(article.get("uid") or article.get("link") or article.get("title") or "") + + +def compact_article(article: dict[str, Any]) -> dict[str, Any]: + return { + "uid": article_uid(article), + "source": article.get("source", ""), + "title": article.get("title", ""), + "link": article.get("link", ""), + "published": article.get("published", ""), + "summary": article.get("summary", "")[:500], + } + + +def main() -> None: + args = parse_args() + track = args.track + limit = max(1, args.limit) + + trend_writer.configure_track(track) + feeds = trend_writer.filter_feeds_for_track(trend_writer.load_feeds()) + articles = trend_writer.fetch_recent_articles(feeds) + if track == "issue" and hasattr(trend_writer, "fetch_github_trending_articles"): + articles.extend(trend_writer.fetch_github_trending_articles()) + + seen = trend_writer.load_seen() + fresh = [article for article in articles if article_uid(article) not in seen] + candidates = fresh[:limit] + + now = datetime.now(timezone.utc) + SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True) + relpath = Path("local_drafts") / "topic_snapshots" / f"{now.strftime('%Y%m%d-%H%M%S')}-{track}.json" + outpath = REPO_ROOT / relpath + outpath.write_text( + json.dumps( + { + "schemaVersion": 1, + "track": track, + "collectedAt": now.isoformat(), + "feedCount": len(feeds), + "totalCount": len(articles), + "freshCount": len(fresh), + "candidates": [compact_article(article) for article in candidates], + }, + ensure_ascii=False, + indent=2, + ), + encoding="utf-8", + ) + + print(f"COLLECTED_FILE={relpath.as_posix()}") + print(f"CANDIDATE_COUNT={len(candidates)}") + print(f"FRESH_COUNT={len(fresh)}") + for index, article in enumerate(candidates[:5], start=1): + print(f"CANDIDATE_{index}={article.get('source', '')} | {article.get('title', '')}") + + +if __name__ == "__main__": + main() diff --git a/scripts/n8n_codex_blog_runner.py b/scripts/n8n_codex_blog_runner.py index a5fdd88..fcd485c 100644 --- a/scripts/n8n_codex_blog_runner.py +++ b/scripts/n8n_codex_blog_runner.py @@ -7,6 +7,7 @@ import os import re import signal +import socketserver import subprocess import threading import urllib.parse @@ -33,6 +34,14 @@ _lock = threading.Lock() +class LocalThreadingHTTPServer(ThreadingHTTPServer): + def server_bind(self) -> None: + socketserver.TCPServer.server_bind(self) + host, port = self.server_address[:2] + self.server_name = str(host) + self.server_port = int(port) + + def _terminate_process_group(process: subprocess.Popen, grace_seconds: float = 5.0) -> None: if process.poll() is not None: return @@ -106,6 +115,21 @@ def _created_files(step: dict[str, Any]) -> list[str]: return _publish_paths(files) +def _collected_file(step: dict[str, Any]) -> str: + stdout = str(step.get("stdoutTail") or "") + match = re.search(r"^COLLECTED_FILE=(.+)$", stdout, flags=re.MULTILINE) + if not match: + return "" + files = _publish_paths([match.group(1)]) + return files[0] if files else "" + + +def _collected_count(step: dict[str, Any]) -> int: + stdout = str(step.get("stdoutTail") or "") + match = re.search(r"^CANDIDATE_COUNT=(\d+)$", stdout, flags=re.MULTILINE) + return int(match.group(1)) if match else 0 + + def _frontmatter_value(lines: list[str], key: str) -> str: prefix = f"{key}:" for line in lines[:40]: @@ -130,6 +154,55 @@ def _latest_scores(count: int) -> list[str]: return [f"{score:.1f}" for _, score in sorted(scores)[-count:]] +def _blog_alert(track: str, ok: bool, paths: list[str], *, failed_step: str = "", error: str = "") -> str: + posts = [Path(path) for path in _publish_paths(paths) if path.startswith("content/posts/")] + scores = _latest_scores(len(posts)) + label = "Gnosys 이슈" if track == "issue" else "Gnosys 기술" + lines = [ + f"{'✅' if ok else '❌'} {label} 자동 발행", + f"상태: {'성공' if ok else '실패'}", + f"점수: {', '.join(score + '점' for score in scores) if scores else 'runner 응답 점수 없음'}", + f"발행글: {len(posts)}개", + ] + if failed_step: + lines.append(f"실패 단계: {failed_step}") + if error: + lines.append(f"오류: {error}") + for index, post in enumerate(posts, start=1): + content = (REPO_ROOT / post).read_text(errors="replace").splitlines() + title = _frontmatter_value(content, "title") or post.stem + description = _frontmatter_value(content, "description") + lines.append(f"\n{index}. {title}") + if description: + lines.append(f"요약: {description}") + lines.append(f"파일: {post.as_posix()}") + return "\n".join(lines) + + +def _collect_alert( + track: str, + ok: bool, + collected_file: str = "", + candidate_count: int = 0, + *, + failed_step: str = "", + error: str = "", +) -> str: + label = "Gnosys 이슈" if track == "issue" else "Gnosys 기술" + lines = [ + f"{'✅' if ok else '❌'} {label} 후보 수집", + f"상태: {'성공' if ok else '실패'}", + f"후보: {candidate_count}개", + f"파일: {collected_file or '없음'}", + "발행글: 0개", + ] + if failed_step: + lines.append(f"실패 단계: {failed_step}") + if error: + lines.append(f"오류: {error}") + return "\n".join(lines) + + def _telegram_chat_id(token: str, env: dict[str, str]) -> str: chat_id = env.get("TELEGRAM_CHAT_ID", "").strip() if chat_id: @@ -150,21 +223,12 @@ def _send_telegram_notification(track: str, paths: list[str], env: dict[str, str if not token: return {"name": "send telegram notification", "command": ["telegram", "sendMessage"], "exitCode": 1, "startedAt": started_at, "stdoutTail": "", "stderrTail": "TELEGRAM_BOT_TOKEN missing"} - posts = [Path(path) for path in _publish_paths(paths) if path.startswith("content/posts/")] - scores = _latest_scores(len(posts)) - lines = [f"n8n blog publish complete ({track})", f"published posts: {len(posts)}", ""] - for index, post in enumerate(posts): - content = (REPO_ROOT / post).read_text(errors="replace").splitlines() - score = scores[index] if index < len(scores) else "unknown" - lines.append(f"- {_frontmatter_value(content, 'title')} ({score}점)") - description = _frontmatter_value(content, "description") - if description: - lines.append(f" {description}") - data = urllib.parse.urlencode({"chat_id": _telegram_chat_id(token, env), "text": "\n".join(lines)}) + text = _blog_alert(track, True, paths) + data = urllib.parse.urlencode({"chat_id": _telegram_chat_id(token, env), "text": text}) request = urllib.request.Request(f"https://api.telegram.org/bot{token}/sendMessage", data=data.encode("utf-8")) with urllib.request.urlopen(request, timeout=15) as response: response.read() - return {"name": "send telegram notification", "command": ["telegram", "sendMessage"], "exitCode": 0, "startedAt": started_at, "stdoutTail": "\n".join(lines), "stderrTail": ""} + return {"name": "send telegram notification", "command": ["telegram", "sendMessage"], "exitCode": 0, "startedAt": started_at, "stdoutTail": text, "stderrTail": ""} def _publish_blog_changes(track: str, paths: list[str], env: dict[str, str]) -> list[dict[str, Any]]: @@ -212,6 +276,7 @@ def run_blog_job( notify: bool = False, publish_paths: list[str] | None = None, publish_only: bool = False, + collect_only: bool = False, ) -> tuple[int, dict[str, Any]]: if not _lock.acquire(blocking=False): return 409, {"ok": False, "error": "blog runner is already running"} @@ -227,20 +292,39 @@ def run_blog_job( commands = [] if "BLOG_RUNNER_PYTHON" not in os.environ: commands.append(("prepare python venv", [SYSTEM_PYTHON_BIN, "-m", "venv", str(VENV_DIR)])) - commands.extend([ - ("install dependencies", [PYTHON_BIN, "-m", "pip", "install", "-q", "-r", "scripts/requirements.txt"]), - ("verify humanizer gate", [PYTHON_BIN, "-B", "scripts/test_humanizer_gate.py"]), - ("build hugo site", [HUGO_BIN, "--minify"]), - ]) - if not publish_only: + commands.append(("install dependencies", [PYTHON_BIN, "-m", "pip", "install", "-q", "-r", "scripts/requirements.txt"])) + if collect_only: + commands.append(("collect topic candidates", [PYTHON_BIN, "-B", "scripts/collect_topics.py", "--track", track])) + else: + commands.extend([ + ("verify humanizer gate", [PYTHON_BIN, "-B", "scripts/test_humanizer_gate.py"]), + ("build hugo site", [HUGO_BIN, "--minify"]), + ]) + if not collect_only and not publish_only: commands.insert(-1, ("write post with codex", [PYTHON_BIN, "scripts/trend_writer.py", "--track", track])) for name, command in commands: step = _run_step(name, command, env) steps.append(step) if step["exitCode"] != 0: - return 500, {"ok": False, "track": track, "failedStep": name, "steps": steps} - + alert = _collect_alert(track, False, failed_step=name) if collect_only else _blog_alert(track, False, [], failed_step=name) + return 500, {"ok": False, "track": track, "collectOnly": collect_only, "failedStep": name, "alert": alert, "steps": steps} + + if collect_only: + collect_step = next(step for step in steps if step["name"] == "collect topic candidates") + collected_file = _collected_file(collect_step) + candidate_count = _collected_count(collect_step) + return 200, { + "ok": True, + "track": track, + "collectOnly": True, + "collectedFile": collected_file, + "candidateCount": candidate_count, + "alert": _collect_alert(track, True, collected_file, candidate_count), + "steps": steps, + } + + paths: list[str] = [] if publish or publish_only: paths = list(publish_paths or []) if not publish_only: @@ -248,18 +332,21 @@ def run_blog_job( for step in _publish_blog_changes(track, paths, env): steps.append(step) if step["exitCode"] != 0: - return 500, {"ok": False, "track": track, "failedStep": step["name"], "steps": steps} + return 500, {"ok": False, "track": track, "failedStep": step["name"], "createdFiles": paths, "alert": _blog_alert(track, False, paths, failed_step=step["name"]), "steps": steps} if notify: step = _send_telegram_notification(track, paths, env) steps.append(step) if step["exitCode"] != 0: - return 500, {"ok": False, "track": track, "failedStep": step["name"], "steps": steps} + return 500, {"ok": False, "track": track, "failedStep": step["name"], "createdFiles": paths, "alert": _blog_alert(track, False, paths, failed_step=step["name"]), "steps": steps} - return 200, {"ok": True, "track": track, "steps": steps} + return 200, {"ok": True, "track": track, "createdFiles": paths, "alert": _blog_alert(track, True, paths), "steps": steps} except subprocess.TimeoutExpired as exc: - return 504, {"ok": False, "track": track, "error": f"timeout after {TIMEOUT_SECONDS}s", "step": exc.cmd, "steps": steps} + error = f"timeout after {TIMEOUT_SECONDS}s" + alert = _collect_alert(track, False, error=error) if collect_only else _blog_alert(track, False, [], error=error) + return 504, {"ok": False, "track": track, "collectOnly": collect_only, "error": error, "step": exc.cmd, "alert": alert, "steps": steps} except Exception as exc: - return 500, {"ok": False, "track": track, "error": str(exc), "steps": steps} + alert = _collect_alert(track, False, error=str(exc)) if collect_only else _blog_alert(track, False, [], error=str(exc)) + return 500, {"ok": False, "track": track, "collectOnly": collect_only, "error": str(exc), "alert": alert, "steps": steps} finally: _lock.release() @@ -307,6 +394,7 @@ def do_POST(self) -> None: publish=bool(body.get("publish")), notify=bool(body.get("notify")), publish_only=bool(body.get("publishOnly")), + collect_only=bool(body.get("collectOnly")), publish_paths=[str(path) for path in publish_paths], ) _json_response(self, status, payload) @@ -316,7 +404,7 @@ def main() -> None: if HOST not in {"127.0.0.1", "localhost", "::1"} and not TOKEN: raise SystemExit("BLOG_RUNNER_TOKEN is required when BLOG_RUNNER_HOST is not loopback") - server = ThreadingHTTPServer((HOST, PORT), BlogRunnerHandler) + server = LocalThreadingHTTPServer((HOST, PORT), BlogRunnerHandler) print(f"n8n Codex blog runner listening on http://{HOST}:{PORT}", flush=True) server.serve_forever() diff --git a/scripts/test_humanizer_gate.py b/scripts/test_humanizer_gate.py index e59afd9..dbd16c8 100644 --- a/scripts/test_humanizer_gate.py +++ b/scripts/test_humanizer_gate.py @@ -72,6 +72,28 @@ def _call_name(node: ast.AST) -> str: assert "회사 기술 블로그나 공식 문서는 주장 검증과 사례 보강에만 사용" in source assert "issue 트랙" in source assert "tech 트랙" in source +# Named failure taxonomy: rubric and judges must reference concrete pattern codes. +assert "실패 패턴 점검 목록" in source +assert "stale_structure" in source +assert "weak_community_angle" in source +assert "thin_evidence" in source +assert "generic_conclusion" in source +assert "missing_counterpoint" in source +assert "single_source_summary" in source +assert "ai_tone" in source +assert "구조 다양성 원칙" in source +# Flexible functional-beat structure replaced the fixed H2 skeleton. +assert "기능적 비트" in source +assert "## 왜 지금 이슈인가" not in source +assert "## 커뮤니티에서 갈리는 지점" not in source +assert "## 아키텍처 관점에서 볼 점" not in source +assert "## 실무에서 볼 점" not in source +assert "## 무슨 일이 있었나" not in source +# AI-tell guards exist both at generation time and in the humanizer gate. +assert source.count("상투적 전환어") >= 2 +assert source.count("수사 의문문") >= 2 +assert "귀추가 주목된다" in source +assert "새 사실, 새 수치, 새 경험담을 추가하지 말 것" in source assert "대표 원문은 중심축" not in source assert "GEMINI" not in source assert "google.genai" not in source @@ -99,3 +121,12 @@ def _call_name(node: ast.AST) -> str: ] assert candidate_calls.index("generate_post") < candidate_calls.index("humanize_post") + +runner_source = Path(__file__).with_name("n8n_codex_blog_runner.py").read_text(encoding="utf-8") +collector_source = Path(__file__).with_name("collect_topics.py").read_text(encoding="utf-8") + +assert "collectOnly" in runner_source +assert "scripts/collect_topics.py" in runner_source +assert "candidateCount" in runner_source +assert "COLLECTED_FILE" in collector_source +assert "CANDIDATE_COUNT" in collector_source diff --git a/scripts/trend_writer.py b/scripts/trend_writer.py index b7aeec1..6d8bc35 100644 --- a/scripts/trend_writer.py +++ b/scripts/trend_writer.py @@ -77,12 +77,13 @@ "contrarian-check", ) ISSUE_CODE_RULES = ( - ("single_source_summary", ("단일", "요약", "single", "source", "회사", "article")), - ("weak_community_angle", ("커뮤니티", "논쟁", "반응", "hn", "reddit", "github")), - ("generic_conclusion", ("결론", "일반적", "generic", "뻔", "추상")), - ("missing_counterpoint", ("반론", "counter", "찬반", "균형", "우려")), - ("thin_evidence", ("근거", "수치", "사례", "evidence", "출처")), - ("stale_structure", ("구조", "반복", "나열", "템플릿", "흐름")), + ("single_source_summary", ("single_source_summary", "단일", "요약", "single", "source", "회사", "article")), + ("weak_community_angle", ("weak_community_angle", "커뮤니티", "논쟁", "반응", "hn", "reddit", "github")), + ("generic_conclusion", ("generic_conclusion", "결론", "일반적", "generic", "뻔", "추상")), + ("missing_counterpoint", ("missing_counterpoint", "반론", "counter", "찬반", "균형", "우려")), + ("thin_evidence", ("thin_evidence", "근거", "수치", "사례", "evidence", "출처")), + ("stale_structure", ("stale_structure", "구조", "반복", "나열", "템플릿", "흐름", "고정", "획일", "천편일률")), + ("ai_tone", ("ai_tone", "ai 말투", "챗봇", "기계적", "번역투", "상투적", "클리셰", "수사 의문문", "양비론")), ) MAX_ARTICLES_TO_SCORE = 20 # LLM 혁신도 평가 최대 기사 수 MAX_BODY_CHARS = 10000 # 본문 최대 글자 수 (토큰 절약) @@ -199,11 +200,11 @@ QUALITY_RUBRIC = """ 100점 기준 평가목록: -- 독자 문제와 검색 의도 7 +- 독자 문제와 검색 의도 6 - 핵심 주장/관점 7 -- 제목/메타/슬러그 적합성 5 +- 제목/메타/슬러그 적합성 4 - 사실관계/기술 정확성 10 -- 시점/범위/전제 명시 5 +- 시점/범위/전제 명시 4 - 근거/출처/검증 8 - 구체적 사례와 재현 가능한 맥락 7 - 보안/데이터/운영/플랫폼 리스크 6 @@ -211,9 +212,22 @@ - 대안/트레이드오프/한계 5 - 초반 흡입력/갈등 또는 긴장 제시 5 - 사례에서 원칙으로 끌어올리는 힘 6 -- 구조/문단/전환 흐름 8 +- 주제 맞춤 구조: 소제목 문구·순서·개수가 이 글의 주제에서 나온 설계인가 6 +- 개성 있는 문체: 단정하는 문장, 짧은 문단, 사람이 쓴 리듬 5 - 문장 압축력과 한국어 자연스러움 6 - 결말 회수와 여운/행동 유도 7 + +실패 패턴 점검 목록 (평가 시 각 항목을 하나씩 확인하고, 발견하면 패턴 이름을 blocker 문장에 그대로 인용): +- stale_structure: 다른 글에도 그대로 붙일 수 있는 범용 소제목 세트를 순서까지 그대로 반복한 템플릿형 구조 +- weak_community_angle: 커뮤니티/업계에서 실제로 갈린 반응 없이 "반응이 뜨겁다" 수준으로 뭉갠 서술 +- thin_evidence: 핵심 주장에 수치·사례·출처가 붙지 않고 단정만 남은 서술 +- generic_conclusion: 어느 주제에 붙여도 되는 결론("지켜볼 필요가 있다"류)으로 끝나는 마무리 +- missing_counterpoint: 반대 관점이나 한계를 한 번도 다루지 않은 일방적 전개 +- single_source_summary: 선정 글감 하나를 순서대로 요약한 수준에 그친 글 +- ai_tone: 상투적 전환어 반복, 수사 의문문 남발, 알맹이 없는 양비론, 클리셰 마무리, 모든 문단의 불릿화 같은 기계적 문체 + +구조 다양성 원칙: 소제목 문구·순서·개수가 이전 글들과 다른 것은 결함이 아니라 가점 요인입니다. +고정 템플릿을 따르지 않았다는 이유로 감점하지 말고, 주제에 맞게 새로 설계된 구조인지로 평가하세요. 초안 작성 단계부터 95점 이상을 목표로 삼고, 부족한 항목은 본문 안에서 보강하세요. """ @@ -426,53 +440,36 @@ def build_quality_record( def post_structure() -> str: if CURRENT_TRACK == "issue": return """ -> **한 줄 요약** 글에서 말하고 싶은 핵심을 딱 한두 문장으로 - -## 무슨 일이 있었나 -- 사건, 당사자, 날짜, 정책/제품/플랫폼 범위를 구체적으로 정리 -- 확인된 사실과 아직 추정인 해석을 분리 - -## 왜 사람들이 반응했나 -- 커뮤니티가 불편해한 지점, 기대한 지점, 오해가 생긴 지점을 정리 -- 단순 찬반보다 신뢰, 권한, 비용, 사용성, 규제 리스크로 나눠 설명 - -## 내가 보는 핵심 -- 이번 이슈를 하나의 원칙이나 반복되는 패턴으로 끌어올림 -- 과장하지 말고, 그래도 놓치면 안 되는 지점을 분명히 말함 - -## 앞으로 볼 기준 -- 독자가 다음 뉴스를 볼 때 확인할 체크포인트 -- 결론은 첫 문단의 긴장을 회수하면서 여운 있게 마무리 - -## 참고 자료 -- [선정 글감] 링크를 포함하고, 실제로 본문에 사용한 보조 레퍼런스만 추가 +구조는 고정 템플릿이 아니라 기능적 비트의 집합입니다. 아래 비트는 소제목 라벨이 아니라 이 글이 소화해야 할 역할 목록입니다. + +1. 도입: 이 이슈 고유의 긴장이나 질문 하나를 첫 문단에서 구체적으로 세운다. "이런 이슈가 있었다"식 일반 서두 금지. +2. 사실 정리: 사건, 당사자, 시점, 정책/제품/플랫폼 범위를 구체적으로. 확인된 사실과 아직 추정인 해석을 분리. +3. 갈린 반응: 커뮤니티가 불편해한 지점과 기대한 지점을 신뢰, 권한, 비용, 사용성, 규제 리스크 같은 축으로 정리. +4. 관점: 이번 이슈를 하나의 원칙이나 반복되는 패턴으로 끌어올리고, 놓치면 안 되는 지점을 분명하게 단정. +5. 회수: 도입에서 세운 긴장을 결론에서 직접 회수해 답한다. 불릿 나열형 "정리" 섹션으로 끝내지 말 것. +6. ## 참고 자료: 실제로 본문에 사용한 링크만. + +구조 설계 규칙: +- 소제목(H2)은 위 비트의 라벨을 그대로 옮기지 말고, 이 글의 주제에 대한 구체적 주장이나 질문으로 새로 짓는다. +- 비트 두 개를 한 섹션으로 합치거나 순서를 바꿔도 된다. H2는 참고 자료를 제외하고 3~5개 사이에서 주제가 요구하는 만큼만 쓴다. +- 범용 소제목("무슨 일이 있었나", "왜 사람들이 반응했나", "정리")을 재사용하지 않는다. 매 글의 소제목 구성은 달라야 한다. """ return """ -> **한 줄 요약** 글에서 말하고 싶은 핵심을 딱 한두 문장으로 - -## 왜 지금 이슈인가 -- GitHub, Hacker News, 개발자 커뮤니티에서 말이 붙을 만한 배경 -- 단순 뉴스가 아니라 실무 문제와 어떻게 연결되는지 - -## 커뮤니티에서 갈리는 지점 -- 찬성/반대, 기대/우려, 기존 방식과 새 방식의 차이를 정리 -- 특정 회사 관점으로 몰지 말고 여러 레퍼런스를 주제별로 엮을 것 - -## 아키텍처 관점에서 볼 점 -- 시스템 설계, 데이터 흐름, 장애 격리, 성능, 운영 복잡도 관점으로 설명 -- 코드 스니펫 또는 Mermaid 다이어그램 포함 **[필수]** - -## 실무에서 볼 점 -- 도입 전에 확인할 조건, 트레이드오프, 실패하기 쉬운 지점 -- 비슷한 상황을 겪어본 경험이 있다면 과장 없이 자연스럽게 언급 - -## 정리 -- 핵심 메시지를 간결하게 마무리 -- 독자가 당장 확인해볼 것 한 가지 - -## 참고 자료 -- [선정 글감] 링크를 포함하고, 실제로 본문에 사용한 보조 레퍼런스만 추가 +구조는 고정 템플릿이 아니라 기능적 비트의 집합입니다. 아래 비트는 소제목 라벨이 아니라 이 글이 소화해야 할 역할 목록입니다. + +1. 도입: 이 주제 고유의 긴장이나 질문 하나를 첫 문단에서 구체적으로 세운다. 핵심을 한두 문장으로 요약한 인용구(>)로 열어도 좋다. +2. 맥락: 왜 지금 이 주제에 GitHub/Hacker News/개발자 커뮤니티에서 말이 붙는지, 실무 문제와 어떻게 연결되는지. +3. 갈리는 지점: 찬성/반대, 기대/우려, 기존 방식과 새 방식의 차이. 특정 회사 관점으로 몰지 말고 여러 레퍼런스를 주제별로 엮을 것. +4. 아키텍처: 시스템 설계, 데이터 흐름, 장애 격리, 성능, 운영 복잡도 관점. 코드 스니펫 또는 Mermaid 다이어그램 1개 필수. +5. 실무 조건: 도입 전에 확인할 조건, 트레이드오프, 실패하기 쉬운 지점. +6. 회수: 도입에서 세운 긴장을 결론에서 직접 회수해 답한다. 불릿 나열형 "정리" 섹션으로 끝내지 말 것. +7. ## 참고 자료: 실제로 본문에 사용한 링크만. + +구조 설계 규칙: +- 소제목(H2)은 위 비트의 라벨을 그대로 옮기지 말고, 이 글의 주제에 대한 구체적 주장이나 질문으로 새로 짓는다. 검색될 법한 키워드를 소제목에 자연스럽게 담는다. +- 비트 두 개를 한 섹션으로 합치거나 순서를 바꿔도 된다. H2는 참고 자료를 제외하고 3~6개 사이에서 주제가 요구하는 만큼만 쓴다. +- 범용 소제목("왜 지금 이슈인가", "커뮤니티에서 갈리는 지점", "아키텍처 관점에서 볼 점", "실무에서 볼 점", "정리")을 재사용하지 않는다. 매 글의 소제목 구성은 달라야 한다. """ @@ -1158,8 +1155,8 @@ def generate_post( ━━━ 글쓰기 방식 ━━━ 1. **쟁점 내용 충실히 전달**: 선정된 글감과 보조 레퍼런스의 핵심 개념, 수치, 사례는 정확하게 소화해서 설명 -2. **나만의 시각 추가**: 자료 묶음을 읽고 든 생각, 실무에서 겪을 법한 유사 상황, 동의하거나 의문을 가진 부분을 자연스럽게 녹임 -3. **실무 관점 코멘트**: "실제로 이런 상황에서는", "현업에서 비슷한 고민을 하다 보면" 같은 자연스러운 코멘트 삽입 — 나이/연차/회사 이름은 절대 언급하지 말 것 +2. **관점을 세우고 단정하기**: 자료를 근거로 하나의 판단을 평서문으로 분명히 말하고 그 판단 뒤에 선다. 모든 문장을 "~일 수도 있다", "~로 보인다"로 흐리지 말 것. 단, 단정의 근거는 반드시 선정 글감과 보조 레퍼런스 안에서 가져올 것 — 자료에 없는 경험이나 사례를 지어내지 말 것 +3. **실무 관점 코멘트**: 도입 조건, 운영 리스크, 확인 포인트를 구체적으로 짚되, 직접 겪은 것처럼 쓰지 말 것 — 나이/연차/회사 이름도 절대 언급하지 말 것 4. **코드/다이어그램**: 원문 코드가 있으면 포함. 없으면 개념을 설명하는 Mermaid 다이어그램 1개 직접 작성 ━━━ 절대 금지 ━━━ @@ -1172,6 +1169,12 @@ def generate_post( - 굵은 글씨는 소제목 안에서만, 또는 정말 핵심 명령어/코드 옆에서만 허용 - 일반 문장 안에서 `**X**와 **Y**` 식으로 키워드를 굵게 처리하지 말 것 - 작은따옴표로 단어를 감싸는 패턴 (`'규모'`, `'정확성'`) — 그냥 단어 그대로 쓸 것 +- 상투적 전환어 남발: 또한/게다가/뿐만 아니라/결론적으로/정리하자면/요컨대/다양한 관점에서 같은 표현을 문단마다 반복하는 것 +- 알맹이 없는 양비론: "장점도 있고 단점도 있다"식으로 판단 없이 끝내는 것 — 근거가 기우는 쪽을 분명히 말할 것 +- 수사 의문문 남발: "~일까?", "~이 아닐까?" 반복 사용 — 글 전체에서 최대 1회 +- 클리셰 마무리 문구: "지켜볼 필요가 있다", "귀추가 주목된다", "앞으로가 기대된다" 등 +- 모든 문단을 불릿 목록으로 정리하려는 강박 — 불릿은 진짜 목록에만 쓰고, 논지는 문장으로 전개할 것 +- 문단 시작 패턴의 획일성: 연속된 문단을 같은 접속사/같은 문형으로 시작하는 것 ━━━ SEO 최적화 ━━━ - 핵심 키워드를 본문 앞부분(첫 200자 이내)에 자연스럽게 포함 @@ -1179,39 +1182,18 @@ def generate_post( - 독자가 실제로 검색할 법한 질문("왜 X를 써야 할까?", "X vs Y 차이점") 형식의 소제목 활용 ━━━ 글쓰기 스타일 ━━━ -- 문단 호흡 짧게 (한 문단 3문장 이내) -- 불릿(-), **굵은 글씨**, 인용블록(>), 표 적극 활용 +- 확신 있는 문장: 주장을 평서문으로 단정하고, 근거를 바로 뒤에 붙인다. 조건이 필요하면 조건을 구체적으로 명시하고 얼버무리지 않는다 +- 문단은 대부분 2~4문장. 강조하고 싶은 한 문장은 의도적으로 한 줄 문단으로 세운다 +- 대구와 대조: 두 대상을 같은 문형으로 맞세우는 문장(예: "A는 X를 샀고, B는 X를 잃었다")을 핵심 지점에서 한두 번만 사용 — 매 문장에 쓰면 효과가 죽는다 +- 막연한 추상어 대신 구체적 수치, 이름, 동작을 쓴다 +- 불릿(-), 인용블록(>), 표는 필요한 곳에만 사용하고, 굵은 글씨는 남발하지 않는다 - 기술 용어 첫 등장 시 영문 병기: "서킷 브레이커(Circuit Breaker)" - 분량: **최소 2,500자** (깊이 있게 쓸 것) ━━━ 포스트 구조 ━━━ {post_structure()} -issue 트랙이면 아래 기술 글 기본 구조는 무시하세요. tech 트랙이면 아래 구조를 참고해도 됩니다. - -> **한 줄 요약** — 이 글에서 말하고 싶은 핵심을 딱 한두 문장으로 - -## 왜 지금 이슈인가 -- GitHub, Hacker News, 개발자 커뮤니티에서 말이 붙을 만한 배경 -- 단순 뉴스가 아니라 실무 문제와 어떻게 연결되는지 - -## 커뮤니티에서 갈리는 지점 -- 찬성/반대, 기대/우려, 기존 방식과 새 방식의 차이를 정리 -- 특정 회사 관점으로 몰지 말고 여러 레퍼런스를 주제별로 엮을 것 - -## 아키텍처 관점에서 볼 점 -- 시스템 설계, 데이터 흐름, 장애 격리, 성능, 운영 복잡도 관점으로 설명 -- 코드 스니펫 또는 Mermaid 다이어그램 포함 **[필수]** - -## 실무에서 볼 점 -- 도입 전에 확인할 조건, 트레이드오프, 실패하기 쉬운 지점 -- 비슷한 상황을 겪어본 경험이 있다면 과장 없이 자연스럽게 언급 - -## 정리 -- 핵심 메시지를 간결하게 마무리 -- 독자가 당장 확인해볼 것 한 가지 - -## 참고 자료 +━━━ 참고 자료(마지막 H2) 작성 규칙 ━━━ - [선정 글감] [{article['title']}]({article['link']}) — {article['source']} - 보조 레퍼런스에서 실제로 활용한 링크를 `- [관련] 제목 — 출처` 형식으로 추가 - 보조 레퍼런스가 없으면 관련 링크를 추가하지 말 것 @@ -1240,6 +1222,13 @@ def humanize_post(body: str) -> str: 규칙: - 의미, 코드블록, 링크, 표, 제목 구조는 보존 - 과장된 의미 부여, 광고 문구, 막연한 출처, AI 단어, not only/but 패턴, 3개 나열 강박, em dash 남발, 굵은 글씨 남발, 이모지, 챗봇 말투, 지식 컷오프 문구, 뻔한 긍정 결론 제거 +- 상투적 전환어 남발 제거: 또한/게다가/뿐만 아니라/결론적으로/정리하자면/요컨대/다양한 관점에서 같은 표현이 문단마다 반복되면 삭제하거나 문장 연결을 다시 쓰기 +- 알맹이 없는 양비론 제거: "장점도 있고 단점도 있다"식 결론 없는 균형 잡기는 본문 근거가 기우는 쪽으로 정리 +- 수사 의문문 정리: "~일까?", "~이 아닐까?"가 글 전체에서 2회 이상이면 평서문으로 교체 +- 클리셰 마무리 제거: "지켜볼 필요가 있다", "귀추가 주목된다", "앞으로가 기대된다"류 문장은 도입의 질문에 답하는 구체적 문장으로 교체 +- 불릿 강박 완화: 논지 전개까지 불릿으로 쪼갠 부분은 문장으로 풀기 (진짜 목록은 불릿 유지) +- 문단 시작 획일성 해소: 연속된 문단이 같은 접속사나 같은 문형으로 시작하면 시작을 다르게 바꾸기 +- 새 사실, 새 수치, 새 경험담을 추가하지 말 것 - 문장은 실제 사람이 쓴 한국어 블로그처럼 구체적으로 정리 - 출력은 교정된 Markdown 본문만. frontmatter, 설명, 변경 요약은 쓰지 말 것 @@ -1344,7 +1333,10 @@ def evaluate_post(article: dict, body: str, supporting_context: str, judge_index 선호 차이, 더 좋아질 여지, 사소한 문장 polish는 95점 미만 사유가 아닙니다. 95점 미만을 주려면 게시를 막는 구체적 publish blocker를 blockers에 최소 1개 적어야 합니다. publish blocker가 없다면 score는 반드시 95점 이상이고 publishable은 true입니다. -blockers는 사실 오류, 핵심 주장 검증 불가, 제목/본문 불일치, 구조 붕괴, 게시 품질 미달 문체, 출처가 필요한 핵심 주장 누락, 중복/환각/명백한 생성 흔적에만 씁니다. +blockers는 사실 오류, 핵심 주장 검증 불가, 제목/본문 불일치, 구조 붕괴, 게시 품질 미달 문체, 출처가 필요한 핵심 주장 누락, 중복/환각/명백한 생성 흔적, 그리고 아래 루브릭의 실패 패턴에만 씁니다. +막연한 총평으로 점수를 매기지 말고, 루브릭의 실패 패턴 7가지(stale_structure, weak_community_angle, thin_evidence, generic_conclusion, missing_counterpoint, single_source_summary, ai_tone)를 하나씩 점검하세요. +패턴이 실제로 보이면 blocker 문장 안에 해당 패턴 이름을 그대로 인용하고, 본문의 어느 부분이 그런지 근거를 짧게 붙이세요. +소제목 구성이 다른 글과 다르거나 고정 템플릿을 따르지 않는 것은 감점 사유가 아닙니다. 주제에 맞게 구조를 새로 설계했다면 오히려 가점하세요. 본문에는 frontmatter와 제목이 없으므로 제목/메타/슬러그 항목은 본문이 검색 의도와 제목 생성에 충분한지로 평가하세요. {QUALITY_RUBRIC} From 3a1ce8ec885958227cb8ed5e329672a8c6b17f60 Mon Sep 17 00:00:00 2001 From: n8n Blog Runner Date: Sat, 4 Jul 2026 20:43:07 +0900 Subject: [PATCH 2/5] fix: stabilize n8n blog runner publish path --- scripts/n8n_codex_blog_runner.py | 173 +++++++++++++++++++++++++------ 1 file changed, 139 insertions(+), 34 deletions(-) diff --git a/scripts/n8n_codex_blog_runner.py b/scripts/n8n_codex_blog_runner.py index fcd485c..c67c853 100644 --- a/scripts/n8n_codex_blog_runner.py +++ b/scripts/n8n_codex_blog_runner.py @@ -6,9 +6,11 @@ import json import os import re +import shutil import signal import socketserver import subprocess +import tempfile import threading import urllib.parse import urllib.request @@ -30,6 +32,7 @@ HUGO_BIN = os.environ.get("BLOG_RUNNER_HUGO", "/opt/homebrew/bin/hugo") PUBLISH_BRANCH = os.environ.get("BLOG_RUNNER_PUBLISH_BRANCH", "main") VALID_TRACKS = {"issue", "tech"} +ENV_FILE = Path(os.environ.get("BLOG_RUNNER_ENV_FILE", str(Path.home() / ".config" / "blog-runner" / "env"))) _lock = threading.Lock() @@ -68,11 +71,24 @@ def _json_response(handler: BaseHTTPRequestHandler, status: int, payload: dict[s handler.wfile.write(data) -def _run_step(name: str, command: list[str], env: dict[str, str]) -> dict[str, Any]: +def _load_env_file(env: dict[str, str]) -> None: + if not ENV_FILE.exists(): + return + for raw_line in ENV_FILE.read_text(errors="replace").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + key = key.strip() + if key and not env.get(key): + env[key] = value.strip().strip("'\"") + + +def _run_step(name: str, command: list[str], env: dict[str, str], cwd: Path = REPO_ROOT) -> dict[str, Any]: started_at = datetime.now(timezone.utc).isoformat() process = subprocess.Popen( command, - cwd=REPO_ROOT, + cwd=cwd, env=env, text=True, stdout=subprocess.PIPE, @@ -217,55 +233,113 @@ def _telegram_chat_id(token: str, env: dict[str, str]) -> str: raise RuntimeError("TELEGRAM_CHAT_ID missing and no Telegram updates found; send /start to the bot first") -def _send_telegram_notification(track: str, paths: list[str], env: dict[str, str]) -> dict[str, Any]: - started_at = datetime.now(timezone.utc).isoformat() +def _send_telegram_text(text: str, env: dict[str, str]) -> None: token = env.get("TELEGRAM_BOT_TOKEN", "").strip() if not token: - return {"name": "send telegram notification", "command": ["telegram", "sendMessage"], "exitCode": 1, "startedAt": started_at, "stdoutTail": "", "stderrTail": "TELEGRAM_BOT_TOKEN missing"} - - text = _blog_alert(track, True, paths) + raise RuntimeError("TELEGRAM_BOT_TOKEN missing") data = urllib.parse.urlencode({"chat_id": _telegram_chat_id(token, env), "text": text}) request = urllib.request.Request(f"https://api.telegram.org/bot{token}/sendMessage", data=data.encode("utf-8")) with urllib.request.urlopen(request, timeout=15) as response: response.read() + + +def _send_telegram_notification(track: str, paths: list[str], env: dict[str, str]) -> dict[str, Any]: + started_at = datetime.now(timezone.utc).isoformat() + text = _blog_alert(track, True, paths) + try: + _send_telegram_text(text, env) + except Exception as exc: + return {"name": "send telegram notification", "command": ["telegram", "sendMessage"], "exitCode": 1, "startedAt": started_at, "stdoutTail": "", "stderrTail": exc.__class__.__name__} return {"name": "send telegram notification", "command": ["telegram", "sendMessage"], "exitCode": 0, "startedAt": started_at, "stdoutTail": text, "stderrTail": ""} +def _copy_publish_files(publish_paths: list[str], publish_worktree: Path) -> None: + for rel in publish_paths: + source = REPO_ROOT / rel + target = publish_worktree / rel + target.parent.mkdir(parents=True, exist_ok=True) + if source.is_dir(): + shutil.copytree(source, target, dirs_exist_ok=True) + else: + shutil.copy2(source, target) + + def _publish_blog_changes(track: str, paths: list[str], env: dict[str, str]) -> list[dict[str, Any]]: publish_paths = _publish_paths(paths + ["scripts/.seen_articles.json", "scripts/.seen_articles_issue.json"]) timestamp = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M %Z") message = f"feat: publish n8n blog posts ({timestamp})" body = f"- track: {track}\n- files: {', '.join(publish_paths)}" steps = [] - commands = [ - ("git configure bot", ["git", "config", "user.name", "n8n Blog Runner"]), - ("git configure email", ["git", "config", "user.email", "bot@gnosyslambda.github.io"]), - ("git fetch main", ["git", "fetch", "origin", f"{PUBLISH_BRANCH}:refs/remotes/origin/{PUBLISH_BRANCH}"]), - ("verify fast-forward publish", ["git", "merge-base", "--is-ancestor", f"origin/{PUBLISH_BRANCH}", "HEAD"]), - ("stage published files", ["git", "add", "--", *publish_paths]), - ] - for name, command in commands: - step = _run_step(name, command, env) - steps.append(step) - if step["exitCode"] != 0: + temp_dir = Path(tempfile.mkdtemp(prefix="blog-runner-publish-")) + worktree_created = False + try: + for name, command in [ + ("git fetch publish branch", ["git", "fetch", "origin", f"refs/heads/{PUBLISH_BRANCH}:refs/remotes/origin/{PUBLISH_BRANCH}"]), + ("create publish worktree", ["git", "worktree", "add", "--detach", str(temp_dir), f"origin/{PUBLISH_BRANCH}"]), + ]: + step = _run_step(name, command, env) + steps.append(step) + if step["exitCode"] != 0: + return steps + if name == "create publish worktree": + worktree_created = True + + started_at = datetime.now(timezone.utc).isoformat() + try: + _copy_publish_files(publish_paths, temp_dir) + except Exception as exc: + steps.append({ + "name": "copy publish files", + "command": ["copy", *publish_paths], + "exitCode": 1, + "startedAt": started_at, + "stdoutTail": "", + "stderrTail": exc.__class__.__name__, + }) return steps - check = _run_step("check staged changes", ["git", "diff", "--cached", "--quiet"], env) - steps.append(check) - if check["exitCode"] == 0: - return steps - if check["exitCode"] == 1: - check["exitCode"] = 0 - check["stdoutTail"] = "staged changes present" - else: - return steps - for name, command in [ - ("commit published files", ["git", "commit", "-m", message, "-m", body]), - ("push published files", ["git", "push", "origin", f"HEAD:{PUBLISH_BRANCH}"]), - ]: - step = _run_step(name, command, env) - steps.append(step) - if step["exitCode"] != 0: + steps.append({ + "name": "copy publish files", + "command": ["copy", *publish_paths], + "exitCode": 0, + "startedAt": started_at, + "stdoutTail": f"copied {len(publish_paths)} files", + "stderrTail": "", + }) + + for name, command in [ + ("git configure publish bot", ["git", "config", "user.name", "n8n Blog Runner"]), + ("git configure publish email", ["git", "config", "user.email", "bot@gnosyslambda.github.io"]), + ("build publish worktree", [HUGO_BIN, "--minify"]), + ("stage published files", ["git", "add", "--", *publish_paths]), + ]: + step = _run_step(name, command, env, cwd=temp_dir) + steps.append(step) + if step["exitCode"] != 0: + return steps + + check = _run_step("check staged changes", ["git", "diff", "--cached", "--quiet"], env, cwd=temp_dir) + steps.append(check) + if check["exitCode"] == 0: return steps + if check["exitCode"] == 1: + check["exitCode"] = 0 + check["stdoutTail"] = "staged changes present" + else: + return steps + + for name, command in [ + ("commit published files", ["git", "commit", "-m", message, "-m", body]), + ("push published files", ["git", "push", "origin", f"HEAD:{PUBLISH_BRANCH}"]), + ]: + step = _run_step(name, command, env, cwd=temp_dir) + steps.append(step) + if step["exitCode"] != 0: + return steps + finally: + if worktree_created: + subprocess.run(["git", "worktree", "remove", "--force", str(temp_dir)], cwd=REPO_ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + else: + shutil.rmtree(temp_dir, ignore_errors=True) return steps @@ -282,6 +356,7 @@ def run_blog_job( return 409, {"ok": False, "error": "blog runner is already running"} env = os.environ.copy() + _load_env_file(env) env["PATH"] = "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:" + env.get("PATH", "") env["BLOG_TRACK"] = track if force: @@ -362,6 +437,36 @@ def do_GET(self) -> None: _json_response(self, 200, {"ok": True, "service": "n8n-codex-blog-runner"}) def do_POST(self) -> None: + if self.path == "/notify": + if TOKEN and self.headers.get("X-Blog-Runner-Token") != TOKEN: + _json_response(self, 401, {"ok": False, "error": "unauthorized"}) + return + + length = int(self.headers.get("Content-Length", "0") or "0") + raw_body = self.rfile.read(length).decode("utf-8") if length else "" + try: + body = json.loads(raw_body) if raw_body else {} + except json.JSONDecodeError: + body = {} + if not isinstance(body, dict): + body = {} + + text = str(body.get("text") or "").strip() + if not text: + _json_response(self, 400, {"ok": False, "error": "text is required"}) + return + + env = os.environ.copy() + _load_env_file(env) + try: + _send_telegram_text(text, env) + except Exception as exc: + _json_response(self, 502, {"ok": False, "error": "telegram send failed", "errorType": exc.__class__.__name__}) + return + + _json_response(self, 200, {"ok": True}) + return + if self.path != "/run": _json_response(self, 404, {"ok": False, "error": "not found"}) return From 8fd9bdef072e06fa98d297e601a86a31df6bff4a Mon Sep 17 00:00:00 2001 From: n8n Blog Runner Date: Sun, 5 Jul 2026 21:05:08 +0900 Subject: [PATCH 3/5] feat: cap tech blog to 2 posts/day (TREND_DAILY_POST_LIMIT); 1 topic per run via env --- scripts/trend_writer.py | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/scripts/trend_writer.py b/scripts/trend_writer.py index 6d8bc35..533207a 100644 --- a/scripts/trend_writer.py +++ b/scripts/trend_writer.py @@ -1792,6 +1792,24 @@ def _check_cooldown() -> None: log.warning(f"쿨다운 체크 실패 (무시하고 계속): {e}") +DAILY_POST_LIMIT = int(os.environ.get("TREND_DAILY_POST_LIMIT", "2")) + + +def _check_daily_limit() -> None: + """오늘(KST) 발행 수가 일일 상한에 도달했으면 이번 실행을 건너뜁니다.""" + kst = timezone(timedelta(hours=9)) + today = datetime.now(tz=kst).strftime("%Y-%m-%d") + try: + count = len(list(POSTS_DIR.glob(f"{today}-*.md"))) + except Exception as exc: + log.warning(f"일일 상한 체크 실패 (무시하고 계속): {exc}") + return + if count >= DAILY_POST_LIMIT: + log.info(f"⏭️ 오늘 이미 {count}개 발행 (일일 상한 {DAILY_POST_LIMIT}) — 이번 실행은 건너뜁니다.") + sys.exit(0) + log.info(f"📅 오늘 발행 {count}/{DAILY_POST_LIMIT}개 — 계속") + + def create_post_for_article(article: dict, articles: list[dict]) -> Path: log.info(f"🧵 주제 작성 시작: [{article['source']}] {article['title']}") @@ -1841,9 +1859,10 @@ def main(): f"judge_timeout={JUDGE_TIMEOUT_SECONDS}s, max_codex_calls≈{max_quality_calls + (TOPICS_PER_RUN * 3)}" ) - # 0. 쿨다운 체크: FORCE=true 환경변수가 있으면 스킵 + # 0. 쿨다운 + 일일 상한 체크: FORCE_RUN 환경변수가 있으면 스킵 if not os.environ.get("FORCE_RUN"): _check_cooldown() + _check_daily_limit() # 1. seen 로드 (타임스탬프 기반, 만료된 항목 자동 제외) seen = load_seen() From b198f6eff74a2e87fdf5fe3f91a8206b299b772a Mon Sep 17 00:00:00 2001 From: n8n Blog Runner Date: Sun, 5 Jul 2026 23:41:03 +0900 Subject: [PATCH 4/5] feat: rich HTML Telegram alerts with per-post score, issues, and live link --- scripts/n8n_codex_blog_runner.py | 81 +++++++++++++++++++++++++------- 1 file changed, 64 insertions(+), 17 deletions(-) diff --git a/scripts/n8n_codex_blog_runner.py b/scripts/n8n_codex_blog_runner.py index c67c853..e25093d 100644 --- a/scripts/n8n_codex_blog_runner.py +++ b/scripts/n8n_codex_blog_runner.py @@ -170,29 +170,71 @@ def _latest_scores(count: int) -> list[str]: return [f"{score:.1f}" for _, score in sorted(scores)[-count:]] +_ISSUE_LABELS = { + "single_source_summary": "단일 출처 요약", + "weak_community_angle": "커뮤니티 반응 약함", + "generic_conclusion": "일반적 결론", + "missing_counterpoint": "반론 부재", + "thin_evidence": "근거 부족", + "stale_structure": "구조 진부", + "ai_tone": "AI 말투", +} + + +def _html_escape(value: Any) -> str: + return str("" if value is None else value).replace("&", "&").replace("<", "<").replace(">", ">") + + +def _post_history() -> dict[str, dict[str, Any]]: + """saved_post_relpath -> {avg, issue_codes} from the latest 'selected' quality entry.""" + path = SCRIPT_DIR / ".quality_history.jsonl" + out: dict[str, dict[str, Any]] = {} + if path.exists(): + for line in path.read_text(errors="replace").splitlines(): + try: + item = json.loads(line) + except json.JSONDecodeError: + continue + rel = item.get("saved_post_relpath") + if item.get("status") == "selected" and rel: + out[rel] = {"avg": (item.get("scores") or {}).get("avg"), "issue_codes": item.get("issue_codes") or []} + return out + + def _blog_alert(track: str, ok: bool, paths: list[str], *, failed_step: str = "", error: str = "") -> str: posts = [Path(path) for path in _publish_paths(paths) if path.startswith("content/posts/")] - scores = _latest_scores(len(posts)) label = "Gnosys 이슈" if track == "issue" else "Gnosys 기술" - lines = [ - f"{'✅' if ok else '❌'} {label} 자동 발행", - f"상태: {'성공' if ok else '실패'}", - f"점수: {', '.join(score + '점' for score in scores) if scores else 'runner 응답 점수 없음'}", - f"발행글: {len(posts)}개", - ] - if failed_step: - lines.append(f"실패 단계: {failed_step}") - if error: - lines.append(f"오류: {error}") - for index, post in enumerate(posts, start=1): + if not ok: + lines = [f"🔴 {_html_escape(label)} 발행 실패"] + if failed_step: + lines.append(f"⛔ 단계: {_html_escape(failed_step)}") + if error: + lines.append(f"⚠️ 오류: {_html_escape(error[:400])}") + return "\n".join(lines) + published = bool(posts) + header = "✅ 발행 완료" if published else "📝 검토 대기 · 미발행" + hist = _post_history() + out = [f"🔔 {header}", f"🗂 블로그: {_html_escape(label)}", ""] + for post in posts: content = (REPO_ROOT / post).read_text(errors="replace").splitlines() title = _frontmatter_value(content, "title") or post.stem description = _frontmatter_value(content, "description") - lines.append(f"\n{index}. {title}") + info = hist.get(post.as_posix(), {}) + avg = info.get("avg") + emoji = "🟢" if isinstance(avg, (int, float)) and avg >= 90 else ("🟡" if isinstance(avg, (int, float)) and avg >= 85 else "🔴") + out.append(f"{emoji} {_html_escape(title)}") + if isinstance(avg, (int, float)): + out.append(f" ├ 점수: {avg:.1f}점") if description: - lines.append(f"요약: {description}") - lines.append(f"파일: {post.as_posix()}") - return "\n".join(lines) + out.append(f" ├ 📝 {_html_escape(description[:200])}") + labels = [_ISSUE_LABELS.get(code, code) for code in (info.get("issue_codes") or [])][:4] + if labels: + out.append(f" ├ 📊 지적: {_html_escape(', '.join(labels))}") + out.append(f' └ 🔗 글 보기') + out.append("") + if not posts: + out.append("발행된 글 없음 (품질 기준 미달)") + return "\n".join(out).strip() def _collect_alert( @@ -237,7 +279,12 @@ def _send_telegram_text(text: str, env: dict[str, str]) -> None: token = env.get("TELEGRAM_BOT_TOKEN", "").strip() if not token: raise RuntimeError("TELEGRAM_BOT_TOKEN missing") - data = urllib.parse.urlencode({"chat_id": _telegram_chat_id(token, env), "text": text}) + data = urllib.parse.urlencode({ + "chat_id": _telegram_chat_id(token, env), + "text": text, + "parse_mode": "HTML", + "disable_web_page_preview": "true", + }) request = urllib.request.Request(f"https://api.telegram.org/bot{token}/sendMessage", data=data.encode("utf-8")) with urllib.request.urlopen(request, timeout=15) as response: response.read() From f0013d7d84015d4282a72b6672c82525171cf968 Mon Sep 17 00:00:00 2001 From: n8n Blog Runner Date: Mon, 6 Jul 2026 14:32:26 +0900 Subject: [PATCH 5/5] fix: sync blog quality gate criteria --- scripts/n8n_codex_blog_runner.py | 2 +- scripts/trend_writer.py | 22 +++++++++++----------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/scripts/n8n_codex_blog_runner.py b/scripts/n8n_codex_blog_runner.py index e25093d..ec827d2 100644 --- a/scripts/n8n_codex_blog_runner.py +++ b/scripts/n8n_codex_blog_runner.py @@ -233,7 +233,7 @@ def _blog_alert(track: str, ok: bool, paths: list[str], *, failed_step: str = "" out.append(f' └ 🔗 글 보기') out.append("") if not posts: - out.append("발행된 글 없음 (품질 기준 미달)") + out.append("발행된 글 없음 (검토 대기/자동 발행 스킵)") return "\n".join(out).strip() diff --git a/scripts/trend_writer.py b/scripts/trend_writer.py index 533207a..85cef08 100644 --- a/scripts/trend_writer.py +++ b/scripts/trend_writer.py @@ -56,9 +56,9 @@ POST_VARIANTS = min(5, max(1, int(os.environ.get("POST_VARIANTS", "5")))) POST_REPAIR_VARIANTS = min(POST_VARIANTS, max(1, int(os.environ.get("POST_REPAIR_VARIANTS", "2")))) POST_JUDGES = min(5, max(1, int(os.environ.get("POST_JUDGES", "2")))) -POST_MIN_SCORE = float(os.environ.get("POST_MIN_SCORE", "95")) -POST_REVIEW_MIN_SCORE = float(os.environ.get("POST_REVIEW_MIN_SCORE", "90")) -POST_POLISH_MIN_SCORE = float(os.environ.get("POST_POLISH_MIN_SCORE", "93")) +POST_MIN_SCORE = float(os.environ.get("POST_MIN_SCORE", "85")) +POST_REVIEW_MIN_SCORE = float(os.environ.get("POST_REVIEW_MIN_SCORE", "80")) +POST_POLISH_MIN_SCORE = float(os.environ.get("POST_POLISH_MIN_SCORE", "85")) POST_MAX_ROUNDS = max(1, int(os.environ.get("POST_MAX_ROUNDS", "2"))) CANDIDATE_TIMEOUT_SECONDS = int(os.environ.get("CANDIDATE_TIMEOUT_SECONDS", str(CODEX_TIMEOUT_SECONDS))) JUDGE_TIMEOUT_SECONDS = int(os.environ.get("JUDGE_TIMEOUT_SECONDS", str(CODEX_TIMEOUT_SECONDS))) @@ -228,7 +228,7 @@ 구조 다양성 원칙: 소제목 문구·순서·개수가 이전 글들과 다른 것은 결함이 아니라 가점 요인입니다. 고정 템플릿을 따르지 않았다는 이유로 감점하지 말고, 주제에 맞게 새로 설계된 구조인지로 평가하세요. -초안 작성 단계부터 95점 이상을 목표로 삼고, 부족한 항목은 본문 안에서 보강하세요. +초안 작성 단계부터 85점 이상을 목표로 삼고, 부족한 항목은 본문 안에서 보강하세요. """ @@ -1110,7 +1110,7 @@ def generate_post( variant_note = f""" 이번 출력은 후보 {variant_index}/{total_variants}입니다. - 같은 자료에서 출발하되, 다른 후보와 제목감, 초반 긴장, 결론 관점을 다르게 잡으세요. -- 억지로 튀지 말고, 95점 이상 평가를 받을 수 있는 가장 설득력 있는 한 가지 관점에 집중하세요. +- 억지로 튀지 말고, 85점 이상 평가를 받을 수 있는 가장 설득력 있는 한 가지 관점에 집중하세요. """ feedback_note = "" if quality_feedback: @@ -1329,10 +1329,10 @@ def evaluate_post(article: dict, body: str, supporting_context: str, judge_index prompt = f"""{track_context()} 당신은 블로그 글 품질을 평가하는 독립 judge 에이전트 {judge_index}입니다. -95점 이상은 완벽한 글이 아니라, 사람이 바로 게시해도 되는 수준입니다. -선호 차이, 더 좋아질 여지, 사소한 문장 polish는 95점 미만 사유가 아닙니다. -95점 미만을 주려면 게시를 막는 구체적 publish blocker를 blockers에 최소 1개 적어야 합니다. -publish blocker가 없다면 score는 반드시 95점 이상이고 publishable은 true입니다. +85점 이상은 완벽한 글이 아니라, 사람이 바로 게시해도 되는 수준입니다. +선호 차이, 더 좋아질 여지, 사소한 문장 polish는 85점 미만 사유가 아닙니다. +85점 미만을 주려면 게시를 막는 구체적 publish blocker를 blockers에 최소 1개 적어야 합니다. +publish blocker가 없다면 score는 반드시 85점 이상이고 publishable은 true입니다. blockers는 사실 오류, 핵심 주장 검증 불가, 제목/본문 불일치, 구조 붕괴, 게시 품질 미달 문체, 출처가 필요한 핵심 주장 누락, 중복/환각/명백한 생성 흔적, 그리고 아래 루브릭의 실패 패턴에만 씁니다. 막연한 총평으로 점수를 매기지 말고, 루브릭의 실패 패턴 7가지(stale_structure, weak_community_angle, thin_evidence, generic_conclusion, missing_counterpoint, single_source_summary, ai_tone)를 하나씩 점검하세요. 패턴이 실제로 보이면 blocker 문장 안에 해당 패턴 이름을 그대로 인용하고, 본문의 어느 부분이 그런지 근거를 짧게 붙이세요. @@ -1357,7 +1357,7 @@ def evaluate_post(article: dict, body: str, supporting_context: str, judge_index {{ "score": 0, "publishable": false, -"blockers": ["95점 미만이면 게시를 막는 구체적 결함. 없으면 빈 배열"], +"blockers": ["85점 미만이면 게시를 막는 구체적 결함. 없으면 빈 배열"], "top_3_fixes": ["blocker를 제거하고 점수를 올릴 가장 큰 수정 3개 이하"], "do_not_change": ["이미 좋은 부분 3개 이하"], "minor_suggestions": ["게시를 막지 않는 선택적 개선점"], @@ -1623,7 +1623,7 @@ def save_review_candidate(article: dict, candidate: dict) -> Path: {candidate["body"]} """ filepath.write_text(content, encoding="utf-8") - log.info(f"🗂️ 90점 이상 수정 후보 저장: {filepath}") + log.info(f"🗂️ 80점 이상 수정 후보 저장: {filepath}") return filepath