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..ec827d2 100644
--- a/scripts/n8n_codex_blog_runner.py
+++ b/scripts/n8n_codex_blog_runner.py
@@ -6,8 +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
@@ -29,10 +32,19 @@
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()
+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
@@ -59,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,
@@ -106,6 +131,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 +170,97 @@ 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/")]
+ label = "Gnosys 이슈" if track == "issue" else "Gnosys 기술"
+ 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")
+ 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:
+ 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(
+ 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:
@@ -144,27 +275,40 @@ 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"}
-
- 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)})
+ raise RuntimeError("TELEGRAM_BOT_TOKEN missing")
+ 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()
- return {"name": "send telegram notification", "command": ["telegram", "sendMessage"], "exitCode": 0, "startedAt": started_at, "stdoutTail": "\n".join(lines), "stderrTail": ""}
+
+
+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]]:
@@ -173,35 +317,76 @@ def _publish_blog_changes(track: str, paths: list[str], env: dict[str, str]) ->
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
@@ -212,11 +397,13 @@ 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"}
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:
@@ -227,20 +414,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 +454,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()
@@ -275,6 +484,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
@@ -307,6 +546,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 +556,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..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)))
@@ -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,10 +212,23 @@
- 대안/트레이드오프/한계 5
- 초반 흡입력/갈등 또는 긴장 제시 5
- 사례에서 원칙으로 끌어올리는 힘 6
-- 구조/문단/전환 흐름 8
+- 주제 맞춤 구조: 소제목 문구·순서·개수가 이 글의 주제에서 나온 설계인가 6
+- 개성 있는 문체: 단정하는 문장, 짧은 문단, 사람이 쓴 리듬 5
- 문장 압축력과 한국어 자연스러움 6
- 결말 회수와 여운/행동 유도 7
-초안 작성 단계부터 95점 이상을 목표로 삼고, 부족한 항목은 본문 안에서 보강하세요.
+
+실패 패턴 점검 목록 (평가 시 각 항목을 하나씩 확인하고, 발견하면 패턴 이름을 blocker 문장에 그대로 인용):
+- stale_structure: 다른 글에도 그대로 붙일 수 있는 범용 소제목 세트를 순서까지 그대로 반복한 템플릿형 구조
+- weak_community_angle: 커뮤니티/업계에서 실제로 갈린 반응 없이 "반응이 뜨겁다" 수준으로 뭉갠 서술
+- thin_evidence: 핵심 주장에 수치·사례·출처가 붙지 않고 단정만 남은 서술
+- generic_conclusion: 어느 주제에 붙여도 되는 결론("지켜볼 필요가 있다"류)으로 끝나는 마무리
+- missing_counterpoint: 반대 관점이나 한계를 한 번도 다루지 않은 일방적 전개
+- single_source_summary: 선정 글감 하나를 순서대로 요약한 수준에 그친 글
+- ai_tone: 상투적 전환어 반복, 수사 의문문 남발, 알맹이 없는 양비론, 클리셰 마무리, 모든 문단의 불릿화 같은 기계적 문체
+
+구조 다양성 원칙: 소제목 문구·순서·개수가 이전 글들과 다른 것은 결함이 아니라 가점 요인입니다.
+고정 템플릿을 따르지 않았다는 이유로 감점하지 말고, 주제에 맞게 새로 설계된 구조인지로 평가하세요.
+초안 작성 단계부터 85점 이상을 목표로 삼고, 부족한 항목은 본문 안에서 보강하세요.
"""
@@ -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개 사이에서 주제가 요구하는 만큼만 쓴다.
+- 범용 소제목("왜 지금 이슈인가", "커뮤니티에서 갈리는 지점", "아키텍처 관점에서 볼 점", "실무에서 볼 점", "정리")을 재사용하지 않는다. 매 글의 소제목 구성은 달라야 한다.
"""
@@ -1113,7 +1110,7 @@ def generate_post(
variant_note = f"""
이번 출력은 후보 {variant_index}/{total_variants}입니다.
- 같은 자료에서 출발하되, 다른 후보와 제목감, 초반 긴장, 결론 관점을 다르게 잡으세요.
-- 억지로 튀지 말고, 95점 이상 평가를 받을 수 있는 가장 설득력 있는 한 가지 관점에 집중하세요.
+- 억지로 튀지 말고, 85점 이상 평가를 받을 수 있는 가장 설득력 있는 한 가지 관점에 집중하세요.
"""
feedback_note = ""
if quality_feedback:
@@ -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, 설명, 변경 요약은 쓰지 말 것
@@ -1340,11 +1329,14 @@ 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입니다.
-blockers는 사실 오류, 핵심 주장 검증 불가, 제목/본문 불일치, 구조 붕괴, 게시 품질 미달 문체, 출처가 필요한 핵심 주장 누락, 중복/환각/명백한 생성 흔적에만 씁니다.
+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 문장 안에 해당 패턴 이름을 그대로 인용하고, 본문의 어느 부분이 그런지 근거를 짧게 붙이세요.
+소제목 구성이 다른 글과 다르거나 고정 템플릿을 따르지 않는 것은 감점 사유가 아닙니다. 주제에 맞게 구조를 새로 설계했다면 오히려 가점하세요.
본문에는 frontmatter와 제목이 없으므로 제목/메타/슬러그 항목은 본문이 검색 의도와 제목 생성에 충분한지로 평가하세요.
{QUALITY_RUBRIC}
@@ -1365,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": ["게시를 막지 않는 선택적 개선점"],
@@ -1631,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
@@ -1800,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']}")
@@ -1849,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()