1#!/bin/sh2# Classify the corpus with independent processes on two nodes (M3U96b local + M3U96a remote); macOS pools hang with pyarrow.3# Usage: sh taxonomy/run_classify_shards.sh [N_total] — waits for the pass-2 chain (CHAIN_DONE) before preparing the input.4cd ~/wp44_uqo; . .venv/bin/activate; ulimit -n 65536; N=${1:-56}; HALF=$((N/2)); R=M3U96a5while ! grep -q CHAIN_DONE logs/define_corpus_v2.log 2>/dev/null; do sleep 60; done6rm -rf data/processed/classification; mkdir -p logs/classify7python taxonomy/classify.py --root ~/wp44_uqo --ontology taxonomy/ontology.yaml --prepare > logs/classify/prepare.log 2>&18ssh $R "mkdir -p ~/wp44_uqo/data/processed/classification ~/wp44_uqo/taxonomy ~/wp44_uqo/logs/classify; rm -rf ~/wp44_uqo/data/processed/classification/*"9rsync -aq data/processed/classification/input $R:~/wp44_uqo/data/processed/classification/10rsync -aq taxonomy/classify.py taxonomy/ontology.yaml taxonomy/__init__.py $R:~/wp44_uqo/taxonomy/11i=0; while [ $i -lt $HALF ]; do12 nohup python taxonomy/classify.py --root ~/wp44_uqo --ontology taxonomy/ontology.yaml --run --shard $i/$N > logs/classify/shard_$i.log 2>&1 &13 i=$((i+1)); done14ssh $R "cd ~/wp44_uqo; . .venv/bin/activate; i=$HALF; while [ \$i -lt $N ]; do nohup python taxonomy/classify.py --root ~/wp44_uqo --ontology taxonomy/ontology.yaml --run --shard \$i/$N > logs/classify/shard_\$i.log 2>&1 & i=\$((i+1)); done; echo remote started"15wait16while ssh $R "pgrep -f 'classify.py --root' >/dev/null"; do sleep 60; done17rsync -aq $R:~/wp44_uqo/data/processed/classification/cats/ data/processed/classification/cats/18rsync -aq $R:~/wp44_uqo/data/processed/classification/primary/ data/processed/classification/primary/19python - <<EOF220import duckdb, json, datetime as dt, yaml21con = duckdb.connect(); o = yaml.safe_load(open("taxonomy/ontology.yaml"))22n = con.sql("select count(distinct work_id) from read_parquet('data/processed/classification/primary/*.parquet')").fetchone()[0]23tot = con.sql("select count(*) from read_parquet('data/processed/classification/input/chunk=*/*.parquet')").fetchone()[0]24json.dump({"ontology_version": o.get("version"), "works": tot, "classified": n, "when": dt.datetime.now(dt.UTC).isoformat()}, open("data/processed/classification/classification_report.json", "w"), indent=1)25print("classified", n, "of", tot)26EOF227echo CLASSIFY_DONE >> logs/classify/prepare.log28