chore: add basic notify scripts for bluesky,github,twitter

10 months ago · 402981f16e
3 changed files with 255 additions and 0 deletions
--- a/tools/ci/ci_tools/social/notify_bluesky.py
+++ b/tools/ci/ci_tools/social/notify_bluesky.py
@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+# MIT License
+#
+# Copyright (c) 2025 Eugenio Parodi <ceccopierangiolieugenio AT googlemail DOT com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import os,sys
+import requests
+from datetime import datetime
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+sys.path.append(current_dir)
+from social_common import get_social_data, SocialData, get_env_var
+
+identifier=get_env_var('BLUESKY_APP_IDENTIFIER')
+password=get_env_var('BLUESKY_APP_PWD')
+
+# Step 1: Authenticate and get access token
+auth_response = requests.post(
+    "https://bsky.social/xrpc/com.atproto.server.createSession",
+    headers = {
+        "Content-Type": "application/json"
+    },
+    json={
+        "identifier": identifier,
+        "password": password
+    }
+)
+
+auth_data = auth_response.json()
+# print(auth_data)
+access_token = auth_data["accessJwt"]
+did = auth_data["did"]
+
+# Step 2: Post a message
+headers = {
+    "Authorization": f"Bearer {access_token}",
+    "Content-Type": "application/json"
+}
+
+post_data = {
+    "repo": did,
+    "collection": "app.bsky.feed.post",
+    "record": {
+        "$type": "app.bsky.feed.post",
+        "text": "Hello from my Python bot!",
+        "createdAt": datetime.now().isoformat() + "Z"
+    }
+}
+
+post_response = requests.post(
+    "https://bsky.social/xrpc/com.atproto.repo.createRecord",
+    headers=headers,
+    json=post_data
+)
+
+print(post_response.status_code)
+print(post_response.json())
--- a/tools/ci/ci_tools/social/notify_github_discussion.py
+++ b/tools/ci/ci_tools/social/notify_github_discussion.py
@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+# MIT License
+#
+# Copyright (c) 2025 Eugenio Parodi <ceccopierangiolieugenio AT googlemail DOT com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import os, sys
+import requests
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+sys.path.append(current_dir)
+from social_common import get_social_data, SocialData, get_env_var
+
+
+# === CONFIGURATION ===
+GITHUB_TOKEN = get_env_var('GH_DISCUSSION_TOKEN')
+REPO_OWNER = "ceccopierangiolieugenio"
+REPO_NAME = "pyTermTk"
+DICSUSSION_CATEGORY="announcements"
+DISCUSSION_TITLE = "Your Announcement Title"
+DISCUSSION_BODY = "This is the content of your announcement."
+
+# === FUNCTIONS ===
+
+def get_repo_id(owner, repo, token):
+    query = f"""
+    query {{
+      repository(owner: "{owner}", name: "{repo}") {{
+        id
+      }}
+    }}
+    """
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json"
+    }
+    response = requests.post('https://api.github.com/graphql', json={'query': query}, headers=headers)
+    return response.json()['data']['repository']['id']
+
+def get_category_id(repo_id, token):
+    query = f"""
+    query {{
+      node(id: "{repo_id}") {{
+        ... on Repository {{
+          discussionCategories(first: 10) {{
+            nodes {{
+              id
+              name
+            }}
+          }}
+        }}
+      }}
+    }}
+    """
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json"
+    }
+    response = requests.post('https://api.github.com/graphql', json={'query': query}, headers=headers)
+    categories = response.json()['data']['node']['discussionCategories']['nodes']
+    for category in categories:
+        if category['name'].lower() == DICSUSSION_CATEGORY:
+            return category['id']
+    raise Exception("Announcements category not found")
+
+def create_discussion(repo_id, category_id, title, body, token):
+    mutation = f"""
+    mutation {{
+      createDiscussion(input: {{
+        repositoryId: "{repo_id}",
+        categoryId: "{category_id}",
+        title: "{title}",
+        body: "{body}"
+      }}) {{
+        discussion {{
+          id
+          url
+        }}
+      }}
+    }}
+    """
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json"
+    }
+    response = requests.post('https://api.github.com/graphql', json={'query': mutation}, headers=headers)
+    data = response.json()
+    print("DEBUG: Create Discussion response:", data)  # Add this line
+    return data['data']['createDiscussion']['discussion']['url']
+
+# === MAIN EXECUTION ===
+
+try:
+    repo_id = get_repo_id(REPO_OWNER, REPO_NAME, GITHUB_TOKEN)
+    print(f"{repo_id=}")
+    category_id = get_category_id(repo_id, GITHUB_TOKEN)
+    print(f"{category_id=}")
+    discussion_url = create_discussion(repo_id, category_id, DISCUSSION_TITLE, DISCUSSION_BODY, GITHUB_TOKEN)
+    print(f"{discussion_url=}")
+    print(f"✅ Discussion created successfully: {discussion_url}")
+except Exception as e:
+    print(f"❌ Error: {e}")
--- a/tools/ci/ci_tools/social/notify_twitter.py
+++ b/tools/ci/ci_tools/social/notify_twitter.py
@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+# MIT License
+#
+# Copyright (c) 2025 Eugenio Parodi <ceccopierangiolieugenio AT googlemail DOT com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import os,sys
+import json
+import requests
+import argparse
+from requests_oauthlib import OAuth1Session
+
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+sys.path.append(current_dir)
+from social_common import get_social_data, SocialData, get_env_var
+
+def send_twitter_message(version: str, data:SocialData):
+    api_key = get_env_var("X_API_KEY")
+    api_secret = get_env_var("X_API_SECRET")
+    access_token = get_env_var("X_ACCESS_TOKEN")
+    access_token_secret = get_env_var("X_ACCESS_TOKEN_SECRET")
+
+    message = get_env_var("MESSAGE")
+
+    twitter = OAuth1Session(api_key, api_secret, access_token, access_token_secret)
+
+    payload = {"text": "Hello from OAuth 1.0a!"}
+    response = twitter.post("https://api.twitter.com/2/tweets", json=payload)
+
+    print(response.status_code)
+    print(response.json())
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Send a Twitter/X notification.")
+    parser.add_argument("app", type=str, help="The application name.")
+    parser.add_argument("version", type=str, help="The application version.")
+    args = parser.parse_args()
+
+    data = get_social_data(args.app)
+    if not data:
+        raise ValueError(f"app: {args.app} is not recognised")
+
+    send_twitter_message(args.version, data)