266 lines
9.2 KiB
Python
266 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Nextcloud Talk Bot — Polling-based bridge to Claude
|
||
Polls Talk API for new messages, gets AI reply with knowledge-base context, posts back.
|
||
"""
|
||
|
||
import base64
|
||
import glob
|
||
import json
|
||
import os
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
# === CONFIG ===
|
||
NC_URL = "http://10.0.0.230:11001"
|
||
NC_USER = "admin"
|
||
NC_PASS = "1qaz!QAZ"
|
||
BOT_USER = "maximka"
|
||
BOT_PASS = "MaximkaBot2026"
|
||
CONVERSATION_TOKEN = "aecax6yg"
|
||
|
||
CLIPROXY_URL = "http://localhost:8317/v1/messages"
|
||
CLIPROXY_KEY = "sk-clawdbot-proxy"
|
||
|
||
KB_PATH = "/root/clawd/knowledge-base"
|
||
MAX_KB_SIZE = 80000 # ~80KB max for knowledge base in prompt
|
||
|
||
# Chat history for context
|
||
chat_history = []
|
||
MAX_HISTORY = 10 # last N message pairs
|
||
|
||
|
||
def load_knowledge_base():
|
||
"""Load all .md files from knowledge-base repo into context string"""
|
||
kb_parts = []
|
||
total_size = 0
|
||
|
||
# Priority order: projects first, then snippets, decisions, notes
|
||
patterns = [
|
||
f"{KB_PATH}/README.md",
|
||
f"{KB_PATH}/CLAUDE.md",
|
||
f"{KB_PATH}/projects/**/*.md",
|
||
f"{KB_PATH}/snippets/*.md",
|
||
f"{KB_PATH}/decisions/*.md",
|
||
f"{KB_PATH}/notes/README.md",
|
||
]
|
||
|
||
seen = set()
|
||
for pattern in patterns:
|
||
for fpath in sorted(glob.glob(pattern, recursive=True)):
|
||
if fpath in seen:
|
||
continue
|
||
seen.add(fpath)
|
||
|
||
try:
|
||
with open(fpath, "r") as f:
|
||
content = f.read()
|
||
if total_size + len(content) > MAX_KB_SIZE:
|
||
break
|
||
rel = os.path.relpath(fpath, KB_PATH)
|
||
kb_parts.append(f"### {rel}\n{content}")
|
||
total_size += len(content)
|
||
except Exception:
|
||
continue
|
||
|
||
return "\n\n---\n\n".join(kb_parts)
|
||
|
||
|
||
def build_system_prompt():
|
||
"""Build system prompt with knowledge base"""
|
||
kb = load_knowledge_base()
|
||
kb_size = len(kb)
|
||
|
||
prompt = (
|
||
"Ты Максимка — AI-ассистент DevOps-инженера Олега Батлаева. "
|
||
"Отвечай кратко и по делу. Используй русский язык. "
|
||
"Ты работаешь в Nextcloud Talk.\n\n"
|
||
"ВАЖНО: Ниже приведена полная база знаний инфраструктуры Олега. "
|
||
"Ты УЖЕ ЗНАЕШЬ всю его инфраструктуру — серверы, IP-адреса, пароли, проекты, решения. "
|
||
"Когда тебя спрашивают про серверы, сеть, настройки — ВСЕГДА отвечай на основе этих данных. "
|
||
"НЕ говори что у тебя нет доступа — информация уже есть ниже. "
|
||
"Отвечай конкретно, с IP-адресами, именами и деталями из базы знаний.\n\n"
|
||
)
|
||
|
||
if kb:
|
||
prompt += f"## База знаний ({kb_size} символов)\n\n{kb}"
|
||
|
||
return prompt
|
||
|
||
|
||
def nc_request(path, method="GET", data=None, user=None, password=None):
|
||
"""Nextcloud OCS API request"""
|
||
url = f"{NC_URL}/ocs/v2.php/apps/spreed/api/{path}"
|
||
payload = json.dumps(data).encode() if data else None
|
||
req = urllib.request.Request(url, data=payload, method=method)
|
||
req.add_header("OCS-APIRequest", "true")
|
||
req.add_header("Accept", "application/json")
|
||
req.add_header("Content-Type", "application/json")
|
||
creds = base64.b64encode(f"{user or NC_USER}:{password or NC_PASS}".encode()).decode()
|
||
req.add_header("Authorization", f"Basic {creds}")
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
return json.loads(resp.read())
|
||
|
||
|
||
def get_last_message_id():
|
||
"""Get the highest message ID in the conversation"""
|
||
result = nc_request(f"v1/chat/{CONVERSATION_TOKEN}?lookIntoFuture=0&limit=1")
|
||
messages = result.get("ocs", {}).get("data", [])
|
||
return max((m.get("id", 0) for m in messages), default=0)
|
||
|
||
|
||
def poll_new_messages(last_id):
|
||
"""Long-poll for new messages after last_id"""
|
||
try:
|
||
result = nc_request(
|
||
f"v1/chat/{CONVERSATION_TOKEN}?lookIntoFuture=1&limit=20"
|
||
f"&lastKnownMessageId={last_id}&timeout=30"
|
||
)
|
||
return result.get("ocs", {}).get("data", [])
|
||
except urllib.error.HTTPError as e:
|
||
if e.code == 304:
|
||
return []
|
||
raise
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def join_room():
|
||
"""Join conversation as bot user"""
|
||
try:
|
||
nc_request(
|
||
f"v4/room/{CONVERSATION_TOKEN}/participants/active",
|
||
method="POST", user=BOT_USER, password=BOT_PASS
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def send_message(text, reply_to=None):
|
||
"""Send message as bot user"""
|
||
join_room()
|
||
data = {"message": text[:32000]}
|
||
if reply_to:
|
||
data["replyTo"] = reply_to
|
||
nc_request(
|
||
f"v1/chat/{CONVERSATION_TOKEN}",
|
||
method="POST", data=data, user=BOT_USER, password=BOT_PASS
|
||
)
|
||
|
||
|
||
def get_ai_reply(message, author, system_prompt):
|
||
"""Get reply from Claude via cliproxy"""
|
||
global chat_history
|
||
|
||
# Build messages: KB as first user/assistant pair + history + new message
|
||
messages = [
|
||
{"role": "user", "content": f"[СИСТЕМНЫЙ КОНТЕКСТ]\n{system_prompt}\n[/СИСТЕМНЫЙ КОНТЕКСТ]\n\nПодтверди что ты Максимка и готов отвечать на вопросы."},
|
||
{"role": "assistant", "content": "Я Максимка — AI-ассистент Олега. База знаний загружена. Готов отвечать на вопросы об инфраструктуре, серверах и проектах."},
|
||
]
|
||
messages.extend(chat_history)
|
||
messages.append({"role": "user", "content": f"{author}: {message}"})
|
||
|
||
payload = json.dumps({
|
||
"model": "claude-sonnet-4-6",
|
||
"max_tokens": 2048,
|
||
"system": system_prompt,
|
||
"messages": messages
|
||
}).encode()
|
||
|
||
req = urllib.request.Request(CLIPROXY_URL, data=payload, method="POST")
|
||
req.add_header("Content-Type", "application/json")
|
||
req.add_header("x-api-key", CLIPROXY_KEY)
|
||
req.add_header("anthropic-version", "2023-06-01")
|
||
|
||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||
result = json.loads(resp.read())
|
||
content = result.get("content", [])
|
||
if content and isinstance(content, list):
|
||
reply = content[0].get("text", "")
|
||
else:
|
||
reply = str(content)
|
||
|
||
# Update history
|
||
chat_history.append({"role": "user", "content": f"{author}: {message}"})
|
||
chat_history.append({"role": "assistant", "content": reply})
|
||
|
||
# Trim history
|
||
while len(chat_history) > MAX_HISTORY * 2:
|
||
chat_history.pop(0)
|
||
chat_history.pop(0)
|
||
|
||
return reply
|
||
|
||
|
||
def main():
|
||
print("🤖 Максимка Talk Bot v5 (with Knowledge Base)")
|
||
print(f" NC: {NC_URL} | Chat: {CONVERSATION_TOKEN}")
|
||
|
||
# Git pull knowledge base
|
||
os.system(f"cd {KB_PATH} && git pull --quiet 2>/dev/null")
|
||
|
||
# Load knowledge base
|
||
system_prompt = build_system_prompt()
|
||
print(f" Knowledge base: {len(system_prompt)} chars in system prompt")
|
||
|
||
# Start from latest message
|
||
last_id = get_last_message_id()
|
||
print(f" Starting from message ID: {last_id}")
|
||
print(f" Listening...\n")
|
||
|
||
while True:
|
||
try:
|
||
messages = poll_new_messages(last_id)
|
||
|
||
for msg in messages:
|
||
msg_id = msg.get("id", 0)
|
||
if msg_id <= last_id:
|
||
continue
|
||
last_id = msg_id
|
||
|
||
actor_id = msg.get("actorId", "")
|
||
actor_type = msg.get("actorType", "")
|
||
actor_name = msg.get("actorDisplayName", "")
|
||
text = msg.get("message", "").strip()
|
||
msg_type = msg.get("messageType", "")
|
||
|
||
# Skip: own, system, bots, empty
|
||
if actor_id == BOT_USER:
|
||
continue
|
||
if actor_type == "bots":
|
||
continue
|
||
if msg_type in ("system", "command"):
|
||
continue
|
||
if not text:
|
||
continue
|
||
|
||
# Special commands
|
||
if text.lower() == "/reload":
|
||
os.system(f"cd {KB_PATH} && git pull --quiet 2>/dev/null")
|
||
system_prompt = build_system_prompt()
|
||
send_message(f"🔄 База знаний обновлена ({len(system_prompt)} символов)", reply_to=msg_id)
|
||
print(f"[RELOAD] KB reloaded: {len(system_prompt)} chars")
|
||
continue
|
||
|
||
print(f"[IN] {actor_name} (id:{msg_id}): {text[:80]}")
|
||
|
||
try:
|
||
reply = get_ai_reply(text, actor_name, system_prompt)
|
||
if reply:
|
||
send_message(reply, reply_to=msg_id)
|
||
print(f"[OUT] → {reply[:80]}...")
|
||
except Exception as e:
|
||
print(f"[ERR] Reply failed: {e}")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\nBye!")
|
||
break
|
||
except Exception as e:
|
||
print(f"[ERR] Poll: {e}")
|
||
time.sleep(3)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|