From 69e9d0d46226e0e4bec6ffe7657173050031d53c Mon Sep 17 00:00:00 2001 From: Michael Peter Christen Date: Sun, 20 Sep 2026 23:44:16 +0200 Subject: [PATCH] added mediawiki/wikipedia to YaCyPack extraction/production and document frequency statistic file generator for advanced ranking --- .gitignore | 7 + ir/mediawiki/README.md | 224 ++++++ ir/mediawiki/download_wikipedia_dump.sh | 87 +++ .../extract_wikipedia_dump_to_markdown.py | 641 ++++++++++++++++++ ir/mediawiki/markdown_to_term_frequency.py | 170 +++++ ir/mediawiki/requirements.txt | 7 + .../term_frequency_to_document_frequency.py | 187 +++++ ir/mediawiki/write_parquet.py | 98 +++ 8 files changed, 1421 insertions(+) create mode 100644 ir/mediawiki/README.md create mode 100755 ir/mediawiki/download_wikipedia_dump.sh create mode 100755 ir/mediawiki/extract_wikipedia_dump_to_markdown.py create mode 100755 ir/mediawiki/markdown_to_term_frequency.py create mode 100644 ir/mediawiki/requirements.txt create mode 100755 ir/mediawiki/term_frequency_to_document_frequency.py create mode 100755 ir/mediawiki/write_parquet.py diff --git a/.gitignore b/.gitignore index 82b8bcf2b..c5a656a35 100644 --- a/.gitignore +++ b/.gitignore @@ -37,3 +37,10 @@ yacy.log /lib vagrant_yacy/.vagrant/ .gradle +grafic_examples +/ir/mediawiki/_dumps/ +/ir/mediawiki/__pycache__/ +/ir/mediawiki/test_*.py +/ir/mediawiki/.venv/ +/ir/mediawiki/_markdown/ +/ir/mediawiki/_statistics/ diff --git a/ir/mediawiki/README.md b/ir/mediawiki/README.md new file mode 100644 index 000000000..639b15cd9 --- /dev/null +++ b/ir/mediawiki/README.md @@ -0,0 +1,224 @@ +# Wikipedia word statistics for YaCy + +How common is a word across articles? This small toolkit turns Wikipedia dumps +into a shared reference for that question: download an edition, extract its +articles, clean the text, and count the words. + +YaCy could use these statistics to give rare query terms more weight when +combining search results from different peers. The tools below work today; +publishing a dataset and using it in YaCy's ranking are still future work. +Wikipedia provides a useful reference corpus, but whether it improves web search +needs to be measured. + +## Get started + +Run these commands from `ir/mediawiki`. You need Python 3.9+, Bash, `curl`, +and `shasum`. Installed `lbzip2`/`bzip2` and `pigz`/`gzip` are used for streaming +decompression and compression; Python's standard library is the fallback. + +```sh +python3 -m venv .venv +source .venv/bin/activate +python3 -m pip install -r requirements.txt +``` + +Dependencies are pinned in [requirements.txt](requirements.txt). If pip builds +the wikitext parser from source, a C compiler is also needed. + +## From Wikipedia to statistics + +First, download an edition: + +```sh +./download_wikipedia_dump.sh de +``` + +Use `de` for German, `it` for Italian, or another Wikipedia language code. +The downloader selects the latest completed +[Wikimedia current-content export](https://wikitech.wikimedia.org/wiki/MediaWiki_Content_File_Exports) +and checks its SHA-256 checksums. Large exports arrive in several parts; all +parts belong to the same snapshot and are needed. +Rerunning the command skips files whose checksums match, resumes `.part` downloads, +and downloads completed files again only if their checksums fail. + +Then use the date in the downloaded filenames: + +```sh +python3 extract_wikipedia_dump_to_markdown.py de 2026-09-01 +python3 markdown_to_term_frequency.py de 2026-09-01 +python3 term_frequency_to_document_frequency.py de 2026-09-01 +``` + +The direct converter reads every dump part and writes one Markdown YaCy Pack, +without creating wikitext files. A reader streams page records to worker processes; +one writer collects the results. Bounded queues limit memory use, and completion +signals drain all pending work before the final pack is published. + +Shared helpers and the Markdown renderer live inside +`extract_wikipedia_dump_to_markdown.py`; the statistics scripts import its helpers. +There is no separate wikitext extraction step. + +Each step leaves its results available to inspect: + +| Location | Contents | +| --- | --- | +| `_dumps/` | Compressed dump parts and checksums, stored directly in this folder | +| `_markdown/` | One gzip-compressed YaCy Pack per language and snapshot | +| `_statistics/dewiki-2026-09-01-term-frequency.jsonl` | Word counts within each article | +| `_statistics/dewiki-2026-09-01-document-frequency.csv` | Number of articles containing each retained word | + +Files stay directly in these folders. Dump filenames include their +page range; the Markdown pack combines all parts of the selected language/date: + +```text +_dumps/dewiki-2026-09-01-p3442045p7820211.xml.bz2 +_markdown/YaCy-Pack_scroll-common-web_wikipedia-de_20260901.jsonl.gz +``` + +The Markdown output is a gzip-compressed YaCy Pack, matching the existing packs: +Elasticsearch Bulk action/document pairs using YaCy's field names. `title` is a +one-element array; the other fields are strings. `last_modified` comes from the +dump's `` and is omitted when absent. + +```jsonl +{"index":{}} +{"url_s":"https://de.wikipedia.org/wiki/YaCy","title":["YaCy"],"text_t":"# YaCy\n\nEine Suchmaschine …\n","language_s":"de","last_modified":"2026-08-01T08:49:22Z"} +``` + +No `collection_sxt` or Wikipedia page ID is written. YaCy derives its own document +ID from `url_s`. + +Pack names follow `YaCy-Pack_--__.jsonl.gz`. +We use `scroll` (encyclopedia), `web`, and the generator's default tier `common`. +Use `--tier uncommon`, `rare`, `epic`, or `legendary` to choose another tier; +the tier is not calculated from file size. The slug is `wikipedia-`; +the date is the dump snapshot date for reproducible names. +Keep only one tier per language/date when computing statistics. + +The converter streams all dump parts into one compressed pack: after interruption, +the entire unfinished `.part` pack restarts. Completed packs are skipped. +Use `--overwrite` to regenerate a pack, including older packs without revision +timestamps. The dump date is never substituted for a missing revision timestamp. +The final pack is replaced only after successful conversion. + +Dump discovery rejects unfinished downloads, invalid or overlapping part ranges, +and missing or extra parts relative to the downloaded checksum manifest, when present. +The downloader verifies checksums; extraction does not hash the dump files again. +Gaps between page IDs are valid. Without a manifest, missing parts cannot reliably +be detected. +Statistics require exactly one completed combined pack for the language/date; +old packs with page ranges are ignored and are not deleted. + +Both statistics stages refuse existing outputs and unfinished `.part` files. +Move these aside before rerunning. Regenerating a Markdown pack does not regenerate +its statistics automatically. + +YaCy can import these packs through `DATA/PACKS/load/`. For Elasticsearch, decompress +and submit complete action/document pairs to `//_bulk` as NDJSON, in batches +within the server's request-size limit. With the requested empty `index` action, +Elasticsearch assigns IDs, so re-uploading creates additional documents. +See the [Bulk API documentation](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk). + +Markdown conversion and both statistics stages use all available CPU cores by default. +Add `--jobs 4` to limit concurrency. Every script shows usage +with no arguments, `-h`, or `--help`. +Term counting streams the pack through a bounded queue; workers decode JSON and +count each article independently. Completed counts are written continuously, +without keeping the whole corpus in memory. +TF-to-DF also decodes and counts batches in parallel, then merges their counters +in one process. Its final vocabulary stays in RAM; `--jobs 1` runs it sequentially. + +### Which articles and words are included? + +Extraction includes namespace `0` pages with available wikitext, including +redirects, disambiguation pages, and empty articles. Talk pages, user pages, +templates, categories, and other namespaces are skipped. The progress display +therefore shows more pages read than articles written. + +Titles are stored as UTF-8 data rather than filenames, so filesystem restrictions +and filename collisions do not affect article titles. + +Markdown keeps prose, headings, emphasis, lists, table text, and TeX formulas. +Every Markdown record's `text_t` starts with the original article title as a `#` heading. +It removes reference tags, URL targets, file/category embeds, recognized navigation +containers, and decorative markup. Localized file/category namespace names are +read from dump metadata. Link labels are retained. Common reference-section titles +in English, German, French, Italian, and Spanish are omitted; other languages and +unrecognized section names may retain that material. +Only a small set of presentation templates (`lang`, `nowrap`, `nobr`, `small`, +`smaller`, `big`, `larger`, `math`, `mvar`, `var`, `abbr`) is supported. Other +templates are omitted, so some template-generated content is lost. Templates are +not fetched or expanded through MediaWiki; the original dumps remain available +for comparison. + +Counting reads only the Pack document records, excluding the `index` action lines +and metadata. It lowercases `text_t` and treats sequences of Unicode letters, +numbers, and attached combining marks as words. Punctuation, apostrophes, +hyphens, and underscores separate words. The token pattern is +`[\p{L}\p{N}][\p{L}\p{M}\p{N}]*`. +There is no stemming, stopword removal, accent removal, or Unicode normalization. +Article titles, section headings, numbers, and TeX command names also contribute tokens. +Languages without word spaces may need a more suitable tokenizer before these +counts are used for language-specific evaluation. + +## Reading the results + +**Term frequency** counts occurrences inside one article. **Document frequency** +counts articles containing the term, once per article regardless of repetition. + +For example, the term-frequency JSONL could contain: + +```jsonl +{"Solar":{"solar":2,"energy":1}} +{"Wind":{"wind":1,"energy":1}} +{"Battery":{"solar":1,"battery":1,"energy":1}} +{"Empty":{}} +``` + +With the current filtering thresholds, its document-frequency CSV is: + +```text +term;count +_;4 +``` + +Here, `solar` occurs three times across two articles, so its document frequency +is `2`. All four terms are omitted from the CSV because their counts are below 10. +The reserved `_` row records all four documents, including the empty one. +Rows sort by document frequency from lowest to highest, alphabetically for ties, +with the `_` row last. + +The generated CSV files start with `term;count` and use UTF-8, semicolons, +integer counts, LF line endings, and no byte-order mark. +A nonnumeric term is retained when **its document frequency is at least 10 AND it +sorts at or after `aa`, OR its document frequency exceeds 100** (Python's Unicode +lexicographic order). For example, `aa` with count 9 and `a` with count 100 are +omitted; `aa` with count 10 and `a` with count 101 are retained. +Full occurrence counts remain in the JSONL. Evaluation should use a count of +**1 for any missing term**. No document-frequency JSON is generated. + +The TF-to-DF step also excludes terms consisting only of decimal digits after +removing dots and commas, including `42`, `3.14`, `3,14`, and `1.234,56`. +Unicode decimal digits are included in this rule; mixed terms such as `3d` are +subject to the count-and-order filter above. +The term-frequency JSONL and total document count are unaffected. + +JSONL article keys retain their original titles and case. Word keys are lowercase. +Record order can vary between runs; the counts are independent of worker order. + +## Export document frequencies as Parquet + +```sh +python3 write_parquet.py de 2026-09-01 +``` + +This reads the document-frequency CSV and writes +`_statistics/dewiki-2026-09-01-document-frequency.parquet` using PyArrow. +The Parquet rows are sorted by term; the CSV keeps its count ordering. +Both columns (`term`: UTF-8 string, `count`: 64-bit integer) and the `_` row are +preserved. Parquet uses prefix encoding for terms and ZSTD compression at level 22. +Use `--compression-level 19` for faster writing at a lower compression level. + +The converter loads and sorts the full table in RAM, including extra sorting +workspace. It writes to `.part` and publishes the final filename only on success. +Existing outputs are refused; move them aside before rerunning. diff --git a/ir/mediawiki/download_wikipedia_dump.sh b/ir/mediawiki/download_wikipedia_dump.sh new file mode 100755 index 000000000..174fbdd78 --- /dev/null +++ b/ir/mediawiki/download_wikipedia_dump.sh @@ -0,0 +1,87 @@ +#!/usr/bin/env bash +set -euo pipefail + +usage() { + cat <<'EOF' +Usage: download_wikipedia_dump.sh LANGUAGE + +Download the latest completed Wikipedia current-content dump for a language. + +Examples: + ./download_wikipedia_dump.sh de German Wikipedia + ./download_wikipedia_dump.sh en English Wikipedia + ./download_wikipedia_dump.sh simple Simple English Wikipedia + +Files stay compressed directly in _dumps/ next to this script. +Requires curl and shasum. Reruns skip verified files and resume partial downloads. +No arguments, -h, and --help all show this help. +EOF +} + +if [[ $# -eq 0 ]] || [[ $# -eq 1 && ( $1 == -h || $1 == --help ) ]]; then + usage + exit 0 +fi +if [[ $# -ne 1 ]] || [[ ! $1 =~ ^[a-z][a-z0-9]*([-_][a-z0-9]+)*$ ]]; then + echo "Error: supply one lowercase Wikipedia language code, such as de or en." >&2 + usage >&2 + exit 1 +fi +for command in curl shasum; do + command -v "$command" >/dev/null || { echo "Error: $command is required." >&2; exit 1; } +done + +wiki="${1//-/_}wiki" +base="https://dumps.wikimedia.org/other/mediawiki_content_current/$wiki" +dump_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)/_dumps" +mkdir -p "$dump_dir" +manifest=$(mktemp "$dump_dir/.SHA256SUMS.XXXXXX") +trap 'rm -f "$manifest"' EXIT + +# Wikimedia publishes SHA256SUMS only when an export is complete. +# https://wikitech.wikimedia.org/wiki/MediaWiki_Content_File_Exports +dates=$(curl --fail --silent --show-error --location "$base/" | + sed -n 's/.*href="\([0-9]\{4\}-[0-9]\{2\}-[0-9]\{2\}\)\/".*/\1/p' | + sort -ru) +export_url= +for date in $dates; do + candidate="$base/$date/xml/bzip2" + status=$(curl --silent --show-error --location --output "$manifest" \ + --write-out '%{http_code}' "$candidate/SHA256SUMS") + case "$status" in + 200) export_url="$candidate"; break ;; + 404) continue ;; + *) echo "Error: HTTP $status fetching $candidate/SHA256SUMS" >&2; exit 1 ;; + esac +done +if [[ -z "$export_url" || ! -s "$manifest" ]]; then + echo "Error: no completed dump found for $wiki." >&2 + exit 1 +fi + +checksum_file="$wiki-$date-SHA256SUMS" +cp "$manifest" "$dump_dir/$checksum_file" +cd "$dump_dir" +echo "Downloading $wiki ($date) to $dump_dir" +while read -r checksum filename; do + [[ -n "$filename" ]] || continue + # Export filenames are plain basenames, not paths. + if [[ ! $filename =~ ^[a-zA-Z0-9_-]+\.xml\.bz2$ ]]; then + echo "Error: unexpected dump filename: $filename" >&2 + exit 1 + fi + if [[ -f "$filename" ]]; then + echo "Checking $filename" + if printf '%s %s\n' "$checksum" "$filename" | shasum -a 256 --check; then + echo "Already downloaded; skipping $filename" + continue + fi + echo "Checksum mismatch; downloading $filename again" + fi + echo "Downloading $filename" + curl --fail --location --show-error --continue-at - \ + --output "$filename.part" "$export_url/$filename" + printf '%s %s\n' "$checksum" "$filename.part" | shasum -a 256 --check + mv "$filename.part" "$filename" +done < "$checksum_file" +echo "Download complete: $dump_dir" diff --git a/ir/mediawiki/extract_wikipedia_dump_to_markdown.py b/ir/mediawiki/extract_wikipedia_dump_to_markdown.py new file mode 100755 index 000000000..666d57a99 --- /dev/null +++ b/ir/mediawiki/extract_wikipedia_dump_to_markdown.py @@ -0,0 +1,641 @@ +#!/usr/bin/env python3 +"""Stream XML dump pages through parallel workers directly into one YaCy Pack.""" + +import sys +import json +import time +import queue +import argparse +import threading +import multiprocessing +import xml.etree.ElementTree as ET +from xml.parsers import expat +from xml.sax.saxutils import quoteattr +from collections import Counter +from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait + +import bz2 +import gzip +import io +import os +import re +import shutil +import subprocess +import tempfile +from contextlib import contextmanager +from datetime import date, datetime +from pathlib import Path +from urllib.parse import quote + +try: + import mwparserfromhell as mw +except ImportError: + mw = None + +BASE_DIR = Path(__file__).resolve().parent +PACK_TIERS = ('common', 'uncommon', 'rare', 'epic', 'legendary') + + +def available_cores(): + if hasattr(os, 'process_cpu_count'): + return os.process_cpu_count() or 1 + if hasattr(os, 'sched_getaffinity'): + return len(os.sched_getaffinity(0)) or 1 + return os.cpu_count() or 1 + + +def language_code(value): + if not re.fullmatch(r'[a-z][a-z0-9]*(?:[-_][a-z0-9]+)*', value): + raise argparse.ArgumentTypeError('expected a lowercase Wikipedia language code') + return value.replace('_', '-') + + +def dump_date(value): + try: + if not re.fullmatch(r'\d{4}-\d{2}-\d{2}', value): + raise ValueError() + date.fromisoformat(value) + except ValueError: + raise argparse.ArgumentTypeError('expected a valid snapshot date, YYYY-MM-DD') from None + return value + + +def revision_timestamp(value): + try: + if not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z', value): + raise ValueError() + datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ') + except ValueError: + raise ValueError(f'Invalid revision timestamp: {value!r}') from None + return value + + +def snapshot_prefix(language, snapshot): + return f"{language_code(language).replace('-', '_')}wiki-{dump_date(snapshot)}" + + +def pack_path(folder, language, snapshot, tier='common'): + if tier not in PACK_TIERS: + raise ValueError(f'Unknown pack tier: {tier}') + return Path(folder) / (f'YaCy-Pack_scroll-{tier}-web_wikipedia-{language_code(language)}_' + f'{dump_date(snapshot).replace("-", "")}.jsonl.gz') + + +def find_parts(folder, language, snapshot): + folder = Path(folder) + prefix = snapshot_prefix(language, snapshot) + pattern = re.compile(re.escape(prefix) + r'-p(\d+)p(\d+)\.xml\.bz2') + parts = {p.name: p for p in folder.glob(f'{prefix}-*.xml.bz2') + if pattern.fullmatch(p.name) and p.is_file()} + if list(folder.glob(f'{prefix}-*.xml.bz2.part')): + raise ValueError(f'Unfinished dump downloads for {prefix}') + manifest = folder / f'{prefix}-SHA256SUMS' + if manifest.is_file(): + expected = set() + for line in manifest.read_text(encoding='utf-8').splitlines(): + if not line.strip(): + continue + fields = line.split() + if (len(fields) != 2 or not re.fullmatch(r'[a-fA-F0-9]{64}', fields[0]) + or not pattern.fullmatch(fields[1])): + raise ValueError(f'Invalid checksum manifest entry: {line!r}') + if fields[1] in expected: + raise ValueError(f'Duplicate dump part in {manifest}: {fields[1]}') + expected.add(fields[1]) + if not expected or expected != set(parts): + raise ValueError(f'Dump parts do not match {manifest}; rerun the downloader') + if not parts: + raise ValueError(f'No dump parts for {prefix} in {folder}') + ordered = sorted(parts.values(), key=lambda p: int(pattern.fullmatch(p.name)[1])) + previous_end = None + for part in ordered: + start, end = map(int, pattern.fullmatch(part.name).groups()) + # Deleted/nonexistent page IDs can leave gaps even in a complete export. + if start > end or (previous_end is not None and start <= previous_end): + raise ValueError(f'Invalid or overlapping dump ranges at {part.name}') + previous_end = end + return ordered + + +def find_containers(base, language, snapshot, folder='_markdown'): + directory = Path(base) / folder + legacy_marker = directory / f'{snapshot_prefix(language, snapshot)}.incomplete' + if legacy_marker.exists(): + raise ValueError(f'Incomplete Markdown input: {legacy_marker}') + parts = [] + for tier in PACK_TIERS: + target = pack_path(directory, language, snapshot, tier) + for suffix in ('.part', '.incomplete'): + if target.with_name(target.name + suffix).exists(): + raise ValueError(f'Incomplete Markdown input: {target}') + if target.is_file(): + parts.append(target) + if len(parts) != 1: + raise ValueError(f'Expected exactly one combined Markdown pack for ' + f'{language}/{snapshot}, found {len(parts)}') + return parts + + +@contextmanager +def decompressed(path): + """Yield a binary stream; check subprocess exit codes and gzip/bzip2 checksums.""" + path = Path(path) + names = ('lbzip2', 'bzip2') if path.suffix == '.bz2' else ('pigz', 'gzip') + executable = next((found for name in names if (found := shutil.which(name))), None) + if path.suffix not in ('.bz2', '.gz'): + with path.open('rb') as stream: + yield stream + elif executable: + with tempfile.TemporaryFile() as errors: + process = subprocess.Popen([executable, '-dc', str(path)], + stdout=subprocess.PIPE, stderr=errors) + try: + yield process.stdout + process.stdout.close() + if process.wait(): + errors.seek(0) + raise ValueError(f'Cannot decompress {path}: ' + f'{errors.read().decode("utf-8", "replace").strip()}') + finally: + process.stdout.close() + if process.poll() is None: + process.terminate() + process.wait() + else: + opener = bz2.open if path.suffix == '.bz2' else gzip.open + with opener(path, 'rb') as stream: + yield stream + + +@contextmanager +def compressed_output(target): + """Restart an unfinished pack; atomically replace the final file on success only.""" + target = Path(target) + partial = target.with_name(target.name + '.part') + executable = shutil.which('pigz') or shutil.which('gzip') + if executable: + with partial.open('wb') as destination, tempfile.TemporaryFile() as errors: + process = subprocess.Popen([executable, '-c'], stdin=subprocess.PIPE, + stdout=destination, stderr=errors) + stream = io.TextIOWrapper(process.stdin, encoding='utf-8', newline='\n') + try: + yield stream + stream.close() + if process.wait(): + errors.seek(0) + raise ValueError(f'Cannot compress {target}: ' + f'{errors.read().decode("utf-8", "replace").strip()}') + finally: + if process.poll() is None: + process.terminate() + try: + stream.close() + except BrokenPipeError: + pass + process.wait() + else: + with gzip.open(partial, 'wt', encoding='utf-8', newline='\n') as stream: + yield stream + partial.replace(target) + + +def render_wikitext(text, hidden_namespaces=()): + """Render useful article text without fetching templates or external resources.""" + if mw is None: + raise ValueError('Install dependencies with: python3 -m pip install -r requirements.txt') + hidden = {'file', 'image', 'media', 'category'} | { + name.replace('_', ' ').strip().casefold() for name in hidden_namespaces} + reference_sections = { + 'references', 'notes', 'footnotes', 'external links', 'bibliography', + 'literature', 'sources', 'further reading', 'see also', + 'einzelnachweise', 'anmerkungen', 'weblinks', 'literatur', 'quellen', 'siehe auch', + 'références', 'notes et références', 'liens externes', 'bibliographie', 'voir aussi', + 'note', 'bibliografia', 'collegamenti esterni', 'voci correlate', + 'referencias', 'enlaces externos', 'bibliografía', 'véase también', + } + + def render(code, sections=False): + output, skipped_level = [], None + for node in code.nodes: + if isinstance(node, mw.nodes.Heading): + title = render(node.title).strip() + if sections: + if skipped_level is not None and node.level > skipped_level: + continue + skipped_level = None + if title.casefold() in reference_sections: + skipped_level = node.level + continue + output.append('\n\n' + '#' * max(2, node.level) + ' ' + title + '\n\n') + continue + if skipped_level is not None: + continue + if isinstance(node, mw.nodes.Text): + output.append(str(node)) + elif isinstance(node, mw.nodes.HTMLEntity): + output.append(node.normalize()) + elif isinstance(node, mw.nodes.Wikilink): + target = str(node.title).strip() + prefix = target.split(':', 1)[0].replace('_', ' ').casefold() + if ':' in target and not target.startswith(':') and prefix in hidden: + continue + output.append(render(node.text) if node.text is not None + else target.lstrip(':').replace('_', ' ')) + elif isinstance(node, mw.nodes.ExternalLink): + output.append(render(node.title) if node.title is not None else '') + elif isinstance(node, mw.nodes.Argument): + output.append(render(node.default) if node.default is not None else '') + elif isinstance(node, mw.nodes.Template): + name = str(node.name).replace('_', ' ').strip().casefold() + # A deliberately small set of presentation-only templates. + parameter = '2' if name == 'lang' else '1' + if name in {'lang', 'nowrap', 'nobr', 'small', 'smaller', 'big', 'larger', + 'math', 'mvar', 'var', 'abbr'} and node.has(parameter): + value = render(node.get(parameter).value) + output.append('$' + value + '$' if name in {'math', 'mvar'} else value) + else: + output.append(' ') + elif isinstance(node, mw.nodes.Tag): + tag = str(node.tag).strip().casefold() + classes = str(node.get('class').value).casefold().split() if node.has('class') else [] + if (tag in {'ref', 'references', 'gallery', 'imagemap', 'timeline', 'score', + 'script', 'style', 'img', 'noinclude'} + or any(c in {'navbox', 'vertical-navbox', 'metadata', 'infobox', + 'toc', 'noprint', 'mw-editsection'} for c in classes)): + continue + if tag == 'math': + output.append('$' + str(node.contents).strip() + '$') + continue + if tag in {'nowiki', 'pre', 'syntaxhighlight', 'source', 'code'}: + output.append(str(node.contents)) + continue + if node.wiki_markup in {'*', '#', ';', ':'}: + output.append('\x00' + node.wiki_markup) + continue + value = render(node.contents) + if tag in {'b', 'strong', 'i', 'em'}: + marker = '**' if tag in {'b', 'strong'} else '*' + output.append(marker + value + marker if value else '') + elif tag in {'br', 'hr'}: + output.append('\n') + elif tag in {'td', 'th'}: + if str(node.wiki_markup) == '|' and value.startswith('+'): + value = value[1:] + output.append(value.strip() + ' | ') + elif tag == 'li': + output.append('\n- ' + value.strip() + '\n') + elif tag in {'table', 'tr', 'caption', 'p', 'div', 'ul', 'ol', 'dl', 'dt', 'dd'}: + output.append('\n' + value.strip() + '\n') + else: + output.append(value) + # Comments and unknown node types contribute no text. + return ''.join(output) + + rendered = render(mw.parse(text), sections=True) + + def list_prefix(match): + marks = match[1].replace('\x00', '') + marker = '1. ' if marks[-1] == '#' else '- ' + return ' ' * (len(marks) - 1) + marker + + rendered = re.sub(r'(?m)^(\x00[*#;:](?:\x00[*#;:])*)[ \t]*', list_prefix, rendered) + rendered = rendered.replace('\x00', '') + rendered = re.sub(r'__[A-Z][A-Z0-9_]*__', '', rendered) + rendered = re.sub(r'[ \t]+\n', '\n', rendered) + return re.sub(r'\n{3,}', '\n\n', rendered).strip() + + +def convert_batch(articles, language): + """Return newline-terminated Bulk action/document pairs for one worker batch.""" + language = language_code(language) + lines = [] + for article in articles: + title = article['title'] + body = render_wikitext(article['text'], article.get('hidden_namespaces', ())) + document = { + 'url_s': f'https://{language}.wikipedia.org/wiki/' + quote(title.replace(' ', '_'), safe='/:'), + 'title': [title], + 'text_t': '# ' + title + '\n' + ('\n' + body + '\n' if body else ''), + 'language_s': language, + } + if article.get('last_modified'): + document['last_modified'] = revision_timestamp(article['last_modified']) + lines.extend(('{"index":{}}\n', json.dumps(document, ensure_ascii=False, separators=(',', ':')) + '\n')) + return lines + + +class Stopped(Exception): + """Unblock queue operations when another stage fails.""" + + +def put(items, value, stopped): + while not stopped.is_set(): + try: + items.put(value, timeout=0.1) + return + except queue.Full: + pass + raise Stopped() + + +def get(items, stopped): + while not stopped.is_set(): + try: + return items.get(timeout=0.1) + except queue.Empty: + pass + raise Stopped() + + +def frame_pages(stream, emit, stopped, chunk_size=1024 * 1024): + """Validate XML boundaries without building trees or interpreting articles.""" + parser = expat.ParserCreate(encoding='UTF-8') + buffer = bytearray() + base, depth, page_start, number = 0, 0, None, 0 + opening, closing, siteinfo = b'', b'', b'' + + def start(name, attributes): + nonlocal depth, page_start, opening, closing + if depth == 0: + if name.split(':')[-1] != 'mediawiki': + raise ValueError('Expected a MediaWiki XML export') + # Carry inherited namespaces into each standalone page fragment. + attrs = ''.join(f' {key}={quoteattr(value)}' for key, value in attributes.items()) + opening = f'<{name}{attrs}>'.encode('utf-8') + closing = f''.encode('utf-8') + elif depth == 1 and name.split(':')[-1] in {'page', 'siteinfo'}: + page_start = parser.CurrentByteIndex + depth += 1 + + def end(name): + nonlocal depth, page_start, number, siteinfo + depth -= 1 + if depth == 1 and page_start is not None: + position = parser.CurrentByteIndex - base + # For , Expat reports the position after the whole tag. + end_tag = ('', position) + 1 + raw = bytes(buffer[page_start - base:position]) + if name.split(':')[-1] == 'siteinfo': + metadata = ET.fromstring(opening + raw + closing)[0] + # Keep only namespace metadata needed to suppress localized file/category links. + for child in list(metadata): + if child.tag.rsplit('}', 1)[-1] not in {'namespaces', 'namespacealiases'}: + metadata.remove(child) + siteinfo = ET.tostring(metadata, encoding='utf-8') + else: + number += 1 + emit(number, opening + siteinfo + raw + closing) + page_start = None + + def reject_doctype(*args): + raise ValueError('MediaWiki dumps with DTD declarations are not supported') + + parser.StartElementHandler = start + parser.EndElementHandler = end + parser.StartDoctypeDeclHandler = reject_doctype + while True: + if stopped.is_set(): + raise Stopped() + chunk = stream.read(chunk_size) + buffer.extend(chunk) + parser.Parse(chunk, not chunk) + if not chunk: + break + keep = page_start if page_start is not None else parser.CurrentByteIndex + del buffer[:keep - base] + base = keep + return number + + +def article_record(data, counts): + """Select namespace-zero wikitext and preserve the current revision timestamp.""" + root = ET.fromstring(data) + namespace = root.tag.removesuffix('mediawiki') + page = root.find(namespace + 'page') + if page is None: + raise ValueError('Expected a page in the MediaWiki export namespace') + q = lambda name: namespace + name + counts['pages'] += 1 + page_namespace = page.findtext(q('ns')) + if page_namespace is None: + raise ValueError('Page is missing its namespace') + if page_namespace != '0': + counts['other namespaces'] += 1 + return None + title, page_id = page.findtext(q('title')), page.findtext(q('id')) + if not title or not page_id or not page_id.isascii() or not page_id.isdecimal(): + raise ValueError('Article is missing a title or numeric page ID') + revisions = page.findall(q('revision')) + if len(revisions) != 1: + raise ValueError(f'{title!r}: expected one current revision, got {len(revisions)}') + content = revisions[0] + slots = content.find(q('slots')) + if slots is not None: + main = [slot for slot in slots if slot.get('role') == 'main'] + if len(main) != 1: + raise ValueError(f'{title!r}: expected one main content slot') + content = main[0] + if content.findtext(q('model')) not in (None, 'wikitext'): + counts['other content models'] += 1 + return None + text = content.find(q('text')) + if text is None or 'deleted' in text.attrib: + counts['unavailable text'] += 1 + return None + if len(text): + raise ValueError(f'{title!r}: unexpected XML elements inside article text') + record = {'id': int(page_id), 'title': title, 'text': text.text or ''} + siteinfo = root.find(q('siteinfo')) + if siteinfo is not None: + record['hidden_namespaces'] = [element.text for element in siteinfo.iter() + if element.get('key', element.get('id')) in {'6', '14'} + and element.text] + timestamp = revisions[0].findtext(q('timestamp')) + if timestamp is not None: + record['last_modified'] = revision_timestamp(timestamp) + counts['articles'] += 1 + return record + + +def convert_pages(batch, language): + counts, lines = Counter(), [] + for source, number, data in batch: + try: + article = article_record(data, counts) + if article is not None: + lines.extend(convert_batch([article], language)) + except Exception as error: + raise ValueError(f'{source}, page {number}: {error}') from error + return counts, ''.join(lines) + + +def read_dumps(parts, inputs, stopped): + batch, byte_count = [], 0 + + def emit(number, data): + nonlocal batch, byte_count + batch.append((part.name, number, data)) + byte_count += len(data) + if len(batch) >= 32 or byte_count >= 262144: + put(inputs, batch, stopped) + batch, byte_count = [], 0 + + for index, part in enumerate(parts, 1): + print(f'[{index}/{len(parts)}] Reading {part.name}', flush=True) + try: + with decompressed(part) as stream: + frame_pages(stream, emit, stopped) + except (expat.ExpatError, ValueError) as error: + raise ValueError(f'{part.name}: {error}') from error + if batch: + put(inputs, batch, stopped) + # FIFO: all batches precede the reader's poison pill. + put(inputs, None, stopped) + + +def write_pack(target, outputs, stopped, counts, legacy_markers): + last_report = 0 + started = time.monotonic() + with compressed_output(target) as stream: + for marker in legacy_markers: + marker.unlink(missing_ok=True) + while True: + result = get(outputs, stopped) + if result is None: + break + batch_counts, lines = result + stream.write(lines) + counts.update(batch_counts) + if counts['pages'] - last_report >= 10000: + rate = counts['articles'] / max(time.monotonic() - started, 1e-9) + print(f" {counts['pages']:,} pages processed; {counts['articles']:,} articles written " + f'({rate:.0f}/s)', + flush=True) + last_report = counts['pages'] + + +def run_pipeline(parts, target, language, jobs, legacy_markers=()): + stopped = threading.Event() + inputs, outputs = queue.Queue(maxsize=jobs * 2), queue.Queue(maxsize=jobs * 2) + errors, counts = queue.Queue(), Counter() + + def guarded(function, *args): + try: + function(*args) + except Stopped: + pass + except BaseException as error: + errors.put(error) + stopped.set() + + reader = threading.Thread(target=guarded, args=(read_dumps, parts, inputs, stopped), + name='dump-reader') + writer = threading.Thread(target=guarded, + args=(write_pack, target, outputs, stopped, counts, legacy_markers), + name='pack-writer') + pending = set() + reader.start() + writer.start() + try: + with ProcessPoolExecutor(max_workers=jobs, + mp_context=multiprocessing.get_context('spawn')) as pool: + try: + exhausted = False + while pending or not exhausted: + if stopped.is_set(): + raise Stopped() + while not exhausted and len(pending) < jobs * 2: + batch = get(inputs, stopped) + if batch is None: + exhausted = True + else: + pending.add(pool.submit(convert_pages, batch, language)) + if pending: + done, pending = wait(pending, timeout=0.1, return_when=FIRST_COMPLETED) + for future in done: + put(outputs, future.result(), stopped) + except BaseException: + stopped.set() + for future in pending: + future.cancel() + raise + # Executor shutdown sends worker sentinels and joins the processes. + # Only then may the writer consume its poison pill and publish the pack. + reader.join() + put(outputs, None, stopped) + writer.join() + if not errors.empty(): + raise errors.get() + except Stopped: + if not errors.empty(): + raise errors.get() from None + raise RuntimeError('Conversion stopped before completion') from None + finally: + stopped.set() + reader.join() + writer.join() + return counts + + +def convert(language, snapshot, jobs=None, base=BASE_DIR, tier='common', overwrite=False): + if mw is None: + raise ValueError('Install dependencies with: python3 -m pip install -r requirements.txt') + jobs = available_cores() if jobs is None else jobs + if jobs < 1: + raise ValueError('Worker count must be positive') + parts = find_parts(base / '_dumps', language, snapshot) + output = base / '_markdown' + target = pack_path(output, language, snapshot, tier) + marker = target.with_name(target.name + '.incomplete') + legacy_marker = output / f'{snapshot_prefix(language, snapshot)}.incomplete' + if (target.is_file() and not overwrite and not marker.exists() + and not target.with_name(target.name + '.part').exists()): + legacy_marker.unlink(missing_ok=True) + print(f'Already complete; skipping {target.name}', flush=True) + return Counter() + output.mkdir(parents=True, exist_ok=True) + started = time.monotonic() + print(f'Using {jobs} worker processes; writing to {target}', flush=True) + counts = run_pipeline(parts, target, language, jobs, (marker, legacy_marker)) + print(f"Complete: {counts['pages']:,} pages, {counts['articles']:,} articles " + f'in {time.monotonic() - started:.1f}s\n{target}', flush=True) + return counts + + +def main(argv=None): + parser = argparse.ArgumentParser( + description='Convert Wikipedia XML dumps directly to a Markdown YaCy Pack.', + epilog='Example: python3 extract_wikipedia_dump_to_markdown.py de 2026-09-01\n' + 'Reads _dumps/*.xml.bz2; writes one _markdown/YaCy-Pack_*.jsonl.gz.\n' + 'Creates no intermediate wikitext files. Completed packs are skipped;\n' + 'unfinished .part packs restart in full. Install dependencies from requirements.txt.', + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument('language', type=language_code) + parser.add_argument('date', type=dump_date) + parser.add_argument('--jobs', type=int, default=None, + help='worker processes (default: all available logical CPU cores)') + parser.add_argument('--tier', choices=PACK_TIERS, default='common') + parser.add_argument('--overwrite', action='store_true', help='regenerate a completed pack') + args = sys.argv[1:] if argv is None else argv + if not args: + parser.print_help() + return 0 + args = parser.parse_args(args) + if args.jobs is not None and args.jobs < 1: + parser.error('--jobs must be positive') + try: + convert(args.language, args.date, args.jobs, tier=args.tier, overwrite=args.overwrite) + except KeyboardInterrupt: + print('Interrupted; output ending in .part is unfinished.', file=sys.stderr) + return 130 + except (OSError, ValueError, RuntimeError) as error: + print(f'Error: {error}. Output ending in .part is unfinished.', file=sys.stderr) + return 1 + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/ir/mediawiki/markdown_to_term_frequency.py b/ir/mediawiki/markdown_to_term_frequency.py new file mode 100755 index 000000000..53547417e --- /dev/null +++ b/ir/mediawiki/markdown_to_term_frequency.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Count terms per Markdown article in a YaCy Pack and stream them to JSONL.""" + +import os +import sys +import json +import time +import argparse +import multiprocessing +from collections import Counter +from extract_wikipedia_dump_to_markdown import (BASE_DIR, decompressed, dump_date, find_containers, + language_code) +from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait + +try: + import regex +except ImportError: + regex = None + +# A word starts with a letter/number; combining marks stay attached to it. +WORD = regex.compile(r"[\p{L}\p{N}][\p{L}\p{M}\p{N}]*", regex.VERSION1) if regex else None + + +def word_frequencies(text): + if WORD is None: + raise ValueError("Install dependencies with: python3 -m pip install -r requirements.txt") + return Counter(match[0] for match in WORD.finditer(text.lower())) + + +def available_cores(): + if hasattr(os, "process_cpu_count"): + return os.process_cpu_count() or 1 + if hasattr(os, "sched_getaffinity"): + return len(os.sched_getaffinity(0)) or 1 + return os.cpu_count() or 1 + + +def raw_pack_batches(parts, size=64, max_bytes=262144): + """Read unchanged bytes; keep Bulk action/document pairs in bounded batches.""" + for part in parts: + batch, byte_count, first_line = [], 0, 1 + with decompressed(part) as stream: + for number, line in enumerate(stream, 1): + batch.append(line) + byte_count += len(line) + # One oversized article is allowed; never split a Bulk pair. + if number % 2 == 0 and (len(batch) >= size * 2 or byte_count >= max_bytes): + yield part, first_line, batch + batch, byte_count, first_line = [], 0, number + 1 + if batch: + yield part, first_line, batch + + +def count_batch(batch): + """Decode, validate, count and encode in a worker with private article counters.""" + source, first_line, raw_lines = batch + output = [] + for offset, line in enumerate(raw_lines): + try: + if offset % 2 == 0: + if line != b'{"index":{}}\n' and json.loads(line.decode('utf-8')) != {'index': {}}: + raise ValueError('expected {"index":{}} action') + continue + article = json.loads(line.decode('utf-8')) + if (not isinstance(article, dict) or not isinstance(article.get('title'), list) + or len(article['title']) != 1 or not isinstance(article['title'][0], str) + or not article['title'][0] or not isinstance(article.get('text_t'), str) + or not isinstance(article.get('url_s'), str) or not article['url_s']): + raise ValueError('expected url_s, one title, and text_t') + except (ValueError, TypeError) as error: + raise ValueError(f'Invalid pack record at {source}:{first_line + offset}: {error}') from error + counts = word_frequencies(article['text_t']) + output.append(json.dumps({article['title'][0]: counts}, ensure_ascii=False, + separators=(',', ':')) + '\n') + if len(raw_lines) % 2: + raise ValueError(f'Invalid pack record at {source}:{first_line + len(raw_lines) - 1}: ' + 'missing document after index action') + return len(output), ''.join(output).encode('utf-8') + + +def write_term_frequencies(destination, batches, jobs): + completed, last_report, started = 0, 0, time.monotonic() + pending = set() + try: + with destination.open("xb", buffering=1024 * 1024) as stream: + with ProcessPoolExecutor(max_workers=jobs, + mp_context=multiprocessing.get_context("spawn")) as pool: + try: + exhausted = False + while pending or not exhausted: + while not exhausted and len(pending) < jobs * 2: + batch = next(batches, None) + if batch is None: + exhausted = True + else: + pending.add(pool.submit(count_batch, batch)) + if not pending: + break + done, pending = wait(pending, return_when=FIRST_COMPLETED) + for future in done: + count, encoded = future.result() + stream.write(encoded) + completed += count + if completed - last_report >= 1000: + print(f" {completed:,} articles counted " + f"({completed / (time.monotonic() - started):.0f}/s)", flush=True) + last_report = completed + finally: + for future in pending: + future.cancel() + finally: + batches.close() + return completed + + +def compute(language, snapshot, jobs=None, base=BASE_DIR): + if WORD is None: + raise ValueError("Install dependencies with: python3 -m pip install -r requirements.txt") + parts = find_containers(base, language, snapshot, '_markdown') + jobs = available_cores() if jobs is None else jobs + if jobs < 1: + raise ValueError("Worker count must be positive") + output = base / "_statistics" + output.mkdir(parents=True, exist_ok=True) + prefix = f"{language.replace('-', '_')}wiki-{snapshot}" + target = output / f"{prefix}-term-frequency.jsonl" + partial = target.with_suffix(target.suffix + ".part") + for path in (target, partial): + if path.exists(): + raise ValueError(f"Output already exists: {path}. Move it aside before rerunning.") + batches = raw_pack_batches(parts) + started = time.monotonic() + print(f"Counting article terms with {jobs} worker processes into {partial}", flush=True) + count = write_term_frequencies(partial, batches, jobs) + partial.replace(target) + print(f"Complete: {count:,} documents in {time.monotonic() - started:.1f}s\n{target}", flush=True) + return count + + +def main(argv=None): + parser = argparse.ArgumentParser( + description="Count terms per Markdown article and write term-frequency JSONL.", + epilog="Example: python3 markdown_to_term_frequency.py de 2026-09-01\n" + "Reads matching _markdown/YaCy-Pack_*.jsonl.gz; writes _statistics/wiki--term-frequency.jsonl.\n" + "Each line maps an article title to its term counts. Output remains .part until complete.", + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("language", type=language_code) + parser.add_argument("date", type=dump_date) + parser.add_argument("--jobs", type=int, default=None, + help="worker processes (default: all available logical CPU cores)") + args = sys.argv[1:] if argv is None else argv + if not args: + parser.print_help() + return 0 + args = parser.parse_args(args) + if args.jobs is not None and args.jobs < 1: + parser.error("--jobs must be positive") + try: + compute(args.language, args.date, args.jobs) + except KeyboardInterrupt: + print("Interrupted; outputs ending in .part are unfinished.", file=sys.stderr) + return 130 + except (OSError, ValueError, RuntimeError) as error: + print(f"Error: {error}. Outputs ending in .part are unfinished.", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ir/mediawiki/requirements.txt b/ir/mediawiki/requirements.txt new file mode 100644 index 000000000..d5d6abd48 --- /dev/null +++ b/ir/mediawiki/requirements.txt @@ -0,0 +1,7 @@ +# Wikitext rendering inside extract_wikipedia_dump_to_markdown.py. +mwparserfromhell==0.7.2 +# Unicode word boundaries for markdown_to_term_frequency.py; retains Python 3.9 support. +regex==2025.7.34 +# Parquet conversion; Arrow 21 is the last release supporting Python 3.9. +pyarrow==21.0.0; python_version < "3.10" +pyarrow==25.0.1; python_version >= "3.10" diff --git a/ir/mediawiki/term_frequency_to_document_frequency.py b/ir/mediawiki/term_frequency_to_document_frequency.py new file mode 100755 index 000000000..1c96008a1 --- /dev/null +++ b/ir/mediawiki/term_frequency_to_document_frequency.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Transform per-article term-frequency JSONL into document-frequency CSV.""" + +import os +import sys +import csv +import time +import json +import argparse +import multiprocessing +from collections import Counter +from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait +from extract_wikipedia_dump_to_markdown import BASE_DIR, dump_date, language_code + +DOCUMENT_COUNT_TERM = "_" # Reserved for corpus size; the tokenizer never emits it. + + +def available_cores(): + if hasattr(os, 'process_cpu_count'): + return os.process_cpu_count() or 1 + if hasattr(os, 'sched_getaffinity'): + return len(os.sched_getaffinity(0)) or 1 + return os.cpu_count() or 1 + + +def raw_term_batches(source, size=512, max_bytes=1048576): + """Bound queued input by document count and bytes, allowing one large record.""" + batch, byte_count, first_line = [], 0, 1 + with source.open('rb') as stream: + for number, line in enumerate(stream, 1): + batch.append(line) + byte_count += len(line) + if len(batch) >= size or byte_count >= max_bytes: + yield source, first_line, batch + batch, byte_count, first_line = [], 0, number + 1 + if batch: + yield source, first_line, batch + + +def count_batch(batch): + """Decode and count document presence in a worker-local Counter.""" + source, first_line, lines = batch + frequencies = Counter() + for number, line in enumerate(lines, first_line): + try: + record = json.loads(line.decode('utf-8')) + if not isinstance(record, dict) or len(record) != 1: + raise ValueError("expected one article object") + title, counts = next(iter(record.items())) + if not title or not isinstance(counts, dict): + raise ValueError("expected an article title and word-count object") + if any(not word or word == DOCUMENT_COUNT_TERM or type(count) is not int or count <= 0 + for word, count in counts.items()): + raise ValueError("word counts must be positive integers with nonempty, non-reserved words") + except (ValueError, TypeError) as error: + raise ValueError(f"Invalid term-frequency record at {source}:{number}: {error}") from error + # Count each key once per document, regardless of its term frequency. + frequencies.update(term for term in counts + if not term.replace('.', '').replace(',', '').isdecimal()) + return frequencies, len(lines) + + +def partial_counts(batches, jobs): + """Map bounded batches to processes, returning completed counters promptly.""" + if jobs == 1: + for batch in batches: + yield count_batch(batch) + return + with ProcessPoolExecutor(max_workers=jobs, + mp_context=multiprocessing.get_context('spawn')) as pool: + pending = set() + try: + exhausted = False + while pending or not exhausted: + while not exhausted and len(pending) < jobs * 2: + batch = next(batches, None) + if batch is None: + exhausted = True + else: + pending.add(pool.submit(count_batch, batch)) + if not pending: + break + done, pending = wait(pending, return_when=FIRST_COMPLETED) + for future in done: + yield future.result() + finally: + for future in pending: + future.cancel() + + +def document_frequencies(source, jobs=None): + """Merge partial counters in one process; no shared mutable counter or locks.""" + jobs = available_cores() if jobs is None else jobs + if jobs < 1: + raise ValueError('Worker count must be positive') + frequencies = Counter() + documents, last_report = 0, 0 + batches = raw_term_batches(source) + results = partial_counts(batches, jobs) + try: + for counts, records in results: + frequencies.update(counts) + documents += records + if documents - last_report >= 10000: + print(f" {documents:,} JSONL records read; {len(frequencies):,} distinct terms", flush=True) + last_report = documents + finally: + results.close() + batches.close() + return frequencies, documents + + +def write_document_frequencies(stream, frequencies, documents): + """Write retained terms by ascending document frequency, then corpus size.""" + terms = sorted((term for term, count in frequencies.items() if (count >= 10 and term >= 'aa') or count > 100), + key=lambda term: (frequencies[term], term)) + writer = csv.writer(stream, delimiter=";", lineterminator="\n") + writer.writerow(('term', 'count')) + writer.writerows((term, frequencies[term]) for term in terms) + writer.writerow((DOCUMENT_COUNT_TERM, documents)) + return len(terms) + + +def compute(language, snapshot, base=BASE_DIR, jobs=None): + jobs = available_cores() if jobs is None else jobs + if jobs < 1: + raise ValueError('Worker count must be positive') + output = base / "_statistics" + prefix = f"{language.replace('-', '_')}wiki-{snapshot}" + source = output / f"{prefix}-term-frequency.jsonl" + if source.with_suffix(source.suffix + ".part").exists(): + raise ValueError(f"Term-frequency input is incomplete: {source}") + if not source.is_file(): + raise ValueError(f"No term-frequency input: {source}") + target = output / f"{prefix}-document-frequency.csv" + partial = target.with_suffix(target.suffix + ".part") + for path in (target, partial): + if path.exists(): + raise ValueError(f"Output already exists: {path}. Move it aside before rerunning.") + started = time.monotonic() + print(f"Reading {source} to count document frequencies with {jobs} workers", flush=True) + # Reserve the output before reading; a failed run leaves only a .part file. + with partial.open("x", encoding="utf-8", newline="") as stream: + frequencies, records = document_frequencies(source, jobs) + retained = write_document_frequencies(stream, frequencies, records) + partial.replace(target) + print(f"Complete: {records:,} documents, {len(frequencies):,} distinct terms, " + f"{retained:,} terms stored in CSV " + f"in {time.monotonic() - started:.1f}s\n{target}", flush=True) + return records, len(frequencies) + + +def main(argv=None): + parser = argparse.ArgumentParser( + description="Transform term-frequency JSONL into document-frequency CSV.", + epilog="Example: python3 term_frequency_to_document_frequency.py de 2026-09-01\n" + "Reads _statistics/wiki--term-frequency.jsonl;\n" + "writes _statistics/wiki--document-frequency.csv.\n" + "CSV starts with the header term;count and uses _ for total documents.\n" + "Retain terms with count >= 10 and term >= 'aa', or count > 100.\n" + "Rows sort by ascending document frequency, alphabetically for ties; _ comes last.\n" + "Numeric terms (decimal digits with optional dots/commas) are excluded.", + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("language", type=language_code) + parser.add_argument("date", type=dump_date) + parser.add_argument('--jobs', type=int, default=None, + help='worker processes (default: all available logical CPU cores; 1 runs sequentially)') + args = sys.argv[1:] if argv is None else argv + if not args: + parser.print_help() + return 0 + args = parser.parse_args(args) + if args.jobs is not None and args.jobs < 1: + parser.error('--jobs must be positive') + try: + compute(args.language, args.date, jobs=args.jobs) + except KeyboardInterrupt: + print("Interrupted; outputs ending in .part are unfinished.", file=sys.stderr) + return 130 + except (OSError, ValueError, RuntimeError) as error: + print(f"Error: {error}. Outputs ending in .part are unfinished.", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ir/mediawiki/write_parquet.py b/ir/mediawiki/write_parquet.py new file mode 100755 index 000000000..607396499 --- /dev/null +++ b/ir/mediawiki/write_parquet.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Convert document-frequency CSV to Parquet sorted by term, with prefix encoding.""" + +import argparse +import sys +import time +from extract_wikipedia_dump_to_markdown import BASE_DIR, dump_date, language_code + +try: + import pyarrow as pa + import pyarrow.csv as csv + import pyarrow.parquet as pq +except ImportError: + pa = None + + +def convert(language, snapshot, base=BASE_DIR, compression_level=22): + if pa is None: + raise ValueError('Install dependencies with: python3 -m pip install -r requirements.txt') + if not 1 <= compression_level <= 22: + raise ValueError('ZSTD compression level must be between 1 and 22') + prefix = f"{language.replace('-', '_')}wiki-{snapshot}" + source = base / '_statistics' / f'{prefix}-document-frequency.csv' + if source.with_suffix('.csv.part').exists(): + raise ValueError(f'Document-frequency input is incomplete: {source}') + if not source.is_file(): + raise ValueError(f'No document-frequency input: {source}') + target = source.with_suffix('.parquet') + partial = target.with_suffix('.parquet.part') + for path in (target, partial): + if path.exists(): + raise ValueError(f'Output already exists: {path}. Move it aside before rerunning.') + started = time.monotonic() + # Reserve the output before reading; never publish a failed conversion. + with partial.open('xb') as stream: + print(f'Reading {source}', flush=True) + table = csv.read_csv( + source, + parse_options=csv.ParseOptions(delimiter=';'), + convert_options=csv.ConvertOptions( + column_types={'term': pa.string(), 'count': pa.int64()}, + strings_can_be_null=False, + ), + ) + if table.column_names != ['term', 'count']: + raise ValueError('Expected CSV header: term;count') + if any(column.null_count for column in table.columns): + raise ValueError('Terms and counts must not be null') + print(f'Sorting {table.num_rows:,} rows by term in memory', flush=True) + table = table.sort_by([('term', 'ascending')]) + print(f'Writing {partial} with ZSTD level {compression_level}', flush=True) + pq.write_table( + table, stream, + compression='zstd', compression_level=compression_level, + use_dictionary=False, + column_encoding={'term': 'DELTA_BYTE_ARRAY', 'count': 'DELTA_BINARY_PACKED'}, + row_group_size=262144, + sorting_columns=pq.SortingColumn.from_ordering(table.schema, [('term', 'ascending')]), + ) + partial.replace(target) + print(f'Complete: {table.num_rows:,} rows in {time.monotonic() - started:.1f}s\n{target}', flush=True) + return table.num_rows + + +def main(argv=None): + parser = argparse.ArgumentParser( + description='Convert document-frequency CSV to term-sorted, ZSTD-compressed Parquet.', + epilog='Example: python3 write_parquet.py de 2026-09-01\n' + 'Reads _statistics/wiki--document-frequency.csv (header: term;count).\n' + 'Writes the same filename with .parquet; the CSV is unchanged.\n' + 'Loads and sorts the full table in RAM, including additional sorting workspace.\n' + 'Unfinished output ends in .part; existing outputs are not overwritten.', + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument('language', type=language_code) + parser.add_argument('date', type=dump_date) + parser.add_argument('--compression-level', type=int, default=22, metavar='1..22', + help='ZSTD compression level (default: 22; 22 is slowest)') + args = sys.argv[1:] if argv is None else argv + if not args: + parser.print_help() + return 0 + args = parser.parse_args(args) + if not 1 <= args.compression_level <= 22: + parser.error('--compression-level must be between 1 and 22') + try: + convert(args.language, args.date, compression_level=args.compression_level) + except KeyboardInterrupt: + print('Interrupted; output ending in .part is unfinished.', file=sys.stderr) + return 130 + except (OSError, ValueError, RuntimeError, MemoryError) as error: + print(f'Error: {error}. Output ending in .part is unfinished.', file=sys.stderr) + return 1 + return 0 + + +if __name__ == '__main__': + sys.exit(main())