ぼざろクリーチャーシリーズ DB 兼 API(自分用)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

1046 lines
34 KiB

  1. """
  2. 日次で実行し,ぼざクリ DB を最新に更新する.
  3. """
  4. from __future__ import annotations
  5. import json
  6. import os
  7. import random
  8. import string
  9. import time
  10. from dataclasses import dataclass
  11. from datetime import date, datetime, timedelta
  12. from typing import Any, Self, Type, TypedDict, cast
  13. import mysql.connector
  14. import requests
  15. from mysql.connector.connection import MySQLConnectionAbstract
  16. class DbNull:
  17. def __new__ (
  18. cls,
  19. ):
  20. delattr (cls, '__init__')
  21. DbNullType = Type[DbNull]
  22. class VideoSearchParam (TypedDict):
  23. q: str
  24. targets: str
  25. _sort: str
  26. fields: str
  27. _limit: int
  28. jsonFilter: str
  29. class VideoResult (TypedDict):
  30. contentId: str
  31. title: str
  32. tags: str
  33. description: str | None
  34. viewCounter: int
  35. startTime: str
  36. class CommentResult (TypedDict):
  37. id: str
  38. no: int
  39. vposMs: int
  40. body: str
  41. commands: list[str]
  42. userId: str
  43. isPremium: bool
  44. score: int
  45. postedAt: str
  46. nicoruCount: int
  47. nicoruId: Any
  48. source: str
  49. isMyPost: bool
  50. class CommentRow (TypedDict):
  51. id: int
  52. video_id: int
  53. comment_no: int
  54. user_id: int
  55. content: str
  56. posted_at: datetime
  57. nico_count: int
  58. vpos_ms: int | None
  59. class TagRow (TypedDict):
  60. id: int
  61. name: str
  62. class UserRow (TypedDict):
  63. id: int
  64. code: str
  65. class VideoRow (TypedDict):
  66. id: int
  67. code: str
  68. title: str
  69. description: str
  70. uploaded_at: datetime
  71. deleted_at: datetime | None
  72. class VideoHistoryRow (TypedDict):
  73. id: int
  74. video_id: int
  75. fetched_at: date
  76. views_count: int
  77. class VideoTagRow (TypedDict):
  78. id: int
  79. video_id: int
  80. tag_id: int
  81. tagged_at: date
  82. untagged_at: date | None
  83. def main (
  84. ) -> None:
  85. conn = mysql.connector.connect (user = os.environ['MYSQL_USER'],
  86. password = os.environ['MYSQL_PASS'],
  87. database = 'nizika_nico')
  88. if not isinstance (conn, MySQLConnectionAbstract):
  89. raise TypeError
  90. now = datetime.now ()
  91. video_dao = VideoDao (conn)
  92. tag_dao = TagDao (conn)
  93. video_tag_dao = VideoTagDao (conn)
  94. video_history_dao = VideoHistoryDao (conn)
  95. comment_dao = CommentDao (conn)
  96. user_dao = UserDao (conn)
  97. api_data = search_nico_by_tags (['伊地知ニジカ', 'ぼざろクリーチャーシリーズ'])
  98. update_tables (video_dao, tag_dao, video_tag_dao, video_history_dao, comment_dao, user_dao,
  99. api_data, now)
  100. conn.commit ()
  101. conn.close ()
  102. def update_tables (
  103. video_dao: VideoDao,
  104. tag_dao: TagDao,
  105. video_tag_dao: VideoTagDao,
  106. video_history_dao: VideoHistoryDao,
  107. comment_dao: CommentDao,
  108. user_dao: UserDao,
  109. api_data: list[VideoResult],
  110. now: datetime,
  111. ) -> None:
  112. video_ids: list[int] = []
  113. for datum in api_data:
  114. tag_names: list[str] = datum['tags'].split ()
  115. video = VideoDto (code = datum['contentId'],
  116. title = datum['title'],
  117. description = datum['description'] or '',
  118. uploaded_at = datetime.fromisoformat (datum['startTime']))
  119. video_dao.upsert (video, False)
  120. if video.id_ is not None:
  121. video_ids.append (video.id_)
  122. video_history = VideoHistoryDto (video_id = video.id_,
  123. fetched_at = now,
  124. views_count = datum['viewCounter'])
  125. video_history_dao.insert (video_history)
  126. tag_ids: list[int] = []
  127. video_tags = video_tag_dao.fetch_alive_by_video_id (video.id_, False)
  128. for vt in video_tags:
  129. tag = tag_dao.find (vt.tag_id)
  130. if (tag is not None
  131. and (tag.name not in tag_names)
  132. and (tag.id_ is not None)):
  133. tag_ids.append (tag.id_)
  134. video_tag_dao.untag_all (video.id_, tag_ids, now)
  135. tags: list[TagDto] = []
  136. for tag_name in tag_names:
  137. tag = tag_dao.fetch_by_name (tag_name)
  138. if tag is None:
  139. tag = TagDto (name = tag_name)
  140. tag_dao.insert (tag)
  141. if video.id_ is not None and tag.id_ is not None:
  142. video_tag = video_tag_dao.fetch_alive_by_ids (video.id_, tag.id_, False)
  143. if video_tag is None:
  144. video_tag = VideoTagDto (video_id = video.id_,
  145. tag_id = tag.id_,
  146. tagged_at = now)
  147. video_tag_dao.insert (video_tag, False)
  148. for com in fetch_comments (video.code):
  149. user = user_dao.fetch_by_code (com['userId'])
  150. if user is None:
  151. user = UserDto (code = com['userId'])
  152. user_dao.insert (user)
  153. if video.id_ is not None and user.id_ is not None:
  154. comment = CommentDto (video_id = video.id_,
  155. comment_no = com['no'],
  156. user_id = user.id_,
  157. content = com['body'],
  158. posted_at = datetime.fromisoformat (com['postedAt']),
  159. nico_count = com['nicoruCount'],
  160. vpos_ms = com['vposMs'])
  161. comment_dao.upsert (comment, False)
  162. alive_video_codes = [d['contentId'] for d in api_data]
  163. lost_video_ids: list[int] = []
  164. videos = video_dao.fetch_alive ()
  165. for video in videos:
  166. if video.id_ is not None and video.code not in alive_video_codes:
  167. lost_video_ids.append (video.id_)
  168. video_dao.delete (lost_video_ids, now)
  169. def fetch_comments (
  170. video_code: str,
  171. ) -> list[CommentResult]:
  172. time.sleep (1.2)
  173. headers = { 'X-Frontend-Id': '6',
  174. 'X-Frontend-Version': '0' }
  175. action_track_id = (
  176. ''.join (random.choice (string.ascii_letters + string.digits)
  177. for _ in range (10))
  178. + '_'
  179. + str (random.randrange (10 ** 12, 10 ** 13)))
  180. url = (f"https://www.nicovideo.jp/api/watch/v3_guest/{ video_code }"
  181. + f"?actionTrackId={ action_track_id }")
  182. res = requests.post (url, headers = headers, timeout = 60).json ()
  183. try:
  184. nv_comment = res['data']['comment']['nvComment']
  185. except KeyError:
  186. return []
  187. if nv_comment is None:
  188. return []
  189. headers = { 'X-Frontend-Id': '6',
  190. 'X-Frontend-Version': '0',
  191. 'Content-Type': 'application/json' }
  192. params = { 'params': nv_comment['params'],
  193. 'additionals': { },
  194. 'threadKey': nv_comment['threadKey'] }
  195. url = nv_comment['server'] + '/v1/threads'
  196. res = (requests.post (url, json.dumps (params),
  197. headers = headers,
  198. timeout = 60)
  199. .json ())
  200. try:
  201. return res['data']['threads'][1]['comments']
  202. except (IndexError, KeyError):
  203. return []
  204. def search_nico_by_tag (
  205. tag: str,
  206. ) -> list[VideoResult]:
  207. return search_nico_by_tags ([tag])
  208. def search_nico_by_tags (
  209. tags: list[str],
  210. ) -> list[VideoResult]:
  211. today = datetime.now ()
  212. url = ('https://snapshot.search.nicovideo.jp'
  213. + '/api/v2/snapshot/video/contents/search')
  214. result_data: list[VideoResult] = []
  215. to = datetime (2022, 12, 3)
  216. while to <= today:
  217. time.sleep (1.2)
  218. until = to + timedelta (days = 14)
  219. query_filter = json.dumps ({ 'type': 'or',
  220. 'filters': [
  221. { 'type': 'range',
  222. 'field': 'startTime',
  223. 'from': '%04d-%02d-%02dT00:00:00+09:00' % (to.year, to.month, to.day),
  224. 'to': '%04d-%02d-%02dT23:59:59+09:00' % (until.year, until.month, until.day),
  225. 'include_lower': True }] })
  226. params: VideoSearchParam = { 'q': ' OR '.join (tags),
  227. 'targets': 'tagsExact',
  228. '_sort': '-viewCounter',
  229. 'fields': ('contentId,'
  230. 'title,'
  231. 'tags,'
  232. 'description,'
  233. 'viewCounter,'
  234. 'startTime'),
  235. '_limit': 100,
  236. 'jsonFilter': query_filter }
  237. res = requests.get (url, params = cast (dict[str, int | str], params), timeout = 60).json ()
  238. try:
  239. result_data += res['data']
  240. except KeyError:
  241. pass
  242. to = until + timedelta (days = 1)
  243. return result_data
  244. class VideoDao:
  245. def __init__ (
  246. self,
  247. conn: MySQLConnectionAbstract,
  248. ):
  249. self.conn = conn
  250. def find (
  251. self,
  252. video_id: int,
  253. with_relation_tables: bool,
  254. ) -> VideoDto | None:
  255. with self.conn.cursor (dictionary = True) as c:
  256. c.execute ("""
  257. SELECT
  258. id,
  259. code,
  260. title,
  261. description,
  262. uploaded_at,
  263. deleted_at
  264. FROM
  265. videos
  266. WHERE
  267. id = %s
  268. ORDER BY
  269. id""", (video_id,))
  270. row = cast (VideoRow | None, c.fetchone ())
  271. if row is None:
  272. return None
  273. return self._create_dto_from_row (row, with_relation_tables)
  274. def fetch_all (
  275. self,
  276. with_relation_tables: bool,
  277. ) -> list[VideoDto]:
  278. with self.conn.cursor (dictionary = True) as c:
  279. c.execute ("""
  280. SELECT
  281. id,
  282. code,
  283. title,
  284. description,
  285. uploaded_at,
  286. deleted_at
  287. FROM
  288. videos
  289. ORDER BY
  290. id""")
  291. videos: list[VideoDto] = []
  292. for row in cast (list[VideoRow], c.fetchall ()):
  293. videos.append (self._create_dto_from_row (row, with_relation_tables))
  294. return videos
  295. def fetch_alive (
  296. self,
  297. ) -> list[VideoDto]:
  298. with self.conn.cursor (dictionary = True) as c:
  299. c.execute ("""
  300. SELECT
  301. id,
  302. code,
  303. title,
  304. description,
  305. uploaded_at,
  306. deleted_at
  307. FROM
  308. videos
  309. WHERE
  310. deleted_at IS NULL""")
  311. videos: list[VideoDto] = []
  312. for row in cast (list[VideoRow], c.fetchall ()):
  313. videos.append (self._create_dto_from_row (row, False))
  314. return videos
  315. def upsert (
  316. self,
  317. video: VideoDto,
  318. with_relation_tables: bool,
  319. ) -> None:
  320. deleted_at: datetime | DbNullType | None = video.deleted_at
  321. if deleted_at is None:
  322. raise TypeError ('未実装')
  323. if deleted_at is DbNull:
  324. deleted_at = None
  325. deleted_at = cast (datetime | None, deleted_at)
  326. with self.conn.cursor (dictionary = True) as c:
  327. c.execute ("""
  328. INSERT INTO
  329. videos(
  330. code,
  331. title,
  332. description,
  333. uploaded_at,
  334. deleted_at)
  335. VALUES
  336. (
  337. %s,
  338. %s,
  339. %s,
  340. %s,
  341. %s)
  342. ON DUPLICATE KEY UPDATE
  343. code = VALUES(code),
  344. title = VALUES(title),
  345. description = VALUES(description),
  346. uploaded_at = VALUES(uploaded_at),
  347. deleted_at = VALUES(deleted_at)""", (video.code,
  348. video.title,
  349. video.description,
  350. video.uploaded_at,
  351. deleted_at))
  352. video.id_ = c.lastrowid
  353. if with_relation_tables:
  354. if video.video_tags is not None:
  355. VideoTagDao (self.conn).upsert_all (video.video_tags, False)
  356. if video.comments is not None:
  357. CommentDao (self.conn).upsert_all (video.comments, False)
  358. if video.video_histories is not None:
  359. VideoHistoryDao (self.conn).upsert_all (video.video_histories)
  360. def upsert_all (
  361. self,
  362. videos: list[VideoDto],
  363. with_relation_tables: bool,
  364. ) -> None:
  365. for video in videos:
  366. self.upsert (video, with_relation_tables)
  367. def delete (
  368. self,
  369. video_ids: list[int],
  370. at: datetime,
  371. ) -> None:
  372. if not video_ids:
  373. return
  374. with self.conn.cursor (dictionary = True) as c:
  375. c.execute ("""
  376. UPDATE
  377. videos
  378. SET
  379. deleted_at = %%s
  380. WHERE
  381. id IN (%s)""" % ', '.join (['%s'] * len (video_ids)), (at, *video_ids))
  382. def _create_dto_from_row (
  383. self,
  384. row: VideoRow,
  385. with_relation_tables: bool,
  386. ) -> VideoDto:
  387. video = VideoDto (id_ = row['id'],
  388. code = row['code'],
  389. title = row['title'],
  390. description = row['description'],
  391. uploaded_at = row['uploaded_at'],
  392. deleted_at = row['deleted_at'] or DbNull)
  393. if with_relation_tables and video.id_ is not None:
  394. video.video_tags = VideoTagDao (self.conn).fetch_by_video_id (video.id_, False)
  395. for i in range (len (video.video_tags)):
  396. video.video_tags[i].video = video
  397. video.comments = CommentDao (self.conn).fetch_by_video_id (video.id_, False)
  398. for i in range (len (video.comments)):
  399. video.comments[i].video = video
  400. video.video_histories = VideoHistoryDao (self.conn).fetch_by_video_id (video.id_, False)
  401. for i in range (len (video.video_histories)):
  402. video.video_histories[i].video = video
  403. return video
  404. @dataclass (slots = True)
  405. class VideoDto:
  406. code: str
  407. title: str
  408. description: str
  409. uploaded_at: datetime
  410. id_: int | None = None
  411. deleted_at: datetime | DbNullType = DbNull
  412. video_tags: list[VideoTagDto] | None = None
  413. comments: list[CommentDto] | None = None
  414. video_histories: list[VideoHistoryDto] | None = None
  415. class VideoTagDao:
  416. def __init__ (
  417. self,
  418. conn: MySQLConnectionAbstract,
  419. ):
  420. self.conn = conn
  421. def fetch_by_video_id (
  422. self,
  423. video_id: int,
  424. with_relation_tables: bool,
  425. ) -> list[VideoTagDto]:
  426. with self.conn.cursor (dictionary = True) as c:
  427. c.execute ("""
  428. SELECT
  429. id,
  430. video_id,
  431. tag_id,
  432. tagged_at,
  433. untagged_at
  434. FROM
  435. video_tags
  436. WHERE
  437. video_id = %s
  438. ORDER BY
  439. id""", (video_id,))
  440. video_tags: list[VideoTagDto] = []
  441. for row in cast (list[VideoTagRow], c.fetchall ()):
  442. video_tags.append (self._create_dto_from_row (row, with_relation_tables))
  443. return video_tags
  444. def fetch_alive_by_video_id (
  445. self,
  446. video_id: int,
  447. with_relation_tables: bool,
  448. ) -> list[VideoTagDto]:
  449. with self.conn.cursor (dictionary = True) as c:
  450. c.execute ("""
  451. SELECT
  452. id,
  453. video_id,
  454. tag_id,
  455. tagged_at,
  456. untagged_at
  457. FROM
  458. video_tags
  459. WHERE
  460. video_id = %s
  461. AND (untagged_at IS NULL)
  462. ORDER BY
  463. id""", (video_id,))
  464. video_tags: list[VideoTagDto] = []
  465. for row in cast (list[VideoTagRow], c.fetchall ()):
  466. video_tags.append (self._create_dto_from_row (row, with_relation_tables))
  467. return video_tags
  468. def fetch_alive_by_ids (
  469. self,
  470. video_id: int,
  471. tag_id: int,
  472. with_relation_tables: bool,
  473. ) -> VideoTagDto | None:
  474. with self.conn.cursor (dictionary = True) as c:
  475. c.execute ("""
  476. SELECT
  477. id,
  478. video_id,
  479. tag_id,
  480. tagged_at,
  481. untagged_at
  482. FROM
  483. video_tags
  484. WHERE
  485. video_id = %s
  486. AND tag_id = %s""", (video_id, tag_id))
  487. row = cast (VideoTagRow, c.fetchone ())
  488. if row is None:
  489. return None
  490. return self._create_dto_from_row (row, with_relation_tables)
  491. def insert (
  492. self,
  493. video_tag: VideoTagDto,
  494. with_relation_tables: bool,
  495. ) -> None:
  496. untagged_at: date | DbNullType | None = video_tag.untagged_at
  497. if untagged_at is None:
  498. raise TypeError ('未実装')
  499. if untagged_at is DbNull:
  500. untagged_at = None
  501. untagged_at = cast (date | None, untagged_at)
  502. with self.conn.cursor (dictionary = True) as c:
  503. c.execute ("""
  504. INSERT INTO
  505. video_tags(
  506. video_id,
  507. tag_id,
  508. tagged_at,
  509. untagged_at)
  510. VALUES
  511. (
  512. %s,
  513. %s,
  514. %s,
  515. %s)""", (video_tag.video_id, video_tag.tag_id,
  516. video_tag.tagged_at, untagged_at))
  517. video_tag.id_ = c.lastrowid
  518. if with_relation_tables:
  519. if video_tag.video is not None:
  520. VideoDao (self.conn).upsert (video_tag.video, True)
  521. if video_tag.tag is not None:
  522. TagDao (self.conn).upsert (video_tag.tag)
  523. def upsert (
  524. self,
  525. video_tag: VideoTagDto,
  526. with_relation_tables: bool,
  527. ) -> None:
  528. untagged_at: date | DbNullType | None = video_tag.untagged_at
  529. if untagged_at is None:
  530. raise TypeError ('未実装')
  531. if untagged_at is DbNull:
  532. untagged_at = None
  533. untagged_at = cast (date | None, untagged_at)
  534. with self.conn.cursor (dictionary = True) as c:
  535. c.execute ("""
  536. INSERT INTO
  537. video_tags(
  538. video_id,
  539. tag_id,
  540. tagged_at,
  541. untagged_at)
  542. VALUES
  543. (
  544. %s,
  545. %s,
  546. %s,
  547. %s)
  548. ON DUPLICATE KEY UPDATE
  549. video_id = VALUES(video_id),
  550. tag_id = VALUES(tag_id),
  551. tagged_at = VALUES(tagged_at),
  552. untagged_at = VALUES(untagged_at)""", (video_tag.video_id,
  553. video_tag.tag_id,
  554. video_tag.tagged_at,
  555. untagged_at))
  556. video_tag.id_ = c.lastrowid
  557. if with_relation_tables:
  558. if video_tag.video is not None:
  559. VideoDao (self.conn).upsert (video_tag.video, True)
  560. if video_tag.tag is not None:
  561. TagDao (self.conn).upsert (video_tag.tag)
  562. def upsert_all (
  563. self,
  564. video_tags: list[VideoTagDto],
  565. with_relation_tables: bool,
  566. ) -> None:
  567. for video_tag in video_tags:
  568. self.upsert (video_tag, with_relation_tables)
  569. def untag_all (
  570. self,
  571. video_id: int,
  572. tag_ids: list[int],
  573. now: datetime,
  574. ) -> None:
  575. if not tag_ids:
  576. return
  577. with self.conn.cursor (dictionary = True) as c:
  578. c.execute ("""
  579. UPDATE
  580. video_tags
  581. SET
  582. untagged_at = %%s
  583. WHERE
  584. video_id = %%s
  585. AND tag_ids IN (%s)""" % ', '.join (['%s'] * len (tag_ids)), (now, video_id, *tag_ids))
  586. def _create_dto_from_row (
  587. self,
  588. row: VideoTagRow,
  589. with_relation_tables: bool,
  590. ) -> VideoTagDto:
  591. video_tag = VideoTagDto (id_ = row['id'],
  592. video_id = row['video_id'],
  593. tag_id = row['tag_id'],
  594. tagged_at = row['tagged_at'],
  595. untagged_at = row['untagged_at'] or DbNull)
  596. if with_relation_tables:
  597. video_tag.video = VideoDao (self.conn).find (video_tag.video_id, True)
  598. video_tag.tag = TagDao (self.conn).find (video_tag.tag_id)
  599. return video_tag
  600. @dataclass (slots = True)
  601. class VideoTagDto:
  602. video_id: int
  603. tag_id: int
  604. tagged_at: date
  605. id_: int | None = None
  606. untagged_at: date | DbNullType = DbNull
  607. video: VideoDto | None = None
  608. tag: TagDto | None = None
  609. class TagDao:
  610. def __init__ (
  611. self,
  612. conn: MySQLConnectionAbstract,
  613. ):
  614. self.conn = conn
  615. def find (
  616. self,
  617. tag_id: int,
  618. ) -> TagDto | None:
  619. with self.conn.cursor (dictionary = True) as c:
  620. c.execute ("""
  621. SELECT
  622. id,
  623. name
  624. FROM
  625. tags
  626. WHERE
  627. id = %s""", (tag_id,))
  628. row = cast (TagRow | None, c.fetchone ())
  629. if row is None:
  630. return None
  631. return self._create_dto_from_row (row)
  632. def fetch_by_name (
  633. self,
  634. tag_name: str,
  635. ) -> TagDto | None:
  636. with self.conn.cursor (dictionary = True) as c:
  637. c.execute ("""
  638. SELECT
  639. id,
  640. name
  641. FROM
  642. tags
  643. WHERE
  644. name = %s""", (tag_name,))
  645. row = cast (TagRow | None, c.fetchone ())
  646. if row is None:
  647. return None
  648. return self._create_dto_from_row (row)
  649. def insert (
  650. self,
  651. tag: TagDto,
  652. ) -> None:
  653. with self.conn.cursor (dictionary = True) as c:
  654. c.execute ("""
  655. INSERT INTO
  656. tags(name)
  657. VALUES
  658. (%s)""", (tag.name,))
  659. tag.id_ = c.lastrowid
  660. def upsert (
  661. self,
  662. tag: TagDto,
  663. ) -> None:
  664. with self.conn.cursor (dictionary = True) as c:
  665. c.execute ("""
  666. INSERT INTO
  667. tags(name)
  668. VALUES
  669. (%s)
  670. ON DUPLICATE KEY UPDATE
  671. name = VALUES(name)""", (tag.name,))
  672. tag.id_ = c.lastrowid
  673. def _create_dto_from_row (
  674. self,
  675. row: TagRow,
  676. ) -> TagDto:
  677. return TagDto (id_ = row['id'],
  678. name = row['name'])
  679. @dataclass (slots = True)
  680. class TagDto:
  681. name: str
  682. id_: int | None = None
  683. class VideoHistoryDao:
  684. def __init__ (
  685. self,
  686. conn: MySQLConnectionAbstract,
  687. ):
  688. self.conn = conn
  689. def fetch_by_video_id (
  690. self,
  691. video_id: int,
  692. with_relation_tables: bool,
  693. ) -> list[VideoHistoryDto]:
  694. with self.conn.cursor (dictionary = True) as c:
  695. c.execute ("""
  696. SELECT
  697. id,
  698. video_id,
  699. fetched_at,
  700. views_count
  701. FROM
  702. video_histories
  703. WHERE
  704. video_id = %s""", (video_id,))
  705. video_histories: list[VideoHistoryDto] = []
  706. for row in cast (list[VideoHistoryRow], c.fetchall ()):
  707. video_histories.append (self._create_dto_from_row (row, with_relation_tables))
  708. return video_histories
  709. def insert (
  710. self,
  711. video_history: VideoHistoryDto,
  712. ) -> None:
  713. with self.conn.cursor (dictionary = True) as c:
  714. c.execute ("""
  715. INSERT INTO
  716. video_histories(
  717. video_id,
  718. fetched_at,
  719. views_count)
  720. VALUES
  721. (
  722. %s,
  723. %s,
  724. %s)""", (video_history.video_id,
  725. video_history.fetched_at,
  726. video_history.views_count))
  727. def upsert (
  728. self,
  729. video_history: VideoHistoryDto,
  730. ) -> None:
  731. with self.conn.cursor (dictionary = True) as c:
  732. c.execute ("""
  733. INSERT INTO
  734. video_histories(
  735. video_id,
  736. fetched_at,
  737. views_count)
  738. VALUES
  739. (
  740. %s,
  741. %s,
  742. %s)
  743. ON DUPLICATE KEY UPDATE
  744. video_id,
  745. fetched_at,
  746. views_count""", (video_history.video_id,
  747. video_history.fetched_at,
  748. video_history.views_count))
  749. def upsert_all (
  750. self,
  751. video_histories: list[VideoHistoryDto],
  752. ) -> None:
  753. for video_history in video_histories:
  754. self.upsert (video_history)
  755. def _create_dto_from_row (
  756. self,
  757. row: VideoHistoryRow,
  758. with_relation_tables: bool,
  759. ) -> VideoHistoryDto:
  760. video_history = VideoHistoryDto (id_ = row['id'],
  761. video_id = row['video_id'],
  762. fetched_at = row['fetched_at'],
  763. views_count = row['views_count'])
  764. if with_relation_tables:
  765. video_history.video = VideoDao (self.conn).find (video_history.video_id, True)
  766. return video_history
  767. @dataclass (slots = True)
  768. class VideoHistoryDto:
  769. video_id: int
  770. fetched_at: date
  771. views_count: int
  772. id_: int | None = None
  773. video: VideoDto | None = None
  774. class CommentDao:
  775. def __init__ (
  776. self,
  777. conn: MySQLConnectionAbstract,
  778. ):
  779. self.conn = conn
  780. def fetch_by_video_id (
  781. self,
  782. video_id: int,
  783. with_relation_tables: bool,
  784. ) -> list[CommentDto]:
  785. with self.conn.cursor (dictionary = True) as c:
  786. c.execute ("""
  787. SELECT
  788. id,
  789. video_id,
  790. comment_no,
  791. user_id,
  792. content,
  793. posted_at,
  794. nico_count,
  795. vpos_ms
  796. FROM
  797. comments
  798. WHERE
  799. video_id = %s""", (video_id,))
  800. comments: list[CommentDto] = []
  801. for row in cast (list[CommentRow], c.fetchall ()):
  802. comments.append (self._create_dto_from_row (row, with_relation_tables))
  803. return comments
  804. def upsert (
  805. self,
  806. comment: CommentDto,
  807. with_relation_tables: bool,
  808. ) -> None:
  809. vpos_ms: int | DbNullType | None = comment.vpos_ms
  810. if vpos_ms is None:
  811. raise TypeError ('未実装')
  812. if vpos_ms is DbNull:
  813. vpos_ms = None
  814. vpos_ms = cast (int | None, vpos_ms)
  815. with self.conn.cursor (dictionary = True) as c:
  816. c.execute ("""
  817. INSERT INTO
  818. comments(
  819. video_id,
  820. comment_no,
  821. user_id,
  822. content,
  823. posted_at,
  824. nico_count,
  825. vpos_ms)
  826. VALUES
  827. (
  828. %s,
  829. %s,
  830. %s,
  831. %s,
  832. %s,
  833. %s,
  834. %s)
  835. ON DUPLICATE KEY UPDATE
  836. video_id = VALUES(video_id),
  837. comment_no = VALUES(comment_no),
  838. user_id = VALUES(user_id),
  839. content = VALUES(content),
  840. posted_at = VALUES(posted_at),
  841. nico_count = VALUES(nico_count),
  842. vpos_ms = VALUES(vpos_ms)""", (comment.video_id,
  843. comment.comment_no,
  844. comment.user_id,
  845. comment.content,
  846. comment.posted_at,
  847. comment.nico_count,
  848. vpos_ms))
  849. def upsert_all (
  850. self,
  851. comments: list[CommentDto],
  852. with_relation_tables: bool,
  853. ) -> None:
  854. for comment in comments:
  855. self.upsert (comment, with_relation_tables)
  856. def _create_dto_from_row (
  857. self,
  858. row: CommentRow,
  859. with_relation_tables: bool,
  860. ) -> CommentDto:
  861. comment = CommentDto (id_ = row['id'],
  862. video_id = row['video_id'],
  863. comment_no = row['comment_no'],
  864. user_id = row['user_id'],
  865. content = row['content'],
  866. posted_at = row['posted_at'],
  867. nico_count = row['nico_count'],
  868. vpos_ms = row['vpos_ms'] or DbNull)
  869. if with_relation_tables:
  870. comment.video = VideoDao (self.conn).find (comment.video_id, True)
  871. return comment
  872. @dataclass (slots = True)
  873. class CommentDto:
  874. video_id: int
  875. comment_no: int
  876. user_id: int
  877. content: str
  878. posted_at: datetime
  879. id_: int | None = None
  880. nico_count: int = 0
  881. vpos_ms: int | DbNullType = DbNull
  882. video: VideoDto | None = None
  883. user: UserDto | None = None
  884. class UserDao:
  885. def __init__ (
  886. self,
  887. conn: MySQLConnectionAbstract,
  888. ):
  889. self.conn = conn
  890. def fetch_by_code (
  891. self,
  892. user_code: str
  893. ) -> UserDto | None:
  894. with self.conn.cursor (dictionary = True) as c:
  895. c.execute ("""
  896. SELECT
  897. id,
  898. code
  899. FROM
  900. users
  901. WHERE
  902. code = %s""", (user_code,))
  903. row = cast (UserRow | None, c.fetchone ())
  904. if row is None:
  905. return None
  906. return self._create_dto_from_row (row)
  907. def insert (
  908. self,
  909. user: UserDto,
  910. ) -> None:
  911. with self.conn.cursor (dictionary = True) as c:
  912. c.execute ("""
  913. INSERT INTO
  914. users(code)
  915. VALUES
  916. (%s)""", (user.code,))
  917. user.id_ = c.lastrowid
  918. def _create_dto_from_row (
  919. self,
  920. row: UserRow,
  921. ) -> UserDto:
  922. return UserDto (id_ = row['id'],
  923. code = row['code'])
  924. @dataclass (slots = True)
  925. class UserDto:
  926. code: str
  927. id_: int | None = None
  928. if __name__ == '__main__':
  929. main ()