285 lines
10 KiB
Python
285 lines
10 KiB
Python
import os
|
|
import re
|
|
import datetime
|
|
import requests
|
|
import spotipy
|
|
import sys
|
|
from spotipy.oauth2 import SpotifyOAuth
|
|
from dotenv import load_dotenv
|
|
from flask import Flask, request, jsonify
|
|
|
|
# --- KONFIGURAATIO ---
|
|
|
|
# Määritetään .cache-tiedoston polku aina skriptin omaan kansioon
|
|
CACHE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".cache")
|
|
|
|
def lataa_asetukset():
|
|
"""Lataa ympäristömuuttujat .env-tiedostosta."""
|
|
load_dotenv()
|
|
return {
|
|
"SPOTIPY_CLIENT_ID": os.getenv("SPOTIPY_CLIENT_ID"),
|
|
"SPOTIPY_CLIENT_SECRET": os.getenv("SPOTIPY_CLIENT_SECRET"),
|
|
"SPOTIPY_REDIRECT_URI": os.getenv("SPOTIPY_REDIRECT_URI"),
|
|
"PLAYLIST_ID": os.getenv("PLAYLIST_ID"),
|
|
"WIKI_API_URL": os.getenv("WIKI_API_URL"),
|
|
"WIKI_PAGE_TITLE": os.getenv("WIKI_PAGE_TITLE"),
|
|
"WIKI_USERNAME": os.getenv("WIKI_USERNAME"),
|
|
"WIKI_PASSWORD": os.getenv("WIKI_PASSWORD"),
|
|
"WEBHOOK_SECRET": os.getenv("WEBHOOK_SECRET", "salainen_avain"), # Lisätty turva
|
|
}
|
|
|
|
# --- WIKI-YHTEYS ---
|
|
|
|
class WikiClient:
|
|
"""Luokka MediaWiki-rajapinnan kanssa toimimiseen."""
|
|
|
|
def __init__(self, api_url, username, password):
|
|
self.api_url = api_url
|
|
self.username = username
|
|
self.password = password
|
|
self.session = requests.Session()
|
|
|
|
def kirjaudu(self):
|
|
"""Kirjautuu Wikiin ja palauttaa True jos onnistuu."""
|
|
token_params = {
|
|
"action": "query",
|
|
"meta": "tokens",
|
|
"type": "login",
|
|
"format": "json"
|
|
}
|
|
|
|
try:
|
|
# Vaihe 1: Haetaan kirjautumistoken
|
|
response = self.session.get(url=self.api_url, params=token_params)
|
|
response.raise_for_status()
|
|
login_token = response.json()['query']['tokens']['logintoken']
|
|
|
|
# Vaihe 2: Kirjaudutaan sisään
|
|
login_params = {
|
|
"action": "login",
|
|
"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",
|
|
"prop": "revisions",
|
|
"rvprop": "content",
|
|
"rvslots": "main",
|
|
"titles": otsikko,
|
|
"format": "json"
|
|
}
|
|
|
|
try:
|
|
response = self.session.get(url=self.api_url, params=params)
|
|
response.raise_for_status()
|
|
pages = response.json()["query"]["pages"]
|
|
page_id = list(pages.keys())[0]
|
|
|
|
if page_id == "-1":
|
|
print(f"Virhe: Wiki-sivua '{otsikko}' ei löytynyt!")
|
|
return ""
|
|
|
|
return pages[page_id]["revisions"][0]["slots"]["main"]["*"]
|
|
except Exception as e:
|
|
print(f"Virhe sivun noudossa: {e}")
|
|
return ""
|
|
|
|
# --- TEKSTINKÄSITTELY ---
|
|
|
|
def erista_viikon_osio(teksti, viikko_numero):
|
|
"""Eristää tietyn viikon osion koko sivun tekstistä (== Viikko X ==)."""
|
|
viikko_str = f"{viikko_numero:02d}"
|
|
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, "Ei linkkejä"
|
|
|
|
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,
|
|
cache_path=CACHE_PATH,
|
|
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, f"Päivitetty {len(linkit[:100])} kappaletta"
|
|
except Exception as e:
|
|
err_msg = str(e)
|
|
if "invalid_grant" in err_msg or "Refresh token revoked" in err_msg:
|
|
print("\n[!] Spotifyn luvitus on vanhentunut tai mitätöity (Refresh token revoked).")
|
|
print(" Suorita uudelleenluvitus ajamalla terminaalissa:")
|
|
print(" source .venv/bin/activate && python levyraati.py --auth\n")
|
|
else:
|
|
print(f"Virhe Spotifyn päivityksessä: {e}")
|
|
return False, err_msg
|
|
|
|
def suorita_luvitus(asetukset):
|
|
"""Suorittaa Spotifyn OAuth-luvituksen interaktiivisesti terminaalissa."""
|
|
scope = "playlist-modify-private playlist-modify-public"
|
|
auth_manager = SpotifyOAuth(
|
|
client_id=asetukset["SPOTIPY_CLIENT_ID"],
|
|
client_secret=asetukset["SPOTIPY_CLIENT_SECRET"],
|
|
redirect_uri=asetukset["SPOTIPY_REDIRECT_URI"],
|
|
scope=scope,
|
|
cache_path=CACHE_PATH,
|
|
open_browser=False
|
|
)
|
|
|
|
# Poistetaan aiempi vanhentunut cache-tiedosto
|
|
if os.path.exists(CACHE_PATH):
|
|
try:
|
|
os.remove(CACHE_PATH)
|
|
print(f"Poistettu vanha luvitustiedosto ({CACHE_PATH}).")
|
|
except OSError as e:
|
|
print(f"Huom: vanhan luvitustiedoston poisto epäonnistui: {e}")
|
|
|
|
auth_url = auth_manager.get_authorize_url()
|
|
print("\n=================== SPOTIFYN LUVITUS ===================")
|
|
print("1. Avaa seuraava osoite selaimessasi:\n")
|
|
print(f" {auth_url}\n")
|
|
print("2. Kirjaudu Spotifyhin ja hyväksy sovelluksen pyytämät oikeudet.")
|
|
print("3. Selain ohjautuu osoitteeseen:")
|
|
print(f" {asetukset['SPOTIPY_REDIRECT_URI']}?code=...")
|
|
print(" (Vaikka selain näyttäisi virhesivua kuten 404, osoite on oikea!)")
|
|
print("4. Kopioi selaimen osoiteriviltä koko URL ja liitä se alle.")
|
|
print("========================================================\n")
|
|
|
|
try:
|
|
uudelleenohjattu_url = input("Liitä uudelleenohjattu URL tähän: ").strip()
|
|
if not uudelleenohjattu_url:
|
|
print("Virhe: URL oli tyhjä.")
|
|
return False
|
|
|
|
code = auth_manager.parse_response_code(uudelleenohjattu_url)
|
|
auth_manager.get_access_token(code, as_dict=False, check_cache=False)
|
|
print(f"\n[+] Token tallennettu onnistuneesti tiedostoon {CACHE_PATH}!")
|
|
|
|
# Testataan yhteys
|
|
sp = spotipy.Spotify(auth_manager=auth_manager)
|
|
kayttaja = sp.me()
|
|
print(f"[+] Spotify-yhteys toimii! Kirjautunut käyttäjä: {kayttaja.get('display_name')} ({kayttaja.get('id')})")
|
|
|
|
if asetukset.get("PLAYLIST_ID"):
|
|
soittolista = sp.playlist(asetukset["PLAYLIST_ID"])
|
|
print(f"[+] Pääsy soittolistaan kunnossa: '{soittolista.get('name')}'")
|
|
|
|
print("\nValmista! Taustapalvelin ja cron voivat nyt päivittää soittolistaa.")
|
|
return True
|
|
except Exception as e:
|
|
print(f"\n[-] Virhe luvituksessa: {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 False, "Kirjautuminen epäonnistui"
|
|
|
|
wiki_teksti = wiki.hae_sivun_sisalto(asetukset["WIKI_PAGE_TITLE"])
|
|
if not wiki_teksti:
|
|
return False, "Sivun haku epäonnistui"
|
|
|
|
# 2. Oikean viikon osion eristäminen
|
|
viikko_teksti = erista_viikon_osio(wiki_teksti, nykyinen_viikko)
|
|
if not viikko_teksti:
|
|
print(f"Huom: Viikon {nykyinen_viikko:02d} osiota ei löytynyt sivulta.")
|
|
return False, f"Viikon {nykyinen_viikko} osiota ei löytynyt"
|
|
|
|
# 3. Linkkien kerääminen ja Spotifyn päivittäminen
|
|
spotify_linkit = poimi_spotify_linkit(viikko_teksti)
|
|
success, msg = paivita_spotify_soittolista(spotify_linkit, asetukset)
|
|
return success, msg
|
|
|
|
# --- FLASK PALVELIN ---
|
|
|
|
app = Flask(__name__)
|
|
|
|
@app.route('/webhook', methods=['POST'])
|
|
def webhook():
|
|
asetukset = lataa_asetukset()
|
|
|
|
# Yksinkertainen turvatarkistus: vaaditaan salainen avain otsakkeessa tai parametrina
|
|
secret = request.args.get('secret') or request.headers.get('X-Webhook-Secret')
|
|
if secret != asetukset["WEBHOOK_SECRET"]:
|
|
return jsonify({"status": "error", "message": "Invalid secret"}), 403
|
|
|
|
print("Webhook vastaanotettu! Käynnistetään päivitys...")
|
|
success, msg = aja_paivitys()
|
|
|
|
if success:
|
|
return jsonify({"status": "success", "message": msg}), 200
|
|
else:
|
|
return jsonify({"status": "error", "message": msg}), 500
|
|
|
|
if __name__ == '__main__':
|
|
if len(sys.argv) > 1 and sys.argv[1] == '--server':
|
|
print("Käynnistetään Webhook-palvelin portissa 5000...")
|
|
# Huom: Tuotannossa käytä WSGI-palvelinta (esim. gunicorn)
|
|
app.run(host='0.0.0.0', port=5000)
|
|
elif len(sys.argv) > 1 and sys.argv[1] in ('--auth', '--reauth'):
|
|
asetukset = lataa_asetukset()
|
|
suorita_luvitus(asetukset)
|
|
else:
|
|
aja_paivitys()
|