#!/bin/sh
# Classify the corpus with independent processes on two nodes (M3U96b local + M3U96a remote); macOS pools hang with pyarrow.
# Usage: sh taxonomy/run_classify_shards.sh [N_total]  — waits for the pass-2 chain (CHAIN_DONE) before preparing the input.
cd ~/wp44_uqo; . .venv/bin/activate; ulimit -n 65536; N=${1:-56}; HALF=$((N/2)); R=M3U96a
while ! grep -q CHAIN_DONE logs/define_corpus_v2.log 2>/dev/null; do sleep 60; done
rm -rf data/processed/classification; mkdir -p logs/classify
python taxonomy/classify.py --root ~/wp44_uqo --ontology taxonomy/ontology.yaml --prepare > logs/classify/prepare.log 2>&1
ssh $R "mkdir -p ~/wp44_uqo/data/processed/classification ~/wp44_uqo/taxonomy ~/wp44_uqo/logs/classify; rm -rf ~/wp44_uqo/data/processed/classification/*"
rsync -aq data/processed/classification/input $R:~/wp44_uqo/data/processed/classification/
rsync -aq taxonomy/classify.py taxonomy/ontology.yaml taxonomy/__init__.py $R:~/wp44_uqo/taxonomy/
i=0; while [ $i -lt $HALF ]; 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
ssh $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"
wait
while ssh $R "pgrep -f 'classify.py --root' >/dev/null"; do sleep 60; done
rsync -aq $R:~/wp44_uqo/data/processed/classification/cats/ data/processed/classification/cats/
rsync -aq $R:~/wp44_uqo/data/processed/classification/primary/ data/processed/classification/primary/
python - <<EOF2
import duckdb, json, datetime as dt, yaml
con = duckdb.connect(); o = yaml.safe_load(open("taxonomy/ontology.yaml"))
n = con.sql("select count(distinct work_id) from read_parquet('data/processed/classification/primary/*.parquet')").fetchone()[0]
tot = con.sql("select count(*) from read_parquet('data/processed/classification/input/chunk=*/*.parquet')").fetchone()[0]
json.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)
print("classified", n, "of", tot)
EOF2
echo CLASSIFY_DONE >> logs/classify/prepare.log