from fastapi import FastAPI, HTTPException, Depends, APIRouter,Query,Body from sqlalchemy.engine.reflection import Inspector from common.security import valid_access_token from pydantic import BaseModel,Extra,Field from datetime import datetime from typing import List, Optional,Any,Dict from sqlalchemy import create_engine, Column, Integer, String, Boolean, MetaData, Table, \ inspect, exists,or_,text,insert,asc,desc from sqlalchemy.orm import Session from pypinyin import lazy_pinyin, Style from database import get_db from models import * import random import pandas as pd from sqlalchemy import text from fastapi.responses import StreamingResponse from io import BytesIO router = APIRouter() metadata = MetaData() class ReportField(BaseModel): name: str def get_next_event_id(db: Session): while True: random_10_digit_number = random.randint(1000000000, 9999999999) reportId = 'report' + str(random_10_digit_number) it_exists = db.query( exists().where(ReportManagement.report_id == reportId) ).scalar() if it_exists == False: return reportId # # class ReportQuery(BaseModel): # report_id: str class TableStructure(BaseModel): column_name: str comment: str # 表数据模型 class TableData(BaseModel): row_data: dict #详情 字段校验 @router.get("/report_structure/{report_id}") async def get_report_structure( report_id: str, db: Session = Depends(get_db), creator_id = Depends(valid_access_token) ): # 查询 ReportManagement 表以获取 data_table_name report = db.query(ReportManagement).filter( ReportManagement.report_id == report_id, ReportManagement.creator_id == creator_id ).first() if not report: return {"code": 404, "msg": "Report not found"} # raise HTTPException(status_code=404, detail="Report not found") data_table_name = report.data_table_name # 查询对应表的表结构 table_structure_query = db.execute( text(""" SELECT COLUMN_NAME, COLUMN_COMMENT, ORDINAL_POSITION FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = :table_name AND TABLE_SCHEMA = (SELECT DATABASE()) ORDER BY ORDINAL_POSITION """), {"table_name": data_table_name} ) table_structures = [] column_order = [] # 用于存储字段的顺序 for row in table_structure_query.fetchall(): if row[0] not in ['collect_status', 'create_id', 'id', 'user_id', 'add_time','temporarily_store']: table_structures.append(TableStructure(column_name=row[0], comment=row[1])) column_order.append(row[0]) # 保存字段顺序 # 查询表中的数据，排除指定字段 excluded_columns = ['collect_status', 'create_id', 'id', 'user_id', 'add_time','temporarily_store'] columns_to_select = ", ".join(column_order) # 使用字段顺序 table_data_with_headers = [] # print("字段：",columns_to_select) # print(len(columns_to_select)) # table_data_query = if (len(columns_to_select)) != 0: table_data_query = db.execute( text(f"SELECT {columns_to_select} FROM {data_table_name}") ) # 将查询结果转换为字典列表 table_data = [dict(zip(column_order, row)) for row in table_data_query.fetchall()] # 构造表头的字段名和字段注释作为第一行 table_headers = {col.column_name: col.comment for col in table_structures} # 在表数据中添加表头备注作为第一行 table_data_with_headers = [ table_headers # 表头备注 ] + table_data # 表数据 # 查询已提交和未提交的用户状态，并获取用户昵称 user_submission_status = [] # 查询所有相关用户 users = db.query(FormSubmission.user_id).filter( FormSubmission.report_id == report_id ).distinct() user_ids = [user[0] for user in users.all()] reported = db.query(FormSubmission).filter( FormSubmission.report_id == report_id, FormSubmission.submission_status == 1 ).all() num_reported = len(reported) reported_user_ids = [submission.user_id for submission in reported] unreported = db.query(FormSubmission).filter( FormSubmission.report_id == report_id, FormSubmission.submission_status == 0 ).all() # print(unreported) num_unreported = len(unreported) # 查询每个用户的提交状态和昵称 for user_id in user_ids: user = db.query(SysUser.nick_name).filter(SysUser.user_id == user_id).first() if user: nick_name = user[0] else: nick_name = "未知用户" submission_status = db.query(FormSubmission.submission_status).filter( FormSubmission.report_id == report_id, FormSubmission.user_id == user_id ).first() if submission_status: submission_status = str(submission_status[0]) else: submission_status = "0" # 默认为未提交 user_submission_status.append({ "name": nick_name, "submission_status": submission_status, "user_id":user_id }) # 构造返回结果 result = { "code": 200, "msg": "查询成功", "report_info": { "id": report.id, "report_id": report.report_id, "table_name": report.table_name, "data_table_name": report.data_table_name, "start_time": report.start_time, "end_time": report.end_time, "status": report.status, "issued_status": report.issued_status, "period_type": report.period_type, "creator_name": report.creator_name, "num_reporters": len(user_ids), "creator_id": creator_id, "created_at": report.created_at, "updated_at": report.updated_at, "num_reported": num_reported, "num_unreported": num_unreported, "creator_phone": report.creator_phone, "user_filling_status": user_submission_status, # 用户提交状态列表 "user_ids": user_ids }, "table_structure": table_structures, "table_data": table_data_with_headers # 添加表数据 } return result # 函数用于将中文转换为拼音首字母缩写 def to_first_letter(chinese_str: str) -> str: return ''.join([p[0][0] for p in lazy_pinyin(chinese_str, style=Style.FIRST_LETTER)]).lower() # 动态创建表 def create_dynamic_table(table_name: str, field_names: List[str], db: Session): metadata = MetaData() inspector = Inspector.from_engine(db.bind) # 检查表是否已存在 if inspector.has_table(table_name): return {"code": 500, "msg": "表已存在"} table = Table(table_name, metadata, Column('id', Integer, primary_key=True, comment="主键ID"), Column('user_id', Integer, comment="用户ID"), Column('create_id', Integer, comment="创建者ID"), Column('collect_status', Boolean, comment="收取结果"), Column('add_time', DateTime, server_default=func.now(), comment="添加时间"), Column('temporarily_store', Integer, server_default='0', comment="暂存状态"), extend_existing=True ) # 为每个字段动态添加列 for i, field_name in enumerate(field_names, start=1): column_name = f"col{i}" table.append_column(Column(column_name, String(255), comment=field_name)) # 创建表 try: metadata.create_all(bind=db.bind) except Exception as e: db.rollback() return {"code": 500, "msg": "创建表失败"} return {"code": 200, "msg": "表创建成功"} ####表名必填，其它非必填 class ReportCreate(BaseModel): table_name: str = Field(..., description="表单名称，必填") end_time: Optional[str] = Field(None, description="结束时间，非必填，格式为 ISO8601") issued_status: str = Field(..., description="发布状态，必填") creator_name: Optional[str] = Field(None, description="创建者姓名，非必填") creator_phone: Optional[str] = Field(None, description="创建者电话，非必填") user_ids: Optional[List[int]] = Field(None, description="用户 ID 列表，非必填") period_type: Optional[str] = Field(None, description="周期，非必填") field_names: Optional[List[str]] = Field(None, description="字段名称列表，非必填") @router.post("/report/") def create_report_and_table(report: ReportCreate, db: Session = Depends(get_db), creator_id=Depends(valid_access_token)): try: # 验证 table_name 是否必填 if not report.table_name: return {"code": 400, "msg": "表单名称（table_name）是必填项"} # 获取当前时间并格式化为 YYYYMMDDHHMMSS current_time_str = datetime.now().strftime("%Y%m%d%H%M%S") # 动态生成 data_table_name table_name_pinyin = ''.join(lazy_pinyin(report.table_name, style=Style.FIRST_LETTER)).lower() data_table_name = f"tbxt_{table_name_pinyin}_{current_time_str}" # 根据 issued_status 判断必填字段 if report.issued_status == "2": # 发布状态 # 验证发布状态下必填字段 if not all([report.end_time, report.creator_name, report.creator_phone, report.user_ids, report.field_names]): missing_fields = [] if not report.end_time: missing_fields.append("end_time") if not report.creator_name: missing_fields.append("creator_name") if not report.creator_phone: missing_fields.append("creator_phone") if not report.user_ids: missing_fields.append("user_ids") if not report.field_names: missing_fields.append("field_names") return {"code": 400, "msg": f"发布状态下，以下字段为必填项：{', '.join(missing_fields)}"} elif report.issued_status != "1": # 如果 issued_status 不是 1 或 2，返回错误 return {"code": 400, "msg": "issued_status 必须为 1（待发布）或 2（已发布）"} # 登记填报管理 new_report = ReportManagement( report_id=get_next_event_id(db), table_name=report.table_name, data_table_name=data_table_name, start_time=datetime.now(), end_time=datetime.fromisoformat(report.end_time) if report.end_time else None, status=0, issued_status=report.issued_status, collection_status=0, # 未收取 period_type=report.period_type, creator_name=report.creator_name, creator_id=creator_id, creator_phone=report.creator_phone, num_reporters=len(report.user_ids) if report.user_ids else 0, issued_time=datetime.now() if report.issued_status == "2" else None ) db.add(new_report) db.commit() db.refresh(new_report) # 如果有字段信息，则动态创建新表 if report.field_names and len(report.field_names) > 0: create_dynamic_table(data_table_name, report.field_names, db) # 为每个用户创建填报记录 if report.user_ids: for user_id in report.user_ids: submission = FormSubmission( report_id=new_report.report_id, user_id=user_id, submission_status=0 # 默认状态为未填报 ) db.add(submission) db.commit() return { "code": 200, "msg": "创建成功" } except Exception as e: raise HTTPException(status_code=400, detail=str(e)) # 定义请求体的 Pydantic 模型 class ReportQuery(BaseModel): table_name: Optional[str] = Field(None, description="Table name filter") status: Optional[str] = Field(None, description="Status filter (comma-separated values, e.g., '1,2,3')") start_time: Optional[datetime] = Field(None, description="Start time filter") end_time: Optional[datetime] = Field(None, description="End time filter") issued_status: Optional[str] = Field(None, description="Issued status filter (comma-separated values, e.g., '0,1')") page: int = Field(1, gt=0, description="Page number for pagination") pageSize: int = Field(10, gt=0, description="Page size for pagination") @router.post("/select") async def select_report( db: Session = Depends(get_db), query: ReportQuery = Body(..., description="Report query parameters in the request body"), creator_id: str = Depends(valid_access_token) ): # 构建查询 data_query = db.query(ReportManagement).filter(ReportManagement.creator_id == creator_id) # 应用过滤条件 if query.table_name: data_query = data_query.filter(ReportManagement.table_name.ilike(f"%{query.table_name}%")) if query.start_time and query.end_time: data_query = data_query.filter( ReportManagement.start_time >= query.start_time, ReportManagement.end_time <= query.end_time ) # 处理 status 和 issued_status 的逗号分隔字符串 if query.status: status_list = [int(s) for s in query.status.split(",")] data_query = data_query.filter(ReportManagement.status.in_(status_list)) if query.issued_status: issued_status_list = [int(s) for s in query.issued_status.split(",")] data_query = data_query.filter(ReportManagement.issued_status.in_(issued_status_list)) data_query = data_query.order_by( asc(ReportManagement.issued_status), # 按 issued_status 升序 desc(ReportManagement.created_at) # 按 created_at 降序 ) # 计算总数 total_count = data_query.count() # 分页查询 offset = (query.page - 1) * query.pageSize data = data_query.offset(offset).limit(query.pageSize).all() # 构造结果 result_items = [] for item in data: current_time = datetime.now() is_filling_ended = 0 if item.end_time < current_time: is_filling_ended = 2 result_item = { "id": item.id, "report_id": item.report_id, "table_name": item.table_name, "data_table_name": item.data_table_name, "start_time": item.start_time, "end_time": item.end_time, "status": item.status, "issued_status": item.issued_status, "period_type": item.period_type, "creator_name": item.creator_name, "creator_id": creator_id, "created_at": item.created_at, "creator_phone": item.creator_phone, "updated_at": item.updated_at, "num_reporters": item.num_reporters, "is_filling_ended":is_filling_ended } result_items.append(result_item) result = { "code": 200, "msg": "查询成功", "total": total_count, "pages": (total_count + query.pageSize - 1) // query.pageSize, "currentPage": query.page, "pageSize": query.pageSize, "data": result_items } return result def update_table_fields(table_name: str, field_names: List[str], db: Session): inspector = inspect(db.bind) # 检查表是否存在 if not inspector.has_table(table_name): raise HTTPException(status_code=400, detail="表不存在，无法更新字段") # return {"code": 403, "msg": "未找到对应的填报ID"} # 获取现有表的列信息 existing_columns = inspector.get_columns(table_name) existing_column_names = {col['name'] for col in existing_columns} # 定义需要保留的基础字段 columns_to_keep = {'id', 'user_id', 'create_id', 'collect_status', 'add_time', 'temporarily_store'} existing_column_names -= columns_to_keep # 排除基础字段 # 将新字段名转换为 col1, col2 这种形式 new_field_names = {f"col{i}" for i in range(1, len(field_names) + 1)} # 确定需要删除的字段（现有字段中不存在于新字段列表中的字段） columns_to_drop = existing_column_names - new_field_names # 确定需要添加的字段（新字段列表中不存在于现有字段中的字段） columns_to_add = new_field_names - existing_column_names # 删除不再需要的字段 for column_name in columns_to_drop: try: db.execute(text(f"ALTER TABLE {table_name} DROP COLUMN {column_name}")) except Exception as e: db.rollback() raise HTTPException(status_code=400, detail=f"删除字段失败：{str(e)}") # 添加新字段 for i, field_name in enumerate(field_names, start=1): column_name = f"col{i}" if column_name in columns_to_add: try: db.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {column_name} VARCHAR(255) COMMENT '{field_name}'")) except Exception as e: db.rollback() return {"code": 500, "msg": "添加字段失败，请联系管理员排查"} return {"code": 200, "msg": "字段更新成功"} # class ReportCreate(BaseModel): # table_name: str = Field(..., description="表单名称，必填") # end_time: str = Field(..., description="结束时间，必填，格式为 ISO8601") # # status: str = Field(..., description="状态，必填") # issued_status: str = Field(..., description="发布状态，必填") # creator_name: str = Field(..., description="创建者姓名，必填") # creator_phone: str = Field(..., description="创建者电话，必填") # user_ids: List[int] = Field(..., description="用户 ID 列表，必填") # # period_type: Optional[str] = Field(None, description="周期，非必填") # field_names: Optional[List[str]] = Field(None, description="字段名称列表，非必填") class ReportUpdate(BaseModel): table_name: Optional[str] = None end_time: Optional[str] = None status: Optional[int] = None issued_status: Optional[str] = None period_type: Optional[str] = None creator_phone: Optional[str] = None creator_name: Optional[str] = None user_ids: Optional[List[int]] = Field(None, description="字段名称列表，非必填") # comments: Optional[Dict[str, str]] = None new_fields: Optional[List[str]] = Field(None, description="字段名称列表，非必填") class Config: extra = 'allow' def table_exists(db: Session, table_name: str) -> bool: inspector = inspect(db.bind) return inspector.has_table(table_name) # 删除表（如果存在） def drop_table_if_exists(db: Session, table_name: str): inspector = inspect(db.bind) if inspector.has_table(table_name): try: # 删除表 db.execute(text(f"DROP TABLE {table_name}")) # 清理 MetaData 中的表定义 metadata.reflect(bind=db.bind) if table_name in metadata.tables: del metadata.tables[table_name] except Exception as e: db.rollback() return {"code": 500, "msg": "联系管理员排查"} # raise HTTPException(status_code=400, detail=f"删除表失败：{str(e)}") #修改 @router.put("/report/{report_id}/") async def update_report( report_id: str, update_data: ReportUpdate, db: Session = Depends(get_db), creator_id = Depends(valid_access_token) ): # creator_id = '1' # 假设creator_id已经通过某种方式验证 # 这里得添加下，如果是已经下发的，那就不能修改了，只有待发布（暂存状态）才可以修改 暂存：保存至“待发布”状态 #.filter(ReportManagement.creator_id == creator_id) report = db.query(ReportManagement).filter(ReportManagement.report_id == report_id, ReportManagement.creator_id == creator_id).first() if not report: return {"code": 404, "msg": "Report not found"} # raise HTTPException(status_code=404, detail="Report not found") if report.issued_status in ['2', 2]: return {"code": 400, "msg": "当前表单已发布，无法修改"} # raise HTTPException(status_code=400, detail="当前表单已发布，无法修改") if report.collection_status in ['2',2]: return {"code": 400, "msg": "当前表单已收取，无法修改"} # raise HTTPException(status_code=400, detail="当前表单已收取，无法修改") # 这里要添加发布状态判断，如果是暂存状态则跳过，发布状态则要提供字段信息 # if update_data.issued_status in ['2', 2]: # 检查表是否存在 table_name = report.data_table_name if not table_exists(db, table_name): # 如果表不存在，根据 new_fields 创建表 if not update_data.new_fields: # raise HTTPException(status_code=400, detail="表不存在且未提供字段信息，无法创建表") return {"code": 400, "msg": "表不存在且未提供字段信息，无法创建表"} create_dynamic_table(table_name, update_data.new_fields, db) elif table_exists(db, table_name) and update_data.new_fields: update_table_fields(table_name, update_data.new_fields, db) elif update_data.issued_status in ['0', 0]: if update_data.new_fields: if len(update_data.new_fields)>0: table_name = report.data_table_name if not table_exists(db, table_name): create_dynamic_table(table_name, update_data.new_fields, db) elif table_exists(db, table_name) and update_data.new_fields: update_table_fields(table_name, update_data.new_fields, db) # 更新 creator_phone 和 creator_name if update_data.creator_phone: report.creator_phone = update_data.creator_phone if update_data.creator_name: report.creator_name = update_data.creator_name # 删除旧的填报人 ID，并添加新的填报人 ID if update_data.user_ids: # 删除旧的填报记录 db.query(FormSubmission).filter(FormSubmission.report_id == report.report_id).delete() # 添加新的填报记录 for user_id in update_data.user_ids: submission = FormSubmission( report_id=report.report_id, user_id=user_id, submission_status=0 # 默认状态为未填报 ) db.add(submission) report.num_reporters = len(update_data.user_ids) # 更新字段 if update_data.table_name: report.table_name = update_data.table_name if update_data.status is not None: report.status = update_data.status if update_data.period_type: report.period_type = update_data.period_type if update_data.end_time: report.end_time = datetime.fromisoformat(update_data.end_time) if update_data.issued_status: if update_data.issued_status in [2,"2"]: # 更新issued_status为2 # 这里要判断是否 report.issued_status = 2 report.issued_time = datetime.utcnow() if update_data.status: # print(11111111) # 更新create_time为当前时间 report.status = update_data.status # if update_data. current_time = datetime.now() report.updated_at = current_time #先判断状态，未发布/暂存 的可以修改 #先看看传入参数有没有修改字段，有修改字段的时候，先看看有没有表，如果没有表，没有的话就读取表单里面的表名，再读取字段参数进行建表 #如果已经有表了，那就读取表单里的表名，再读取字段参数，然后删掉原有的表，再进行建表 db.commit() db.refresh(report) return { "code": 200, "msg": "操作成功" } #发布 @router.put("/report/{report_id}/update_status/") async def update_report_status_and_time( report_id: str, db: Session = Depends(get_db), creator_id = Depends(valid_access_token) ): # 查询要修改的记录 report = db.query(ReportManagement).filter(ReportManagement.report_id == report_id).first() if not report: return {"code": 404, "msg": "Report not found"} # raise HTTPException(status_code=404, detail="Report not found") # 验证请求者ID if str(report.creator_id) != str(creator_id): return {"code": 403, "msg": "没有权限操作"} # raise HTTPException(status_code=403, detail="没有权限操作") if report.issued_status ==2: return {"code": 400, "msg": "不可重复发布"} # raise HTTPException(status_code=403, detail="不可重复发布") data_table_name = report.data_table_name # 查询对应表的表结构 table_structure_query = db.execute( text(""" SELECT COLUMN_NAME, COLUMN_COMMENT, ORDINAL_POSITION FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = :table_name AND TABLE_SCHEMA = (SELECT DATABASE()) ORDER BY ORDINAL_POSITION """), {"table_name": data_table_name} ) table_structures = [] column_order = [] # 用于存储字段的顺序 for row in table_structure_query.fetchall(): if row[0] not in ['collect_status', 'create_id', 'id', 'user_id', 'add_time','temporarily_store']: table_structures.append(TableStructure(column_name=row[0], comment=row[1])) column_order.append(row[0]) # 保存字段顺序 # print(table_structures) # 查询所有相关用户 users = db.query(FormSubmission.user_id).filter( FormSubmission.report_id == report_id ).distinct() user_ids = [user[0] for user in users.all()] # print(user_ids) if len(table_structures) == 0 or len(user_ids) ==0: # raise HTTPException(status_code=400, detail=str('信息未填写完整，无法发布')) return {"code": 400, "msg": "信息未填写完整，无法发布"} # 更新issued_status为2 report.issued_status = 2 # 更新create_time为当前时间 report.issued_time = datetime.utcnow() try: db.commit() db.refresh(report) return { "code": 200, "msg": "操作成功" } except Exception as e: db.rollback() # raise HTTPException(status_code=400, detail=str(e)) return {"code": 500, "msg": "未知报错"} class TaskQuery(BaseModel): submission_status: Optional[List[int]] = None table_name: Optional[str] = None page: int = 1 # 分页参数：当前页码，默认为1 pageSize: int = 10 # 分页参数：每页大小，默认为10 @router.post("/my_filling") async def get_user_tasks( db: Session = Depends(get_db), query: TaskQuery = Body(...), user_id: str = Depends(valid_access_token), ): # 检查用户ID是否提供 if not user_id: # raise HTTPException(status_code=400, detail="用户ID是必填项") return {"code": 400, "msg": "用户ID是必填项"} # 查询用户的所有任务信息 user_tasks = db.query(ReportManagement, FormSubmission).join( FormSubmission, ReportManagement.report_id == FormSubmission.report_id ).filter( FormSubmission.user_id == user_id ) # 如果提供了填报结果列表，则过滤结果 if query.submission_status: user_tasks = user_tasks.filter(FormSubmission.submission_status.in_(query.submission_status)) if query.table_name: user_tasks = user_tasks.filter(ReportManagement.table_name.ilike(f'%{query.table_name}%')) # 按 submission_status 升序排序 user_tasks = user_tasks.order_by(FormSubmission.submission_status.asc()) # 计算总数 total_count = user_tasks.count() # 分页查询 offset = (query.page - 1) * query.pageSize tasks = user_tasks.offset(offset).limit(query.pageSize).all() # 收取状态 # 构造返回结果 result_items = [] for report, submission in tasks: result_item = { "user_id": user_id, "table_name": report.table_name, "report_id": report.report_id, "submission_status": submission.submission_status, "start_time": report.start_time, "end_time": report.end_time, "collection_status":report.collection_status } result_items.append(result_item) # 构造分页结果 result = { "code": 200, "msg": "查询成功", "total": total_count, "totalPages": (total_count + query.pageSize - 1) // query.pageSize, "page": query.page, "pageSize": query.pageSize, "data": result_items } return result @router.get("/report_fields") @router.post("/report_fields") async def get_report_fields( db: Session = Depends(get_db), # user_id: str = Query(None, description="用户ID"), report_id: str = Query(None, description="填报ID"), user_id = Depends(valid_access_token) ): # 检查用户ID和填报ID是否提供 if not user_id or not report_id: # raise HTTPException(status_code=400, detail="用户ID和填报ID是必填项") return {"code": 400, "msg": "用户ID和填报ID是必填项"} # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter(ReportManagement.report_id == report_id).first() if not report: # raise HTTPException(status_code=404, detail="未找到对应的填报ID") return {"code": 404, "msg": "未找到对应的填报ID"} data_table_name = report.data_table_name if not data_table_name: # raise HTTPException(status_code=404, detail="未找到对应的数据表名称") return {"code": 404, "msg": "未找到对应的数据表名称"} # 检查用户是否有权限访问填报数据 submission = db.query(FormSubmission).filter( FormSubmission.report_id == report_id, FormSubmission.user_id == user_id ).first() if not submission: # raise HTTPException(status_code=403, detail="没有权限访问这个填报数据") return {"code": 403, "msg": "没有权限访问这个填报数据"} # 使用SQLAlchemy的inspect功能来获取表的字段信息 inspector = inspect(db.bind) columns = inspector.get_columns(data_table_name) # 构造返回结果 result_fields = [] for column in columns: if column['name'] not in ['collect_status', 'create_id', 'id', 'user_id','temporarily_store']: result_field = { "field_name": column['name'], "field_comment": column.get('comment', '无注释') } result_fields.append(result_field) # 构造返回结果 result_items = [] excluded_columns = ['id', 'user_id', 'create_id', 'collect_status','temporarily_store'] # 构建查询SQL query_sql = text(f""" SELECT * FROM {data_table_name} WHERE user_id = :user_id """) result = db.execute(query_sql, {"user_id": user_id}) rows = result.fetchall() # 提取列名和列注释 column_names = [column['name'] for column in columns] column_comments = {column['name']: column['comment'] for column in columns if 'comment' in column} # 添加字段名和字段注释作为第一行 first_row = {column: column_comments.get(column, '') for column in column_names if column not in excluded_columns} result_items.append(first_row) # print(result_items) for row in rows: # 过滤掉不需要的列 # print(row) filtered_row = {column: row[idx] for idx, column in enumerate(column_names) if column not in excluded_columns} # print(filtered_row) result_items.append(filtered_row) # 返回用户ID、填报ID和字段信息 return { "code":200, "msg":"查询成功", "user_id": user_id, "report_id": report_id, "fields": result_fields, "data":result_items } class DataEntry(BaseModel): data: List[Dict[str, Any]] # 数据列表，每个元素是一个字典，包含字段名和值 class DataItem(BaseModel): pass # 用于动态接收键值对 class SubmitData(BaseModel): # user_id: int report_id: str data: List[Dict[str, str]] # 数据列表，每个元素是一个字典，包含字段名和值 class Config: arbitrary_types_allowed = True @router.post("/submit_data") async def submit_data( db: Session = Depends(get_db), submit_data: SubmitData = Body(...), user_id = Depends(valid_access_token) ): # 检查用户ID和填报ID是否提供 if not user_id or not submit_data.report_id: return {"code": 400, "msg": "填报ID是必填项"} # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter(ReportManagement.report_id == submit_data.report_id).first() if not report: return {"code": 400, "msg": "未找到对应的填报ID"} data_table_name = report.data_table_name if not data_table_name: return {"code": 400, "msg": "未找到对应的数据表名称"} if report.issued_status not in [2, '2']: return {"code": 400, "msg": "当前未发布，不可填写"} is_collection = report.collection_status if is_collection == 2 or is_collection == '2': return {"code": 400, "msg": "管理员已收取信息，无法填写"} # 检查用户是否有权限填报 submission = db.query(FormSubmission).filter( FormSubmission.report_id == submit_data.report_id, FormSubmission.user_id == str(user_id) # 确保user_id是字符串类型 ).first() if not submission: return {"code": 400, "msg": "用户没有填报权限"} current_time = datetime.now() if report.end_time < current_time: return {"code": 400, "msg": "填写时间已过"} # 将数据写入数据库 for item in submit_data.data: # 构造插入SQL语句 columns = ', '.join(list(item.keys()) + ['create_id', 'user_id', 'collect_status']) values = ', '.join( [f":{k}" for k in item.keys()] + [f"'{report.creator_id}'", f"'{user_id}'", '1'] ) sql = f"INSERT INTO {data_table_name} ({columns}) VALUES ({values})" # 执行插入操作 db.execute(text(sql), item) # 修改成已填写状态 submission.submission_status = 1 db.add(submission) # 删除暂存的记录 delete_sql = f"DELETE FROM {data_table_name} WHERE temporarily_store = 1 AND user_id = '{user_id}'" db.execute(text(delete_sql)) # 提交事务 db.commit() return { "code": 200, "msg": "数据提交成功" } #数据保存 覆盖旧的，删掉旧记录 @router.post("/save_data") async def save_data( db: Session = Depends(get_db), save_data: SubmitData = Body(...), user_id = Depends(valid_access_token) ): # 检查用户ID和填报ID是否提供 if not user_id or not save_data.report_id: return {"code": 400, "msg": "填报ID是必填项"} # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter(ReportManagement.report_id == save_data.report_id).first() if not report: return {"code": 400, "msg": "未找到对应的填报ID"} data_table_name = report.data_table_name if not data_table_name: return {"code": 400, "msg": "未找到对应的数据表名称"} if report.issued_status not in [2, '2']: return {"code": 400, "msg": "当前未发布，不可填写"} is_collection = report.collection_status if is_collection == 2 or is_collection == '2': return {"code": 400, "msg": "管理员已收取信息，无法填写"} # 检查用户是否有权限填报 submission = db.query(FormSubmission).filter( FormSubmission.report_id == save_data.report_id, FormSubmission.user_id == str(user_id) # 确保user_id是字符串类型 ).first() if not submission: return {"code": 400, "msg": "用户没有填报权限"} current_time = datetime.now() if report.end_time < current_time: return {"code": 400, "msg": "填写时间已过"} # 检查目标表是否包含 temporarily_store 字段 inspector = inspect(db.bind) columns = inspector.get_columns(data_table_name) column_names = [col['name'] for col in columns] if 'temporarily_store' not in column_names: return {"code": 400, "msg": "目标表中缺失字段 'temporarily_store'，保存失败"} delete_sql = f"DELETE FROM {data_table_name} WHERE temporarily_store = 1 AND user_id = '{user_id}'" db.execute(text(delete_sql)) # 将数据写入数据库 for item in save_data.data: # 构造插入SQL语句 columns = ', '.join(list(item.keys()) + ['create_id', 'user_id', 'collect_status', 'temporarily_store']) values = ', '.join( [f":{k}" for k in item.keys()] + [f"'{report.creator_id}'", f"'{user_id}'", '1', '1'] ) sql = f"INSERT INTO {data_table_name} ({columns}) VALUES ({values})" # 执行插入操作 db.execute(text(sql), item) # 提交事务 db.commit() return { "code": 200, "msg": "数据暂存成功" } class SubmissionQuery(BaseModel): # user_id: int # 用户ID，必须是整数 report_id: str # 填报ID，必须是字符串 @router.post("/submission_status") async def get_submission_status( db: Session = Depends(get_db), query: SubmissionQuery = Body(...), user_id = Depends(valid_access_token) ): # 检查用户ID和填报ID是否提供 # if not user_id or not query.report_id: # return {"code": 400, "msg": "用户ID和填报ID是必填项"} # raise HTTPException(status_code=400, detail="用户ID和填报ID是必填项") # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter(ReportManagement.report_id == query.report_id).first() if not report: return {"code": 400, "msg": "未找到对应的填报ID"} # raise HTTPException(status_code=404, detail="未找到对应的填报ID") data_table_name = report.data_table_name if not data_table_name: return {"code": 400, "msg": "未找到对应的数据表名称"} # raise HTTPException(status_code=404, detail="未找到对应的数据表名称") # 获取填报情况 submission = db.query(FormSubmission).filter( FormSubmission.report_id == query.report_id, FormSubmission.user_id == str(user_id) # 确保user_id是字符串类型 ).first() if not submission: return {"code": 404, "msg": "未找到对应的填报"} # raise HTTPException(status_code=404, detail="未找到对应的填报情况") start_time = report.start_time end_time = report.end_time start_time_str = start_time.strftime('%Y-%m-%d %H:%M:%S') end_time_str = end_time.strftime('%Y-%m-%d %H:%M:%S') # 使用SQLAlchemy的inspect功能来获取表的列信息 inspector = Inspector.from_engine(db.bind) columns = inspector.get_columns(data_table_name) # 提取列名和列注释 column_names = [column['name'] for column in columns] column_comments = {column['name']: column['comment'] for column in columns if 'comment' in column} # 构造返回结果 result_items = [] excluded_columns = ['id', 'user_id', 'create_id', 'collect_status','temporarily_store'] # 构建查询SQL query_sql = text(f""" SELECT * FROM {data_table_name} WHERE user_id = :user_id """) result = db.execute(query_sql, {"user_id": user_id}) rows = result.fetchall() # 添加字段名和字段注释作为第一行 first_row = {column: column_comments.get(column, '') for column in column_names if column not in excluded_columns} result_items.append(first_row) for row in rows: # 过滤掉不需要的列 filtered_row = {column: row[idx] for idx, column in enumerate(column_names) if column not in excluded_columns} result_items.append(filtered_row) result = { "code": 200, 'msg': '查询成功', 'start_time':start_time_str, "end_time":end_time_str, "data": result_items} return result # 辅助函数：检查表是否存在 def table_exists1(inspector: Inspector, table_name: str) -> bool: return inspector.has_table(table_name) # 辅助函数：根据字段备注获取表中所有匹配字段名 def get_columns_with_comment_like( inspector: Inspector, table_name: str, comment_like: str ) -> List[str]: # 检查表是否存在 if not table_exists1(inspector, table_name): print(f"表 {table_name} 不存在") return [] # 或者可以选择抛出异常 columns = inspector.get_columns(table_name) matching_columns = [column['name'] for column in columns if column.get('comment') and comment_like in column['comment']] return matching_columns # 辅助函数：检查是否有字段备注匹配 def has_matching_column_comments( inspector: Inspector, table_name: str, comment_like: str ) -> bool: return bool(get_columns_with_comment_like(inspector, table_name, comment_like)) #【数据档案管理】-列表 @router.post("/reports_by_creator") @router.get("/reports_by_creator") async def get_reports_by_creator( field_comment: Optional[str] = Query(None, description="Optional comment of the field to match"), page: int = Query(default=1, gt=0), # 分页参数：当前页码，默认为1 pageSize: int = Query(default=10, gt=0), # 分页参数：每页大小，默认为10 db: Session = Depends(get_db), creator_id=Depends(valid_access_token) ): # 获取数据库Inspector inspector: Inspector = inspect(db.bind) # 查询 ReportManagement 表以获取所有相关的记录 query = db.query(ReportManagement).filter(ReportManagement.creator_id == creator_id) query = query.order_by( asc(ReportManagement.collection_status) # 按 collection_status 升序 ) # 计算总数 total_count = query.count() # 分页查询 offset = (page - 1) * pageSize reports = query.offset(offset).limit(pageSize).all() # 存储结果 results = [] for report in reports: # 检查是否存在 data_table_name if not report.data_table_name: continue # 如果没有 data_table_name，跳过这个 report # 检查是否有字段备注匹配 field_comment if field_comment and not has_matching_column_comments(inspector, report.data_table_name, field_comment): continue # 如果没有匹配的字段备注，跳过这个 report # 如果匹配成功，添加到结果中 collection_time_str = report.collection_time.isoformat().replace('T', ' ') if report.collection_time else None #在这里判断数据库 #收取状态 collection_status = report.collection_status issued_status = report.issued_status #结束时间 end_time = report.end_time current_time = datetime.now() if end_time < current_time and collection_status in [0,'0'] and issued_status in [2,'2']: # print("符合自动收取") report.collection_status=2 report.collection_time = current_time db.add(report) db.commit() db.refresh(report) results.append({ "table_name": report.table_name, "collection_status": report.collection_status, "collection_time": collection_time_str, "report_id": report.report_id }) # 如果没有找到任何记录，抛出404异常 if not results: return {"code": 404, "msg": "没有找到与该创建人ID相关的记录"} # raise HTTPException(status_code=404, detail="没有找到与该创建人ID相关的记录") # 构造分页结果 result = { "code": 200, "msg": "查询成功", "total": total_count, "totalPages": (total_count + pageSize - 1) // pageSize, "page": page, "pageSize": pageSize, "data": results } return result @router.put("/update_collection_status/") async def update_collection_status( # creator_id: str, report_id: str, new_status: int = Query(..., description="New collection status, must be 0, 1, or 2"), db: Session = Depends(get_db), creator_id = Depends(valid_access_token) ): # 检查 new_status 是否为允许的值之一 if new_status not in (0, 1, 2): return {"code": 400, "msg": "Invalid collection status value"} # raise HTTPException(status_code=400, detail="Invalid collection status value") # 查询 ReportManagement 表以获取对应记录 report = db.query(ReportManagement).filter( ReportManagement.creator_id == creator_id, ReportManagement.report_id == report_id ).first() # 如果没有找到记录，返回404 if not report: return {"code": 404, "msg": "Report not found"} # raise HTTPException(status_code=404, detail="Report not found") if report.collection_status == 2 or report.collection_status == '2': return {"code": 404, "msg": "当前已收取，无需重复收取"} # raise HTTPException(status_code=404, detail="当前已收取，无需重复收取") current_time_str = datetime.now().strftime("%Y%m%d%H%M%S") # 更新 collection_status report.collection_status = new_status report.collection_time = current_time_str db.add(report) db.commit() db.refresh(report) return { "code":200, "msg": "更新成功", "new_status": new_status } class ReportQuery(BaseModel): # creator_id: str # 创建人ID，必须是字符串 report_id: str # 填报ID，必须是字符串 @router.get("/dataArchiveDetails/") async def get_records_by_creator_and_report( query: ReportQuery = Depends(), db: Session = Depends(get_db), creator_id = Depends(valid_access_token) ): # 查询 ReportManagement 表以获取对应记录 report = db.query(ReportManagement).filter( ReportManagement.creator_id == creator_id, ReportManagement.report_id == query.report_id ).first() # 如果没有找到记录，返回404 if not report: return {"code": 404, "msg": "Report not found"} # 如果没有 data_table_name，返回404 if not report.data_table_name: return {"code": 404, "msg": "Data table name not found"} # 使用SQLAlchemy的inspect功能来获取表的列信息 inspector = inspect(db.bind) columns = inspector.get_columns(report.data_table_name) # 提取列名和列注释 column_names = [column['name'] for column in columns] column_comments = {column['name']: column['comment'] for column in columns if 'comment' in column} # 构造字段信息 columns_info = [] for column in column_names: if column not in ['id', 'user_id', 'create_id', 'collect_status', 'temporarily_store']: columns_info.append({ 'prop': column, 'label': column_comments.get(column, '') }) # 构造查询SQL语句 if 'temporarily_store' in column_names: # 如果存在 temporarily_store 字段，过滤掉暂存状态的记录 query_sql = text(f""" SELECT w.*, u.user_name FROM {report.data_table_name} w LEFT JOIN sys_user u ON w.user_id = u.user_id WHERE w.temporarily_store != 1 """) else: query_sql = text(f""" SELECT w.*, u.user_name FROM {report.data_table_name} w LEFT JOIN sys_user u ON w.user_id = u.user_id """) result = db.execute(query_sql) rows = result.fetchall() # 构造返回结果 rows_data = [] for row in rows: # 过滤掉不需要的列，并添加到结果中 filtered_row = {column: row[idx] for idx, column in enumerate(column_names) if column not in ['id', 'user_id', 'create_id', 'collect_status', 'temporarily_store']} filtered_row['user_name'] = row[-1] # 添加用户昵称 rows_data.append(filtered_row) # 获取报告的开始和结束时间，并格式化为字符串 start_time_str = report.start_time.strftime('%Y-%m-%d %H:%M:%S') if report.start_time else None end_time_str = report.end_time.strftime('%Y-%m-%d %H:%M:%S') if report.end_time else None return { "code": 200, 'msg': '查询成功', 'start_time': start_time_str, "end_time": end_time_str, "columns": columns_info, "rows": rows_data } from fastapi import status @router.get("/export_to_excel") @router.post("/export_to_excel") async def export_to_excel( report_id: str = Query(..., description="填报ID"), db: Session = Depends(get_db), creator_id: str = Depends(valid_access_token) ): # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter( ReportManagement.report_id == report_id, ReportManagement.creator_id == creator_id ).first() if not report: # raise HTTPException(status_code=404, detail="未找到对应的填报ID") return {"code": 403, "msg": "未找到对应的填报ID"} data_table_name = report.data_table_name if not data_table_name: # raise HTTPException(status_code=404, detail="未找到对应的数据表名称") return {"code": 404, "msg": "未找到对应的数据表名称"} # 获取表结构（用户填报的字段） inspector = inspect(db.bind) columns = inspector.get_columns(data_table_name) # 提取用户填报的字段注释 user_report_columns = [col for col in columns if col['name'] not in ['id', 'create_id', 'collect_status', 'add_time', 'user_id','temporarily_store']] column_comments = [col.get('comment', '') for col in user_report_columns] # 构建查询SQL，关联 sys_user 表获取 nick_name query_sql = f""" SELECT su.nick_name AS user_name, {', '.join([f'rd.{col["name"]}' for col in user_report_columns])} FROM {data_table_name} rd JOIN sys_user su ON rd.user_id = su.user_id """ # 使用 text 包装查询字符串 result = db.execute(text(query_sql)) rows = result.fetchall() # 将查询结果转换为 DataFrame df = pd.DataFrame(rows, columns=["user_name"] + column_comments) # 将 DataFrame 导出为 Excel 文件 output = BytesIO() with pd.ExcelWriter(output, engine='openpyxl') as writer: df.to_excel(writer, index=False, sheet_name='填报数据') # 设置响应头 output.seek(0) headers = { 'Content-Disposition': 'attachment; filename="report_data.xlsx"', 'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' } # 返回文件流 return StreamingResponse(output, headers=headers)