# ───────────────────────────────────────────────────────────────────────────── # File: app.py # BuffTEKS Hub — Discord OAuth + RBAC, tickets, and VIP join form # Non-members are redirected to /join (VIP portal) after Discord login. # ───────────────────────────────────────────────────────────────────────────── from __future__ import annotations import os, time, requests from datetime import datetime from pathlib import Path from typing import Optional import json as _json import hmac import hashlib from flask_wtf import CSRFProtect from flask_limiter import Limiter from flask_limiter.util import get_remote_address from flask import ( Flask, render_template, request, redirect, url_for, flash, session, jsonify, abort ) from flask_sqlalchemy import SQLAlchemy from sqlalchemy import case from dotenv import load_dotenv # ───────────────────────────────────────────────────────────────────────────── # Env + App # ───────────────────────────────────────────────────────────────────────────── ROOT = Path(__file__).parent load_dotenv(ROOT / ".env") app = Flask( __name__, static_folder="static", static_url_path="/static", template_folder="templates" ) app.secret_key = os.environ.get("APP_SECRET_KEY", "dev") csrf = CSRFProtect(app) limiter = Limiter( get_remote_address, app=app, ) # Branding BRAND = os.environ.get("SITE_BRAND", "BuffTEKS") TAGLINE = os.environ.get("SITE_TAGLINE", "Student Engineers. Real Projects. Community Impact.") ACCENT = os.environ.get("SITE_ACCENT", "#8a2be2") # Discord config DISCORD_API = "https://discord.com/api/v10" DISCORD_GUILD_ID = os.environ.get("DISCORD_GUILD_ID", "") DISCORD_CHANNEL_ID = os.environ.get("DISCORD_CHANNEL_ID", "") # optional DISCORD_BOT_TOKEN = os.environ.get("DISCORD_BOT_TOKEN", "") # for reads / role lookups DISCORD_WEBHOOK_URL = os.environ.get("DISCORD_WEBHOOK_URL", "") # for announcements DISCORD_CLIENT_ID = os.environ.get("DISCORD_CLIENT_ID", "") DISCORD_CLIENT_SECRET = os.environ.get("DISCORD_CLIENT_SECRET", "") OAUTH_REDIRECT_URI = os.environ.get("OAUTH_REDIRECT_URI", "http://localhost:5000/auth/discord/callback") # Roles ADMIN_ROLE_IDS = {r.strip() for r in os.environ.get("ADMIN_ROLE_IDS", "").split(",") if r.strip()} MEMBER_ROLE_IDS = {r.strip() for r in os.environ.get("MEMBER_ROLE_IDS", "").split(",") if r.strip()} ROLE_TTL_SECONDS = int(os.environ.get("ROLE_TTL_SECONDS", "600")) # DB os.makedirs(app.instance_path, exist_ok=True) DEFAULT_SQLITE_PATH = (Path(app.instance_path) / 'buffteks.db').as_posix() DB_URL = os.environ.get('DATABASE_URL', f'sqlite:///{DEFAULT_SQLITE_PATH}') app.config["SQLALCHEMY_DATABASE_URI"] = DB_URL app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False db = SQLAlchemy(app) def getenv(name: str, default: Optional[str] = None): return os.environ.get(name, default) # NOW load env-based configuration REPO_EVENT_CHANNEL_MAP = _json.loads(getenv("REPO_EVENT_CHANNEL_MAP", "{}")) GITHUB_WEBHOOK_SECRET = getenv("GITHUB_WEBHOOK_SECRET", "") DEFAULT_DISCORD_WEBHOOK = getenv("DISCORD_WEBHOOK_URL", "") # ───────────────────────────────────────────────────────────────────────────── # Models # ───────────────────────────────────────────────────────────────────────────── class Ticket(db.Model): __tablename__ = "tickets" id = db.Column(db.Integer, primary_key=True) title = db.Column(db.String(200), nullable=False) description = db.Column(db.Text, nullable=False) status = db.Column(db.String(20), default="submitted") # submitted/triage/in_progress/awaiting_review/done + side priority = db.Column(db.String(20), default="normal") # low, normal, high, urgent labels = db.Column(db.Text, default="[]") # JSON list[str] assignee_id = db.Column(db.String(40)) # Discord user id assignee_name = db.Column(db.String(120)) # cached display name created_by_id = db.Column(db.String(40)) # Discord user id created_by_name = db.Column(db.String(120)) # helpers due_at = db.Column(db.DateTime) points = db.Column(db.Integer) checklist = db.Column(db.Text, default="[]") # JSON [{text, checked}] blocked_reason = db.Column(db.Text) info_request = db.Column(db.Text) sprint = db.Column(db.String(50)) watchers = db.Column(db.Text, default="[]") # JSON list[str] created_at = db.Column(db.DateTime, default=datetime.utcnow) updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) comments = db.relationship("TicketComment", backref="ticket", cascade="all, delete-orphan") def label_list(self) -> list[str]: try: return json.loads(self.labels or "[]") except Exception: return [] def checklist_items(self) -> list[dict]: try: return json.loads(self.checklist or "[]") except Exception: return [] def watcher_ids(self) -> list[str]: try: return json.loads(self.watchers or "[]") except Exception: return [] class TicketComment(db.Model): __tablename__ = "ticket_comments" id = db.Column(db.Integer, primary_key=True) ticket_id = db.Column(db.Integer, db.ForeignKey("tickets.id", ondelete="CASCADE"), nullable=False) author_id = db.Column(db.String(40)) author_name = db.Column(db.String(120)) body = db.Column(db.Text, nullable=False) created_at = db.Column(db.DateTime, default=datetime.utcnow) class Attachment(db.Model): __tablename__ = "attachments" id = db.Column(db.Integer, primary_key=True) ticket_id = db.Column(db.Integer, db.ForeignKey("tickets.id", ondelete="CASCADE"), nullable=False) filename = db.Column(db.String(255)) url = db.Column(db.Text) kind = db.Column(db.String(30)) uploaded_by_id = db.Column(db.String(40)) uploaded_by_name = db.Column(db.String(120)) created_at = db.Column(db.DateTime, default=datetime.utcnow) class AuditEvent(db.Model): __tablename__ = "audit_events" id = db.Column(db.Integer, primary_key=True) ticket_id = db.Column(db.Integer, db.ForeignKey("tickets.id", ondelete="CASCADE"), nullable=False) actor_id = db.Column(db.String(40)) actor_name = db.Column(db.String(120)) event_type = db.Column(db.String(40)) # created, updated, comment, assign, status payload = db.Column(db.Text, default="{}") created_at = db.Column(db.DateTime, default=datetime.utcnow) # VIP: Non-member intake form submissions class NonMemberApplication(db.Model): __tablename__ = "nonmember_applications" id = db.Column(db.Integer, primary_key=True) discord_id = db.Column(db.String(40), nullable=False) discord_username = db.Column(db.String(120), nullable=False) first_name = db.Column(db.String(100), nullable=False) last_name = db.Column(db.String(100), nullable=False) major = db.Column(db.String(120), nullable=False) student_email = db.Column(db.String(200), nullable=False) # NEW VIP fields commitment = db.Column(db.String(200)) # which team/project they will join commit_message = db.Column(db.String(200)) # fun commit line next_url = db.Column(db.String(500)) created_at = db.Column(db.DateTime, default=datetime.utcnow) # ───────────────────────────────────────────────────────────────────────────── # Lifecycle + permissions # ───────────────────────────────────────────────────────────────────────────── STATUS_FLOW = ["submitted", "triage", "in_progress", "awaiting_review", "done"] SIDE_STATUSES = {"needs_more_info", "blocked", "cancelled"} STATUS_CHOICES = set(STATUS_FLOW) | SIDE_STATUSES PRIORITY_CHOICES = {"low", "normal", "high", "urgent"} STATUS_TRANSITIONS = { ("submitted", "triage"): "admin", ("triage", "in_progress"): "admin", ("in_progress", "awaiting_review"): "assignee_or_admin", ("awaiting_review", "done"): "admin", ("awaiting_review", "needs_more_info"): "admin", ("needs_more_info", "in_progress"): "assignee_or_admin", ("in_progress", "blocked"): "assignee_or_admin", ("blocked", "in_progress"): "assignee_or_admin", } REQUIRED_FIELDS = { "awaiting_review": ["checklist"], "blocked": ["blocked_reason"], "needs_more_info": ["info_request"], } # ───────────────────────────────────────────────────────────────────────────── # Helpers (Discord + RBAC) # ───────────────────────────────────────────────────────────────────────────── def discord_auth_headers(): if not DISCORD_BOT_TOKEN: raise RuntimeError("Missing DISCORD_BOT_TOKEN") return {"Authorization": f"Bot {DISCORD_BOT_TOKEN}"} def user_is_in_guild(user_id: str) -> bool: if not user_id: return False url = f"{DISCORD_API}/guilds/{DISCORD_GUILD_ID}/members/{user_id}" r = requests.get(url, headers=discord_auth_headers(), timeout=10) return r.status_code == 200 def fetch_member_roles(discord_user_id: str) -> list[str]: if not discord_user_id: return [] url = f"{DISCORD_API}/guilds/{DISCORD_GUILD_ID}/members/{discord_user_id}" r = requests.get(url, headers=discord_auth_headers(), timeout=10) if r.status_code != 200: return [] return r.json().get("roles", []) or [] def compute_site_roles(discord_role_ids: set[str]) -> list[str]: site = set() if discord_role_ids & ADMIN_ROLE_IDS: site.add("admin") if discord_role_ids & MEMBER_ROLE_IDS or discord_role_ids: site.add("member") return [r for r in ["admin", "member"] if r in site] def session_user() -> Optional[dict]: return session.get("discord_user") def ensure_roles_fresh(force: bool = False): u = session_user() if not u: return now = int(time.time()) if not force and now - u.get("roles_ts", 0) < ROLE_TTL_SECONDS and u.get("site_roles"): return roles = set(fetch_member_roles(u["id"])) u["site_roles"] = compute_site_roles(roles) u["roles_ts"] = now session["discord_user"] = u def user_has_any(*required: str, force_refresh: bool = False) -> bool: u = session_user() if not u: return False ensure_roles_fresh(force=force_refresh) return bool(set(u.get("site_roles", [])).intersection(required)) def announce_to_discord(content: str) -> bool: if not DISCORD_WEBHOOK_URL: return False r = requests.post(DISCORD_WEBHOOK_URL, json={"content": content[:1900]}, timeout=10) return r.status_code in (200, 204) def announce_ticket(event: str, t: Ticket, mention: bool = False, extra: str | None = None): link = url_for("ticket_detail", ticket_id=t.id, _external=True) parts = [] if event == "created": parts.append(f"🆕 **Ticket #{t.id} — {t.title}** ({t.priority})") elif event == "updated": parts.append(f"✏️ **Updated Ticket #{t.id} — {t.title}**") elif event == "assigned": who = t.assignee_name or (t.assignee_id and f"<@{t.assignee_id}>") or "Unassigned" parts.append(f"🧭 **Assigned #{t.id} to {who}**") elif event == "status": parts.append(f"🔄 **Status: #{t.id} → `{t.status}`**") elif event == "comment": parts.append(f"💬 **Comment on #{t.id} — {t.title}**") if extra: parts.append(extra) if mention and t.assignee_id: parts.append(f"<@{t.assignee_id}>") parts.append(link) announce_to_discord("\n".join(parts)) # ───────────────────────────────────────────────────────────────────────────── # Template globals # ───────────────────────────────────────────────────────────────────────────── @app.context_processor def inject_globals(): return dict(brand=BRAND, tagline=TAGLINE, accent=ACCENT) # ───────────────────────────────────────────────────────────────────────────── # Global gate — require sign-in; non-members → /join (VIP portal) # ───────────────────────────────────────────────────────────────────────────── SAFE_ENDPOINTS = { "discord_login", "discord_callback", "logout", "static", "join", "join_thanks", "favicon", "github_webhook", } @app.before_request def require_guild_membership(): ep = (request.endpoint or "") if ep in SAFE_ENDPOINTS or ep.startswith("static"): return u = session_user() if not u: target = request.full_path if request.query_string else request.path return redirect(url_for("discord_login", next=target)) ensure_roles_fresh() if not user_is_in_guild(u.get("id")): target = request.full_path if request.query_string else request.path return redirect(url_for("join", next=target)) # ───────────────────────────────────────────────────────────────────────────── # Auth (Discord OAuth2) # ───────────────────────────────────────────────────────────────────────────── @app.route("/auth/discord/login") def discord_login(): next_url = request.args.get("next") or request.referrer or url_for("tickets") session["post_login_redirect"] = next_url params = { "client_id": DISCORD_CLIENT_ID, "response_type": "code", "redirect_uri": OAUTH_REDIRECT_URI, "scope": "identify", "prompt": "none", } q = "&".join([f"{k}={requests.utils.quote(v)}" for k, v in params.items() if v]) return redirect(f"https://discord.com/oauth2/authorize?{q}") @app.route("/auth/discord/callback") def discord_callback(): code = request.args.get("code") if not code: flash("No code from Discord.", "error") return redirect(url_for("tickets")) tok = requests.post( f"{DISCORD_API}/oauth2/token", data={ "client_id": DISCORD_CLIENT_ID, "client_secret": DISCORD_CLIENT_SECRET, "grant_type": "authorization_code", "code": code, "redirect_uri": OAUTH_REDIRECT_URI, }, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=10, ) if tok.status_code != 200: flash("Discord login failed.", "error") return redirect(url_for("tickets")) access_token = tok.json().get("access_token") u = requests.get( f"{DISCORD_API}/users/@me", headers={"Authorization": f"Bearer {access_token}"}, timeout=10, ) if u.status_code != 200: flash("Discord login failed.", "error") return redirect(url_for("tickets")) user = u.json() session["discord_user"] = { "id": user["id"], "username": user.get("global_name") or user["username"], } if not user_is_in_guild(user["id"]): return redirect(url_for("join", next=session.get("post_login_redirect", url_for("tickets")))) role_ids = set(fetch_member_roles(user["id"])) session["discord_user"]["site_roles"] = compute_site_roles(role_ids) session["discord_user"]["roles_ts"] = int(time.time()) return redirect(session.pop("post_login_redirect", url_for("tickets"))) @app.route("/auth/logout") def logout(): session.pop("discord_user", None) return redirect(url_for("tickets")) # ───────────────────────────────────────────────────────────────────────────── # VIP Join form # ───────────────────────────────────────────────────────────────────────────── @app.route("/join", methods=["GET", "POST"]) def join(): u = session_user() if not u: target = request.full_path if request.query_string else request.path return redirect(url_for("discord_login", next=target)) # Already a member? let them through if user_is_in_guild(u.get("id")): return redirect(request.args.get("next") or url_for("tickets")) next_url = request.args.get("next") or session.get("post_login_redirect") or url_for("tickets") if request.method == "POST": first = (request.form.get("first_name") or "").strip() last = (request.form.get("last_name") or "").strip() major = (request.form.get("major") or "").strip() email = (request.form.get("student_email") or "").strip() # Be lenient with the project/team field name to survive template drift def get_commitment(form): for k in ("commitment", "project_team", "project", "team"): v = (form.get(k) or "").strip() if v: return v return "" commit_to = get_commitment(request.form) commit_msg = (request.form.get("commit_message") or "").strip() next_post = (request.form.get("next") or next_url).strip() missing = [k for k,v in { "First name": first, "Last name": last, "Major": major, "Student email": email, "Project/team": commit_to }.items() if not v] if missing: flash("Please complete all required fields: " + ", ".join(missing), "error") return render_template("join_form.html", user=u, next_url=next_post, vip=True, first_name=first, last_name=last, major=major, student_email=email, commitment=commit_to, commit_message=commit_msg) app_row = NonMemberApplication( discord_id=u["id"], discord_username=u["username"], first_name=first, last_name=last, major=major, student_email=email, commitment=commit_to, commit_message=commit_msg, next_url=next_post ) try: db.session.add(app_row) db.session.commit() except Exception as e: db.session.rollback() try: app_row.commitment = None app_row.commit_message = None db.session.add(app_row) db.session.commit() flash("Saved without VIP extras (run DB migration for commitment/commit_message).", "error") except Exception as e2: flash("Could not save your request. An admin has been notified.", "error") try: announce_to_discord(f"⚠️ VIP form DB error for {u['username']} ({u['id']}): {e2}") except Exception: pass return render_template("join_form.html", user=u, next_url=next_post, vip=True, first_name=first, last_name=last, major=major, student_email=email, commitment=commit_to, commit_message=commit_msg) try: announce_to_discord( "🧾 **New BuffTEKS VIP Access Request**\n" f"- Discord: {u['username']} (ID {u['id']})\n" f"- Name: {first} {last}\n" f"- Major: {major}\n" f"- Student Email: {email}\n" f"- Commitment: {commit_to or '(missing)'}\n" f"- Commit Msg: {commit_msg or '(none)'}" ) except Exception: pass return redirect(url_for("join_thanks")) # GET return render_template("join_form.html", user=u, next_url=next_url, vip=True) # GET return render_template("join_form.html", user=u, next_url=next_url, vip=True) @app.route("/join/thanks") def join_thanks(): return render_template("join_thanks.html") # ───────────────────────────────────────────────────────────────────────────── # Core pages # ───────────────────────────────────────────────────────────────────────────── @app.route("/") def home(): return redirect(url_for("tickets")) @app.route("/tickets") def tickets(): status = request.args.get("status") assignee = request.args.get("assignee") q = request.args.get("q", "").strip() qry = Ticket.query if status: qry = qry.filter(Ticket.status == status) if assignee: qry = qry.filter(Ticket.assignee_id == assignee) if q: like = f"%{q}%" qry = qry.filter((Ticket.title.ilike(like)) | (Ticket.description.ilike(like))) items = qry.order_by( case((Ticket.status == "open", 0), (Ticket.status == "in_progress", 1), (Ticket.status == "done", 2), else_=3), case((Ticket.priority == "urgent", 3), (Ticket.priority == "high", 2), (Ticket.priority == "normal", 1), else_=0).desc(), Ticket.updated_at.desc(), ).all() user = session_user() return render_template("tickets.html", tickets=items, user=user) @app.route("/tickets/") def ticket_detail(ticket_id: int): t = Ticket.query.get_or_404(ticket_id) user = session_user() can_manage = bool(user) and user_has_any("admin", force_refresh=True) can_update_status = bool(user) and ( user_has_any("admin", force_refresh=True) or (t.assignee_id and user.get("id") == t.assignee_id) ) return render_template("ticket_detail.html", t=t, user=user, can_manage=can_manage, can_update_status=can_update_status) # ───────────────────────────────────────────────────────────────────────────── # Admin: create/update tickets # ───────────────────────────────────────────────────────────────────────────── @app.route("/admin/tickets/new", methods=["GET", "POST"]) def admin_new_ticket(): if not (session_user() and user_has_any("admin", force_refresh=True)): abort(403) if request.method == "POST": form = request.form title = form.get("title", "").strip() desc = form.get("description", "").strip() priority = form.get("priority", "normal").strip() labels = [s.strip() for s in (form.get("labels", "").split(",") if form.get("labels") else []) if s.strip()] assignee_id = form.get("assignee_id", "").strip() or None assignee_name = form.get("assignee_name", "").strip() or None if not title or not desc: flash("Title and description are required.", "error") return redirect(url_for("admin_new_ticket")) if priority not in PRIORITY_CHOICES: priority = "normal" u = session_user() t = Ticket( title=title, description=desc, priority=priority, labels=json.dumps(labels), assignee_id=assignee_id, assignee_name=assignee_name, created_by_id=u.get("id"), created_by_name=u.get("username"), status="submitted", ) db.session.add(t); db.session.commit() db.session.add(AuditEvent(ticket_id=t.id, actor_id=u.get("id"), actor_name=u.get("username"), event_type="created", payload=json.dumps({"priority":priority,"labels":labels}))) db.session.commit() announce_ticket("created", t, mention=bool(assignee_id)) flash("Ticket created.", "ok") return redirect(url_for("ticket_detail", ticket_id=t.id)) return render_template("admin_ticket_new.html") @app.route("/admin/tickets//edit", methods=["GET", "POST"]) def admin_edit_ticket(ticket_id: int): if not (session_user() and user_has_any("admin", force_refresh=True)): abort(403) t = Ticket.query.get_or_404(ticket_id) if request.method == "POST": form = request.form old = {"status": t.status, "assignee_id": t.assignee_id, "priority": t.priority} t.title = form.get("title", t.title).strip() t.description = form.get("description", t.description).strip() pri = form.get("priority", t.priority).strip() t.priority = pri if pri in PRIORITY_CHOICES else t.priority st = form.get("status", t.status).strip() t.status = st if st in STATUS_CHOICES else t.status labels = [s.strip() for s in (form.get("labels", "").split(",") if form.get("labels") else []) if s.strip()] t.labels = json.dumps(labels) t.assignee_id = form.get("assignee_id", "").strip() or None t.assignee_name = form.get("assignee_name", "").strip() or None db.session.commit() u = session_user() db.session.add(AuditEvent(ticket_id=t.id, actor_id=u.get("id"), actor_name=u.get("username"), event_type="updated", payload=json.dumps({"old":old,"new":{"status":t.status,"priority":t.priority}}))) db.session.commit() announce_ticket("updated", t) if old["assignee_id"] != t.assignee_id: announce_ticket("assigned", t, mention=True) if old["status"] != t.status: announce_ticket("status", t, mention=(t.status=="awaiting_review")) flash("Saved.", "ok") return redirect(url_for("ticket_detail", ticket_id=t.id)) return render_template("admin_ticket_edit.html", t=t) # ───────────────────────────────────────────────────────────────────────────── # APIs: comments, status, assign # ───────────────────────────────────────────────────────────────────────────── @app.post("/api/tickets//comment") def api_comment(ticket_id: int): user = session_user() if not (user and user_is_in_guild(user.get("id"))): return jsonify({"ok": False, "error": "Not authorized"}), 401 t = Ticket.query.get_or_404(ticket_id) data = request.get_json(silent=True) or {} body = (data.get("body") or "").strip() if not body: return jsonify({"ok": False, "error": "Empty comment"}), 400 c = TicketComment(ticket_id=t.id, author_id=user.get("id"), author_name=user.get("username"), body=body) db.session.add(c) db.session.add(AuditEvent(ticket_id=t.id, actor_id=user.get("id"), actor_name=user.get("username"), event_type="comment", payload=json.dumps({"length":len(body)}))) db.session.commit() announce_ticket("comment", t) return jsonify({"ok": True, "comment": { "id": c.id, "author_name": c.author_name, "body": c.body, "created_at": c.created_at.isoformat(), }}) @app.post("/api/tickets//status") def api_status(ticket_id: int): user = session_user() if not (user and user_is_in_guild(user.get("id"))): return jsonify({"ok": False, "error": "Not authorized"}), 401 t = Ticket.query.get_or_404(ticket_id) data = request.get_json(silent=True) or {} new_status = (data.get("status") or "").strip() if new_status not in STATUS_CHOICES: return jsonify({"ok": False, "error": "Invalid status"}), 400 old = t.status or "submitted" allowed = False role = STATUS_TRANSITIONS.get((old, new_status)) is_admin = user_has_any("admin", force_refresh=True) is_assignee = t.assignee_id and user.get("id") == t.assignee_id if role == "admin" and is_admin: allowed = True elif role == "assignee_or_admin" and (is_admin or is_assignee): allowed = True elif old == new_status: allowed = True else: if is_admin: allowed = True if not allowed: return jsonify({"ok": False, "error": "Forbidden"}), 403 t.status = new_status db.session.add(AuditEvent(ticket_id=t.id, actor_id=user.get("id"), actor_name=user.get("username"), event_type="status", payload=json.dumps({"old":old,"new":new_status}))) db.session.commit() announce_ticket("status", t, mention=(new_status=="awaiting_review")) return jsonify({"ok": True, "status": t.status}) @app.context_processor def inject_globals(): from datetime import timezone import math def reltime(dt): if not dt: return "" now = datetime.utcnow().replace(tzinfo=None) diff = (now - dt).total_seconds() past = diff >= 0 s = abs(diff) for unit, secs in [("yr", 31536000), ("mo", 2592000), ("d", 86400), ("h", 3600), ("m", 60)]: if s >= secs: n = int(s // secs) return f"{n}{unit}{'' if n==1 else 's'} {'ago' if past else 'from now'}" return "just now" STATUS_COLORS = { "submitted": "bg-indigo-500/20 border-indigo-500/30 text-indigo-200", "triage": "bg-sky-500/20 border-sky-500/30 text-sky-200", "in_progress": "bg-amber-500/20 border-amber-500/30 text-amber-200", "awaiting_review": "bg-fuchsia-500/20 border-fuchsia-500/30 text-fuchsia-200", "done": "bg-emerald-500/20 border-emerald-500/30 text-emerald-200", "needs_more_info": "bg-pink-500/20 border-pink-500/30 text-pink-200", "blocked": "bg-red-500/20 border-red-500/30 text-red-200", "cancelled": "bg-slate-500/20 border-slate-500/30 text-slate-300", "open": "bg-indigo-500/20 border-indigo-500/30 text-indigo-200", } PRIORITY_COLORS = { "low": "bg-slate-500/20 border-slate-500/30 text-slate-200", "normal": "bg-zinc-500/20 border-zinc-500/30 text-zinc-200", "high": "bg-orange-500/20 border-orange-500/30 text-orange-200", "urgent": "bg-red-600/25 border-red-500/40 text-red-200", } def status_class(s): return STATUS_COLORS.get(s, "bg-white/10 border-white/20 text-white/80") def priority_class(p): return PRIORITY_COLORS.get(p, "bg-white/10 border-white/20 text-white/80") def checklist_progress(t): try: items = t.checklist_items() total = len(items) if not total: return (0, 0) done = sum(1 for i in items if i.get("checked")) return (done, total) except Exception: return (0, 0) def has_endpoint(name: str) -> bool: return name in app.view_functions return dict( brand=BRAND, tagline=TAGLINE, accent=ACCENT, reltime=reltime, status_class=status_class, priority_class=priority_class, checklist_progress=checklist_progress, has_endpoint=has_endpoint ) @app.context_processor def inject_globals(): def has_endpoint(name: str) -> bool: return name in app.view_functions return dict( brand=BRAND, tagline=TAGLINE, accent=ACCENT, has_endpoint=has_endpoint, # <-- new ) @app.post("/api/tickets//assign") def api_assign(ticket_id: int): u = session_user() if not (u and user_has_any("admin", force_refresh=True)): return jsonify({"ok": False, "error": "Admins only"}), 403 t = Ticket.query.get_or_404(ticket_id) data = request.get_json(silent=True) or {} assignee_id = (data.get("assignee_id") or "").strip() or None assignee_name = (data.get("assignee_name") or "").strip() or None t.assignee_id = assignee_id t.assignee_name = assignee_name db.session.commit() link = url_for("ticket_detail", ticket_id=t.id, _external=True) mention = f" <@{assignee_id}>" if assignee_id else "" announce_to_discord(f"🧭 **Assignment: Ticket #{t.id} — {t.title}** → {assignee_name or assignee_id or 'Unassigned'}{mention}\n{link}") return jsonify({"ok": True, "assignee_id": t.assignee_id, "assignee_name": t.assignee_name}) # ───────────────────────────────────────────────────────────────────────────── # Favicon + Bootstrap # ───────────────────────────────────────────────────────────────────────────── @app.route("/favicon.ico") def favicon(): return ("", 204) def init_db(): with app.app_context(): db.create_all() @app.post("/webhooks/github") @csrf.exempt @limiter.exempt def github_webhook(): event = request.headers.get("X-GitHub-Event") payload = request.get_json(silent=True) or {} # Identify repo repo = payload.get("repository", {}).get("full_name") if not repo: return jsonify({"ok": False, "error": "No repository info"}), 400 # Look up channel mapping for this repo repo_cfg = REPO_EVENT_CHANNEL_MAP.get(repo, {}) # Look up webhook for this specific event webhook = repo_cfg.get(event) if not webhook: # No configured channel for this event → ignore safely return jsonify({ "ok": True, "note": f"No channel configured for event `{event}` on repo `{repo}`" }), 200 # Format message message = format_github_event(event, payload) # Send to Discord discord_webhook_send(webhook, message) return jsonify({"ok": True}), 200 def format_github_event(event: str, p: dict) -> str: repo = p.get("repository", {}).get("full_name", "Unknown Repo") if event == "push": pusher = p.get("pusher", {}).get("name") commits = p.get("commits", []) commit_lines = "\n".join( f"- `{c.get('id','')[:7]}` {c.get('message','').strip()} — {c.get('author',{}).get('name','')}" for c in commits ) return ( f"📦 **Push to `{repo}`** by **{pusher}**\n" f"{commit_lines or '(no commit messages)'}" ) if event == "issues": action = p.get("action") issue = p.get("issue", {}) return ( f"🐛 **Issue {action} — #{issue.get('number')}**\n" f"**{issue.get('title')}**\n" f"{issue.get('html_url')}" ) if event == "pull_request": action = p.get("action") pr = p.get("pull_request", {}) return ( f"🔀 **PR {action} — #{pr.get('number')}**\n" f"**{pr.get('title')}**\n" f"{pr.get('html_url')}" ) if event == "release": r = p.get("release", {}) return ( f"🚀 **New Release `{r.get('tag_name')}`**\n" f"**{r.get('name')}**\n" f"{r.get('html_url')}" ) # Fallback return f"ℹ️ Event `{event}` received from `{repo}`" def discord_webhook_send(url: str, content: str): if not url: return try: requests.post(url, json={"content": content[:1900]}, timeout=10) except Exception as e: print("Discord webhook error:", e) def send_discord(msg: str, repo: str): # Decide channel (repo-specific or fallback) webhook = REPO_CHANNEL_MAP.get(repo, DEFAULT_DISCORD_WEBHOOK) if not webhook: print(f"No webhook for repo {repo}") return try: requests.post(webhook, json={"content": msg}) except Exception as e: print("Discord error:", e) # ───────────────────────────────────────────────────────────────────────────── # Templates (written at import-time so Gunicorn has them) # ───────────────────────────────────────────────────────────────────────────── BASE_HTML = r""" {% block title %}{{ brand }} · Hub{% endblock %} {% set is_admin = (session.get('discord_user') and 'admin' in session.get('discord_user',{}).get('site_roles',[])) %}
{{ brand }}
{% block content %}{% endblock %}
""" JOIN_FORM_HTML = r"""{% extends "base.html" %} {% block title %}BuffTEKS VIP Server Access — {{ brand }}{% endblock %} {% block content %}

BuffTEKS VIP Server Access

Hi {{ user.username }}! The BuffTEKS VIP Server is our private collaboration space for active members.

To gain access, you’ll: 1) join BuffTEKS, 2) perform the Git Commit Ritual, and 3) commit to a project team.

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %}
{% for cat,msg in messages %}
{{ msg }}
{% endfor %}
{% endif %} {% endwith %}
$ git add me
$ git commit -m "{{ commit_message or 'chore: joined BuffTEKS, ready to contribute' }}"
$ git push origin greatness
  
{% endblock %} """ JOIN_THANKS_HTML = r"""{% extends "base.html" %} {% block title %}Thanks — {{ brand }}{% endblock %} {% block content %}

Thanks!

Your VIP request has been submitted. A BuffTEKS officer will contact you soon.

{% endblock %} """ TICKETS_HTML = r"""{% extends "base.html" %} {% block title %}Tickets — {{ brand }}{% endblock %} {% block content %}

Tickets

{{ tagline }}

{% endblock %} """ TICKET_DETAIL_HTML = r"""{% extends "base.html" %} {% block title %}#{{ t.id }} — {{ t.title }} · {{ brand }}{% endblock %} {% block content %}

#{{ t.id }} · {{ t.title }}

{{ t.status.replace('_',' ').title() }} Priority: {{ t.priority }} {% for label in t.label_list() %}{{ label }}{% endfor %}
{% if can_manage %} Edit {% endif %}

{{ t.description }}


Comments

{% for c in t.comments %}
{{ c.author_name }} · {{ c.created_at.strftime('%Y-%m-%d %H:%M') }} UTC
{{ c.body }}
{% else %}
No comments yet.
{% endfor %}
{% if user %}
{% endif %}
{% endblock %} """ ADMIN_TICKET_NEW_HTML = r"""{% extends "base.html" %} {% block title %}New Ticket — {{ brand }}{% endblock %} {% block content %}

Create Ticket

{% endblock %} """ ADMIN_TICKET_EDIT_HTML = r"""{% extends "base.html" %} {% block title %}Edit Ticket — {{ brand }}{% endblock %} {% block content %}

Edit Ticket #{{ t.id }}

{% endblock %} """ def _ensure_template_files(): tpl_dir = ROOT / "templates" tpl_dir.mkdir(parents=True, exist_ok=True) files = { tpl_dir / "base.html": BASE_HTML, tpl_dir / "join_form.html": JOIN_FORM_HTML, tpl_dir / "join_thanks.html": JOIN_THANKS_HTML, tpl_dir / "tickets.html": TICKETS_HTML, tpl_dir / "ticket_detail.html": TICKET_DETAIL_HTML, tpl_dir / "admin_ticket_new.html": ADMIN_TICKET_NEW_HTML, tpl_dir / "admin_ticket_edit.html": ADMIN_TICKET_EDIT_HTML, } for p, content in files.items(): if content and not p.exists(): p.write_text(content, encoding="utf-8") # Write templates + init DB at import time so Gunicorn workers are ready _ensure_template_files() init_db() if __name__ == "__main__": app.run(debug=True, port=5000)