Приложения
3. Основные вопросы, подлежащие разработке.
Выпускная квалификационная работа была посвящена разработке и обоснованию проекта автоматизации аналитического учета и формирования управленческой отчетности в образовательной организации. Актуальность выбранной темы обусловлена ростом объема разнородных данных, используемых в управлении образовательным процессом и финансово-экономической деятельностью, а также необходимостью повышения оперативности, достоверности и сопоставимости аналитической информации, предоставляемой руководству и ответственным подразделениям.
В ходе выполнения работы была достигнута поставленная цель – разработан проект автоматизации аналитического учета, включающий информационную, программную и организационную составляющие, а также выполнено экономическое обоснование его внедрения. Для достижения цели последовательно решены задачи, соответствующие структуре выпускной квалификационной работы.
По результатам анализа предметной области, выполненного в первой главе, установлено, что существующая практика формирования аналитической отчетности характеризуется высокой долей ручных операций, фрагментарностью источников данных и отсутствием единого информационного пространства. В ходе исследования было выявлено, что данные, необходимые для принятия управленческих решений, формируются в различных учетных системах (учебных, финансовых, приемных), при этом используются различные справочники и классификаторы, что приводит к расхождениям показателей, дополнительным трудозатратам на сверку информации и снижению доверия к итоговой отчетности.
Также в первой главе определено, что значительная часть аналитических операций выполняется с использованием офисных приложений, что не позволяет обеспечить должный уровень контроля качества данных, историю изменений и воспроизводимость расчетов. Проведенный анализ показал, что действующая организация аналитического учета не в полной мере соответствует требованиям к оперативности и прозрачности управления, особенно в условиях роста контингента обучающихся и увеличения числа образовательных программ.
По итогам первой главы сформулирован вывод о целесообразности внедрения автоматизированной системы аналитического учета, основанной на интеграции данных из существующих информационных систем, централизованном хранилище данных и использовании инструментов бизнес-аналитики для формирования отчетности и ключевых показателей эффективности.
Во второй главе выпускной квалификационной работы была разработана проектная часть, включающая описание архитектуры информационной системы, информационного и программного обеспечения, а также сценариев взаимодействия пользователей с системой. В рамках проектирования была выбрана модель жизненного цикла, соответствующая характеру задачи автоматизации, и обоснован модульный подход к построению программного обеспечения, позволяющий обеспечить масштабируемость и сопровождаемость решения.
В ходе проектирования информационного обеспечения была разработана информационная модель, включающая нормативно-справочную, входную, оперативную и результатную информацию. Определен состав основных справочников, классификаторов и показателей, используемых в аналитических расчетах. Особое внимание уделено обеспечению целостности данных и их сопоставимости при интеграции из различных источников. Разработанная структура базы данных обеспечивает хранение истории изменений и формирование аналитических витрин, что является важным условием для корректного расчета показателей и анализа динамики.
Программное обеспечение задачи спроектировано с учетом разделения функций между модулями интеграции, обработки данных, расчетов и представления информации. Описанные сценарии диалога пользователей отражают реальные бизнес-процессы формирования и анализа отчетности и ориентированы на различные категории пользователей – от аналитиков и сотрудников подразделений до руководителей. Разработанные экранные формы и макеты отчетов демонстрируют возможность оперативного доступа к информации и гибкой настройки аналитических разрезов.
Отдельное внимание во второй главе уделено вопросам качества данных и информационной безопасности. В проекте предусмотрены механизмы контроля загрузок, мониторинга ошибок и журналирования действий пользователей, что позволяет повысить надежность системы и снизить риски принятия управленческих решений на основе некорректной информации. Проведенное описание испытаний разработанного решения показало, что система соответствует функциональным требованиям, обеспечивает корректную интеграцию с источниками данных и готова к этапу опытной эксплуатации.
Таким образом, по результатам второй главы можно сделать вывод, что разработанный проект автоматизации является логически завершенным, технологически реализуемым и соответствует требованиям, предъявляемым к современным информационным системам аналитического учета в образовательных организациях.
В третьей главе выполнено обоснование экономической эффективности проекта. Для этого была выбрана методика, основанная на сравнении базового и проектного вариантов выполнения операций обработки аналитической информации. В расчетах учтены трудовые и стоимостные показатели, а также затраты на создание и внедрение проектируемой системы.
Результаты расчетов показали, что внедрение автоматизированной системы позволяет существенно сократить трудоемкость процессов формирования отчетности за счет уменьшения доли ручного труда и исключения дублирующих операций. Экономический эффект достигается преимущественно за счет снижения затрат на оплату труда специалистов, занятых подготовкой аналитической информации, при незначительном увеличении расходов на машинное время и поддержку программных компонентов.
Расчет срока окупаемости показал, что проект относится к экономически целесообразным: вложения в разработку и внедрение автоматизированной системы компенсируются за счет получаемого экономического эффекта в приемлемые для информационных проектов сроки. Помимо прямого экономического эффекта, в работе обоснован значительный косвенный эффект, выражающийся в повышении качества данных, сокращении времени подготовки отчетов и улучшении информационного обеспечения управленческих решений.
По итогам третьей главы сделан вывод о том, что проект автоматизации аналитического учета является не только технически и организационно обоснованным, но и экономически эффективным, что подтверждает целесообразность его внедрения в практику деятельности образовательной организации.
В целом по результатам выпускной квалификационной работы можно сформулировать следующие обобщающие выводы. Во-первых, автоматизация аналитического учета на основе интеграции данных и использования хранилища данных позволяет перейти от фрагментарной отчетности к единому информационному пространству, обеспечивающему сопоставимость и достоверность показателей. Во-вторых, внедрение проектируемой системы способствует снижению нагрузки на специалистов и повышению прозрачности процессов управления. В-третьих, использование инструментов бизнес-аналитики обеспечивает удобство представления информации и расширяет возможности анализа данных без привлечения дополнительных ресурсов.
На основе полученных результатов могут быть сформулированы практические рекомендации. Рекомендуется внедрение системы на первом этапе в режиме опытной эксплуатации в одном или нескольких подразделениях с последующим масштабированием на всю организацию. Целесообразно также разработать регламенты работы с аналитической системой, включая порядок обновления справочников, контроля качества данных и использования отчетности в управленческой деятельности. Дополнительно рекомендуется рассмотреть возможность расширения функциональности системы за счет подключения новых источников данных и внедрения прогнозной аналитики.
Перспективы дальнейших исследований и развития проекта связаны с углублением аналитических возможностей системы, использованием методов интеллектуального анализа данных, а также автоматизацией поддержки принятия управленческих решений. Дальнейшая работа в данном направлении представляется целесообразной, поскольку развитие цифровых технологий и рост объемов информации требуют постоянного совершенствования инструментов аналитического учета и управления.
Таким образом, выполненная выпускная квалификационная работа является завершенным исследованием, имеет практическую направленность и может быть использована в деятельности образовательной организации при внедрении и развитии автоматизированных систем аналитического учета и управленческой отчетности.
СПИСОК ИСПОЛЬЗОВАННОЙ ЛИТЕРАТУРЫ
Балашов А. И. Экономика информационных систем : учебное пособие для вузов. — М. : Юрайт, 2023. — 256 с.
Бархатова Е. А. Экономическая эффективность автоматизированных информационных систем : учебник. — М. : Инфра-М, 2022. — 312 с.
Бочкарёв А. А. Управление ИТ-проектами : теория и практика. — СПб. : Питер, 2023. — 368 с.
Бухалков М. И. Организация и нормирование труда : учебник. — М. : Инфра-М, 2022. — 416 с.
Виханский О. С. Стратегическое управление : учебник. — М. : Экономистъ, 2021. — 560 с.
Гаврилов Л. П. Информационные технологии в экономике : учебник для вузов. — М. : Юрайт, 2023. — 381 с.
Голубков Е. П. Экономика и управление в цифровой среде. — М. : Юрайт, 2024. — 298 с.
ГОСТ Р 7.0.100–2018. Система стандартов по информации, библиотечному и издательскому делу. Библиографическая запись. Библиографическое описание. Общие требования и правила составления. — М. : Стандартинформ, 2018.
ГОСТ 34.601–90. Информационная технология. Комплекс стандартов на автоматизированные системы. Стадии создания. — М. : Издательство стандартов, 1990.
Дубровский В. Ж. Экономическая эффективность ИТ-проектов. — М. : КНОРУС, 2022. — 240 с.
Ершов А. А. Проектирование корпоративных информационных систем : учебное пособие. — М. : Юрайт, 2023. — 304 с.
Ильин В. В. Экономика автоматизированных систем управления. — М. : Инфра-М, 2021. — 352 с.
Ковалёв В. В. Экономический анализ : учебник. — М. : Проспект, 2022. — 544 с.
Липаев В. В. Надёжность и качество программного обеспечения. — М. : СИНТЕГ, 2021. — 432 с.
Назаров А. А. Информационная безопасность : учебник для вузов. — М. : Юрайт, 2024. — 389 с.
Орлов А. И. Экономические методы оценки эффективности ИТ-решений. — М. : Финансы и статистика, 2022. — 288 с.
Sommerville I. Software Engineering. — 10th ed. — Boston : Pearson, 2021. — 816 p.
Laudon K. C., Laudon J. P. Management Information Systems: Managing the Digital Firm. — 17th ed. — Pearson, 2023. — 640 p.
Pressman R. S., Maxim B. R. Software Engineering: A Practitioner’s Approach. — 9th ed. — New York : McGraw-Hill, 2022. — 976 p.
ISO/IEC 12207:2017. Systems and software engineering — Software life cycle processes. — Geneva : ISO, 2017.
Электронные ресурсы
Министерство науки и высшего образования Российской Федерации : официальный сайт. — URL: https://minobrnauki.gov.ru (дата обращения: 26.12.2025).
Федеральная служба государственной статистики Российской Федерации. — URL: https://rosstat.gov.ru (дата обращения: 26.12.2025).
Gartner Research. Information Systems and Analytics Reports. — URL: https://www.gartner.com (дата обращения: 26.12.2025).
Microsoft Power BI Documentation. — URL: https://learn.microsoft.com/power-bi (дата обращения: 26.12.2025).
Приложение А — Листинги программного кода (полный код проекта)
А.0. Структура проекта
analytics_is/
README.md
requirements.txt
.env.example
docker-compose.yml
sql/
00_create_schema.sql
10_views_bi.sql
app/
__init__.py
config.py
db.py
models.py
etl_sources.py
etl_load.py
kpi_calc.py
api.py
audit.py
utils.py
data/
sample_ref_department.csv
sample_ref_program.csv
sample_ref_student.csv
sample_fact_finance.csv
sample_fact_enrollment.csv
sample_fact_academic.csv
sample_fact_applicants.csv
sample_ref_document.csv
А.1. README.md
# Проектируемая система аналитического учета (ETL → DWH → KPI → BI)
Состав:
- PostgreSQL: аналитическое хранилище (10 таблиц по ER-модели)
- ETL загрузка из файловых выгрузок (как типовой способ интеграции с LMS/1C/CRM/порталом)
- Расчет KPI (в т.ч. KPI_DEBT_AMOUNT: задолженность)
- API (FastAPI) для BI/дашбордов и управленческих отчетов
- Аудит операций расчета и загрузки
## Быстрый запуск (локально)
1) Создать окружение:
python -m venv venv
venv\Scripts\activate (Windows) / source venv/bin/activate (Linux/Mac)
2) Установить зависимости:
pip install -r requirements.txt
3) Подготовить .env:
скопировать .env.example в .env и указать DB_URL
4) Создать БД и таблицы:
psql "postgresql://..." -f sql/00_create_schema.sql
psql "postgresql://..." -f sql/10_views_bi.sql
5) Загрузить тестовые данные:
python -m app.etl_load --load-samples
6) Рассчитать KPI:
python -m app.kpi_calc --period 2025-Осень --user system
7) Запустить API:
uvicorn app.api:app --host 0.0.0.0 --port 8000
API:
- GET /health
- GET /kpi/debt?period=2025-Осень
- GET /kpi/debt/by-program?period=2025-Осень
- GET /students/debt?period=2025-Осень&limit=50
А.2. requirements.txt
fastapi==0.115.0
uvicorn==0.30.6
psycopg2-binary==2.9.9
python-dotenv==1.0.1
pydantic==2.8.2
А.3. .env.example
DB_URL=postgresql://postgres:postgres@localhost:5432/analytics_is
А.4. docker-compose.yml (если хотите запуск через Docker)
services:
db:
image: postgres:16
environment:
POSTGRES_DB: analytics_is
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
volumes:
pgdata:
А.5. SQL: создание схемы и таблиц (10 таблиц по ER)
sql/00_create_schema.sql
-- PostgreSQL DWH schema for university analytics IS
CREATE SCHEMA IF NOT EXISTS dwh;
-- 1) ref_department
CREATE TABLE IF NOT EXISTS dwh.ref_department (
department_id BIGSERIAL PRIMARY KEY,
department_code VARCHAR(20) NOT NULL UNIQUE,
department_name VARCHAR(200) NOT NULL,
department_type VARCHAR(50) NOT NULL,
parent_department_id BIGINT NULL REFERENCES dwh.ref_department(department_id),
valid_from DATE NOT NULL DEFAULT CURRENT_DATE,
valid_to DATE NULL
);
-- 2) ref_program
CREATE TABLE IF NOT EXISTS dwh.ref_program (
program_id BIGSERIAL PRIMARY KEY,
department_id BIGINT NOT NULL REFERENCES dwh.ref_department(department_id),
program_code VARCHAR(30) NOT NULL UNIQUE,
program_name VARCHAR(255) NOT NULL,
education_level VARCHAR(50) NULL,
study_form VARCHAR(30) NULL,
valid_from DATE NOT NULL DEFAULT CURRENT_DATE,
valid_to DATE NULL
);
-- 3) ref_student
CREATE TABLE IF NOT EXISTS dwh.ref_student (
student_id BIGSERIAL PRIMARY KEY,
program_id BIGINT NOT NULL REFERENCES dwh.ref_program(program_id),
fio VARCHAR(255) NOT NULL,
birth_date DATE NULL,
gender CHAR(1) NULL,
study_status VARCHAR(30) NOT NULL,
admission_date DATE NULL,
expulsion_date DATE NULL
);
-- 4) ref_document
CREATE TABLE IF NOT EXISTS dwh.ref_document (
document_id BIGSERIAL PRIMARY KEY,
department_id BIGINT NULL REFERENCES dwh.ref_department(department_id),
doc_type VARCHAR(50) NOT NULL,
doc_number VARCHAR(50) NOT NULL,
doc_date DATE NOT NULL,
description VARCHAR(255) NULL
);
-- 5) fact_enrollment
CREATE TABLE IF NOT EXISTS dwh.fact_enrollment (
enrollment_id BIGSERIAL PRIMARY KEY,
student_id BIGINT NOT NULL REFERENCES dwh.ref_student(student_id),
event_type VARCHAR(30) NOT NULL, -- зачисление/перевод/отчисление
event_date DATE NOT NULL,
period VARCHAR(20) NOT NULL, -- 2025-Осень
group_code VARCHAR(20) NULL,
document_id BIGINT NULL REFERENCES dwh.ref_document(document_id)
);
-- 6) fact_academic
CREATE TABLE IF NOT EXISTS dwh.fact_academic (
academic_id BIGSERIAL PRIMARY KEY,
student_id BIGINT NOT NULL REFERENCES dwh.ref_student(student_id),
discipline_name VARCHAR(200) NOT NULL,
attestation_type VARCHAR(50) NOT NULL,
grade VARCHAR(10) NOT NULL,
attendance_percent NUMERIC(5,2) NULL,
period VARCHAR(20) NOT NULL
);
-- 7) fact_finance
CREATE TABLE IF NOT EXISTS dwh.fact_finance (
finance_id BIGSERIAL PRIMARY KEY,
student_id BIGINT NOT NULL REFERENCES dwh.ref_student(student_id),
contract_id VARCHAR(50) NULL,
period VARCHAR(20) NOT NULL,
charge_amount NUMERIC(12,2) NOT NULL DEFAULT 0,
paid_amount NUMERIC(12,2) NOT NULL DEFAULT 0,
discount_amount NUMERIC(12,2) NOT NULL DEFAULT 0,
debt_amount NUMERIC(12,2) NOT NULL DEFAULT 0
);
-- 8) fact_applicants
CREATE TABLE IF NOT EXISTS dwh.fact_applicants (
applicant_id BIGSERIAL PRIMARY KEY,
program_id BIGINT NOT NULL REFERENCES dwh.ref_program(program_id),
apply_date DATE NOT NULL,
funnel_stage VARCHAR(50) NOT NULL,
status VARCHAR(30) NOT NULL,
source VARCHAR(50) NULL
);
-- 9) agg_kpi
CREATE TABLE IF NOT EXISTS dwh.agg_kpi (
kpi_id BIGSERIAL PRIMARY KEY,
kpi_code VARCHAR(50) NOT NULL,
period VARCHAR(20) NOT NULL,
department_id BIGINT NOT NULL REFERENCES dwh.ref_department(department_id),
program_id BIGINT NULL REFERENCES dwh.ref_program(program_id),
kpi_value NUMERIC(14,4) NOT NULL,
unit VARCHAR(20) NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_agg_kpi_code_period ON dwh.agg_kpi(kpi_code, period);
-- 10) audit_log
CREATE TABLE IF NOT EXISTS dwh.audit_log (
audit_id BIGSERIAL PRIMARY KEY,
event_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
user_id VARCHAR(50) NULL,
action VARCHAR(100) NOT NULL,
object_name VARCHAR(50) NULL,
object_id VARCHAR(50) NULL,
kpi_id BIGINT NULL REFERENCES dwh.agg_kpi(kpi_id),
details VARCHAR(500) NULL
);
CREATE INDEX IF NOT EXISTS idx_audit_time ON dwh.audit_log(event_time);
sql/10_views_bi.sql (BI-представления)
CREATE SCHEMA IF NOT EXISTS bi;
-- Общая задолженность по подразделениям и программам
CREATE OR REPLACE VIEW bi.v_debt_by_program AS
SELECT
k.period,
d.department_name,
p.program_name,
SUM(k.kpi_value) AS total_debt
FROM dwh.agg_kpi k
JOIN dwh.ref_department d ON d.department_id = k.department_id
LEFT JOIN dwh.ref_program p ON p.program_id = k.program_id
WHERE k.kpi_code = 'KPI_DEBT_AMOUNT'
GROUP BY k.period, d.department_name, p.program_name;
-- ТОП студентов по задолженности (на основе fact_finance)
CREATE OR REPLACE VIEW bi.v_students_debt AS
SELECT
f.period,
s.student_id,
s.fio,
p.program_name,
d.department_name,
SUM(f.debt_amount) AS debt_amount
FROM dwh.fact_finance f
JOIN dwh.ref_student s ON s.student_id = f.student_id
JOIN dwh.ref_program p ON p.program_id = s.program_id
JOIN dwh.ref_department d ON d.department_id = p.department_id
GROUP BY f.period, s.student_id, s.fio, p.program_name, d.department_name;
А.6. Python-код приложения (полный)
app/config.py
from dataclasses import dataclass
import os
from dotenv import load_dotenv
load_dotenv()
@dataclass(frozen=True)
class Settings:
db_url: str
def get_settings() -> Settings:
db_url = os.getenv("DB_URL", "").strip()
if not db_url:
raise RuntimeError("DB_URL is not set. Create .env based on .env.example")
return Settings(db_url=db_url)
app/db.py
import psycopg2
from psycopg2.extras import RealDictCursor
from contextlib import contextmanager
from .config import get_settings
@contextmanager
def get_conn():
settings = get_settings()
conn = psycopg2.connect(settings.db_url)
try:
yield conn
finally:
conn.close()
@contextmanager
def get_cursor(conn, dict_rows: bool = False):
cur = conn.cursor(cursor_factory=RealDictCursor if dict_rows else None)
try:
yield cur
finally:
cur.close()
app/utils.py
from decimal import Decimal, ROUND_HALF_UP
def dec2(x) -> Decimal:
if x is None:
return Decimal("0.00")
if isinstance(x, Decimal):
return x.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
return Decimal(str(x)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
def safe_str(x) -> str:
return "" if x is None else str(x).strip()
app/audit.py
from .db import get_cursor
def write_audit(conn, user_id: str | None, action: str, object_name: str | None = None,
object_id: str | None = None, kpi_id: int | None = None, details: str | None = None) -> None:
with get_cursor(conn) as cur:
cur.execute(
"""
INSERT INTO dwh.audit_log(user_id, action, object_name, object_id, kpi_id, details)
VALUES (%s, %s, %s, %s, %s, %s)
""",
(user_id, action, object_name, object_id, kpi_id, details)
)
conn.commit()
app/etl_sources.py (источники данных как выгрузки CSV)
import csv
from pathlib import Path
DATA_DIR = Path(__file__).resolve().parents[1] / "data"
def read_csv(filename: str):
path = DATA_DIR / filename
if not path.exists():
raise FileNotFoundError(f"Missing file: {path}")
with path.open("r", encoding="utf-8", newline="") as f:
reader = csv.DictReader(f, delimiter=",")
for row in reader:
yield {k.strip(): (v.strip() if isinstance(v, str) else v) for k, v in row.items()}
app/etl_load.py (полная загрузка справочников и фактов в DWH)
import argparse
from decimal import Decimal
from .db import get_conn, get_cursor
from .etl_sources import read_csv
from .utils import dec2, safe_str
from .audit import write_audit
def upsert_departments(conn):
rows = list(read_csv("sample_ref_department.csv"))
with get_cursor(conn) as cur:
for r in rows:
cur.execute("""
INSERT INTO dwh.ref_department(department_code, department_name, department_type)
VALUES (%s, %s, %s)
ON CONFLICT (department_code) DO UPDATE
SET department_name = EXCLUDED.department_name,
department_type = EXCLUDED.department_type
""", (safe_str(r["department_code"]), safe_str(r["department_name"]), safe_str(r["department_type"])))
conn.commit()
def upsert_programs(conn):
rows = list(read_csv("sample_ref_program.csv"))
with get_cursor(conn) as cur:
for r in rows:
# find department_id by code
cur.execute("SELECT department_id FROM dwh.ref_department WHERE department_code=%s", (safe_str(r["department_code"]),))
dep = cur.fetchone()
if not dep:
raise RuntimeError(f"Unknown department_code in ref_program: {r['department_code']}")
department_id = dep[0]
cur.execute("""
INSERT INTO dwh.ref_program(department_id, program_code, program_name, education_level, study_form)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (program_code) DO UPDATE
SET department_id = EXCLUDED.department_id,
program_name = EXCLUDED.program_name,
education_level = EXCLUDED.education_level,
study_form = EXCLUDED.study_form
""", (department_id, safe_str(r["program_code"]), safe_str(r["program_name"]),
safe_str(r.get("education_level")), safe_str(r.get("study_form"))))
conn.commit()
def upsert_students(conn):
rows = list(read_csv("sample_ref_student.csv"))
with get_cursor(conn) as cur:
for r in rows:
cur.execute("SELECT program_id FROM dwh.ref_program WHERE program_code=%s", (safe_str(r["program_code"]),))
prog = cur.fetchone()
if not prog:
raise RuntimeError(f"Unknown program_code in ref_student: {r['program_code']}")
program_id = prog[0]
# We treat external_student_id as stable identifier; store it in object_id field? Here: map by fio+birth_date if needed.
# For simplicity in demo, we UPSERT by fio+birth_date (acceptable in учебном примере).
fio = safe_str(r["fio"])
birth_date = r.get("birth_date") or None
study_status = safe_str(r.get("study_status") or "обучается")
cur.execute("""
INSERT INTO dwh.ref_student(program_id, fio, birth_date, gender, study_status, admission_date, expulsion_date)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""", (program_id, fio, birth_date, safe_str(r.get("gender")) or None,
study_status, r.get("admission_date") or None, r.get("expulsion_date") or None))
conn.commit()
def upsert_documents(conn):
rows = list(read_csv("sample_ref_document.csv"))
with get_cursor(conn) as cur:
for r in rows:
dep_code = safe_str(r.get("department_code"))
department_id = None
if dep_code:
cur.execute("SELECT department_id FROM dwh.ref_department WHERE department_code=%s", (dep_code,))
dep = cur.fetchone()
if dep:
department_id = dep[0]
cur.execute("""
INSERT INTO dwh.ref_document(department_id, doc_type, doc_number, doc_date, description)
VALUES (%s, %s, %s, %s, %s)
""", (department_id, safe_str(r["doc_type"]), safe_str(r["doc_number"]), r["doc_date"], safe_str(r.get("description")) or None))
conn.commit()
def load_fact_finance(conn):
rows = list(read_csv("sample_fact_finance.csv"))
with get_cursor(conn) as cur:
for r in rows:
# resolve student_id by fio (demo)
fio = safe_str(r["fio"])
cur.execute("SELECT student_id FROM dwh.ref_student WHERE fio=%s ORDER BY student_id DESC LIMIT 1", (fio,))
st = cur.fetchone()
if not st:
raise RuntimeError(f"Unknown student (fio) in fact_finance: {fio}")
student_id = st[0]
period = safe_str(r["period"])
charge = dec2(r.get("charge_amount"))
paid = dec2(r.get("paid_amount"))
disc = dec2(r.get("discount_amount"))
debt = charge - paid - disc
if debt < Decimal("0.00"):
debt = Decimal("0.00")
cur.execute("""
INSERT INTO dwh.fact_finance(student_id, contract_id, period, charge_amount, paid_amount, discount_amount, debt_amount)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""", (student_id, safe_str(r.get("contract_id")) or None, period, charge, paid, disc, debt))
conn.commit()
def load_fact_enrollment(conn):
rows = list(read_csv("sample_fact_enrollment.csv"))
with get_cursor(conn) as cur:
for r in rows:
fio = safe_str(r["fio"])
cur.execute("SELECT student_id FROM dwh.ref_student WHERE fio=%s ORDER BY student_id DESC LIMIT 1", (fio,))
st = cur.fetchone()
if not st:
raise RuntimeError(f"Unknown student (fio) in fact_enrollment: {fio}")
student_id = st[0]
doc_id = None
doc_number = safe_str(r.get("doc_number"))
if doc_number:
cur.execute("SELECT document_id FROM dwh.ref_document WHERE doc_number=%s ORDER BY document_id DESC LIMIT 1", (doc_number,))
doc = cur.fetchone()
if doc:
doc_id = doc[0]
cur.execute("""
INSERT INTO dwh.fact_enrollment(student_id, event_type, event_date, period, group_code, document_id)
VALUES (%s, %s, %s, %s, %s, %s)
""", (student_id, safe_str(r["event_type"]), r["event_date"], safe_str(r["period"]), safe_str(r.get("group_code")) or None, doc_id))
conn.commit()
def load_fact_academic(conn):
rows = list(read_csv("sample_fact_academic.csv"))
with get_cursor(conn) as cur:
for r in rows:
fio = safe_str(r["fio"])
cur.execute("SELECT student_id FROM dwh.ref_student WHERE fio=%s ORDER BY student_id DESC LIMIT 1", (fio,))
st = cur.fetchone()
if not st:
raise RuntimeError(f"Unknown student (fio) in fact_academic: {fio}")
student_id = st[0]
cur.execute("""
INSERT INTO dwh.fact_academic(student_id, discipline_name, attestation_type, grade, attendance_percent, period)
VALUES (%s, %s, %s, %s, %s, %s)
""", (student_id, safe_str(r["discipline_name"]), safe_str(r["attestation_type"]), safe_str(r["grade"]),
r.get("attendance_percent") or None, safe_str(r["period"])))
conn.commit()
def load_fact_applicants(conn):
rows = list(read_csv("sample_fact_applicants.csv"))
with get_cursor(conn) as cur:
for r in rows:
cur.execute("SELECT program_id FROM dwh.ref_program WHERE program_code=%s", (safe_str(r["program_code"]),))
prog = cur.fetchone()
if not prog:
raise RuntimeError(f"Unknown program_code in fact_applicants: {r['program_code']}")
program_id = prog[0]
cur.execute("""
INSERT INTO dwh.fact_applicants(program_id, apply_date, funnel_stage, status, source)
VALUES (%s, %s, %s, %s, %s)
""", (program_id, r["apply_date"], safe_str(r["funnel_stage"]), safe_str(r["status"]), safe_str(r.get("source")) or None))
conn.commit()
def load_all_samples(conn, user="system"):
upsert_departments(conn)
write_audit(conn, user, "ETL_UPSERT", "ref_department", details="Loaded sample_ref_department.csv")
upsert_programs(conn)
write_audit(conn, user, "ETL_UPSERT", "ref_program", details="Loaded sample_ref_program.csv")
upsert_students(conn)
write_audit(conn, user, "ETL_INSERT", "ref_student", details="Loaded sample_ref_student.csv")
upsert_documents(conn)
write_audit(conn, user, "ETL_INSERT", "ref_document", details="Loaded sample_ref_document.csv")
load_fact_finance(conn)
write_audit(conn, user, "ETL_INSERT", "fact_finance", details="Loaded sample_fact_finance.csv")
load_fact_enrollment(conn)
write_audit(conn, user, "ETL_INSERT", "fact_enrollment", details="Loaded sample_fact_enrollment.csv")
load_fact_academic(conn)
write_audit(conn, user, "ETL_INSERT", "fact_academic", details="Loaded sample_fact_academic.csv")
load_fact_applicants(conn)
write_audit(conn, user, "ETL_INSERT", "fact_applicants", details="Loaded sample_fact_applicants.csv")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--load-samples", action="store_true", help="Load CSV samples from /data")
parser.add_argument("--user", default="system")
args = parser.parse_args()
with get_conn() as conn:
if args.load_samples:
load_all_samples(conn, user=args.user)
print("OK: samples loaded")
if __name__ == "__main__":
main()
app/kpi_calc.py (полный расчет KPI_DEBT_AMOUNT + агрегация по программе/подразделению)
import argparse
from decimal import Decimal
from .db import get_conn, get_cursor
from .audit import write_audit
from .utils import dec2
KPI_DEBT = "KPI_DEBT_AMOUNT"
UNIT_RUB = "RUB"
def calc_kpi_debt_amount(conn, period: str, user: str | None = "system") -> int:
"""
Расчет KPI_DEBT_AMOUNT:
1) берем задолженность по каждому студенту из fact_finance (period)
2) приводим отрицательные значения к 0 (переплата отдельно)
3) агрегируем по программе и подразделению
4) пишем в agg_kpi
Возвращает количество записей KPI, добавленных за период.
"""
with get_cursor(conn) as cur:
# очистка KPI за период (чтобы расчет был идемпотентным)
cur.execute("""
DELETE FROM dwh.agg_kpi
WHERE kpi_code=%s AND period=%s
""", (KPI_DEBT, period))
conn.commit()
write_audit(conn, user, "KPI_CLEAR", "agg_kpi", details=f"Cleared {KPI_DEBT} for period={period}")
with get_cursor(conn) as cur:
# Агрегация задолженности по программе/подразделению
cur.execute("""
WITH student_debt AS (
SELECT
f.student_id,
CASE WHEN SUM(f.debt_amount) < 0 THEN 0 ELSE SUM(f.debt_amount) END AS debt_amount
FROM dwh.fact_finance f
WHERE f.period=%s
GROUP BY f.student_id
)
SELECT
d.department_id,
p.program_id,
SUM(sd.debt_amount) AS total_debt
FROM student_debt sd
JOIN dwh.ref_student s ON s.student_id = sd.student_id
JOIN dwh.ref_program p ON p.program_id = s.program_id
JOIN dwh.ref_department d ON d.department_id = p.department_id
GROUP BY d.department_id, p.program_id
ORDER BY d.department_id, p.program_id
""", (period,))
rows = cur.fetchall()
inserted = 0
with get_cursor(conn) as cur:
for (department_id, program_id, total_debt) in rows:
value = dec2(total_debt)
cur.execute("""
INSERT INTO dwh.agg_kpi(kpi_code, period, department_id, program_id, kpi_value, unit)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING kpi_id
""", (KPI_DEBT, period, department_id, program_id, value, UNIT_RUB))
kpi_id = cur.fetchone()[0]
inserted += 1
write_audit(conn, user, "KPI_INSERT", "agg_kpi", object_id=str(kpi_id), kpi_id=kpi_id,
details=f"{KPI_DEBT} period={period} dep={department_id} prog={program_id} value={value}")
conn.commit()
return inserted
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--period", required=True, help="e.g. 2025-Осень")
parser.add_argument("--user", default="system")
args = parser.parse_args()
with get_conn() as conn:
n = calc_kpi_debt_amount(conn, args.period, args.user)
print(f"OK: inserted {n} KPI rows for period={args.period}")
if __name__ == "__main__":
main()
app/api.py (полное API для BI/дашбордов)
from fastapi import FastAPI, Query
from .db import get_conn, get_cursor
app = FastAPI(title="University Analytics IS API", version="1.0")
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/kpi/debt")
def kpi_debt(period: str = Query(..., description="Например: 2025-Осень")):
with get_conn() as conn:
with get_cursor(conn, dict_rows=True) as cur:
cur.execute("""
SELECT period, SUM(kpi_value) AS total_debt
FROM dwh.agg_kpi
WHERE kpi_code='KPI_DEBT_AMOUNT' AND period=%s
GROUP BY period
""", (period,))
row = cur.fetchone()
return row or {"period": period, "total_debt": 0}
@app.get("/kpi/debt/by-program")
def kpi_debt_by_program(period: str = Query(...)):
with get_conn() as conn:
with get_cursor(conn, dict_rows=True) as cur:
cur.execute("""
SELECT
d.department_name,
p.program_name,
SUM(k.kpi_value) AS total_debt
FROM dwh.agg_kpi k
JOIN dwh.ref_department d ON d.department_id=k.department_id
JOIN dwh.ref_program p ON p.program_id=k.program_id
WHERE k.kpi_code='KPI_DEBT_AMOUNT' AND k.period=%s
GROUP BY d.department_name, p.program_name
ORDER BY d.department_name, p.program_name
""", (period,))
return cur.fetchall()
@app.get("/students/debt")
def students_debt(period: str = Query(...), limit: int = Query(50, ge=1, le=500)):
with get_conn() as conn:
with get_cursor(conn, dict_rows=True) as cur:
cur.execute("""
SELECT *
FROM bi.v_students_debt
WHERE period=%s
ORDER BY debt_amount DESC
LIMIT %s
""", (period, limit))
return cur.fetchall()
Выпускная квалификационная работа выполнена мной совершенно самостоятельно. Все использованные в работе материалы и концепции из опубликованной литературы и других источников имеют ссылки на них. Выпускная квалификационная работа прошла проверку на корректность заимствования в системе «Антиплагиат.ВУЗ».
Настоящим подтверждаю, что даю разрешение Университету «Синергия» на размещение полного текста моей выпускной квалификационной работы и отзыва о работе в период ее подготовки в электронно-библиотечной системе Университета «Синергия».
«___» ______________ 20__ г.
