ainfo

385def main() -> None:
386    app()

 45@app.command()
 46def run(
 47    url: str,
 48    render_js: bool = typer.Option(
 49        False, help="Render pages using a headless browser before extraction",
 50    ),
 51    use_llm: bool = typer.Option(
 52        False, help="Use an LLM instead of regex for contact extraction",
 53    ),
 54    summarize: bool = typer.Option(
 55        False, help="Summarize page content using the LLM",
 56    ),
 57    summary_language: str = typer.Option(
 58        "German",
 59        "--summary-language",
 60        help="Language used for LLM summaries",
 61        envvar="AINFO_SUMMARY_LANGUAGE",
 62    ),
 63    summary_prompt: str | None = typer.Option(
 64        None,
 65        "--summary-prompt",
 66        help="Custom instruction supplied to the LLM when summarising",
 67        envvar="AINFO_SUMMARY_PROMPT",
 68    ),
 69    summary_prompt_file: Path | None = typer.Option(
 70        None,
 71        "--summary-prompt-file",
 72        help="Read the summary prompt from PATH",
 73    ),
 74    extract: list[str] = typer.Option(
 75        [], "--extract", "-e", help="Additional extractors to run",
 76    ),
 77    output: Path | None = typer.Option(
 78        None, "--output", "-o", help="Write JSON results to PATH.",
 79    ),
 80    json_output: bool = typer.Option(
 81        False, "--json", help="Print extracted data as JSON to stdout",
 82    ),
 83    include_text: bool = typer.Option(
 84        True,
 85        "--text/--no-text",
 86        help="Include page text in the results",
 87    ),
 88) -> None:
 89    """Fetch ``url`` and display extracted text and optional information."""
 90
 91    if summary_prompt is not None and summary_prompt_file is not None:
 92        raise typer.BadParameter(
 93            "Use either --summary-prompt or --summary-prompt-file, not both"
 94        )
 95
 96    custom_summary_prompt = summary_prompt
 97    if summary_prompt_file is not None:
 98        try:
 99            custom_summary_prompt = summary_prompt_file.read_text(encoding="utf-8")
100        except OSError as exc:
101            raise typer.BadParameter(
102                f"Unable to read summary prompt file: {exc}"
103            ) from exc
104
105    raw = fetch_data(url, render_js=render_js)
106    document = parse_data(raw, url=url)
107    text: str | None = None
108    if include_text or summarize:
109        text = extract_text(document)
110
111    results: dict[str, object] = {}
112    if include_text and text is not None:
113        results["text"] = text
114
115    needs_llm = summarize or (use_llm and "contacts" in extract)
116
117    if needs_llm:
118        with LLMService() as llm:
119            for name in extract:
120                func = AVAILABLE_EXTRACTORS.get(name)
121                if func is None:
122                    raise typer.BadParameter(f"Unknown extractor: {name}")
123                if name == "contacts":
124                    results[name] = func(
125                        document, method="llm" if use_llm else "regex", llm=llm
126                    )
127                else:
128                    results[name] = func(document)
129            if summarize and text is not None:
130                results["summary"] = llm.summarize(
131                    text, language=summary_language, prompt=custom_summary_prompt
132                )
133    else:
134        for name in extract:
135            func = AVAILABLE_EXTRACTORS.get(name)
136            if func is None:
137                raise typer.BadParameter(f"Unknown extractor: {name}")
138            if name == "contacts":
139                results[name] = func(document, method="regex", llm=None)
140            else:
141                results[name] = func(document)
142
143    if output is not None:
144        serialisable = {
145            k: (v.model_dump() if isinstance(v, ContactDetails) else v)
146            for k, v in results.items()
147        }
148        output.write_text(json.dumps(serialisable))
149
150    if json_output:
151        serialisable = {
152            k: (v.model_dump() if isinstance(v, ContactDetails) else v)
153            for k, v in results.items()
154        }
155        typer.echo(json.dumps(serialisable))
156    else:
157        if include_text and text is not None:
158            typer.echo(text)
159        for name in extract:
160            value = results.get(name)
161            if name == "contacts" and isinstance(value, ContactDetails):
162                output_results(value)
163            else:
164                typer.echo(f"{name}:")
165                if isinstance(value, dict):
166                    for key, items in value.items():
167                        typer.echo(f"  {key}: {', '.join(items)}")
168                elif isinstance(value, list):
169                    for item in value:
170                        typer.echo(f"  - {item}")
171                elif value is not None:
172                    typer.echo(f"  {value}")
173        if summarize and "summary" in results:
174            typer.echo("summary:")
175            typer.echo(results["summary"])

178@app.command()
179def crawl(
180    url: str,
181    depth: int = 1,
182    render_js: bool = typer.Option(
183        False, help="Render pages using a headless browser before extraction",
184    ),
185    use_llm: bool = typer.Option(
186        False, help="Use an LLM instead of regex for contact extraction",
187    ),
188    extract: list[str] = typer.Option(
189        [], "--extract", "-e", help="Additional extractors to run",
190    ),
191    output: Path | None = typer.Option(
192        None, "--output", "-o", help="Write JSON results to PATH.",
193    ),
194    json_output: bool = typer.Option(
195        False, "--json", help="Print aggregated results as JSON to stdout",
196    ),
197    include_text: bool = typer.Option(
198        True,
199        "--text/--no-text",
200        help="Include page text in the results",
201    ),
202) -> None:
203    """Crawl ``url`` up to ``depth`` levels and extract text and data."""
204
205    method = "llm" if use_llm else "regex"
206    aggregated_results: dict[str, dict[str, object]] = {}
207
208    async def _crawl(llm: LLMService | None = None) -> None:
209        async for link, raw in crawl_urls(url, depth, render_js=render_js):
210            document = parse_data(raw, url=link)
211            page_results: dict[str, object] = {}
212            text = ""
213            if include_text:
214                text = extract_text(document)
215                page_results["text"] = text
216            for name in extract:
217                func = AVAILABLE_EXTRACTORS.get(name)
218                if func is None:
219                    raise typer.BadParameter(f"Unknown extractor: {name}")
220                if name == "contacts":
221                    page_results[name] = func(document, method=method, llm=llm)
222                else:
223                    page_results[name] = func(document)
224            aggregated_results[link] = page_results
225            if not json_output:
226                typer.echo(f"Results for {link}:")
227                if include_text:
228                    typer.echo(text)
229                for name in extract:
230                    value = page_results.get(name)
231                    if name == "contacts" and isinstance(value, ContactDetails):
232                        output_results(value)
233                    else:
234                        typer.echo(f"{name}: {value}")
235                typer.echo()
236
237    if use_llm:
238        with LLMService() as llm:
239            asyncio.run(_crawl(llm))
240    else:
241        asyncio.run(_crawl())
242
243    if output is not None:
244        serialisable = {
245            url: {
246                k: (v.model_dump() if isinstance(v, ContactDetails) else v)
247                for k, v in res.items()
248            }
249            for url, res in aggregated_results.items()
250        }
251        output.write_text(json.dumps(serialisable))
252    if json_output:
253        serialisable = {
254            url: {
255                k: (v.model_dump() if isinstance(v, ContactDetails) else v)
256                for k, v in res.items()
257            }
258            for url, res in aggregated_results.items()
259        }
260        typer.echo(json.dumps(serialisable))

24def fetch_data(url: str, render_js: bool = False) -> str | asyncio.Task[str]:
25    """Fetch raw HTML from ``url``.
26
27    The function adapts to the surrounding asynchronous environment. If no
28    event loop is running, the coroutine is executed immediately and the HTML
29    is returned. When called while an event loop is already running, the
30    coroutine is scheduled on that loop and an :class:`asyncio.Task` is
31    returned. For fully asynchronous workflows use :func:`async_fetch_data`.
32
33    Parameters
34    ----------
35    url:
36        The address to retrieve.
37    render_js:
38        Whether to render the page with a headless browser so that any
39        JavaScript on the page executes before the HTML is returned.
40
41    Returns
42    -------
43    str | asyncio.Task[str]
44        The HTML body of the page or a task that resolves to it.
45    """
46
47    try:
48        loop = asyncio.get_running_loop()
49    except RuntimeError:
50        return asyncio.run(_fetch(url, render_js))
51    else:
52        return loop.create_task(_fetch(url, render_js))

18async def async_fetch_data(url: str, render_js: bool = False) -> str:
19    """Fetch raw HTML from ``url`` asynchronously."""
20
21    return await _fetch(url, render_js)

10def parse_data(raw: str, url: str | None = None) -> Document:
11    """Parse raw HTML into a :class:`~ainfo.models.Document`.
12
13    Parameters
14    ----------
15    raw:
16        The raw HTML string.
17    url:
18        Optional source URL associated with the HTML.
19    """
20
21    return parse_html(raw, url=url)

 79def extract_information(
 80    doc: Document,
 81    method: str = "regex",
 82    llm: LLMService | None = None,
 83    instruction: str | None = None,
 84    model: str | None = None,
 85) -> ContactDetails:
 86    """Extract contact details from a parsed document.
 87
 88    Parameters
 89    ----------
 90    doc:
 91        Parsed :class:`Document` to process.
 92    method:
 93        ``"regex"`` to use the built-in regular expressions or ``"llm"`` to
 94        delegate extraction to an LLM service.
 95    llm:
 96        Instance of :class:`LLMService` required when ``method`` is ``"llm"``.
 97    """
 98
 99    logger.info("Extracting contact information using %s", method)
100    text = extract_text(doc, content_only=False)
101    if method == "llm":
102        if llm is None:
103            msg = "LLMService instance required when method='llm'"
104            raise ValueError(msg)
105        instruction = instruction or (
106            "Extract any email addresses, phone numbers, street addresses and "
107            "social media profiles from the following text. Respond in JSON "
108            "with keys 'emails', 'phone_numbers', 'addresses' and "
109            "'social_media'."
110          )
111        response = llm.extract(text, instruction, model=model)
112        try:
113            data = json.loads(response)
114        except Exception:
115            data = {}
116        return ContactDetails(
117            emails=data.get("emails", []),
118            phone_numbers=data.get("phone_numbers", []),
119            addresses=data.get("addresses", []),
120            social_media=data.get("social_media", []),
121        )
122
123    # Default to regex based extraction
124    return ContactDetails(
125        emails=extract_emails(doc),
126        phone_numbers=extract_phone_numbers(text),
127        addresses=extract_addresses(text),
128        social_media=extract_social_profiles(text),
129    )

43def extract_text(
44    doc: Document,
45    joiner: str = " ",
46    as_list: bool = False,
47    *,
48    content_only: bool = True,
49) -> str | list[str]:
50    """Extract and clean the main textual content from ``doc``.
51
52    Parameters
53    ----------
54    doc:
55        Parsed :class:`Document` to process.
56    joiner:
57        String used to join individual text fragments when ``as_list`` is
58        ``False``. Defaults to a single space.
59    as_list:
60        When ``True`` return a list of text fragments instead of a single
61        string.
62    content_only:
63        When ``True`` include only nodes identified as primary content. Set to
64        ``False`` to include navigation and footer text as well.
65    """
66
67    logger.info("Extracting text from document")
68    parts = [
69        re.sub(r"\s+", " ", p).strip()
70        for p in _gather_text(doc.nodes, content_only=content_only)
71    ]
72    if as_list:
73        return [p for p in parts if p]
74    filtered = [p for p in parts if p]
75    text = joiner.join(filtered)
76    return text.strip()

132def extract_custom(
133    doc: Document,
134    patterns: dict[str, str] | None = None,
135    *,
136    llm: LLMService | None = None,
137    prompt: str | None = None,
138    model: str | None = None,
139) -> dict[str, list[str]]:
140    """Extract arbitrary information from ``doc``.
141
142    The extraction can be performed either using regular expression
143    ``patterns`` or delegated to an LLM service when ``llm`` is provided.
144
145    Parameters
146    ----------
147    doc:
148        Parsed :class:`Document` to search.
149    patterns:
150        Mapping of field names to regular expression patterns. Required when
151        ``llm`` is ``None``.
152    llm:
153        Optional :class:`LLMService` used to perform extraction via a large
154        language model.
155    prompt:
156        Custom prompt supplied to the LLM. It should describe the desired JSON
157        structure, for example ``"Extract product names as a list under the key
158        'products'"``. If omitted a generic instruction is used.
159    model:
160        Identifier of the model to use when ``llm`` is provided.
161
162    Returns
163    -------
164    dict[str, list[str]]
165        A mapping of field names to lists of extracted strings.
166    """
167
168    logger.info("Extracting custom information")
169    text = extract_text(doc)
170    if llm is not None:
171        instruction = prompt or "Extract the requested information as JSON."
172        response = llm.extract(text, instruction, model=model)
173        try:
174            data = json.loads(response)
175        except Exception:
176            data = {}
177        results: dict[str, list[str]] = {}
178        for key, value in data.items():
179            if isinstance(value, list):
180                results[key] = value
181            elif value is not None:
182                results[key] = [value]
183        return results
184
185    if patterns is None:
186        msg = "patterns required when llm is None"
187        raise ValueError(msg)
188
189    results: dict[str, list[str]] = {}
190    for key, pattern in patterns.items():
191        regex = re.compile(pattern, re.IGNORECASE)
192        matches = [m.group(0) for m in regex.finditer(text)]
193        results[key] = list(dict.fromkeys(matches))
194    return results

322def extract_site(
323    url: str,
324    *,
325    depth: int = 0,
326    render_js: bool = False,
327    extract: list[str] | None = None,
328    include_text: bool = False,
329    use_llm: bool = False,
330    llm: LLMService | None = None,
331    dedupe: bool = True,
332) -> dict[str, dict[str, object]] | asyncio.Task[dict[str, dict[str, object]]]:
333    """Synchronously run :func:`async_extract_site` when no event loop exists.
334
335    When called from within a running event loop a task is scheduled instead.
336    """
337
338    try:
339        loop = asyncio.get_running_loop()
340    except RuntimeError:
341        if use_llm and llm is None:
342            with LLMService() as managed_llm:
343                return asyncio.run(
344                    async_extract_site(
345                        url,
346                        depth=depth,
347                        render_js=render_js,
348                        extract=extract,
349                        include_text=include_text,
350                        use_llm=True,
351                        llm=managed_llm,
352                        dedupe=dedupe,
353                    )
354                )
355        return asyncio.run(
356            async_extract_site(
357                url,
358                depth=depth,
359                render_js=render_js,
360                extract=extract,
361                include_text=include_text,
362                use_llm=use_llm,
363                llm=llm,
364                dedupe=dedupe,
365            )
366        )
367    else:
368        if use_llm and llm is None:
369            msg = "llm must be provided when use_llm=True inside an event loop"
370            raise RuntimeError(msg)
371        return loop.create_task(
372            async_extract_site(
373                url,
374                depth=depth,
375                render_js=render_js,
376                extract=extract,
377                include_text=include_text,
378                use_llm=use_llm,
379                llm=llm,
380                dedupe=dedupe,
381            )
382        )

263async def async_extract_site(
264    url: str,
265    *,
266    depth: int = 0,
267    render_js: bool = False,
268    extract: list[str] | None = None,
269    include_text: bool = False,
270    use_llm: bool = False,
271    llm: LLMService | None = None,
272    dedupe: bool = True,
273) -> dict[str, dict[str, object]]:
274    """Crawl ``url`` up to ``depth`` levels and run extractors on each page.
275
276    Results are returned as a mapping of page URL to the extracted data.
277    Duplicate pages are skipped by comparing a SHA-256 hash of their HTML
278    content. Only pages on the same domain as ``url`` are processed.
279    """
280
281    extract_names = list(extract or ["contacts"])
282    method = "llm" if use_llm else "regex"
283    if use_llm and llm is None:
284        msg = "llm service required when use_llm=True"
285        raise ValueError(msg)
286
287    start_domain = urlparse(url).netloc
288    results: dict[str, dict[str, object]] = {}
289    seen_hashes: set[str] = set()
290
291    async for link, raw in crawl_urls(url, depth, render_js=render_js):
292        if urlparse(link).netloc != start_domain:
293            continue
294
295        if dedupe:
296            digest = hashlib.sha256(raw.encode("utf-8", errors="ignore")).hexdigest()
297            if digest in seen_hashes:
298                logger.debug("Skipping %s due to duplicate content hash", link)
299                continue
300            seen_hashes.add(digest)
301
302        document = parse_data(raw, url=link)
303        page_results: dict[str, object] = {}
304
305        if include_text:
306            page_results["text"] = extract_text(document)
307
308        for name in extract_names:
309            func = AVAILABLE_EXTRACTORS.get(name)
310            if func is None:
311                raise ValueError(f"Unknown extractor: {name}")
312            if name == "contacts":
313                page_results[name] = func(document, method=method, llm=llm)
314            else:
315                page_results[name] = func(document)
316
317        results[link] = page_results
318
319    return results