Merge pull request #44 from chenmozhijin/dev

0.9.1

Author: chenmozhijin

.github/workflows/test.yml

@@ -64,6 +64,9 @@ jobs:
                 mkdir -p tests/artifacts/reports
                 cp -r htmlcov ./tests/artifacts/reports
                 cp coverage.xml ./tests/artifacts/reports
+                export log_dir=$(python build_helper.py --task log_dir)
+                cp -r $log_dir ./tests/artifacts
+              shell: bash
 
             - name: Upload artifacts
               uses: actions/upload-artifact@v4

LDDC/common/data/cache.py

@@ -1,5 +1,6 @@
 # SPDX-FileCopyrightText: Copyright (C) 2024-2025 沉默の金 <cmzj@cmzj.org>
 # SPDX-License-Identifier: GPL-3.0-only
+import atexit
 from collections.abc import Callable
 from typing import Any, Literal, ParamSpec, TypeVar, overload
 
@@ -8,7 +9,7 @@
 from LDDC.common.paths import cache_dir
 
 cache = Cache(cache_dir, sqlitecache_size=512)
-cache_version = 5
+cache_version = 6
 if "version" not in cache or cache["version"] != cache_version:
     cache.clear()
 cache["version"] = cache_version
@@ -54,23 +55,42 @@ def cached_call(
     cache.set(key, result, expire=expire)
     return result
 
+
 @overload
 def get_cached_func(
-    func: Callable[P, T], typed: bool = True, ignore: set[int | str] | None = None, expire: int | None = None,
-) -> Callable[P, T]:...
+    func: Callable[P, T],
+    typed: bool = True,
+    ignore: set[int | str] | None = None,
+    expire: int | None = None,
+) -> Callable[P, T]: ...
+
 
 @overload
 def get_cached_func(
-    func: Callable[P, T], typed: bool = True, ignore: set[int | str] | None = None, expire: int | None = None, with_status: Literal[False] = False,
-) -> Callable[P, T]:...
+    func: Callable[P, T],
+    typed: bool = True,
+    ignore: set[int | str] | None = None,
+    expire: int | None = None,
+    with_status: Literal[False] = False,
+) -> Callable[P, T]: ...
+
 
 @overload
 def get_cached_func(
-    func: Callable[P, T], typed: bool = True, ignore: set[int | str] | None = None, expire: int | None = None, with_status: Literal[True] = True,
-) -> Callable[P, tuple[T, bool]]:...
+    func: Callable[P, T],
+    typed: bool = True,
+    ignore: set[int | str] | None = None,
+    expire: int | None = None,
+    with_status: Literal[True] = True,
+) -> Callable[P, tuple[T, bool]]: ...
+
 
 def get_cached_func(
-    func: Callable[P, T], typed: bool = True, ignore: set[int | str] | None = None, expire: int | None = None, with_status: bool = False,
+    func: Callable[P, T],
+    typed: bool = True,
+    ignore: set[int | str] | None = None,
+    expire: int | None = None,
+    with_status: bool = False,
 ) -> Callable[P, T] | Callable[P, tuple[T, bool]]:
     cache_settings = {"typed": typed, "ignore": ignore if ignore else set(), "expire": expire}
     if with_status:
@@ -151,3 +171,12 @@ def _buildcache_key(
         key += type_sig
 
     return key
+
+
+def _atexit() -> None:
+    cache["version"] = cache_version
+    cache.expire()
+    cache.close()
+
+
+atexit.register(_atexit)

LDDC/common/task_manager.py

@@ -12,7 +12,7 @@
 
 from .logger import logger
 from .models import P, T
-from .thread import in_other_thread, threadpool
+from .thread import in_other_thread, is_exited, threadpool
 
 
 class TaskManager:
@@ -252,7 +252,7 @@ def __init__(self) -> None:
 
     @property
     def is_stopped(self) -> bool:
-        return self.taskmanager.is_finished(self.task_type, self.taskid)
+        return self.taskmanager.is_finished(self.task_type, self.taskid) or is_exited()
 
     def finished_task(self) -> None:
         self.taskmanager.finished_task(self.task_type, self.taskid)

LDDC/common/thread.py

@@ -2,15 +2,26 @@
 # SPDX-License-Identifier: GPL-3.0-only
 from collections.abc import Callable, Iterable
 from functools import partial, wraps
-from threading import Lock
+from threading import Event, Lock
 from typing import Any
 
 from PySide6.QtCore import QCoreApplication, QEvent, QObject, QRunnable, Qt, QThread, QThreadPool, Signal, Slot
 
 from .logger import logger
 from .models import P, T
 
-threadpool = QThreadPool()
+exit_event = Event()
+
+
+def is_exited() -> bool:
+    return exit_event.is_set()
+
+
+def set_exited() -> None:
+    exit_event.set()
+
+
+threadpool = QThreadPool(QCoreApplication.instance())
 if threadpool.maxThreadCount() < 8:
     threadpool.setMaxThreadCount(8)
 
@@ -76,8 +87,13 @@ def __init__(self, func: Callable[..., T], emitter: SignalEmitter) -> None:
         self.emitter = emitter
 
     def run(self) -> None:
+        if is_exited():
+            return
         try:
-            self.emitter.success.emit(self.func())  # 成功时发射信号
+            result = self.func()
+            if is_exited():
+                return
+            self.emitter.success.emit(result)  # 成功时发射信号
         except Exception as e:
             self.emitter.error.emit(e)  # 失败时发射信号
             logger.exception(e)

LDDC/common/version.py

@@ -3,7 +3,7 @@
 
 """版本信息与处理模块"""
 
-__version__ = "v0.9.0"
+__version__ = "v0.9.1"
 import re
 from typing import Literal
 

LDDC/core/api/lyrics/kg.py

@@ -4,6 +4,7 @@
 # ruff: noqa: S311 S324
 import hashlib
 import json
+import random
 import time
 from base64 import b64decode, b64encode
 from datetime import datetime, timedelta, timezone
@@ -137,7 +138,8 @@ def request(
                 "uuid": "-",
                 # "uuid": "15e772e1213bdd0718d0c1d10d64e06f",
                 "mid": mid,
-                "dfid": self.dfid,
+                # "dfid": self.dfid,
+                "dfid": "-",
                 "clientver": "11070",
                 "platform": "AndroidFilter",
                 **params,
@@ -181,7 +183,11 @@ def search(self, keyword: str, search_type: SearchType, page: int = 1) -> APIRes
             "page": page,
         }
         url, module = SEARCH_TYPE_MAPPING[search_type]
-        data = self.request(url, params, module, headers={"x-router": "complexsearch.kugou.com"})
+        try:
+            data = self.request(url, params, module, headers={"x-router": "complexsearch.kugou.com"})
+        except APIRequestError:
+            logger.exception("kg API请求错误,尝试使用旧接口")
+            return self._old_search(keyword, search_type, page)
 
         if not data["data"]["lists"]:
             return APIResultList(
@@ -202,7 +208,7 @@ def search(self, keyword: str, search_type: SearchType, page: int = 1) -> APIRes
                     [
                         SongInfo(
                             source=self.source,
-                            id=info["ID"],
+                            id=str(info["ID"]),
                             hash=info["FileHash"],
                             title=info["SongName"],
                             subtitle=info["Auxiliary"],
@@ -232,7 +238,7 @@ def search(self, keyword: str, search_type: SearchType, page: int = 1) -> APIRes
                         SongListInfo(
                             source=self.source,
                             type=SongListType.ALBUM,
-                            id=info["albumid"],
+                            id=str(info["albumid"]),
                             title=info["albumname"],
                             imgurl=info["img"],
                             songcount=info["songcount"],
@@ -267,7 +273,7 @@ def search(self, keyword: str, search_type: SearchType, page: int = 1) -> APIRes
                         SongListInfo(
                             source=self.source,
                             type=SongListType.SONGLIST,
-                            id=info["gid"],
+                            id=str(info["gid"]),
                             title=info["specialname"],
                             imgurl=info["img"],
                             songcount=info["song_count"],
@@ -297,6 +303,155 @@ def search(self, keyword: str, search_type: SearchType, page: int = 1) -> APIRes
             case _:
                 raise NotImplementedError
 
+    def _old_search(self, keyword: str, search_type: SearchType, page: int = 1) -> APIResultList[SongInfo] | APIResultList[SongListInfo]:
+        """备用搜索API"""
+        domain = random.choice(["mobiles.kugou.com", "msearchcdn.kugou.com", "mobilecdnbj.kugou.com", "msearch.kugou.com"])
+        pagesize = 20
+
+        match search_type:
+            case SearchType.SONG:
+                url = f"http://{domain}/api/v3/search/song"
+                params = {
+                    "showtype": "14",
+                    "highlight": "",
+                    "pagesize": "30",
+                    "tag_aggr": "1",
+                    "plat": "0",
+                    "sver": "5",
+                    "keyword": keyword,
+                    "correct": "1",
+                    "api_ver": "1",
+                    "version": "9108",
+                    "page": page,
+                }
+            case SearchType.SONGLIST:
+                url = f"http://{domain}/api/v3/search/special"
+                params = {
+                    "version": "9108",
+                    "highlight": "",
+                    "keyword": keyword,
+                    "pagesize": "20",
+                    "filter": "0",
+                    "page": page,
+                    "sver": "2",
+                }
+            case SearchType.ALBUM:
+                url = f"http://{domain}/api/v3/search/album"
+                params = {
+                    "version": "9108",
+                    "iscorrection": "1",
+                    "highlight": "",
+                    "plat": "0",
+                    "keyword": keyword,
+                    "pagesize": "20",
+                    "page": page,
+                    "sver": "2",
+                }
+
+        response = self.client.get(url, params=params, timeout=3)
+        response.raise_for_status()
+        data = response.json()
+        start_index = (page - 1) * pagesize
+        match search_type:
+            case SearchType.SONG:
+                return APIResultList(
+                    [
+                        SongInfo(
+                            source=self.source,
+                            id=str(info["album_audio_id"]),
+                            hash=info["hash"],
+                            title=info["songname"],
+                            subtitle=info["topic"],
+                            artist=Artist(info["singername"].split("、")),
+                            album=info["album_name"],
+                            duration=info["duration"] * 1000,
+                            language=LANGUAGE_MAPPING.get(info["trans_param"].get("language"), Language.OTHER),
+                        )
+                        for info in data["data"]["info"]
+                    ],
+                    SearchInfo(
+                        source=self.source,
+                        keyword=keyword,
+                        search_type=search_type,
+                        page=page,
+                    ),
+                    (
+                        start_index,
+                        start_index + len(data["data"]["info"]) - 1,
+                        data["data"]["total"] if len(data["data"]["info"]) == pagesize else start_index + len(data["data"]["info"]),
+                    ),
+                )
+            case SearchType.SONGLIST:
+                return APIResultList(
+                    [
+                        SongListInfo(
+                            source=self.source,
+                            type=SongListType.SONGLIST,
+                            id=str(info["specialid"]),
+                            title=info["specialname"],
+                            imgurl=info["imgurl"],
+                            songcount=info["songcount"],
+                            publishtime=int(
+                                datetime.strptime(info["publishtime"], "%Y-%m-%d %H:%M:%S")
+                                .replace(tzinfo=timezone(timedelta(hours=8)))
+                                .astimezone(timezone.utc)
+                                .timestamp(),
+                            )
+                            if info["publishtime"] != "0000-00-00 00:00:00"
+                            else None,
+                            author=info["nickname"],
+                        )
+                        for info in data["data"]["info"]
+                    ],
+                    SearchInfo(
+                        source=self.source,
+                        keyword=keyword,
+                        search_type=search_type,
+                        page=page,
+                    ),
+                    (
+                        start_index,
+                        start_index + len(data["data"]["info"]) - 1,
+                        data["data"]["total"] if len(data["data"]["info"]) == pagesize else start_index + len(data["data"]["info"]),
+                    ),
+                )
+            case SearchType.ALBUM:
+                return APIResultList(
+                    [
+                        SongListInfo(
+                            source=self.source,
+                            type=SongListType.ALBUM,
+                            id=str(info["albumid"]),
+                            title=info["albumname"],
+                            imgurl=info["imgurl"],
+                            songcount=info["songcount"],
+                            publishtime=int(
+                                datetime.strptime(info["publishtime"], "%Y-%m-%d %H:%M:%S")
+                                .replace(tzinfo=timezone(timedelta(hours=8)))
+                                .astimezone(timezone.utc)
+                                .timestamp(),
+                            )
+                            if info["publishtime"] != "0000-00-00 00:00:00"
+                            else None,
+                            author=info["singer"],
+                        )
+                        for info in data["data"]["info"]
+                    ],
+                    SearchInfo(
+                        source=self.source,
+                        keyword=keyword,
+                        search_type=search_type,
+                        page=page,
+                    ),
+                    (
+                        start_index,
+                        start_index + len(data["data"]["info"]) - 1,
+                        data["data"]["total"] if len(data["data"]["info"]) == pagesize else start_index + len(data["data"]["info"]),
+                    ),
+                )
+            case _:
+                raise NotImplementedError
+
     def get_songlist(self, songlist_info: SongListInfo) -> APIResultList[SongInfo]:
         match songlist_info.type:
             case SongListType.ALBUM:
@@ -402,7 +557,7 @@ def get_lyricslist(self, song_info: SongInfo) -> APIResultList[LyricInfo]:
             [
                 LyricInfo(
                     source=self.source,
-                    id=lyric["id"],
+                    id=str(lyric["id"]),
                     accesskey=lyric["accesskey"],
                     creator=lyric["nickname"],
                     duration=lyric["duration"],