Initial commit
Some checks failed
Build and Run VSTU Public TG bot / build_and_run (push) Failing after 6s
Some checks failed
Build and Run VSTU Public TG bot / build_and_run (push) Failing after 6s
This commit is contained in:
52
data_manager.py
Normal file
52
data_manager.py
Normal file
@@ -0,0 +1,52 @@
|
||||
import aiohttp
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class DataManager:
|
||||
def __init__(self, facultets_url, groups_url):
|
||||
self.facultets_url = facultets_url
|
||||
self.groups_url = groups_url
|
||||
self.facs = {} # {id: {name: ...}}
|
||||
self.groups = {} # {id: {real_name: ..., facultet_tech: ..., ...}}
|
||||
|
||||
async def refresh_cache(self, connector):
|
||||
async with aiohttp.ClientSession(connector=connector) as session:
|
||||
async def fetch(url):
|
||||
async with session.get(url) as r: return await r.json()
|
||||
try:
|
||||
self.facs = await fetch(self.facultets_url)
|
||||
self.groups = (await fetch(self.groups_url))['groups']
|
||||
logger.info(f"Loaded {len(self.facs)} facs and {len(self.groups)} groups")
|
||||
except Exception as e:
|
||||
logger.error(f"Cache refresh failed: {e}")
|
||||
|
||||
def get_groups_by_fac(self, fac_id):
|
||||
return [
|
||||
(gid, g['real_name']) for gid, g in self.groups.items()
|
||||
if g.get('facultet_tech') == fac_id or g.get('facultet_recognized') == fac_id
|
||||
]
|
||||
|
||||
def get_files_by_fac(self, fac_id):
|
||||
files = {} # {uniqpath*: display_name}
|
||||
for g in self.groups.values():
|
||||
if g.get('facultet_tech') == fac_id or g.get('facultet_recognized') == fac_id:
|
||||
for ex in g.get('excels', []):
|
||||
fn = ex['uniqpath'].replace("vstu.ru/rasp?dep=", "")
|
||||
files[fn] = ex['display_filename']
|
||||
return list(files.items())
|
||||
|
||||
def get_excel_by_uniqpath(self, uniqid):
|
||||
for g in self.groups.values():
|
||||
for ex in g.get('excels', []):
|
||||
if uniqid == ex['uniqpath']:
|
||||
return ex
|
||||
|
||||
return None
|
||||
|
||||
def find_group_by_name(self, name):
|
||||
name = name.lower()
|
||||
for gid, g in self.groups.items():
|
||||
if g['real_name'].lower() == name:
|
||||
return gid, g['real_name']
|
||||
return None, None
|
||||
Reference in New Issue
Block a user