resolve #14 check producer cat; #13 some more async stuff

main.py

@@ -18,7 +18,7 @@
 from utils.helpers import prompt_choices, prompt_response, only_canonical_videos, get_video, \
     prompt_multiline
 from utils.image import write_to_file, download_thumbnail, pick_color
-from utils.mgp import get_producer_templates
+from utils.mgp import get_producer_templates, get_producer_cats, get_producer_info
 from utils.name_converter import name_to_cat, name_to_chinese
 from utils.string import auto_lj, is_empty, datetime_to_ymd, assert_str_exists, join_string
 from utils.vocadb import get_song_by_name
@@ -109,7 +109,7 @@ def create_intro(song: Song):
             f"""是由{join_string(song.creators.producers_str()[:1],
                                inner_wrapper=('[[', ']]'),
                                mapper=auto_lj)}""" +
-            videos_to_str2(videos) + "的日文原创歌曲。" +
+            videos_to_str2(videos) + "的[[VOCALOID]]日文原创歌曲。" +
             f"""由{join_string(song.creators.vocalists_str(),
                               outer_wrapper=('[[', ']]'),
                               mapper=name_to_chinese)}演唱。""" +
@@ -163,15 +163,18 @@ def create_lyrics(song: Song):
 
 
 def create_end(song: Song):
-    lst = asyncio.run(get_producer_templates(song.creators.producers))
-    producer_string = join_string(lst, outer_wrapper=("{{Template:", "}}\n"))
-    return (producer_string + """== 注释与外部链接 ==
+    list_templates, list_cats = asyncio.run(get_producer_info(song.creators.producers))
+    producer_templates = join_string(list_templates, deliminator="",
+                                     outer_wrapper=("{{Template:", "}}\n"))
+    producer_cats = join_string(list_cats, deliminator="",
+                                outer_wrapper=("[[Category:", "作品]]\n"))
+    vocalist_cat = join_string(song.creators.vocalists_str(),
+                               deliminator="", mapper=name_to_cat,
+                               outer_wrapper=('[[分类:', '歌曲]]\n'))
+    return (producer_templates + """== 注释与外部链接 ==
 <references/>
 [[分类:日本音乐作品]]
-[[分类:使用VOCALOID的歌曲]]\n""" +
-            join_string(song.creators.vocalists_str(),
-                        deliminator="\n", mapper=name_to_cat,
-                        outer_wrapper=('[[分类:', '歌曲]]')))
+[[分类:使用VOCALOID的歌曲]]\n""" + vocalist_cat + producer_cats)
 
 
 def get_video_bilibili() -> Union[Video, None]:

tests/utils/test_mgp.py

@@ -1,8 +1,9 @@
 import asyncio
+import unittest
 from unittest import TestCase
 
 from models.creators import Person
-from utils.mgp import producer_template_exists, get_producer_templates
+from utils.mgp import producer_template_exists, get_producer_templates, producer_cat_exists, get_producer_cats
 
 
 class TestProducerTemplate(TestCase):
@@ -21,3 +22,22 @@ def test_get_producer_templates(self):
                      Person("谁也不是", ["胡话P", "LyricsKai"]),
                      Person("什么鬼", ["HarryP", "针原翼"])]
         self.assertEquals(['Wowaka', 'HarryP'], asyncio.run(get_producer_templates(producers)))
+
+
+class TestProducerCat(TestCase):
+    def test_cat_exists(self):
+        true = ["wowaka"]
+        for t in true:
+            self.assertTrue(producer_cat_exists(t))
+        false = ["黑幕", "这是一个不存在的分类"]
+        for t in false:
+            self.assertFalse(producer_cat_exists(t))
+
+    def test_get_producer_cats(self):
+        producers = [Person("黑幕", ["wowakaP"]),
+                     Person("谁也不是", ["胡话P", "LyricsKai"])]
+        self.assertEquals(['wowaka'], asyncio.run(get_producer_cats(producers)))
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/mgp.py

@@ -1,18 +1,19 @@
 import asyncio
 import logging
 import urllib
-from typing import Union, List
+from typing import Union, List, Tuple
 
 import requests
 from bs4 import BeautifulSoup
 
 from models.creators import Person
 
-BASE = "https://zh.moegirl.org.cn/Template:{}"
+BASE_TEMPLATE = "https://zh.moegirl.org.cn/Template:{}"
+BASE_CAT = "https://zh.moegirl.org.cn/Category:{}作品"
 
 
 def producer_template_exists(name: str) -> bool:
-    url = BASE.format(urllib.parse.quote(name))
+    url = BASE_TEMPLATE.format(urllib.parse.quote(name))
     soup = BeautifulSoup(requests.get(url).text, "html.parser")
     result = soup.find("div", {"id", "mw-normal-catlinks"})
     if not result:
@@ -24,20 +25,49 @@ def producer_template_exists(name: str) -> bool:
     return False
 
 
+def producer_cat_exists(name: str) -> bool:
+    url = BASE_CAT.format(urllib.parse.quote(name))
+    soup = BeautifulSoup(requests.get(url).text, "html.parser")
+    result = soup.find("div", {"id": "topicpath"})
+    return not not result
+
+
 async def check_template_names(names: List[str]) -> Union[str, None]:
     for name in names:
         if producer_template_exists(name):
             return name
     return None
 
 
-async def get_producer_templates(producers: List[Person]) -> List[str]:
-    logging.info("Fetching producer templates for " + ", ".join([p.name for p in producers]))
+async def check_cat_names(names: List[str]) -> Union[str, None]:
+    for name in names:
+        if producer_cat_exists(name):
+            return name
+    return None
+
+
+async def producer_checker(producers: List[Person], task):
     result = []
     for producer in producers:
         names = [producer.name, *producer.name_eng]
         names.extend([name[:-1] for name in names if name[-1] == 'P'])
-        result.append(asyncio.create_task(check_template_names(names)))
+        result.append(asyncio.create_task(task(names)))
     result = [await r for r in result]
     result = [r for r in result if r]
     return result
+
+
+async def get_producer_templates(producers: List[Person]) -> List[str]:
+    logging.info("Fetching producer templates for " + ", ".join([p.name for p in producers]))
+    return await producer_checker(producers, check_template_names)
+
+
+async def get_producer_cats(producers: List[Person]) -> List[str]:
+    logging.info("Fetching producer categories for " + ", ".join([p.name for p in producers]))
+    return await producer_checker(producers, check_cat_names)
+
+
+async def get_producer_info(producers: List[Person]) -> Tuple[List[str], List[str]]:
+    task1 = asyncio.create_task(get_producer_templates(producers))
+    task2 = asyncio.create_task(get_producer_cats(producers))
+    return await task1, await task2