144 lines
4.3 KiB
Python
144 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal owner-report producer for ClawTeam-style worker checkpoints.
|
|
|
|
Writes ~/.clawteam/owner-reports/pending/<report_id>.md using explicit checkpoint
|
|
fields and a human-readable message suitable for direct Telegram delivery.
|
|
|
|
This intentionally stays tiny:
|
|
- no daemon
|
|
- no event bus
|
|
- no parser for arbitrary logs
|
|
- just explicit fields in -> pending markdown out
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from owner_report_consumer import OWNER_REPORT_ROOT
|
|
|
|
PENDING_DIR = OWNER_REPORT_ROOT / "pending"
|
|
|
|
|
|
def _slug(value: str) -> str:
|
|
slug = re.sub(r"[^a-zA-Z0-9._-]+", "-", value.strip()).strip("-._")
|
|
return slug or "report"
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now().astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
def build_message(*, team: str, worker: str, task_id: str, progress: str, done: str, next_step: str, status: str, source: str | None, report_kind: str) -> str:
|
|
headline = f"🔔 [{team}] {worker}"
|
|
if report_kind == "leader-final":
|
|
headline = f"✅ [{team}] final"
|
|
|
|
lines = [
|
|
headline,
|
|
done,
|
|
]
|
|
|
|
if next_step.strip():
|
|
lines.append(f"→ {next_step}")
|
|
|
|
tech = [
|
|
f"task={task_id}",
|
|
f"status={status}",
|
|
f"progress={progress}",
|
|
]
|
|
if source:
|
|
tech.append(f"source={source}")
|
|
lines.append(" | ".join(tech))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def build_report_body(*, report_id: str, team: str, worker: str, task_id: str, progress: str, done: str, next_step: str, status: str, source: str | None, created_at: str, message: str, report_kind: str) -> str:
|
|
fields: list[tuple[str, str | None]] = [
|
|
("report_id", report_id),
|
|
("team", team),
|
|
("worker", worker),
|
|
("task_id", task_id),
|
|
("progress", progress),
|
|
("done", done),
|
|
("next", next_step),
|
|
("status", status),
|
|
("report_kind", report_kind),
|
|
("source", source),
|
|
("created_at", created_at),
|
|
("message", json.dumps(message, ensure_ascii=False)),
|
|
]
|
|
return "\n".join(f"{k}: {v}" for k, v in fields if v is not None) + "\n"
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="Create one pending owner report from explicit checkpoint fields")
|
|
ap.add_argument("--team", required=True)
|
|
ap.add_argument("--worker", required=True)
|
|
ap.add_argument("--task-id", required=True)
|
|
ap.add_argument("--progress", required=True)
|
|
ap.add_argument("--done", required=True)
|
|
ap.add_argument("--next", dest="next_step", required=True)
|
|
ap.add_argument("--status", required=True)
|
|
ap.add_argument("--source")
|
|
ap.add_argument("--report-kind", choices=["checkpoint", "leader-final"], default="checkpoint")
|
|
ap.add_argument("--report-id", help="Optional explicit report_id / filename stem")
|
|
ap.add_argument("--created-at", default=_now_iso())
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
report_id = args.report_id or f"{_slug(args.team)}-{_slug(args.worker)}-{_slug(args.task_id)}-{_slug(args.report_kind)}"
|
|
message = build_message(
|
|
team=args.team,
|
|
worker=args.worker,
|
|
task_id=args.task_id,
|
|
progress=args.progress,
|
|
done=args.done,
|
|
next_step=args.next_step,
|
|
status=args.status,
|
|
source=args.source,
|
|
report_kind=args.report_kind,
|
|
)
|
|
body = build_report_body(
|
|
report_id=report_id,
|
|
team=args.team,
|
|
worker=args.worker,
|
|
task_id=args.task_id,
|
|
progress=args.progress,
|
|
done=args.done,
|
|
next_step=args.next_step,
|
|
status=args.status,
|
|
source=args.source,
|
|
created_at=args.created_at,
|
|
message=message,
|
|
report_kind=args.report_kind,
|
|
)
|
|
|
|
path = PENDING_DIR / f"{report_id}.md"
|
|
|
|
result = {
|
|
"ok": True,
|
|
"report_id": report_id,
|
|
"path": str(path),
|
|
"message": message,
|
|
"dry_run": args.dry_run,
|
|
}
|
|
|
|
if args.dry_run:
|
|
result["body"] = body
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
PENDING_DIR.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(body, encoding="utf-8")
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|