1
0
forked from bot/app

first commit

This commit is contained in:
远野千束 2024-03-18 18:21:56 +08:00
parent 9e9f6e4ad6
commit 51cb1a87b8
24 changed files with 229 additions and 139 deletions

19
.gitignore vendored
View File

@ -1,16 +1,7 @@
# idea
plugin/
.venv/
.cache/
data/
# config
config.yml
# external plugins
/plugins/
# pyc/pyo
**/*.pyc
**/*.pyo
# data
/data/
```
config.example.yml

1
CNAME
View File

@ -1 +0,0 @@
bot.liteyuki.icu

View File

@ -19,11 +19,10 @@
## 手动安装和部署
1. 安装`Git`和`Python3.10+`后,使用命令`git clone https://github.com/snowykami/LiteyukiBot` 克隆项目至本地。
一定要安装GitBot自带功能需要git支持
2. 切换到轻雪目录,使用`pip install -r requirements.txt`安装依赖
3. `python main.py`启动!
1. 前置:`Git`和`Python3.10+`
2. 使用命令`git clone https://github.com/snowykami/LiteyukiBot`
3. 切换到轻雪目录,使用`pip install -r requirements.txt`安装依赖
4. `python main.py`启动!
## 一键部署脚本(复制到本地保存执行)

View File

@ -1,4 +0,0 @@
{
"name": "Liteyuki Default",
"version": "1.0"
}

View File

@ -1,4 +0,0 @@
{
"name": "Liteyuki Language Pack",
"version": "1.0"
}

19
main.py
View File

@ -1,7 +1,16 @@
from src.liteyuki import *
import nonebot
from nonebot.adapters.onebot import v11, v12
from src.utils.config import load_from_yaml
if __name__ == '__main__':
liteyuki = Liteyuki()
app = liteyuki.get_asgi()
liteyuki.run(app="main:app")
nonebot.init(**load_from_yaml("config.yml").get("nonebot", {}))
adapters = [v11.Adapter, v12.Adapter]
driver = nonebot.get_driver()
for adapter in adapters:
driver.register_adapter(adapter)
nonebot.load_plugin("src.plugins.liteyuki_plugin_main")
if __name__ == "__main__":
nonebot.run()

View File

@ -1,5 +1,5 @@
nonebot2[fastapi]
nonebot-adapter-onebot==2.4.1
pydantic==1.10.8
yaml==0.2.5
pyyaml==6.0
arclet-alconna==2.0.0a1
nonebot2[fastapi]==2.2.1
nonebot-adapter-onebot==2.4.3
nonebot-plugin-alconna==0.41.0
pydantic==2.6.4

View File

@ -1,33 +0,0 @@
import os
import yaml
from nonebot import logger
def load_config() -> dict[str, any]:
"""
Load config from config.yml
:return:
"""
config = {
'host': '0.0.0.0',
'port': 20216,
'nickname': ['Liteyuki'],
'command_start': [''],
}
if not os.path.exists('config.yml'):
logger.warning('warn.config_file_not_found')
with open('config.yml', 'w', encoding='utf-8') as f:
f.write(yaml.dump(config, indent=4))
else:
try:
with open('config.yml', 'r', encoding='utf-8') as f:
config.update(yaml.load(f, Loader=yaml.FullLoader))
logger.success('success.config_loaded')
# 格式化后写入
with open('config.yml', 'w', encoding='utf-8') as f:
f.write(yaml.dump(config, indent=4))
except Exception as e:
logger.error(f'error.load_config: {e}')
return config

View File

View File

@ -1,45 +0,0 @@
from typing import Any, Optional
import nonebot
from nonebot import DOTENV_TYPE
from nonebot.adapters.onebot.v11 import Adapter as OnebotV11Adapter
from nonebot.adapters.onebot.v12 import Adapter as OnebotV12Adapter
from src.api.utils import load_config
app = None
adapters = [
OnebotV11Adapter,
OnebotV12Adapter
]
class Liteyuki:
def __init__(self, *, _env_file: Optional[DOTENV_TYPE] = None, **kwargs: Any):
print(
'\033[34m' + r''' __ ______ ________ ________ __ __ __ __ __ __ ______
/ | / |/ |/ |/ \ / |/ | / |/ | / |/ |
$$ | $$$$$$/ $$$$$$$$/ $$$$$$$$/ $$ \ /$$/ $$ | $$ |$$ | /$$/ $$$$$$/
$$ | $$ | $$ | $$ |__ $$ \/$$/ $$ | $$ |$$ |/$$/ $$ |
$$ | $$ | $$ | $$ | $$ $$/ $$ | $$ |$$ $$< $$ |
$$ | $$ | $$ | $$$$$/ $$$$/ $$ | $$ |$$$$$ \ $$ |
$$ |_____ _$$ |_ $$ | $$ |_____ $$ | $$ \__$$ |$$ |$$ \ _$$ |_
$$ |/ $$ | $$ | $$ | $$ | $$ $$/ $$ | $$ |/ $$ |
$$$$$$$$/ $$$$$$/ $$/ $$$$$$$$/ $$/ $$$$$$/ $$/ $$/ $$$$$$/ ''' + '\033[0m'
)
kwargs = load_config()
self.nonebot = nonebot
self.nonebot.init(_env_file=_env_file, **kwargs)
self.driver = self.nonebot.get_driver()
def run(self, *args, **kwargs):
for adapter in adapters:
self.driver.register_adapter(adapter)
self.nonebot.load_plugin('src.liteyuki_main') # Load main plugin
self.nonebot.load_plugins('src/builtin') # Load builtin plugins
self.nonebot.load_plugins('plugins') # Load custom plugins
# Todo: load from database
self.nonebot.run(*args, **kwargs)
def get_asgi(self):
return self.nonebot.get_asgi()

View File

@ -1,16 +0,0 @@
from nonebot import on_command
from nonebot.adapters.onebot.v11.event import MessageEvent
from nonebot.permission import SUPERUSER
import os
folders = ['plugins']
for folder in folders:
if not os.path.exists(folder):
os.makedirs(folder)
echo = on_command('echo', permission=SUPERUSER, priority=5)
@echo.handle()
async def _(event: MessageEvent):
print(event.get_message())
await echo.finish(event.get_message())

View File

@ -0,0 +1,19 @@
import nonebot
from nonebot.plugin import PluginMetadata
__author__ = "snowykami"
__plugin_meta__ = PluginMetadata(
name="轻雪主程序",
description="轻雪主程序插件,包含了许多初始化的功能",
usage="",
homepage="https://github.com/snowykami/LiteyukiBot",
)
fastapi_app = nonebot.get_app()
@fastapi_app.get("/")
async def root():
return {
"message": "Hello LiteyukiBot!",
}

View File

View File

@ -0,0 +1,18 @@
from nonebot.plugin import PluginMetadata
from .manager import *
from .installer import *
__author__ = "snowykami"
__plugin_meta__ = PluginMetadata(
name="轻雪插件管理",
description="本地插件管理和插件商店支持,支持启用/停用,安装/卸载插件",
usage=(
"lnpm list\n"
"lnpm enable/disable <plugin_name>\n"
"lnpm search <keywords...>\n"
"lnpm install/uninstall <plugin_name>\n"
),
type="application",
homepage="https://github.com/snowykami/LiteyukiBot",
)

View File

@ -0,0 +1 @@
LNPM_COMMAND_START = "lnpm"

View File

@ -0,0 +1,16 @@
import nonebot.plugin
from nonebot import on_command
from src.utils.adapter import MessageEvent
from src.utils.language import get_user_lang
list_plugins = on_command("list-plugin", aliases={"列出插件"}, priority=0)
toggle_plugin = on_command("enable-plugin", aliases={"启用插件", "禁用插件", "disable-plugin"}, priority=0)
@list_plugins.handle()
async def _(event: MessageEvent):
lang = get_user_lang(event.user_id)
reply = lang.get("npm.current_plugins")
for plugin in nonebot.get_loaded_plugins():
reply += f"\n- {plugin.name}"
await list_plugins.finish(reply)

7
src/utils/adapter.py Normal file
View File

@ -0,0 +1,7 @@
from nonebot.adapters.onebot import v11, v12
Bot = v11.Bot | v12.Bot
GroupMessageEvent = v11.GroupMessageEvent | v12.GroupMessageEvent
PrivateMessageEvent = v11.PrivateMessageEvent | v12.PrivateMessageEvent
MessageEvent = v11.MessageEvent | v12.MessageEvent

6
src/utils/config.py Normal file
View File

@ -0,0 +1,6 @@
from yaml import load, FullLoader
def load_from_yaml(file: str) -> dict:
with open(file, 'r', encoding='utf-8') as f:
return load(f, Loader=FullLoader)

View File

@ -69,8 +69,8 @@ class BaseORMAdapter(ABC):
raise NotImplementedError
class SqliteORMAdapter(BaseORMAdapter):
"""SQLiteORM适配器严禁使用FORIEGNID和JSON作为主键前缀严禁使用$ID:作为字符串值前缀
class SqliteORMDatabase(BaseORMAdapter):
"""SQLiteORM适配器严禁使用`FORIEGNID``JSON`作为主键前缀,严禁使用`$ID:`作为字符串值前缀
Attributes:
@ -167,7 +167,7 @@ class SqliteORMAdapter(BaseORMAdapter):
value_list.append(f'{self.ID}:{value.__class__.__name__}:{self.save(value)}')
elif isinstance(value, BaseIterable):
key_list.append(f'{self.JSON}{field}')
value_list.append(self.flat(value))
value_list.append(self._flat(value))
else:
key_list.append(field)
value_list.append(value)
@ -178,7 +178,7 @@ class SqliteORMAdapter(BaseORMAdapter):
self.conn.commit()
return ids[0] if len(ids) == 1 else tuple(ids)
def flat(self, data: Iterable) -> str:
def _flat(self, data: Iterable) -> str:
"""扁平化数据,返回扁平化对象
Args:
@ -192,7 +192,7 @@ class SqliteORMAdapter(BaseORMAdapter):
if isinstance(v, LiteModel):
return_data[f'{self.FOREIGNID}{k}'] = f'{self.ID}:{v.__class__.__name__}:{self.save(v)}'
elif isinstance(v, BaseIterable):
return_data[f'{self.JSON}{k}'] = self.flat(v)
return_data[f'{self.JSON}{k}'] = self._flat(v)
else:
return_data[k] = v
@ -202,7 +202,7 @@ class SqliteORMAdapter(BaseORMAdapter):
if isinstance(v, LiteModel):
return_data.append(f'{self.ID}:{v.__class__.__name__}:{self.save(v)}')
elif isinstance(v, BaseIterable):
return_data.append(self.flat(v))
return_data.append(self._flat(v))
else:
return_data.append(v)
else:
@ -305,4 +305,4 @@ class SqliteORMAdapter(BaseORMAdapter):
new_d = d
return new_d
return load(data)
return load(data)

16
src/utils/data_manager.py Normal file
View File

@ -0,0 +1,16 @@
import os
from src.utils.data import LiteModel, SqliteORMDatabase as DB
DATA_PATH = "data/liteyuki"
user_db = DB(os.path.join(DATA_PATH, 'users.ldb'))
class UserModel(LiteModel):
id: str
username: str
lang: str
user_db.auto_migrate(UserModel)

111
src/utils/language.py Normal file
View File

@ -0,0 +1,111 @@
"""
语言模块,添加对多语言的支持
"""
import json
import os
from typing import Any
import nonebot
from src.utils.data_manager import UserModel, user_db
_language_data = {
"en": {
"name": "English",
}
}
def load_from_lang(file_path: str, lang_code: str = None):
"""
从lang文件中加载语言数据用于简单的文本键值对
Args:
file_path: lang文件路径
lang_code: 语言代码如果为None则从文件名中获取
"""
try:
if lang_code is None:
lang_code = os.path.basename(file_path).split('.')[0]
with open(file_path, 'r', encoding='utf-8') as file:
data = {}
for line in file:
line = line.strip()
if not line or line.startswith('#'): # 空行或注释
continue
key, value = line.split('=', 1)
data[key.strip()] = value.strip()
if lang_code not in _language_data:
_language_data[lang_code] = {}
_language_data[lang_code].update(data)
except Exception as e:
nonebot.logger.error(f"Failed to load language data from {file_path}: {e}")
def load_from_json(file_path: str, lang_code: str = None):
"""
从json文件中加载语言数据可以定义一些变量
Args:
lang_code: 语言代码如果为None则从文件名中获取
file_path: json文件路径
"""
try:
if lang_code is None:
lang_code = os.path.basename(file_path).split('.')[0]
with open(file_path, 'r', encoding='utf-8') as file:
data = json.load(file)
if lang_code not in _language_data:
_language_data[lang_code] = {}
_language_data[lang_code].update(data)
except Exception as e:
nonebot.logger.error(f"Failed to load language data from {file_path}: {e}")
def load_from_dict(data: dict, lang_code: str):
"""
从字典中加载语言数据
Args:
lang_code: 语言代码
data: 字典数据
"""
if lang_code not in _language_data:
_language_data[lang_code] = {}
_language_data[lang_code].update(data)
class Language:
def __init__(self, lang_code: str = "en", fallback_lang_code: str = "en"):
self.lang_code = lang_code
self.fallback_lang_code = fallback_lang_code
def get(self, item: str, *args) -> str | Any:
"""
获取当前语言文本
Args:
item: 文本键
*args: 格式化参数
Returns:
str: 当前语言的文本
"""
try:
if self.lang_code in _language_data and item in _language_data[self.lang_code]:
return _language_data[self.lang_code][item].format(*args)
if self.fallback_lang_code in _language_data and item in _language_data[self.fallback_lang_code]:
return _language_data[self.fallback_lang_code][item].format(*args)
return item
except Exception as e:
nonebot.logger.error(f"Failed to get language text or format: {e}")
return item
def get_user_lang(user_id: str) -> Language:
"""
获取用户的语言代码
"""
user = user_db.first(UserModel, "id = ?", user_id, default=UserModel(id=user_id, username="Unknown", lang="en"))
return Language(user.lang)