Files
ordinaceprojekt/SběrDatRůzné/DailySudoku/stahni_sudoku.py
T
Vladimir Buzalka c9903646f1 notebookvb
2026-05-08 13:30:47 +02:00

98 lines
3.1 KiB
Python

"""
Stáhne daily Sudoku puzzle data ze solitaire.org a uloží do MySQL.
"""
import asyncio
import json
import sys
sys.stdout.reconfigure(encoding="utf-8")
sys.stderr.reconfigure(encoding="utf-8")
from datetime import date, timedelta
from pathlib import Path
from playwright.async_api import async_playwright
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "Knihovny"))
from mysql_db import connect_mysql
URL = "https://www.solitaire.org/daily-sudoku/"
DIFFICULTIES = ["easy", "medium", "hard", "expert"]
async def fetch_all_levels() -> dict:
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context(viewport={"width": 1280, "height": 900})
page = await context.new_page()
print(f"Načítám {URL} ...")
await page.goto(URL, wait_until="networkidle", timeout=60_000)
game_url = None
for frame in page.frames:
if frame.url != page.url and frame.url.strip() not in ("", "about:blank"):
game_url = frame.url
break
if not game_url:
iframe_src = await page.get_attribute("iframe", "src")
if iframe_src:
game_url = iframe_src if iframe_src.startswith("http") else f"https://www.solitaire.org{iframe_src}"
await page.close()
game_page = await context.new_page()
target_url = game_url if game_url else URL
print(f"Načítám hru: {target_url} ...")
await game_page.goto(target_url, wait_until="networkidle", timeout=60_000)
raw = await game_page.evaluate("() => JSON.stringify(gameLevels)")
await browser.close()
return json.loads(raw)
def save_to_mysql(game_levels: dict, start_date: date, end_date: date):
conn = connect_mysql(database="puzzle")
cur = conn.cursor()
inserted = 0
d = start_date
while d <= end_date:
mmdd = d.strftime("%m-%d")
date_str = d.strftime("%Y-%m-%d")
for diff in DIFFICULTIES:
if diff not in game_levels or mmdd not in game_levels[diff]:
continue
entry = game_levels[diff][mmdd]
board = entry["board"]
solution = entry["solution"]
cur.execute(
"INSERT IGNORE INTO puzzles (game_type, difficulty, puzzle_date, puzzle, solution, extra, source) "
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
("sudoku", diff, date_str, board, solution,
json.dumps({"grid_size": 9}), "solitaire.org"),
)
if cur.rowcount > 0:
inserted += 1
d += timedelta(days=1)
cur.close()
conn.close()
return inserted
async def main():
game_levels = await fetch_all_levels()
total = sum(len(game_levels.get(d, {})) for d in DIFFICULTIES)
print(f"gameLevels: {total} záznamů")
inserted = save_to_mysql(game_levels, date(2026, 1, 1), date(2026, 12, 31))
print(f"MySQL: vloženo {inserted} nových řádků (celý rok 2026)")
if __name__ == "__main__":
asyncio.run(main())