format code

applications/api/__init__.py

@@ -0,0 +1,7 @@
+
+# feishu_sheet
+from .async_feishu_api import FeishuBotApi
+from .async_feishu_api import FeishuSheetApi
+
+feishu_robot = FeishuBotApi()
+feishu_sheet = FeishuSheetApi()

applications/api/async_feishu_api.py

@@ -0,0 +1,224 @@
+import json
+from applications.utils import AsyncHttPClient
+
+
+class Feishu:
+    # 服务号分组群发监测机器人
+    server_account_publish_monitor_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/380fdecf-402e-4426-85b6-7d9dbd2a9f59"
+
+    # 外部服务号投流监测机器人
+    outside_gzh_monitor_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/0899d43d-9f65-48ce-a419-f83ac935bf59"
+
+    # 长文 daily 报警机器人
+    long_articles_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/b44333f2-16c0-4cb1-af01-d135f8704410"
+
+    # 测试环境报警机器人
+    long_articles_bot_dev = "https://open.feishu.cn/open-apis/bot/v2/hook/f32c0456-847f-41f3-97db-33fcc1616bcd"
+
+    def __init__(self):
+        self.token = None
+        self.headers = {"Content-Type": "application/json"}
+        self.mention_all = {
+            "content": "<at id=all></at>\n",
+            "tag": "lark_md",
+        }
+        self.not_mention = {}
+
+    async def fetch_token(self):
+        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/"
+        post_data = {
+            "app_id": "cli_a51114cf8bf8d00c",
+            "app_secret": "cNoTAqMpsAm7mPBcpCAXFfvOzCNL27fe",
+        }
+        async with AsyncHttPClient(default_headers=self.headers) as client:
+            response = await client.post(url=url, data=post_data)
+
+        tenant_access_token = response["tenant_access_token"]
+        self.token = tenant_access_token
+
+
+class FeishuSheetApi(Feishu):
+
+    async def prepend_value(self, sheet_token, sheet_id, ranges, values):
+        insert_value_url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{}/values_prepend".format(
+            sheet_token
+        )
+        headers = {
+            "Authorization": "Bearer " + self.token,
+            "contentType": "application/json; charset=utf-8",
+        }
+        body = {
+            "valueRange": {"range": "{}!{}".format(sheet_id, ranges), "values": values}
+        }
+        async with AsyncHttPClient() as client:
+            response = await client.post(
+                url=insert_value_url, json=body, headers=headers
+            )
+
+        print(response)
+
+    async def insert_value(self, sheet_token, sheet_id, ranges, values):
+        insert_value_url = (
+            "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{}/values".format(
+                sheet_token
+            )
+        )
+        headers = {
+            "Authorization": "Bearer " + self.token,
+            "contentType": "application/json; charset=utf-8",
+        }
+        body = {
+            "valueRange": {"range": "{}!{}".format(sheet_id, ranges), "values": values}
+        }
+        async with AsyncHttPClient() as client:
+            response = await client.put(
+                url=insert_value_url, json=body, headers=headers
+            )
+
+
+class FeishuBotApi(Feishu):
+
+    @classmethod
+    def create_feishu_columns_sheet(
+        cls,
+        sheet_type,
+        sheet_name,
+        display_name,
+        width="auto",
+        vertical_align="top",
+        horizontal_align="left",
+        number_format=None,
+    ):
+        match sheet_type:
+            case "plain_text":
+                return {
+                    "name": sheet_name,
+                    "display_name": display_name,
+                    "width": width,
+                    "data_type": "text",
+                    "vertical_align": vertical_align,
+                    "horizontal_align": horizontal_align,
+                }
+
+            case "lark_md":
+                return {
+                    "name": sheet_name,
+                    "display_name": display_name,
+                    "data_type": "lark_md",
+                }
+
+            case "number":
+                return {
+                    "name": sheet_name,
+                    "display_name": display_name,
+                    "data_type": "number",
+                    "format": number_format,
+                    "width": width,
+                }
+
+            case "date":
+                return {
+                    "name": sheet_name,
+                    "display_name": display_name,
+                    "data_type": "date",
+                    "date_format": "YYYY/MM/DD",
+                }
+
+            case "options":
+                return {
+                    "name": sheet_name,
+                    "display_name": display_name,
+                    "data_type": "options",
+                }
+
+            case _:
+                return {
+                    "name": sheet_name,
+                    "display_name": display_name,
+                    "width": width,
+                    "data_type": "text",
+                    "vertical_align": vertical_align,
+                    "horizontal_align": horizontal_align,
+                }
+
+    # 表格形式
+    def create_feishu_table(self, title, columns, rows, mention):
+        table_base = {
+            "header": {
+                "template": "blue",
+                "title": {"content": title, "tag": "plain_text"},
+            },
+            "elements": [
+                self.mention_all if mention else self.not_mention,
+                {
+                    "tag": "table",
+                    "page_size": len(rows) + 1,
+                    "row_height": "low",
+                    "header_style": {
+                        "text_align": "left",
+                        "text_size": "normal",
+                        "background_style": "grey",
+                        "text_color": "default",
+                        "bold": True,
+                        "lines": 1,
+                    },
+                    "columns": columns,
+                    "rows": rows,
+                },
+            ],
+        }
+        return table_base
+
+    def create_feishu_bot_obj(self, title, mention, detail):
+        """
+        create feishu bot object
+        """
+        return {
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": self.mention_all if mention else self.not_mention,
+                },
+                {
+                    "tag": "div",
+                    "text": {
+                        "content": json.dumps(detail, ensure_ascii=False, indent=4),
+                        "tag": "lark_md",
+                    },
+                },
+            ],
+            "header": {"title": {"content": title, "tag": "plain_text"}},
+        }
+
+    # bot
+    async def bot(self, title, detail, mention=True, table=False, env="prod"):
+        match env:
+            case "dev":
+                url = self.long_articles_bot_dev
+            case "prod":
+                url = self.long_articles_bot
+            case "outside_gzh_monitor":
+                url = self.outside_gzh_monitor_bot
+            case "server_account_publish_monitor":
+                url = self.server_account_publish_monitor_bot
+            case _:
+                url = self.long_articles_bot_dev
+
+        headers = {"Content-Type": "application/json"}
+        if table:
+            card = self.create_feishu_table(
+                title=title,
+                columns=detail["columns"],
+                rows=detail["rows"],
+                mention=mention,
+            )
+        else:
+            card = self.create_feishu_bot_obj(
+                title=title, mention=mention, detail=detail
+            )
+
+        payload = {"msg_type": "interactive", "card": card}
+        async with AsyncHttPClient() as client:
+            res = await client.post(url=url, headers=headers, data=json.dumps(payload))
+
+        return res

applications/service/__init__.py

@@ -1 +1,6 @@
 from .get_cover import GetCoverService
+from .response import Response
+from .response import TaskScheduleResponse
+
+
+task_schedule_response = TaskScheduleResponse()

applications/service/response.py

@@ -7,3 +7,13 @@ class Response:
     @classmethod
     def error_response(cls, error_code, error_message):
         return {"code": error_code, "status": "error", "message": error_message}
+
+
+class TaskScheduleResponse:
+    @classmethod
+    async def fail_response(cls, error_code, error_message):
+        return {"code": error_code, "status": "error", "message": error_message}
+
+    @classmethod
+    async def success_response(cls, data):
+        return {"code": 0, "status": "task execute successfully", "data": data}

applications/utils/__init__.py

@@ -1,3 +1,4 @@
 from .get_cover import fetch_channel_info
 from .get_cover import fetch_aigc_cover
 from .get_cover import fetch_long_video_cover
+from .async_http_client import AsyncHttPClient

applications/utils/async_http_client.py

@@ -0,0 +1,99 @@
+import aiohttp
+from typing import Optional, Union, Dict, Any
+
+
+class AsyncHttPClient:
+    def __init__(
+        self,
+        timeout: int = 10,
+        max_connections: int = 100,
+        default_headers: Optional[Dict[str, str]] = None,
+    ):
+        """
+        简化版异步 HTTP 客户端
+
+        :param timeout: 请求超时时间（秒）
+        :param max_connections: 连接池最大连接数
+        :param default_headers: 默认请求头
+        """
+        self.timeout = aiohttp.ClientTimeout(total=timeout)
+        self.connector = aiohttp.TCPConnector(limit=max_connections)
+        self.default_headers = default_headers or {}
+        self.session = None
+
+    async def __aenter__(self):
+        self.session = aiohttp.ClientSession(
+            connector=self.connector, timeout=self.timeout, headers=self.default_headers
+        )
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.session.close()
+
+    async def request(
+        self,
+        method: str,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+        json: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], str]:
+        """核心请求方法"""
+        request_headers = {**self.default_headers, **(headers or {})}
+
+        try:
+            async with self.session.request(
+                method,
+                url,
+                params=params,
+                data=data,
+                json=json,
+                headers=request_headers,
+            ) as response:
+                response.raise_for_status()
+                content_type = response.headers.get("Content-Type", "")
+
+                if "application/json" in content_type:
+                    return await response.json()
+                return await response.text()
+
+        except aiohttp.ClientResponseError as e:
+            print(f"HTTP error: {e.status} {e.message}")
+            raise
+        except aiohttp.ClientError as e:
+            print(f"Network error: {str(e)}")
+            raise
+
+    async def get(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], str]:
+        """GET 请求"""
+        return await self.request("GET", url, params=params, headers=headers)
+
+    async def post(
+        self,
+        url: str,
+        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+        json: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], str]:
+        """POST 请求"""
+        return await self.request("POST", url, data=data, json=json, headers=headers)
+
+    async def put(
+            self,
+            url: str,
+            data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+            json: Optional[Dict[str, Any]] = None,
+            headers: Optional[Dict[str, str]] = None
+    ) -> Union[Dict[str, Any], str]:
+        """
+        PUT 请求
+
+        通常用于更新资源，可以发送表单数据或 JSON 数据
+        """
+        return await self.request("PUT", url, data=data, json=json, headers=headers)

routes/blueprint.py

@@ -1,6 +1,8 @@
 from quart import Blueprint, jsonify, request
 from applications.service import GetCoverService
 
+from tasks.task_scheduler import TaskScheduler
+
 server_blueprint = Blueprint("api", __name__, url_prefix="/api")
 
 
@@ -12,4 +14,10 @@ def server_routes(pools):
         task = GetCoverService(pools, params)
         return jsonify(await task.deal())
 
+    @server_blueprint.route("/run_task", methods=["POST"])
+    async def run_task():
+        data = await request.get_json()
+        task_scheduler = TaskScheduler(data)
+        return jsonify(await task_scheduler.deal())
+
     return server_blueprint

tasks/monitor_tasks/__init__.py

@@ -0,0 +1 @@
+from .kimi_balance import check_kimi_balance

tasks/monitor_tasks/kimi_balance.py

@@ -0,0 +1,33 @@
+import traceback
+from typing import Dict
+from applications.api import feishu_robot
+from applications.utils import AsyncHttPClient
+
+# const
+BALANCE_LIMIT_THRESHOLD = 100.0
+
+
+async def check_kimi_balance() -> Dict:
+    url = "https://api.moonshot.cn/v1/users/me/balance"
+    headers = {
+        "Authorization": "Bearer sk-5DqYCa88kche6nwIWjLE1p4oMm8nXrR9kQMKbBolNAWERu7q",
+        "Content-Type": "application/json; charset=utf-8",
+    }
+    async with AsyncHttPClient() as client:
+        response = await client.get(url, headers=headers)
+
+    try:
+        balance = response["data"]["available_balance"]
+        if balance < BALANCE_LIMIT_THRESHOLD:
+            await feishu_robot.bot(
+                title="kimi余额小于 {} 块".format(BALANCE_LIMIT_THRESHOLD),
+                detail={"balance": balance},
+            )
+    except Exception as e:
+        error_stack = traceback.format_exc()
+        await feishu_robot.bot(
+            title="kimi余额接口处理失败，数据结构异常",
+            detail={"error": str(e), "error_msg": error_stack},
+        )
+
+    return response

tasks/task_scheduler.py

@@ -0,0 +1,27 @@
+from applications.service import task_schedule_response
+from tasks.monitor_tasks import check_kimi_balance
+
+
+class TaskScheduler:
+    def __init__(self, data):
+        self.data = data
+
+    async def deal(self):
+        task_name = self.data.get("task_name")
+        if not task_name:
+            return await task_schedule_response.fail_response(
+                error_code="4002",
+                error_message="task_name must be input"
+            )
+
+        match task_name:
+            case "check_kimi_balance":
+                response = await check_kimi_balance()
+                return task_schedule_response.success_response(response)
+            case _:
+                return await task_schedule_response.fail_response(
+                    error_code="4001",
+                    error_message="wrong task name input"
+                )
+
+