From edf0ad3045dcb0f0871b0baf0bc8b969c4b59228 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BE=B3=E4=BD=8F=E5=8F=8B=E7=A8=9C?= Date: Sun, 8 Dec 2019 00:11:00 +0900 Subject: [PATCH] first --- .gitignore | 4 + .python-version | 1 + README.md | 10 ++ qiita_iine_collect/check_new_iine_dev.py | 121 +++++++++++++++++ qiita_notification/send_new_iine_dev.py | 158 +++++++++++++++++++++++ 5 files changed, 294 insertions(+) create mode 100644 .gitignore create mode 100644 .python-version create mode 100644 README.md create mode 100644 qiita_iine_collect/check_new_iine_dev.py create mode 100644 qiita_notification/send_new_iine_dev.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..574b264 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__ +env +.mypy_cache +.vscode \ No newline at end of file diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..424e179 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.6.8 diff --git a/README.md b/README.md new file mode 100644 index 0000000..8a71ef1 --- /dev/null +++ b/README.md @@ -0,0 +1,10 @@ +# Qiita notification job with AWS Lambda and Dynamo DB Stream. +this repository includes two code, which apply to Lambda one by one. + +## /qiita_iine_collect/check_new_iine_dev.py +- collect all articles iine by Qiita API v2 +- update logs in Dynamo DB to stream differences, which is target of notification + +## /qiita_notification/send_notification.py +- get stream data of Dynamo DB +- notify via LINE Notify \ No newline at end of file diff --git a/qiita_iine_collect/check_new_iine_dev.py b/qiita_iine_collect/check_new_iine_dev.py new file mode 100644 index 0000000..2f4e1c6 --- /dev/null +++ b/qiita_iine_collect/check_new_iine_dev.py @@ -0,0 +1,121 @@ +import os +from math import ceil +from typing import List, Dict, Any, Union, Tuple +import json +from urllib.request import Request +from urllib import request, parse, error +from http.client import HTTPResponse +import boto3 +from botocore.exceptions import ClientError + + +class Response(): + """Http Response Object""" + + def __init__(self, res: HTTPResponse): + self.body = self._json(res) + self.status_code = self._status_code(res) + self.headers = self._headers(res) + + def _json(self, res: HTTPResponse): + return json.loads(res.read()) + + def _status_code(self, res: HTTPResponse) -> int: + return res.status + + def _headers(self, res: HTTPResponse) -> Dict[str, str]: + return dict(res.getheaders()) + + +def req_get(url, headers=None, params=None) -> Response: + """get request. simplified request function of Requests + :return: Response object + """ + if params: + url = '{}?{}'.format(url, parse.urlencode(params)) + + req = Request(url, headers=headers, method='GET') + + with request.urlopen(req) as res: + response = Response(res) + return response + + +def serialize_response(response: Response) -> List[Dict[str, Any]]: + """serialize response of Qiita API v2 + :param response: + :return: + """ + keys = ['id', 'title', 'likes_count'] + return [ + {f: resp.get(f) for f in keys} for resp in response.body + ] + + +def get_item(url: str, headers: Dict[str, str], **param) -> List[Dict[str, Any]]: + """get a item by Qiita API v2 and return the list of serialized response (dictionary)""" + response = req_get(url, headers=headers, params=param) + return serialize_response(response) + + +def get_items(token: str, per_page=1, url='https://qiita.com/api/v2/authenticated_user/items') -> List[Dict[str, Any]]: + """ページネーションして認証ユーザの全ての記事を取得する + :return: 記事のリスト + """ + headers = {'Authorization': 'Bearer {}'.format(token)} + + response: Response = req_get(url, headers=headers, params={'page': 1, 'per_page': per_page}) + items = serialize_response(response) + tot_count = int(response.headers['Total-Count']) + tot_pages = ceil(tot_count / per_page) + if tot_pages <= 1: + return items + + for page in range(2, tot_pages + 1): + items += get_item(url, headers, page=page, per_page=per_page) + return items + + +def update_logs(items: List[Dict[str, Any]]): + """Update the number of iine in Dynamo DB + If item ID do not exist in Dynamo DB, insert them in it + """ + dynamodb = boto3.resource('dynamodb') + + table = dynamodb.Table('iine_qiita_logs') + + for item in items: + ids = item.get('id') + title = item.get('title') + iine = item.get('likes_count') + + try: + response = table.update_item( + Key={ + 'ids': ids + }, + UpdateExpression="set iine = :newiine, title = :title", + ConditionExpression="attribute_not_exists(ids) or iine <> :newiine", + ExpressionAttributeValues={ + ":newiine": iine, + ":title": title + }, + ) + except ClientError as e: + if e.response['Error']['Code'] == "ConditionalCheckFailedException": + print(e.response['Error']['Message']) + else: + raise + + +def main(client, content): + """this is handler function for Lambda""" + qiita_token: str = os.environ['QIITA_TOKEN'] + url: str = os.environ['QIITA_URL'] + per_page = int(os.environ['PER_PAGE']) + + items: List[Dict[str, Any]] = get_items(qiita_token, per_page=per_page, url=url) + update_logs(items) + return { + 'statusCode': 200 + } diff --git a/qiita_notification/send_new_iine_dev.py b/qiita_notification/send_new_iine_dev.py new file mode 100644 index 0000000..696692f --- /dev/null +++ b/qiita_notification/send_new_iine_dev.py @@ -0,0 +1,158 @@ +import json +import os +from math import ceil +from typing import List, Dict, Any, Union, Tuple +import json +from urllib.request import Request +from urllib import request, parse, error +from http.client import HTTPResponse + + +class Response(): + """Http Response Object""" + + def __init__(self, res: HTTPResponse): + self.body = self._json(res) + self.status_code = self._status_code(res) + self.headers = self._headers(res) + + def _json(self, res: HTTPResponse): + return json.loads(res.read()) + + def _status_code(self, res: HTTPResponse) -> int: + return res.status + + def _headers(self, res: HTTPResponse) -> Dict[str, str]: + return dict(res.getheaders()) + + +def req_get(url: str, headers=None, params=None) -> Response: + """get request. simplified request function of Requests + :return: Response object + """ + if params: + url = '{}?{}'.format(url, parse.urlencode(params)) + + req = Request(url, headers=headers, method='GET') + + with request.urlopen(req) as res: + response = Response(res) + return response + + +def req_post(url: str, data: Dict[str, Any], headers=None, params=None) -> Response: + """post request. simplified request function of Requests + :return: Response object + """ + if headers.get('Content-Type') == 'application/x-www-form-urlencoded': + encoded_data = parse.urlencode(data).encode() + + else: + encoded_data = json.dumps(data).encode() + + req = Request(url, data=encoded_data, headers=headers, method='POST') + + with request.urlopen(req) as res: + response = Response(res) + return response + + +def serialize_record(record: Dict[str, Any]) -> Dict[str, Any]: + """serialize data of Dynamo DB Stream + :return: + """ + if record.get('eventName') != 'MODIFY': + return {} + + past = record.get('dynamodb', {}).get('OldImage') + past_iine = int(past.get('iine', {}).get('N', 0)) + ids = past.get('ids', {}).get('S', '') + + new = record.get('dynamodb', {}).get('NewImage') + title = new.get('title', {}).get('S', '') + return { + 'ids': ids, + 'title': title, + 'past_iine': past_iine + } + + +def serialize_response_name(response: Response, num: int, title: str) -> Dict[str, Any]: + """serialize iine data of Qiita API v2 + :param response: + :return: + """ + size = len(response.body) - num + if size <= 0: + users: List[str] = [] + + new_iine = response.body[:size] + users = [ + resp.get('user', {}).get('id') for resp in new_iine + ] + return { + 'title': title, + 'users': users + } + + +def get_new_iine(item: Dict[str, Any], token: str) -> Dict[str, Any]: + """HTTP request to Qiita API v2 + :params: + :return: + """ + headers = {'Authorization': 'Bearer {}'.format(token)} + ids = item.get('ids', '') + past_iine = item.get('past_iine', 0) + url = f'https://qiita.com/api/v2/items/{ids}/likes' + + response = req_get(url, headers=headers) + title: str = item.get('title', '') + resp = serialize_response_name(response, past_iine, title) + return resp + + +def deserialize_response_name(response: Dict[str, Any], max_length=20) -> str: + """deserialize text for LINE Notify + :param max_length: max sentence length + :return: + """ + names = ", ".join(response.get('users', [])) + title = response.get('title', '') + title = f"{title}" if len(title) <= max_length else f"{title[:max_length]}..." + return f"\n{names}が「{title}」にいいねしました。" + + +def send_notification(message: str, token: str): + """send notification by LINE notify""" + url = 'https://notify-api.line.me/api/notify' + + headers = { + 'Authorization': 'Bearer {}'.format(token), + 'Content-Type': 'application/x-www-form-urlencoded' + } + msg = { + 'message': message + } + response = req_post(url, data=msg, headers=headers) + return response.body + + +def lambda_handler(event, context): + """main handler for Lambda""" + qiita_token = os.environ["QIITA_TOKEN"] + line_token = os.environ["LINE_TOKEN"] + + records = event.get('Records', []) + for record in records: + serialized_data = serialize_record(record) + if not serialized_data: + continue + new_iines = get_new_iine(serialized_data, qiita_token) + if len(new_iines.get('users')) == 0: + continue + send_notification(deserialize_response_name(new_iines), line_token) + + return { + 'statusCode': 200, + }