ember/mediabias-plotter - Change IJS5OIDK4YRQVHTN2MZWP3SFLS7CJMIMCDMWZBLDV5PDNMEU7MYAC

multithread phase1, compress the output

Created by ember on January 10, 2021

IJS5OIDK4YRQVHTN2MZWP3SFLS7CJMIMCDMWZBLDV5PDNMEU7MYAC

Dependencies

In channels

main

Change contents

Insertion in src/main.rs at line 4 [4.10]
[3.70]
[3.70]
```
    sync::Arc,
```
Insertion in src/main.rs at line 10 [4.10]
[5.20]
[3.94]
```
use parking_lot::RwLock;
```

Replacement in src/main.rs at line 15 [4.10]

B:BD[3.188] → [5.21:107]

use tokio::sync::broadcast;
use tracing::{error, info, metadata::LevelFilter, Level};

[3.188]

[5.107]

use tokio::{sync::broadcast, task::JoinHandle};
use tracing::{error, info, metadata::LevelFilter, warn, Level};

Replacement in src/main.rs at line 88 [4.10]

B:BD[6.324] → [6.324:390]

    cl: &Client,
    confirmed: &mut HashMap<Url, PageCandidate>,

[6.324]

[6.390]

    cl: Arc<Client>,
    confirmed: Arc<RwLock<HashMap<Url, PageCandidate>>>,

Replacement in src/main.rs at line 91 [4.10]

B:BD[6.416] → [5.1200:1787]

    let mut pending: HashSet<Url> = if confirmed.is_empty() {
        info!("starting from root, with no cache");
        iter::once("https://mediabiasfactcheck.com/".try_into()?).collect()
    } else {
        let mut new = HashSet::new();
        confirmed
            .values()
            .map(|v| {
                find_new_links(
                    std::str::from_utf8(&v.data).expect("nonutf8"),
                    &confirmed,
                    &mut new,
                )
            })
            .for_each(|_| ());
        info!("scraping through {:?}", new);
        new

[6.416]

[5.1787]

    let mut pending: HashSet<Url> = {
        let confirmed = confirmed.read();
        if confirmed.is_empty() {
            info!("starting from root, with no cache");
            iter::once("https://mediabiasfactcheck.com/".try_into()?).collect()
        } else {
            let mut new = HashSet::new();
            confirmed
                .values()
                .map(|v| {
                    if let Ok(utf8) = std::str::from_utf8(&v.data) {
                        if let Err(e) = find_new_links(utf8, &confirmed, &mut new) {
                            warn!("error finding links on {}: {}", v.final_url, e);
                        }
                    }
                })
                .for_each(|_| ());
            info!("scraping through {:?}", new);
            new
        }

Insertion in src/main.rs at line 113 [4.10]

[3.644]

[5.1795]

    let (pending_tx, mut pending_rx) = tokio::sync::mpsc::unbounded_channel();
    pending
        .drain()
        .for_each(|pnd| pending_tx.send(pnd.clone()).expect("couldn't send ):"));

Replacement in src/main.rs at line 121 [4.10]

B:BD[3.827] → [6.417:516]

    while let Some(next_url) = pending.iter().cloned().next() {
        pending.remove(&next_url);

[3.827]

[3.916]

    while let Some(next_url) = pending_rx.recv().await {
        let cl = cl.clone();
        let confirmed = confirmed.clone();
        let pending_tx = pending_tx.clone();
        let _: JoinHandle<anyhow::Result<()>> = tokio::task::spawn(async move {
            required_sleep().await;

Replacement in src/main.rs at line 128 [4.10]

B:BD[3.917] → [6.517:549]

        required_sleep().await;

[3.917]

[5.1886]

            let pg = cl.get(next_url.clone()).send().await?;
            let final_url = pg.url().clone();

Replacement in src/main.rs at line 131 [4.10]

∅:D[5.1887] → [6.549:687]

B:BD[6.549] → [6.549:687]

        let pg = cl.get(next_url.clone()).send().await?;
        let final_url = pg.url().clone();
        let data = pg.bytes().await?;

[5.1887]

[5.1888]

            let data = pg.bytes().await?;

Replacement in src/main.rs at line 133 [4.10]

∅:D[5.1889] → [6.1265:1344]

B:BD[3.1711] → [6.1265:1344]

B:BD[6.1344] → [2.0:98]

        confirmed.insert(next_url.clone(), PageCandidate { final_url, data });
        println!("finished {}/{}: {}", confirmed.len(), confirmed.len()+pending.len(), next_url);

[5.1889]

[6.1393]

            let mut pending = HashSet::new();
            if let Ok(utf8) = std::str::from_utf8(&data) {
                if let Err(e) = find_new_links(utf8, &confirmed.read(), &mut pending) {
                    warn!("error finding links on {}: {}", final_url, e);
                }
            }
            pending
                .drain()
                .for_each(|pnd| pending_tx.send(pnd.clone()).expect("couldn't send ):"));
            confirmed
                .write()
                .insert(next_url.clone(), PageCandidate { final_url, data });
            println!(
                "finished {}/{}: {}",
                confirmed.read().len(),
                confirmed.read().len() + pending.len(),
                next_url
            );
            Ok(())
        });

Insertion in src/main.rs at line 161 [4.10]

[3.1739]

fn consider_page(s: &Selector, p: &PageCandidate) -> anyhow::Result<Option<MediaOutlet>> {
    let doc = Html::parse_document(std::str::from_utf8(&p.data)?);
    for last_img_elems in doc.select(s) {
        // these might be the bias and actuality urls!
    }
    Ok(None)
}

Insertion in src/main.rs at line 204 [4.10]

[6.2164]

[6.2203]


    fn phase<T, U>(fname: &str, func: impl FnOnce(&mut T) -> anyhow::Result<U>) -> anyhow::Result<U>
    where
        T: for<'de> Deserialize<'de> + Serialize + Default,
    {
        let f = std::fs::File::open(fname).map_err(|e| anyhow!(e));
        let mut rehydrated = f
            .and_then(|f| {
                bincode::deserialize_from(std::io::BufReader::new(f)).map_err(|e| anyhow!(e))
            })
            .unwrap_or(T::default());

Insertion in src/main.rs at line 216 [4.10]

[6.2204]

[5.2242]

        let res = func(&mut rehydrated);
        bincode::serialize_into(
            std::io::BufWriter::new(
                std::fs::File::create(fname).expect("couldn't create phase output file ):"),
            ),
            &rehydrated,
        )?;
        res
    }

Replacement in src/main.rs at line 233 [4.10]
B:BD[6.2443] → [5.2311:2341]
```
        let mut confirmed = {
```
[6.2443]
[5.2341]
```
        let confirmed = {
```

Replacement in src/main.rs at line 240 [4.10]

B:BD[6.2626] → [5.2570:2652]

        let r = all_urls_on_site_everywhere(stop, &mut cl, &mut confirmed).await;

[6.2626]

[5.2652]

        let confirmed = Arc::new(RwLock::new(confirmed));
        let r = all_urls_on_site_everywhere(stop, Arc::new(cl), confirmed.clone()).await;

Replacement in src/main.rs at line 247 [4.10]

B:BD[6.2734] → [6.2734:2847]

            std::io::BufWriter::new(std::fs::File::create("phase1.blob").expect("sad")),
            &confirmed,

[6.2734]

[6.2847]

            brotli::CompressorWriter::new(
                std::io::BufWriter::new(std::fs::File::create("phase1.blob.brotli").expect("sad")),
                4096 * 64,
                8,
                22,
            ),
            &*confirmed.read(),

Insertion in src/main.rs at line 256 [4.10]

[6.2865]


    if let Some(phase2) = args.subcommand_matches("find_media_outlets") {
        let all_pages: HashMap<Url, PageCandidate> = {
            let f = std::fs::File::open("phase1.blob").map_err(|e| anyhow!(e));
            f.and_then(|f| {
                bincode::deserialize_from(std::io::BufReader::new(f)).map_err(|e| anyhow!(e))
            })?
        };

Insertion in src/main.rs at line 265 [4.10]

[6.2866]

        let sel = Selector::parse("header.entry-header > h2 > img + img")
            .expect("fix the phase2 selector");
        let mut outlets: HashMap<Url, MediaOutlet> = {
            let f = std::fs::File::open("phase2.blob").map_err(|e| anyhow!(e));
            f.and_then(|f| {
                bincode::deserialize_from(std::io::BufReader::new(f)).map_err(|e| anyhow!(e))
            })?
        };
        for (orig_url, candidate) in all_pages.into_iter() {
            if !outlets.contains_key(&orig_url) {
                continue;
            } // a filter would borrow outlets during the iterations
            if let Some(outlet) = consider_page(&sel, &candidate)? {
                println!("found outlet!");
                outlets.insert(orig_url, outlet);
            }
        }
    }

Deletion in Cargo.toml at line 2 [7.20]
B:BD[7.31] → [7.31:76]
```
name = "mediabias-plotter"
version = "0.1.0"
```
Insertion in Cargo.toml at line 4 [7.20]
[7.139]
[7.139]
```
name = "mediabias-plotter"
version = "0.1.0"
```

Insertion in Cargo.toml at line 10 [7.20]

[7.253]

[3.1795]

anyhow = "1.0.37"
async-stream = "0.3"
bincode = "1.3"
brotli = "3"
bytes = {version = "1", features = ["serde"]}
chrono = {version = "0.4.19", features = ["serde"]}
clap = "3.0.0-beta.2"
parking_lot = "0.11"

Insertion in Cargo.toml at line 19 [7.20]
[3.1814]
[3.1814]
```
scraper = "0.12"
```

Replacement in Cargo.toml at line 21 [7.20]

B:BD[3.1831] → [3.1831:1885]

∅:D[3.1885] → [7.319:414]

B:BD[7.319] → [7.319:414]

chrono = { version = "0.4.19", features = ["serde"] }
serde = { version = "1", features = ["derive"] }
url = { version = "2", features = ["serde"] }

[3.1831]

[3.1886]

serde = {version = "1", features = ["derive"]}

Replacement in Cargo.toml at line 23 [7.20]

B:BD[3.1909] → [3.1909:1995]

B:BD[3.1995] → [6.2914:3000]

async-stream = "0.3"
anyhow = "1.0.37"
tokio = { version = "1", features = ["full"] }
scraper = "0.12"
bytes = { version = "1", features = ["serde"]}
clap = "3.0.0-beta.2"

[3.1909]

[6.3000]

tokio = {version = "1", features = ["full"]}

Replacement in Cargo.toml at line 25 [7.20]

B:BD[6.3016] → [6.3016:3061]

bincode = "1.3"
tracing-subscriber = "0.2.15"

[6.3016]

tracing-subscriber = "0.2.15"
url = {version = "2", features = ["serde"]}

Insertion in Cargo.lock at line 3 [7.450]

[7.542]

[[package]]
name = "alloc-no-stdlib"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5192ec435945d87bc2f70992b4d818154b5feede43c09fb7592146374eac90a6"

Insertion in Cargo.lock at line 10 [7.450]

[7.554]

[6.3062]

name = "alloc-stdlib"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "697ed7edc0f1711de49ce108c541623a0af97c6c60b2f6e2b65229847ac843c2"
dependencies = [
 "alloc-no-stdlib",
]
[[package]]

Insertion in Cargo.lock at line 107 [7.450]

[7.2050]

[3.2918]


[[package]]
name = "brotli"
version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f29919120f08613aadcd4383764e00526fc9f18b6c0895814faeed0dd78613e"
dependencies = [
 "alloc-no-stdlib",
 "alloc-stdlib",
 "brotli-decompressor",
]

Insertion in Cargo.lock at line 120 [7.450]

[3.2931]

name = "brotli-decompressor"
version = "2.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1052e1c3b8d4d80eb84a8b94f0a1498797b5fb96314c001156a1c761940ef4ec"
dependencies = [
 "alloc-no-stdlib",
 "alloc-stdlib",
]
[[package]]

Insertion in Cargo.lock at line 673 [7.450]
[6.4924]
[6.4924]
```
 "brotli",
```
Insertion in Cargo.lock at line 677 [7.450]
[6.4944]
[3.6599]
```
 "parking_lot",
```