【Python】Discord上で動作するAIチャットボットを作成する【NL②】

# Discord Bot Token
DISCORD_TOKEN=your_discord_token_here

# OpenAI API Key
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_MODEL=gpt-5-nano

# Guild and Channel IDs to monitor (format: guild_id:channel_id,guild_id:channel_id)
TARGET_PAIRS=123456789:987654321

# Data Paths
DATABASE_TXT=./data/database.txt
PRICE_DB_CSV=./data/price-db.csv
DAILY_LIMIT_STORE=./data/daily_limit.json

# Features
WEB_SEARCH=1
WEB_SEARCH_RESULTS=5
DEBUG=0

# Rate Limits
DAILY_LIMIT_DEFAULT=5
# format: role_id:limit,role_id:limit
DAILY_LIMIT_BY_ROLE=111222333:10

import os
import re
import csv
import json
import asyncio
from datetime import datetime, timedelta, timezone
from typing import List, Tuple, Dict, Any, Callable, Optional, Set

import requests
import discord
from discord import Intents
from dotenv import load_dotenv

# URLおよび空白制御用の正規表現
URL_RE = re.compile(r"(https?://[^\s<>()\"']+)", re.IGNORECASE)
SPACE_RE = re.compile(r"[ \t]+")

# サーバー状況確認用APIのプレースホルダ（例: Minecraftサーバーやウェブサービスなど）
# オリジナル版では特定のサーバーAPIが指定されていました。
SERVER_STATUS_API_URL = os.getenv("SERVER_STATUS_API_URL", "https://api.example.com/status")

def _safe_int(v: str, default: int) -> int:
    """文字列を安全に整数に変換する"""
    try:
        return int(v)
    except Exception:
        return default

def _dbg(enabled: bool, s: str) -> None:
    """デバッグ情報が出力可能であれば表示する"""
    if enabled:
        print(s, flush=True)

def _extract_urls(text: str) -> List[str]:
    """テキストからURLを抽出する"""
    if not text:
        return []
    urls = URL_RE.findall(text)
    out = []
    seen = set()
    for u in urls:
        u2 = u.rstrip(".,;:)]}＞】」』”’\"")
        if u2 and u2 not in seen:
            seen.add(u2)
            out.append(u2)
    return out

def _remove_urls_from_text(text: str) -> str:
    """テキストからURLを削除し、改行を整理する"""
    if not text:
        return ""
    t = URL_RE.sub("", text)
    t = re.sub(r"\n{3,}", "\n\n", t).strip()
    return t

def _wrap_url_no_embed(url: str) -> str:
    """Discordで埋め込み表示されないようにURLを<>で囲む"""
    u = (url or "").strip()
    if not u:
        return ""
    if u.startswith("<") and u.endswith(">"):
        return u
    return "<" + u + ">"

def _dedupe_preserve(seq: List[str]) -> List[str]:
    """順序を維持したままリストを重複排除する"""
    seen = set()
    out = []
    for x in seq:
        if not x:
            continue
        if x not in seen:
            seen.add(x)
            out.append(x)
    return out

def _make_progress(mention: str, label: str, filled: int, total: int = 5) -> str:
    """進捗状況を示す進捗バーを生成する"""
    filled = max(0, min(filled, total))
    bar = "■" * filled + "□" * (total - filled)
    return f"{mention}\n\n{label} [{bar}]"

def _final_message(mention: str, answer_body: str, ref_urls: List[str]) -> str:
    """最終的な回答メッセージを整形する"""
    refs = "\n".join([_wrap_url_no_embed(u) for u in ref_urls]) if ref_urls else "（なし）"
    return (
        f"{mention}\n\n"
        f"生成完了 [■■■■■]\n"
        f"----------------------------------------\n"
        f"{answer_body.strip()}\n\n"
        f"[参考リンク]\n"
        f"{refs}\n"
        f"----------------------------------------"
    )

def _tokenize_ja_mixed(text: str) -> List[str]:
    """日本語と英数字が混在したテキストを簡易的にトークナイズする"""
    t = (text or "").lower().strip()
    t = re.sub(
        r"[^\w\u3040-\u30ff\u3400-\u9fff\uFF10-\uFF19\uFF21-\uFF3A\uFF41-\uFF5A]+",
        " ",
        t,
        flags=re.UNICODE,
    )
    t = SPACE_RE.sub(" ", t).strip()
    if not t:
        return []
    parts = t.split(" ")
    tokens = []
    for p in parts:
        if not p:
            continue
        if len(p) == 1 and not p.isdigit():
            continue
        tokens.append(p)
    return tokens[:40]

def _read_text_file(path: str, limit_chars: int) -> str:
    """テキストファイルを読み込み、必要に応じて文字数を制限する"""
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            data = f.read()
        if limit_chars > 0 and len(data) > limit_chars:
            return data[:limit_chars]
        return data
    except Exception:
        return ""

def _split_paragraphs(db_text: str) -> List[str]:
    """テキストを段落ごとに分割する"""
    if not db_text:
        return []
    t = db_text.replace("\r\n", "\n").replace("\r", "\n")
    parts = re.split(r"\n{2,}", t)
    paras = []
    for p in parts:
        p2 = p.strip()
        if p2:
            paras.append(p2)
    return paras

def _score_paragraph(p: str, tokens: List[str]) -> int:
    """キーワードに基づいて段落の関連度をスコアリングする"""
    if not p or not tokens:
        return 0
    pl = p.lower()
    s = 0
    for tok in tokens:
        if tok and tok in pl:
            s += 3
    # 関連度を高めるためのキーワード
    boosts = [
        "faq", "ルール", "規約", "参加", "接続", "状況",
        "人数", "コマンド", "ログイン", "テクスチャ", "サポート"
    ]
    for b in boosts:
        if b in pl:
            s += 1
    if "http://" in pl or "https://" in pl:
        s += 1
    return s

def _select_relevant_db_paragraphs(db_text: str, user_text: str, max_paras: int = 8) -> List[str]:
    """ユーザーの入力に対して関連性の高い段落を抽出する"""
    paras = _split_paragraphs(db_text)
    tokens = _tokenize_ja_mixed(user_text)
    scored = []
    for p in paras:
        s = _score_paragraph(p, tokens)
        if s > 0:
            scored.append((s, p))
    scored.sort(key=lambda x: x[0], reverse=True)
    return [p for _, p in scored[:max_paras]]

def _extract_urls_from_paras(paras: List[str], max_urls: int = 10) -> List[str]:
    """段落リストからURLを抽出する"""
    urls: List[str] = []
    for p in paras or []:
        urls.extend(_extract_urls(p))
    urls = _dedupe_preserve(urls)
    return urls[:max_urls]

def _load_csv_rows(csv_path: str, user_text: str, max_rows: int = 30) -> Tuple[List[Dict[str, str]], List[str]]:
    """CSVから関連する行を検索・抽出する"""
    if not csv_path or not os.path.exists(csv_path):
        return [], []
    tokens = _tokenize_ja_mixed(user_text)
    urls: List[str] = []

    try:
        with open(csv_path, "r", encoding="utf-8", errors="ignore", newline="") as f:
            reader = csv.DictReader(f)
            all_rows = []
            for r in reader:
                all_rows.append({k: (v.strip() if isinstance(v, str) else "") for k, v in r.items()})
    except Exception:
        return [], []

    def row_text(r: Dict[str, str]) -> str:
        return " ".join([str(v) for v in r.values() if v])

    scored = []
    for r in all_rows:
        txt = row_text(r).lower()
        s = 0
        for tok in tokens:
            if tok and tok in txt:
                s += 2
        name_keys = ["item", "name", "id", "title", "アイテム", "商品名"]
        name_blob = " ".join([r.get(k, "") for k in name_keys if k in r]).lower()
        for tok in tokens[:12]:
            if tok and tok in name_blob:
                s += 3
        if s > 0:
            scored.append((s, r))
    scored.sort(key=lambda x: x[0], reverse=True)
    picked = [r for _, r in scored[:max_rows]]
    for r in picked:
        for v in r.values():
            urls.extend(_extract_urls(v))
    urls = _dedupe_preserve(urls)
    return picked, urls

def _compact_json(obj: Any, max_chars: int = 18000) -> str:
    """JSONをコンパクトな文字列に変換し、必要に応じて切り詰める"""
    try:
        s = json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
        if len(s) > max_chars:
            return s[:max_chars] + "…"
        return s
    except Exception:
        return ""

def _duckduckgo_html_search(query: str, max_results: int, timeout: int) -> Tuple[List[Dict[str, str]], List[str]]:
    """DuckDuckGoでHTML検索を実行し、結果とURLを返す"""
    if not query:
        return [], []
    q = query.strip()
    url = "https://duckduckgo.com/html/"
    params = {"q": q}
    headers = {"User-Agent": "Mozilla/5.0"}
    try:
        r = requests.get(url, params=params, headers=headers, timeout=timeout)
        if r.status_code != 200:
            return [], []
        html = r.text
    except Exception:
        return [], []

    results: List[Dict[str, str]] = []
    urls: List[str] = []
    anchors = re.findall(
        r'<a[^>]+class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)</a>',
        html,
        flags=re.IGNORECASE | re.DOTALL,
    )
    for href, title_html in anchors[: max_results * 2]:
        title = re.sub(r"<.*?>", "", title_html)
        title = re.sub(r"\s+", " ", title).strip()
        href = href.strip()
        if not href.startswith("http"):
            continue
        results.append({"title": title[:200], "url": href})
        urls.append(href)
        if len(results) >= max_results:
            break
    urls = _dedupe_preserve(urls)
    return results, urls

def _build_openai_input(text: str, image_urls: List[str], max_images: int) -> List[Dict[str, Any]]:
    """OpenAI API用の入力を構築する"""
    content: List[Dict[str, Any]] = [{"type": "input_text", "text": text}]
    n = 0
    for u in (image_urls or []):
        if not u:
            continue
        content.append({"type": "input_image", "image_url": u})
        n += 1
        if n >= max_images:
            break
    return [{"role": "user", "content": content}]

def _extract_text_from_responses_json(data: Dict[str, Any]) -> str:
    """OpenAI APIのレスポンスからテキストを抽出する"""
    ot = data.get("output_text")
    if isinstance(ot, str) and ot.strip():
        return ot.strip()

    parts: List[str] = []
    out = data.get("output", [])
    if isinstance(out, list):
        for item in out:
            if not isinstance(item, dict):
                continue
            content = item.get("content", [])
            if isinstance(content, list):
                for c in content:
                    if not isinstance(c, dict):
                        continue
                    t = c.get("type")
                    if t in ("output_text", "text"):
                        txt = c.get("text", "")
                        if isinstance(txt, str) and txt.strip():
                            parts.append(txt)
    return "\n".join([p for p in parts if p]).strip()

def _openai_responses_call(
    api_key: str,
    model: str,
    instructions: str,
    input_text: str,
    input_image_urls: Optional[List[str]] = None,
    max_input_images: int = 2,
    timeout: int = 45,
    max_output_tokens: int = 900,
) -> Tuple[str, Dict[str, Any], int, str]:
    """OpenAI APIを呼び出し、モデルの回答を生成する"""
    url = "https://api.openai.com/v1/responses"
    headers = {"Authorization": "Bearer " + api_key, "Content-Type": "application/json"}

    payload = {
        "model": model,
        "instructions": instructions,
        "input": _build_openai_input(input_text, input_image_urls or [], max_input_images),
        "text": {"format": {"type": "text"}},
        "max_output_tokens": max_output_tokens,
        "store": False,
    }

    try:
        r = requests.post(url, headers=headers, json=payload, timeout=timeout)
        req_id = r.headers.get("x-request-id", "") or ""
        status = r.status_code
        try:
            data = r.json()
        except Exception:
            data = {"error": {"message": "jsonではないレスポンス", "raw": r.text[:1200]}}
        return _extract_text_from_responses_json(data), data, status, req_id
    except Exception as e:
        return "", {"error": {"message": "リクエスト失敗: " + repr(e)}}, 0, ""

async def _run_blocking(fn: Callable[[], Any]) -> Any:
    """ブロッキング処理を別スレッドで実行する"""
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, fn)

def _parse_target_pairs(val: str) -> Set[Tuple[int, int]]:
    """GuildID:ChannelID の形式の文字列をパースする"""
    pairs: Set[Tuple[int, int]] = set()
    if not val:
        return pairs
    for chunk in val.split(","):
        chunk = chunk.strip()
        if not chunk:
            continue
        if ":" not in chunk:
            continue
        a, b = chunk.split(":", 1)
        try:
            gid = int(a.strip())
            cid = int(b.strip())
            if gid > 0 and cid > 0:
                pairs.add((gid, cid))
        except Exception:
            pass
    return pairs

def _looks_unknown(answer: str) -> bool:
    """AIが回答を知らないと判断しているかを確認する"""
    t = (answer or "").strip()
    if not t:
        return True
    head = t.replace("\n", " ")[:180]
    # AIが回答できない場合のキーワード（日本語・英語）
    keys = ["分かりません", "わかりません", "情報がありません", "不明", "判断できません", "I don't know", "unsure"]
    for k in keys:
        if k.lower() in head.lower():
            return True
    if len(t) < 40:
        return True
    return False

def _needs_server_status(user_text: str) -> bool:
    """ユーザーがサーバーの状況について尋ねているかを確認する"""
    t = (user_text or "").lower()
    keys = ["状況", "オンライン", "人数", "鯖落ち", "status", "ping"]
    for k in keys:
        if k in t:
            return True
    return False

def _fetch_server_status(timeout: int = 8) -> Dict[str, Any]:
    """サーバーの状況を取得する（ダミーのAPI呼び出し）"""
    try:
        r = requests.get(SERVER_STATUS_API_URL, timeout=timeout)
        if r.status_code != 200:
            return {"ok": False, "http": r.status_code}
        return {"ok": True, "data": r.json()}
    except Exception as e:
        return {"ok": False, "error": repr(e)}

def _is_thread_message(message: discord.Message) -> bool:
    """スレッド内のメッセージかどうかを判定する"""
    try:
        ch = message.channel
        if isinstance(ch, discord.Thread):
            return True
        t = getattr(ch, "type", None)
        if t in (
            discord.ChannelType.public_thread,
            discord.ChannelType.private_thread,
            discord.ChannelType.news_thread,
        ):
            return True
    except Exception:
        pass
    return False

def _is_reply_message(message: discord.Message) -> bool:
    """返信メッセージかどうかを判定する"""
    try:
        return message.reference is not None
    except Exception:
        return False

def _parse_role_daily_limits(val: str) -> Dict[int, int]:
    """ロールごとの利用制限をパースする"""
    out: Dict[int, int] = {}
    if not val:
        return out
    for chunk in val.split(","):
        c = chunk.strip()
        if not c:
            continue
        if ":" not in c:
            continue
        a, b = c.split(":", 1)
        try:
            rid = int(a.strip())
            lim = int(b.strip())
            if rid > 0:
                out[rid] = lim
        except Exception:
            pass
    return out

JST = timezone(timedelta(hours=9), "JST")

def _today_key() -> str:
    """今日の日付文字列を生成する"""
    return datetime.now(JST).strftime("%Y-%m-%d")

class DailyLimiter:
    """ユーザーごとの1日の利用制限を管理するクラス"""
    def __init__(self, store_path: str, debug: bool = False):
        self.store_path = store_path
        self.debug = debug
        self._lock = asyncio.Lock()
        self._cache: Dict[str, Any] = {"date": "", "users": {}}

    def _load_sync(self) -> Dict[str, Any]:
        """制限情報をファイルから読み込む"""
        try:
            if not self.store_path or not os.path.exists(self.store_path):
                return {"date": "", "users": {}}
            with open(self.store_path, "r", encoding="utf-8") as f:
                data = json.load(f)
            return data if isinstance(data, dict) else {"date": "", "users": {}}
        except Exception:
            return {"date": "", "users": {}}

    def _save_sync(self, data: Dict[str, Any]) -> None:
        """制限情報をファイルに保存する"""
        try:
            if not self.store_path:
                return
            os.makedirs(os.path.dirname(self.store_path) or ".", exist_ok=True)
            tmp = self.store_path + ".tmp"
            with open(tmp, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, separators=(",", ":"))
            os.replace(tmp, self.store_path)
        except Exception as e:
            _dbg(self.debug, "LIMIT_SAVE_ERROR %r" % e)

    async def ensure_today(self) -> None:
        """キャッシュされたデータが今日のものか確認する"""
        async with self._lock:
            today = _today_key()
            if self._cache.get("date") == today:
                return
            data = await _run_blocking(lambda: self._load_sync())
            if data.get("date") != today:
                data = {"date": today, "users": {}}
                await _run_blocking(lambda: self._save_sync(data))
            self._cache = data

    async def get_count(self, user_id: int) -> int:
        """ユーザーの今日の利用回数を取得する"""
        await self.ensure_today()
        async with self._lock:
            return int(self._cache.get("users", {}).get(str(user_id), 0))

    async def inc(self, user_id: int, delta: int = 1) -> int:
        """ユーザーの利用回数を加算する"""
        await self.ensure_today()
        async with self._lock:
            users = self._cache.get("users", {})
            key = str(user_id)
            cur = int(users.get(key, 0)) + int(delta)
            users[key] = cur
            self._cache["users"] = users
            await _run_blocking(lambda: self._save_sync(self._cache))
            return cur

class SupportBot(discord.Client):
    """サポートAIボットの本体プログラム"""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

        self.debug = os.getenv("DEBUG", "0").strip() == "1"
        self.api_key = os.getenv("OPENAI_API_KEY", "").strip()
        self.model = os.getenv("OPENAI_MODEL", "gpt-4o")

        self.db_path = os.getenv("DATABASE_TXT", "./data/database.txt")
        self.price_csv_path = os.getenv("PRICE_DB_CSV", "./data/price-db.csv")

        self.max_db_chars = _safe_int(os.getenv("MAX_DB_CHARS", "200000"), 200000)
        self.max_price_rows = _safe_int(os.getenv("MAX_PRICE_ROWS", "30"), 30)

        self.web_search = os.getenv("WEB_SEARCH", "1").strip() == "1"
        self.web_search_results = _safe_int(os.getenv("WEB_SEARCH_RESULTS", "5"), 5)
        self.web_search_timeout = _safe_int(os.getenv("WEB_SEARCH_TIMEOUT", "8"), 8)

        self.openai_timeout = _safe_int(os.getenv("OPENAI_TIMEOUT", "45"), 45)
        self.openai_max_output_tokens = _safe_int(os.getenv("OPENAI_MAX_OUTPUT_TOKENS", "2000"), 2000)

        self.target_pairs = _parse_target_pairs(os.getenv("TARGET_PAIRS", "").strip())
        self.daily_limit_default = _safe_int(os.getenv("DAILY_LIMIT_DEFAULT", "0"), 0)
        self.daily_limit_by_role = _parse_role_daily_limits(os.getenv("DAILY_LIMIT_BY_ROLE", "").strip())
        self.daily_limit_store = os.getenv("DAILY_LIMIT_STORE", "./data/daily_limit.json").strip()
        self.limiter = DailyLimiter(self.daily_limit_store, debug=self.debug)

    def _is_target(self, guild_id: int, channel_id: int) -> bool:
        """ボットが反応すべきギルドとチャンネルかを確認する"""
        return (guild_id, channel_id) in self.target_pairs

    def _effective_daily_limit(self, member: discord.Member) -> int:
        """メンバーに適用される1日の利用回数制限を取得する"""
        lim = self.daily_limit_default
        try:
            for r in getattr(member, "roles", []) or []:
                rid = getattr(r, "id", None)
                if rid in self.daily_limit_by_role:
                    lim = max(lim, self.daily_limit_by_role[rid])
        except Exception:
            pass
        return lim

    async def on_ready(self):
        _dbg(self.debug, f"ログイン完了 ユーザー={self.user} ID={self.user.id}")

    async def on_message(self, message: discord.Message):
        """メッセージを受信した際の処理"""
        if message.author.bot or not message.guild:
            return
        if _is_thread_message(message) or _is_reply_message(message):
            return
        if not self._is_target(message.guild.id, message.channel.id):
            return

        mention = message.author.mention
        user_text = message.content or ""

        member = message.author if isinstance(message.author, discord.Member) else None
        if member:
            lim = self._effective_daily_limit(member)
            if lim > 0:
                cnt = await self.limiter.get_count(member.id)
                if cnt >= lim:
                    await message.channel.send(f"{mention}\n\n1日の利用上限に達しました。時間を置いてから再度お試しください。")
                    return

        # 添付画像の処理
        image_urls = []
        for att in (message.attachments or []):
            ct = (att.content_type or "").lower()
            if ct.startswith("image/"):
                image_urls.append(att.url)

        status_msg = await message.channel.send(_make_progress(mention, "情報取得中...", 1))

        try:
            # 1. データベース検索
            db_text_full = _read_text_file(self.db_path, self.max_db_chars)
            db_hit_paras = _select_relevant_db_paragraphs(db_text_full, user_text)
            
            # 2. CSVデータ検索
            price_rows, _ = _load_csv_rows(self.price_csv_path, user_text)

            # 3. サーバー状況取得
            server_status = None
            if _needs_server_status(user_text):
                server_status = await _run_blocking(_fetch_server_status)

            await status_msg.edit(content=_make_progress(mention, "回答生成中...", 3))

            # 利用制限のカウントアップ
            if member and lim > 0:
                await self.limiter.inc(member.id, 1)

            ctx = {
                "user_message": user_text,
                "database_hits": db_hit_paras,
                "csv_hits": price_rows,
                "server_status": server_status,
            }

            instructions = (
                "あなたはRPGプロジェクト（RPG-Project）のナレッジベースを用いた、親切なサポートAIである。\n"
                "提供される文脈（CONTEXT）に基づいて、ユーザーの質問に回答せよ。\n"
                "ルール:\n"
                "- 本文にURLを直接含めない（参考リンクとして後述すること）。\n"
                "- 簡潔かつ丁寧に回答すること。\n"
                "- 不確かな情報は推測で答えず、正直にわからないと伝えること。\n"
            )

            input_text = f"### CONTEXT\n{_compact_json(ctx)}\n\n### USER QUERY\n{user_text}"

            answer, _, _, _ = await _run_blocking(lambda: _openai_responses_call(
                api_key=self.api_key,
                model=self.model,
                instructions=instructions,
                input_text=input_text,
                input_image_urls=image_urls,
                timeout=self.openai_timeout,
                max_output_tokens=self.openai_max_output_tokens,
            ))

            # 回答が不明な場合のウェブ検索フォールバック
            if _looks_unknown(answer) and self.web_search:
                await status_msg.edit(content=_make_progress(mention, "ウェブ検索中...", 4))
                web_results, web_urls = await _run_blocking(lambda: _duckduckgo_html_search(user_text, self.web_search_results, self.web_search_timeout))
                ctx["web_search_results"] = web_results
                
                answer, _, _, _ = await _run_blocking(lambda: _openai_responses_call(
                    api_key=self.api_key,
                    model=self.model,
                    instructions=instructions,
                    input_text=f"### UPDATED CONTEXT\n{_compact_json(ctx)}\n\n### USER QUERY\n{user_text}",
                    input_image_urls=image_urls,
                ))

            body = _remove_urls_from_text(answer).strip()
            final_text = _final_message(mention, body, []) # 必要に応じてURLを追加可能
            
            await status_msg.edit(content=final_text)

        except Exception as e:
            _dbg(self.debug, f"HANDLER_ERROR {e}")
            await status_msg.edit(content=f"{mention}\n\n内部エラーが発生しました。")

def main():
    """メインエントリーポイント"""
    load_dotenv()
    token = os.getenv("DISCORD_TOKEN", "").strip()
    if not token or not os.getenv("OPENAI_API_KEY"):
        print("環境変数（トークンやAPIキー）が不足しています。")
        return

    intents = Intents.default()
    intents.message_content = True
    bot = SupportBot(intents=intents)
    bot.run(token)

if __name__ == "__main__":
    main()