From d7176305b5658db05d2e2713c4b389d2081e3b51 Mon Sep 17 00:00:00 2001
From: Kyomotoi <kyomotoiowo@gmail.com>
Date: Tue, 5 Apr 2022 15:18:00 +0800
Subject: =?UTF-8?q?=F0=9F=9A=9A=20=E6=8F=92=E4=BB=B6=20=20=E9=87=8D?=
=?UTF-8?q?=E5=91=BD=E5=90=8D=E4=B8=BA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
ATRI/plugins/chat/__init__.py | 74 ---------------------
ATRI/plugins/chat/data_source.py | 136 ---------------------------------------
ATRI/plugins/kimo/__init__.py | 59 +++++++++++++++++
ATRI/plugins/kimo/data_source.py | 136 +++++++++++++++++++++++++++++++++++++++
test/test_plugin_chat.py | 84 ------------------------
test/test_plugin_kimo.py | 43 +++++++++++++
6 files changed, 238 insertions(+), 294 deletions(-)
delete mode 100644 ATRI/plugins/chat/__init__.py
delete mode 100644 ATRI/plugins/chat/data_source.py
create mode 100644 ATRI/plugins/kimo/__init__.py
create mode 100644 ATRI/plugins/kimo/data_source.py
delete mode 100644 test/test_plugin_chat.py
create mode 100644 test/test_plugin_kimo.py
diff --git a/ATRI/plugins/chat/__init__.py b/ATRI/plugins/chat/__init__.py
deleted file mode 100644
index 3cb5dbb..0000000
--- a/ATRI/plugins/chat/__init__.py
+++ /dev/null
@@ -1,74 +0,0 @@
-from random import choice
-
-from nonebot.matcher import Matcher
-from nonebot.params import ArgPlainText, CommandArg
-from nonebot.adapters.onebot.v11 import MessageEvent, Message
-from nonebot.adapters.onebot.v11.helpers import Cooldown
-
-from ATRI.utils.apscheduler import scheduler
-from .data_source import Chat
-
-
-_chat_flmt_notice = choice(["慢...慢一..点❤", "冷静1下", "歇会歇会~~", "我开始为你以后的伴侣担心了..."])
-
-
-chat = Chat().on_message("文爱", "闲聊（文爱")
-
-
-@chat.handle([Cooldown(3, prompt=_chat_flmt_notice)])
-async def _chat(event: MessageEvent):
- user_id = event.get_user_id()
- msg = str(event.message)
- repo = await Chat().deal(msg, user_id)
- try:
- await chat.finish(repo)
- except Exception:
- return
-
-
-my_name_is = Chat().on_command("叫我", "更改闲聊（文爱）时的称呼", aliases={"我是"}, priority=1)
-
-
-@my_name_is.handle([Cooldown(3, prompt=_chat_flmt_notice)])
-async def _name(matcher: Matcher, args: Message = CommandArg()):
- msg = args.extract_plain_text()
- if msg:
- matcher.set_arg("name", args)
-
-
-@my_name_is.got("name", "欧尼酱想让咱如何称呼呢！0w0")
-async def _deal_name(event: MessageEvent, new_name: str = ArgPlainText("name")):
- user_id = event.get_user_id()
- repo = choice(
- [
- f"好~w 那咱以后就称呼你为{new_name}！",
- f"噢噢噢！原来你叫{new_name}阿~",
- f"好欸！{new_name}ちゃん~~~",
- "很不错的称呼呢w",
- ]
- )
- Chat().name_is(user_id, new_name)
- await my_name_is.finish(repo)
-
-
-say = Chat().on_command("说", "别人让我说啥就说啥（", priority=1)
-
-
-@say.handle([Cooldown(3, prompt=_chat_flmt_notice)])
-async def _ready_say(matcher: Matcher, args: Message = CommandArg()):
- msg = args.extract_plain_text()
- if msg:
- matcher.set_arg("say", args)
-
-
-@say.got("say", "想要咱复读啥呢...")
-async def _deal_say(msg: str = ArgPlainText("say")):
- await say.finish(msg)
-
-
-@scheduler.scheduled_job("interval", name="闲聊词库检查更新", hours=3, misfire_grace_time=60) # type: ignore
-async def _check_kimo():
- try:
- await Chat().update_data()
- except BaseException:
- pass
diff --git a/ATRI/plugins/chat/data_source.py b/ATRI/plugins/chat/data_source.py
deleted file mode 100644
index 70c6345..0000000
--- a/ATRI/plugins/chat/data_source.py
+++ /dev/null
@@ -1,136 +0,0 @@
-import os
-import json
-from pathlib import Path
-from jieba import posseg
-from random import choice, shuffle
-
-from ATRI.service import Service
-from ATRI.rule import to_bot, is_in_service
-from ATRI.log import logger as log
-from ATRI.utils import request
-from ATRI.exceptions import ReadFileError, WriteError
-
-
-CHAT_PATH = Path(".") / "data" / "database" / "chat"
-os.makedirs(CHAT_PATH, exist_ok=True)
-KIMO_URL = "https://cdn.jsdelivr.net/gh/Kyomotoi/AnimeThesaurus/data.json"
-
-
-class Chat(Service):
- def __init__(self):
- Service.__init__(
- self, "闲聊", "好像有点涩？", rule=to_bot() & is_in_service("闲聊"), priority=5
- )
-
- @staticmethod
- async def _request(url: str) -> dict:
- res = await request.get(url)
- data = res.json()
- return data
-
- @classmethod
- async def _generate_data(cls) -> None:
- file_name = "kimo.json"
- path = CHAT_PATH / file_name
- if not path.is_file():
- log.warning("未检测到词库，生成中")
- data = await cls._request(KIMO_URL)
- try:
- with open(path, "w", encoding="utf-8") as w:
- w.write(json.dumps(data, indent=4))
- log.info("生成完成")
- except WriteError:
- raise WriteError("Writing kimo words failed!")
-
- @classmethod
- async def _load_data(cls) -> dict:
- file_name = "kimo.json"
- path = CHAT_PATH / file_name
- if not path.is_file():
- await cls._generate_data()
-
- with open(path, "r", encoding="utf-8") as r:
- data = json.loads(r.read())
- return data
-
- @classmethod
- async def update_data(cls) -> None:
- log.info("更新闲聊词库ing...")
- file_name = "kimo.json"
- path = CHAT_PATH / file_name
- if not path.is_file():
- await cls._generate_data()
-
- updata_data = await cls._request(KIMO_URL)
- data = json.loads(path.read_bytes())
- for i in updata_data:
- if i not in data:
- data[i] = updata_data[i]
-
- with open(path, "w", encoding="utf-8") as w:
- w.write(json.dumps(data, indent=4))
- log.info("闲聊词库更新完成")
-
- @staticmethod
- def name_is(user_id: str, new_name: str):
- file_name = "users.json"
- path = CHAT_PATH / file_name
- if not path.is_file():
- with open(path, "w", encoding="utf-8") as w:
- w.write(json.dumps({}))
- data = {}
-
- data = json.loads(path.read_bytes())
- data[user_id] = new_name
- try:
- with open(path, "w", encoding="utf-8") as w:
- w.write(json.dumps(data, indent=4))
- except ReadFileError:
- raise ReadFileError("Update user name failed!")
-
- @staticmethod
- def load_name(user_id: str) -> str:
- file_name = "users.json"
- path = CHAT_PATH / file_name
- if not path.is_file():
- with open(path, "w", encoding="utf-8") as w:
- w.write(json.dumps({}))
- return "你"
-
- data = json.loads(path.read_bytes())
- try:
- result = data[user_id]
- except BaseException:
- result = "你"
- return result
-
- @classmethod
- async def deal(cls, msg: str, user_id: str) -> str:
- keywords = posseg.lcut(msg)
- shuffle(keywords)
-
- data = await cls._load_data()
-
- repo = str()
- for i in keywords:
- a = i.word
- b = list(a)
- try:
- if b[0] == b[1]:
- a = b[0]
- except BaseException:
- pass
- if a in data:
- repo = data.get(a, str())
-
- if not repo:
- temp_data = list(data)
- shuffle(temp_data)
- for i in temp_data:
- if i in msg:
- repo = data.get(i, str())
-
- a = choice(repo) if type(repo) is list else repo
- user_name = cls.load_name(user_id)
- repo = a.replace("你", user_name)
- return repo
diff --git a/ATRI/plugins/kimo/__init__.py b/ATRI/plugins/kimo/__init__.py
new file mode 100644
index 0000000..b1c5698
--- /dev/null
+++ b/ATRI/plugins/kimo/__init__.py
@@ -0,0 +1,59 @@
+from random import choice
+
+from nonebot.matcher import Matcher
+from nonebot.params import ArgPlainText, CommandArg
+from nonebot.adapters.onebot.v11 import MessageEvent, Message
+from nonebot.adapters.onebot.v11.helpers import Cooldown
+
+from ATRI.utils.apscheduler import scheduler
+from .data_source import Kimo
+
+
+_chat_flmt_notice = choice(["慢...慢一..点❤", "冷静1下", "歇会歇会~~", "我开始为你以后的伴侣担心了..."])
+
+
+kimo = Kimo().on_message("文爱", "闲聊（文爱")
+
+
+@kimo.handle([Cooldown(3, prompt=_chat_flmt_notice)])
+async def _chat(event: MessageEvent):
+ user_id = event.get_user_id()
+ msg = str(event.message)
+ repo = await Kimo().deal(msg, user_id)
+ try:
+ await kimo.finish(repo)
+ except Exception:
+ return
+
+
+my_name_is = Kimo().on_command("叫我", "更改kimo时的称呼", aliases={"我是"}, priority=1)
+
+
+@my_name_is.handle([Cooldown(3, prompt=_chat_flmt_notice)])
+async def _name(matcher: Matcher, args: Message = CommandArg()):
+ msg = args.extract_plain_text()
+ if msg:
+ matcher.set_arg("name", args)
+
+
+@my_name_is.got("name", "欧尼酱想让咱如何称呼呢！0w0")
+async def _deal_name(event: MessageEvent, new_name: str = ArgPlainText("name")):
+ user_id = event.get_user_id()
+ repo = choice(
+ [
+ f"好~w 那咱以后就称呼你为{new_name}！",
+ f"噢噢噢！原来你叫{new_name}阿~",
+ f"好欸！{new_name}ちゃん~~~",
+ "很不错的称呼呢w",
+ ]
+ )
+ Kimo().name_is(user_id, new_name)
+ await my_name_is.finish(repo)
+
+
+@scheduler.scheduled_job("interval", name="kimo词库检查更新", hours=3, misfire_grace_time=60) # type: ignore
+async def _check_kimo():
+ try:
+ await Kimo().update_data()
+ except BaseException:
+ pass
diff --git a/ATRI/plugins/kimo/data_source.py b/ATRI/plugins/kimo/data_source.py
new file mode 100644
index 0000000..6dd448a
--- /dev/null
+++ b/ATRI/plugins/kimo/data_source.py
@@ -0,0 +1,136 @@
+import os
+import json
+from pathlib import Path
+from jieba import posseg
+from random import choice, shuffle
+
+from ATRI.service import Service
+from ATRI.rule import to_bot, is_in_service
+from ATRI.log import logger as log
+from ATRI.utils import request
+from ATRI.exceptions import ReadFileError, WriteError
+
+
+CHAT_PATH = Path(".") / "data" / "database" / "kimo"
+os.makedirs(CHAT_PATH, exist_ok=True)
+KIMO_URL = "https://cdn.jsdelivr.net/gh/Kyomotoi/AnimeThesaurus/data.json"
+
+
+class Kimo(Service):
+ def __init__(self):
+ Service.__init__(
+ self, "kimo", "好像有点涩？", rule=to_bot() & is_in_service("kimo"), priority=5
+ )
+
+ @staticmethod
+ async def _request(url: str) -> dict:
+ res = await request.get(url)
+ data = res.json()
+ return data
+
+ @classmethod
+ async def _generate_data(cls) -> None:
+ file_name = "kimo.json"
+ path = CHAT_PATH / file_name
+ if not path.is_file():
+ log.warning("未检测到词库，生成中")
+ data = await cls._request(KIMO_URL)
+ try:
+ with open(path, "w", encoding="utf-8") as w:
+ w.write(json.dumps(data, indent=4))
+ log.info("生成完成")
+ except WriteError:
+ raise WriteError("Writing kimo words failed!")
+
+ @classmethod
+ async def _load_data(cls) -> dict:
+ file_name = "kimo.json"
+ path = CHAT_PATH / file_name
+ if not path.is_file():
+ await cls._generate_data()
+
+ with open(path, "r", encoding="utf-8") as r:
+ data = json.loads(r.read())
+ return data
+
+ @classmethod
+ async def update_data(cls) -> None:
+ log.info("更新闲聊词库ing...")
+ file_name = "kimo.json"
+ path = CHAT_PATH / file_name
+ if not path.is_file():
+ await cls._generate_data()
+
+ updata_data = await cls._request(KIMO_URL)
+ data = json.loads(path.read_bytes())
+ for i in updata_data:
+ if i not in data:
+ data[i] = updata_data[i]
+
+ with open(path, "w", encoding="utf-8") as w:
+ w.write(json.dumps(data, indent=4))
+ log.info("kimo词库更新完成")
+
+ @staticmethod
+ def name_is(user_id: str, new_name: str):
+ file_name = "users.json"
+ path = CHAT_PATH / file_name
+ if not path.is_file():
+ with open(path, "w", encoding="utf-8") as w:
+ w.write(json.dumps({}))
+ data = {}
+
+ data = json.loads(path.read_bytes())
+ data[user_id] = new_name
+ try:
+ with open(path, "w", encoding="utf-8") as w:
+ w.write(json.dumps(data, indent=4))
+ except ReadFileError:
+ raise ReadFileError("Update user name failed!")
+
+ @staticmethod
+ def load_name(user_id: str) -> str:
+ file_name = "users.json"
+ path = CHAT_PATH / file_name
+ if not path.is_file():
+ with open(path, "w", encoding="utf-8") as w:
+ w.write(json.dumps({}))
+ return "你"
+
+ data = json.loads(path.read_bytes())
+ try:
+ result = data[user_id]
+ except BaseException:
+ result = "你"
+ return result
+
+ @classmethod
+ async def deal(cls, msg: str, user_id: str) -> str:
+ keywords = posseg.lcut(msg)
+ shuffle(keywords)
+
+ data = await cls._load_data()
+
+ repo = str()
+ for i in keywords:
+ a = i.word
+ b = list(a)
+ try:
+ if b[0] == b[1]:
+ a = b[0]
+ except BaseException:
+ pass
+ if a in data:
+ repo = data.get(a, str())
+
+ if not repo:
+ temp_data = list(data)
+ shuffle(temp_data)
+ for i in temp_data:
+ if i in msg:
+ repo = data.get(i, str())
+
+ a = choice(repo) if type(repo) is list else repo
+ user_name = cls.load_name(user_id)
+ repo = a.replace("你", user_name)
+ return repo
diff --git a/test/test_plugin_chat.py b/test/test_plugin_chat.py
deleted file mode 100644
index e04f1d3..0000000
--- a/test/test_plugin_chat.py
+++ /dev/null
@@ -1,84 +0,0 @@
-import pytest
-from nonebug import App
-from nonebot.adapters.onebot.v11 import MessageSegment
-
-from .utils import make_fake_message, make_fake_event
-
-
-@pytest.mark.asyncio
-async def test_chat(app: App):
- from ATRI.plugins.chat import chat
-
- Message = make_fake_message()
-
- async with app.test_matcher(chat) as ctx:
- bot = ctx.create_bot()
-
- msg = Message("爱你")
- event = make_fake_event(_message=msg, _to_me=True)()
-
- ctx.receive_event(bot, event)
- ctx.should_call_send(event, "是…是嘛(脸红）呐，其实咱也……", True)
-
-
-@pytest.mark.asyncio
-async def test_my_name_is(app: App):
- from ATRI.plugins.chat import my_name_is
-
- Message = make_fake_message()
-
- async with app.test_matcher(my_name_is) as ctx:
- bot = ctx.create_bot()
-
- msg = Message("叫我")
- event = make_fake_event(_message=msg, _to_me=True)()
-
- ctx.receive_event(bot, event)
- ctx.should_call_send(event, "欧尼酱想让咱如何称呼呢！0w0", True)
-
- msg = Message("欧尼酱")
- event = make_fake_event(_message=msg)()
-
- ctx.receive_event(bot, event)
- ctx.should_call_send(event, "好~w 那咱以后就称呼你为欧尼酱！", True)
-
-
-@pytest.mark.asyncio
-async def test_say(app: App):
- from ATRI.plugins.chat import say
-
- Message = make_fake_message()
-
- async with app.test_matcher(say) as ctx:
- bot = ctx.create_bot()
-
- msg = Message("说")
- event = make_fake_event(_message=msg, _to_me=True)()
-
- ctx.receive_event(bot, event)
- ctx.should_call_send(event, "想要咱复读啥呢...", True)
-
- msg = Message("nya~")
- event = make_fake_event(_message=msg)()
-
- ctx.receive_event(bot, event)
- ctx.should_call_send(event, "nya~", True)
-
- async with app.test_matcher(say) as ctx:
- bot = ctx.create_bot()
-
- msg = Message("说")
- event = make_fake_event(_message=msg, _to_me=True)()
-
- ctx.receive_event(bot, event)
- ctx.should_call_send(event, "想要咱复读啥呢...", True)
-
- msg = Message(
- MessageSegment.image(
- "https://cdn.jsdelivr.net/gh/Kyomotoi/CDN@master/noting/88674944_p0.png"
- )
- )
- event = make_fake_event(_message=msg, _to_me=True)()
-
- ctx.receive_event(bot, event)
- ctx.should_call_send(event, "不要...", True)
diff --git a/test/test_plugin_kimo.py b/test/test_plugin_kimo.py
new file mode 100644
index 0000000..ad6ff0b
--- /dev/null
+++ b/test/test_plugin_kimo.py
@@ -0,0 +1,43 @@
+import pytest
+from nonebug import App
+from nonebot.adapters.onebot.v11 import MessageSegment
+
+from .utils import make_fake_message, make_fake_event
+
+
+@pytest.mark.asyncio
+async def test_chat(app: App):
+ from ATRI.plugins.kimo import kimo
+
+ Message = make_fake_message()
+
+ async with app.test_matcher(kimo) as ctx:
+ bot = ctx.create_bot()
+
+ msg = Message("爱你")
+ event = make_fake_event(_message=msg, _to_me=True)()
+
+ ctx.receive_event(bot, event)
+ ctx.should_call_send(event, "是…是嘛(脸红）呐，其实咱也……", True)
+
+
+@pytest.mark.asyncio
+async def test_my_name_is(app: App):
+ from ATRI.plugins.kimo import my_name_is
+
+ Message = make_fake_message()
+
+ async with app.test_matcher(my_name_is) as ctx:
+ bot = ctx.create_bot()
+
+ msg = Message("叫我")
+ event = make_fake_event(_message=msg, _to_me=True)()
+
+ ctx.receive_event(bot, event)
+ ctx.should_call_send(event, "欧尼酱想让咱如何称呼呢！0w0", True)
+
+ msg = Message("欧尼酱")
+ event = make_fake_event(_message=msg)()
+
+ ctx.receive_event(bot, event)
+ ctx.should_call_send(event, "好~w 那咱以后就称呼你为欧尼酱！", True)
--
cgit v1.2.3