Skill System

The skill system provides formal abstractions for Android automation. It is defined in marketing_system/skills/base.py (262 lines) and includes five core classes.

Base Classes

ActionResult

Return value from action execution:

@dataclass
class ActionResult:
    success: bool
    data: dict = field(default_factory=dict)
    error: str = ""
    duration_ms: float = 0.0

ActionResult is truthy when success=True, so you can use it in conditionals: if result: ...

Element

UI locator with fallback chain:

class Element:
    # Locator fields (tried in this order):
    content_desc: str
    text: str
    resource_id: str
    class_name: str
    x: int  # absolute fallback
    y: int

    def find(self, device, xml) -> tuple[int, int] | None:
        """Try each locator in priority order, return center coords."""

Action (Abstract)

Atomic operation with pre/post validation:

class Action:
    name: str
    description: str
    max_retries: int = 2
    retry_delay: float = 1.0

    def precondition(self, dev, xml) -> bool: ...   # Override to validate state
    def execute(self, dev, **kwargs) -> ActionResult: ...  # Must implement
    def postcondition(self, dev, xml) -> bool: ...   # Override to verify success
    def rollback(self, dev) -> None: ...             # Override to undo on failure
    def run(self, dev, **kwargs) -> ActionResult: ...  # Orchestrator

Workflow

Composed action sequence:

class Workflow:
    name: str
    description: str
    params: dict

    def steps(self) -> list: ...   # Override: return [(action_name, params), ...]
    def run(self, dev) -> ActionResult: ...  # Execute all steps in order

Skill

Top-level container that loads from YAML:

class Skill:
    name: str
    app_package: str
    version: str
    elements: dict[str, Element]

    def register_action(self, action_cls): ...
    def register_workflow(self, workflow_cls): ...
    def get_action(self, name, device) -> Action: ...
    def get_workflow(self, name, device, **params) -> Workflow: ...
    def list_actions(self) -> list[str]: ...
    def list_workflows(self) -> list[str]: ...

    @classmethod
    def from_yaml(cls, path) -> Skill: ...  # Load skill.yaml + elements.yaml

Execution Flow

Action.run()

precondition() -> False? -> return failure immediately
      | True
      v
execute() -> loop up to max_retries (default 2):
    |  success + postcondition() True? -> return success
    |  failure? -> rollback(), retry after retry_delay (default 1.0s)
    v
All retries exhausted -> return failure with last error

Workflow.run()

for action in steps():
    result = action.run()
    if not result:
        return failure (completed_steps count + failed_step name)
return success (total completed_steps)

Skill Directory Structure

skills/tiktok/
  skill.yaml           # metadata: name, version, app_package, description
  elements.yaml        # 41 UI elements with fallback locator chains
  __init__.py          # load() function registers actions + workflows
  actions/
    __init__.py
    core.py            # OpenApp, NavigateToProfile, TapSearch, TypeAndSearch, DismissPopup
    engagement.py      # LikePost, CommentOnPost, FollowUser, ScrollFeed, TapUser, etc.
  workflows/
    __init__.py
    upload_video.py
    crawl_hashtag.py
    send_dm.py
    engage_fyp.py
    ... (9 total)

TikTok Skill Reference

13 Actions

Action	File	Has Postcondition
`open_app`	core.py	Yes (checks TIKTOK_PKG in dumpsys)
`navigate_to_profile`	core.py	Yes (checks followers/Following)
`tap_search`	core.py	Yes (checks search box RID)
`type_and_search`	core.py	No
`dismiss_popup`	core.py	No
`like_post`	engagement.py	No
`comment_on_post`	engagement.py	No
`follow_user`	engagement.py	No
`scroll_feed`	engagement.py	No
`tap_user`	engagement.py	No
`tap_message_button`	engagement.py	No
`type_message`	engagement.py	No
`tap_send`	engagement.py	No

9 Workflows

Workflow	Wraps
`upload_video`	`bots/tiktok/upload.py` (43-step flow)
`crawl_hashtag`	`bots/tiktok/scraper.py --tab top`
`crawl_users`	`bots/tiktok/scraper.py --tab users`
`send_dm`	`bots/tiktok/outreach.py`
`engage_fyp`	`bots/tiktok/engage.py`
`scan_inbox`	`bots/tiktok/inbox_scanner.py`
`scrape_analytics`	`bots/tiktok/perf_scanner.py`
`continuous_scrape`	`bots/tiktok/continuous_scrape.py`
`publish_draft`	`bots/tiktok/upload.py --publish-draft`

Base Skill (9 shared actions)

The _base skill provides generic actions usable by any app: tap_element, swipe_direction, type_text, wait, back, home, launch_app, take_screenshot, and press_enter.

Subprocess Execution

When skills are executed via the scheduler or API, they run as subprocesses:

python3 skills/_run_skill.py \
  --skill tiktok \
  --workflow engage_fyp \
  --device L9AIB7603188953 \
  --params '{"duration": 60}'

This keeps the server responsive and provides process isolation.

Files

File	Lines	Purpose
`skills/base.py`	262	ActionResult, Element, Action, Workflow, Skill
`skills/tiktok/`	—	13 actions + 9 workflows + 41 elements
`skills/_base/`	—	9 shared actions
`skills/instagram/`	—	Skeleton (skill.yaml only)
`skills/_run_skill.py`	—	CLI runner for job queue

Skills Overview — concepts and quick start
Elements — locator chains deep dive
API: Skill Classes — full class reference