suyiiyii/nonebot-bison
1
0
Fork 0
mirror of https://github.com/suyiiyii/nonebot-bison.git synced 2025-09-03 22:52:25 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

disable update browser

Browse Source
This commit is contained in:
felinae98 2022-02-10 16:17:19 +08:00
parent 69a4f59c04
commit 8e3de419d3
No known key found for this signature in database
GPG Key ID: 00C8B010587FF610
1 changed files with 109 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

171
src/plugins/nonebot_bison/utils.py
View File

@ -1,36 +1,48 @@
import asyncio import asyncio
import base64 import base64
from html import escape
import os import os
import platform import platform
import re import re
import subprocess import subprocess
import sys import sys
from html import escape
from pathlib import Path
from time import asctime from time import asctime
from typing import Awaitable, Callable, Optional, Union from typing import Awaitable, Callable, Optional, Union
from bs4 import BeautifulSoup as bs
import nonebot import nonebot
from bs4 import BeautifulSoup as bs
from nonebot.adapters.cqhttp.message import MessageSegment from nonebot.adapters.cqhttp.message import MessageSegment
from nonebot.log import logger from nonebot.log import default_format, logger
from nonebot.log import default_format
from playwright._impl._driver import compute_driver_executable from playwright._impl._driver import compute_driver_executable
from playwright.async_api import Browser, Page, Playwright, async_playwright from playwright.async_api import Browser, Page, Playwright, async_playwright
from uvicorn.loops import asyncio as _asyncio
from uvicorn import config from uvicorn import config
from uvicorn.loops import asyncio as _asyncio
from .plugin_config import plugin_config from .plugin_config import plugin_config
class Singleton(type): class Singleton(type):
_instances = {} _instances = {}
def __call__(cls, *args, **kwargs): def __call__(cls, *args, **kwargs):
if cls not in cls._instances: if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls] return cls._instances[cls]
@nonebot.get_driver().on_startup @nonebot.get_driver().on_startup
def download_browser(): def download_browser():
if not plugin_config.bison_browser and not plugin_config.bison_use_local: if not plugin_config.bison_browser and not plugin_config.bison_use_local:
system = platform.system()
if system == "Linux":
browser_path = Path.home() / ".cache" / "ms-playwright"
elif system == "Windows":
browser_path = Path.home() / "AppData" / "Local" / "ms-playwright"
else:
raise RuntimeError("platform not supported")
if browser_path.exists() and os.listdir(str(browser_path)):
logger.warning("Browser Exists, skip")
env = os.environ.copy() env = os.environ.copy()
driver_executable = compute_driver_executable() driver_executable = compute_driver_executable()
env["PW_CLI_TARGET_LANG"] = "python" env["PW_CLI_TARGET_LANG"] = "python"
@ -38,65 +50,84 @@ def download_browser():
class Render(metaclass=Singleton): class Render(metaclass=Singleton):
def __init__(self): def __init__(self):
self.lock = asyncio.Lock() self.lock = asyncio.Lock()
self.browser: Browser self.browser: Browser
self.interval_log = '' self.interval_log = ""
self.remote_browser = False self.remote_browser = False
async def get_browser(self, playwright: Playwright) -> Browser: async def get_browser(self, playwright: Playwright) -> Browser:
if plugin_config.bison_browser: if plugin_config.bison_browser:
if plugin_config.bison_browser.startswith('local:'): if plugin_config.bison_browser.startswith("local:"):
path = plugin_config.bison_browser.split('local:', 1)[1] path = plugin_config.bison_browser.split("local:", 1)[1]
return await playwright.chromium.launch( return await playwright.chromium.launch(
executable_path=path, args=['--no-sandbox']) executable_path=path, args=["--no-sandbox"]
if plugin_config.bison_browser.startswith('ws:'): )
if plugin_config.bison_browser.startswith("ws:"):
self.remote_browser = True self.remote_browser = True
return await playwright.chromium.connect(plugin_config.bison_browser) return await playwright.chromium.connect(plugin_config.bison_browser)
if plugin_config.bison_browser.startswith('wsc:'): if plugin_config.bison_browser.startswith("wsc:"):
self.remote_browser = True self.remote_browser = True
return await playwright.chromium.connect_over_cdp( return await playwright.chromium.connect_over_cdp(
'ws:' + plugin_config.bison_browser[4:] "ws:" + plugin_config.bison_browser[4:]
) )
raise RuntimeError('bison_BROWSER error') raise RuntimeError("bison_BROWSER error")
if plugin_config.bison_use_local: if plugin_config.bison_use_local:
return await playwright.chromium.launch( return await playwright.chromium.launch(
executable_path='/usr/bin/chromium', args=['--no-sandbox']) executable_path="/usr/bin/chromium", args=["--no-sandbox"]
return await playwright.chromium.launch(args=['--no-sandbox']) )
return await playwright.chromium.launch(args=["--no-sandbox"])
async def close_browser(self): async def close_browser(self):
if not self.remote_browser: if not self.remote_browser:
await self.browser.close() await self.browser.close()
async def render(self, url: str, viewport: Optional[dict] = None, target: Optional[str] = None, async def render(
operation: Optional[Callable[[Page], Awaitable[None]]] = None) -> Optional[bytes]: self,
url: str,
viewport: Optional[dict] = None,
target: Optional[str] = None,
operation: Optional[Callable[[Page], Awaitable[None]]] = None,
) -> Optional[bytes]:
retry_times = 0 retry_times = 0
self.interval_log = '' self.interval_log = ""
while retry_times < 3: while retry_times < 3:
try: try:
return await asyncio.wait_for(self.do_render(url, viewport, target, operation), 20) return await asyncio.wait_for(
self.do_render(url, viewport, target, operation), 20
)
except asyncio.TimeoutError: except asyncio.TimeoutError:
retry_times += 1 retry_times += 1
logger.warning("render error {}\n".format(retry_times) + self.interval_log) logger.warning(
self.interval_log = '' "render error {}\n".format(retry_times) + self.interval_log
)
self.interval_log = ""
# if self.browser: # if self.browser:
# await self.browser.close() # await self.browser.close()
# self.lock.release() # self.lock.release()
def _inter_log(self, message: str) -> None: def _inter_log(self, message: str) -> None:
self.interval_log += asctime() + '' + message + '\n' self.interval_log += asctime() + "" + message + "\n"
async def do_render(self, url: str, viewport: Optional[dict] = None, target: Optional[str] = None, async def do_render(
operation: Optional[Callable[[Page], Awaitable[None]]] = None) -> Optional[bytes]: self,
url: str,
viewport: Optional[dict] = None,
target: Optional[str] = None,
operation: Optional[Callable[[Page], Awaitable[None]]] = None,
) -> Optional[bytes]:
async with self.lock: async with self.lock:
async with async_playwright() as playwright: async with async_playwright() as playwright:
self.browser = await self.get_browser(playwright) self.browser = await self.get_browser(playwright)
self._inter_log('open browser') self._inter_log("open browser")
if viewport: if viewport:
constext = await self.browser.new_context( constext = await self.browser.new_context(
viewport={'width': viewport['width'], 'height': viewport['height']}, viewport={
device_scale_factor=viewport.get('deviceScaleFactor', 1)) "width": viewport["width"],
"height": viewport["height"],
},
device_scale_factor=viewport.get("deviceScaleFactor", 1),
)
page = await constext.new_page() page = await constext.new_page()
else: else:
page = await self.browser.new_page() page = await self.browser.new_page()
@ -104,51 +135,57 @@ class Render(metaclass=Singleton):
await operation(page) await operation(page)
else: else:
await page.goto(url) await page.goto(url)
self._inter_log('open page') self._inter_log("open page")
if target: if target:
target_ele = page.locator(target) target_ele = page.locator(target)
if not target_ele: if not target_ele:
return None return None
data = await target_ele.screenshot(type='jpeg') data = await target_ele.screenshot(type="jpeg")
else: else:
data = await page.screenshot(type='jpeg') data = await page.screenshot(type="jpeg")
self._inter_log('screenshot') self._inter_log("screenshot")
await page.close() await page.close()
self._inter_log('close page') self._inter_log("close page")
await self.close_browser() await self.close_browser()
self._inter_log('close browser') self._inter_log("close browser")
assert(isinstance(data, bytes)) assert isinstance(data, bytes)
return data return data
async def text_to_pic(self, text: str) -> Optional[bytes]: async def text_to_pic(self, text: str) -> Optional[bytes]:
lines = text.split('\n') lines = text.split("\n")
parsed_lines = list(map(lambda x: '<p>{}</p>'.format(escape(x)), lines)) parsed_lines = list(map(lambda x: "<p>{}</p>".format(escape(x)), lines))
html_text = '<div style="width:17em;padding:1em">{}</div>'.format(''.join(parsed_lines)) html_text = '<div style="width:17em;padding:1em">{}</div>'.format(
url = 'data:text/html;charset=UTF-8;base64,{}'.format(base64.b64encode(html_text.encode()).decode()) "".join(parsed_lines)
data = await self.render(url, target='div') )
url = "data:text/html;charset=UTF-8;base64,{}".format(
base64.b64encode(html_text.encode()).decode()
)
data = await self.render(url, target="div")
return data return data
async def text_to_pic_cqcode(self, text:str) -> MessageSegment: async def text_to_pic_cqcode(self, text: str) -> MessageSegment:
data = await self.text_to_pic(text) data = await self.text_to_pic(text)
# logger.debug('file size: {}'.format(len(data))) # logger.debug('file size: {}'.format(len(data)))
if data: if data:
# logger.debug(code) # logger.debug(code)
return MessageSegment.image(data) return MessageSegment.image(data)
else: else:
return MessageSegment.text('生成图片错误') return MessageSegment.text("生成图片错误")
async def parse_text(text: str) -> MessageSegment: async def parse_text(text: str) -> MessageSegment:
'return raw text if don\'t use pic, otherwise return rendered opcode' "return raw text if don't use pic, otherwise return rendered opcode"
if plugin_config.bison_use_pic: if plugin_config.bison_use_pic:
render = Render() render = Render()
return await render.text_to_pic_cqcode(text) return await render.text_to_pic_cqcode(text)
else: else:
return MessageSegment.text(text) return MessageSegment.text(text)
def html_to_text(html: str, query_dict: dict = {}) -> str: def html_to_text(html: str, query_dict: dict = {}) -> str:
html = re.sub(r'<br\s*/?>', '<br>\n', html) html = re.sub(r"<br\s*/?>", "<br>\n", html)
html = html.replace('</p>', '</p>\n') html = html.replace("</p>", "</p>\n")
soup = bs(html, 'html.parser') soup = bs(html, "html.parser")
if query_dict: if query_dict:
node = soup.find(**query_dict) node = soup.find(**query_dict)
else: else:
@ -158,7 +195,6 @@ def html_to_text(html: str, query_dict: dict = {}) -> str:
class Filter: class Filter:
def __init__(self) -> None: def __init__(self) -> None:
self.level: Union[int, str] = "DEBUG" self.level: Union[int, str] = "DEBUG"
@ -168,36 +204,47 @@ class Filter:
if module: if module:
module_name = getattr(module, "__module_name__", module_name) module_name = getattr(module, "__module_name__", module_name)
record["name"] = module_name.split(".")[0] record["name"] = module_name.split(".")[0]
levelno = logger.level(self.level).no if isinstance(self.level, levelno = (
str) else self.level logger.level(self.level).no if isinstance(self.level, str) else self.level
)
nonebot_warning_level = logger.level("WARNING").no nonebot_warning_level = logger.level("WARNING").no
return record["level"].no >= levelno if record["name"] != "nonebot" \ return (
else record["level"].no >= nonebot_warning_level record["level"].no >= levelno
if record["name"] != "nonebot"
else record["level"].no >= nonebot_warning_level
)
if plugin_config.bison_filter_log: if plugin_config.bison_filter_log:
logger.remove() logger.remove()
default_filter = Filter() default_filter = Filter()
logger.add(sys.stdout, logger.add(
colorize=True, sys.stdout,
diagnose=False, colorize=True,
filter=default_filter, diagnose=False,
format=default_format) filter=default_filter,
format=default_format,
)
config = nonebot.get_driver().config config = nonebot.get_driver().config
logger.success("Muted info & success from nonebot") logger.success("Muted info & success from nonebot")
default_filter.level = ( default_filter.level = (
"DEBUG" if config.debug else ("DEBUG" if config.debug else "INFO")
"INFO") if config.log_level is None else config.log_level if config.log_level is None
else config.log_level
)
# monkey patch # monkey patch
def asyncio_setup(): def asyncio_setup():
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
@property @property
def should_reload(self): def should_reload(self):
return False return False
if platform.system() == "Windows": if platform.system() == "Windows":
_asyncio.asyncio_setup = asyncio_setup _asyncio.asyncio_setup = asyncio_setup
config.Config.should_reload = should_reload # type:ignore config.Config.should_reload = should_reload # type:ignore
logger.warning('检测到当前为 Windows 系统，已自动注入猴子补丁') logger.warning("检测到当前为 Windows 系统，已自动注入猴子补丁")