""" uc_salary_pipeline.py ===================== All-in-one pipeline: download UC employee wage data from ucinvestments/ UC-wages (https://github.com/ucinvestments/UC-wages), classify every employee into one of 8 analytical categories (see classifier.py), and write a per-employee CSV. USAGE ----- # Download every campus for every available year and classify: python uc_salary_pipeline.py # A subset: python uc_salary_pipeline.py --years 2023 2024 --campuses UCB UCLA # Classify already-downloaded data without hitting the network: python uc_salary_pipeline.py --skip-download DATA ACQUISITION ---------------- Downloading is delegated to uc_wages_downloader.py, which fetches raw JSON from ucinvestments/UC-wages into --data-dir (default uc_wages_data/), skipping files already present. See that module's docstring for details. CLASSIFICATION --------------- Delegated entirely to classifier.py — see that module's docstring for the category definitions. This script only downloads data, maps UC-wages' JSON fields onto the (title, gross_pay) shape classifier.classify_title() expects, and writes the result. OUTPUT ------ One row per employee per year: year, campus, title, category, category_label, gross_pay, regular_pay, overtime_pay, other_pay, first_name, last_name """ import argparse import csv import json import logging import re import sys from pathlib import Path import classifier import uc_wages_downloader as downloader # --------------------------------------------------------------------------- # LOGGING # --------------------------------------------------------------------------- logging.basicConfig( format="%(asctime)s [%(levelname)s] %(message)s", level=logging.INFO, handlers=[ logging.FileHandler("uc_salary_pipeline.log", encoding="utf-8"), logging.StreamHandler(sys.stdout), ], ) log = logging.getLogger(__name__) OUTPUT_COLUMNS = [ "year", "campus", "title", "category", "category_label", "gross_pay", "regular_pay", "overtime_pay", "other_pay", "first_name", "last_name", ] def _dollar(s) -> float: try: return float(re.sub(r"[$,\s]", "", str(s))) except (ValueError, TypeError): return 0.0 def parse_uc_wages_record(record: dict) -> dict | None: """Map one UC-wages JSON record onto the fields classify_title() needs.""" title = (record.get("title") or "").strip() if not title: return None regular = _dollar(record.get("basepay") or 0) overtime = _dollar(record.get("overtimepay") or 0) other = _dollar(record.get("adjustpay") or 0) gross = _dollar(record.get("grosspay") or 0) if gross == 0: gross = regular + overtime + other return { "title": title, "gross_pay": gross, "regular_pay": regular, "overtime_pay": overtime, "other_pay": other, "first_name": (record.get("firstname") or "").strip(), "last_name": (record.get("lastname") or "").strip(), } def process_file(path: Path, campus: str, year: int, writer: csv.writer, senior_threshold: float, support_threshold: float) -> int: """Classify every record in one UC-wages JSON file, writing a CSV row per employee. Returns the number of rows written.""" with open(path, encoding="utf-8") as fh: data = json.load(fh) count = 0 for record in data.get("records", []): parsed = parse_uc_wages_record(record) if parsed is None: continue category = classifier.classify_title( parsed["title"], gross_pay=parsed["gross_pay"], senior_threshold=senior_threshold, support_threshold=support_threshold, ) writer.writerow([ year, campus, parsed["title"], category, classifier.CATEGORY_LABELS[category], round(parsed["gross_pay"], 2), round(parsed["regular_pay"], 2), round(parsed["overtime_pay"], 2), round(parsed["other_pay"], 2), parsed["first_name"], parsed["last_name"], ]) count += 1 return count # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser( description="Download and classify UC employee wage data into 8 categories.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument("--years", nargs="+", type=int, default=downloader.DEFAULT_YEARS, help=f"Year(s) to process (default: 2010-" f"{downloader.DEFAULT_YEARS[-1]}).") parser.add_argument("--campuses", nargs="+", choices=list(downloader.LOCATIONS.keys()), default=list(downloader.LOCATIONS.keys()), metavar="CAMPUS", help="Campuses to process (default: all). Choices: " + " ".join(downloader.LOCATIONS.keys())) parser.add_argument("--data-dir", default="uc_wages_data", help="Local cache directory for downloaded JSON " "(default: uc_wages_data/).") parser.add_argument("--output", default="uc_salary_classified.csv", help="Output CSV path (default: uc_salary_classified.csv).") parser.add_argument("--skip-download", action="store_true", help="Classify whatever is already cached in " "--data-dir without hitting the network.") parser.add_argument("--senior-threshold", type=float, default=classifier.SENIOR_STAFF_THRESHOLD, help=f"Gross pay threshold for SENIOR_PROF_STAFF " f"(default ${classifier.SENIOR_STAFF_THRESHOLD:,}).") parser.add_argument("--support-threshold", type=float, default=classifier.SUPPORT_STAFF_THRESHOLD, help=f"Gross pay below this -> SUPPORT_STAFF " f"(default ${classifier.SUPPORT_STAFF_THRESHOLD:,}).") args = parser.parse_args() data_dir = Path(args.data_dir) if not args.skip_download: downloader.download_all(args.years, args.campuses, data_dir) else: log.info("--skip-download set, classifying cached data only.") output_path = Path(args.output) total_rows = 0 with open(output_path, "w", newline="", encoding="utf-8") as out_fh: writer = csv.writer(out_fh) writer.writerow(OUTPUT_COLUMNS) for year in sorted(args.years): for campus in args.campuses: path = downloader.out_path(campus, year, data_dir) if not path.exists(): log.info("No data for %s %s, skipping.", campus, year) continue count = process_file(path, campus, year, writer, args.senior_threshold, args.support_threshold) total_rows += count log.info("%s %s: classified %d records", campus, year, count) if total_rows == 0: log.error("No data processed.") sys.exit(1) log.info("Total rows written: %d -> %s", total_rows, output_path) if __name__ == "__main__": main()