mirror of
https://github.com/Faridik/Carolyn.git
synced 2026-07-08 13:21:53 +00:00
132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
import os.path
|
||
from googleapiclient.discovery import _MEDIA_SIZE_BIT_SHIFTS, build
|
||
from google_auth_oauthlib.flow import InstalledAppFlow
|
||
from google.auth.transport.requests import Request
|
||
from google.oauth2.credentials import Credentials
|
||
|
||
from models.student import Student
|
||
from models.group import Group
|
||
from models.group import StudentNotFound
|
||
from models.assignment import Assignment
|
||
|
||
STUDENT_NUMBERS = 0
|
||
STUDENT_NAMES = 1
|
||
GROUP_IDS = 2 # for StudentList and AssignmentList
|
||
TELEGRAM_IDS = 3
|
||
ASSIGNMENT_NAMES = 1
|
||
ASSIGNMENT_SUBJECT = 3
|
||
ASSIGNMENT_WEIGHTS = 4
|
||
ASSIGNMENT_RANGES = 5
|
||
ASSIGNMENT_ALLOWS = 6
|
||
|
||
class Manager:
|
||
"""Позволяет выполнять операции с Google таблицами."""
|
||
|
||
def __init__(self, spreadsheet_id: str='1GKfmLwDVGjcpXFQAUyxM4RMQ5Nx6HAZ5VWenz2z--d0',
|
||
scopes: list=['https://www.googleapis.com/auth/spreadsheets.readonly']):
|
||
|
||
self.spreadsheet_id = spreadsheet_id
|
||
|
||
creds = None
|
||
|
||
if os.path.exists('token.json'):
|
||
creds = Credentials.from_authorized_user_file('token.json', scopes)
|
||
# If there are no (valid) credentials available, let the user log in.
|
||
if not creds or not creds.valid:
|
||
if creds and creds.expired and creds.refresh_token:
|
||
creds.refresh(Request())
|
||
else:
|
||
flow = InstalledAppFlow.from_client_secrets_file(
|
||
'client_secret.json', scopes)
|
||
creds = flow.run_local_server(port=0)
|
||
# Save the credentials for the next run
|
||
with open('token.json', 'w') as token:
|
||
token.write(creds.to_json())
|
||
|
||
service = build('sheets', 'v4', credentials=creds)
|
||
self.sheet = service.spreadsheets()
|
||
|
||
|
||
|
||
|
||
def get_values(self, range_name: str = 'StudentList') -> list: # TODO: change subject
|
||
"""Получить значения с таблицы."""
|
||
|
||
result = self.sheet.values().get(spreadsheetId=self.spreadsheet_id,
|
||
range=range_name).execute()
|
||
|
||
return result.get('values', [])
|
||
|
||
|
||
|
||
|
||
def get_group(self, group_id: int) -> Group:
|
||
"""Получить группу со студентами."""
|
||
|
||
values = self.get_values()
|
||
|
||
group = Group(group_id)
|
||
|
||
f = lambda row: group_id == int(row[GROUP_IDS])
|
||
for row in filter(f, values):
|
||
group.add_student(Student(int(row[STUDENT_NUMBERS]), row[STUDENT_NAMES],
|
||
group_id, row[TELEGRAM_IDS]))
|
||
|
||
return group
|
||
|
||
|
||
|
||
|
||
def get_student_by_name(self, name: str) -> Student:
|
||
"""Получить студента по имени."""
|
||
|
||
values = self.get_values()
|
||
|
||
f = lambda row: name == row[STUDENT_NAMES]
|
||
try:
|
||
row = next(filter(f, values))
|
||
return Student(int(row[STUDENT_NUMBERS]), name,
|
||
row[GROUP_IDS], row[TELEGRAM_IDS])
|
||
except StopIteration:
|
||
raise StudentNotFound("Не могу найти студента по имени")
|
||
|
||
|
||
|
||
|
||
def get_student_by_tg_id(self, tg_id: int) -> Student:
|
||
"""Получить студента по telegram id."""
|
||
|
||
values = self.get_values()
|
||
|
||
f = lambda row: tg_id == row[TELEGRAM_IDS]
|
||
try:
|
||
row = next(filter(f, values))
|
||
return Student(int(row[STUDENT_NUMBERS]), row[STUDENT_NAMES],
|
||
row[GROUP_IDS], tg_id)
|
||
except StopIteration:
|
||
raise StudentNotFound("Не могу найти студента по telegram id")
|
||
|
||
|
||
|
||
|
||
def set_assignments_for_student(self, student: Student):
|
||
"""Добавить задания студенту."""
|
||
|
||
values = self.get_values('AssignmentList')
|
||
|
||
f = lambda row: student.group_id == int(row[GROUP_IDS]) # TODO add subject
|
||
for row in filter(f,values):
|
||
assignment_value = self.get_values(row[ASSIGNMENT_RANGES])
|
||
ass_values = list(map(float, assignment_value[student.number]))
|
||
student.add_assignment(Assignment(row[ASSIGNMENT_NAMES], ass_values,
|
||
row[ASSIGNMENT_WEIGHTS]))
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|