Initial: Link analyzer bot for #remora channel
This commit is contained in:
190
bot.py
Normal file
190
bot.py
Normal file
@@ -0,0 +1,190 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Discord bot for #remora channel - analyzes links in real-time
|
||||
Posts summaries, adds to Tududi inbox, maintains JSON history
|
||||
"""
|
||||
|
||||
import discord
|
||||
import os
|
||||
import json
|
||||
import re
|
||||
import requests
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load .env file
|
||||
load_dotenv()
|
||||
|
||||
# Config
|
||||
CHANNEL_ID = 1467557082583535729
|
||||
TRACKER_FILE = Path(__file__).parent / "tracker.json"
|
||||
TUDUDI_API_URL = os.getenv("TUDUDI_API_URL", "https://todo.dilain.com/api/v1")
|
||||
TUDUDI_API_KEY = os.getenv("TUDUDI_API_KEY")
|
||||
GATEWAY_URL = os.getenv("OPENCLAW_GATEWAY", "http://127.0.0.1:18789")
|
||||
GATEWAY_TOKEN = os.getenv("OPENCLAW_GATEWAY_TOKEN")
|
||||
|
||||
# Load or init tracker
|
||||
def load_tracker():
|
||||
if TRACKER_FILE.exists():
|
||||
with open(TRACKER_FILE) as f:
|
||||
return json.load(f)
|
||||
return {
|
||||
"channel_id": CHANNEL_ID,
|
||||
"processed_message_ids": [],
|
||||
"links": []
|
||||
}
|
||||
|
||||
def save_tracker(data):
|
||||
with open(TRACKER_FILE, "w") as f:
|
||||
json.dump(data, f, indent=2)
|
||||
|
||||
# Detect links in text
|
||||
def extract_urls(text):
|
||||
url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+'
|
||||
return re.findall(url_pattern, text)
|
||||
|
||||
# Fetch and analyze URL
|
||||
def analyze_url(url):
|
||||
"""Fetch URL and create summary"""
|
||||
try:
|
||||
print(f" 📥 Fetching: {url}")
|
||||
response = requests.get(url, timeout=5, headers={
|
||||
'User-Agent': 'Mozilla/5.0'
|
||||
})
|
||||
content = response.text[:2000] # First 2k chars
|
||||
|
||||
# Extract title
|
||||
title_match = re.search(r'<title[^>]*>([^<]+)</title>', content, re.IGNORECASE)
|
||||
title = title_match.group(1).strip() if title_match else url.split('/')[-1]
|
||||
|
||||
# Simple content type detection
|
||||
link_type = "webpage"
|
||||
if "github.com" in url:
|
||||
link_type = "GitHub"
|
||||
elif "reddit.com" in url:
|
||||
link_type = "Reddit"
|
||||
elif "youtube.com" in url or "youtu.be" in url:
|
||||
link_type = "YouTube"
|
||||
elif "tiktok.com" in url:
|
||||
link_type = "TikTok"
|
||||
elif "twitter.com" in url or "x.com" in url:
|
||||
link_type = "Twitter/X"
|
||||
|
||||
return {
|
||||
"title": title,
|
||||
"type": link_type,
|
||||
"status": "ok"
|
||||
}
|
||||
except Exception as e:
|
||||
print(f" ❌ Error fetching: {e}")
|
||||
return {
|
||||
"title": "Couldn't fetch",
|
||||
"type": "unknown",
|
||||
"status": "error",
|
||||
"error": str(e)
|
||||
}
|
||||
|
||||
# Send to Tududi inbox
|
||||
def add_to_tududi(title, url, link_type):
|
||||
"""Add to Tududi inbox with summary"""
|
||||
try:
|
||||
if not TUDUDI_API_KEY:
|
||||
print(" ⚠️ TUDUDI_API_KEY not set")
|
||||
return False
|
||||
|
||||
content = f"📌 {link_type}: {title}\n🔗 {url}"
|
||||
|
||||
response = requests.post(
|
||||
f"{TUDUDI_API_URL}/inbox",
|
||||
headers={
|
||||
"Authorization": f"Bearer {TUDUDI_API_KEY}",
|
||||
"Content-Type": "application/json"
|
||||
},
|
||||
json={"content": content},
|
||||
timeout=5
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
print(f" ✅ Added to Tududi: {title}")
|
||||
return True
|
||||
else:
|
||||
print(f" ⚠️ Tududi error: {response.status_code}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f" ❌ Tududi error: {e}")
|
||||
return False
|
||||
|
||||
# Discord bot
|
||||
intents = discord.Intents.default()
|
||||
intents.message_content = True
|
||||
|
||||
class LinkAnalyzerBot(discord.Client):
|
||||
async def on_ready(self):
|
||||
print(f"✅ Bot logged in as {self.user}")
|
||||
print(f"📍 Watching channel #remora ({CHANNEL_ID})")
|
||||
|
||||
async def on_message(self, message):
|
||||
# Ignore bot's own messages
|
||||
if message.author == self.user:
|
||||
return
|
||||
|
||||
# Only process #remora channel
|
||||
if message.channel.id != CHANNEL_ID:
|
||||
return
|
||||
|
||||
# Check for URLs
|
||||
urls = extract_urls(message.content)
|
||||
if not urls:
|
||||
return
|
||||
|
||||
# Skip if already processed
|
||||
tracker = load_tracker()
|
||||
if message.id in tracker["processed_message_ids"]:
|
||||
return
|
||||
|
||||
print(f"🔗 New link from {message.author}: {message.content}")
|
||||
|
||||
# Process each URL
|
||||
for url in urls:
|
||||
print(f" Processing: {url}")
|
||||
|
||||
# Analyze
|
||||
analysis = analyze_url(url)
|
||||
|
||||
# Add to Tududi
|
||||
add_to_tududi(analysis["title"], url, analysis["type"])
|
||||
|
||||
# Prepare response
|
||||
summary = f"📌 **{analysis['type']}**: {analysis['title']}"
|
||||
if analysis["status"] == "error":
|
||||
summary += f"\n⚠️ {analysis['error']}"
|
||||
|
||||
# Post summary in channel
|
||||
await message.reply(summary, mention_author=False)
|
||||
|
||||
# Add to tracker
|
||||
tracker["links"].append({
|
||||
"url": url,
|
||||
"title": analysis["title"],
|
||||
"type": analysis["type"],
|
||||
"author": str(message.author),
|
||||
"message_id": message.id,
|
||||
"date": datetime.now().isoformat(),
|
||||
"tududi": True
|
||||
})
|
||||
|
||||
# Update processed IDs
|
||||
tracker["processed_message_ids"].append(message.id)
|
||||
save_tracker(tracker)
|
||||
|
||||
# Main
|
||||
if __name__ == "__main__":
|
||||
token = os.getenv("DISCORD_BOT_TOKEN")
|
||||
if not token:
|
||||
print("❌ DISCORD_BOT_TOKEN not set!")
|
||||
print("Set it: export DISCORD_BOT_TOKEN='your_token'")
|
||||
exit(1)
|
||||
|
||||
bot = LinkAnalyzerBot(intents=intents)
|
||||
bot.run(token)
|
||||
Reference in New Issue
Block a user