You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
128 lines
4.8 KiB
128 lines
4.8 KiB
import logging |
|
import random |
|
from typing import List, Dict, Tuple, Union, Set |
|
from collections import defaultdict |
|
|
|
from .requests import LoginRequest, PostLoginRequest, PulseRequest, SendMsgRequest |
|
from ..commons import BotMessage, Message, AbstractResponse |
|
|
|
|
|
class Interlocutor: |
|
|
|
def __init__(self, nick: str, age: int, city: str, is_male: bool, conv_id: str): |
|
self.nick = nick |
|
self.age = age |
|
self.is_male = is_male |
|
self.city = city |
|
self.id = conv_id |
|
|
|
@classmethod |
|
def from_string(cls, str): |
|
# 47130922100412004Rastapopoulos |
|
# 47 (age) 1 (sexe) 30922 (city id) 100412(conv id) |
|
age = int(str[:2]) |
|
is_male = int(str[2:3]) in (1, 6) |
|
city_id = str[3:8] |
|
conv_id = str[8:14] |
|
nick = str[17:] |
|
return cls(nick, age, city_id, is_male, conv_id) |
|
|
|
def to_botmessage(self): |
|
sex_indic = "un homme" if self.is_male else "une femme" |
|
return BotMessage("Conversation avec %s, %s de %i ans" %(self.nick, sex_indic, self.age)) |
|
|
|
def __eq__(self, other): |
|
return other.nick == self.nick |
|
|
|
def __hash__(self): |
|
return hash(self.nick) |
|
|
|
|
|
class CocoClient: |
|
|
|
def __init__(self): |
|
self.interlocutors = set() # type: Set[Interlocutor] |
|
self.current_interlocutor = None # type: Interlocutor |
|
self.histories = defaultdict(list) # type:defaultdict[Interlocutor,List[Tuple]] |
|
|
|
self.user_id = None # type:str |
|
self.user_pass = None # type:str |
|
self.nick = None # type:str |
|
self.is_connected = False |
|
|
|
def _format_history(self, interlocutor: Interlocutor): |
|
if interlocutor in self.histories: |
|
return [BotMessage("💬 %s: %s" % (nick, msg)) |
|
for nick, msg in self.histories[interlocutor][:5]] |
|
else: |
|
return [] |
|
|
|
def __process_and_format_received_msg(self, received_msgs): |
|
out = [] |
|
for user_code, msg in received_msgs: |
|
user = Interlocutor.from_string(user_code) |
|
self.interlocutors.add(user) |
|
self.histories[user].append((user.nick, msg)) |
|
logging.info("Msg from %s : %s" % (user.nick, msg)) |
|
|
|
if self.current_interlocutor is not None and user == self.current_interlocutor: |
|
out.append(Message("💬 %s: %s" % (user.nick, msg))) |
|
else: |
|
out.append(BotMessage("💬 %s: %s" % (user.nick, msg))) |
|
return out |
|
|
|
def connect(self, nick: str, age: int, is_female: bool, zip_code: str): |
|
self.interlocutors = set() |
|
self.nick = nick |
|
self.histories = defaultdict(list) |
|
self.current_interlocutor = None |
|
login_req = LoginRequest(nick, age, is_female, zip_code) |
|
self.user_id, self.user_pass = login_req.retrieve() |
|
logging.info("Logged in to coco as %s" % self.nick) |
|
post_login_req = PostLoginRequest(self.user_id, self.user_pass) |
|
post_login_req.retrieve() |
|
logging.info("Post login successful") |
|
self.is_connected = True |
|
|
|
def pulse(self) -> List[AbstractResponse]: |
|
pulse_req = PulseRequest(self.user_id, self.user_pass) |
|
received_msg = pulse_req.retrieve() |
|
return self.__process_and_format_received_msg(received_msg) |
|
|
|
def send_msg(self, msg: str) -> List[AbstractResponse]: |
|
if self.current_interlocutor is not None: |
|
sendmsg_req = SendMsgRequest(self.user_id, self.user_pass, self.current_interlocutor.id, msg) |
|
output = sendmsg_req.retrieve() |
|
self.histories[self.current_interlocutor].append((self.nick, msg)) |
|
out_msg = Message("💬 %s: %s" % (self.nick, msg)) |
|
|
|
if output: |
|
return [out_msg] + self.__process_and_format_received_msg(output) |
|
else: |
|
return [out_msg] |
|
else: |
|
return [BotMessage("Il faut sélectionner une conversation d'abord pd")] |
|
|
|
def switch_conv(self, nick: str=None) -> Union[List[BotMessage], BotMessage]: |
|
if not self.interlocutors: |
|
return BotMessage("Pas de conversations en cours") |
|
|
|
new_interlocutor = None |
|
if nick is not None: |
|
for usr in self.interlocutors: |
|
if usr.nick.upper() == nick.upper(): |
|
new_interlocutor = usr |
|
break |
|
else: |
|
new_interlocutor = random.choice(list(self.interlocutors)) |
|
|
|
if new_interlocutor is None: |
|
return BotMessage("Impossible de trouver l'utilisateur") |
|
else: |
|
self.current_interlocutor = new_interlocutor |
|
return [new_interlocutor.to_botmessage()] + \ |
|
self._format_history(self.current_interlocutor) |
|
|
|
def list_convs(self): |
|
return BotMessage("Conversations : " + ", ".join(["%s(%i)" % (usr.nick, usr.age) |
|
for usr in self.interlocutors])) |