PamGram/core/services/template/services.py

import asyncio
from typing import Optional
from urllib.parse import urlencode, urljoin, urlsplit
from uuid import uuid4

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from jinja2 import Environment, FileSystemLoader, Template
from playwright.async_api import ViewportSize

from core.application import Application
from core.base_service import BaseService
from core.config import config as application_config
from core.dependence.aiobrowser import AioBrowser
from core.services.template.cache import HtmlToFileIdCache, TemplatePreviewCache
from core.services.template.error import QuerySelectorNotFound
from core.services.template.models import FileType, RenderResult
from utils.const import PROJECT_ROOT
from utils.log import logger

__all__ = ("TemplateService", "TemplatePreviewer")


class TemplateService(BaseService):
    def __init__(
        self,
        app: Application,
        browser: AioBrowser,
        html_to_file_id_cache: HtmlToFileIdCache,
        preview_cache: TemplatePreviewCache,
        template_dir: str = "resources",
    ):
        self._browser = browser
        self.template_dir = PROJECT_ROOT / template_dir

        self._jinja2_env = Environment(
            loader=FileSystemLoader(template_dir),
            enable_async=True,
            autoescape=True,
            auto_reload=application_config.debug,
        )
        self.using_preview = application_config.debug and application_config.webserver.enable

        if self.using_preview:
            self.previewer = TemplatePreviewer(self, preview_cache, app.web_app)

        self.html_to_file_id_cache = html_to_file_id_cache

    def get_template(self, template_name: str) -> Template:
        return self._jinja2_env.get_template(template_name)

    async def render_async(self, template_name: str, template_data: dict) -> str:
        """模板渲染
        :param template_name: 模板文件名
        :param template_data: 模板数据
        """
        loop = asyncio.get_event_loop()
        start_time = loop.time()
        template = self.get_template(template_name)
        html = await template.render_async(**template_data)
        logger.debug("%s 模板渲染使用了 %s", template_name, str(loop.time() - start_time))
        return html

    async def render(
        self,
        template_name: str,
        template_data: dict,
        viewport: Optional[ViewportSize] = None,
        full_page: bool = True,
        evaluate: Optional[str] = None,
        query_selector: Optional[str] = None,
        file_type: FileType = FileType.PHOTO,
        ttl: int = 24 * 60 * 60,
        caption: Optional[str] = None,
        parse_mode: Optional[str] = None,
        filename: Optional[str] = None,
    ) -> RenderResult:
        """模板渲染成图片
        :param template_name: 模板文件名
        :param template_data: 模板数据
        :param viewport: 截图大小
        :param full_page: 是否长截图
        :param evaluate: 页面加载后运行的 js
        :param query_selector: 截图选择器
        :param file_type: 缓存的文件类型
        :param ttl: 缓存时间
        :param caption: 图片描述
        :param parse_mode: 图片描述解析模式
        :param filename: 文件名字
        :return:
        """
        loop = asyncio.get_event_loop()
        start_time = loop.time()
        template = self.get_template(template_name)

        if self.using_preview:
            preview_url = await self.previewer.get_preview_url(template_name, template_data)
            logger.debug("调试模板 URL: \n%s", preview_url)

        html = await template.render_async(**template_data)
        logger.debug("%s 模板渲染使用了 %s", template_name, str(loop.time() - start_time))

        file_id = await self.html_to_file_id_cache.get_data(html, file_type.name)
        if file_id and not application_config.debug:
            logger.debug("%s 命中缓存，返回 file_id[%s]", template_name, file_id)
            return RenderResult(
                html=html,
                photo=file_id,
                file_type=file_type,
                cache=self.html_to_file_id_cache,
                ttl=ttl,
                caption=caption,
                parse_mode=parse_mode,
                filename=filename,
            )

        browser = await self._browser.get_browser()
        start_time = loop.time()
        page = await browser.new_page(viewport=viewport)
        uri = (PROJECT_ROOT / template.filename).as_uri()
        await page.goto(uri)
        await page.set_content(html, wait_until="networkidle")
        if evaluate:
            await page.evaluate(evaluate)
        clip = None
        if query_selector:
            try:
                card = await page.query_selector(query_selector)
                if not card:
                    raise QuerySelectorNotFound
                clip = await card.bounding_box()
                if not clip:
                    raise QuerySelectorNotFound
            except QuerySelectorNotFound:
                logger.warning("未找到 %s 元素", query_selector)
        png_data = await page.screenshot(clip=clip, full_page=full_page)
        await page.close()
        logger.debug("%s 图片渲染使用了 %s", template_name, str(loop.time() - start_time))
        return RenderResult(
            html=html,
            photo=png_data,
            file_type=file_type,
            cache=self.html_to_file_id_cache,
            ttl=ttl,
            caption=caption,
            parse_mode=parse_mode,
            filename=filename,
        )


class TemplatePreviewer(BaseService, load=application_config.webserver.enable and application_config.debug):
    def __init__(
        self,
        template_service: TemplateService,
        cache: TemplatePreviewCache,
        web_app: FastAPI,
    ):
        self.web_app = web_app
        self.template_service = template_service
        self.cache = cache
        self.register_routes()

    async def get_preview_url(self, template: str, data: dict):
        """获取预览 URL"""
        components = urlsplit(application_config.webserver.url)
        path = urljoin("/preview/", template)
        query = {}

        # 如果有数据，暂存在 redis 中
        if data:
            key = str(uuid4())
            await self.cache.set_data(key, data)
            query["key"] = key

        # noinspection PyProtectedMember
        return components._replace(path=path, query=urlencode(query)).geturl()

    def register_routes(self):
        """注册预览用到的路由"""

        @self.web_app.get("/preview/{path:path}")
        async def preview_template(path: str, key: Optional[str] = None):  # pylint: disable=W0612
            # 如果是 /preview/ 开头的静态文件，直接返回内容。比如使用相对链接 ../ 引入的静态资源
            if not path.endswith((".html", ".jinja2")):
                full_path = self.template_service.template_dir / path
                if not full_path.is_file():
                    raise HTTPException(status_code=404, detail=f"Template '{path}' not found")
                return FileResponse(full_path)

            # 取回暂存的渲染数据
            data = await self.cache.get_data(key) if key else {}
            if key and data is None:
                raise HTTPException(status_code=404, detail=f"Template data {key} not found")

            # 渲染 jinja2 模板
            html = await self.template_service.render_async(path, data)
            # 将本地 URL file:// 修改为 HTTP url，因为浏览器内不允许加载本地文件
            # file:///project_dir/cache/image.jpg => /cache/image.jpg
            html = html.replace(PROJECT_ROOT.as_uri(), "")
            return HTMLResponse(html)

        # 其他静态资源
        for name in ["cache", "resources"]:
            directory = PROJECT_ROOT / name
            directory.mkdir(exist_ok=True)
            self.web_app.mount(f"/{name}", StaticFiles(directory=PROJECT_ROOT / name), name=name)