ニコニコ

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):

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')