96 lines
3.3 KiB
Python
Executable File
96 lines
3.3 KiB
Python
Executable File
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = ["nanobot-ai"]
|
|
# ///
|
|
"""Nightly compact-memory auto run, triggered by the nanobot user crontab.
|
|
|
|
Runs the compact-memory skill in auto mode via the Nanobot Python API with a
|
|
fresh, never-reused `session_key` per invocation, then delivers the result
|
|
directly to Telegram. This bypasses nanobot's cron/jobs.json `agent_turn`
|
|
mechanism entirely, so the turn never gets appended to the user's live chat
|
|
session (which caused context contamination across nights) and needs no
|
|
delivery-gating evaluator (there is none for bound cron jobs in nanobot-ai
|
|
0.2.2 — whatever the agent answers would otherwise go straight to the chat).
|
|
|
|
Same pattern as skills/remind/scripts/remind_send.py and
|
|
skills/detach/scripts/tasks-daemon.py: external script, direct Telegram Bot
|
|
API delivery, no nanobot channel pipeline involved.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import traceback
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from nanobot import Nanobot
|
|
|
|
CONFIG = Path.home() / ".nanobot" / "config.json"
|
|
FALLBACK_CHAT_ID = "8826147089"
|
|
TIMEOUT_SECONDS = 10 * 60
|
|
|
|
GOAL = (
|
|
"Read skills/compact-memory/SKILL.md and run its Auto mode to audit and "
|
|
"compact memory/MEMORY.md. Perform every step yourself using your file "
|
|
"tools — there is no script to run."
|
|
)
|
|
|
|
|
|
def _telegram_config() -> tuple[str, str]:
|
|
"""Return (bot token, chat id). Chat id reads channels.telegram.allowFrom[0], with a constant fallback."""
|
|
data = json.loads(CONFIG.read_text(encoding="utf-8"))
|
|
telegram = data["channels"]["telegram"]
|
|
allow_from = telegram.get("allowFrom") or []
|
|
chat_id = str(allow_from[0]) if allow_from else FALLBACK_CHAT_ID
|
|
return telegram["token"], chat_id
|
|
|
|
|
|
def _send_telegram(text: str, token: str, chat_id: str) -> None:
|
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
|
payload = urllib.parse.urlencode({"chat_id": chat_id, "text": text}).encode()
|
|
req = urllib.request.Request(url, data=payload, method="POST")
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
resp.read()
|
|
|
|
|
|
async def _run_audit(session_key: str) -> str:
|
|
bot = Nanobot.from_config()
|
|
result = await bot.run(GOAL, session_key=session_key)
|
|
return result.content or ""
|
|
|
|
|
|
def main() -> int:
|
|
session_key = f"compact-memory-auto:{datetime.now():%Y%m%d-%H%M%S}"
|
|
|
|
try:
|
|
content = asyncio.run(asyncio.wait_for(_run_audit(session_key), timeout=TIMEOUT_SECONDS))
|
|
except asyncio.TimeoutError:
|
|
print(f"compact_memory_auto: timeout after {TIMEOUT_SECONDS // 60} min (session={session_key})", file=sys.stderr)
|
|
return 1
|
|
except Exception as e:
|
|
print(f"compact_memory_auto: run failed (session={session_key}): {e}\n{traceback.format_exc()}", file=sys.stderr)
|
|
return 1
|
|
|
|
if not content.strip():
|
|
print(f"compact_memory_auto: empty response (session={session_key})", file=sys.stderr)
|
|
return 1
|
|
|
|
token, chat_id = _telegram_config()
|
|
try:
|
|
_send_telegram(content, token, chat_id)
|
|
except Exception as e:
|
|
print(f"compact_memory_auto: telegram delivery failed (session={session_key}): {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|