feat(rust): dlp --since — filter URLs by message timestamp
Rust-only extension (no python counterpart): --since 5d / 12h / 2026-10-04 keeps only URLs posted at/after the cutoff. URLs without a timestamp (e.g. --urls-file lists) are kept with a warning.
This commit is contained in:
@@ -90,6 +90,9 @@ struct DlpArgs {
|
|||||||
/// process only one host class
|
/// process only one host class
|
||||||
#[arg(long)]
|
#[arg(long)]
|
||||||
only: Option<String>,
|
only: Option<String>,
|
||||||
|
/// only URLs posted at/after this point: e.g. 5d, 12h, 2026-10-04
|
||||||
|
#[arg(long, value_name = "WHEN")]
|
||||||
|
since: Option<String>,
|
||||||
/// download at most N pending URLs per host class (test runs)
|
/// download at most N pending URLs per host class (test runs)
|
||||||
#[arg(long, value_name = "N")]
|
#[arg(long, value_name = "N")]
|
||||||
limit: Option<usize>,
|
limit: Option<usize>,
|
||||||
@@ -343,6 +346,7 @@ impl From<&DlpArgs> for orchestrate::RunOpts {
|
|||||||
limit: a.limit,
|
limit: a.limit,
|
||||||
shuffle: a.shuffle,
|
shuffle: a.shuffle,
|
||||||
only: a.only.clone(),
|
only: a.only.clone(),
|
||||||
|
since: a.since.clone(),
|
||||||
dry_run: a.dry_run,
|
dry_run: a.dry_run,
|
||||||
fresh: a.fresh,
|
fresh: a.fresh,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ use std::thread;
|
|||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use anyhow::{Context as _, anyhow};
|
use anyhow::{Context as _, anyhow};
|
||||||
|
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
|
||||||
use serde_json::Value;
|
use serde_json::Value;
|
||||||
use sha2::{Digest, Sha256};
|
use sha2::{Digest, Sha256};
|
||||||
|
|
||||||
@@ -36,6 +37,9 @@ pub struct RunOpts {
|
|||||||
pub limit: Option<usize>,
|
pub limit: Option<usize>,
|
||||||
pub shuffle: bool,
|
pub shuffle: bool,
|
||||||
pub only: Option<String>,
|
pub only: Option<String>,
|
||||||
|
/// only URLs posted at/after this point (Rust-only extension: "5d",
|
||||||
|
/// "12h", "2026-10-04", or an RFC 3339 timestamp)
|
||||||
|
pub since: Option<String>,
|
||||||
pub dry_run: bool,
|
pub dry_run: bool,
|
||||||
pub fresh: bool,
|
pub fresh: bool,
|
||||||
}
|
}
|
||||||
@@ -53,12 +57,56 @@ impl Default for RunOpts {
|
|||||||
limit: None,
|
limit: None,
|
||||||
shuffle: false,
|
shuffle: false,
|
||||||
only: None,
|
only: None,
|
||||||
|
since: None,
|
||||||
dry_run: false,
|
dry_run: false,
|
||||||
fresh: false,
|
fresh: false,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Parse a `--since` value: relative ("5d", "12h", "30m") or absolute
|
||||||
|
/// (date, naive datetime, or RFC 3339). Naive times are read as UTC.
|
||||||
|
pub fn parse_since(s: &str) -> anyhow::Result<DateTime<Utc>> {
|
||||||
|
let t = s.trim();
|
||||||
|
for (suffix, dur) in [
|
||||||
|
("d", chrono::Duration::hours(24)),
|
||||||
|
("h", chrono::Duration::hours(1)),
|
||||||
|
("m", chrono::Duration::minutes(1)),
|
||||||
|
] {
|
||||||
|
if let Some(num) = t.strip_suffix(suffix) {
|
||||||
|
let n: i32 = num.parse().map_err(|_| anyhow!("bad --since value: {s}"))?;
|
||||||
|
return Ok(Utc::now() - dur * n);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if let Ok(dt) = DateTime::parse_from_rfc3339(t) {
|
||||||
|
return Ok(dt.with_timezone(&Utc));
|
||||||
|
}
|
||||||
|
if let Ok(dt) = NaiveDateTime::parse_from_str(t, "%Y-%m-%dT%H:%M:%S%.f") {
|
||||||
|
return Ok(dt.and_utc());
|
||||||
|
}
|
||||||
|
if let Ok(dt) = NaiveDateTime::parse_from_str(t, "%Y-%m-%d %H:%M:%S") {
|
||||||
|
return Ok(dt.and_utc());
|
||||||
|
}
|
||||||
|
if let Ok(d) = NaiveDate::parse_from_str(t, "%Y-%m-%d") {
|
||||||
|
return Ok(d.and_hms_opt(0, 0, 0).unwrap().and_utc());
|
||||||
|
}
|
||||||
|
anyhow::bail!("bad --since value: {s} (use e.g. 5d, 12h, 2026-10-04, or RFC 3339)")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Parse a message timestamp from the export; None -> keep-with-warning.
|
||||||
|
pub fn parse_message_ts(ts: &str) -> Option<DateTime<Utc>> {
|
||||||
|
if let Ok(dt) = DateTime::parse_from_rfc3339(ts) {
|
||||||
|
return Some(dt.with_timezone(&Utc));
|
||||||
|
}
|
||||||
|
if let Ok(dt) = NaiveDateTime::parse_from_str(ts, "%Y-%m-%dT%H:%M:%S%.f") {
|
||||||
|
return Some(dt.and_utc());
|
||||||
|
}
|
||||||
|
if let Ok(d) = NaiveDate::parse_from_str(ts, "%Y-%m-%d") {
|
||||||
|
return Some(d.and_hms_opt(0, 0, 0).unwrap().and_utc());
|
||||||
|
}
|
||||||
|
None
|
||||||
|
}
|
||||||
|
|
||||||
/// sha256 of a file, streamed (python sha256_of).
|
/// sha256 of a file, streamed (python sha256_of).
|
||||||
pub fn sha256_of(path: &Path) -> anyhow::Result<String> {
|
pub fn sha256_of(path: &Path) -> anyhow::Result<String> {
|
||||||
let mut file =
|
let mut file =
|
||||||
@@ -327,6 +375,37 @@ pub fn process_channel(
|
|||||||
.filter(|f| opts.only.as_ref().is_none_or(|o| o.as_str() == f.host))
|
.filter(|f| opts.only.as_ref().is_none_or(|o| o.as_str() == f.host))
|
||||||
.collect();
|
.collect();
|
||||||
|
|
||||||
|
let found: Vec<FoundUrl> = if let Some(since) = &opts.since {
|
||||||
|
let cutoff = parse_since(since)?;
|
||||||
|
let before = found.len();
|
||||||
|
let kept: Vec<FoundUrl> = found
|
||||||
|
.into_iter()
|
||||||
|
.filter(|f| match &f.timestamp {
|
||||||
|
Some(ts) => match parse_message_ts(ts) {
|
||||||
|
Some(dt) => dt >= cutoff,
|
||||||
|
None => {
|
||||||
|
log::warn!("Unparseable timestamp {ts:?} for {} - keeping it", f.url);
|
||||||
|
true
|
||||||
|
}
|
||||||
|
},
|
||||||
|
None => {
|
||||||
|
// e.g. --urls-file lists carry no attribution
|
||||||
|
log::warn!("No timestamp for {} - keeping it", f.url);
|
||||||
|
true
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
log::info!(
|
||||||
|
"--since {}: kept {} of {} URLs (rest are older)",
|
||||||
|
since,
|
||||||
|
kept.len(),
|
||||||
|
before
|
||||||
|
);
|
||||||
|
kept
|
||||||
|
} else {
|
||||||
|
found
|
||||||
|
};
|
||||||
|
|
||||||
let mut gofile_urls: Vec<FoundUrl> = Vec::new();
|
let mut gofile_urls: Vec<FoundUrl> = Vec::new();
|
||||||
let mut z0807_urls: Vec<FoundUrl> = Vec::new();
|
let mut z0807_urls: Vec<FoundUrl> = Vec::new();
|
||||||
for f in &found {
|
for f in &found {
|
||||||
@@ -1025,4 +1104,38 @@ mod tests {
|
|||||||
fs::write(&p, r#"[{"channel_id": "2"}]"#).unwrap();
|
fs::write(&p, r#"[{"channel_id": "2"}]"#).unwrap();
|
||||||
assert!(load_subscriptions(&p).is_err());
|
assert!(load_subscriptions(&p).is_err());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn parse_since_relative_and_absolute() {
|
||||||
|
let now = Utc::now();
|
||||||
|
let cutoff = parse_since("5d").unwrap();
|
||||||
|
assert!(cutoff <= now);
|
||||||
|
assert!(cutoff > now - chrono::Duration::days(6));
|
||||||
|
assert!(parse_since("12h").unwrap() > now - chrono::Duration::hours(13));
|
||||||
|
assert_eq!(
|
||||||
|
parse_since("2026-10-04").unwrap().to_rfc3339(),
|
||||||
|
"2026-10-04T00:00:00+00:00"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
parse_since("2026-10-04T09:30:00Z").unwrap().to_rfc3339(),
|
||||||
|
"2026-10-04T09:30:00+00:00"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
parse_since("2026-10-04 09:30:00").unwrap().to_rfc3339(),
|
||||||
|
"2026-10-04T09:30:00+00:00"
|
||||||
|
);
|
||||||
|
assert!(parse_since("nonsense").is_err());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn parse_message_ts_handles_export_formats() {
|
||||||
|
assert_eq!(
|
||||||
|
parse_message_ts("2026-01-01T00:00:00")
|
||||||
|
.unwrap()
|
||||||
|
.to_rfc3339(),
|
||||||
|
"2026-01-01T00:00:00+00:00"
|
||||||
|
);
|
||||||
|
assert!(parse_message_ts("2026-01-01T00:00:00.123000+00:00").is_some());
|
||||||
|
assert!(parse_message_ts("garbage").is_none());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user