Tee scriptistä modulaarisempi
This commit is contained in:
parent
00c0fd5189
commit
6f77e4580c
1 changed files with 160 additions and 115 deletions
275
levyraati.py
275
levyraati.py
|
|
@ -1,141 +1,186 @@
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import datetime
|
||||||
|
import requests
|
||||||
import spotipy
|
import spotipy
|
||||||
from spotipy.oauth2 import SpotifyOAuth
|
from spotipy.oauth2 import SpotifyOAuth
|
||||||
import requests
|
|
||||||
import re
|
|
||||||
import os
|
|
||||||
import datetime
|
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
|
|
||||||
load_dotenv()
|
# --- KONFIGURAATIO ---
|
||||||
|
|
||||||
CLIENT_ID = os.getenv("SPOTIPY_CLIENT_ID")
|
def lataa_asetukset():
|
||||||
CLIENT_SECRET = os.getenv("SPOTIPY_CLIENT_SECRET")
|
"""Lataa ympäristömuuttujat .env-tiedostosta."""
|
||||||
REDIRECT_URI = os.getenv("SPOTIPY_REDIRECT_URI")
|
load_dotenv()
|
||||||
PLAYLIST_ID = os.getenv("PLAYLIST_ID")
|
return {
|
||||||
|
"SPOTIPY_CLIENT_ID": os.getenv("SPOTIPY_CLIENT_ID"),
|
||||||
WIKI_API_URL = os.getenv("WIKI_API_URL")
|
"SPOTIPY_CLIENT_SECRET": os.getenv("SPOTIPY_CLIENT_SECRET"),
|
||||||
WIKI_PAGE_TITLE = os.getenv("WIKI_PAGE_TITLE")
|
"SPOTIPY_REDIRECT_URI": os.getenv("SPOTIPY_REDIRECT_URI"),
|
||||||
WIKI_USERNAME = os.getenv("WIKI_USERNAME")
|
"PLAYLIST_ID": os.getenv("PLAYLIST_ID"),
|
||||||
WIKI_PASSWORD = os.getenv("WIKI_PASSWORD")
|
"WIKI_API_URL": os.getenv("WIKI_API_URL"),
|
||||||
|
"WIKI_PAGE_TITLE": os.getenv("WIKI_PAGE_TITLE"),
|
||||||
def hae_wiki_sisalto():
|
"WIKI_USERNAME": os.getenv("WIKI_USERNAME"),
|
||||||
"""Hakee MediaWiki-sivun raakatekstin APIn kautta."""
|
"WIKI_PASSWORD": os.getenv("WIKI_PASSWORD"),
|
||||||
|
|
||||||
# Käytetään Session-oliota, jotta kirjautumiskeksit (cookies) pysyvät tallessa
|
|
||||||
session = requests.Session()
|
|
||||||
|
|
||||||
# Vaihe 1: Haetaan kirjautumistoken (login token)
|
|
||||||
token_params = {
|
|
||||||
"action": "query",
|
|
||||||
"meta": "tokens",
|
|
||||||
"type": "login",
|
|
||||||
"format": "json"
|
|
||||||
}
|
}
|
||||||
|
|
||||||
try:
|
# --- WIKI-YHTEYS ---
|
||||||
response = session.get(url=WIKI_API_URL, params=token_params)
|
|
||||||
data = response.json()
|
class WikiClient:
|
||||||
login_token = data['query']['tokens']['logintoken']
|
"""Luokka MediaWiki-rajapinnan kanssa toimimiseen."""
|
||||||
|
|
||||||
# Vaihe 2: Kirjaudutaan sisään tunnuksilla ja tokenilla (POST-pyyntö)
|
def __init__(self, api_url, username, password):
|
||||||
login_params = {
|
self.api_url = api_url
|
||||||
"action": "login",
|
self.username = username
|
||||||
"lgname": WIKI_USERNAME,
|
self.password = password
|
||||||
"lgpassword": WIKI_PASSWORD,
|
self.session = requests.Session()
|
||||||
"lgtoken": login_token,
|
|
||||||
|
def kirjaudu(self):
|
||||||
|
"""Kirjautuu Wikiin ja palauttaa True jos onnistuu."""
|
||||||
|
token_params = {
|
||||||
|
"action": "query",
|
||||||
|
"meta": "tokens",
|
||||||
|
"type": "login",
|
||||||
"format": "json"
|
"format": "json"
|
||||||
}
|
}
|
||||||
|
|
||||||
login_response = session.post(url=WIKI_API_URL, data=login_params)
|
try:
|
||||||
login_data = login_response.json()
|
# Vaihe 1: Haetaan kirjautumistoken
|
||||||
|
response = self.session.get(url=self.api_url, params=token_params)
|
||||||
if login_data.get("login", {}).get("result") != "Success":
|
response.raise_for_status()
|
||||||
print(f"Wikin kirjautuminen epäonnistui: {login_data}")
|
login_token = response.json()['query']['tokens']['logintoken']
|
||||||
return ""
|
|
||||||
|
|
||||||
print("Kirjauduttu Wikiin onnistuneesti!")
|
# Vaihe 2: Kirjaudutaan sisään
|
||||||
|
login_params = {
|
||||||
# Vaihe 3: Haetaan vihdoin itse sivun sisältö
|
"action": "login",
|
||||||
page_params = {
|
"lgname": self.username,
|
||||||
|
"lgpassword": self.password,
|
||||||
|
"lgtoken": login_token,
|
||||||
|
"format": "json"
|
||||||
|
}
|
||||||
|
|
||||||
|
login_response = self.session.post(url=self.api_url, data=login_params)
|
||||||
|
login_response.raise_for_status()
|
||||||
|
|
||||||
|
if login_response.json().get("login", {}).get("result") == "Success":
|
||||||
|
print("Kirjauduttu Wikiin onnistuneesti!")
|
||||||
|
return True
|
||||||
|
|
||||||
|
print(f"Wikin kirjautuminen epäonnistui: {login_response.json()}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Virhe Wiki-kirjautumisessa: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
def hae_sivun_sisalto(self, otsikko):
|
||||||
|
"""Hakee annetun sivun raakatekstin."""
|
||||||
|
params = {
|
||||||
"action": "query",
|
"action": "query",
|
||||||
"prop": "revisions",
|
"prop": "revisions",
|
||||||
"rvprop": "content",
|
"rvprop": "content",
|
||||||
"rvslots": "main",
|
"rvslots": "main",
|
||||||
"titles": WIKI_PAGE_TITLE,
|
"titles": otsikko,
|
||||||
"format": "json"
|
"format": "json"
|
||||||
}
|
}
|
||||||
|
|
||||||
page_response = session.get(url=WIKI_API_URL, params=page_params)
|
try:
|
||||||
page_data = page_response.json()
|
response = self.session.get(url=self.api_url, params=params)
|
||||||
|
response.raise_for_status()
|
||||||
pages = page_data["query"]["pages"]
|
pages = response.json()["query"]["pages"]
|
||||||
page_id = list(pages.keys())[0]
|
page_id = list(pages.keys())[0]
|
||||||
|
|
||||||
if page_id == "-1":
|
|
||||||
print("Virhe: Wiki-sivua ei löytynyt!")
|
|
||||||
return ""
|
|
||||||
|
|
||||||
return pages[page_id]["revisions"][0]["slots"]["main"]["*"]
|
if page_id == "-1":
|
||||||
|
print(f"Virhe: Wiki-sivua '{otsikko}' ei löytynyt!")
|
||||||
except Exception as e:
|
return ""
|
||||||
print(f"Virhe Wiki-yhteydessä: {e}")
|
|
||||||
return ""
|
return pages[page_id]["revisions"][0]["slots"]["main"]["*"]
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Virhe sivun noudossa: {e}")
|
||||||
|
return ""
|
||||||
|
|
||||||
def paivita_soittolista():
|
# --- TEKSTINKÄSITTELY ---
|
||||||
# 1. Haetaan teksti Wikistä
|
|
||||||
nykyinen_viikko_int = datetime.date.today().isocalendar()[1]
|
def erista_viikon_osio(teksti, viikko_numero):
|
||||||
nykyinen_viikko = f"{nykyinen_viikko_int:02d}"
|
"""Eristää tietyn viikon osion koko sivun tekstistä (== Viikko X ==)."""
|
||||||
print(f"Kuluva viikko on {nykyinen_viikko}")
|
viikko_str = f"{viikko_numero:02d}"
|
||||||
wiki_teksti = hae_wiki_sisalto()
|
pattern = rf"==\s*Viikko {viikko_str}\s*==(.*?(?===|\Z))"
|
||||||
|
match = re.search(pattern, teksti, re.IGNORECASE | re.DOTALL)
|
||||||
|
|
||||||
|
if match:
|
||||||
|
return match.group(1).strip()
|
||||||
|
return None
|
||||||
|
|
||||||
|
def poimi_spotify_linkit(teksti):
|
||||||
|
"""Etsii tekstistä Spotify-kappaleiden linkit hakasulkeiden sisältä."""
|
||||||
|
linkit = []
|
||||||
|
for rivi in teksti.split('\n'):
|
||||||
|
# Etsitään [http...] muotoisia linkkejä
|
||||||
|
match = re.search(r'\[(https?://[^\s\]]+)', rivi)
|
||||||
|
if match:
|
||||||
|
url = match.group(1)
|
||||||
|
if "spotify.com" in url.lower():
|
||||||
|
linkit.append(url)
|
||||||
|
print(f"Löydetty Spotify-linkki: {url}")
|
||||||
|
else:
|
||||||
|
print(f"Ohitettu muu linkki: {url}")
|
||||||
|
return linkit
|
||||||
|
|
||||||
|
# --- SPOTIFY-PÄIVITYS ---
|
||||||
|
|
||||||
|
def paivita_spotify_soittolista(linkit, asetukset):
|
||||||
|
"""Päivittää Spotify-soittolistan annetuilla linkeillä."""
|
||||||
|
if not linkit:
|
||||||
|
print("\nEi päivitettäviä linkkejä. Soittolistaa ei muutettu.")
|
||||||
|
return False
|
||||||
|
|
||||||
|
scope = "playlist-modify-private playlist-modify-public"
|
||||||
|
try:
|
||||||
|
sp = spotipy.Spotify(auth_manager=SpotifyOAuth(
|
||||||
|
client_id=asetukset["SPOTIPY_CLIENT_ID"],
|
||||||
|
client_secret=asetukset["SPOTIPY_CLIENT_SECRET"],
|
||||||
|
redirect_uri=asetukset["SPOTIPY_REDIRECT_URI"],
|
||||||
|
scope=scope,
|
||||||
|
open_browser=False
|
||||||
|
))
|
||||||
|
|
||||||
|
# Spotify sallii enintään 100 kappaletta per kutsu
|
||||||
|
sp.playlist_replace_items(asetukset["PLAYLIST_ID"], linkit[:100])
|
||||||
|
print(f"Soittolista päivitetty onnistuneesti ({len(linkit[:100])} kappaletta).")
|
||||||
|
return True
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Virhe Spotifyn päivityksessä: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# --- PÄÄOHJELMA ---
|
||||||
|
|
||||||
|
def aja_paivitys():
|
||||||
|
"""Suorittaa koko päivitysprosessin vaihe kerrallaan."""
|
||||||
|
asetukset = lataa_asetukset()
|
||||||
|
nykyinen_viikko = datetime.date.today().isocalendar()[1]
|
||||||
|
print(f"--- Aloitetaan päivitys (Viikko {nykyinen_viikko:02d}) ---")
|
||||||
|
|
||||||
|
# 1. Wiki-yhteys ja sisällön haku
|
||||||
|
wiki = WikiClient(
|
||||||
|
asetukset["WIKI_API_URL"],
|
||||||
|
asetukset["WIKI_USERNAME"],
|
||||||
|
asetukset["WIKI_PASSWORD"]
|
||||||
|
)
|
||||||
|
|
||||||
|
if not wiki.kirjaudu():
|
||||||
|
return
|
||||||
|
|
||||||
|
wiki_teksti = wiki.hae_sivun_sisalto(asetukset["WIKI_PAGE_TITLE"])
|
||||||
if not wiki_teksti:
|
if not wiki_teksti:
|
||||||
return
|
return
|
||||||
|
|
||||||
# 3. Eristetään kuluvan viikon osio tekstistä säännöllisellä lausekkeella (regex)
|
# 2. Oikean viikon osion eristäminen
|
||||||
# Etsii esim. "== Viikko 10 ==" ja nappaa kaiken tekstin sen jälkeen
|
viikko_teksti = erista_viikon_osio(wiki_teksti, nykyinen_viikko)
|
||||||
# aina seuraavaan "==" merkkiin asti tai sivun loppuun.
|
if not viikko_teksti:
|
||||||
viikko_pattern = rf"==\s*Viikko {nykyinen_viikko}\s*==(.*?(?===|\Z))"
|
print(f"Huom: Viikon {nykyinen_viikko:02d} osiota ei löytynyt sivulta.")
|
||||||
match = re.search(viikko_pattern, wiki_teksti, re.IGNORECASE | re.DOTALL)
|
|
||||||
|
|
||||||
if not match:
|
|
||||||
print(f"\nWiki-sivulta ei löytynyt otsikkoa '== Viikko {nykyinen_viikko} =='. Soittolistaa ei päivitetty.")
|
|
||||||
return
|
|
||||||
|
|
||||||
kuluvan_viikon_teksti = match.group(1)
|
|
||||||
|
|
||||||
# 4. Poimitaan suorat Spotify-linkit koodista
|
|
||||||
track_urls = []
|
|
||||||
|
|
||||||
# Käydään wiki-teksti läpi rivi riviltä
|
|
||||||
for rivi in kuluvan_viikon_teksti.split('\n'):
|
|
||||||
# Etsitään vain rivit, joissa lukee if '' in rivi:
|
|
||||||
# Säännöllinen lauseke (regex), joka etsii http- tai https-alkuisen
|
|
||||||
# linkin heti avaavan hakasulkeen [ jälkeen
|
|
||||||
match = re.search(r'\[(https?://[^\s\]]+)', rivi)
|
|
||||||
|
|
||||||
if match:
|
|
||||||
url = match.group(1) # Napataan pelkkä linkki
|
|
||||||
if "spotify.com" in url.lower():
|
|
||||||
track_urls.append(url)
|
|
||||||
print(f"Löydettiin linkki: {url}")
|
|
||||||
else:
|
|
||||||
print(f"Ohitettiin muun palvelun linkki: {url}")
|
|
||||||
|
|
||||||
# 5. Tarkistetaan löytyikö mitään
|
|
||||||
if not track_urls:
|
|
||||||
print("\nYhtään Spotify-linkkiä ei löytynyt Wiki-sivulta. Soittolistaa ei päivitetty.")
|
|
||||||
return
|
return
|
||||||
|
|
||||||
# 6. Kirjaudutaan sisään
|
# 3. Linkkien kerääminen ja Spotifyn päivittäminen
|
||||||
scope = "playlist-modify-private playlist-modify-public"
|
spotify_linkit = poimi_spotify_linkit(viikko_teksti)
|
||||||
sp = spotipy.Spotify(auth_manager=SpotifyOAuth(client_id=CLIENT_ID,
|
paivita_spotify_soittolista(spotify_linkit, asetukset)
|
||||||
client_secret=CLIENT_SECRET,
|
|
||||||
redirect_uri=REDIRECT_URI,
|
|
||||||
scope=scope, open_browser=False))
|
|
||||||
|
|
||||||
# 7. Päivitetään soittolista
|
|
||||||
sp.playlist_replace_items(PLAYLIST_ID, track_urls[:100])
|
|
||||||
print("Soittolista päivitetty onnistuneesti!")
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
paivita_soittolista()
|
aja_paivitys()
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue