diff --git a/main.py b/main.py index 1c1f522..badf8b1 100644 --- a/main.py +++ b/main.py @@ -1,16 +1,17 @@ -from datetime import datetime, timedelta import io import json import time import sys +from datetime import datetime, timedelta +import requests from atproto import Client, models from bs4 import BeautifulSoup from requests.exceptions import Timeout -import requests -from ai.talk import Talk import account +import nico +from ai.talk import Talk def check_notifications ( @@ -90,27 +91,29 @@ def main ( parent = records[0]['strong_ref'], root = records[-1]['strong_ref'])) - for datum in [e for e in get_nico_deerjika () - if e['contentId'] not in watched_videos]: - watched_videos += [datum['contentId']] - - uri = f"https://www.nicovideo.jp/watch/{ datum['contentId'] }" - (title, description, thumbnail) = get_embed_info (uri) - try: - upload = client.com.atproto.repo.upload_blob ( - io.BytesIO (requests.get (thumbnail, - timeout = 60).content)) - thumb = upload.blob - except Timeout: - thumb = None - - embed_external = models.AppBskyEmbedExternal.Main ( - external = models.AppBskyEmbedExternal.External ( - title = title, - description = description, - thumb = upload.blob, - uri = uri)) - client.post (Talk.main (f""" + latest_deerjika = nico.get_latest_deerjika () + if latest_deerjika is not None: + for datum in [e for e in [latest_deerjika]] + if e['contentId'] not in watched_videos]: + watched_videos += [datum['contentId']] + + uri = f"https://www.nicovideo.jp/watch/{ datum['contentId'] }" + (title, description, thumbnail) = get_embed_info (uri) + try: + upload = client.com.atproto.repo.upload_blob ( + io.BytesIO (requests.get (thumbnail, + timeout = 60).content)) + thumb = upload.blob + except Timeout: + thumb = None + + embed_external = models.AppBskyEmbedExternal.Main ( + external = models.AppBskyEmbedExternal.External ( + title = title, + description = description, + thumb = upload.blob, + uri = uri)) + client.post (Talk.main (f""" ニコニコに『{ datum['title'] }』という動画がアップされました。 つけられたタグは「{ '」、「'.join (datum['tags']) }」です。 概要には次のように書かれています： @@ -118,7 +121,7 @@ def main ( { datum['description'] } ``` このことについて、みんなに告知するとともに、ニジカちゃんの感想を教えてください。 """), - embed = embed_external) + embed = embed_external) if now.hour == 14 and has_got_snack_time: has_got_snack_time = False @@ -156,42 +159,6 @@ def main ( time.sleep (60) -def get_nico_deerjika (): - URL = ('https://snapshot.search.nicovideo.jp/api/v2/snapshot/video' - '/contents/search') - - now = datetime.now () - base = now - timedelta (hours = 24) - - params = { 'q': '伊地知ニジカ', - 'targets': 'tags', - '_sort': '-startTime', - 'fields': 'contentId,title,description,tags,startTime', - '_limit': 20, - 'jsonFilter': json.dumps ({ 'type': 'or', - 'filters': [{ - 'type': 'range', - 'field': 'startTime', - 'from': ('%04d-%02d-%02dT%02d:%02d:00+09:00' - % (base.year, base.month, base.day, - base.hour, base.minute)), - 'to': ('%04d-%02d-%02dT23:59:59+09:00' - % (now.year, now.month, now.day)), - 'include_lower': True }] }) } - - try: - res = requests.get (URL, params = params, timeout = 60).json () - except Timeout: - return [] - - data = [] - for datum in res['data']: - datum['tags'] = datum['tags'].split () - data.append (datum) - - return data - - def get_embed_info ( url: str ) -> (str, str, str): diff --git a/nico.py b/nico.py new file mode 100644 index 0000000..f8f3516 --- /dev/null +++ b/nico.py @@ -0,0 +1,128 @@ +""" +ニコニコのニジカ動画取得モヂュール +""" + +from typing import TypedDict + +import requests +from bs4 import BeautifulSoup +from requests.exceptions import Timeout + + +class VideoInfo (TypedDict): + contentId: str + title: str + tags: list[str] + description: str + + +def get_nico_deerjika ( +) -> list: + URL = ('https://snapshot.search.nicovideo.jp/api/v2/snapshot/video' + '/contents/search') + + now = datetime.now () + base = now - timedelta (hours = 24) + + params = { 'q': '伊地知ニジカ', + 'targets': 'tags', + '_sort': '-startTime', + 'fields': 'contentId,title,description,tags,startTime', + '_limit': 20, + 'jsonFilter': json.dumps ({ 'type': 'or', + 'filters': [{ + 'type': 'range', + 'field': 'startTime', + 'from': ('%04d-%02d-%02dT05:00:00+09:00' + % (base.year, base.month, base.day, + base.hour, base.minute)), + 'to': ('%04d-%02d-%02dT05:00:00+09:00' + % (now.year, now.month, now.day)), + 'include_lower': True }] }) } + + try: + res = requests.get (URL, params = params, timeout = 60).json () + except Timeout: + return [] + + data = [] + for datum in res['data']: + datum['tags'] = datum['tags'].split () + data.append (datum) + + return data + + pass + + +def get_latest_deerjika ( +) -> VideoInfo | None: + tag = '伊地知ニジカ' + url = f"https://www.nicovideo.jp/tag/{ tag }" + + params = { 'sort': 'f', + 'order': 'd' } + + video_info = { } + + bs = get_bs_from_url (url, params) + if bs is None: + return None + + try: + video = (bs.find_all ('ul', class_ = 'videoListInner')[1] + .find ('li', class_ = 'item')) + + video_info['contentId'] = video['data-video-id'] + except Exception: + return None + + bs = get_bs_from_url ('https://www.nicovideo.jp/watch/' + + video_info['contentId']) + if bs is None: + return None + + try: + video_info['title'] = '-'.join (bs.find ('title').text.split ('-')[:(-1)])[:(-1)] + + tags = bs.find ('meta', attrs = { 'name': 'keywords' }).get ('content') + video_info['tags'] = tags.split (',') + + video_info['description'] = bs.find ('meta', attrs = { 'name': 'description' }).get ('content') + except Exception: + return None + + return video_info + + +def get_bs_from_url ( + url: str, + params: dict = { }, +) -> BeautifulSoup | None: + """ + URL から BeautifulSoup インスタンス生成 + + Parameters + ---------- + url: str + 捜査する URL + params: dict + パラメータ + + Return + ------ + BeautifulSoup | None + BeautifulSoup オブゼクト（失敗したら None） + """ + + try: + req = requests.get (url, params = params, timeout = 60) + except Timeout: + return None + + if req.status_code != 200: + return None + + req.encoding = req.apparent_encoding + + return BeautifulSoup (req.text, 'html.parser')