#!/usr/bin/env python3
"""
SEO Ranking Monitor for ai4managers.net
Checks Google search rankings for target keywords.
Updates data.json with results. Sends summary via Telegram.
Cron (weekly Sunday 6:17am):
17 6 * * 0 /home/clawd/agents-claude-env/bin/python3 /home/clawd/playgrounds/seo-monitor-ai4m/check-rankings.py >> /home/clawd/playgrounds/seo-monitor-ai4m/cron.log 2>&1
"""
import json
import sys
import time
import urllib.request
import urllib.parse
from datetime import datetime, timedelta, timezone
from pathlib import Path
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
DOMAIN = "ai4managers.net"
DATA_FILE = Path(__file__).parent / "data.json"
MAX_RESULTS = 50
PAUSE_BETWEEN = 8.0
# Telegram notification
TELEGRAM_TOKEN_FILE = Path.home() / ".claude/channels/telegram/.env"
TELEGRAM_CHAT_ID = "8286896218"
KEYWORDS = [
{"keyword": "Design OS IA managers", "target": "/blog/el-metodo-design-os"},
{"keyword": "AI Agent Squad gestión equipos", "target": "/blog/que-es-un-ai-agent-squad"},
{"keyword": "automatización IA para managers", "target": "/blog/5-senales-de-que-tu-flujo"},
{"keyword": "recuperar tiempo con agentes IA", "target": "/blog/como-los-managers-recuperan"},
{"keyword": "VibeCoding programar sin código IA", "target": "/blog/vibecoding-para-managers"},
{"keyword": "ROI inteligencia artificial management", "target": "/blog/el-roi-de-la-ia"},
{"keyword": "IA para managers no técnicos", "target": "/"},
{"keyword": "AI4Managers", "target": "/"},
]
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def load_data() -> dict:
if DATA_FILE.exists():
with open(DATA_FILE, "r") as f:
return json.load(f)
return {
"lastChecked": None, "nextCheck": None,
"geoScore": {"current": 47, "previous": None, "nextAudit": "2026-04-13"},
"indexation": {"sitemap": 18, "indexed": "pending", "newPosts": 6},
"keywords": [], "posts": [],
"platforms": {"google": 52, "chatgpt": 28, "perplexity": 35, "gemini": 45, "bing": 33},
"actionsCompleted": [], "actionsPending": [],
}
def save_data(data: dict) -> None:
DATA_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(DATA_FILE, "w") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def search_google(query: str, num_results: int = MAX_RESULTS) -> list[str]:
try:
from googlesearch import search
except ImportError:
print("ERROR: googlesearch-python not installed.", file=sys.stderr)
sys.exit(1)
urls = []
try:
for url in search(query, num_results=num_results, lang="es", sleep_interval=2):
urls.append(url)
except Exception as e:
print(f" WARNING: search failed for '{query}': {e}", file=sys.stderr)
return urls
def find_position(urls: list[str], domain: str) -> int | None:
for i, url in enumerate(urls, start=1):
if domain in url:
return i
return None
def send_email(subject: str, body: str) -> None:
"""Send monitoring results via AgentMail (milesofroberto@agentmail.to)."""
try:
from agentmail import AgentMail
client = AgentMail(api_key="am_us_8392836cd0a7490664687e31ca896b9f5cf3b18ebd392ca2ceebbad5f532dcf9")
client.inboxes.messages.send(
"milesofroberto@agentmail.to",
to="aguirrerjg@gmail.com",
subject=subject,
text=body
)
print("Email notification sent")
except Exception as e:
print(f"WARN: Email send failed: {e}")
def send_telegram(text: str) -> None:
try:
token = None
if TELEGRAM_TOKEN_FILE.exists():
for line in TELEGRAM_TOKEN_FILE.read_text().splitlines():
if line.startswith("TELEGRAM_BOT_TOKEN="):
token = line.split("=", 1)[1].strip()
if not token:
print("WARN: No Telegram token found")
return
data = urllib.parse.urlencode({"chat_id": TELEGRAM_CHAT_ID, "text": text, "parse_mode": "HTML"}).encode()
req = urllib.request.Request(f"https://api.telegram.org/bot{token}/sendMessage", data=data)
urllib.request.urlopen(req, timeout=10)
print("Telegram notification sent")
except Exception as e:
print(f"WARN: Telegram send failed: {e}")
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def run():
now = datetime.now(timezone.utc)
today_str = now.strftime("%Y-%m-%d")
iso_now = now.strftime("%Y-%m-%dT%H:%M:%SZ")
next_week = (now + timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%SZ")
data = load_data()
data["lastChecked"] = iso_now
data["nextCheck"] = next_week
existing_kw = {e["keyword"]: e for e in data.get("keywords", [])}
results_summary = []
for i, kw in enumerate(KEYWORDS):
keyword = kw["keyword"]
target = kw["target"]
print(f"[{i+1}/{len(KEYWORDS)}] Searching: {keyword} ...", end=" ", flush=True)
urls = search_google(keyword)
position = find_position(urls, DOMAIN)
entry = existing_kw.get(keyword)
if entry is None:
entry = {"keyword": keyword, "target": target, "googlePosition": None, "bingPosition": None, "history": []}
existing_kw[keyword] = entry
prev_position = entry.get("googlePosition")
entry["googlePosition"] = position
entry["target"] = target
entry.setdefault("history", [])
entry["history"].append({"date": today_str, "position": position})
entry["history"] = entry["history"][-52:]
if prev_position is not None and position is not None:
delta = prev_position - position
movement = f"+{delta}" if delta > 0 else str(delta)
else:
movement = "new"
pos_str = str(position) if position else ">50"
print(f"position={pos_str} ({movement})")
results_summary.append((keyword, pos_str, movement))
if i < len(KEYWORDS) - 1:
time.sleep(PAUSE_BETWEEN)
kw_order = [kw["keyword"] for kw in KEYWORDS]
ordered = [existing_kw[k] for k in kw_order if k in existing_kw]
extras = [existing_kw[k] for k in existing_kw if k not in kw_order]
data["keywords"] = ordered + extras
save_data(data)
# Print summary
print("\n" + "=" * 60)
print(f"SEO Monitor — {DOMAIN} — {today_str}")
print("=" * 60)
print(f"{'Keyword':<45} {'Pos':>5} {'Move':>6}")
print("-" * 60)
for keyword, pos, move in results_summary:
print(f"{keyword:<45} {pos:>5} {move:>6}")
print("-" * 60)
ranked = sum(1 for _, p, _ in results_summary if p != ">50")
print(f"Ranked in top 50: {ranked}/{len(results_summary)}")
print(f"Data saved to: {DATA_FILE}")
# Send Telegram notification
lines = [f"📊 SEO Monitor — {DOMAIN}", f"📅 {today_str}", ""]
for keyword, pos, move in results_summary:
icon = "🟢" if pos != ">50" else "⚪"
move_str = f" ({move})" if move != "new" else ""
lines.append(f"{icon} {keyword}: {pos}{move_str}")
lines.append(f"\n🎯 Top 50: {ranked}/{len(results_summary)}")
wins = [(kw, p) for kw, p, m in results_summary if p != ">50"]
if wins:
lines.append("\n🏆 Keywords rankeadas:")
for kw, p in wins:
lines.append(f" • {kw} → posición {p}")
send_telegram("\n".join(lines))
# Send email
email_lines = [f"SEO Monitor — {DOMAIN} — {today_str}\n"]
for keyword, pos, move in results_summary:
icon = "✅" if pos != ">50" else "⬜"
move_str = f" ({move})" if move != "new" else ""
email_lines.append(f"{icon} {keyword}: {pos}{move_str}")
email_lines.append(f"\nTop 50: {ranked}/{len(results_summary)}")
if wins:
email_lines.append("\n🏆 Keywords rankeadas:")
for kw, p in wins:
email_lines.append(f" • {kw} → posición {p}")
send_email(f"📊 SEO Monitor — {DOMAIN} — {today_str}", "\n".join(email_lines))
if __name__ == "__main__":
run()