Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -29,5 +29,7 @@ You can find a copy of this license [here](LICENSE).
## TODOs πŸ‘¨πŸ»β€πŸ’»

- [x] Implement custom search engine (CSE) API for image searching.
- [ ] Add pagination for image search results (currently only first 10 results are shown).
- [x] Add pagination for image search results (currently only first 10 results are shown).
- [ ] Add suggestions for image search queries that are misspelled.
- [ ] Add search functionality in private chat with bot.
- [ ] Add more commands to inline query (e.g. safe search, per page, ...).
14 changes: 7 additions & 7 deletions api/cse.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,15 +61,15 @@ def _request(self, method: str, endpoint: str, **kwargs) -> Response:
status_code = r.status_code
if status_code in (401, 403):
raise AuthError(
"Authentication error occurred, API response: %s" % r.text
)
else:
raise CSEAPIError(
"An error occurred while requesting the API, status code: %s, response: %s" % (
status_code, r.text
),
"Authentication error occurred, API response: %s" % r.text,
status_code
)
raise CSEAPIError(
"An error occurred while requesting the API, status code: %s, response: %s" % (
status_code, r.text
),
status_code
)
return r

def search(
Expand Down
67 changes: 51 additions & 16 deletions bot.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,9 @@
import telebot
from telebot import types

import texts
from api.cse import CSEAPIError, GoogleSearchEngine, SearchResult
from ext import parse_query

TG_API_TOKEN = "<YOUR_TELEGRAM_API_TOKEN>"
GOOGLE_API_KEY = "<YOUR_GOOGLE_API_KEY>"
Expand All @@ -29,40 +31,71 @@ def start_message(message: types.Message) -> None:
"""Handle `/start` command."""
first_name = message.from_user.first_name
chat_id = message.from_user.id
text = (
f"Hey [{first_name}](tg://user?id={chat_id})!\n\n"
"I'm **Imarch** πŸ€–, a bot for searching any kind images on Google 🌐.\n\n"
"⁉️ How to use me?\n\n"
"Just type my name followed by your query and I'll do the rest πŸ˜‰\n\n"
"For example to search for images of a cat, type in below command in inline mode:\n"
"`@ImarchBot cat`\n\n"
"Remember you can only use me in inline mode πŸ™‚"
bot.send_message(
chat_id,
texts.START_MSG.format(first_name=first_name, chat_id=chat_id),
parse_mode="Markdown"
)


# help command
@bot.message_handler(commands=['help'])
def help_message(message: types.Message) -> None:
"""Handle `/help` command."""
chat_id = message.from_user.id
message_id = message.message_id
kb = [
[
types.InlineKeyboardButton(
"Search now πŸ”Ž",
switch_inline_query_current_chat=" "
)
]
]
bot.send_message(
chat_id,
texts.HELP_MSG,
parse_mode="Markdown",
reply_to_message_id=message_id,
reply_markup=types.InlineKeyboardMarkup(kb)
)
bot.send_message(chat_id, text, parse_mode="Markdown")


# handle inline queries
@bot.inline_handler(func=lambda query: len(query.query) > 0)
def inline_query_handler(inline_query: types.InlineQuery) -> None:
"""Handle every inline query that is not empty."""
query = inline_query.query
query_id = inline_query.id
parsed_query = parse_query(inline_query.query)
# query string without commands
query_text = parsed_query.query
query_id = str(inline_query.id)
results = []
not_found = types.InlineQueryResultArticle(
id=str(uuid4()),
title="⚠️ No results found",
description="Couldn't find any results for your query πŸ˜”",
description=texts.NOT_FOUND_MSG,
input_message_content=types.InputTextMessageContent(
message_text="not_found_result"
)
)
page = 1
# handle query commands
if parsed_query.commands:
for command in parsed_query.commands:
if command.name.lower() == "page":
try:
value = abs(int(command.value))
page = value if value > 1 else 1
except ValueError:
continue
try:
search_result: SearchResult = cse.search(query, only_image=True)
search_result: SearchResult = cse.search(
query_text, page, only_image=True)
except CSEAPIError as e:
logger.error(f"Error while searching for {query!r}: {e}")
logger.error(f"Error while searching for {query_text!r}: {e}")
bot.answer_inline_query(query_id, [])
else:
# for every item in search result that has image, add it to results
# for every item in search result that has image attribute, add it to results
if search_result.items:
for item in search_result.items:
if item.image:
Expand Down Expand Up @@ -91,10 +124,12 @@ def message_handler(message: types.Message) -> None:
message_id = message.message_id
if text == "not_found_result":
bot.delete_message(chat_id, message_id)
else:
bot.send_message(chat_id, texts.NOT_FOUND_MSG)


def start_polling() -> None:
"""Start polling and responding to messages."""
"""Start polling and responding to every message."""
logger.info("Bot polling started...")
bot.infinity_polling()
while True:
Expand Down
43 changes: 43 additions & 0 deletions ext.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
import re
from dataclasses import dataclass
from typing import List, Optional


@dataclass(frozen=True)
class InlineCommand:
"""Represents a command parsed from an inline query string."""
name: str
"""Name of the command."""
value: Optional[str] = None
"""Value of the command (if any)."""


@dataclass(frozen=True)
class ParsedQuery:
"""Represents a parsed query object from an inline query string."""
text: str
"""The original query string that was parsed."""
query: str
"""The query string without the commands."""
commands: Optional[List[InlineCommand]] = None
"""A list of tuples of commands and their arguments from the query."""


def parse_query(text: str) -> ParsedQuery:
"""Return a parsed query object from the given query string.

Args:
- text (`str`): The query string to parse.

Returns:
- `ParsedQuery`: A parsed query object.
"""
cmd_pattern = r"([0-9a-zA-Z_]+):([^\s]*)"
commands = re.findall(cmd_pattern, text)
if commands:
commands = [
InlineCommand(name, value)
for name, value in commands
]
query = re.sub(cmd_pattern, "", text).strip()
return ParsedQuery(text, query, commands)
22 changes: 22 additions & 0 deletions texts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
START_MSG = (
"Hey [{first_name}](tg://user?id={chat_id})!\n\n"
"I'm *Imarch* πŸ€–, a bot for searching any kind images on Google 🌐.\n\n"
"Send /help to get started and see the instructions πŸ“–."
)
HELP_MSG = (
"πŸ“– *Imarch Bot Usage*\n\n"
"To search for images, just type my username and the query you want to search πŸ”Ž\n"
"\n*Examples* πŸ§ͺ \n\n"
"πŸ”Έ `@ImarchBot cat` - search for images of cats\n"
"πŸ”Έ `@ImarchBot cat page:2` - search for images of cats on page 2\n"
"\n❗ Beside the query, you can add commands to change the search results behavior.\n\n"
"*Supported commands*:\n\n"
"πŸ”Έ `page:<number>`: Change the page of the search results (default: 1)\n"
"\nπŸ’‘ *Note:*\n\n"
"The search results are paginated. You can change the page "
"of the search results by adding a command to the query.\n"

)
NOT_FOUND_MSG = (
"Sorry, I couldn't find any results for your query πŸ˜”"
)