fix ide warnings

This commit is contained in:
jxxghp
2025-03-28 19:43:55 +08:00
parent 58fa8064ad
commit b7ce5663a3
3 changed files with 58 additions and 58 deletions
-1
View File
@@ -1,4 +1,3 @@
import re
from typing import Any, Generator, List, Optional, Tuple, Union from typing import Any, Generator, List, Optional, Tuple, Union
from app import schemas from app import schemas
+54 -51
View File
@@ -4,7 +4,7 @@ import random
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum from enum import Enum
from typing import Optional, Union from typing import Optional, Union, List
from app.core.config import settings from app.core.config import settings
from app.log import logger from app.log import logger
@@ -25,8 +25,8 @@ class Category(Enum):
Others = "Others" Others = "Others"
@classmethod @classmethod
def _missing_(self, value): def _missing_(cls, value):
return self.Others return cls.Others
class Type(Enum): class Type(Enum):
@@ -38,8 +38,8 @@ class Type(Enum):
Directory = "Directory" Directory = "Directory"
@classmethod @classmethod
def _missing_(self, value): def _missing_(cls, value):
return self.Video return cls.Video
@dataclass @dataclass
@@ -131,14 +131,14 @@ class Api:
:return: 成功返回token 否则返回None :return: 成功返回token 否则返回None
""" """
if ( if (
res := self.__request_api( res := self.__request_api(
"/login", "/login",
data={ data={
"username": username, "username": username,
"password": password, "password": password,
"app_name": "trimemedia-web", "app_name": "trimemedia-web",
}, },
) )
) and res.success: ) and res.success:
self._token = res.data.get("token") self._token = res.data.get("token")
return self._token return self._token
@@ -148,7 +148,7 @@ class Api:
退出账号 退出账号
""" """
if (res := self.__request_api("/user/logout", method="post")) and res.success: if (res := self.__request_api("/user/logout", method="post")) and res.success:
if res.data == True: if res.data:
self._token = None self._token = None
return True return True
return False return False
@@ -173,9 +173,9 @@ class Api:
当前用户信息 当前用户信息
""" """
if (res := self.__request_api("/user/info")) and res.success: if (res := self.__request_api("/user/info")) and res.success:
user = User("", "") _user = User("", "")
user.__dict__.update(res.data) _user.__dict__.update(res.data)
return user return _user
return None return None
def mediadb_sum(self) -> Optional[MediaDbSumary]: def mediadb_sum(self) -> Optional[MediaDbSumary]:
@@ -183,17 +183,17 @@ class Api:
媒体数量统计 媒体数量统计
""" """
if (res := self.__request_api("/mediadb/sum")) and res.success: if (res := self.__request_api("/mediadb/sum")) and res.success:
sum = MediaDbSumary() sums = MediaDbSumary()
sum.__dict__.update(res.data) sums.__dict__.update(res.data)
return sum return sums
return None return None
def mediadb_list(self) -> Optional[MediaDbSumary]: def mediadb_list(self) -> Optional[List[MediaDb]]:
""" """
媒体库列表(普通用户) 媒体库列表(普通用户)
""" """
if (res := self.__request_api("/mediadb/list")) and res.success: if (res := self.__request_api("/mediadb/list")) and res.success:
items = [] _items = []
for info in res.data: for info in res.data:
mdb = MediaDb( mdb = MediaDb(
guid=info.get("guid"), guid=info.get("guid"),
@@ -204,8 +204,8 @@ class Api:
for poster in info.get("posters", []) for poster in info.get("posters", [])
], ],
) )
items.append(mdb) _items.append(mdb)
return items return _items
return None return None
def __build_img_api_url(self, img_path: Optional[str]) -> Optional[str]: def __build_img_api_url(self, img_path: Optional[str]) -> Optional[str]:
@@ -220,7 +220,7 @@ class Api:
媒体库列表(管理员) 媒体库列表(管理员)
""" """
if (res := self.__request_api("/mdb/list")) and res.success: if (res := self.__request_api("/mdb/list")) and res.success:
items = [] _items = []
for info in res.data: for info in res.data:
mdb = MediaDb( mdb = MediaDb(
guid=info.get("guid"), guid=info.get("guid"),
@@ -232,8 +232,8 @@ class Api:
], ],
dir_list=info.get("dir_list"), dir_list=info.get("dir_list"),
) )
items.append(mdb) _items.append(mdb)
return items return _items
return None return None
def mdb_scanall(self) -> bool: def mdb_scanall(self) -> bool:
@@ -241,7 +241,7 @@ class Api:
扫描所有媒体库 扫描所有媒体库
""" """
if (res := self.__request_api("/mdb/scanall", method="post")) and res.success: if (res := self.__request_api("/mdb/scanall", method="post")) and res.success:
if res.data == True: if res.data:
self._token = None self._token = None
return True return True
return False return False
@@ -251,9 +251,9 @@ class Api:
扫描指定媒体库 扫描指定媒体库
""" """
if ( if (
res := self.__request_api(f"/mdb/scan/{mdb.guid}", data={}) res := self.__request_api(f"/mdb/scan/{mdb.guid}", data={})
) and res.success: ) and res.success:
if res.data == True: if res.data:
self._token = None self._token = None
return True return True
return False return False
@@ -274,18 +274,20 @@ class Api:
return item return item
def item_list( def item_list(
self, self,
guid: Optional[str] = None, guid: Optional[str] = None,
type: Optional[list[Type]] = [Type.Movie, Type.TV, Type.Directory, Type.Video], type=None,
exclude_grouped_video=True, exclude_grouped_video=True,
page=1, page=1,
page_size=22, page_size=22,
sort_by="create_time", sort_by="create_time",
sort="DESC", sort="DESC",
) -> Optional[list[Item]]: ) -> Optional[list[Item]]:
""" """
媒体列表 媒体列表
""" """
if type is None:
type = [Type.Movie, Type.TV, Type.Directory, Type.Video]
post = { post = {
"tags": {"type": type} if type else {}, "tags": {"type": type} if type else {},
"sort_type": sort, "sort_type": sort,
@@ -307,7 +309,7 @@ class Api:
搜索影片、演员 搜索影片、演员
""" """
if ( if (
res := self.__request_api("/search/list", params={"q": keywords}) res := self.__request_api("/search/list", params={"q": keywords})
) and res.success: ) and res.success:
return [self.__build_item(info) for info in res.data] return [self.__build_item(info) for info in res.data]
return None return None
@@ -368,7 +370,7 @@ class Api:
return f"nonce={nonce}&timestamp={ts}&sign={sign}" return f"nonce={nonce}&timestamp={ts}&sign={sign}"
def __request_api( def __request_api(
self, api: str, method: str = None, params: dict = None, data: dict = None self, api: str, method: str = None, params: dict = None, data: dict = None
): ):
""" """
请求飞牛影视API 请求飞牛影视API
@@ -388,13 +390,14 @@ class Api:
def default(self, obj): def default(self, obj):
if isinstance(obj, Type): if isinstance(obj, Type):
return obj.value return obj.value
return super().default(self, obj) return super().default(obj)
if not self._host or not api: if not self._host or not api:
return None return None
if api[0] != "/": if not api.startswith("/"):
api = "/" + api api_path = f"{self._api_path}/{api}"
api_path = self._api_path + api else:
api_path = self._api_path + api
url = self._host + api_path url = self._host + api_path
if method is None: if method is None:
method = "get" if data is None else "post" method = "get" if data is None else "post"
@@ -430,17 +433,17 @@ class Api:
if __name__ == "__main__": if __name__ == "__main__":
api = Api("http://192.168.1.49:5666/", "16CCEB3D-AB42-077D-36A1-F355324E4237") fnApi = Api("http://192.168.1.49:5666/", "16CCEB3D-AB42-077D-36A1-F355324E4237")
api.login("adad", "123456") fnApi.login("adad", "123456")
logger.debug(f"token={api.token}") logger.debug(f"token={fnApi.token}")
user = api.user_info() user = fnApi.user_info()
logger.debug(user) logger.debug(user)
mediadbs = api.mdb_list() mediadbs = fnApi.mdb_list()
logger.debug(mediadbs) logger.debug(mediadbs)
items = api.item_list(mediadbs[0].guid, page=1, page_size=0) items = fnApi.item_list(mediadbs[0].guid, page=1, page_size=0)
logger.debug(items) logger.debug(items)
api.logout() fnApi.logout()
+4 -6
View File
@@ -1,4 +1,3 @@
import json
from pathlib import Path from pathlib import Path
from typing import Any, Dict, Generator, List, Optional, Tuple, Union from typing import Any, Dict, Generator, List, Optional, Tuple, Union
@@ -306,14 +305,13 @@ class TrimeMedia:
if path is None: if path is None:
return None return None
for lib in self._libraries.values(): for lib in self._libraries.values():
for dir in lib.dir_list or []: for d in lib.dir_list or []:
if is_subpath(path, Path(dir)): if is_subpath(path, Path(d)):
return lib return lib
return None return None
# TODO 飞牛似乎还没有这个功能
def get_webhook_message(self, body: any) -> Optional[schemas.WebhookEventInfo]: def get_webhook_message(self, body: any) -> Optional[schemas.WebhookEventInfo]:
return None pass
def get_iteminfo(self, itemid: str) -> Optional[schemas.MediaServerItem]: def get_iteminfo(self, itemid: str) -> Optional[schemas.MediaServerItem]:
""" """
@@ -442,7 +440,7 @@ class TrimeMedia:
yield self.__build_media_server_item(item) yield self.__build_media_server_item(item)
return None return None
def get_play_url(self, item_id: str) -> str: def get_play_url(self, item_id: str) -> Optional[str]:
""" """
获取媒体的外网播放链接 获取媒体的外网播放链接