🎉 first commit

.gitignore

@@ -159,4 +159,7 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/
+
+data/
+wakatime.db

README.md

@@ -0,0 +1,61 @@
+<!-- markdownlint-disable MD033 MD036 MD041 MD045 -->
+<div align="center">
+  <a href="https://v2.nonebot.dev/store">
+    <img src="./docs/NoneBotPlugin.svg" width="300" alt="logo">
+  </a>
+</div>
+
+<div align="center">
+
+# NoneBot-Plugin-Wakatime
+
+_✨ NoneBot Wakatime 查询插件✨_
+
+<a href="">
+  <img src="https://img.shields.io/pypi/v/nonebot-plugin-wakatime.svg" alt="pypi"
+</a>
+<img src="https://img.shields.io/badge/python-3.10+-blue.svg" alt="python">
+<a href="https://pdm.fming.dev">
+  <img src="https://img.shields.io/endpoint?url=https%3A%2F%2Fcdn.jsdelivr.net%2Fgh%2Fpdm-project%2F.github%2Fbadge.json" alt="pdm-managed">
+</a>
+<a href="https://github.com/nonebot/plugin-alconna">
+  <img src="https://img.shields.io/badge/Alconna-resolved-2564C2" alt="alc-resolved">
+</a>
+
+</div>
+
+## 📖 介绍
+
+NoneBot Wakatime 查询插件。将你的代码统计嵌入 Bot 中
+
+## ⚙️ 配置
+
+在项目的配置文件中添加下表中的可选配置
+
+> [!note]
+> `client_id` 和 `client_secret` 均从 [WakaTime App](https://wakatime.com/apps) 获取
+
+|             配置项             | 必填 |             默认值             |
+|:---------------------------:|:--:|:---------------------------:|
+|     wakatime__client_id     | 是  |              无              |
+|   wakatime__client_secret   | 是  |              无              |
+|      wakatime__api_url      | 否  | https://wakatime.com/api/v1 |
+| wakatime__background_source | 否  |           default           |
+
+
+## 🎉 使用
+
+> [!note]
+> 请注意你的 `COMMAND_START` 以及上述配置项。
+
+### 绑定账号
+
+```shell
+/wakatime -b|--bind|bind [code]
+```
+
+### 查询信息
+
+```shell
+/wakatime [@]
+```

nonebot_plugin_wakatime/__init__.py

@@ -0,0 +1,122 @@
+from urllib.parse import parse_qs
+
+from nonebot import require
+from nonebot.log import logger
+from httpx import ConnectTimeout
+from sqlalchemy.exc import InterfaceError
+from nonebot.internal.adapter import Event
+from nonebot.plugin import PluginMetadata, inherit_supported_adapters
+
+require("nonebot_plugin_orm")
+require("nonebot_plugin_alconna")
+require("nonebot_plugin_htmlrender")
+from nonebot_plugin_orm import async_scoped_session
+from nonebot_plugin_alconna.uniseg import At, Button, UniMessage
+from nonebot_plugin_alconna import Args, Match, Option, Alconna, MsgTarget, on_alconna
+
+from .apis import API
+from . import migrations
+from .models import User
+from .shema import WakaTime
+from .render_pic import render
+from .config import Config, config
+
+__plugin_meta__ = PluginMetadata(
+    name="谁是卷王",
+    description="将代码统计嵌入 Bot 中",
+    usage="/wakatime",
+    config=Config,
+    homepage="https://github.com/KomoriDev/nonebot-plugin-wakatime",
+    supported_adapters=inherit_supported_adapters("nonebot_plugin_alconna"),
+    extra={
+        "unique_name": "WakaTime",
+        "orm_version_location": migrations,
+        "author": "Komorebi <mute231010@gmail.com>",
+        "version": "0.1.0",
+    },
+)
+
+wakatime = on_alconna(
+    Alconna(
+        "wakatime",
+        Args["target?", At | int],
+        Option("-b|--bind|bind", Args["code?", str], help_text="绑定 wakatime"),
+    ),
+    aliases={"waka"},
+    use_cmd_start=True,
+)
+
+
+@wakatime.assign("$main")
+async def _(event: Event, target: Match[At | int]):
+    if target.available:
+        if isinstance(target, At):
+            target_name = "他"
+            target_id = target.target
+        else:
+            target_name = "他"
+            target_id = target.result
+    else:
+        target_name = "你"
+        target_id = event.get_user_id()
+
+    try:
+        user_info = await API.get_user_info(target_id)
+        stats_info = await API.get_user_stats(target_id)
+        all_time_since_today = await API.get_all_time_since_today(target_id)
+    except InterfaceError:
+        await UniMessage.text(
+            f"{target_name}还没有绑定 Wakatime 账号！请私聊我并使用 /bind 命令进行绑定"
+        ).finish(at_sender=True)
+    except ConnectTimeout:
+        await (
+            UniMessage.text("网络超时，再试试叭")
+            .keyboard(Button("input", "重试", text="/wakatime"))
+            .finish(at_sender=True)
+        )
+
+    result = WakaTime(
+        user=user_info, stats=stats_info, all_time_since_today=all_time_since_today
+    )
+    image = await render(result)
+    await UniMessage.image(raw=image).finish(at_sender=True)
+
+
+@wakatime.assign("bind")
+async def _(
+    code: Match[str], event: Event, msg_target: MsgTarget, session: async_scoped_session
+):
+
+    if await session.get(User, event.get_user_id()):
+        await UniMessage("已绑定过 wakatime 账号").finish(at_sender=True)
+
+    if not msg_target.private:
+        await UniMessage("绑定指令只允许在私聊中使用").finish(at_sender=True)
+
+    if not code.available:
+        auth_url = (
+            f"https://wakatime.com/oauth/authorize?client_id={config.client_id}&response_type=code"
+            f"&redirect_uri=https://wakatime.com/dashboard"
+        )
+
+        await (
+            UniMessage.text(f"前往该页面绑定 wakatime 账号：{auth_url}")
+            # .keyboard(Button("link", label="即刻前往", url=auth_url))
+            .finish(at_sender=True)
+        )
+
+    resp = await API.bind_user(code.result)
+
+    if resp.status_code == 200:
+        parsed_data = parse_qs(resp.text)
+        user = User(
+            user_id=event.get_user_id(),
+            platform=msg_target.adapter,
+            access_token=parsed_data["access_token"][0],
+        )
+        session.add(user)
+        await session.commit()
+        await UniMessage("绑定成功").finish(at_sender=True)
+
+    logger.error(f"用户 {event.get_user_id()} 绑定失败。状态码：{resp.status_code}")
+    await UniMessage("绑定失败").finish(at_sender=True)

nonebot_plugin_wakatime/apis/__init__.py

@@ -0,0 +1 @@
+from .request import API as API

nonebot_plugin_wakatime/apis/request.py

@@ -0,0 +1,97 @@
+from typing import Literal, TypeAlias
+
+import httpx
+from httpx import Response
+from sqlalchemy import select
+from nonebot_plugin_orm import get_session
+
+from ..models import User
+from ..config import config
+from ..shema import Stats, Users
+
+api_url = config.api_url
+TimeScope: TypeAlias = Literal[
+    "last_7_days", "last_30_days", "last_6_months", "last_year", "all_time"
+]
+
+
+class API:
+
+    _access_token_cache: dict[str, str] = {}
+
+    @classmethod
+    async def get_access_token(cls, user_id: str) -> str:
+        """Get the access token from database"""
+        if user_id in cls._access_token_cache:
+            return cls._access_token_cache[user_id]
+
+        session = get_session()
+        async with session.begin():
+            stmt = select(User).where(User.user_id == user_id)
+            user = (await session.execute(stmt)).scalar()
+
+            if not user:
+                ...
+
+            cls._access_token_cache[user_id] = user.access_token
+            return user.access_token
+
+    @classmethod
+    async def bind_user(cls, code: str) -> Response:
+        """Get the secret access token.
+
+        Args:
+            code: The code parameter in the callback address
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                "https://wakatime.com/oauth/token",
+                data={
+                    "client_id": config.client_id,
+                    "client_secret": config.client_secret,
+                    "redirect_uri": "https://wakatime.com/dashboard",
+                    "grant_type": "authorization_code",
+                    "code": code,
+                },
+            )
+        return response
+
+    @classmethod
+    async def get_user_info(cls, user_id: str) -> Users:
+        """Get user's information"""
+        access_token = await cls.get_access_token(user_id)
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{api_url}/users/current",
+                headers={"Authorization": f"Bearer {access_token}"},
+            )
+        return Users(**(response.json()["data"]))
+
+    @classmethod
+    async def get_user_stats(
+        cls, user_id: str, scope: TimeScope = "last_7_days"
+    ) -> Stats:
+        """Get user's coding activity.
+
+        Args:
+            user_id: user id
+            scope: "last_7_days" "last_30_days" "last_6_months" "last_year" "all_time"
+        """
+        assess_token = await cls.get_access_token(user_id)
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{api_url}/users/current/stats/{scope}",
+                headers={"Authorization": f"Bearer {assess_token}"},
+            )
+        return Stats(**(response.json()["data"]))
+
+    @classmethod
+    async def get_all_time_since_today(cls, user_id: str) -> str:
+        """Get the total time logged since account created."""
+        assess_token = await cls.get_access_token(user_id)
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{api_url}/users/current/all_time_since_today",
+                headers={"Authorization": f"Bearer {assess_token}"},
+            )
+        return response.json()["data"]["text"]

nonebot_plugin_wakatime/config.py

@@ -0,0 +1,29 @@
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel
+from nonebot.plugin import get_plugin_config
+
+RESOURCES_DIR: Path = Path(__file__).parent / "resources"
+TEMPLATES_DIR: Path = RESOURCES_DIR / "templates"
+
+
+class ScopedConfig(BaseModel):
+
+    client_id: str
+    """Your App ID from https://wakatime.com/apps"""
+    client_secret: str
+    """Your App Secret from https://wakatime.com/apps"""
+    api_url: str = "https://wakatime.com/api/v1"
+    """wakatime api"""
+    background_source: Literal["default", "LoliAPI", "Lolicon"] = "default"
+    """Background Source"""
+
+
+class Config(BaseModel):
+
+    wakatime: ScopedConfig
+    """Wakatime Plugin Config"""
+
+
+config = get_plugin_config(Config).wakatime

nonebot_plugin_wakatime/migrations/a27298ba954d_first_revision.py

@@ -0,0 +1,42 @@
+"""first revision
+
+迁移 ID: a27298ba954d
+父迁移:
+创建时间: 2024-07-28 15:04:34.928642
+
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "a27298ba954d"
+down_revision: str | Sequence[str] | None = None
+branch_labels: str | Sequence[str] | None = ("nonebot_plugin_wakatime",)
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = "") -> None:
+    if name:
+        return
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "wakatime",
+        sa.Column("user_id", sa.String(), nullable=False),
+        sa.Column("platform", sa.String(), nullable=False),
+        sa.Column("access_token", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("user_id", name=op.f("pk_wakatime")),
+        info={"bind_key": "nonebot_plugin_wakatime"},
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade(name: str = "") -> None:
+    if name:
+        return
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("wakatime")
+    # ### end Alembic commands ###

nonebot_plugin_wakatime/models.py

@@ -0,0 +1,14 @@
+from nonebot_plugin_orm import Model
+from sqlalchemy.orm import Mapped, mapped_column
+
+
+class User(Model):
+
+    __tablename__ = "wakatime"
+
+    user_id: Mapped[str] = mapped_column(primary_key=True)
+    """User ID"""
+    platform: Mapped[str]
+    """User Account Platform"""
+    access_token: Mapped[str]
+    """Wakatime Access Token"""