-
Notifications
You must be signed in to change notification settings - Fork 29
Expand file tree
/
Copy path__init__.py
More file actions
87 lines (75 loc) · 2.7 KB
/
__init__.py
File metadata and controls
87 lines (75 loc) · 2.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
import json
from typing import Optional
from lagrange.info import AppInfo
from typing_extensions import Literal
import asyncio
from .client.client import Client as Client
# from .client.server_push.msg import msg_push_handler
# from .client.server_push.service import server_kick_handler
from .utils.log import log as log
from .utils.log import install_loguru as install_loguru
from .utils.sign import sign_provider
from .info import InfoManager
from .info.app import app_list
from .utils.binary.protobuf.models import evaluate_all
class Lagrange:
client: Client
def __init__(
self,
uin: int,
protocol: Literal["linux", "macos", "windows", "custom"] = "linux",
sign_url: Optional[str] = None,
device_info_path="./device.json",
signinfo_path="./sig.bin",
custom_protocol_path="./protocol.json",
):
self.im = InfoManager(uin, device_info_path, signinfo_path)
self.uin = uin
self.sign = sign_provider(sign_url) if sign_url else None
self.events = {}
self.log = log
self._protocol = protocol
self._protocol_path = custom_protocol_path
def subscribe(self, event, handler):
self.events[event] = handler
async def login(self, client: Client):
if self.im.sig_info.d2:
if not await client.register():
return await client.login()
return True
else:
return await client.login()
async def run(self):
if self._protocol == "custom":
log.root.debug("load custom protocol from %s" % self._protocol_path)
with open(self._protocol_path, "r") as f:
proto = json.loads(f.read())
app_info = AppInfo.load_custom(proto)
else:
app_info = app_list[self._protocol]
log.root.info(f"AppInfo: platform={app_info.os}, ver={app_info.build_version}({app_info.sub_app_id})")
with self.im as im:
self.client = Client(
self.uin,
app_info,
im.device,
im.sig_info,
self.sign,
)
for event, handler in self.events.items():
self.client.events.subscribe(event, handler)
self.client.connect()
status = await self.login(self.client)
if not status:
log.login.error("Login failed")
return
await self.client.wait_closed()
def launch(self):
try:
asyncio.run(self.run())
except KeyboardInterrupt:
self.client._task_clear()
log.root.info("Program exited by user")
else:
log.root.info("Program exited normally")
evaluate_all()