sqlinsight-pro/deploy/forum/configure.py

#!/usr/bin/env python3
"""Sets up the SQLInsight Pro forum on a fresh Discourse through its API. Safe to run again: it only changes what
differs from the wanted state.

  DISCOURSE_URL=https://forum.sqlinsight.seosoft.dk DISCOURSE_API_KEY=... python3 deploy/forum/configure.py

Create the key inside the container:  ./launcher enter app  then
  cd /var/www/discourse && su discourse -c 'RAILS_ENV=production bundle exec rake api_key:create_master[forum-setup]'
Optional: COMMUNITY_URL (the community site, linked from the forum), CONTACT_EMAIL.
"""
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

BASE = os.environ.get("DISCOURSE_URL", "").rstrip("/")
KEY = os.environ.get("DISCOURSE_API_KEY", "")
COMMUNITY = os.environ.get("COMMUNITY_URL", "https://sqlinsight.seosoft.dk").rstrip("/")
CONTACT = os.environ.get("CONTACT_EMAIL", "")

SETTINGS = {
    "title": "SQLInsight Pro Forum",
    "site_description": "Spørgsmål, svar og erfaringer om SQL Server-overvågning med SQLInsight Pro.",
    "short_site_description": "SQL Server-overvågning med dyb ekspertise",
    "company_name": "SEOSoft",
    "default_locale": "da",
    "login_required": "false",
    "solved_enabled": "true",
    "tagging_enabled": "true",
    "top_menu": "latest|categories|new|top",
    "contact_url": f"{COMMUNITY}/support",
}
if CONTACT:
    SETTINGS["contact_email"] = CONTACT

# name, slug, colour, description, accepted answers (solved), only staff may start topics
CATEGORIES = [
    ("Annonceringer", "annonceringer", "0E7490",
     "Nye versioner, regelpakker og vigtige SQL Server-opdateringer. Kun SEOSoft opretter emner her; alle kan svare.", False, True),
    ("Spørgsmål og svar", "spoergsmaal", "155E75",
     "Spørg om fund, anbefalinger og SQL Server generelt. Markér det svar, der løste det.", True, False),
    ("Installation og drift", "installation", "1570EF",
     "MSI, rettigheder, repository-databasen, opdateringer, Linux og Managed Instance.", True, False),
    ("Performance og tuning", "performance", "E04F16",
     "Wait stats, Query Store, planer, indexes og blocking. Del gerne målingerne fra fundet.", True, False),
    ("Regler", "regler", "079455",
     "Forslag til nye regler og community-regler i SQL (se dokumentationen om egne regler).", False, False),
    ("Idéer og feedback", "ideer", "B54708",
     "Hvad mangler SQLInsight Pro? Fejl oprettes som issues på Forgejo.", False, False),
]

WELCOME_TITLE = "Velkommen til SQLInsight Pro-forummet"
WELCOME = f"""SQLInsight Pro er gratis SQL Server-overvågning med dyb ekspertise: hvert fund har en forklaring, målingerne bag og en konkret anbefaling.

**Her kan du**
- stille spørgsmål om installation, fund og tuning i [Spørgsmål og svar](/c/spoergsmaal)
- dele erfaringer og foreslå nye [regler](/c/regler)
- følge nye versioner og regelpakker i [Annonceringer](/c/annonceringer)

**Gode råd til et hurtigt svar**
- Skriv SQL Server-version og CU (står på instansens side i SQLInsight Pro).
- Kopiér fundets titel, målinger og regel-id (fx `SQLI-DB-007`).
- Fjern servernavne, logins og forespørgselstekst, du ikke vil dele offentligt.
- Markér det svar, der løste problemet, så andre kan finde det.

Dokumentation og download: [{COMMUNITY.removeprefix('https://')}]({COMMUNITY}) · Betalt support med svartid: [{COMMUNITY.removeprefix('https://')}/support]({COMMUNITY}/support)
"""


def call(method, path, data=None):
    body = urllib.parse.urlencode(data, doseq=True).encode() if data is not None else None
    request = urllib.request.Request(BASE + path, data=body, method=method, headers={
        "Api-Key": KEY, "Api-Username": "system", "Accept": "application/json",
    })
    for _ in range(10):
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                text = response.read().decode()
                return json.loads(text) if text else {}
        except urllib.error.HTTPError as error:
            body = error.read().decode()
            if error.code == 429:  # Discourse rate limit: wait as told, then try again
                time.sleep(json.loads(body).get("extras", {}).get("wait_seconds", 10) + 1)
                continue
            sys.exit(f"{method} {path}: {error.code} {body[:300]}")
    sys.exit(f"{method} {path}: stadig begrænset efter flere forsøg")


def main():
    if not BASE or not KEY:
        sys.exit(__doc__)

    current = {s["setting"]: s.get("value") for s in call("GET", "/admin/site_settings.json")["site_settings"]}
    for name, value in SETTINGS.items():
        if name not in current:
            print(f"  (indstillingen {name} findes ikke i denne Discourse-version)")
        elif str(current[name]) != value:
            call("PUT", f"/admin/site_settings/{name}.json", {name: value})
            print(f"indstilling  {name} = {value}")

    categories = {c["slug"]: c for c in call("GET", "/categories.json")["category_list"]["categories"]}
    for name, slug, colour, description, solved, staff_only in CATEGORIES:
        fields = {
            "name": name, "slug": slug, "color": colour, "text_color": "FFFFFF",
            "custom_fields[enable_accepted_answers]": "true" if solved else "false",
        }
        if staff_only:
            fields.update({"permissions[everyone]": "2", "permissions[staff]": "1"})  # everyone: reply; staff: create
        if slug in categories:
            call("PUT", f"/categories/{categories[slug]['id']}.json", fields)
            category_id = categories[slug]["id"]
            print(f"kategori     {name} (opdateret)")
        else:
            category_id = call("POST", "/categories.json", fields)["category"]["id"]
            print(f"kategori     {name} (oprettet)")
        # The description is the first post of the category's "About" topic.
        about = call("GET", f"/c/{category_id}/show.json")["category"].get("topic_url")
        if about:
            topic = call("GET", about + ".json")
            post = topic["post_stream"]["posts"][0]
            if post.get("raw") != description and description not in (post.get("cooked") or ""):
                call("PUT", f"/posts/{post['id']}.json", {"post[raw]": description})

    category_id = next(c["id"] for c in call("GET", "/categories.json")["category_list"]["categories"] if c["slug"] == "annonceringer")
    announcements = call("GET", f"/c/annonceringer/{category_id}/l/latest.json")["topic_list"]["topics"]
    if not any(t["title"] == WELCOME_TITLE for t in announcements):
        topic = call("POST", "/posts.json", {"title": WELCOME_TITLE, "raw": WELCOME, "category": category_id})
        call("PUT", f"/t/{topic['topic_id']}/status.json", {"status": "pinned_globally", "enabled": "true"})
        print("emne         velkomst (oprettet og fastgjort)")
    # Discourse's own English welcome topic: no longer pinned or listed; ours replaces it.
    for topic in call("GET", "/latest.json")["topic_list"]["topics"]:
        if topic.get("pinned_globally") and topic["title"] != WELCOME_TITLE and topic["title"].startswith("Welcome to"):
            call("PUT", f"/t/{topic['id']}/status.json", {"status": "pinned_globally", "enabled": "false"})
            call("PUT", f"/t/{topic['id']}/status.json", {"status": "visible", "enabled": "false"})
            print("emne         Discourses engelske velkomst skjult")

    feedback = next((c for c in call("GET", "/categories.json")["category_list"]["categories"] if c["slug"] == "site-feedback"), None)
    if feedback and feedback["name"] != "Om forummet":
        call("PUT", f"/categories/{feedback['id']}.json", {"name": "Om forummet", "slug": "site-feedback", "color": feedback["color"]})
        print("kategori     Site Feedback omdøbt til Om forummet")
    print("Forummet er sat op.")


if __name__ == "__main__":
    main()