cairn/import_corpus.py at main · ghzhost/cairn · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
"""Corpus import scaffold: turn a .txt or .html file into a reviewable,
front-matter markdown scaffold — never a corpus input format Cairn reads.

`cairn index` reads exactly one format: markdown with a minimal front-matter
block (see `cairn/corpus.py`). This script never changes that, and it is not
wired into `cairn index` or any runtime path — it is a one-time, offline,
stdlib-only preprocessing convenience for migrating existing text (PDF-
derived text, a web page, a plain notice) into that format, with a
mandatory human-review step before the output is a real corpus document.

Nothing here guesses a doc id and ships it quietly: the placeholder id is
prefixed `review-` and stays that way until a human renames it. Ids are
citation-load-bearing (see `cairn/corpus.py`, `DOC_ID`) and should never be
auto-assigned as if they were final.

After writing the scaffold, this script loads it back through
`cairn.corpus.load_document` — the exact function `cairn index` calls — and
prints the passage boundaries that call actually produced, so what an
author reviews is exactly what indexing would do with it, not a second,
possibly-drifting idea of what "a paragraph" means.

Usage:
    python3 import_corpus.py notice.txt -o corpus/mine/notice.md --lang en
    python3 import_corpus.py page.html -o corpus/mine/page.md --title "..."
"""

from __future__ import annotations

import argparse
import re
import sys
from html.parser import HTMLParser
from pathlib import Path

from cairn.corpus import CorpusError, load_document

_SLUG_RE = re.compile(r"[^a-z0-9._:-]+")


def slugify(text: str) -> str:
    """A string that satisfies `cairn.corpus.DOC_ID`: starts with a letter,
    holds only letters, digits, `.`, `_`, `:`, `-`."""
    slug = _SLUG_RE.sub("-", text.strip().lower()).strip("-")
    if not slug or not slug[0].isalpha():
        slug = f"doc-{slug}" if slug else "doc"
    return slug


class _ParagraphExtractor(HTMLParser):
    """Naive HTML-to-paragraphs: text is split on block-level tag
    boundaries; everything inside `<script>`/`<style>`/`<nav>`/`<head>` is
    dropped. Not a general-purpose HTML reader — good enough for a page a
    human is about to review paragraph by paragraph regardless.
    """

    _BLOCK_TAGS = {
        "p", "div", "li", "h1", "h2", "h3", "h4", "h5", "h6", "br",
        "tr", "section", "article", "header", "footer",
    }
    _SKIP_TAGS = {"script", "style", "head", "nav"}

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.paragraphs: list[str] = []
        self.title: str | None = None
        self._current: list[str] = []
        self._skip_depth = 0
        self._in_title = False

    def _flush(self) -> None:
        text = " ".join(" ".join(self._current).split())
        if text:
            self.paragraphs.append(text)
        self._current = []

    def handle_starttag(self, tag: str, attrs: list) -> None:
        if tag in self._SKIP_TAGS:
            self._skip_depth += 1
        if tag == "title":
            self._in_title = True
        if tag in self._BLOCK_TAGS:
            self._flush()

    def handle_endtag(self, tag: str) -> None:
        if tag in self._SKIP_TAGS:
            self._skip_depth = max(0, self._skip_depth - 1)
        if tag == "title":
            self._in_title = False
        if tag in self._BLOCK_TAGS:
            self._flush()

    def handle_data(self, data: str) -> None:
        if self._in_title:
            # Checked ahead of `_skip_depth`: `<title>` lives inside `<head>`,
            # which is itself a skip tag for body text, but the title is not
            # body text and must not be swallowed by that same guard.
            self.title = (self.title or "") + data
            return
        if self._skip_depth:
            return
        self._current.append(data)

    def close(self) -> None:
        self._flush()
        super().close()


def extract_html(text: str) -> tuple[list[str], str | None]:
    parser = _ParagraphExtractor()
    parser.feed(text)
    parser.close()
    title = parser.title.strip() if parser.title else None
    return parser.paragraphs, (title or None)


def extract_text(text: str) -> list[str]:
    """Plain text, already assumed paragraph-delimited by blank lines — the
    same convention `cairn.corpus._chunk` reads."""
    blocks = re.split(r"\n\s*\n", text.strip())
    return [" ".join(b.split()) for b in blocks if b.strip()]


def build_scaffold(paragraphs: list[str], *, doc_id: str, title: str, lang: str) -> str:
    """The scaffold's front matter carries one extra key, `review`, that
    `cairn.corpus` never reads (only `id`, `title`, `lang`, and `synthetic`
    are) — front matter accepts unknown keys silently, which is exactly
    what an inert, human-visible marker needs. The review reminder itself
    is never written into the body: `cairn.corpus._chunk` would turn any
    body text into a real, retrievable, scored passage the moment this file
    is indexed, and a reviewer's own note becoming a quotable "passage" is
    the last thing this scaffold should risk.
    """
    body = "\n\n".join(paragraphs)
    return (
        "---\n"
        f"id: {doc_id}\n"
        f"title: {title}\n"
        f"lang: {lang}\n"
        "synthetic: false\n"
        "review: unreviewed\n"
        "---\n"
        f"{body}\n"
    )


def scaffold_one(
    src: Path,
    out_path: Path,
    *,
    doc_id: str | None,
    title: str | None,
    lang: str,
) -> tuple[int, int]:
    """Scaffold one input file to `out_path`. Returns `(exit_code,
    paragraph_count)` — the single source both `--batch` and the one-file
    path go through, so scaffolding a directory of files is never a second,
    drifting idea of what scaffolding one file does.
    """
    if not src.is_file():
        print(f"import_corpus: error: no such file: {src}", file=sys.stderr)
        return 1, 0
    raw = src.read_text(encoding="utf-8", errors="replace")

    if src.suffix.lower() in (".htm", ".html"):
        paragraphs, html_title = extract_html(raw)
        default_title = html_title or src.stem
    else:
        paragraphs = extract_text(raw)
        default_title = src.stem

    if not paragraphs:
        print(f"import_corpus: error: no paragraph text extracted from {src}", file=sys.stderr)
        return 1, 0

    resolved_title = title or default_title
    resolved_id = doc_id or f"review-{slugify(resolved_title)}"

    scaffold = build_scaffold(paragraphs, doc_id=resolved_id, title=resolved_title, lang=lang)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(scaffold, encoding="utf-8")

    print(f"Wrote {out_path} ({len(paragraphs)} paragraph(s) extracted)")
    print(
        "REVIEW REQUIRED before this is a real corpus document: check the doc id "
        f"(still prefixed 'review-' unless --id was given: {resolved_id!r}), the title, "
        "the language, the synthetic flag, and every paragraph boundary shown "
        "below. Then delete the 'review: unreviewed' front-matter line — it is "
        "inert to Cairn, a marker for a human only."
    )

    try:
        doc = load_document(out_path)
    except CorpusError as exc:
        print(
            f"WARNING: the scaffold does not load as a valid document: {exc}",
            file=sys.stderr,
        )
        return 1, len(paragraphs)

    print(f"Chunk preview ({len(doc.passages)} passage(s), via cairn.corpus.load_document):")
    for p in doc.passages:
        preview = " ".join(p.text.split())
        if len(preview) > 70:
            preview = preview[:69] + "…"
        print(f"  {p.passage_id}: {preview}")
    return 0, len(paragraphs)


def _batch_sources(src_dir: Path) -> list[Path]:
    """Every `.txt`/`.html` file directly in `src_dir` — non-recursive, the
    same flat-directory convention `cairn.corpus.corpus_paths` uses for the
    real corpus, so batch output maps predictably onto a real corpus layout.
    """
    return sorted(
        p
        for p in src_dir.iterdir()
        if p.is_file() and p.suffix.lower() in (".txt", ".html", ".htm")
    )


def run_batch(src_dir: Path, out_dir: Path, *, lang: str) -> int:
    if not src_dir.is_dir():
        print(f"import_corpus: error: not a directory: {src_dir}", file=sys.stderr)
        return 1
    sources = _batch_sources(src_dir)
    if not sources:
        print(f"import_corpus: error: no .txt or .html files in {src_dir}", file=sys.stderr)
        return 1

    failed = 0
    total_paragraphs = 0
    for src in sources:
        print(f"--- {src.name} ---")
        code, paragraph_count = scaffold_one(
            src, out_dir / f"{src.stem}.md", doc_id=None, title=None, lang=lang
        )
        if code != 0:
            failed += 1
        else:
            total_paragraphs += paragraph_count
        print()

    print(
        f"Batch: {len(sources) - failed}/{len(sources)} file(s) scaffolded, "
        f"{total_paragraphs} paragraph(s) total."
    )
    if failed:
        print(f"{failed} file(s) failed to scaffold — see the errors above.", file=sys.stderr)
    print("REVIEW REQUIRED for every file above before any of them is a real corpus document.")
    return 1 if failed else 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Scaffold front-matter markdown corpus document(s) from .txt or .html."
    )
    parser.add_argument(
        "input", help="a .txt or .html file, or (with --batch) a directory of them"
    )
    parser.add_argument(
        "-o",
        "--output",
        required=True,
        help="path to write the .md scaffold to (with --batch: the output directory)",
    )
    parser.add_argument(
        "--batch",
        action="store_true",
        help=(
            "treat 'input' as a directory: scaffold every .txt/.html file in it "
            "(non-recursive) into --output, one .md per source file. --id and "
            "--title do not apply — each file's id/title is derived the same way "
            "the one-file path derives them when neither is given."
        ),
    )
    parser.add_argument(
        "--id",
        dest="doc_id",
        default=None,
        help="doc id (default: review-<slug of title or filename>); not valid with --batch",
    )
    parser.add_argument(
        "--title",
        default=None,
        help=(
            "document title (default: <title> tag for HTML, filename for text); "
            "not valid with --batch"
        ),
    )
    parser.add_argument("--lang", default="en", help="language code (default: en)")
    args = parser.parse_args(argv)

    if args.batch:
        if args.doc_id or args.title:
            print(
                "import_corpus: error: --id and --title are not valid with --batch",
                file=sys.stderr,
            )
            return 1
        return run_batch(Path(args.input), Path(args.output), lang=args.lang)

    code, _ = scaffold_one(
        Path(args.input),
        Path(args.output),
        doc_id=args.doc_id,
        title=args.title,
        lang=args.lang,
    )
    return code


if __name__ == "__main__":
    sys.exit(main())