Auto-Assign Tasks After Analysis

import csv
import os
from datetime import date
from pathlib import Path
from tempfile import NamedTemporaryFile

from urllib3.util import parse_url

import mindbridgeapi as mbapi

# --- Configure inputs (edit for your environment) ---
# Replace with the full analysis URL from your browser.
# Host and path IDs must match your tenant.
analysis_url = (
    "https://{your-tenant}.mindbridge.ai"
    "/app/organization/{ORG_ID}"
    "/engagement/{ENGAGEMENT_ID}"
    "/analysis/{ANALYSIS_RESULT_ID}"
    "/analyze/financial-statements"
    "?productCode=GENERAL_LEDGER"
)

data_table_query = {
    "effective_date": {
        "$gte": date(2022, 4, 1),
        "$lt": date(2022, 6, 1),
    },
    "risk": {"$gte": 15_00},
}
data_table_logical_name = "gl_journal_lines"
column_for_assigning = "company_code"

# Keys must match values in your data column.
# Values must be real user emails in your tenant.
assigned_user_map = {
    "001": "reviewer-one@yourcompany.com",
    "002": "reviewer-two@yourcompany.com",
}

# --- Step 1: Parse the analysis URL ---
parsed_url = parse_url(analysis_url)
mindbridge_url = parsed_url.host

parsed_url_path = parsed_url.path.split("/")
analysis_idx = parsed_url_path.index("analysis")
analysis_result_id = parsed_url_path[analysis_idx + 1]

# --- Step 2: Connect, load analysis, resolve assignees ---
# Load the API token, connect, fetch the analysis and
# data tables, print logical names, restart data tables
# for a fresh export, and resolve each assignee email
# to a user ID. Errors if an email is missing or
# ambiguous.

token = os.environ.get("MINDBRIDGE_API_TOKEN", "")

server = mbapi.Server(url=mindbridge_url, token=token)

analysis_result = server.analysis_results.get_by_id(
    analysis_result_id
)
analysis = server.analyses.get_by_id(analysis_result.analysis_id)

print("Available data tables:")
for data_table in analysis.data_tables:
    print(
        f"- logical_name: {data_table.logical_name} "
        f"(type: {data_table.type}, id: {data_table.id})"
    )

server.analyses.restart_data_tables(analysis)
data_table = next(
    dt
    for dt in analysis.data_tables
    if dt.logical_name == data_table_logical_name
)
print(
    f"Using logical_name: {data_table.logical_name} "
    f"(type: {data_table.type}, id: {data_table.id})"
)

assigned_user_map_id = {}
for key, assigned_user_email in assigned_user_map.items():
    results = server.users.get(json={"email": assigned_user_email})
    user, *others = results
    if not user:
        raise ValueError(
            f"User with email '{assigned_user_email}' "
            "not found."
        )
    if others:
        raise Exception(
            f"Multiple users found with email "
            f"'{assigned_user_email}'. "
            "Please resolve duplicates."
        )
    assigned_user_map_id[key] = user.id

# --- Step 3: Export data and assign tasks ---
# Export the selected table to CSV. For each row, create
# an entry task for the mapped user unless a task already
# exists for the same analysis, result, transaction, and
# row (safe to rerun).

with NamedTemporaryFile(delete=False) as temp_file:
    temp_file_path = Path(temp_file.name)

print(f"Exporting to: {temp_file_path}")
async_result = server.data_tables.export(
    data_table,
    fields=[
        "rowid",
        "txid",
        "risk",
        "effective_date",
        column_for_assigning,
    ],
    query=data_table_query,
)
server.data_tables.wait_for_export(async_result)
temp_file_path = server.data_tables.download(
    async_result, output_file_path=temp_file_path
)

with temp_file_path.open(newline="", encoding="utf_8") as infile:
    reader = csv.DictReader(infile)
    print("Assigning tasks for the following rows:")
    for row in reader:
        row_id = row["rowid"]
        transaction_id = row["txid"]
        print(
            f"- row_id: {row_id}, "
            f"transaction_id: {transaction_id}, "
            f"risk: {row['risk']}, "
            f"effective_date: {row['effective_date']}"
        )
        assigned_id = assigned_user_map_id[row[column_for_assigning]]
        task = mbapi.TaskItem(
            row_id=row_id,
            transaction_id=transaction_id,
            type=mbapi.TaskType.ENTRY,
            status=mbapi.TaskStatus.OPEN,
            engagement_id=analysis.engagement_id,
            analysis_result_id=analysis_result.id,
            audit_areas=["Audit Area 1"],
            assigned_id=assigned_id,
        )
        existing_task = next(
            server.tasks.get(
                json={
                    "analysisId": analysis.id,
                    "analysisResultId": task.analysis_result_id,
                    "transactionId": task.transaction_id,
                    "rowId": task.row_id,
                }
            ),
            None,
        )
        if existing_task:
            print(
                f"   - Task already exists "
                f"(ID: {existing_task.id})"
            )
        else:
            created_task = server.tasks.create(task)
            print(f"    - Created task (ID: {created_task.id})")

print("Task assignment complete.")
temp_file_path.unlink()