127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""
|
|
Feedback API — pure Python backend, no Streamlit imports.
|
|
Called directly from app/feedback.py now; wrappable in a FastAPI route later.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import platform
|
|
import re
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
_ROOT = Path(__file__).parent.parent
|
|
_EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}")
|
|
_PHONE_RE = re.compile(r"(\+?1[\s\-.]?)?\(?\d{3}\)?[\s\-.]?\d{3}[\s\-.]?\d{4}")
|
|
|
|
|
|
def mask_pii(text: str) -> str:
|
|
"""Redact email addresses and phone numbers from text."""
|
|
text = _EMAIL_RE.sub("[email redacted]", text)
|
|
text = _PHONE_RE.sub("[phone redacted]", text)
|
|
return text
|
|
|
|
|
|
def collect_context(page: str) -> dict:
|
|
"""Collect app context: page, version, tier, LLM backend, OS, timestamp."""
|
|
# App version from git
|
|
try:
|
|
version = subprocess.check_output(
|
|
["git", "describe", "--tags", "--always"],
|
|
cwd=_ROOT, text=True, timeout=5,
|
|
).strip()
|
|
except Exception:
|
|
version = "dev"
|
|
|
|
# Tier from user.yaml
|
|
tier = "unknown"
|
|
try:
|
|
user = yaml.safe_load((_ROOT / "config" / "user.yaml").read_text()) or {}
|
|
tier = user.get("tier", "unknown")
|
|
except Exception:
|
|
pass
|
|
|
|
# LLM backend from llm.yaml
|
|
llm_backend = "unknown"
|
|
try:
|
|
llm = yaml.safe_load((_ROOT / "config" / "llm.yaml").read_text()) or {}
|
|
llm_backend = llm.get("provider", "unknown")
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"page": page,
|
|
"version": version,
|
|
"tier": tier,
|
|
"llm_backend": llm_backend,
|
|
"os": platform.platform(),
|
|
"timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
|
}
|
|
|
|
|
|
def collect_logs(n: int = 100, log_path: Path | None = None) -> str:
|
|
"""Return last n lines of the Streamlit log, with PII masked."""
|
|
path = log_path or (_ROOT / ".streamlit.log")
|
|
if not path.exists():
|
|
return "(no log file found)"
|
|
lines = path.read_text(errors="replace").splitlines()
|
|
return mask_pii("\n".join(lines[-n:]))
|
|
|
|
|
|
def collect_listings(db_path: Path | None = None, n: int = 5) -> list[dict]:
|
|
"""Return the n most-recent job listings — title, company, url only."""
|
|
import sqlite3
|
|
from scripts.db import DEFAULT_DB
|
|
path = db_path or DEFAULT_DB
|
|
conn = sqlite3.connect(path)
|
|
conn.row_factory = sqlite3.Row
|
|
rows = conn.execute(
|
|
"SELECT title, company, url FROM jobs ORDER BY id DESC LIMIT ?", (n,)
|
|
).fetchall()
|
|
conn.close()
|
|
return [{"title": r["title"], "company": r["company"], "url": r["url"]} for r in rows]
|
|
|
|
|
|
def build_issue_body(form: dict, context: dict, attachments: dict) -> str:
|
|
"""Assemble the Forgejo issue markdown body from form data, context, and attachments."""
|
|
_TYPE_LABELS = {"bug": "🐛 Bug", "feature": "✨ Feature Request", "other": "💬 Other"}
|
|
lines: list[str] = [
|
|
f"## {_TYPE_LABELS.get(form.get('type', 'other'), '💬 Other')}",
|
|
"",
|
|
form.get("description", ""),
|
|
"",
|
|
]
|
|
|
|
if form.get("type") == "bug" and form.get("repro"):
|
|
lines += ["### Reproduction Steps", "", form["repro"], ""]
|
|
|
|
if context:
|
|
lines += ["### Context", ""]
|
|
for k, v in context.items():
|
|
lines.append(f"- **{k}:** {v}")
|
|
lines.append("")
|
|
|
|
if attachments.get("logs"):
|
|
lines += [
|
|
"<details>",
|
|
"<summary>App Logs (last 100 lines)</summary>",
|
|
"",
|
|
"```",
|
|
attachments["logs"],
|
|
"```",
|
|
"</details>",
|
|
"",
|
|
]
|
|
|
|
if attachments.get("listings"):
|
|
lines += ["### Recent Listings", ""]
|
|
for j in attachments["listings"]:
|
|
lines.append(f"- [{j['title']} @ {j['company']}]({j['url']})")
|
|
lines.append("")
|
|
|
|
if attachments.get("submitter"):
|
|
lines += ["---", f"*Submitted by: {attachments['submitter']}*"]
|
|
|
|
return "\n".join(lines)
|