Manager

psengine.analyst_notes.note_mgr.AnalystNoteMgr

AnalystNoteMgr(rf_token: str = None)

Manages requests for Recorded Future analyst notes.

PARAMETER	DESCRIPTION
rf_token	Recorded Future API token. TYPE: str DEFAULT: None

Source code in psengine/analyst_notes/note_mgr.py

def __init__(self, rf_token: str = None):
    """Initializes the `AnalystNoteMgr` object.

    Args:
        rf_token (str, optional): Recorded Future API token.
    """
    self.log = logging.getLogger(__name__)
    self.rf_client = RFClient(api_token=rf_token) if rf_token else RFClient()

delete

delete(note_id: str) -> bool

Delete an analyst note.

PARAMETER	DESCRIPTION
note_id	The ID of the analyst note to look up. TYPE: str

Endpoint

/analystnote/delete/{note_id}

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
AnalystNoteDeleteError	If connection error occurs.

RETURNS	DESCRIPTION
bool	True if delete is successful, False otherwise.

Source code in psengine/analyst_notes/note_mgr.py

@debug_call
@validate_call
@connection_exceptions(
    ignore_status_code=[404], exception_to_raise=AnalystNoteDeleteError, on_ignore_return=False
)
def delete(
    self, note_id: Annotated[str, Doc('The ID of the analyst note to look up.')]
) -> Annotated[bool, Doc('True if delete is successful, False otherwise.')]:
    """Delete an analyst note.

    Endpoint:
        `/analystnote/delete/{note_id}`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        AnalystNoteDeleteError: If connection error occurs.
    """
    if not note_id.startswith('doc:'):
        note_id = f'doc:{note_id}'

    self.log.info(f'Deleting note {note_id}')
    self.rf_client.request('delete', url=EP_ANALYST_NOTE_DELETE.format(note_id))
    return True

fetch_attachment

fetch_attachment(note_id: str) -> tuple[bytes, str]

Get an analyst note attachment.

To work with the attachment is the same regardless of the file extension.

PARAMETER	DESCRIPTION
note_id	The ID of the note. TYPE: str

Endpoint

/analystnote/attachment/{note_id}

Example

Fetch and save an attachment from an analyst note:

from psengine.analyst_notes import save_attachment

# Note with PDF attachment
attachment, extension = note_mgr.fetch_attachment('tPtLVw')
save_attachment('tPtLVw', attachment, extension)

# Note with YAR attachment
attachment, extension = note_mgr.fetch_attachment('oJeqDP')
save_attachment('oJeqDP', attachment, extension)

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
AnalystNoteAttachmentError	If connection error occurs.

RETURNS	DESCRIPTION
tuple[bytes, str]	A tuple containing the file content (bytes) and the file extension (str).

Source code in psengine/analyst_notes/note_mgr.py

@debug_call
@validate_call
@connection_exceptions(
    ignore_status_code=[404],
    exception_to_raise=AnalystNoteAttachmentError,
    on_ignore_return=(b'', None),
)
def fetch_attachment(
    self,
    note_id: Annotated[str, Doc('The ID of the note.')],
) -> Annotated[
    tuple[bytes, str],
    Doc('A tuple containing the file content (bytes) and the file extension (str).'),
]:
    """Get an analyst note attachment.

    To work with the attachment is the same regardless of the file extension.

    Endpoint:
        `/analystnote/attachment/{note_id}`

    Example:
        Fetch and save an attachment from an analyst note:

        ```python
        from psengine.analyst_notes import save_attachment

        # Note with PDF attachment
        attachment, extension = note_mgr.fetch_attachment('tPtLVw')
        save_attachment('tPtLVw', attachment, extension)

        # Note with YAR attachment
        attachment, extension = note_mgr.fetch_attachment('oJeqDP')
        save_attachment('oJeqDP', attachment, extension)
        ```

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        AnalystNoteAttachmentError: If connection error occurs.
    """
    if not note_id.startswith('doc:'):
        note_id = f'doc:{note_id}'

    self.log.info(f"Looking up analyst note's {note_id} attachment")
    response = self.rf_client.request('get', EP_ANALYST_NOTE_ATTACHMENT.format(note_id))

    content_disp = response.headers.get('Content-Disposition')
    ext = re.findall(r'filename=.*\.(\w+)', content_disp)

    ext = ext[-1] if ext else ''

    return response.content, ext

lookup

lookup(
    note_id: str,
    tagged_text: bool = False,
    serialization: str = 'full',
) -> AnalystNote

Look up an analyst note by ID.

PARAMETER	DESCRIPTION
note_id	The ID of the analyst note to look up. TYPE: str
tagged_text	Add RF IDs to the note entities. TYPE: bool DEFAULT: False
serialization	The serialization type of the payload. TYPE: str DEFAULT: 'full'

Endpoint

/analystnote/lookup/{note_id}

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
AnalystNoteLookupError	If API error occurs.

RETURNS	DESCRIPTION
AnalystNote	The requested note.

Source code in psengine/analyst_notes/note_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[404], exception_to_raise=AnalystNoteLookupError)
def lookup(
    self,
    note_id: Annotated[str, Doc('The ID of the analyst note to look up.')],
    tagged_text: Annotated[bool, Doc('Add RF IDs to the note entities.')] = False,
    serialization: Annotated[str, Doc('The serialization type of the payload.')] = 'full',
) -> Annotated[AnalystNote, Doc('The requested note.')]:
    """Look up an analyst note by ID.

    Endpoint:
        `/analystnote/lookup/{note_id}`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        AnalystNoteLookupError: If API error occurs.
    """
    if not note_id.startswith('doc:'):
        note_id = f'doc:{note_id}'

    data = {'tagged_text': tagged_text, 'serialization': serialization}
    self.log.info(f'Looking up analyst note: {note_id}')
    response = self.rf_client.request(
        'post', EP_ANALYST_NOTE_LOOKUP.format(note_id), data=data
    ).json()
    return AnalystNote.model_validate(response)

preview

preview(
    title: str,
    text: str,
    published: Optional[str] = None,
    topic: Union[str, list[str], None] = None,
    context_entities: Optional[list[str]] = None,
    note_entities: Optional[list[str]] = None,
    validation_urls: Optional[list[str]] = None,
    source: Optional[str] = None,
) -> AnalystNotePreviewOut

Preview of the AnalystNote. It does not create a note; it just returns how the note will look.

PARAMETER	DESCRIPTION
title	The title of the note. TYPE: str
text	The text of the note. TYPE: str
published	The date when the note was published. TYPE: Optional[str] DEFAULT: None
topic	The topic of the note. TYPE: Union[str, list[str], None] DEFAULT: None
context_entities	The context entities of the note. TYPE: Optional[list[str]] DEFAULT: None
note_entities	The note entities of the note. TYPE: Optional[list[str]] DEFAULT: None
validation_urls	The validation URLs of the note. TYPE: Optional[list[str]] DEFAULT: None
source	The source of the note. TYPE: Optional[str] DEFAULT: None

Endpoint

/analystnote/preview

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
AnalystNotePreviewRequest	If connection error occurs.

RETURNS	DESCRIPTION
AnalystNotePreviewOut	The note that will be created.

Source code in psengine/analyst_notes/note_mgr.py

@debug_call
@validate_call
@connection_exceptions(
    ignore_status_code=[404],
    exception_to_raise=AnalystNotePreviewError,
)
def preview(
    self,
    title: Annotated[str, Doc('The title of the note.')],
    text: Annotated[str, Doc('The text of the note.')],
    published: Annotated[Optional[str], Doc('The date when the note was published.')] = None,
    topic: Annotated[Union[str, list[str], None], Doc('The topic of the note.')] = None,
    context_entities: Annotated[
        Optional[list[str]], Doc('The context entities of the note.')
    ] = None,
    note_entities: Annotated[Optional[list[str]], Doc('The note entities of the note.')] = None,
    validation_urls: Annotated[
        Optional[list[str]], Doc('The validation URLs of the note.')
    ] = None,
    source: Annotated[Optional[str], Doc('The source of the note.')] = None,
) -> Annotated[AnalystNotePreviewOut, Doc('The note that will be created.')]:
    """Preview of the AnalystNote. It does not create a note; it just returns how the note
    will look.

    Endpoint:
        `/analystnote/preview`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        AnalystNotePreviewRequest: If connection error occurs.
    """
    if topic:
        topic = topic if isinstance(topic, list) else [topic]

    data = {
        'attributes': {
            'title': title,
            'text': text,
            'published': published,
            'context_entities': context_entities,
            'note_entities': note_entities,
            'validation_urls': validation_urls,
            'topic': topic,
        },
        'source': source,
    }

    note = AnalystNotePreviewIn.model_validate(data)
    self.log.info(f'Previewing note: {note.attributes.title}')
    resp = self.rf_client.request('post', EP_ANALYST_NOTE_PREVIEW, data=note.json()).json()

    return AnalystNotePreviewOut.model_validate(resp)

publish

publish(
    title: str,
    text: str,
    published: Optional[str] = None,
    topic: Union[str, list[str], None] = None,
    context_entities: Optional[list[str]] = None,
    note_entities: Optional[list[str]] = None,
    validation_urls: Optional[list[str]] = None,
    source: Optional[str] = None,
    note_id: Optional[str] = None,
) -> AnalystNotePublishOut

Publish data. This method creates a note and returns its ID.

PARAMETER	DESCRIPTION
title	The title of the note. TYPE: str
text	The text of the note. TYPE: str
published	The date when the note was published. TYPE: Optional[str] DEFAULT: None
topic	The topic of the note. TYPE: Union[str, list[str], None] DEFAULT: None
context_entities	The context entities of the note. TYPE: Optional[list[str]] DEFAULT: None
note_entities	The note entities of the note. TYPE: Optional[list[str]] DEFAULT: None
validation_urls	The validation URLs of the note. TYPE: Optional[list[str]] DEFAULT: None
source	The source of the note. TYPE: Optional[str] DEFAULT: None
note_id	The ID of the note. Use if you want to modify an existing note. TYPE: Optional[str] DEFAULT: None

Endpoint

/analystnote/publish

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
AnalystNotePublishError	If connection error occurs.

RETURNS	DESCRIPTION
AnalystNotePublishOut	The published note.

Source code in psengine/analyst_notes/note_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[404], exception_to_raise=AnalystNotePublishError)
def publish(
    self,
    title: Annotated[str, Doc('The title of the note.')],
    text: Annotated[str, Doc('The text of the note.')],
    published: Annotated[Optional[str], Doc('The date when the note was published.')] = None,
    topic: Annotated[Union[str, list[str], None], Doc('The topic of the note.')] = None,
    context_entities: Annotated[
        Optional[list[str]], Doc('The context entities of the note.')
    ] = None,
    note_entities: Annotated[Optional[list[str]], Doc('The note entities of the note.')] = None,
    validation_urls: Annotated[
        Optional[list[str]], Doc('The validation URLs of the note.')
    ] = None,
    source: Annotated[Optional[str], Doc('The source of the note.')] = None,
    note_id: Annotated[
        Optional[str], Doc('The ID of the note. Use if you want to modify an existing note.')
    ] = None,
) -> Annotated[AnalystNotePublishOut, Doc('The published note.')]:
    """Publish data. This method creates a note and returns its ID.

    Endpoint:
        `/analystnote/publish`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        AnalystNotePublishError: If connection error occurs.
    """
    if topic:
        topic = topic if isinstance(topic, list) else [topic]

    data = {
        'attributes': {
            'title': title,
            'text': text,
            'published': published,
            'context_entities': context_entities,
            'note_entities': note_entities,
            'validation_urls': validation_urls,
            'topic': topic,
        },
        'source': source,
        'note_id': note_id,
    }
    note = AnalystNotePublishIn.model_validate(data)
    self.log.info(f'Publishing note: {note.attributes.title}')
    resp = self.rf_client.request('post', EP_ANALYST_NOTE_PUBLISH, data=note.json()).json()
    return AnalystNotePublishOut.model_validate(resp)

search

search(
    published: Optional[str] = None,
    entity: Optional[str] = None,
    author: Optional[str] = None,
    title: Optional[str] = None,
    topic: Optional[Union[str, list]] = None,
    label: Optional[str] = None,
    source: Optional[str] = None,
    serialization: Optional[str] = None,
    tagged_text: Optional[bool] = None,
    max_results: Optional[int] = Field(
        ge=1, le=1000, default=DEFAULT_LIMIT
    ),
    notes_per_page: Optional[int] = Field(
        ge=1, le=1000, default=NOTES_PER_PAGE
    ),
) -> list[AnalystNote]

Execute a search for the analyst notes based on the parameters provided. Every parameter that has not been set up will be discarded.

If more than one topic is specified, a search for each topic is executed and the AnalystNotes will be deduplicated.

max_results is the maximum number of references, not notes.

PARAMETER	DESCRIPTION
published	Notes published after a date. TYPE: Optional[str] DEFAULT: None
entity	An entity the note refers to, RF ID. TYPE: Optional[str] DEFAULT: None
author	An author of the note, RF ID. TYPE: Optional[str] DEFAULT: None
title	A title of the note. TYPE: Optional[str] DEFAULT: None
topic	A topic of the note, RF ID. TYPE: Optional[Union[str, list]] DEFAULT: None
label	A label of the note, by name. TYPE: Optional[str] DEFAULT: None
source	The source of the note. TYPE: Optional[str] DEFAULT: None
serialization	An entity serializer (id, min, full, raw). TYPE: Optional[str] DEFAULT: None
tagged_text	Should the text contain tags. TYPE: Optional[bool] DEFAULT: None
max_results	The maximum number of references (not notes), max 1000. TYPE: Optional[int] DEFAULT: Field(ge=1, le=1000, default=DEFAULT_LIMIT)
notes_per_page	The number of notes for each paged request. TYPE: Optional[int] DEFAULT: Field(ge=1, le=1000, default=NOTES_PER_PAGE)

Endpoint

/analystnote/search

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
AnalystNoteSearchError	If API error occurs.

RETURNS	DESCRIPTION
list[AnalystNote]	A list of deduplicated AnalystNote objects.

Source code in psengine/analyst_notes/note_mgr.py

@debug_call
@validate_call
def search(
    self,
    published: Annotated[Optional[str], Doc('Notes published after a date.')] = None,
    entity: Annotated[Optional[str], Doc('An entity the note refers to, RF ID.')] = None,
    author: Annotated[Optional[str], Doc('An author of the note, RF ID.')] = None,
    title: Annotated[Optional[str], Doc('A title of the note.')] = None,
    topic: Annotated[Optional[Union[str, list]], Doc('A topic of the note, RF ID.')] = None,
    label: Annotated[Optional[str], Doc('A label of the note, by name.')] = None,
    source: Annotated[Optional[str], Doc('The source of the note.')] = None,
    serialization: Annotated[
        Optional[str], Doc('An entity serializer (id, min, full, raw).')
    ] = None,
    tagged_text: Annotated[Optional[bool], Doc('Should the text contain tags.')] = None,
    max_results: Annotated[
        Optional[int],
        Doc('The maximum number of references (not notes), max 1000.'),
    ] = Field(ge=1, le=1000, default=DEFAULT_LIMIT),
    notes_per_page: Annotated[
        Optional[int], Doc('The number of notes for each paged request.')
    ] = Field(ge=1, le=1000, default=NOTES_PER_PAGE),
) -> Annotated[list[AnalystNote], Doc('A list of deduplicated AnalystNote objects.')]:
    """Execute a search for the analyst notes based on the parameters provided.
    Every parameter that has not been set up will be discarded.

    If more than one topic is specified, a search for each topic is executed and the
    `AnalystNotes` will be deduplicated.

    `max_results` is the maximum number of references, not notes.

    Endpoint:
        `/analystnote/search`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        AnalystNoteSearchError: If API error occurs.
    """
    responses = []
    topic = None if topic == [] else topic
    data = {
        'published': published,
        'entity': entity,
        'author': author,
        'title': title,
        'topic': topic,
        'label': label,
        'source': source,
        'serialization': serialization,
        'taggedText': tagged_text,
        'limit': min(max_results, notes_per_page),
    }
    data = {key: val for key, val in data.items() if val is not None}

    max_results = DEFAULT_LIMIT if max_results is None else max_results

    responses = []
    if isinstance(topic, list) and len(topic):
        for t in topic:
            data['topic'] = t
            responses.append(self._search(data, max_results))
        return list(set(chain.from_iterable(responses)))

    return list(set(self._search(data, max_results)))