Browse Source

chore: test

pull/454/head
Parodi, Eugenio 🌶 6 months ago
parent
commit
857b2aa0b7
  1. 8
      .github/workflows/_scratchpad.yml
  2. 6
      .github/workflows/release.yml
  3. 195
      tools/ci/ci_tools/social/notify_bluesky.py
  4. 1
      tools/ci/pyproject.toml

8
.github/workflows/_scratchpad.yml

@ -43,4 +43,10 @@ jobs:
GH_DISCUSSION_TOKEN: ${{ secrets.GH_DISCUSSION_TOKEN }}
run: |
# notify-discord ${{ matrix.name }} v${{ matrix.version }}
notify-gh-discussion ${{ matrix.name }} v${{ matrix.version }}
# notify-gh-discussion ${{ matrix.name }} v${{ matrix.version }}
- name: Notify ${{ matrix.name }} on Bluesky
env:
BLUESKY_APP_PWD: ${{ secrets.BLUESKY_APP_PWD }}
BLUESKY_APP_IDENTIFIER: ${{ secrets.BLUESKY_APP_IDENTIFIER }}
run: |
notify-bluesky ${{ matrix.name }} v${{ matrix.version }}

6
.github/workflows/release.yml

@ -293,3 +293,9 @@ jobs:
GH_DISCUSSION_TOKEN: ${{ secrets.GH_DISCUSSION_TOKEN }}
run: |
notify-gh-discussion ${{ matrix.name }} v${{ matrix.version }}
- name: Notify ${{ matrix.name }} on Bluesky
env:
BLUESKY_APP_PWD: ${{ secrets.BLUESKY_APP_PWD }}
BLUESKY_APP_IDENTIFIER: ${{ secrets.BLUESKY_APP_IDENTIFIER }}
run: |
notify-bluesky ${{ matrix.name }} v${{ matrix.version }}

195
tools/ci/ci_tools/social/notify_bluesky.py

@ -23,53 +23,174 @@
import os,sys
import requests
import argparse
from datetime import datetime
from pprint import pprint
from typing import Dict
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')
import re
from typing import List, Dict
def _parse_mentions(text: str) -> List[Dict]:
spans = []
# regex based on: https://atproto.com/specs/handle#handle-identifier-syntax
mention_regex = rb"[$|\W](@([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)"
text_bytes = text.encode("UTF-8")
for m in re.finditer(mention_regex, text_bytes):
spans.append({
"start": m.start(1),
"end": m.end(1),
"handle": m.group(1)[1:].decode("UTF-8")
})
return spans
def _parse_urls(text: str) -> List[Dict]:
spans = []
# partial/naive URL regex based on: https://stackoverflow.com/a/3809435
# tweaked to disallow some training punctuation
url_regex = rb"[$|\W](https?:\/\/(www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b([-a-zA-Z0-9()@:%_\+.~#?&//=]*[-a-zA-Z0-9@%_\+~#//=])?)"
text_bytes = text.encode("UTF-8")
for m in re.finditer(url_regex, text_bytes):
spans.append({
"start": m.start(1),
"end": m.end(1),
"url": m.group(1).decode("UTF-8"),
})
return spans
# Parse facets from text and resolve the handles to DIDs
def _parse_facets(text: str) -> List[Dict]:
facets = []
for m in _parse_mentions(text):
resp = requests.get(
"https://bsky.social/xrpc/com.atproto.identity.resolveHandle",
params={"handle": m["handle"]},
)
# If the handle can't be resolved, just skip it!
# It will be rendered as text in the post instead of a link
if resp.status_code == 400:
continue
did = resp.json()["did"]
facets.append({
"index": {
"byteStart": m["start"],
"byteEnd": m["end"],
},
"features": [{"$type": "app.bsky.richtext.facet#mention", "did": did}],
})
for u in _parse_urls(text):
facets.append({
"index": {
"byteStart": u["start"],
"byteEnd": u["end"],
},
"features": [
{
"$type": "app.bsky.richtext.facet#link",
# NOTE: URI ("I") not URL ("L")
"uri": u["url"],
}
],
})
return facets
def _get_facet(txt:str, slice:str, content:Dict ) -> Dict:
start = txt.index(slice)
end = start + len(slice)
return {
"index": { "byteStart": start, "byteEnd": end },
"features": [ content ]
}
def main():
parser = argparse.ArgumentParser(description="Send a Discord 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()
# Step 1: Authenticate and get access token
auth_response = requests.post(
"https://bsky.social/xrpc/com.atproto.server.createSession",
data = get_social_data(args.app)
if not data:
raise ValueError(f"app: {args.app} is not recognised")
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"
},
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"
text = f"{args.app} v{args.version} Released\n\n{data.link}\n\n#pyTermTk #TUI #Python #Linux #terminal"
post_data = {
"repo": did,
"collection": "app.bsky.feed.post",
"record": {
"$type": "app.bsky.feed.post",
"text":text,
"facets": [
_get_facet(
text, data.link,
{ "uri": data.link , "$type": "app.bsky.richtext.facet#link" }
),
_get_facet(
text, '#pyTermTk',
{ "tag": 'pyTermTk' , "$type": "app.bsky.richtext.facet#tag" }
),
_get_facet(
text, '#TUI',
{ "tag": 'TUI' , "$type": "app.bsky.richtext.facet#tag" }
),
_get_facet(
text, '#Python',
{ "tag": 'Python' , "$type": "app.bsky.richtext.facet#tag" }
),
_get_facet(
text, '#Linux',
{ "tag": 'Linux' , "$type": "app.bsky.richtext.facet#tag" }
),
_get_facet(
text, '#terminal',
{ "tag": 'terminal' , "$type": "app.bsky.richtext.facet#tag" }
),
],
"createdAt": datetime.now().isoformat() + "Z"
}
}
}
post_response = requests.post(
"https://bsky.social/xrpc/com.atproto.repo.createRecord",
headers=headers,
json=post_data
)
pprint(post_data)
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())
print(post_response.status_code)
print(post_response.json())
if __name__ == "__main__":
main()

1
tools/ci/pyproject.toml

@ -23,6 +23,7 @@
[project.scripts]
release-helper = "ci_tools.release_helper:main"
notify-discord = "ci_tools.social.notify_discord:main"
notify-bluesky = "ci_tools.social.notify_bluesky:main"
notify-gh-discussion = "ci_tools.social.notify_github_discussion:main"
[tool.setuptools]

Loading…
Cancel
Save