1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
// SPDX-FileCopyrightText: 2021-2025 Robin Vobruba <hoijui.quaero@gmail.com>
//
// SPDX-License-Identifier: AGPL-3.0-or-later
mod cli;
mod stream_test;
use async_std::{fs, io, path};
use clap::crate_name;
use cli_utils::BoxError;
use fs4::async_std::AsyncFileExt;
use futures::{pin_mut, Stream, StreamExt};
use std::pin::pin;
use thiserror::Error;
// use futures::future::select_all;
use futures::stream::select_all;
use okh_scraper::settings::{self, SettingsError};
use cli_utils::logging;
use tracing::instrument;
// use okh_scraper::Settings;
use tracing_subscriber::filter::LevelFilter;
// use std::collections::HashMap;
#[allow(clippy::print_stdout)]
fn print_version_and_exit(quiet: bool) {
if !quiet {
print!("{} ", clap::crate_name!());
}
println!("{}", okh_scraper::VERSION);
std::process::exit(0);
}
#[derive(Debug, Clone, Copy)]
pub enum LockingErrorKind {
CreateFile,
OpenFile,
Locking,
Unlocking,
}
#[derive(Error, Debug)]
pub enum Error {
#[error("Failed to initialize the logging/tracing system (1/2): {0}")]
LoggingInit(#[from] tracing_subscriber::util::TryInitError),
#[error("Failed to initialize the logging/tracing system (2/2): {0}")]
LoggingInit2(#[from] tracing_subscriber::reload::Error),
#[error("Failed to ensure exclusivity (locking): Is there already a scraper running (on the same machine or in an (other) container/VM?)? - error-kind: {0:#?}, lock-file: '{1:#?}'")]
Locking(LockingErrorKind, path::PathBuf, Option<BoxError>),
#[error("Failed to initialize/construct the configuration: {0}")]
Settings(#[from] SettingsError),
#[error("Database/Workdir '{0:#?}' does not exist")]
MissingDatabaseDir(path::PathBuf),
}
macro_rules! lock_file {
($path_var:ident, $file_var:ident) => {
if !$path_var.exists().await {
fs::File::create(&$path_var).await.map_err(|err| {
Error::Locking(
LockingErrorKind::CreateFile,
$path_var.clone(),
Some(err.into()),
)
})?;
}
tracing::debug!("Preparing to lock file '{}' ...", $path_var.display());
let $file_var = fs::File::open(&$path_var).await.map_err(|err| {
Error::Locking(
LockingErrorKind::OpenFile,
$path_var.clone(),
Some(err.into()),
)
})?;
if !$file_var.try_lock_exclusive().map_err(|err| {
Error::Locking(
LockingErrorKind::Locking,
$path_var.clone(),
Some(err.into()),
)
})? {
return Err(Error::Locking(
LockingErrorKind::Locking,
$path_var.clone(),
None,
));
}
tracing::debug!("Obtained lock on file '{}'.", $path_var.display());
};
}
macro_rules! unlock_file {
($path_var:ident, $file_var:ident) => {
tracing::trace!("Releasing lock on file '{}' ...", $path_var.display());
$file_var.unlock().map_err(|err| {
Error::Locking(
LockingErrorKind::Unlocking,
$path_var.clone(),
Some(err.into()),
)
})?;
tracing::info!("Released lock on file '{}'.", $path_var.display());
};
}
#[tokio::main]
#[instrument]
async fn main() -> Result<(), Error> {
let log_reload_handle = logging::setup(crate_name!())?;
let args = cli::args_matcher().get_matches();
let quiet = args.get_flag(cli::A_L_QUIET);
let version = args.get_flag(cli::A_L_VERSION);
if version {
print_version_and_exit(quiet);
}
let verbose = args.get_flag(cli::A_L_VERBOSE);
let log_level = if verbose {
// LevelFilter::DEBUG
LevelFilter::TRACE
} else if quiet {
LevelFilter::WARN
} else {
LevelFilter::INFO
};
logging::set_log_level_tracing(&log_reload_handle, log_level)?;
let list = args.get_flag(cli::A_L_LIST);
let src = args.get_one::<String>(cli::A_L_INPUT).cloned();
let dst = args.get_one::<String>(cli::A_L_OUTPUT).cloned();
// With this we try to enforce
// that at most one scraper instance is running at a time,
// on this system.
let system_lock_file_path = path::PathBuf::from("/tmp/okh-scraper.lock");
lock_file!(system_lock_file_path, system_lock_file);
let run_settings = settings::load()?;
// create the async version of the workdir path
let workdir = path::PathBuf::from(&run_settings.database.path);
// ensure the workdir exists
if !workdir.exists().await {
return Err(Error::MissingDatabaseDir(workdir));
}
// With this we try to enforce
// that at most one scraper instance is running at a time,
// that uses this workdir.
//
// We need both these lock files,
// because otherwise one might run the scraper once (or more times) in a container,
// and at the same time on the host machine, using the same workdir,
// which would mess up the storage completely.
//
// If we had only this one nad not the former,
// One might run the scraper multiple times on the same machine,
// using different workdirs,
// which would under-cut the API requests per minute quotas.
//
// NOTE This way, one could still run one scraper in a container
// and one on the host, using different workdirs.
// -> Don't do that!
let workdir_lock_file_path = workdir.join("okh-scraper-workdir.lock");
lock_file!(workdir_lock_file_path, workdir_lock_file);
let mut scrape_streams = Vec::new();
// TODO Parallelize this loop
tracing::info!("Setting up scrapers ...");
for (scraper_id, scraper) in run_settings.scrapers {
tracing::info!("- Setting up scraper {scraper_id} ...");
scrape_streams.push(scraper.scrape().await);
}
// stream_test::test().await;
// let scrapers = run_settings.scrapers.into_iter().map(|(scraper_id, scraper)| async move {scraper.scrape_all().await});
let mut projects = select_all(scrape_streams);
// pin_mut!(projects); // needed for iteration
while let Some(project_or_err) = projects.next().await {
match project_or_err {
Ok(proj) => println!("Scraped project: {}", proj.id),
Err(err) => {
if err.aborts() {
println!("Scraping error:\n{err}");
} else {
println!("Scraping failed:\n{err}");
}
}
}
}
unlock_file!(system_lock_file_path, system_lock_file);
unlock_file!(workdir_lock_file_path, workdir_lock_file);
// scrapers::appropedia::scrape_all().await?;
// scrapers::oshwa::scrape_all().await?;
// if list {
// let detected_vars = replacer::extract_from_file(src.as_deref())?;
// tools::write_to_file(detected_vars, dst.as_deref())?;
// } else {
// let mut vars = HashMap::new();
// // enlist environment variables
// if args.get_flag(cli::A_L_ENVIRONMENT) {
// tools::append_env(&mut vars);
// }
// // enlist variables from files
// if let Some(var_files) = args.get_many::<String>(cli::A_L_VARIABLES_FILE) {
// for var_file in var_files {
// let mut reader = cli_utils::create_input_reader(Some(var_file))?;
// vars.extend(key_value::parse_vars_file_reader(&mut reader)?);
// }
// }
// // enlist variables provided on the CLI
// if let Some(variables) = args.get_many::<String>(cli::A_L_VARIABLE) {
// for key_value in variables {
// let pair = key_value::Pair::parse(key_value)?;
// vars.insert(pair.key.to_owned(), pair.value.to_owned());
// }
// }
// let fail_on_missing = args.get_flag(cli::A_L_FAIL_ON_MISSING_VALUES);
// let settings = settings! {
// vars: vars,
// fail_on_missing: fail_on_missing
// };
// replacer::replace_in_file(src.as_deref(), dst.as_deref(), &settings)?;
// }
Ok(())
}