"""Facts for the report "Business Insider: 47% of newsroom titles carry a rank word". python3 tools/report_rank_words.py python3 tools/report_rank_words.py --sanity python3 tools/report_rank_words.py --coverage Reads the published base (data/journalists.csv.gz, platform_domain == 0 and media_typed == 1, 127,602 rows) and rebuilds the outlet table the way tools/export.py does, by importing that module's helpers, so outlet names, countries and types are identical to the site's. One lexicon carries the report: the narrow set, ten words — chief, executive, deputy, senior, head of, director, vp, vice president, principal, lead. Every count, share, ranking and cut in this file is on the narrow set. A wider set, the narrow set plus bare "head" and "managing", is carried in the wide_* columns as a second reading and is never used for a ranking. Junior words (assistant, associate, junior, trainee) are counted in their own columns and are never part of the rank count or the language check. Rank words are matched over the job title only, lowercased and whitespace-collapsed, never over the profile headline. Writes the dataset CSVs under static/data/business-insider-47-percent-of-newsroom-titles-carry-a-rank-word/ and prints every number with its denominator. Counts between 1 and 4 print "<5". No person-level row is written: every CSV is a count per outlet, country, type, role or title string. """ import collections import csv import gzip import os import re import shutil import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) sys.path.insert(0, os.path.join(ROOT, "tools")) import export as E # noqa: E402 (tools/export.py) SLUG = "business-insider-47-percent-of-newsroom-titles-carry-a-rank-word" OUT = os.path.join(ROOT, "static", "data", SLUG) EXPECT_BASE = 127602 OUTLET_MIN_ROWS = 100 # the cut for the outlet ranking COUNTRY_MIN_ROWS = 200 # the cut the site uses for a country page TITLE_MIN_COUNT = 5 # the cut for title_frequency.csv _WS = re.compile(r"\s+") # ── the lexicon ────────────────────────────────────────────────────── # One rule: a word is in the lexicon when it places the holder above or below # a plain title (a "senior reporter" and an "assistant editor" both carry a # rank; a "reporter" and an "editor" do not). Words that name a desk, a beat, # a medium or a contract (national, digital, freelance, staff) are not ranks # and are out. # # The narrow set is the report's set. "head" counts only as "head of"; bare # "head" and "managing" are in the wide set only. NARROW_WORD_RES = [ ("senior", r"\b(senior|snr|sr\.?)\b"), ("head of", r"\bhead\s+of\b"), ("director", r"\bdirector\b"), ("chief", r"\b(chief|editor[\s\-]in[\s\-]chief)\b"), ("deputy", r"\bdeputy\b"), ("executive", r"\b(executive|exec)\b"), ("lead", r"\blead\b"), ("vp", r"\b(vp|vice\s+president)\b"), ("principal", r"\bprincipal\b"), ] NARROW_WORD_ORDER = [w for w, _ in NARROW_WORD_RES] NARROW_WORD_RE = {w: re.compile(p) for w, p in NARROW_WORD_RES} # The wide set: the narrow set plus the two words that are only sometimes a # rank. Kept as a second column, never used to rank. WIDE_EXTRA_RES = [ ("head (bare)", r"\bhead\b"), ("managing", r"\bmanaging\b"), ] WIDE_EXTRA_ORDER = [w for w, _ in WIDE_EXTRA_RES] WIDE_EXTRA_RE = {w: re.compile(p) for w, p in WIDE_EXTRA_RES} JUNIOR_WORD_RES = [ ("assistant", r"\b(assistant|asst\.?)\b"), ("associate", r"\b(associate|assoc\.?)\b"), ("junior", r"\b(junior|jr\.?)\b"), ("trainee", r"\b(trainee|apprentice)\b"), ] JUNIOR_WORD_ORDER = [w for w, _ in JUNIOR_WORD_RES] JUNIOR_WORD_RE = {w: re.compile(p) for w, p in JUNIOR_WORD_RES} # Combined matchers. NARROW_RE = re.compile("|".join("(?:%s)" % p for _, p in NARROW_WORD_RES)) WIDE_RE = re.compile("|".join( "(?:%s)" % p for _, p in (NARROW_WORD_RES + WIDE_EXTRA_RES))) JUNIOR_RE = re.compile("|".join("(?:%s)" % p for _, p in JUNIOR_WORD_RES)) # English job-noun check for the language item. ENGLISH_JOB_RE = re.compile(r"\b(editor|reporter|journalist)\b") # The four outlets whose commonest titles are listed. TITLE_OUTLETS = ["Sveriges Radio", "SVT", "Business Insider", "Daily Mail"] def cell(n): return "<5" if 0 < n < 5 else str(n) def pct(part, whole): return round(100.0 * part / whole, 1) if whole else None def fmt(x): return "" if x is None else ("%.1f" % x) def norm_title(text): return _WS.sub(" ", (text or "").lower().replace("&", " and ")).strip() # ── load ───────────────────────────────────────────────────────────── def load(): records = E.load_outlet_records() rows_by_outlet = collections.Counter() domains_by_outlet = collections.defaultdict(set) company_names = collections.defaultdict(collections.Counter) country_of_outlet = collections.defaultdict(collections.Counter) narrow_by_outlet = collections.Counter() wide_by_outlet = collections.Counter() junior_by_outlet = collections.Counter() any_by_outlet = collections.Counter() word_by_outlet = collections.defaultdict(collections.Counter) titles_by_outlet = collections.defaultdict(collections.Counter) english_by_outlet = collections.Counter() no_english_by_outlet = collections.Counter() rows_by_country = collections.Counter() narrow_by_country = collections.Counter() wide_by_country = collections.Counter() junior_by_country = collections.Counter() any_by_country = collections.Counter() rows_by_role = collections.Counter() narrow_by_role = collections.Counter() wide_by_role = collections.Counter() junior_by_role = collections.Counter() any_by_role = collections.Counter() title_counts = collections.Counter() word_all = collections.Counter() junior_word_all = collections.Counter() narrow_all = wide_all = junior_all = any_all = 0 blank_titles = 0 stamps = [] base = dropped_platform = dropped_not_media = 0 with gzip.open(os.path.join(ROOT, "data", "journalists.csv.gz"), "rt", newline="") as fh: for row in csv.DictReader(fh): if row["platform_domain"] == "1": dropped_platform += 1 continue if row["media_typed"] != "1": dropped_not_media += 1 continue base += 1 key = row["outlet"] code = row["country_code"] role = row["role"] raw_title = _WS.sub(" ", (row["title"] or "").strip()) title = norm_title(row["title"]) if not title: blank_titles += 1 if row.get("fetched_at"): stamps.append(row["fetched_at"]) rows_by_outlet[key] += 1 domains_by_outlet[key].add(row["company_domain"]) if row["company_name"]: company_names[key][row["company_name"]] += 1 if code: country_of_outlet[key][code] += 1 rows_by_country[code] += 1 rows_by_role[role] += 1 titles_by_outlet[key][raw_title] += 1 title_counts[raw_title] += 1 nar = bool(NARROW_RE.search(title)) wid = bool(WIDE_RE.search(title)) jun = bool(JUNIOR_RE.search(title)) if nar: narrow_all += 1 narrow_by_outlet[key] += 1 narrow_by_country[code] += 1 narrow_by_role[role] += 1 if wid: wide_all += 1 wide_by_outlet[key] += 1 wide_by_country[code] += 1 wide_by_role[role] += 1 if jun: junior_all += 1 junior_by_outlet[key] += 1 junior_by_country[code] += 1 junior_by_role[role] += 1 if nar or jun: any_all += 1 any_by_outlet[key] += 1 any_by_country[code] += 1 any_by_role[role] += 1 for word in NARROW_WORD_ORDER: if NARROW_WORD_RE[word].search(title): word_all[word] += 1 word_by_outlet[key][word] += 1 for word in WIDE_EXTRA_ORDER: if WIDE_EXTRA_RE[word].search(title): word_all[word] += 1 word_by_outlet[key][word] += 1 for word in JUNIOR_WORD_ORDER: if JUNIOR_WORD_RE[word].search(title): junior_word_all[word] += 1 word_by_outlet[key][word] += 1 # the language rule, item 3: no narrow rank word and no English # job noun. Junior words play no part. if ENGLISH_JOB_RE.search(title): english_by_outlet[key] += 1 if not nar and not ENGLISH_JOB_RE.search(title): no_english_by_outlet[key] += 1 outlets = {} for key, count in rows_by_outlet.items(): domains = sorted(domains_by_outlet[key]) recs = [r for d in domains for r in records.get(d, [])] modal = company_names[key].most_common(1) name = E.outlet_name(key, domains, recs, modal[0][0] if modal else "") entry = { "key": key, "name": name, "rows": count, "domains": domains, "country_code": E.outlet_country_code_for(key, recs) or E.fallback_country_code(country_of_outlet[key]), "type": E.outlet_type(recs), "parent": bool(set(domains) & E.PARENT_DOMAINS), "row_countries": country_of_outlet[key], } entry["country"] = E.COUNTRIES.get(entry["country_code"], "") outlets[key] = entry return dict( base=base, dropped_platform=dropped_platform, dropped_not_media=dropped_not_media, blank_titles=blank_titles, outlets=outlets, rows_by_outlet=rows_by_outlet, narrow_by_outlet=narrow_by_outlet, wide_by_outlet=wide_by_outlet, junior_by_outlet=junior_by_outlet, any_by_outlet=any_by_outlet, word_by_outlet=word_by_outlet, titles_by_outlet=titles_by_outlet, english_by_outlet=english_by_outlet, no_english_by_outlet=no_english_by_outlet, rows_by_country=rows_by_country, narrow_by_country=narrow_by_country, wide_by_country=wide_by_country, junior_by_country=junior_by_country, any_by_country=any_by_country, rows_by_role=rows_by_role, narrow_by_role=narrow_by_role, wide_by_role=wide_by_role, junior_by_role=junior_by_role, any_by_role=any_by_role, word_all=word_all, junior_word_all=junior_word_all, narrow_all=narrow_all, wide_all=wide_all, junior_all=junior_all, any_all=any_all, title_counts=title_counts, stamps=stamps, ) class _Tee: def __init__(self, path): self.fh = open(path, "w", encoding="utf-8") self.out = sys.stdout def write(self, text): self.out.write(text) self.fh.write(text) def flush(self): self.out.flush() self.fh.flush() def main(): os.makedirs(OUT, exist_ok=True) sys.stdout = _Tee(os.path.join(OUT, "facts_console.txt")) D = load() base = D["base"] outlets = D["outlets"] dom = lambda k: ";".join(outlets[k]["domains"]) # noqa: E731 # ── item 9: the funnel ─────────────────────────────────────────── print("# funnel") print("rows in data/journalists.csv.gz read: %d" % (base + D["dropped_platform"] + D["dropped_not_media"])) print("dropped, platform_domain == 1: %d" % D["dropped_platform"]) print("dropped, media_typed == 0: %d" % D["dropped_not_media"]) print("published base: %d (expected %d)" % (base, EXPECT_BASE)) assert base == EXPECT_BASE, base print("rows with a blank job title in the base: %d" % D["blank_titles"]) stamps = sorted(D["stamps"]) print("fetched_at over the base: rows with a stamp %d of %d" % (len(stamps), base)) print(" min %s" % stamps[0]) print(" median %s" % stamps[len(stamps) // 2]) print(" max %s" % stamps[-1]) by_month = collections.Counter(s[:7] for s in stamps) print(" by month: %s" % ", ".join("%s %d" % (m, n) for m, n in sorted(by_month.items()))) # ── item 1: the lexicon and the base share ─────────────────────── print("\n# lexicon, matched over the job title only (lowercased, " "whitespace collapsed, & -> and)") print("rule: a word is in the lexicon when it puts the holder above or " "below a plain title.") print("PRIMARY: the narrow set. Every ranking, cut and headline number " "below is on it.") for word in NARROW_WORD_ORDER: print(" narrow %-12s %s" % (word, NARROW_WORD_RE[word].pattern)) print("SECONDARY: the wide set = the narrow set plus these two. Reported " "in wide_* columns only, never ranked on.") for word in WIDE_EXTRA_ORDER: print(" wide+ %-12s %s" % (word, WIDE_EXTRA_RE[word].pattern)) print("junior words, their own columns, never part of the rank count:") for word in JUNIOR_WORD_ORDER: print(" junior %-12s %s" % (word, JUNIOR_WORD_RE[word].pattern)) print("\n# item 1: share of the base") for label, n in (("rank word, NARROW SET (the report's)", D["narrow_all"]), ("rank word, wide set (secondary)", D["wide_all"]), ("junior word", D["junior_all"]), ("any rank word (narrow or junior)", D["any_all"])): print(" %-38s %7d of %d %4.1f%%" % (label, n, base, 100.0 * n / base)) print(" the plan's 19,273 of 127,602, 15.1%%, on the narrow set: %s" % ("confirmed" if D["narrow_all"] == 19273 else "NOT confirmed, got %d" % D["narrow_all"])) print(" Indeed Hiring Lab, 2026-07-23: 14% of US job postings are for " "senior positions, under Indeed's own seniority classification.") print(" Our narrow-set figure is %.1f%% of newsroom job titles carrying " "one of ten rank words. Different populations and different " "measures; a scale check, not a comparison." % (100.0 * D["narrow_all"] / base)) print(" per word, over the base of %d:" % base) for word in NARROW_WORD_ORDER: n = D["word_all"][word] print(" narrow %-12s %6d %4.1f%%" % (word, n, 100.0 * n / base)) for word in WIDE_EXTRA_ORDER: n = D["word_all"][word] print(" wide+ %-12s %6d %4.1f%%" % (word, n, 100.0 * n / base)) for word in JUNIOR_WORD_ORDER: n = D["junior_word_all"][word] print(" junior %-12s %6d %4.1f%%" % (word, n, 100.0 * n / base)) # ── item 2: the outlet ranking ─────────────────────────────────── big = [k for k, n in D["rows_by_outlet"].items() if n >= OUTLET_MIN_ROWS] big_named = [k for k in big if outlets[k]["name"]] def key_narrow(k): return (-D["narrow_by_outlet"][k] / D["rows_by_outlet"][k], -D["rows_by_outlet"][k]) ranked202 = sorted(big_named, key=key_narrow) no_parent = [k for k in big_named if not outlets[k]["parent"]] ranked = sorted(no_parent, key=key_narrow) # the primary ranking rank_of = {k: i for i, k in enumerate(ranked, 1)} rows_in_big = sum(D["rows_by_outlet"][k] for k in big_named) rows_in_194 = sum(D["rows_by_outlet"][k] for k in ranked) print("\n# item 2: outlets with %d+ editorial rows" % OUTLET_MIN_ROWS) print("outlets in the base: %d" % len(outlets)) print("outlets with %d+ rows: %d" % (OUTLET_MIN_ROWS, len(big))) print("of those with a publishable name: %d, holding %d rows (%.1f%% of %d)" % (len(big_named), rows_in_big, 100.0 * rows_in_big / base, base)) unnamed = [k for k in big if not outlets[k]["name"]] print("with no publishable name, excluded: %d (%s)" % (len(unnamed), ", ".join(sorted(unnamed)[:10]) or "none")) parents = [k for k in ranked202 if outlets[k]["parent"]] print("parent-company domains among the %d: %d (%s)" % (len(ranked202), len(parents), ", ".join("%s %s" % (outlets[k]["name"], k) for k in parents))) print("THE RANKING: the %d outlets left once the parent-company domains " "are dropped, holding %d rows (%.1f%% of %d). Ranked 1..%d by the " "exact unrounded narrow-set share, ties by editorial rows " "descending." % (len(ranked), rows_in_194, 100.0 * rows_in_194 / base, base, len(ranked))) untyped = [k for k in ranked if not outlets[k]["country"] or not outlets[k]["type"]] print("in the ranking but missing a country or a type, so without a site " "page: %d (%s)" % (len(untyped), ", ".join( "%s: country %r type %r" % (outlets[k]["name"], outlets[k]["country"], outlets[k]["type"]) for k in untyped) or "none")) head = ["rank", "outlet", "outlet_key", "outlet_domain", "country", "outlet_type", "editorial_rows", "rank_count", "rank_share_pct", "wide_rank_count", "wide_rank_share_pct", "junior_rank_count", "junior_rank_share_pct", "any_rank_count", "any_rank_share_pct"] def outlet_row(i, k, parent_col=False): e = outlets[k] n = D["rows_by_outlet"][k] row = [i, e["name"], k, dom(k), e["country"], e["type"], n] if parent_col: row.insert(6, "yes" if e["parent"] else "no") return row + [ cell(D["narrow_by_outlet"][k]), fmt(pct(D["narrow_by_outlet"][k], n)), cell(D["wide_by_outlet"][k]), fmt(pct(D["wide_by_outlet"][k], n)), cell(D["junior_by_outlet"][k]), fmt(pct(D["junior_by_outlet"][k], n)), cell(D["any_by_outlet"][k]), fmt(pct(D["any_by_outlet"][k], n))] path = os.path.join(OUT, "outlets_ranked_194.csv") with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(head) for i, k in enumerate(ranked, 1): w.writerow(outlet_row(i, k)) print("wrote %s (%d outlets)" % (os.path.basename(path), len(ranked))) path = os.path.join(OUT, "outlets_100plus_rank_share.csv") with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) h = list(head) h.insert(6, "parent_company_domain") w.writerow(h) for i, k in enumerate(ranked202, 1): w.writerow(outlet_row(i, k, parent_col=True)) print("wrote %s (%d outlets, parent-company domains flagged)" % (os.path.basename(path), len(ranked202))) def line(i, k): e = outlets[k] n = D["rows_by_outlet"][k] print("%3d. %-32s %-16s %-14s rows %5d narrow %5s %5s%% " "wide %5s %5s%% junior %5s%% any %5s%%" % (i, e["name"][:32], e["country"][:16], (e["type"] or "")[:14], n, cell(D["narrow_by_outlet"][k]), fmt(pct(D["narrow_by_outlet"][k], n)), cell(D["wide_by_outlet"][k]), fmt(pct(D["wide_by_outlet"][k], n)), fmt(pct(D["junior_by_outlet"][k], n)), fmt(pct(D["any_by_outlet"][k], n)))) print("\ntop 10 of the %d, narrow set:" % len(ranked)) for i, k in enumerate(ranked[:10], 1): line(i, k) print("bottom 10 of the %d, narrow set:" % len(ranked)) for i, k in enumerate(ranked[-10:], len(ranked) - 9): line(i, k) bottom10 = ranked[-10:] bcast = [k for k in bottom10 if outlets[k]["type"] == "broadcasters"] print("broadcasters among the bottom ten: %d of 10 (%s)" % (len(bcast), ", ".join(outlets[k]["name"] for k in bcast))) print("types of the bottom ten: %s" % ", ".join("%s %s" % (outlets[k]["name"], outlets[k]["type"] or "(untyped)") for k in bottom10)) print("\nfor reference only, the same ranking with the 8 parent-company " "domains kept (%d outlets):" % len(ranked202)) print(" top 10: %s" % ", ".join( "%d. %s %s%%" % (i, outlets[k]["name"], fmt(pct(D["narrow_by_outlet"][k], D["rows_by_outlet"][k]))) for i, k in enumerate(ranked202[:10], 1))) # named checks by_name = {} for k in ranked: by_name.setdefault(outlets[k]["name"], k) by_name202 = {} for k in big_named: by_name202.setdefault(outlets[k]["name"], k) print("\nnamed checks, narrow set, rank out of %d:" % len(ranked)) for name in ("Business Insider", "Sveriges Radio", "BFMTV", "Columbia Daily Spectator", "Law360", "Daily Mail", "Moneycontrol", "The Indian Express", "Press Trust of India", "SVT", "DR - Danmarks Radio", "Aftonbladet", "TV 2", "CNN-News18", "The New York Times", "BBC"): k = by_name.get(name) or by_name202.get(name) if not k: print(" %-26s NOT FOUND" % name) continue n = D["rows_by_outlet"][k] print(" %-26s rank %4s of %d narrow %5s/%d %5s%% wide %5s %5s%% " "country %s type %s" % (name, rank_of.get(k, "n/a (parent)"), len(ranked), cell(D["narrow_by_outlet"][k]), n, fmt(pct(D["narrow_by_outlet"][k], n)), cell(D["wide_by_outlet"][k]), fmt(pct(D["wide_by_outlet"][k], n)), outlets[k]["country"] or "(none)", outlets[k]["type"] or "(none)")) bi = by_name.get("Business Insider") print(" Business Insider rank 1 at 135/286: %s" % ("confirmed" if rank_of.get(bi) == 1 and D["narrow_by_outlet"][bi] == 135 and D["rows_by_outlet"][bi] == 286 else "NOT confirmed")) last = ranked[-1] print(" last of the %d is %s: %s" % (len(ranked), outlets[last]["name"], "BFMTV confirmed" if outlets[last]["name"] == "BFMTV" else "NOT BFMTV")) # ── item 3: which word does the work ───────────────────────────── ends = ranked[:10] + ranked[-10:] all_words = NARROW_WORD_ORDER + WIDE_EXTRA_ORDER + JUNIOR_WORD_ORDER path = os.path.join(OUT, "rank_word_breakdown_top_bottom.csv") with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(["end", "rank", "outlet", "outlet_key", "outlet_domain", "country", "outlet_type", "editorial_rows", "rank_count", "rank_share_pct", "wide_rank_count", "wide_rank_share_pct"] + ["w_" + re.sub(r"[^a-z0-9]+", "_", x).strip("_") for x in all_words]) for k in ends: e = outlets[k] n = D["rows_by_outlet"][k] i = rank_of[k] w.writerow(["top" if i <= 10 else "bottom", i, e["name"], k, dom(k), e["country"], e["type"], n, cell(D["narrow_by_outlet"][k]), fmt(pct(D["narrow_by_outlet"][k], n)), cell(D["wide_by_outlet"][k]), fmt(pct(D["wide_by_outlet"][k], n))] + [cell(D["word_by_outlet"][k][x]) for x in all_words]) print("\n# item 3: which rank word does the work, top 10 and bottom 10 " "of the %d" % len(ranked)) print("%-30s %6s %s" % ("outlet", "rows", " ".join("%10s" % x for x in NARROW_WORD_ORDER))) for k in ends: print("%-30s %6d %s" % (outlets[k]["name"][:30], D["rows_by_outlet"][k], " ".join("%10s" % cell(D["word_by_outlet"][k][x]) for x in NARROW_WORD_ORDER))) # ── item 4: country, type, role ────────────────────────────────── path = os.path.join(OUT, "rank_share_by_country.csv") countries = [(c, n) for c, n in D["rows_by_country"].most_common() if c and n >= COUNTRY_MIN_ROWS and E.COUNTRIES.get(c)] countries.sort(key=lambda t: (-D["narrow_by_country"][t[0]] / t[1], -t[1])) crank = {c: i for i, (c, _) in enumerate(countries, 1)} with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(["rank", "country_code", "country", "editorial_rows", "rank_count", "rank_share_pct", "wide_rank_count", "wide_rank_share_pct", "junior_rank_count", "junior_rank_share_pct", "any_rank_share_pct"]) for i, (c, n) in enumerate(countries, 1): w.writerow([i, c, E.COUNTRIES[c], n, cell(D["narrow_by_country"][c]), fmt(pct(D["narrow_by_country"][c], n)), cell(D["wide_by_country"][c]), fmt(pct(D["wide_by_country"][c], n)), cell(D["junior_by_country"][c]), fmt(pct(D["junior_by_country"][c], n)), fmt(pct(D["any_by_country"][c], n))]) print("\n# item 4a: by country, %d+ rows, %d countries, narrow set" % (COUNTRY_MIN_ROWS, len(countries))) for i, (c, n) in enumerate(countries, 1): print("%3d. %-22s rows %6d narrow %6s %5s%% wide %5s%% junior %5s%%" % (i, E.COUNTRIES[c], n, cell(D["narrow_by_country"][c]), fmt(pct(D["narrow_by_country"][c], n)), fmt(pct(D["wide_by_country"][c], n)), fmt(pct(D["junior_by_country"][c], n)))) top_c, top_n = countries[0] bot_c, bot_n = countries[-1] print("country extremes on the narrow set: top %s %d/%d %s%%, bottom %s " "%d/%d %s%%" % (E.COUNTRIES[top_c], D["narrow_by_country"][top_c], top_n, fmt(pct(D["narrow_by_country"][top_c], top_n)), E.COUNTRIES[bot_c], D["narrow_by_country"][bot_c], bot_n, fmt(pct(D["narrow_by_country"][bot_c], bot_n)))) # by type and by role, over the whole base rows_by_type = collections.Counter() narrow_by_type = collections.Counter() wide_by_type = collections.Counter() junior_by_type = collections.Counter() any_by_type = collections.Counter() for k, e in outlets.items(): t = e["type"] or "(untyped)" rows_by_type[t] += D["rows_by_outlet"][k] narrow_by_type[t] += D["narrow_by_outlet"][k] wide_by_type[t] += D["wide_by_outlet"][k] junior_by_type[t] += D["junior_by_outlet"][k] any_by_type[t] += D["any_by_outlet"][k] named_types = sum(n for t, n in rows_by_type.items() if t != "(untyped)") print("\n# item 4b: by outlet type, denominator each type's own rows, " "narrow set") print("the five named buckets hold %d rows; the (untyped) bucket holds " "%d; together %d, the whole base" % (named_types, rows_by_type["(untyped)"], named_types + rows_by_type["(untyped)"])) path = os.path.join(OUT, "rank_share_by_type_and_role.csv") with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(["cut", "segment", "editorial_rows", "rank_count", "rank_share_pct", "wide_rank_count", "wide_rank_share_pct", "junior_rank_count", "junior_rank_share_pct", "any_rank_share_pct"]) for t, n in sorted(rows_by_type.items(), key=lambda kv: -narrow_by_type[kv[0]] / kv[1]): w.writerow(["outlet_type", t, n, cell(narrow_by_type[t]), fmt(pct(narrow_by_type[t], n)), cell(wide_by_type[t]), fmt(pct(wide_by_type[t], n)), cell(junior_by_type[t]), fmt(pct(junior_by_type[t], n)), fmt(pct(any_by_type[t], n))]) print(" %-14s rows %7d narrow %6s %5s%% wide %5s%% " "junior %5s%%" % (t, n, cell(narrow_by_type[t]), fmt(pct(narrow_by_type[t], n)), fmt(pct(wide_by_type[t], n)), fmt(pct(junior_by_type[t], n)))) print("\n# item 4c: by role, denominator each role's own rows, " "narrow set") for r, n in sorted(D["rows_by_role"].items(), key=lambda kv: -D["narrow_by_role"][kv[0]] / kv[1]): w.writerow(["role", r, n, cell(D["narrow_by_role"][r]), fmt(pct(D["narrow_by_role"][r], n)), cell(D["wide_by_role"][r]), fmt(pct(D["wide_by_role"][r], n)), cell(D["junior_by_role"][r]), fmt(pct(D["junior_by_role"][r], n)), fmt(pct(D["any_by_role"][r], n))]) print(" %-18s rows %7d narrow %6s %5s%% wide %5s%% " "junior %5s%%" % (r, n, cell(D["narrow_by_role"][r]), fmt(pct(D["narrow_by_role"][r], n)), fmt(pct(D["wide_by_role"][r], n)), fmt(pct(D["junior_by_role"][r], n)))) w.writerow(["all", "published base", base, D["narrow_all"], fmt(pct(D["narrow_all"], base)), D["wide_all"], fmt(pct(D["wide_all"], base)), D["junior_all"], fmt(pct(D["junior_all"], base)), fmt(pct(D["any_all"], base))]) print(" editors against reporters, narrow set: editor %d of %d (%s%%), " "reporter %d of %d (%s%%)" % (D["narrow_by_role"]["editor"], D["rows_by_role"]["editor"], fmt(pct(D["narrow_by_role"]["editor"], D["rows_by_role"]["editor"])), D["narrow_by_role"]["reporter"], D["rows_by_role"]["reporter"], fmt(pct(D["narrow_by_role"]["reporter"], D["rows_by_role"]["reporter"])))) # ── item 5: the Nordic floor ───────────────────────────────────── print("\n# item 5: the Nordic countries, narrow set, denominator each " "country's rows, rank out of the %d countries" % len(countries)) for c in ("SE", "NO", "DK", "FI"): n = D["rows_by_country"][c] if not n: print(" %s: no rows" % c) continue print(" %-10s rank %2s of %d rows %6d narrow %5s %5s%% " "wide %5s%% junior %5s%%" % (E.COUNTRIES.get(c, c), crank.get(c, "n/a"), len(countries), n, cell(D["narrow_by_country"][c]), fmt(pct(D["narrow_by_country"][c], n)), fmt(pct(D["wide_by_country"][c], n)), fmt(pct(D["junior_by_country"][c], n)))) four = [(name, by_name.get(name) or by_name202.get(name)) for name in TITLE_OUTLETS] path = os.path.join(OUT, "common_titles_four_outlets.csv") with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(["outlet", "outlet_key", "outlet_domain", "country", "editorial_rows", "rank", "title", "rows_with_this_title", "share_of_outlet_rows_pct", "narrow_rank_word", "wide_rank_word"]) for name, k in four: if not k: print(" %s NOT FOUND" % name) continue n = D["rows_by_outlet"][k] print("\n ten commonest exact job titles at %s (%d editorial " "rows)" % (outlets[k]["name"], n)) for i, (title, c) in enumerate( D["titles_by_outlet"][k].most_common(10), 1): t = norm_title(title) nar = bool(NARROW_RE.search(t)) wid = bool(WIDE_RE.search(t)) w.writerow([outlets[k]["name"], k, dom(k), outlets[k]["country"], n, i, title, cell(c), fmt(pct(c, n)), "yes" if nar else "no", "yes" if wid else "no"]) print(" %2d. %-46s %5s %5s%% %s" % (i, title[:46], cell(c), fmt(pct(c, n)), "narrow rank word" if nar else ("wide only" if wid else ""))) # ── item 6/3: the language check ───────────────────────────────── bottom20 = ranked[-20:] path = os.path.join(OUT, "bottom20_language_check.csv") print("\n# item 6: the bottom 20 of the %d, language check, narrow set" % len(ranked)) print("rule: a title counts when it carries no rank word from the narrow " "set AND none of the English job nouns editor, reporter, " "journalist. No junior word is used. The denominator is every row " "at the bottom-20 outlets, all of which are tested.") tested = flagged = eng = 0 with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(["rank", "outlet", "outlet_key", "outlet_domain", "country", "country_code", "outlet_type", "editorial_rows", "rank_count", "rank_share_pct", "titles_with_editor_reporter_or_journalist", "english_job_noun_share_pct", "titles_with_no_rank_word_and_no_english_job_noun", "no_rank_word_no_job_noun_share_pct"]) for k in bottom20: e = outlets[k] n = D["rows_by_outlet"][k] i = rank_of[k] tested += n flagged += D["no_english_by_outlet"][k] eng += D["english_by_outlet"][k] w.writerow([i, e["name"], k, dom(k), e["country"], e["country_code"], e["type"], n, cell(D["narrow_by_outlet"][k]), fmt(pct(D["narrow_by_outlet"][k], n)), cell(D["english_by_outlet"][k]), fmt(pct(D["english_by_outlet"][k], n)), cell(D["no_english_by_outlet"][k]), fmt(pct(D["no_english_by_outlet"][k], n))]) print("%3d. %-30s %-16s rows %5d narrow %5s%% " "editor/reporter/journalist %5s %5s%% no rank word and no " "job noun %5s %5s%%" % (i, e["name"][:30], e["country"][:16], n, fmt(pct(D["narrow_by_outlet"][k], n)), cell(D["english_by_outlet"][k]), fmt(pct(D["english_by_outlet"][k], n)), cell(D["no_english_by_outlet"][k]), fmt(pct(D["no_english_by_outlet"][k], n)))) print("rows tested (every row at the 20 outlets): %d" % tested) print("titles with no narrow rank word and no English job noun: %d of %d " "(%.1f%%)" % (flagged, tested, 100.0 * flagged / tested)) print("titles carrying editor, reporter or journalist: %d of %d (%.1f%%)" % (eng, tested, 100.0 * eng / tested)) # ── item 7: the NYT and the BBC ────────────────────────────────── print("\n# item 7: the NYT and the BBC on the narrow set, all editorial " "rows") for name in ("The New York Times", "BBC"): k = next((key for key in outlets if outlets[key]["name"] == name), None) if not k: print(" %s NOT FOUND" % name) continue n = D["rows_by_outlet"][k] print(" %-20s rows %5d narrow %4d %4.1f%% wide %4d %4.1f%% " "junior %4d %4.1f%% any %4.1f%% rank %s of %d" % (name, n, D["narrow_by_outlet"][k], pct(D["narrow_by_outlet"][k], n), D["wide_by_outlet"][k], pct(D["wide_by_outlet"][k], n), D["junior_by_outlet"][k], pct(D["junior_by_outlet"][k], n), pct(D["any_by_outlet"][k], n), rank_of.get(k, "n/a"), len(ranked))) print(" per word: %s" % ", ".join("%s %s" % (x, cell(D["word_by_outlet"][k][x])) for x in NARROW_WORD_ORDER + WIDE_EXTRA_ORDER)) # ── item 8: the country disagreement scan ──────────────────────── print("\n# country check: outlets with %d+ rows whose published country " "differs from the country of 80%%+ of their own rows" % OUTLET_MIN_ROWS) disagree = [] for k in big_named: e = outlets[k] c = e["row_countries"] total = sum(c.values()) if not total: continue code, n = c.most_common(1)[0] if n / total >= 0.80 and code != e["country_code"]: disagree.append((e["name"], k, e["country_code"] or "(none)", code, n, total, D["rows_by_outlet"][k])) disagree.sort(key=lambda t: -t[6]) for name, k, pub, own, n, total, rows in disagree: print(" %-30s %-22s published %s, %d of %d rows with a country " "(%.0f%%) say %s, %d editorial rows" % (name[:30], k, pub, n, total, 100.0 * n / total, own, rows)) print(" %d outlets" % len(disagree)) # ── item 8b: the title frequency table ─────────────────────────── path = os.path.join(OUT, "title_frequency.csv") freq = [(t, n) for t, n in D["title_counts"].items() if n >= TITLE_MIN_COUNT] freq.sort(key=lambda kv: (-kv[1], kv[0].lower())) with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(["title", "count", "narrow_rank_word", "wide_rank_word"]) for t, n in freq: nt = norm_title(t) w.writerow([t, n, "true" if NARROW_RE.search(nt) else "false", "true" if WIDE_RE.search(nt) else "false"]) covered = sum(n for _, n in freq) nar_rows = sum(n for t, n in freq if NARROW_RE.search(norm_title(t))) print("\n# title frequency table") print("distinct job-title strings in the base: %d" % len(D["title_counts"])) print("strings occurring %d or more times: %d, covering %d of %d rows " "(%.1f%%)" % (TITLE_MIN_COUNT, len(freq), covered, base, 100.0 * covered / base)) print("of those rows, carrying a narrow rank word: %d (%.1f%% of the " "covered rows)" % (nar_rows, 100.0 * nar_rows / covered)) # ── the lexicon, shipped ───────────────────────────────────────── with open(os.path.join(OUT, "lexicon.txt"), "w", encoding="utf-8") as fh: fh.write("Rank-word lexicon, JournalistLabs.\n" "Matched over the job title only, lowercased, whitespace " "collapsed, & read as and.\nA word is in the lexicon when it " "puts the holder above or below a plain title.\n\n" "NARROW SET — the report's set. Every count, share and " "ranking uses it.\n") for word in NARROW_WORD_ORDER: fh.write(" %-12s %s\n" % (word, NARROW_WORD_RE[word].pattern)) fh.write("\nWIDE SET — the narrow set plus the two words below. " "Reported in the wide_* columns\nas a second reading; never " "used to rank.\n") for word in WIDE_EXTRA_ORDER: fh.write(" %-12s %s\n" % (word, WIDE_EXTRA_RE[word].pattern)) fh.write("\njunior words, counted in their own columns and never part " "of the rank count\n") for word in JUNIOR_WORD_ORDER: fh.write(" %-12s %s\n" % (word, JUNIOR_WORD_RE[word].pattern)) fh.write("\nany rank word = a narrow rank word or a junior word.\n") fh.write("\nEnglish job-noun test, used in the language check\n %s\n" % ENGLISH_JOB_RE.pattern) fh.write("\nLanguage check rule: a title counts when it carries no " "narrow rank word and\nno English job noun. No country list " "and no junior word is used.\n") shutil.copyfile(os.path.abspath(__file__), os.path.join(OUT, "report_rank_words.py")) print("\nwrote lexicon.txt and copied report_rank_words.py into the " "dataset") print("wrote 8 CSVs to %s" % OUT) # ── the sanity check ───────────────────────────────────────────────── SANITY_URLS = [ # A public share of senior roles, same unit as ours (a share), different # measure: Indeed's own seniority classification of US job postings. ("indeed-hiring-lab-seniority", "https://www.hiringlab.org/2026/07/23/the-labor-market-is-tilting-toward-seniority/"), # A growth figure, not a share; kept as context, never compared with ours. ("globaltimes-linkedin-title-growth", "https://www.globaltimes.cn/content/925396.shtml"), ("wikipedia-job-title-inflation", "https://en.wikipedia.org/wiki/Job_title_inflation"), # Tried and unusable, recorded so the failures are on the record. ("economist-job-title-inflation", "https://www.economist.com/business/2022/12/07/the-scourge-of-job-title-inflation"), ("bbc-worklife-job-title-inflation", "https://www.bbc.com/worklife/article/20220826-the-rise-of-job-title-inflation"), ] COVERAGE_URLS = [ ("wikipedia-bbc-news", "https://en.wikipedia.org/wiki/BBC_News", r"[^.]{0,220}(?:journalist|staff|employe|newsroom)[^.]{0,180}\."), ("wikipedia-nyt", "https://en.wikipedia.org/wiki/The_New_York_Times", r"[^.]{0,220}(?:journalist|newsroom|staff|reporter)[^.]{0,180}\."), ] def _fetch_and_print(name, url, pattern, limit=10): import html as htmlmod from proxy_fetch import fetch_response_with_fallback try: r = fetch_response_with_fallback( url, "journalistlabs_rank_words_sanity", timeout=45) except Exception as exc: # noqa: BLE001 print("=== %s FAILED %s" % (name, exc)) return txt = re.sub(r"|", "", r.text, flags=re.S) txt = re.sub(r"\s+", " ", htmlmod.unescape(re.sub(r"<[^>]+>", " ", txt))) print("=== %s HTTP %d %s" % (name, r.status_code, url)) shown = 0 for m in re.compile(pattern, re.I).findall(txt): if re.search(r"\d[\d,]{2,}|\d+(?:\.\d+)?\s?(?:%|percent|per cent)", m): print(" " + m.strip()[:320]) shown += 1 if shown >= limit: break if not shown: print(" no sentence with a number matched") def sanity(): """Fetched through the Webshare rotating gateway, never from this IP.""" os.makedirs(OUT, exist_ok=True) sys.stdout = _Tee(os.path.join(OUT, "sanity_console.txt")) sys.path.insert(0, os.path.join(os.path.dirname(ROOT), "ai_sending_tool_v2")) pat = (r"[^.]{0,220}(?:senior|seniority|vice president|chief|" r"title inflation|job title)[^.]{0,180}\.") for name, url in SANITY_URLS: _fetch_and_print(name, url, pat, limit=8) def coverage(): """The published newsroom sizes the coverage section cites, fetched through the Webshare rotating gateway, never from this IP.""" os.makedirs(OUT, exist_ok=True) sys.stdout = _Tee(os.path.join(OUT, "coverage_console.txt")) sys.path.insert(0, os.path.join(os.path.dirname(ROOT), "ai_sending_tool_v2")) for name, url, pat in COVERAGE_URLS: _fetch_and_print(name, url, pat, limit=12) if __name__ == "__main__": if "--sanity" in sys.argv: sanity() elif "--coverage" in sys.argv: coverage() else: main()