Compare commits

..
Author SHA1 Message Date
HibiKier 3263748234 Update README.md 2022-01-06 00:32:57 +08:00
HibiKier b8149c0308 Update auth_hook.py 2022-01-06 00:32:19 +08:00
HibiKier 75abe23896 Update __version__ 2022-01-05 23:43:19 +08:00
HibiKier ad7a3e6261 Update update_info.json 2022-01-05 23:43:05 +08:00
HibiKier f1e359c666 Update README.md 2022-01-05 23:18:12 +08:00
HibiKier 0bfd71295e Delete x 2022-01-05 23:15:54 +08:00
HibiKier 379ad77791 Add files via upload 2022-01-05 23:15:41 +08:00
HibiKier 236bc40153 Create x 2022-01-05 23:14:05 +08:00
HibiKier c2a54a8164 Update update_info.json 2022-01-05 23:07:03 +08:00
HibiKier 4dbab21853 Update README.md 2022-01-05 22:57:29 +08:00
HibiKier 3063c5bc02 Update README.md 2022-01-05 22:57:08 +08:00
1058 changed files with 41115 additions and 50049 deletions
-20
View File
@@ -1,20 +0,0 @@
.devcontainer/
.github/
.vscode/
.idea/
.pytest_cache/
.ruff_cache/
.venv/
docs_image/
k8s/
tests/
.dockerignore
.editorconfig
.gitignore
.pre-commit-config.yaml
.prettier*
.env.dev
docker-compose.yml
Dockerfile
LICENSE
*.md
+4 -65
View File
@@ -1,3 +1,5 @@
SUPERUSERS=[""]
COMMAND_START=[""]
@@ -6,72 +8,9 @@ SESSION_RUNNING_EXPRESSION="别急呀,小真寻要宕机了!QAQ"
NICKNAME=["真寻", "小真寻", "绪山真寻", "小寻子"]
SESSION_EXPIRE_TIMEOUT=00:00:30
SESSION_EXPIRE_TIMEOUT=30
ALCONNA_USE_COMMAND_START=True
# 全局图片统一使用bytes发送,当真寻与协议端不在同一服务器上时为True
IMAGE_TO_BYTES = True
# 回复消息时自称
SELF_NICKNAME="小真寻"
# 官bot appid:bot账号
QBOT_ID_DATA = '{
}'
# 数据库配置
# 示例: "postgres://user:password@127.0.0.1:5432/database"
# 示例: "mysql://user:password@127.0.0.1:3306/database"
# 示例: "sqlite:data/db/zhenxun.db" 在data目录下建立db文件夹
DB_URL = ""
# 系统代理
# SYSTEM_PROXY = "http://127.0.0.1:7890"
PLATFORM_SUPERUSERS = '
{
"qq": [""],
"dodo": [""]
}
'
DRIVER=~fastapi+~httpx+~websockets
# LOG_LEVEL=DEBUG
DEBUG=False
# 服务器和端口
HOST = 127.0.0.1
PORT = 8080
# kook adapter toekn
# kaiheila_bots =[{"token": ""}]
# # discode adapter
# DISCORD_BOTS='
# [
# {
# "token": "",
# "intent": {
# "guild_messages": true,
# "direct_messages": true
# },
# "application_commands": {"*": ["*"]}
# }
# ]
# '
# DISCORD_PROXY=''
# # dodo adapter
# DODO_BOTS='
# [
# {
# "client_id": "",
# "token": ""
# }
# ]
# '
# application_commands的{"*": ["*"]}代表将全部应用命令注册为全局应用命令
# {"admin": ["123", "456"]}则代表将admin命令注册为id是123、456服务器的局部命令,其余命令不注册
-98
View File
@@ -1,98 +0,0 @@
name: Bug 反馈
title: "Bug: "
description: 提交 Bug 反馈以帮助我们改进代码
labels: [ "bug" ]
body:
- type: checkboxes
id: checklist
attributes:
label: 提交前检查项
description: 在提交问题之前,请确认以下事项:
options:
- label: 我已搜索相关的 issue,但没有找到类似的问题
required: true
- label: 我已更新到最新版本(包括但不限于真寻本体,插件以及相关依赖),问题仍然存在
required: true
- label: 我已仔细阅读文档,确认我的配置正确
required: true
- type: dropdown
id: env-os
attributes:
label: 操作系统
description: 选择运行 zhenxun_bot 的系统
options:
- Windows
- MacOS
- Linux
- Other
validations:
required: true
- type: input
id: env-python-ver
attributes:
label: Python 版本
description: 填写运行 zhenxun_bot 的 Python 版本
placeholder: e.g. 3.11.0
validations:
required: true
- type: input
id: env-zhenxun-ver
attributes:
label: zhenxun_bot 版本
description: 填写 zhenxun_bot 版本
placeholder: e.g. 0.1.0
validations:
required: true
- type: input
id: env-adapter
attributes:
label: 适配器
description: 填写使用的适配器以及版本
placeholder: e.g. OneBot v11 2.2.2
validations:
required: true
- type: input
id: env-protocol
attributes:
label: 协议端
description: 填写连接 zhenxun_bot 的协议端及版本
placeholder: e.g. NapCat V4.0.3
validations:
required: true
- type: textarea
id: describe
attributes:
label: 描述问题
description: 清晰简洁地说明问题是什么
validations:
required: true
- type: textarea
id: reproduction
attributes:
label: 复现步骤
description: 提供能复现此问题的详细操作步骤
placeholder: |
1. 首先……
2. 然后……
3. 发生……
validations:
required: true
- type: textarea
id: expected
attributes:
label: 期望的结果
description: 清晰简洁地描述你期望发生的事情
- type: textarea
id: logs
attributes:
label: 截图或日志(请勿包含敏感信息如密码、令牌等)
description: 提供有助于诊断问题的任何日志和截图
-1
View File
@@ -1 +0,0 @@
blank_issues_enabled: false
-18
View File
@@ -1,18 +0,0 @@
name: 文档改进
title: "Docs: 描述"
description: 文档错误及改进意见反馈
labels: ["documentation"]
body:
- type: textarea
id: problem
attributes:
label: 描述问题或主题
validations:
required: true
- type: textarea
id: improve
attributes:
label: 需做出的修改
validations:
required: true
@@ -1,27 +0,0 @@
name: 功能建议
title: "Feature: 功能描述"
description: 提出关于项目新功能的想法
labels: [ "enhancement" ]
body:
- type: textarea
id: problem
attributes:
label: 希望能解决的问题
description: 在使用中遇到什么问题而需要新的功能?
validations:
required: true
- type: textarea
id: feature
attributes:
label: 描述所需要的功能
description: 请说明需要的功能或解决方法
validations:
required: true
- type: checkboxes
id: checklist
attributes:
label: 我有能力且愿意为这个功能贡献代码
options:
- label: 我有能力且愿意为这个功能贡献代码
-40
View File
@@ -1,40 +0,0 @@
name: Setup Python
description: Setup Python
inputs:
python-version:
description: Python version
required: false
default: "3.10"
env-dir:
description: Environment directory
required: false
default: "."
no-root:
description: Do not install package in the environment
required: false
default: "false"
runs:
using: "composite"
steps:
- name: Install poetry
run: pipx install poetry
shell: bash
- uses: actions/setup-python@v5
with:
python-version: ${{ inputs.python-version }}
cache: "poetry"
cache-dependency-path: |
./poetry.lock
${{ inputs.env-dir }}/poetry.lock
- run: |
cd ${{ inputs.env-dir }}
if [ "${{ inputs.no-root }}" = "true" ]; then
poetry install --all-extras --no-root
else
poetry install --all-extras
fi
shell: bash
-77
View File
@@ -1,77 +0,0 @@
template: $CHANGES
name-template: "v$RESOLVED_VERSION"
tag-template: "v$RESOLVED_VERSION"
exclude-labels:
- reverted
- no-changelog
- skip-changelog
- invalid
autolabeler:
- label: "bug"
title:
- "/:bug:.+/"
- "/🐛.+/"
- label: "enhancement"
title:
- "/:sparkles:.+/"
- "/✨.+/"
- label: "ci"
files:
- .github/**/*
- label: "breaking-change"
title:
- "/.+!:.+/"
- label: "documentation"
files:
- "*.md"
- label: "dependencies"
files:
- "pyproject.toml"
- "requirements.txt"
- "poetry.lock"
title:
- "/:wrench:.+/"
- "/🔧.+/"
- label: "resources"
files:
- resources/**/*
categories:
- title: 💥 破坏性变更
labels:
- breaking-change
- title: 🚀 新功能
labels:
- enhancement
- title: 🐛 Bug 修复
labels:
- bug
- title: 📝 文档更新
labels:
- documentation
- title: 👻 自动化程序
labels:
- chore
- internal
- maintenance
- title: 🚦 测试
labels:
- test
- tests
- title: 📦 依赖更新
labels:
- dependencies
collapse-after: 15
- title: 💫 杂项
change-template: "- $TITLE @$AUTHOR (#$NUMBER)"
change-title-escapes: '\<*_&' # You can add # and @ to disable mentions, and add ` to disable code blocks.
version-resolver:
major:
labels:
- "major"
minor:
labels:
- "minor"
patch:
labels:
- "patch"
default: patch
-84
View File
@@ -1,84 +0,0 @@
name: 检查bot是否运行正常
on:
push:
branches: ["main"]
paths:
- zhenxun/**
- tests/**
- .github/workflows/bot_check.yml
- bot.py
pull_request:
branches: ["main"]
paths:
- zhenxun/**
- tests/**
- .github/workflows/bot_check.yml
- bot.py
jobs:
bot-check:
runs-on: ubuntu-latest
name: bot check
steps:
- uses: actions/checkout@v4
- name: Setup Python
id: setup_python
uses: actions/setup-python@v5
with:
python-version: "3.10"
- name: Install Poetry
run: pip install poetry
# Poetry cache depends on OS, Python version and Poetry version.
- name: Cache Poetry cache
id: cache-poetry
uses: actions/cache@v3
with:
path: ~/.cache/pypoetry
key: poetry-cache-${{ runner.os }}-${{ steps.setup_python.outputs.python-version }}-${{ hashFiles('pyproject.toml') }}
- name: Cache playwright cache
id: cache-playwright
uses: actions/cache@v3
with:
path: ~/.cache/ms-playwright
key: playwright-cache-${{ runner.os }}-${{ steps.setup_python.outputs.python-version }}
- name: Cache Data cache
uses: actions/cache@v3
with:
path: data
key: data-cache-${{ runner.os }}-${{ steps.setup_python.outputs.python-version }}
- name: Install dependencies
if: steps.cache-poetry.outputs.cache-hit != 'true'
run: |
rm -rf poetry.lock
poetry source remove aliyun
poetry install --no-root
- name: Install playwright
if: steps.cache-playwright.outputs.cache-hit != 'true'
run: |
poetry run sudo apt-get update
poetry run sudo apt-get install -y libgstreamer-plugins-base1.0-0 libgstreamer1.0-0 gstreamer1.0-plugins-base gstreamer1.0-plugins-good gstreamer1.0-plugins-bad gstreamer1.0-libav flite x264 libx264-dev
poetry run pip install playwright
poetry run playwright install-deps
poetry run playwright install
- name: Run tests
run: poetry run pytest --cov=zhenxun --cov-report xml
- name: Check bot run
id: bot_check_run
run: |
mv scripts/bot_check.py bot_check.py
sed -i "s|^.*\?DB_URL.*|DB_URL=\"${{ env.DB_URL }}\"|g" .env.dev
sed -i "s/^.*\?LOG_LEVEL.*/LOG_LEVEL=${{ env.LOG_LEVEL }}/g" .env.dev
poetry run python3 bot_check.py
env:
DB_URL: "sqlite://:memory:"
LOG_LEVEL: DEBUG
-94
View File
@@ -1,94 +0,0 @@
# For most projects, this workflow file will not need changing; you simply need
# to commit it to your repository.
#
# You may wish to alter this file to override the set of languages analyzed,
# or to provide custom queries or build logic.
#
# ******** NOTE ********
# We have attempted to detect the languages in your repository. Please check
# the `language` matrix defined below to confirm you have the correct set of
# supported CodeQL languages.
#
name: "CodeQL Code Security Analysis"
on:
push:
branches: [ "main" ]
pull_request:
branches: [ "main" ]
schedule:
- cron: '45 21 * * 2'
jobs:
analyze:
name: Analyze (${{ matrix.language }})
# Runner size impacts CodeQL analysis time. To learn more, please see:
# - https://gh.io/recommended-hardware-resources-for-running-codeql
# - https://gh.io/supported-runners-and-hardware-resources
# - https://gh.io/using-larger-runners (GitHub.com only)
# Consider using larger runners or machines with greater resources for possible analysis time improvements.
runs-on: ${{ (matrix.language == 'swift' && 'macos-latest') || 'ubuntu-latest' }}
permissions:
# required for all workflows
security-events: write
# required to fetch internal or private CodeQL packs
packages: read
# only required for workflows in private repositories
actions: read
contents: read
strategy:
fail-fast: false
matrix:
include:
- language: python
build-mode: none
- language: javascript-typescript
build-mode: none
# CodeQL supports the following values keywords for 'language': 'c-cpp', 'csharp', 'go', 'java-kotlin', 'javascript-typescript', 'python', 'ruby', 'swift'
# Use `c-cpp` to analyze code written in C, C++ or both
# Use 'java-kotlin' to analyze code written in Java, Kotlin or both
# Use 'javascript-typescript' to analyze code written in JavaScript, TypeScript or both
# To learn more about changing the languages that are analyzed or customizing the build mode for your analysis,
# see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/customizing-your-advanced-setup-for-code-scanning.
# If you are analyzing a compiled language, you can modify the 'build-mode' for that language to customize how
# your codebase is analyzed, see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/codeql-code-scanning-for-compiled-languages
steps:
- name: Checkout repository
uses: actions/checkout@v4
# Initializes the CodeQL tools for scanning.
- name: Initialize CodeQL
uses: github/codeql-action/init@v3
with:
languages: ${{ matrix.language }}
build-mode: ${{ matrix.build-mode }}
# If you wish to specify custom queries, you can do so here or in a config file.
# By default, queries listed here will override any specified in a config file.
# Prefix the list here with "+" to use these queries and those in the config file.
# For more details on CodeQL's query packs, refer to: https://docs.github.com/en/code-security/code-scanning/automatically-scanning-your-code-for-vulnerabilities-and-errors/configuring-code-scanning#using-queries-in-ql-packs
# queries: security-extended,security-and-quality
# If the analyze step fails for one of the languages you are analyzing with
# "We were unable to automatically build your code", modify the matrix above
# to set the build mode to "manual" for that language. Then modify this step
# to build your code.
# ℹ️ Command-line programs to run using the OS shell.
# 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun
- if: matrix.build-mode == 'manual'
shell: bash
run: |
echo 'If you are using a "manual" build mode for one or more of the' \
'languages you are analyzing, replace this with the commands to build' \
'your code, for example:'
echo ' make bootstrap'
echo ' make release'
exit 1
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@v3
with:
category: "/language:${{matrix.language}}"
-11
View File
@@ -1,11 +0,0 @@
name: Sequential Lint and Type Check
on: [push, pull_request]
jobs:
ruff-call:
uses: ./.github/workflows/ruff.yml
pyright-call:
needs: ruff-call
uses: ./.github/workflows/pyright.yml
-55
View File
@@ -1,55 +0,0 @@
name: Pyright Lint
on:
workflow_call:
workflow_dispatch:
inputs:
python-version:
description: "Python version"
required: false
type: choice
options:
- "all"
- "3.10"
- "3.11"
- "3.12"
default: "all"
debug-mode:
description: "enable debug mode"
required: false
type: boolean
default: false
jobs:
pyright:
name: Pyright Lint
runs-on: ubuntu-latest
concurrency:
group: pyright-${{ github.ref }}-${{ matrix.env }}
cancel-in-progress: true
strategy:
matrix:
env: [pydantic-v1, pydantic-v2]
fail-fast: false
steps:
- uses: actions/checkout@v4
- name: Setup Python environment
uses: ./.github/actions/setup-python
with:
env-dir: ./envs/${{ matrix.env }}
no-root: true
- run: |
(cd ./envs/${{ matrix.env }} && echo "$(poetry env info --path)/bin" >> $GITHUB_PATH)
if [ "${{ matrix.env }}" = "pydantic-v1" ]; then
sed -i 's/PYDANTIC_V2 = true/PYDANTIC_V2 = false/g' ./pyproject.toml
fi
shell: bash
- name: Run Pyright Check
uses: jakebailey/pyright-action@v2
with:
pylance-version: latest-release
-18
View File
@@ -1,18 +0,0 @@
name: Release Drafter
on:
push:
branches:
- main
- dev
pull_request:
types: [opened, reopened, synchronize]
jobs:
update_release_draft:
name: Update Release Draft
runs-on: ubuntu-latest
steps:
- uses: release-drafter/release-drafter@v6
env:
GITHUB_TOKEN: ${{ secrets.GH_TOKEN }}
-20
View File
@@ -1,20 +0,0 @@
name: Ruff Lint
on:
workflow_call:
jobs:
ruff:
name: Ruff Lint
runs-on: ubuntu-latest
concurrency:
group: ruff-${{ github.ref }}
cancel-in-progress: true
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Install Ruff
uses: astral-sh/ruff-action@v3
- name: Run Ruff Check
run: ruff check
-73
View File
@@ -1,73 +0,0 @@
name: Update Version
on:
push:
paths:
- .github/workflows/update_version_pr.yml
- zhenxun/**
- resources/**
- bot.py
branches:
- main
- dev
jobs:
update-version:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
token: ${{ secrets.GH_TOKEN }}
- name: Read current version
id: read_version
run: |
version_line=$(grep '__version__' __version__)
version=$(echo $version_line | sed -E 's/__version__:\s*v([0-9]+\.[0-9]+\.[0-9]+)(-.+)?/\1/')
echo "Current version: $version"
echo "current_version=$version" >> $GITHUB_OUTPUT
- name: Check for version file changes
id: check_diff
run: |
if git diff --name-only HEAD~1 HEAD | grep -q '__version__'; then
echo "Version file has changes"
echo "version_changed=true" >> $GITHUB_OUTPUT
else
echo "Version file has no changes"
echo "version_changed=false" >> $GITHUB_OUTPUT
fi
- name: Get commit hash
id: get_commit_hash
run: echo "commit_hash=$(git rev-parse --short HEAD)" >> $GITHUB_OUTPUT
- name: Update version file
id: update_version
if: steps.check_diff.outputs.version_changed == 'false'
run: |
current_version="${{ steps.read_version.outputs.current_version }}"
commit_hash="${{ steps.get_commit_hash.outputs.commit_hash }}"
new_version="v${current_version}-${commit_hash}"
echo "new_version=$new_version" >> $GITHUB_OUTPUT
echo "Updating version to: $new_version"
echo "__version__: $new_version" > __version__
- name: Check updated version
if: steps.check_diff.outputs.version_changed == 'false'
run: cat __version__
- name: Create or update PR
if: steps.check_diff.outputs.version_changed == 'false'
uses: peter-evans/create-pull-request@v7
with:
token: ${{ secrets.GH_TOKEN }}
branch: create-pr/update_version
title: ":tada: chore(version): 自动更新版本到 ${{ steps.update_version.outputs.new_version }}"
body: "This PR updates the version file."
commit-message: ":tada: chore(version): Update version to ${{ steps.update_version.outputs.new_version }}"
add-paths: __version__
author: "AkashiCoin <i@loli.vet>"
committer: "${{ github.actor }} <${{ github.actor_id }}+${{ github.actor }}@users.noreply.github.com>"
labels: automated-update
+3 -16
View File
@@ -113,7 +113,6 @@ venv/
ENV/
env.bak/
venv.bak/
.env.dev
# Spyder project settings
.spyderproject
@@ -139,22 +138,10 @@ dmypy.json
# Cython debug symbols
cython_debug/
demo.py
test.py
server_ip.py
server.py
member_activity_handle.py
Yu-Gi-Oh/
search_image/
black_word/
csgo/
fantasy_card/
data/
log/
backup/
extensive_plugin/
test/
bot.py
.idea/
resources/
/configs/config.py
configs/config.yaml
.vscode/launch.json
plugins_/
-4
View File
@@ -1,4 +0,0 @@
MD013: false
MD024: # 重复标题
siblings_only: true
MD033: false # 允许 html
-16
View File
@@ -1,16 +0,0 @@
default_install_hook_types: [pre-commit]
ci:
autofix_commit_msg: ":rotating_light: auto fix by pre-commit hooks"
autofix_prs: true
autoupdate_branch: main
autoupdate_schedule: monthly
autoupdate_commit_msg: ":arrow_up: auto update by pre-commit hooks"
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.8.2
hooks:
- id: ruff
args: [--fix]
stages: [pre-commit]
- id: ruff-format
stages: [pre-commit]
-8
View File
@@ -1,8 +0,0 @@
{
"recommendations": [
"charliermarsh.ruff",
"esbenp.prettier-vscode",
"ms-python.python",
"ms-python.vscode-pylance"
]
}
-62
View File
@@ -1,62 +0,0 @@
{
"C_Cpp.errorSquiggles": "enabled",
"terminal.integrated.env.linux": {
"PYTHONPATH": "${workspaceFolder}${pathSeparator}${env:PYTHONPATH}"
},
"cSpell.words": [
"aiofiles",
"Alconna",
"arclet",
"Arparma",
"displayname",
"flmt",
"getbbox",
"hibiapi",
"httpx",
"jsdelivr",
"kaiheila",
"lolicon",
"Mahiro",
"nonebot",
"onebot",
"pixiv",
"qbot",
"Setu",
"tobytes",
"ujson",
"unban",
"Uninfo",
"userinfo",
"zhenxun"
],
"python.analysis.autoImportCompletions": true,
"python.testing.pytestArgs": ["tests"],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true,
"[python]": {
"editor.defaultFormatter": "charliermarsh.ruff",
"editor.wordBasedSuggestions": "allDocuments",
"editor.formatOnType": true,
"editor.formatOnSave": true,
"editor.codeActionsOnSave": {
"source.fixAll.ruff": "explicit",
"source.organizeImports": "explicit"
}
},
"ruff.format.preview": false,
"isort.check": true,
"ruff.importStrategy": "useBundled",
"ruff.organizeImports": false,
"[javascript]": {
"editor.defaultFormatter": "esbenp.prettier-vscode"
},
"[json]": {
"editor.defaultFormatter": "esbenp.prettier-vscode"
},
"[yaml]": {
"editor.defaultFormatter": "esbenp.prettier-vscode"
},
"[markdown]": {
"editor.defaultFormatter": "esbenp.prettier-vscode"
}
}
-87
View File
@@ -1,87 +0,0 @@
# zhenxun_bot 贡献者公约
## 我们的承诺
身为社区成员、贡献者和负责人,我们承诺使社区参与者不受骚扰,无论其年龄、体型、可见或不可见的缺陷、族裔、性征、性别认同和表达、经验水平、教育程度、社会与经济地位、国籍、相貌、种族、种姓、肤色、宗教信仰、性倾向或性取向如何。
我们承诺以有助于建立开放、友善、多样化、包容、健康社区的方式行事和互动。
## 我们的准则
有助于为我们的社区创造积极环境的行为例子包括但不限于:
* 表现出对他人的同情和善意
* 尊重不同的主张、观点和感受
* 提出和大方接受建设性意见
* 承担责任并向受我们错误影响的人道歉
* 注重社区共同诉求,而非个人得失
不当行为例子包括:
* 使用情色化的语言或图像,及性引诱或挑逗
* 嘲弄、侮辱或诋毁性评论,以及人身或政治攻击
* 公开或私下的骚扰行为
* 未经他人明确许可,公布他人的私人信息,如物理或电子邮件地址
* 其他有理由认定为违反职业操守的不当行为
## 责任和权力
社区负责人有责任解释和落实我们所认可的行为准则,并妥善公正地对他们认为不当、威胁、冒犯或有害的任何行为采取纠正措施。
社区负责人有权力和责任删除、编辑或拒绝或拒绝与本行为准则不相符的评论(comment)、提交(commits)、代码、维基(wiki)编辑、议题(issues)或其他贡献,并在适当时机知采取措施的理由。
## 适用范围
本行为准则适用于所有社区场合,也适用于在公共场所代表社区时的个人。
代表社区的情形包括使用官方电子邮件地址、通过官方社交媒体帐户发帖或在线上或线下活动中担任指定代表。
## 监督
辱骂、骚扰或其他不可接受的行为可通过 775757368@qq.com 向负责监督的社区负责人报告。
所有投诉都将得到及时和公平的审查和调查。
所有社区负责人都有义务尊重任何事件报告者的隐私和安全。
## 处理方针
社区负责人将遵循下列社区处理方针来明确他们所认定违反本行为准则的行为的处理方式:
### 1. 纠正
**社区影响**:使用不恰当的语言或其他在社区中被认定为不符合职业道德或不受欢迎的行为。
**处理意见**:由社区负责人发出非公开的书面警告,明确说明违规行为的性质,并解释举止如何不妥。或将要求公开道歉。
### 2. 警告
**社区影响**:单个或一系列违规行为。
**处理意见**:警告并对连续性行为进行处理。在指定时间内,不得与相关人员互动,包括主动与行为准则执行者互动。这包括避免在社区场所和外部渠道中的互动。违反这些条款可能会导致临时或永久封禁。
### 3. 临时封禁
**社区影响**: 严重违反社区准则,包括持续的不当行为。
**处理意见**: 在指定时间内,暂时禁止与社区进行任何形式的互动或公开交流。在此期间,不得与相关人员进行公开或私下互动,包括主动与行为准则执行者互动。违反这些条款可能会导致永久封禁。
### 4. 永久封禁
**社区影响**:行为模式表现出违反社区准则,包括持续的不当行为、骚扰个人或攻击或贬低某个类别的个体。
**处理意见**:永久禁止在社区内进行任何形式的公开互动。
## 参见
本行为准则改编自 [Contributor Covenant][homepage] 2.1 版, 参见 [https://www.contributor-covenant.org/version/2/1/code_of_conduct.html][v2.1]。
社区处理方针灵感来源于 [Mozilla's code of conduct enforcement ladder][Mozilla CoC]。
有关本行为准则的常见问题的答案,参见 [https://www.contributor-covenant.org/faq][FAQ]。
其他语言翻译参见 [https://www.contributor-covenant.org/translations][translations]。
[homepage]: https://www.contributor-covenant.org
[v2.1]: https://www.contributor-covenant.org/version/2/1/code_of_conduct.html
[Mozilla CoC]: https://github.com/mozilla/diversity
[FAQ]: https://www.contributor-covenant.org/faq
[translations]: https://www.contributor-covenant.org/translations
-95
View File
@@ -1,95 +0,0 @@
# zhenxun_bot 贡献指南
首先,感谢你愿意为 zhenxun_bot 贡献自己的一份力量!
本指南旨在引导你更规范地向 zhenxun_bot 提交贡献,请务必认真阅读。
## 提交 Issue
在提交 Issue 前,我们建议你先查看 [已有的 Issues](https://github.com/HibiKier/zhenxun_bot/issues),以防重复提交。
### 报告问题、故障与漏洞
如果你在使用过程中发现问题并确信是由 zhenxun_bot 引起的,欢迎提交 Issue。
请使用我们提供的 **Bug 反馈** 模板,并尽可能详细地描述:
- 问题描述
- 重现步骤
- 你的环境信息(如操作系统、依赖版本等)
### 建议功能
如果你有新的功能需求或改进建议,欢迎提出。
请使用 **功能建议** 模板,并详细描述你所需要的特性,可能的话可以提出你认为可行的解决方案。
### 文档相关
如果你觉得文档有误或缺乏更新,欢迎提出。
请使用 **文档改进** 模板,并详细描述问题或主题,希望我们做出的修改
## Pull Request
### 分支管理
请从 `main` 分支创建新功能分支,例如:
- 新功能:`feature/功能描述`
- 问题修复:`bugfix/问题描述`
### 代码风格
zhenxun_bot 使用 `pre-commit` 进行代码格式化和检查,请在提交前确保代码通过检查。
```bash
# 在安装项目依赖后安装 pre-commit 钩子
pre-commit install
```
> 未通过 `pre-commit` 检查的代码将无法合并。
### Commit 规范
请确保你的每一个 commit 都能清晰地描述其意图,一个 commit 尽量只有一个目的。
我们建议遵循 [gitmoji](https://gitmoji.dev/) 的 commit message 格式,在创建 commit 时请牢记这一点。
### 工作流程概述
`main` 分支为 zhenxun_bot 的主分支,在任何情况下都请不要直接修改 `main` 分支,而是创建一个目标分支为 `main` 的 Pull Request 来提交修改。Pull Request 标题请尽量清晰,以便维护者进行审核。
如果你不是 zhenxun_bot 团队的成员,可在 fork 本仓库后,向本仓库的 `main` 分支发起 Pull Request,注意遵循先前提到的 commit message 规范创建 commit。我们将在 code review 通过后合并你的贡献。
### 撰写文档
如果你对文档有改进建议,欢迎提交 Pull Request 或者 Issue。
[//]: # (我们使用 Markdown 编写文档,建议遵循以下规范:)
[//]: # ()
[//]: # (1. 中文与英文、数字、半角符号之间需要有空格。例:`zhenxun_bot 是一个高效的聊天机器人。`)
[//]: # (2. 若非英文整句,使用全角标点符号。例:`现在你可以看到机器人回复你:“Hello,世界!”。`)
[//]: # (3. 直引号`「」`和弯引号`“”`都可接受,但同一份文件里应使用同种引号。)
[//]: # (4. **不要使用斜体**,你不需要一种与粗体不同的强调。)
[//]: # (5. 文档中应以“我们”指代开发者,以“用户”指代机器人的使用者。)
[//]: # ()
[//]: # (如果你需要编辑器检查 Markdown 规范,可以在 VSCode 中安装 `markdownlint` 扩展。)
### 参与开发
zhenxun_bot 的代码风格遵循 [PEP 8](https://www.python.org/dev/peps/pep-0008/) 与 [PEP 484](https://www.python.org/dev/peps/pep-0484/) 规范,请确保你的代码风格和项目已有的代码保持一致,变量命名清晰,有适当的注释与测试代码。
> 暂未搭建测试框架,因此暂不要求添加测试代码。
## 项目沟通
如有关于贡献流程的疑问或需要进一步指导,请通过 [QQ群](https://jq.qq.com/?_wv=1027&k=u8PgBkMZ) 联系我们。
再次感谢你的贡献!
-67
View File
@@ -1,67 +0,0 @@
FROM python:3.11-bookworm AS requirements-stage
WORKDIR /tmp
ENV POETRY_HOME="/opt/poetry" PATH="${PATH}:/opt/poetry/bin"
RUN curl -sSL https://install.python-poetry.org | python - -y && \
poetry self add poetry-plugin-export
COPY ./pyproject.toml ./poetry.lock* /tmp/
RUN poetry export \
-f requirements.txt \
--output requirements.txt \
--without-hashes \
--without-urls
FROM python:3.11-bookworm AS build-stage
WORKDIR /wheel
COPY --from=requirements-stage /tmp/requirements.txt /wheel/requirements.txt
# RUN python3 -m pip config set global.index-url https://mirrors.aliyun.com/pypi/simple
RUN pip wheel --wheel-dir=/wheel --no-cache-dir --requirement /wheel/requirements.txt
FROM python:3.11-bookworm AS metadata-stage
WORKDIR /tmp
RUN --mount=type=bind,source=./.git/,target=/tmp/.git/ \
git describe --tags --exact-match > /tmp/VERSION 2>/dev/null \
|| git rev-parse --short HEAD > /tmp/VERSION \
&& echo "Building version: $(cat /tmp/VERSION)"
FROM python:3.11-slim-bookworm
WORKDIR /app/zhenxun
ENV TZ=Asia/Shanghai PYTHONUNBUFFERED=1
#COPY ./scripts/docker/start.sh /start.sh
#RUN chmod +x /start.sh
EXPOSE 8080
RUN apt update && \
apt install -y --no-install-recommends curl fontconfig fonts-noto-color-emoji \
&& apt clean \
&& fc-cache -fv \
&& apt-get purge -y --auto-remove curl \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖项和应用代码
COPY --from=build-stage /wheel /wheel
COPY . .
RUN pip install --no-cache-dir --no-index --find-links=/wheel -r /wheel/requirements.txt && rm -rf /wheel
RUN playwright install --with-deps chromium \
&& rm -rf /var/lib/apt/lists/* /tmp/*
COPY --from=metadata-stage /tmp/VERSION /app/VERSION
VOLUME ["/app/zhenxun/data", "/app/zhenxun/resources", "/app/zhenxun/log"]
CMD ["python", "bot.py"]
+239 -297
View File
@@ -1,134 +1,194 @@
<!-- markdownlint-disable MD033 MD041 -->
<div align=center>
<div align=center><img width="320" height="320" src="https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/logo.png"/></div>
<img width="250" height="312" src=./docs_image/tt.jpg alt="zhenxun_bot"/>
![maven](https://img.shields.io/badge/python-3.8%2B-blue)
![maven](https://img.shields.io/badge/nonebot-2.0.0-yellow)
![maven](https://img.shields.io/badge/go--cqhttp-1.0.0-red)
</div>
# 绪山真寻Bot
****
此项目基于 Nonebot2 和 go-cqhttp 开发,以 postgresql 作为数据库的QQ群娱乐机器人
## 关于
用爱发电,某些功能学习借鉴了大佬们的代码,因为绪山真寻实在太可爱了因此开发了
绪山真寻bot,实现了一些对群友的娱乐功能和实用功能(大概)。
<div align=center>
<a href="./LICENSE">
<img src="https://img.shields.io/badge/license-AGPL3.0-FE7D37" alt="license">
</a>
<a href="https://www.python.org">
<img src="https://img.shields.io/badge/Python-3.10%20%7C%203.11%20%7C%203.12-blue" alt="python">
</a>
<a href="https://nonebot.dev/">
<img src="https://img.shields.io/badge/nonebot-v2.1.3-EA5252" alt="nonebot">
</a>
<a href="https://onebot.dev/">
<img src="https://img.shields.io/badge/OneBot-v11-black?style=social&logo=data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAEAAAABABAMAAABYR2ztAAAAIVBMVEUAAAAAAAADAwMHBwceHh4UFBQNDQ0ZGRkoKCgvLy8iIiLWSdWYAAAAAXRSTlMAQObYZgAAAQVJREFUSMftlM0RgjAQhV+0ATYK6i1Xb+iMd0qgBEqgBEuwBOxU2QDKsjvojQPvkJ/ZL5sXkgWrFirK4MibYUdE3OR2nEpuKz1/q8CdNxNQgthZCXYVLjyoDQftaKuniHHWRnPh2GCUetR2/9HsMAXyUT4/3UHwtQT2AggSCGKeSAsFnxBIOuAggdh3AKTL7pDuCyABcMb0aQP7aM4AnAbc/wHwA5D2wDHTTe56gIIOUA/4YYV2e1sg713PXdZJAuncdZMAGkAukU9OAn40O849+0ornPwT93rphWF0mgAbauUrEOthlX8Zu7P5A6kZyKCJy75hhw1Mgr9RAUvX7A3csGqZegEdniCx30c3agAAAABJRU5ErkJggg==" alt="onebot">
</a>
<a href="https://onebot.dev/">
<img src="https://img.shields.io/badge/OneBot-v12-black?style=social&logo=data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAEAAAABABAMAAABYR2ztAAAAIVBMVEUAAAAAAAADAwMHBwceHh4UFBQNDQ0ZGRkoKCgvLy8iIiLWSdWYAAAAAXRSTlMAQObYZgAAAQVJREFUSMftlM0RgjAQhV+0ATYK6i1Xb+iMd0qgBEqgBEuwBOxU2QDKsjvojQPvkJ/ZL5sXkgWrFirK4MibYUdE3OR2nEpuKz1/q8CdNxNQgthZCXYVLjyoDQftaKuniHHWRnPh2GCUetR2/9HsMAXyUT4/3UHwtQT2AggSCGKeSAsFnxBIOuAggdh3AKTL7pDuCyABcMb0aQP7aM4AnAbc/wHwA5D2wDHTTe56gIIOUA/4YYV2e1sg713PXdZJAuncdZMAGkAukU9OAn40O849+0ornPwT93rphWF0mgAbauUrEOthlX8Zu7P5A6kZyKCJy75hhw1Mgr9RAUvX7A3csGqZegEdniCx30c3agAAAABJRU5ErkJggg==" alt="onebot">
</a>
<a href="https://bot.q.qq.com/wiki/">
<img src="https://img.shields.io/badge/QQ-Bot-lightgrey?style=social&logo=data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAxMTIuODIgMTMwLjg5Ij48ZyBkYXRhLW5hbWU9IuWbvuWxgiAyIj48ZyBkYXRhLW5hbWU9IuWbvuWxgiAxIj48cGF0aCBkPSJNNTUuNjMgMTMwLjhjLTcgMC0xMy45LjA4LTIwLjg2IDAtMTkuMTUtLjI1LTMxLjcxLTExLjQtMzQuMjItMzAuMy00LjA3LTMwLjY2IDE0LjkzLTU5LjIgNDQuODMtNjYuNjQgMi0uNTEgNS4yMS0uMzEgNS4yMS0xLjYzIDAtMi4xMy4xNC0yLjEzLjE0LTUuNTcgMC0uODktMS4zLTEuNDYtMi4yMi0yLjMxLTYuNzMtNi4yMy03LjY3LTEzLjQxLTEtMjAuMTggNS40LTUuNTIgMTEuODctNS40IDE3LjgtLjU5IDYuNDkgNS4yNiA2LjMxIDEzLjA4LS44NiAyMS0uNjguNzQtMS43OCAxLjYtMS43OCAyLjY3djQuMjFjMCAxLjM1IDIuMiAxLjYyIDQuNzkgMi4zNSAzMS4wOSA4LjY1IDQ4LjE3IDM0LjEzIDQ1IDY2LjM3LTEuNzYgMTguMTUtMTQuNTYgMzAuMjMtMzIuNyAzMC42My04LjAyLjE5LTE2LjA3LS4wMS0yNC4xMy0uMDF6IiBmaWxsPSIjMDI5OWZlIi8+PHBhdGggZD0iTTMxLjQ2IDExOC4zOGMtMTAuNS0uNjktMTYuOC02Ljg2LTE4LjM4LTE3LjI3LTMtMTkuNDIgMi43OC0zNS44NiAxOC40Ni00Ny44MyAxNC4xNi0xMC44IDI5Ljg3LTEyIDQ1LjM4LTMuMTkgMTcuMjUgOS44NCAyNC41OSAyNS44MSAyNCA0NS4yOS0uNDkgMTUuOS04LjQyIDIzLjE0LTI0LjM4IDIzLjUtNi41OS4xNC0xMy4xOSAwLTE5Ljc5IDAiIGZpbGw9IiNmZWZlZmUiLz48cGF0aCBkPSJNNDYuMDUgNzkuNThjLjA5IDUgLjIzIDkuODItNyA5Ljc3LTcuODItLjA2LTYuMS01LjY5LTYuMjQtMTAuMTktLjE1LTQuODItLjczLTEwIDYuNzMtOS44NHM2LjM3IDUuNTUgNi41MSAxMC4yNnoiIGZpbGw9IiMxMDlmZmUiLz48cGF0aCBkPSJNODAuMjcgNzkuMjdjLS41MyAzLjkxIDEuNzUgOS42NC01Ljg4IDEwLTcuNDcuMzctNi44MS00LjgyLTYuNjEtOS41LjItNC4zMi0xLjgzLTEwIDUuNzgtMTAuNDJzNi41OSA0Ljg5IDYuNzEgOS45MnoiIGZpbGw9IiMwODljZmUiLz48L2c+PC9nPjwvc3ZnPg==" alt="QQ">
</a>
<a href="https://github.com/psf/black">
<img src="https://img.shields.io/badge/code%20style-black-000000.svg?logo=python&logoColor=edb641" alt="black">
</a>
<a href="https://github.com/Microsoft/pyright">
<img src="https://img.shields.io/badge/types-pyright-797952.svg?logo=python&logoColor=edb641" alt="pyright">
</a>
<a href="https://github.com/astral-sh/ruff">
<img src="https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/charliermarsh/ruff/main/assets/badge/v2.json" alt="ruff">
</a>
</div>
如果该项目的图片等等侵犯猫豆腐老师权益请联系我删除!
<div align=center>
如果希望有个地方讨论绪山真寻Bot,或者有问题或建议,可以发送issues或加入[ <strong>[是真寻酱哒](https://jq.qq.com/?_wv=1027&k=u8PgBkMZ) </strong>]
[![tencent-qq](https://img.shields.io/badge/%E7%BE%A4-是真寻酱哒-red?style=logo=tencent-qq)](https://qm.qq.com/q/mRNtLSl6uc)
[![tencent-qq](https://img.shields.io/badge/%E7%BE%A4-真寻的技术群-c73e7e?style=logo=tencent-qq)](https://qm.qq.com/q/YYYt5rkMYc)
## 声明
此项目仅用于学习交流,请勿用于非法用途
</div>
## 未完成的文档
<div align=center>
# [传送门](https://hibikier.github.io/zhenxun_bot/)
[文档](https://hibikier.github.io/zhenxun_bot/)
## 真寻的帮助
请对真寻说: '真寻帮助' or '管理员帮助' or '超级用户帮助' or '真寻帮助 指令'
<br>
普通帮助图片: <br>
![x](https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/docs_image/3238573864-836268675-E2FFBB2AC143EAF4DDDF150438508721.png)
</div>
## 提供符合真寻标准的插件仓库
<div align=center>
[AkashiCoin/nonebot_plugins_zhenxun_bot](https://github.com/AkashiCoin/nonebot_plugins_zhenxun_bot)
## 绪山真寻 Bot
## 来点优点?
一.作为bot:
* 实现了许多功能,且提供了大量功能管理命令
* __..... 更多详细请通过`传送门`查看文档!__
</div>
二.作为框架?:
* 通过Config配置项将所有插件配置统计保存至config.yaml,利于统一用户修改
* 方便增删插件,原生nonebot2 matcher,不需要额外修改,仅仅通过简单的配置属性就可以生成`帮助图片`和`帮助信息`
* 提供了cd,阻塞,每日次数等限制,仅仅通过简单的属性就可以生成一个限制,例如:`__plugin_cd_limit__`
* 基本完善的权限管理
* __..... 更多详细请通过`传送门`查看文档!__
<div align=center>
“真寻是<strong>[椛椛](https://github.com/FloatTech/ZeroBot-Plugin)</strong>的好朋友!”
🎉 喜欢真寻,于是真寻就来了!🎉
本项目符合 [OneBot](https://github.com/howmanybots/onebot) 标准,可基于以下项目与机器人框架/平台进行交互
| 项目地址 | 平台 | 核心作者 | 备注 |
| :-----------------------------------------------------------: | :--: | :----------------------: | :--: |
| [LLOneBot](https://github.com/LLOneBot/LLOneBot) | NTQQ | linyuchen | 可用 |
| [Napcat](https://github.com/NapNeko/NapCatQQ) | NTQQ | NapNeko | 可用 |
| [Lagrange.Core](https://github.com/LagrangeDev/Lagrange.Core) | NTQQ | LagrangeDev/Linwenxuan04 | 可用 |
</div>
<div align=center>
<img width="100%" src="https://starify.komoridevs.icu/api/starify?owner=HibiKier&repo=zhenxun_bot" alt="starify" />
<img src="https://api.star-history.com/svg?repos=HibiKier/zhenxun_bot&type=Timeline" alt="Star Trend" width="800" />
</div>
## 🤝 帮助页面
## 功能列表
<details>
<summary>点击展开查看图片</summary>
<img width="300" height="auto" src="./docs_image/zhenxun_help.png" alt="zhenxun_help"/>
<img width="300" height="auto" src="./docs_image/html_help.png" alt="html_help"/>
<img width="300" height="auto" src="./docs_image/help.png" alt="help"/>
<summary>已实现的功能</summary>
### 已实现的常用功能
- [x] 昵称系统(群与群与私聊分开.)
- [x] 图灵AI(会把'你'等关键字替换为你的昵称),且带有 [AnimeThesaurus](https://github.com/Kyomotoi/AnimeThesaurus),够味
- [x] 签到/我的签到/好感度排行/好感度总排行(影响色图概率和开箱次数,支持配置)
- [x] 发送某文件夹下的随机图片(支持自定义,默认:美图,萝莉,壁纸)
- [x] 色图(这不是基础功能嘛喂)
- [x] coser
- [x] 黑白草图生成器
- [x] 鸡汤/语录
- [x] 骂我(钉宫语音)
- [x] 戳一戳(概率发送美图,钉宫语音或者戳回去)
- [x] 模拟开箱/我的开箱/群开箱统计/我的金色/设置cookie(csgo,内置爬虫脚本,需要提前抓取数据和图片,需要session,可能需要代理,阿里云服务器等ip也许已经被ban了(我无代理访问失败),如果访问太多账号API调用可能被禁止访问api!)
- [x] 鲁迅说过
- [x] 构造假消息(自定义的分享链接)
- [x] 商店/我的金币/购买道具/使用道具
- [x] 8种手游抽卡 (查看 [nonebot_plugin_gamedraw](https://github.com/HibiKier/nonebot_plugin_gamedraw))
- [x] 我有一个朋友想问问..(借鉴pcrbot插件)
- [x] 原神黄历
- [x] 原神今日素材
- [x] 原神资源查询 (借鉴[Genshin_Impact_bot](https://github.com/H-K-Y/Genshin_Impact_bot)插件)
- [x] 原神便笺查询
- [x] 原神玩家查询
- [x] 金币红包
- [x] 微博热搜
- [x] B站主播/UP/番剧订阅
- [x] pil对图片的一些操作
- [x] BUFF饰品底价查询(需要session)
- [x] 天气查询
- [x] 疫情查询
- [x] bt磁力搜索(咳咳,这功能我想dddd)
- [x] reimu搜索(上车) (使用[XUN_Langskip](https://github.com/Angel-Hair/XUN_Bot)的插件)
- [x] 靠图识番 (使用[XUN_Langskip](https://github.com/Angel-Hair/XUN_Bot)的插件)
- [x] 以图搜图 (使用[nonebot_plugin_picsearcher](https://github.com/synodriver/nonebot_plugin_picsearcher)插件)
- [x] 搜番
- [x] 点歌 [nonebot_plugin_songpicker2](https://github.com/maxesisn/nonebot_plugin_songpicker2)插件(删除了选歌和评论)
- [x] epic免费游戏
- [x] p站排行榜
- [x] p站搜图
- [x] 翻译(日英韩)
- [x] pix图库(一个自己的图库,含有增删查改,黑名单等命令)
- [x] 查看当前群欢迎消息
- [x] 查看该群自己的权限
- [x] 我的信息(只是为了看看什么时候入群)
- [x] 更新信息(如果继续更新的话)
- [x] go-cqhttp最新版下载和上传(不需要请删除)
- [x] 撤回
- [x] 滴滴滴-(用户对超级用户发送消息)
- [x] 金币红包/金币排行
- [x] 俄罗斯轮盘/胜场排行/败场排行/欧洲人排行/慈善家排行
- [x] 网易云热评
- [x] 念首古诗
- [x] 获取b站视频封面
- [x] 通过PID获取图片
- [x] 功能统计可视化
- [x] 关于
### 已实现的管理员功能
- [x] 更新群组成员信息
- [x] 95%的群功能开关
- [x] 查看群内被动技能状态
- [x] 自定义群欢迎消息(是真寻的不是管家的!)
- [x] .ban/.unban(支持设置ban时长)= 黑白名单
- [x] 刷屏禁言相关:刷屏检测设置/设置禁言时长/设置检测次数
- [x] 上传图片/连续上传图片 (上传图片至指定图库)
- [x] 移动图片 (同上)
- [x] 删除图片 (同上)
- [x] 群内B站订阅
- [x] 群词条
- [x] 休息吧/醒来
### 已实现的超级用户功能
- [x] 添加/删除权限(是真寻的管理员权限,不是群管理员)
- [x] 开启/关闭指定群的广播通知
- [x] 广播
- [x] 自检(检查系统状态)
- [x] 所有群组/所有好友
- [x] 退出指定群
- [x] 更新好友信息/更新群信息
- [x] /t(对用户进行回复或发送消息)
- [x] 上传/删除/修改商品(需要编写对应的商品功能)
- [x] 节日红包发送
- [x] 修改群权限
- [x] ban
- [x] 更新色图
- [x] 更新价格/更加图片(csgo开箱)
- [x] 重载原神/方舟/赛马娘/坎公骑冠剑卡池
- [x] 更新原神今日素材/更新原神资源信息
- [x] PIX相关操作
- [x] 检查更新真寻
- [x] 重启
- [x] 添加/删除/查看群白名单
- [x] 功能开关(更多设置)
- [x] 功能状态
- [x] b了
- [x] 执行sql
- [x] 重载配置
- [x] 清理临时数据
- [x] 增删群认证
- [x] 同意/拒绝好友/群聊请求
#### 超级用户的被动技能
- [x] 邀请入群提醒(别人邀请真寻入群)
- [x] 添加好友提醒(别人添加真寻好友)
### 已实现的被动技能
- [x] 进群欢迎消息
- [x] 群早晚安
- [x] 每日开箱重置提醒
- [x] b站转发解析(解析b站分享信息,支持bv,bilibili链接,b站手机端转发卡片,cv,b23.tv),且5分钟内不解析相同url
- [x] 丢人爬(爬表情包)
- [x] epic通知(每日发送epic免费游戏链接)
- [x] 原神黄历提醒
- [x] 复读
### 已实现的看不见的技能!
- [x] 刷屏禁言检测
- [x] 功能调用统计
- [x] 检测恶意触发命令(将被最高权限ban掉30分钟,只有最高权限(9级)可以进行unban)
- [x] 自动同意好友请求,加群请求将会提醒管理员,退群提示,加群欢迎等等
- [x] 群聊时间检测(当群聊最后一人发言时间大于当前36小时后将关闭该群所有通知(即被动技能))
- [x] 群管理员监控,自动为新晋管理员增加权限,为失去群管理员的用户删除权限
- [x] 群权限系统
- [x] 定时更新权限
</details>
## 📦 这是一份扩展
## 详细配置请前往文档,以下为最简部署和配置,如果你有基础并学习过nonebot2的话
### 1. 体验一下?
这是一个免费的,版本为 dev 的 zhenxun,你可以通过 [napcat](https://github.com/NapNeko/NapCatQQ) 或 [拉格朗日](https://github.com/LagrangeDev/Lagrange.Core) 以及 [matcha](https://github.com/A-kirami/matcha) 等直接连接用于体验与测试
(球球了测试君!)
## 简单部署
```text
Url: ws://test.zhenxun.org:8080/onebot/v11/ws
AccessToken: PUBLIC_ZHENXUN_TEST
注:你无法获得超级用户权限
```
### 2. 额外扩展
# 配置gocq
<div align=center>
“不要害怕,你的背后还有千千万万的 <strong>伙伴</strong> 啊!”
在 https://github.com/Mrs4s/go-cqhttp 下载Releases最新版本,运行后选择反向代理,
后将gocq的配置文件config.yml中的universal改为universal: ws://127.0.0.1:8080/cqhttp/ws
| 项目名称 | 主要用途 | 仓库作者 | 备注 |
| :--------------------------------------------------------------------: | :------: | :-------------------------------------------------: | :---------------------------------------------------: |
| [插件库](https://github.com/zhenxun-org/zhenxun_bot_plugins) | 插件 | [zhenxun-org](https://github.com/zhenxun-org) | 原 plugins 文件夹插件 |
| [插件索引库](https://github.com/zhenxun-org/zhenxun_bot_plugins_index) | 插件 | [zhenxun-org](https://github.com/zhenxun-org) | 扩展插件索引库 |
| [一键安装](https://github.com/soloxiaoye2022/zhenxun_bot-deploy) | 安装 | [soloxiaoye2022](https://github.com/soloxiaoye2022) | 第三方 |
| [WebUi](https://github.com/HibiKier/zhenxun_bot_webui) | 管理 | [hibikier](https://github.com/HibiKier) | 基于真寻 WebApi 的 webui 实现 [预览](#-webui界面展示) |
| [安卓 app(WebUi)](https://github.com/YuS1aN/zhenxun_bot_android_ui) | 安装 | [YuS1aN](https://github.com/YuS1aN) | 第三方 |
</div>
## 🥰 ~~来点优点?~~ 可爱难道还不够吗
- 实现了许多功能,且提供了大量功能管理命令,进行了多平台适配,兼容 nb2 商店插件
- 拥有完善可用的 webui
- 通过 Config 配置项将所有插件配置统计保存至 config.yaml,利于统一用户修改
- 方便增删插件,原生 nonebot2 matcher,不需要额外修改,仅仅通过简单的配置属性就可以生成`帮助图片`和`帮助信息`
- 提供了 cd,阻塞,每日次数等限制,仅仅通过简单的属性就可以生成一个限制,例如:`PluginCdBlock` 等
- **更多详细请通过 [传送门](https://hibikier.github.io/zhenxun_bot/) 查看文档!**
## 🛠️ 简单部署
```bash
# 获取代码
git clone https://github.com/HibiKier/zhenxun_bot.git
@@ -136,224 +196,106 @@ git clone https://github.com/HibiKier/zhenxun_bot.git
cd zhenxun_bot
# 安装依赖
pip install poetry # 安装 poetry
poetry install # 安装依赖
pip install -r requirements.txt
# 进行基础配置
####请查看 配置 部分####
# 开始运行
poetry run python bot.py
python bot.py
```
## 📝 简单配置
## 简单配置
> [!TIP]
> config.yaml 需要启动一次 Bot 后生成
```
1.在.env.dev文件中
1.在 .env.dev 文件中填写你的机器人配置项
SUPERUSERS = [""] # 填写你的QQ
2.在 configs/config.yaml 文件中修改你需要修改的插件配置项
2.在configs/config.py文件中
* 数据库配置
<details>
<summary>数据库地址(DB_URL)配置说明</summary>
3.在configs/config.yaml文件中 # 该文件需要启动一次后生成
* 修改插件配置项
DB_URL 是基于 Tortoise ORM 的数据库连接字符串,用于指定项目所使用的数据库。以下是 DB_URL 的组成部分以及示例:
```
格式为: `<数据库类型>://<用户名>:<密码>@<主机>:<端口>/<数据库名>?<参数>`
- 数据库类型:表示数据库类型,例如 postgres、mysql、sqlite 等。
- 用户名:数据库的用户名,例如 root。
- 密码:数据库的密码,例如 123456。
- 主机:数据库的主机地址,例如 127.0.0.1(本地)或远程服务器 IP。
- 端口:数据库的端口号,例如:PostgreSQL:5432, MySQL:3306
- 数据库名:指定要使用的数据库名称,例如 zhenxun。
- 参数(可选):用于传递额外的配置,例如字符集设置。
</details>
## 📋 功能列表
## 更新
> [!NOTE]
> 真寻原 `plugins` 插件文件夹已迁移至 [插件仓库](https://github.com/zhenxun-org/zhenxun_bot_plugins) ,现在本体仅保留核心功能
### 2021/1/5 \[v0.0.8.0]
<details>
<summary>内置功能</summary>
* 提供金币消费hook,可在plugins2settings.yaml中配置该功能需要消费的金币
* 商店插件将作为内置插件移动至basic_plugins
* 商店插件通过export提供了方法,不需要修改商店插件代码添加商品数据和生效方法
* 修改了hook插件顺序,主要以auth_hook为主
* 修改商店图片样式
* 取消每次启动更新城市列表(首次除外),采用定时更新,加快bot启动速度
* 取消每次启动时截取今日素材,采用调用时截取保存,加快bot启动速度
* 更新色图时当图片404时会删除并替换
* 疫情消息回复改为图片
* 修复商店折扣和限时时间无法生效
* 修复原神玩家查询尘歌壶缺少图片
* 修复cd限制重复导致所有含有cd插件非超级用户无法触发
### 🔧 基础功能
### 2021/12/26
- 昵称系统(群与群与私聊分开)
- 签到/我的签到/好感度排行/好感度总排行(影响色图概率和开箱次数,支持配置)
- 商店/我的金币/购买道具/使用道具/金币排行(完整的商店添加/购买/使用流程)
- 查看当前群欢迎消息
- 个人信息查看(群组内权限,聊天频率等)
- 消息撤回
- 功能统计可视化
- 关于
- 三种样式的帮助菜单
* 修复群词条问题 空格 会被录入导致不断回复
* 修复米游社app替换api导致无法正常查询
### 🛠️ 管理员功能
### 2021/12/24
- 管理员帮助
- 更新群组成员信息
- 95%的群功能开关
- 查看群内被动技能状态
- 自定义群欢迎消息(是真寻的不是管家的!)
- ban/unban(支持设置 ban 时长)= 群组及用户的黑名单
- 休息吧/醒来(群组内真寻状态)
* 支持国际疫情数据查询 [@pull/99](https://github.com/HibiKier/zhenxun_bot/pull/99)
### 🧑‍💼 超级用户功能
### 2021/12/20
- 超级用户帮助
- 添加/删除权限(是真寻的管理员权限,不是群管理员)
- 群组管理,退群指令等
- 广播
- 自检(检查系统状态)
- 所有群组/所有好友
- 退出指定群
- 更新好友信息/更新群信息
- 修改群权限
- 检查更新
- 重启
- 添加/删除/查看群白名单
- 功能开关(更多设置)
- 功能状态
- 执行 SQL
- 重载配置
- 清理临时数据
- 增删群认证
- 同意/拒绝好友/群聊请求
- 添加/移除/更新插件/插件商店(plugins 库以及扩展库)
- WebUI API(对真寻前端的支持)
* 只有发布小于存储时间的新动态/视频的时候才获取并推送 [@pull/96](https://github.com/HibiKier/zhenxun_bot/pull/96)
#### 🛡️ 超级用户的被动技能
### 2021/12/16 \[v0.0.7.0]
- 邀请入群提醒(别人邀请真寻入群,可配置自动同意)
* 提供了真寻群聊功能总开关和对应默认配置项,命令:休息吧 醒来
* 新增原神玩家查询,原神便笺查询
* 群功能管理提供全部开启/关闭命令:开启/关闭全部功能
* 提供主要数据自动备份,且提供自定义配置项
* 提供命令:关于,用于介绍Bot之类的
* 新增命令exec,用于执行sql语句
* 签到提供参数 "all",用于签到所有群聊
* Ban提醒提供cd
* 本地图库提供配置项SHOW_ID,用于设置发送图片时是否显示id
* 色图和PIX提供配置项SHOW_INFO,用于设置发送图片时是否显示图片信息
* 所有被动技能提供了进群默认状态配置项
* 修复添加权限第二种添加形式无法正确添加正确的权限
* 修复签到获取好感度卡时金币不会增加
* 修复当红包数量不合法时依旧扣除金币
* 修复金币红包再次使用塞红包时无法正确退回上次未开完的金币
* 修复 滴滴滴- 只包含图片时不会发送至管理员
* 修复添加权限等级错误
* 修复群词条以bot名称为开头时无法正确触发
* 修改了权限插件加载顺序防止小概率优先加载权限插件引起报错
* 本地图库新图库会统一建立在resource/img/image_management文件夹下,如果该文件夹内未找到图库,会从上级目录查找(即:resource/img/)
- 添加好友提醒(别人添加真寻好友,可配置自动同意)
<br>
### 🤖 被动技能
__..... 更多更新信息请查看文档__
- 群早晚安
### 👻 看不见的技能
- 功能调用统计
- 聊天记录统计
- 检测恶意触发命令(将被最高权限 ban 掉 30 分钟,只有最高权限(9 级)可以进行 unban)
- 自动同意好友/群组请求,加群请求将会提醒管理员,退群提示,加群欢迎等等
- 群聊时间检测(当群聊最后一人发言时间大于当前 48 小时后将关闭该群所有通知(即被动技能))
- 群管理员监控,自动为新晋管理员增加权限,为失去群管理员的用户删除权限
- 群权限系统
- 定时更新权限
- 自动配置重载
- 强制入群保护
- 自定备份(可配置)
- 笨蛋检测(当使用功能名称当指令时真寻会跳出来狠狠嘲笑并帮助)
</details>
## 💖 赞助
<details>
<summary>爱发电</summary>
<a href="https://afdian.com/a/HibiKier">
<img width="365px" height="450px" src=./docs_image/afd.jpg>
</a>
</details>
### 赞助名单
(可以告诉我你的 **github** 地址,我偷偷换掉 0v|)
[Zer](https://afdian.com/u/6bccdb2a60b411ec9ad452540025c377) [爱发电用户\_HTjk](https://afdian.com/u/6c7d0208064511ec8d7b52540025c377) [shenghuo2](https://afdian.com/u/bca13286102111eda2a052540025c377) [术樱](https://afdian.com/u/414da63a09a311ec8eb752540025c377) [飞火](https://afdian.com/u/404135f48ed711ec962152540025c377) [shenqi](https://afdian.net/u/fa923a8cfe3d11eba61752540025c377) [A_Kyuu](https://afdian.net/u/b83954fc2c1211eba9eb52540025c377) [疯狂混沌](https://afdian.net/u/789a2f9200cd11edb38352540025c377) [投冥](https://afdian.net/a/144514mm) [茶喵](https://afdian.net/u/fd22382eac4d11ecbfc652540025c377) [AemokpaTNR](https://afdian.net/u/1169bb8c8a9611edb0c152540025c377) [爱发电用户\_wrxn](https://afdian.net/u/4aa03d20db4311ecb1e752540025c377) [qqw](https://afdian.net/u/b71db4e2cc3e11ebb76652540025c377) [溫一壺月光下酒](https://afdian.net/u/ad667a5c650c11ed89bf52540025c377) [伝木](https://afdian.net/u/246b80683f9511edba7552540025c377) [阿奎](https://afdian.net/u/da41f72845d511ed930d52540025c377) [醉梦尘逸](https://afdian.net/u/bc11d2683cd011ed99b552540025c377) [Abc](https://afdian.net/u/870dc10a3cd311ed828852540025c377) [本喵无敌哒](https://afdian.net/u/dffaa9005bc911ebb69b52540025c377) [椎名冬羽](https://afdian.net/u/ca1ebd64395e11ed81b452540025c377) [kaito](https://afdian.net/u/a055e20a498811eab1f052540025c377) [笑柒 XIAO_Q7](https://afdian.net/u/4696db5c529111ec84ea52540025c377) [请问一份爱多少钱](https://afdian.net/u/f57ef6602dbd11ed977f52540025c377) [咸鱼鱼鱼鱼](https://afdian.net/u/8e39b9a400e011ed9f4a52540025c377) [Kafka](https://afdian.net/u/41d66798ef6911ecbc5952540025c377) [墨然](https://afdian.net/u/8aa5874a644d11eb8a6752540025c377) [爱发电用户\_T9e4](https://afdian.net/u/2ad1bb82f3a711eca22852540025c377) [笑柒 XIAO_Q7](https://afdian.net/u/4696db5c529111ec84ea52540025c377) [noahzark](https://afdian.net/a/noahzark) [腊条](https://afdian.net/u/f739c4d69eca11eba94b52540025c377) [ze roller](https://afdian.net/u/0e599e96257211ed805152540025c377) [爱发电用户\_4jrf](https://afdian.net/u/6b2cdcc817c611ed949152540025c377) [爱发电用户\_TBsd](https://afdian.net/u/db638b60217911ed9efd52540025c377) [烟寒若雨](https://afdian.net/u/067bd2161eec11eda62b52540025c377) [ln](https://afdian.net/u/b51914ba1c6611ed8a4e52540025c377) [爱发电用户\_b9S4](https://afdian.net/u/3d8f30581a2911edba6d52540025c377) [爱发电用户\_c58s](https://afdian.net/u/a6ad8dda195e11ed9a4152540025c377) [爱发电用户\_eNr9](https://afdian.net/u/05fdb41c0c9a11ed814952540025c377) [MangataAkihi](https://github.com/Sakuracio) [炀](https://afdian.net/u/69b76e9ec77b11ec874f52540025c377) [爱发电用户\_Bc6j](https://afdian.net/u/8546be24f44111eca64052540025c377) [大魔王](https://github.com/xipesoy) [CopilotLaLaLa](https://github.com/CopilotLaLaLa) [嘿小欧](https://afdian.net/u/daa4bec4f24911ec82e552540025c377) [回忆的秋千](https://afdian.net/u/e315d9c6f14f11ecbeef52540025c377) [十年くん](https://github.com/shinianj) [哇](https://afdian.net/u/9b266244f23911eca19052540025c377) [yajiwa](https://github.com/yajiwa) [爆金币](https://afdian.net/u/0d78879ef23711ecb22452540025c377)...
## 📜 贡献指南
欢迎查看我们的 [贡献指南](CONTRIBUTING.md) 和 [行为守则](CODE_OF_CONDUCT.md) 以了解如何参与贡献。
## ❔ 需要帮助?
> [!TIP]
> 发起 [issue](https://github.com/HibiKier/zhenxun_bot/issues/new/choose) 前,我们希望你能够阅读过或者了解 [提问的智慧](https://github.com/ryanhanwu/How-To-Ask-Questions-The-Smart-Way/blob/main/README-zh_CN.md)
>
> - 善用[搜索引擎](https://www.google.com/)
> - 查阅 issue 中是否有类似问题,如果没有请按照模板发起 issue
欢迎前往 [issue](https://github.com/HibiKier/zhenxun_bot/issues/new/choose) 中提出你遇到的问题,或者加入我们的 [用户群](https://qm.qq.com/q/mRNtLSl6uc) 或 [技术群](https://qm.qq.com/q/YYYt5rkMYc)与我们联系
## 🛠️ 进度追踪
Project [zhenxun_bot](https://github.com/users/HibiKier/projects/2)
## 🌟 特别感谢
首席设计师:[酥酥/coldly-ss](https://github.com/coldly-ss)
## 🙏 感谢
## Todo
- [ ] docker容器
- [ ] web管理
## 感谢
[botuniverse / onebot](https://github.com/botuniverse/onebot) :超棒的机器人协议
[Mrs4s / go-cqhttp](https://github.com/Mrs4s/go-cqhttp) :cqhttp 的 golang 实现,轻量、原生跨平台.
[nonebot / nonebot2](https://github.com/nonebot/nonebot2) :跨平台 Python 异步机器人框架
[Angel-Hair / XUN_Bot](https://github.com/Angel-Hair/XUN_Bot) :一个基于 NoneBot 和酷 Q 的功能性 QQ 机器人
[pcrbot / cappuccilo_plugins](https://github.com/pcrbot/cappuccilo_plugins) :hoshino 插件合集
[Mrs4s / go-cqhttp](https://github.com/Mrs4s/go-cqhttp) :cqhttp的golang实现,轻量、原生跨平台.
[nonebot / nonebot2](https://github.com/nonebot/nonebot2) :跨平台Python异步机器人框架
[Angel-Hair / XUN_Bot](https://github.com/Angel-Hair/XUN_Bot) :一个基于NoneBot和酷Q的功能性QQ机器人
[pcrbot / cappuccilo_plugins](https://github.com/pcrbot/cappuccilo_plugins) :hoshino插件合集
[MeetWq /nonebot-plugin-withdraw](https://github.com/MeetWq/nonebot-plugin-withdraw) :A simple withdraw plugin for Nonebot2
[maxesisn / nonebot_plugin_songpicker2](https://github.com/maxesisn/nonebot_plugin_songpicker2) :适用于 nonebot2 的点歌插件
[maxesisn / nonebot_plugin_songpicker2](https://github.com/maxesisn/nonebot_plugin_songpicker2) :适用于nonebot2的点歌插件
[nonepkg / nonebot-plugin-manager](https://github.com/nonepkg/nonebot-plugin-manager) :Nonebot Plugin Manager base on import hook
[H-K-Y / Genshin_Impact_bot](https://github.com/H-K-Y/Genshin_Impact_bot) :原神 bot,这是一个基于 nonebot 和 HoshinoBot 的原神娱乐及信息查询插件
[NothAmor / nonebot2_luxun_says](https://github.com/NothAmor/nonebot2_luxun_says) :基于 nonebot2 机器人框架的鲁迅说插件
[Kyomotoi / AnimeThesaurus](https://github.com/Kyomotoi/AnimeThesaurus) :一个~~特二刺螈~~(文爱)的适用于任何 bot 的词库
[Ailitonia / omega-miya](https://github.com/Ailitonia/omega-miya) :基于 nonebot2 的 qq 机器人
[KimigaiiWuyi / GenshinUID](https://github.com/KimigaiiWuyi/GenshinUID) :一个基于 HoshinoBot/NoneBot2 的原神 UID 查询插件
## 📊 统计与活跃贡献者
<a href="https://next.ossinsight.io/widgets/official/compose-last-28-days-stats?repo_id=368008334" target="_blank" style="display: block" align="center">
<picture>
<source media="(prefers-color-scheme: dark)" srcset="https://next.ossinsight.io/widgets/official/compose-last-28-days-stats/thumbnail.png?repo_id=368008334&image_size=auto&color_scheme=dark" width="800" height="auto">
<img alt="Performance Stats of HibiKier/zhenxun_bot - Last 28 days" src="https://next.ossinsight.io/widgets/official/compose-last-28-days-stats/thumbnail.png?repo_id=368008334&image_size=auto&color_scheme=light" width="800" height="auto">
</picture>
</a>
<a href="https://next.ossinsight.io/widgets/official/compose-recent-active-contributors?repo_id=368008334&limit=30" target="_blank" style="display: block" align="center">
<picture>
<source media="(prefers-color-scheme: dark)" srcset="https://next.ossinsight.io/widgets/official/compose-recent-active-contributors/thumbnail.png?repo_id=368008334&limit=30&image_size=auto&color_scheme=dark" width="800" height="auto">
<img alt="Active Contributors of HibiKier/zhenxun_bot - Last 28 days" src="https://next.ossinsight.io/widgets/official/compose-recent-active-contributors/thumbnail.png?repo_id=368008334&limit=30&image_size=auto&color_scheme=light" width="800" height="auto">
</picture>
</a>
## 👨‍💻 开发者
感谢以下开发者对 绪山真寻 Bot 作出的贡献:
<a href="https://github.com/HibiKier/zhenxun_bot/graphs/contributors" style="display: block" align="center">
<img src="https://contrib.rocks/image?repo=HibiKier/zhenxun_bot&max=1000" alt="contributors"/>
</a>
## 📸 WebUI 界面展示
<div style="display: flex; flex-wrap: wrap; justify-content: space-between;">
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui00.png" alt="webui00" style="width: 100%; height: auto;">
</div>
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui01.png" alt="webui01" style="width: 100%; height: auto;">
</div>
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui02.png" alt="webui02" style="width: 100%; height: auto;">
</div>
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui03.png" alt="webui03" style="width: 100%; height: auto;">
</div>
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui04.png" alt="webui04" style="width: 100%; height: auto;">
</div>
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui05.png" alt="webui05" style="width: 100%; height: auto;">
</div>
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui06.png" alt="webui06" style="width: 100%; height: auto;">
</div>
<div style="width: 48%; margin-bottom: 10px;">
<img src="./docs_image/webui07.png" alt="webui07" style="width: 100%; height: auto;">
</div>
</div>
[H-K-Y / Genshin_Impact_bot](https://github.com/H-K-Y/Genshin_Impact_bot) :原神bot,这是一个基于nonebot和HoshinoBot的原神娱乐及信息查询插件
[NothAmor / nonebot2_luxun_says](https://github.com/NothAmor/nonebot2_luxun_says) :基于nonebot2机器人框架的鲁迅说插件
[Kyomotoi / AnimeThesaurus](https://github.com/Kyomotoi/AnimeThesaurus) :一个~~特二刺螈~~(文爱)的适用于任何bot的词库
[Ailitonia / omega-miya](https://github.com/Ailitonia/omega-miya) :基于nonebot2的qq机器人
[KimigaiiWuyi / GenshinUID]("https://github.com/KimigaiiWuyi/GenshinUID") :一个基于HoshinoBot/NoneBot2的原神UID查询插件
+1 -1
View File
@@ -1 +1 @@
__version__: v0.2.4-2c97eea
__version__: v0.0.8.1
View File
+30
View File
@@ -0,0 +1,30 @@
from configs.config import Config
import nonebot
Config.add_plugin_config(
"admin_bot_manage:custom_welcome_message",
"SET_GROUP_WELCOME_MESSAGE_LEVEL [LEVEL]",
2,
name="群管理员操作",
help_="设置群欢迎消息权限",
default_value=2,
)
Config.add_plugin_config(
"admin_bot_manage:switch_rule",
"CHANGE_GROUP_SWITCH_LEVEL [LEVEL]",
2,
help_="开关群功能权限",
default_value=2,
)
Config.add_plugin_config(
"admin_bot_manage",
"ADMIN_DEFAULT_AUTH",
5,
help_="默认群管理员权限",
default_value=5
)
nonebot.load_plugins("basic_plugins/admin_bot_manage")
+38
View File
@@ -0,0 +1,38 @@
from nonebot import on_notice
from services.log import logger
from nonebot.adapters.cqhttp import Bot, GroupAdminNoticeEvent
from nonebot.typing import T_State
from models.level_user import LevelUser
from models.group_member_info import GroupInfoUser
from configs.config import Config
__zx_plugin_name__ = "群管理员变动监测 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
admin_notice = on_notice(priority=5)
@admin_notice.handle()
async def _(bot: Bot, event: GroupAdminNoticeEvent, state: T_State):
try:
nickname = (
await GroupInfoUser.get_member_info(event.user_id, event.group_id)
).user_name
except AttributeError:
nickname = event.user_id
if event.sub_type == "set":
await LevelUser.set_level(
event.user_id,
event.group_id,
Config.get_config("admin_bot_manage", "ADMIN_DEFAULT_AUTH"),
)
logger.info(
f"为新晋管理员 {nickname}({event.user_id}) "
f"添加权限等级:{Config.get_config('admin_bot_manage', 'ADMIN_DEFAULT_AUTH')}"
)
elif event.sub_type == "unset":
await LevelUser.delete_level(event.user_id, event.group_id)
logger.info(f"将非管理员 {nickname}({event.user_id}) 取消权限等级")
+51
View File
@@ -0,0 +1,51 @@
from nonebot import on_command
from utils.utils import get_message_text, get_message_imgs
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent
from .data_source import custom_group_welcome
from nonebot.adapters.cqhttp.permission import GROUP
from configs.config import Config
from services.log import logger
__zx_plugin_name__ = "自定义进群欢迎消息 [Admin]"
__plugin_usage__ = """
usage:
指令:
自定义进群欢迎消息 ?[文本] ?[图片]
示例:自定义进群欢迎消息 欢迎新人![图片]
Note:可以通过[at]来确认是否艾特新成员
示例:自定义进群欢迎消息 欢迎你[at]
""".strip()
__plugin_des__ = '简易的自定义群欢迎消息'
__plugin_cmd__ = ['自定义群欢迎消息 ?[文本] ?[图片]']
__plugin_version__ = 0.1
__plugin_author__ = 'HibiKier'
__plugin_settings__ = {
"admin_level": Config.get_config("admin_bot_manage", "SET_GROUP_WELCOME_MESSAGE_LEVEL"),
}
custom_welcome = on_command(
"自定义进群欢迎消息",
aliases={"自定义欢迎消息", "自定义群欢迎消息", "设置群欢迎消息"},
permission=GROUP,
priority=5,
block=True,
)
@custom_welcome.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
try:
msg = get_message_text(event.json())
imgs = get_message_imgs(event.json())
if not msg and not imgs:
await custom_welcome.finish(__plugin_usage__)
await custom_welcome.send(
await custom_group_welcome(msg, imgs, event.user_id, event.group_id),
at_sender=True,
)
logger.info(f"USER {event.user_id} GROUP {event.group_id} 自定义群欢迎消息:{msg}")
except Exception as e:
logger.error(f"自定义进群欢迎消息发生错误 {type(e)}:{e}")
await custom_welcome.send("发生了一些未知错误...")
+319
View File
@@ -0,0 +1,319 @@
from typing import List
from nonebot.adapters.cqhttp.message import MessageSegment
from services.log import logger
from configs.path_config import DATA_PATH
from utils.message_builder import image
from utils.utils import get_local_proxy, get_bot
from pathlib import Path
from models.group_member_info import GroupInfoUser
from datetime import datetime
from services.db_context import db
from models.level_user import LevelUser
from configs.config import Config
from utils.manager import group_manager, plugins2settings_manager, plugins_manager
from utils.image_utils import BuildImage
from utils.http_utils import AsyncHttpx
import asyncio
import time
import os
try:
import ujson as json
except ModuleNotFoundError:
import json
async def group_current_status(group_id: int) -> str:
"""
获取当前所有通知的开关
:param group_id: 群号
"""
rst = "[被动技能 状态]\n"
_data = group_manager.get_task_data()
for task in _data.keys():
rst += f'{_data[task]}: {"√" if await group_manager.check_group_task_status(group_id, task) else "×"}\n'
return rst.strip()
custom_welcome_msg_json = (
Path() / "data" / "custom_welcome_msg" / "custom_welcome_msg.json"
)
async def custom_group_welcome(
msg: str, imgs: List[str], user_id: int, group_id: int
) -> str:
"""
替换群欢迎消息
:param msg: 欢迎消息文本
:param imgs: 欢迎消息图片,只取第一张
:param user_id: 用户id,用于log记录
:param group_id: 群号
"""
img_result = ""
img = imgs[0] if imgs else ""
result = ""
if os.path.exists(DATA_PATH + f"custom_welcome_msg/{group_id}.jpg"):
os.remove(DATA_PATH + f"custom_welcome_msg/{group_id}.jpg")
if not custom_welcome_msg_json.exists():
custom_welcome_msg_json.parent.mkdir(parents=True, exist_ok=True)
data = {}
else:
try:
data = json.load(open(custom_welcome_msg_json, "r"))
except FileNotFoundError:
data = {}
try:
if msg:
data[str(group_id)] = str(msg)
json.dump(
data, open(custom_welcome_msg_json, "w"), indent=4, ensure_ascii=False
)
logger.info(f"USER {user_id} GROUP {group_id} 更换群欢迎消息 {msg}")
result += msg
if img:
await AsyncHttpx.download_file(
img, DATA_PATH + f"custom_welcome_msg/{group_id}.jpg"
)
img_result = image(abspath=DATA_PATH + f"custom_welcome_msg/{group_id}.jpg")
logger.info(f"USER {user_id} GROUP {group_id} 更换群欢迎消息图片")
except Exception as e:
logger.error(f"GROUP {group_id} 替换群消息失败 e:{e}")
return "替换群消息失败.."
return f"替换群欢迎消息成功:\n{result}" + img_result
task_data = None
async def change_group_switch(cmd: str, group_id: int, is_super: bool = False):
global task_data
"""
修改群功能状态
:param cmd: 功能名称
:param group_id: 群号
:param is_super: 是否位超级用户,超级用户用于私聊开关功能状态
"""
if not task_data:
task_data = group_manager.get_task_data()
group_help_file = Path(DATA_PATH) / "group_help" / f"{group_id}.png"
status = cmd[:2]
cmd = cmd[2:]
type_ = "plugin"
modules = plugins2settings_manager.get_plugin_module(cmd, True)
if cmd == "全部被动":
for task in task_data:
if status == "开启":
if not await group_manager.check_group_task_status(group_id, task):
await group_manager.open_group_task(group_id, task)
else:
if await group_manager.check_group_task_status(group_id, task):
await group_manager.close_group_task(group_id, task)
if group_help_file.exists():
group_help_file.unlink()
return f"已 {status} 全部被动技能!"
if cmd == "全部功能":
for f in plugins2settings_manager.get_data():
if status == "开启":
group_manager.unblock_plugin(f, group_id)
else:
group_manager.block_plugin(f, group_id)
return f"已 {status} 全部功能!"
if cmd in [task_data[x] for x in task_data.keys()]:
type_ = "task"
modules = [x for x in task_data.keys() if task_data[x] == cmd]
for module in modules:
if is_super:
module = f"{module}:super"
if status == "开启":
if type_ == "task":
if await group_manager.check_group_task_status(group_id, module):
return f"被动 {task_data[module]} 正处于开启状态!不要重复开启."
await group_manager.open_group_task(group_id, module)
else:
if group_manager.get_plugin_status(module, group_id):
return f"功能 {cmd} 正处于开启状态!不要重复开启."
group_manager.unblock_plugin(module, group_id)
else:
if type_ == "task":
if not await group_manager.check_group_task_status(group_id, module):
return f"被动 {task_data[module]} 正处于关闭状态!不要重复关闭."
await group_manager.close_group_task(group_id, module)
else:
if not group_manager.get_plugin_status(module, group_id):
return f"功能 {cmd} 正处于关闭状态!不要重复关闭."
group_manager.block_plugin(module, group_id)
if group_help_file.exists():
group_help_file.unlink()
if is_super:
for file in os.listdir(Path(DATA_PATH) / "group_help"):
file = Path(DATA_PATH) / "group_help" / file
file.unlink()
else:
_help_image = Path(DATA_PATH) / "group_help" / f"{group_id}.png"
if _help_image.exists():
_help_image.unlink()
return f"{status} {cmd} 功能!"
def set_plugin_status(cmd: str, block_type: str = "all"):
"""
设置插件功能状态(超级用户使用)
:param cmd: 功能名称
:param block_type: 限制类型, 'all': 私聊+群里, 'private': 私聊, 'group': 群聊
"""
status = cmd[:2]
cmd = cmd[2:]
module = plugins2settings_manager.get_plugin_module(cmd)
if status == "开启":
plugins_manager.unblock_plugin(module)
else:
plugins_manager.block_plugin(module, block_type=block_type)
for file in os.listdir(Path(DATA_PATH) / "group_help"):
file = Path(DATA_PATH) / "group_help" / file
file.unlink()
async def get_plugin_status():
"""
获取功能状态
"""
return await asyncio.get_event_loop().run_in_executor(None, _get_plugin_status)
def _get_plugin_status() -> MessageSegment:
"""
合成功能状态图片
"""
rst = "\t功能\n"
flag_str = "状态".rjust(4) + "\n"
for module in plugins_manager.get_data():
flag = plugins_manager.get_plugin_block_type(module)
flag = flag.upper() + " CLOSE" if flag else "OPEN"
try:
plugin_name = plugins_manager.get(module)["plugin_name"]
if (
"[Hidden]" in plugin_name
or "[Admin]" in plugin_name
or "[Superuser]" in plugin_name
):
continue
rst += f"{plugin_name}"
except KeyError:
rst += f"{module}"
if plugins_manager.get(module)["error"]:
rst += "[ERROR]"
rst += "\n"
flag_str += f"{flag}\n"
height = len(rst.split("\n")) * 24
a = BuildImage(250, height, font_size=20)
a.text((10, 10), rst)
b = BuildImage(200, height, font_size=20)
b.text((10, 10), flag_str)
A = BuildImage(500, height)
A.paste(a)
A.paste(b, (270, 0))
return image(b64=A.pic2bs4())
async def update_member_info(group_id: int, remind_superuser: bool = False) -> bool:
"""
更新群成员信息
:param group_id: 群号
:param remind_superuser: 失败信息提醒超级用户
"""
bot = get_bot()
_group_user_list = await bot.get_group_member_list(group_id=group_id)
_error_member_list = []
_exist_member_list = []
# try:
for user_info in _group_user_list:
if user_info["card"] == "":
nickname = user_info["nickname"]
else:
nickname = user_info["card"]
async with db.transaction():
# 更新权限
if (
user_info["role"]
in [
"owner",
"admin",
]
and not await LevelUser.is_group_flag(user_info["user_id"], group_id)
):
await LevelUser.set_level(
user_info["user_id"],
user_info["group_id"],
Config.get_config("admin_bot_manage", "ADMIN_DEFAULT_AUTH"),
)
if str(user_info["user_id"]) in bot.config.superusers:
await LevelUser.set_level(
user_info["user_id"], user_info["group_id"], 9
)
user = await GroupInfoUser.get_member_info(
user_info["user_id"], user_info["group_id"]
)
if user:
if user.user_name != nickname:
await user.update(user_name=nickname).apply()
logger.info(
f"用户{user_info['user_id']} 所属{user_info['group_id']} 更新群昵称成功"
)
_exist_member_list.append(int(user_info["user_id"]))
continue
join_time = datetime.strptime(
time.strftime(
"%Y-%m-%d %H:%M:%S", time.localtime(user_info["join_time"])
),
"%Y-%m-%d %H:%M:%S",
)
if await GroupInfoUser.add_member_info(
user_info["user_id"],
user_info["group_id"],
nickname,
join_time,
):
_exist_member_list.append(int(user_info["user_id"]))
logger.info(f"用户{user_info['user_id']} 所属{user_info['group_id']} 更新成功")
else:
_error_member_list.append(
f"用户{user_info['user_id']} 所属{user_info['group_id']} 更新失败\n"
)
_del_member_list = list(
set(_exist_member_list).difference(
set(await GroupInfoUser.get_group_member_id_list(group_id))
)
)
if _del_member_list:
for del_user in _del_member_list:
if await GroupInfoUser.delete_member_info(del_user, group_id):
logger.info(f"退群用户{del_user} 所属{group_id} 已删除")
else:
logger.info(f"退群用户{del_user} 所属{group_id} 删除失败")
if _error_member_list and remind_superuser:
result = ""
for error_user in _error_member_list:
result += error_user
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]), message=result[:-1]
)
return True
def set_group_bot_status(group_id: int, status: bool) -> str:
"""
设置群聊bot开关状态
:param group_id: 群号
:param status: 状态
"""
if status:
if group_manager.check_group_bot_status(group_id):
return "我还醒着呢!"
group_manager.turn_on_group_bot_status(group_id)
return "呜..醒来了..."
else:
group_manager.shutdown_group_bot_status(group_id)
# for x in group_manager.get_task_data():
# group_manager.close_group_task(group_id, x)
return "那我先睡觉了..."
+42
View File
@@ -0,0 +1,42 @@
from nonebot.adapters.cqhttp import Bot, Event
from nonebot.typing import T_State
from utils.manager import group_manager, plugins2settings_manager
from utils.utils import get_message_text
from services.log import logger
cmd = []
def switch_rule(bot: Bot, event: Event, state: T_State) -> bool:
"""
检测文本是否是关闭功能命令
:param bot: pass
:param event: pass
:param state: pass
"""
global cmd
try:
if not cmd:
cmd = ["关闭全部被动", "开启全部被动", "开启全部功能", "关闭全部功能"]
_data = group_manager.get_task_data()
for key in _data:
cmd.append(f"开启{_data[key]}")
cmd.append(f"关闭{_data[key]}")
cmd.append(f"开启 {_data[key]}")
cmd.append(f"关闭 {_data[key]}")
_data = plugins2settings_manager.get_data()
for key in _data:
try:
for x in _data[key]["cmd"]:
cmd.append(f"开启{x}")
cmd.append(f"关闭{x}")
cmd.append(f"开启 {x}")
cmd.append(f"关闭 {x}")
except KeyError:
pass
msg = get_message_text(event.json()).split()
msg = msg[0] if msg else ""
return msg in cmd
except Exception as e:
logger.error(f"检测是否为功能开关命令发生错误 {type(e)}: {e}")
return False
+125
View File
@@ -0,0 +1,125 @@
from nonebot import on_command, on_message, on_regex
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent, MessageEvent, GROUP
from .data_source import (
change_group_switch,
set_plugin_status,
get_plugin_status,
group_current_status,
set_group_bot_status
)
from services.log import logger
from configs.config import NICKNAME, Config
from utils.utils import get_message_text, is_number
from nonebot.permission import SUPERUSER
from .rule import switch_rule
import re
__zx_plugin_name__ = "群功能开关 [Admin]"
__plugin_usage__ = """
usage:
群内功能与被动技能开关
指令:
开启/关闭[功能]
群被动状态
开启全部被动
关闭全部被动
醒来/休息吧
示例:开启/关闭色图
""".strip()
__plugin_superuser_usage__ = """
usage:
功能总开关与指定群禁用
指令:
功能状态
开启/关闭[功能] [group]
开启/关闭[功能] ['private'/'group']
""".strip()
__plugin_des__ = "群内功能开关"
__plugin_cmd__ = [
"开启/关闭[功能]",
"群被动状态",
"开启全部被动",
"关闭全部被动",
"醒来/休息吧",
"功能状态 [_superuser]",
"开启/关闭[功能] [group] [_superuser]",
"开启/关闭[功能] ['private'/'group'] [_superuser]",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"admin_level": Config.get_config("admin_bot_manage", "CHANGE_GROUP_SWITCH_LEVEL"),
"cmd": ["开启功能", "关闭功能", "开关"]
}
switch_rule_matcher = on_message(rule=switch_rule, priority=4, block=True)
plugins_status = on_command("功能状态", permission=SUPERUSER, priority=5, block=True)
group_task_status = on_command("群被动状态", permission=GROUP, priority=5, block=True)
group_status = on_regex("^(休息吧|醒来)$", permission=GROUP, priority=5, block=True)
@switch_rule_matcher.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
_cmd = get_message_text(event.json()).split()[0]
if isinstance(event, GroupMessageEvent):
await switch_rule_matcher.send(await change_group_switch(_cmd, event.group_id))
logger.info(f"USER {event.user_id} GROUP {event.group_id} 使用群功能管理命令 {_cmd}")
else:
if str(event.user_id) in bot.config.superusers:
block_type = " ".join(get_message_text(event.json()).split()[1:])
block_type = block_type if block_type else "a"
if is_number(block_type):
if not int(block_type) in [
g["group_id"]
for g in await bot.get_group_list(self_id=int(bot.self_id))
]:
await switch_rule_matcher.finish(f"{NICKNAME}未加入群聊:{block_type}")
await change_group_switch(_cmd, int(block_type), True)
group_name = (await bot.get_group_info(group_id=int(block_type)))[
"group_name"
]
await switch_rule_matcher.send(
f"已禁用群聊 {group_name}({block_type}) 的 {_cmd[2:]} 功能"
)
elif block_type in ["all", "private", "group", "a", "p", "g"]:
block_type = "all" if block_type == "a" else block_type
block_type = "private" if block_type == "p" else block_type
block_type = "group" if block_type == "g" else block_type
set_plugin_status(_cmd, block_type)
if block_type == "all":
await switch_rule_matcher.send(f"已{_cmd[:2]}功能:{_cmd[2:]}")
elif block_type == "private":
await switch_rule_matcher.send(f"已在私聊中{_cmd[:2]}功能:{_cmd[2:]}")
else:
await switch_rule_matcher.send(f"已在群聊中{_cmd[:2]}功能:{_cmd[2:]}")
else:
await switch_rule_matcher.finish("格式错误:关闭[功能] [group]/[p/g]")
logger.info(f"USER {event.user_id} 使用功能管理命令 {_cmd} | {block_type}")
@plugins_status.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
await plugins_status.send(await get_plugin_status())
@group_task_status.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
await group_task_status.send(await group_current_status(event.group_id))
@group_status.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
r = re.search("^(休息吧|醒来)$", get_message_text(event.json()))
if r:
if r.group(1) == "休息吧":
msg = set_group_bot_status(event.group_id, False)
else:
msg = set_group_bot_status(event.group_id, True)
await group_status.send(msg)
logger.info(f"USER {event.user_id} GROUP {event.group_id} 使用总开关命令:{r.group(1)}")
+51
View File
@@ -0,0 +1,51 @@
from utils.utils import scheduler, get_bot
from .data_source import update_member_info
from services.log import logger
from models.group_info import GroupInfo
from asyncpg.exceptions import ConnectionDoesNotExistError, UndefinedColumnError
__zx_plugin_name__ = '管理方面定时任务 [Hidden]'
__plugin_usage__ = '无'
__plugin_des__ = '成员信息和管理权限的定时更新'
__plugin_version__ = 0.1
__plugin_author__ = 'HibiKier'
# 自动更新群员信息
@scheduler.scheduled_job(
"cron",
hour=2,
minute=1,
)
async def _():
bot = get_bot()
if bot:
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
for g in gl:
try:
await update_member_info(g)
logger.info(f"更新群组 g:{g} 成功")
except Exception as e:
logger.error(f"更新群组错误 g:{g} e:{e}")
# 快速更新群员信息以及管理员权限
@scheduler.scheduled_job(
"interval",
minutes=5,
)
async def _():
try:
bot = get_bot()
if bot:
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
all_group = [x.group_id for x in await GroupInfo.get_all_group()]
for g in gl:
if g not in all_group:
await update_member_info(g, False)
logger.info(f"快速更新群信息以及权限:{g}")
except (IndexError, ConnectionDoesNotExistError, UndefinedColumnError):
pass
+41
View File
@@ -0,0 +1,41 @@
from nonebot import on_command, on_notice
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent, GROUP, GroupIncreaseNoticeEvent
from .data_source import update_member_info
__zx_plugin_name__ = "更新群组成员列表 [Admin]"
__plugin_usage__ = """
usage:
更新群组成员的基本信息
指令:
更新群组成员列表/更新群组成员信息
""".strip()
__plugin_des__ = '更新群组成员列表'
__plugin_cmd__ = ['更新群组成员列表']
__plugin_version__ = 0.1
__plugin_author__ = 'HibiKier'
__plugin_settings__ = {
"admin_level": 1,
}
refresh_member_group = on_command(
"更新群组成员列表", aliases={"更新群组成员信息"}, permission=GROUP, priority=5, block=True
)
@refresh_member_group.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
if await update_member_info(event.group_id):
await refresh_member_group.finish("更新群员信息成功!", at_sender=True)
else:
await refresh_member_group.finish("更新群员信息失败!", at_sender=True)
group_increase_handle = on_notice(priority=1, block=False)
@group_increase_handle.handle()
async def _(bot: Bot, event: GroupIncreaseNoticeEvent, state: dict):
if event.user_id == int(bot.self_id):
await update_member_info(event.group_id)
+27
View File
@@ -0,0 +1,27 @@
from nonebot import on_command
from nonebot.typing import T_State
from nonebot.adapters import Bot
from nonebot.adapters.cqhttp import GroupMessageEvent
from utils.message_builder import image
from .data_source import create_help_image, admin_help_image
__zx_plugin_name__ = '管理帮助 [Admin]'
__plugin_usage__ = '管理员帮助,在群内回复“管理员帮助”'
__plugin_version__ = 0.1
__plugin_author__ = 'HibiKier'
__plugin_settings__ = {
"admin_level": 1,
}
admin_help = on_command("管理员帮助", aliases={"管理帮助"}, priority=5, block=True)
if admin_help_image.exists():
admin_help_image.unlink()
@admin_help.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
if not admin_help_image.exists():
await create_help_image()
await admin_help.send(image('admin_help_img.png'))
+94
View File
@@ -0,0 +1,94 @@
from utils.image_utils import BuildImage
from configs.path_config import IMAGE_PATH
from services.log import logger
from utils.utils import get_matchers
from utils.manager import group_manager
from nonebot.adapters.cqhttp import Bot
from pathlib import Path
from nonebot import Driver
import asyncio
import nonebot
driver: Driver = nonebot.get_driver()
background = Path(IMAGE_PATH) / "background" / "0.png"
admin_help_image = Path(IMAGE_PATH) / 'admin_help_img.png'
@driver.on_bot_connect
async def init_task(bot: Bot = None):
if not group_manager.get_task_data():
await group_manager.init_group_task()
logger.info(f'已成功加载 {len(group_manager.get_task_data())} 个被动技能.')
async def create_help_image():
"""
创建管理员帮助图片
"""
await asyncio.get_event_loop().run_in_executor(
None, _create_help_image
)
def _create_help_image():
"""
创建管理员帮助图片
"""
_matchers = get_matchers()
_plugin_name_list = []
width = 0
_plugin_level = {}
for matcher in _matchers:
_plugin = nonebot.plugin.get_plugin(matcher.module)
_module = _plugin.module
try:
plugin_name = _module.__getattribute__("__zx_plugin_name__")
except AttributeError:
continue
try:
if (
"[admin]" in plugin_name.lower()
and plugin_name not in _plugin_name_list
and plugin_name != "管理帮助 [Admin]"
):
_plugin_name_list.append(plugin_name)
plugin_settings = _module.__getattribute__("__plugin_settings__")
plugin_des = _module.__getattribute__("__plugin_des__")
plugin_cmd = _module.__getattribute__("__plugin_cmd__")
plugin_cmd = [x for x in plugin_cmd if "[_superuser]" not in x]
admin_level = int(plugin_settings["admin_level"])
if _plugin_level.get(admin_level):
_plugin_level[admin_level].append(
f"[{admin_level}] {plugin_des} -> " + " / ".join(plugin_cmd)
)
else:
_plugin_level[admin_level] = [
f"[{admin_level}] {plugin_des} -> " + " / ".join(plugin_cmd)
]
x = len(f"[{admin_level}] {plugin_des} -> " + " / ".join(plugin_cmd)) * 23
width = width if width > x else x
except AttributeError:
logger.warning(f"获取管理插件 {matcher.module}: {plugin_name} 设置失败...")
help_str = "* 注: ‘*’ 代表可有多个相同参数 ‘?’ 代表可省略该参数 *\n\n" \
"[权限等级] 管理员帮助:\n\n"
x = list(_plugin_level.keys())
x.sort()
for level in x:
for help_ in _plugin_level[level]:
help_str += f"\t{help_}\n\n"
help_str += '-----[被动技能开关]-----\n\n'
task_data = group_manager.get_task_data()
for i, x in enumerate(task_data.keys()):
help_str += f'{i+1}.开启/关闭{task_data[x]}\n\n'
height = len(help_str.split("\n")) * 33
A = BuildImage(width, height, font_size=24)
_background = BuildImage(width, height, background=background)
A.text((150, 110), help_str)
A.paste(_background, alpha=True)
A.save(admin_help_image)
logger.info(f'已成功加载 {len(_plugin_name_list)} 条管理员命令')
+251
View File
@@ -0,0 +1,251 @@
from utils.message_builder import image
from utils.utils import scheduler, get_bot
from nonebot import on_message
from services.log import logger
from models.group_info import GroupInfo
from models.friend_user import FriendUser
from nonebot.adapters.cqhttp.exception import ActionFailed
from configs.config import NICKNAME, Config
from utils.manager import group_manager
from pathlib import Path
import shutil
__zx_plugin_name__ = "定时任务相关 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_task__ = {'zwa': '早晚安'}
Config.add_plugin_config(
"_task",
"DEFAULT_ZWA",
True,
help_="被动 早晚安 进群默认开关状态",
default_value=True,
)
Config.add_plugin_config(
"_backup",
"BACKUP_FLAG",
True,
help_="是否开启文件备份",
default_value=True
)
Config.add_plugin_config(
"_backup",
"BACKUP_DIR_OR_FILE",
['data/black_word', 'data/configs', 'data/statistics', 'data/word_bank', 'data/manager', 'configs'],
name="文件备份",
help_="备份的文件夹或文件",
default_value=[]
)
cx = on_message(priority=9, block=False)
# 早上好
@scheduler.scheduled_job(
"cron",
hour=6,
minute=1,
)
async def _():
try:
bot = get_bot()
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
for g in gl:
if await group_manager.check_group_task_status(g, 'zwa'):
result = image("zao.jpg", "zhenxun")
try:
await bot.send_group_msg(group_id=g, message="早上好" + result)
except ActionFailed:
logger.warning(f"{g} 群被禁言中,无法发送早安")
except Exception as e:
logger.error(f"早晚安错误 e:{e}")
# 睡觉了
@scheduler.scheduled_job(
"cron",
hour=23,
minute=59,
)
async def _():
try:
bot = get_bot()
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
for g in gl:
if await group_manager.check_group_task_status(g, 'zwa'):
result = image("sleep.jpg", "zhenxun")
try:
await bot.send_group_msg(
group_id=g, message=f"{NICKNAME}要睡觉了,你们也要早点睡呀" + result
)
except ActionFailed:
logger.warning(f"{g} 群被禁言中,无法发送晚安")
except Exception as e:
logger.error(f"早晚安错误 e:{e}")
# 自动更新群组信息
@scheduler.scheduled_job(
"cron",
hour=3,
minute=1,
)
async def _():
try:
bot = get_bot()
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
for g in gl:
group_info = await bot.get_group_info(group_id=g)
await GroupInfo.add_group_info(
group_info["group_id"],
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
)
logger.info(f"自动更新群组 {g} 信息成功")
except Exception as e:
logger.error(f"自动更新群组信息错误 e:{e}")
# 自动更新好友信息
@scheduler.scheduled_job(
"cron",
hour=3,
minute=1,
)
async def _():
try:
bot = get_bot()
fl = await bot.get_friend_list()
for f in fl:
if await FriendUser.add_friend_info(f["user_id"], f["nickname"]):
logger.info(f'自动更新好友 {f["user_id"]} 信息成功')
else:
logger.warning(f'自动更新好友 {f["user_id"]} 信息失败')
except Exception as e:
logger.error(f"自动更新群组信息错误 e:{e}")
# 自动备份
@scheduler.scheduled_job(
"cron",
hour=3,
minute=25,
)
async def _():
if Config.get_config("_backup", "BACKUP_FLAG"):
_backup_path = Path() / 'backup'
_backup_path.mkdir(exist_ok=True, parents=True)
for x in Config.get_config("_backup", "BACKUP_DIR_OR_FILE"):
try:
path = Path(x)
_p = _backup_path / x
if path.exists():
if path.is_dir():
if _p.exists():
shutil.rmtree(_p, ignore_errors=True)
shutil.copytree(x, _p)
else:
if _p.exists():
_p.unlink()
shutil.copy(x, _p)
logger.info(f'已完成自动备份:{x}')
except Exception as e:
logger.error(f"自动备份文件 {x} 发生错误 {type(e)}:{e}")
# 一次性任务
# 固定时间触发,仅触发一次:
#
# from datetime import datetime
#
# @nonebot.scheduler.scheduled_job(
# 'date',
# run_date=datetime(2021, 1, 1, 0, 0),
# # timezone=None,
# )
# async def _():
# await bot.send_group_msg(group_id=123456,
# message="2021,新年快乐!")
# 定期任务
# 从 start_date 开始到 end_date 结束,根据类似 Cron
#
# 的规则触发任务:
#
# @nonebot.scheduler.scheduled_job(
# 'cron',
# # year=None,
# # month=None,
# # day=None,
# # week=None,
# day_of_week="mon,tue,wed,thu,fri",
# hour=7,
# # minute=None,
# # second=None,
# # start_date=None,
# # end_date=None,
# # timezone=None,
# )
# async def _():
# await bot.send_group_msg(group_id=123456,
# message="起床啦!")
# 间隔任务
#
# interval 触发器
#
# 从 start_date 开始,每间隔一段时间触发,到 end_date 结束:
#
# @nonebot.scheduler.scheduled_job(
# 'interval',
# # weeks=0,
# # days=0,
# # hours=0,
# minutes=5,
# # seconds=0,
# # start_date=time.now(),
# # end_date=None,
# )
# async def _():
# has_new_item = check_new_item()
# if has_new_item:
# await bot.send_group_msg(group_id=123456,
# message="XX有更新啦!")
# 动态的计划任务
# import datetime
#
# from apscheduler.triggers.date import DateTrigger # 一次性触发器
# # from apscheduler.triggers.cron import CronTrigger # 定期触发器
# # from apscheduler.triggers.interval import IntervalTrigger # 间隔触发器
# from nonebot import on_command, scheduler
#
# @on_command('赖床')
# async def _(session: CommandSession):
# await session.send('我会在5分钟后再喊你')
#
# # 制作一个“5分钟后”触发器
# delta = datetime.timedelta(minutes=5)
# trigger = DateTrigger(
# run_date=datetime.datetime.now() + delta
# )
#
# # 添加任务
# scheduler.add_job(
# func=session.send, # 要添加任务的函数,不要带参数
# trigger=trigger, # 触发器
# args=('不要再赖床啦!',), # 函数的参数列表,注意:只有一个值时,不能省略末尾的逗号
# # kwargs=None,
# misfire_grace_time=60, # 允许的误差时间,建议不要省略
# # jobstore='default', # 任务储存库,在下一小节中说明
# )
+207
View File
@@ -0,0 +1,207 @@
from nonebot import on_command
from models.ban_user import BanUser
from models.level_user import LevelUser
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot
from nonebot.adapters.cqhttp import GroupMessageEvent, PrivateMessageEvent, MessageEvent
from utils.utils import get_message_at, get_message_text, is_number
from configs.config import NICKNAME, Config
from nonebot.permission import SUPERUSER
from services.log import logger
__zx_plugin_name__ = "封禁Ban用户 [Admin]"
__plugin_usage__ = """
usage:
将用户拉入或拉出黑名单
指令:
.ban [at] ?[小时] ?[分钟]
.unban
示例:.ban @user
示例:.ban @user 6
示例:.ban @user 3 10
示例:.unban @user
""".strip()
__plugin_superuser_usage__ = """
usage:
b了=屏蔽用户消息,相当于最上级.ban
跨群ban以及跨群b了
指令:
b了 [at/qq]
.ban [user_id] ?[小时] ?[分钟]
示例:b了 @user
示例:b了 1234567
示例:.ban 12345567
""".strip()
__plugin_des__ = '你被逮捕了!丢进小黑屋!'
__plugin_cmd__ = ['.ban [at] ?[小时] ?[分钟]', '.unban [at]', 'b了 [at] [_superuser]']
__plugin_version__ = 0.1
__plugin_author__ = 'HibiKier'
__plugin_settings__ = {
"admin_level": Config.get_config("ban", "BAN_LEVEL"),
"cmd": ['.ban', '.unban', 'ban', 'unban']
}
__plugin_configs__ = {
"BAN_LEVEL [LEVEL]": {
"value": 5,
"help": "ban/unban所需要的管理员权限等级",
"default_value": 5
}
}
ban = on_command(
".ban",
aliases={".unban", "/ban", "/unban"},
priority=5,
block=True,
)
super_ban = on_command('b了', permission=SUPERUSER, priority=5, block=True)
@ban.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
result = ""
qq = get_message_at(event.json())
if qq:
qq = qq[0]
user_name = await bot.get_group_member_info(group_id=event.group_id, user_id=qq)
user_name = user_name['card'] if user_name['card'] else user_name['nickname']
msg = get_message_text(event.json())
if msg:
msg = msg.split()
if len(msg) == 2:
if not is_number(msg[0].strip()) or not is_number(msg[1].strip()):
await ban.finish("参数必须是数字!", at_sender=True)
time = int(msg[0]) * 60 * 60 + int(msg[1]) * 60
else:
if not is_number(msg[0].strip()):
await ban.finish("参数必须是数字!", at_sender=True)
time = int(msg[0]) * 60 * 60
else:
time = -1
if state["_prefix"]["raw_command"] in [".ban", "/ban"]:
if (
await LevelUser.get_user_level(event.user_id, event.group_id)
<= await LevelUser.get_user_level(qq, event.group_id)
and str(event.user_id) not in bot.config.superusers
):
await ban.finish(
f"您的权限等级比对方低或相等, {NICKNAME}不能为您使用此功能!",
at_sender=True,
)
if await BanUser.ban(
qq, await LevelUser.get_user_level(event.user_id, event.group_id), time
):
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 将 USER {qq} 封禁 时长 {time/60} 分钟"
)
result = f"已经将 {user_name} 加入{NICKNAME}的黑名单了!"
if time != -1:
result += f"将在 {time/60} 分钟后解封"
else:
time = await BanUser.check_ban_time(qq)
if is_number(time):
time = abs(int(time))
if time < 60:
time = str(time) + " 秒"
else:
time = str(int(time / 60)) + " 分钟"
else:
time += " 分钟"
result = f"{user_name} 已在黑名单!预计 {time}后解封"
else:
if (
await BanUser.check_ban_level(
qq, await LevelUser.get_user_level(event.user_id, event.group_id)
)
and str(event.user_id) not in bot.config.superusers
):
await ban.finish(
f"ban掉 {user_name} 的管理员权限比您高,无法进行unban", at_sender=True
)
if await BanUser.unban(qq):
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 将 USER {qq} 解禁"
)
result = f"已经把 {user_name} 从黑名单中删除了!"
else:
result = f"{user_name} 不在黑名单!"
else:
await ban.finish("艾特人了吗??", at_sender=True)
await ban.finish(result, at_sender=True)
@ban.handle()
async def _(bot: Bot, event: PrivateMessageEvent, state: T_State):
if str(event.user_id) in bot.config.superusers:
msg = get_message_text(event.json())
msg = msg.split()
if is_number(msg[0]):
qq = int(msg[0])
if state["_prefix"]["raw_command"] in [".ban", "/ban"]:
hour = 0
minute = 0
if len(msg) > 1 and is_number(msg[1]):
hour = int(msg[1])
if len(msg) > 2 and is_number(msg[2]):
minute = int(msg[2])
time = hour * 60 * 60 + minute * 60
time = time if time else -1
if await BanUser.ban(
qq, 9, time
):
logger.info(
f"USER {event.user_id} 将 USER {qq} 封禁 时长 {time/60} 分钟"
)
result = f"已经将 {qq} 加入{NICKNAME}的黑名单了!"
if time != -1:
result += f"将在 {time/60} 分钟后解封"
else:
result += f"将在 ∞ 分钟后解封"
await ban.send(result)
else:
time = await BanUser.check_ban_time(qq)
if is_number(time):
time = abs(int(time))
if time < 60:
time = str(time) + " 秒"
else:
time = str(int(time / 60)) + " 分钟"
else:
time += " 分钟"
await ban.send(f"{qq} 已在黑名单!预计 {time}后解封")
else:
if await BanUser.unban(qq):
logger.info(
f"USER {event.user_id} 将 USER {qq} 解禁"
)
result = f"已经把 {qq} 从黑名单中删除了!"
else:
result = f"{qq} 不在黑名单!"
await ban.send(result)
else:
await ban.finish('qq号必须是数字!\n格式:.ban [qq] [hour]? [minute]?', at_sender=True)
@super_ban.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
if isinstance(event, GroupMessageEvent):
qq = get_message_at(event.json())
else:
qq = get_message_text(event.json())
if not is_number(qq):
await super_ban.finish("对象qq必须为纯数字...")
qq = [qq]
if qq:
qq = qq[0]
user = await bot.get_group_member_info(group_id=event.group_id, user_id=qq)
user_name = user['card'] if user['card'] else user['nickname']
if not await BanUser.ban(qq, 10, 99999999):
await BanUser.unban(qq)
await BanUser.ban(qq, 10, 99999999)
await ban.send(f"已将 {user_name} 拉入黑名单!")
else:
await super_ban.send('需要艾特被super ban的对象..')
+67
View File
@@ -0,0 +1,67 @@
from nonebot import on_command
from nonebot.typing import T_State
from nonebot.adapters import Bot, Event
from nonebot.permission import SUPERUSER
import asyncio
from utils.utils import get_message_text, get_message_imgs
from services.log import logger
from utils.message_builder import image
from utils.manager import group_manager
from configs.config import Config
__zx_plugin_name__ = "广播 [Superuser]"
__plugin_usage__ = """
usage:
指令:
广播- ?[消息] ?[图片]
示例:广播- 你们好!
""".strip()
__plugin_des__ = "昭告天下!"
__plugin_cmd__ = ["广播-"]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_task__ = {"broadcast": "广播"}
Config.add_plugin_config(
"_task",
"DEFAULT_BROADCAST",
True,
help_="被动 广播 进群默认开关状态",
default_value=True,
)
broadcast = on_command("广播-", priority=1, permission=SUPERUSER, block=True)
@broadcast.handle()
async def _(bot: Bot, event: Event, state: T_State):
msg = get_message_text(event.json())
imgs = get_message_imgs(event.json())
rst = ""
for img in imgs:
rst += image(img)
sid = bot.self_id
gl = await bot.get_group_list(self_id=sid)
gl = [
g["group_id"]
for g in gl
if await group_manager.check_group_task_status(g["group_id"], "broadcast")
]
g_cnt = len(gl)
cnt = 0
error = ""
x = 0.25
for g in gl:
cnt += 1
if cnt / g_cnt > x:
await broadcast.send(f"已播报至 {int(cnt / g_cnt * 100)}% 的群聊")
x += 0.25
try:
await bot.send_group_msg(self_id=sid, group_id=g, message=msg + rst)
logger.info(f"GROUP {g} 投递广播成功")
except Exception as e:
logger.error(f"GROUP {g} 投递广播失败:{type(e)}")
error += f"GROUP {g} 投递广播失败:{type(e)}\n"
await asyncio.sleep(0.5)
await broadcast.send(f"已播报至 100% 的群聊")
if error:
await broadcast.send(f"播报时错误:{error}")
+200
View File
@@ -0,0 +1,200 @@
from nonebot import on_notice, on_request
from configs.path_config import IMAGE_PATH, DATA_PATH
from utils.message_builder import image
from models.group_member_info import GroupInfoUser
from datetime import datetime
from services.log import logger
from nonebot.adapters.cqhttp import (
Bot,
GroupIncreaseNoticeEvent,
GroupDecreaseNoticeEvent,
)
from nonebot.adapters.cqhttp.exception import ActionFailed
from utils.manager import group_manager, plugins2settings_manager, requests_manager
from configs.config import NICKNAME
from models.group_info import GroupInfo
from utils.utils import FreqLimiter
from configs.config import Config
from pathlib import Path
import random
import os
try:
import ujson as json
except ModuleNotFoundError:
import json
__zx_plugin_name__ = "群事件处理 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_task__ = {"group_welcome": "进群欢迎", "refund_group_remind": "退群提醒"}
Config.add_plugin_config(
"invite_manager", "message", f"请不要未经同意就拉{NICKNAME}入群!告辞!", help_="强制拉群后进群回复的内容.."
)
Config.add_plugin_config(
"invite_manager", "flag", True, help_="被强制拉群后是否直接退出", default_value=True
)
Config.add_plugin_config(
"invite_manager", "welcome_msg_cd", 5, help_="群欢迎消息cd", default_value=5
)
Config.add_plugin_config(
"_task",
"DEFAULT_GROUP_WELCOME",
True,
help_="被动 进群欢迎 进群默认开关状态",
default_value=True,
)
Config.add_plugin_config(
"_task",
"DEFAULT_REFUND_GROUP_REMIND",
True,
help_="被动 退群提醒 进群默认开关状态",
default_value=True,
)
_flmt = FreqLimiter(Config.get_config("invite_manager", "welcome_msg_cd"))
# 群员增加处理
group_increase_handle = on_notice(priority=1, block=False)
# 群员减少处理
group_decrease_handle = on_notice(priority=1, block=False)
# (群管理)加群同意请求
add_group = on_request(priority=1, block=False)
@group_increase_handle.handle()
async def _(bot: Bot, event: GroupIncreaseNoticeEvent, state: dict):
if event.user_id == int(bot.self_id):
group = await GroupInfo.get_group_info(event.group_id)
# 群聊不存在或被强制拉群,退出该群
if (not group or group.group_flag == 0) and Config.get_config(
"invite_manager", "flag"
):
try:
msg = Config.get_config("invite_manager", "message")
if msg:
await bot.send_group_msg(group_id=event.group_id, message=msg)
await bot.set_group_leave(group_id=event.group_id)
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]),
message=f"触发强制入群保护,已成功退出群聊 {event.group_id}..",
)
logger.info(f"强制拉群或未有群信息,退出群聊 {group} 成功")
requests_manager.remove_request("group", event.group_id)
except Exception as e:
logger.info(f"强制拉群或未有群信息,退出群聊 {group} 失败 e:{e}")
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]),
message=f"触发强制入群保护,退出群聊 {event.group_id} 失败..",
)
# 默认群功能开关
elif event.group_id not in group_manager["group_manager"].keys():
data = plugins2settings_manager.get_data()
for plugin in data.keys():
if not data[plugin]["default_status"]:
group_manager.block_plugin(plugin, event.group_id)
else:
join_time = datetime.now()
user_info = await bot.get_group_member_info(
group_id=event.group_id, user_id=event.user_id
)
if await GroupInfoUser.add_member_info(
user_info["user_id"],
user_info["group_id"],
user_info["nickname"],
join_time,
):
logger.info(f"用户{user_info['user_id']} 所属{user_info['group_id']} 更新成功")
else:
logger.info(f"用户{user_info['user_id']} 所属{user_info['group_id']} 更新失败")
# 群欢迎消息
if await group_manager.check_group_task_status(
event.group_id, "group_welcome"
) and _flmt.check(event.group_id):
_flmt.start_cd(event.group_id)
msg = ""
img = ""
at_flag = False
custom_welcome_msg_json = (
Path() / "data" / "custom_welcome_msg" / "custom_welcome_msg.json"
)
if custom_welcome_msg_json.exists():
data = json.load(open(custom_welcome_msg_json, "r"))
if data.get(str(event.group_id)):
msg = data[str(event.group_id)]
if msg.find("[at]") != -1:
msg = msg.replace("[at]", "")
at_flag = True
if os.path.exists(DATA_PATH + f"custom_welcome_msg/{event.group_id}.jpg"):
img = image(
abspath=DATA_PATH + f"custom_welcome_msg/{event.group_id}.jpg"
)
if msg or img:
await group_increase_handle.send(
"\n" + msg.strip() + img, at_sender=at_flag
)
else:
await group_increase_handle.send(
"新人快跑啊!!本群现状↓(快使用自定义!)"
+ image(random.choice(os.listdir(IMAGE_PATH + "qxz/")), "qxz")
)
@group_decrease_handle.handle()
async def _(bot: Bot, event: GroupDecreaseNoticeEvent, state: dict):
# 被踢出群
if event.sub_type == "kick_me":
group_id = event.group_id
operator_id = event.operator_id
try:
operator_name = (
await GroupInfoUser.get_member_info(event.operator_id, event.group_id)
).user_name
except AttributeError:
operator_name = "None"
group = await GroupInfo.get_group_info(group_id)
group_name = group.group_name if group else ""
coffee = int(list(bot.config.superusers)[0])
await bot.send_private_msg(
user_id=coffee,
message=f"****呜..一份踢出报告****\n"
f"我被 {operator_name}({operator_id})\n"
f"踢出了 {group_name}({group_id})\n"
f"日期:{str(datetime.now()).split('.')[0]}",
)
return
if event.user_id == int(bot.self_id):
group_manager.delete_group(event.group_id)
return
try:
user_name = (
await GroupInfoUser.get_member_info(event.user_id, event.group_id)
).user_name
except AttributeError:
user_name = str(event.user_id)
if await GroupInfoUser.delete_member_info(event.user_id, event.group_id):
logger.info(f"用户{user_name}, qq={event.user_id} 所属{event.group_id} 删除成功")
else:
logger.info(f"用户{user_name}, qq={event.user_id} 所属{event.group_id} 删除失败")
if await group_manager.check_group_task_status(
event.group_id, "refund_group_remind"
):
rst = ""
if event.sub_type == "leave":
rst = f"{user_name}离开了我们..."
if event.sub_type == "kick":
operator = await bot.get_group_member_info(
user_id=event.operator_id, group_id=event.group_id
)
operator_name = (
operator["card"] if operator["card"] else operator["nickname"]
)
rst = f"{user_name} 被 {operator_name} 送走了."
try:
await group_decrease_handle.send(f"{rst}")
except ActionFailed:
return
+73
View File
@@ -0,0 +1,73 @@
from nonebot import on_command
from nonebot.adapters.cqhttp import (
Bot,
MessageEvent,
GroupMessageEvent
)
from nonebot.typing import T_State
from nonebot.rule import to_me
from configs.path_config import IMAGE_PATH, DATA_PATH
from utils.message_builder import image
from .data_source import create_help_img, get_plugin_help
from utils.utils import get_message_text
from pathlib import Path
import os
__zx_plugin_name__ = "帮助"
group_help_path = Path(DATA_PATH) / "group_help"
help_image = Path(IMAGE_PATH) / "help.png"
simple_help_image = Path(IMAGE_PATH) / "simple_help.png"
if help_image.exists():
help_image.unlink()
if simple_help_image.exists():
simple_help_image.unlink()
group_help_path.mkdir(exist_ok=True, parents=True)
for x in os.listdir(group_help_path):
group_help_image = group_help_path / x
group_help_image.unlink()
_help = on_command("详细功能", rule=to_me(), aliases={"详细帮助"}, priority=1, block=True)
simple_help = on_command("功能", rule=to_me(), aliases={"help", "帮助"}, priority=1, block=True)
@_help.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
if not help_image.exists():
if help_image.exists():
help_image.unlink()
if simple_help_image.exists():
simple_help_image.unlink()
await create_help_img(None, help_image, simple_help_image)
await _help.finish(image("help.png"))
@simple_help.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json())
is_super = False
if msg:
if '-super' in msg:
if str(event.user_id) in bot.config.superusers:
is_super = True
msg = msg.replace('-super', '').strip()
msg = get_plugin_help(msg, is_super)
if msg:
await _help.send(image(b64=msg))
else:
await _help.send("没有此功能的帮助信息...")
else:
if isinstance(event, GroupMessageEvent):
_image_path = group_help_path / f"{event.group_id}.png"
if not _image_path.exists():
await create_help_img(event.group_id, help_image, _image_path)
await simple_help.send(image(_image_path))
else:
if not simple_help_image.exists():
if help_image.exists():
help_image.unlink()
if simple_help_image.exists():
simple_help_image.unlink()
await create_help_img(None, help_image, simple_help_image)
await _help.finish(image("simple_help.png"))
+367
View File
@@ -0,0 +1,367 @@
from utils.image_utils import BuildImage
from configs.path_config import IMAGE_PATH
from utils.manager import (
plugins2settings_manager,
admin_manager,
plugins_manager,
group_manager,
)
from typing import Optional
from services.log import logger
from pathlib import Path
from utils.utils import get_matchers
import random
import asyncio
import nonebot
import os
random_bk_path = Path(IMAGE_PATH) / "background" / "help" / "simple_help"
background = Path(IMAGE_PATH) / "background" / "0.png"
async def create_help_img(
group_id: Optional[int], help_image: Path, simple_help_image: Path
):
"""
生成帮助图片
:param group_id: 群号
:param help_image: 图片路径
:param simple_help_image: 简易帮助图片路径
"""
return await asyncio.get_event_loop().run_in_executor(
None, _create_help_img, group_id, help_image, simple_help_image
)
def _create_help_img(
group_id: Optional[int], help_image: Path, simple_help_image: Path
):
"""
生成帮助图片
:param group_id: 群号
:param help_image: 图片路径
:param simple_help_image: 简易帮助图片路径
"""
_matchers = get_matchers()
width = 0
matchers_data = {}
_des_tmp = {}
_plugin_name_tmp = []
_tmp = []
tmp_img = BuildImage(0, 0, plain_text="1", font_size=24)
font_height = tmp_img.h
# 插件分类
for matcher in _matchers:
plugin_name = None
_plugin = nonebot.plugin.get_plugin(matcher.module)
_module = _plugin.module
try:
plugin_name = _module.__getattribute__("__zx_plugin_name__")
try:
plugin_des = _module.__getattribute__("__plugin_des__")
except AttributeError:
plugin_des = "_"
if (
"[hidden]" in plugin_name.lower()
or "[admin]" in plugin_name.lower()
or "[superuser]" in plugin_name.lower()
or plugin_name in _plugin_name_tmp
or plugin_name == "帮助"
):
continue
plugin_type = ("normal",)
text_type = 0
if plugins2settings_manager.get(
matcher.module
) and plugins2settings_manager[matcher.module].get("plugin_type"):
plugin_type = tuple(
plugins2settings_manager.get_plugin_data(matcher.module)[
"plugin_type"
]
)
else:
try:
plugin_type = _module.__getattribute__("__plugin_type__")
except AttributeError:
pass
if len(plugin_type) > 1:
try:
text_type = int(plugin_type[1])
except ValueError as e:
logger.warning(f"生成列向帮助排列失败 {plugin_name}: {type(e)}: {e}")
plugin_type = plugin_type[0]
else:
plugin_type = plugin_type[0]
try:
plugin_cmd = _module.__getattribute__("__plugin_cmd__")
plugin_cmd = [x for x in plugin_cmd if "[_superuser]" not in x]
except AttributeError:
plugin_cmd = []
if plugin_type not in matchers_data.keys():
matchers_data[plugin_type] = {}
if plugin_des in _des_tmp.keys():
try:
matchers_data[plugin_type][_des_tmp[plugin_des]]["cmd"] = (
matchers_data[plugin_type][_des_tmp[plugin_des]]["cmd"]
+ plugin_cmd
)
except KeyError as e:
logger.warning(f"{type(e)}: {e}")
else:
matchers_data[plugin_type][plugin_name] = {
"module": matcher.module,
"des": plugin_des,
"cmd": plugin_cmd,
"text_type": text_type,
}
try:
if text_type == 0:
x = tmp_img.getsize(
f'{plugin_name}: {matchers_data[plugin_type][plugin_name]["des"]} ->'
+ " / ".join(matchers_data[plugin_type][plugin_name]["cmd"])
)[0]
width = width if width > x else x
except KeyError:
pass
if plugin_des not in _des_tmp:
_des_tmp[plugin_des] = plugin_name
except AttributeError as e:
if plugin_name not in _plugin_name_tmp:
logger.warning(f"获取功能 {matcher.module}: {plugin_name} 设置失败...e:{e}")
if plugin_name not in _plugin_name_tmp:
_plugin_name_tmp.append(plugin_name)
help_img_list = []
simple_help_img_list = []
types = list(matchers_data.keys())
types.sort()
ix = 0
# 详细帮助
for type_ in types:
keys = list(matchers_data[type_].keys())
keys.sort()
help_str = f"{type_ if type_ != 'normal' else '功能'}:\n\n"
simple_help_str = f"{type_ if type_ != 'normal' else '功能'}:\n\n"
for i, k in enumerate(keys):
# 禁用flag
flag = True
if plugins_manager.get_plugin_status(
matchers_data[type_][k]["module"], "all"
):
flag = False
if group_id:
flag = flag and plugins_manager.get_plugin_status(
matchers_data[type_][k]["module"], "group"
)
simple_help_str += (
f"{i+1}.{k}<|_|~|>"
f"{group_manager.get_plugin_status(matchers_data[type_][k]['module'], group_id) if group_id else '_'}|"
f"{flag}\n"
)
if matchers_data[type_][k]["text_type"] == 1:
_x = tmp_img.getsize(
f"{i+1}".rjust(5)
+ f'.{k}: {matchers_data[type_][k]["des"]} {"->" if matchers_data[type_][k]["cmd"] else ""} '
)[0]
_str = (
f"{i+1}".rjust(5)
+ f'.{k}: {matchers_data[type_][k]["des"]} {"->" if matchers_data[type_][k]["cmd"] else ""} '
)
_str += matchers_data[type_][k]["cmd"][0] + "\n"
for c in matchers_data[type_][k]["cmd"][1:]:
_str += "".rjust(int(_x * 0.125) + 1) + f"{c}\n"
help_str += _str
else:
help_str += (
f"{i+1}".rjust(5)
+ f'.{k}: {matchers_data[type_][k]["des"]} {"->" if matchers_data[type_][k]["cmd"] else ""} '
+ " / ".join(matchers_data[type_][k]["cmd"])
+ "\n"
)
height = len(help_str.split("\n")) * (font_height + 5)
simple_height = len(simple_help_str.split("\n")) * (font_height + 5)
A = BuildImage(
width + 150, height, font_size=24, color="white" if not ix % 2 else "black"
)
A.text((10, 10), help_str, (255, 255, 255) if ix % 2 else (0, 0, 0))
# 生成各个分类的插件简易帮助图片
simple_width = 0
for x in [
tmp_img.getsize(x.split("<|_|~|>")[0])[0]
for x in simple_help_str.split("\n")
]:
simple_width = simple_width if simple_width > x else x
bk = BuildImage(simple_width + 20, simple_height, font_size=24, color="#6495ED")
B = BuildImage(
simple_width + 20,
simple_height,
font_size=24,
color="white" if not ix % 2 else "black",
)
# 切分,判断插件开关状态
_s_height = 10
for _s in simple_help_str.split("\n"):
text_color = (255, 255, 255) if ix % 2 else (0, 0, 0)
_line_flag = False
if "<|_|~|>" in _s:
_x = _s.split("<|_|~|>")
_flag_sp = _x[-1].split("|")
if group_id:
if _flag_sp[0].lower() != "true":
text_color = (252, 75, 13)
if _flag_sp[1].lower() == "true":
_line_flag = True
_s = _x[0]
B.text((10, _s_height), _s, text_color)
if _line_flag:
B.line(
(
7,
_s_height + int(B.getsize(_s)[1] / 2) + 2,
B.getsize(_s)[0] + 11,
_s_height + int(B.getsize(_s)[1] / 2) + 2,
),
(236, 66, 7),
3,
)
_s_height += B.getsize("1")[1] + 5
# B.text((10, 10), simple_help_str, (255, 255, 255) if ix % 2 else (0, 0, 0))
bk.paste(B, center_type="center")
bk.transparent(2)
ix += 1
help_img_list.append(A)
simple_help_img_list.append(bk)
height = 0
for img in help_img_list:
height += img.h
if not group_id:
A = BuildImage(width + 150, height + 50, font_size=24)
A.text(
(10, 10), '* 注: ‘*’ 代表可有多个相同参数 ‘?’ 代表可省略该参数 *\n\n" "功能名: 功能简介 -> 指令\n\n'
)
current_height = 50
for img in help_img_list:
A.paste(img, (0, current_height))
current_height += img.h
A.save(help_image)
# 详细帮助生成完毕
# 简易帮助图片合成
height = 0
width = 0
for img in simple_help_img_list:
if img.h > height:
height = img.h
width += img.w + 10
B = BuildImage(width + 100, height + 250, font_size=24)
width, _ = get_max_width_or_paste(simple_help_img_list, B)
bk = None
random_bk = os.listdir(random_bk_path)
if random_bk:
bk = random.choice(random_bk)
x = max(width + 50, height + 250)
B = BuildImage(
x,
x,
font_size=24,
color="#FFEFD5",
background=random_bk_path / bk,
)
B.filter("GaussianBlur", 10)
_, B = get_max_width_or_paste(simple_help_img_list, B, True)
w = 10
h = 10
for msg in ["目前支持的功能列表:", "可以通过 ‘帮助[功能名称]’ 来获取对应功能的使用方法", "或者使用 ‘详细帮助’ 来获取所有功能方法"]:
text = BuildImage(
0,
0,
plain_text=msg,
font_size=24,
font="yuanshen.ttf",
)
B.paste(text, (w, h), True)
h += 50
if msg == "目前支持的功能列表:":
w += 50
B.paste(
BuildImage(
0,
0,
plain_text="注: 红字代表功能被群管理员禁用,红线代表功能正在维护",
font_size=24,
font="yuanshen.ttf",
font_color=(231, 74, 57)
),
(300, 10),
True,
)
B.save(simple_help_image)
def get_max_width_or_paste(
simple_help_img_list: list, B: BuildImage = None, is_paste: bool = False
) -> "int, BuildImage":
"""
获取最大宽度,或直接贴图
:param simple_help_img_list: 简单帮助图片列表
:param B: 背景图
:param is_paste: 是否直接贴图
"""
current_width = 50
current_height = 180
max_width = simple_help_img_list[0].w
for i in range(len(simple_help_img_list)):
try:
if is_paste and B:
B.paste(simple_help_img_list[i], (current_width, current_height), True)
current_height += simple_help_img_list[i].h + 40
if current_height + simple_help_img_list[i + 1].h > B.h - 10:
current_height = 180
current_width += max_width + 30
max_width = 0
elif simple_help_img_list[i].w > max_width:
max_width = simple_help_img_list[i].w
except IndexError:
pass
if current_width > simple_help_img_list[0].w + 50:
current_width += simple_help_img_list[-1].w
return current_width, B
def get_plugin_help(msg: str, is_super: bool = False) -> Optional[str]:
"""
获取功能的帮助信息
:param msg: 功能cmd
:param is_super: 是否为超级用户
"""
module = plugins2settings_manager.get_plugin_module(msg)
if not module:
module = admin_manager.get_plugin_module(msg)
if module:
try:
plugin = nonebot.plugin.get_plugin(module)
if plugin:
if is_super:
result = plugin.module.__getattribute__(
"__plugin_superuser_usage__"
)
else:
result = plugin.module.__getattribute__("__plugin_usage__")
if result:
width = 0
for x in result.split("\n"):
_width = len(x) * 24
width = width if width > _width else _width
height = len(result.split("\n")) * 45
A = BuildImage(width, height, font_size=24)
bk = BuildImage(
width,
height,
background=Path(IMAGE_PATH) / "background" / "1.png",
)
A.paste(bk, alpha=True)
A.text((int(width * 0.048), int(height * 0.21)), result)
return A.pic2bs4()
except AttributeError:
pass
return None
+36
View File
@@ -0,0 +1,36 @@
from configs.config import Config
Config.add_plugin_config(
"hook",
"CHECK_NOTICE_INFO_CD",
300,
name="基础hook配置",
help_="群检测,个人权限检测等各种检测提示信息cd",
default_value=300
)
Config.add_plugin_config(
"hook",
"MALICIOUS_BAN_TIME",
30,
help_="恶意命令触发检测触发后ban的时长(分钟)",
default_value=30
)
Config.add_plugin_config(
"hook",
"MALICIOUS_CHECK_TIME",
5,
help_="恶意命令触发检测规定时间内(秒)",
default_value=5
)
Config.add_plugin_config(
"hook",
"MALICIOUS_BAN_COUNT",
6,
help_="恶意命令触发检测最大触发次数",
default_value=6
)
+378
View File
@@ -0,0 +1,378 @@
from nonebot.matcher import Matcher
from nonebot.message import run_preprocessor, run_postprocessor, IgnoredException
from nonebot.adapters.cqhttp.exception import ActionFailed
from models.friend_user import FriendUser
from models.group_member_info import GroupInfoUser
from models.bag_user import BagUser
from utils.manager import (
plugins2settings_manager,
admin_manager,
group_manager,
plugins_manager,
plugins2cd_manager,
plugins2block_manager,
plugins2count_manager
)
from .utils import set_block_limit_false, status_message_manager
from nonebot.typing import T_State
from typing import Optional
from nonebot.adapters.cqhttp import (
Bot,
MessageEvent,
GroupMessageEvent,
PokeNotifyEvent,
PrivateMessageEvent,
Message,
Event
)
from configs.config import Config
from models.ban_user import BanUser
from utils.utils import FreqLimiter
from utils.message_builder import at
from models.level_user import LevelUser
import nonebot
_flmt = FreqLimiter(Config.get_config("hook", "CHECK_NOTICE_INFO_CD"))
_flmt_g = FreqLimiter(Config.get_config("hook", "CHECK_NOTICE_INFO_CD"))
_flmt_s = FreqLimiter(Config.get_config("hook", "CHECK_NOTICE_INFO_CD"))
_flmt_c = FreqLimiter(Config.get_config("hook", "CHECK_NOTICE_INFO_CD"))
ignore_rst_module = ["ai", "poke", "dialogue"]
# 权限检测
@run_preprocessor
async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
module = matcher.module
plugins2info_dict = plugins2settings_manager.get_data()
# 功能的金币检测 #######################################
# 功能的金币检测 #######################################
# 功能的金币检测 #######################################
cost_gold = 0
if isinstance(event, GroupMessageEvent) and plugins2settings_manager.get_plugin_data(module).get('cost_gold'):
cost_gold = plugins2settings_manager.get_plugin_data(module).get('cost_gold')
if await BagUser.get_gold(event.user_id, event.group_id) < cost_gold:
await send_msg(f"金币不足..该功能需要{cost_gold}金币..", bot, event)
raise IgnoredException(f"{module} 金币限制...")
# 当插件不阻塞超级用户时,超级用户提前扣除金币
if str(event.user_id) in bot.config.superusers and not plugins2info_dict[module]["limit_superuser"]:
await BagUser.spend_gold(event.user_id, event.group_id, cost_gold)
try:
if (
(not isinstance(event, MessageEvent) and module != "poke")
or await BanUser.is_ban(event.user_id)
and str(event.user_id) not in bot.config.superusers
) or (
str(event.user_id) in bot.config.superusers
and plugins2info_dict.get(module)
and not plugins2info_dict[module]["limit_superuser"]
):
return
except AttributeError:
pass
# 超级用户命令
try:
_plugin = nonebot.plugin.get_plugin(module)
_module = _plugin.module
plugin_name = _module.__getattribute__("__zx_plugin_name__")
if (
"[superuser]" in plugin_name.lower()
and str(event.user_id) in bot.config.superusers
):
return
except AttributeError:
pass
# 群黑名单检测 群总开关检测
if isinstance(event, GroupMessageEvent) or matcher.module == "poke":
try:
if group_manager.get_group_level(event.group_id) < 0:
raise IgnoredException("群黑名单")
if not group_manager.check_group_bot_status(event.group_id):
try:
if str(event.get_message()) != "醒来":
raise IgnoredException("功能总开关关闭状态")
except ValueError:
raise IgnoredException("功能总开关关闭状态")
except AttributeError:
pass
if module in admin_manager.keys() and matcher.priority not in [1, 9]:
if isinstance(event, GroupMessageEvent):
# 个人权限
if (
not await LevelUser.check_level(
event.user_id,
event.group_id,
admin_manager.get_plugin_level(module),
)
and admin_manager.get_plugin_level(module) > 0
):
try:
if _flmt.check(event.user_id):
_flmt.start_cd(event.user_id)
await bot.send_group_msg(
group_id=event.group_id,
message=f"{at(event.user_id)}你的权限不足喔,该功能需要的权限等级:"
f"{admin_manager.get_plugin_level(module)}",
)
except ActionFailed:
pass
set_block_limit_false(event, module)
if event.is_tome():
status_message_manager.add(event.group_id)
raise IgnoredException("权限不足")
else:
if not await LevelUser.check_level(
event.user_id, 0, admin_manager.get_plugin_level(module)
):
try:
await bot.send_private_msg(
user_id=event.user_id,
message=f"你的权限不足喔,该功能需要的权限等级:{admin_manager.get_plugin_level(module)}",
)
except ActionFailed:
pass
set_block_limit_false(event, module)
if event.is_tome():
status_message_manager.add(event.user_id)
raise IgnoredException("权限不足")
if module in plugins2info_dict.keys() and matcher.priority not in [1, 9]:
# 戳一戳单独判断
if isinstance(event, GroupMessageEvent) or (
isinstance(event, PokeNotifyEvent) and event.group_id
):
if status_message_manager.get(event.group_id) is None:
status_message_manager.delete(event.group_id)
if plugins2info_dict[module]["level"] > group_manager.get_group_level(
event.group_id
):
try:
if _flmt_g.check(event.user_id) and module not in ignore_rst_module:
_flmt_g.start_cd(event.user_id)
await bot.send_group_msg(
group_id=event.group_id, message="群权限不足..."
)
except ActionFailed:
pass
if event.is_tome():
status_message_manager.add(event.group_id)
set_block_limit_false(event, module)
raise IgnoredException("群权限不足")
# 插件状态
if not group_manager.get_plugin_status(module, event.group_id):
try:
if module not in ignore_rst_module and _flmt_s.check(
event.group_id
):
_flmt_s.start_cd(event.group_id)
await bot.send_group_msg(
group_id=event.group_id, message="该群未开启此功能.."
)
except ActionFailed:
pass
if event.is_tome():
status_message_manager.add(event.group_id)
set_block_limit_false(event, module)
raise IgnoredException("未开启此功能...")
# 管理员禁用
if not group_manager.get_plugin_status(f"{module}:super", event.group_id):
try:
if (
_flmt_s.check(event.group_id)
and module not in ignore_rst_module
):
_flmt_s.start_cd(event.group_id)
await bot.send_group_msg(
group_id=event.group_id, message="管理员禁用了此群该功能..."
)
except ActionFailed:
pass
if event.is_tome():
status_message_manager.add(event.group_id)
set_block_limit_false(event, module)
raise IgnoredException("管理员禁用了此群该功能...")
# 群聊禁用
if not plugins_manager.get_plugin_status(module, block_type="group"):
try:
if (
_flmt_c.check(event.group_id)
and module not in ignore_rst_module
):
_flmt_c.start_cd(event.group_id)
await bot.send_group_msg(
group_id=event.group_id, message="该功能在群聊中已被禁用..."
)
except ActionFailed:
pass
if event.is_tome():
status_message_manager.add(event.group_id)
set_block_limit_false(event, module)
raise IgnoredException("该插件在群聊中已被禁用...")
else:
# 私聊禁用
if not plugins_manager.get_plugin_status(module, block_type="private"):
try:
if _flmt_c.check(event.user_id):
_flmt_c.start_cd(event.user_id)
await bot.send_private_msg(
user_id=event.user_id, message="该功能在私聊中已被禁用..."
)
except ActionFailed:
pass
if event.is_tome():
status_message_manager.add(event.user_id)
set_block_limit_false(event, module)
raise IgnoredException("该插件在私聊中已被禁用...")
# 维护
if not plugins_manager.get_plugin_status(module, block_type="all"):
if isinstance(
event, GroupMessageEvent
) and group_manager.check_group_is_white(event.group_id):
return
try:
if isinstance(event, GroupMessageEvent):
if (
_flmt_c.check(event.group_id)
and module not in ignore_rst_module
):
_flmt_c.start_cd(event.group_id)
await bot.send_group_msg(
group_id=event.group_id, message="此功能正在维护..."
)
else:
await bot.send_private_msg(
user_id=event.user_id, message="此功能正在维护..."
)
except ActionFailed:
pass
if event.is_tome():
id_ = (
event.group_id
if isinstance(event, GroupMessageEvent)
else event.user_id
)
status_message_manager.add(id_)
set_block_limit_false(event, module)
raise IgnoredException("此功能正在维护...")
# 以下为限制检测 #######################################################
# 以下为限制检测 #######################################################
# 以下为限制检测 #######################################################
# 以下为限制检测 #######################################################
# 以下为限制检测 #######################################################
# 以下为限制检测 #######################################################
# Cd
if plugins2cd_manager.check_plugin_cd_status(module):
plugin_cd_data = plugins2cd_manager.get_plugin_cd_data(module)
check_type = plugin_cd_data["check_type"]
limit_type = plugin_cd_data["limit_type"]
rst = plugin_cd_data["rst"]
if (
(isinstance(event, PrivateMessageEvent) and check_type == "private")
or (isinstance(event, GroupMessageEvent) and check_type == "group")
or plugins2cd_manager.get_plugin_data(module).get("check_type") == "all"
):
cd_type_ = event.user_id
if limit_type == "group" and isinstance(event, GroupMessageEvent):
cd_type_ = event.group_id
if not plugins2cd_manager.check(module, cd_type_):
if rst:
rst = await init_rst(rst, event)
await send_msg(rst, bot, event)
raise IgnoredException(f"{module} 正在cd中...")
else:
plugins2cd_manager.start_cd(module, cd_type_)
# Block
if plugins2block_manager.check_plugin_block_status(module):
plugin_block_data = plugins2block_manager.get_plugin_block_data(module)
check_type = plugin_block_data["check_type"]
limit_type = plugin_block_data["limit_type"]
rst = plugin_block_data["rst"]
if (
(isinstance(event, PrivateMessageEvent) and check_type == "private")
or (isinstance(event, GroupMessageEvent) and check_type == "group")
or check_type == "all"
):
block_type_ = event.user_id
if limit_type == "group" and isinstance(event, GroupMessageEvent):
block_type_ = event.group_id
if plugins2block_manager.check(block_type_, module):
if rst:
rst = await init_rst(rst, event)
await send_msg(rst, bot, event)
raise IgnoredException(f"{event.user_id}正在调用{module}....")
else:
plugins2block_manager.set_true(block_type_, module)
# Count
if (
plugins2count_manager.check_plugin_count_status(module)
and event.user_id not in bot.config.superusers
):
plugin_count_data = plugins2count_manager.get_plugin_count_data(module)
limit_type = plugin_count_data["limit_type"]
rst = plugin_count_data["rst"]
count_type_ = event.user_id
if limit_type == "group" and isinstance(event, GroupMessageEvent):
count_type_ = event.group_id
if not plugins2count_manager.check(module, count_type_):
if rst:
rst = await init_rst(rst, event)
await send_msg(rst, bot, event)
raise IgnoredException(f"{module} count次数限制...")
else:
plugins2count_manager.increase(module, count_type_)
# 功能花费的金币 #######################################
# 功能花费的金币 #######################################
if cost_gold:
await BagUser.spend_gold(event.user_id, event.group_id, cost_gold)
async def send_msg(rst: str, bot: Bot, event: MessageEvent):
"""
发送信息
:param rst: pass
:param bot: pass
:param event: pass
"""
rst = await init_rst(rst, event)
try:
if isinstance(event, GroupMessageEvent):
status_message_manager.add(event.group_id)
await bot.send_group_msg(group_id=event.group_id, message=Message(rst))
else:
status_message_manager.add(event.user_id)
await bot.send_private_msg(user_id=event.user_id, message=Message(rst))
except ActionFailed:
pass
# 解除命令block阻塞
@run_postprocessor
async def _(
matcher: Matcher,
exception: Optional[Exception],
bot: Bot,
event: Event,
state: T_State,
):
if not isinstance(event, MessageEvent) and matcher.module != "poke":
return
module = matcher.module
set_block_limit_false(event, module)
async def init_rst(rst: str, event: MessageEvent):
if "[uname]" in rst:
uname = event.sender.card if event.sender.card else event.sender.nickname
rst = rst.replace("[uname]", uname)
if "[nickname]" in rst:
if isinstance(event, GroupMessageEvent):
nickname = await GroupInfoUser.get_group_member_nickname(
event.user_id, event.group_id
)
else:
nickname = await FriendUser.get_friend_nickname(event.user_id)
rst = rst.replace("[nickname]", nickname)
if "[at]" in rst and isinstance(event, GroupMessageEvent):
rst = rst.replace("[at]", str(at(event.user_id)))
return rst
+86
View File
@@ -0,0 +1,86 @@
from nonebot.matcher import Matcher
from nonebot.message import run_preprocessor, IgnoredException
from nonebot.adapters.cqhttp.exception import ActionFailed
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import (
Bot,
MessageEvent,
GroupMessageEvent,
)
from configs.config import Config
from models.ban_user import BanUser
from utils.utils import is_number, static_flmt, FreqLimiter
from utils.message_builder import at
Config.add_plugin_config(
"hook",
"BAN_RESULT",
"才不会给你发消息.",
help_="对被ban用户发送的消息",
)
_flmt = FreqLimiter(300)
# 检查是否被ban
@run_preprocessor
async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
try:
if (
await BanUser.is_super_ban(event.user_id)
and str(event.user_id) not in bot.config.superusers
):
raise IgnoredException("用户处于超级黑名单中")
except AttributeError:
pass
if not isinstance(event, MessageEvent):
return
if matcher.type == "message" and matcher.priority not in [1, 9]:
if (
await BanUser.is_ban(event.user_id)
and str(event.user_id) not in bot.config.superusers
):
time = await BanUser.check_ban_time(event.user_id)
if is_number(time):
time = abs(int(time))
if time < 60:
time = str(time) + " 秒"
else:
time = str(int(time / 60)) + " 分钟"
else:
time = str(time) + " 分钟"
if isinstance(event, GroupMessageEvent):
if not static_flmt.check(event.user_id):
raise IgnoredException("用户处于黑名单中")
static_flmt.start_cd(event.user_id)
if matcher.priority != 9:
try:
ban_result = Config.get_config("hook", "BAN_RESULT")
if ban_result and _flmt.check(event.user_id):
_flmt.start_cd(event.user_id)
await bot.send_group_msg(
group_id=event.group_id,
message=at(event.user_id)
+ ban_result
+ f" 在..在 {time} 后才会理你喔",
)
except ActionFailed:
pass
else:
if not static_flmt.check(event.user_id):
raise IgnoredException("用户处于黑名单中")
static_flmt.start_cd(event.user_id)
if matcher.priority != 9:
try:
ban_result = Config.get_config("hook", "BAN_RESULT")
if ban_result:
await bot.send_private_msg(
user_id=event.user_id,
message=at(event.user_id)
+ ban_result
+ f" 在..在 {time}后才会理你喔",
)
except ActionFailed:
pass
raise IgnoredException("用户处于黑名单中")
+54
View File
@@ -0,0 +1,54 @@
from nonebot.matcher import Matcher
from nonebot.message import run_preprocessor, IgnoredException
from nonebot.adapters.cqhttp.exception import ActionFailed
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import (
Bot,
MessageEvent,
GroupMessageEvent,
)
from configs.config import Config
from models.ban_user import BanUser
from utils.utils import BanCheckLimiter
from utils.message_builder import at
from services.log import logger
_blmt = BanCheckLimiter(
Config.get_config("hook", "MALICIOUS_CHECK_TIME"),
Config.get_config("hook", "MALICIOUS_BAN_COUNT"),
)
# 恶意触发命令检测
@run_preprocessor
async def _(matcher: Matcher, bot: Bot, event: GroupMessageEvent, state: T_State):
if not isinstance(event, MessageEvent):
return
if matcher.type == "message" and matcher.priority not in [1, 9]:
if state["_prefix"]["raw_command"]:
if _blmt.check(f'{event.user_id}{state["_prefix"]["raw_command"]}'):
if await BanUser.ban(
event.user_id,
9,
Config.get_config("hook", "MALICIOUS_BAN_TIME") * 60,
):
logger.info(f"USER {event.user_id} 触发了恶意触发检测")
if isinstance(event, GroupMessageEvent):
try:
await bot.send_group_msg(
group_id=event.group_id,
message=at(event.user_id) + "检测到恶意触发命令,您将被封禁 30 分钟",
)
except ActionFailed:
pass
else:
try:
await bot.send_private_msg(
user_id=event.user_id,
message=at(event.user_id) + "检测到恶意触发命令,您将被封禁 30 分钟",
)
except ActionFailed:
pass
raise IgnoredException("检测到恶意触发命令")
_blmt.add(f'{event.user_id}{state["_prefix"]["raw_command"]}')
+41
View File
@@ -0,0 +1,41 @@
from nonebot.matcher import Matcher
from nonebot.message import run_preprocessor, IgnoredException
from nonebot.typing import T_State
from .utils import status_message_manager
from nonebot.adapters.cqhttp import (
Bot,
MessageEvent,
PrivateMessageEvent,
GroupMessageEvent,
)
# 为什么AI会自己和自己聊天
@run_preprocessor
async def _(matcher: Matcher, bot: Bot, event: PrivateMessageEvent, state: T_State):
if not isinstance(event, MessageEvent):
return
if event.user_id == int(bot.self_id):
raise IgnoredException("为什么AI会自己和自己聊天")
# 有命令就别说话了
@run_preprocessor
async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
if not isinstance(event, MessageEvent):
return
if matcher.type == "message" and matcher.module == "ai":
if (
isinstance(event, GroupMessageEvent)
and not status_message_manager.check(event.group_id)
):
status_message_manager.delete(event.group_id)
raise IgnoredException("有命令就别说话了")
elif (
isinstance(event, PrivateMessageEvent)
and not status_message_manager.check(event.user_id)
):
print(status_message_manager)
status_message_manager.delete(event.user_id)
raise IgnoredException("有命令就别说话了")
+48
View File
@@ -0,0 +1,48 @@
from nonebot.adapters.cqhttp import GroupMessageEvent, PrivateMessageEvent
from utils.manager import plugins2block_manager, StaticData
import time
class StatusMessageManager(StaticData):
def __init__(self):
super().__init__(None)
def add(self, id_: int):
self._data[id_] = time.time()
def delete(self, id_: int):
if self._data.get(id_):
del self._data[id_]
def check(self, id_: int, t: int = 30) -> bool:
if self._data.get(id_):
if time.time() - self._data[id_] > t:
del self._data[id_]
return True
return False
return True
status_message_manager = StatusMessageManager()
def set_block_limit_false(event, module):
"""
设置用户block为false
:param event: event
:param module: 插件模块
"""
if plugins2block_manager.check_plugin_block_status(module):
plugin_block_data = plugins2block_manager.get_plugin_block_data(module)
check_type = plugin_block_data["check_type"]
limit_type = plugin_block_data["limit_type"]
if not (
(isinstance(event, GroupMessageEvent) and check_type == "private")
or (isinstance(event, PrivateMessageEvent) and check_type == "group")
):
block_type_ = event.user_id
if limit_type == "group" and isinstance(event, GroupMessageEvent):
block_type_ = event.group_id
plugins2block_manager.set_false(block_type_, module)
+28
View File
@@ -0,0 +1,28 @@
from nonebot.matcher import Matcher
from nonebot.message import run_postprocessor
from typing import Optional
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, Event
from utils.manager import withdraw_message_manager
import asyncio
# 消息撤回
@run_postprocessor
async def _(
matcher: Matcher,
exception: Optional[Exception],
bot: Bot,
event: Event,
state: T_State,
):
tasks = []
for id_, time in withdraw_message_manager.data:
tasks.append(asyncio.ensure_future(_withdraw_message(bot, id_, time)))
withdraw_message_manager.remove((id_, time))
await asyncio.gather(*tasks)
async def _withdraw_message(bot: Bot, id_: int, time: int):
await asyncio.sleep(time)
await bot.delete_msg(message_id=id_, self_id=int(bot.self_id))
+61
View File
@@ -0,0 +1,61 @@
from .init_group_manager import init_group_manager, group_manager
from .init_plugins_config import init_plugins_config
from .init_plugins_data import init_plugins_data, plugins_manager
from .init_none_plugin_count_manager import init_none_plugin_count_manager
from .init_plugins_resources import init_plugins_resources
from .init_plugins_settings import init_plugins_settings
from .init_plugins_limit import (
init_plugins_block_limit,
init_plugins_count_limit,
init_plugins_cd_limit,
)
from .init import init
from .check_plugin_status import check_plugin_status
from utils.manager import admin_manager
from nonebot.adapters.cqhttp import Bot
from configs.path_config import DATA_PATH
from services.log import logger
from pathlib import Path
from nonebot import Driver
import nonebot
__zx_plugin_name__ = "初始化插件数据 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
driver: Driver = nonebot.get_driver()
@driver.on_startup
def _():
"""
初始化数据
"""
_flag = False
config_file = Path(DATA_PATH) / "configs" / "plugins2config.yaml"
if not config_file.exists():
_flag = True
init()
init_plugins_settings(DATA_PATH)
init_plugins_cd_limit(DATA_PATH)
init_plugins_block_limit(DATA_PATH)
init_plugins_count_limit(DATA_PATH)
init_plugins_data(DATA_PATH)
init_plugins_config(DATA_PATH)
init_plugins_resources()
init_none_plugin_count_manager()
x = group_manager.get_super_old_data()
if x:
for key in x.keys():
plugins_manager.block_plugin(key, block_type=x[key])
if _flag:
raise Exception("首次运行,已在configs目录下生成配置文件config.yaml,修改后重启即可...")
logger.info("初始化数据完成...")
@driver.on_bot_connect
async def _(bot: Bot):
await init_group_manager()
await check_plugin_status(bot)
+18
View File
@@ -0,0 +1,18 @@
from utils.manager import plugins_manager
from nonebot.adapters.cqhttp import Bot
async def check_plugin_status(bot: Bot):
"""
遍历查看插件加载情况
"""
rst = ""
for plugin in plugins_manager.keys():
data = plugins_manager.get(plugin)
if data.get("error") or data.get("error") is None:
rst += f'{plugin}:{data["plugin_name"]}\n'
if rst:
rst = "以下插件加载失败..\n" + rst
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]), message=rst[:-1]
)
+14
View File
@@ -0,0 +1,14 @@
from utils.manager import plugins2settings_manager
def init():
if plugins2settings_manager.get("update_pic"):
plugins2settings_manager["update_picture"] = plugins2settings_manager["update_pic"]
plugins2settings_manager.delete("update_pic")
if plugins2settings_manager.get("white2black_img"):
plugins2settings_manager["white2black_image"] = plugins2settings_manager["white2black_img"]
plugins2settings_manager.delete("white2black_img")
if plugins2settings_manager.get("send_img"):
plugins2settings_manager["send_image"] = plugins2settings_manager["send_img"]
plugins2settings_manager.delete("send_img")
+73
View File
@@ -0,0 +1,73 @@
from pathlib import Path
from utils.manager import group_manager
from services.db_context import db
from asyncpg.exceptions import DuplicateColumnError
from services.log import logger
try:
import ujson as json
except ModuleNotFoundError:
import json
try:
from models.group_remind import GroupRemind
except ModuleNotFoundError:
pass
async def init_group_manager():
"""
旧数据格式替换为新格式
初始化数据
"""
old_group_level_file = Path() / "data" / "manager" / "group_level.json"
old_plugin_list_file = Path() / "data" / "manager" / "plugin_list.json"
if old_group_level_file.exists():
data = json.load(open(old_group_level_file, "r", encoding="utf8"))
for key in data.keys():
group = key
level = data[key]
group_manager.set_group_level(group, level)
old_group_level_file.unlink()
group_manager.save()
if old_plugin_list_file.exists():
data = json.load(open(old_plugin_list_file, "r", encoding="utf8"))
for plugin in data.keys():
for group in data[plugin].keys():
if group == "default" and not data[plugin]["default"]:
group_manager.block_plugin(plugin)
elif not data[plugin][group]:
group_manager.block_plugin(plugin, group)
old_plugin_list_file.unlink()
old_data_table = Path() / "models" / "group_remind.py"
try:
if old_data_table.exists():
b = {
"hy": "group_welcome",
"kxcz": "open_case_reset_remind",
"zwa": "zwa",
"blpar": "bilibili_parse",
"epic": "epic_free_game",
"pa": "pa",
"almanac": "genshin_alc",
}
for group in group_manager.get_data()["group_manager"]:
for remind in b:
try:
status = await GroupRemind.get_status(int(group), remind)
if status is not None:
if status:
await group_manager.open_group_task(group, b[remind])
logger.info(f"读取旧数据-->{group} 开启 {b[remind]}")
else:
await group_manager.close_group_task(group, b[remind])
logger.info(f"读取旧数据-->{group} 关闭 {b[remind]}")
except Exception as e:
pass
query = db.text("DROP TABLE group_reminds;")
await db.first(query)
old_data_table.unlink()
logger.info("旧数据读取完毕,删除了舍弃表 group_reminds...")
except (ModuleNotFoundError, DuplicateColumnError):
pass
group_manager.save()
@@ -0,0 +1,45 @@
from utils.manager import (
none_plugin_count_manager,
plugins2count_manager,
plugins2cd_manager,
plugins2settings_manager,
plugins2block_manager,
plugins_manager,
resources_manager
)
from services.log import logger
from utils.utils import get_matchers
try:
import ujson as json
except ModuleNotFoundError:
import json
def init_none_plugin_count_manager():
"""
清除已删除插件数据
"""
modules = [x.module for x in get_matchers()]
for module in none_plugin_count_manager.keys():
if module not in modules:
none_plugin_count_manager.add_count(module)
else:
none_plugin_count_manager.reset(module)
if none_plugin_count_manager.check(module):
try:
plugin_name = plugins_manager.get(module)["plugin_name"]
except (AttributeError, KeyError):
plugin_name = ""
try:
plugins2settings_manager.delete(module)
plugins2count_manager.delete(module)
plugins2cd_manager.delete(module)
plugins2block_manager.delete(module)
plugins_manager.delete(module)
resources_manager.remove_resource(module)
logger.info(f"{module}:{plugin_name} 插件疑似已删除,清除对应插件数据...")
except Exception as e:
logger.error(
f"{module}:{plugin_name} 插件疑似已删除,清除对应插件数据失败...{type(e)}:{e}"
)
+150
View File
@@ -0,0 +1,150 @@
from pathlib import Path
from ruamel.yaml import round_trip_load, round_trip_dump, YAML
from utils.manager import admin_manager, plugins_manager
from configs.config import Config
from services.log import logger
from utils.utils import get_matchers
from ruamel import yaml
import nonebot
_yaml = YAML(typ="safe")
def init_plugins_config(data_path):
"""
初始化插件数据配置
"""
plugins2config_file = Path(data_path) / "configs" / "plugins2config.yaml"
plugins2config_file.parent.mkdir(parents=True, exist_ok=True)
_data = {}
if plugins2config_file.exists():
_data = _yaml.load(open(plugins2config_file, "r", encoding="utf8"))
_matchers = get_matchers()
for matcher in _matchers:
_plugin = nonebot.plugin.get_plugin(matcher.module)
try:
_module = _plugin.module
except AttributeError:
continue
try:
plugin_version = _module.__getattribute__("__plugin_version__")
except AttributeError:
plugin_version = None
try:
plugin_configs = _module.__getattribute__("__plugin_configs__")
except AttributeError:
continue
# 插件配置版本更新或为Version为None或不在存储配置内
if (
plugin_version is None
or (
_data.get(matcher.module)
and _data[matcher.module].keys() != plugin_configs.keys()
)
or plugin_version > plugins_manager.get(matcher.module)["version"]
or matcher.module not in _data.keys()
):
for key in plugin_configs:
if isinstance(plugin_configs[key], dict):
Config.add_plugin_config(
matcher.module,
key,
plugin_configs[key].get("value"),
help_=plugin_configs[key].get("help"),
default_value=plugin_configs[key].get("default_value"),
_override=True,
)
else:
Config.add_plugin_config(matcher.module, key, plugin_configs[key])
else:
plugin_configs = _data[matcher.module]
for key in plugin_configs:
Config.add_plugin_config(
matcher.module,
key,
plugin_configs[key]["value"],
help_=plugin_configs[key]["help"],
default_value=plugin_configs[key]["default_value"],
_override=True,
)
if not Config.is_empty():
Config.save()
_data = round_trip_load(open(plugins2config_file, encoding="utf8"))
for plugin in _data.keys():
try:
plugin_name = plugins_manager.get(plugin)["plugin_name"]
except (AttributeError, TypeError):
plugin_name = plugin
_data[plugin].yaml_set_start_comment(plugin_name, indent=2)
# 初始化未设置的管理员权限等级
for k, v in Config.get_admin_level_data():
admin_manager.set_admin_level(k, v)
# 存完插件基本设置
with open(plugins2config_file, "w", encoding="utf8") as wf:
round_trip_dump(
_data, wf, indent=2, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
# 再开始读取用户配置
user_config_file = Path() / "configs" / "config.yaml"
_data = {}
_tmp_data = {}
if user_config_file.exists():
with open(user_config_file, "r", encoding="utf8") as f:
_data = _yaml.load(f)
# 数据替换
for plugin in Config.keys():
_tmp_data[plugin] = {}
for k in Config[plugin].keys():
if _data.get(plugin) and k in _data[plugin].keys():
Config.set_config(plugin, k, _data[plugin][k])
if level2module := Config.get_level2module(plugin, k):
admin_manager.set_admin_level(level2module, _data[plugin][k])
_tmp_data[plugin][k] = Config.get_config(plugin, k)
Config.save()
temp_file = Path() / "configs" / "temp_config.yaml"
try:
with open(temp_file, "w", encoding="utf8") as wf:
yaml.dump(
_tmp_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
with open(temp_file, "r", encoding="utf8") as rf:
_data = round_trip_load(rf)
# 添加注释
for plugin in _data.keys():
rst = ""
plugin_name = None
try:
plugin_data = Config.get(plugin)
for x in list(Config.get(plugin).keys()):
try:
_x = plugin_data[x].get("name")
if _x:
plugin_name = _x
except AttributeError:
pass
except (KeyError, AttributeError):
plugin_name = None
if not plugin_name:
try:
plugin_name = plugins_manager.get(plugin)["plugin_name"]
except (AttributeError, TypeError):
plugin_name = plugin
plugin_name = (
plugin_name.replace("[Hidden]", "")
.replace("[Superuser]", "")
.replace("[Admin]", "")
.strip()
)
rst += plugin_name + "\n"
for k in _data[plugin].keys():
rst += f'{k}: {Config[plugin][k]["help"]}' + "\n"
_data[plugin].yaml_set_start_comment(rst[:-1], indent=2)
with open(Path() / "configs" / "config.yaml", "w", encoding="utf8") as wf:
round_trip_dump(
_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
except Exception as e:
logger.error(f"生成简易配置注释错误 {type(e)}:{e}")
if temp_file.exists():
temp_file.unlink()
+82
View File
@@ -0,0 +1,82 @@
from pathlib import Path
from ruamel.yaml import YAML
from utils.manager import plugins_manager
from utils.utils import get_matchers
import nonebot
try:
import ujson as json
except ModuleNotFoundError:
import json
_yaml = YAML(typ="safe")
def init_plugins_data(data_path):
"""
初始化插件数据信息
"""
plugin2data_file = Path(data_path) / "manager" / "plugin_manager.json"
plugin2data_file.parent.mkdir(parents=True, exist_ok=True)
_data = {}
if plugin2data_file.exists():
_data = json.load(open(plugin2data_file, "r", encoding="utf8"))
_matchers = get_matchers()
for matcher in _matchers:
_plugin = nonebot.plugin.get_plugin(matcher.module)
try:
_module = _plugin.module
except AttributeError:
if matcher.module not in _data.keys():
plugins_manager.add_plugin_data(
matcher.module, matcher.module, error=True
)
else:
plugins_manager.set_module_data(matcher.module, "error", True)
plugin_data = plugins_manager.get(matcher.module)
if plugin_data:
plugins_manager.set_module_data(
matcher.module, "version", plugin_data.get("version")
)
else:
try:
plugin_version = _module.__getattribute__("__plugin_version__")
except AttributeError:
plugin_version = None
try:
plugin_name = _module.__getattribute__("__zx_plugin_name__")
except AttributeError:
plugin_name = matcher.module
try:
plugin_author = _module.__getattribute__("__plugin_author__")
except AttributeError:
plugin_author = None
if matcher.module in plugins_manager.keys():
plugins_manager.set_module_data(matcher.module, "error", False)
if matcher.module not in plugins_manager.keys():
plugins_manager.add_plugin_data(
matcher.module,
plugin_name=plugin_name,
author=plugin_author,
version=plugin_version,
)
elif plugins_manager[matcher.module]["version"] is None or (
plugin_version is not None
and plugin_version > plugins_manager[matcher.module]["version"]
):
plugins_manager.set_module_data(
matcher.module, "plugin_name", plugin_name
)
plugins_manager.set_module_data(matcher.module, "author", plugin_author)
plugins_manager.set_module_data(
matcher.module, "version", plugin_version
)
if matcher.module in _data.keys():
plugins_manager.set_module_data(
matcher.module, "error", _data[matcher.module]["error"]
)
plugins_manager.set_module_data(
matcher.module, "plugin_name", _data[matcher.module]["plugin_name"]
)
plugins_manager.save()
+157
View File
@@ -0,0 +1,157 @@
from pathlib import Path
from ruamel.yaml import round_trip_load, round_trip_dump, YAML
from utils.manager import (
plugins2cd_manager,
plugins2block_manager,
plugins2count_manager,
)
from utils.utils import get_matchers
from ruamel import yaml
import nonebot
_yaml = YAML(typ="safe")
def init_plugins_cd_limit(data_path):
"""
加载 cd 限制
"""
plugins2cd_file = Path(data_path) / "configs" / "plugins2cd.yaml"
plugins2cd_file.parent.mkdir(exist_ok=True, parents=True)
_data = {}
_matchers = get_matchers()
for matcher in _matchers:
if not plugins2cd_manager.get_plugin_cd_data(matcher.module):
_plugin = nonebot.plugin.get_plugin(matcher.module)
try:
_module = _plugin.module
plugin_cd_limit = _module.__getattribute__("__plugin_cd_limit__")
plugins2cd_manager.add_cd_limit(
matcher.module, data_dict=plugin_cd_limit
)
except AttributeError:
pass
if not plugins2cd_manager.keys():
plugins2cd_manager.add_cd_limit(
"这是一个示例"
)
_tmp_data = {"PluginCdLimit": plugins2cd_manager.get_data()}
with open(plugins2cd_file, "w", encoding="utf8") as wf:
yaml.dump(_tmp_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
_data = round_trip_load(open(plugins2cd_file, encoding="utf8"))
_data["PluginCdLimit"].yaml_set_start_comment(
"""# 需要cd的功能
# 自定义的功能需要cd也可以在此配置
# key:模块名称
# cd:cd 时长(秒)
# status:此限制的开关状态
# check_type:'private'/'group'/'all',限制私聊/群聊/全部
# limit_type:监听对象,以user_id或group_id作为键来限制,'user':用户id,'group':群id
# 示例:'user':用户N秒内触发1次,'group':群N秒内触发1次
# rst:回复的话,可以添加[at],[uname],[nickname]来对应艾特,用户群名称,昵称系统昵称
# rst 为 "" 或 None 时则不回复
# rst示例:"[uname]你冲的太快了,[nickname]先生,请稍后再冲[at]"
# rst回复:"老色批你冲的太快了,欧尼酱先生,请稍后再冲@老色批"
# 用户昵称↑ 昵称系统的昵称↑ 艾特用户↑""",
indent=2,
)
with open(plugins2cd_file, "w", encoding="utf8") as wf:
round_trip_dump(_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
plugins2cd_manager.reload_cd_limit()
def init_plugins_block_limit(data_path):
"""
加载阻塞限制
"""
plugins2block_file = Path(data_path) / "configs" / "plugins2block.yaml"
plugins2block_file.parent.mkdir(exist_ok=True, parents=True)
_data = {}
_matchers = get_matchers()
for matcher in _matchers:
if not plugins2block_manager.get_plugin_block_data(matcher.module):
_plugin = nonebot.plugin.get_plugin(matcher.module)
try:
_module = _plugin.module
plugin_block_limit = _module.__getattribute__("__plugin_block_limit__")
plugins2block_manager.add_block_limit(
matcher.module, data_dict=plugin_block_limit
)
except AttributeError:
pass
if not plugins2block_manager.keys():
plugins2block_manager.add_block_limit(
"这是一个示例"
)
_tmp_data = {"PluginBlockLimit": plugins2block_manager.get_data()}
with open(plugins2block_file, "w", encoding="utf8") as wf:
yaml.dump(_tmp_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
_data = round_trip_load(open(plugins2block_file, encoding="utf8"))
_data["PluginBlockLimit"].yaml_set_start_comment(
"""# 用户调用阻塞
# 即 当用户调用此功能还未结束时
# 用发送消息阻止用户重复调用此命令直到该命令结束
# key:模块名称
# status:此限制的开关状态
# check_type:'private'/'group'/'all',限制私聊/群聊/全部
# limit_type:监听对象,以user_id或group_id作为键来限制,'user':用户id,'group':群id
# 示例:'user':阻塞用户,'group':阻塞群聊
# rst:回复的话,可以添加[at],[uname],[nickname]来对应艾特,用户群名称,昵称系统昵称
# rst 为 "" 或 None 时则不回复
# rst示例:"[uname]你冲的太快了,[nickname]先生,请稍后再冲[at]"
# rst回复:"老色批你冲的太快了,欧尼酱先生,请稍后再冲@老色批"
# 用户昵称↑ 昵称系统的昵称↑ 艾特用户↑""",
indent=2,
)
with open(plugins2block_file, "w", encoding="utf8") as wf:
round_trip_dump(_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
plugins2block_manager.reload_block_limit()
def init_plugins_count_limit(data_path):
"""
加载次数限制
"""
plugins2count_file = Path(data_path) / "configs" / "plugins2count.yaml"
plugins2count_file.parent.mkdir(exist_ok=True, parents=True)
_data = {}
_matchers = get_matchers()
for matcher in _matchers:
if not plugins2count_manager.get_plugin_count_data(matcher.module):
_plugin = nonebot.plugin.get_plugin(matcher.module)
try:
_module = _plugin.module
plugin_count_limit = _module.__getattribute__("__plugin_count_limit__")
plugins2count_manager.add_count_limit(
matcher.module, data_dict=plugin_count_limit
)
except AttributeError:
pass
if not plugins2count_manager.keys():
plugins2count_manager.add_count_limit(
"这是一个示例"
)
_tmp_data = {"PluginCountLimit": plugins2count_manager.get_data()}
with open(plugins2count_file, "w", encoding="utf8") as wf:
yaml.dump(_tmp_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
_data = round_trip_load(open(plugins2count_file, encoding="utf8"))
_data["PluginCountLimit"].yaml_set_start_comment(
"""# 命令每日次数限制
# 即 用户/群聊 每日可调用命令的次数 [数据内存存储,重启将会重置]
# 每日调用直到 00:00 刷新
# key:模块名称
# max_count: 每日调用上限
# status:此限制的开关状态
# limit_type:监听对象,以user_id或group_id作为键来限制,'user':用户id,'group':群id
# 示例:'user':用户上限,'group':群聊上限
# rst:回复的话,可以添加[at],[uname],[nickname]来对应艾特,用户群名称,昵称系统昵称
# rst 为 "" 或 None 时则不回复
# rst示例:"[uname]你冲的太快了,[nickname]先生,请稍后再冲[at]"
# rst回复:"老色批你冲的太快了,欧尼酱先生,请稍后再冲@老色批"
# 用户昵称↑ 昵称系统的昵称↑ 艾特用户↑""",
indent=2,
)
with open(plugins2count_file, "w", encoding="utf8") as wf:
round_trip_dump(_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
plugins2count_manager.reload_count_limit()
+43
View File
@@ -0,0 +1,43 @@
from utils.manager import resources_manager
from utils.utils import get_matchers
from services.log import logger
from pathlib import Path
import nonebot
def init_plugins_resources():
"""
资源文件路径的移动
"""
_tmp = []
for matcher in get_matchers():
if matcher.module not in _tmp:
_tmp.append(matcher.module)
_plugin = nonebot.plugin.get_plugin(matcher.module)
try:
_module = _plugin.module
except AttributeError:
logger.warning(f"插件 {matcher.module} 加载失败...,资源控制未加载...")
else:
try:
resources = _module.__getattribute__("__plugin_resources__")
except AttributeError:
pass
else:
path = Path(_module.__getattribute__("__file__")).parent
for resource in resources.keys():
resources_manager.add_resource(matcher.module, (path / resource).absolute(), resources[resource])
resources_manager.save()
resources_manager.start_move()
+128
View File
@@ -0,0 +1,128 @@
from pathlib import Path
from ruamel.yaml import round_trip_load, round_trip_dump, YAML
from utils.manager import plugins2settings_manager, admin_manager
from services.log import logger
from utils.utils import get_matchers
from ruamel import yaml
import nonebot
_yaml = YAML(typ="safe")
def init_plugins_settings(data_path: str):
"""
初始化插件设置,从插件中获取 __zx_plugin_name__,__plugin_cmd__,__plugin_settings__
"""
plugins2settings_file = Path(data_path) / "configs" / "plugins2settings.yaml"
plugins2settings_file.parent.mkdir(exist_ok=True, parents=True)
_matchers = get_matchers()
_tmp_module = {}
_tmp = []
for x in plugins2settings_manager.keys():
try:
_plugin = nonebot.plugin.get_plugin(x)
_module = _plugin.module
plugin_name = _module.__getattribute__("__zx_plugin_name__")
_tmp_module[x] = plugin_name
except (KeyError, AttributeError) as e:
logger.warning(f"配置文件 模块:{x} 获取 plugin_name 失败...{e}")
_tmp_module[x] = ""
for matcher in _matchers:
if matcher.module not in plugins2settings_manager.keys():
_plugin = nonebot.plugin.get_plugin(matcher.module)
try:
_module = _plugin.module
except AttributeError:
logger.warning(f"插件 {matcher.module} 加载失败...,插件控制未加载.")
else:
try:
plugin_name = _module.__getattribute__("__zx_plugin_name__")
if "[admin]" in plugin_name.lower():
try:
admin_settings = _module.__getattribute__(
"__plugin_settings__"
)
level = admin_settings["admin_level"]
cmd = admin_settings.get("cmd")
except (AttributeError, KeyError):
level = 5
cmd = None
if level is None:
level = 5
admin_manager.add_admin_plugin_settings(
matcher.module, cmd, level
)
if (
"[hidden]" in plugin_name.lower()
or "[admin]" in plugin_name.lower()
or "[superuser]" in plugin_name.lower()
or matcher.module in plugins2settings_manager.keys()
):
continue
except AttributeError:
if matcher.module not in _tmp:
logger.warning(
f"获取插件 {matcher.module} __zx_plugin_name__ 失败...,插件控制未加载."
)
else:
try:
_tmp_module[matcher.module] = plugin_name
plugin_settings = _module.__getattribute__(
"__plugin_settings__"
)
if plugin_settings.get('cost_gold') is None:
plugin_settings['cost_gold'] = 0
if (
plugin_settings["cmd"] is not None
and plugin_name not in plugin_settings["cmd"]
):
plugin_settings["cmd"].append(plugin_name)
if plugins2settings_manager.get(
matcher.module
) and plugins2settings_manager[matcher.module].get(
"plugin_type"
):
plugin_type = tuple(
plugins2settings_manager.get_plugin_data(
matcher.module
)["plugin_type"]
)
else:
try:
plugin_type = _module.__getattribute__(
"__plugin_type__"
)
except AttributeError:
plugin_type = ("normal",)
if plugin_settings and matcher.module:
plugins2settings_manager.add_plugin_settings(
matcher.module,
plugin_type=plugin_type,
**plugin_settings,
)
except AttributeError:
pass
_tmp.append(matcher.module)
_tmp_data = {"PluginSettings": plugins2settings_manager.get_data()}
with open(plugins2settings_file, "w", encoding="utf8") as wf:
yaml.dump(_tmp_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
_data = round_trip_load(open(plugins2settings_file, encoding="utf8"))
_data["PluginSettings"].yaml_set_start_comment(
"""# 模块与对应命令和对应群权限
# 用于生成帮助图片 和 开关功能
# key:模块名称
# level:需要的群等级
# default_status:加入群时功能的默认开关状态
# limit_superuser: 功能状态是否限制超级用户
# cmd: 关闭[cmd] 都会触发命令 关闭对应功能,cmd列表第一个词为统计的功能名称
# plugin_type: 帮助类别 示例:('原神相关',) 或 ('原神相关', 1),1代表帮助命令列向排列,否则为横向排列""",
indent=2,
)
for plugin in _data["PluginSettings"].keys():
_data["PluginSettings"][plugin].yaml_set_start_comment(
f"{plugin}:{_tmp_module[plugin]}", indent=2
)
with open(plugins2settings_file, "w", encoding="utf8") as wf:
round_trip_dump(_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
logger.info(f"已成功加载 {len(plugins2settings_manager.get_data())} 个非限制插件.")
+149
View File
@@ -0,0 +1,149 @@
from nonebot import on_request, on_message
from nonebot.adapters.cqhttp import (
Bot,
FriendRequestEvent,
GroupRequestEvent,
MessageEvent,
)
from models.friend_user import FriendUser
from datetime import datetime
from configs.config import NICKNAME, Config
from nonebot.adapters.cqhttp.exception import ActionFailed
from utils.manager import requests_manager
from models.group_info import GroupInfo
from utils.utils import scheduler
import asyncio
import time
import re
__zx_plugin_name__ = "好友群聊处理请求 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_configs__ = {
"AUTO_ADD_FRIEND": {"value": False, "help": "是否自动同意好友添加", "default_value": False}
}
friend_req = on_request(priority=5, block=True)
group_req = on_request(priority=5, block=True)
x = on_message(priority=9, block=False)
exists_data = {"private": {}, "group": {}}
@friend_req.handle()
async def _(bot: Bot, event: FriendRequestEvent, state: dict):
global exists_data
if exists_data["private"].get(event.user_id):
if time.time() - exists_data["private"][event.user_id] < 60 * 5:
return
exists_data["private"][event.user_id] = time.time()
user = await bot.get_stranger_info(user_id=event.user_id)
nickname = user["nickname"]
sex = user["sex"]
age = str(user["age"])
comment = event.comment
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]),
message=f"*****一份好友申请*****\n"
f"昵称:{nickname}({event.user_id})\n"
f"自动同意:{'√' if Config.get_config('invite_manager', 'AUTO_ADD_FRIEND') else '×'}\n"
f"日期:{str(datetime.now()).split('.')[0]}\n"
f"备注:{event.comment}",
)
if Config.get_config("invite_manager", "AUTO_ADD_FRIEND"):
await bot.set_friend_add_request(flag=event.flag, approve=True)
await FriendUser.add_friend_info(user["user_id"], user["nickname"])
else:
requests_manager.add_request(
event.user_id,
"private",
event.flag,
nickname=nickname,
sex=sex,
age=age,
comment=comment,
)
@group_req.handle()
async def _(bot: Bot, event: GroupRequestEvent, state: dict):
global exists_data
if event.sub_type == "invite":
if str(event.user_id) in bot.config.superusers:
try:
if await GroupInfo.get_group_info(event.group_id):
await GroupInfo.set_group_flag(event.group_id, 1)
else:
group_info = await bot.get_group_info(group_id=event.group_id)
await GroupInfo.add_group_info(
group_info["group_id"],
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
1,
)
await bot.set_group_add_request(
flag=event.flag, sub_type="invite", approve=True
)
except ActionFailed:
pass
else:
user = await bot.get_stranger_info(user_id=event.user_id)
sex = user["sex"]
age = str(user["age"])
if exists_data["group"].get(f"{event.user_id}:{event.group_id}"):
if (
time.time()
- exists_data["group"][f"{event.user_id}:{event.group_id}"]
< 60 * 5
):
return
exists_data["group"][f"{event.user_id}:{event.group_id}"] = time.time()
nickname = await FriendUser.get_user_name(event.user_id)
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]),
message=f"*****一份入群申请*****\n"
f"申请人:{nickname}({event.user_id})\n"
f"群聊:{event.group_id}\n"
f"邀请日期:{str(datetime.now()).split('.')[0]}",
)
await bot.send_private_msg(
user_id=event.user_id,
message=f"想要邀请我偷偷入群嘛~已经提醒{NICKNAME}的管理员大人了\n"
"请确保已经群主或群管理沟通过!\n"
"等待管理员处理吧!",
)
requests_manager.add_request(
event.user_id,
"group",
event.flag,
invite_group=event.group_id,
nickname=nickname,
sex=sex,
age=age,
)
@x.handle()
async def _(bot: Bot, event: MessageEvent, state: dict):
await asyncio.sleep(0.1)
r = re.search(r'groupcode="(.*?)"', str(event.get_message()))
if r:
group_id = int(r.group(1))
else:
return
r = re.search(r'groupname="(.*?)"', str(event.get_message()))
if r:
group_name = r.group(1)
else:
group_name = "None"
requests_manager.set_group_name(group_name, group_id)
@scheduler.scheduled_job(
"interval",
minutes=5,
)
async def _():
global exists_data
exists_data = {"private": {}, "group": {}}
+201
View File
@@ -0,0 +1,201 @@
from nonebot import on_command
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent, PrivateMessageEvent
from nonebot.rule import to_me
from utils.utils import get_message_text
from models.group_member_info import GroupInfoUser
from models.friend_user import FriendUser
import random
from models.ban_user import BanUser
from services.log import logger
from configs.config import NICKNAME, Config
__zx_plugin_name__ = "昵称系统"
__plugin_usage__ = f"""
usage:
个人昵称系统,群聊 与 私聊 昵称相互独立
指令:
以后叫我 [昵称]
{NICKNAME}我是谁
""".strip()
__plugin_des__ = "区区昵称,才不想叫呢!"
__plugin_cmd__ = ["以后叫我 [昵称]", f"{NICKNAME}我是谁"]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_configs__ = {
"BLACK_WORD": {
"value": ["爸", "爹", "爷"],
"help": "昵称所屏蔽的关键词,会被替换为 *",
"default_value": None
}
}
nickname = on_command(
"nickname",
aliases={"以后叫我", "以后请叫我", "称呼我", "以后请称呼我", "以后称呼我", "叫我", "请叫我"},
rule=to_me(),
priority=5,
block=True,
)
my_nickname = on_command(
"my_name", aliases={"我叫什么", "我是谁", "我的名字"}, rule=to_me(), priority=5, block=True
)
cancel_nickname = on_command("取消昵称", rule=to_me(), priority=5, block=True)
@nickname.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
msg = get_message_text(event.json())
if not msg:
await nickname.finish("叫你空白?叫你虚空?叫你无名??", at_sender=True)
if len(msg) > 10:
await nickname.finish("昵称可不能超过10个字!", at_sender=True)
if msg in bot.config.superusers:
await nickname.finish("笨蛋!休想占用我的名字!#", at_sender=True)
_tmp = ""
black_word = Config.get_config("nickname", "BLACK_WORD")
for x in msg:
_tmp += "*" if x in black_word else x
msg = _tmp
if isinstance(event, GroupMessageEvent):
if await GroupInfoUser.set_group_member_nickname(
event.user_id, event.group_id, msg
):
if len(msg) < 5:
if random.random() < 0.3:
msg = "~".join(msg)
await nickname.send(
random.choice(
[
f"好啦好啦,我知道啦,{msg},以后就这么叫你吧",
f"嗯嗯,{NICKNAME}记住你的昵称了哦,{msg}",
f"好突然,突然要叫你昵称什么的...{msg}..",
f"{NICKNAME}会好好记住的{msg}的,放心吧",
f"好..好.,那窝以后就叫你{msg}了.",
]
)
)
logger.info(f"USER {event.user_id} GROUP {event.group_id} 设置群昵称 {msg}")
else:
await nickname.send("设置昵称失败,请更新群组成员信息!", at_sender=True)
logger.warning(f"USER {event.user_id} GROUP {event.group_id} 设置群昵称 {msg} 失败")
else:
if await FriendUser.set_friend_nickname(event.user_id, msg):
await nickname.send(
random.choice(
[
f"好啦好啦,我知道啦,{msg},以后就这么叫你吧",
f"嗯嗯,{NICKNAME}记住你的昵称了哦,{msg}",
f"好突然,突然要叫你昵称什么的...{msg}..",
f"{NICKNAME}会好好记住的{msg}的,放心吧",
f"好..好.,那窝以后就叫你{msg}了.",
]
)
)
logger.info(f"USER {event.user_id} 设置昵称 {msg}")
else:
await nickname.send("设置昵称失败了,明天再来试一试!或联系管理员更新好友!", at_sender=True)
logger.warning(f"USER {event.user_id} 设置昵称 {msg} 失败")
@my_nickname.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
try:
nickname_ = await GroupInfoUser.get_group_member_nickname(
event.user_id, event.group_id
)
except AttributeError:
nickname_ = ""
if nickname_:
await my_nickname.send(
random.choice(
[
f"我肯定记得你啊,你是{nickname_}啊",
f"我不会忘记你的,你也不要忘记我!{nickname_}",
f"哼哼,{NICKNAME}记忆力可是很好的,{nickname_}",
f"嗯?你是失忆了嘛...{nickname_}..",
f"不要小看{NICKNAME}的记忆力啊!笨蛋{nickname_}!QAQ",
f"哎?{nickname_}..怎么了吗..突然这样问..",
]
)
)
else:
nickname_ = event.sender.card if event.sender.card else event.sender.nickname
await my_nickname.send(
random.choice(
["没..没有昵称嘛,{}", "啊,你是{}啊,我想叫你的昵称!", "是{}啊,有什么事吗?", "你是{}?"]
).format(nickname_)
)
@my_nickname.handle()
async def _(bot: Bot, event: PrivateMessageEvent, state: T_State):
nickname_ = await FriendUser.get_friend_nickname(event.user_id)
if nickname_:
await my_nickname.send(
random.choice(
[
f"我肯定记得你啊,你是{nickname_}啊",
f"我不会忘记你的,你也不要忘记我!{nickname_}",
f"哼哼,{NICKNAME}记忆力可是很好的,{nickname_}",
f"嗯?你是失忆了嘛...{nickname_}..",
f"不要小看{NICKNAME}的记忆力啊!笨蛋{nickname_}!QAQ",
f"哎?{nickname_}..怎么了吗..突然这样问..",
]
)
)
else:
nickname_ = (await bot.get_stranger_info(user_id=event.user_id))["nickname"]
await my_nickname.send(
random.choice(
["没..没有昵称嘛,{}", "啊,你是{}啊,我想叫你的昵称!", "是{}啊,有什么事吗?", "你是{}?"]
).format(nickname_)
)
@cancel_nickname.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
nickname_ = await GroupInfoUser.get_group_member_nickname(
event.user_id, event.group_id
)
if nickname_:
await cancel_nickname.send(
random.choice(
[
f"呜..{NICKNAME}睡一觉就会忘记的..和梦一样..{nickname_}",
f"窝知道了..{nickname_}..",
f"是{NICKNAME}哪里做的不好嘛..好吧..晚安{nickname_}",
f"呃,{nickname_},下次我绝对绝对绝对不会再忘记你!",
f"可..可恶!{nickname_}!太可恶了!呜",
]
)
)
await GroupInfoUser.set_group_member_nickname(event.user_id, event.group_id, "")
await BanUser.ban(event.user_id, 9, 60)
else:
await cancel_nickname.send("你在做梦吗?你没有昵称啊", at_sender=True)
@cancel_nickname.handle()
async def _(bot: Bot, event: PrivateMessageEvent, state: T_State):
nickname_ = await FriendUser.get_friend_nickname(event.user_id)
if nickname_:
await cancel_nickname.send(
random.choice(
[
f"呜..{NICKNAME}睡一觉就会忘记的..和梦一样..{nickname_}",
f"窝知道了..{nickname_}..",
f"是{NICKNAME}哪里做的不好嘛..好吧..晚安{nickname_}",
f"呃,{nickname_},下次我绝对绝对绝对不会再忘记你!",
f"可..可恶!{nickname_}!太可恶了!呜",
]
)
)
await FriendUser.get_user_name(event.user_id)
await BanUser.ban(event.user_id, 9, 60)
else:
await cancel_nickname.send("你在做梦吗?你没有昵称啊", at_sender=True)
+115
View File
@@ -0,0 +1,115 @@
from nonebot import Driver
from services.db_context import db
from asyncpg.exceptions import DuplicateColumnError
from models.group_info import GroupInfo
from nonebot.adapters.cqhttp import Bot
from services.log import logger
from configs.path_config import TEXT_PATH
from asyncio.exceptions import TimeoutError
from utils.http_utils import AsyncHttpx
from utils.utils import scheduler
from pathlib import Path
import nonebot
try:
import ujson as json
except ModuleNotFoundError:
import json
driver: Driver = nonebot.get_driver()
@driver.on_startup
async def update_city():
"""
部分插件需要中国省份城市
这里直接更新,避免插件内代码重复
"""
china_city = Path(TEXT_PATH) / "china_city.json"
data = {}
if not china_city.exists():
try:
res = await AsyncHttpx.get(
"http://www.weather.com.cn/data/city3jdata/china.html", timeout=5
)
res.encoding = "utf8"
provinces_data = json.loads(res.text)
for province in provinces_data.keys():
data[provinces_data[province]] = []
res = await AsyncHttpx.get(
f"http://www.weather.com.cn/data/city3jdata/provshi/{province}.html",
timeout=5,
)
res.encoding = "utf8"
city_data = json.loads(res.text)
for city in city_data.keys():
data[provinces_data[province]].append(city_data[city])
with open(china_city, "w", encoding="utf8") as f:
json.dump(data, f, indent=4, ensure_ascii=False)
logger.info("自动更新城市列表完成.....")
except TimeoutError:
logger.warning("自动更新城市列表超时.....")
except ValueError:
logger.warning("自动城市列表失败.....")
except Exception as e:
logger.error(f"自动城市列表未知错误 {type(e)}:{e}")
@driver.on_startup
async def _():
"""
数据库表结构变换
"""
sql_str = [
"ALTER TABLE group_info ADD group_flag Integer NOT NULL DEFAULT 0;" # group_info表添加一个group_flag
]
for sql in sql_str:
try:
query = db.text(sql)
await db.first(query)
except DuplicateColumnError:
pass
@driver.on_bot_connect
async def _(bot: Bot):
"""
版本某些需要的变换
"""
# 清空不存在的群聊信息,并将已所有已存在的群聊group_flag设置为1(认证所有已存在的群)
if not await GroupInfo.get_group_info(114514):
# 标识符,该功能只需执行一次
await GroupInfo.add_group_info(114514, "114514", 114514, 114514, 1)
group_list = await bot.get_group_list()
group_list = [g["group_id"] for g in group_list]
_gl = [x.group_id for x in await GroupInfo.get_all_group()]
if 114514 in _gl:
_gl.remove(114514)
for group_id in _gl:
if group_id in group_list:
if await GroupInfo.get_group_info(group_id):
await GroupInfo.set_group_flag(group_id, 1)
else:
group_info = await bot.get_group_info(group_id=group_id)
await GroupInfo.add_group_info(
group_info["group_id"],
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
1,
)
logger.info(f"已将群聊 {group_id} 添加认证...")
else:
await GroupInfo.delete_group_info(group_id)
logger.info(f"移除不存在的群聊信息:{group_id}")
# 自动更新城市列表
@scheduler.scheduled_job(
"cron",
hour=6,
minute=1,
)
async def _():
await update_city()
+14
View File
@@ -0,0 +1,14 @@
import nonebot
from configs.config import Config
Config.add_plugin_config(
"shop",
"IMPORT_DEFAULT_SHOP_GOODS",
True,
help_="导入商店自带的三个商品",
default_value=True
)
nonebot.load_plugins("basic_plugins/shop")
+86
View File
@@ -0,0 +1,86 @@
from nonebot import on_command
from services.log import logger
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent
from nonebot.typing import T_State
from utils.utils import get_message_text, is_number
from models.bag_user import BagUser
from services.db_context import db
from nonebot.adapters.cqhttp.permission import GROUP
from models.goods_info import GoodsInfo
__zx_plugin_name__ = "商店 - 购买道具"
__plugin_usage__ = """
usage:
购买道具
指令:
购买 [序号或名称] ?[数量=1]
示例:购买 好感双倍加持卡Ⅰ
示例:购买 1 4
""".strip()
__plugin_des__ = "商店 - 购买道具"
__plugin_cmd__ = ["购买 [序号或名称] ?[数量=1]"]
__plugin_type__ = ('商店',)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["商店", "购买道具"],
}
buy = on_command("购买", aliases={"购买道具"}, priority=5, block=True, permission=GROUP)
@buy.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
goods = None
if get_message_text(event.json()) in ["神秘药水"]:
await buy.finish("你们看看就好啦,这是不可能卖给你们的~", at_sender=True)
goods_lst = await GoodsInfo.get_all_goods()
goods_name_lst = [x.goods_name for x in goods_lst]
msg = get_message_text(event.json()).split()
num = 1
if len(msg) > 1:
if is_number(msg[1]) and int(msg[1]) > 0:
num = int(msg[1])
else:
await buy.finish("购买的数量要是数字且大于0!", at_sender=True)
# print(msg, num)
if is_number(msg[0]):
msg = int(msg[0])
if msg > len(goods_lst) or msg < 1:
await buy.finish("请输入正确的商品id!", at_sender=True)
goods = goods_lst[msg - 1]
else:
if msg[0] in goods_name_lst:
for i in range(len(goods_name_lst)):
if msg[0] == goods_name_lst[i]:
goods = goods_lst[i]
break
else:
await buy.finish("请输入正确的商品名称!")
else:
await buy.finish("请输入正确的商品名称!", at_sender=True)
async with db.transaction():
if (
await BagUser.get_gold(event.user_id, event.group_id)
) < goods.goods_price * num * goods.goods_discount:
await buy.finish("您的金币好像不太够哦", at_sender=True)
if await BagUser.buy_props(event.user_id, event.group_id, goods, num):
await buy.send(
f"花费 {goods.goods_price * num * goods.goods_discount} 金币购买 {goods.goods_name} ×{num} 成功!",
at_sender=True,
)
logger.info(
f"USER {event.user_id} GROUP {event.group_id} "
f"花费 {goods.goods_price*num} 金币购买 {goods.goods_name} ×{num} 成功!"
)
else:
await buy.send(f"{goods.goods_name} 购买失败!", at_sender=True)
logger.info(
f"USER {event.user_id} GROUP {event.group_id} "
f"花费 {goods.goods_price * num * goods.goods_discount} 金币购买 {goods.goods_name} ×{num} 失败!"
)
+52
View File
@@ -0,0 +1,52 @@
from nonebot import on_command
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent
from nonebot.typing import T_State
from nonebot.adapters.cqhttp.permission import GROUP
from utils.data_utils import init_rank
from models.bag_user import BagUser
from utils.message_builder import image
from utils.utils import get_message_text, is_number
__zx_plugin_name__ = "商店 - 我的金币"
__plugin_usage__ = """
usage:
我的金币
指令:
我的金币
""".strip()
__plugin_des__ = "商店 - 我的金币"
__plugin_cmd__ = ["我的金币"]
__plugin_type__ = ("商店",)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["商店", "我的金币"],
}
my_gold = on_command("我的金币", priority=5, block=True, permission=GROUP)
gold_rank = on_command("金币排行", priority=5, block=True, permission=GROUP)
@my_gold.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
await my_gold.finish(await BagUser.get_my_total_gold(event.user_id, event.group_id))
@gold_rank.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
num = get_message_text(event.json())
if is_number(num) and 51 > int(num) > 10:
num = int(num)
else:
num = 10
all_users = await BagUser.get_all_users(event.group_id)
all_user_id = [user.user_qq for user in all_users]
all_user_data = [user.gold for user in all_users]
rank_image = await init_rank("金币排行", all_user_id, all_user_data, event.group_id, num)
if rank_image:
await gold_rank.finish(image(b64=rank_image.pic2bs4()))
+52
View File
@@ -0,0 +1,52 @@
from nonebot import on_command
from services.log import logger
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent
from nonebot.typing import T_State
from models.bag_user import BagUser
from nonebot.adapters.cqhttp.permission import GROUP
__zx_plugin_name__ = "商店 - 我的道具"
__plugin_usage__ = """
usage:
我的道具
指令:
我的道具
""".strip()
__plugin_des__ = "商店 - 我的道具"
__plugin_cmd__ = ["我的道具"]
__plugin_type__ = ('商店',)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["商店", "我的道具"],
}
my_props = on_command("我的道具", priority=5, block=True, permission=GROUP)
@my_props.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
props = await BagUser.get_props(event.user_id, event.group_id)
if props:
pname_list = []
pnum_list = []
rst = ""
props = props[:-1].split(",")
for p in props:
if p != "":
if p in pname_list:
pnum_list[pname_list.index(p)] += 1
else:
pname_list.append(p)
pnum_list.append(1)
for i in range(len(pname_list)):
rst += f"{i+1}.{pname_list[i]}\t×{pnum_list[i]}\n"
await my_props.send("\n" + rst[:-1], at_sender=True)
logger.info(f"USER {event.user_id} GROUP {event.group_id} 查看我的道具")
else:
await my_props.finish("您的背包里没有任何的道具噢~", at_sender=True)
+26
View File
@@ -0,0 +1,26 @@
from utils.utils import scheduler
from models.bag_user import BagUser
from services.log import logger
__zx_plugin_name__ = "每日金币重置 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
# 重置每日金币
@scheduler.scheduled_job(
"cron",
hour=0,
minute=1,
)
async def _():
try:
user_list = await BagUser.get_all_users()
for user in user_list:
await user.update(
get_today_gold=0,
spend_today_gold=0,
).apply()
except Exception as e:
logger.error(f"重置每日金币错误 e:{e}")
+133
View File
@@ -0,0 +1,133 @@
from nonebot import on_command
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent, MessageEvent
from nonebot.typing import T_State
from configs.path_config import IMAGE_PATH
from utils.message_builder import image
from .data_source import create_shop_help, delete_goods, update_goods, registered_goods, parse_goods_info
from nonebot.permission import SUPERUSER
from utils.utils import get_message_text, is_number
from nonebot.plugin import export
from services.log import logger
import os
__zx_plugin_name__ = "商店"
__plugin_usage__ = """
usage:
商店项目,这可不是奸商
指令:
商店
""".strip()
__plugin_superuser_usage__ = """
usage:
商品操作
指令:
添加商品 name:[名称] price:[价格] des:[描述] ?discount:[折扣](小数) ?limit_time:[限时时间](小时)
删除商品 [名称或序号]
修改商品 name:[名称或序号] price:[价格] des:[描述] discount:[折扣] limit_time:[限时]
示例:添加商品 name:萝莉酒杯 price:9999 des:普通的酒杯,但是里面.. discount:0.4 limit_time:90
示例:添加商品 name:可疑的药 price:5 des:效果未知
示例:删除商品 2
示例:修改商品 name:1 price:900 修改序号为1的商品的价格为900
* 修改商品只需添加需要值即可 *
""".strip()
__plugin_des__ = "商店系统[金币回收计划]"
__plugin_cmd__ = [
"商店",
"添加商品 name:[名称] price:[价格] des:[描述] ?discount:[折扣](小数) ?limit_time:[限时时间](小时)) [_superuser]",
"删除商品 [名称或序号] [_superuser]",
"修改商品 name:[名称或序号] price:[价格] des:[描述] discount:[折扣] limit_time:[限时] [_superuser]",
]
__plugin_type__ = ('商店',)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["商店"],
}
__plugin_block_limit__ = {
"limit_type": "group"
}
# 导出方法供其他插件使用
export = export()
export.registered_goods = registered_goods
export.delete_goods = delete_goods
export.update_goods = update_goods
shop_help = on_command("商店", priority=5, block=True)
shop_add_goods = on_command("添加商品", priority=5, permission=SUPERUSER, block=True)
shop_del_goods = on_command("删除商品", priority=5, permission=SUPERUSER, block=True)
shop_update_goods = on_command("修改商品", priority=5, permission=SUPERUSER, block=True)
@shop_help.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
await shop_help.send(image(b64=await create_shop_help()))
@shop_add_goods.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json())
if msg:
data = parse_goods_info(msg)
if isinstance(data, str):
await shop_add_goods.finish(data)
if not data.get("name") or not data.get("price") or not data.get("des"):
await shop_add_goods.finish("name:price:des 参数不可缺少!")
if await registered_goods(**data):
await shop_add_goods.send(f"添加商品 {data['name']} 成功!\n"
f"名称:{data['name']}\n"
f"价格:{data['price']}金币\n"
f"简介:{data['des']}\n"
f"折扣:{data.get('discount')}\n"
f"限时:{data.get('limit_time')}", at_sender=True)
logger.info(f"USER {event.user_id} 添加商品 {msg} 成功")
else:
await shop_add_goods.send(f"添加商品 {msg} 失败了...", at_sender=True)
logger.warning(f"USER {event.user_id} 添加商品 {msg} 失败")
@shop_del_goods.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json())
if msg:
name = ""
id_ = 0
if is_number(msg):
id_ = int(msg)
else:
name = msg
rst, goods_name, code = await delete_goods(name, id_)
if code == 200:
await shop_del_goods.send(f"删除商品 {goods_name} 成功了...", at_sender=True)
if os.path.exists(f"{IMAGE_PATH}/shop_help.png"):
os.remove(f"{IMAGE_PATH}/shop_help.png")
logger.info(f"USER {event.user_id} 删除商品 {goods_name} 成功")
else:
await shop_del_goods.send(f"删除商品 {goods_name} 失败了...", at_sender=True)
logger.info(f"USER {event.user_id} 删除商品 {goods_name} 失败")
@shop_update_goods.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json())
if msg:
data = parse_goods_info(msg)
if isinstance(data, str):
await shop_add_goods.finish(data)
if not data.get("name"):
await shop_add_goods.finish("name 参数不可缺少!")
flag, name, text = await update_goods(**data)
if flag:
await shop_update_goods.send(f"修改商品 {name} 成功了...\n{text}", at_sender=True)
logger.info(f"USER {event.user_id} 修改商品 {name} 数据 {text} 成功")
else:
await shop_update_goods.send(f"修改商品 {name} 失败了...", at_sender=True)
logger.info(f"USER {event.user_id} 修改商品 {name} 数据 {text} 失败")
@@ -0,0 +1,281 @@
from models.goods_info import GoodsInfo
from utils.image_utils import BuildImage
from models.sign_group_user import SignGroupUser
from utils.utils import is_number
from configs.path_config import IMAGE_PATH
from typing import Optional, Union
from configs.config import Config
from nonebot import Driver
from nonebot.plugin import require
import nonebot
import time
driver: Driver = nonebot.get_driver()
use = require("use")
@driver.on_startup
async def init_default_shop_goods():
"""
导入内置的三个商品
"""
async def sign_card_1(**kwargs):
user_id = kwargs['user_id']
group_id = kwargs['group_id']
user = await SignGroupUser.ensure(user_id, group_id)
await user.update(add_probability=0.1).apply()
async def sign_card_2(**kwargs):
user_id = kwargs['user_id']
group_id = kwargs['group_id']
user = await SignGroupUser.ensure(user_id, group_id)
await user.update(add_probability=0.2).apply()
async def sign_card_3(**kwargs):
user_id = kwargs['user_id']
group_id = kwargs['group_id']
user = await SignGroupUser.ensure(user_id, group_id)
await user.update(add_probability=0.3).apply()
if Config.get_config("shop", "IMPORT_DEFAULT_SHOP_GOODS"):
await registered_goods(
"好感度双倍加持卡Ⅰ", 30, "下次签到双倍好感度概率 + 10%(谁才是真命天子?)(同类商品将覆盖)"
)
use.registered_use("好感度双倍加持卡Ⅰ", sign_card_1)
await registered_goods("好感度双倍加持卡Ⅱ", 150, "下次签到双倍好感度概率 + 20%(平平庸庸)(同类商品将覆盖)")
use.registered_use("好感度双倍加持卡Ⅱ", sign_card_2)
await registered_goods(
"好感度双倍加持卡Ⅲ", 250, "下次签到双倍好感度概率 + 30%(金币才是真命天子!)(同类商品将覆盖)"
)
use.registered_use("好感度双倍加持卡Ⅲ", sign_card_3)
# 创建商店界面
async def create_shop_help() -> str:
"""
制作商店图片
:return: 图片base64
"""
goods_lst = await GoodsInfo.get_all_goods()
idx = 1
_dc = {}
font_h = BuildImage(0, 0).getsize("正")[1]
h = 10
_list = []
for goods in goods_lst:
if goods.goods_limit_time == 0 or time.time() < goods.goods_limit_time:
h += len(goods.goods_description.strip().split("\n")) * font_h + 80
_list.append(goods)
A = BuildImage(1000, h, color="#f9f6f2")
current_h = 0
for goods in _list:
bk = BuildImage(
700, 80, font_size=15, color="#f9f6f2", font="CJGaoDeGuo.otf"
)
goods_image = BuildImage(
600, 80, font_size=20, color="#a29ad6", font="CJGaoDeGuo.otf"
)
name_image = BuildImage(
580, 40, font_size=25, color="#e67b6b", font="CJGaoDeGuo.otf"
)
await name_image.atext(
(15, 0), f"{idx}.{goods.goods_name}", center_type="by_height"
)
await name_image.aline((380, -5, 280, 45), "#a29ad6", 5)
await name_image.atext((390, 0), "售价:", center_type="by_height")
await name_image.atext(
(440, 0), str(goods.goods_price), (255, 255, 255), center_type="by_height"
)
await name_image.atext(
(
440
+ BuildImage(0, 0, plain_text=str(goods.goods_price), font_size=25).w,
0,
),
" 金币",
center_type="by_height",
)
await name_image.acircle_corner(5)
await goods_image.apaste(name_image, (0, 5), True, center_type="by_width")
await goods_image.atext((15, 50), f"简介:{goods.goods_description}")
await goods_image.acircle_corner(20)
await bk.apaste(goods_image, alpha=True)
# 添加限时图标和时间
if goods.goods_limit_time > 0:
_limit_time_logo = BuildImage(40, 40, background=f"{IMAGE_PATH}/other/time.png")
await bk.apaste(_limit_time_logo, (600, 0), True)
await bk.apaste(BuildImage(0, 0, plain_text="限时!", font_size=23, font="CJGaoDeGuo.otf"), (640, 10), True)
limit_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(goods.goods_limit_time)).split()
y_m_d = limit_time[0]
_h_m = limit_time[1].split(":")
h_m = _h_m[0] + "时 " + _h_m[1] + "分"
await bk.atext((605, 38), str(y_m_d))
await bk.atext((615, 57), str(h_m))
await bk.aline((550, -1, 710, -1), "#a29ad6", 5)
await bk.aline((550, 80, 710, 80), "#a29ad6", 5)
idx += 1
await A.apaste(bk, (0, current_h), True)
current_h += 90
w = 1000
h = A.h + 230 + 100
h = 1000 if h < 1000 else h
shop_logo = BuildImage(100, 100, background=f"{IMAGE_PATH}/other/shop_text.png")
shop = BuildImage(w, h, font_size=20, color="#f9f6f2")
shop.paste(A, (20, 230))
zx_img = BuildImage(0, 0, background=f"{IMAGE_PATH}/zhenxun/toukan.png")
zx_img.replace_color_tran(((240, 240, 240), (255, 255, 255)), (249, 246, 242))
await shop.apaste(zx_img, (780, 100))
await shop.apaste(shop_logo, (450, 30), True)
shop.text(
(int((1000 - shop.getsize("注【通过 序号 或者 商品名称 购买】")[0]) / 2), 170),
"注【通过 序号 或者 商品名称 购买】",
)
shop.text((20, h - 100), "神秘药水\t\t售价:9999999金币\n\t\t鬼知道会有什么效果~")
return shop.pic2bs4()
async def registered_goods(
name: str,
price: int,
des: str,
discount: Optional[float] = 1,
limit_time: Optional[int] = 0,
**kwargs,
):
"""
添加商品
例如: 折扣:可选参数↓ 限时时间:可选,单位为小时
添加商品 name:萝莉酒杯 price:9999 des:普通的酒杯,但是里面.. discount:0.4 limit_time:90
添加商品 name:可疑的药 price:5 des:效果未知
:param name: 商品名称
:param price: 商品价格
:param des: 商品简介
:param discount: 商品折扣
:param limit_time: 商品限时销售时间,单位为小时
:param kwargs: kwargs
:return: 是否添加成功
"""
if kwargs:
name = kwargs.get("name")
price = kwargs.get("price")
des = kwargs.get("des")
discount = kwargs.get("discount")
limit_time = kwargs.get("time_limit")
limit_time = float(limit_time) if limit_time else limit_time
discount = discount if discount is None else 1
limit_time = int(time.time() + limit_time * 60 * 60) if limit_time is not None and limit_time != 0 else 0
return await GoodsInfo.add_goods(
name, int(price), des, float(discount), limit_time
)
# 删除商品
async def delete_goods(name: str, id_: int) -> "str, str, int":
"""
删除商品
:param name: 商品名称
:param id_: 商品id
:return: 删除状况
"""
goods_lst = await GoodsInfo.get_all_goods()
if id_:
if id_ < 1 or id_ > len(goods_lst):
return "序号错误,没有该序号商品...", "", 999
goods_name = goods_lst[id_ - 1].goods_name
if await GoodsInfo.delete_goods(goods_name):
return f"删除商品 {goods_name} 成功!", goods_name, 200
else:
return f"删除商品 {goods_name} 失败!", goods_name, 999
if name:
if await GoodsInfo.delete_goods(name):
return f"删除商品 {name} 成功!", name, 200
else:
return f"删除商品 {name} 失败!", name, 999
# 更新商品信息
async def update_goods(**kwargs) -> "str, str, int":
"""
更新商品信息
:param kwargs: kwargs
:return: 更新状况
"""
if kwargs:
goods_lst = await GoodsInfo.get_all_goods()
if is_number(kwargs["name"]):
if int(kwargs["name"]) < 1 or int(kwargs["name"]) > len(goods_lst):
return "序号错误,没有该序号的商品...", "", 999
goods = goods_lst[int(kwargs["name"]) - 1]
else:
goods = await GoodsInfo.get_goods_info(kwargs["name"])
if not goods:
return "名称错误,没有该名称的商品...", "", 999
name = goods.goods_name
price = goods.goods_price
des = goods.goods_description
discount = goods.goods_discount
limit_time = goods.goods_limit_time
new_time = 0
tmp = ""
if kwargs.get("price"):
tmp += f'价格:{price} --> {kwargs["price"]}\n'
price = kwargs["price"]
if kwargs.get("des"):
tmp += f'描述:{des} --> {kwargs["des"]}\n'
des = kwargs["des"]
if kwargs.get("discount"):
tmp += f'折扣:{discount} --> {kwargs["discount"]}\n'
discount = kwargs["discount"]
if kwargs.get("limit_time"):
new_time = time.strftime(
"%Y-%m-%d %H:%M:%S",
time.localtime(time.time() + int(kwargs["limit_time"] * 60 * 60)),
)
tmp += f"折扣至: {new_time}\n"
limit_time = kwargs["limit_time"]
return (
await GoodsInfo.update_goods(
name,
int(price),
des,
float(discount),
int(time.time() + limit_time * 60 * 60 if limit_time != 0 and new_time else 0),
),
name,
tmp[:-1],
)
def parse_goods_info(msg: str) -> Union[dict, str]:
"""
解析格式数据
:param msg: 消息
:return: 解析完毕的数据data
"""
if "name:" not in msg:
return "必须指定修改的商品名称或序号!"
data = {}
for x in msg.split():
sp = x.split(":", maxsplit=1)
if str(sp[1]).strip():
sp[1] = sp[1].strip()
print(sp)
if sp[0] == "name":
data["name"] = sp[1]
elif sp[0] == "price":
if not is_number(sp[1]) or int(sp[1]) < 0:
return "price参数不合法,必须大于等于0!"
data["price"] = sp[1]
elif sp[0] == "des":
data["des"] = sp[1]
elif sp[0] == "discount":
if not is_number(sp[1]) or float(sp[1]) < 0:
return "discount参数不合法,必须大于0!"
data["discount"] = sp[1]
elif sp[0] == "limit_time":
if not is_number(sp[1]) or float(sp[1]) < 0:
return "limit_time参数不合法,必须大于0!"
data["limit_time"] = sp[1]
return data
+76
View File
@@ -0,0 +1,76 @@
from nonebot import on_command
from services.log import logger
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent
from nonebot.typing import T_State
from utils.utils import is_number, get_message_text
from models.bag_user import BagUser
from nonebot.adapters.cqhttp.permission import GROUP
from services.db_context import db
from nonebot.plugin import export
from .data_source import effect, registered_use
__zx_plugin_name__ = "商店 - 使用道具"
__plugin_usage__ = """
usage:
普通的使用道具
指令:
使用道具 [序号或道具名称]
* 序号以 ”我的道具“ 为准 *
""".strip()
__plugin_des__ = "商店 - 使用道具"
__plugin_cmd__ = ["使用道具 [序号或道具名称]"]
__plugin_type__ = ('商店',)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["商店", "使用道具"],
}
# 导出方法供其他插件使用
export = export()
export.registered_use = registered_use
use_props = on_command("使用道具", priority=5, block=True, permission=GROUP)
@use_props.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
msg = get_message_text(event.json())
if msg in ["", "帮助"]:
await use_props.finish(__plugin_usage__)
props = await BagUser.get_props(event.user_id, event.group_id)
if props:
async with db.transaction():
pname_list = []
props = props[:-1].split(",")
for p in props:
if p != "":
if p not in pname_list:
pname_list.append(p)
if is_number(msg):
if 0 < int(msg) <= len(pname_list):
name = pname_list[int(msg) - 1]
else:
await use_props.finish("仔细看看自己的道具仓库有没有这个道具?", at_sender=True)
else:
if msg not in pname_list:
await use_props.finish("道具名称错误!", at_sender=True)
name = msg
if await BagUser.del_props(
event.user_id, event.group_id, name
) and await effect(event.user_id, event.group_id, name):
await use_props.send(f"使用道具 {name} 成功!", at_sender=True)
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 使用道具 {name} 成功"
)
else:
await use_props.send(f"使用道具 {name} 失败!", at_sender=True)
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 使用道具 {name} 失败"
)
else:
await use_props.send("您的背包里没有任何的道具噢~", at_sender=True)
+45
View File
@@ -0,0 +1,45 @@
import asyncio
from services.log import logger
_use_func_data = {}
async def effect(user_id: int, group_id: int, goods_name: str) -> bool:
"""
商品生效
:param user_id: 用户id
:param group_id: 群号
:param goods_name: 商品名称
:return: 使用是否成功
"""
# 优先使用注册的商品插件
try:
if _use_func_data.get(goods_name):
_kwargs = _use_func_data[goods_name]["kwargs"]
_kwargs["goods_name"] = goods_name
_kwargs["user_id"] = user_id
_kwargs["group_id"] = group_id
if asyncio.iscoroutinefunction(_use_func_data[goods_name]["func"]):
await _use_func_data[goods_name]["func"](
**_kwargs,
)
else:
_use_func_data[goods_name]["func"](
**_kwargs,
)
return True
except Exception as e:
logger.error(f"use 商品生效函数effect 发生错误 {type(e)}:{e}")
return False
def registered_use(goods_name: str, func, **kwargs):
"""
注册商品使用方法
:param goods_name: 商品名称
:param func: 使用函数
:param kwargs: kwargs
"""
if goods_name in _use_func_data.keys():
raise ValueError("该商品使用函数已被注册!")
_use_func_data[goods_name] = {"func": func, "kwargs": kwargs}
+10
View File
@@ -0,0 +1,10 @@
import nonebot
nonebot.load_plugins('basic_plugins/super_cmd')
+152
View File
@@ -0,0 +1,152 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent, Message
from nonebot.rule import to_me
from utils.utils import get_message_text, is_number
from utils.manager import requests_manager
from utils.message_builder import image
from models.group_info import GroupInfo
__zx_plugin_name__ = "显示所有好友群组 [Superuser]"
__plugin_usage__ = """
usage:
显示所有好友群组
指令:
查看所有好友/查看所有群组
同意好友请求 [id]
拒绝好友请求 [id]
同意群聊请求 [id]
拒绝群聊请求 [id]
查看所有请求
清空所有请求
""".strip()
__plugin_des__ = "显示所有好友群组"
__plugin_cmd__ = [
"查看所有好友/查看所有群组",
"同意好友请求 [id]",
"拒绝好友请求 [id]",
"同意群聊请求 [id]",
"拒绝群聊请求 [id]",
"查看所有请求",
"清空所有请求",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
cls_group = on_command(
"查看所有群组", rule=to_me(), permission=SUPERUSER, priority=1, block=True
)
cls_friend = on_command(
"查看所有好友", rule=to_me(), permission=SUPERUSER, priority=1, block=True
)
friend_handle = on_command(
"同意好友请求", aliases={"拒绝好友请求"}, permission=SUPERUSER, priority=1, block=True
)
group_handle = on_command(
"同意群聊请求", aliases={"拒绝群聊请求"}, permission=SUPERUSER, priority=1, block=True
)
clear_request = on_command("清空所有请求", permission=SUPERUSER, priority=1, block=True)
cls_request = on_command("查看所有请求", permission=SUPERUSER, priority=1, block=True)
@cls_group.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
gl = await bot.get_group_list(self_id=int(bot.self_id))
msg = ["{group_id} {group_name}".format_map(g) for g in gl]
msg = "\n".join(msg)
msg = f"bot:{bot.self_id}\n| 群号 | 群名 | 共{len(gl)}个群\n" + msg
await bot.send_private_msg(
self_id=int(bot.self_id),
user_id=int(list(bot.config.superusers)[0]),
message=msg,
)
@cls_friend.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
gl = await bot.get_friend_list(self_id=int(bot.self_id))
msg = ["{user_id} {nickname}".format_map(g) for g in gl]
msg = "\n".join(msg)
msg = f"| QQ号 | 昵称 | 共{len(gl)}个好友\n" + msg
await bot.send_private_msg(
self_id=int(bot.self_id),
user_id=int(list(bot.config.superusers)[0]),
message=msg,
)
@friend_handle.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
id_ = get_message_text(event.json())
if is_number(id_):
id_ = int(id_)
if state["_prefix"]["raw_command"][:2] == "同意":
if await requests_manager.approve(bot, id_, "private"):
await friend_handle.send("同意好友请求成功..")
else:
await friend_handle.send("同意好友请求失败,可能是未找到此id的请求..")
else:
if await requests_manager.refused(bot, id_, "private"):
await friend_handle.send("拒绝好友请求成功..")
else:
await friend_handle.send("拒绝好友请求失败,可能是未找到此id的请求..")
else:
await friend_handle.send("id必须为纯数字!")
@group_handle.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
id_ = get_message_text(event.json())
if is_number(id_):
id_ = int(id_)
if state["_prefix"]["raw_command"][:2] == "同意":
rid = await requests_manager.approve(bot, id_, "group")
if rid:
await friend_handle.send("同意群聊请求成功..")
if await GroupInfo.get_group_info(rid):
await GroupInfo.set_group_flag(rid, 1)
else:
group_info = await bot.get_group_info(group_id=rid)
await GroupInfo.add_group_info(
rid,
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
1
)
else:
await friend_handle.send("同意群聊请求失败,可能是未找到此id的请求..")
else:
if await requests_manager.refused(bot, id_, "group"):
await friend_handle.send("拒绝群聊请求成功..")
else:
await friend_handle.send("拒绝群聊请求失败,可能是未找到此id的请求..")
else:
await friend_handle.send("id必须为纯数字!")
@cls_request.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
_str = ""
for type_ in ["private", "group"]:
msg = await requests_manager.show(type_)
if msg:
_str += image(b64=msg)
else:
_str += "没有任何好友请求.." if type_ == "private" else "没有任何群聊请求.."
if type_ == "private":
_str += '\n--------------------\n'
await cls_request.send(Message(_str))
@clear_request.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
requests_manager.clear()
await cls_request.send("已清空所有好友/群聊请求..")
+67
View File
@@ -0,0 +1,67 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent
from configs.path_config import TEMP_PATH
from nonebot.rule import to_me
from utils.utils import scheduler
from services.log import logger
from utils.manager import resources_manager
import asyncio
import time
import os
__zx_plugin_name__ = "清理临时数据 [Superuser]"
__plugin_usage__ = """
usage:
清理临时数据
指令:
清理临时数据
""".strip()
__plugin_des__ = "清理临时数据"
__plugin_cmd__ = [
"清理临时数据",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
clear_data = on_command(
"清理临时数据", rule=to_me(), permission=SUPERUSER, priority=1, block=True
)
resources_manager.add_temp_dir(TEMP_PATH)
@clear_data.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
await clear_data.send("开始清理临时数据....")
size = await asyncio.get_event_loop().run_in_executor(None, _clear_data)
await clear_data.send("共清理了 {:.2f}MB 的数据...".format(size / 1024 / 1024))
def _clear_data() -> float:
size = 0
for dir_ in resources_manager.get_temp_data_dir():
if dir_.exists():
for file in os.listdir(dir_):
file = dir_ / file
try:
if time.time() - os.path.getatime(file) > 300:
file_size = os.path.getsize(file)
file.unlink()
size += file_size
except Exception as e:
logger.error(f"清理临时数据错误...{type(e)}:{e}")
return float(size)
@scheduler.scheduled_job(
"cron",
hour=1,
minute=1,
)
async def _():
size = await asyncio.get_event_loop().run_in_executor(None, _clear_data)
logger.info("自动清理临时数据完成," + "共清理了 {:.2f}MB 的数据...".format(size / 1024 / 1024))
+74
View File
@@ -0,0 +1,74 @@
# async def open_remind(group: int, name: str) -> str:
# _name = ""
# if name == "zwa":
# _name = "早晚安"
# if name == "dz":
# _name = "地震播报"
# if name == "hy":
# _name = "群欢迎"
# if name == "kxcz":
# _name = "开箱重置提醒"
# if name == "gb":
# _name = "广播"
# if await GroupRemind.get_status(group, name):
# return f"该群已经开启过 {_name} 通知,请勿重复开启!"
# if await GroupRemind.set_status(group, name, True):
# return f"成功开启 {_name} 通知!0v0"
# else:
# return f"开启 {_name} 通知失败了..."
#
#
# async def close_remind(group: int, name: str) -> str:
# _name = ""
# if name == "zwa":
# _name = "早晚安"
# if name == "dz":
# _name = "地震播报"
# if name == "hy":
# _name = "群欢迎"
# if name == "kxcz":
# _name = "开箱重置提醒"
# if name == "gb":
# _name = "广播"
# if not await GroupRemind.get_status(group, name):
# return f"该群已经取消过 {_name} 通知,请勿重复取消!"
# if await GroupRemind.set_status(group, name, False):
# return f"成功关闭 {_name} 通知!0v0"
# else:
# return f"关闭 {_name} 通知失败了..."
# cmd_list = ['总开关', '签到', '发送图片', '色图', '黑白草图', 'coser', '鸡汤/语录', '骂我', '开箱', '鲁迅说', '假消息', '商店系统',
# '操作图片', '查询皮肤', '天气', '疫情', '识番', '搜番', '点歌', 'pixiv', 'rss', '方舟一井', '查干员', '骰子娘', '原神一井']
#
#
# def check_group_switch_json(group_id):
# if not os.path.exists(DATA_PATH + f'rule/group_switch/'):
# os.mkdir(DATA_PATH + f'rule/group_switch/')
# if not os.path.exists(DATA_PATH + f'rule/group_switch/{group_id}.json'):
# with open(DATA_PATH + f'rule/group_switch/{group_id}.json', 'w', encoding='utf8') as f:
# data = {}
# for cmd in cmd_list:
# data[cmd] = True
# f.write(json.dumps(data, ensure_ascii=False))
# else:
# with open(DATA_PATH + f'rule/group_switch/{group_id}.json', 'r', encoding='utf8') as f:
# try:
# data = json.load(f)
# except ValueError:
# data = {}
# if len(data.keys()) - 1 != len(cmd_list):
# for cmd in cmd_list:
# if cmd not in data.keys():
# data[cmd] = True
# with open(DATA_PATH + f'rule/group_switch/{group_id}.json', 'w', encoding='utf8') as wf:
# wf.write(json.dumps(data, ensure_ascii=False))
# reload(data)
# for file in os.listdir(DATA_PATH + 'group_help'):
# os.remove(DATA_PATH + f'group_help/{file}')
def reload(data):
static_group_dict = data
+38
View File
@@ -0,0 +1,38 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent
from nonebot.rule import to_me
from services.db_context import db
from utils.utils import get_message_text
from services.log import logger
__zx_plugin_name__ = "执行sql [Superuser]"
__plugin_usage__ = """
usage:
执行一段sql语句
指令:
exec [sql语句]
""".strip()
__plugin_des__ = "执行一段sql语句"
__plugin_cmd__ = [
"exec [sql语句]",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
exec_ = on_command("exec", rule=to_me(), permission=SUPERUSER, priority=1, block=True)
@exec_.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
sql = get_message_text(event.json())
async with db.transaction():
try:
query = db.text(sql)
await db.first(query)
await exec_.send("执行 sql 语句成功.")
except Exception as e:
await exec_.send(f"执行 sql 语句失败 {type(e)}:{e}")
logger.error(f"执行 sql 语句失败 {type(e)}:{e}")
+200
View File
@@ -0,0 +1,200 @@
from nonebot import on_command, on_regex
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent, GROUP, GroupMessageEvent
from nonebot.rule import to_me
from utils.utils import get_message_text, is_number
from utils.manager import group_manager, plugins2settings_manager
from models.group_info import GroupInfo
from services.log import logger
from configs.config import NICKNAME
from nonebot.adapters.cqhttp.exception import ActionFailed
__zx_plugin_name__ = "管理群操作 [Superuser]"
__plugin_usage__ = """
usage:
群权限 | 群白名单 | 退出群 操作
指令:
退群 [group_id]
修改群权限 [group_id] [等级]
添加群白名单 *[group_id]
删除群白名单 *[group_id]
添加群认证 *[group_id]
删除群认证 *[group_id]
查看群白名单
""".strip()
__plugin_des__ = "管理群操作"
__plugin_cmd__ = [
"退群 [group_id]",
"修改群权限 [group_id] [等级]",
"添加群白名单 *[group_id]",
"删除群白名单 *[group_id]",
"添加群认证 *[group_id]",
"删除群认证 *[group_id]",
"查看群白名单",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
del_group = on_command("退群", rule=to_me(), permission=SUPERUSER, priority=1, block=True)
add_group_level = on_command("修改群权限", priority=1, permission=SUPERUSER, block=True)
my_group_level = on_command(
"查看群权限", aliases={"群权限"}, priority=5, permission=GROUP, block=True
)
what_up_group_level = on_regex(
".*?(提高|提升|升高|增加|加上)(.*?)群权限.*?",
rule=to_me(),
priority=5,
permission=GROUP,
block=True,
)
manager_group_whitelist = on_command(
"添加群白名单", aliases={"删除群白名单"}, priority=1, permission=SUPERUSER, block=True
)
show_group_whitelist = on_command(
"查看群白名单", priority=1, permission=SUPERUSER, block=True
)
group_auth = on_command(
"添加群认证", aliases={"删除群认证"}, priority=1, permission=SUPERUSER, block=True
)
@del_group.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
group_id = get_message_text(event.json())
if group_id:
if is_number(group_id):
try:
await bot.set_group_leave(group_id=int(group_id))
logger.info(f"退出群聊 {group_id} 成功")
await del_group.send(f"退出群聊 {group_id} 成功", at_sender=True)
group_manager.delete_group(int(group_id))
await GroupInfo.delete_group_info(int(group_id))
except Exception as e:
logger.info(f"退出群聊 {group_id} 失败 e:{e}")
else:
await del_group.finish(f"请输入正确的群号", at_sender=True)
else:
await del_group.finish(f"请输入群号", at_sender=True)
@add_group_level.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json())
group_id = 0
level = 0
if not msg:
await add_group_level.finish("用法:修改群权限 [group] [level]")
msg = msg.split(" ")
if len(msg) < 2:
await add_group_level.finish("参数不完全..[group] [level]")
if is_number(msg[0]) and is_number(msg[1]):
group_id = msg[0]
level = int(msg[1])
else:
await add_group_level.finish("参数错误...group和level必须是数字..")
old_level = group_manager.get_group_level(group_id)
group_manager.set_group_level(group_id, level)
await add_group_level.send("修改成功...", at_sender=True)
if level > -1:
await bot.send_group_msg(
group_id=int(group_id), message=f"管理员修改了此群权限:{old_level} -> {level}"
)
logger.info(f"{event.user_id} 修改了 {group_id} 的权限:{level}")
@my_group_level.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
level = group_manager.get_group_level(event.group_id)
tmp = ""
data = plugins2settings_manager.get_data()
for module in data:
if data[module]["level"] > level:
plugin_name = data[module]["cmd"][0]
if plugin_name == "pixiv":
plugin_name = "搜图 p站排行"
tmp += f"{plugin_name}\n"
if tmp:
tmp = "\n目前无法使用的功能:\n" + tmp
await my_group_level.finish(f"当前群权限:{level}{tmp}")
@what_up_group_level.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
await what_up_group_level.finish(
f"[此功能用于防止内鬼,如果引起不便那真是抱歉了]\n" f"目前提高群权限的方法:\n" f"\t1.管理员修改权限"
)
@manager_group_whitelist.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json()).split()
all_group = [
g["group_id"] for g in await bot.get_group_list(self_id=int(bot.self_id))
]
group_list = []
for group in msg:
if is_number(group) and int(group) in all_group:
group_list.append(int(group))
if group_list:
for group in group_list:
if state["_prefix"]["raw_command"] in ["添加群白名单"]:
group_manager.add_group_white_list(group)
else:
group_manager.delete_group_white_list(group)
group_list = [str(x) for x in group_list]
await manager_group_whitelist.send(
"已成功将 " + "\n".join(group_list) + " " + state["_prefix"]["raw_command"]
)
else:
await manager_group_whitelist.send(f"添加失败,请检查{NICKNAME}是否已加入这些群聊或重复添加/删除群白单名")
@show_group_whitelist.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
x = group_manager.get_group_white_list()
x = [str(g) for g in x]
if x:
await show_group_whitelist.send("目前的群白名单:\n" + "\n".join(x))
else:
await show_group_whitelist.send("没有任何群在群白名单...")
@group_auth.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json()).split()
for group_id in msg:
if not is_number(group_id):
await group_auth.send(f"{group_id}非纯数字,已跳过该项..")
group_id = int(group_id)
if state["_prefix"]["raw_command"][:2] == "添加":
if await GroupInfo.get_group_info(group_id):
await GroupInfo.set_group_flag(group_id, 1)
else:
try:
group_info = await bot.get_group_info(group_id=group_id)
except ActionFailed:
group_info = {
"group_id": group_id,
"group_name": "_",
"max_member_count": -1,
"member_count": -1,
}
await GroupInfo.add_group_info(
group_info["group_id"],
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
1,
)
else:
if await GroupInfo.get_group_info(group_id):
await GroupInfo.set_group_flag(group_id, 0)
await group_auth.send(
f'已为 {group_id} {state["_prefix"]["raw_command"][:2]}群认证..'
)
+44
View File
@@ -0,0 +1,44 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from nonebot.rule import to_me
from utils.manager import (
plugins2cd_manager,
plugins2settings_manager,
plugins2block_manager,
group_manager,
)
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent
__zx_plugin_name__ = "重载插件配置 [Superuser]"
__plugin_usage__ = """
usage:
重载插件配置
plugins2settings,
plugins2cd
plugins2block
group_manager
指令:
重载插件配置
""".strip()
__plugin_des__ = "重载插件配置"
__plugin_cmd__ = [
"重载插件配置",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
reload_plugins_manager = on_command(
"重载配置", rule=to_me(), permission=SUPERUSER, priority=1, block=True
)
@reload_plugins_manager.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
plugins2settings_manager.reload()
plugins2cd_manager.reload()
plugins2block_manager.reload()
group_manager.reload()
await reload_plugins_manager.send("重载完成...")
+88
View File
@@ -0,0 +1,88 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from models.level_user import LevelUser
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, GroupMessageEvent, Message
from utils.utils import get_message_at, get_message_text, is_number
from services.log import logger
from utils.message_builder import at
__zx_plugin_name__ = "用户权限管理 [Superuser]"
__plugin_usage__ = """
usage:
增删改用户的权限
指令:
添加权限 [at] [权限]
添加权限 [qq] [group_id] [权限]
删除权限 [at]
""".strip()
__plugin_des__ = "增删改用户的权限"
__plugin_cmd__ = [
"添加权限 [at] [权限]",
"添加权限 [qq] [group_id] [权限]",
"删除权限 [at]",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
super_cmd = on_command(
"添加管理",
aliases={"删除管理", "添加权限", "删除权限"},
priority=1,
permission=SUPERUSER,
block=True,
)
@super_cmd.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
group_id = -1
level = 0
try:
args = get_message_text(event.json()).split()
qq = get_message_at(event.json())
flag = -1
if not qq:
if len(args) > 2:
if is_number(args[0]) and is_number(args[1]) and is_number(args[2]):
qq = int(args[0])
group_id = int(args[1])
level = int(args[2])
flag = 1
else:
await super_cmd.finish("所有参数必须是数字!", at_sender=True)
else:
await super_cmd.finish(
"权限参数不完全\n\t格式:添加/删除权限 [at] [level]"
"\n\t格式:添加/删除权限 [qq] [group_id] [level]",
at_sender=True,
)
else:
if not is_number(args[0]):
await super_cmd.finish("所有参数必须是数字!", at_sender=True)
level = int(args[0])
qq = qq[0]
group_id = event.group_id
flag = 2
if state["_prefix"]["raw_command"][:2] == "添加":
if await LevelUser.set_level(qq, group_id, level, 1):
result = "添加管理成功, 权限: " + str(level)
else:
result = "管理已存在, 更新权限: " + str(level)
else:
if await LevelUser.delete_level(qq, event.group_id):
result = "删除管理成功!"
else:
result = "该账号无管理权限!"
if flag == 2:
await super_cmd.send(result)
elif flag == 1:
await bot.send_group_msg(
group_id=group_id,
message=Message(f"{at(qq)}管理员修改了你的权限" f"\n--------\n你当前的权限等级:{level}"),
)
await super_cmd.send("修改成功")
except Exception as e:
await super_cmd.send("执行指令失败!")
logger.error(f"执行指令失败 e:{e}")
+56
View File
@@ -0,0 +1,56 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent
from nonebot.rule import to_me
from utils.utils import get_message_text, is_number
from services.log import logger
from utils.manager import group_manager
__zx_plugin_name__ = "超级用户被动开关 [Superuser]"
__plugin_usage__ = """
usage:
超级用户被动开关
指令:
开启/关闭广播通知
""".strip()
__plugin_des__ = "超级用户被动开关"
__plugin_cmd__ = [
"开启/关闭广播通知",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
oc_gb = on_command(
"开启广播通知",
aliases={"关闭广播通知"},
rule=to_me(),
permission=SUPERUSER,
priority=1,
block=True,
)
@oc_gb.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
group = get_message_text(event.json())
if group:
if is_number(group):
group = int(group)
for g in await bot.get_group_list():
if g["group_id"] == group:
break
else:
await oc_gb.finish("没有加入这个群...", at_sender=True)
if state["_prefix"]["raw_command"] == "开启广播通知":
logger.info(f"USER {event.user_id} 开启了 GROUP {group} 的广播")
await oc_gb.finish(await group_manager.open_group_task(group, "broadcast",), at_sender=True)
else:
logger.info(f"USER {event.user_id} 关闭了 GROUP {group} 的广播")
await oc_gb.finish(await group_manager.close_group_task(group, "broadcast"), at_sender=True)
else:
await oc_gb.finish("请输入正确的群号", at_sender=True)
else:
await oc_gb.finish("请输入要关闭广播的群号", at_sender=True)
+71
View File
@@ -0,0 +1,71 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent
from nonebot.rule import to_me
from utils.utils import get_bot
from services.log import logger
from models.group_info import GroupInfo
from models.friend_user import FriendUser
__zx_plugin_name__ = "更新群/好友信息 [Superuser]"
__plugin_usage__ = """
usage:
更新群/好友信息
指令:
更新群信息
更新好友信息
""".strip()
__plugin_des__ = "更新群/好友信息"
__plugin_cmd__ = [
"更新群信息",
"更新好友信息",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
update_group_info = on_command(
"更新群信息", rule=to_me(), permission=SUPERUSER, priority=1, block=True
)
update_friend_info = on_command(
"更新好友信息", rule=to_me(), permission=SUPERUSER, priority=1, block=True
)
@update_group_info.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
bot = get_bot()
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
num = 0
rst = ""
for g in gl:
group_info = await bot.get_group_info(group_id=g)
if await GroupInfo.add_group_info(
group_info["group_id"],
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
):
num += 1
logger.info(f"自动更新群组 {g} 信息成功")
else:
logger.info(f"自动更新群组 {g} 信息失败")
rst += f"{g} 更新失败\n"
await update_group_info.send(f"成功更新了 {num} 个群的信息\n{rst[:-1]}")
@update_friend_info.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
num = 0
rst = ""
fl = await get_bot().get_friend_list()
for f in fl:
if await FriendUser.add_friend_info(f["user_id"], f["nickname"]):
logger.info(f'自动更新好友 {f["user_id"]} 信息成功')
num += 1
else:
logger.warning(f'自动更新好友 {f["user_id"]} 信息失败')
rst += f'{f["user_id"]} 更新失败\n'
await update_friend_info.send(f"成功更新了 {num} 个好友的信息\n{rst[:-1]}")
+30
View File
@@ -0,0 +1,30 @@
from nonebot import on_command
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from nonebot.adapters import Bot, Event
from nonebot.rule import to_me
from configs.path_config import IMAGE_PATH
from utils.message_builder import image
from .data_source import create_help_image
from pathlib import Path
__zx_plugin_name__ = '超级用户帮助 [Superuser]'
superuser_help_image = Path(IMAGE_PATH) / 'superuser_help.png'
if superuser_help_image.exists():
superuser_help_image.unlink()
super_help = on_command(
"超级用户帮助", rule=to_me(), priority=1, permission=SUPERUSER, block=True
)
@super_help.handle()
async def _(bot: Bot, event: Event, state: T_State):
if not superuser_help_image.exists():
await create_help_image()
x = image(superuser_help_image)
await super_help.finish(x)
+80
View File
@@ -0,0 +1,80 @@
from utils.image_utils import BuildImage
from configs.path_config import IMAGE_PATH
from services.log import logger
from utils.utils import get_matchers
from nonebot.adapters.cqhttp import Bot
from pathlib import Path
from nonebot import Driver
import asyncio
import nonebot
driver: Driver = nonebot.get_driver()
background = Path(IMAGE_PATH) / "background" / "0.png"
superuser_help_image = Path(IMAGE_PATH) / "superuser_help.png"
@driver.on_bot_connect
async def create_help_image(bot: Bot = None):
"""
创建超级用户帮助图片
"""
await asyncio.get_event_loop().run_in_executor(None, _create_help_image)
def _create_help_image():
"""
创建管理员帮助图片
"""
_matchers = get_matchers()
_plugin_name_list = []
width = 0
help_str = "超级用户帮助\n\n* 注: ‘*’ 代表可有多个相同参数 ‘?’ 代表可省略该参数 *\n\n"
tmp_img = BuildImage(0, 0, plain_text='1', font_size=24)
for matcher in _matchers:
plugin_name = ""
try:
_plugin = nonebot.plugin.get_plugin(matcher.module)
_module = _plugin.module
try:
plugin_name = _module.__getattribute__("__zx_plugin_name__")
except AttributeError:
continue
is_superuser_usage = False
try:
_ = _module.__getattribute__("__plugin_superuser_usage__")
is_superuser_usage = True
except AttributeError:
pass
if (
("[superuser]" in plugin_name.lower() or is_superuser_usage)
and plugin_name != "超级用户帮助 [Superuser]"
and plugin_name not in _plugin_name_list
and "[hidden]" not in plugin_name.lower()
):
_plugin_name_list.append(plugin_name)
try:
plugin_des = _module.__getattribute__("__plugin_des__")
except AttributeError:
plugin_des = '_'
plugin_cmd = _module.__getattribute__("__plugin_cmd__")
if is_superuser_usage:
plugin_cmd = [x for x in plugin_cmd if "[_superuser]" in x]
plugin_cmd = " / ".join(plugin_cmd).replace('[_superuser]', '').strip()
help_str += f"{plugin_des} -> {plugin_cmd}\n\n"
x = tmp_img.getsize(f"{plugin_des} -> {plugin_cmd}")[0]
width = width if width > x else x
except Exception as e:
logger.warning(
f"获取超级用户插件 {matcher.module}: {plugin_name} 设置失败... {type(e)}:{e}"
)
height = len(help_str.split("\n")) * 33
width += 500
A = BuildImage(width, height, font_size=24)
_background = BuildImage(width, height, background=background)
A.text((300, 140), help_str)
A.paste(_background, alpha=True)
A.save(superuser_help_image)
logger.info(f"已成功加载 {len(_plugin_name_list)} 条超级用户命令")
+35
View File
@@ -0,0 +1,35 @@
from nonebot import on_command
from nonebot.adapters.cqhttp import Bot, Event
from nonebot.typing import T_State
from utils.message_builder import image
__zx_plugin_name__ = "更新信息"
__plugin_usage__ = """
usage:
更新信息
指令:
更新信息
""".strip()
__plugin_des__ = "当前版本的更新信息"
__plugin_cmd__ = ["更新信息"]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["更新信息"],
}
update_info = on_command("更新信息", aliases={"更新日志"}, priority=5, block=True)
@update_info.handle()
async def _(bot: Bot, event: Event, state: T_State):
img = image("update_info.png")
if img:
await update_info.finish(image("update_info.png"))
else:
await update_info.finish("目前没有更新信息哦")
+10 -17
View File
@@ -1,27 +1,20 @@
import nonebot
from nonebot.adapters.cqhttp import Bot as CQHTTPBot
from services.db_context import init, disconnect
# from nonebot.adapters.discord import Adapter as DiscordAdapter
# from nonebot.adapters.dodo import Adapter as DoDoAdapter
# from nonebot.adapters.kaiheila import Adapter as KaiheilaAdapter
from nonebot.adapters.onebot.v11 import Adapter as OneBotV11Adapter
nonebot.init()
driver = nonebot.get_driver()
driver.register_adapter(OneBotV11Adapter)
# driver.register_adapter(KaiheilaAdapter)
# driver.register_adapter(DoDoAdapter)
# driver.register_adapter(DiscordAdapter)
from zhenxun.services.db_context import disconnect, init
driver.register_adapter("cqhttp", CQHTTPBot)
config = driver.config
driver.on_startup(init)
driver.on_shutdown(disconnect)
# nonebot.load_builtin_plugins("echo")
nonebot.load_plugins("zhenxun/builtin_plugins")
nonebot.load_plugins("zhenxun/plugins")
# 优先加载定时任务
nonebot.load_plugin("nonebot_plugin_apscheduler")
nonebot.load_plugins("basic_plugins")
nonebot.load_plugins("plugins")
# 最后加载权限控制
nonebot.load_plugins("basic_plugins/hooks")
if __name__ == "__main__":
+24
View File
@@ -0,0 +1,24 @@
from typing import Optional
from pathlib import Path
from .utils import ConfigsManager
# 回复消息名称
NICKNAME: str = "小真寻"
# 数据库(必要)
# 如果填写了bind就不需要再填写后面的字段了#)
# 示例:"bind": "postgresql://user:password@127.0.0.1:5432/database"
bind: str = "" # 数据库连接链接
sql_name: str = "postgresql"
user: str = "" # 数据用户名
password: str = "" # 数据库密码
address: str = "" # 数据库地址
port: str = "" # 数据库端口
database: str = "" # 数据库名称
# 代理,例如 "http://127.0.0.1:7890"
SYSTEM_PROXY: Optional[str] = None # 全局代理
Config = ConfigsManager(Path() / "data" / "configs" / "plugins2config.yaml")
+41
View File
@@ -0,0 +1,41 @@
from pathlib import Path
# 图片路径
IMAGE_PATH = Path("resources/img/")
# 音频路径
VOICE_PATH = Path("resources/voice/")
# 文本路径
TEXT_PATH = Path("resources/txt/")
# 日志路径
LOG_PATH = Path("log/")
# 字体路径
FONT_PATH = Path("resources/ttf/")
# 数据路径
DATA_PATH = Path("data/")
# 临时图片路径
TEMP_PATH = Path("resources/img/temp/")
def init_path():
global IMAGE_PATH, VOICE_PATH, TEXT_PATH, LOG_PATH, FONT_PATH, DATA_PATH, TEMP_PATH
IMAGE_PATH.mkdir(parents=True, exist_ok=True)
VOICE_PATH.mkdir(parents=True, exist_ok=True)
TEXT_PATH.mkdir(parents=True, exist_ok=True)
LOG_PATH.mkdir(parents=True, exist_ok=True)
FONT_PATH.mkdir(parents=True, exist_ok=True)
DATA_PATH.mkdir(parents=True, exist_ok=True)
TEMP_PATH.mkdir(parents=True, exist_ok=True)
IMAGE_PATH = str(IMAGE_PATH.absolute()) + '/'
VOICE_PATH = str(VOICE_PATH.absolute()) + '/'
TEXT_PATH = str(TEXT_PATH.absolute()) + '/'
LOG_PATH = str(LOG_PATH.absolute()) + '/'
FONT_PATH = str(FONT_PATH.absolute()) + '/'
DATA_PATH = str(DATA_PATH.absolute()) + '/'
TEMP_PATH = str(TEMP_PATH.absolute()) + '/'
init_path()
+175
View File
@@ -0,0 +1,175 @@
from typing import Optional, Any, Union
from pathlib import Path
from ruamel.yaml import YAML
from ruamel import yaml
class ConfigsManager:
"""
插件配置 与 资源 管理器
"""
def __init__(self, file: Path):
self._data: dict = {}
self._admin_level_data = []
if file:
file.parent.mkdir(exist_ok=True, parents=True)
self.file = file
if file.exists():
_yaml = YAML()
with open(file, "r", encoding="utf8") as f:
self._data = _yaml.load(f)
def add_plugin_config(
self,
module: str,
key: str,
value: Optional[Any],
*,
name: Optional[str] = None,
help_: Optional[str] = None,
default_value: Optional[str] = None,
_override: bool = False,
):
"""
为插件添加一个配置,不会被覆盖,只有第一个生效
:param module: 模块
:param key: 键
:param value: 值
:param name: 插件名称
:param help_: 配置注解
:param default_value: 默认值
:param _override: 覆盖前值
"""
if (
not (module in self._data.keys() and self._data[module].get(key))
or _override
):
_module = None
if ":" in module:
module = module.split(":")
_module = module[-1]
module = module[0]
if "[LEVEL]" in key and _module:
key = key.replace("[LEVEL]", "").strip()
self._admin_level_data.append((_module, value))
if self._data.get(module) is None:
self._data[module] = {}
key = key.upper()
self._data[module][key] = {
"value": value,
"name": name.strip() if isinstance(name, str) else name,
"help": help_.strip() if isinstance(help_, str) else help_,
"default_value": default_value,
"level_module": _module,
}
def remove_plugin_config(self, module: str):
"""
为插件删除一个配置
:param module: 模块名
"""
if module in self._data.keys():
del self._data[module]
def set_config(self, module: str, key: str, value: str):
"""
设置配置值
:param module: 模块名
:param key: 配置名称
:param value: 值
"""
if module in self._data.keys():
if self._data[module].get(key) is not None:
self._data[module][key]["value"] = value
def set_help(self, module: str, key: str, help_: str):
"""
设置配置注释
:param module: 模块名
:param key: 配置名称
:param help_: 注释文本
"""
if module in self._data.keys():
if self._data[module].get(key) is not None:
self._data[module][key]["help"] = help_
def set_default_value(self, module: str, key: str, value: str):
"""
设置配置默认值
:param module: 模块名
:param key: 配置名称
:param value: 值
"""
if module in self._data.keys():
if self._data[module].get(key) is not None:
self._data[module][key]["default_value"] = value
def get_config(self, module: str, key: str, default: Optional[Any] = None) -> Optional[Any]:
"""
获取指定配置值
:param module: 模块名
:param key: 配置名称
:param default: 没有key值内容的默认返回值
"""
key = key.upper()
if module in self._data.keys():
for key in [key, f"{key} [LEVEL]"]:
if self._data[module].get(key) is not None:
if self._data[module][key]["value"] is None:
return self._data[module][key]["default_value"]
return self._data[module][key]["value"]
if default is not None:
return default
return None
def get_level2module(self, module: str, key: str) -> Optional[str]:
"""
获取指定key所绑定的module,一般为权限等级
:param module: 模块名
:param key: 配置名称
:return:
"""
if self._data.get(module) is not None:
if self._data[module].get(key) is not None:
return self._data[module][key].get("level_module")
def get(self, key: str):
"""
获取插件配置数据
:param key: 名称
"""
if key in self._data.keys():
return self._data[key]
def save(self, path: Union[str, Path] = None):
"""
保存数据
:param path: 路径
"""
path = path if path else self.file
with open(path, "w", encoding="utf8") as f:
yaml.dump(
self._data, f, indent=2, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
def get_admin_level_data(self):
"""
获取管理插件等级
"""
return self._admin_level_data
def is_empty(self) -> bool:
return not bool(self._data)
def keys(self):
return self._data.keys()
def __str__(self):
return str(self._data)
def __setitem__(self, key, value):
self._data[key] = value
def __getitem__(self, key):
return self._data[key]
+157 -1452
View File
File diff suppressed because it is too large Load Diff
Binary file not shown.

After

Width:  |  Height:  |  Size: 772 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 84 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 799 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.3 MiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 694 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 188 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 315 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 352 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 279 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 182 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 228 KiB

Some files were not shown because too many files have changed in this diff Show More