feat: add frequency page
This commit is contained in:
@@ -1,16 +1,17 @@
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
from ppgee import PPGEE
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
async def main():
|
||||
|
||||
async def main() -> None:
|
||||
cpf = "00011122233"
|
||||
async with PPGEE(cpf, cpf) as ppgee:
|
||||
response = await ppgee.frequency()
|
||||
if "Opção não disponível" in response:
|
||||
print("Not ready yet")
|
||||
async with PPGEE(user=cpf, password=cpf) as ppgee:
|
||||
frequency_page = await ppgee.frequency()
|
||||
print(frequency_page.history())
|
||||
await frequency_page.confirm()
|
||||
await asyncio.sleep(5)
|
||||
|
||||
|
||||
|
||||
27
poetry.lock
generated
27
poetry.lock
generated
@@ -59,6 +59,21 @@ docs = ["furo", "sphinx", "zope.interface", "sphinx-notfound-page"]
|
||||
tests = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "mypy (>=0.900,!=0.940)", "pytest-mypy-plugins", "zope.interface", "cloudpickle"]
|
||||
tests_no_zope = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "mypy (>=0.900,!=0.940)", "pytest-mypy-plugins", "cloudpickle"]
|
||||
|
||||
[[package]]
|
||||
name = "beautifulsoup4"
|
||||
version = "4.11.1"
|
||||
description = "Screen-scraping library"
|
||||
category = "main"
|
||||
optional = false
|
||||
python-versions = ">=3.6.0"
|
||||
|
||||
[package.dependencies]
|
||||
soupsieve = ">1.2"
|
||||
|
||||
[package.extras]
|
||||
html5lib = ["html5lib"]
|
||||
lxml = ["lxml"]
|
||||
|
||||
[[package]]
|
||||
name = "charset-normalizer"
|
||||
version = "2.1.1"
|
||||
@@ -193,6 +208,14 @@ python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
|
||||
python-http-client = ">=3.2.1"
|
||||
starkbank-ecdsa = ">=2.0.1"
|
||||
|
||||
[[package]]
|
||||
name = "soupsieve"
|
||||
version = "2.3.2.post1"
|
||||
description = "A modern CSS selector implementation for Beautiful Soup."
|
||||
category = "main"
|
||||
optional = false
|
||||
python-versions = ">=3.6"
|
||||
|
||||
[[package]]
|
||||
name = "starkbank-ecdsa"
|
||||
version = "2.0.3"
|
||||
@@ -224,7 +247,7 @@ multidict = ">=4.0"
|
||||
[metadata]
|
||||
lock-version = "1.1"
|
||||
python-versions = "^3.10"
|
||||
content-hash = "e2c81cf40eb115ecc3c2e19723d94097e237dbda967e5eaf06660c758e73e915"
|
||||
content-hash = "04315ad95e7d31d6046cbac5bc97855e2fa45eb9c6c2f16c2cc0dfc166e1bcf1"
|
||||
|
||||
[metadata.files]
|
||||
aiohttp = []
|
||||
@@ -232,6 +255,7 @@ aiosignal = []
|
||||
async-timeout = []
|
||||
atomicwrites = []
|
||||
attrs = []
|
||||
beautifulsoup4 = []
|
||||
charset-normalizer = []
|
||||
colorama = []
|
||||
frozenlist = []
|
||||
@@ -245,6 +269,7 @@ pyparsing = []
|
||||
pytest = []
|
||||
python-http-client = []
|
||||
sendgrid = []
|
||||
soupsieve = []
|
||||
starkbank-ecdsa = []
|
||||
wcwidth = []
|
||||
yarl = []
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import aiohttp
|
||||
import asyncio
|
||||
import logging
|
||||
from ppgee.http import HttpClient
|
||||
from ppgee.pages import FrequencyPage
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -13,26 +13,32 @@ class PPGEE:
|
||||
self.session: aiohttp.ClientSession
|
||||
self.http: HttpClient
|
||||
|
||||
async def __aenter__(self):
|
||||
async def start(self):
|
||||
self.session = aiohttp.ClientSession()
|
||||
self.http = HttpClient(self.session)
|
||||
|
||||
async def close(self):
|
||||
if self.session:
|
||||
await self.session.close()
|
||||
|
||||
async def __aenter__(self):
|
||||
await self.start()
|
||||
await self.login()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb) -> None:
|
||||
await self.logoff()
|
||||
await asyncio.sleep(1)
|
||||
if self.session:
|
||||
await self.session.close()
|
||||
await self.close()
|
||||
|
||||
async def login(self) -> str:
|
||||
logger.info("Logging in...")
|
||||
return await self.http.login(self.user, self.password)
|
||||
|
||||
async def frequency(self) -> str:
|
||||
return await self.http.frequency()
|
||||
|
||||
async def frequency_confirmation(self) -> str:
|
||||
return await self.http.frequency_confirmation()
|
||||
async def frequency(self) -> FrequencyPage:
|
||||
logger.info("Requesting frequency page...")
|
||||
html = await self.http.frequency()
|
||||
return FrequencyPage(html, self.http.frequency_confirmation)
|
||||
|
||||
async def logoff(self) -> str:
|
||||
logger.info("Logging off...")
|
||||
return await self.http.logoff()
|
||||
|
||||
1
ppgee/pages/__init__.py
Normal file
1
ppgee/pages/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
from .frequency import FrequencyPage
|
||||
51
ppgee/pages/frequency.py
Normal file
51
ppgee/pages/frequency.py
Normal file
@@ -0,0 +1,51 @@
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
import logging
|
||||
|
||||
from ppgee.parser import parse_frequency_history
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class HistoryEntry:
|
||||
year: int
|
||||
month: int
|
||||
asked: datetime
|
||||
confirmed: datetime | None
|
||||
|
||||
|
||||
class History(list[HistoryEntry]):
|
||||
def __str__(self) -> str:
|
||||
out: list[str] = []
|
||||
for entry in self:
|
||||
e = [entry.year, entry.month, entry.asked, entry.confirmed]
|
||||
out.append("\t".join(map(str, e)))
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
class FrequencyPage:
|
||||
def __init__(self, html, confirmation_callback):
|
||||
self.html = html
|
||||
self._history = History()
|
||||
self.confirmation_callback = confirmation_callback
|
||||
|
||||
def _build_history(self):
|
||||
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
|
||||
|
||||
def is_available(self):
|
||||
if "Opção não disponível" in self.html:
|
||||
return False
|
||||
return True
|
||||
|
||||
async def confirm(self) -> None:
|
||||
if self.is_available():
|
||||
logger.info("Requesting frequency confirmation...")
|
||||
await self.confirmation_callback()
|
||||
46
ppgee/parser.py
Normal file
46
ppgee/parser.py
Normal file
@@ -0,0 +1,46 @@
|
||||
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
|
||||
@@ -7,6 +7,7 @@ authors = ["tiagovla <tiagovla@gmail.com>"]
|
||||
[tool.poetry.dependencies]
|
||||
python = "^3.10"
|
||||
aiohttp = "^3.8.1"
|
||||
beautifulsoup4 = "^4.11.1"
|
||||
|
||||
[tool.poetry.dev-dependencies]
|
||||
pytest = "^5.2"
|
||||
|
||||
Reference in New Issue
Block a user