xzbautomarkapi/admin/api/endpoints/homepage/statistics.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import datetime
from collections import defaultdict

from fastapi import Depends
from sqlalchemy import select, func, desc
from sqlalchemy.ext.asyncio import AsyncSession

from crud.marktask import crud_task
from crud.resource import crud_work_resource, crud_exam_resource
from crud.school import crud_school, crud_class
from crud.sysdata.region import crud_region
from crud.user import crud_teacher, crud_student
from models.marktask import MarkTask
from models.resource import WorkResource, ExamResource
from models.sysdata.region import Region
from models.user import Admin
from utils.depends import get_async_db, get_current_user


async def statistic_resource(db: AsyncSession = Depends(get_async_db),
                             current_user: Admin = Depends(get_current_user)):
    # 作业资源统计
    stmt = select(WorkResource.subject, func.count()).select_from(WorkResource)\
        .group_by(WorkResource.subject)
    total_work = 0
    work_resource_dist = []
    for x in await crud_work_resource.execute_v2(db, stmt):
        total_work += x[1]
        temp = {"sub": x[0], "count": x[1]}
        work_resource_dist.append(temp)

    # 考试资源统计
    stmt = select(ExamResource.subject, func.count()).select_from(ExamResource)\
        .group_by(ExamResource.subject)
    total_exam = 0
    exam_resource_dist = []
    for x in await crud_work_resource.execute_v2(db, stmt):
        total_exam += x[1]
        temp = {"sub": x[0], "count": x[1]}
        exam_resource_dist.append(temp)

    # 最终结果
    data = {
        "all": {
            "total": total_work + total_exam,
            "work": total_work,
            "exam": total_exam
        },
        "work_dist": work_resource_dist,
        "exam_dist": exam_resource_dist
    }

    return {"data": data}


async def resource_date_dist(db: AsyncSession = Depends(get_async_db),
                             current_user: Admin = Depends(get_current_user)):
    today = datetime.date.today()
    start_date = datetime.datetime(today.year, 1, 1, 0, 0, 0)
    temp = defaultdict(int)
    for x in [(WorkResource, crud_work_resource), (ExamResource, crud_exam_resource)]:
        stmt = select(
            func.concat(func.year(x[0].created_at), ".",
                        func.month(x[0].created_at)).label("month"),
            func.count(),
        ).select_from(x[0]).where(x[0].created_at >= start_date).group_by("month")
        for y in await x[1].execute_v2(db, stmt):
            temp[y[0]] += y[1]

    data = [{"date": k, "count": v} for k, v in temp.items()]
    data.sort(key=lambda z: z["date"])
    return {"data": data}


async def statistic_marktask(db: AsyncSession = Depends(get_async_db),
                             current_user: Admin = Depends(get_current_user)):
    data = {"work": 0, "exam": 0, "completed": 0, "ongoing": 0, "total": 0}
    # 按任务类型统计
    stmt = select(MarkTask.mtype, func.count()).select_from(MarkTask).group_by(MarkTask.mtype)
    for x in await crud_task.execute_v2(db, stmt):
        data[x[0]] += x[1]
        data["total"] += x[1]

    # 按批阅进度统计
    stmt = select(MarkTask.status, func.count()).select_from(MarkTask).where(MarkTask.status > 0)\
        .group_by(MarkTask.status)

    for item in await crud_task.execute_v2(db, stmt):
        if item[0] == 2:
            data["completed"] += item[1]
        else:
            data["ongoing"] += item[1]
    return {"data": data}


async def get_top10_school_by_task(db: AsyncSession = Depends(get_async_db),
                                   current_user: Admin = Depends(get_current_user)):
    stmt = select(MarkTask.school_name, func.count().label("school_count"))\
        .select_from(MarkTask).group_by(MarkTask.school_name)\
        .order_by(desc("school_count")).limit(10)
    data = [{"key": x[0], "val": x[1]} for x in await crud_task.execute_v2(db, stmt)]
    return {"data": data}


async def get_top10_subject_by_task(db: AsyncSession = Depends(get_async_db),
                                    current_user: Admin = Depends(get_current_user)):
    stmt = select(MarkTask.subject, func.count().label("subject_count")).select_from(MarkTask)\
        .group_by(MarkTask.subject).order_by(desc("subject_count")).limit(10)
    data = [{"key": x[0], "val": x[1]} for x in await crud_task.execute_v2(db, stmt)]
    return {"data": data}


async def statistic_school(db: AsyncSession = Depends(get_async_db),
                           current_user: Admin = Depends(get_current_user)):
    school_count = await crud_school.count(db)
    class_count = await crud_class.count(db)
    teacher_count = await crud_teacher.count(db)
    student_count = await crud_student.count(db)
    data = {
        "school": school_count,
        "class": class_count,
        "teacher": teacher_count,
        "student": student_count
    }
    return {"data": data}


async def school_distribute(db: AsyncSession = Depends(get_async_db),
                            current_user: Admin = Depends(get_current_user)):
    # 查询学校
    _, db_schools = await crud_school.find_all(db)
    temp = defaultdict(list)
    area_codes = []
    for item in db_schools:
        area = item.area_code[14:]
        temp[area].append({"id": item.id, "name": item.name, "area": item.area_name})
        area_codes.append(area)

    # 根据地区查询经纬度
    if area_codes:
        _, db_regions = await crud_region.find_all(db, filters=[Region.code.in_(area_codes)])
        for region in db_regions:
            for school in temp[region.code]:
                school["position"] = (region.lng, region.lat)

    data = []
    for item in temp.values():
        data.extend(item)
    return {"data": data}