Code Example
Process data to ERP
An example how to use Aiviro to create a story, which at first reads emails from a mailbox and extracts the data from the email’s attachment, then it continues by adding this data into the ERP. In the end, all opened windows are closed by robot and a report is sent.
Main Story
An example how to use Aiviro, which handles the main loop with all the required actions and steps. To see more information about used classes, please see
BaseScenario, ShellCommands, InvoiceReader.Download here
main_story_example.pyfrom collections.abc import Generator
from pathlib import Path
import aiviro
from aiviro.modules.config import YAMLConfig
from aiviro.modules.pdf import create_pdf_robot
from aiviro.modules.powershell import ShellCommands
from aiviro.modules.reader import InvoiceData, InvoiceReader
from docs.code_examples.erp_handler_example import ERPHandler
class MainStoryXX(aiviro.BaseScenario):
"""
This class inherits from BaseScenario and implements the main logic of the story.
This class handles a process of parsing attachments from an email and adding this
data to an ERP system. In the end, all opened windows are closed by a shell command
and a report is generated and sent to the list of recipients defined in the config file.
"""
def __init__(self, config: YAMLConfig):
super().__init__(config)
self.pdf_robot = create_pdf_robot()
# name of the robot is set in the config file
self.r = self.robot(robot_name="rdp_robot")
self.erp_handler = ERPHandler(
rdp_robot=self.r, export_folder=self.tmp_folder.unique_folder
)
self.invoice_reader = InvoiceReader(self.pdf_robot)
self.shell_commands = ShellCommands(self.r)
self.report = self.report_builder()
def _before_run(self) -> None:
"""
e.g. wait until robot is connected to the RDP machine
"""
self.r.wait_until_ready()
def _after_run(self) -> None:
"""
e.g. close all opened applications, send a report etc.
"""
self.shell_commands.close_all_open_windows().execute()
self.send_report_via_aiviro_service(self.config.recipients)
def _run(self):
"""
This method handles the main loop of the story
"""
for parsed_data, pdf_path in self._get_data_from_emails_pdf():
try:
self.erp_handler.handle_story_xx(data=parsed_data, pdf_path=pdf_path)
except Exception as e: # noqa: BLE001
self.report.error(
f"An error occurred when processing PDF file {pdf_path} - {e}"
)
else:
self.report.successful(
f"Data from PDF file {pdf_path} was successfully added to ERP"
)
@aiviro.step
def _get_data_from_emails_pdf(
self,
) -> Generator[tuple[InvoiceData, Path], None, None]:
"""
All the required parameters for email extractor can be set in the
.yaml config file , for example:
- maximum number of emails to be processed by setting variable
'max_valid_emails'
- conditions for email attachments by setting variable 'attachment_conditions',
e.g. if email contains pdf files within its attachments, then each
attachment is checked if it is a pdf file, otherwise it is skipped
"""
for _email, attachments in self.email_extractor().extract_all():
for one_attachment in attachments:
if one_attachment.suffix != ".pdf":
continue
data = self._parse_file(one_attachment)
yield data, one_attachment
@aiviro.step
def _parse_file(self, one_attachment: Path) -> InvoiceData:
# set required pdf file as a source for the pdf robot
self.pdf_robot.parse(one_attachment)
return self.invoice_reader.parse()
ERP Handler
An example how to use Aiviro to handle actions in ERP system. In this case, Aiviro robot is used to add parsed data from the pdf file to the ERP system.
Download here
erp_handler_example.pyimport datetime
import pathlib
import aiviro
from aiviro.modules.reader import InvoiceData
class ERPHandler:
def __init__(self, rdp_robot: "aiviro.RDPRobot", export_folder: pathlib.Path):
self.r = rdp_robot
self._export_folder = export_folder
@aiviro.step
def handle_story_xx(self, data: InvoiceData, pdf_path: pathlib.Path) -> None:
"""Method called to handle whole process of adding data to ERP"""
self._start_erp()
self._open_invoices_with_new_record()
self._write_to_new_record(data)
self._attach_pdf(pdf_path)
self._save_new_record()
self._close_erp()
@aiviro.step
def _start_erp(self) -> None:
"""Method to start ERP program via WIN+R shortcut"""
self.r.start_process(r"path\to\your\erp\program.exe")
@aiviro.step
def _open_invoices_with_new_record(self) -> None:
"""Method to open invoices tab in ERP"""
self.r.double_click(aiviro.Text("Přijaté faktury", element_index=0))
self.r.click(aiviro.Text("Nový záznam", element_index=0))
@aiviro.step
def _write_to_new_record(self, data: InvoiceData) -> None:
"""Method to write provided data to new record"""
self.r.type_text(
element=aiviro.Input("Číslo dokladu", element_index=0),
text_to_type=data.invoice_id.value,
)
self.r.type_text(
element=aiviro.Input("Celková částka", element_index=0),
text_to_type=str(data.total_amount.value),
)
self.r.type_text(
element=aiviro.Input("Celková částka bez DPH", element_index=0),
text_to_type=str(data.total_amount_without_tax.value),
)
self.r.type_text(
element=aiviro.Input("Daňový doklad", element_index=0),
text_to_type=datetime.date.strftime(data.tax_date.value, "%d.%m.%Y"),
)
@aiviro.step
def _attach_pdf(self, pdf_path: pathlib.Path) -> None:
"""Method to attach pdf to the new record"""
self.r.transfer_files_to_guests_clipboard(pdf_path)
self.r.click(aiviro.Button("Připojit ze schránky", element_index=0))
@aiviro.step
def _save_new_record(self) -> None:
"""Method to save new record simply by clicking on the Save button.
And transfer generated files from guest clipboard to the export folder on the host machine.
"""
self.r.click(aiviro.Button("Uložit", element_index=0))
self.r.click(aiviro.Text("Nový záznam", element_index=0))
self.r.transfer_files_from_guests_clipboard(self._export_folder, copy=True)
@aiviro.step
def _close_erp(self) -> None:
"""Method to close ERP program"""
self.r.click(aiviro.Button("Zavřít", element_index=0))
self.r.click(aiviro.Button("Ano", element_index=0))
Run Editor Scenario
An example how to run a scenario exported from the Aiviro Editor inside a
BaseScenario. The Editor exports each flow as a standalone module defining a main(robot, input_variables) entry-point, which is loaded and executed via _run_app_script(). This way the exported flow runs together with Core’s logging, metrics, reporting and error handling.Download here
run_editor_scenario.pyimport pathlib
import aiviro
from aiviro.modules.config import YAMLConfig
class RunEditorScenario(aiviro.BaseScenario):
"""Runs a scenario exported from the Aiviro Editor.
The Editor exports each flow as a standalone Python module that defines a
``main(robot, input_variables)`` entry-point. :meth:`~.BaseScenario._run_app_script`
loads that module dynamically and calls its ``main`` function, so the exported
script runs inside a regular Core scenario together with its logging, metrics,
reporting and error handling.
"""
#: folder with the scripts exported from the Editor
EXPORTED_SCRIPTS = pathlib.Path(__file__).parent / "exported"
def __init__(self, config: YAMLConfig):
super().__init__(config)
# the robot name is defined in the .yaml config file
self.r = self.robot(robot_name="web_robot")
def _before_run(self) -> None:
self.r.wait_until_ready()
def _run(self) -> None:
# ``input_variables`` are forwarded to the exported script's ``main`` function,
# where they are exposed as the flow's input variables
self._run_app_script(
self.EXPORTED_SCRIPTS / "login_flow.py",
robot=self.r,
input_variables={"username": "aiviro", "password": "secret"},
)
if __name__ == "__main__":
# using the context manager closes the scenario automatically on exit
main_config = YAMLConfig("main_config.yaml")
with RunEditorScenario(main_config) as scenario:
scenario.start()
The exported script referenced above only needs to expose a main function with the following signature:
def main(robot, input_variables):
# automation logic generated by the Editor
...