Annotation For src/core.rs
Logged in as anonymous

Lines of src/core.rs from check-in b4af85e31b that are changed by the sequence of edits moving toward check-in 5a4aab7687:

                         1: use crate::{
                         2: 	command,
                         3: 	sql::Db,
                         4: };
                         5: 
                         6: use std::{
                         7: 	borrow::Cow,
                         8: 	collections::{
                         9: 		BTreeMap,
                        10: 		HashSet,
                        11: 	},
                        12: 	sync::{
                        13: 		Arc,
                        14: 		Mutex
                        15: 	},
                        16: };
                        17: 
                        18: use anyhow::{
                        19: 	anyhow,
                        20: 	bail,
                        21: 	Result,
                        22: };
                        23: use async_std::task;
                        24: use chrono::DateTime;
                        25: use tgbot::{
                        26: 	api::Client,
                        27: 	handler::UpdateHandler,
                        28: 	types::{
                        29: 		Bot,
                        30: 		ChatPeerId,
                        31: 		Command,
                        32: 		GetBot,
                        33: 		Message,
                        34: 		ParseMode,
                        35: 		SendMessage,
                        36: 		Update,
                        37: 		UpdateType,
                        38: 		UserPeerId,
                        39: 	},
                        40: };
                        41: 
                        42: #[derive(Clone)]
                        43: pub struct Core {
                        44: 	owner_chat: ChatPeerId,
                        45: 	// max_delay: u16,
                        46: 	pub tg: Client,
                        47: 	pub me: Bot,
                        48: 	pub db: Db,
                        49: 	sources: Arc<Mutex<HashSet<Arc<i32>>>>,
                        50: 	http_client: reqwest::Client,
                        51: }
                        52: 
                        53: impl Core {
                        54: 	pub async fn new(settings: config::Config) -> Result<Core> {
                        55: 		let owner_chat = ChatPeerId::from(settings.get_int("owner")?);
                        56: 		let api_key = settings.get_string("api_key")?;
                        57: 		let tg = Client::new(&api_key)?;
                        58: 
                        59: 		let mut client = reqwest::Client::builder();
                        60: 		if let Ok(proxy) = settings.get_string("proxy") {
                        61: 			let proxy = reqwest::Proxy::all(proxy)?;
                        62: 			client = client.proxy(proxy);
                        63: 		}
                        64: 		let http_client = client.build()?;
                        65: 		let me = tg.execute(GetBot).await?;
                        66: 		let core = Core {
                        67: 			tg,
                        68: 			me,
                        69: 			owner_chat,
                        70: 			db: Db::new(&settings.get_string("pg")?)?,
                        71: 			sources: Arc::new(Mutex::new(HashSet::new())),
                        72: 			http_client,
                        73: 			// max_delay: 60,
                        74: 		};
                        75: 		let clone = core.clone();
                        76: 		task::spawn(async move {
                        77: 			loop {
                        78: 				let delay = match &clone.autofetch().await {
                        79: 					Err(err) => {
                        80: 						if let Err(err) = clone.send(format!("šŸ›‘ {err:?}"), None, None).await {
                        81: 							eprintln!("Autofetch error: {err:?}");
                        82: 						};
                        83: 						std::time::Duration::from_secs(60)
                        84: 					},
                        85: 					Ok(time) => *time,
                        86: 				};
                        87: 				task::sleep(delay).await;
                        88: 			}
                        89: 		});
                        90: 		Ok(core)
                        91: 	}
                        92: 
                        93: 	pub async fn send <S>(&self, msg: S, target: Option<ChatPeerId>, mode: Option<ParseMode>) -> Result<Message>
                        94: 	where S: Into<String> {
                        95: 		let msg = msg.into();
                        96: 
                        97: 		let mode = mode.unwrap_or(ParseMode::Html);
                        98: 		let target = target.unwrap_or(self.owner_chat);
                        99: 		Ok(self.tg.execute(
                       100: 			SendMessage::new(target, msg)
                       101: 				.with_parse_mode(mode)
                       102: 		).await?)
                       103: 	}
                       104: 
                       105: 	pub async fn check (&self, id: i32, real: bool) -> Result<String> {
                       106: 		let mut posted: i32 = 0;
                       107: 		let mut conn = self.db.begin().await?;
                       108: 
                       109: 		let id = {
                       110: 			let mut set = self.sources.lock().unwrap();
                       111: 			match set.get(&id) {
                       112: 				Some(id) => id.clone(),
                       113: 				None => {
                       114: 					let id = Arc::new(id);
                       115: 					set.insert(id.clone());
                       116: 					id.clone()
                       117: 				},
                       118: 			}
                       119: 		};
                       120: 		let count = Arc::strong_count(&id);
                       121: 		if count == 2 {
                       122: 			let source = conn.get_source(*id, self.owner_chat).await?;
                       123: 			conn.set_scrape(*id).await?;
                       124: 			let destination = ChatPeerId::from(match real {
                       125: 				true => source.channel_id,
                       126: 				false => source.owner,
                       127: 			});
                       128: 			let mut this_fetch: Option<DateTime<chrono::FixedOffset>> = None;
                       129: 			let mut posts: BTreeMap<DateTime<chrono::FixedOffset>, String> = BTreeMap::new();
                       130: 
                       131: 			let response = self.http_client.get(&source.url).send().await?;
                       132: 			let status = response.status();
                       133: 			let content = response.bytes().await?;
                       134: 			match rss::Channel::read_from(&content[..]) {
                       135: 				Ok(feed) => {
                       136: 					for item in feed.items() {
                       137: 						if let Some(link) = item.link() {
                       138: 							let date = match item.pub_date() {
                       139: 								Some(feed_date) => DateTime::parse_from_rfc2822(feed_date),
                       140: 								None => DateTime::parse_from_rfc3339(&item.dublin_core_ext().unwrap().dates()[0]),
                       141: 							}?;
                       142: 							let url = link;
                       143: 							posts.insert(date, url.to_string());
                       144: 						}
                       145: 					};
                       146: 				},
                       147: 				Err(err) => match err {
                       148: 					rss::Error::InvalidStartTag => {
                       149: 						match atom_syndication::Feed::read_from(&content[..]) {
                       150: 							Ok(feed) => {
                       151: 								for item in feed.entries() {
                       152: 									let date = item.published().unwrap();
                       153: 									let url = item.links()[0].href();
                       154: 									posts.insert(*date, url.to_string());
                       155: 								};
                       156: 							},
                       157: 							Err(err) => {
                       158: 								bail!("Unsupported or mangled content:\n{:?}\n{err:#?}\n{status:#?}\n", &source.url)
                       159: 							},
                       160: 						}
                       161: 					},
                       162: 					rss::Error::Eof => (),
                       163: 					_ => bail!("Unsupported or mangled content:\n{:?}\n{err:#?}\n{status:#?}\n", &source.url)
                       164: 				}
                       165: 			};
                       166: 			for (date, url) in posts.iter() {
                       167: 				let post_url: Cow<str> = match source.url_re {
                       168: 					Some(ref x) => sedregex::ReplaceCommand::new(x)?.execute(url),
                       169: 					None => url.into(),
                       170: 				};
                       171: 				if let Some(exists) = conn.exists(&post_url, *id).await? {
                       172: 					if ! exists {
                       173: 						if this_fetch.is_none() || *date > this_fetch.unwrap() {
                       174: 							this_fetch = Some(*date);
                       175: 						};
                       176: 						self.send( match &source.iv_hash {
                       177: 							Some(hash) => format!("<a href=\"https://t.me/iv?url={post_url}&rhash={hash}\"> </a>{post_url}"),
                       178: 							None => format!("{post_url}"),
                       179: 						}, Some(destination), Some(ParseMode::Html)).await?;
                       180: 						conn.add_post(*id, date, &post_url).await?;
                       181: 					};
                       182: 				};
                       183: 				posted += 1;
                       184: 			};
                       185: 			posts.clear();
                       186: 		};
                       187: 		Ok(format!("Posted: {posted}"))
                       188: 	}
                       189: 
                       190: 	async fn autofetch(&self) -> Result<std::time::Duration> {
                       191: 		let mut delay = chrono::Duration::minutes(1);
                       192: 		let now = chrono::Local::now();
                       193: 		let queue = {
                       194: 			let mut conn = self.db.begin().await?;
                       195: 			conn.get_queue().await?
                       196: 		};
                       197: 		for row in queue {
                       198: 			if let Some(next_fetch) = row.next_fetch {
                       199: 				if next_fetch < now {
                       200: 					if let (Some(owner), Some(source_id)) = (row.owner, row.source_id) {
                       201: 						let clone = Core {
                       202: 							owner_chat: ChatPeerId::from(owner),
                       203: 							..self.clone()
                       204: 						};
                       205: 						let source = {
                       206: 							let mut conn = self.db.begin().await?;
                       207: 							match conn.get_one(owner, source_id).await {
                       208: 								Ok(Some(source)) => source.to_string(),
                       209: 								Ok(None) => "Source not found in database?".to_string(),
                       210: 								Err(err) => format!("Failed to fetch source data:\n{err}"),
                       211: 							}
                       212: 						};
                       213: 						task::spawn(async move {
                       214: 							if let Err(err) = clone.check(source_id, true).await {
b4af85e31b 2025-06-29  215: 								if let Err(err) = clone.send(&format!("{source}\nšŸ›‘ {err:?}"), None, None).await {
                       216: 									eprintln!("Check error: {err:?}");
                       217: 									// clone.disable(&source_id, owner).await.unwrap();
                       218: 								};
                       219: 							};
                       220: 						});
                       221: 					}
                       222: 				} else if next_fetch - now < delay {
                       223: 					delay = next_fetch - now;
                       224: 				}
                       225: 			}
                       226: 		};
                       227: 		Ok(delay.to_std()?)
                       228: 	}
                       229: 
                       230: 	pub async fn list (&self, owner: UserPeerId) -> Result<String> {
                       231: 		let mut reply: Vec<String> = vec![];
                       232: 		reply.push("Channels:".into());
                       233: 		let mut conn = self.db.begin().await?;
                       234: 		for row in conn.get_list(owner).await? {
                       235: 			reply.push(row.to_string());
                       236: 		};
                       237: 		Ok(reply.join("\n\n"))
                       238: 	}
                       239: }
                       240: 
                       241: impl UpdateHandler for Core {
                       242: 	async fn handle (&self, update: Update) {
                       243: 		if let UpdateType::Message(msg) = update.update_type {
                       244: 			if let Ok(cmd) = Command::try_from(msg) {
                       245: 				let msg = cmd.get_message();
                       246: 				let words = cmd.get_args();
                       247: 				let command = cmd.get_name();
                       248: 				let res = match command {
                       249: 					"/check" | "/clean" | "/enable" | "/delete" | "/disable" => command::command(self, command, msg, words).await,
                       250: 					"/start" => command::start(self, msg).await,
                       251: 					"/list" => command::list(self, msg).await,
                       252: 					"/add" | "/update" => command::update(self, command, msg, words).await,
                       253: 					any => Err(anyhow!("Unknown command: {any}")),
                       254: 				};
                       255: 				if let Err(err) = res {
                       256: 					if let Err(err2) = self.send(format!("\\#error\n```\n{err:?}\n```"),
                       257: 						Some(msg.chat.get_id()),
                       258: 						Some(ParseMode::MarkdownV2)
                       259: 					).await{
                       260: 						dbg!(err2);
                       261: 					};
                       262: 				}
                       263: 			};
                       264: 		};
                       265: 	}
                       266: }