Annotation For src/core.rs
Logged in as anonymous

Origin for each line in src/core.rs from check-in 1bd041d00f:

0340541002 2025-04-24    1: use crate::{
0340541002 2025-04-24    2: 	command,
0340541002 2025-04-24    3: 	sql::Db,
0340541002 2025-04-24    4: };
1c444d34ff 2024-08-28    5: 
9171c791eb 2021-11-13    6: use std::{
c1e27b74ed 2021-11-13    7: 	borrow::Cow,
9171c791eb 2021-11-13    8: 	collections::{
9171c791eb 2021-11-13    9: 		BTreeMap,
9171c791eb 2021-11-13   10: 		HashSet,
9171c791eb 2021-11-13   11: 	},
1c444d34ff 2024-08-28   12: 	num::TryFromIntError,
1c444d34ff 2024-08-28   13: 	sync::{
1c444d34ff 2024-08-28   14: 		Arc,
1c444d34ff 2024-08-28   15: 		Mutex
1c444d34ff 2024-08-28   16: 	},
1c444d34ff 2024-08-28   17: };
1c444d34ff 2024-08-28   18: 
e624ef9d66 2025-04-20   19: use anyhow::{
1bd041d00f 2025-05-20   20: 	anyhow,
e624ef9d66 2025-04-20   21: 	bail,
e624ef9d66 2025-04-20   22: 	Result,
e624ef9d66 2025-04-20   23: };
e624ef9d66 2025-04-20   24: use async_std::task;
e624ef9d66 2025-04-20   25: use chrono::DateTime;
e624ef9d66 2025-04-20   26: use frankenstein::{
1bd041d00f 2025-05-20   27: 	AsyncTelegramApi,
1bd041d00f 2025-05-20   28: 	Error as FrankError,
1bd041d00f 2025-05-20   29: 	ParseMode,
e624ef9d66 2025-04-20   30: 	client_reqwest::Bot,
e624ef9d66 2025-04-20   31: 	methods::{
e624ef9d66 2025-04-20   32: 		GetUpdatesParams,
e624ef9d66 2025-04-20   33: 		SendMessageParams
e624ef9d66 2025-04-20   34: 	},
e624ef9d66 2025-04-20   35: 	types::{
e624ef9d66 2025-04-20   36: 		AllowedUpdate,
e624ef9d66 2025-04-20   37: 		MessageEntityType,
e624ef9d66 2025-04-20   38: 		User,
e624ef9d66 2025-04-20   39: 	},
e624ef9d66 2025-04-20   40: 	updates::UpdateContent,
e624ef9d66 2025-04-20   41: };
e624ef9d66 2025-04-20   42: use thiserror::Error;
e624ef9d66 2025-04-20   43: 
1c444d34ff 2024-08-28   44: #[derive(Error, Debug)]
1c444d34ff 2024-08-28   45: pub enum RssError {
a2880e5100 2025-04-19   46: 	// #[error(transparent)]
a2880e5100 2025-04-19   47: 	// Tg(#[from] TgError),
1c444d34ff 2024-08-28   48: 	#[error(transparent)]
1c444d34ff 2024-08-28   49: 	Int(#[from] TryFromIntError),
79c91a5357 2024-08-02   50: }
9171c791eb 2021-11-13   51: 
9171c791eb 2021-11-13   52: #[derive(Clone)]
9171c791eb 2021-11-13   53: pub struct Core {
e624ef9d66 2025-04-20   54: 	owner_chat: i64,
1bd041d00f 2025-05-20   55: 	max_delay: u16,
a2880e5100 2025-04-19   56: 	pub tg: Bot,
e624ef9d66 2025-04-20   57: 	pub me: User,
0340541002 2025-04-24   58: 	pub db: Db,
9171c791eb 2021-11-13   59: 	sources: Arc<Mutex<HashSet<Arc<i32>>>>,
45e34762e4 2023-05-28   60: 	http_client: reqwest::Client,
9171c791eb 2021-11-13   61: }
9171c791eb 2021-11-13   62: 
9171c791eb 2021-11-13   63: impl Core {
0340541002 2025-04-24   64: 	pub async fn new(settings: config::Config) -> Result<Core> {
e624ef9d66 2025-04-20   65: 		let owner_chat = settings.get_int("owner")?;
28da2e2a00 2022-03-12   66: 		let api_key = settings.get_string("api_key")?;
e624ef9d66 2025-04-20   67: 		let tg = Bot::new(&api_key);
45e34762e4 2023-05-28   68: 
45e34762e4 2023-05-28   69: 		let mut client = reqwest::Client::builder();
9d8a6738fd 2023-07-30   70: 		if let Ok(proxy) = settings.get_string("proxy") {
45e34762e4 2023-05-28   71: 			let proxy = reqwest::Proxy::all(proxy)?;
45e34762e4 2023-05-28   72: 			client = client.proxy(proxy);
45e34762e4 2023-05-28   73: 		}
45e34762e4 2023-05-28   74: 		let http_client = client.build()?;
e624ef9d66 2025-04-20   75: 		let me = tg.get_me().await?;
e624ef9d66 2025-04-20   76: 		let me = me.result;
0340541002 2025-04-24   77: 		let core = Core {
f988dfd28f 2022-02-13   78: 			tg,
e624ef9d66 2025-04-20   79: 			me,
e624ef9d66 2025-04-20   80: 			owner_chat,
0340541002 2025-04-24   81: 			db: Db::new(&settings.get_string("pg")?)?,
9171c791eb 2021-11-13   82: 			sources: Arc::new(Mutex::new(HashSet::new())),
45e34762e4 2023-05-28   83: 			http_client,
1bd041d00f 2025-05-20   84: 			max_delay: 60,
0340541002 2025-04-24   85: 		};
0340541002 2025-04-24   86: 		let mut clone = core.clone();
cb86e770f9 2022-03-15   87: 		task::spawn(async move {
26339860ce 2022-02-13   88: 			loop {
26339860ce 2022-02-13   89: 				let delay = match &clone.autofetch().await {
26339860ce 2022-02-13   90: 					Err(err) => {
e624ef9d66 2025-04-20   91: 						if let Err(err) = clone.send(format!("šŸ›‘ {err:?}"), None, None).await {
e624ef9d66 2025-04-20   92: 							eprintln!("Autofetch error: {err:?}");
26339860ce 2022-02-13   93: 						};
cb86e770f9 2022-03-15   94: 						std::time::Duration::from_secs(60)
26339860ce 2022-02-13   95: 					},
26339860ce 2022-02-13   96: 					Ok(time) => *time,
9171c791eb 2021-11-13   97: 				};
cb86e770f9 2022-03-15   98: 				task::sleep(delay).await;
9171c791eb 2021-11-13   99: 			}
e624ef9d66 2025-04-20  100: 		});
e624ef9d66 2025-04-20  101: 		Ok(core)
e624ef9d66 2025-04-20  102: 	}
e624ef9d66 2025-04-20  103: 
0340541002 2025-04-24  104: 	pub async fn stream(&mut self) -> Result<()> {
e624ef9d66 2025-04-20  105: 		let mut offset: i64 = 0;
e624ef9d66 2025-04-20  106: 		let mut params = GetUpdatesParams {
e624ef9d66 2025-04-20  107: 			offset: None,
e624ef9d66 2025-04-20  108: 			limit: Some(100),
e624ef9d66 2025-04-20  109: 			timeout: Some(300),
e624ef9d66 2025-04-20  110: 			allowed_updates: Some(vec![AllowedUpdate::Message]),
e624ef9d66 2025-04-20  111: 		};
e624ef9d66 2025-04-20  112: 		loop {
e624ef9d66 2025-04-20  113: 			let updates = self.tg.get_updates(&params).await?.result;
e624ef9d66 2025-04-20  114: 			if updates.is_empty() {
e624ef9d66 2025-04-20  115: 				offset = 0;
e624ef9d66 2025-04-20  116: 				params.offset = None;
e624ef9d66 2025-04-20  117: 				continue;
e624ef9d66 2025-04-20  118: 			}
e624ef9d66 2025-04-20  119: 			for update in updates {
e624ef9d66 2025-04-20  120: 				if i64::from(update.update_id) >= offset {
e624ef9d66 2025-04-20  121: 					offset = i64::from(update.update_id) + 1;
e624ef9d66 2025-04-20  122: 					params.offset = Some(offset);
e624ef9d66 2025-04-20  123: 				}
e624ef9d66 2025-04-20  124: 				if let UpdateContent::Message(msg) = update.content {
e624ef9d66 2025-04-20  125: 					if let Some(text) = msg.text {
e624ef9d66 2025-04-20  126: 						if let Some(entities) = msg.entities {
e624ef9d66 2025-04-20  127: 							let chars: Vec<u16> = text.encode_utf16().collect();
e624ef9d66 2025-04-20  128: 							for entity in entities {
e624ef9d66 2025-04-20  129: 								if entity.type_field == MessageEntityType::BotCommand && entity.offset != 0 {
e624ef9d66 2025-04-20  130: 									bail!("commands should be at message start");
e624ef9d66 2025-04-20  131: 								};
e624ef9d66 2025-04-20  132: 								let cmd = String::from_utf16_lossy(&chars[entity.offset as usize..entity.length as usize]);
e624ef9d66 2025-04-20  133: 								let words: Vec<&str> = text.split_whitespace().collect();
1bd041d00f 2025-05-20  134: 								let res = match cmd.as_ref() {
1bd041d00f 2025-05-20  135: 									"/check" | "/clean" | "/enable" | "/delete" | "/disable" => command::command(self, msg.chat.id, words).await,
1bd041d00f 2025-05-20  136: 									"/start" => command::start(self, msg.chat.id).await,
1bd041d00f 2025-05-20  137: 									"/list" => command::list(self, msg.chat.id).await,
1bd041d00f 2025-05-20  138: 									"/add" | "/update" => command::update(self, msg.chat.id, words).await,
1bd041d00f 2025-05-20  139: 									any => Err(anyhow!("Unknown command: {any}")),
e624ef9d66 2025-04-20  140: 								};
1bd041d00f 2025-05-20  141: 								if let Err(err) = res {
1bd041d00f 2025-05-20  142: 									if let Err(err2) = self.send(format!("\\#error\n```\n{err:?}\n```"),
1bd041d00f 2025-05-20  143: 										Some(msg.chat.id),
1bd041d00f 2025-05-20  144: 										Some(ParseMode::MarkdownV2)
1bd041d00f 2025-05-20  145: 									).await{
1bd041d00f 2025-05-20  146: 										dbg!(err2);
1bd041d00f 2025-05-20  147: 									};
1bd041d00f 2025-05-20  148: 								}
e624ef9d66 2025-04-20  149: 							};
e624ef9d66 2025-04-20  150: 						};
e624ef9d66 2025-04-20  151: 					};
e624ef9d66 2025-04-20  152: 				};
e624ef9d66 2025-04-20  153: 			}
e624ef9d66 2025-04-20  154: 		}
e624ef9d66 2025-04-20  155: 	}
e624ef9d66 2025-04-20  156: 
e624ef9d66 2025-04-20  157: 	pub async fn send <S>(&self, msg: S, target: Option<i64>, mode: Option<ParseMode>) -> Result<()>
e624ef9d66 2025-04-20  158: 	where S: Into<String> {
e624ef9d66 2025-04-20  159: 		let msg = msg.into();
e624ef9d66 2025-04-20  160: 
e624ef9d66 2025-04-20  161: 		let mode = mode.unwrap_or(ParseMode::Html);
e624ef9d66 2025-04-20  162: 		let target = target.unwrap_or(self.owner_chat);
e624ef9d66 2025-04-20  163: 		let send = SendMessageParams::builder()
e624ef9d66 2025-04-20  164: 			.chat_id(target)
e624ef9d66 2025-04-20  165: 			.text(msg)
e624ef9d66 2025-04-20  166: 			.parse_mode(mode)
e624ef9d66 2025-04-20  167: 			.build();
1bd041d00f 2025-05-20  168: 		loop {
1bd041d00f 2025-05-20  169: 			match self.tg.send_message(&send).await {
1bd041d00f 2025-05-20  170: 				Ok(_) => break,
1bd041d00f 2025-05-20  171: 				Err(err) => match err {
1bd041d00f 2025-05-20  172: 					FrankError::Api(ref resp) => {
1bd041d00f 2025-05-20  173: 						if resp.error_code == 429 {
1bd041d00f 2025-05-20  174: 							let mut my_delay = self.max_delay;
1bd041d00f 2025-05-20  175: 							if let Some(params) = resp.parameters {
1bd041d00f 2025-05-20  176: 								if let Some(delay) = params.retry_after {
1bd041d00f 2025-05-20  177: 									if delay < my_delay {
1bd041d00f 2025-05-20  178: 										my_delay = delay;
1bd041d00f 2025-05-20  179: 									}
1bd041d00f 2025-05-20  180: 								}
1bd041d00f 2025-05-20  181: 							}
1bd041d00f 2025-05-20  182: 							task::sleep(std::time::Duration::from_secs(my_delay.into())).await;
1bd041d00f 2025-05-20  183: 						} else {
1bd041d00f 2025-05-20  184: 							return Err(err.into());
1bd041d00f 2025-05-20  185: 						}
1bd041d00f 2025-05-20  186: 					},
1bd041d00f 2025-05-20  187: 					_ => return Err(err.into()),
1bd041d00f 2025-05-20  188: 				},
1bd041d00f 2025-05-20  189: 			}
1bd041d00f 2025-05-20  190: 		}
e624ef9d66 2025-04-20  191: 		Ok(())
e624ef9d66 2025-04-20  192: 	}
e624ef9d66 2025-04-20  193: 
0340541002 2025-04-24  194: 	pub async fn check (&mut self, id: &i32, owner: i64, real: bool) -> Result<String> {
e624ef9d66 2025-04-20  195: 		let mut posted: i32 = 0;
0340541002 2025-04-24  196: 		let mut conn = self.db.begin().await?;
285ce2bc31 2024-09-17  197: 
9171c791eb 2021-11-13  198: 		let id = {
9171c791eb 2021-11-13  199: 			let mut set = self.sources.lock().unwrap();
9171c791eb 2021-11-13  200: 			match set.get(id) {
9171c791eb 2021-11-13  201: 				Some(id) => id.clone(),
9171c791eb 2021-11-13  202: 				None => {
9171c791eb 2021-11-13  203: 					let id = Arc::new(*id);
9171c791eb 2021-11-13  204: 					set.insert(id.clone());
9171c791eb 2021-11-13  205: 					id.clone()
9171c791eb 2021-11-13  206: 				},
9171c791eb 2021-11-13  207: 			}
9171c791eb 2021-11-13  208: 		};
9171c791eb 2021-11-13  209: 		let count = Arc::strong_count(&id);
9171c791eb 2021-11-13  210: 		if count == 2 {
0340541002 2025-04-24  211: 			let source = conn.get_source(*id, owner).await?;
61899b5618 2025-04-26  212: 			conn.set_scrape(*id).await?;
9171c791eb 2021-11-13  213: 			let destination = match real {
e624ef9d66 2025-04-20  214: 				true => source.channel_id,
e624ef9d66 2025-04-20  215: 				false => source.owner,
9171c791eb 2021-11-13  216: 			};
9171c791eb 2021-11-13  217: 			let mut this_fetch: Option<DateTime<chrono::FixedOffset>> = None;
9171c791eb 2021-11-13  218: 			let mut posts: BTreeMap<DateTime<chrono::FixedOffset>, String> = BTreeMap::new();
45e34762e4 2023-05-28  219: 
9910c2209c 2023-08-04  220: 			let response = self.http_client.get(&source.url).send().await?;
613a665847 2021-11-15  221: 			let status = response.status();
613a665847 2021-11-15  222: 			let content = response.bytes().await?;
9171c791eb 2021-11-13  223: 			match rss::Channel::read_from(&content[..]) {
9171c791eb 2021-11-13  224: 				Ok(feed) => {
9171c791eb 2021-11-13  225: 					for item in feed.items() {
f988dfd28f 2022-02-13  226: 						if let Some(link) = item.link() {
f988dfd28f 2022-02-13  227: 							let date = match item.pub_date() {
f988dfd28f 2022-02-13  228: 								Some(feed_date) => DateTime::parse_from_rfc2822(feed_date),
f988dfd28f 2022-02-13  229: 								None => DateTime::parse_from_rfc3339(&item.dublin_core_ext().unwrap().dates()[0]),
f988dfd28f 2022-02-13  230: 							}?;
f988dfd28f 2022-02-13  231: 							let url = link;
26339860ce 2022-02-13  232: 							posts.insert(date, url.to_string());
9171c791eb 2021-11-13  233: 						}
9171c791eb 2021-11-13  234: 					};
9171c791eb 2021-11-13  235: 				},
9171c791eb 2021-11-13  236: 				Err(err) => match err {
9171c791eb 2021-11-13  237: 					rss::Error::InvalidStartTag => {
2c36f015d8 2024-11-23  238: 						match atom_syndication::Feed::read_from(&content[..]) {
2c36f015d8 2024-11-23  239: 							Ok(feed) => {
2c36f015d8 2024-11-23  240: 								for item in feed.entries() {
2c36f015d8 2024-11-23  241: 									let date = item.published().unwrap();
2c36f015d8 2024-11-23  242: 									let url = item.links()[0].href();
2c36f015d8 2024-11-23  243: 									posts.insert(*date, url.to_string());
2c36f015d8 2024-11-23  244: 								};
2c36f015d8 2024-11-23  245: 							},
2c36f015d8 2024-11-23  246: 							Err(err) => {
e624ef9d66 2025-04-20  247: 								bail!("Unsupported or mangled content:\n{:?}\n{err:#?}\n{status:#?}\n", &source.url)
2c36f015d8 2024-11-23  248: 							},
2c36f015d8 2024-11-23  249: 						}
9171c791eb 2021-11-13  250: 					},
9171c791eb 2021-11-13  251: 					rss::Error::Eof => (),
e624ef9d66 2025-04-20  252: 					_ => bail!("Unsupported or mangled content:\n{:?}\n{err:#?}\n{status:#?}\n", &source.url)
9171c791eb 2021-11-13  253: 				}
9171c791eb 2021-11-13  254: 			};
9171c791eb 2021-11-13  255: 			for (date, url) in posts.iter() {
9910c2209c 2023-08-04  256: 				let post_url: Cow<str> = match source.url_re {
3dc9cddd4d 2023-08-04  257: 					Some(ref x) => sedregex::ReplaceCommand::new(x)?.execute(url),
659724c658 2021-12-08  258: 					None => url.into(),
659724c658 2021-12-08  259: 				};
0340541002 2025-04-24  260: 				if let Some(exists) = conn.exists(&post_url, *id).await? {
9910c2209c 2023-08-04  261: 					if ! exists {
9910c2209c 2023-08-04  262: 						if this_fetch.is_none() || *date > this_fetch.unwrap() {
9910c2209c 2023-08-04  263: 							this_fetch = Some(*date);
9910c2209c 2023-08-04  264: 						};
e624ef9d66 2025-04-20  265: 						self.send( match &source.iv_hash {
e624ef9d66 2025-04-20  266: 							Some(hash) => format!("<a href=\"https://t.me/iv?url={post_url}&rhash={hash}\"> </a>{post_url}"),
e624ef9d66 2025-04-20  267: 							None => format!("{post_url}"),
e624ef9d66 2025-04-20  268: 						}, Some(destination), Some(ParseMode::Html)).await?;
0340541002 2025-04-24  269: 						conn.add_post(*id, date, &post_url).await?;
659724c658 2021-12-08  270: 					};
9171c791eb 2021-11-13  271: 				};
9171c791eb 2021-11-13  272: 				posted += 1;
9171c791eb 2021-11-13  273: 			};
9171c791eb 2021-11-13  274: 			posts.clear();
9171c791eb 2021-11-13  275: 		};
0340541002 2025-04-24  276: 		Ok(format!("Posted: {posted}"))
0340541002 2025-04-24  277: 	}
0340541002 2025-04-24  278: 
0340541002 2025-04-24  279: 	async fn autofetch(&mut self) -> Result<std::time::Duration> {
0340541002 2025-04-24  280: 		let mut delay = chrono::Duration::minutes(1);
0340541002 2025-04-24  281: 		let now = chrono::Local::now();
0340541002 2025-04-24  282: 		let mut conn = self.db.begin().await?;
0340541002 2025-04-24  283: 		for row in conn.get_queue().await? {
9910c2209c 2023-08-04  284: 			if let Some(next_fetch) = row.next_fetch {
9910c2209c 2023-08-04  285: 				if next_fetch < now {
9910c2209c 2023-08-04  286: 					if let (Some(owner), Some(source_id)) = (row.owner, row.source_id) {
0340541002 2025-04-24  287: 						let mut clone = Core {
e624ef9d66 2025-04-20  288: 							owner_chat: owner,
9910c2209c 2023-08-04  289: 							..self.clone()
9910c2209c 2023-08-04  290: 						};
9910c2209c 2023-08-04  291: 						task::spawn(async move {
9910c2209c 2023-08-04  292: 							if let Err(err) = clone.check(&source_id, owner, true).await {
e624ef9d66 2025-04-20  293: 								if let Err(err) = clone.send(&format!("šŸ›‘ {err:?}"), None, None).await {
e624ef9d66 2025-04-20  294: 									eprintln!("Check error: {err:?}");
79c91a5357 2024-08-02  295: 									// clone.disable(&source_id, owner).await.unwrap();
9910c2209c 2023-08-04  296: 								};
9910c2209c 2023-08-04  297: 							};
9910c2209c 2023-08-04  298: 						});
9910c2209c 2023-08-04  299: 					}
9910c2209c 2023-08-04  300: 				} else if next_fetch - now < delay {
9910c2209c 2023-08-04  301: 					delay = next_fetch - now;
9910c2209c 2023-08-04  302: 				}
9910c2209c 2023-08-04  303: 			}
9910c2209c 2023-08-04  304: 		};
26339860ce 2022-02-13  305: 		Ok(delay.to_std()?)
26339860ce 2022-02-13  306: 	}
26339860ce 2022-02-13  307: 
0340541002 2025-04-24  308: 	pub async fn list (&mut self, owner: i64) -> Result<String> {
26339860ce 2022-02-13  309: 		let mut reply: Vec<Cow<str>> = vec![];
26339860ce 2022-02-13  310: 		reply.push("Channels:".into());
0340541002 2025-04-24  311: 		let mut conn = self.db.begin().await?;
0340541002 2025-04-24  312: 		for row in conn.get_list(owner).await? {
79c91a5357 2024-08-02  313: 			reply.push(format!("\n\\#ļøāƒ£ {} \\*ļøāƒ£ `{}` {}\nšŸ”— `{}`", row.source_id, row.channel,
9910c2209c 2023-08-04  314: 				match row.enabled {
26339860ce 2022-02-13  315: 					true  => "šŸ”„ enabled",
26339860ce 2022-02-13  316: 					false => "ā›” disabled",
9910c2209c 2023-08-04  317: 				}, row.url).into());
9910c2209c 2023-08-04  318: 			if let Some(hash) = &row.iv_hash {
e624ef9d66 2025-04-20  319: 				reply.push(format!("IV: `{hash}`").into());
26339860ce 2022-02-13  320: 			}
9910c2209c 2023-08-04  321: 			if let Some(re) = &row.url_re {
e624ef9d66 2025-04-20  322: 				reply.push(format!("RE: `{re}`").into());
a7f91033c0 2021-11-13  323: 			}
a7f91033c0 2021-11-13  324: 		};
a7f91033c0 2021-11-13  325: 		Ok(reply.join("\n"))
e624ef9d66 2025-04-20  326: 	}
9171c791eb 2021-11-13  327: }