[P-Reinforce] Global knowledge consolidation, massive deduplication (5,249 files), and high-density wikification (45 nodes)

This commit is contained in:
Antigravity Agent
2026-05-05 15:28:22 +09:00
parent a7d1e60ccf
commit dd01e01bea
3430 changed files with 42739 additions and 52263 deletions
+12
View File
@@ -0,0 +1,12 @@
# 🧭 CEO (Chief Executive Agent) 개인 메모리
_CEO 에이전트만 읽고 쓰는 개인 노트. 학습·교훈·자주 쓰는 패턴이 누적됩니다._
## 학습 기록
- [2026-05-01] [모닝 브리핑] 오늘 날짜는 2026-05-01입니다. 회사 목표(goals.md)와 지금까지의 의사결정 로그를 바탕으로 오늘 우리 회사가 우선순위로 처리해야 할 작업 3가지를 결정하고, 각 작업을 적절한 에이전트에게 분배하세요. → 보고서 sessions/2026-05-01T08-07/_report.md
- [2026-05-01] [자율 사이클 — 2026-05-01] 1인 기업 24시간 운영 중. 회사 목표·각 에이전트의 개인 목표(_agents/{id}/goal.md)·최근 의사결정·메모리를 검토해서 지금 가장 가치 있는 단일 작업 1개를 결정하고, 적절한 1~2명 에이전트에게 분배해서 실행하세요. 같은 산출물을 반복하지 마세요 — 메모리에 비슷한 항목이 24시간 내에 있으면 다른 각도로 진전시키세요. → 보고서 sessions/2026-05-01T08-57/_report.md
- [2026-05-01] [자율 사이클 — 2026-05-01] 1인 기업 24시간 운영 중. 회사 목표·각 에이전트의 개인 목표(_agents/{id}/goal.md)·최근 의사결정·메모리를 검토해서 지금 가장 가치 있는 단일 작업 1개를 결정하고, 적절한 1~2명 에이전트에게 분배해서 실행하세요. 같은 산출물을 반복하지 마세요 — 메모리에 비슷한 항목이 24시간 내에 있으면 다른 각도로 진전시키세요. → 보고서 sessions/2026-05-01T09-44/_report.md
- [2026-05-01] [자율 사이클 — 2026-05-01] 1인 기업 24시간 운영 중. 회사 목표·각 에이전트의 개인 목표(_agents/{id}/goal.md)·최근 의사결정·메모리를 검토해서 지금 가장 가치 있는 단일 작업 1개를 결정하고, 적절한 1~2명 에이전트에게 분배해서 실행하세요. 같은 산출물을 반복하지 마세요 — 메모리에 비슷한 항목이 24시간 내에 있으면 다른 각도로 진전시키세요. → 보고서 sessions/2026-05-01T10-02/_report.md
- [2026-05-01] [자율 사이클 — 2026-05-01] 1인 기업 24시간 운영 중. 회사 목표·각 에이전트의 개인 목표(_agents/{id}/goal.md)·최근 의사결정·메모리를 검토해서 지금 가장 가치 있는 단일 작업 1개를 결정하고, 적절한 1~2명 에이전트에게 분배해서 실행하세요. 같은 산출물을 반복하지 마세요 — 메모리에 비슷한 항목이 24시간 내에 있으면 다른 각도로 진전시키세요. → 보고서 sessions/2026-05-01T10-17/_report.md
- [2026-05-01] [자율 사이클 — 2026-05-01] 1인 기업 24시간 운영 중. 회사 목표·각 에이전트의 개인 목표(_agents/{id}/goal.md)·최근 의사결정·메모리를 검토해서 지금 가장 가치 있는 단일 작업 1개를 결정하고, 적절한 1~2명 에이전트에게 분배해서 실행하세요. 같은 산출물을 반복하지 마세요 — 메모리에 비슷한 항목이 24시간 내에 있으면 다른 각도로 진전시키세요. → 보고서 sessions/2026-05-01T11-39/_report.md
@@ -0,0 +1,5 @@
# 🔍 Researcher 페르소나 디테일
_여기에 Researcher 에이전트에게 주고 싶은 추가 지시·말투·취향·예시 등을 자유롭게 적으세요._
_매 호출 시 시스템 프롬프트에 자동 주입됩니다. (git에 동기화됨)_
@@ -0,0 +1,4 @@
{
"INTERVAL_HOURS": 2,
"TOTAL_RUN_HOURS": 8
}
@@ -0,0 +1,29 @@
# 🌙 오토 플래너
트렌드 스나이퍼를 정해진 간격으로 반복 실행해서 패턴 데이터를 쌓아주는 무인 작업자예요. 한 번 트렌드를 보면 지금 잘 되는 영상 한 장만 보이지만, 8시간 동안 2시간마다 4번 보면 "어떤 키워드의 후크가 시간이 지나도 계속 살아남는지"가 보이기 시작합니다 — 자는 동안에 그 작업을 대신해줍니다.
## 어떻게 도와주나요?
- ⏰ N시간마다 `[[trend_sniper]].py`를 자동 실행 (스나이퍼 결과는 매번 sessions/에 누적)
- 🛌 잘 때 켜두면 아침에 4~5번분의 트렌드 스냅샷이 쌓여 있어요
- 📊 같은 키워드라도 시간대별로 어떤 영상이 새로 떠오르는지 비교 가능
## 어떤 상황에 켜면 좋나요?
- 새 채널 컨셉을 결정하기 전, 며칠치 트렌드를 누적해서 보고 싶을 때
- 회사 일/외출 중 백그라운드에서 데이터만 모아두고 싶을 때
- 특정 키워드의 알고리즘 반응이 시간대마다 다른지 확인하고 싶을 때
## 시작하기 전 체크
- 트렌드 스나이퍼 도구가 먼저 설정돼 있어야 해요 (YouTube API 키, 키워드 목록 등)
- 첫 실행 전에 트렌드 스나이퍼를 한 번 수동으로 돌려서 정상 작동 확인을 권장합니다
## 설정값 (auto_planner.json)
- `INTERVAL_HOURS` — 몇 시간마다 실행할지 (기본 2)
- `TOTAL_RUN_HOURS` — 총 가동 시간 (기본 8 → 8시간 동안 4회 실행)
## 실행 방법
패널의 [▶ 실행]을 누르면 시작됩니다. 또는 터미널에서:
```bash
python auto_planner.py
```
⚠️ 이 스크립트는 끝날 때까지 터미널을 점유해요. 백그라운드로 돌리려면 별도 창에서 실행하세요. 중단하려면 Ctrl+C.
@@ -0,0 +1,43 @@
#!/usr/bin/env python3
"""Auto Planner — runs trend_sniper.py on a fixed interval for a chosen
duration (e.g. overnight). Reads its config from auto_planner.json."""
import os, json, time, datetime, subprocess, sys
HERE = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(HERE, "auto_planner.json")
SNIPER_PATH = os.path.join(HERE, "trend_sniper.py")
def load_config():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
print(f"❌ 설정 파일을 읽을 수 없어요: {CONFIG_PATH}\n{e}")
sys.exit(1)
def main():
cfg = load_config()
interval_h = float(cfg.get("INTERVAL_HOURS", 2))
total_h = float(cfg.get("TOTAL_RUN_HOURS", 8))
print(f"\n🚀 [오토 플래너] {total_h}시간 동안 {interval_h}시간마다 트렌드 분석 실행")
if not os.path.exists(SNIPER_PATH):
print(f"❌ trend_sniper.py를 찾을 수 없어요: {SNIPER_PATH}")
sys.exit(1)
start = time.time()
loop = 0
while True:
if time.time() - start > total_h * 3600:
print("\n☀️ 목표 가동 시간을 채웠어요. 종료합니다.")
break
loop += 1
ts = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"\n[{ts}] 🤖 {loop}회차 트렌드 스나이핑")
try:
subprocess.run([sys.executable, SNIPER_PATH], check=False)
except Exception as e:
print(f"❌ 실행 실패: {e}")
print(f"⏳ 다음 실행: {interval_h}시간 후")
time.sleep(interval_h * 3600)
if __name__ == "__main__":
main()
@@ -0,0 +1,5 @@
{
"VIDEOS_PER_CHANNEL": 5,
"COMMENTS_PER_VIDEO": 20,
"LOOKBACK_DAYS": 14
}
@@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""Comment Harvester — for every channel in WATCHED_CHANNELS, pulls the most
recent N videos and their top M comments. Appends the results to the agent's
memory.md so the YouTube agent can reference real audience reactions on the
next think step.
Reads from youtube_account.json (api key, watched channels) and
comment_harvester.json (volume settings)."""
import os, json, sys, time, datetime
HERE = os.path.dirname(os.path.abspath(__file__))
ACCOUNT = os.path.join(HERE, "youtube_account.json")
CONFIG = os.path.join(HERE, "comment_harvester.json")
# memory.md lives one level up — under _agents/youtube/
MEMORY = os.path.abspath(os.path.join(HERE, "..", "memory.md"))
REPORT = os.path.join(HERE, "comment_harvester_report.md")
def _load(p):
with open(p, "r", encoding="utf-8") as f:
return json.load(f)
def _resolve_channel_id(youtube, handle):
h = handle.lstrip("@")
try:
r = youtube.search().list(part="snippet", q=h, type="channel", maxResults=1).execute()
items = r.get("items", [])
if items:
return items[0]["snippet"]["channelId"], items[0]["snippet"]["title"]
except Exception as e:
print(f"⚠️ {handle} 채널 조회 실패: {e}")
return None, None
def main():
if not os.path.exists(ACCOUNT):
print("❌ youtube_account.json이 없어요. 먼저 그 도구로 설정.")
sys.exit(1)
acct = _load(ACCOUNT)
cfg = _load(CONFIG) if os.path.exists(CONFIG) else {}
api_key = (acct.get("YOUTUBE_API_KEY") or "").strip()
watched = acct.get("WATCHED_CHANNELS") or []
if not api_key:
print("❌ YOUTUBE_API_KEY 비어있음.")
sys.exit(1)
if not watched:
print("❌ WATCHED_CHANNELS가 비어있어요. youtube_account.json에 핸들 목록을 넣어주세요.")
print(' 예: "WATCHED_CHANNELS": ["@channel_a", "@channel_b"]')
sys.exit(1)
vids_per = int(cfg.get("VIDEOS_PER_CHANNEL", 5))
cmts_per = int(cfg.get("COMMENTS_PER_VIDEO", 20))
lookback = int(cfg.get("LOOKBACK_DAYS", 14))
try:
from googleapiclient.discovery import build
except ImportError:
print("❌ pip install google-api-python-client")
sys.exit(1)
youtube = build("youtube", "v3", developerKey=api_key)
after = (datetime.datetime.utcnow() - datetime.timedelta(days=lookback)).isoformat("T") + "Z"
harvested = []
for ch in watched:
cid, ctitle = _resolve_channel_id(youtube, ch)
if not cid:
continue
print(f"📡 [{ch}] 최근 영상 {vids_per}개 가져오는 중...")
sr = youtube.search().list(part="snippet", channelId=cid, maxResults=vids_per,
order="date", publishedAfter=after, type="video").execute()
for it in sr.get("items", []):
vid = it["id"]["videoId"]
vtitle = it["snippet"]["title"]
print(f" 💬 {vtitle[:60]}")
try:
cr = youtube.commentThreads().list(part="snippet", videoId=vid,
maxResults=cmts_per, order="relevance",
textFormat="plainText").execute()
except Exception as e:
msg = str(e)
if "commentsDisabled" in msg or "disabled" in msg.lower():
continue
print(f" ⚠️ 댓글 가져오기 실패: {e}")
continue
comments = []
for ci in cr.get("items", []):
top = ci["snippet"]["topLevelComment"]["snippet"]
comments.append({
"author": top.get("authorDisplayName", ""),
"likes": int(top.get("likeCount", 0)),
"text": (top.get("textDisplay", "") or "")[:280],
})
harvested.append({
"channel": ch, "channel_title": ctitle,
"video": vtitle, "video_id": vid, "comments": comments,
})
if not harvested:
print("⚠️ 수집된 댓글 없음.")
sys.exit(0)
ts = time.strftime('%Y-%m-%d %H:%M')
md_lines = [f"\n## 💬 시청자 댓글 수집 — {ts}"]
for h in harvested:
md_lines.append(f"\n### {h['channel_title']} ({h['channel']}) — {h['video']}")
md_lines.append(f"https://youtu.be/{h['video_id']}")
for c in h["comments"][:10]:
md_lines.append(f"- ({c['likes']}❤) **{c['author']}**: {c['text']}")
block = "\n".join(md_lines)
# Append to memory so the agent uses these comments next think.
os.makedirs(os.path.dirname(MEMORY), exist_ok=True)
if not os.path.exists(MEMORY):
with open(MEMORY, "w", encoding="utf-8") as f:
f.write("# YouTube 에이전트 — 메모리\n\n")
with open(MEMORY, "a", encoding="utf-8") as f:
f.write("\n" + block + "\n")
with open(REPORT, "a", encoding="utf-8") as f:
f.write("\n" + block + "\n\n---\n")
print(f"\n✅ 메모리에 추가: {MEMORY}")
print(f"✅ 보고서: {REPORT}")
print(f" {len(harvested)}개 영상 · 평균 {sum(len(h['comments']) for h in harvested)//max(len(harvested),1)}개 댓글")
if __name__ == "__main__":
main()
@@ -0,0 +1,4 @@
{
"TOP_N_PER_CHANNEL": 5,
"LOOKBACK_DAYS": 30
}
@@ -0,0 +1,20 @@
# 🔭 경쟁 채널 분석
`[[youtube_account]].json``COMPETITOR_CHANNELS`에 적은 경쟁 채널들의 최근 떡상 영상을 모아서, 로컬 LLM에게 **지시문 형식**의 다음 액션 브리프를 받아옵니다 — "이거 해야합니다 / 저거 해야합니다 / 이건 절대 하지 마세요" 형태로 나옵니다.
## 어떻게 도와주나요?
- 🔭 경쟁 채널마다 최근 N개 인기 영상(view 기준) 수집
- 🧠 로컬 LLM이 패턴을 읽고 4섹션으로 브리프 작성:
- 1) 지금 당장 해야 하는 것 3개
- 2) 이번 주 시도할 것 3개 (제목 후보 포함)
- 3) 절대 하지 말 것 1개
- 4) 다음 영상 핵심 한 줄
- 📨 텔레그램 설정돼있으면 자동 푸시
## 시작하기 전 체크
- `youtube_account.json``COMPETITOR_CHANNELS` 채워두기
- 로컬 LLM(Ollama/LM Studio)이 켜져 있어야 함
## 설정값 (competitor[[_brief]].json)
- `TOP_N_PER_CHANNEL` — 채널마다 상위 영상 몇 개 (기본 5)
- `LOOKBACK_DAYS` — 며칠치 (기본 30)
@@ -0,0 +1,156 @@
#!/usr/bin/env python3
"""Competitor Brief — for every channel in COMPETITOR_CHANNELS, pulls their
recent top-performing videos and asks the local LLM for a *prescriptive*
brief: what should YOU do next, given what's working for them.
Reads youtube_account.json (api key, competitors, ollama, model) and
competitor_brief.json (volume)."""
import os, json, sys, time, datetime
HERE = os.path.dirname(os.path.abspath(__file__))
ACCOUNT = os.path.join(HERE, "youtube_account.json")
CONFIG = os.path.join(HERE, "competitor_brief.json")
REPORT = os.path.join(HERE, "competitor_brief_report.md")
def _load(p):
with open(p, "r", encoding="utf-8") as f:
return json.load(f)
def _resolve_channel_id(youtube, handle):
h = handle.lstrip("@")
try:
r = youtube.search().list(part="snippet", q=h, type="channel", maxResults=1).execute()
items = r.get("items", [])
if items:
return items[0]["snippet"]["channelId"], items[0]["snippet"]["title"]
except Exception:
pass
return None, None
def _push_telegram(account, text):
token = (account.get("TELEGRAM_BOT_TOKEN") or "").strip()
chat = (account.get("TELEGRAM_CHAT_ID") or "").strip()
if not token or not chat:
return
try:
import requests
requests.post(f"https://api.telegram.org/bot{token}/sendMessage",
json={"chat_id": chat, "text": text[:4000], "parse_mode": "Markdown"},
timeout=10)
except Exception:
pass
def main():
if not os.path.exists(ACCOUNT):
print("❌ youtube_account.json이 없어요.")
sys.exit(1)
acct = _load(ACCOUNT)
cfg = _load(CONFIG) if os.path.exists(CONFIG) else {}
api_key = (acct.get("YOUTUBE_API_KEY") or "").strip()
competitors = acct.get("COMPETITOR_CHANNELS") or []
if not api_key:
print("❌ YOUTUBE_API_KEY 비어있음.")
sys.exit(1)
if not competitors:
print("❌ COMPETITOR_CHANNELS가 비어있어요. youtube_account.json에 채워주세요.")
sys.exit(1)
top_n = int(cfg.get("TOP_N_PER_CHANNEL", 5))
lookback = int(cfg.get("LOOKBACK_DAYS", 30))
ollama_url = (acct.get("OLLAMA_URL") or "http://127.0.0.1:11434").rstrip("/")
model = acct.get("MODEL") or ""
try:
from googleapiclient.discovery import build
import requests
except ImportError:
print("❌ pip install google-api-python-client requests")
sys.exit(1)
youtube = build("youtube", "v3", developerKey=api_key)
after = (datetime.datetime.utcnow() - datetime.timedelta(days=lookback)).isoformat("T") + "Z"
snapshot = []
for ch in competitors:
cid, ctitle = _resolve_channel_id(youtube, ch)
if not cid:
print(f"⚠️ {ch} 채널 못 찾음")
continue
print(f"🔭 [{ch}] 최근 영상 분석 중...")
sr = youtube.search().list(part="snippet", channelId=cid, maxResults=top_n,
order="viewCount", publishedAfter=after, type="video").execute()
ids = [it["id"]["videoId"] for it in sr.get("items", [])]
if not ids:
continue
st = youtube.videos().list(part="statistics,snippet", id=",".join(ids)).execute()
for it in st.get("items", []):
stats = it.get("statistics", {})
snip = it.get("snippet", {})
snapshot.append({
"channel": ctitle,
"title": snip.get("title", ""),
"views": int(stats.get("viewCount", 0)),
"published": snip.get("publishedAt", "")[:10],
})
if not snapshot:
print("❌ 데이터 수집 실패.")
sys.exit(1)
snapshot.sort(key=lambda r: r["views"], reverse=True)
data_text = "\n".join(f"[{r['channel']}] {r['views']:,}회 · {r['published']} · {r['title']}"
for r in snapshot[:25])
if not model:
try:
r = requests.get(f"{ollama_url}/api/tags", timeout=5)
r.raise_for_status()
models = [m["name"] for m in r.json().get("models", [])]
if not models:
print("❌ 로컬 LLM에 모델이 없어요.")
sys.exit(1)
model = models[0]
except Exception as e:
print(f"❌ LLM 연결 실패: {e}")
sys.exit(1)
prompt = f"""당신은 유튜브 알고리즘 전략가입니다. 아래는 경쟁 채널들의 최근 {lookback}일간 상위 영상 데이터입니다.
[경쟁 데이터]
{data_text}
이 채널 운영자에게 **지시문 형식**으로 다음을 작성하세요. 모호한 조언 금지, 구체적이고 실행 가능한 지시.
## 1) 지금 당장 해야 하는 것 (3개)
- 각 항목: "~을(를) 하세요. 왜냐하면 …"
## 2) 이번 주 안에 시도해야 하는 것 (3개)
- 각 항목: 구체적 영상 제목 후보 또는 후크 문장 포함
## 3) 절대 하지 말아야 할 것 (1개)
- 경쟁사 데이터에서 보이는 함정 패턴
## 4) 한 줄 요약
- 다음 영상의 핵심 컨셉을 한 문장으로
"""
print("🧠 [LLM 분석 중...]")
try:
r = requests.post(f"{ollama_url}/api/generate",
json={"model": model, "prompt": prompt, "stream": False},
timeout=240)
r.raise_for_status()
brief = r.json().get("response", "").strip()
except Exception as e:
print(f"❌ LLM 실패: {e}")
sys.exit(1)
ts = time.strftime('%Y-%m-%d %H:%M')
out = f"# 🔭 경쟁 채널 브리프 — {ts}\n\n채널: {', '.join(competitors)} · 최근 {lookback}\n\n{brief}\n"
print("\n" + "="*60)
print(out)
print("="*60)
with open(REPORT, "a", encoding="utf-8") as f:
f.write("\n\n" + out + "\n---\n")
print(f"\n✅ 보고서: {REPORT}")
_push_telegram(acct, out)
if __name__ == "__main__":
main()
@@ -0,0 +1,4 @@
{
"LOOKBACK_DAYS": 30,
"TOP_N": 10
}
@@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""My Videos Check — pulls your own channel's recent uploads, computes a
view-count baseline (median of last N), and flags which videos are above /
below the line. Outputs a short report. Optionally pings Telegram.
Reads YOUTUBE_API_KEY + MY_CHANNEL_HANDLE/ID from youtube_account.json.
Reads LOOKBACK_DAYS / TOP_N from my_videos_check.json."""
import os, json, sys, time, datetime
HERE = os.path.dirname(os.path.abspath(__file__))
ACCOUNT = os.path.join(HERE, "youtube_account.json")
CONFIG = os.path.join(HERE, "my_videos_check.json")
REPORT = os.path.join(HERE, "my_videos_check_report.md")
def _load(p):
with open(p, "r", encoding="utf-8") as f:
return json.load(f)
def _resolve_channel_id(youtube, handle, channel_id):
if channel_id:
return channel_id
if not handle:
return None
h = handle.lstrip("@")
try:
r = youtube.search().list(part="snippet", q=h, type="channel", maxResults=1).execute()
items = r.get("items", [])
if items:
return items[0]["snippet"]["channelId"]
except Exception as e:
print(f"⚠️ 채널 ID 조회 실패: {e}")
return None
def _push_telegram(account, text):
token = (account.get("TELEGRAM_BOT_TOKEN") or "").strip()
chat = (account.get("TELEGRAM_CHAT_ID") or "").strip()
if not token or not chat:
return
try:
import requests
requests.post(
f"https://api.telegram.org/bot{token}/sendMessage",
json={"chat_id": chat, "text": text, "parse_mode": "Markdown"},
timeout=10,
)
print("📨 텔레그램으로 보고 전송")
except Exception as e:
print(f"⚠️ 텔레그램 전송 실패: {e}")
def main():
if not os.path.exists(ACCOUNT):
print("❌ youtube_account.json이 없어요. 같은 폴더에서 youtube_account 도구를 먼저 실행/설정하세요.")
sys.exit(1)
acct = _load(ACCOUNT)
cfg = _load(CONFIG) if os.path.exists(CONFIG) else {}
api_key = (acct.get("YOUTUBE_API_KEY") or "").strip()
handle = (acct.get("MY_CHANNEL_HANDLE") or "").strip()
chan_id = (acct.get("MY_CHANNEL_ID") or "").strip()
if not api_key:
print("❌ YOUTUBE_API_KEY가 비어있어요. youtube_account.json에 채워주세요.")
sys.exit(1)
if not (handle or chan_id):
print("❌ MY_CHANNEL_HANDLE 또는 MY_CHANNEL_ID 중 하나는 채워야 해요.")
sys.exit(1)
lookback = int(cfg.get("LOOKBACK_DAYS", 30))
top_n = int(cfg.get("TOP_N", 10))
try:
from googleapiclient.discovery import build
except ImportError:
print("❌ google-api-python-client 미설치. pip install google-api-python-client requests")
sys.exit(1)
youtube = build("youtube", "v3", developerKey=api_key)
cid = _resolve_channel_id(youtube, handle, chan_id)
if not cid:
print("❌ 채널 ID를 찾지 못했어요. youtube_account.json의 핸들/ID 확인.")
sys.exit(1)
print(f"🎬 [내 영상 체크] 채널 {handle or cid} 최근 {top_n}개 분석 중...")
after = (datetime.datetime.utcnow() - datetime.timedelta(days=lookback)).isoformat("T") + "Z"
sr = youtube.search().list(part="snippet", channelId=cid, maxResults=top_n,
order="date", publishedAfter=after, type="video").execute()
vids = [(it["id"]["videoId"], it["snippet"]["title"], it["snippet"]["publishedAt"])
for it in sr.get("items", [])]
if not vids:
print(f"⚠️ 최근 {lookback}일 안에 업로드한 영상이 없어요.")
sys.exit(0)
stats = youtube.videos().list(part="statistics", id=",".join(v[0] for v in vids)).execute()
sm = {it["id"]: it["statistics"] for it in stats.get("items", [])}
rows = []
for vid, title, pub in vids:
s = sm.get(vid, {})
views = int(s.get("viewCount", 0))
likes = int(s.get("likeCount", 0))
comments = int(s.get("commentCount", 0))
rows.append({"id": vid, "title": title, "pub": pub[:10], "views": views, "likes": likes, "comments": comments})
rows.sort(key=lambda r: r["views"], reverse=True)
views_list = sorted([r["views"] for r in rows])
median = views_list[len(views_list)//2] if views_list else 0
print("\n" + "="*60)
print(f"중간값(median) 조회수: {median:,}")
print("="*60)
for r in rows:
marker = "🔥" if r["views"] >= median * 1.5 else ("👍" if r["views"] >= median else "🥶")
print(f"{marker} {r['views']:>7,}회 · {r['pub']} · {r['title'][:60]}")
print(f" https://youtu.be/{r['id']}")
above = [r for r in rows if r["views"] >= median * 1.5]
below = [r for r in rows if r["views"] < median * 0.5]
summary_lines = [
f"# 🎬 내 채널 체크 — {time.strftime('%Y-%m-%d %H:%M')}",
f"채널: {handle or cid} · 최근 {lookback}일 · 영상 {len(rows)}",
f"조회수 중간값: **{median:,}**",
"",
f"## 🔥 떡상 (중간값×1.5 이상) — {len(above)}",
]
for r in above[:5]:
summary_lines.append(f"- {r['views']:,}회 · {r['title']}")
summary_lines.append(f"\n## 🥶 부진 (중간값×0.5 미만) — {len(below)}")
for r in below[:5]:
summary_lines.append(f"- {r['views']:,}회 · {r['title']}")
summary_lines.append("\n## 다음 액션 (제안)")
if above:
summary_lines.append(f"- 🔥 떡상한 영상의 후크/제목 패턴을 트렌드 스나이퍼 결과와 교차 분석")
if below:
summary_lines.append(f"- 🥶 부진 영상은 썸네일 A/B 또는 제목 리네이밍 후보")
summary_lines.append("- 댓글 수집기를 돌려서 시청자 반응 키워드 확인")
summary = "\n".join(summary_lines)
with open(REPORT, "a", encoding="utf-8") as f:
f.write("\n\n" + summary + "\n\n---\n")
print(f"\n✅ 보고서: {REPORT}")
_push_telegram(acct, summary)
if __name__ == "__main__":
main()
@@ -0,0 +1 @@
{}
@@ -0,0 +1,55 @@
#!/usr/bin/env python3
"""Telegram Notify — small wrapper that sends a message to your Telegram bot.
Two modes:
1. No CLI arg → sends a connectivity test ("✅ 텔레그램 연결 정상").
2. With CLI arg(s) → sends those as the message body. Other tools can call
this script to push their summaries.
Reads TELEGRAM_BOT_TOKEN / TELEGRAM_CHAT_ID from youtube_account.json."""
import os, json, sys, time
HERE = os.path.dirname(os.path.abspath(__file__))
ACCOUNT = os.path.join(HERE, "youtube_account.json")
def main():
if not os.path.exists(ACCOUNT):
print("❌ youtube_account.json이 없어요.")
sys.exit(1)
with open(ACCOUNT, "r", encoding="utf-8") as f:
acct = json.load(f)
token = (acct.get("TELEGRAM_BOT_TOKEN") or "").strip()
chat = (acct.get("TELEGRAM_CHAT_ID") or "").strip()
if not token or not chat:
print("❌ TELEGRAM_BOT_TOKEN 또는 TELEGRAM_CHAT_ID가 비어있어요.")
print(" 봇 만들기: Telegram에서 @BotFather → /newbot")
print(" chat_id 찾기: 봇한테 메시지 한 번 보내고")
print(" https://api.telegram.org/bot<TOKEN>/getUpdates 열기")
sys.exit(1)
if len(sys.argv) > 1:
body = " ".join(sys.argv[1:])
else:
body = f"✅ 텔레그램 연결 정상 — {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n이 메시지가 보이면 다른 YouTube 도구들도 자동으로 보고를 보낼 수 있어요."
try:
import requests
except ImportError:
print("❌ pip install requests")
sys.exit(1)
try:
r = requests.post(
f"https://api.telegram.org/bot{token}/sendMessage",
json={"chat_id": chat, "text": body, "parse_mode": "Markdown"},
timeout=15,
)
r.raise_for_status()
print(f"✅ 전송 OK ({len(body)}자)")
except Exception as e:
print(f"❌ 전송 실패: {e}")
if "Bad Request" in str(e):
print(" chat_id가 정확한지, 봇과 한 번이라도 대화를 시작했는지 확인하세요.")
sys.exit(1)
if __name__ == "__main__":
main()
@@ -0,0 +1,8 @@
{
"TARGET_KEYWORDS": [
"유튜브 자동화",
"AI 비즈니스",
"마케팅 트렌드",
"생산성 툴"
]
}
@@ -0,0 +1,21 @@
# 🎯 트렌드 스나이퍼
유튜브 Data API로 최근 30일 떡상 영상을 수집하고, 로컬 LLM(Ollama/LM Studio)으로 패턴을 분석해 다음 영상 기획안(제목·썸네일·후크)을 도출합니다.
## 필요한 것
- Python 3 + `pip install google-api-python-client requests`
- `[[youtube_account]].json``YOUTUBE_API_KEY` 채우기 (한 번만)
- 로컬 LLM (Ollama 또는 LM Studio)이 켜져 있어야 함
## 설정값 (trend_sniper.json)
- `TARGET_KEYWORDS` — 분석할 키워드 배열
- (API 키·Ollama URL·모델은 공유 `youtube_account.json`에서 자동 로드)
## 실행 방법
패널의 [▶ 실행] 버튼을 누르거나 터미널에서:
```bash
python trend_sniper.py
```
## 출력
같은 폴더에 `trend_sniper_report.md` 누적 저장.
@@ -0,0 +1,151 @@
#!/usr/bin/env python3
"""Trend Sniper — pulls top YouTube videos for target keywords, asks a local
LLM (Ollama/LM Studio) to extract the algorithmic patterns, and writes a
planning report next to this script.
Shared keys (API key, OLLAMA_URL, MODEL) come from youtube_account.json so
you only set them once. Per-tool keys (TARGET_KEYWORDS) come from
trend_sniper.json. If a key exists in both, trend_sniper.json wins.
Requires: pip install google-api-python-client requests
"""
import os, json, time, random, datetime, sys
HERE = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(HERE, "trend_sniper.json")
ACCOUNT_PATH = os.path.join(HERE, "youtube_account.json")
REPORT_PATH = os.path.join(HERE, "trend_sniper_report.md")
def load_config():
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
print(f"❌ 설정 파일을 읽을 수 없어요: {CONFIG_PATH}\n{e}")
sys.exit(1)
def load_account():
try:
if os.path.exists(ACCOUNT_PATH):
with open(ACCOUNT_PATH, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def _shared(cfg, acct, key, default=""):
"""Per-tool config wins; falls back to shared account; finally default."""
v = cfg.get(key)
if v not in (None, "", []):
return v
v = acct.get(key)
if v not in (None, "", []):
return v
return default
def main():
cfg = load_config()
acct = load_account()
api_key = (_shared(cfg, acct, "YOUTUBE_API_KEY") or "").strip()
if not api_key:
print("⚠️ YOUTUBE_API_KEY가 비어있어요. youtube_account.json 또는 trend_sniper.json에 입력하세요.")
print(" 발급: https://console.cloud.google.com/ → YouTube Data API v3 사용 설정 → 사용자 인증 정보 → API 키")
sys.exit(1)
target_keywords = cfg.get("TARGET_KEYWORDS", [])
if not target_keywords:
print("⚠️ TARGET_KEYWORDS가 비어있어요. 분석할 키워드를 1개 이상 추가하세요.")
sys.exit(1)
ollama_url = (_shared(cfg, acct, "OLLAMA_URL", "http://127.0.0.1:11434") or "http://127.0.0.1:11434").rstrip("/")
model = _shared(cfg, acct, "MODEL", "") or ""
pick = min(2, len(target_keywords))
chosen = random.sample(target_keywords, pick)
try:
from googleapiclient.discovery import build
except ImportError:
print("❌ google-api-python-client가 설치되지 않았어요.")
print(" 설치: pip install google-api-python-client requests")
sys.exit(1)
try:
import requests
except ImportError:
print("❌ requests가 설치되지 않았어요. pip install requests")
sys.exit(1)
print(f"\n🎯 [트렌드 스나이퍼] 키워드 {chosen} 스캔 시작...")
youtube = build('youtube', 'v3', developerKey=api_key)
last_month = (datetime.datetime.utcnow() - datetime.timedelta(days=30)).isoformat("T") + "Z"
sniper_data = []
for q in chosen:
print(f"📡 [{q}] 검색 중...")
try:
req = youtube.search().list(
part="snippet", q=q, maxResults=5, order="viewCount",
publishedAfter=last_month, type="video"
)
res = req.execute()
for item in res.get('items', []):
title = item['snippet']['title']
channel = item['snippet']['channelTitle']
sniper_data.append(f"[{q}] 채널: {channel} | 제목: {title}")
except Exception as e:
print(f"❌ 검색 오류 ({q}): {e}")
if not sniper_data:
print("❌ 수집된 데이터 없음. API 키 한도/네트워크 확인.")
sys.exit(1)
data_text = "\n".join(sniper_data)
prompt = f"""당신은 유튜브 알고리즘 마스터마인드입니다. 아래는 최근 30일 떡상 영상입니다.
[키워드] {', '.join(chosen)}
[데이터]
{data_text}
분석해서 마크다운 보고서를 작성하세요. 반드시 3섹션:
1. 🌍 트렌드 해킹 분석 — 어떤 패턴이 조회수를 끌고 있는지
2. 🎯 빈집 털기 전략 — 차별화 가능한 틈새 주제
3. 🎬 파괴적 영상 기획안 — 썸네일 카피, 제목 3개, 후킹 오프닝(첫 5초)
"""
print("🧠 [LLM 분석 중...]")
if not model:
# Try first available model
try:
r = requests.get(f"{ollama_url}/api/tags", timeout=5)
r.raise_for_status()
models = [m["name"] for m in r.json().get("models", [])]
if not models:
print("❌ 로컬 LLM에 설치된 모델이 없어요. Ollama/LM Studio에서 모델을 풀(pull)하세요.")
sys.exit(1)
model = models[0]
except Exception as e:
print(f"❌ 로컬 LLM 연결 실패 ({ollama_url}): {e}")
sys.exit(1)
try:
r = requests.post(
f"{ollama_url}/api/generate",
json={"model": model, "prompt": prompt, "stream": False},
timeout=180,
)
r.raise_for_status()
report = r.json().get("response", "").strip()
except Exception as e:
print(f"❌ LLM 호출 실패: {e}")
sys.exit(1)
print("\n" + "="*60)
print(report)
print("="*60)
with open(REPORT_PATH, "a", encoding="utf-8") as f:
now = time.strftime('%Y-%m-%d %H:%M:%S')
f.write(f"\n\n# 🎯 트렌드 스나이핑 보고서 — {now}\n")
f.write(f"## 📡 키워드: {', '.join(chosen)}\n\n")
f.write(report)
f.write("\n\n---\n")
print(f"\n✅ 보고서 저장: {REPORT_PATH}")
if __name__ == "__main__":
main()
@@ -0,0 +1,11 @@
{
"YOUTUBE_API_KEY": "",
"MY_CHANNEL_HANDLE": "",
"MY_CHANNEL_ID": "",
"WATCHED_CHANNELS": [],
"COMPETITOR_CHANNELS": [],
"TELEGRAM_BOT_TOKEN": "",
"TELEGRAM_CHAT_ID": "",
"OLLAMA_URL": "http://127.0.0.1:11434",
"MODEL": ""
}
@@ -0,0 +1,23 @@
# 🔑 계정 / 채널 (공유 설정)
여기 한 번만 채워두면 다른 모든 YouTube 도구(트렌드 스나이퍼·내 영상 체크·댓글 수집기·경쟁 채널 분석·텔레그램 보고)가 이 값을 그대로 가져다 씁니다. 매번 도구마다 같은 키를 넣지 않아도 돼요.
## 채워야 하는 항목
| 키 | 설명 | 채우는 법 |
|---|---|---|
| `YOUTUBE_API_KEY` | YouTube Data API v3 키 | [console.cloud.google.com](https://console.cloud.google.com/) → 프로젝트 → "YouTube Data API v3" 사용 설정 → 사용자 인증 정보 → API 키. 무료 한도 충분(하루 10,000 단위). |
| `MY_CHANNEL_HANDLE` | 본인 채널 @핸들 | 예: `@mychannel`. 핸들 또는 ID 둘 중 하나만 채우면 됨. |
| `MY_CHANNEL_ID` | 본인 채널 ID (UCxxxx) | 핸들로 못 잡힐 때 백업용. studio.youtube.com → 설정 → 채널에서 확인. |
| `WATCHED_CHANNELS` | 댓글 수집 대상 채널 핸들 목록 | 예: `["@channel_a", "@channel_b"]`. 댓글 수집기가 이 채널들 최근 영상의 댓글을 메모리로 가져옵니다. |
| `COMPETITOR_CHANNELS` | 경쟁 채널 분석 대상 | 같은 형식. 경쟁 채널 분석 도구가 패턴을 뽑아 다음 액션을 추천합니다. |
| `TELEGRAM_BOT_TOKEN` | 텔레그램 봇 토큰 | @BotFather에서 /newbot으로 봇 만들고 받은 `123:ABC...` 형식 토큰. 비워두면 보고 알림 OFF. |
| `TELEGRAM_CHAT_ID` | 본인 chat_id | 봇한테 아무 메시지 보낸 뒤 `https://api.telegram.org/bot<TOKEN>/getUpdates` 열어서 `chat.id` 확인. |
| `OLLAMA_URL` | 로컬 LLM 주소 | 기본 `http://127.0.0.1:11434`. LM Studio면 보통 `http://127.0.0.1:1234`. |
| `MODEL` | 분석에 쓸 모델 이름 | 비워두면 첫 번째로 발견된 모델을 자동 선택. |
## 실행하면?
입력값이 제대로 들어왔는지 확인 리포트만 출력합니다 (실제 데이터 호출 X). 키가 비어있으면 알려줍니다.
## 어디 저장되나?
이 폴더의 `youtube_account.json` 한 파일에. 브레인 폴더 안이라 GitHub 백업도 같이 됩니다.
@@ -0,0 +1,46 @@
#!/usr/bin/env python3
"""YouTube Account / Channels — shared config for every YouTube tool.
This script doesn't fetch anything by itself. It's listed in the agent panel
so you can click ⚙️ once and fill in your API key, channel, watched
channels, etc. — and every other tool will read from here.
Running it just prints a sanity-check report so you can confirm the values
are loaded correctly (without leaking the full API key)."""
import os, json, sys
HERE = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(HERE, "youtube_account.json")
def load():
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f)
def main():
cfg = load()
api = (cfg.get("YOUTUBE_API_KEY") or "").strip()
masked = (api[:4] + "" + api[-3:]) if len(api) >= 8 else ("(빈 값)" if not api else "(짧음)")
print("─── YouTube 계정 / 채널 설정 ───")
print(f" API 키 : {masked}")
print(f" 내 채널 핸들 : {cfg.get('MY_CHANNEL_HANDLE') or '(없음)'}")
print(f" 내 채널 ID : {cfg.get('MY_CHANNEL_ID') or '(없음)'}")
watched = cfg.get('WATCHED_CHANNELS') or []
print(f" 감시 채널 ({len(watched)}개) : {', '.join(watched) if watched else '(없음)'}")
competitors = cfg.get('COMPETITOR_CHANNELS') or []
print(f" 경쟁 채널 ({len(competitors)}개): {', '.join(competitors) if competitors else '(없음)'}")
tg_bot = (cfg.get('TELEGRAM_BOT_TOKEN') or '').strip()
tg_chat = (cfg.get('TELEGRAM_CHAT_ID') or '').strip()
if tg_bot and tg_chat:
print(f" 텔레그램 : 연결됨 (chat {tg_chat})")
else:
print(f" 텔레그램 : 미설정 (보고 알림 비활성)")
print(f" Ollama URL : {cfg.get('OLLAMA_URL') or 'http://127.0.0.1:11434'}")
print(f" 분석 모델 : {cfg.get('MODEL') or '(자동 선택)'}")
if not api:
print("\n⚠️ API 키가 비어있어요. 다른 도구들이 동작하지 않습니다.")
print(" 발급: https://console.cloud.google.com/ → YouTube Data API v3")
sys.exit(1)
print("\n✅ 공유 설정 로드 OK. 다른 도구들이 이 값을 자동으로 사용합니다.")
if __name__ == "__main__":
main()