Marcos
128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
"""pragent pilot — daily feedback report delivery.
|
|
|
|
Calls `feedback_analyze.analyze()` and posts the markdown report as a
|
|
comment on a single long-lived "feedback roll-up" issue in
|
|
`gitea_admin/pragent`. Comments are append-only history — one comment per
|
|
run, timestamped in the body. This keeps every report in one place, easy
|
|
to scroll, and avoids the issue-explosion of "one issue per day".
|
|
|
|
If the issue doesn't exist yet, create it. Subsequent runs just add a
|
|
new comment.
|
|
|
|
Designed for the daily K8s CronJob (`k8s/pragent-feedback-cronjob.yaml`)
|
|
but runnable from CLI for ad-hoc checks.
|
|
|
|
Env:
|
|
GITEA_API in-cluster Gitea base URL
|
|
PRAGENT_BOT_TOKEN bot token (Write collaborator on gitea_admin/pragent)
|
|
PRAGENT_FEEDBACK_DB path to SQLite (default /data/feedback.db)
|
|
PRAGENT_FEEDBACK_ISSUE_REPO default gitea_admin/pragent
|
|
PRAGENT_FEEDBACK_ISSUE_TITLE default "pragent feedback roll-up"
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
import ai_review
|
|
|
|
from feedback_analyze import analyze
|
|
|
|
log = logging.getLogger("pragent.feedback.post")
|
|
|
|
|
|
REPO_DEFAULT = "gitea_admin/pragent"
|
|
TITLE_DEFAULT = "pragent feedback roll-up"
|
|
|
|
|
|
def _find_or_create_issue(api: str, token: str, repo: str, title: str) -> int:
|
|
"""Locate the open issue with this title; create one if missing.
|
|
|
|
Gitea's issue search is via `GET /repos/{o}/{r}/issues?state=open&q=...`
|
|
(q matches title + body). We filter client-side for the exact title
|
|
to avoid query-text false matches.
|
|
"""
|
|
status, raw = ai_review.gitea_get(api, repo, "issues?state=open&per_page=50", token)
|
|
if status == 200:
|
|
try:
|
|
for issue in json.loads(raw):
|
|
if issue.get("title") == title:
|
|
# NB: the comment URL needs the per-repo `number`, not the
|
|
# global `id`. `id=60 num=8` for an early-N create; we want
|
|
# `num=8` for `/repos/o/r/issues/8/comments`.
|
|
return int(issue["number"])
|
|
except (json.JSONDecodeError, ValueError, KeyError):
|
|
pass
|
|
# Create
|
|
status, raw = ai_review.gitea_post(
|
|
api, repo, "issues", token,
|
|
{"title": title, "body": "pragent feedback roll-up — auto-created."},
|
|
)
|
|
if status not in (200, 201):
|
|
raise RuntimeError(f"issue create failed: HTTP {status} body={raw[:200]!r}")
|
|
return int(json.loads(raw)["number"])
|
|
|
|
|
|
def _post_comment(api: str, token: str, repo: str, issue_number: int, body: str) -> int:
|
|
status, raw = ai_review.gitea_post(
|
|
api, repo, f"issues/{issue_number}/comments", token, {"body": body},
|
|
)
|
|
if status not in (200, 201):
|
|
raise RuntimeError(f"comment post failed: HTTP {status} body={raw[:200]!r}")
|
|
return json.loads(raw)["id"]
|
|
|
|
|
|
def deliver(
|
|
*, api: str, token: str, db_path: str,
|
|
repo: str = REPO_DEFAULT, title: str = TITLE_DEFAULT,
|
|
since_ts: int | None = None,
|
|
) -> dict:
|
|
"""Build the report and post it as a comment. Returns a stats dict."""
|
|
report = analyze(db_path, since_ts=since_ts)
|
|
issue_id = _find_or_create_issue(api, token, repo, title)
|
|
comment_id = _post_comment(api, token, repo, issue_id, report)
|
|
return {
|
|
"repo": repo, "issue_id": issue_id, "comment_id": comment_id,
|
|
"report_bytes": len(report.encode()),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(
|
|
description="Post the daily feedback report to Gitea.",
|
|
)
|
|
p.add_argument("--api", default=os.environ.get(
|
|
"GITEA_API", "http://gitea-http.gitea.svc.cluster.local:3000",
|
|
))
|
|
p.add_argument("--token", default=os.environ.get("PRAGENT_BOT_TOKEN", ""))
|
|
p.add_argument("--db", default=os.environ.get(
|
|
"PRAGENT_FEEDBACK_DB", "/data/feedback.db",
|
|
))
|
|
p.add_argument("--repo", default=os.environ.get(
|
|
"PRAGENT_FEEDBACK_ISSUE_REPO", REPO_DEFAULT,
|
|
))
|
|
p.add_argument("--title", default=os.environ.get(
|
|
"PRAGENT_FEEDBACK_ISSUE_TITLE", TITLE_DEFAULT,
|
|
))
|
|
p.add_argument("--since", type=int, default=None,
|
|
help="Unix timestamp; only include findings posted since")
|
|
args = p.parse_args()
|
|
|
|
if not args.token:
|
|
print("PRAGENT_BOT_TOKEN required", flush=True)
|
|
return 2
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
stats = deliver(
|
|
api=args.api, token=args.token, db_path=args.db,
|
|
repo=args.repo, title=args.title, since_ts=args.since,
|
|
)
|
|
print(json.dumps(stats), flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main()) |