qsc20001102/astrbot_plugin_jx3
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加避雷功能

Browse Source
This commit is contained in:
qsc
2026-02-07 14:29:46 +08:00
parent 57857aa84e
commit e502ebd5d5
8 changed files with 475 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/bilei_data.py
+220
View File
@@ -0,0 +1,220 @@
# pyright: reportArgumentType=false
# pyright: reportAttributeAccessIssue=false
# pyright: reportIndexIssue=false
# pyright: reportOptionalMemberAccess=false
from datetime import datetime
from typing import Dict, Any, Optional, List, Union
from astrbot.api import logger
from astrbot.api import AstrBotConfig
from .sqlite import AsyncSQLiteDB
from .fun_basic import load_template
class BiLeidata:
def __init__(self,sqlite:AsyncSQLiteDB):
# 引用sqlite
self._sql_db = sqlite
def _init_return_data(self) -> Dict[str, Any]:
"""初始化标准的返回数据结构"""
return {
"code": 0,
"msg": "功能函数未执行",
"data": {}
}
# --- 业务功能函数 ---
async def add(self,name: str, text: str ,user: str) -> Dict[str, Any]:
"""避雷添加"""
return_data = self._init_return_data()
# 获取系统时间
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 添加数据
try:
await self._sql_db.insert(
"bilei",
{
"name": name,
"text": text,
"time": now,
"user": user,
}
)
except FileNotFoundError as e:
logger.error(f"添加避雷失败: {e}")
return_data["msg"] = "添加避雷失败"
return return_data
return_data["data"] = (
"避雷添加成功\n"
f"避雷名称：{name}\n"
f"避雷备注：{text}\n"
f"添加时间：{now}\n"
f"记录人：{user}\n"
)
return_data["code"] = 200
return return_data
async def all(self) -> Dict[str, Any]:
"""避雷查看"""
return_data = self._init_return_data()
# 查询数据
try:
data = await self._sql_db.select_all("bilei")
except FileNotFoundError as e:
logger.error(f"查看避雷失败: {e}")
return_data["msg"] = "查看避雷失败"
return return_data
if not data:
return_data["msg"] = "未找到避雷数据"
return return_data
# 加载模板
try:
return_data["temp"] = await load_template("bilei.html")
except FileNotFoundError as e:
logger.error(f"加载模板失败: {e}")
return_data["msg"] = "系统错误：模板文件不存在"
return return_data
# 数据处理
return_data["data"]["lists"] = data
return_data["code"] = 200
return return_data
async def select(self, name:str) -> Dict[str, Any]:
"""避雷查询 名称"""
return_data = self._init_return_data()
# 模糊拼接
like_name = f"%{name}%"
# 查询数据
try:
data = await self._sql_db.select_all(
"bilei",
"name LIKE ?",
(like_name,)
)
except FileNotFoundError as e:
logger.error(f"查询避雷失败: {e}")
return_data["msg"] = "查询避雷失败"
return return_data
if not data:
return_data["msg"] = "未查询到避雷数据"
return return_data
# 加载模板
try:
return_data["temp"] = await load_template("bilei.html")
except FileNotFoundError as e:
logger.error(f"加载模板失败: {e}")
return_data["msg"] = "系统错误：模板文件不存在"
return return_data
# 数据处理
return_data["data"]["lists"] = data
return_data["code"] = 200
return return_data
async def update(self, id:int, name: str, text: str ,user: str) -> Dict[str, Any]:
"""避雷修改 ID 名称 备注"""
return_data = self._init_return_data()
data = await self._sql_db.select_one(
"bilei",
"id=?",
(id,)
)
if not data:
return_data["msg"] = "没有当前ID"
return return_data
# 获取系统时间
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 修改数据
try:
await self._sql_db.update(
"bilei",
{
"name": name,
"text": text,
"time": now,
"user": user,
},
"id=?",
(id,)
)
except FileNotFoundError as e:
logger.error(f"避雷修改失败: {e}")
return_data["msg"] = "避雷修改失败"
return return_data
return_data["data"] = (
"避雷修改成功\n"
f"ID：{id}\n"
f"避雷名称：{name}\n"
f"避雷备注：{text}\n"
f"修改时间：{now}\n"
f"修改人：{user}\n"
)
return_data["code"] = 200
return return_data
async def delete(self, id:int) -> Dict[str, Any]:
"""避雷删除 ID"""
return_data = self._init_return_data()
data = await self._sql_db.select_one(
"bilei",
"id=?",
(id,)
)
if not data:
return_data["msg"] = "没有当前ID"
return return_data
# 删除
try:
await self._sql_db.delete(
"bilei",
"id=?",
(id,)
)
except FileNotFoundError as e:
logger.error(f"避雷删除失败: {e}")
return_data["msg"] = "避雷删除失败"
return return_data
return_data["data"] = f"避雷删除成功。ID：{id}"
return_data["code"] = 200
return return_data
core/jx3_commands.py
+30 -4
View File
@@ -22,12 +22,13 @@ from astrbot.core.utils.session_waiter import (
from .jx3_data import JX3Service
from .async_task import AsyncTask
from .bilei_data import BiLeidata
class JX3Commands(Star):
def __init__(self, jx3_data:JX3Service, at:AsyncTask, server:str):
def __init__(self, jx3_data:JX3Service, at:AsyncTask, bilei:BiLeidata, server:str):
self.jx3fun = jx3_data
self.jx3at = at
self.bilie = bilei
self.server = server
@@ -307,9 +308,34 @@ class JX3Commands(Star):
return await self.plain_msg(event, lambda: self.jx3fun.pianzhi(qq))
async def jx3_bagua(self, event: AstrMessageEvent,type: str):
async def jx3_bagua(self, event: AstrMessageEvent,name: str,text: str):
"""剑三 八卦 类型"""
return await self.plain_msg(event, lambda: self.jx3fun.bagua(type))
return await self.plain_msg(event, lambda: self.jx3fun.bagua(name))
async def bilei_add(self, event: AstrMessageEvent,name: str, text: str):
"""避雷添加 名称 备注"""
return await self.plain_msg(event, lambda: self.bilie.add(name,text,event.get_sender_name()))
async def bilei_all(self, event: AstrMessageEvent):
"""避雷查看"""
return await self.T2I_image_msg(event, self.bilie.all)
async def bilei_select(self, event: AstrMessageEvent, name:str):
"""避雷查询"""
return await self.T2I_image_msg(event, lambda: self.bilie.select(name))
async def bilei_update(self, event: AstrMessageEvent, id:int, name: str, text: str):
"""避雷修改 ID 名称 备注"""
return await self.plain_msg(event, lambda: self.bilie.update(id,name,text,event.get_sender_name()))
async def bilei_delete(self, event: AstrMessageEvent, id:int):
"""避雷删除 ID"""
return await self.plain_msg(event, lambda: self.bilie.delete(id))
async def jx3_kaifhujiank(self, event: AstrMessageEvent):
core/jx3_data.py
+6 -3
View File
@@ -11,15 +11,17 @@ from astrbot.api import logger
from astrbot.api import AstrBotConfig
from .request import APIClient
from .sqlite import AsyncSQLiteDB
from .fun_basic import load_template,gold_to_string,week_to_num,compare_date_str
class JX3Service:
def __init__(self, api_config, config:AstrBotConfig):
self._api = APIClient()
# 获取API配置文件
# 引用API配置文件
self._api_config = api_config
# 获取插件配置文件
# 引用插件配置文件
self._config = config
# 获取配置中的 Token
self.token = self._config.get("jx3api_token", "")
if self.token == "":
@@ -1483,4 +1485,5 @@ class JX3Service:
return_data["code"] = 200
return return_data
core/sqlite.py
+69
View File
@@ -0,0 +1,69 @@
import aiosqlite
from typing import Any, Dict, List, Optional, Tuple
class AsyncSQLiteDB:
def __init__(self, db_path: str = "data.db"):
self.db_path = db_path
self.conn: Optional[aiosqlite.Connection] = None
# ======================
# 生命周期
# ======================
async def connect(self):
self.conn = await aiosqlite.connect(self.db_path)
self.conn.row_factory = aiosqlite.Row
async def close(self):
if self.conn:
await self.conn.close()
# ======================
# 基础执行
# ======================
async def execute(self, sql: str, params: Tuple = ()):
async with self.conn.execute(sql, params):
await self.conn.commit()
async def fetch_one(self, sql: str, params: Tuple = ()) -> Optional[Dict[str, Any]]:
async with self.conn.execute(sql, params) as cursor:
row = await cursor.fetchone()
return dict(row) if row else None
async def fetch_all(self, sql: str, params: Tuple = ()) -> List[Dict[str, Any]]:
async with self.conn.execute(sql, params) as cursor:
rows = await cursor.fetchall()
return [dict(r) for r in rows]
# ======================
# CRUD
# ======================
async def insert(self, table: str, data: Dict[str, Any]):
keys = ", ".join(data.keys())
placeholders = ", ".join(["?"] * len(data))
sql = f"INSERT INTO {table} ({keys}) VALUES ({placeholders})"
await self.execute(sql, tuple(data.values()))
async def update(self, table: str, data: Dict[str, Any], where: str, params: Tuple):
set_clause = ", ".join([f"{k}=?" for k in data.keys()])
sql = f"UPDATE {table} SET {set_clause} WHERE {where}"
await self.execute(sql, tuple(data.values()) + params)
async def delete(self, table: str, where: str, params: Tuple):
sql = f"DELETE FROM {table} WHERE {where}"
await self.execute(sql, params)
async def select_one(self, table: str, where: str = "", params: Tuple = ()):
sql = f"SELECT * FROM {table}"
if where:
sql += f" WHERE {where}"
return await self.fetch_one(sql, params)
async def select_all(self, table: str, where: str = "", params: Tuple = ()):
sql = f"SELECT * FROM {table}"
if where:
sql += f" WHERE {where}"
return await self.fetch_all(sql, params)