docs: integrate 72 technical fragments into 5 high-density clusters and archive originals
This commit is contained in:
Antigravity Agent
@@ -1,4 +0,0 @@
|
||||
{
|
||||
"INTERVAL_HOURS": 2,
|
||||
"TOTAL_RUN_HOURS": 8
|
||||
}
|
||||
@@ -1,43 +0,0 @@
|
||||
#!/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()
|
||||
@@ -1,5 +0,0 @@
|
||||
{
|
||||
"VIDEOS_PER_CHANNEL": 5,
|
||||
"COMMENTS_PER_VIDEO": 20,
|
||||
"LOOKBACK_DAYS": 14
|
||||
}
|
||||
@@ -1,122 +0,0 @@
|
||||
#!/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()
|
||||
@@ -1,4 +0,0 @@
|
||||
{
|
||||
"TOP_N_PER_CHANNEL": 5,
|
||||
"LOOKBACK_DAYS": 30
|
||||
}
|
||||
@@ -1,156 +0,0 @@
|
||||
#!/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()
|
||||
@@ -1,4 +0,0 @@
|
||||
{
|
||||
"LOOKBACK_DAYS": 30,
|
||||
"TOP_N": 10
|
||||
}
|
||||
@@ -1,141 +0,0 @@
|
||||
#!/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()
|
||||
@@ -1 +0,0 @@
|
||||
{}
|
||||
@@ -1,55 +0,0 @@
|
||||
#!/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()
|
||||
@@ -1,8 +0,0 @@
|
||||
{
|
||||
"TARGET_KEYWORDS": [
|
||||
"유튜브 자동화",
|
||||
"AI 비즈니스",
|
||||
"마케팅 트렌드",
|
||||
"생산성 툴"
|
||||
]
|
||||
}
|
||||
@@ -1,151 +0,0 @@
|
||||
#!/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()
|
||||
@@ -1,11 +0,0 @@
|
||||
{
|
||||
"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": ""
|
||||
}
|
||||
@@ -1,46 +0,0 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user