552 lines
16 KiB
Python
Executable file
552 lines
16 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
import argparse
|
|
import configparser
|
|
import dataclasses
|
|
import enum
|
|
import os
|
|
import select
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
import xml.etree.ElementTree
|
|
|
|
if os.name != 'nt':
|
|
import termios
|
|
else:
|
|
import msvcrt
|
|
|
|
@dataclasses.dataclass
|
|
class Metadata:
|
|
lähde: str
|
|
edellinen: int | None
|
|
seuraava: int | None
|
|
alasivuja: int
|
|
|
|
def asetustiedosto():
|
|
if os.name == 'nt':
|
|
asetuskansio = os.path.join(os.environ['APPDATA'], 'pytekstitv')
|
|
else:
|
|
asetuskansio = os.path.expanduser('~/.config')
|
|
if 'XDG_CONFIG_HOME' in os.environ:
|
|
asetuskansio = os.environ['XDG_CONFIG_HOME']
|
|
return os.path.join(asetuskansio, 'pytekstitv.conf')
|
|
|
|
def apikutsu(apiavain, sivu):
|
|
url = f'https://external.api.yle.fi/v1/teletext/pages/{sivu}.xml?{apiavain}'
|
|
with urllib.request.urlopen(url) as f:
|
|
return xml.etree.ElementTree.parse(f)
|
|
|
|
class Signaali:
|
|
def __init__(self):
|
|
self.kirjoituspää, self.lukupää = socket.socketpair()
|
|
|
|
def lähetä(self):
|
|
self.kirjoituspää.send(b'.')
|
|
|
|
def kuittaa(self):
|
|
self.lukupää.recv(1)
|
|
|
|
def odota(self):
|
|
select.select([self.lukupää], [], []) # Odota kunnes dataa on luettavaksi
|
|
self.kuittaa()
|
|
|
|
def fileno(self):
|
|
return self.lukupää.fileno()
|
|
|
|
class Jono:
|
|
def __init__(self):
|
|
self.jono_lukko = threading.Lock()
|
|
self.jono = []
|
|
self.signaali = Signaali()
|
|
|
|
def lähetä(self, viesti):
|
|
with self.jono_lukko:
|
|
self.jono.append(viesti)
|
|
self.signaali.lähetä()
|
|
|
|
def vastaanota(self):
|
|
self.signaali.odota()
|
|
with self.jono_lukko:
|
|
return self.jono.pop(0)
|
|
|
|
def tyhjä(self):
|
|
luettava, _, _ = select.select([self.signaali], [], [], 0)
|
|
return len(luettava) == 0
|
|
|
|
def fileno(self):
|
|
return self.signaali.fileno()
|
|
|
|
class sivustatus(enum.Enum):
|
|
puuttuu, ei_sivua, epäonnistui, ok = range(4)
|
|
|
|
class LataajaSäie(threading.Thread):
|
|
def __init__(self, apiavain, välimuisti):
|
|
threading.Thread.__init__(self)
|
|
self.apiavain = apiavain
|
|
self.välimuisti = välimuisti
|
|
self.liian_tiuhaan = 0
|
|
|
|
def lataa(self, sivu):
|
|
if self.välimuisti.status(sivu) == sivustatus.ok:
|
|
return
|
|
try:
|
|
puu = apikutsu(self.apiavain, sivu)
|
|
self.välimuisti.lisää(sivu, prosessoi_sivu(puu))
|
|
self.liian_tiuhaan = 0
|
|
except urllib.error.HTTPError as virhe:
|
|
if virhe.code == 401:
|
|
self.liian_tiuhaan += 1
|
|
time.sleep(2 ** self.liian_tiuhaan)
|
|
if self.liian_tiuhaan > 4:
|
|
self.välimuisti.lisää(sivu, sivustatus.epäonnistui)
|
|
self.liian_tiuhaan = 0
|
|
elif virhe.code == 404:
|
|
self.välimuisti.lisää(sivu, sivustatus.ei_sivua)
|
|
self.liian_tiuhaan = 0
|
|
|
|
def run(self):
|
|
taustalla = []
|
|
while True:
|
|
while self.välimuisti.pyynnöt.tyhjä() and len(taustalla) > 0:
|
|
self.lataa(taustalla.pop())
|
|
|
|
match self.välimuisti.pyynnöt.vastaanota():
|
|
case None:
|
|
break
|
|
case [sivu, signaali]:
|
|
self.lataa(sivu)
|
|
signaali.lähetä()
|
|
taustalla = []
|
|
case sivu:
|
|
taustalla.append(sivu)
|
|
|
|
class Välimuisti:
|
|
def __init__(self, apiavain):
|
|
self.sivut_lukko = threading.Lock()
|
|
self.sivut = {}
|
|
self.pyynnöt = Jono()
|
|
LataajaSäie(apiavain, self).start()
|
|
|
|
def lisää(self, sivu, data):
|
|
with self.sivut_lukko:
|
|
self.sivut[sivu] = data
|
|
|
|
def status(self, sivu):
|
|
with self.sivut_lukko:
|
|
if sivu not in self.sivut:
|
|
return sivustatus.puuttuu
|
|
elif isinstance(self.sivut[sivu], sivustatus):
|
|
return self.sivut[sivu]
|
|
else:
|
|
return sivustatus.ok
|
|
|
|
def lataa(self, sivu, signaali):
|
|
self.pyynnöt.lähetä((sivu, signaali))
|
|
|
|
def linkki(self, sivu):
|
|
if sivu is not None and self.status(sivu) != sivustatus.ok:
|
|
self.pyynnöt.lähetä(sivu)
|
|
|
|
def hae(self, sivu):
|
|
metadata, alasivut = self.sivut[sivu]
|
|
return metadata, alasivut
|
|
|
|
# with-lauseketta varten
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, _1, _2, _3):
|
|
self.pyynnöt.lähetä(None)
|
|
# Heitä poikkeus uudestaan
|
|
return False
|
|
|
|
värit = {
|
|
'black': 30,
|
|
'red': 31,
|
|
'green': 32,
|
|
'yellow': 93,
|
|
'blue': 34,
|
|
'magenta': 95,
|
|
'cyan': 96,
|
|
'white': 97,
|
|
}
|
|
|
|
def grafiikkamerkki(arvo):
|
|
assert 0x20 <= arvo <= 0x7f
|
|
if arvo == 0x20: return '\u00a0' # No-Break Space
|
|
elif arvo < 0x35: return chr(0x1fb00 - 1 - 0x20 + arvo)
|
|
elif arvo == 0x35: return '\u258c' # Left Half Block
|
|
elif arvo <= 0x3f: return chr(0x1fb00 - 2 - 0x20 + arvo)
|
|
elif arvo < 0x60: return chr(arvo)
|
|
elif arvo < 0x6a: return chr(0x1fb00 - 2 - 0x40 + arvo)
|
|
elif arvo == 0x6a: return '\u2590' # Right Half Block
|
|
elif arvo < 0x7f: return chr(0x1fb00 - 3 - 0x40 + arvo)
|
|
elif arvo == 0x7f: return '\u2588' # Full Block
|
|
|
|
def ohjauskoodit(jakso):
|
|
grafiikka = False
|
|
väri = värit.get(jakso.get('fg'))
|
|
if väri == None:
|
|
assert jakso.get('fg')[0] == 'g'
|
|
väri = värit[jakso.get('fg')[1:]]
|
|
grafiikka = True
|
|
tausta = värit[jakso.get('bg')] + 10
|
|
teksti = jakso.text
|
|
if grafiikka:
|
|
teksti = ''.join(grafiikkamerkki(ord(merkki)) for merkki in teksti)
|
|
return f'\x1b[{väri};{tausta}m{teksti}'
|
|
|
|
def prosessoi_sivu(puu):
|
|
sivu = puu.getroot().find('page')
|
|
metadata = Metadata(
|
|
lähde = puu.getroot().get('network'),
|
|
edellinen = int(sivu.get('prevpg')) if sivu.get('prevpg') is not None else None,
|
|
seuraava = int(sivu.get('nextpg')) if sivu.get('nextpg') is not None else None,
|
|
alasivuja = int(sivu.get('subpagecount')),
|
|
)
|
|
|
|
alasivut = []
|
|
for alasivunumero in range(1, metadata.alasivuja + 1):
|
|
alasivu = []
|
|
for rivinumero in range(1, 24 + 1):
|
|
rivi = sivu.find(f'subpage[@number="{alasivunumero}"]/content[@type="structured"]/line[@number="{rivinumero}"]')
|
|
jaksot = [ohjauskoodit(jakso) for jakso in rivi]
|
|
alasivu.append(''.join(jaksot))
|
|
alasivut.append(alasivu)
|
|
return metadata, alasivut
|
|
|
|
def luo_käyttöliittymä(metadata, sivunumero, alasivunumero, sivunumero_syöte):
|
|
rivit = []
|
|
if len(sivunumero_syöte) == 0:
|
|
rivit.append(f'Sivu {sivunumero}')
|
|
else:
|
|
rivit.append(f'Sivu {sivunumero_syöte}{"_"*(3 - len(sivunumero_syöte))}')
|
|
if metadata.alasivuja != 1:
|
|
rivit.append(f'Alasivu {alasivunumero}/{metadata.alasivuja}')
|
|
else:
|
|
rivit.append('')
|
|
edellinen = metadata.edellinen if metadata.edellinen else ' '
|
|
seuraava = metadata.seuraava if metadata.seuraava else ' '
|
|
rivit.append(f'<<< {edellinen} {seuraava} >>>')
|
|
rivit.append('')
|
|
rivit.append(f'Sisällön lähde: {metadata.lähde}')
|
|
return rivit
|
|
|
|
def näytä_alasivu(metadata, alasivut, sivunumero, alasivunumero, sivunumero_syöte):
|
|
print('\x1b[1;1H\x1b[2J', end='') # Kursori yläkulmaan, tyhjennä näyttö
|
|
alasivu = alasivut[alasivunumero-1]
|
|
käyttöliittymä = luo_käyttöliittymä(metadata, sivunumero, alasivunumero, sivunumero_syöte)
|
|
korkeus = max(len(alasivu), len(käyttöliittymä))
|
|
for rivi in range(korkeus):
|
|
sisältö_rivi = alasivu[rivi] if rivi < len(alasivu) else ' '*40
|
|
käyttöliittymä_rivi = käyttöliittymä[rivi] if rivi < len(käyttöliittymä) else ''
|
|
print(sisältö_rivi, end='')
|
|
print('\x1b[0m ', end='')
|
|
print(käyttöliittymä_rivi, end='')
|
|
print('' if rivi == 23 else '\n', end='', flush = True)
|
|
|
|
class Sekvensoija:
|
|
def __init__(self):
|
|
self.puskuri = []
|
|
|
|
def syötä(self, merkki):
|
|
self.puskuri.append(merkki)
|
|
return self.käsittele()
|
|
|
|
def tyhjennä(self):
|
|
syöte = b''.join(self.puskuri)
|
|
self.puskuri = []
|
|
return syöte
|
|
|
|
class ANSISekvensoija(Sekvensoija):
|
|
def käsittele(self):
|
|
if self.puskuri[0] != b'\x1b':
|
|
# Yksittäinen merkki
|
|
return self.tyhjennä()
|
|
|
|
if len(self.puskuri) > 1 and self.puskuri[1] != b'[':
|
|
# Alt+näppäin, <esc><merkki>
|
|
return self.tyhjennä()
|
|
|
|
sarja1_alku = 2
|
|
# Oletus, että näppäinkoodi ei voi olla yli kolminumeroinen
|
|
sarja1_loppu = self.numerosarja(sarja1_alku, 3)
|
|
|
|
if len(self.puskuri) > sarja1_loppu and self.puskuri[sarja1_loppu] != b';':
|
|
# Erikoisnäppäin, <esc>[<muokkausnäppäimet><merkki>
|
|
return self.tyhjennä()
|
|
|
|
sarja2_alku = sarja1_loppu + 1
|
|
# Muokkausnäppäinkentän arvo voi olla maksimissaan 16
|
|
sarja2_loppu = self.numerosarja(sarja2_alku, 2)
|
|
|
|
if len(self.puskuri) > sarja2_loppu:
|
|
# Erikoisnäppäin, <esc>[<näppäinkoodi>;<muokkausnäppäimet>~
|
|
# Vaihtoehtoisesti virheellinen sarja, <esc>[<numeroita>;<numeroita><ei-tilde>
|
|
return self.tyhjennä()
|
|
|
|
return b''
|
|
|
|
def numerosarja(self, alku, maksimipituus):
|
|
pituus = min(maksimipituus, len(self.puskuri) - alku)
|
|
while pituus > 0:
|
|
if all(ord('0') <= tavu <= ord('9') for (tavu,) in self.puskuri[alku:alku + pituus]):
|
|
return alku + pituus
|
|
pituus -= 1
|
|
return alku
|
|
|
|
class WindowsSekvensoija(Sekvensoija):
|
|
def käsittele(self):
|
|
if self.puskuri[0] != b'\xe0':
|
|
# Yksittäinen merkki
|
|
return self.tyhjennä()
|
|
|
|
if len(self.puskuri) == 2:
|
|
# Erikoisnäppäin, \xe0 ja näppäinkoodi
|
|
return self.tyhjennä()
|
|
|
|
return b''
|
|
|
|
class toimet(enum.Enum):
|
|
kumita, poistu, seuraava, edellinen, seuraava_sivu, edellinen_sivu, seuraava_alasivu, edellinen_alasivu, muu = range(9)
|
|
|
|
näppäinsarjat = {
|
|
b'\x7f': toimet.kumita, # Askelpalautin *nix:illä
|
|
b'\x08': toimet.kumita, # Askelpalautin Windowsilla
|
|
|
|
b'q': toimet.poistu,
|
|
b'\x03': toimet.poistu, # Ctrl+C
|
|
|
|
b' ': toimet.seuraava,
|
|
b'f': toimet.seuraava,
|
|
b'\x1b[6~': toimet.seuraava, # Page Down *nix:illä
|
|
b'\xe0Q': toimet.seuraava, # Page Down Windowsilla
|
|
|
|
b'b': toimet.edellinen,
|
|
b'\x1b[5~': toimet.edellinen, # Page Up *nix:illä
|
|
b'\xe0I': toimet.edellinen, # Page Up Windowsilla
|
|
|
|
b'+': toimet.seuraava_alasivu,
|
|
b'\x1b[B': toimet.seuraava_alasivu, # Nuoli alas *nix:illä
|
|
b'\xe0P': toimet.seuraava_alasivu, # Nuoli alas Windowsilla
|
|
|
|
b'-': toimet.edellinen_alasivu,
|
|
b'\x1b[A': toimet.edellinen_alasivu, # Nuoli ylös *nix:illä
|
|
b'\xe0H': toimet.edellinen_alasivu, # Nuoli ylös Windowsilla
|
|
|
|
b'n': toimet.seuraava_sivu,
|
|
b'\x1b[C': toimet.seuraava_sivu, # Nuoli oikealle *nix:illä
|
|
b'\xe0M': toimet.seuraava_sivu, # Nuoli oikealle Windowsilla
|
|
|
|
b'p': toimet.edellinen_sivu,
|
|
b'\x1b[D': toimet.edellinen_sivu, # Nuoli vasemmalle *nix:illä
|
|
b'\xe0K': toimet.edellinen_sivu, # Nuoli vasemmalle Windowsilla
|
|
}
|
|
|
|
class SyöteSäie(threading.Thread):
|
|
def __init__(self, syötteet):
|
|
threading.Thread.__init__(self)
|
|
self.syötteet = syötteet
|
|
|
|
def run(self):
|
|
try:
|
|
if os.name != 'nt':
|
|
sekvensoija = ANSISekvensoija()
|
|
else:
|
|
sekvensoija = WindowsSekvensoija()
|
|
|
|
while True:
|
|
if os.name != 'nt':
|
|
näppäin = sys.stdin.buffer.raw.read(1)
|
|
else:
|
|
näppäin = msvcrt.getch()
|
|
|
|
syöte = sekvensoija.syötä(näppäin)
|
|
if syöte == b'': continue
|
|
|
|
if syöte in näppäinsarjat:
|
|
toimi = näppäinsarjat[syöte]
|
|
elif syöte in b'0123456789':
|
|
toimi = syöte.decode()
|
|
else:
|
|
toimi = toimet.muu
|
|
|
|
# Tämä on ennen toimen lähetystä, ettei poistu-toimea
|
|
# lähetettäisi kahdesti (tässä ja finally-blokissa)
|
|
if toimi == toimet.poistu:
|
|
break
|
|
|
|
self.syötteet.lähetä(toimi)
|
|
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
finally:
|
|
self.syötteet.lähetä(toimet.poistu)
|
|
|
|
class päivitys(enum.Enum):
|
|
ei, piirrä, lataa = range(3)
|
|
|
|
def pääsilmukka(syöte, välimuisti, sivu_ladattu, sivunumero, alasivunumero = 1):
|
|
käynnissä = True
|
|
päivitä = päivitys.lataa
|
|
|
|
edellinen_sivunumero = None
|
|
sivunumero_syöte = ''
|
|
while käynnissä:
|
|
if päivitä == päivitys.lataa:
|
|
status = välimuisti.status(sivunumero)
|
|
if status == sivustatus.ok:
|
|
metadata, alasivut = välimuisti.hae(sivunumero)
|
|
välimuisti.linkki(metadata.edellinen)
|
|
välimuisti.linkki(metadata.seuraava)
|
|
päivitä = päivitys.piirrä
|
|
elif status == sivustatus.puuttuu:
|
|
välimuisti.lataa(sivunumero, sivu_ladattu)
|
|
päivitä = päivitys.ei
|
|
else:
|
|
sivunumero = edellinen_sivunumero
|
|
päivitä = päivitys.piirrä
|
|
|
|
if päivitä == päivitys.piirrä:
|
|
if alasivunumero < 0:
|
|
alasivunumero = metadata.alasivuja + 1 + alasivunumero
|
|
näytä_alasivu(metadata, alasivut, sivunumero, alasivunumero, sivunumero_syöte)
|
|
päivitä = päivitys.ei
|
|
|
|
luettava, _, _ = select.select([syöte.fileno(), sivu_ladattu.fileno()], [], [])
|
|
|
|
for fd in luettava:
|
|
if fd == sivu_ladattu.fileno():
|
|
sivu_ladattu.kuittaa()
|
|
päivitä = päivitys.lataa
|
|
continue
|
|
|
|
assert fd == syöte.fileno()
|
|
match syöte.vastaanota():
|
|
case ('0' | '1' | '2' | '3' | '4' | '5' | '6' | '7' | '8' | '9') as numero:
|
|
sivunumero_syöte += numero
|
|
päivitä = päivitys.piirrä
|
|
if len(sivunumero_syöte) == 3:
|
|
edellinen_sivunumero = sivunumero
|
|
sivunumero = int(sivunumero_syöte)
|
|
alasivunumero = 1
|
|
sivunumero_syöte = ''
|
|
päivitä = päivitys.lataa
|
|
|
|
case toimet.kumita if len(sivunumero_syöte) > 0:
|
|
sivunumero_syöte = sivunumero_syöte[:-1]
|
|
päivitä = päivitys.piirrä
|
|
|
|
case _ if len(sivunumero_syöte) > 0:
|
|
sivunumero_syöte = ''
|
|
päivitä = päivitys.piirrä
|
|
|
|
case toimet.poistu:
|
|
käynnissä = False
|
|
|
|
case toimet.seuraava_alasivu if alasivunumero < metadata.alasivuja:
|
|
alasivunumero += 1
|
|
päivitä = päivitys.piirrä
|
|
|
|
case toimet.edellinen_alasivu if alasivunumero > 1:
|
|
alasivunumero -= 1
|
|
päivitä = päivitys.piirrä
|
|
|
|
case toimet.seuraava_sivu if metadata.seuraava is not None:
|
|
edellinen_sivunumero = sivunumero
|
|
sivunumero = metadata.seuraava
|
|
alasivunumero = 1
|
|
päivitä = päivitys.lataa
|
|
|
|
case toimet.edellinen_sivu if metadata.edellinen is not None:
|
|
edellinen_sivunumero = sivunumero
|
|
sivunumero = metadata.edellinen
|
|
alasivunumero = 1
|
|
päivitä = päivitys.lataa
|
|
|
|
case toimet.seuraava if alasivunumero < metadata.alasivuja:
|
|
alasivunumero += 1
|
|
päivitä = päivitys.piirrä
|
|
case toimet.seuraava if metadata.seuraava is not None:
|
|
edellinen_sivunumero = sivunumero
|
|
sivunumero = metadata.seuraava
|
|
alasivunumero = 1
|
|
päivitä = päivitys.lataa
|
|
|
|
case toimet.edellinen if alasivunumero > 1:
|
|
alasivunumero -= 1
|
|
päivitä = päivitys.piirrä
|
|
case toimet.edellinen if metadata.edellinen is not None:
|
|
edellinen_sivunumero = sivunumero
|
|
sivunumero = metadata.edellinen
|
|
alasivunumero = -1
|
|
päivitä = päivitys.lataa
|
|
|
|
def pää():
|
|
parseri = argparse.ArgumentParser()
|
|
parseri.add_argument('sivu', nargs='?')
|
|
sivunumero = parseri.parse_args().sivu
|
|
|
|
asetukset = configparser.ConfigParser()
|
|
asetukset.read(asetustiedosto())
|
|
|
|
if 'yle' not in asetukset or 'apiavain' not in asetukset['yle']:
|
|
print('Virhe: API-avain puuttuu.\n', file=sys.stderr)
|
|
print('1. Hanki API-avain osoitteesta https://tunnus.yle.fi/api-avaimet', file=sys.stderr)
|
|
print(f'2. Lisää tiedostoon {asetustiedosto()} osio:', file=sys.stderr)
|
|
print('[yle]', file=sys.stderr)
|
|
print('apiavain=app_id=…&app_key=…', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if sivunumero == None:
|
|
sivunumero = 100
|
|
else:
|
|
try:
|
|
sivunumero = int(sivunumero)
|
|
except ValueError:
|
|
print('Virhe: virheellinen sivunumero', file=sys.stderr)
|
|
sys.exit(1)
|
|
if sivunumero < 100 or 899 < sivunumero:
|
|
print('Virhe: sivunumeron tulee olla väliltä 100-899', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
with Välimuisti(asetukset['yle']['apiavain']) as välimuisti:
|
|
sivu_ladattu = Signaali()
|
|
välimuisti.lataa(sivunumero, sivu_ladattu)
|
|
sivu_ladattu.odota()
|
|
if välimuisti.status(sivunumero) == sivustatus.ei_sivua:
|
|
print(f'Virhe: Sivua {sivunumero} ei löydy', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
print('\x1b[?1049h', end='') # Vaihtoehtoinen näyttöpuskuri
|
|
print('\x1b[?25l', end='') # Piilota kursori
|
|
|
|
syötteet = Jono()
|
|
SyöteSäie(syötteet).start()
|
|
|
|
try:
|
|
if os.name != 'nt':
|
|
pääte = sys.stdout.fileno()
|
|
vanhat_attribuutit = termios.tcgetattr(pääte)
|
|
iflag, oflag, cflag, lflag, ispeed, ospeed, cc = vanhat_attribuutit
|
|
lflag &= ~termios.ECHO
|
|
lflag &= ~termios.ICANON
|
|
uudet_attribuutit = [iflag, oflag, cflag, lflag, ispeed, ospeed, cc]
|
|
termios.tcsetattr(pääte, termios.TCSADRAIN, uudet_attribuutit)
|
|
|
|
pääsilmukka(syötteet, välimuisti, sivu_ladattu, sivunumero)
|
|
|
|
finally:
|
|
if os.name != 'nt':
|
|
termios.tcsetattr(pääte, termios.TCSADRAIN, vanhat_attribuutit)
|
|
|
|
finally:
|
|
print('\x1b[?25h', end='') # Palauta kursori näkyväksi
|
|
print('\x1b[?1049l', end='', flush=True) # Palauta normaali näyttöpuskuri
|
|
|
|
if __name__ == '__main__':
|
|
pää()
|