Это асинхронное ядро ICQ-клиента, реализующее протокол OSCAR. Оно не содержит графического интерфейса и предназначено для встраивания в другие приложения. Библиотека написана на Python с использованием asyncio.
Библиотека использует только стандартную библиотеку Python (asyncio, logging, struct, time, re, dataclasses, enum, typing). Никаких внешних зависимостей не требуется.
import asyncio
from icq_core import ICQClient, Status
async def main():
# Инициализация клиента
client = ICQClient("12345678", "password")
# Назначение обработчиков событий
client.on_contact_online = lambda c: print(f"{c.display_name} онлайн")
client.on_message = lambda m: print(f"{m.sender_uin}: {m.text}")
client.on_connected = lambda: print("Подключено!")
client.on_disconnected = lambda: print("Отключено")
# Запуск клиента
await client.run()
if __name__ == "__main__":
asyncio.run(main())Основной класс для взаимодействия с сервером ICQ.
client = ICQClient(uin: str, password: str, server: str = SERVER, port: int = PORT)uin: Ваш UIN (номер ICQ).password: Ваш пароль.server: Адрес сервера (по умолчанию195.66.114.37).port: Порт сервера (по умолчанию5190).
client.my_nick: Ваш никнейм (заполняется послеrequest_my_info()илиsave_my_info()).client.my_info: ОбъектUserInfoс вашей полной анкетой.client.contacts: Словарь{uin: Contact}со списком контактов.client.groups: Словарь{group_id: Group}со списком групп.
Представляет контакт в списке.
uin: UIN контакта.name: Отображаемое имя (ник).group_id: ID группы, в которой находится контакт.item_id: Внутренний ID элемента в SSI (Server-Side Storage).status: Текущий статус (Status.ONLINE,Status.AWAYи т.д.).status_msg: Сообщение статуса.client: Название клиента контакта (например, "QIP 2005a", "Jimm").xstatus: Расширенный статус (xStatus), например, "coffee", "music".xstatus_msg: Сообщение расширенного статуса.display_name: Свойство, возвращающееnameилиuin, если имя пустое.is_online: Свойство, возвращающееTrue, если контакт онлайн.
Представляет группу контактов.
group_id: ID группы.name: Название группы.
Предствует сообщение.
sender_uin: UIN отправителя.text: Текст сообщения.timestamp: Время получения (unix timestamp).is_outgoing:True, если сообщение исходящее (в данной реализации используется только для входящих).
Представляет полную анкету пользователя.
uin,nick,first_name,last_name,email,city,state,phone,fax,address,cell_phone,age,gender("M" или "F"),home_page,birthday("DD.MM.YYYY"),about.- Поля работы:
work_city,work_state,work_phone,work_fax,work_addr,work_name(компания),work_dep(отдел),work_pos(должность). full_name: Свойство, возвращающее "Имя Фамилия".
Представляет результат поиска пользователя.
uin,nick,first_name,last_name,email,auth_req(требуется ли авторизация),online,gender,age.
Статусы пользователя:
Status.ONLINEStatus.AWAYStatus.DND(Do Not Disturb)Status.NA(Not Available)Status.FREEStatus.OFFLINE
Назначьте функции этим атрибутам клиента для обработки событий. Функции могут быть как обычными, так и асинхронными (async def).
on_connected(): Вызывается при успешном подключении и инициализации сессии.on_disconnected(): Вызывается при отключении от сервера.on_roster(groups: list[Group], contacts: list[Contact]): Вызывается после получения полного списка контактов и групп.on_contact_online(contact: Contact): Вызывается, когда контакт появляется в сети.on_contact_offline(contact: Contact): Вызывается, когда контакт уходит из сети.on_contact_status(contact: Contact): Вызывается при изменении статуса контакта (включая онлайн/оффлайн).on_message(message: Message): Вызывается при получении нового сообщения.on_typing(uin: str, is_typing: bool): Вызывается, когда контакт начинает или заканчивает печатать сообщение.on_error(exc: Exception): Вызывается при возникновении ошибки (например, проблемы с соединением).on_xstatus_updated(contact: Contact): Вызывается при обновлении xStatus контакта.on_my_info(info: UserInfo): Вызывается после получения собственной анкеты черезrequest_my_info().on_user_info(info: UserInfo): Вызывается после получения анкеты другого пользователя черезrequest_user_info().on_search_result(result: SearchResult): Вызывается для каждого найденного пользователя в процессе поиска.on_search_done(results: list[SearchResult]): Вызывается после завершения поиска, содержит полный список результатов.on_offline_message(message: Message): Вызывается для каждого оффлайн-сообщения, полученного из очереди при входе.on_auth_request(uin: str, message: str): Вызывается, когда кто-то запрашивает разрешение добавить вас в свой список контактов.on_auth_reply(uin: str, granted: bool, message: str): Вызывается, когда пользователь отвечает на ваш запрос авторизации (granted=Trueесли разрешено).on_you_were_added(uin: str): Вызывается, когда кто-то добавил вас в свой список контактов (без запроса, если у него открытые настройки приватности).
Все методы, взаимодействующие с сетью, являются асинхронными и должны вызываться с await.
await client.set_status(status: Status, message: str = "")Устанавливает ваш статус и сообщение статуса.
await client.set_xstatus(name: str, title: str = "", desc: str = "")Устанавливает расширенный статус (xStatus).
name: Название xStatus (например, "coffee", "music", "angry"). Полный список см. вXSTATUS_BY_NAMEилиXSTATUS_TABLEв исходном коде.title: Заголовок xStatus (опционально).desc: Описание xStatus (опционально).
await client.send_message(to_uin: str, text: str)Отправляет текстовое сообщение пользователю с указанным UIN. Длинные сообщения автоматически разбиваются на части.
await client.send_typing(to_uin: str, is_typing: bool = True)Отправляет уведомление о наборе текста ("печатает...").
await client.add_contact(uin: str, nick: str = "", group_id: int = 0) -> boolДобавляет пользователя в список контактов. Возвращает True при успехе.
uin: UIN добавляемого пользователя.nick: Отображаемое имя (если пусто, будет использован UIN).group_id: ID группы, в которую добавить контакт.
await client.remove_contact(uin: str) -> boolУдаляет пользователя из списка контактов. Возвращает True при успехе.
await client.rename_contact(uin: str, new_nick: str) -> boolПереименовывает контакт в списке. Возвращает True при успехе.
await client.move_contact(uin: str, new_group_id: int) -> boolПеремещает контакт в другую группу. Возвращает True при успехе.
await client.create_group(name: str) -> tuple[bool, Optional[Group]]Создает новую группу контактов. Возвращает кортеж (успех, объект_группы).
await client.delete_group(group_id: int) -> boolУдаляет группу. Группа должна быть пустой. Возвращает True при успехе.
await client.request_my_info()Запрашивает вашу собственную анкету с сервера. Результат будет доступен в client.my_info и через колбэк on_my_info().
await client.request_user_info(uin: str)Запрашивает анкету указанного пользователя. Результат придет через колбэк on_user_info().
await client.save_my_info(info: UserInfo) -> boolСохраняет изменения в вашей анкете на сервере. Возвращает True при успехе.
results = await client.search_users(
uin: str = "",
nick: str = "",
first_name: str = "",
last_name: str = "",
email: str = "",
city: str = "",
keyword: str = "",
only_online: bool = False,
timeout: float = 30.0
) -> list[SearchResult]Выполняет поиск пользователей по заданным критериям. Результаты также поступают через колбэки on_search_result() (по одному) и on_search_done() (все сразу). Возвращает список всех найденных результатов.
await client.send_auth_request(to_uin: str, message: str = "")Отправляет запрос на авторизацию (добавление в список) пользователю. Используйте, если при добавлении контакта получили auth_req=True.
await client.send_auth_reply(to_uin: str, granted: bool, message: str = "")Отвечает на входящий запрос авторизации.
granted:Trueчтобы разрешить,Falseчтобы отклонить.
await client.request_offline_messages()Запрашивает очередь оффлайн-сообщений. Обычно вызывается автоматически после подключения. Сообщения приходят через колбэк on_offline_message().
await client.request_xstatus(to_uin: str)Запрашивает текущий xStatus у контакта. Ответ придет через обновление контакта и колбэк on_xstatus_updated(). Обычно вызывается автоматически при появлении контакта онлайн.
await client.run()Основной метод для запуска клиента. Подключается к серверу, проходит аутентификацию и входит в цикл обработки сообщений. Блокирует выполнение до отключения.
await client.stop()Корректно останавливает клиент и закрывает соединение.
import asyncio
from icq_core import ICQClient, Status, UserInfo
async def main():
client = ICQClient("12345678", "password")
async def on_connected():
print("Успешно подключено!")
# Запросить свою анкету
await client.request_my_info()
# Запросить оффлайн сообщения
await client.request_offline_messages()
def on_my_info(info: UserInfo):
print(f"Мой ник: {info.nick}")
print(f"Мое имя: {info.full_name}")
async def on_message(message):
print(f"Сообщение от {message.sender_uin}: {message.text}")
# Эхо-ответ
if message.text.lower() == "привет":
await client.send_message(message.sender_uin, "Привет! Как дела?")
def on_contact_online(contact):
print(f"Контакт онлайн: {contact.display_name} ({contact.client})")
def on_auth_request(uin, message):
print(f"Запрос авторизации от {uin}: {message}")
# Автоматически принимаем все запросы
asyncio.create_task(client.send_auth_reply(uin, True, "Добро пожаловать!"))
client.on_connected = on_connected
client.on_my_info = on_my_info
client.on_message = on_message
client.on_contact_online = on_contact_online
client.on_auth_request = on_auth_request
try:
await client.run()
except KeyboardInterrupt:
await client.stop()
if __name__ == "__main__":
asyncio.run(main())- Асинхронность: Все сетевые операции асинхронны. Не блокируйте основной поток длительными операциями в колбэках. Если нужно выполнить долгую задачу, используйте
asyncio.create_task(). - Обработка ошибок: Используйте колбэк
on_errorдля логирования проблем с соединением. - Безопасность: Никогда не храните пароли в открытом виде в коде. Используйте переменные окружения или файлы конфигурации.
- Лимиты: ICQ имеет лимиты на частоту запросов. Не отправляйте слишком много сообщений или запросов информации за короткое время.
- Кодировка: Библиотека пытается автоматически определять кодировку сообщений (UTF-8, CP1251, UTF-16BE). В большинстве случаев это работает корректно.
- xStatus: Поддержка xStatus реализована через протокол xTraz. Не все клиенты ICQ его поддерживают или отображают одинаково.