archivebox.index package

Submodules

archivebox.index.csv module

archivebox.index.csv.links_to_csv(links: List[archivebox.index.schema.Link], cols: Optional[List[str]] = None, header: bool = True, separator: str = ',', ljust: int = 0) → str

archivebox.index.csv.to_csv(obj: Any, cols: List[str], separator: str = ',', ljust: int = 0) → str

archivebox.index.html module

archivebox.index.html.parse_html_main_index(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Iterator[str]

parse an archive index html file and return the list of urls

archivebox.index.html.generate_index_from_links(links: List[archivebox.index.schema.Link], with_headers: bool)

archivebox.index.html.main_index_template(links: List[archivebox.index.schema.Link], template: str = 'static_index.html') → str

render the template for the entire main index

archivebox.index.html.write_html_link_details(link: archivebox.index.schema.Link, out_dir: Optional[str] = None) → None

archivebox.index.html.link_details_template(link: archivebox.index.schema.Link) → str

archivebox.index.html.render_django_template(template: str, context: Mapping[str, str]) → str

render a given html template string with the given template content

archivebox.index.html.snapshot_icons(snapshot) → str

archivebox.index.json module

archivebox.index.json.generate_json_index_from_links(links: List[archivebox.index.schema.Link], with_headers: bool)

archivebox.index.json.parse_json_main_index(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Iterator[archivebox.index.schema.Link]

parse an archive index json file and return the list of links

archivebox.index.json.write_json_link_details(link: archivebox.index.schema.Link, out_dir: Optional[str] = None) → None

write a json file with some info about the link

archivebox.index.json.parse_json_link_details(out_dir: Union[pathlib.Path, str], guess: Optional[bool] = False) → Optional[archivebox.index.schema.Link]

load the json link index from a given directory

archivebox.index.json.parse_json_links_details(out_dir: Union[pathlib.Path, str]) → Iterator[archivebox.index.schema.Link]

read through all the archive data folders and return the parsed links

class archivebox.index.json.ExtendedEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: json.encoder.JSONEncoder

Extended json serializer that supports serializing several model fields and objects

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

archivebox.index.json.to_json(obj: Any, indent: Optional[int] = 4, sort_keys: bool = True, cls=<class 'archivebox.index.json.ExtendedEncoder'>) → str

archivebox.index.schema module

WARNING: THIS FILE IS ALL LEGACY CODE TO BE REMOVED.

DO NOT ADD ANY NEW FEATURES TO THIS FILE, NEW CODE GOES HERE: core/models.py

exception archivebox.index.schema.ArchiveError(message, hints=None)

Bases: Exception

class archivebox.index.schema.ArchiveResult(cmd: List[str], pwd: Union[str, NoneType], cmd_version: Union[str, NoneType], output: Union[str, Exception, NoneType], status: str, start_ts: datetime.datetime, end_ts: datetime.datetime, index_texts: Union[List[str], NoneType] = None, schema: str = 'ArchiveResult')

Bases: object

index_texts = None

schema = 'ArchiveResult'

typecheck() → None

classmethod guess_ts(dict_info)

classmethod from_json(json_info, guess=False)

to_dict(*keys) → dict

to_json(indent=4, sort_keys=True) → str

to_csv(cols: Optional[List[str]] = None, separator: str = ', ', ljust: int = 0) → str

classmethod field_names()

duration

class archivebox.index.schema.Link(timestamp: str, url: str, title: Union[str, NoneType], tags: Union[str, NoneType], sources: List[str], history: Dict[str, List[archivebox.index.schema.ArchiveResult]] = <factory>, updated: Union[datetime.datetime, NoneType] = None, schema: str = 'Link')

Bases: object

updated = None

schema = 'Link'

overwrite(**kwargs)

pure functional version of dict.update that returns a new instance

typecheck() → None

as_snapshot()

classmethod from_json(json_info, guess=False)

to_json(indent=4, sort_keys=True) → str

to_csv(cols: Optional[List[str]] = None, separator: str = ', ', ljust: int = 0) → str

snapshot_id

classmethod field_names()

link_dir

archive_path

archive_size

url_hash

scheme

extension

domain

path

basename

base_url

bookmarked_date

updated_date

archive_dates

oldest_archive_date

newest_archive_date

num_outputs

num_failures

is_static

is_archived

latest_outputs(status: str = None) → Dict[str, Union[str, Exception, None]]

get the latest output that each archive method produced for link

canonical_outputs() → Dict[str, Optional[str]]

predict the expected output paths that should be present after archiving

archivebox.index.sql module

archivebox.index.sql.parse_sql_main_index(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Iterator[archivebox.index.schema.Link]

archivebox.index.sql.remove_from_sql_main_index(snapshots: django.db.models.query.QuerySet, atomic: bool = False, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → None

archivebox.index.sql.write_link_to_sql_index(link: archivebox.index.schema.Link)

archivebox.index.sql.write_sql_main_index(links: List[archivebox.index.schema.Link], out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → None

archivebox.index.sql.write_sql_link_details(link: archivebox.index.schema.Link, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → None

archivebox.index.sql.list_migrations(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → List[Tuple[bool, str]]

archivebox.index.sql.apply_migrations(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → List[str]

archivebox.index.sql.get_admins(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → List[str]

Module contents

archivebox.index.merge_links(a: archivebox.index.schema.Link, b: archivebox.index.schema.Link) → archivebox.index.schema.Link

deterministially merge two links, favoring longer field values over shorter, and “cleaner” values over worse ones.

archivebox.index.validate_links(links: Iterable[archivebox.index.schema.Link]) → List[archivebox.index.schema.Link]

archivebox.index.archivable_links(links: Iterable[archivebox.index.schema.Link]) → Iterable[archivebox.index.schema.Link]

remove chrome://, about:// or other schemed links that cant be archived

archivebox.index.fix_duplicate_links(sorted_links: Iterable[archivebox.index.schema.Link]) → Iterable[archivebox.index.schema.Link]

ensures that all non-duplicate links have monotonically increasing timestamps

archivebox.index.sorted_links(links: Iterable[archivebox.index.schema.Link]) → Iterable[archivebox.index.schema.Link]

archivebox.index.links_after_timestamp(links: Iterable[archivebox.index.schema.Link], resume: Optional[float] = None) → Iterable[archivebox.index.schema.Link]

archivebox.index.lowest_uniq_timestamp(used_timestamps: collections.OrderedDict, timestamp: str) → str

resolve duplicate timestamps by appending a decimal 1234, 1234 -> 1234.1, 1234.2

archivebox.index.timed_index_update(out_path: pathlib.Path)

archivebox.index.write_main_index(links: List[archivebox.index.schema.Link], out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → None

Writes links to sqlite3 file for a given list of links

archivebox.index.load_main_index(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs'), warn: bool = True) → List[archivebox.index.schema.Link]

parse and load existing index with any new links from import_path merged in

archivebox.index.load_main_index_meta(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Optional[dict]

archivebox.index.parse_links_from_source(source_path: str, root_url: Optional[str] = None, parser: str = 'auto') → Tuple[List[archivebox.index.schema.Link], List[archivebox.index.schema.Link]]

archivebox.index.fix_duplicate_links_in_index(snapshots: django.db.models.query.QuerySet, links: Iterable[archivebox.index.schema.Link]) → Iterable[archivebox.index.schema.Link]

Given a list of in-memory Links, dedupe and merge them with any conflicting Snapshots in the DB.

archivebox.index.dedupe_links(snapshots: django.db.models.query.QuerySet, new_links: List[archivebox.index.schema.Link]) → List[archivebox.index.schema.Link]

The validation of links happened at a different stage. This method will focus on actual deduplication and timestamp fixing.

archivebox.index.write_link_details(link: archivebox.index.schema.Link, out_dir: Optional[str] = None, skip_sql_index: bool = False) → None

archivebox.index.load_link_details(link: archivebox.index.schema.Link, out_dir: Optional[str] = None) → archivebox.index.schema.Link

check for an existing link archive in the given directory, and load+merge it into the given link dict

archivebox.index.q_filter(snapshots: django.db.models.query.QuerySet, filter_patterns: List[str], filter_type: str = 'exact') → django.db.models.query.QuerySet

archivebox.index.search_filter(snapshots: django.db.models.query.QuerySet, filter_patterns: List[str], filter_type: str = 'search') → django.db.models.query.QuerySet

archivebox.index.snapshot_filter(snapshots: django.db.models.query.QuerySet, filter_patterns: List[str], filter_type: str = 'exact') → django.db.models.query.QuerySet

archivebox.index.get_indexed_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

indexed links without checking archive status or data directory validity

archivebox.index.get_archived_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

indexed links that are archived with a valid data directory

archivebox.index.get_unarchived_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

indexed links that are unarchived with no data directory or an empty data directory

archivebox.index.get_present_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

dirs that actually exist in the archive/ folder

archivebox.index.get_valid_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

dirs with a valid index matched to the main index and archived content

archivebox.index.get_invalid_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

dirs that are invalid for any reason: corrupted/duplicate/orphaned/unrecognized

archivebox.index.get_duplicate_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

dirs that conflict with other directories that have the same link URL or timestamp

archivebox.index.get_orphaned_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

dirs that contain a valid index but aren’t listed in the main index

archivebox.index.get_corrupted_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

dirs that don’t contain a valid index and aren’t listed in the main index

archivebox.index.get_unrecognized_folders(snapshots, out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Dict[str, Optional[archivebox.index.schema.Link]]

dirs that don’t contain recognizable archive data and aren’t listed in the main index

archivebox.index.is_valid(link: archivebox.index.schema.Link) → bool

archivebox.index.is_corrupt(link: archivebox.index.schema.Link) → bool

archivebox.index.is_archived(link: archivebox.index.schema.Link) → bool

archivebox.index.is_unarchived(link: archivebox.index.schema.Link) → bool

archivebox.index.fix_invalid_folder_locations(out_dir: pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/archivebox/checkouts/v0.6.0/docs')) → Tuple[List[str], List[str]]