跳转至

utils.session

Session 管理

ApiConfig

Bases: TypedDict

API 配置

Session 

Session(*, credential: Credential | None = None, enable_sign: bool = False, **kwargs)

Bases: AsyncClient

Session 类,用于管理 QQ 音乐的登录态和 API 请求

Source code in qqmusic_api/utils/session.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(self, *, credential: Credential | None = None, enable_sign: bool = False, **kwargs) -> None:
    super().__init__(**kwargs)
    self.credential = credential or Credential()
    self.headers = httpx.Headers(
        {
            "User-Agent": self.UA_DEFAULT,
            "Referer": self.HOST,
        }
    )
    self.timeout = 20
    self.api_config = ApiConfig(
        version="13.2.5.8",
        version_code=13020508,
        enable_sign=enable_sign,
        endpoint="https://u.y.qq.com/cgi-bin/musicu.fcg",
        enc_endpoint="https://u.y.qq.com/cgi-bin/musics.fcg",
    )
    self.qimei = get_qimei(get_cached_device(), self.api_config["version"])["q36"]

musicid property 

musicid: int

获取 musicid

musickey property 

musickey: str

获取 musickey

SessionManager 

SessionManager()

Session 管理类,负责管理不同事件循环的 Session

初始化 SessionManager

Source code in qqmusic_api/utils/session.py 
76
77
78
79
80
81
def __init__(self) -> None:
    """初始化 SessionManager"""
    self.session_pool: dict[asyncio.AbstractEventLoop, Session] = {}  # 存储全局 Session 池
    self.context_stack: dict[
        asyncio.AbstractEventLoop, deque[Session]
    ] = {}  # 使用 deque 存储 async with 上下文中的 Session 栈

get 

get() -> Session

获取当前事件循环的 Session

Source code in qqmusic_api/utils/session.py 
83
84
85
86
87
88
89
90
91
92
93
94
def get(self) -> Session:
    """获取当前事件循环的 Session"""
    loop = get_loop()
    # 优先从上下文栈中获取当前事件循环的 Session
    if self.context_stack.get(loop):
        return self.context_stack[loop][-1]
    # 如果上下文栈中没有,返回全局池中的 Session
    session = self.session_pool.get(loop, None)
    if not session:
        session = Session()
        self.session_pool[loop] = session
    return session

set 

set(session: Session) -> None

设置当前事件循环的 Session(全局池)

Source code in qqmusic_api/utils/session.py 
 96
 97
 98
 99
100
101
102
def set(self, session: Session) -> None:
    """设置当前事件循环的 Session(全局池)"""
    loop = get_loop()
    # 如果当前事件循环正在 async with 中,不能直接设置 Session
    if self.context_stack.get(loop):
        raise Exception("不能在 `async with` 块中设置 Session.")
    self.session_pool[loop] = session

push_to_stack 

push_to_stack(loop: AbstractEventLoop, session: Session) -> None

将 Session 推入当前事件循环的 async with 上下文栈

Source code in qqmusic_api/utils/session.py 
104
105
106
107
108
def push_to_stack(self, loop: asyncio.AbstractEventLoop, session: Session) -> None:
    """将 Session 推入当前事件循环的 async with 上下文栈"""
    if loop not in self.context_stack:
        self.context_stack[loop] = deque()  # 初始化 deque
    self.context_stack[loop].append(session)

pop_from_stack 

pop_from_stack(loop: AbstractEventLoop) -> None

从当前事件循环的 async with 上下文栈中弹出 Session

Source code in qqmusic_api/utils/session.py 
110
111
112
113
def pop_from_stack(self, loop: asyncio.AbstractEventLoop) -> None:
    """从当前事件循环的 async with 上下文栈中弹出 Session"""
    if self.context_stack.get(loop):
        self.context_stack[loop].pop()

create_session 

create_session(credential: Credential | None = None, enable_sign: bool = False) -> Session

创建新的 Session

Source code in qqmusic_api/utils/session.py 
115
116
117
118
119
def create_session(self, credential: Credential | None = None, enable_sign: bool = False) -> Session:
    """创建新的 Session"""
    session = Session(credential=credential, enable_sign=enable_sign)
    self.session_pool[get_loop()] = session
    return session

reset 

reset() -> None

重置当前事件循环的 Session

Source code in qqmusic_api/utils/session.py 
121
122
123
124
125
def reset(self) -> None:
    """重置当前事件循环的 Session"""
    loop = get_loop()
    self.session_pool.pop(loop, None)
    self.context_stack.pop(loop, None)

get_session 

get_session() -> Session

获取当前事件循环的 Session

Source code in qqmusic_api/utils/session.py 
132
133
134
def get_session() -> Session:
    """获取当前事件循环的 Session"""
    return session_manager.get()

set_session 

set_session(session: Session) -> None

设置当前事件循环的 Session

Source code in qqmusic_api/utils/session.py 
137
138
139
def set_session(session: Session) -> None:
    """设置当前事件循环的 Session"""
    session_manager.set(session)

create_session 

create_session(credential: Credential | None = None, enable_sign: bool = False) -> Session

创建新的 Session

PARAMETER DESCRIPTION
credential

凭据

TYPE: Credential | None DEFAULT: None

enable_sign

是否启用 sign

TYPE: bool DEFAULT: False

Source code in qqmusic_api/utils/session.py 
142
143
144
145
146
147
148
149
def create_session(credential: Credential | None = None, enable_sign: bool = False) -> Session:
    """创建新的 Session

    Args:
        credential: 凭据
        enable_sign: 是否启用 sign
    """
    return session_manager.create_session(credential=credential, enable_sign=enable_sign)

set_session_credential 

set_session_credential(credential: Credential)

设置当前 Session 的凭据

Source code in qqmusic_api/utils/session.py 
152
153
154
def set_session_credential(credential: Credential):
    """设置当前 Session 的凭据"""
    session_manager.get().credential = credential

get_loop 

get_loop() -> AbstractEventLoop

获取当前事件循环

Source code in qqmusic_api/utils/session.py 
157
158
159
160
161
162
def get_loop() -> asyncio.AbstractEventLoop:
    """获取当前事件循环"""
    try:
        return asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.new_event_loop()