135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
from sqlalchemy.orm import Session
|
|
from typing import Dict, Any, List
|
|
from app.repositories.submission_repository import SubmissionRepository
|
|
from app.repositories.form_repository import FormRepository
|
|
|
|
|
|
class AnalyticsService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
self.submission_repo = SubmissionRepository(db)
|
|
self.form_repo = FormRepository(db)
|
|
|
|
def get_form_report(self, form_id: int) -> Dict[str, Any]:
|
|
"""Полный отчет по форме"""
|
|
form = self.form_repo.get_by_id(form_id)
|
|
if not form:
|
|
raise ValueError("Form not found")
|
|
|
|
# Базовая статистика
|
|
total_submissions = self.submission_repo.count(form_id=form_id)
|
|
|
|
# Аналитика по каждому полю
|
|
fields_analytics = {}
|
|
for form_field in form.fields:
|
|
field = form_field.field
|
|
stats = self.submission_repo.get_field_statistics(form_id, field.name)
|
|
|
|
fields_analytics[field.name] = {
|
|
"label": field.label,
|
|
"type": field.field_type,
|
|
"completion_rate": self._calculate_completion_rate(form_id, field.name),
|
|
"unique_values": len(stats),
|
|
"distribution": stats,
|
|
"null_count": self._get_null_count(form_id, field.name)
|
|
}
|
|
|
|
# Временная аналитика
|
|
daily_stats = self.submission_repo.get_daily_submissions(form_id)
|
|
|
|
return {
|
|
"form_name": form.name,
|
|
"total_submissions": total_submissions,
|
|
"fields_analytics": fields_analytics,
|
|
"daily_statistics": daily_stats,
|
|
"peak_hours": self._get_peak_hours(form_id),
|
|
"submission_trend": self._calculate_trend(daily_stats)
|
|
}
|
|
|
|
def _calculate_completion_rate(self, form_id: int, field_name: str) -> float:
|
|
"""Процент заполнения поля"""
|
|
total = self.submission_repo.count(form_id=form_id)
|
|
if total == 0:
|
|
return 0.0
|
|
|
|
filled = self.submission_repo.count(
|
|
form_id=form_id,
|
|
filters={f"data->>'{field_name}'": "is not null"}
|
|
)
|
|
return (filled / total) * 100
|
|
|
|
def _get_null_count(self, form_id: int, field_name: str) -> int:
|
|
"""Количество пустых значений в поле"""
|
|
return self.submission_repo.count(
|
|
form_id=form_id,
|
|
filters={f"data->>'{field_name}'": "is null"}
|
|
)# app/services/analytics_service.py (базовая версия)
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func
|
|
from datetime import datetime, timedelta
|
|
from app.models.submission import Submission
|
|
from app.models.form import Form
|
|
|
|
class AnalyticsService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def get_form_report(self, form_id: int, days: int = 30):
|
|
form = self.db.query(Form).filter(Form.id == form_id).first()
|
|
if not form:
|
|
raise ValueError("Form not found")
|
|
|
|
start_date = datetime.utcnow() - timedelta(days=days)
|
|
submissions = self.db.query(Submission).filter(
|
|
Submission.form_id == form_id,
|
|
Submission.submitted_at >= start_date
|
|
).all()
|
|
|
|
return {
|
|
"form_id": form_id,
|
|
"form_name": form.name,
|
|
"total_submissions": len(submissions),
|
|
"unique_submitters": len(set(s.submitted_by for s in submissions if s.submitted_by)),
|
|
"avg_completion_time": None,
|
|
"completion_rate": 100.0,
|
|
"fields_stats": [],
|
|
"daily_stats": [],
|
|
"peak_hours": {},
|
|
"trend": "stable",
|
|
"last_submission_at": submissions[-1].submitted_at if submissions else None,
|
|
"first_submission_at": submissions[0].submitted_at if submissions else None
|
|
}
|
|
|
|
def get_field_statistics(self, form_id: int, field_name: str):
|
|
return None
|
|
|
|
def get_global_overview(self, days: int):
|
|
return {"total_forms": 0, "total_submissions": 0, "active_forms": 0}
|
|
|
|
def _get_peak_hours(self, form_id: int) -> Dict[str, int]:
|
|
"""Часы пик заполнения"""
|
|
from sqlalchemy import func
|
|
|
|
results = self.db.query(
|
|
func.extract('hour', Submission.submitted_at).label('hour'),
|
|
func.count(Submission.id).label('count')
|
|
).filter(
|
|
Submission.form_id == form_id
|
|
).group_by('hour').all()
|
|
|
|
return {int(r.hour): r.count for r in results}
|
|
|
|
def _calculate_trend(self, daily_stats: List[Dict]) -> str:
|
|
"""Расчет тренда (рост/падение/стабильно)"""
|
|
if len(daily_stats) < 2:
|
|
return "insufficient_data"
|
|
|
|
recent_avg = sum(d['count'] for d in daily_stats[-7:]) / 7
|
|
previous_avg = sum(d['count'] for d in daily_stats[-14:-7]) / 7
|
|
|
|
if recent_avg > previous_avg * 1.1:
|
|
return "increasing"
|
|
elif recent_avg < previous_avg * 0.9:
|
|
return "decreasing"
|
|
else:
|
|
return "stable" |