Loading obakasan.py +20 −5 Changes for obakasan.py: 20 added lines, 5 removed lines. Original line number Diff line number Diff line #!/usr/bin/env python3 import pymysql import os import json from typing import Any from datetime import datetime from pathlib import Path import pymysql from apscheduler.schedulers.blocking import BlockingScheduler from pymysql.cursors import DictCursor Loading @@ -28,14 +31,26 @@ def connect() -> pymysql.Connection: ) def get_views() -> None: def get_views() -> list[dict[str, Any]]: with connect() as connection: with connection.cursor() as cursor: cursor.execute(WP_POSTVIEWS_GET_COUNTS_SQL) print(cursor.fetchone()) return cursor.fetchall() def get_and_save_views() -> None: now = datetime.now() now_str = now.isoformat().split(".")[0].replace(":", "_") + "Z" views = get_views() base_dir = Path("./instance/wp_postviews") base_dir.mkdir(exist_ok=True, parents=True) with open(base_dir / f"{now_str}.json", "w") as f: json.dump(views, f, ensure_ascii=False, indent=4) if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job(get_views, "interval", minutes=1) scheduler.add_job(get_and_save_views, "cron", hour=0, minute=0) scheduler.start() Loading
obakasan.py +20 −5 Changes for obakasan.py: 20 added lines, 5 removed lines. Original line number Diff line number Diff line #!/usr/bin/env python3 import pymysql import os import json from typing import Any from datetime import datetime from pathlib import Path import pymysql from apscheduler.schedulers.blocking import BlockingScheduler from pymysql.cursors import DictCursor Loading @@ -28,14 +31,26 @@ def connect() -> pymysql.Connection: ) def get_views() -> None: def get_views() -> list[dict[str, Any]]: with connect() as connection: with connection.cursor() as cursor: cursor.execute(WP_POSTVIEWS_GET_COUNTS_SQL) print(cursor.fetchone()) return cursor.fetchall() def get_and_save_views() -> None: now = datetime.now() now_str = now.isoformat().split(".")[0].replace(":", "_") + "Z" views = get_views() base_dir = Path("./instance/wp_postviews") base_dir.mkdir(exist_ok=True, parents=True) with open(base_dir / f"{now_str}.json", "w") as f: json.dump(views, f, ensure_ascii=False, indent=4) if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job(get_views, "interval", minutes=1) scheduler.add_job(get_and_save_views, "cron", hour=0, minute=0) scheduler.start()