diff --git a/.ai/notes6-answers.txt b/.ai/notes6-answers.txt
new file mode 100644
index 0000000..f90b781
--- /dev/null
+++ b/.ai/notes6-answers.txt
@@ -0,0 +1,17 @@
+Answers:
+
+1. API surface: (lookup, retrieve_record) as clean public APIs that callers compose themselves
+
+2. What should the return type look like? Something like a dataclass FetchResult(payload, http_headers, warc_headers, surt_key, timestamp)
+
+3. HTML parser integration:
+
+```python pseudocode
+import ccnget
+from selectolax.lexbor import LexborHTMLParser
+resp = ccnget.fetch("http://example.com")
+html = resp.payload
+tree = LexborHTMLParser(html)
+```
+
+4. Backward compatibility: I like the CLI interface pretty well now, would prefer to keep it the same
diff --git a/.ai/notes6.txt b/.ai/notes6.txt
new file mode 100644
index 0000000..659dad5
--- /dev/null
+++ b/.ai/notes6.txt
@@ -0,0 +1,9 @@
+source: src/ccnget/geturl.py
+run: uv run ccnget ...
+test: make test
+
+This command line is working well, but now I want to use it as a library.
+
+Please analyze how this might be refactored to better support use as a module with various html parsers such as https://github.com/rushter/selectolax
+
+as a caller of the API, I might like to get access to the response headers, or the WARC headers
diff --git a/Makefile b/Makefile
index 95db7ac..8922fdb 100644
--- a/Makefile
+++ b/Makefile
@@ -8,7 +8,7 @@ help:
@echo " make test Run static checks followed immediately by pytest"
@echo " make clean Wipe out test tool cache tracking footprints"
@echo " make init Initialize new project with uv and test setup"
- @echo " make man Create man page"
+ @echo " make mandoc Create man page and pydoc markdown for api.md"
check:
@echo "\n— [An extremely fast Python linter and code formatter](https://docs.astral.sh/ruff/)"
@@ -48,6 +48,8 @@ checkdeps:
testpackages:
uv add --dev ruff bandit vulture refurb ty pytest interrogate argparse-manpage
+mandoc: man doc
+
man:
mkdir -p man
uv run argparse-manpage \
@@ -59,6 +61,9 @@ man:
--include man/__envars.inc \
> man/ccnget.1
+doc:
+ uvx pydoc-markdown
+
export GIT_CEILING_DIRECTORIES # can influence `uv init` behaviour
pyproject.toml:
uv init --package .
diff --git a/README.md b/README.md
index f4fd853..753414f 100644
--- a/README.md
+++ b/README.md
@@ -116,6 +116,11 @@ Besides the code in this repository, code needed to make this work is in
* [`brian-learns/cdx-cc-news` dataset Files tab](https://huggingface.co/datasets/brian-learns/cdx-cc-news/tree/main) to build cdxj and rocksdb indexes
* [`brian-learns/cc-news-cdx-server` hf spaces Files tab](https://huggingface.co/spaces/brian-learns/cc-news-cdx-server/tree/main) for the lookup endpoint
+## See Also
+ * [`samples`](./samples/) directory with example using `duckdb` to query the parquet files, and sort of random samples of the data
+ * [`man`](./man/) man page for the command line
+ * [`api.md`](./api.md) pydoc markdown for use as a python module
+
## License
BSD 3-Clause for the code in this revision control repository.
diff --git a/api.md b/api.md
new file mode 100644
index 0000000..8f83da5
--- /dev/null
+++ b/api.md
@@ -0,0 +1,292 @@
+
+
+# ccnget
+
+ccnget -- lookup URLs and get archived pages from Common Crawl News.
+
+Library usage
+-------------
+>>> import ccnget
+>>> result = ccnget.fetch("http://example.com")
+>>> print(result.surt_key, result.timestamp)
+>>> # Parse with any HTML parser
+>>> from selectolax.lexbor import LexborHTMLParser
+>>> tree = LexborHTMLParser(result.payload)
+
+Or use the lower-level API:
+>>> lr = ccnget.lookup("http://example.com", limit=5)
+>>> for entry in lr.entries:
+... result = ccnget.retrieve(entry.warc_path, entry.offset, entry.length)
+
+
+
+# ccnget.geturl
+
+CLI entry-point for ccnget.
+
+Uses the library API (ccnget.api) for all logic.
+
+
+
+#### limited\_int
+
+```python
+def limited_int(val_str: str) -> int
+```
+
+Checks that input is an integer between 1 and 100.
+
+
+
+#### lookup\_cmd
+
+```python
+def lookup_cmd(args: argparse.Namespace) -> None
+```
+
+Execute the lookup subcommand.
+
+
+
+#### retrieve\_cmd
+
+```python
+def retrieve_cmd(args: argparse.Namespace) -> None
+```
+
+Execute the retrieve subcommand.
+
+
+
+#### fetch\_cmd
+
+```python
+def fetch_cmd(args: argparse.Namespace) -> None
+```
+
+Execute the fetch subcommand: lookup then retrieve the first result.
+
+
+
+#### get\_version
+
+```python
+def get_version() -> str
+```
+
+Get version from pyproject.toml
+
+
+
+#### get\_parser
+
+```python
+def get_parser() -> argparse.ArgumentParser
+```
+
+Build and return the ArgumentParser for ccnget.
+
+
+
+#### main
+
+```python
+def main(argv: Optional[list[str]] = None) -> None
+```
+
+Parse CLI arguments and dispatch to subcommands.
+
+
+
+# ccnget.api
+
+Public library API for ccnget.
+
+Lookup and retrieve archived web pages from the Common Crawl News dataset.
+
+Example
+-------
+>>> import ccnget
+>>> result = ccnget.fetch("http://example.com")
+>>> print(result.surt_key, result.timestamp)
+>>> tree = LexborHTMLParser(result.payload)
+
+
+
+## LookupEntry Objects
+
+```python
+@dataclass
+class LookupEntry()
+```
+
+One CDX index hit returned by the lookup API.
+
+
+
+## LookupResult Objects
+
+```python
+@dataclass
+class LookupResult()
+```
+
+Result of a CDX index lookup.
+
+
+
+## FetchResult Objects
+
+```python
+@dataclass
+class FetchResult()
+```
+
+Result of fetching an archived page.
+
+Attributes
+----------
+payload : bytes
+ Raw response body (typically HTML).
+http_headers : dict[str, str]
+ HTTP response headers from inside the WARC record.
+warc_headers : dict[str, str]
+ WARC record headers.
+surt_key : str
+ SURT-formatted URL key from the CDX index.
+timestamp : str
+ WARC timestamp (YYYYMMDDhhmmss).
+warc_path : str
+ Path to the WARC file on Common Crawl storage.
+
+
+
+#### lookup
+
+```python
+def lookup(url: str,
+ *,
+ exact: bool = False,
+ limit: int = 10,
+ at: str | None = None,
+ cdx_url: str = CDX_LOOKUP_URL) -> LookupResult
+```
+
+Search the CC-NEWS CDX index for *url*.
+
+Parameters
+----------
+url : str
+ URL to search for.
+exact : bool
+ Require exact match.
+limit : int
+ Maximum number of results (1-100).
+at : str | None
+ Timestamp filter (YYYYMMDDhhmmss).
+cdx_url : str
+ Override the CDX lookup endpoint.
+
+Returns
+-------
+LookupResult
+
+
+
+#### retrieve
+
+```python
+def retrieve(warc_path: str,
+ offset: int,
+ length: int,
+ *,
+ base_url: str = CC_CRAWL_BASE_URL,
+ surt_key: str = "",
+ timestamp: str = "") -> FetchResult
+```
+
+Retrieve a single WARC record via byte-range request.
+
+Parameters
+----------
+warc_path : str
+ Path within Common Crawl storage (e.g. ``crawl-data/CC-NEWS/...``).
+offset : int
+ Byte offset of the record.
+length : int
+ Byte length of the record.
+base_url : str
+ Override the Common Crawl base URL.
+surt_key : str
+ SURT key from the CDX index (populated by ``fetch()``).
+timestamp : str
+ Timestamp from the CDX index (populated by ``fetch()``).
+
+Returns
+-------
+FetchResult
+
+
+
+#### fetch
+
+```python
+def fetch(url: str,
+ *,
+ exact: bool = False,
+ at: str | None = None,
+ cdx_url: str = CDX_LOOKUP_URL,
+ base_url: str = CC_CRAWL_BASE_URL) -> FetchResult
+```
+
+Lookup *url* in the CDX index and retrieve the first archived result.
+
+Convenience wrapper around :func:`lookup` + :func:`retrieve`.
+
+Parameters
+----------
+url : str
+ URL to search for.
+exact : bool
+ Require exact match.
+at : str | None
+ Timestamp filter (YYYYMMDDhhmmss).
+cdx_url : str
+ Override the CDX lookup endpoint.
+base_url : str
+ Override the Common Crawl base URL.
+
+Returns
+-------
+FetchResult
+
+
+
+## CcngetError Objects
+
+```python
+class CcngetError(Exception)
+```
+
+Base exception for ccnget.
+
+
+
+## NotFoundError Objects
+
+```python
+class NotFoundError(CcngetError)
+```
+
+Raised when a URL has no matches in the CDX index.
+
+
+
+## NoRecordError Objects
+
+```python
+class NoRecordError(CcngetError)
+```
+
+Raised when a WARC segment contains no response record.
+
diff --git a/man/ccnget.1 b/man/ccnget.1
index 0ac4c85..e33d2d3 100644
--- a/man/ccnget.1
+++ b/man/ccnget.1
@@ -1,4 +1,4 @@
-.TH CCNGET "1" "2026\-08\-02" "ccnget 0.1.2" "Generated Python Manual"
+.TH CCNGET "1" "2026\-08\-02" "ccnget 0.1.3" "Generated Python Manual"
.SH NAME
ccnget
.SH SYNOPSIS
diff --git a/pyproject.toml b/pyproject.toml
index cb9938d..c083b29 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "ccnget"
-version = "0.1.2"
+version = "0.1.3"
description = "lookup urls and get files from Common Crawl News"
readme = "README.md"
authors = [
@@ -33,6 +33,10 @@ dev = [
"vulture>=2.16",
]
+[tool.pydoc-markdown.renderer]
+type = "markdown"
+filename = "api.md"
+
[tool.build_manpages]
manpages = [
"man/ccnget.1:function=get_parser:module=ccnget.geturl",
diff --git a/src/ccnget/__init__.py b/src/ccnget/__init__.py
index e69de29..49d7459 100644
--- a/src/ccnget/__init__.py
+++ b/src/ccnget/__init__.py
@@ -0,0 +1,40 @@
+"""ccnget -- lookup URLs and get archived pages from Common Crawl News.
+
+Library usage
+-------------
+>>> import ccnget
+>>> result = ccnget.fetch("http://example.com")
+>>> print(result.surt_key, result.timestamp)
+>>> # Parse with any HTML parser
+>>> from selectolax.lexbor import LexborHTMLParser
+>>> tree = LexborHTMLParser(result.payload)
+
+Or use the lower-level API:
+>>> lr = ccnget.lookup("http://example.com", limit=5)
+>>> for entry in lr.entries:
+... result = ccnget.retrieve(entry.warc_path, entry.offset, entry.length)
+"""
+
+from ccnget.api import (
+ CcngetError,
+ FetchResult,
+ LookupEntry,
+ LookupResult,
+ NoRecordError,
+ NotFoundError,
+ fetch,
+ lookup,
+ retrieve,
+)
+
+__all__ = [
+ "CcngetError",
+ "FetchResult",
+ "LookupEntry",
+ "LookupResult",
+ "NoRecordError",
+ "NotFoundError",
+ "fetch",
+ "lookup",
+ "retrieve",
+]
diff --git a/src/ccnget/api.py b/src/ccnget/api.py
new file mode 100644
index 0000000..327664e
--- /dev/null
+++ b/src/ccnget/api.py
@@ -0,0 +1,266 @@
+"""Public library API for ccnget.
+
+Lookup and retrieve archived web pages from the Common Crawl News dataset.
+
+Example
+-------
+>>> import ccnget
+>>> result = ccnget.fetch("http://example.com")
+>>> print(result.surt_key, result.timestamp)
+>>> tree = LexborHTMLParser(result.payload)
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass, field
+from io import BytesIO
+from typing import Any
+
+import requests
+from dotenv import load_dotenv
+from warcio.archiveiterator import ArchiveIterator
+
+load_dotenv()
+
+logger: logging.Logger = logging.getLogger(__name__)
+
+CDX_LOOKUP_URL: str = os.environ.get(
+ "CDX_LOOKUP_URL",
+ "https://brian-learns-cc-news-cdx-server.hf.space/lookup",
+)
+CC_CRAWL_BASE_URL: str = os.environ.get(
+ "CC_CRAWL_BASE_URL",
+ "https://data.commoncrawl.org",
+)
+
+
+# ── Data classes ──────────────────────────────────────────────────────────
+
+
+@dataclass
+class LookupEntry:
+ """One CDX index hit returned by the lookup API."""
+
+ surt_key: str
+ timestamp: str
+ warc_path: str
+ offset: int
+ length: int
+ # Extra fields from the API (if any) are stored here
+ extra: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class LookupResult:
+ """Result of a CDX index lookup."""
+
+ url: str
+ entries: list[LookupEntry]
+
+
+@dataclass
+class FetchResult:
+ """Result of fetching an archived page.
+
+ Attributes
+ ----------
+ payload : bytes
+ Raw response body (typically HTML).
+ http_headers : dict[str, str]
+ HTTP response headers from inside the WARC record.
+ warc_headers : dict[str, str]
+ WARC record headers.
+ surt_key : str
+ SURT-formatted URL key from the CDX index.
+ timestamp : str
+ WARC timestamp (YYYYMMDDhhmmss).
+ warc_path : str
+ Path to the WARC file on Common Crawl storage.
+ """
+
+ payload: bytes
+ http_headers: dict[str, str]
+ warc_headers: dict[str, str]
+ surt_key: str
+ timestamp: str
+ warc_path: str
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────
+
+
+def _headers_to_dict(headers: Any) -> dict[str, str]:
+ """Convert a warcio Header object to a plain dict."""
+ if headers is None:
+ return {}
+ # warcio Header objects have a .headers list of (name, value) tuples
+ return dict(headers.headers)
+
+
+def _entry_from_dict(d: dict[str, Any]) -> LookupEntry:
+ """Build a LookupEntry from a raw CDX JSON dict."""
+ known = {"surt_key", "timestamp", "warc_path", "offset", "length"}
+ base = {k: d[k] for k in known if k in d}
+ extra = {k: v for k, v in d.items() if k not in known}
+ return LookupEntry(extra=extra, **base) # type: ignore[arg-type]
+
+
+# ── Public API ────────────────────────────────────────────────────────────
+
+
+def lookup(
+ url: str,
+ *,
+ exact: bool = False,
+ limit: int = 10,
+ at: str | None = None,
+ cdx_url: str = CDX_LOOKUP_URL,
+) -> LookupResult:
+ """Search the CC-NEWS CDX index for *url*.
+
+ Parameters
+ ----------
+ url : str
+ URL to search for.
+ exact : bool
+ Require exact match.
+ limit : int
+ Maximum number of results (1-100).
+ at : str | None
+ Timestamp filter (YYYYMMDDhhmmss).
+ cdx_url : str
+ Override the CDX lookup endpoint.
+
+ Returns
+ -------
+ LookupResult
+ """
+ params = {"url": url, "exact": exact, "limit": limit, "at": at}
+ logger.debug("Requesting %s with params %s", cdx_url, params)
+
+ response = requests.get(cdx_url, params=params, timeout=30)
+ if response.status_code == 404:
+ raise NotFoundError(f"No match for {url} in {cdx_url}")
+ response.raise_for_status()
+
+ data = response.json()
+ entries = [_entry_from_dict(r) for r in data.get("results", [])]
+ return LookupResult(url=url, entries=entries)
+
+
+def retrieve(
+ warc_path: str,
+ offset: int,
+ length: int,
+ *,
+ base_url: str = CC_CRAWL_BASE_URL,
+ surt_key: str = "",
+ timestamp: str = "",
+) -> FetchResult:
+ """Retrieve a single WARC record via byte-range request.
+
+ Parameters
+ ----------
+ warc_path : str
+ Path within Common Crawl storage (e.g. ``crawl-data/CC-NEWS/...``).
+ offset : int
+ Byte offset of the record.
+ length : int
+ Byte length of the record.
+ base_url : str
+ Override the Common Crawl base URL.
+ surt_key : str
+ SURT key from the CDX index (populated by ``fetch()``).
+ timestamp : str
+ Timestamp from the CDX index (populated by ``fetch()``).
+
+ Returns
+ -------
+ FetchResult
+ """
+ warc_url = f"{base_url}/{warc_path}"
+ start = offset
+ end = start + length - 1
+
+ headers = {"Range": f"bytes={start}-{end}"}
+ logger.debug("Requesting %s Range: bytes=%d-%d", warc_url, start, end)
+
+ response = requests.get(warc_url, headers=headers, timeout=60)
+ response.raise_for_status()
+
+ for record in ArchiveIterator(BytesIO(response.content)):
+ if record.rec_type == "response":
+ payload = record.content_stream().read()
+ logger.debug("WARC Headers:\n%s", record.rec_headers)
+ return FetchResult(
+ payload=payload,
+ http_headers=_headers_to_dict(record.http_headers),
+ warc_headers=_headers_to_dict(record.rec_headers),
+ surt_key=surt_key,
+ timestamp=timestamp,
+ warc_path=warc_path,
+ )
+
+ raise NoRecordError(f"No response record found in WARC data at {warc_path}:{offset}")
+
+
+def fetch(
+ url: str,
+ *,
+ exact: bool = False,
+ at: str | None = None,
+ cdx_url: str = CDX_LOOKUP_URL,
+ base_url: str = CC_CRAWL_BASE_URL,
+) -> FetchResult:
+ """Lookup *url* in the CDX index and retrieve the first archived result.
+
+ Convenience wrapper around :func:`lookup` + :func:`retrieve`.
+
+ Parameters
+ ----------
+ url : str
+ URL to search for.
+ exact : bool
+ Require exact match.
+ at : str | None
+ Timestamp filter (YYYYMMDDhhmmss).
+ cdx_url : str
+ Override the CDX lookup endpoint.
+ base_url : str
+ Override the Common Crawl base URL.
+
+ Returns
+ -------
+ FetchResult
+ """
+ result = lookup(url, exact=exact, at=at, limit=1, cdx_url=cdx_url)
+ if not result.entries:
+ raise NotFoundError(f"No archived results for {url}")
+
+ first = result.entries[0]
+ logger.info("Found: %s at %s", first.surt_key, first.timestamp)
+ return retrieve(
+ first.warc_path,
+ first.offset,
+ first.length,
+ base_url=base_url,
+ surt_key=first.surt_key,
+ timestamp=first.timestamp,
+ )
+
+
+# ── Exceptions ────────────────────────────────────────────────────────────
+
+
+class CcngetError(Exception):
+ """Base exception for ccnget."""
+
+
+class NotFoundError(CcngetError):
+ """Raised when a URL has no matches in the CDX index."""
+
+
+class NoRecordError(CcngetError):
+ """Raised when a WARC segment contains no response record."""
diff --git a/src/ccnget/geturl.py b/src/ccnget/geturl.py
index 761f1fe..354dc54 100644
--- a/src/ccnget/geturl.py
+++ b/src/ccnget/geturl.py
@@ -1,33 +1,27 @@
+"""CLI entry-point for ccnget.
+
+Uses the library API (ccnget.api) for all logic.
+"""
+
+from __future__ import annotations
+
import argparse
import json
import logging
-import os
import sys
from importlib.metadata import PackageNotFoundError, version
-from io import BytesIO
-from pathlib import Path
from typing import Optional
-import requests
-from dotenv import load_dotenv
-from warcio.archiveiterator import ArchiveIterator
-
-load_dotenv()
+from ccnget.api import CDX_LOOKUP_URL, NotFoundError
+from ccnget.api import fetch as api_fetch
+from ccnget.api import lookup as api_lookup
+from ccnget.api import retrieve as api_retrieve
logger: logging.Logger = logging.getLogger(__name__)
-CDX_LOOKUP_URL = os.environ.get(
- "CDX_LOOKUP_URL",
- "https://brian-learns-cc-news-cdx-server.hf.space/lookup",
-)
-CC_CRAWL_BASE_URL = os.environ.get(
- "CC_CRAWL_BASE_URL",
- "https://data.commoncrawl.org",
-)
-
-def limited_int(val_str):
- """Checks that input is an integer between 1 and 1000."""
+def limited_int(val_str: str) -> int:
+ """Checks that input is an integer between 1 and 100."""
try:
val = int(val_str)
except ValueError:
@@ -39,87 +33,79 @@ def limited_int(val_str):
return val
-def handle_lookup_404(response: requests.Response, url: str) -> None:
- """Handle a 404 from the CDX lookup by printing a clean error and exiting."""
- if response.status_code == 404:
+def lookup_cmd(args: argparse.Namespace) -> None:
+ """Execute the lookup subcommand."""
+ try:
+ result = api_lookup(
+ args.url,
+ exact=args.exact,
+ limit=args.limit,
+ at=args.at,
+ )
+ except NotFoundError:
print(
- f"ccnget: no match for {url} in {CDX_LOOKUP_URL}",
+ f"ccnget: no match for {args.url} in {CDX_LOOKUP_URL}",
file=sys.stderr,
)
sys.exit(1)
- response.raise_for_status()
-
-def lookup_cmd(args: argparse.Namespace) -> None:
- """Execute the lookup subcommand."""
- params = {
- "url": args.url,
- "exact": args.exact,
- "limit": args.limit,
- "at": args.at,
+ # Build a JSON-serialisable dict matching the old format
+ output = {
+ "url": result.url,
+ "results": [
+ {
+ "surt_key": e.surt_key,
+ "timestamp": e.timestamp,
+ "warc_path": e.warc_path,
+ "offset": e.offset,
+ "length": e.length,
+ **e.extra,
+ }
+ for e in result.entries
+ ],
}
-
- logger.debug("Requesting %s with params %s", CDX_LOOKUP_URL, params)
-
- response = requests.get(CDX_LOOKUP_URL, params=params, timeout=30)
- handle_lookup_404(response, args.url)
-
- print(json.dumps(response.json(), indent=2))
-
-
-def retrieve_record(warc_path: str, offset: int, length: int, output: Optional[str] = None) -> None:
- """Retrieve a WARC record and write to stdout or file."""
- warc_url = f"{CC_CRAWL_BASE_URL}/{warc_path}"
- start = offset
- end = start + length - 1
-
- headers = {"Range": f"bytes={start}-{end}"}
- logger.debug("Requesting %s Range: bytes=%d-%d", warc_url, start, end)
-
- response = requests.get(warc_url, headers=headers, timeout=60)
- response.raise_for_status()
-
- for record in ArchiveIterator(BytesIO(response.content)):
- logger.debug(f"WARC Headers:\n{record.rec_headers}")
- if record.rec_type == "response":
- payload = record.content_stream().read()
- if output:
- Path(output).write_bytes(payload)
- logger.info("Wrote %d bytes to %s", len(payload), output)
- else:
- sys.stdout.buffer.write(payload)
- return
-
- logger.error("No response record found in WARC data")
+ print(json.dumps(output, indent=2))
def retrieve_cmd(args: argparse.Namespace) -> None:
"""Execute the retrieve subcommand."""
- retrieve_record(args.warc_path, args.offset, args.length, args.output)
+ result = api_retrieve(
+ args.warc_path,
+ args.offset,
+ args.length,
+ )
+ if args.output:
+ from pathlib import Path
-def fetch_cmd(args: argparse.Namespace) -> None:
- """Execute the fetch subcommand: lookup then retrieve the first result."""
- params = {
- "url": args.url,
- "exact": args.exact,
- "at": args.at,
- "limit": 1,
- }
+ Path(args.output).write_bytes(result.payload)
+ logger.info("Wrote %d bytes to %s", len(result.payload), args.output)
+ else:
+ sys.stdout.buffer.write(result.payload)
- logger.debug("Looking up %s", args.url)
- response = requests.get(CDX_LOOKUP_URL, params=params, timeout=30)
- handle_lookup_404(response, args.url)
- results = response.json().get("results", [])
+def fetch_cmd(args: argparse.Namespace) -> None:
+ """Execute the fetch subcommand: lookup then retrieve the first result."""
+ try:
+ result = api_fetch(
+ args.url,
+ exact=args.exact,
+ at=args.at,
+ )
+ except NotFoundError:
+ print(
+ f"ccnget: no match for {args.url} in {CDX_LOOKUP_URL}",
+ file=sys.stderr,
+ )
+ sys.exit(1)
- if not results:
- logger.error("No results found for %s", args.url)
- return
+ if args.output:
+ from pathlib import Path
- first = results[0]
- logger.info("Found: %s at %s", first["surt_key"], first["timestamp"])
- retrieve_record(first["warc_path"], first["offset"], first["length"], args.output)
+ Path(args.output).write_bytes(result.payload)
+ logger.info("Wrote %d bytes to %s", len(result.payload), args.output)
+ else:
+ sys.stdout.buffer.write(result.payload)
def get_version() -> str:
diff --git a/tests/test_geturl.py b/tests/test_geturl.py
index 4d5ce0a..60a7a42 100644
--- a/tests/test_geturl.py
+++ b/tests/test_geturl.py
@@ -1,13 +1,29 @@
import argparse
-import json
from io import BytesIO
from unittest.mock import MagicMock, patch
import pytest
-from warcio.archiveiterator import ArchiveIterator
from warcio.warcwriter import WARCWriter
-from ccnget.geturl import fetch_cmd, lookup_cmd, limited_int, main, retrieve_cmd
+from ccnget.api import (
+ CcngetError,
+ FetchResult,
+ LookupEntry,
+ LookupResult,
+ NoRecordError,
+ NotFoundError,
+ _entry_from_dict,
+)
+from ccnget.api import (
+ fetch as api_fetch,
+)
+from ccnget.api import (
+ lookup as api_lookup,
+)
+from ccnget.api import (
+ retrieve as api_retrieve,
+)
+from ccnget.geturl import fetch_cmd, limited_int, lookup_cmd, main, retrieve_cmd
def _make_warc_response(payload: bytes) -> bytes:
@@ -27,6 +43,9 @@ def _make_warc_response(payload: bytes) -> bytes:
return buf.getvalue()
+# ── CLI helpers ────────────────────────────────────────────────────────────
+
+
class TestLimitedInt:
def test_valid_value(self):
assert limited_int("50") == 50
@@ -50,8 +69,11 @@ def test_non_integer_raises(self):
limited_int("abc")
+# ── CLI parsing tests ─────────────────────────────────────────────────────
+
+
class TestMainParsing:
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_lookup_subcommand(self, mock_get, capsys):
mock_response = MagicMock()
mock_response.json.return_value = {"results": []}
@@ -62,7 +84,7 @@ def test_lookup_subcommand(self, mock_get, capsys):
captured = capsys.readouterr()
assert '"results"' in captured.out
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_retrieve_subcommand(self, mock_get, capsys):
warc_content = _make_warc_response(b"test")
mock_response = MagicMock()
@@ -90,8 +112,11 @@ def test_retrieve_missing_required_fails(self):
assert exc_info.value.code != 0
+# ── CLI command tests ─────────────────────────────────────────────────────
+
+
class TestLookupCmd:
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_lookup_calls_api(self, mock_get):
mock_response = MagicMock()
mock_response.json.return_value = {"results": []}
@@ -108,7 +133,7 @@ def test_lookup_calls_api(self, mock_get):
assert call_args[1]["params"]["limit"] == 10
assert call_args[1]["params"]["at"] is None
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_lookup_exact_flag(self, mock_get):
mock_response = MagicMock()
mock_response.json.return_value = {"results": []}
@@ -120,7 +145,7 @@ def test_lookup_exact_flag(self, mock_get):
call_args = mock_get.call_args
assert call_args[1]["params"]["exact"] is True
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_lookup_at_parameter(self, mock_get):
mock_response = MagicMock()
mock_response.json.return_value = {"results": []}
@@ -132,9 +157,22 @@ def test_lookup_at_parameter(self, mock_get):
call_args = mock_get.call_args
assert call_args[1]["params"]["at"] == "20240101120000"
+ @patch("ccnget.api.requests.get")
+ def test_lookup_404_clean_exit(self, mock_get, capsys):
+ mock_response = MagicMock()
+ mock_response.status_code = 404
+ mock_get.return_value = mock_response
+
+ args = argparse.Namespace(url="http://nonexistent.example", exact=False, limit=10, at=None)
+ with pytest.raises(SystemExit):
+ lookup_cmd(args)
+
+ captured = capsys.readouterr()
+ assert "no match for http://nonexistent.example" in captured.err
+
class TestRetrieveCmd:
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_retrieve_writes_to_stdout(self, mock_get, capsys):
warc_content = _make_warc_response(b"test")
@@ -153,7 +191,7 @@ def test_retrieve_writes_to_stdout(self, mock_get, capsys):
captured = capsys.readouterr()
assert b"test" in captured.out.encode()
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_retrieve_writes_to_file(self, mock_get, tmp_path):
warc_content = _make_warc_response(b"file test")
@@ -173,7 +211,7 @@ def test_retrieve_writes_to_file(self, mock_get, tmp_path):
assert output_file.exists()
assert output_file.read_bytes() == b"file test"
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_retrieve_uses_range_header(self, mock_get):
warc_content = _make_warc_response(b"test")
@@ -196,12 +234,18 @@ def test_retrieve_uses_range_header(self, mock_get):
class TestFetchCmd:
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_fetch_retrieves_first_result(self, mock_get, capsys):
lookup_response = MagicMock()
lookup_response.json.return_value = {
"results": [
- {"surt_key": "com,example)/", "timestamp": "20170101000000", "warc_path": "test.warc.gz", "offset": 100, "length": 50}
+ {
+ "surt_key": "com,example)/",
+ "timestamp": "20170101000000",
+ "warc_path": "test.warc.gz",
+ "offset": 100,
+ "length": 50,
+ }
]
}
@@ -218,12 +262,18 @@ def test_fetch_retrieves_first_result(self, mock_get, capsys):
assert b"fetched" in captured.out.encode()
assert mock_get.call_count == 2
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_fetch_with_at_parameter(self, mock_get, capsys):
lookup_response = MagicMock()
lookup_response.json.return_value = {
"results": [
- {"surt_key": "com,example)/", "timestamp": "20170101000000", "warc_path": "test.warc.gz", "offset": 100, "length": 50}
+ {
+ "surt_key": "com,example)/",
+ "timestamp": "20170101000000",
+ "warc_path": "test.warc.gz",
+ "offset": 100,
+ "length": 50,
+ }
]
}
@@ -239,12 +289,18 @@ def test_fetch_with_at_parameter(self, mock_get, capsys):
call_args = mock_get.call_args_list[0]
assert call_args[1]["params"]["at"] == "20240101120000"
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_fetch_with_output_file(self, mock_get, tmp_path):
lookup_response = MagicMock()
lookup_response.json.return_value = {
"results": [
- {"surt_key": "com,example)/", "timestamp": "20170101000000", "warc_path": "test.warc.gz", "offset": 200, "length": 75}
+ {
+ "surt_key": "com,example)/",
+ "timestamp": "20170101000000",
+ "warc_path": "test.warc.gz",
+ "offset": 200,
+ "length": 75,
+ }
]
}
@@ -261,13 +317,166 @@ def test_fetch_with_output_file(self, mock_get, tmp_path):
assert output_file.exists()
assert output_file.read_bytes() == b"file output"
- @patch("ccnget.geturl.requests.get")
+ @patch("ccnget.api.requests.get")
def test_fetch_no_results(self, mock_get, capsys):
lookup_response = MagicMock()
lookup_response.json.return_value = {"results": []}
mock_get.return_value = lookup_response
args = argparse.Namespace(url="http://nonexistent.example", exact=False, output=None, at=None)
- fetch_cmd(args)
+ with pytest.raises(SystemExit):
+ fetch_cmd(args)
assert mock_get.call_count == 1
+ captured = capsys.readouterr()
+ assert "no match for http://nonexistent.example" in captured.err
+
+
+# ── Library API tests ─────────────────────────────────────────────────────
+
+
+class TestEntryFromDict:
+ def test_basic_entry(self):
+ d = {
+ "surt_key": "com,example)/",
+ "timestamp": "20170101000000",
+ "warc_path": "crawl-data/test.warc.gz",
+ "offset": 100,
+ "length": 500,
+ }
+ entry = _entry_from_dict(d)
+ assert entry.surt_key == "com,example)/"
+ assert entry.timestamp == "20170101000000"
+ assert entry.warc_path == "crawl-data/test.warc.gz"
+ assert entry.offset == 100
+ assert entry.length == 500
+ assert entry.extra == {}
+
+ def test_extra_fields(self):
+ d = {
+ "surt_key": "com,example)/",
+ "timestamp": "20170101000000",
+ "warc_path": "test.warc.gz",
+ "offset": 0,
+ "length": 100,
+ "original_url": "http://example.com",
+ "mime_type": "text/html",
+ }
+ entry = _entry_from_dict(d)
+ assert entry.extra["original_url"] == "http://example.com"
+ assert entry.extra["mime_type"] == "text/html"
+
+
+class TestLookup:
+ @patch("ccnget.api.requests.get")
+ def test_lookup_returns_result(self, mock_get):
+ mock_response = MagicMock()
+ mock_response.json.return_value = {
+ "results": [
+ {
+ "surt_key": "com,example)/",
+ "timestamp": "20170101000000",
+ "warc_path": "test.warc.gz",
+ "offset": 100,
+ "length": 50,
+ }
+ ]
+ }
+ mock_get.return_value = mock_response
+
+ result = api_lookup("http://example.com")
+ assert isinstance(result, LookupResult)
+ assert result.url == "http://example.com"
+ assert len(result.entries) == 1
+ assert isinstance(result.entries[0], LookupEntry)
+ assert result.entries[0].surt_key == "com,example)/"
+
+ @patch("ccnget.api.requests.get")
+ def test_lookup_404_raises(self, mock_get):
+ mock_response = MagicMock()
+ mock_response.status_code = 404
+ mock_get.return_value = mock_response
+
+ with pytest.raises(NotFoundError, match="No match for"):
+ api_lookup("http://nonexistent.example")
+
+ @patch("ccnget.api.requests.get")
+ def test_lookup_custom_cdx_url(self, mock_get):
+ mock_response = MagicMock()
+ mock_response.json.return_value = {"results": []}
+ mock_get.return_value = mock_response
+
+ api_lookup("http://example.com", cdx_url="http://custom-cdx/lookup")
+ assert mock_get.call_args[0][0] == "http://custom-cdx/lookup"
+
+
+class TestRetrieve:
+ @patch("ccnget.api.requests.get")
+ def test_retrieve_returns_fetch_result(self, mock_get):
+ warc_content = _make_warc_response(b"test")
+ mock_response = MagicMock()
+ mock_response.content = warc_content
+ mock_get.return_value = mock_response
+
+ result = api_retrieve("test.warc.gz", 0, 100)
+ assert isinstance(result, FetchResult)
+ assert result.payload == b"test"
+ assert result.warc_path == "test.warc.gz"
+ assert isinstance(result.http_headers, dict)
+ assert isinstance(result.warc_headers, dict)
+
+ @patch("ccnget.api.requests.get")
+ def test_retrieve_custom_base_url(self, mock_get):
+ warc_content = _make_warc_response(b"test")
+ mock_response = MagicMock()
+ mock_response.content = warc_content
+ mock_get.return_value = mock_response
+
+ api_retrieve("test.warc.gz", 0, 100, base_url="http://custom-crawl")
+ assert mock_get.call_args[0][0] == "http://custom-crawl/test.warc.gz"
+
+
+class TestFetch:
+ @patch("ccnget.api.requests.get")
+ def test_fetch_returns_fetch_result(self, mock_get):
+ lookup_response = MagicMock()
+ lookup_response.json.return_value = {
+ "results": [
+ {
+ "surt_key": "com,example)/",
+ "timestamp": "20170101000000",
+ "warc_path": "test.warc.gz",
+ "offset": 100,
+ "length": 50,
+ }
+ ]
+ }
+
+ warc_content = _make_warc_response(b"fetched")
+ retrieve_response = MagicMock()
+ retrieve_response.content = warc_content
+
+ mock_get.side_effect = [lookup_response, retrieve_response]
+
+ result = api_fetch("http://example.com")
+ assert isinstance(result, FetchResult)
+ assert result.payload == b"fetched"
+ assert result.surt_key
+ assert result.timestamp
+ assert mock_get.call_count == 2
+
+ @patch("ccnget.api.requests.get")
+ def test_fetch_no_results_raises(self, mock_get):
+ mock_response = MagicMock()
+ mock_response.json.return_value = {"results": []}
+ mock_get.return_value = mock_response
+
+ with pytest.raises(NotFoundError, match="No archived results"):
+ api_fetch("http://nonexistent.example")
+
+
+class TestExceptions:
+ def test_hierarchy(self):
+ assert issubclass(NotFoundError, CcngetError)
+ assert issubclass(NoRecordError, CcngetError)
+ assert issubclass(CcngetError, Exception)
diff --git a/uv.lock b/uv.lock
index 87c2f84..cc637d1 100644
--- a/uv.lock
+++ b/uv.lock
@@ -79,7 +79,7 @@ wheels = [
[[package]]
name = "ccnget"
-version = "0.1.2"
+version = "0.1.3"
source = { editable = "." }
dependencies = [
{ name = "python-dotenv" },