### Create and Manipulate Tasks in Cerebro with Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Provides Python code examples for creating tasks, setting their properties like priority, progress, and planned time, and establishing relationships between tasks. It also shows how to link tasks and allocate users to them using the pycerebro library.

```python
from pycerebro import database, dbtypes
import datetime

db = database.Database()
db.connect('username', 'password')

# Get root projects
root_tasks = db.root_tasks()
project = next(t for t in root_tasks if t[dbtypes.TASK_DATA_NAME] == 'Test Project')
project_id = project[dbtypes.TASK_DATA_ID]

# Create a new task
task_id = db.add_task(project_id, 'New Animation Task', activity_id=0)

# Set task properties in batch
db.task_set_priority(task_id, dbtypes.TASK_PRIORITY_HIGH)
db.task_set_progress(task_id, 25)
db.task_set_planned_time(task_id, 40.0)  # 40 hours

# Set task start time (days from 2000-01-01 UTC)
now = datetime.datetime.utcnow()
base_date = datetime.datetime(2000, 1, 1)
days_offset = (now - base_date).total_seconds() / (24 * 60 * 60)
db.task_set_start(task_id, days_offset)

# Set finish time (3 days from now)
db.task_set_finish(task_id, days_offset + 3)

# Create subtasks
subtask1 = db.add_task(task_id, 'Modeling')
subtask2 = db.add_task(task_id, 'Rigging')

# Link tasks (subtask1 must complete before subtask2 starts)
link_id = db.set_link_tasks(subtask1, subtask2)

# Allocate users to tasks
users = db.users()
artist = next(u for u in users if u[dbtypes.USER_DATA_LOGIN] == 'artist01')
db.task_set_allocated(subtask1, artist[dbtypes.USER_DATA_ID])
```

--------------------------------

### Python: CerebroHQ Hashtag Management

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Provides code examples for managing hashtags on tasks, messages, and attachments within CerebroHQ. This includes setting, getting, and removing hashtags for each entity type. Requires an initialized and connected `db` object from `pycerebro.database`.

```python
from pycerebro import database, dbtypes

db = database.Database()
db.connect('username', 'password')

task_id = 12345
message_id = 67890

# Set task hashtags
hashtags = {'animation', 'character', 'priority'}
db.task_set_hashtags(task_id, hashtags)

# Get task hashtags
existing_tags = db.task_hashtags(task_id)
print(f"Task hashtags: {existing_tags}")

# Remove specific hashtags
db.task_remove_hashtags(task_id, {'priority'})

# Set message hashtags
db.message_set_hashtags(message_id, {'review', 'approved'})

# Remove message hashtags
db.message_remove_hashtags(message_id, {'approved'})

# Set attachment hashtags (for ATTACHMENT_TAG_FILE or ATTACHMENT_TAG_LINK)
attachment_id = 11111
db.attachment_set_hashtags(attachment_id, {'final', 'render'})

# Get and remove attachment hashtags
attachment_tags = db.attachment_hashtags(attachment_id)
db.attachment_remove_hashtags(attachment_id, {'render'})

```

--------------------------------

### Front-End Plugin Event Handling - Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Provides an example of creating client-side plugins in Cerebro HQ that respond to application events. It defines handler functions (`before_event`, `after_event`, `error_event`) to react to task changes, message additions, and file attachments, allowing for event modification or cancellation.

```python
# File: front-plugins/my_plugin/event.py
"""
Event handler for client-side plugin.
Responds to before_event, after_event, and error_event.
"""

import cerebro

def before_event(event):
    """Called before data is changed"""
    event_type = event.type()

    if event_type == cerebro.EVENT_TASK_CHANGE:
        task = event.task()
        print(f"Task about to change: {task.name()}")

        # Get old values
        old_progress = task.progress()

        # Can modify event or cancel it
        # event.cancel()

    elif event_type == cerebro.EVENT_MESSAGE_ADD:
        message = event.message()
        print(f"Message about to be added to task {message.task().name()}")

def after_event(event):
    """Called after data has been successfully changed"""
    event_type = event.type()

    if event_type == cerebro.EVENT_TASK_CHANGE:
        task = event.task()
        print(f"Task changed: {task.name()}")
        print(f"New progress: {task.progress()}%")

        # Perform post-change actions
        if task.progress() == 100:
            print("Task completed!")

    elif event_type == cerebro.EVENT_ATTACHMENT_ADD:
        attachment = event.attachment()
        print(f"File attached: {attachment.name()}")

def error_event(error, event):
    """Called when an error occurs during data change"""
    event_type = event.type()
    error_message = str(error)

    print(f"Error occurred on event type {event_type}: {error_message}")

    if event_type == cerebro.EVENT_TASK_CHANGE:
        task = event.task()
        print(f"Failed to change task: {task.name()}")
```

--------------------------------

### File Storage Operations (Upload/Download) - Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Shows how to upload and download files using the Cerebro HQ's `Cargador` file storage service. It includes initialization, setting proxies, importing files with a task URL, converting hash formats using `cclib`, and downloading files.

```python
from pycerebro import cargador, cclib

# Initialize Cargador connection
Carga = cargador.Cargador(
    _host='storage.example.com',
    _rpc_port=4040,
    _http_port=4080,
    _http_proxy=''  # Optional proxy
)

# Set proxy if needed
Carga.set_proxy('proxy.example.com:8080')

# Import file to storage
file_path = '/local/path/file.mp4'
task_url = 'Test Project/Animation Task'
hash_base64 = Carga.import_file(file_path, task_url)
print(f"File uploaded with hash: {hash_base64}")

# Convert hash formats
hash_base16 = cclib.hash64_16(hash_base64)
print(f"Hash in base16: {hash_base16}")

# Download file from storage
download_path = '/local/path/downloaded_file.mp4'
success = Carga.download_file(download_path, hash_base64)
if success:
    print("File downloaded successfully")

# XML-RPC methods (inherited from ServerProxy)
status_info = Carga.statusInfo()
print(f"Cargador status: {status_info}")
```

--------------------------------

### Python: CerebroHQ Core Task and File Operations

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Demonstrates core CerebroHQ functionalities including creating file storage connections, adding task definitions, attaching files and thumbnails, creating reports with work times, and adding reviews and notes. Uses the `pycerebro` library and assumes a `db` object is initialized and connected.

```python
from pycerebro import database, dbtypes

# Assuming db is already initialized and connected
# db = database.Database()
# db.connect('username', 'password')

# Create Cargador connection for file storage
carga = cargador.Cargador('storage.example.com', 4040, 4080)

# Add task definition
task_id = 12345
definition_id = db.add_definition(task_id, '<h1>Task Description</h1><p>Complete the character animation.</p>')

# Attach file to definition
file_path = '/path/to/animation.mp4'
thumbnails = ['/path/to/thumb1.jpg', '/path/to/thumb2.jpg', '/path/to/thumb3.jpg']
attachment_id = db.add_attachment(
    message_id=definition_id,
    carga=carga,
    filename=file_path,
    thumbnails=thumbnails,
    description='Final animation render',
    as_link=False  # False=import to storage, True=link only
)

# Add report with work time
parent_message_id = db.task_definition(task_id)[dbtypes.MESSAGE_DATA_ID]
report_id = db.add_report(
    task_id=task_id,
    message_id=parent_message_id,
    html_text='<p>Completed animation sequence</p>',
    minutes=180  # 3 hours
)

# Attach rendered frames
db.add_attachment(report_id, carga, '/renders/shot_001.exr', [], 'Shot 001 render', False)

# Add review message
review_id = db.add_review(task_id, report_id, '<p>Looks good, approved!</p>', minutes=30)

# Add note
note_id = db.add_note(task_id, review_id, '<p>Remember to update the rig.</p>')

```

--------------------------------

### Import Tasks and Properties from Excel using Python

Source: https://github.com/cerebrohq/cerebro-plugins/blob/master/end-plugins/example/README.md

This script demonstrates the process of importing tasks and their properties from an Excel file into the Cerebro system. It requires an Excel template and uses a library like Pandas for reading the file.

```python
import pandas as pd
# Assuming cerebrohq is in the Python path or installed
# from cerebro.db import DB
# from cerebro.task import Task

def import_tasks_from_excel(input_filename='import.xlsx'):
    try:
        df = pd.read_excel(input_filename)
        print(f"Successfully read {input_filename}")
    except FileNotFoundError:
        print(f"Error: The file {input_filename} was not found.")
        return
    except Exception as e:
        print(f"Error reading Excel file: {e}")
        return

    print("Connecting to database...")
    # db = DB()
    # db.connect()

    imported_count = 0
    for index, row in df.iterrows():
        try:
            # Placeholder for actual task creation logic
            # Task.create(db, name=row['Name'], description=row['Description'], ...)
            print(f"Importing task: {row.get('Name', 'Unnamed Task')}")
            imported_count += 1
        except Exception as e:
            print(f"Error importing row {index}: {e}")

    print(f"Import process completed. {imported_count} tasks processed.")

if __name__ == "__main__":
    print("Running Excel import script...")
    import_tasks_from_excel()

```

--------------------------------

### Connect to Cerebro Database with Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Demonstrates how to establish a connection to the Cerebro database using the pycerebro library. It covers connecting via a running Cerebro client or directly using credentials, and includes methods for checking connection status and retrieving the current user ID.

```python
from pycerebro import database

db = database.Database()

# Method 1: Connect via running Cerebro client (preserves client connection)
status = db.connect_from_cerebro_client()
if status != 0:
    # Status codes: 0=success, 1=client running but not logged in, 2=client not running
    # Fallback to credentials
    db.connect('username', 'password')

# Method 2: Direct connection with credentials
db.connect('username', 'password')

# Method 3: Direct connection to specific server
db.connect('username', 'password', primary_url='https://server.example.com')

# Check connection status
if db.is_connected():
    print(f"Connected with token: {db.token}")
    user_id = db.current_user_id()
    print(f"Current user ID: {user_id}")
```

--------------------------------

### Create Tasks and Messages via Python API

Source: https://github.com/cerebrohq/cerebro-plugins/blob/master/end-plugins/example/README.md

Demonstrates establishing database connections, creating tasks, linking them, and creating messages with file attachments using Python, bypassing the Cerebro GUI. This script is useful for programmatic task and message generation.

```python
import os
import sys

# Assuming cerebrohq is in the Python path or installed
# from cerebro.db import DB
# from cerebro.task import Task
# from cerebro.message import Message

def create_sample_tasks():
    # Placeholder for actual Cerebro API calls
    print("Connecting to database...")
    # db = DB()
    # db.connect()

    print("Creating tasks...")
    # task1 = Task.create(db, name='Task 1', description='First task')
    # task2 = Task.create(db, name='Task 2', description='Second task', depends_on=[task1.id])

    print("Creating message with attachment...")
    # message = Message.create(db, subject='Report Data', body='See attached report.')
    # attachment_path = '/path/to/your/report.pdf' # Example attachment path
    # if os.path.exists(attachment_path):
    #     message.attach_file(attachment_path)
    #     print(f"Attached file: {attachment_path}")
    # else:
    #     print(f"Attachment not found: {attachment_path}")

    print("Tasks and message created successfully.")

if __name__ == "__main__":
    # Example of how to run this script
    # Ensure your environment is set up to run Cerebro scripts
    print("Running task creation script...")
    create_sample_tasks()

```

--------------------------------

### Query Tasks and Retrieve Data in Cerebro with Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Illustrates how to query tasks, retrieve task data, and access related information such as children, allocated users, attachments, and messages using the pycerebro library. It also shows how to fetch tasks by URL or by a set of IDs.

```python
from pycerebro import database, dbtypes

db = database.Database()
db.connect('username', 'password')

# Get user's assigned tasks
user_id = db.current_user_id()
todo_tasks = db.to_do_task_list(user_id, with_done_task=False)

for task in todo_tasks:
    task_id = task[dbtypes.TASK_DATA_ID]
    task_name = task[dbtypes.TASK_DATA_NAME]
    progress = task[dbtypes.TASK_DATA_PROGRESS]
    status_id = task[dbtypes.TASK_DATA_STATUS]

    print(f"Task: {task_name} ({progress}%)")

    # Get task details
    children = db.task_children(task_id)
    allocated_users = db.task_allocated(task_id)
    attachments = db.task_attachments(task_id)
    messages = db.task_messages(task_id)

    # Get task by URL
    task_url = '/Test Project/Animation/Character Rig'
    task_data = db.task_by_url(task_url)

    # Access specific task
    single_task = db.task(task_id)
    if single_task:
        print(f"Priority: {single_task[dbtypes.TASK_DATA_PRIORITY]}")
        print(f"Budget: {single_task[dbtypes.TASK_DATA_BUDGET]}")

# Query multiple tasks efficiently
task_ids = {101, 102, 103}
tasks = db.tasks(task_ids)
```

--------------------------------

### Create Reports with File Attachments via Python API

Source: https://github.com/cerebrohq/cerebro-plugins/blob/master/end-plugins/example/README.md

This script demonstrates connecting to the database, creating messages, and attaching files to them without using the Cerebro GUI. It's designed for scenarios where reports need to be generated programmatically from external software like Nuke or Maya.

```python
import os
import sys

# Assuming cerebrohq is in the Python path or installed
# from cerebro.db import DB
# from cerebro.message import Message

def create_report_message(report_title, report_body, attachment_path):
    print("Connecting to database...")
    # db = DB()
    # db.connect()

    print(f"Creating report message: {report_title}")
    # message = Message.create(db, subject=report_title, body=report_body)

    if os.path.exists(attachment_path):
        # message.attach_file(attachment_path)
        print(f"Attached file: {attachment_path}")
    else:
        print(f"Attachment not found: {attachment_path}")

    print("Report message created successfully.")

if __name__ == "__main__":
    # Example usage:
    # Assume you have a generated report file at '/path/to/your/report.pdf'
    # and you are running this script from an environment that can execute Cerebro Python scripts.
    
    example_report_title = "Daily Render Report"
    example_report_body = "Please find the daily render statistics and logs attached."
    example_attachment_path = "/path/to/your/report.pdf" # Replace with actual path

    print("Running report creation script...")
    # create_report_message(example_report_title, example_report_body, example_attachment_path)
    print("Script finished. Uncomment API calls to enable functionality.")

```

--------------------------------

### Python: CerebroHQ Task Copying and Duplication

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Shows how to copy and duplicate tasks in CerebroHQ, including their associated properties and relationships. Defines various copy flags to control which elements are duplicated, such as sub-tasks, internal links, tags, assigned users, and events. Requires an initialized and connected `db` object.

```python
from pycerebro import database, dbtypes

db = database.Database()
db.connect('username', 'password')

# Define copy flags
flags = (
    dbtypes.COPY_TASKS_SUB_TASKS |
    dbtypes.COPY_TASKS_INTERNAL_LINKS |
    dbtypes.COPY_TASKS_TAGS |
    dbtypes.COPY_TASKS_ASSIGNED_USERS |
    dbtypes.COPY_TASKS_EVENTS
)

# Example of how to use flags (actual copy operation not shown)
# copied_task_ids = db.copy_tasks(task_ids_to_copy, flags=flags)

```

--------------------------------

### HTML to Text Conversion Service Module

Source: https://github.com/cerebrohq/cerebro-plugins/blob/master/end-plugins/example/README.md

A service module used by other scripts (like excel_export.py) to convert HTML content to plain text. This is typically used for cleaning up text before processing or display.

```python
from bs4 import BeautifulSoup

def html_to_text(html_content):
    """Converts HTML content to plain text using BeautifulSoup."""
    if not html_content:
        return ""
    try:
        soup = BeautifulSoup(html_content, 'html.parser')
        return soup.get_text()
    except Exception as e:
        print(f"Error converting HTML to text: {e}")
        return html_content # Return original content if conversion fails

if __name__ == '__main__':
    sample_html = """<html><body><h1>Title</h1><p>This is a <b>bold</b> paragraph.</p><ul><li>Item 1</li><li>Item 2</li></ul></body></html>"""
    text_output = html_to_text(sample_html)
    print("--- Original HTML ---")
    print(sample_html)
    print("\n--- Converted Text ---")
    print(text_output)

```

--------------------------------

### Custom SQL Queries - Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Illustrates how to execute custom SQL queries in Cerebro HQ using the `pycerebro.database` module. It covers read-only queries with parameters, executing queries to retrieve user lists based on flags, and fetching activities and statuses.

```python
from pycerebro import database, dbtypes

db = database.Database()
db.connect('username', 'password')

# Execute read-only query
query = '''
    SELECT uid, name, progress
    FROM "taskQuery_11"(?)
    WHERE progress < ?
'''
task_ids = {100, 101, 102}
threshold = 50

results = db.execute(query, task_ids, threshold)
for row in results:
    task_id, task_name, progress = row
    print(f"Task {task_name}: {progress}% complete")

# Execute with parameters
query_with_params = '''
    SELECT * FROM "userList"()
    WHERE "userGetFlags"(uid) & ? != 0
'''
user_flag = 1 << dbtypes.USER_FLAG_IS_RESOURCE
resources = db.execute(query_with_params, user_flag)

# Get activities
activities = db.activities()
for activity in activities:
    activity_id = activity[dbtypes.ACTIVITY_DATA_ID]
    activity_name = activity[dbtypes.ACTIVITY_DATA_NAME]
    print(f"Activity: {activity_name} (ID: {activity_id})")

# Get all statuses
statuses = db.statuses()
for status in statuses:
    status_id = status[dbtypes.STATUS_DATA_ID]
    status_name = status[dbtypes.STATUS_DATA_NAME]
    print(f"Status: {status_name}")
```

--------------------------------

### Export Task Properties to Excel using Python

Source: https://github.com/cerebrohq/cerebro-plugins/blob/master/end-plugins/example/README.md

This script demonstrates how to export task properties from the Cerebro system into an Excel file. It requires a database connection and likely utilizes a library like Pandas or XlsxWriter for Excel generation.

```python
import pandas as pd
# Assuming cerebrohq is in the Python path or installed
# from cerebro.db import DB
# from cerebro.task import Task

def export_tasks_to_excel(output_filename='tasks_export.xlsx'):
    print("Connecting to database...")
    # db = DB()
    # db.connect()

    print("Fetching tasks...")
    # tasks = Task.all(db) # Example: Fetch all tasks
    # task_data = []
    # for task in tasks:
    #     task_data.append({
    #         'ID': task.id,
    #         'Name': task.name,
    #         'Description': task.description,
    #         'Status': task.status,
    #         'Created': task.created_at,
    #         'Dependencies': ','.join(map(str, task.dependencies)) # Example property
    #     })

    # Placeholder data if actual fetching is not available
    task_data = [
        {'ID': 1, 'Name': 'Task A', 'Description': 'Description A', 'Status': 'Open', 'Created': '2023-01-01', 'Dependencies': ''},
        {'ID': 2, 'Name': 'Task B', 'Description': 'Description B', 'Status': 'In Progress', 'Created': '2023-01-02', 'Dependencies': '1'}
    ]

    if not task_data:
        print("No tasks found to export.")
        return

    df = pd.DataFrame(task_data)

    try:
        df.to_excel(output_filename, index=False)
        print(f"Successfully exported tasks to {output_filename}")
    except Exception as e:
        print(f"Error exporting to Excel: {e}")

if __name__ == "__main__":
    print("Running Excel export script...")
    export_tasks_to_excel()

```

--------------------------------

### Python: CerebroHQ Batch Task Operations

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Demonstrates efficient batch operations on multiple tasks in CerebroHQ. This includes retrieving tasks, extracting IDs, and then modifying priorities, statuses, progress, planned times, flags, allocated users, and hashtags for a collection of tasks simultaneously. Utilizes `pycerebro.database` and `pycerebro.dbtypes`.

```python
from pycerebro import database, dbtypes

db = database.Database()
db.connect('username', 'password')

# Get multiple tasks
user_id = db.current_user_id()
tasks = db.to_do_task_list(user_id, with_done_task=False)

# Extract task IDs
task_ids = {task[dbtypes.TASK_DATA_ID] for task in tasks[:10]}

# Batch operations on multiple tasks
db.task_set_priority(task_ids, dbtypes.TASK_PRIORITY_ABOVE_NORMAL)
db.task_set_status(task_ids, status_id=5)
db.task_set_progress(task_ids, 50)
db.task_set_planned_time(task_ids, 8.0)

# Set flags on multiple tasks
db.task_set_flag(task_ids, dbtypes.TASK_FLAG_CLOSED, is_set=True)

# Allocate users to multiple tasks
artist_ids = {101, 102, 103}
db.task_set_allocated(task_ids, artist_ids)

# Remove users from multiple tasks
db.task_remove_allocated(task_ids, {101})

# Set hashtags on multiple tasks
db.task_set_hashtags(task_ids, {'batch', 'updated'})

```

--------------------------------

### Python: CerebroHQ Task Tag Management

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Illustrates how to manage task tags and custom fields in CerebroHQ. This includes fetching available project tags, setting string, integer, float, and enumeration tags on tasks, retrieving existing task tags, and resetting tag values. It utilizes the `pycerebro.database` and `pycerebro.dbtypes` modules.

```python
from pycerebro import database, dbtypes

db = database.Database()
db.connect('username', 'password')

project_id = 100
task_id = 12345

# Get available tags for project
tags = db.project_tags(project_id)
for tag in tags:
    tag_id = tag[dbtypes.TAG_DATA_ID]
    tag_name = tag[dbtypes.TAG_DATA_NAME]
    tag_type = tag[dbtypes.TAG_DATA_TYPE]
    print(f"Tag: {tag_name} (Type: {tag_type})")

# Set string tag
string_tag_id = 50
db.task_set_tag_string(task_id, string_tag_id, 'Character A')

# Set integer tag
int_tag_id = 51
db.task_set_tag_int(task_id, int_tag_id, 42)

# Set float tag
float_tag_id = 52
db.task_set_tag_float(task_id, float_tag_id, 3.14159)

# Set enumeration tag
enum_tag_id = 53
enums = db.tag_enums(enum_tag_id)
selected_enum = enums[0][dbtypes.TAG_ENUM_DATA_ID]
db.task_set_tag_enum(task_id, selected_enum, is_set=True)

# Get task tags
task_tags = db.task_tags(task_id)
for tag_value in task_tags:
    print(f"Tag value: {tag_value[dbtypes.TASK_TAG_DATA_VALUE]}")

# Reset tag value
db.task_tag_reset(task_id, string_tag_id)

```

--------------------------------

### Add Messages and Attach Files to Tasks in Cerebro with Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Provides Python code for adding messages to tasks and attaching files, including support for thumbnails, using the pycerebro and cargador libraries. It demonstrates how to interact with the database and file storage services.

```python
from pycerebro import database, dbtypes, cargador, cclib
import os

db = database.Database()
db.connect('username', 'password')
```

--------------------------------

### Copy Tasks Multiple Times (Replication) - Python

Source: https://context7.com/cerebrohq/cerebro-plugins/llms.txt

Demonstrates how to copy single or multiple tasks in Cerebro HQ using the `db.copy_tasks` function. This function takes a parent task ID and a list of tasks to create, optionally including flags. It returns a list of newly created task IDs.

```python
source_task_id = 12345
parent_task_id = 100
tasks_to_create = [
    (source_task_id, 'Shot 001'),
    (source_task_id, 'Shot 002'),
    (source_task_id, 'Shot 003'),
    (source_task_id, 'Shot 004'),
]

new_task_ids = db.copy_tasks(parent_task_id, tasks_to_create, flags)
print(f"Created {len(new_task_ids)} new tasks: {new_task_ids}")

# Copy multiple different tasks
tasks_to_copy = [
    (111, 'Animation Copy'),
    (222, 'Rigging Copy'),
    (333, 'Modeling Copy'),
]
copied_ids = db.copy_tasks(parent_task_id, tasks_to_copy, flags)
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.