python-fastapi-mm-zhcs-yj-api/routers/api/eventManagement/checkin.py

from fastapi import APIRouter, Request, Depends, HTTPException, Query
from sqlalchemy.exc import IntegrityError
from fastapi.responses import HTMLResponse, FileResponse, StreamingResponse, RedirectResponse
from fastapi.responses import JSONResponse
from database import get_db
from sqlalchemy import text, exists, and_, or_, not_
from sqlalchemy.orm import Session
from sqlalchemy.sql import func
from models import *
import json
import random
from sqlalchemy import create_engine, select
from typing import Optional
from utils.StripTagsHTMLParser import *
from common.db import db_event_management, db_user, db_area, db_emergency_plan
from common.security import valid_access_token
import traceback
from utils import *
from datetime import datetime, timedelta
import qrcode
from PIL import Image
from io import BytesIO
from config import settings
from common import YzyApi
from extensions import logger

router = APIRouter()

@router.get('/qrcode', response_class=StreamingResponse)
async def get_qrcode(
    request: Request,
    event_id: str,
    db: Session = Depends(get_db)
):
    url = str(request.url)
    url = url.replace("qrcode", "qrcode2") + "&r="+rnd()

    # if settings.IS_STAGE:
    #     url = url.replace(request.url.hostname, "120.241.74.139:8086")
    #    logger.info('替换到互联网ip: {}', url)   

    qr = qrcode.QRCode(
        version=1,
        error_correction=qrcode.constants.ERROR_CORRECT_M,
        box_size=10,
        border=0,
    )
    qr.add_data(url)
    qr.make(fit=True)
    image = qr.make_image()

    buf = BytesIO()
    image.save(buf, 'png')
    img_data = buf.getvalue()
    return StreamingResponse(BytesIO(img_data), media_type="image/png")

@router.get('/qrcode2')
async def get_qrcode2(
    event_id: str,
    db: Session = Depends(get_db)
):    
    redirect_url = "/signPage?event_id={}".format(event_id) # 业务页面
    logger.info("redirect_url: {}", redirect_url)
    # qrcode_str = f"http://19.155.220.209/api/event_management/event?event_id={event_id}"
    detail_url = YzyApi.format_redirect_url(redirect_url)

    return RedirectResponse(detail_url)

@router.post("/getInfo")
async def getInfo(
    request: Request, 
    db: Session = Depends(get_db), 
    body = Depends(remove_xss_json), 
    user_id = Depends(valid_access_token)
):
    event_id = body['event_id']

    row = db.query(SysUser).filter(SysUser.user_id == user_id).first()
    user_info = get_model_dict(row)

    yzy_account = user_info['yzy_account']

    contact_info = db.query(EmergencyContactInfo).filter(and_(EmergencyContactInfo.del_flag == "0", EmergencyContactInfo.yue_gov_ease_phone == yzy_account)).first()
    contact_info = get_model_dict(contact_info)
    nick_name = contact_info['contact_name']
    dept_id = contact_info['unit_id']
    dept_name = contact_info['unit_name']
    duties = contact_info['position']

    # row = db.query(SysDept).filter(SysDept.dept_id == user_info['dept_id']).first()
    # dept_info = get_model_dict(row)

    user_name = user_info['user_name']
    # nick_name = user_info['nick_name']
    # dept_id = user_info['dept_id']
    # dept_name = dept_info['dept_name']
    # duties = ''
    sign_time = ''
    phone = user_info['phonenumber']

    row = db.query(EventCheckin).filter(and_(EventCheckin.event_id == event_id, EventCheckin.phone == phone, EventCheckin.del_flag == '0')).first()
    if row is not None:
        check_info = get_model_dict(row)
        user_name = check_info['user_name']
        nick_name = check_info['nick_name']
        dept_id = check_info['dept_id']
        dept_name = check_info['dept_name']
        duties = check_info['duties']
        sign_time = get_datetime_str(check_info['sign_time'])
        phone = check_info['phone']
        print('---------------', sign_time)

    return {
        'code': 200,
        'msg': '查询成功',
        'data': {
            'user_id': user_info['user_id'],
            'user_name': user_name,
            'nick_name': nick_name,
            'dept_id': dept_id,
            'dept_name': dept_name,
            'duties': duties,
            'sign_time': sign_time,
            'phone': phone
        }
    }


@router.post("/check")
async def check(
    request: Request, 
    db: Session = Depends(get_db), 
    body = Depends(remove_xss_json), 
    # user_id = Depends(valid_access_token)
):
    time.sleep(2.0)
    
    event_id = body['event_id']
    nick_name = body['nick_name']
    dept_name = body['dept_name']
    phone = body['phone']
    duties = body['duties']
    type_ = body['type']

    dept_id = 0
    yzy_account = ''
    contact_info = db.query(EmergencyContactInfo).filter(and_(EmergencyContactInfo.del_flag == "0", EmergencyContactInfo.yue_gov_ease_phone == phone)).first()
    if contact_info is not None:
        yzy_account = phone
        contact_info = get_model_dict(contact_info)
        dept_id = contact_info['unit_id']

    if type_ == '1':
        # 签名
        row = db.query(EventCheckin).filter(and_(EventCheckin.event_id == event_id, EventCheckin.phone == phone)).first()
        if row is None:
            event_checkin = EventCheckin(
                event_id = event_id,
                user_id = 0,
                user_name = '',
                nick_name = nick_name,
                dept_id = dept_id,
                dept_name = dept_name,
                sign_time = datetime.now(),
                yzy_account = yzy_account,
                duties = duties,
                phone = phone,
                del_flag = '0'
            )
            db.add(event_checkin)
            db.commit()
        else:
            row.sign_time = datetime.now()
            row.nick_name = nick_name
            row.dept_name = dept_name
            row.duties = duties
            row.phone = phone
            row.del_flag = '0'
            db.commit()

        return {
            'code': 200,
            'msg': '签到成功'
        }

    elif type_ == '2':
        # 取消签名
        row = db.query(EventCheckin).filter(and_(EventCheckin.event_id == event_id, EventCheckin.phone == phone)).first()
        if row is None:
            return {
                'code': 500,
                'msg': '用户并未签名，无法签退'
            }
        
        row.sign_time = datetime.now()
        row.del_flag = '1'
        db.commit()

        return {
            'code': 200,
            'msg': '签退成功'
        }
    
@router.get('/list')
async def get_event_list(
    event_id: str,
    page: int = Query(1, gt=0, description='页码'),
    page_size: int = Query(10, gt=0, description='pageSize'),
    db: Session = Depends(get_db)
):
    try:
        # 应用查询条件
        where = and_(EventCheckin.del_flag == '0', EventCheckin.event_id == event_id)
        # 计算总条目数
        q = db.query(func.count(EventCheckin.id))
        q = q.filter(where)
        total = q.scalar()
        
        # 执行分页查询
        q = db.query(EventCheckin)
        q = q.filter(where)
        rows = q.order_by(EventCheckin.id.desc()).offset((page - 1) * page_size).limit(page_size).all()
        data = [
            {
                "event_id": row.event_id,
                "user_id": row.user_id,
                "user_name": row.user_name,
                "nick_name": row.nick_name,
                "dept_id": row.dept_id,
                "dept_name": row.dept_name,
                "sign_time": get_datetime_str(row.sign_time),
                "yzy_account": row.yzy_account,
                "duties": row.duties,
                "phone": row.phone
            }
            for row in rows
        ]

        # 返回结果
        return {
            "code": 200,
            "msg": "查询成功",
            "data": data,
            "total": total
        }

    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))