723 lines
26 KiB
Python
723 lines
26 KiB
Python
"""Async client for the Librus Synergia portal.
|
|
|
|
Adapted from dani3l0/librusik (lib/api/librus.py, stable branch):
|
|
https://github.com/dani3l0/librusik
|
|
|
|
The OAuth flow, API host and scraping logic follow that implementation.
|
|
Changes here are limited to defensive parsing, week selection for the
|
|
timetable and a few helpers the web frontend needs.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from http.cookies import SimpleCookie
|
|
|
|
import aiohttp
|
|
import yarl
|
|
from bs4 import BeautifulSoup
|
|
|
|
BROWSER_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:130.0) "
|
|
"Gecko/20100101 Firefox/130.0"
|
|
)
|
|
}
|
|
SYNERGIA_URL = yarl.URL("https://synergia.librus.pl/")
|
|
OAUTH_CLIENT_ID = 46
|
|
|
|
DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
|
|
|
|
|
|
class LibrusError(Exception):
|
|
"""Raised when Synergia returns an unusable response."""
|
|
|
|
|
|
class Librus:
|
|
def __init__(self, session=None, request_timeout=8, retries=3):
|
|
self.host = "https://synergia.librus.pl/gateway/api/2.0/"
|
|
self.msg_host = "https://synergia.librus.pl"
|
|
self.headers = session
|
|
self.request_timeout = request_timeout
|
|
self.retries = retries
|
|
# Set by mktoken(): True only when Librus itself rejected the
|
|
# credentials, as opposed to a network error or an outage page.
|
|
self.credentials_rejected = False
|
|
|
|
# ------------------------------------------------------------------ auth
|
|
|
|
def client(self):
|
|
# The cookie jar is pinned to HTTPS librus.pl so session cookies are
|
|
# never leaked to another host through a redirect.
|
|
session = aiohttp.ClientSession(
|
|
headers=BROWSER_HEADERS,
|
|
timeout=aiohttp.ClientTimeout(total=self.request_timeout),
|
|
)
|
|
if self.headers:
|
|
cookies = SimpleCookie()
|
|
for name, value in self.headers.items():
|
|
cookies[name] = value
|
|
cookies[name]["domain"] = "librus.pl"
|
|
cookies[name]["secure"] = True
|
|
session.cookie_jar.update_cookies(cookies, SYNERGIA_URL)
|
|
return session
|
|
|
|
async def activate_api_access(self):
|
|
try:
|
|
async with self.client() as session:
|
|
resp = await session.get(self.host + "Auth/TokenInfo")
|
|
if resp.status != 200:
|
|
return False
|
|
identifier = (await resp.json())["UserIdentifier"]
|
|
resp2 = await session.get(
|
|
self.host + "Auth/UserInfo/" + str(identifier)
|
|
)
|
|
return resp2.status == 200
|
|
except Exception:
|
|
return False
|
|
|
|
async def mktoken(self, login, passw):
|
|
"""Perform the Synergia OAuth dance and keep the resulting cookies."""
|
|
self.credentials_rejected = False
|
|
|
|
# Existing cookies may still be valid (e.g. after a reload).
|
|
if self.headers and await self.get_data("Me"):
|
|
return True
|
|
|
|
try:
|
|
async with aiohttp.ClientSession(
|
|
headers=BROWSER_HEADERS,
|
|
timeout=aiohttp.ClientTimeout(total=self.request_timeout),
|
|
) as session:
|
|
# Synergia has to hand us a session before the OAuth grant
|
|
# has anything to bind to.
|
|
await session.get(
|
|
"https://synergia.librus.pl/loguj/portalRodzina?v=%d"
|
|
% int(time.time())
|
|
)
|
|
|
|
form = aiohttp.FormData()
|
|
form.add_field("action", "login")
|
|
form.add_field("login", login)
|
|
form.add_field("pass", passw)
|
|
resp = await session.post(
|
|
"https://api.librus.pl/OAuth/Authorization?client_id=%d"
|
|
% OAUTH_CLIENT_ID,
|
|
data=form,
|
|
)
|
|
try:
|
|
payload = await resp.json(content_type=None)
|
|
except Exception:
|
|
payload = None
|
|
if not isinstance(payload, dict):
|
|
# Not the JSON verdict at all (outage / captcha /
|
|
# maintenance page).
|
|
return False
|
|
if payload.get("status") != "ok":
|
|
self.credentials_rejected = True
|
|
return False
|
|
|
|
# The device check redirects onwards through Grant and back
|
|
# into Synergia.
|
|
await session.get(
|
|
"https://api.librus.pl/OAuth/Authorization/2FA?client_id=%d"
|
|
% OAUTH_CLIENT_ID
|
|
)
|
|
|
|
jar = session.cookie_jar.filter_cookies(
|
|
yarl.URL("https://synergia.librus.pl")
|
|
)
|
|
self.headers = {
|
|
key: morsel.value for key, morsel in jar.items()
|
|
}
|
|
if "DZIENNIKSID" not in self.headers or "SDZIENNIKSID" not in self.headers:
|
|
return False
|
|
return await self.activate_api_access()
|
|
except Exception:
|
|
return False
|
|
|
|
# --------------------------------------------------------------- helpers
|
|
|
|
async def curl(self, link):
|
|
status = 0
|
|
for _ in range(self.retries):
|
|
try:
|
|
async with self.client() as session:
|
|
async with session.get(link) as resp:
|
|
status = resp.status
|
|
if status in (200, 401, 403):
|
|
return {"code": status, "text": await resp.text()}
|
|
except Exception:
|
|
pass
|
|
return {"code": status, "text": None}
|
|
|
|
async def get_data(self, method):
|
|
try:
|
|
r = await self.curl(self.host + method)
|
|
if r["code"] == 200 and r["text"]:
|
|
return json.loads(r["text"])
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
# ------------------------------------------------------------------- data
|
|
|
|
async def get_subjects(self):
|
|
r = await self.get_data("Subjects")
|
|
if r is None:
|
|
return None
|
|
return {x["Id"]: x["Name"] for x in r.get("Subjects", [])}
|
|
|
|
async def get_teachers(self):
|
|
r = await self.get_data("Users")
|
|
if r is None:
|
|
return None
|
|
return {
|
|
i["Id"]: {
|
|
"FirstName": i.get("FirstName", ""),
|
|
"LastName": i.get("LastName", ""),
|
|
}
|
|
for i in r.get("Users", [])
|
|
}
|
|
|
|
async def get_categories(self):
|
|
r = await self.get_data("Grades/Categories")
|
|
if r is None:
|
|
return None
|
|
categories = {}
|
|
for i in r.get("Categories", []):
|
|
categories[i["Id"]] = {
|
|
"Name": i.get("Name", "?"),
|
|
"Weight": i.get("Weight", "none"),
|
|
}
|
|
return categories
|
|
|
|
async def get_categories2(self):
|
|
r = await self.get_data("HomeWorks/Categories")
|
|
if r is None:
|
|
return None
|
|
return {
|
|
i["Id"]: {"Name": i.get("Name", "?")}
|
|
for i in r.get("Categories", [])
|
|
}
|
|
|
|
async def get_comments(self):
|
|
r = await self.get_data("Grades/Comments")
|
|
if r is None:
|
|
return None
|
|
return {i["Id"]: {"Text": i.get("Text", "")} for i in r.get("Comments", [])}
|
|
|
|
async def get_grades(self):
|
|
response, categories, comments, subjects, teachers = await asyncio.gather(
|
|
self.get_data("Grades"),
|
|
self.get_categories(),
|
|
self.get_comments(),
|
|
self.get_subjects(),
|
|
self.get_teachers(),
|
|
)
|
|
if None in (response, categories, comments, subjects, teachers):
|
|
return None
|
|
|
|
grades = {name: [] for name in subjects.values()}
|
|
for x in response.get("Grades", []):
|
|
try:
|
|
subject = subjects[x["Subject"]["Id"]]
|
|
category = categories[x["Category"]["Id"]]
|
|
except (KeyError, TypeError):
|
|
continue
|
|
|
|
comment = "No comment"
|
|
try:
|
|
comment = comments[x["Comments"][0]["Id"]]["Text"]
|
|
except (KeyError, IndexError, TypeError):
|
|
pass
|
|
|
|
teacher = {"FirstName": "unknown", "LastName": "teacher"}
|
|
try:
|
|
teacher = teachers[x["AddedBy"]["Id"]]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
|
|
grades[subject].append(
|
|
{
|
|
"Grade": x.get("Grade", "?"),
|
|
"Weight": category["Weight"],
|
|
"Comment": comment,
|
|
"Category": category["Name"],
|
|
"isFinal": bool(x.get("IsFinal") or x.get("IsFinalProposition")),
|
|
"isSemester": bool(
|
|
x.get("IsSemester") or x.get("IsSemesterProposition")
|
|
),
|
|
"separator": bool(x.get("IsSemester")),
|
|
"underlined": bool(x.get("IsSemester") or x.get("IsFinal")),
|
|
"Semester": x.get("Semester"),
|
|
"AddDate": x.get("AddDate"),
|
|
"AddedBy": {
|
|
"FirstName": teacher.get("FirstName", ""),
|
|
"LastName": teacher.get("LastName", ""),
|
|
},
|
|
}
|
|
)
|
|
return grades
|
|
|
|
async def get_me(self):
|
|
g, u, w, n = await asyncio.gather(
|
|
self.get_data("Me"),
|
|
self.get_data("UserProfile"),
|
|
self.get_teachers(),
|
|
self.get_data("Classes"),
|
|
)
|
|
if None in (g, u, w, n):
|
|
return None
|
|
try:
|
|
tutor = w[n["Class"]["ClassTutor"]["Id"]]
|
|
return {
|
|
"FirstName": g["Me"]["Account"]["FirstName"],
|
|
"LastName": g["Me"]["Account"]["LastName"],
|
|
"Tutor": tutor,
|
|
"SchoolYearStarts": n["Class"]["BeginSchoolYear"],
|
|
"SchoolYearMiddles": n["Class"]["EndFirstSemester"],
|
|
"SchoolYearEnds": n["Class"]["EndSchoolYear"],
|
|
"Type": u["UserProfile"]["UnitType"].title(),
|
|
"Class": "%s %s" % (n["Class"]["Number"], n["Class"]["Symbol"].upper()),
|
|
}
|
|
except (KeyError, TypeError):
|
|
return None
|
|
|
|
async def get_school(self):
|
|
data = await self.get_data("Schools")
|
|
return data.get("School") if data else None
|
|
|
|
async def get_free_days(self):
|
|
r = await self.get_data("SchoolFreeDays")
|
|
if r is None:
|
|
return None
|
|
days = []
|
|
for item in r.get("SchoolFreeDays", []):
|
|
item = dict(item)
|
|
item.pop("Id", None)
|
|
item.pop("Units", None)
|
|
days.append(item)
|
|
return days
|
|
|
|
async def get_teacher_free_days(self):
|
|
r = await self.get_data("TeacherFreeDays")
|
|
teachers = await self.get_teachers()
|
|
if r is None or teachers is None:
|
|
return None
|
|
days = []
|
|
for item in r.get("TeacherFreeDays", []):
|
|
item = dict(item)
|
|
item.pop("Id", None)
|
|
try:
|
|
item["Teacher"] = teachers[item["Teacher"]["Id"]]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
days.append(item)
|
|
return days
|
|
|
|
async def get_lucky_number(self):
|
|
try:
|
|
return (await self.get_data("LuckyNumbers"))["LuckyNumber"]["LuckyNumber"]
|
|
except (KeyError, TypeError):
|
|
return None
|
|
|
|
async def get_behaviour_grade(self):
|
|
try:
|
|
ids = [
|
|
"nothing",
|
|
"wzorowe",
|
|
"bardzo dobre",
|
|
"dobre",
|
|
"poprawne",
|
|
"nieodpowiednie",
|
|
"naganne",
|
|
"nothing",
|
|
]
|
|
raw = (await self.get_data("BehaviourGrades"))["Grades"][0]
|
|
teacher = (await self.get_teachers())[raw["Teacher"]["Id"]]
|
|
return {
|
|
"Grade": ids[raw["GradeType"]["Id"]],
|
|
"Teacher": {
|
|
"FirstName": teacher.get("FirstName", ""),
|
|
"LastName": teacher.get("LastName", ""),
|
|
},
|
|
}
|
|
except (KeyError, IndexError, TypeError):
|
|
return None
|
|
|
|
async def get_classrooms(self):
|
|
d = await self.get_data("TimetableEntries")
|
|
if d is None:
|
|
return {}
|
|
classrooms = {}
|
|
for x in d.get("TimetableEntries", []):
|
|
classroom = x.get("Classroom")
|
|
if classroom and "Id" in classroom:
|
|
classrooms[str(classroom["Id"])] = {
|
|
"Name": classroom.get("Name", ""),
|
|
"Symbol": classroom.get("Symbol", classroom.get("Name", "")),
|
|
}
|
|
return classrooms
|
|
|
|
async def get_timetable(self, week=0):
|
|
"""Return a timetable. week=0 is the current week, 1 the next one."""
|
|
today = datetime.now()
|
|
week_start = today - timedelta(days=today.weekday()) + timedelta(weeks=week)
|
|
week_end = week_start + timedelta(days=6)
|
|
|
|
date_from = week_start.strftime("%Y-%m-%d")
|
|
date_to = week_end.strftime("%Y-%m-%d")
|
|
|
|
r, additional, classrooms = await asyncio.gather(
|
|
self.get_data("Timetables?weekStart=%s" % date_from),
|
|
self.get_data(
|
|
"Timetables/OtherActivitiesRegister?dateFrom=%s&dateTo=%s"
|
|
"&hideOutdatedEntries=false" % (date_from, date_to)
|
|
),
|
|
self.get_classrooms(),
|
|
)
|
|
|
|
if r is None:
|
|
return None
|
|
if "Timetable" not in r:
|
|
return {
|
|
"Timetable": {},
|
|
"weekStart": date_from,
|
|
"weekEnd": date_to,
|
|
"week": week,
|
|
}
|
|
|
|
time_to_lesson = {}
|
|
for date_key, entries in r["Timetable"].items():
|
|
for entry in entries or []:
|
|
if entry:
|
|
time_to_lesson[entry[0].get("HourFrom")] = entry[0].get(
|
|
"LessonNo", "?"
|
|
)
|
|
|
|
timetable = {day: [] for day in DAY_NAMES}
|
|
|
|
for date_str, entries in r["Timetable"].items():
|
|
if not entries:
|
|
continue
|
|
try:
|
|
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
|
except ValueError:
|
|
continue
|
|
day_name = DAY_NAMES[dt.weekday()]
|
|
|
|
for entry_list in entries:
|
|
if not entry_list:
|
|
continue
|
|
lesson = entry_list[0]
|
|
classroom_name = "?"
|
|
try:
|
|
room = classrooms.get(str(lesson["Classroom"]["Id"]), {})
|
|
classroom_name = room.get("Symbol") or room.get("Name") or "?"
|
|
except (KeyError, TypeError):
|
|
pass
|
|
try:
|
|
teacher = {
|
|
"FirstName": lesson["Teacher"].get("FirstName", ""),
|
|
"LastName": lesson["Teacher"].get("LastName", ""),
|
|
}
|
|
except (KeyError, TypeError):
|
|
teacher = {"FirstName": "", "LastName": ""}
|
|
|
|
subject = lesson.get("Subject") or {}
|
|
timetable[day_name].append(
|
|
{
|
|
"Lesson": lesson.get("LessonNo", "?"),
|
|
"Subject": subject.get("Name", "Lesson"),
|
|
"isSubstitution": bool(lesson.get("IsSubstitutionClass")),
|
|
"isCancelled": bool(lesson.get("IsCanceled")),
|
|
"Teacher": teacher,
|
|
"HourFrom": lesson.get("HourFrom", ""),
|
|
"HourTo": lesson.get("HourTo", ""),
|
|
"Classroom": classroom_name,
|
|
"Date": date_str,
|
|
}
|
|
)
|
|
|
|
for item in (additional or {}).get("data", []) or []:
|
|
try:
|
|
dt = datetime.strptime(item["date"], "%Y-%m-%d")
|
|
except (KeyError, ValueError):
|
|
continue
|
|
day_name = DAY_NAMES[dt.weekday()]
|
|
|
|
parts = (item.get("teacherName") or "").split(" ")
|
|
last = parts[0] if parts else ""
|
|
first = " ".join(parts[1:]) if len(parts) > 1 else ""
|
|
|
|
classroom = item.get("classroom") or {}
|
|
timetable[day_name].append(
|
|
{
|
|
"Lesson": time_to_lesson.get(item.get("startTime"), "-"),
|
|
"Subject": item.get("title", "Activity"),
|
|
"isSubstitution": False,
|
|
"isCancelled": False,
|
|
"Teacher": {"FirstName": first, "LastName": last},
|
|
"HourFrom": item.get("startTime", ""),
|
|
"HourTo": item.get("endTime", ""),
|
|
"Classroom": classroom.get("symbol", "?"),
|
|
"Date": item["date"],
|
|
"isAdditional": True,
|
|
}
|
|
)
|
|
|
|
final = {}
|
|
for day in DAY_NAMES:
|
|
if timetable[day]:
|
|
timetable[day].sort(key=lambda x: x.get("HourFrom") or "")
|
|
final[day] = timetable[day]
|
|
|
|
return {
|
|
"Timetable": final,
|
|
"weekStart": date_from,
|
|
"weekEnd": date_to,
|
|
"week": week,
|
|
}
|
|
|
|
async def get_exams(self):
|
|
r, teachers, categories, subjects = await asyncio.gather(
|
|
self.get_data("HomeWorks"),
|
|
self.get_teachers(),
|
|
self.get_categories2(),
|
|
self.get_subjects(),
|
|
)
|
|
if None in (r, teachers, categories, subjects):
|
|
return None
|
|
|
|
exams = []
|
|
for x in r.get("HomeWorks", []):
|
|
try:
|
|
lesson = subjects[x["Subject"]["Id"]]
|
|
except (KeyError, TypeError):
|
|
lesson = "Lesson %s" % x.get("LessonNo", "?")
|
|
try:
|
|
added_by = teachers[x["CreatedBy"]["Id"]]
|
|
except (KeyError, TypeError):
|
|
added_by = {"FirstName": "", "LastName": ""}
|
|
try:
|
|
category = categories[x["Category"]["Id"]]["Name"]
|
|
except (KeyError, TypeError):
|
|
category = "?"
|
|
|
|
exams.append(
|
|
{
|
|
"Lesson": lesson,
|
|
"AddedBy": added_by,
|
|
"Type": category,
|
|
"StartTime": x.get("TimeFrom"),
|
|
"EndTime": x.get("TimeTo"),
|
|
"Date": x.get("Date"),
|
|
"AddDate": x.get("AddDate"),
|
|
"Content": x.get("Content", ""),
|
|
}
|
|
)
|
|
return exams
|
|
|
|
async def get_attendances(self):
|
|
g, u, w, n, o = await asyncio.gather(
|
|
self.get_data("Attendances"),
|
|
self.get_teachers(),
|
|
self.get_data("Lessons"),
|
|
self.get_subjects(),
|
|
self.get_data("Attendances/Types"),
|
|
)
|
|
if None in (g, u, w, n, o):
|
|
return None
|
|
|
|
lessons = {
|
|
a["Id"]: a.get("Subject", {}).get("Id")
|
|
for a in w.get("Lessons", [])
|
|
}
|
|
types = {
|
|
a["Id"]: {
|
|
"isPresence": a.get("IsPresenceKind", True),
|
|
"Type": a.get("Name", "?"),
|
|
"Short": a.get("Short", "?"),
|
|
}
|
|
for a in o.get("Types", [])
|
|
}
|
|
|
|
attends = []
|
|
for x in g.get("Attendances", []):
|
|
try:
|
|
subject = n[lessons[x["Lesson"]["Id"]]]
|
|
except (KeyError, TypeError):
|
|
subject = "?"
|
|
try:
|
|
type_info = types[x["Type"]["Id"]]
|
|
except (KeyError, TypeError):
|
|
type_info = {"Type": "?", "Short": "?", "isPresence": True}
|
|
try:
|
|
added_by = u[x["AddedBy"]["Id"]]
|
|
except (KeyError, TypeError):
|
|
added_by = {"FirstName": "", "LastName": ""}
|
|
|
|
attends.append(
|
|
{
|
|
"Lesson": subject,
|
|
"Type": type_info["Type"],
|
|
"Short": type_info["Short"],
|
|
"isPresence": type_info["isPresence"],
|
|
"Added": x.get("AddDate"),
|
|
"Date": x.get("Date"),
|
|
"AddedBy": added_by,
|
|
}
|
|
)
|
|
return attends
|
|
|
|
async def get_parent_teacher_conferences(self):
|
|
data = await self.get_data("ParentTeacherConferences")
|
|
return (data or {}).get("ParentTeacherConferences", [])
|
|
|
|
async def get_notifications(self, days=14):
|
|
grades, exams, attendances, types, conferences = await asyncio.gather(
|
|
self.get_data("Grades"),
|
|
self.get_data("HomeWorks"),
|
|
self.get_data("Attendances"),
|
|
self.get_data("Attendances/Types"),
|
|
self.get_parent_teacher_conferences(),
|
|
)
|
|
presence = {
|
|
str(x["Id"]): x.get("IsPresenceKind", True)
|
|
for x in (types or {}).get("Types", [])
|
|
}
|
|
absences = 0
|
|
for attendance in (attendances or {}).get("Attendances", []):
|
|
try:
|
|
absences += not presence[str(attendance["Type"]["Id"])]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
return {
|
|
"grades": len((grades or {}).get("Grades", [])),
|
|
"exams": len((exams or {}).get("HomeWorks", [])),
|
|
"absences": absences,
|
|
"conferences": len(conferences or []),
|
|
}
|
|
|
|
# --------------------------------------------------------------- messages
|
|
|
|
async def get_messages(self):
|
|
try:
|
|
async with self.client() as session:
|
|
resp = await session.get("%s/wiadomosci" % self.msg_host)
|
|
html = await resp.text()
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
table = soup.find("table", class_="decorated stretch")
|
|
if table is None:
|
|
return []
|
|
messages = []
|
|
for row in table.find_all("tr"):
|
|
try:
|
|
cols = row.find_all("td")
|
|
if len(cols) < 5:
|
|
continue
|
|
sender = cols[2].get_text(" ", strip=True)
|
|
if "(" in sender:
|
|
sender = sender[: sender.index("(")].strip()
|
|
href = cols[2].find("a", href=True)["href"][1:]
|
|
messages.append(
|
|
{
|
|
"from": sender,
|
|
"subject": cols[3].get_text(" ", strip=True),
|
|
"date": cols[4].get_text(" ", strip=True),
|
|
"link": href[href.index("/") + 1 :].replace("/", "-"),
|
|
}
|
|
)
|
|
except Exception:
|
|
continue
|
|
return messages
|
|
except Exception:
|
|
return []
|
|
|
|
async def get_message(self, link):
|
|
try:
|
|
async with self.client() as session:
|
|
resp = await session.get(
|
|
"%s/wiadomosci/%s"
|
|
% (self.msg_host, link.replace("-", "/"))
|
|
)
|
|
html = await resp.text()
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
content_div = soup.find("div", class_="container-message-content")
|
|
content = content_div.decode_contents() if content_div else ""
|
|
|
|
details = {}
|
|
for row in soup.find_all("tr"):
|
|
cells = row.find_all("td")
|
|
for i, cell in enumerate(cells):
|
|
label = cell.get_text(" ", strip=True).lower().rstrip(":")
|
|
if label in ("temat", "od", "data", "przeczytano") and i + 1 < len(cells):
|
|
details[label] = cells[i + 1].get_text(" ", strip=True)
|
|
|
|
attachments = []
|
|
seen = set()
|
|
for img in soup.find_all("img"):
|
|
onclick = str(img.get("onclick") or "")
|
|
match = re.search(
|
|
r"""(['"])([^'"]*pobierz_zalacznik[^'"]*)\1""", onclick
|
|
)
|
|
if not match:
|
|
continue
|
|
source = match.group(2)
|
|
if source.startswith("/"):
|
|
source = self.msg_host + source
|
|
nice = source.split("/pobierz_zalacznik/")[-1]
|
|
if nice in seen:
|
|
continue
|
|
seen.add(nice)
|
|
name = nice
|
|
row = img.find_parent("tr")
|
|
if row:
|
|
first_cell = row.find("td")
|
|
if first_cell and first_cell.get_text(strip=True):
|
|
name = first_cell.get_text(" ", strip=True)
|
|
attachments.append(
|
|
{
|
|
"name": name,
|
|
"source": source,
|
|
"direct": "/get",
|
|
"nice": nice,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"subject": details.get("temat", "(no subject)"),
|
|
"from": details.get("od", ""),
|
|
"date": details.get("data", ""),
|
|
"read": details.get("przeczytano", ""),
|
|
"content": content,
|
|
"attachments": attachments,
|
|
}
|
|
except Exception as exc: # keep the UI alive on markup changes
|
|
return {
|
|
"subject": "Could not load message",
|
|
"from": "system",
|
|
"date": "",
|
|
"read": "",
|
|
"content": "<p>%s</p>" % repr(exc),
|
|
"attachments": [],
|
|
}
|
|
|
|
async def download_file(self, link):
|
|
try:
|
|
async with self.client() as session:
|
|
resp = await session.get(
|
|
"%s/wiadomosci/pobierz_zalacznik/%s"
|
|
% (self.msg_host, link)
|
|
)
|
|
resp_f = await session.get("%s/get" % resp.url)
|
|
return {
|
|
"headers": resp_f.headers,
|
|
"content": await resp_f.read(),
|
|
}
|
|
except Exception:
|
|
return None
|