103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
from ncatbot.core import BotClient, GroupMessage, PrivateMessage
|
||
from ncatbot.utils import get_log
|
||
from ncatbot.plugin import BasePlugin, CompatibleEnrollment
|
||
from dotenv import load_dotenv
|
||
import requests
|
||
import os
|
||
|
||
# bot = BotClient()
|
||
bot = CompatibleEnrollment
|
||
_log = get_log()
|
||
|
||
|
||
class GetFile(BasePlugin):
|
||
name = "GetFile"
|
||
version = "0.0.1"
|
||
author = "pythagodzilla"
|
||
info = "A plugin to get images and videos"
|
||
|
||
async def on_load(self):
|
||
_log.info(f"Plugin {self.name} loaded! ")
|
||
|
||
@bot.private_event()
|
||
async def on_receive_images_private_event(self, msg: PrivateMessage):
|
||
if msg.message[0]["type"] == "image":
|
||
file_name = msg.message[0]["data"]["file"]
|
||
_log.info(f"Name of the image is: {file_name}")
|
||
|
||
image_url = await self.api.get_image(file_name)
|
||
_log.info(f"Image's information collected by \"get_image\" is: {image_url}")
|
||
real_url = image_url['data']['url']
|
||
_log.info(f"Image's real url is: {real_url}")
|
||
|
||
response = requests.get(real_url)
|
||
img = response.content
|
||
with open(f"../data/pics/{file_name}", 'wb') as f:
|
||
f.write(img)
|
||
|
||
@bot.group_event()
|
||
async def on_receive_images_group_event(self, msg: GroupMessage):
|
||
if msg.message[0]["type"] == "image":
|
||
file_name = msg.message[0]["data"]["file"]
|
||
_log.info(f"Name of the image is: {file_name}")
|
||
|
||
image_url = await self.api.get_image(file_name)
|
||
_log.info(f"Image's information collected by \"get_image\" is: {image_url}")
|
||
real_url = image_url['data']['url']
|
||
_log.info(f"Image's real url is: {real_url}")
|
||
|
||
response = requests.get(real_url)
|
||
img = response.content
|
||
with open(f"../data/pics/{file_name}", 'wb') as f:
|
||
f.write(img)
|
||
|
||
@bot.group_event()
|
||
async def on_receive_videos_group_event(self, msg: GroupMessage):
|
||
if msg.message[0]["type"] == "video":
|
||
file_name = msg.message[0]["data"]["file"]
|
||
_log.info(f"Name of the video is: {file_name}")
|
||
|
||
video_url = await self.api.get_file(file_name)
|
||
_log.info(f"Video's information collected by \"get_video\" is: {video_url}")
|
||
real_url = video_url['data']['url']
|
||
_log.info(f"Video's real url is: {real_url}")
|
||
|
||
response = requests.get(real_url)
|
||
video = response.content
|
||
with open(f"../data/videos/{file_name}", 'wb') as f:
|
||
f.write(video)
|
||
|
||
@bot.private_event()
|
||
async def on_receive_videos_private_event(self, msg: PrivateMessage):
|
||
if msg.message[0]["type"] == "video":
|
||
file_name = msg.message[0]["data"]["file"]
|
||
_log.info(f"Name of the video is: {file_name}")
|
||
|
||
video_url = await self.api.get_file(file_name)
|
||
_log.info(f"Video's information collected by \"get_video\" is: {video_url}")
|
||
real_url = video_url['data']['url']
|
||
_log.info(f"Video's real url is: {real_url}")
|
||
|
||
response = requests.get(real_url)
|
||
video = response.content
|
||
with open(f"../data/videos/{file_name}", 'wb') as f:
|
||
f.write(video)
|
||
|
||
|
||
"""
|
||
当单独运行这个程序时，会连接napcat并保存图片和视频到本地。
|
||
|
||
"""
|
||
|
||
if __name__ == "__main__":
|
||
load_dotenv(dotenv_path=r"../../../.env")
|
||
bot = BotClient()
|
||
api = bot.run(
|
||
bt_uin=os.getenv("BOT_ACCOUNT"),
|
||
root=os.getenv("ROOT_ACCOUNT"),
|
||
ws_uri=os.getenv("WS_URI"),
|
||
ws_token=os.getenv("WS_TOKEN"),
|
||
webui_uri=os.getenv("WEBUI_URI"),
|
||
webui_token=os.getenv("WEBUI_TOKEN")
|
||
)
|