rag_dcbot.py

import os
os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "300" 

import discord, requests, ollama, urllib.parse
from discord import app_commands
from discord.ext import commands
from dotenv import load_dotenv
from langchain_community.vectorstores import FAISS
from embeddings import EmbeddingGemmaEmbeddings
from duckduckgo_search import DDGS

load_dotenv()
TOKEN = os.getenv("DISCORD_TOKEN")
WEATHER_KEY = os.getenv("OPENWEATHER_API_KEY")
TDX_ID = os.getenv("TDX_CLIENT_ID")
TDX_SECRET = os.getenv("TDX_CLIENT_SECRET")

SCAM_KEYWORDS = ["免費領取", "飆股", "投資獲利", "加賴領取", "點擊連結", "中獎通知", "穩賺不賠"]
BAN_WORDS = ["@everyone", "@here"]

intents = discord.Intents.all()
bot = commands.Bot(command_prefix="!", intents=intents)

embed_model = EmbeddingGemmaEmbeddings()
mem = {}
db_med = None
db_food = None

if os.path.exists("db_medical"):
    db_med = FAISS.load_local("db_medical", embed_model, allow_dangerous_deserialization=True)
if os.path.exists("db_food"):
    db_food = FAISS.load_local("db_food", embed_model, allow_dangerous_deserialization=True)

def get_coordinates(address):
    try:
        url = "https://nominatim.openstreetmap.org/search"
        params = {"q": address, "format": "json", "limit": 1, "countrycodes": "tw"}
        headers = {"User-Agent": "DiscordBot_Traffic_Assistant/1.0"}
        r = requests.get(url, params=params, headers=headers).json()
        if r and len(r) > 0:
            return r[0]['lat'], r[0]['lon'], r[0]['display_name']
        return None, None, None
    except:
        return None, None, None

def get_tdx_token():
    if not TDX_ID or not TDX_SECRET: return None
    try:
        url = "https://tdx.transportdata.tw/auth/realms/TDXConnect/protocol/openid-connect/token"
        data = {"grant_type": "client_credentials", "client_id": TDX_ID, "client_secret": TDX_SECRET}
        return requests.post(url, data=data).json().get("access_token")
    except:
        return None

def search_web(query):
    try:
        results = DDGS().text(keywords=query, region='tw-tzh', max_results=5)
        context = ""
        for res in results:
            context += f"標題: {res['title']}\n摘要: {res['body']}\n連結: {res['href']}\n\n"
        return context
    except:
        return "無法連線至搜尋引擎"

async def run_llm(itx, user_msg, system_role, context_data="", uid=None):
    h = mem.get(uid, "") if uid else ""
    full_prompt = f"""
    [System Instruction]
    {system_role}
    [Context Info]
    {context_data}
    [Chat History]
    {h}
    [User Question]
    {user_msg}
    [Requirement]
    Please answer in Traditional Chinese.
    Do NOT use emojis.
    """
    r = ollama.chat(model="llama3.2:3b", messages=[{'role':'user','content':full_prompt}])['message']['content']
    if any(w in r for w in BAN_WORDS):
        r = f"偵測到不合理使用，請 {itx.user.mention} 勿濫用本系統"
    if uid:
        mem[uid] = (h + f"\nUser:{user_msg}\nAI:{r}")[-800:]
    await itx.followup.send(r)

@bot.event
async def on_ready():
    print(f"Login: {bot.user}")

@bot.event
async def on_member_join(member):
    channel = member.guild.system_channel
    if channel is not None:
        msg = f"歡迎 {member.mention} 加入\n請輸入 / 來查看功能"
        await channel.send(msg)

@bot.event
async def on_message(msg):
    if msg.author == bot.user: return
    if any(keyword in msg.content for keyword in SCAM_KEYWORDS):
        try:
            await msg.delete()
            await msg.channel.send(f"{msg.author.mention} 偵測到詐騙或惡意廣告，已自動攔截")
        except: pass
        return
    await bot.process_commands(msg)

@bot.command()
async def sync(ctx):
    await bot.tree.sync()
    await ctx.send("指令已同步")

@bot.tree.command(name="chat", description="與三小助手閒聊")
async def chat(itx: discord.Interaction, msg: str):
    await itx.response.defer()
    role = "你是一個樂於助人的助手，請用簡短的繁體中文回答。"
    await run_llm(itx, msg, role, uid=itx.user.id)

@bot.tree.command(name="explore", description="搜尋附近餐廳或住宿")
@app_commands.choices(
    category=[
        app_commands.Choice(name="餐廳/美食", value="美食餐廳"),
        app_commands.Choice(name="旅館/住宿", value="飯店住宿"),
    ],
    budget=[
        app_commands.Choice(name="平價 ($)", value="平價"),
        app_commands.Choice(name="中價位 ($$)", value="中價位"),
        app_commands.Choice(name="高價位 ($$$)", value="高級奢華"),
    ]
)
async def explore(itx: discord.Interaction, location: str, category: app_commands.Choice[str], budget: app_commands.Choice[str]):
    await itx.response.defer()
    search_query = f"{location} {budget.value} {category.value} 推薦評價"
    real_data = search_web(search_query)
    role = f"你是一位旅遊與美食達人，請根據 [Context Info] 推薦 3 間真實存在的{category.value}。"
    await run_llm(itx, f"推薦 {location} 的 {budget.value} {category.value}", role, context_data=real_data, uid=itx.user.id)

@bot.tree.command(name="love", description="情感諮詢")
async def love(itx: discord.Interaction, msg: str):
    await itx.response.defer()
    role = "你是一位心理諮詢師，請用同理心回答人際關係或情感問題。"
    await run_llm(itx, msg, role, uid=itx.user.id)

@bot.tree.command(name="uber", description="生成前往指定地點的 Uber 連結")
async def uber(itx: discord.Interaction, destination: str, nickname: str = None):
    await itx.response.defer()
    lat, lon, full_name = get_coordinates(destination)
    encoded_dest = urllib.parse.quote(destination)
    
    base_url = "https://m.uber.com/ul/?action=setPickup"
    if lat and lon:
        uber_link = f"{base_url}&dropoff[latitude]={lat}&dropoff[longitude]={lon}&dropoff[formatted_address]={encoded_dest}"
    else:
        uber_link = f"{base_url}&dropoff[formatted_address]={encoded_dest}"
    
    if nickname:
        uber_link += f"&dropoff[nickname]={urllib.parse.quote(nickname)}"

    embed_res = discord.Embed(
        title="Uber 叫車連結已生成",
        description=f"目的地: {nickname if nickname else destination}",
        color=0x000000
    )
    embed_res.add_field(name="詳細地址", value=full_name if full_name else destination, inline=False)
    embed_res.add_field(name="叫車服務", value=f"[點我開啟 Uber App]({uber_link})", inline=False)
    
    await itx.followup.send(embed=embed_res)

@bot.tree.command(name="weather", description="查詢城市天氣")
async def weather(itx: discord.Interaction, city: str):
    await itx.response.defer()
    try:
        url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={WEATHER_KEY}&units=metric&lang=zh_tw"
        d = requests.get(url).json()
        await itx.followup.send(f"{d['name']} 溫度: {d['main']['temp']}C 狀況: {d['weather'][0]['description']}")
    except:
        await itx.followup.send("查詢失敗")

if __name__ == "__main__":
    bot.run(TOKEN)