hermes-dashboard/services/task_project_linker.py

"""Task <-> Project linker.

Maps kanban tasks to managed projects by title keywords + workspace_path
substring matching. Used by /projects/<name>, /timeline, and /tasks/<id>.
"""

import logging
import sqlite3
import subprocess
from typing import Any

import config
from extensions import get_db

logger = logging.getLogger(__name__)

_TITLE_KEYWORDS: list[tuple[str, str]] = [
    ("blog-app", "blog-app"),
    ("novel-app", "novel-app"),
    ("novel-writer", "novel-app"),
    ("hermes-dashboard", "hermes-dashboard"),
    ("gitea", "hermes-dashboard"),
    ("omo-openagent", "omo-openagent"),
    ("omo-config", "omo-config"),
    ("opencode", "omo-config"),
]


def link_task_to_project(task_id: str) -> str | None:
    """Return project name for a task_id, or None if no match.

    Match priority: workspace_path substring > title keyword.
    """
    try:
        db = get_db()
        row = db.execute(
            "SELECT title, workspace_path FROM tasks WHERE id = ?", (task_id,)
        ).fetchone()
    except sqlite3.Error:
        logger.exception("link_task_to_project db error for %s", task_id)
        return None

    if not row:
        return None

    title = (row["title"] or "").lower()
    workspace = (row["workspace_path"] or "").lower()

    for proj in config.MANAGED_PROJECTS:
        proj_path = proj["path"].lower()
        if proj_path and proj_path in workspace:
            return proj["name"]

    for keyword, project_name in _TITLE_KEYWORDS:
        if keyword in title:
            return project_name

    return None


def link_project_tasks(project_name: str) -> list[dict[str, Any]]:
    """Return all tasks linked to a given project, newest first."""
    try:
        db = get_db()
        rows = db.execute(
            "SELECT id, title, status, created_at, started_at, completed_at, "
            "workspace_path, result FROM tasks ORDER BY created_at DESC"
        ).fetchall()
    except sqlite3.Error:
        logger.exception("link_project_tasks db error")
        return []

    proj = next((p for p in config.MANAGED_PROJECTS if p["name"] == project_name), None)
    proj_path = (proj["path"] if proj else "").lower()

    matched: list[dict[str, Any]] = []
    for r in rows:
        title = (r["title"] or "").lower()
        workspace = (r["workspace_path"] or "").lower()

        is_match = False
        if proj_path and proj_path in workspace:
            is_match = True
        else:
            for keyword, pname in _TITLE_KEYWORDS:
                if pname == project_name and keyword in title:
                    is_match = True
                    break

        if not is_match:
            continue

        duration_s = None
        if r["started_at"] and r["completed_at"]:
            duration_s = int(r["completed_at"]) - int(r["started_at"])

        matched.append({
            "task_id": r["id"],
            "title": r["title"],
            "status": r["status"],
            "created_at": r["created_at"],
            "started_at": r["started_at"],
            "completed_at": r["completed_at"],
            "duration_s": duration_s,
            "outcome_summary": (r["result"] or "")[:200],
            "workspace_path": r["workspace_path"],
        })

    return matched


def _run_git(cwd: str, args: list[str]) -> str:
    try:
        result = subprocess.run(
            ["git"] + args, cwd=cwd,
            capture_output=True, text=True, check=False, timeout=5,
        )
        return result.stdout.strip() if result.returncode == 0 else ""
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
        return ""


def link_project_commits(
    project_name: str,
    limit: int = 20,
    since_ts: int | None = None,
    until_ts: int | None = None,
) -> list[dict[str, Any]]:
    """Return recent commits for a project, optionally filtered by time window."""
    proj = next((p for p in config.MANAGED_PROJECTS if p["name"] == project_name), None)
    if not proj:
        return []
    path = proj["path"]

    fmt = "%H|%an|%ct|%s"
    args = ["log", f"-{limit}", f"--format={fmt}"]
    if since_ts:
        args.append(f"--since=@{since_ts}")
    if until_ts:
        args.append(f"--until=@{until_ts}")

    out = _run_git(path, args)
    if not out:
        return []

    commits: list[dict[str, Any]] = []
    for line in out.splitlines():
        parts = line.split("|", 3)
        if len(parts) < 4:
            continue
        full_hash, author, ts, subject = parts
        commits.append({
            "hash": full_hash,
            "short": full_hash[:7],
            "author": author,
            "ts": int(ts) if ts.isdigit() else 0,
            "subject": subject,
            "gitea_url": f"http://localhost:3000/admin/{proj['gitea_repo']}/commit/{full_hash}",
        })
    return commits


def get_task_files_changed(
    project_name: str,
    since_ts: int | None,
    until_ts: int | None,
) -> dict[str, list[str]]:
    """Return {commit_short: [file1, file2, ...]} for commits in time window."""
    proj = next((p for p in config.MANAGED_PROJECTS if p["name"] == project_name), None)
    if not proj or not since_ts or not until_ts:
        return {}
    path = proj["path"]

    args = [
        "log",
        f"--since=@{since_ts}",
        f"--until=@{until_ts}",
        "--format=%h",
        "--name-only",
    ]
    out = _run_git(path, args)
    if not out:
        return {}

    result: dict[str, list[str]] = {}
    current_hash: str | None = None
    for line in out.splitlines():
        line = line.strip()
        if not line:
            continue
        if len(line) <= 10 and "/" not in line and "." not in line and not line.startswith(" "):
            if line.isalnum():
                current_hash = line
                result.setdefault(current_hash, [])
                continue
        if current_hash:
            result[current_hash].append(line)
    return result