-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
57 lines (44 loc) · 1.6 KB
/
main.py
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
"""Contains the orchestrator object and the entry point to tun the robots."""
import asyncio
from bots.base import State
from bots.text import TextRobot
class Orchestrator:
"""
Orchestrates all the flow. It is responsible for run each one of the robots.
"""
state = State()
# Registered robots.
text_robot = TextRobot()
async def ask_and_return_search_term(self) -> None:
"""
Asks for the user write an search term and stores it in the state.
"""
search_term: str = input('\nType a search term: ')
self.state.update(search_term=search_term)
async def ask_and_return_prefix(self) -> None:
"""
Asks for the user select an prefix and stores it in the state.
"""
prefixes = ('who is', 'what is', 'The history of')
prefixes_with_indexes: str = '\n'.join(list(map(
lambda index_and_prefix_tuple: '[{0}] {1}'.format(
*index_and_prefix_tuple
),
enumerate(iterable=prefixes, start=1)
)))
prefix_index: int = int(input(
f'\n{prefixes_with_indexes}\nChoose an option: '
))
self.state.update(prefix=prefixes[prefix_index - 1])
print('\n')
async def start(self) -> None:
"""
Starts the orchestrator asking for the user the needed data and run the
bots.
"""
await self.ask_and_return_search_term()
await self.ask_and_return_prefix()
await self.text_robot.run()
if __name__ == '__main__':
orchestrator = Orchestrator()
asyncio.run(orchestrator.start())