import os import re import apprise from flask import Flask, request, jsonify from loguru import logger import requests import sentry_sdk DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL') DISCORD_WEBHOOK_ID = DISCORD_WEBHOOK_URL.split('/')[-2] DISCORD_WEBHOOK_TOKEN = DISCORD_WEBHOOK_URL.split('/')[-1] SENTRY_DSN = os.environ.get('SENTRY_DSN') LINEAR_API_URL = 'https://api.linear.app/graphql' LINEAR_API_KEY = os.environ.get('LINEAR_API_KEY') sentry_sdk.init( dsn=SENTRY_DSN, # Set traces_sample_rate to 1.0 to capture 100% # of transactions for performance monitoring. traces_sample_rate=1.0, # Set profiles_sample_rate to 1.0 to profile 100% # of sampled transactions. # We recommend adjusting this value in production. profiles_sample_rate=1.0, ) app = Flask(__name__) @app.route('/linear/issue', methods=['POST']) def linear_issue(): """ https://developers.linear.app/docs/graphql/webhooks#the-webhook-payload """ if request.headers.get('Linear-Event') != 'Issue': logger.error('Invalid event type: %s' % request.headers.get('Linear-Event')) return jsonify({'message': 'Invalid event type'}), 400 data = request.json logger.info('Received issue webhook: %s' % data) # {'action': 'update', 'actor': {'id': '38c20f6d-8088-461c-9ea3-9f36e185cb62', 'name': 'Ching'}, 'createdAt': '2024-03-20T09:23:00.785Z', 'data': {'id': '3f0d5021-eda8-486a-93b8-a2bd9ec461a3', 'createdAt': '2024-03-20T07:11:44.535Z', 'updatedAt': '2024-03-20T09:23:00.785Z', 'number': 21, 'title': 'etsttse', 'priority': 0, 'boardOrder': 0, 'sortOrder': -7967.13, 'completedAt': '2024-03-20T09:23:00.773Z', 'labelIds': [], 'teamId': '1f28d52c-c91a-4c48-8ca8-96425dfd6516', 'previousIdentifiers': [], 'creatorId': '38c20f6d-8088-461c-9ea3-9f36e185cb62', 'assigneeId': '38c20f6d-8088-461c-9ea3-9f36e185cb62', 'stateId': '5dbc5296-8275-4271-a595-bae6465f17c9', 'priorityLabel': 'No priority', 'botActor': {'id': '5c07d33f-5e8f-484b-8100-67908589ec45', 'type': 'workflow', 'name': 'Linear', 'avatarUrl': 'https://static.linear.app/assets/pwa/icon_maskable_512.png'}, 'identifier': 'TUN-21', 'url': 'https://linear.app/tunpok/issue/TUN-21/etsttse', 'assignee': {'id': '38c20f6d-8088-461c-9ea3-9f36e185cb62', 'name': 'Ching'}, 'state': {'id': '5dbc5296-8275-4271-a595-bae6465f17c9', 'color': '#5e6ad2', 'name': 'Done', 'type': 'completed'}, 'team': {'id': '1f28d52c-c91a-4c48-8ca8-96425dfd6516', 'key': 'TUN', 'name': 'Dev'}, 'subscriberIds': ['38c20f6d-8088-461c-9ea3-9f36e185cb62'], 'labels': [], 'description': 'sdgsgesg', 'descriptionData': '{"type":"doc","content":[{"type":"paragraph","content":[{"type":"text","text":"sdgsgesg"}]}]}'}, 'updatedFrom': {'updatedAt': '2024-03-20T09:21:05.589Z', 'sortOrder': -62.54, 'completedAt': None, 'stateId': '4f02237e-e233-410a-a4f2-3c5ff75e6927', 'canceledAt': '2024-03-20T09:21:05.572Z'}, 'url': 'https://linear.app/tunpok/issue/TUN-21/etsttse', 'type': 'Issue', 'organizationId': '60b84a77-cde4-47dd-8f56-df41efc3a899', 'webhookTimestamp': 1710926580871, 'webhookId': '76f3898f-8fb2-4d79-8c42-4ef926434fff'} if not 'action' in data: logger.error('Invalid issue data: %s' % data) return jsonify({'message': 'Invalid issue data'}), 400 if data['action'] not in ['update', 'create']: logger.warning('Ignoring issue action: %s' % data['action']) return jsonify({'message': 'Ignoring issue action'}), 200 # Send Discord message apobj = apprise.Apprise() apobj.add(f'discord://{DISCORD_WEBHOOK_ID}/{DISCORD_WEBHOOK_TOKEN}/?avatar=No&format=markdown&url={data["data"]["url"]}') apobj.asset.app_id = None if data['action'] == 'update': if not data['updatedFrom'].get('stateId'): logger.warning('Ignoring issue changes') return jsonify({'message': 'Ignoring issue changes'}), 200 title = data['data']['identifier'] + ' - ' + data['data']['title'] body = f"状态变更。\n#Status\n{data['data']['state']['name']}" notify_type = apprise.NotifyType.INFO if data['data']['state']['type'] == 'completed': notify_type = apprise.NotifyType.SUCCESS apobj.notify(body=body, title=title, notify_type=notify_type) elif data['action'] == 'create': if not data['data'].get('botActor', {}).get('subType') == 'discord': title = data['data']['identifier'] + ' - ' + data['data']['title'] body = f"创建 issue。\n#Status\n{data['data']['state']['name']}" notify_type = apprise.NotifyType.INFO for label in data['data']['labels']: if label['name'] == 'Bug': notify_type = apprise.NotifyType.FAILURE break apobj.notify(body=body, title=title, notify_type=notify_type) return jsonify({'message': 'Ok'}), 200 def _attach_url_to_issue(issue_id, url, title=None): data = {'query': 'mutation { attachmentLinkURL (title: "%s" url: "%s" issueId: "%s") { success } }' % (title, url, issue_id)} headers = {'Authorization': LINEAR_API_KEY} resp = requests.post(LINEAR_API_URL, json=data, headers=headers) if resp.status_code != 200: logger.error('Failed to attach url to issue: %s' % resp.text) return False if not resp.json().get('data', {}).get('attachmentLinkURL', {}).get('success'): logger.error('Failed to attach url to issue: %s' % resp.text) return False return True def _get_labels(): headers = {'Authorization': LINEAR_API_KEY} data = {'query': '{issueLabels { edges { node { id name } }}}'} resp = requests.post(LINEAR_API_URL, json=data, headers=headers) labels = [] for label in resp.json().get('data').get('issueLabels').get('edges'): labels.append(label['node']) return labels @app.route('/gitea/push', methods=['POST']) def gitea_push(): """ https://docs.gitea.io/en-us/webhooks/ """ data = request.json logger.info('Received gitea push webhook: %s' % data) if request.headers.get('X-Gitea-Event') != 'push': logger.error('Invalid event type: %s' % request.headers.get('X-Gitea-Event')) return jsonify({'message': 'Invalid event type'}), 400 headers = {'Authorization': LINEAR_API_KEY} # get all workflow states from linear state_query = {'query': '{workflowStates(includeArchived:false) { edges { node { id name type } }}}'} state_resp = requests.post(LINEAR_API_URL, json=state_query, headers=headers) if state_resp.status_code != 200: logger.error('Failed to get workflow states: %s' % state_resp.text) return jsonify({'message': 'Failed to get workflow states'}), 500 states = state_resp.json().get('data').get('workflowStates').get('edges') completed_state = None for state in states: if state['node']['type'] == 'completed': completed_state = state['node']['id'] break if not completed_state: logger.error('Failed to get completed state') return jsonify({'message': 'Failed to get completed state'}), 500 # check if the commit message contains a linear issue id like 'TUN-21' # if yes, update the issue state to completed for commit in data['commits']: issues = re.findall(r'TUN-\d+', commit['message']) if issues: for issue_id in issues: data_ = {'query': '{ issue(id: "%s") { title state { name type }}}' % issue_id} resp = requests.post(LINEAR_API_URL, json=data_, headers=headers) if resp.status_code != 200: logger.error('Failed to get issue: %s' % resp.text) continue issue_data = resp.json().get('data') if not issue_data: logger.error('Issue not found: %s' % issue_id) continue _attach_url_to_issue(issue_id, commit['url'], f'Gitea Commit: {commit["id"][:7]}') issue_data = issue_data['issue'] if issue_data['state']['type'] == 'completed': continue update_data = {'query': 'mutation { issueUpdate(input: { stateId: "%s" } id: "%s") { success } }' % (completed_state, issue_id)} update_resp = requests.post(LINEAR_API_URL, json=update_data, headers=headers) if update_resp.status_code != 200: logger.error('Failed to update issue: %s' % update_resp.text) continue if not update_resp.json().get('data', {}).get('issueUpdate', {}).get('success'): logger.error('Failed to update issue: %s' % update_resp.text) continue return jsonify({'message': 'Ok'}), 200 @app.route('/sentry/event', methods=['POST']) def sentry_event(): """ https://docs.sentry.io/product/integrations/integration-platform/webhooks/ """ data = request.json logger.info('Received sentry event webhook: %s' % data) event = request.headers.get('Sentry-Hook-Resource') if event == 'issue': if data.get('action') == 'created': title = f"{data['issue']['shortId']} {data['issue']['title']}" body = data['issue']['culprit'] # create a new issue in linear create_data = {'query': 'mutation { issueCreate(input: { title: "%s" description: "%s" teamId: "1f28d52c-c91a-4c48-8ca8-96425dfd6516" assigneeId: "38c20f6d-8088-461c-9ea3-9f36e185cb62" labelIds: ["b40ea30a-e48b-4511-bde7-0a3c732cf752"]}) { issue { id } } }' % (title, body)} resp = requests.post(LINEAR_API_URL, json=create_data, headers={'Authorization': LINEAR_API_KEY}) if resp.status_code != 200: logger.error('Failed to create issue: %s' % resp.text) return jsonify({'message': 'Failed to create issue'}), 500 issue_id = resp.json().get('data').get('issueCreate').get('issue').get('id') _attach_url_to_issue(issue_id, data['issue']['web_url'], title) return jsonify({'message': 'Ok'}), 200 if __name__ == "__main__": app.run(host="0.0.0.0", port=5000)