wxpy.api.chats.chats 源代码

# coding: utf-8
from __future__ import unicode_literals

import logging
import time
from collections import Counter

from wxpy.utils import match_attributes, match_name
from wxpy.compatible import *

logger = logging.getLogger(__name__)


[文档]class Chats(list): """ 多个聊天对象的合集,可用于搜索或统计 """ def __init__(self, chat_list=None, source=None): if chat_list: super(Chats, self).__init__(chat_list) self.source = source def __add__(self, other): return Chats(super(Chats, self).__add__(other or list()))
[文档] def search(self, keywords=None, **attributes): """ 在聊天对象合集中进行搜索 .. note:: | 搜索结果为一个 :class:`Chats (列表) <Chats>` 对象 | 建议搭配 :any:`ensure_one()` 使用 :param keywords: 聊天对象的名称关键词 :param attributes: 属性键值对,键可以是 sex(性别), province(省份), city(城市) 等。例如可指定 province='广东' :return: 匹配的聊天对象合集 :rtype: :class:`wxpy.Chats` """ def match(chat): if not match_name(chat, keywords): return if not match_attributes(chat, **attributes): return return True return Chats(filter(match, self), self.source)
[文档] def stats(self, attribs=('sex', 'province', 'city')): """ 统计各属性的分布情况 :param attribs: 需统计的属性列表或元组 :return: 统计结果 """ def attr_stat(objects, attr_name): return Counter(list(map(lambda x: getattr(x, attr_name), objects))) from wxpy.utils import ensure_list attribs = ensure_list(attribs) ret = dict() for attr in attribs: ret[attr] = attr_stat(self, attr) return ret
[文档] def stats_text(self, total=True, sex=True, top_provinces=10, top_cities=10): """ 简单的统计结果的文本 :param total: 总体数量 :param sex: 性别分布 :param top_provinces: 省份分布 :param top_cities: 城市分布 :return: 统计结果文本 """ from .group import Group from wxpy.api.consts import FEMALE from wxpy.api.consts import MALE from wxpy.api.bot import Bot def top_n_text(attr, n): top_n = list(filter(lambda x: x[0], stats[attr].most_common()))[:n] top_n = ['{}: {} ({:.2%})'.format(k, v, v / len(self)) for k, v in top_n] return '\n'.join(top_n) stats = self.stats() text = str() if total: if self.source: if isinstance(self.source, Bot): user_title = '微信好友' nick_name = self.source.self.nick_name elif isinstance(self.source, Group): user_title = '群成员' nick_name = self.source.nick_name else: raise TypeError('source should be Bot or Group') text += '{nick_name} 共有 {total}{user_title}\n\n'.format( nick_name=nick_name, total=len(self), user_title=user_title ) else: text += '共有 {} 位用户\n\n'.format(len(self)) if sex and self: males = stats['sex'].get(MALE, 0) females = stats['sex'].get(FEMALE, 0) text += '男性: {males} ({male_rate:.1%})\n女性: {females} ({female_rate:.1%})\n\n'.format( males=males, male_rate=males / len(self), females=females, female_rate=females / len(self), ) if top_provinces and self: text += 'TOP {} 省份\n{}\n\n'.format( top_provinces, top_n_text('province', top_provinces) ) if top_cities and self: text += 'TOP {} 城市\n{}\n\n'.format( top_cities, top_n_text('city', top_cities) ) return text
[文档] def add_all(self, interval=3, verify_content=''): """ 将合集中的所有用户加为好友,请小心应对调用频率限制! :param interval: 间隔时间(秒) :param verify_content: 验证说明文本 """ to_add = self[:] while to_add: adding = to_add.pop(0) logger.info('Adding {}'.format(adding)) ret = adding.add(verify_content=verify_content) logger.info(ret) logger.info('Waiting for {} seconds'.format(interval)) if to_add: time.sleep(interval)