www.PierreHenry.dev - A (very) local automation hub for capturing ideas, prioritising tasks, planning today, running weekly reviews, and exporting everything to Markdown.

automation_hub.py

#!/usr/bin/env python3
"""
Author: PierreHenry.dev

automation_hub.py - A local automation hub for capturing ideas, prioritising tasks, planning today,
running weekly reviews, and exporting everything to Markdown.

Usage examples:
    python automation_hub.py add "Build a Node feature flag article" --type idea --impact 5 --effort 2
    python automation_hub.py add "Ship YouTube thumbnail workflow" --type task --impact 5 --effort 3 --due 2026-03-20
    python automation_hub.py inbox
    python automation_hub.py today
    python automation_hub.py next
    python automation_hub.py review
    python automation_hub.py done ITEM_ID
    python automation_hub.py export

Data is stored in:
    ~/.automation_hub/data.json
"""

from __future__ import annotations

import argparse
import json
import sys
import textwrap
import uuid
from dataclasses import asdict, dataclass, field
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List, Optional


APP_DIR = Path.home() / ".automation_hub"
DATA_FILE = APP_DIR / "data.json"
EXPORT_FILE = APP_DIR / "dashboard.md"

DATE_FORMAT = "%Y-%m-%d"
VALID_TYPES = {"task", "idea"}
VALID_STATUSES = {"inbox", "active", "done", "archived"}


@dataclass
class Item:
    id: str
    title: str
    item_type: str = "task"
    status: str = "inbox"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat(timespec="seconds"))
    updated_at: str = field(default_factory=lambda: datetime.now().isoformat(timespec="seconds"))
    due_date: Optional[str] = None
    impact: int = 3
    effort: int = 3
    energy: int = 3
    context: str = ""
    notes: str = ""
    tags: List[str] = field(default_factory=list)
    project: str = ""
    score: float = 0.0

    def touch(self) -> None:
        self.updated_at = datetime.now().isoformat(timespec="seconds")

    def recalculate_score(self) -> None:
        self.score = calculate_score(self)

    def short_line(self) -> str:
        due = f" due:{self.due_date}" if self.due_date else ""
        tags = f" tags:{','.join(self.tags)}" if self.tags else ""
        project = f" project:{self.project}" if self.project else ""
        return (
            f"[{self.id}] {self.title} "
            f"(type:{self.item_type} status:{self.status} score:{self.score:.2f}"
            f" impact:{self.impact} effort:{self.effort} energy:{self.energy}{due}{project}{tags})"
        )


def ensure_storage() -> None:
    APP_DIR.mkdir(parents=True, exist_ok=True)
    if not DATA_FILE.exists():
        save_data({"items": [], "meta": {"created_at": datetime.now().isoformat(timespec="seconds")}})


def load_data() -> Dict[str, Any]:
    ensure_storage()
    try:
        with DATA_FILE.open("r", encoding="utf-8") as file:
            data = json.load(file)
    except json.JSONDecodeError:
        print("Error: data file is corrupted. Back it up and remove it, or repair the JSON.", file=sys.stderr)
        sys.exit(1)

    data.setdefault("items", [])
    data.setdefault("meta", {})
    return data


def save_data(data: Dict[str, Any]) -> None:
    ensure_storage()
    with DATA_FILE.open("w", encoding="utf-8") as file:
        json.dump(data, file, indent=2, ensure_ascii=False)


def parse_date(value: Optional[str]) -> Optional[date]:
    if not value:
        return None
    try:
        return datetime.strptime(value, DATE_FORMAT).date()
    except ValueError:
        print(f"Error: invalid date '{value}'. Expected format: {DATE_FORMAT}", file=sys.stderr)
        sys.exit(1)


def normalise_tags(tags: Optional[List[str]]) -> List[str]:
    if not tags:
        return []
    clean_tags: List[str] = []
    for tag in tags:
        for piece in tag.split(","):
            cleaned = piece.strip().lower()
            if cleaned and cleaned not in clean_tags:
                clean_tags.append(cleaned)
    return clean_tags


def calculate_score(item: Item) -> float:
    """
    Higher is better.

    Heuristics:
    - higher impact increases score
    - lower effort increases score
    - lower energy increases score
    - due soon increases score
    - tasks slightly prioritised over ideas for actionability
    """
    impact_weight = item.impact * 3.0
    effort_weight = (6 - item.effort) * 1.8
    energy_weight = (6 - item.energy) * 1.2
    type_bonus = 1.2 if item.item_type == "task" else 0.6
    urgency_bonus = 0.0

    if item.due_date:
        due = parse_date(item.due_date)
        if due is not None:
            days_left = (due - date.today()).days
            if days_left < 0:
                urgency_bonus = 6.0
            elif days_left == 0:
                urgency_bonus = 5.0
            elif days_left <= 2:
                urgency_bonus = 4.0
            elif days_left <= 7:
                urgency_bonus = 2.5
            elif days_left <= 14:
                urgency_bonus = 1.0

    return round(impact_weight + effort_weight + energy_weight + type_bonus + urgency_bonus, 2)


def item_from_dict(data: Dict[str, Any]) -> Item:
    item = Item(
        id=data["id"],
        title=data["title"],
        item_type=data.get("item_type", "task"),
        status=data.get("status", "inbox"),
        created_at=data.get("created_at", datetime.now().isoformat(timespec="seconds")),
        updated_at=data.get("updated_at", datetime.now().isoformat(timespec="seconds")),
        due_date=data.get("due_date"),
        impact=int(data.get("impact", 3)),
        effort=int(data.get("effort", 3)),
        energy=int(data.get("energy", 3)),
        context=data.get("context", ""),
        notes=data.get("notes", ""),
        tags=list(data.get("tags", [])),
        project=data.get("project", ""),
        score=float(data.get("score", 0.0)),
    )
    item.recalculate_score()
    return item


def items_from_data(data: Dict[str, Any]) -> List[Item]:
    return [item_from_dict(raw) for raw in data.get("items", [])]


def write_items(data: Dict[str, Any], items: List[Item]) -> None:
    data["items"] = [asdict(item) for item in items]
    data["meta"]["last_updated_at"] = datetime.now().isoformat(timespec="seconds")
    save_data(data)


def generate_id() -> str:
    return uuid.uuid4().hex[:8]


def find_item(items: List[Item], item_id: str) -> Item:
    for item in items:
        if item.id == item_id:
            return item
    print(f"Error: no item found with id '{item_id}'.", file=sys.stderr)
    sys.exit(1)


def validate_score_range(name: str, value: int) -> int:
    if value < 1 or value > 5:
        print(f"Error: {name} must be between 1 and 5.", file=sys.stderr)
        sys.exit(1)
    return value


def validate_type(item_type: str) -> str:
    if item_type not in VALID_TYPES:
        print(f"Error: type must be one of {sorted(VALID_TYPES)}.", file=sys.stderr)
        sys.exit(1)
    return item_type


def validate_status(status: str) -> str:
    if status not in VALID_STATUSES:
        print(f"Error: status must be one of {sorted(VALID_STATUSES)}.", file=sys.stderr)
        sys.exit(1)
    return status


def add_item(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)

    item = Item(
        id=generate_id(),
        title=args.title.strip(),
        item_type=validate_type(args.type),
        status="inbox",
        due_date=args.due,
        impact=validate_score_range("impact", args.impact),
        effort=validate_score_range("effort", args.effort),
        energy=validate_score_range("energy", args.energy),
        context=args.context.strip(),
        notes=args.notes.strip(),
        tags=normalise_tags(args.tags),
        project=args.project.strip(),
    )
    item.recalculate_score()

    items.append(item)
    write_items(data, items)

    print("Added:")
    print(item.short_line())


def list_items(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)

    filtered = items

    if args.status:
        filtered = [item for item in filtered if item.status == args.status]
    if args.type:
        filtered = [item for item in filtered if item.item_type == args.type]
    if args.project:
        filtered = [item for item in filtered if item.project.lower() == args.project.lower()]
    if args.tag:
        filtered = [item for item in filtered if args.tag.lower() in item.tags]

    if args.sort == "score":
        filtered.sort(key=lambda item: item.score, reverse=True)
    elif args.sort == "due":
        filtered.sort(key=lambda item: (item.due_date is None, item.due_date or "9999-12-31", -item.score))
    elif args.sort == "updated":
        filtered.sort(key=lambda item: item.updated_at, reverse=True)
    else:
        filtered.sort(key=lambda item: item.created_at, reverse=True)

    if not filtered:
        print("No items found.")
        return

    for item in filtered:
        print(item.short_line())
        if args.verbose:
            if item.context:
                print(f"  context: {item.context}")
            if item.project:
                print(f"  project: {item.project}")
            if item.notes:
                wrapped = textwrap.fill(item.notes, width=88, initial_indent="  notes: ", subsequent_indent="         ")
                print(wrapped)
            if item.tags:
                print(f"  tags: {', '.join(item.tags)}")


def set_status(args: argparse.Namespace, new_status: str) -> None:
    data = load_data()
    items = items_from_data(data)
    item = find_item(items, args.id)

    item.status = validate_status(new_status)
    item.touch()
    item.recalculate_score()

    write_items(data, items)
    print(f"Updated {item.id} to status '{item.status}'.")


def edit_item(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)
    item = find_item(items, args.id)

    changed = False

    if args.title is not None:
        item.title = args.title.strip()
        changed = True
    if args.type is not None:
        item.item_type = validate_type(args.type)
        changed = True
    if args.status is not None:
        item.status = validate_status(args.status)
        changed = True
    if args.due is not None:
        item.due_date = args.due
        changed = True
    if args.impact is not None:
        item.impact = validate_score_range("impact", args.impact)
        changed = True
    if args.effort is not None:
        item.effort = validate_score_range("effort", args.effort)
        changed = True
    if args.energy is not None:
        item.energy = validate_score_range("energy", args.energy)
        changed = True
    if args.context is not None:
        item.context = args.context.strip()
        changed = True
    if args.notes is not None:
        item.notes = args.notes.strip()
        changed = True
    if args.project is not None:
        item.project = args.project.strip()
        changed = True
    if args.tags is not None:
        item.tags = normalise_tags(args.tags)
        changed = True

    if not changed:
        print("No changes provided.")
        return

    item.touch()
    item.recalculate_score()
    write_items(data, items)

    print("Updated:")
    print(item.short_line())


def plan_today(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)

    candidates = [item for item in items if item.status in {"inbox", "active"}]
    candidates.sort(key=lambda item: item.score, reverse=True)

    selected: List[Item] = []
    total_effort = 0
    effort_limit = args.effort_limit

    for item in candidates:
        if total_effort + item.effort <= effort_limit:
            selected.append(item)
            total_effort += item.effort
        if len(selected) >= args.max_items:
            break

    if not selected:
        print("No suitable items found for today.")
        return

    print(f"Today plan for {date.today().isoformat()}")
    print(f"Effort budget used: {total_effort}/{effort_limit}")
    print("")

    for index, item in enumerate(selected, start=1):
        if item.status == "inbox":
            item.status = "active"
            item.touch()
            item.recalculate_score()
        print(f"{index}. {item.short_line()}")

    write_items(data, items)


def next_actions(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)

    candidates = [item for item in items if item.status in {"inbox", "active"}]
    if args.project:
        candidates = [item for item in candidates if item.project.lower() == args.project.lower()]
    if args.tag:
        candidates = [item for item in candidates if args.tag.lower() in item.tags]

    candidates.sort(key=lambda item: item.score, reverse=True)
    top = candidates[: args.limit]

    if not top:
        print("No next actions found.")
        return

    print("Next actions")
    print("")

    for index, item in enumerate(top, start=1):
        print(f"{index}. {item.short_line()}")
        next_step = build_next_step(item)
        print(f"   next step: {next_step}")


def build_next_step(item: Item) -> str:
    title = item.title.lower()

    if "write" in title or "article" in title or "post" in title:
        return "Create an outline with 3 sections and a working title."
    if "video" in title or "youtube" in title:
        return "Draft the hook, the 3 main talking points, and the thumbnail concept."
    if "build" in title or "ship" in title or "feature" in title:
        return "Define the smallest working version and implement the first thin slice."
    if "automation" in title or "script" in title:
        return "Write the CLI shape, input fields, and one working command."
    if item.item_type == "idea":
        return "Turn this idea into one concrete, testable task."
    return "Define the first action that can be finished in under 30 minutes."


def review_week(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)

    done_items = [item for item in items if item.status == "done"]
    active_items = [item for item in items if item.status == "active"]
    inbox_items = [item for item in items if item.status == "inbox"]

    overdue_items = []
    for item in items:
        if item.status in {"inbox", "active"} and item.due_date:
            due = parse_date(item.due_date)
            if due and due < date.today():
                overdue_items.append(item)

    top_candidates = sorted(
        [item for item in items if item.status in {"inbox", "active"}],
        key=lambda item: item.score,
        reverse=True,
    )[:5]

    print(f"Weekly review for {date.today().isoformat()}")
    print("")
    print(f"Done: {len(done_items)}")
    print(f"Active: {len(active_items)}")
    print(f"Inbox: {len(inbox_items)}")
    print(f"Overdue: {len(overdue_items)}")
    print("")

    if overdue_items:
        print("Overdue")
        for item in sorted(overdue_items, key=lambda entry: entry.due_date or ""):
            print(f"- {item.short_line()}")
        print("")

    if top_candidates:
        print("Top priorities")
        for item in top_candidates:
            print(f"- {item.short_line()}")
        print("")

    ideas = [item for item in items if item.item_type == "idea" and item.status != "archived"]
    if ideas:
        ideas.sort(key=lambda item: item.score, reverse=True)
        print("Top ideas")
        for item in ideas[:5]:
            print(f"- {item.short_line()}")


def export_markdown(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)

    by_status: Dict[str, List[Item]] = {status: [] for status in VALID_STATUSES}
    for item in items:
        by_status[item.status].append(item)

    for status_items in by_status.values():
        status_items.sort(key=lambda item: item.score, reverse=True)

    lines: List[str] = []
    lines.append(f"# Automation Hub Dashboard")
    lines.append("")
    lines.append(f"Generated: {datetime.now().isoformat(timespec='seconds')}")
    lines.append("")

    summary = {
        "total": len(items),
        "inbox": len(by_status["inbox"]),
        "active": len(by_status["active"]),
        "done": len(by_status["done"]),
        "archived": len(by_status["archived"]),
    }

    lines.append("## Summary")
    lines.append("")
    for key, value in summary.items():
        lines.append(f"- {key.title()}: {value}")
    lines.append("")

    for status in ["active", "inbox", "done", "archived"]:
        lines.append(f"## {status.title()}")
        lines.append("")
        if not by_status[status]:
            lines.append("_No items_")
            lines.append("")
            continue

        for item in by_status[status]:
            due_text = f", due {item.due_date}" if item.due_date else ""
            project_text = f", project {item.project}" if item.project else ""
            tags_text = f", tags {', '.join(item.tags)}" if item.tags else ""
            lines.append(
                f"- **{item.title}** `[id: {item.id}]` "
                f"(score {item.score:.2f}, type {item.item_type}{due_text}{project_text}{tags_text})"
            )
            if item.context:
                lines.append(f"  - Context: {item.context}")
            if item.notes:
                lines.append(f"  - Notes: {item.notes}")
        lines.append("")

    APP_DIR.mkdir(parents=True, exist_ok=True)
    EXPORT_FILE.write_text("\n".join(lines), encoding="utf-8")

    print(f"Exported dashboard to: {EXPORT_FILE}")


def delete_item(args: argparse.Namespace) -> None:
    data = load_data()
    items = items_from_data(data)
    original_count = len(items)

    items = [item for item in items if item.id != args.id]

    if len(items) == original_count:
        print(f"Error: no item found with id '{args.id}'.", file=sys.stderr)
        sys.exit(1)

    write_items(data, items)
    print(f"Deleted item '{args.id}'.")


def seed_sample_data(args: argparse.Namespace) -> None:
    data = load_data()

    if data.get("items"):
        print("Seed skipped: data file already contains items.")
        return

    sample_items = [
        Item(
            id=generate_id(),
            title="Ship Python task automation CLI",
            item_type="task",
            status="active",
            due_date=(date.today() + timedelta(days=3)).isoformat(),
            impact=5,
            effort=3,
            energy=3,
            context="Personal systems",
            notes="Finish add, list, today, review, and export commands.",
            tags=["python", "automation", "cli"],
            project="automation-hub",
        ),
        Item(
            id=generate_id(),
            title="Write article about feature flags",
            item_type="idea",
            status="inbox",
            due_date=None,
            impact=4,
            effort=2,
            energy=2,
            context="Content engine",
            notes="Angle: confident shipping without crossing fingers.",
            tags=["writing", "engineering", "content"],
            project="content-system",
        ),
        Item(
            id=generate_id(),
            title="Draft YouTube video about daily progress",
            item_type="task",
            status="inbox",
            due_date=(date.today() + timedelta(days=7)).isoformat(),
            impact=4,
            effort=2,
            energy=3,
            context="YouTube",
            notes="Need title, hook, and thumbnail concept.",
            tags=["youtube", "video"],
            project="youtube-channel",
        ),
    ]

    for item in sample_items:
        item.recalculate_score()

    write_items(data, sample_items)
    print("Seeded sample data.")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="automation_hub",
        description="Capture, prioritise, plan, and review your tasks and ideas.",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    add_parser = subparsers.add_parser("add", help="Add a new task or idea")
    add_parser.add_argument("title", help="Title of the item")
    add_parser.add_argument("--type", default="task", choices=sorted(VALID_TYPES), help="Item type")
    add_parser.add_argument("--due", help=f"Due date in {DATE_FORMAT} format")
    add_parser.add_argument("--impact", type=int, default=3, help="Impact from 1 to 5")
    add_parser.add_argument("--effort", type=int, default=3, help="Effort from 1 to 5")
    add_parser.add_argument("--energy", type=int, default=3, help="Energy from 1 to 5")
    add_parser.add_argument("--context", default="", help="Short context")
    add_parser.add_argument("--notes", default="", help="Free text notes")
    add_parser.add_argument("--tags", nargs="*", help="Tags, space or comma separated")
    add_parser.add_argument("--project", default="", help="Project name")
    add_parser.set_defaults(func=add_item)

    inbox_parser = subparsers.add_parser("inbox", help="List inbox items")
    inbox_parser.add_argument("--sort", default="score", choices=["score", "due", "updated", "created"])
    inbox_parser.add_argument("--verbose", action="store_true")
    inbox_parser.set_defaults(
        func=lambda args: list_items(
            argparse.Namespace(
                status="inbox",
                type=None,
                project=None,
                tag=None,
                sort=args.sort,
                verbose=args.verbose,
            )
        )
    )

    list_parser = subparsers.add_parser("list", help="List items")
    list_parser.add_argument("--status", choices=sorted(VALID_STATUSES))
    list_parser.add_argument("--type", choices=sorted(VALID_TYPES))
    list_parser.add_argument("--project")
    list_parser.add_argument("--tag")
    list_parser.add_argument("--sort", default="score", choices=["score", "due", "updated", "created"])
    list_parser.add_argument("--verbose", action="store_true")
    list_parser.set_defaults(func=list_items)

    today_parser = subparsers.add_parser("today", help="Build a plan for today")
    today_parser.add_argument("--max-items", type=int, default=5, help="Maximum number of items to pick")
    today_parser.add_argument("--effort-limit", type=int, default=12, help="Total effort budget for today")
    today_parser.set_defaults(func=plan_today)

    next_parser = subparsers.add_parser("next", help="Show next best actions")
    next_parser.add_argument("--limit", type=int, default=5, help="Number of items to show")
    next_parser.add_argument("--project", help="Filter by project")
    next_parser.add_argument("--tag", help="Filter by tag")
    next_parser.set_defaults(func=next_actions)

    review_parser = subparsers.add_parser("review", help="Run a weekly review")
    review_parser.set_defaults(func=review_week)

    export_parser = subparsers.add_parser("export", help="Export dashboard to Markdown")
    export_parser.set_defaults(func=export_markdown)

    done_parser = subparsers.add_parser("done", help="Mark an item as done")
    done_parser.add_argument("id", help="Item ID")
    done_parser.set_defaults(func=lambda args: set_status(args, "done"))

    activate_parser = subparsers.add_parser("start", help="Mark an item as active")
    activate_parser.add_argument("id", help="Item ID")
    activate_parser.set_defaults(func=lambda args: set_status(args, "active"))

    archive_parser = subparsers.add_parser("archive", help="Archive an item")
    archive_parser.add_argument("id", help="Item ID")
    archive_parser.set_defaults(func=lambda args: set_status(args, "archived"))

    edit_parser = subparsers.add_parser("edit", help="Edit an existing item")
    edit_parser.add_argument("id", help="Item ID")
    edit_parser.add_argument("--title")
    edit_parser.add_argument("--type", choices=sorted(VALID_TYPES))
    edit_parser.add_argument("--status", choices=sorted(VALID_STATUSES))
    edit_parser.add_argument("--due")
    edit_parser.add_argument("--impact", type=int)
    edit_parser.add_argument("--effort", type=int)
    edit_parser.add_argument("--energy", type=int)
    edit_parser.add_argument("--context")
    edit_parser.add_argument("--notes")
    edit_parser.add_argument("--tags", nargs="*")
    edit_parser.add_argument("--project")
    edit_parser.set_defaults(func=edit_item)

    delete_parser = subparsers.add_parser("delete", help="Delete an item permanently")
    delete_parser.add_argument("id", help="Item ID")
    delete_parser.set_defaults(func=delete_item)

    seed_parser = subparsers.add_parser("seed", help="Seed sample data")
    seed_parser.set_defaults(func=seed_sample_data)

    return parser


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()

    if hasattr(args, "due") and args.due is not None:
        parse_date(args.due)

    args.func(args)


if __name__ == "__main__":
    main()

automation_hub.py Quick Usage

python automation_hub.py seed
python automation_hub.py inbox
python automation_hub.py add "Build content pipeline" --type task --impact 5 --effort 3 --energy 3 --project creator-system --tags python automation
python automation_hub.py today
python automation_hub.py next
python automation_hub.py review
python automation_hub.py export