Examples

Login

  1"""Example Code for Login.
  2
  3:Copyright: Copyright (C) 2021-2021  cscs181
  4:License: AGPL-3.0 or later. See `LICENSE`_ for detail.
  5
  6.. _LICENSE:
  7    https://github.com/cscs181/CAI/blob/master/LICENSE
  8"""
  9import os
 10import signal
 11import asyncio
 12import traceback
 13from io import BytesIO
 14from hashlib import md5
 15
 16from PIL import Image
 17
 18import cai
 19from cai.exceptions import (
 20    LoginException,
 21    ApiResponseError,
 22    LoginDeviceLocked,
 23    LoginSliderNeeded,
 24    LoginAccountFrozen,
 25    LoginCaptchaNeeded,
 26)
 27
 28
 29async def run():
 30    account = os.getenv("ACCOUNT", "")
 31    password = os.getenv("PASSWORD")
 32    try:
 33        account = int(account)
 34        assert password
 35    except Exception:
 36        print(
 37            f"Error: account '{account}', password '{password}'"  # type: ignore
 38        )
 39        return
 40
 41    try:
 42        client = await cai.login(account, md5(password.encode()).digest())
 43        print(f"Login Success! Client status: {client.status!r}")
 44    except Exception as e:
 45        await handle_failure(e)
 46
 47
 48async def handle_failure(exception: Exception):
 49    if isinstance(exception, LoginSliderNeeded):
 50        print("Verify url:", exception.verify_url)
 51        ticket = input("Please enter the ticket: ").strip()
 52        try:
 53            await cai.submit_slider_ticket(ticket)
 54            print("Login Success!")
 55            await asyncio.sleep(3)
 56        except Exception as e:
 57            await handle_failure(e)
 58    elif isinstance(exception, LoginCaptchaNeeded):
 59        print("Captcha:")
 60        image = Image.open(BytesIO(exception.captcha_image))
 61        image.show()
 62        captcha = input("Please enter the captcha: ").strip()
 63        try:
 64            await cai.submit_captcha(captcha, exception.captcha_sign)
 65            print("Login Success!")
 66            await asyncio.sleep(3)
 67        except Exception as e:
 68            await handle_failure(e)
 69    elif isinstance(exception, LoginAccountFrozen):
 70        print("Account is frozen!")
 71    elif isinstance(exception, LoginDeviceLocked):
 72        print("Device lock detected!")
 73        way = (
 74            "sms"
 75            if exception.sms_phone
 76            else "url"
 77            if exception.verify_url
 78            else ""
 79        )
 80        if exception.sms_phone and exception.verify_url:
 81            while True:
 82                choice = input(
 83                    f"1. Send sms message to {exception.sms_phone}.\n"
 84                    f"2. Verify device by scanning.\nChoose: "
 85                )
 86                if "1" in choice:
 87                    way = "sms"
 88                    break
 89                elif "2" in choice:
 90                    way = "url"
 91                    break
 92                print(f"'{choice}' is not valid!")
 93        if not way:
 94            print("No way to verify device...")
 95        elif way == "sms":
 96            await cai.request_sms()
 97            print(f"SMS was sent to {exception.sms_phone}!")
 98            sms_code = input("Please enter the sms_code: ").strip()
 99            try:
100                await cai.submit_sms(sms_code)
101            except Exception as e:
102                await handle_failure(e)
103        elif way == "url":
104            await cai.close()
105            print(f"Go to {exception.verify_url} to verify device!")
106            input("Press ENTER after verification to continue login...")
107            try:
108                await cai.login(exception.uin)
109            except Exception as e:
110                await handle_failure(e)
111    elif isinstance(exception, LoginException):
112        print("Login Error:", repr(exception))
113    elif isinstance(exception, ApiResponseError):
114        print("Response Error:", repr(exception))
115    else:
116        print("Unknown Error:", repr(exception))
117        traceback.print_exc()
118
119
120if __name__ == "__main__":
121    close = asyncio.Event()
122
123    async def wait_cleanup():
124        await close.wait()
125        await cai.close_all()
126
127    loop = asyncio.get_event_loop()
128    loop.add_signal_handler(signal.SIGINT, close.set)
129    loop.add_signal_handler(signal.SIGTERM, close.set)
130    loop.create_task(run())
131    loop.run_until_complete(wait_cleanup())

Set Client Status

 1"""Example Code for Set Client Status.
 2
 3:Copyright: Copyright (C) 2021-2021  cscs181
 4:License: AGPL-3.0 or later. See `LICENSE`_ for detail.
 5
 6.. _LICENSE:
 7    https://github.com/cscs181/CAI/blob/master/LICENSE
 8"""
 9import os
10import signal
11import asyncio
12from hashlib import md5
13
14import cai
15from cai.client import OnlineStatus
16
17
18async def run():
19    account = os.getenv("ACCOUNT", "")
20    password = os.getenv("PASSWORD")
21    try:
22        account = int(account)
23        assert password
24    except Exception:
25        print(
26            f"Error: account '{account}', password '{password}'"  # type: ignore
27        )
28        return
29
30    client = await cai.login(account, md5(password.encode()).digest())
31
32    await asyncio.sleep(10)
33    await cai.set_status(OnlineStatus.Qme)
34    print("Current client status: ", client.status)
35
36
37if __name__ == "__main__":
38    close = asyncio.Event()
39
40    async def wait_cleanup():
41        await close.wait()
42        await cai.close_all()
43
44    loop = asyncio.get_event_loop()
45    loop.add_signal_handler(signal.SIGINT, close.set)
46    loop.add_signal_handler(signal.SIGTERM, close.set)
47    loop.create_task(run())
48    loop.run_until_complete(wait_cleanup())

Friend and Group

 1"""Example Code for Friend/Group.
 2
 3:Copyright: Copyright (C) 2021-2021  cscs181
 4:License: AGPL-3.0 or later. See `LICENSE`_ for detail.
 5
 6.. _LICENSE:
 7    https://github.com/cscs181/CAI/blob/master/LICENSE
 8"""
 9import os
10import signal
11import asyncio
12from hashlib import md5
13
14import cai
15
16
17async def run():
18    account = os.getenv("ACCOUNT", "")
19    password = os.getenv("PASSWORD")
20    try:
21        account = int(account)
22        assert password
23    except Exception:
24        print(
25            f"Error: account '{account}', password '{password}'"  # type: ignore
26        )
27        return
28
29    client = await cai.login(account, md5(password.encode()).digest())
30
31    # friend
32    friend_list = await cai.get_friend_list()
33    friend_group_list = await cai.get_friend_group_list()
34    print("========== friends ==========", *friend_list, sep="\n")
35    print("========== friend groups ==========", *friend_group_list, sep="\n")
36    example_friend = friend_list[0]
37    # friend = await cai.get_friend(friend_uin)
38    print("========== example friend ==========")
39    print("uin: ", example_friend.uin)
40    print("nick: ", example_friend.nick)
41    print("remark: ", example_friend.remark)
42    print("group: ", await example_friend.get_group())
43
44    group_list = await cai.get_group_list()
45    print("\n========== group list ==========", *group_list, sep="\n")
46    example_group = group_list[0]
47    # group = await cai.get_group(group_id)
48    print("========== example group ==========")
49    print("group id: ", example_group.group_id)
50    print("group name: ", example_group.group_name)
51    print("group owner: ", example_group.group_owner_uin)
52    example_group_member_list = await example_group.get_members()
53    print(
54        "========== example group members ==========",
55        *example_group_member_list,
56        sep="\n",
57    )
58
59
60if __name__ == "__main__":
61    close = asyncio.Event()
62
63    async def wait_cleanup():
64        await close.wait()
65        await cai.close_all()
66
67    loop = asyncio.get_event_loop()
68    loop.add_signal_handler(signal.SIGINT, close.set)
69    loop.add_signal_handler(signal.SIGTERM, close.set)
70    loop.create_task(run())
71    loop.run_until_complete(wait_cleanup())

Message

 1"""Example Code for Message.
 2
 3:Copyright: Copyright (C) 2021-2021  cscs181
 4:License: AGPL-3.0 or later. See `LICENSE`_ for detail.
 5
 6.. _LICENSE:
 7    https://github.com/cscs181/CAI/blob/master/LICENSE
 8"""
 9import os
10import signal
11import asyncio
12from hashlib import md5
13
14import cai
15from cai.client import Event, Client, GroupMessage, PrivateMessage
16
17
18async def run():
19    account = os.getenv("ACCOUNT", "")
20    password = os.getenv("PASSWORD")
21    try:
22        account = int(account)
23        assert password
24    except Exception:
25        print(
26            f"Error: account '{account}', password '{password}'"  # type: ignore
27        )
28        return
29
30    client = await cai.login(account, md5(password.encode()).digest())
31
32    cai.add_event_listener(listen_message)
33    # cai.add_event_listener(listen_message, uin=account)
34    # client.add_event_listener(listen_message)
35
36
37async def listen_message(client: Client, event: Event):
38    if isinstance(event, PrivateMessage):
39        print("Private Message received from", event.from_uin)
40        print("Private Message elements:", event.message)
41    elif isinstance(event, GroupMessage):
42        print(
43            f"Group Message received from {event.group_name}({event.group_id})"
44        )
45        print("Group Message elements:", event.message)
46
47
48if __name__ == "__main__":
49    close = asyncio.Event()
50
51    async def wait_cleanup():
52        await close.wait()
53        await cai.close_all()
54
55    loop = asyncio.get_event_loop()
56    loop.add_signal_handler(signal.SIGINT, close.set)
57    loop.add_signal_handler(signal.SIGTERM, close.set)
58    loop.create_task(run())
59    loop.run_until_complete(wait_cleanup())