Skip to content

Commit

Permalink
Merge pull request #90 from ayrat555/ayrat555/add-subscription-filters
Browse files Browse the repository at this point in the history 
Add subscription filters
  • Loading branch information
@ayrat555
ayrat555 committed Mar 8, 2021
2 parents 2e0ef31 + a7fe42b commit 45d38bd
Show file tree
Hide file tree
Showing 8 changed files with 239 additions and 82 deletions.
1 change: 1 addition & 0 deletions migrations/2021-03-07-131037_add_filter_words_to_subscriptions/down.sql
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
ALTER TABLE telegram_subscriptions DROP COLUMN filter_words;
1 change: 1 addition & 0 deletions migrations/2021-03-07-131037_add_filter_words_to_subscriptions/up.sql
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
ALTER TABLE telegram_subscriptions ADD COLUMN filter_words text[];
57 changes: 54 additions & 3 deletions src/bot/api.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,20 +17,26 @@ static SET_TIMEZONE: &str = "/set_timezone";
static GET_TIMEZONE: &str = "/get_timezone";
static SET_TEMPLATE: &str = "/set_template";
static GET_TEMPLATE: &str = "/get_template";
static SET_FILTER: &str = "/set_filter";
static GET_FILTER: &str = "/get_filter";
static REMOVE_FILTER: &str = "/remove_filter";
static SET_GLOBAL_TEMPLATE: &str = "/set_global_template";
static GET_GLOBAL_TEMPLATE: &str = "/get_global_template";
static UNSUBSCRIBE: &str = "/unsubscribe";
static HELP: &str = "/help";
static START: &str = "/start";
static OWNER_TELEGRAM_ID: OnceCell<Option<i64>> = OnceCell::new();

static COMMANDS: [&str; 11] = [
static COMMANDS: [&str; 14] = [
SUBSCRIBE,
LIST_SUBSCRIPTIONS,
SET_TIMEZONE,
GET_TIMEZONE,
SET_TEMPLATE,
GET_TEMPLATE,
SET_FILTER,
GET_FILTER,
REMOVE_FILTER,
SET_GLOBAL_TEMPLATE,
GET_GLOBAL_TEMPLATE,
UNSUBSCRIBE,
Expand Down Expand Up @@ -114,8 +120,11 @@ fn commands_string() -> String {
Example: /set_template https://www.badykov.com/feed.xml bot_datebot_spacebot_item_namebot_new_linebot_item_description\n\n\
{} url - get a template for the subscription\n\n\
{} template - set global template. This template will be used for all subscriptions. If the subscription has its own template, the subscription template will be used. See /set_template for available fields.\n\n\
{} - get global template\n",
START, SUBSCRIBE, UNSUBSCRIBE, LIST_SUBSCRIPTIONS, HELP, SET_TIMEZONE, GET_TIMEZONE, SET_TEMPLATE, GET_TEMPLATE, SET_GLOBAL_TEMPLATE, GET_GLOBAL_TEMPLATE
{} - get global template\n\n\
{} url - get a filter for the subscription\n\n\
{} url template - set filter, for example, /set_filter https://www.badykov.com/feed.xml telegram,bots. You'll start receiving posts only containing words in the filter\n\n\
{} url - remove filter\n\n",
START, SUBSCRIBE, UNSUBSCRIBE, LIST_SUBSCRIPTIONS, HELP, SET_TIMEZONE, GET_TIMEZONE, SET_TEMPLATE, GET_TEMPLATE, SET_GLOBAL_TEMPLATE, GET_GLOBAL_TEMPLATE, GET_FILTER, SET_FILTER, REMOVE_FILTER
)
}

Expand Down Expand Up @@ -250,6 +259,17 @@ async fn set_template(api: Api, message: MessageOrChannelPost, data: String) ->
Ok(())
}

async fn set_filter(api: Api, message: MessageOrChannelPost, data: String) -> Result<(), Error> {
let chat_id = get_chat_id(&message);
let semaphored_connection = db::get_semaphored_connection().await;
let db_connection = semaphored_connection.connection;

let response = logic::set_filter(&db_connection, chat_id, data);

api.send(message.text_reply(response)).await?;
Ok(())
}

async fn get_timezone(api: Api, message: MessageOrChannelPost) -> Result<(), Error> {
let chat_id = get_chat_id(&message);
let semaphored_connection = db::get_semaphored_connection().await;
Expand Down Expand Up @@ -298,6 +318,28 @@ async fn get_template(api: Api, message: MessageOrChannelPost, data: String) ->
Ok(())
}

async fn get_filter(api: Api, message: MessageOrChannelPost, data: String) -> Result<(), Error> {
let chat_id = get_chat_id(&message);
let semaphored_connection = db::get_semaphored_connection().await;
let db_connection = semaphored_connection.connection;

let response = logic::get_filter(&db_connection, chat_id, data);

api.send(message.text_reply(response)).await?;
Ok(())
}

async fn remove_filter(api: Api, message: MessageOrChannelPost, data: String) -> Result<(), Error> {
let chat_id = get_chat_id(&message);
let semaphored_connection = db::get_semaphored_connection().await;
let db_connection = semaphored_connection.connection;

let response = logic::remove_filter(&db_connection, chat_id, data);

api.send(message.text_reply(response)).await?;
Ok(())
}

fn process_message(api: Api, orig_message: Message) {
match orig_message.kind {
MessageKind::Text { ref data, .. } => {
Expand Down Expand Up @@ -377,6 +419,15 @@ async fn process_message_or_channel_post(
} else if command.starts_with(GET_TEMPLATE) {
let argument = parse_argument(command, GET_TEMPLATE);
tokio::spawn(get_template(api, message, argument));
} else if command.starts_with(GET_FILTER) {
let argument = parse_argument(command, GET_FILTER);
tokio::spawn(get_filter(api, message, argument));
} else if command.starts_with(REMOVE_FILTER) {
let argument = parse_argument(command, REMOVE_FILTER);
tokio::spawn(remove_filter(api, message, argument));
} else if command.starts_with(SET_FILTER) {
let argument = parse_argument(command, SET_FILTER);
tokio::spawn(set_filter(api, message, argument));
} else if command.starts_with(SET_TEMPLATE) {
let argument = parse_argument(command, SET_TEMPLATE);
tokio::spawn(set_template(api, message, argument));
Expand Down
88 changes: 54 additions & 34 deletions src/bot/deliver_job.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,7 @@ async fn deliver_subscription_updates(
let chat = telegram::find_chat(&connection, chat_id).unwrap();
let delay = delay_period(&chat);

if feed_items.len() < undelivered_count as usize {
if subscription.filter_words.is_none() && feed_items.len() < undelivered_count as usize {
let message = format!(
"You have {} unread items, below {} last items for {}",
undelivered_count,
Expand All @@ -140,18 +140,7 @@ async fn deliver_subscription_updates(
Err(error) => {
let error_message = format!("{}", error);

log::error!("Failed to deliver updates: {} {}", chat_id, error_message);

if bot_blocked(&error_message) {
match telegram::remove_chat(&connection, chat_id) {
Ok(_) => log::info!("Successfully removed chat {}", chat_id),
Err(error) => log::error!("Failed to remove a chat {}", error),
}
};

return Err(DeliverJobError {
msg: format!("Failed to send updates : {}", error),
});
return Err(handle_error(error_message, &connection, chat_id));
}
}
}
Expand All @@ -177,35 +166,66 @@ async fn deliver_subscription_updates(
messages.reverse();

for (message, publication_date) in messages {
match api::send_message(chat_id, message).await {
Ok(_) => {
time::delay_for(delay).await;
update_last_deivered_at(&connection, &subscription, publication_date)?;
()
}
Err(error) => {
let error_message = format!("{}", error);

log::error!("Failed to deliver updates: {}", error_message);
match subscription.filter_words.clone() {
None => match api::send_message(chat_id, message).await {
Ok(_) => {
time::delay_for(delay).await;
update_last_deivered_at(&connection, &subscription, publication_date)?;
()
}
Err(error) => {
let error_message = format!("{}", error);

if bot_blocked(&error_message) {
match telegram::remove_chat(&connection, chat_id) {
Ok(_) => log::info!("Successfully removed chat {}", chat_id),
Err(error) => log::error!("Failed to remove a chat {}", error),
return Err(handle_error(error_message, &connection, chat_id));
}
},
Some(words) => {
let mtch = words
.iter()
.any(|word| message.to_lowercase().contains(word));

if mtch {
match api::send_message(chat_id, message).await {
Ok(_) => {
time::delay_for(delay).await;
update_last_deivered_at(
&connection,
&subscription,
publication_date,
)?;
}
Err(error) => {
let error_message = format!("{}", error);

return Err(handle_error(error_message, &connection, chat_id));
}
}
};

return Err(DeliverJobError {
msg: format!("Failed to send updates : {}", error),
});
} else {
update_last_deivered_at(&connection, &subscription, publication_date)?;
}
}
};
}
}
}

Ok(())
}

fn handle_error(error: String, connection: &PgConnection, chat_id: i64) -> DeliverJobError {
log::error!("Failed to deliver updates: {}", error);

if bot_blocked(&error) {
match telegram::remove_chat(connection, chat_id) {
Ok(_) => log::info!("Successfully removed chat {}", chat_id),
Err(error) => log::error!("Failed to remove a chat {}", error),
}
};

DeliverJobError {
msg: format!("Failed to send updates : {}", error),
}
}

fn update_last_deivered_at(
connection: &PgConnection,
subscription: &TelegramSubscription,
Expand All @@ -232,7 +252,7 @@ fn format_messages(

let templ = match template {
Some(t) => t,
None => "{{bot_feed_name}}\n\n{{bot_item_name}}\n\n{{bot_date}}\n\n{{bot_item_link}}\n\n"
None => "{{bot_feed_name}}\n\n{{bot_item_name}}\n\n{{bot_item_description}}\n\n{{bot_date}}\n\n{{bot_item_link}}\n\n"
.to_string(),
};

Expand Down
Loading

0 comments on commit 45d38bd

Please sign in to comment.