修复部分 Codacy 报警

.github/workflows/ci.yml

@@ -8,12 +8,16 @@ on:
 jobs:
   check:
     runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
     steps:
       - name: Checkout code
         uses: actions/checkout@v7
         with:
           ref: ${{ github.head_ref }}
           fetch-depth: 0
+          persist-credentials: false
 
       - name: Set up Python
         uses: actions/setup-python@v6
@@ -44,13 +48,15 @@ jobs:
           poetry run nb larkhelp-generate zh_hans COMMANDS.md
 
       - name: Commit and push changes
+        env:
+          HEAD_REF: ${{ github.head_ref }}
         run: |
           git add .
           if ! git diff --cached --quiet; then
             git config --global user.name "github-actions[bot]"
             git config --global user.email "41898282+github-actions[bot]@users.noreply.github.com"
             git commit -m "Auto update from GitHub Actions"
-            git push origin HEAD:${{ github.head_ref }}
+            git push origin "HEAD:$HEAD_REF"
           else
             echo "No changes to commit."
           fi

migrations/versions/878331db0e57_add_wakeuprank_risedata_table.py

@@ -28,10 +28,10 @@ def upgrade(name: str = "") -> None:
     bind = op.get_bind()
     dialect = bind.dialect.name
     if dialect == "mysql":
-        table_check = f"SHOW TABLES LIKE '{TABLE_NAME}'"
+        table_check = sa.text("SHOW TABLES LIKE :table_name")
     else:
-        table_check = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{TABLE_NAME}'"
-    result = bind.execute(sa.text(table_check)).fetchall()
+        table_check = sa.text("SELECT name FROM sqlite_master WHERE type='table' AND name=:table_name")
+    result = bind.execute(table_check, {"table_name": TABLE_NAME}).fetchall()
     if result:
         return
 

migrations/versions/b4c5d6e7f8a9_add_diarypost_table.py

@@ -12,6 +12,15 @@
 
 from alembic import op
 import sqlalchemy as sa
+from sqlalchemy import (
+    bindparam,
+    column,
+    delete as sa_delete,
+    func,
+    insert as sa_insert,
+    select,
+    table,
+)
 
 # revision identifiers, used by Alembic.
 revision: str = "b4c5d6e7f8a9"
@@ -31,10 +40,10 @@ def upgrade(name: str = "") -> None:
     bind = op.get_bind()
     dialect = bind.dialect.name
     if dialect == "mysql":
-        table_check = f"SHOW TABLES LIKE '{DIARYPOST_TABLE}'"
+        table_check = sa.text("SHOW TABLES LIKE :table_name")
     else:
-        table_check = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{DIARYPOST_TABLE}'"
-    result = bind.execute(sa.text(table_check)).fetchall()
+        table_check = sa.text("SELECT name FROM sqlite_master WHERE type='table' AND name=:table_name")
+    result = bind.execute(table_check, {"table_name": DIARYPOST_TABLE}).fetchall()
     if result:
         # 表已存在（之前迁移部分执行过），跳过
         return
@@ -55,23 +64,37 @@ def upgrade(name: str = "") -> None:
     # 2. 将 note 表中 context_id='moonlark_diary' 的数据迁移到 diarypost
     # created_time 是 Float 类型的 Unix 时间戳，需要转换为 DateTime
     # MySQL 用 FROM_UNIXTIME()，SQLite 用 datetime(..., 'unixepoch', 'localtime')
+    diarypost = table(
+        DIARYPOST_TABLE,
+        column("content"),
+        column("keywords"),
+        column("created_at"),
+        column("expire_at"),
+    )
+    note = table(
+        NOTE_TABLE,
+        column("content"),
+        column("keywords"),
+        column("created_time"),
+        column("expire_time"),
+        column("context_id"),
+    )
     if dialect == "mysql":
-        created_at_expr = "FROM_UNIXTIME(created_time)"
+        created_at_col = func.from_unixtime(note.c.created_time)
     else:
-        created_at_expr = "datetime(created_time, 'unixepoch', 'localtime')"
-    op.execute(f"""
-        INSERT INTO {DIARYPOST_TABLE} (content, keywords, created_at, expire_at)
-        SELECT
-            content,
-            COALESCE(keywords, ''),
-            {created_at_expr},
-            expire_time
-        FROM {NOTE_TABLE}
-        WHERE context_id = '{DIARY_CONTEXT_ID}'
-    """)
+        created_at_col = func.datetime(note.c.created_time, "unixepoch", "localtime")
+    select_stmt = select(note.c.content, func.coalesce(note.c.keywords, ""), created_at_col, note.c.expire_time).where(
+        note.c.context_id == bindparam("context_id")
+    )
+    insert_stmt = sa_insert(diarypost).from_select(
+        ["content", "keywords", "created_at", "expire_at"],
+        select_stmt,
+    )
+    bind.execute(insert_stmt, {"context_id": DIARY_CONTEXT_ID})
 
     # 3. 删除已迁移的 note 记录
-    op.execute(f"DELETE FROM {NOTE_TABLE} WHERE context_id = '{DIARY_CONTEXT_ID}'")
+    delete_stmt = sa_delete(note).where(note.c.context_id == bindparam("context_id"))
+    bind.execute(delete_stmt, {"context_id": DIARY_CONTEXT_ID})
 
 
 def downgrade(name: str = "") -> None:

migrations/versions/c3d4e5f6a7b8_add_timer_table.py

@@ -28,10 +28,10 @@ def upgrade(name: str = "") -> None:
     bind = op.get_bind()
     dialect = bind.dialect.name
     if dialect == "mysql":
-        table_check = f"SHOW TABLES LIKE '{TABLE_NAME}'"
+        table_check = sa.text("SHOW TABLES LIKE :table_name")
     else:
-        table_check = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{TABLE_NAME}'"
-    result = bind.execute(sa.text(table_check)).fetchall()
+        table_check = sa.text("SELECT name FROM sqlite_master WHERE type='table' AND name=:table_name")
+    result = bind.execute(table_check, {"table_name": TABLE_NAME}).fetchall()
     if result:
         return
 

src/plugins/nonebot_plugin_larkcave/commands/remove.py

@@ -40,7 +40,7 @@ async def _(
         await lang.finish("remove_comment.no_result", user_id, comment_id)
         return
     if not (comment.author == user_id or is_superuser):
-        await lang.reply()
+        await lang.reply("remove_comment.no_permission", user_id)
         await cave.finish()
     await lang.send(
         "remove_comment.info",
@@ -62,13 +62,12 @@ async def _(
     try:
         cave_data = await session.get_one(CaveData, {"id": cave_id})
     except NoResultFound:
-        await lang.reply()
-        await cave.finish()
+        await lang.finish("remove.no_result", user_id, cave_id)
     if not (cave_data.author == user_id or is_superuser):
-        await lang.reply()
+        await lang.reply("remove.no_permission", user_id)
         await cave.finish()
     if not cave_data.public:
-        await lang.reply()
+        await lang.reply("remove.private", user_id, cave_id)
         await cave.finish()
     cave_data.public = False
     session.add(

src/plugins/nonebot_plugin_larksetu/plugins/nonebot_plugin_seturank/__main__.py

@@ -31,7 +31,7 @@ async def _(session: async_scoped_session, user_id: str = get_user_id()) -> None
 
 class SetuRanking(WebRanking):
 
-    async def get_sorted_data(self) -> list[RankingData]:
+    async def get_sorted_data(self, user_id: str) -> list[RankingData]:
         async with get_session() as session:
             result = (
                 (await session.execute(select(models.UserData).order_by(models.UserData.count.desc()))).scalars().all()

src/plugins/nonebot_plugin_quick_math/utils/ranking.py

@@ -17,7 +17,7 @@ async def get_user_list(order_by: Any = QuickMathUser.max_point) -> AsyncGenerat
 
 class RecordRanking(WebRanking):
 
-    async def get_sorted_data(self) -> list[RankingData]:
+    async def get_sorted_data(self, user_id: str) -> list[RankingData]:
         return [
             {
                 "user_id": user.user_id,
@@ -30,7 +30,7 @@ async def get_sorted_data(self) -> list[RankingData]:
 
 class TotalRanking(WebRanking):
 
-    async def get_sorted_data(self) -> list[RankingData]:
+    async def get_sorted_data(self, user_id: str) -> list[RankingData]:
         return [
             {
                 "user_id": user.user_id,

src/plugins/nonebot_plugin_ranking/web.py

@@ -33,11 +33,7 @@ async def get_name(self, user_id: str) -> str:
     async def handle(
         self, request: Request, offset: int = 0, limit: int = 20, user_id: str = get_user_id("-1")
     ) -> RankingResponse:
-        try:
-            data = await self.get_sorted_data(user_id)
-        except TypeError:
-            # 兼容旧版
-            data: list[RankingData] = await self.get_sorted_data()  # type: ignore
+        data = await self.get_sorted_data(user_id)
         index = offset
         return {
             "me": await find_user(data, user_id),