perf: change structure and simplify
This commit is contained in:
@@ -11,9 +11,9 @@
|
|||||||
async def main() -> None:
|
async def main() -> None:
|
||||||
cpf = "00011122233"
|
cpf = "00011122233"
|
||||||
async with PPGEE(user=cpf, password=cpf) as ppgee:
|
async with PPGEE(user=cpf, password=cpf) as ppgee:
|
||||||
frequency_page = await ppgee.frequency()
|
attendency_page = await ppgee.attendency()
|
||||||
print(frequency_page.history())
|
print(attendency_page.history())
|
||||||
await frequency_page.confirm()
|
await attendency_page.confirm()
|
||||||
await asyncio.sleep(5)
|
await asyncio.sleep(5)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -9,9 +9,9 @@ logging.basicConfig(level=logging.INFO)
|
|||||||
async def main() -> None:
|
async def main() -> None:
|
||||||
cpf = "00011122233"
|
cpf = "00011122233"
|
||||||
async with PPGEE(user=cpf, password=cpf) as ppgee:
|
async with PPGEE(user=cpf, password=cpf) as ppgee:
|
||||||
frequency_page = await ppgee.frequency()
|
attendency_page = await ppgee.attendency()
|
||||||
print(frequency_page.history())
|
print(attendency_page.history())
|
||||||
await frequency_page.confirm()
|
await attendency_page.confirm()
|
||||||
await asyncio.sleep(5)
|
await asyncio.sleep(5)
|
||||||
|
|
||||||
|
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
__version__ = "0.1.3"
|
__version__ = "0.1.4"
|
||||||
|
|
||||||
from .client import PPGEE
|
from .client import PPGEE
|
||||||
|
|
||||||
|
|||||||
@@ -21,11 +21,11 @@ async def cli():
|
|||||||
user, password = args.username, args.password
|
user, password = args.username, args.password
|
||||||
try:
|
try:
|
||||||
async with PPGEE(user=user, password=password) as ppgee:
|
async with PPGEE(user=user, password=password) as ppgee:
|
||||||
frequency_page = await ppgee.frequency()
|
attendency_page = await ppgee.attendency()
|
||||||
await asyncio.sleep(1)
|
await asyncio.sleep(1)
|
||||||
if frequency_page.is_available():
|
if attendency_page.is_available():
|
||||||
print("Attendency confirmed.")
|
print("Attendency confirmed.")
|
||||||
await frequency_page.confirm()
|
await attendency_page.confirm()
|
||||||
else:
|
else:
|
||||||
eprint("Attendency not available yet.")
|
eprint("Attendency not available yet.")
|
||||||
await asyncio.sleep(1)
|
await asyncio.sleep(1)
|
||||||
|
|||||||
@@ -1,23 +1,16 @@
|
|||||||
import aiohttp
|
|
||||||
import logging
|
import logging
|
||||||
from ppgee.http import HttpClient
|
|
||||||
from ppgee.pages import FrequencyPage
|
import aiohttp
|
||||||
from functools import wraps
|
|
||||||
from ppgee import errors
|
from ppgee import errors
|
||||||
|
from ppgee.http import HttpClient
|
||||||
|
from ppgee.pages import AttendencyPage, AttendencyHistory, AttendencyHistoryEntry
|
||||||
|
from ppgee.parsers import AttendencyParser
|
||||||
|
from ppgee.permissions import is_logged_check
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def is_logged_check(method):
|
|
||||||
@wraps(method)
|
|
||||||
def wrapper(self, *args, **kwargs):
|
|
||||||
if not self.is_logged:
|
|
||||||
raise Exception("You must be logged in to use this method")
|
|
||||||
return method(self, *args, **kwargs)
|
|
||||||
|
|
||||||
return wrapper
|
|
||||||
|
|
||||||
|
|
||||||
class PPGEE:
|
class PPGEE:
|
||||||
def __init__(self, user: str | None = None, password: str | None = None) -> None:
|
def __init__(self, user: str | None = None, password: str | None = None) -> None:
|
||||||
self.user = user
|
self.user = user
|
||||||
@@ -56,10 +49,16 @@ class PPGEE:
|
|||||||
logger.info("Logged in without credentials")
|
logger.info("Logged in without credentials")
|
||||||
|
|
||||||
@is_logged_check
|
@is_logged_check
|
||||||
async def frequency(self) -> FrequencyPage:
|
async def attendency(self) -> AttendencyPage:
|
||||||
logger.info("Requesting frequency page...")
|
logger.info("Requesting attendency page...")
|
||||||
html = await self.http.frequency()
|
html = await self.http.attendency()
|
||||||
return FrequencyPage(html, self.http.frequency_confirmation)
|
parser = AttendencyParser(html)
|
||||||
|
history = AttendencyHistory(
|
||||||
|
(AttendencyHistoryEntry(**item) for item in parser.history())
|
||||||
|
)
|
||||||
|
return AttendencyPage(
|
||||||
|
history, self.http.attendency_confirmation, parser.availability()
|
||||||
|
)
|
||||||
|
|
||||||
@is_logged_check
|
@is_logged_check
|
||||||
async def logoff(self) -> None:
|
async def logoff(self) -> None:
|
||||||
|
|||||||
@@ -26,10 +26,10 @@ class HttpClient:
|
|||||||
data: dict[str, str] = {"login": user, "senha": password}
|
data: dict[str, str] = {"login": user, "senha": password}
|
||||||
return await self._request("post", f"{URL_BASE}/login.php", data=data)
|
return await self._request("post", f"{URL_BASE}/login.php", data=data)
|
||||||
|
|
||||||
async def frequency(self) -> str:
|
async def attendency(self) -> str:
|
||||||
return await self._request("get", f"{URL_BASE}/afreq.php")
|
return await self._request("get", f"{URL_BASE}/afreq.php")
|
||||||
|
|
||||||
async def frequency_confirmation(self) -> str:
|
async def attendency_confirmation(self) -> str:
|
||||||
today = date.today()
|
today = date.today()
|
||||||
data: dict[str, str] = {
|
data: dict[str, str] = {
|
||||||
"freqano": str(today.year),
|
"freqano": str(today.year),
|
||||||
|
|||||||
@@ -1,5 +1,3 @@
|
|||||||
from .frequency import FrequencyPage
|
from .attendency import AttendencyPage, AttendencyHistory, AttendencyHistoryEntry
|
||||||
|
|
||||||
__all__ = [
|
__all__ = ["AttendencyPage", "AttendencyHistory", "AttendencyHistoryEntry"]
|
||||||
"FrequencyPage",
|
|
||||||
]
|
|
||||||
|
|||||||
@@ -1,21 +1,20 @@
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
from typing import Callable, Awaitable
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
from ppgee.parser import parse_frequency_history
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class HistoryEntry:
|
class AttendencyHistoryEntry:
|
||||||
year: int
|
year: int
|
||||||
month: int
|
month: int
|
||||||
asked: datetime
|
asked: datetime
|
||||||
confirmed: datetime | None
|
confirmed: datetime | None
|
||||||
|
|
||||||
|
|
||||||
class History(list[HistoryEntry]):
|
class AttendencyHistory(list[AttendencyHistoryEntry]):
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
out: list[str] = []
|
out: list[str] = []
|
||||||
for entry in self:
|
for entry in self:
|
||||||
@@ -24,28 +23,24 @@ class History(list[HistoryEntry]):
|
|||||||
return "\n".join(out)
|
return "\n".join(out)
|
||||||
|
|
||||||
|
|
||||||
class FrequencyPage:
|
class AttendencyPage:
|
||||||
def __init__(self, html, confirmation_callback):
|
def __init__(
|
||||||
self.html = html
|
self,
|
||||||
self._history = History()
|
history: AttendencyHistory,
|
||||||
|
confirmation_callback: Callable[[], Awaitable[str]],
|
||||||
|
available: bool,
|
||||||
|
):
|
||||||
|
self._history = history
|
||||||
|
self._available = available
|
||||||
self.confirmation_callback = confirmation_callback
|
self.confirmation_callback = confirmation_callback
|
||||||
|
|
||||||
def _build_history(self):
|
def history(self) -> AttendencyHistory:
|
||||||
result_list = parse_frequency_history(self.html)
|
|
||||||
for item in result_list:
|
|
||||||
self._history.append(HistoryEntry(**item))
|
|
||||||
|
|
||||||
def history(self) -> History:
|
|
||||||
if not self._history:
|
|
||||||
self._build_history()
|
|
||||||
return self._history
|
return self._history
|
||||||
|
|
||||||
def is_available(self):
|
def is_available(self):
|
||||||
if "Opção não disponível" in self.html:
|
return self._available
|
||||||
return False
|
|
||||||
return True
|
|
||||||
|
|
||||||
async def confirm(self) -> None:
|
async def confirm(self) -> None:
|
||||||
if self.is_available():
|
if self.is_available():
|
||||||
logger.info("Requesting frequency confirmation...")
|
logger.info("Requesting attendency confirmation...")
|
||||||
await self.confirmation_callback()
|
await self.confirmation_callback()
|
||||||
@@ -1,46 +0,0 @@
|
|||||||
from datetime import datetime
|
|
||||||
import re
|
|
||||||
|
|
||||||
from bs4 import BeautifulSoup
|
|
||||||
|
|
||||||
MONTH_MAP = {
|
|
||||||
"Janeiro": 1,
|
|
||||||
"Fevereiro": 2,
|
|
||||||
"Março": 3,
|
|
||||||
"Abril": 4,
|
|
||||||
"Maio": 5,
|
|
||||||
"Junho": 6,
|
|
||||||
"Julho": 7,
|
|
||||||
"Agosto": 8,
|
|
||||||
"Setembro": 9,
|
|
||||||
"Outubro": 10,
|
|
||||||
"Novembro": 11,
|
|
||||||
"Dezembro": 12,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def parse_frequency_history(html) -> list[dict]:
|
|
||||||
history = []
|
|
||||||
soup = BeautifulSoup(html, "html.parser")
|
|
||||||
table = soup.find_all("table")[2].find_all("table")[-1]
|
|
||||||
rows = table.find_all("tr")
|
|
||||||
for row in rows[1:]:
|
|
||||||
cols = row.find_all("td")
|
|
||||||
data = [ele.text.strip() for ele in cols]
|
|
||||||
year = int(data[0])
|
|
||||||
month = MONTH_MAP[data[1]]
|
|
||||||
date_asked = datetime.strptime(data[2], "%d/%m/%Y%H:%M")
|
|
||||||
date_confirmed = re.sub(r"[^[0-9:/]]*", "", data[3])
|
|
||||||
if date_confirmed == "":
|
|
||||||
date_confirmed = None
|
|
||||||
else:
|
|
||||||
date_confirmed = datetime.strptime(date_confirmed, "%d/%m/%Y%H:%M")
|
|
||||||
history.append(
|
|
||||||
{
|
|
||||||
"year": year,
|
|
||||||
"month": month,
|
|
||||||
"asked": date_asked,
|
|
||||||
"confirmed": date_confirmed,
|
|
||||||
}
|
|
||||||
)
|
|
||||||
return history
|
|
||||||
5
ppgee/parsers/__init__.py
Normal file
5
ppgee/parsers/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
from .attendency import AttendencyParser
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"AttendencyParser",
|
||||||
|
]
|
||||||
53
ppgee/parsers/attendency.py
Normal file
53
ppgee/parsers/attendency.py
Normal file
@@ -0,0 +1,53 @@
|
|||||||
|
from datetime import datetime
|
||||||
|
import re
|
||||||
|
|
||||||
|
from bs4 import BeautifulSoup
|
||||||
|
|
||||||
|
MONTH_MAP = {
|
||||||
|
"Janeiro": 1,
|
||||||
|
"Fevereiro": 2,
|
||||||
|
"Março": 3,
|
||||||
|
"Abril": 4,
|
||||||
|
"Maio": 5,
|
||||||
|
"Junho": 6,
|
||||||
|
"Julho": 7,
|
||||||
|
"Agosto": 8,
|
||||||
|
"Setembro": 9,
|
||||||
|
"Outubro": 10,
|
||||||
|
"Novembro": 11,
|
||||||
|
"Dezembro": 12,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class AttendencyParser:
|
||||||
|
def __init__(self, html: str) -> None:
|
||||||
|
self._html = html
|
||||||
|
|
||||||
|
def history(self) -> list[dict]:
|
||||||
|
history = []
|
||||||
|
soup = BeautifulSoup(self._html, "html.parser")
|
||||||
|
table = soup.find_all("table")[2].find_all("table")[-1]
|
||||||
|
rows = table.find_all("tr")
|
||||||
|
for row in rows[1:]:
|
||||||
|
cols = row.find_all("td")
|
||||||
|
data = [ele.text.strip() for ele in cols]
|
||||||
|
year = int(data[0])
|
||||||
|
month = MONTH_MAP[data[1]]
|
||||||
|
date_asked = datetime.strptime(data[2], "%d/%m/%Y%H:%M")
|
||||||
|
date_confirmed = re.sub(r"[^[0-9:/]]*", "", data[3])
|
||||||
|
if date_confirmed == "":
|
||||||
|
date_confirmed = None
|
||||||
|
else:
|
||||||
|
date_confirmed = datetime.strptime(date_confirmed, "%d/%m/%Y%H:%M")
|
||||||
|
history.append(
|
||||||
|
{
|
||||||
|
"year": year,
|
||||||
|
"month": month,
|
||||||
|
"asked": date_asked,
|
||||||
|
"confirmed": date_confirmed,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return history
|
||||||
|
|
||||||
|
def availability(self) -> bool:
|
||||||
|
return "Opção não disponível" not in self._html
|
||||||
11
ppgee/permissions.py
Normal file
11
ppgee/permissions.py
Normal file
@@ -0,0 +1,11 @@
|
|||||||
|
from functools import wraps
|
||||||
|
|
||||||
|
|
||||||
|
def is_logged_check(method):
|
||||||
|
@wraps(method)
|
||||||
|
def wrapper(self, *args, **kwargs):
|
||||||
|
if not self.is_logged:
|
||||||
|
raise Exception("You must be logged in to use this method")
|
||||||
|
return method(self, *args, **kwargs)
|
||||||
|
|
||||||
|
return wrapper
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
[tool.poetry]
|
[tool.poetry]
|
||||||
name = "ppgee"
|
name = "ppgee"
|
||||||
version = "0.1.3"
|
version = "0.1.4"
|
||||||
description = ""
|
description = ""
|
||||||
authors = ["tiagovla <tiagovla@gmail.com>"]
|
authors = ["tiagovla <tiagovla@gmail.com>"]
|
||||||
|
|
||||||
|
|||||||
@@ -2,4 +2,4 @@ from ppgee import __version__
|
|||||||
|
|
||||||
|
|
||||||
def test_version():
|
def test_version():
|
||||||
assert __version__ == "0.1.3"
|
assert __version__ == "0.1.4"
|
||||||
|
|||||||
Reference in New Issue
Block a user