Coverage for kpoisk_bot/handlers.py: 57%

1import logging 

2from typing import Any 

3 

4from aiogram import Router, html 

5from aiogram.filters import CommandStart 

6from aiogram.types import ( 

7 ChosenInlineResult, 

8 InlineQuery, 

9 InlineQueryResultArticle, 

10 InputTextMessageContent, 

11 Message, 

12) 

13from httpx import HTTPError 

14from kinopoisk_unofficial_api_client import Client 

15from kinopoisk_unofficial_api_client.api.films import ( 

16 get_api_v2_1_films_search_by_keyword, 

17 get_api_v2_2_films_id, 

18) 

19 

20from .config import TOP_RESULTS_COUNT 

21from .format import FilmFormatter 

22 

23 

24router = Router() 

25 

26 

27@router.message(CommandStart()) 

28async def command_start_handler(message: Message) -> None: 

29 """ 

30 This handler receives messages with `/start` command 

31 """ 

32 await message.answer(f"Hello, {html.bold(message.from_user.full_name)}!") 

33 

34 

35@router.message() 

36async def search_handler(message: Message, api_client: Client) -> Any: 

37 if message.text is None: 

38 return 

39 try: 

40 results = await get_api_v2_1_films_search_by_keyword.asyncio( 

41 client=api_client, keyword=message.text, page=1 

42 ) 

43 for content in map( 

44 lambda film: FilmFormatter(film).as_text_message(), 

45 results.films[:TOP_RESULTS_COUNT], 

46 ): 

47 await message.reply( 

48 content.message_text, 

49 link_preview_options=content.link_preview_options, 

50 entities=content.entities, 

51 parse_mode=content.parse_mode, 

52 ) 

53 except HTTPError as e: 

54 # But not all the types is supported to be copied so need to handle it 

55 logging.error(e) 

56 

57 

58@router.inline_query() 

59async def inline_handler(query: InlineQuery, api_client: Client) -> Any: 

60 if not query.query: 

61 return await query.answer( 

62 [ 

63 InlineQueryResultArticle( 

64 id="EMPTY", 

65 title="Наберите название фильма, чтобы найти его в КиноПоиске", 

66 input_message_content=InputTextMessageContent( 

67 message_text="Этот бот поможет вам найти фильм в КиноПоиске" 

68 ), 

69 ) 

70 ] 

71 ) 

72 

73 try: 

74 api_result = await get_api_v2_1_films_search_by_keyword.asyncio( 

75 client=api_client, keyword=query.query, page=1 

76 ) 

77 results = [ 

78 FilmFormatter(film).as_inline() 

79 for film in api_result.films[:TOP_RESULTS_COUNT] 

80 ] 

81 return await query.answer(results) 

82 except (HTTPError, TypeError) as e: 

83 logging.error(e) 

84 return await query.answer( 

85 [ 

86 InlineQueryResultArticle( 

87 id="ERROR", 

88 title="Упс, что-то пошло не так", 

89 input_message_content=InputTextMessageContent( 

90 message_text="Что-то пошло не так :(" 

91 ), 

92 ) 

93 ] 

94 ) 

95 

96 

97@router.chosen_inline_result() 

98async def chosen_inline_handler(result: ChosenInlineResult, api_client: Client) -> Any: 

99 if result.result_id == "ERROR" or result.result_id == "EMPTY": 

100 return 

101 

102 film = await get_api_v2_2_films_id.asyncio( 

103 id=int(result.result_id), client=api_client 

104 ) 

105 formatter = FilmFormatter(film) 

106 as_message = formatter.as_text_message() 

107 await result.bot.edit_message_text( 

108 inline_message_id=result.inline_message_id, 

109 text=as_message.message_text, 

110 parse_mode=as_message.parse_mode, 

111 entities=as_message.entities, 

112 link_preview_options=as_message.link_preview_options, 

113 ) 

114 await result.bot.edit_message_reply_markup( 

115 inline_message_id=result.inline_message_id, 

116 reply_markup=formatter.create_result_markup(), 

117 )