"""Facts for the report "Journalism has 63 AI job titles on file. The biggest desk is the BBC's five". python3 tools/report_ai_desks.py numbers + dataset CSVs python3 tools/report_ai_desks.py --sanity the outside check, via proxy Reads the published base (data/journalists.csv.gz, platform_domain == 0 and media_typed == 1, 127,602 rows) and rebuilds the outlet table by importing tools/export.py, so outlet names, countries, types and the parent-domain exclusion match the site. Two match rules are reported side by side everywhere: title-only the lexicon matches the job title headline-inclusive the lexicon matches title + " || " + headline The headline is a free-text LinkedIn bio, so it carries ex-jobs, side projects and keyword stuffing. Title-only is the strict rule. Writes the dataset CSVs under static/data/journalism-has-63-ai-job-titles-on-file-biggest-desk-bbc-five/ and prints every number with its denominator. No person name is written to any output. """ import collections import csv import datetime import gzip import os import re import shutil import statistics 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 = "journalism-has-63-ai-job-titles-on-file-biggest-desk-bbc-five" OUT = os.path.join(ROOT, "static", "data", SLUG) EXPECT_BASE = 127602 EXPECT_ROWS_READ = 173133 COUNTRY_MIN_ROWS = 200 # the cut the site uses for a country page TOP_N = 50 _WS = re.compile(r"\s+") def norm(text): """Lowercase, & -> and, whitespace collapsed. The same normalisation data/classify.py uses before it reads a title.""" return _WS.sub(" ", (text or "").lower().replace("&", " and ")).strip() # ── lexicons ───────────────────────────────────────────────────────── AI_RE = re.compile( r"\b(ai|artificial intelligence|machine learning|generative ai|llm" r"|chatgpt|openai)\b") # The scout (tools/scout_titles.py) used a looser first alternative, # a\.?i\.? , which also matches the spelling "A.I.". Reported beside the # strict count so the two runs are comparable. AI_RE_SCOUT = re.compile( r"\b(a\.?i\.?|artificial intelligence|machine learning|generative ai|llm" r"|chatgpt|openai)\b") # A hand-read pass over one more spelling: the vendor stores "A.I." with the # periods dropped, so an A.I. title arrives as "a i". Dutch mastheads write # "a.i." for ad interim, which is not this subject, so every hit is printed # and classified by hand in the facts file rather than counted blind. # # This regex is not a superset of the published lexicon and is not a subset of # it either. It matches 65 titles: 60 of the 63 published rows, plus 5 the # published lexicon does not have. The 3 published rows it misses spell the # subject out as "artificial intelligence" and carry no bare "ai" token. AI_RE_SPACED = re.compile(r"\ba\.?\s?i\.?\b") COMPARATORS = { "crypto": r"\b(crypto|cryptocurrency|bitcoin|blockchain|web3|defi|nft)\b", "cybersecurity": r"\b(cyber ?security|cyber-security)\b", "technology": (r"\b(tech|technology|technolog\w+|software|startups?" r"|gadgets?|consumer tech|silicon valley|semiconductor)\b"), "sports": (r"\b(sports?|football|soccer|basketball|nba|nfl|mlb|nhl" r"|cricket|rugby|tennis|golf|olympics?|motorsport|f1" r"|formula one|athletics)\b"), "climate": (r"\b(climate|environment\w*|sustainab\w+|conservation" r"|wildlife|biodiversity|emissions|net zero)\b"), "health": (r"\b(health\w*|medical|medicine|pharma\w*|hospital|nhs" r"|public health|mental health|biotech)\b"), "science": (r"\b(science|scientific|space|astronomy|physics" r"|research corresponden\w+)\b"), } COMPARATOR_RE = {k: re.compile(v) for k, v in COMPARATORS.items()} # ── the deploy / cover rule ────────────────────────────────────────── # # DEPLOY_RE is the rule the scout used, applied to the title only: the title # names a newsroom function performed with AI (innovation, development, # integration, operations, automation, a programme, a project). COVER_RE is # the reporting-job word list. Deploy wins when a title carries both, because # "AI Validation Journalist" validates AI output for a broadcaster and does # not file AI stories. A title with neither is unclear. DEPLOY_RE = re.compile( r"\b(innovation|development|integration|operations|adoption|validation" r"|transcript|assisted|automation|accelerator|strategy|programme|program" r"|project|intrapreneur|ai-hub|ai hub|news ai|editorial ai" r"|generative video|vfx|data and ai)\b") COVER_RE = re.compile( r"\b(reporter|correspondent|writer|journalist|columnist|critic" r"|reviewer)\b") # The second, broader rule named in the task brief: any rank or function word # beside the AI word counts as deploying. Reported as an alternative cut, not # as the headline split, because it reads "AI Editor" (a desk that covers AI) # as deployment. RANK_DEPLOY_RE = re.compile( r"\b(editor|head|director|lead|operations|innovation|development" r"|integration|product|strategy)\b") SENIOR_RE = re.compile(r"\b(head|director|lead|chief|senior)\b") # ── the borderline flag ────────────────────────────────────────────── # # The rows an adversarial reviewer questioned: the title carries the word but # the job is a data-products or production job rather than a news desk. None # is removed from the 63; each is flagged in ai_titles.csv so a reader can # recount without it. Matched on the normalised title and the published # outlet name, and the count is asserted. # The three Dutch titles among the variant matches whose "a.i." means ad # interim, read by hand off the title and the headline. AD_INTERIM_TITLES = { "managing editor a i", "picture editor art director a i", "visual editor a i", } BORDERLINE = [ ("research editor - special lists data and ai", "Barron's"), ("research editor (data and ai)", "The Wall Street Journal (Dow Jones)"), ("research editor, pep sco, data and ai, risk and compliance", "The Wall Street Journal (Dow Jones)"), ("ai transcript editor", "CalMatters"), ("video editor vfx ai video artist", "Sinclair Broadcast Group"), ] # ── desk-style editor titles ───────────────────────────────────────── # # True when the title is an editor title whose only qualifier is the AI # subject: "AI Editor", "Editor, AI", "Artificial Intelligence Editor", # "Senior Editor for AI", "Editor Overseeing AI Coverage". The test strips the # AI words, the word editor, a rank prefix and the joining words, and asks # whether anything is left. "Senior Editor, AI, Tech, and World" leaves tech # and world, so it is not one. _AI_PHRASE_RE = re.compile( r"\b(artificial intelligence|generative ai|machine learning|chatgpt" r"|openai|llm|a\.?\s?i\.?|ai)\b") _DESK_DROP = { "editor", "editors", "senior", "deputy", "associate", "executive", "managing", "global", "chief", "lead", "assistant", "principal", "acting", "for", "of", "on", "the", "and", "in", "at", "to", "overseeing", "coverage", "covering", "desk", } _PUNCT_RE = re.compile(r"[^a-z0-9 ]+") def title_names_a_desk(title): """The title is an editor title whose only qualifier is the AI subject.""" if not re.search(r"\beditor\b", title): return False rest = _AI_PHRASE_RE.sub(" ", title) tokens = [t for t in _PUNCT_RE.sub(" ", rest).split() if t] return not [t for t in tokens if t not in _DESK_DROP] # The named rows the plan asks to confirm: (outlet key, a string that must # appear in the normalised title). EXPECTED = [ ("BBC", "head of news ai"), ("Reuters", "ai development and integration"), ("mcclatchy.com", "ai innovation in journalism"), ("people.inc", "ai editorial operations"), ("miamiherald.com", "ai assisted reporter"), ] class _Tee: """The console output is kept beside the dataset so every published number has a line to point at.""" 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() # ── load ───────────────────────────────────────────────────────────── def load(): records = E.load_outlet_records() rows = [] read = dropped_platform = dropped_not_media = 0 path = os.path.join(ROOT, "data", "journalists.csv.gz") with gzip.open(path, "rt", newline="") as fh: for row in csv.DictReader(fh): read += 1 if row["platform_domain"] == "1": dropped_platform += 1 continue if row["media_typed"] != "1": dropped_not_media += 1 continue row["_key"] = row["outlet"] or row["company_domain"] row["_title"] = norm(row["title"]) row["_blob"] = row["_title"] + " || " + norm(row["headline"]) rows.append(row) rows_by_outlet = collections.Counter() domains_by_outlet = collections.defaultdict(set) domain_rows = collections.defaultdict(collections.Counter) company_names = collections.defaultdict(collections.Counter) country_of_outlet = collections.defaultdict(collections.Counter) for row in rows: key = row["_key"] rows_by_outlet[key] += 1 domains_by_outlet[key].add(row["company_domain"]) domain_rows[key][row["company_domain"]] += 1 if row["company_name"]: company_names[key][row["company_name"]] += 1 if row["country_code"]: country_of_outlet[key][row["country_code"]] += 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) entry = { "key": key, "rows": count, "domains": domains, "name": E.outlet_name(key, domains, recs, modal[0][0] if modal else ""), "country_code": E.outlet_country_code(recs) or E.fallback_country_code(country_of_outlet[key]), "type": E.outlet_type(recs), "parent": bool(set(domains) & E.PARENT_DOMAINS), # the outlet's own domain, the one most of its rows carry; a # group key can hold several, and the count is printed beside it "primary_domain": domain_rows[key].most_common(1)[0][0], "domain_count": len(domains), } entry["country"] = E.COUNTRIES.get(entry["country_code"], "") outlets[key] = entry def publishable(key): e = outlets[key] return bool(not e["parent"] and e["name"] and e["country"] and e["type"]) ranked = [k for k, _ in rows_by_outlet.most_common() if publishable(k)] return dict(rows=rows, read=read, dropped_platform=dropped_platform, dropped_not_media=dropped_not_media, outlets=outlets, rows_by_outlet=rows_by_outlet, top_keys=ranked[:TOP_N], ranked=ranked) def label(outlets, key): """The published outlet name, or the domain when there is none.""" entry = outlets.get(key) return (entry["name"] if entry and entry["name"] else key) def classify(title): if DEPLOY_RE.search(title): return "deploying" if COVER_RE.search(title): return "covering" return "unclear" def classify_rank_rule(title): if RANK_DEPLOY_RE.search(title): return "deploying" if COVER_RE.search(title): return "covering" return "unclear" # ── main ───────────────────────────────────────────────────────────── def main(): os.makedirs(OUT, exist_ok=True) sys.stdout = _Tee(os.path.join(OUT, "facts_console.txt")) D = load() rows, outlets = D["rows"], D["outlets"] base = len(rows) print("# 8. funnel") print("rows read from data/journalists.csv.gz: %d (expected %d)" % (D["read"], EXPECT_ROWS_READ)) 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 assert D["read"] == EXPECT_ROWS_READ, D["read"] ai_t = [r for r in rows if AI_RE.search(r["_title"])] ai_b = [r for r in rows if AI_RE.search(r["_blob"])] ai_t_scout = [r for r in rows if AI_RE_SCOUT.search(r["_title"])] ai_b_scout = [r for r in rows if AI_RE_SCOUT.search(r["_blob"])] outlets_t = {r["_key"] for r in ai_t} outlets_b = {r["_key"] for r in ai_b} print("AI title-only rows: %d of %d (%.3f%%), at %d outlets of %d" % (len(ai_t), base, 100.0 * len(ai_t) / base, len(outlets_t), len(D["rows_by_outlet"]))) print("AI headline-inclusive rows: %d of %d (%.3f%%), at %d outlets" % (len(ai_b), base, 100.0 * len(ai_b) / base, len(outlets_b))) print("scout lexicon (a\\.?i\\.? first alternative), title-only %d, " "headline-inclusive %d at %d outlets" % (len(ai_t_scout), len(ai_b_scout), len({r["_key"] for r in ai_b_scout}))) # ── 1. every title-only row, printed ───────────────────────────── print() print("# 1. the %d title-only AI rows, printed in full" % len(ai_t)) print("lexicon: %s" % AI_RE.pattern) ai_t.sort(key=lambda r: (label(outlets, r["_key"]).lower(), r["_title"])) for i, r in enumerate(ai_t, 1): r["_row_id"] = i r["_class"] = classify(r["_title"]) r["_class_rank_rule"] = classify_rank_rule(r["_title"]) print("%2d | %s | %s | %s | %s | %s" % (i, r["title"], label(outlets, r["_key"]), E.COUNTRIES.get(r["country_code"], "") or "(none)", r["role"], r["_class"])) # false positives, hand-read print() print("# 1b. hand-read false positives among the %d" % len(ai_t)) print("every row above was read; the checks were: AI as a personal name " "(Ai Weiwei), AI inside another acronym (AIB, AIM), and the Dutch " "a.i. that means ad interim.") print("rows removed: 0") print() print("# 1c. hand-read pass over the spelling 'a i' / 'a.i.' in the title") print("regex: %s" % AI_RE_SPACED.pattern) variant = [r for r in rows if AI_RE_SPACED.search(r["_title"])] in_63 = [r for r in variant if AI_RE.search(r["_title"])] spaced = [r for r in variant if not AI_RE.search(r["_title"])] ai_only = [r for r in ai_t if not AI_RE_SPACED.search(r["_title"])] print("titles matching the variant regex: %d" % len(variant)) print(" of those already in the %d published rows: %d" % (len(ai_t), len(in_63))) print(" new, not in the %d: %d" % (len(ai_t), len(spaced))) print("published rows the variant regex misses: %d (they spell it out as " "'artificial intelligence' and carry no bare 'ai' token)" % len(ai_only)) assert len(variant) == len(in_63) + len(spaced) assert len(ai_t) == len(in_63) + len(ai_only) print("the %d new titles:" % len(spaced)) for r in spaced: print(" NEW | %s | %s | %s | %s | headline: %s" % (r["title"], label(outlets, r["_key"]), E.COUNTRIES.get(r["country_code"], "") or "(none)", r["role"], (r["headline"] or "")[:90])) print("the %d published rows that match only on 'artificial " "intelligence':" % len(ai_only)) for r in ai_only: print(" AI-SPELLED-OUT | %s | %s | %s" % (r["title"], label(outlets, r["_key"]), E.COUNTRIES.get(r["country_code"], "") or "(none)")) print("a variant-inclusive rule, taken raw, gives %d rows at %d outlets" % (len(ai_t) + len(spaced), len({r["_key"] for r in ai_t} | {r["_key"] for r in spaced}))) interim = [r for r in spaced if r["_title"] in AD_INTERIM_TITLES] real = [r for r in spaced if r not in interim] print("read by hand, %d of the %d new titles are the Dutch a.i. that " "means ad interim (%s) and %d are A.I. (%s)" % (len(interim), len(spaced), "; ".join(label(outlets, r["_key"]) for r in interim), len(real), "; ".join(r["title"] for r in real))) print("so the hand-read variant-inclusive count is %d rows at %d outlets" % (len(ai_t) + len(real), len({r["_key"] for r in ai_t} | {r["_key"] for r in real}))) # ── 1d. the borderline rows ────────────────────────────────────── print() print("# 1d. borderline rows, flagged and kept") for r in ai_t: r["_borderline"] = any( r["_title"] == needle and label(outlets, r["_key"]) == outlet for needle, outlet in BORDERLINE) r["_desk_title"] = title_names_a_desk(r["_title"]) flagged = [r for r in ai_t if r["_borderline"]] assert len(flagged) == len(BORDERLINE), [r["title"] for r in flagged] print("flagged: %d of %d; the %d minus the flagged rows is %d" % (len(flagged), len(ai_t), len(ai_t), len(ai_t) - len(flagged))) for r in flagged: print(" BORDERLINE | %s | %s | %s" % (r["title"], label(outlets, r["_key"]), r["_class"])) print("no row is removed; the flag is a column in ai_titles.csv") # ── 1e. desk-style editor titles ───────────────────────────────── print() print("# 1e. titles that name a desk: an editor title whose only " "qualifier is the AI subject") desk_titles = [r for r in ai_t if r["_desk_title"]] print("title_names_a_desk: %d of %d" % (len(desk_titles), len(ai_t))) for r in desk_titles: print(" DESK | %s | %s | first rule %s | alt rule %s" % (r["title"], label(outlets, r["_key"]), r["_class"], r["_class_rank_rule"])) unclear_desk = [r for r in desk_titles if r["_class"] == "unclear"] print("of the %d unclear rows under the published rule, %d are desk-style " "editor titles" % (sum(1 for r in ai_t if r["_class"] == "unclear"), len(unclear_desk))) # ── 2. the desk ranking ────────────────────────────────────────── print() print("# 2. outlets by AI-titled rows") desk = collections.Counter(r["_key"] for r in ai_t) desk_b = collections.Counter(r["_key"] for r in ai_b) print("outlets with at least one AI title: %d" % len(desk)) multi = [(k, n) for k, n in desk.most_common() if n >= 2] print("outlets with two or more: %d" % len(multi)) for k, n in multi: print(" %-38s %d AI titles of %d editorial rows" % (label(outlets, k), n, D["rows_by_outlet"][k])) top_key, top_n = desk.most_common(1)[0] print("largest desk: %s, %d AI titles of %d editorial rows" % (label(outlets, top_key), top_n, D["rows_by_outlet"][top_key])) for r in ai_t: if r["_key"] == top_key: print(" %s | %s | %s | %s" % (r["title"], E.COUNTRIES.get(r["country_code"], ""), r["role"], r["_class"])) ones = sum(1 for k, n in desk.items() if n == 1) print("outlets with exactly one AI title: %d of %d" % (ones, len(desk))) # ── 3. deploy vs cover ─────────────────────────────────────────── print() print("# 3. deploying vs covering, over all %d title-only rows" % len(ai_t)) print("rule: deploy words in the title win over reporting words, because " "a title can carry both.") print("deploy words: %s" % DEPLOY_RE.pattern) print("cover words: %s" % COVER_RE.pattern) split = collections.Counter(r["_class"] for r in ai_t) for cls in ("deploying", "covering", "unclear"): print(" %-10s %d of %d" % (cls, split[cls], len(ai_t))) alt = collections.Counter(r["_class_rank_rule"] for r in ai_t) print("alternative rule (any of editor/head/director/lead/operations/" "innovation/development/integration/product/strategy = deploying):") for cls in ("deploying", "covering", "unclear"): print(" %-10s %d of %d" % (cls, alt[cls], len(ai_t))) disagree = [r for r in ai_t if r["_class"] != r["_class_rank_rule"]] print("rows where the two rules disagree: %d (row_id in ai_titles.csv: %s)" % (len(disagree), ", ".join(str(r["_row_id"]) for r in disagree))) for r in disagree: print(" row_id %2d | %s | %s | first rule %s | alt rule %s" % (r["_row_id"], r["title"], label(outlets, r["_key"]), r["_class"], r["_class_rank_rule"])) print() print("# 3b. the five named rows the plan asks for") for key, needle in EXPECTED: hit = [r for r in ai_t if r["_key"] == key and needle in r["_title"]] print(" %-34s %s" % (label(outlets, key), ("present: %s (%s)" % (hit[0]["title"], hit[0]["_class"])) if hit else "NOT FOUND (%s)" % needle)) # ── 4. the 50 largest newsrooms ────────────────────────────────── print() print("# 4. the %d largest publishable outlets by editorial rows" % TOP_N) print("the %d largest publishable outlets: media-typed, not a parent " "domain, named, typed, with a country. This is the same ranking " "that feeds the site's /outlets/ table; the site gives its own " "outlet page to the first 40 of it, so this set is 10 longer than " "the list of published outlet pages." % TOP_N) keys = D["top_keys"] zero_t = [k for k in keys if desk[k] == 0] zero_b = [k for k in keys if desk_b[k] == 0] for i, k in enumerate(keys, 1): print("%2d | %-38s | %-16s | %5d rows | %d AI titles | %d with headlines" % (i, label(outlets, k), outlets[k]["country"], D["rows_by_outlet"][k], desk[k], desk_b[k])) print("of the %d: %d have no AI title, %d have no AI word even in a " "headline" % (TOP_N, len(zero_t), len(zero_b))) print("zero AI titles: %s" % ", ".join(label(outlets, k) for k in zero_t)) print("zero even with headlines: %s" % ", ".join(label(outlets, k) for k in zero_b)) # the scout's own top-50, which did not exclude parent domains scout_keys = [k for k, _ in D["rows_by_outlet"].most_common(TOP_N)] scout_zero_t = [k for k in scout_keys if desk[k] == 0] scout_zero_b = [k for k in scout_keys if desk_b[k] == 0] print("scout's top 50 (parent domains not excluded, for comparison): " "%d with no AI title, %d with none even in a headline" % (len(scout_zero_t), len(scout_zero_b))) print(" scout zero even with headlines: %s" % ", ".join(label(outlets, k) for k in scout_zero_b)) # ── 5. comparators and countries ───────────────────────────────── print() print("# 5. comparator beats, title-only and headline-inclusive") comp_rows = [] for beat in ["ai"] + sorted(COMPARATORS): rx = AI_RE if beat == "ai" else COMPARATOR_RE[beat] t = [r for r in rows if rx.search(r["_title"])] b = [r for r in rows if rx.search(r["_blob"])] comp_rows.append({ "beat": beat, "title_only": len(t), "title_only_outlets": len({r["_key"] for r in t}), "headline_inclusive": len(b), "headline_inclusive_outlets": len({r["_key"] for r in b}), "lexicon": (AI_RE if beat == "ai" else COMPARATOR_RE[beat]).pattern, }) print(" %-14s title-only %6d of %d | with headlines %6d of %d " "| outlets %d / %d" % (beat, len(t), base, len(b), base, len({r["_key"] for r in t}), len({r["_key"] for r in b}))) print() print("# 5b. AI by country, countries with %d+ rows in the base" % COUNTRY_MIN_ROWS) rows_by_country = collections.Counter(r["country_code"] for r in rows) ai_t_country = collections.Counter(r["country_code"] for r in ai_t) ai_b_country = collections.Counter(r["country_code"] for r in ai_b) country_rows = [] for code, n in rows_by_country.most_common(): if n < COUNTRY_MIN_ROWS or not E.COUNTRIES.get(code): continue country_rows.append({ "country": E.COUNTRIES[code], "country_code": code, "editorial_rows": n, "ai_titles": ai_t_country[code], "ai_titles_per_1000": round(1000.0 * ai_t_country[code] / n, 2), "ai_headline_inclusive": ai_b_country[code], "ai_headline_inclusive_per_1000": round(1000.0 * ai_b_country[code] / n, 2), }) print(" %-22s %6d rows | %2d AI titles (%.2f per 1,000) | %3d with " "headlines (%.2f per 1,000)" % (E.COUNTRIES[code], n, ai_t_country[code], 1000.0 * ai_t_country[code] / n, ai_b_country[code], 1000.0 * ai_b_country[code] / n)) no_country = sum(1 for r in ai_t if not r["country_code"]) print("AI-titled rows with no country on the person row: %d of %d" % (no_country, len(ai_t))) # ── 6. seniority ───────────────────────────────────────────────── print() print("# 6. seniority of the AI-titled rows") senior = [r for r in ai_t if SENIOR_RE.search(r["_title"])] print("carry head, director, lead, chief or senior: %d of %d" % (len(senior), len(ai_t))) word_counts = collections.Counter() for r in ai_t: for word in ("head", "director", "lead", "chief", "senior"): if re.search(r"\b%s\b" % word, r["_title"]): word_counts[word] += 1 for word, n in word_counts.most_common(): print(" %-9s %d" % (word, n)) print(" per-word counts sum to %d over %d distinct rows: a title can " "carry two rank words, and one does. \"Head of News Ai (Senior " "News Editor, Ai)\" at the BBC is counted under head and under " "senior." % (sum(word_counts.values()), len(senior))) two = [r for r in senior if sum(1 for w in ("head", "director", "lead", "chief", "senior") if re.search(r"\b%s\b" % w, r["_title"])) > 1] print(" rows carrying more than one rank word: %d (%s)" % (len(two), "; ".join(r["title"] for r in two))) print("the classifier's own seniority column on the same rows: %s" % dict(collections.Counter(r["seniority"] for r in ai_t))) base_senior = sum(1 for r in rows if SENIOR_RE.search(r["_title"])) print("same word list over the whole base: %d of %d (%.1f%%)" % (base_senior, base, 100.0 * base_senior / base)) # ── 6b. when the rows were fetched ─────────────────────────────── print() print("# 6b. fetched_at on the %d AI-titled rows" % len(ai_t)) fetched = sorted((r["fetched_at"] or "")[:10] for r in ai_t) print("min %s | max %s | median %s" % (fetched[0], fetched[-1], statistics.median_low(fetched))) print("fetched_at is when the profile was scraped, not when the job " "started or ended.") # ── CSVs ───────────────────────────────────────────────────────── write_csv("ai_titles.csv", ["row_id", "title", "outlet", "outlet_domain", "country", "role", "class", "class_alt", "title_names_a_desk", "borderline", "fetched_at"], [[r["_row_id"], r["title"], label(outlets, r["_key"]), r["company_domain"], E.COUNTRIES.get(r["country_code"], ""), r["role"], r["_class"], r["_class_rank_rule"], "yes" if r["_desk_title"] else "no", "yes" if r["_borderline"] else "no", (r["fetched_at"] or "")[:10]] for r in ai_t]) desk_rows = [] for k, n in desk.most_common(): e = outlets[k] desk_rows.append([label(outlets, k), e["primary_domain"], e["domain_count"], e["country"], e["type"], n, desk_b[k], D["rows_by_outlet"][k], round(1000.0 * n / D["rows_by_outlet"][k], 2), "yes" if e["parent"] else "no"]) write_csv("ai_desk_ranking.csv", ["outlet", "outlet_domain", "outlet_domains_on_file", "country", "outlet_type", "ai_titles", "ai_headline_inclusive", "editorial_rows", "ai_titles_per_1000_rows", "parent_domain"], desk_rows) write_csv("top50_newsrooms_ai.csv", ["rank", "outlet", "outlet_domain", "outlet_domains_on_file", "country", "outlet_type", "editorial_rows", "ai_titles", "ai_headline_inclusive"], [[i, label(outlets, k), outlets[k]["primary_domain"], outlets[k]["domain_count"], outlets[k]["country"], outlets[k]["type"], D["rows_by_outlet"][k], desk[k], desk_b[k]] for i, k in enumerate(keys, 1)]) write_csv("ai_by_country.csv", ["country", "country_code", "editorial_rows", "ai_titles", "ai_titles_per_1000", "ai_headline_inclusive", "ai_headline_inclusive_per_1000"], [[c["country"], c["country_code"], c["editorial_rows"], c["ai_titles"], c["ai_titles_per_1000"], c["ai_headline_inclusive"], c["ai_headline_inclusive_per_1000"]] for c in country_rows]) write_csv("beat_title_counts.csv", ["beat", "title_only_rows", "title_only_outlets", "headline_inclusive_rows", "headline_inclusive_outlets", "base_rows", "lexicon"], [[c["beat"], c["title_only"], c["title_only_outlets"], c["headline_inclusive"], c["headline_inclusive_outlets"], base, c["lexicon"]] for c in comp_rows]) lex = os.path.join(OUT, "lexicon.txt") with open(lex, "w", encoding="utf-8") as fh: fh.write("Every regex this report uses. Matched against a lowercased " "string with & replaced by 'and' and whitespace collapsed.\n\n") fh.write("ai (published rule, title-only and headline-inclusive)\n %s\n\n" % AI_RE.pattern) fh.write("ai, the scout's looser variant, reported for comparison\n %s\n\n" % AI_RE_SCOUT.pattern) fh.write("ai, the hand-read 'a i' / 'a.i.' spelling pass\n %s\n\n" % AI_RE_SPACED.pattern) for beat in sorted(COMPARATORS): fh.write("%s\n %s\n\n" % (beat, COMPARATORS[beat])) fh.write("deploying (title words, checked first)\n %s\n\n" % DEPLOY_RE.pattern) fh.write("covering (title words)\n %s\n\n" % COVER_RE.pattern) fh.write("deploying, the alternative rank-word rule\n %s\n\n" % RANK_DEPLOY_RE.pattern) fh.write("seniority words\n %s\n" % SENIOR_RE.pattern) shutil.copyfile(os.path.abspath(__file__), os.path.join(OUT, "report_ai_desks.py")) print("wrote report_ai_desks.py (the script that made every file here)") print() print("wrote %s" % OUT) def write_csv(name, header, rows): write_csv_at(OUT, name, header, rows) def write_csv_at(folder, name, header, rows): with open(os.path.join(folder, name), "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh) w.writerow(header) w.writerows(rows) print("wrote %s (%d rows)" % (name, len(rows))) # ── 7. the outside check ───────────────────────────────────────────── SANITY_URLS = [ ("reuters-institute-2026", "https://reutersinstitute.politics.ox.ac.uk/" "journalism-media-and-technology-trends-and-" "predictions-2026"), ("reuters-institute-2025", "https://reutersinstitute.politics.ox.ac.uk/" "journalism-media-and-technology-trends-and-" "predictions-2025"), ("bbc-ai-principles", "https://www.bbc.co.uk/supplying/working-with-us/" "ai-principles/"), # tried and recorded, both 404 through the proxy on 2026-09-16 ("bbc-about-ai-404", "https://www.bbc.co.uk/aboutthebbc/whatwedo/" "artificial-intelligence"), ("bbc-about-ai-com-404", "https://www.bbc.com/aboutthebbc/whatwedo/" "artificial-intelligence"), ] # The outside figure the check turns on: the Reuters Institute's 2026 trends # report says Newsquest "has created more than 30 'AI-assisted reporters' # across its titles". Newsquest's own rows in the base are counted against it. # # The group domain newsquest.co.uk carries only 7 rows, and Newsquest's # mastheads each have their own domain, so counting the group domain alone # understates the group. The masthead list is built from Wikipedia's "List of # newspapers published by Newsquest", whose blue links are resolved to an # official website through Wikidata's P856, and matched against the base by # company_domain. SANITY_OUTLET = "newsquest" NEWSQUEST_LIST_URL = ("https://en.wikipedia.org/wiki/" "List_of_newspapers_published_by_Newsquest") WIKI_API = "https://en.wikipedia.org/w/api.php" WIKIDATA_API = "https://www.wikidata.org/w/api.php" # Hand-checked drops from the resolved websites: the list page links a few # places and councils beside its mastheads, and P856 on those is a council or # tourist-board site. Every .gov.uk host is dropped with them. NEWSQUEST_NOT_A_MASTHEAD = { "eu-digest.com", "visitpembrokeshire.com", "northwichtc.plus.com", } def newsquest_domains(fetch): """Newsquest masthead domains, from the Wikipedia list via Wikidata P856. Returns (domains, titles_on_list, blue_links, resolved, status_lines). """ import html as htmlmod import json import urllib.parse status = [] r = fetch(NEWSQUEST_LIST_URL, "journalistlabs_ai_sanity", timeout=45) status.append("%s HTTP %d" % (NEWSQUEST_LIST_URL, r.status_code)) linked, redlinks = [], 0 for m in re.finditer( r'rel="mw:WikiLink" href="https://en\.wikipedia\.org/wiki/' r'([^"#]+)"[^>]*title="([^"]+)"', r.text): title = htmlmod.unescape(m.group(2)) if ":" in title or title in ("Newspaper", "Newsquest"): continue if "(page does not exist)" in title: redlinks += 1 continue if title not in linked: linked.append(title) qids = {} for i in range(0, len(linked), 40): url = ("%s?action=query&format=json&prop=pageprops&ppprop=wikibase_item" "&redirects=1&titles=%s" % (WIKI_API, urllib.parse.quote("|".join(linked[i:i + 40])))) resp = fetch(url, "journalistlabs_ai_sanity", timeout=45) for page in resp.json()["query"]["pages"].values(): qid = (page.get("pageprops") or {}).get("wikibase_item") if qid: qids[page["title"]] = qid status.append("blue links on the list: %d (plus %d red links); " "with a Wikidata item: %d" % (len(linked), redlinks, len(qids))) sites = {} ids = sorted(set(qids.values())) for i in range(0, len(ids), 40): url = ("%s?action=wbgetentities&format=json&props=claims&ids=%s" % (WIKIDATA_API, "|".join(ids[i:i + 40]))) resp = fetch(url, "journalistlabs_ai_sanity", timeout=45) for qid, entity in resp.json()["entities"].items(): for claim in entity.get("claims", {}).get("P856", []): value = claim["mainsnak"].get("datavalue", {}).get("value") if value: sites.setdefault(qid, []).append(value) domains, by_title = set(), {} for title, qid in qids.items(): for url in sites.get(qid, []): host = re.sub(r"^https?://", "", url).split("/")[0].lower() host = host[4:] if host.startswith("www.") else host if host.endswith(".gov.uk") or host in NEWSQUEST_NOT_A_MASTHEAD: continue domains.add(host) by_title.setdefault(title, []).append(host) domains.add("newsquest.co.uk") status.append("titles with an official website on Wikidata: %d; " "distinct masthead domains after the hand-checked drops: %d" % (len(by_title), len(domains))) return sorted(domains), linked, redlinks, by_title, status def sanity(): """Fetched through the Webshare rotating gateway, never from this IP.""" import html as htmlmod sys.path.insert(0, os.path.join(os.path.dirname(ROOT), "ai_sending_tool_v2")) from proxy_fetch import fetch_response_with_fallback os.makedirs(OUT, exist_ok=True) sys.stdout = _Tee(os.path.join(OUT, "sanity_console.txt")) base_rows = [] 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" or row["media_typed"] != "1": continue base_rows.append(row) hits = [r for r in base_rows if SANITY_OUTLET in r["company_domain"].lower() or SANITY_OUTLET in (r["outlet"] or "").lower()] ai_hits = [r for r in hits if AI_RE.search(norm(r["title"]))] print("# the group domain, which is what the outlet key holds") print("editorial rows at newsquest.co.uk in the published base: %d" % len(hits)) print("of those with an AI word in the title: %d" % len(ai_hits)) print("the site publishes this key under the modal company name; a hand " "alias in names.OUTLET_NAMES now publishes it as Newsquest.") print("titles on file: %s" % "; ".join(sorted(r["title"] for r in hits))) print() print("# Newsquest's mastheads, which carry their own domains") try: domains, linked, redlinks, by_title, status = newsquest_domains( fetch_response_with_fallback) except Exception as exc: # noqa: BLE001 print("FAILED to build the masthead list: %s" % exc) domains, by_title, status = [], {}, [] for line in status: print(" " + line) if domains: print(" fetched through the proxy on %s" % datetime.date.today().isoformat()) nq_rows = [r for r in base_rows if (r["company_domain"] or "").lower() in set(domains)] per_domain = collections.Counter( (r["company_domain"] or "").lower() for r in nq_rows) nq_ai = [r for r in nq_rows if AI_RE.search(norm(r["title"]))] print("Newsquest domains on the list: %d" % len(domains)) print("of those with at least one editorial row in the base: %d" % len(per_domain)) print("editorial rows at those domains: %d" % len(nq_rows)) print("of those with an AI word in the title: %d" % len(nq_ai)) for dom, n in per_domain.most_common(): print(" %-34s %d rows" % (dom, n)) write_csv_at(OUT, "newsquest_domains.csv", ["domain", "wikipedia_title", "editorial_rows_in_base", "ai_titled_rows"], [[dom, "; ".join(sorted(t for t, ds in by_title.items() if dom in ds)) or ("(the group domain, added by hand)" if dom == "newsquest.co.uk" else ""), per_domain[dom], sum(1 for r in nq_ai if (r["company_domain"] or "").lower() == dom)] for dom in domains]) print() pat = re.compile(r"[^.]{0,200}(?:AI|artificial intelligence)[^.]{0,160}\.") for name, url in SANITY_URLS: try: r = fetch_response_with_fallback(url, "journalistlabs_ai_sanity", timeout=45) except Exception as exc: # noqa: BLE001 print("=== %s FAILED %s" % (name, exc)) print(" %s" % url) continue text = re.sub(r"|", "", r.text, flags=re.S) text = re.sub(r"\s+", " ", htmlmod.unescape(re.sub(r"<[^>]+>", " ", text))) print("=== %s HTTP %d %s" % (name, r.status_code, url)) shown = 0 for m in pat.findall(text): if re.search(r"\b(team|staff|roles?|journalists|people|editors|" r"newsroom|appointed|hired|newsquest|\d+%)\b", m, re.I): print(" " + m.strip()[:300]) shown += 1 if shown >= 8: break if __name__ == "__main__": if "--sanity" in sys.argv: sanity() else: main()