diff --git a/docs/cli.html b/docs/cli.html new file mode 100644 index 0000000..f7de743 --- /dev/null +++ b/docs/cli.html @@ -0,0 +1,142 @@ + + + + + + +osfexport.cli API documentation + + + + + + + + + + + +
+
+
+

Module osfexport.cli

+
+
+
+
+
+
+
+
+

Functions

+
+
+def prompt_pat(project_id='', usetest=False) +
+
+
+ +Expand source code + +
def prompt_pat(project_id='', usetest=False):
+    """
+    Ask for a PAT if exporting a single project or all projects a user has.
+
+    Parameters
+    -------------
+        project_id: str
+            ID of a single project to export.
+            If one provided then ask for a PAT.
+        usetest: bool
+            Flag to indicate whether to use the test/production API server.
+
+    Returns
+    -----------------
+        pat: str
+            Personal Access Token to use to authorise a user.
+
+    Raises
+    -------------------
+        HTTPError, URLError - passed on from is_public method.
+    """
+
+    if usetest:
+        api_host = API_HOST_TEST
+    else:
+        api_host = API_HOST_PROD
+
+    if not project_id:
+        pat = click.prompt(
+            'Please enter your PAT to export all your projects',
+            type=str,
+            hide_input=True
+        )
+    elif not exporter.is_public(f'{api_host}/nodes/{project_id}/'):
+        pat = click.prompt(
+            'Please enter your PAT to export this private project',
+            type=str,
+            hide_input=True
+        )
+    else:
+        pat = ''
+
+    return pat
+
+

Ask for a PAT if exporting a single project or all projects a user has.

+

Parameters

+
project_id: str
+    ID of a single project to export.
+    If one provided then ask for a PAT.
+usetest: bool
+    Flag to indicate whether to use the test/production API server.
+
+

Returns

+
pat: str
+    Personal Access Token to use to authorise a user.
+
+

Raises

+
HTTPError, URLError - passed on from is_public method.
+
+
+
+
+
+
+
+ +
+ + + diff --git a/docs/exporter.html b/docs/exporter.html new file mode 100644 index 0000000..bbcd7c4 --- /dev/null +++ b/docs/exporter.html @@ -0,0 +1,1423 @@ + + + + + + +osfexport.exporter API documentation + + + + + + + + + + + +
+
+
+

Module osfexport.exporter

+
+
+
+
+
+
+
+
+

Functions

+
+
+def call_api(url,
pat,
method='GET',
per_page=100,
filters={},
is_json=True,
usetest=False,
max_tries=5)
+
+
+
+ +Expand source code + +
def call_api(
+        url, pat, method='GET', per_page=100, filters={}, is_json=True,
+        usetest=False, max_tries=5):
+    """Call OSF v2 API methods.
+
+    Parameters
+    ----------
+    url: str
+        URL to API method/resource/query.
+    method: str
+        HTTP method for the request.
+    pat: str
+        Personal Access Token to authorise a user with.
+    per_page: int
+        Number of items to include in a JSON page for API responses.
+        The maximum size is 100.
+    filters: dict
+        Dictionary of query parameters to filter results with.
+
+        Example Input: {'category': 'project', 'title': 'ttt'}
+        Example Query String: ?filter[category]=project&filter[title]=ttt
+    is_json: bool
+        If true, set API version to get correct API responses.
+    usetest: bool
+        If True, use fixed delay of 0.1 seconds for tests.
+        If False, use a random delay between [1, 60] seconds between requests.
+        This spaces out requests over time to give the API chance to recover.
+    max_tries: int
+        Number of attempts to make before raising a 429 error. Default is 5, Limit is 7.
+
+    Throws
+    -------------
+        HTTPError - 429 error if we can't connect to the API after retries.
+        Other errors are thrown immediately.
+
+        URLError - failed to connect to OSF API
+
+    Returns
+    ----------
+        result: HTTPResponse
+            Response to the request from the API.
+    """
+    if (filters or per_page) and method == 'GET':
+        query_string = '&'.join([f'filter[{key}]={value}'
+                                 for key, value in filters.items()
+                                 if not isinstance(value, dict)])
+        if per_page:
+            query_string += f'&page[size]={per_page}'
+        url = f'{url}?{query_string}'
+
+    request = webhelper.Request(url, method=method)
+    request.add_header('Authorization', f'Bearer {pat}')
+
+    version = importlib.metadata.version("osfexport")
+    request.add_header('User-Agent', f'osfexport/{version} (Python)')
+
+    # Pin API version so that JSON has correct format
+    API_VERSION = '2.20'
+    if is_json:
+        request.add_header(
+            'Accept',
+            f'application/vnd.api+json;version={API_VERSION}'
+        )
+
+    if max_tries > 7:
+        max_tries = 7  # Cap retries to reduce requests sent and max delay time
+
+    # Retry requests if we get 429 errors
+    try_count = 0
+    result = None
+    while try_count < max_tries and result is None:
+        try:
+            result = webhelper.urlopen(request)
+        except HTTPError as e:
+            # Other error codes tell us directly something is wrong
+            if e.code == 429:
+                if not usetest:
+                    # Wait longer between requests to give API more recovery time
+                    # Wait random periods to avoid hammering requests all at once
+                    min_wait = (try_count+1)**2
+                    time.sleep(random.uniform(min_wait, 60))
+                else:
+                    time.sleep(0.5)  # Wait constant time for tests
+                try_count += 1
+            else:
+                raise e
+    if result is None:
+        raise HTTPError(
+            url=url,
+            code=429,
+            msg="Too many requests to the OSF API.",
+            hdrs=request.headers,
+            fp=None
+        )
+    return result
+
+

Call OSF v2 API methods.

+

Parameters

+
+
url : str
+
URL to API method/resource/query.
+
method : str
+
HTTP method for the request.
+
pat : str
+
Personal Access Token to authorise a user with.
+
per_page : int
+
Number of items to include in a JSON page for API responses. +The maximum size is 100.
+
filters : dict
+
+

Dictionary of query parameters to filter results with.

+

Example Input: {'category': 'project', 'title': 'ttt'} +Example Query String: ?filter[category]=project&filter[title]=ttt

+
+
is_json : bool
+
If true, set API version to get correct API responses.
+
usetest : bool
+
If True, use fixed delay of 0.1 seconds for tests. +If False, use a random delay between [1, 60] seconds between requests. +This spaces out requests over time to give the API chance to recover.
+
max_tries : int
+
Number of attempts to make before raising a 429 error. Default is 5, Limit is 7.
+
+

Throws

+
HTTPError - 429 error if we can't connect to the API after retries.
+Other errors are thrown immediately.
+
+URLError - failed to connect to OSF API
+
+

Returns

+
result: HTTPResponse
+    Response to the request from the API.
+
+
+
+def explore_file_tree(curr_link, pat, dryrun=True) +
+
+
+ +Expand source code + +
def explore_file_tree(curr_link, pat, dryrun=True):
+    """Explore and get names of files stored in OSF.
+
+    Parameters
+    ----------
+    curr_link: string
+        URL/name to use to get real/mock files and folders.
+    pat: string
+        Personal Access Token to authorise a user.
+    dryrun: bool
+        Flag to indicate whether to use mock JSON files or real API calls.
+
+    Returns
+    ----------
+        files_found: list[str]
+            List of file paths found in the project."""
+
+    FILE_FILTER = {
+        'kind': 'file'
+    }
+    FOLDER_FILTER = {
+        'kind': 'folder'
+    }
+    per_page = 100
+
+    files_found = []
+
+    is_last_page_folders = False
+    while not is_last_page_folders:
+        if dryrun:
+            folders = MockAPIResponse.read(f"{curr_link}_folder")
+        else:
+            folders = json.loads(
+                call_api(
+                    curr_link, pat,
+                    per_page=per_page, filters=FOLDER_FILTER
+                ).read()
+            )
+
+        # Find deepest subfolders first to avoid missing files
+        try:
+            for folder in folders['data']:
+                links = folder['relationships']['files']['links']
+                link = links['related']['href']
+                files_found += explore_file_tree(link, pat, dryrun=dryrun)
+        except KeyError:
+            pass
+
+        # For each folder, loop through pages for its files
+        is_last_page_files = False
+        while not is_last_page_files:
+            if dryrun:
+                files = MockAPIResponse.read(f"{curr_link}_files")
+            else:
+                files = json.loads(
+                    call_api(
+                        curr_link, pat,
+                        per_page=per_page, filters=FILE_FILTER
+                    ).read()
+                )
+            try:
+                for file in files['data']:
+                    size = file['attributes']['size']
+                    size_mb = size / (1024 ** 2)  # Convert bytes to MB
+                    data = (
+                        file['attributes']['materialized_path'],
+                        str(round(size_mb, 2)),
+                        file['links']['download']
+                    )
+                    files_found.append(data)
+            except KeyError:
+                pass
+            curr_link = files['links']['next']
+            if curr_link is None:
+                is_last_page_files = True
+
+        curr_link = folders['links']['next']
+        if curr_link is None:
+            is_last_page_folders = True
+
+    return files_found
+
+

Explore and get names of files stored in OSF.

+

Parameters

+
+
curr_link : string
+
URL/name to use to get real/mock files and folders.
+
pat : string
+
Personal Access Token to authorise a user.
+
dryrun : bool
+
Flag to indicate whether to use mock JSON files or real API calls.
+
+

Returns

+
files_found: list[str]
+    List of file paths found in the project.
+
+
+
+def explore_wikis(link, pat, dryrun=True) +
+
+
+ +Expand source code + +
def explore_wikis(link, pat, dryrun=True):
+    """Get wiki contents for a particular project.
+
+    Parameters:
+    -------------
+    link: str
+        URL to project wikis or name of wikis field to access mock JSON.
+    pat: str
+        Personal Access Token to authenticate a user with.
+    dryrun: bool
+        Flag to indicate whether to use mock JSON files or real API calls.
+
+    Returns
+    ---------------
+    wikis: List of JSON representing wikis for a project."""
+
+    wiki_content = {}
+    is_last_page = False
+    if dryrun:
+        wikis = MockAPIResponse.read('wikis')
+    else:
+        wikis = json.loads(
+            call_api(link, pat).read()
+        )
+
+    while not is_last_page:
+        for wiki in wikis['data']:
+            if dryrun:
+                content = MockAPIResponse.read(wiki['attributes']['name'])
+            else:
+                # Decode Markdown content to allow parsing later on
+                content = call_api(
+                    wiki['links']['download'], pat=pat, is_json=False
+                ).read().decode('utf-8')
+            wiki_content[wiki['attributes']['name']] = content
+
+        # Go to next page of wikis if pagination applied
+        # so that we don't miss wikis
+        link = wikis['links']['next']
+        if not link:
+            is_last_page = True
+        else:
+            if dryrun:
+                wikis = MockAPIResponse.read(link)
+            else:
+                wikis = json.loads(
+                    call_api(link, pat).read()
+                )
+
+    return wiki_content
+
+

Get wiki contents for a particular project.

+

Parameters:

+

link: str +URL to project wikis or name of wikis field to access mock JSON. +pat: str +Personal Access Token to authenticate a user with. +dryrun: bool +Flag to indicate whether to use mock JSON files or real API calls.

+

Returns

+

wikis: List of JSON representing wikis for a project.

+
+
+def extract_project_id(url) +
+
+
+ +Expand source code + +
def extract_project_id(url):
+    """Extract project ID from a given OSF project URL.
+
+    Parameters
+    ----------
+    url: str
+        URL of the OSF project which should contain the project ID. E.g.:
+        - Full URL with parameters (https://osf.io/xyz/?param=value)
+        - API URL (https://api.test.osf.io/v2/nodes/xyz)
+        - Just the ID (xyz)
+        - Empty string
+
+    Returns
+    -------
+    str
+        Project ID extracted from the URL.
+    """
+
+    if not url:
+        return ''
+
+    parts = url.strip("/").split("/")
+    # Handle case of just ID provided
+    if len(parts) == 1:
+        return parts[0]
+
+    # API URLs are of form /nodes/id/...
+    if 'nodes' in parts:
+        idx = parts.index('nodes')
+        if idx + 1 < len(parts):
+            return parts[idx + 1]
+
+    # For regular URLs, extract ID from last path component before query params
+    if '?' in parts[-1]:
+        return parts[-2]
+    else:
+        return parts[-1]
+
+

Extract project ID from a given OSF project URL.

+

Parameters

+
+
url : str
+
URL of the OSF project which should contain the project ID. E.g.: +- Full URL with parameters (https://osf.io/xyz/?param=value) +- API URL (https://api.test.osf.io/v2/nodes/xyz) +- Just the ID (xyz) +- Empty string
+
+

Returns

+
+
str
+
Project ID extracted from the URL.
+
+
+
+def get_affiliated_institutions(project, **kwargs) +
+
+
+ +Expand source code + +
def get_affiliated_institutions(project, **kwargs):
+    dryrun = kwargs.pop('dryrun', True)
+    key = kwargs.pop('key', 'affiliated_institutions')
+    pat = kwargs.pop('pat', '')
+    if not dryrun:
+        # Check relationship exists and can get link to linked data
+        # Otherwise just pass a placeholder dict
+        try:
+            link = project['relationships'][key]['links']['related']['href']
+            json_data = json.loads(
+                call_api(
+                    link, pat,
+                    filters=URL_FILTERS.get(key, {})
+                ).read()
+            )
+        except KeyError:
+            json_data = {'data': None}
+    else:
+        json_data = MockAPIResponse.read(key)
+    values = []
+    for item in json_data['data']:
+        values.append(item['attributes']['name'])
+    values = ', '.join(values)
+    if not values:
+        values = 'NA'
+    return values
+
+
+
+
+def get_category(project, **kwargs) +
+
+
+ +Expand source code + +
def get_category(project, **kwargs):
+    """Get category from a project dictionary"""
+
+    # Define nice representations of categories if needed
+    CATEGORY_STRS = {
+        '': 'Uncategorized',
+        'methods and measures': 'Methods and Measures'
+    }
+    if project['attributes']['category'] in CATEGORY_STRS:
+        return CATEGORY_STRS[project['attributes']['category']]
+    else:
+        return project['attributes']['category'].title()
+
+

Get category from a project dictionary

+
+
+def get_contributors(project, **kwargs) +
+
+
+ +Expand source code + +
def get_contributors(project, **kwargs):
+    """Get contributors from a project dictionary
+
+    Parameters
+    --------------
+        dryrun: bool
+            If True, use test OSF API, otherwise use real API for calls
+        pat: str
+            Personal Access Token to authenticate users with.
+    """
+
+    dryrun = kwargs.pop('dryrun', True)
+    key = kwargs.pop('key', 'contributors')
+    pat = kwargs.pop('pat', '')
+    if not dryrun:
+        # Check relationship exists and can get link to linked data
+        # Otherwise just pass a placeholder dict
+        try:
+            link = project['relationships'][key]['links']['related']['href']
+            json_data = json.loads(
+                call_api(
+                    link, pat,
+                    filters=URL_FILTERS.get(key, {})
+                ).read()
+            )
+        except KeyError:
+            json_data = {'data': None}
+    else:
+        json_data = MockAPIResponse.read(key)
+    values = []
+    for item in json_data['data']:
+        values.append((
+            item['embeds']['users']['data']
+            ['attributes']['full_name'],
+            item['attributes']['bibliographic'],
+            item['embeds']['users']['data']['links']['html']
+        ))
+    return values
+
+

Get contributors from a project dictionary

+

Parameters

+
dryrun: bool
+    If True, use test OSF API, otherwise use real API for calls
+pat: str
+    Personal Access Token to authenticate users with.
+
+
+
+def get_host(is_test) +
+
+
+ +Expand source code + +
def get_host(is_test):
+    """Get API host based on flag.
+
+    Parameters
+    ----------
+    is_test: bool
+        If True, return test API host, otherwise return production host.
+
+    Returns
+    -------
+    str
+        API host URL for the test site or production site.
+    """
+
+    return API_HOST_TEST if is_test else API_HOST_PROD
+
+

Get API host based on flag.

+

Parameters

+
+
is_test : bool
+
If True, return test API host, otherwise return production host.
+
+

Returns

+
+
str
+
API host URL for the test site or production site.
+
+
+
+def get_identifiers(project, **kwargs) +
+
+
+ +Expand source code + +
def get_identifiers(project, **kwargs):
+    dryrun = kwargs.pop('dryrun', True)
+    key = kwargs.pop('key', 'identifiers')
+    pat = kwargs.pop('pat', '')
+    if not dryrun:
+        # Check relationship exists and can get link to linked data
+        # Otherwise just pass a placeholder dict
+        try:
+            link = project['relationships'][key]['links']['related']['href']
+            json_data = json.loads(
+                call_api(
+                    link, pat,
+                    filters=URL_FILTERS.get(key, {})
+                ).read()
+            )
+        except KeyError:
+            json_data = {'data': None}
+    else:
+        json_data = MockAPIResponse.read(key)
+    values = []
+    for item in json_data['data']:
+        values.append(item['attributes']['value'])
+    values = ', '.join(values)
+    return values
+
+
+
+
+def get_license(project, **kwargs) +
+
+
+ +Expand source code + +
def get_license(project, **kwargs):
+    dryrun = kwargs.pop('dryrun', True)
+    key = kwargs.pop('key', 'license')
+    pat = kwargs.pop('pat', '')
+    if not dryrun:
+        # Check relationship exists and can get link to linked data
+        # Otherwise just pass a placeholder dict
+        try:
+            link = project['relationships'][key]['links']['related']['href']
+            json_data = json.loads(
+                call_api(
+                    link, pat,
+                    filters=URL_FILTERS.get(key, {})
+                ).read()
+            )
+        except KeyError:
+            json_data = {'data': None}
+    else:
+        json_data = MockAPIResponse.read(key)
+    if json_data['data'] is not None:
+        return json_data['data']['attributes']['name']
+    else:
+        return None
+
+
+
+
+def get_nodes(pat, page_size=100, dryrun=False, project_id='', usetest=False) +
+
+
+ +Expand source code + +
def get_nodes(pat, page_size=100, dryrun=False, project_id='', usetest=False):
+    """Pull and list projects for a user from the OSF.
+
+    Parameters
+    ----------
+    pat: str
+        Personal Access Token to authorise a user with.
+    page_size: int
+        How many nodes to put onto a page. Default is 100.
+        Possible range is 1-100
+    dryrun: bool
+        If True, use test data from JSON stubs to mock API calls.
+    project_id: str
+        Optional ID for a specific OSF project to export.
+    usetest: bool
+        If True, use test API host, otherwise use production host.
+
+    Returns
+    ----------
+        projects: list[dict]
+            List of all project objects found
+        root_nodes: list[int]
+            List of indexes for root nodes in projects list.
+            These are the nodes to make PDFs for and start from in PDFs.
+    """
+
+    # Set start link and page size filter based on flags
+    api_host = get_host(usetest)
+    node_filter = {}
+    if not dryrun:
+        if project_id:
+            start = f'{api_host}/nodes/{project_id}/'
+        else:
+            start = f'{api_host}/users/me/nodes/'
+            node_filter = {
+                'parent': ''
+            }
+    else:
+        page_size = 4  # Nodes found are hardcoded for --dryrun
+        if project_id:
+            start = project_id
+        else:
+            start = 'nodes'
+
+    results = paginate_json_result(
+        start, get_project_data, dryrun=dryrun, usetest=usetest,
+        pat=pat, filters=node_filter, project_id=project_id, per_page=page_size
+    )
+    if len(results) > 0:
+        l1, l2 = zip(*list(results))
+    else:
+        l1, l2 = (), ()
+    projects = [item for sublist in l1 for item in sublist]
+
+    # After pagination we get indexes of root nodes local to each page
+    # We need to convert these to global indexes before merging the list
+    page_idx = -1
+    for page in l2:
+        page_idx += 1
+        for idx, n in enumerate(page):
+            global_node_idx = page_size*page_idx + n
+            page[idx] = global_node_idx
+    root_nodes = [item for sublist in l2 for item in sublist]
+
+    return projects, root_nodes
+
+

Pull and list projects for a user from the OSF.

+

Parameters

+
+
pat : str
+
Personal Access Token to authorise a user with.
+
page_size : int
+
How many nodes to put onto a page. Default is 100. +Possible range is 1-100
+
dryrun : bool
+
If True, use test data from JSON stubs to mock API calls.
+
project_id : str
+
Optional ID for a specific OSF project to export.
+
usetest : bool
+
If True, use test API host, otherwise use production host.
+
+

Returns

+
projects: list[dict]
+    List of all project objects found
+root_nodes: list[int]
+    List of indexes for root nodes in projects list.
+    These are the nodes to make PDFs for and start from in PDFs.
+
+
+
+def get_project_data(nodes, **kwargs) +
+
+
+ +Expand source code + +
def get_project_data(nodes, **kwargs):
+    """Pull and list projects for a specific JSON API response page.
+
+    Parameters
+    ----------
+    pat: str
+        Personal Access Token to authorise a user with.
+    dryrun: bool
+        If True, use test data from JSON stubs to mock API calls.
+    project_id: str
+        Optional ID for a specific OSF project to export.
+    usetest: bool
+        If True, use test API host, otherwise use production host.
+
+    Returns
+    ----------
+        projects: list[dict]
+            List of dictionaries representing projects.
+    """
+
+    pat = kwargs.pop('pat', '')
+    dryrun = kwargs.pop('dryrun', False)
+    usetest = kwargs.pop('usetest', False)
+    project_id = kwargs.pop('project_id', '')
+
+    api_host = get_host(usetest)
+
+    if not dryrun and project_id:
+        nodes = {'data': [nodes['data']]}
+    elif project_id:
+        # Put data into same format as if multiple nodes found
+        nodes = {'data': [MockAPIResponse.read(project_id)['data']]}
+
+    projects = []
+    root_nodes = []  # Track indexes of start nodes for PDFs
+    added_node_ids = set()  # Track added node IDs to avoid duplicates
+
+    # Dispatch table used to define how to process JSON
+    # Add new field by giving name and function
+    fields = {
+        'metadata': {
+            'title': lambda project, **kwargs: project['attributes']['title'],
+            'id': lambda project, **kwargs: project['id'],
+            'url': lambda project, **kwargs: project['links']['html'],
+            'description': lambda project, **kwargs: project['attributes']['description'],
+            # timestamps are rendered as yyyy-mm-dd hour:minute UTC (24hr)
+            'date_created': lambda project, **kwargs: datetime.datetime.fromisoformat(
+                    project['attributes']['date_created']
+                ).astimezone(
+                    datetime.timezone.utc
+                ).strftime(
+                    '%Y-%m-%d %H:%M %Z'
+                ),
+            'date_modified': lambda project, **kwargs: datetime.datetime.fromisoformat(
+                    project['attributes']['date_modified']
+                ).astimezone(
+                    datetime.timezone.utc
+                ).strftime(
+                    '%Y-%m-%d %H:%M %Z'
+                ),
+            'public': lambda project, **kwargs: project['attributes']['public'],
+            'category': get_category,
+            'tags': get_tags,
+            'resource_type': lambda project, **kwargs: 'NA',
+            'resource_lang': lambda project, **kwargs: 'NA',
+            'affiliated_institutions': get_affiliated_institutions,
+            'identifiers': get_identifiers,
+            'license': get_license,
+            'subjects': get_subjects,
+            'funders': lambda project, **kwargs: [],
+        },
+        'contributors': get_contributors
+    }
+
+    for idx, project in enumerate(nodes['data']):
+        try:
+            if project['id'] in added_node_ids:
+                continue
+            else:
+                added_node_ids.add(project['id'])
+
+            project_data = {
+                'metadata': {}
+            }
+            for field in fields['metadata']:
+                project_data['metadata'][field] = fields['metadata'][field](
+                    project, dryrun=dryrun, key=field, pat=pat
+                )
+            project_data['contributors'] = fields['contributors'](
+                project, dryrun=dryrun, key='contributors', pat=pat
+            )
+
+            # TODO: split into function
+            # Resource type/lang/funding info share specific endpoint
+            # that isn't linked to in user nodes' responses
+            if dryrun:
+                metadata = MockAPIResponse.read('custom_metadata')
+            else:
+                metadata = json.loads(call_api(
+                    f"{api_host}/custom_item_metadata_records/{project['id']}/",
+                    pat
+                ).read())
+            metadata = metadata['data']['attributes']
+            resource_type = metadata['resource_type_general']
+            resource_lang = metadata['language']
+            project_data['metadata']['resource_type'] = resource_type
+            project_data['metadata']['resource_lang'] = resource_lang
+            for funder in metadata['funders']:
+                project_data['metadata']['funders'].append(funder)
+            # =========
+
+            relations = project['relationships']
+
+            # Get list of files in project
+            if dryrun:
+                link = 'root'
+                use_mocks = True
+            else:
+                link = relations['files']['links']['related']['href']
+                link += 'osfstorage/'  # ID for OSF Storage
+                use_mocks = False
+            project_data['files'] = explore_file_tree(link, pat, dryrun=use_mocks)
+
+            project_data['wikis'] = explore_wikis(
+                f'{api_host}/nodes/{project['id']}/wikis/',
+                pat=pat, dryrun=dryrun
+            )
+
+            # Check if parent info has been passed down to save effort
+            # If not then search for links to parent
+            try:
+                project_data['parent'] = project['parent']
+            except KeyError:
+                project_data['parent'] = None
+
+                # In general, start nodes for PDFs have no parents
+                if 'links' not in project['relationships']['parent']:
+                    root_nodes.append(idx)
+                elif project_data['parent'] is None:
+                    parent_link = project['relationships']['parent'][
+                        'links']['related']['href']
+                    try:
+                        if not dryrun:
+                            parent = json.loads(
+                                call_api(
+                                    parent_link,
+                                    pat=pat,
+                                    is_json=True
+                                ).read()
+                            )
+                        else:
+                            parent = MockAPIResponse.read(parent_link)
+                        project_data['parent'] = (
+                            parent['data']['attributes']['title'],
+                            parent['data']['links']['html']
+                        )
+                    except (HTTPError, ValueError):
+                        logging.warning(
+                            f"Warning: Parent of {project_data['metadata']['title']} is private."
+                        )
+                        logging.warning(
+                            "Try to give a PAT beforehand using the --pat flag."
+                        )
+
+            # Projects specified by ID to export also count as start nodes for PDFs
+            # This will be the first node in list of root nodes
+            if project_data['metadata']['id'] == project_id and 0 not in root_nodes:
+                root_nodes.append(idx)
+
+            def get_children(json_page, **kwargs):
+                children = []
+                for child in json_page['data']:
+                    child['parent'] = [
+                        project_data['metadata']['title'],
+                        project_data['metadata']['url']
+                    ]
+                    children.append(child['id'])
+                    nodes['data'].append(child)  # Add to list of nodes to search
+                return children
+
+            children_link = relations['children']['links']['related']['href']
+            children = list(paginate_json_result(
+                children_link, dryrun=dryrun, pat=pat, action=get_children
+            ))
+            newlist = [item for sublist in children for item in sublist]
+            project_data['children'] = newlist
+
+            projects.append(project_data)
+        except (HTTPError, KeyError) as e:
+            if isinstance(e, HTTPError):
+                if e.code == 429:
+                    raise e
+                logging.warning(f"Warning: A project failed to export: {e.code}")
+            else:
+                logging.warning("Warning: A project failed to export: Unexpected API response.")
+            logging.warning("Continuing with exporting other projects...")
+
+    return projects, root_nodes
+
+

Pull and list projects for a specific JSON API response page.

+

Parameters

+
+
pat : str
+
Personal Access Token to authorise a user with.
+
dryrun : bool
+
If True, use test data from JSON stubs to mock API calls.
+
project_id : str
+
Optional ID for a specific OSF project to export.
+
usetest : bool
+
If True, use test API host, otherwise use production host.
+
+

Returns

+
projects: list[dict]
+    List of dictionaries representing projects.
+
+
+
+def get_subjects(project, **kwargs) +
+
+
+ +Expand source code + +
def get_subjects(project, **kwargs):
+    dryrun = kwargs.pop('dryrun', True)
+    key = kwargs.pop('key', 'subjects')
+    pat = kwargs.pop('pat', '')
+    if not dryrun:
+        # Check relationship exists and can get link to linked data
+        # Otherwise just pass a placeholder dict
+        try:
+            link = project['relationships'][key]['links']['related']['href']
+            json_data = json.loads(
+                call_api(
+                    link, pat,
+                    filters=URL_FILTERS.get(key, {})
+                ).read()
+            )
+        except KeyError:
+            raise KeyError()  # Subjects should have a href link
+    else:
+        json_data = MockAPIResponse.read(key)
+    values = []
+    for item in json_data['data']:
+        values.append(item['attributes']['text'])
+    values = ', '.join(values)
+    return values
+
+
+
+
+def get_tags(project, **kwargs) +
+
+
+ +Expand source code + +
def get_tags(project, **kwargs):
+    """Get tags from a project dictionary"""
+
+    if project['attributes']['tags']:
+        return ', '.join(project['attributes']['tags'])
+    else:
+        return 'NA'
+
+

Get tags from a project dictionary

+
+
+def is_public(url) +
+
+
+ +Expand source code + +
def is_public(url):
+    """Return boolean to indicate if a URL is public (True) or not (False).
+    This is mainly used for checking if a project is publicly accessible.
+
+    Parameters
+    ------------
+    url: str
+        The URL to test.
+
+    Returns
+    ----------------
+        is_public: bool
+            Whether we can access the URL without a PAT (i.e. status code 200)
+
+    Raises
+    -------------------
+        HTTPError, URLError
+            If we get a HTTP error code that isn't 401/403, or a connection error.
+    """
+
+    try:
+        result = call_api(
+            url, pat='', method='GET'
+        ).status
+    except (HTTPError, URLError) as e:
+        # Don't raise error if we get a HTTP error with certain codes
+        valid_error_codes = [401, 403]
+        if isinstance(e, HTTPError) and e.code in valid_error_codes:
+            result = e.code
+        else:
+            raise e
+    is_public = result == 200
+    return is_public
+
+

Return boolean to indicate if a URL is public (True) or not (False). +This is mainly used for checking if a project is publicly accessible.

+

Parameters

+
+
url : str
+
The URL to test.
+
+

Returns

+
is_public: bool
+    Whether we can access the URL without a PAT (i.e. status code 200)
+
+

Raises

+
HTTPError, URLError
+    If we get a HTTP error code that isn't 401/403, or a connection error.
+
+
+
+def paginate_json_result(start, action, fail_on_first=True, **kwargs) +
+
+
+ +Expand source code + +
def paginate_json_result(start, action, fail_on_first=True, **kwargs):
+    """Loop through paginated JSON responses and perform an action on each.
+
+    Parameters
+    -------------
+    start: str
+        Link to start looping from
+    action: func
+        Takes in found JSON page and returns a result
+    per_page: int
+        How many items to include on one page. Default is 100.
+        Valid range is from 1-1000.
+    filters: dict
+        Optional key-value dict to filter queries by.
+    is_json:
+        If JSON response expected, add header to specify JSON format.
+    pat: str
+        Personal Access Token to authorise a user.
+    dryrun: bool
+        Flag for whether mock JSON or real API will be used.
+    **kwargs
+        Extra keyword args to pass down to action and call_api.
+
+    Returns
+    ------------------
+    results: deque
+        Queue of results per page
+    
+    Throws
+    ------------------
+    HTTPError, URLError - non-429 HTTP errors which indicate a problem
+    """
+
+    next_link = start
+    is_last_page = False
+    is_first_item = True  # Want to throw error if very first item fails
+    results = deque()
+    per_page = kwargs.pop('per_page', 100)
+    filters = kwargs.pop('filters', {})
+    is_json = kwargs.pop('is_json', True)
+    pat = kwargs.get('pat', '')
+    dryrun = kwargs.get('dryrun', False)
+    while not is_last_page:
+        try:
+            if not dryrun:
+                curr_page = call_api(
+                    next_link, pat, per_page=per_page, filters=filters,
+                    is_json=is_json)
+                # Catch error if call_api is replaced with mock in tests
+                try:
+                    curr_page = curr_page.read()
+                    if is_json:
+                        curr_page = json.loads(curr_page)
+                except AttributeError:
+                    pass
+            else:
+                curr_page = MockAPIResponse.read(next_link)
+            results.append(action(curr_page, **kwargs))
+        except HTTPError as e:
+            if fail_on_first and is_first_item or e.code == 429:
+                raise e
+            else:
+                logging.warning("Warning: Couldn't parse JSON page, skipping to next page...")
+        # Stop if no next link found
+        try:
+            next_link = curr_page['links']['next']
+            is_last_page = not next_link
+        except (KeyError, UnboundLocalError):
+            is_last_page = True
+    return results
+
+

Loop through paginated JSON responses and perform an action on each.

+

Parameters

+
+
start : str
+
Link to start looping from
+
action : func
+
Takes in found JSON page and returns a result
+
per_page : int
+
How many items to include on one page. Default is 100. +Valid range is from 1-1000.
+
filters : dict
+
Optional key-value dict to filter queries by.
+
is_json:
+
If JSON response expected, add header to specify JSON format.
+
pat : str
+
Personal Access Token to authorise a user.
+
dryrun : bool
+
Flag for whether mock JSON or real API will be used.
+
**kwargs
+
Extra keyword args to pass down to action and call_api.
+
+

Returns

+
+
results : deque
+
Queue of results per page
+
+

Throws

+

HTTPError, URLError - non-429 HTTP errors which indicate a problem

+
+
+
+
+

Classes

+
+
+class MockAPIResponse +
+
+
+ +Expand source code + +
class MockAPIResponse:
+    """
+    Simulate OSF API response for testing purposes.
+
+    Attributes
+    ----------------
+
+    JSON_FILES: static
+        Key-value dictionary of IDs and paths to stub JSON files.
+        These are used to generate mock responses to API calls.
+    
+    MARKDOWN_FILES: static
+        Key-value dictionary of IDs and paths to stub Markdown files.
+        These are used to generate mock responses to API calls to get Wiki data.
+    """
+
+    JSON_FILES = {
+        'nodes': os.path.join(
+            STUBS_DIR, 'nodestubs.json'),
+        'nodes2': os.path.join(
+            STUBS_DIR, 'nodestubs2.json'),
+        'x': os.path.join(
+            STUBS_DIR, 'singlenode.json'),
+        'a': os.path.join(
+            STUBS_DIR, 'asingle.json'),
+        'affiliated_institutions': os.path.join(
+            STUBS_DIR, 'institutionstubs.json'),
+        'contributors': os.path.join(
+            STUBS_DIR, 'contributorstubs.json'),
+        'identifiers': os.path.join(
+            STUBS_DIR, 'doistubs.json'),
+        'custom_metadata': os.path.join(
+            STUBS_DIR, 'custommetadatastub.json'),
+        'root_folder': os.path.join(
+            STUBS_DIR, 'files', 'rootfolders.json'),
+        'root_files': os.path.join(
+            STUBS_DIR, 'files', 'rootfiles.json'),
+        'tf1_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf1folders.json'),
+        'tf1-2_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf1folders-2.json'),
+        'tf1-2_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-folders.json'),
+        'tf1_files': os.path.join(
+            STUBS_DIR, 'files', 'tf1files.json'),
+        'tf2_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf2folders.json'),
+        'tf2-second_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-folders.json'),
+        'tf2_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2files.json'),
+        'tf2-second_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-files.json'),
+        'tf2-second-2_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-files-2.json'),
+        'license': os.path.join(
+            STUBS_DIR, 'licensestub.json'),
+        'subjects': os.path.join(
+            STUBS_DIR, 'subjectsstub.json'),
+        'wikis': os.path.join(
+            STUBS_DIR, 'wikis', 'wikistubs.json'),
+        'wikis2': os.path.join(
+            STUBS_DIR, 'wikis', 'wikis2stubs.json'),
+        'x-child-1': os.path.join(
+            STUBS_DIR, 'components', 'x-child-1.json'),
+        'x-child-2': os.path.join(
+            STUBS_DIR, 'components', 'x-child-2.json'),
+        'empty-children': os.path.join(
+            STUBS_DIR, 'components', 'empty-children.json'),
+    }
+
+    MARKDOWN_FILES = {
+        'helloworld': os.path.join(
+            STUBS_DIR, 'wikis', 'helloworld.md'),
+        'home': os.path.join(
+            STUBS_DIR, 'wikis', 'home.md'),
+        'anotherone': os.path.join(
+            STUBS_DIR, 'wikis', 'anotherone.md'),
+    }
+
+    @staticmethod
+    def read(field):
+        """Get mock response for a field.
+
+        Parameters
+        -----------
+            field: str
+                ID associated to a JSON or Markdown mock file.
+                Available fields to mock are listed in class-level
+                JSON_FILES and MARKDOWN_FILES attributes.
+
+        Returns
+        ------------
+            Parsed JSON dictionary or Markdown."""
+
+        if field in MockAPIResponse.JSON_FILES.keys():
+            with open(MockAPIResponse.JSON_FILES[field], 'r') as file:
+                return json.load(file)
+        elif field in MockAPIResponse.MARKDOWN_FILES.keys():
+            with open(MockAPIResponse.MARKDOWN_FILES[field], 'r') as file:
+                return file.read()
+        else:
+            return {'data': {}}
+
+

Simulate OSF API response for testing purposes.

+

Attributes

+
+
JSON_FILES : static
+
Key-value dictionary of IDs and paths to stub JSON files. +These are used to generate mock responses to API calls.
+
MARKDOWN_FILES : static
+
Key-value dictionary of IDs and paths to stub Markdown files. +These are used to generate mock responses to API calls to get Wiki data.
+
+

Class variables

+
+
var JSON_FILES
+
+
+
+
var MARKDOWN_FILES
+
+
+
+
+

Static methods

+
+
+def read(field) +
+
+
+ +Expand source code + +
@staticmethod
+def read(field):
+    """Get mock response for a field.
+
+    Parameters
+    -----------
+        field: str
+            ID associated to a JSON or Markdown mock file.
+            Available fields to mock are listed in class-level
+            JSON_FILES and MARKDOWN_FILES attributes.
+
+    Returns
+    ------------
+        Parsed JSON dictionary or Markdown."""
+
+    if field in MockAPIResponse.JSON_FILES.keys():
+        with open(MockAPIResponse.JSON_FILES[field], 'r') as file:
+            return json.load(file)
+    elif field in MockAPIResponse.MARKDOWN_FILES.keys():
+        with open(MockAPIResponse.MARKDOWN_FILES[field], 'r') as file:
+            return file.read()
+    else:
+        return {'data': {}}
+
+

Get mock response for a field.

+

Parameters

+
field: str
+    ID associated to a JSON or Markdown mock file.
+    Available fields to mock are listed in class-level
+    JSON_FILES and MARKDOWN_FILES attributes.
+
+

Returns

+
Parsed JSON dictionary or Markdown.
+
+
+
+
+
+
+
+ +
+ + + diff --git a/docs/formatter.html b/docs/formatter.html new file mode 100644 index 0000000..1228680 --- /dev/null +++ b/docs/formatter.html @@ -0,0 +1,815 @@ + + + + + + +osfexport.formatter API documentation + + + + + + + + + + + +
+
+
+

Module osfexport.formatter

+
+
+
+
+
+
+
+
+

Functions

+
+
+def explore_project_tree(project, projects, pdf=None) +
+
+
+ +Expand source code + +
def explore_project_tree(project, projects, pdf=None):
+    """Recursively find child projects and write them to a PDF.
+
+    Parameters
+    -----------
+        project: dict
+            Dictionary containing project data to write.
+        projects: list[dict]
+            List of all projects to explore.
+        pdf: PDF
+            PDF object to write to. If None, a new PDF will be created.
+
+    Returns
+    -----------
+        pdf: PDF
+            PDF object with the project and its children written to it."""
+
+    # Start with no PDF at root projects
+    if not pdf:
+        pdf = PDF()
+
+    pdf.set_line_width(0.05)
+    pdf.set_left_margin(10)
+    pdf.set_right_margin(30)
+
+    # Add current project to PDF
+    pdf._write_project_body(project)
+
+    # Do children last so that they come at end of the PDF
+    children = project['children']
+    for child_id in children:
+        child_project = next(
+            (p for p in projects if p['metadata']['id'] == child_id), None
+        )
+        if child_project:
+            pdf = explore_project_tree(
+                child_project, projects, pdf=pdf
+            )
+
+    return pdf
+
+

Recursively find child projects and write them to a PDF.

+

Parameters

+
project: dict
+    Dictionary containing project data to write.
+projects: list[dict]
+    List of all projects to explore.
+pdf: PDF
+    PDF object to write to. If None, a new PDF will be created.
+
+

Returns

+
pdf: PDF
+    PDF object with the project and its children written to it.
+
+
+
+def write_pdf(projects, root_idx, folder='') +
+
+
+ +Expand source code + +
def write_pdf(projects, root_idx, folder=''):
+    """Make PDF for each project.
+
+    Parameters
+    ------------
+        projects: dict[str, str|tuple]
+            Projects found to export into the PDF.
+        root_idx: int
+            Position of root node (no parent) in the projects list.
+            This is used for accessing root projects without sorting the list.
+        folder: str
+            The path to the folder to output the project PDFs in.
+            Default is the current working directory.
+
+    Returns
+    ------------
+        pdf: PDF
+            Project export PDF
+        path: str
+            Path to the PDF file
+    """
+
+    curr_project = projects[root_idx]
+    title = curr_project['metadata']['title']
+    pdf = explore_project_tree(curr_project, projects)
+
+    # Remove spaces in file name for better behaviour on Linux
+    # Add timestamp to allow distinguishing between PDFs at a glance
+    timestamp = pdf.date_printed.strftime(
+        '%Y-%m-%d %H-%M-%S %Z'
+    ).replace(' ', '-')
+    filename = f'{title.replace(' ', '-')}-{timestamp}.pdf'
+
+    if folder:
+        if not os.path.exists(folder):
+            os.mkdir(folder)
+        path = os.path.join(os.getcwd(), folder, filename)
+    else:
+        path = os.path.join(os.getcwd(), filename)
+    pdf.output(path)
+
+    return pdf, path
+
+

Make PDF for each project.

+

Parameters

+
projects: dict[str, str|tuple]
+    Projects found to export into the PDF.
+root_idx: int
+    Position of root node (no parent) in the projects list.
+    This is used for accessing root projects without sorting the list.
+folder: str
+    The path to the folder to output the project PDFs in.
+    Default is the current working directory.
+
+

Returns

+
pdf: PDF
+    Project export PDF
+path: str
+    Path to the PDF file
+
+
+
+
+
+

Classes

+
+
+class HTMLImageSizeCapRenderer +(*args, **kwargs) +
+
+
+ +Expand source code + +
class HTMLImageSizeCapRenderer(HTMLRenderer):
+    """
+    Custom Markdown to HTML renderer which caps image size.
+
+    Attributes
+    ------------------
+    max_width: int
+        Max width of images. Image widths above this get shrunk.
+        Default is 300 pixels.
+    
+    max_height: int
+        Max height of images. Image heights above this get shrunk.
+        default is 300 pixels.
+    
+    """
+
+    max_width = 300
+    max_height = 300
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    def render_image(self, token):
+        """Render an image with a specified size."""
+
+        template = '<img src="{}" alt="{}"{}{}{} />'
+
+        # Cap image size if needed so they can fit on the page
+        try:
+            img_info = get_img_info(token.src)
+        except (urllib.error.HTTPError, PIL.UnidentifiedImageError):
+            return f'<a href="{html.escape(token.src)}">{token.src}</a>'
+
+        if img_info['w'] > HTMLImageSizeCapRenderer.max_width:
+            new_width = HTMLImageSizeCapRenderer.max_width
+        else:
+            new_width = img_info['w']
+        width = ' width="{}"'.format(html.escape(str(new_width)))
+
+        if img_info['h'] > HTMLImageSizeCapRenderer.max_height:
+            new_height = HTMLImageSizeCapRenderer.max_height
+        else:
+            new_height = img_info['h']
+        height = ' height="{}"'.format(html.escape(str(new_height)))
+
+        if token.title:
+            title = ' title="{}"'.format(html.escape(token.title))
+        else:
+            title = ""
+        return template.format(token.src, self.render_to_plain(token), title, width, height)
+
+

Custom Markdown to HTML renderer which caps image size.

+

Attributes

+
+
max_width : int
+
Max width of images. Image widths above this get shrunk. +Default is 300 pixels.
+
max_height : int
+
Max height of images. Image heights above this get shrunk. +default is 300 pixels.
+
+

Args

+
+
extras : list
+
allows subclasses to add even more custom tokens.
+
html_escape_double_quotes : bool
+
whether to also escape double +quotes when HTML-escaping rendered text.
+
html_escape_single_quotes : bool
+
whether to also escape single +quotes when HTML-escaping rendered text.
+
process_html_tokens : bool
+
whether to include HTML tokens in the +processing. If False, HTML markup will be treated as plain +text: e.g. input <br> will be rendered as &lt;br&gt;.
+
**kwargs
+
additional parameters to be passed to the ancestor's +constructor.
+
+

Ancestors

+
    +
  • mistletoe.html_renderer.HtmlRenderer
  • +
  • mistletoe.base_renderer.BaseRenderer
  • +
+

Class variables

+
+
var max_height
+
+
+
+
var max_width
+
+
+
+
+

Methods

+
+
+def render_image(self, token) +
+
+
+ +Expand source code + +
def render_image(self, token):
+    """Render an image with a specified size."""
+
+    template = '<img src="{}" alt="{}"{}{}{} />'
+
+    # Cap image size if needed so they can fit on the page
+    try:
+        img_info = get_img_info(token.src)
+    except (urllib.error.HTTPError, PIL.UnidentifiedImageError):
+        return f'<a href="{html.escape(token.src)}">{token.src}</a>'
+
+    if img_info['w'] > HTMLImageSizeCapRenderer.max_width:
+        new_width = HTMLImageSizeCapRenderer.max_width
+    else:
+        new_width = img_info['w']
+    width = ' width="{}"'.format(html.escape(str(new_width)))
+
+    if img_info['h'] > HTMLImageSizeCapRenderer.max_height:
+        new_height = HTMLImageSizeCapRenderer.max_height
+    else:
+        new_height = img_info['h']
+    height = ' height="{}"'.format(html.escape(str(new_height)))
+
+    if token.title:
+        title = ' title="{}"'.format(html.escape(token.title))
+    else:
+        title = ""
+    return template.format(token.src, self.render_to_plain(token), title, width, height)
+
+

Render an image with a specified size.

+
+
+
+
+class PDF +(url='') +
+
+
+ +Expand source code + +
class PDF(FPDF):
+    """Custom PDF class to implement extra customisation.
+
+    Attributes:
+        date_printed: datetime
+            Date and time when the project was exported.
+        url: str
+            Current URL to include in QR codes.
+        parent_url: str
+            URL of root project to use in component sections.
+        parent_title: str
+            Title of root project to use in component sections.
+    """
+
+    # Global styles for PDF
+    BLUE = (173, 216, 230)
+    HEADINGS_STYLE = FontFace(emphasis="BOLD", fill_color=BLUE)
+    FONT_SIZES = {
+        'h1': 16,  # Project titles
+        'h2': 12,  # Section titles
+        'h3': 10,  # Section sub-titles
+        'h4': 9,  # Body
+        'h5': 8  # Footer
+    }
+    LINK_STYLE = FontFace(emphasis="UNDERLINE", size_pt=FONT_SIZES['h5'])
+    LINE_PADDING = -1  # Gaps between lines
+    TITLE_CELL_WIDTH = 150  # Shorter width to avoid QR code clipping
+    CELL_WIDTH = 180  # Width of text cells
+
+    def __init__(self, url=''):
+        super().__init__()
+        self.date_printed = datetime.datetime.now(
+            datetime.timezone.utc
+        )
+        self.url = url
+        # Setup unicode font for use. Can have 4 styles
+        self.font = 'dejavu-sans'
+        self.add_font(self.font, style="", fname=os.path.join(
+            os.path.dirname(__file__), 'font', 'DejaVuSans.ttf'))
+        self.add_font(self.font, style="b", fname=os.path.join(
+            os.path.dirname(__file__), 'font', 'DejaVuSans-Bold.ttf'))
+        self.add_font(self.font, style="i", fname=os.path.join(
+            os.path.dirname(__file__), 'font', 'DejaVuSans-Oblique.ttf'))
+        self.add_font(self.font, style="bi", fname=os.path.join(
+            os.path.dirname(__file__), 'font', 'DejaVuSans-BoldOblique.ttf'))
+
+    def generate_qr_code(self):
+        """
+        Make a QR code based on the current PDF's URL.
+        """
+
+        qr = qrcode.make(self.url)
+        img_byte_arr = io.BytesIO()
+        qr.save(img_byte_arr, format='PNG')
+        img_byte_arr.seek(0)
+        return img_byte_arr
+
+    def footer(self):
+        """
+        Makes the PDF footer, including Exported at timestamp
+        """
+
+        self.set_y(-15)
+        self.set_x(-30)
+        self.set_font(self.font, size=PDF.FONT_SIZES['h5'])
+        self.cell(0, 10, f"Page: {self.page_no()}", align="C")
+        self.set_x(10)
+        timestamp = self.date_printed.strftime(
+            '%Y-%m-%d %H:%M:%S %Z'
+        )
+        self.cell(0, 10, f"Exported: {timestamp}", align="L")
+        self.set_x(10)
+        self.set_y(-15)
+        qr_img = self.generate_qr_code()
+        self.image(qr_img, w=15, h=15, x=Align.C)
+
+    def _write_list_section(self, key, fielddict):
+        """Handle writing fields of different types inplace to a PDF.
+        Possible types are lists, strings or dictionaries.
+
+        Parameters
+        -----------
+            key: str
+                Name of the field to write.
+            fielddict: dict
+                Dictionary containing the data for each field to include.
+                e.g. {
+                'subjects': 's1, s2, s3',
+                'title': 'title1'
+                }
+        """
+
+        # Set nicer display names for certain PDF fields
+        pdf_display_names = {
+            'identifiers': 'DOI',
+            'funders': 'Support/Funding Information'
+        }
+        if key in pdf_display_names:
+            field_name = pdf_display_names[key]
+        else:
+            field_name = key.replace('_', ' ').title()
+
+        if isinstance(fielddict[key], list):
+            # Create separate paragraphs for more complex attributes
+            self.write(0, '\n')
+            self.set_font(self.font, size=PDF.FONT_SIZES['h3'])
+            self.multi_cell(
+                w=PDF.CELL_WIDTH, h=None,
+                text=f'**{field_name}**\n',
+                align='L', markdown=True, padding=PDF.LINE_PADDING
+            )
+            self.set_font(self.font, size=PDF.FONT_SIZES['h4'])
+            if len(fielddict[key]) > 0:
+                for idx, item in enumerate(fielddict[key]):
+                    for subkey in item.keys():
+                        if subkey in pdf_display_names:
+                            field_name = pdf_display_names[subkey]
+                        else:
+                            field_name = subkey.replace('_', ' ').title()
+
+                        self.multi_cell(
+                            w=PDF.CELL_WIDTH, h=None,
+                            text=f'**{field_name}:** {item[subkey]}\n',
+                            align='L', markdown=True, padding=PDF.LINE_PADDING
+                        )
+                    if idx < len(fielddict[key])-1:
+                        self.ln()
+                        self.set_x(9)
+            else:
+                self.multi_cell(
+                    w=PDF.CELL_WIDTH, h=None,
+                    text='NA',
+                    align='L', markdown=True, padding=PDF.LINE_PADDING
+                )
+        else:
+            # Simple key-value attributes can go on one-line
+            self.multi_cell(
+                w=PDF.CELL_WIDTH,
+                h=None,
+                text=f'**{field_name}:** {fielddict[key]}\n',
+                align='L',
+                markdown=True,
+                padding=PDF.LINE_PADDING
+            )
+
+    def _write_project_body(self, project):
+        """Write inplace the body of a project to the PDF.
+
+        Parameters
+        -----------
+            project: dict
+                Dictionary containing project data to write.
+        """
+
+        self.add_page()
+        self.set_font(self.font, size=PDF.FONT_SIZES['h4'])
+        wikis = project['wikis']
+
+        # Start with parent, project headers and links
+        parent = project['parent']
+        if parent:
+            self.set_font(self.font, size=PDF.FONT_SIZES['h1'], style='B')
+            self.write(h=0, text=f'Parent: {parent[0]}\n')
+            self.set_font(self.font, size=PDF.FONT_SIZES['h3'], style='U')
+            self.write(h=0, text=f'{parent[1]}\n', link=parent[1])
+            self.ln(h=5)
+
+        # Pop URL field to avoid printing it out in Metadata section
+        url = project['metadata'].pop('url', '')
+        self.url = url  # Set current URL to use in QR codes
+        qr_img = self.generate_qr_code()
+        self.image(qr_img, w=30, x=180, y=5)
+        title = project['metadata']['title']
+        self.set_font(self.font, size=PDF.FONT_SIZES['h1'], style='B')
+        self.write(
+            h=0,
+            text=f"{'Component: ' if parent else ''}{title} \n"
+        )
+        self.set_font(self.font, size=PDF.FONT_SIZES['h3'], style='U')
+        self.write(h=0, text=f'{url}\n', link=url)
+        self.ln(h=5)
+
+        # Write title for metadata section, then actual fields
+        self.set_font(self.font, size=PDF.FONT_SIZES['h2'], style='B')
+        self.multi_cell(
+            w=PDF.CELL_WIDTH, h=None, text='1. Project Metadata\n',
+            align='L', padding=PDF.LINE_PADDING)
+        self.set_font(self.font, size=PDF.FONT_SIZES['h4'])
+        for key in project['metadata']:
+            self._write_list_section(key, project['metadata'])
+        self.ln(h=7)
+
+        # Write Contributors in table
+        self.set_x(8)
+        self.set_font(self.font, size=PDF.FONT_SIZES['h2'], style='B')
+        self.multi_cell(w=PDF.CELL_WIDTH, h=None, text='2. Contributors\n', align='L')
+        self.set_font(self.font, size=PDF.FONT_SIZES['h4'])
+        with self.table(
+            headings_style=PDF.HEADINGS_STYLE,
+            col_widths=(0.8, 0.5, 1.2), align="LEFT"
+        ) as table:
+            row = table.row()
+            row.cell('Name')
+            row.cell('Bibliographic?')
+            row.cell('Profile Link')
+            self.set_font(self.font, size=PDF.FONT_SIZES['h5'])
+            for data_row in project['contributors']:
+                row = table.row()
+                for idx, datum in enumerate(data_row):
+                    if datum is True:
+                        datum = 'Yes'
+                    if datum is False:
+                        datum = 'No'
+                    if idx == 2:
+                        row.cell(text=datum, link=datum, style=self.LINK_STYLE)
+                    else:
+                        row.cell(datum)
+        self.ln(h=7)
+
+        # List files stored in storage providers
+        # For now only OSF Storage is involved
+        self.set_x(8)
+        self.set_font(self.font, size=PDF.FONT_SIZES['h2'], style='B')
+        self.multi_cell(w=PDF.CELL_WIDTH, h=None, text='3. Files in Main Project\n', align='L')
+        self.set_font(self.font, size=PDF.FONT_SIZES['h3'], style='B')
+        self.multi_cell(w=PDF.CELL_WIDTH, h=None, text='OSF Storage\n', align='L')
+        self.set_font(self.font, size=PDF.FONT_SIZES['h4'])
+        if len(project['files']) > 0:
+            with self.table(
+                headings_style=PDF.HEADINGS_STYLE,
+                col_widths=(1, 0.3, 1.2), align="LEFT"
+            ) as table:
+                self.set_font(self.font, size=PDF.FONT_SIZES['h4'])
+                row = table.row()
+                row.cell('File Name')
+                row.cell('Size (MB)')
+                row.cell('Download Link')
+                self.set_font(self.font, size=PDF.FONT_SIZES['h5'])
+                for data_row in project['files']:
+                    row = table.row()
+                    for idx, datum in enumerate(data_row):
+                        if datum is True:
+                            datum = 'Yes'
+                        if datum is False or datum is None:
+                            datum = 'N/A'
+                        if idx == 2:
+                            row.cell(text=datum, link=datum, style=self.LINK_STYLE)
+                        else:
+                            row.cell(datum)
+        else:
+            self.write(0, '\n')
+            self.multi_cell(
+                w=PDF.CELL_WIDTH, h=None, text='No files found for this project.\n', align='L'
+            )
+            self.write(0, '\n')
+        self.ln(h=10)
+
+        # Write wikis separately to more easily handle Markdown parsing
+        self._write_wiki_pages(
+            wikis, parent=parent, title=project['metadata']['title']
+        )
+
+    def _write_wiki_pages(self, wikis, title, parent=None):
+        """Write inplace the wiki pages to the PDF.
+
+        Parameters
+        -----------
+            wikis: dict
+                Dictionary containing wiki data to write.
+            title: str
+                Title of the current project.
+            parent: None | tuple[str, str]
+                Optional parent project with (name, url)
+
+        """
+
+        for i, wiki in enumerate(wikis.keys()):
+            self.add_page()
+            if i == 0:
+                self.set_font(self.font, size=PDF.FONT_SIZES['h2'], style='B')
+                title_template = '4. Wiki ({}{}{})'
+                if parent:
+                    header = title_template.format(f'Parent: {parent[0]}', ' | ', title)
+                else:
+                    header = title_template.format('', '', title)
+                self.multi_cell(
+                    w=PDF.CELL_WIDTH, h=None, text=f'{header}\n',
+                    align='L')
+                self.ln()
+            self.set_font(self.font, size=PDF.FONT_SIZES['h2'], style='B')
+            self.multi_cell(w=PDF.CELL_WIDTH, h=None, text=f'{wiki}\n')
+            self.set_font(self.font, size=PDF.FONT_SIZES['h4'])
+            html = markdown(
+                wikis[wiki],
+                renderer=HTMLImageSizeCapRenderer
+            )
+            self.write_html(html)
+
+

Custom PDF class to implement extra customisation.

+

Attributes

+
+
date_printed
+
datetime +Date and time when the project was exported.
+
url
+
str +Current URL to include in QR codes.
+
parent_url
+
str +URL of root project to use in component sections.
+
parent_title
+
str +Title of root project to use in component sections.
+
+

Args

+
+
orientation : str
+
possible values are "portrait" (can be abbreviated "P") +or "landscape" (can be abbreviated "L"). Default to "portrait".
+
unit : str, int, float
+
possible values are "pt", "mm", "cm", "in", or a number. +A point equals 1/72 of an inch, that is to say about 0.35 mm (an inch being 2.54 cm). +This is a very common unit in typography; font sizes are expressed in this unit. +If given a number, then it will be treated as the number of points per unit. +(eg. 72 = 1 in) +Default to "mm".
+
format : str
+
possible values are "a3", "a4", "a5", "letter", "legal" or a tuple +(width, height) expressed in the given unit. Default to "a4".
+
font_cache_dir : Path or str
+
[DEPRECATED since v2.5.1] unused
+
+

Ancestors

+
    +
  • fpdf.fpdf.FPDF
  • +
  • fpdf.graphics_state.GraphicsStateMixin
  • +
  • fpdf.text_region.TextRegionMixin
  • +
+

Class variables

+
+
var BLUE
+
+
+
+
var CELL_WIDTH
+
+
+
+
var FONT_SIZES
+
+
+
+
var HEADINGS_STYLE
+
+
+
+
var LINE_PADDING
+
+
+
+ +
+
+
+
var TITLE_CELL_WIDTH
+
+
+
+
+

Methods

+
+
+def footer(self) +
+
+
+ +Expand source code + +
def footer(self):
+    """
+    Makes the PDF footer, including Exported at timestamp
+    """
+
+    self.set_y(-15)
+    self.set_x(-30)
+    self.set_font(self.font, size=PDF.FONT_SIZES['h5'])
+    self.cell(0, 10, f"Page: {self.page_no()}", align="C")
+    self.set_x(10)
+    timestamp = self.date_printed.strftime(
+        '%Y-%m-%d %H:%M:%S %Z'
+    )
+    self.cell(0, 10, f"Exported: {timestamp}", align="L")
+    self.set_x(10)
+    self.set_y(-15)
+    qr_img = self.generate_qr_code()
+    self.image(qr_img, w=15, h=15, x=Align.C)
+
+

Makes the PDF footer, including Exported at timestamp

+
+
+def generate_qr_code(self) +
+
+
+ +Expand source code + +
def generate_qr_code(self):
+    """
+    Make a QR code based on the current PDF's URL.
+    """
+
+    qr = qrcode.make(self.url)
+    img_byte_arr = io.BytesIO()
+    qr.save(img_byte_arr, format='PNG')
+    img_byte_arr.seek(0)
+    return img_byte_arr
+
+

Make a QR code based on the current PDF's URL.

+
+
+
+
+
+
+ +
+ + + diff --git a/docs/getting_started.rst b/docs/getting_started.rst new file mode 100644 index 0000000..aa7f714 --- /dev/null +++ b/docs/getting_started.rst @@ -0,0 +1,24 @@ +Basic Flow of Exporting +======================= + +1. Users may need to give a string GUID to export a particular project. The easiest way to do this for them is to give the URL to their project. `osfexport` provides a way to project IDs from OSF project URLs with the `exporter.extract_project_id` method. +2. If the user wants to export all projects where they are contributors, or a single private project, they will need to provide a Personal Access Token. In the CLI this is done via the cli.prompt_pat method. +3. Get data for projects for rendering via the OSF API using `exporter.get_nodes`. This will return a dictionary of projects with their attributes (including child projects AKA components), and a list of indexes for the positions of the projects to make PDFs for. +4. Output the project data obtained to a PDF by passing the project dictionary to `formatter.write_pdf` + +Example: + +.. code-block:: python + + import osfexport + + # Need PAT for private projects or exporting all projects, otherwise can be left blank + # pat = + pat = '' + + # Need this part for single projects + project_id = osfexport.exporter.extract_project_id(url) + + projects, indexes_of_main_projects = osfexport.exporter.get_nodes(pat=pat, project_id=project_id) + for project in indexes_of_main_projects: + file, file_path = osfexport.formatter.write_pdf(projects, project, folder='') diff --git a/docs/index.html b/docs/index.html new file mode 100644 index 0000000..3ad667c --- /dev/null +++ b/docs/index.html @@ -0,0 +1,896 @@ + + + + + + +osfexport API documentation + + + + + + + + + + + +
+
+
+

Package osfexport

+
+
+
+
+

Sub-modules

+
+
osfexport.cli
+
+
+
+
osfexport.exporter
+
+
+
+
osfexport.formatter
+
+
+
+
+
+
+
+
+

Functions

+
+
+def call_api(url,
pat,
method='GET',
per_page=100,
filters={},
is_json=True,
usetest=False,
max_tries=5)
+
+
+
+ +Expand source code + +
def call_api(
+        url, pat, method='GET', per_page=100, filters={}, is_json=True,
+        usetest=False, max_tries=5):
+    """Call OSF v2 API methods.
+
+    Parameters
+    ----------
+    url: str
+        URL to API method/resource/query.
+    method: str
+        HTTP method for the request.
+    pat: str
+        Personal Access Token to authorise a user with.
+    per_page: int
+        Number of items to include in a JSON page for API responses.
+        The maximum size is 100.
+    filters: dict
+        Dictionary of query parameters to filter results with.
+
+        Example Input: {'category': 'project', 'title': 'ttt'}
+        Example Query String: ?filter[category]=project&filter[title]=ttt
+    is_json: bool
+        If true, set API version to get correct API responses.
+    usetest: bool
+        If True, use fixed delay of 0.1 seconds for tests.
+        If False, use a random delay between [1, 60] seconds between requests.
+        This spaces out requests over time to give the API chance to recover.
+    max_tries: int
+        Number of attempts to make before raising a 429 error. Default is 5, Limit is 7.
+
+    Throws
+    -------------
+        HTTPError - 429 error if we can't connect to the API after retries.
+        Other errors are thrown immediately.
+
+        URLError - failed to connect to OSF API
+
+    Returns
+    ----------
+        result: HTTPResponse
+            Response to the request from the API.
+    """
+    if (filters or per_page) and method == 'GET':
+        query_string = '&'.join([f'filter[{key}]={value}'
+                                 for key, value in filters.items()
+                                 if not isinstance(value, dict)])
+        if per_page:
+            query_string += f'&page[size]={per_page}'
+        url = f'{url}?{query_string}'
+
+    request = webhelper.Request(url, method=method)
+    request.add_header('Authorization', f'Bearer {pat}')
+
+    version = importlib.metadata.version("osfexport")
+    request.add_header('User-Agent', f'osfexport/{version} (Python)')
+
+    # Pin API version so that JSON has correct format
+    API_VERSION = '2.20'
+    if is_json:
+        request.add_header(
+            'Accept',
+            f'application/vnd.api+json;version={API_VERSION}'
+        )
+
+    if max_tries > 7:
+        max_tries = 7  # Cap retries to reduce requests sent and max delay time
+
+    # Retry requests if we get 429 errors
+    try_count = 0
+    result = None
+    while try_count < max_tries and result is None:
+        try:
+            result = webhelper.urlopen(request)
+        except HTTPError as e:
+            # Other error codes tell us directly something is wrong
+            if e.code == 429:
+                if not usetest:
+                    # Wait longer between requests to give API more recovery time
+                    # Wait random periods to avoid hammering requests all at once
+                    min_wait = (try_count+1)**2
+                    time.sleep(random.uniform(min_wait, 60))
+                else:
+                    time.sleep(0.5)  # Wait constant time for tests
+                try_count += 1
+            else:
+                raise e
+    if result is None:
+        raise HTTPError(
+            url=url,
+            code=429,
+            msg="Too many requests to the OSF API.",
+            hdrs=request.headers,
+            fp=None
+        )
+    return result
+
+

Call OSF v2 API methods.

+

Parameters

+
+
url : str
+
URL to API method/resource/query.
+
method : str
+
HTTP method for the request.
+
pat : str
+
Personal Access Token to authorise a user with.
+
per_page : int
+
Number of items to include in a JSON page for API responses. +The maximum size is 100.
+
filters : dict
+
+

Dictionary of query parameters to filter results with.

+

Example Input: {'category': 'project', 'title': 'ttt'} +Example Query String: ?filter[category]=project&filter[title]=ttt

+
+
is_json : bool
+
If true, set API version to get correct API responses.
+
usetest : bool
+
If True, use fixed delay of 0.1 seconds for tests. +If False, use a random delay between [1, 60] seconds between requests. +This spaces out requests over time to give the API chance to recover.
+
max_tries : int
+
Number of attempts to make before raising a 429 error. Default is 5, Limit is 7.
+
+

Throws

+
HTTPError - 429 error if we can't connect to the API after retries.
+Other errors are thrown immediately.
+
+URLError - failed to connect to OSF API
+
+

Returns

+
result: HTTPResponse
+    Response to the request from the API.
+
+
+
+def extract_project_id(url) +
+
+
+ +Expand source code + +
def extract_project_id(url):
+    """Extract project ID from a given OSF project URL.
+
+    Parameters
+    ----------
+    url: str
+        URL of the OSF project which should contain the project ID. E.g.:
+        - Full URL with parameters (https://osf.io/xyz/?param=value)
+        - API URL (https://api.test.osf.io/v2/nodes/xyz)
+        - Just the ID (xyz)
+        - Empty string
+
+    Returns
+    -------
+    str
+        Project ID extracted from the URL.
+    """
+
+    if not url:
+        return ''
+
+    parts = url.strip("/").split("/")
+    # Handle case of just ID provided
+    if len(parts) == 1:
+        return parts[0]
+
+    # API URLs are of form /nodes/id/...
+    if 'nodes' in parts:
+        idx = parts.index('nodes')
+        if idx + 1 < len(parts):
+            return parts[idx + 1]
+
+    # For regular URLs, extract ID from last path component before query params
+    if '?' in parts[-1]:
+        return parts[-2]
+    else:
+        return parts[-1]
+
+

Extract project ID from a given OSF project URL.

+

Parameters

+
+
url : str
+
URL of the OSF project which should contain the project ID. E.g.: +- Full URL with parameters (https://osf.io/xyz/?param=value) +- API URL (https://api.test.osf.io/v2/nodes/xyz) +- Just the ID (xyz) +- Empty string
+
+

Returns

+
+
str
+
Project ID extracted from the URL.
+
+
+
+def get_nodes(pat, page_size=100, dryrun=False, project_id='', usetest=False) +
+
+
+ +Expand source code + +
def get_nodes(pat, page_size=100, dryrun=False, project_id='', usetest=False):
+    """Pull and list projects for a user from the OSF.
+
+    Parameters
+    ----------
+    pat: str
+        Personal Access Token to authorise a user with.
+    page_size: int
+        How many nodes to put onto a page. Default is 100.
+        Possible range is 1-100
+    dryrun: bool
+        If True, use test data from JSON stubs to mock API calls.
+    project_id: str
+        Optional ID for a specific OSF project to export.
+    usetest: bool
+        If True, use test API host, otherwise use production host.
+
+    Returns
+    ----------
+        projects: list[dict]
+            List of all project objects found
+        root_nodes: list[int]
+            List of indexes for root nodes in projects list.
+            These are the nodes to make PDFs for and start from in PDFs.
+    """
+
+    # Set start link and page size filter based on flags
+    api_host = get_host(usetest)
+    node_filter = {}
+    if not dryrun:
+        if project_id:
+            start = f'{api_host}/nodes/{project_id}/'
+        else:
+            start = f'{api_host}/users/me/nodes/'
+            node_filter = {
+                'parent': ''
+            }
+    else:
+        page_size = 4  # Nodes found are hardcoded for --dryrun
+        if project_id:
+            start = project_id
+        else:
+            start = 'nodes'
+
+    results = paginate_json_result(
+        start, get_project_data, dryrun=dryrun, usetest=usetest,
+        pat=pat, filters=node_filter, project_id=project_id, per_page=page_size
+    )
+    if len(results) > 0:
+        l1, l2 = zip(*list(results))
+    else:
+        l1, l2 = (), ()
+    projects = [item for sublist in l1 for item in sublist]
+
+    # After pagination we get indexes of root nodes local to each page
+    # We need to convert these to global indexes before merging the list
+    page_idx = -1
+    for page in l2:
+        page_idx += 1
+        for idx, n in enumerate(page):
+            global_node_idx = page_size*page_idx + n
+            page[idx] = global_node_idx
+    root_nodes = [item for sublist in l2 for item in sublist]
+
+    return projects, root_nodes
+
+

Pull and list projects for a user from the OSF.

+

Parameters

+
+
pat : str
+
Personal Access Token to authorise a user with.
+
page_size : int
+
How many nodes to put onto a page. Default is 100. +Possible range is 1-100
+
dryrun : bool
+
If True, use test data from JSON stubs to mock API calls.
+
project_id : str
+
Optional ID for a specific OSF project to export.
+
usetest : bool
+
If True, use test API host, otherwise use production host.
+
+

Returns

+
projects: list[dict]
+    List of all project objects found
+root_nodes: list[int]
+    List of indexes for root nodes in projects list.
+    These are the nodes to make PDFs for and start from in PDFs.
+
+
+
+def is_public(url) +
+
+
+ +Expand source code + +
def is_public(url):
+    """Return boolean to indicate if a URL is public (True) or not (False).
+    This is mainly used for checking if a project is publicly accessible.
+
+    Parameters
+    ------------
+    url: str
+        The URL to test.
+
+    Returns
+    ----------------
+        is_public: bool
+            Whether we can access the URL without a PAT (i.e. status code 200)
+
+    Raises
+    -------------------
+        HTTPError, URLError
+            If we get a HTTP error code that isn't 401/403, or a connection error.
+    """
+
+    try:
+        result = call_api(
+            url, pat='', method='GET'
+        ).status
+    except (HTTPError, URLError) as e:
+        # Don't raise error if we get a HTTP error with certain codes
+        valid_error_codes = [401, 403]
+        if isinstance(e, HTTPError) and e.code in valid_error_codes:
+            result = e.code
+        else:
+            raise e
+    is_public = result == 200
+    return is_public
+
+

Return boolean to indicate if a URL is public (True) or not (False). +This is mainly used for checking if a project is publicly accessible.

+

Parameters

+
+
url : str
+
The URL to test.
+
+

Returns

+
is_public: bool
+    Whether we can access the URL without a PAT (i.e. status code 200)
+
+

Raises

+
HTTPError, URLError
+    If we get a HTTP error code that isn't 401/403, or a connection error.
+
+
+
+def paginate_json_result(start, action, fail_on_first=True, **kwargs) +
+
+
+ +Expand source code + +
def paginate_json_result(start, action, fail_on_first=True, **kwargs):
+    """Loop through paginated JSON responses and perform an action on each.
+
+    Parameters
+    -------------
+    start: str
+        Link to start looping from
+    action: func
+        Takes in found JSON page and returns a result
+    per_page: int
+        How many items to include on one page. Default is 100.
+        Valid range is from 1-1000.
+    filters: dict
+        Optional key-value dict to filter queries by.
+    is_json:
+        If JSON response expected, add header to specify JSON format.
+    pat: str
+        Personal Access Token to authorise a user.
+    dryrun: bool
+        Flag for whether mock JSON or real API will be used.
+    **kwargs
+        Extra keyword args to pass down to action and call_api.
+
+    Returns
+    ------------------
+    results: deque
+        Queue of results per page
+    
+    Throws
+    ------------------
+    HTTPError, URLError - non-429 HTTP errors which indicate a problem
+    """
+
+    next_link = start
+    is_last_page = False
+    is_first_item = True  # Want to throw error if very first item fails
+    results = deque()
+    per_page = kwargs.pop('per_page', 100)
+    filters = kwargs.pop('filters', {})
+    is_json = kwargs.pop('is_json', True)
+    pat = kwargs.get('pat', '')
+    dryrun = kwargs.get('dryrun', False)
+    while not is_last_page:
+        try:
+            if not dryrun:
+                curr_page = call_api(
+                    next_link, pat, per_page=per_page, filters=filters,
+                    is_json=is_json)
+                # Catch error if call_api is replaced with mock in tests
+                try:
+                    curr_page = curr_page.read()
+                    if is_json:
+                        curr_page = json.loads(curr_page)
+                except AttributeError:
+                    pass
+            else:
+                curr_page = MockAPIResponse.read(next_link)
+            results.append(action(curr_page, **kwargs))
+        except HTTPError as e:
+            if fail_on_first and is_first_item or e.code == 429:
+                raise e
+            else:
+                logging.warning("Warning: Couldn't parse JSON page, skipping to next page...")
+        # Stop if no next link found
+        try:
+            next_link = curr_page['links']['next']
+            is_last_page = not next_link
+        except (KeyError, UnboundLocalError):
+            is_last_page = True
+    return results
+
+

Loop through paginated JSON responses and perform an action on each.

+

Parameters

+
+
start : str
+
Link to start looping from
+
action : func
+
Takes in found JSON page and returns a result
+
per_page : int
+
How many items to include on one page. Default is 100. +Valid range is from 1-1000.
+
filters : dict
+
Optional key-value dict to filter queries by.
+
is_json:
+
If JSON response expected, add header to specify JSON format.
+
pat : str
+
Personal Access Token to authorise a user.
+
dryrun : bool
+
Flag for whether mock JSON or real API will be used.
+
**kwargs
+
Extra keyword args to pass down to action and call_api.
+
+

Returns

+
+
results : deque
+
Queue of results per page
+
+

Throws

+

HTTPError, URLError - non-429 HTTP errors which indicate a problem

+
+
+def prompt_pat(project_id='', usetest=False) +
+
+
+ +Expand source code + +
def prompt_pat(project_id='', usetest=False):
+    """
+    Ask for a PAT if exporting a single project or all projects a user has.
+
+    Parameters
+    -------------
+        project_id: str
+            ID of a single project to export.
+            If one provided then ask for a PAT.
+        usetest: bool
+            Flag to indicate whether to use the test/production API server.
+
+    Returns
+    -----------------
+        pat: str
+            Personal Access Token to use to authorise a user.
+
+    Raises
+    -------------------
+        HTTPError, URLError - passed on from is_public method.
+    """
+
+    if usetest:
+        api_host = API_HOST_TEST
+    else:
+        api_host = API_HOST_PROD
+
+    if not project_id:
+        pat = click.prompt(
+            'Please enter your PAT to export all your projects',
+            type=str,
+            hide_input=True
+        )
+    elif not exporter.is_public(f'{api_host}/nodes/{project_id}/'):
+        pat = click.prompt(
+            'Please enter your PAT to export this private project',
+            type=str,
+            hide_input=True
+        )
+    else:
+        pat = ''
+
+    return pat
+
+

Ask for a PAT if exporting a single project or all projects a user has.

+

Parameters

+
project_id: str
+    ID of a single project to export.
+    If one provided then ask for a PAT.
+usetest: bool
+    Flag to indicate whether to use the test/production API server.
+
+

Returns

+
pat: str
+    Personal Access Token to use to authorise a user.
+
+

Raises

+
HTTPError, URLError - passed on from is_public method.
+
+
+
+def write_pdf(projects, root_idx, folder='') +
+
+
+ +Expand source code + +
def write_pdf(projects, root_idx, folder=''):
+    """Make PDF for each project.
+
+    Parameters
+    ------------
+        projects: dict[str, str|tuple]
+            Projects found to export into the PDF.
+        root_idx: int
+            Position of root node (no parent) in the projects list.
+            This is used for accessing root projects without sorting the list.
+        folder: str
+            The path to the folder to output the project PDFs in.
+            Default is the current working directory.
+
+    Returns
+    ------------
+        pdf: PDF
+            Project export PDF
+        path: str
+            Path to the PDF file
+    """
+
+    curr_project = projects[root_idx]
+    title = curr_project['metadata']['title']
+    pdf = explore_project_tree(curr_project, projects)
+
+    # Remove spaces in file name for better behaviour on Linux
+    # Add timestamp to allow distinguishing between PDFs at a glance
+    timestamp = pdf.date_printed.strftime(
+        '%Y-%m-%d %H-%M-%S %Z'
+    ).replace(' ', '-')
+    filename = f'{title.replace(' ', '-')}-{timestamp}.pdf'
+
+    if folder:
+        if not os.path.exists(folder):
+            os.mkdir(folder)
+        path = os.path.join(os.getcwd(), folder, filename)
+    else:
+        path = os.path.join(os.getcwd(), filename)
+    pdf.output(path)
+
+    return pdf, path
+
+

Make PDF for each project.

+

Parameters

+
projects: dict[str, str|tuple]
+    Projects found to export into the PDF.
+root_idx: int
+    Position of root node (no parent) in the projects list.
+    This is used for accessing root projects without sorting the list.
+folder: str
+    The path to the folder to output the project PDFs in.
+    Default is the current working directory.
+
+

Returns

+
pdf: PDF
+    Project export PDF
+path: str
+    Path to the PDF file
+
+
+
+
+
+

Classes

+
+
+class MockAPIResponse +
+
+
+ +Expand source code + +
class MockAPIResponse:
+    """
+    Simulate OSF API response for testing purposes.
+
+    Attributes
+    ----------------
+
+    JSON_FILES: static
+        Key-value dictionary of IDs and paths to stub JSON files.
+        These are used to generate mock responses to API calls.
+    
+    MARKDOWN_FILES: static
+        Key-value dictionary of IDs and paths to stub Markdown files.
+        These are used to generate mock responses to API calls to get Wiki data.
+    """
+
+    JSON_FILES = {
+        'nodes': os.path.join(
+            STUBS_DIR, 'nodestubs.json'),
+        'nodes2': os.path.join(
+            STUBS_DIR, 'nodestubs2.json'),
+        'x': os.path.join(
+            STUBS_DIR, 'singlenode.json'),
+        'a': os.path.join(
+            STUBS_DIR, 'asingle.json'),
+        'affiliated_institutions': os.path.join(
+            STUBS_DIR, 'institutionstubs.json'),
+        'contributors': os.path.join(
+            STUBS_DIR, 'contributorstubs.json'),
+        'identifiers': os.path.join(
+            STUBS_DIR, 'doistubs.json'),
+        'custom_metadata': os.path.join(
+            STUBS_DIR, 'custommetadatastub.json'),
+        'root_folder': os.path.join(
+            STUBS_DIR, 'files', 'rootfolders.json'),
+        'root_files': os.path.join(
+            STUBS_DIR, 'files', 'rootfiles.json'),
+        'tf1_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf1folders.json'),
+        'tf1-2_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf1folders-2.json'),
+        'tf1-2_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-folders.json'),
+        'tf1_files': os.path.join(
+            STUBS_DIR, 'files', 'tf1files.json'),
+        'tf2_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf2folders.json'),
+        'tf2-second_folder': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-folders.json'),
+        'tf2_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2files.json'),
+        'tf2-second_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-files.json'),
+        'tf2-second-2_files': os.path.join(
+            STUBS_DIR, 'files', 'tf2-second-files-2.json'),
+        'license': os.path.join(
+            STUBS_DIR, 'licensestub.json'),
+        'subjects': os.path.join(
+            STUBS_DIR, 'subjectsstub.json'),
+        'wikis': os.path.join(
+            STUBS_DIR, 'wikis', 'wikistubs.json'),
+        'wikis2': os.path.join(
+            STUBS_DIR, 'wikis', 'wikis2stubs.json'),
+        'x-child-1': os.path.join(
+            STUBS_DIR, 'components', 'x-child-1.json'),
+        'x-child-2': os.path.join(
+            STUBS_DIR, 'components', 'x-child-2.json'),
+        'empty-children': os.path.join(
+            STUBS_DIR, 'components', 'empty-children.json'),
+    }
+
+    MARKDOWN_FILES = {
+        'helloworld': os.path.join(
+            STUBS_DIR, 'wikis', 'helloworld.md'),
+        'home': os.path.join(
+            STUBS_DIR, 'wikis', 'home.md'),
+        'anotherone': os.path.join(
+            STUBS_DIR, 'wikis', 'anotherone.md'),
+    }
+
+    @staticmethod
+    def read(field):
+        """Get mock response for a field.
+
+        Parameters
+        -----------
+            field: str
+                ID associated to a JSON or Markdown mock file.
+                Available fields to mock are listed in class-level
+                JSON_FILES and MARKDOWN_FILES attributes.
+
+        Returns
+        ------------
+            Parsed JSON dictionary or Markdown."""
+
+        if field in MockAPIResponse.JSON_FILES.keys():
+            with open(MockAPIResponse.JSON_FILES[field], 'r') as file:
+                return json.load(file)
+        elif field in MockAPIResponse.MARKDOWN_FILES.keys():
+            with open(MockAPIResponse.MARKDOWN_FILES[field], 'r') as file:
+                return file.read()
+        else:
+            return {'data': {}}
+
+

Simulate OSF API response for testing purposes.

+

Attributes

+
+
JSON_FILES : static
+
Key-value dictionary of IDs and paths to stub JSON files. +These are used to generate mock responses to API calls.
+
MARKDOWN_FILES : static
+
Key-value dictionary of IDs and paths to stub Markdown files. +These are used to generate mock responses to API calls to get Wiki data.
+
+

Class variables

+
+
var JSON_FILES
+
+
+
+
var MARKDOWN_FILES
+
+
+
+
+

Static methods

+
+
+def read(field) +
+
+
+ +Expand source code + +
@staticmethod
+def read(field):
+    """Get mock response for a field.
+
+    Parameters
+    -----------
+        field: str
+            ID associated to a JSON or Markdown mock file.
+            Available fields to mock are listed in class-level
+            JSON_FILES and MARKDOWN_FILES attributes.
+
+    Returns
+    ------------
+        Parsed JSON dictionary or Markdown."""
+
+    if field in MockAPIResponse.JSON_FILES.keys():
+        with open(MockAPIResponse.JSON_FILES[field], 'r') as file:
+            return json.load(file)
+    elif field in MockAPIResponse.MARKDOWN_FILES.keys():
+        with open(MockAPIResponse.MARKDOWN_FILES[field], 'r') as file:
+            return file.read()
+    else:
+        return {'data': {}}
+
+

Get mock response for a field.

+

Parameters

+
field: str
+    ID associated to a JSON or Markdown mock file.
+    Available fields to mock are listed in class-level
+    JSON_FILES and MARKDOWN_FILES attributes.
+
+

Returns

+
Parsed JSON dictionary or Markdown.
+
+
+
+
+
+
+
+ +
+ + + diff --git a/docs/overview.rst b/docs/overview.rst new file mode 100644 index 0000000..87e9769 --- /dev/null +++ b/docs/overview.rst @@ -0,0 +1,17 @@ +Overview +======== + +The main source code for osfexport is kept in src/osfexport and is structured as such: + +* exporter.py: This handles the main logic for exporting data through the API. +* formatter.py: This handles how to write project data obtained from the API to an output PDF. +* cli.py: This handles the command-line interface, defining what commands and help text are shown to a CLI user. + +Tests are written using the `unittest` framework and are kept in tests/test_clitool.py: + +* TestFormatter: tests the code in formatter.py +* TestCLI: tests code in cli.py +* TestExporter: tests code in exporter.py +* TestAPI: API tests to check we can use the OSF v2 API as expected + +You can run these tests using `python -m unittest tests.test_clitool`. diff --git a/src/osfexport/exporter.py b/src/osfexport/exporter.py index bd29ca2..5159992 100644 --- a/src/osfexport/exporter.py +++ b/src/osfexport/exporter.py @@ -30,7 +30,20 @@ class MockAPIResponse: - """Simulate OSF API response for testing purposes.""" + """ + Simulate OSF API response for testing purposes. + + Attributes + ---------------- + + JSON_FILES: static + Key-value dictionary of IDs and paths to stub JSON files. + These are used to generate mock responses to API calls. + + MARKDOWN_FILES: static + Key-value dictionary of IDs and paths to stub Markdown files. + These are used to generate mock responses to API calls to get Wiki data. + """ JSON_FILES = { 'nodes': os.path.join( @@ -227,7 +240,7 @@ def call_api( Personal Access Token to authorise a user with. per_page: int Number of items to include in a JSON page for API responses. - The maximum is 100. + The maximum size is 100. filters: dict Dictionary of query parameters to filter results with. @@ -245,6 +258,9 @@ def call_api( Throws ------------- HTTPError - 429 error if we can't connect to the API after retries. + Other errors are thrown immediately. + + URLError - failed to connect to OSF API Returns ---------- @@ -307,7 +323,7 @@ def call_api( def paginate_json_result(start, action, fail_on_first=True, **kwargs): - """Loop through paginated JSON responses and perform action on each. + """Loop through paginated JSON responses and perform an action on each. Parameters ------------- @@ -333,6 +349,10 @@ def paginate_json_result(start, action, fail_on_first=True, **kwargs): ------------------ results: deque Queue of results per page + + Throws + ------------------ + HTTPError, URLError - non-429 HTTP errors which indicate a problem """ next_link = start @@ -518,7 +538,7 @@ def get_nodes(pat, page_size=100, dryrun=False, project_id='', usetest=False): Personal Access Token to authorise a user with. page_size: int How many nodes to put onto a page. Default is 100. - Possible range is 1-1000 + Possible range is 1-100 dryrun: bool If True, use test data from JSON stubs to mock API calls. project_id: str @@ -577,7 +597,7 @@ def get_nodes(pat, page_size=100, dryrun=False, project_id='', usetest=False): def get_project_data(nodes, **kwargs): - """Pull and list projects for a user from the OSF. + """Pull and list projects for a specific JSON API response page. Parameters ---------- @@ -777,6 +797,8 @@ def get_children(json_page, **kwargs): def get_category(project, **kwargs): + """Get category from a project dictionary""" + # Define nice representations of categories if needed CATEGORY_STRS = { '': 'Uncategorized', @@ -789,6 +811,8 @@ def get_category(project, **kwargs): def get_tags(project, **kwargs): + """Get tags from a project dictionary""" + if project['attributes']['tags']: return ', '.join(project['attributes']['tags']) else: @@ -796,6 +820,16 @@ def get_tags(project, **kwargs): def get_contributors(project, **kwargs): + """Get contributors from a project dictionary + + Parameters + -------------- + dryrun: bool + If True, use test OSF API, otherwise use real API for calls + pat: str + Personal Access Token to authenticate users with. + """ + dryrun = kwargs.pop('dryrun', True) key = kwargs.pop('key', 'contributors') pat = kwargs.pop('pat', '') diff --git a/src/osfexport/formatter.py b/src/osfexport/formatter.py index 811ddbe..44ad58b 100644 --- a/src/osfexport/formatter.py +++ b/src/osfexport/formatter.py @@ -13,7 +13,20 @@ class HTMLImageSizeCapRenderer(HTMLRenderer): - """Custom Markdown to HTML renderer which caps image size.""" + """ + Custom Markdown to HTML renderer which caps image size. + + Attributes + ------------------ + max_width: int + Max width of images. Image widths above this get shrunk. + Default is 300 pixels. + + max_height: int + Max height of images. Image heights above this get shrunk. + default is 300 pixels. + + """ max_width = 300 max_height = 300 @@ -53,6 +66,7 @@ def render_image(self, token): class PDF(FPDF): """Custom PDF class to implement extra customisation. + Attributes: date_printed: datetime Date and time when the project was exported. @@ -97,6 +111,10 @@ def __init__(self, url=''): os.path.dirname(__file__), 'font', 'DejaVuSans-BoldOblique.ttf')) def generate_qr_code(self): + """ + Make a QR code based on the current PDF's URL. + """ + qr = qrcode.make(self.url) img_byte_arr = io.BytesIO() qr.save(img_byte_arr, format='PNG') @@ -104,6 +122,10 @@ def generate_qr_code(self): return img_byte_arr def footer(self): + """ + Makes the PDF footer, including Exported at timestamp + """ + self.set_y(-15) self.set_x(-30) self.set_font(self.font, size=PDF.FONT_SIZES['h5']) @@ -127,7 +149,11 @@ def _write_list_section(self, key, fielddict): key: str Name of the field to write. fielddict: dict - Dictionary containing the field data. + Dictionary containing the data for each field to include. + e.g. { + 'subjects': 's1, s2, s3', + 'title': 'title1' + } """ # Set nicer display names for certain PDF fields @@ -190,8 +216,6 @@ def _write_project_body(self, project): ----------- project: dict Dictionary containing project data to write. - parent_title: str - Title of the parent project. """ self.add_page() @@ -309,6 +333,11 @@ def _write_wiki_pages(self, wikis, title, parent=None): ----------- wikis: dict Dictionary containing wiki data to write. + title: str + Title of the current project. + parent: None | tuple[str, str] + Optional parent project with (name, url) + """ for i, wiki in enumerate(wikis.keys()): @@ -345,8 +374,6 @@ def explore_project_tree(project, projects, pdf=None): List of all projects to explore. pdf: PDF PDF object to write to. If None, a new PDF will be created. - parent_title: str - Title of the parent project. Returns ----------- @@ -394,8 +421,10 @@ def write_pdf(projects, root_idx, folder=''): Returns ------------ - pdfs: list - List of created PDF files. + pdf: PDF + Project export PDF + path: str + Path to the PDF file """ curr_project = projects[root_idx]