imgapi/imgapi.py at main · siiway/imgapi · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import typing as t
import importlib
import os

from fastapi import Request
from loguru import logger as l

import utils as u

_ImgFunc = t.Union[t.Callable[[Request], str], t.Callable[[Request], t.Awaitable[str]]]
_InitFunc = t.Union[t.Callable[[], None], t.Callable[[], t.Awaitable[None]]]


class ImageAPI:
    '''
    图片 API 基类
    '''
    id: str
    '''唯一 id'''
    horizontal: _ImgFunc | str | None
    '''处理横向图片请求的函数 / url'''
    vertical: _ImgFunc | str | None
    '''处理竖向图片请求的函数 / url'''
    auto: _ImgFunc | str | None
    '''处理自适应图片请求的函数 / url'''
    init: _InitFunc | None
    '''在初始化时执行的函数'''
    cn: bool
    '''是否适用于中国大陆用户'''
    outseas: bool
    '''是否适用于港澳台 / 海外用户'''

    def __init__(
        self,
        id: str,
        horizontal: _ImgFunc | str | None = None,
        vertical: _ImgFunc | str | None = None,
        auto: _ImgFunc | str | None = None,
        init: _InitFunc | None = None,
        cn: bool = False,
        outseas: bool = False
    ):
        '''
        声明一个图片 API

        :param id: 唯一 id, 直接传入 __name__ 以使用文件名
        :param horizontal: 处理横向图片请求的函数 / url
        :param vertical: 处理竖向图片请求的函数 / url
        :param auto: 处理自适应图片请求的函数 / url
        :param init: 在初始化时执行的函数
        :param cn: 是否适用于中国大陆用户
        :param outseas: 是否适用于港澳台 / 海外用户
        '''
        self.id = id.split('.')[-1]
        self.horizontal = horizontal
        self.vertical = vertical
        self.auto = auto
        self.init = init
        self.cn = cn
        self.outseas = outseas


class ImgAPIWrapped(ImageAPI):
    horizontal: _ImgFunc
    vertical: _ImgFunc
    auto: _ImgFunc
    init: _InitFunc


class ImgAPIInit:
    allow_h: set[ImgAPIWrapped] = set()
    allow_v: set[ImgAPIWrapped] = set()
    allow_a: set[ImgAPIWrapped] = set()
    allow_h_cn: set[ImgAPIWrapped] = set()
    allow_v_cn: set[ImgAPIWrapped] = set()
    allow_a_cn: set[ImgAPIWrapped] = set()
    allow_h_outseas: set[ImgAPIWrapped] = set()
    allow_v_outseas: set[ImgAPIWrapped] = set()
    allow_a_outseas: set[ImgAPIWrapped] = set()

    async def load_all(self) -> None:
        p_all = u.perf_counter()
        dirlst = os.listdir('sites/')
        sites = 0
        for n in dirlst:
            name, ext = os.path.splitext(n)
            if ext != '.py' or 'example' in name:
                continue

            p = u.perf_counter()
            module = importlib.import_module(f'sites.{name}')
            for attr in dir(module):
                obj = getattr(module, attr)
                if not isinstance(obj, ImageAPI):
                    continue

                await u.call_init_func(obj.init)

                if obj.horizontal:
                    if obj.cn:
                        self.allow_h_cn.add(obj)  # type: ignore
                    if obj.outseas:
                        self.allow_h_outseas.add(obj) # type: ignore
                    self.allow_h.add(obj)  # type: ignore
                if obj.vertical:
                    if obj.cn:
                        self.allow_v_cn.add(obj)  # type: ignore
                    if obj.outseas:
                        self.allow_v_outseas.add(obj) # type: ignore
                    self.allow_v.add(obj)  # type: ignore
                if obj.auto:
                    if obj.cn:
                        self.allow_a_cn.add(obj)  # type: ignore
                    if obj.outseas:
                        self.allow_a_outseas.add(obj) # type: ignore
                    self.allow_a.add(obj)  # type: ignore

                l.debug(f'Init site {name} from sites/{n} took {p()}ms')
                sites += 1

        l.info(f'Init {sites} sites finished in {p_all()}ms.')
        l.info(f'Loaded: {len(self.allow_h_cn)} / {len(self.allow_h_outseas)} Horizontal, {len(self.allow_v_cn)} / {len(self.allow_v_outseas)} Vertical, {len(self.allow_a_cn)} / {len(self.allow_a_outseas)} Auto (cn / outseas).')
        l.debug(f'allow_h sites: {[i.id for i in self.allow_h_cn]} (cn) / {[i.id for i in self.allow_h_outseas]} (outseas)')
        l.debug(f'allow_v sites: {[i.id for i in self.allow_v_cn]} (cn) / {[i.id for i in self.allow_v_outseas]} (outseas)')
        l.debug(f'allow_a sites: {[i.id for i in self.allow_a_cn]} (cn) / {[i.id for i in self.allow_a_outseas]} (outseas)')