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 from sqlalchemy.orm import Session from pypinyin import lazy_pinyin, Style from database import get_db from models import * import random 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 # 函数用于将中文转换为拼音首字母缩写 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() # class ReportQuery(BaseModel): # report_id: str class TableStructure(BaseModel): column_name: str comment: str #查看详情 @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).first() if not report: 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 FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = :table_name AND TABLE_SCHEMA = (SELECT DATABASE()) """), {"table_name": data_table_name} ) table_structures = [] for row in table_structure_query.fetchall(): if row[0] not in ['collect_status','create_id','id','user_id']: table_structures.append(TableStructure(column_name=row[0], comment=row[1])) # # 查询 user_id 去重后的数量 # distinct_user_count_query = db.execute( # text(""" # SELECT COUNT(DISTINCT user_id) AS distinct_user_count # FROM {} # """.format(data_table_name)) # ) # num_reported = distinct_user_count_query.scalar() # num_unreported = report.num_reporters - num_reported # 构造返回结果,包括 ReportManagement 表中的记录和其他相关信息 # 计算去重用户数量 distinct_users = db.query(FormSubmission.user_id).filter( FormSubmission.report_id == report_id).distinct().count() # 计算已填报数量(填报结果为1) num_reported = db.query(FormSubmission).filter(FormSubmission.report_id == report_id, FormSubmission.submission_status == 1).count() # 计算未填报数量(填报结果为0) num_unreported = db.query(FormSubmission).filter(FormSubmission.report_id == report_id, FormSubmission.submission_status == 0).count() print("xx") print(distinct_users,num_reported,num_unreported) 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": distinct_users, "creator_id": creator_id, "created_at": report.created_at, "updated_at": report.updated_at, "num_reported":num_reported, "num_unreported":num_unreported }, 'table_structure': table_structures } return result # 动态创建表 def create_dynamic_table(table_name: str, field_names: List[str], db: Session): inspector = inspect(db.bind) # 检查表是否已存在 if inspector.has_table(table_name): raise HTTPException(status_code=400, detail="Table already exists") table = Table(table_name, metadata, Column('id', Integer, primary_key=True), Column('user_id', Integer), Column('create_id', Integer), Column('collect_status', Boolean) ) existing_columns = set() for field_name in field_names: column_name = to_first_letter(field_name) # 如果列名已存在,则添加一个唯一的后缀 unique_column_name = column_name suffix = 1 while unique_column_name in existing_columns: unique_column_name = f"{column_name}_{suffix}" suffix += 1 existing_columns.add(unique_column_name) table.append_column(Column(unique_column_name, String(255), comment=field_name)) # 创建表 metadata.create_all(bind=db.bind) class ReportCreate(BaseModel): table_name: str #start_time: str end_time: str status: str issued_status: str period_type: str creator_name: str # creator_id: int creator_phone:str field_names: List[str] # 用户只传递字段名称 user_ids: List[int] # 新建填报和创建新表的接口 @router.post("/report/") def create_report_and_table(report: ReportCreate, db: Session = Depends(get_db), creator_id = Depends(valid_access_token) ): try: # 获取当前时间并格式化为 YYYYMMDDHHMMSS current_time_str = datetime.now().strftime("%Y%m%d%H%M%S") # 动态生成 data_table_name table_name_pinyin='' for i in range(len(lazy_pinyin(report.table_name, style=Style.FIRST_LETTER))): table_name_pinyin += ''.join(lazy_pinyin(report.table_name, style=Style.FIRST_LETTER)[i]).lower() data_table_name = f"{table_name_pinyin}_{current_time_str}" # 动态创建新表 create_dynamic_table(data_table_name, report.field_names, db) # 登记填报管理 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=report.end_time, status=report.status, 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) ) db.add(new_report) db.commit() db.refresh(new_report) # 为每个用户创建填报记录 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') #@router.get("/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)) # 计算总数 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: 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 } 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 # class ReportUpdate(BaseModel): # table_name: str = None # status: int = None # # issued_status: int = None # period_type: str = None # end_time: str = None # comments: dict = None # 字典,键为字段名,值为新的备注 # new_fields: List[dict] = None class ReportUpdate(BaseModel): table_name: Optional[str] = None status: Optional[int] = None period_type: Optional[str] = None end_time: Optional[str] = None comments: Optional[Dict[str, str]] = None new_fields: Optional[List[Dict[str, str]]] = None class Config: extra = 'allow' # 动态添加字段到表 def add_column_if_not_exists(table_name: str, column_name: str, column_type: str, comment: str, db: Session): inspector = inspect(db.bind) columns = inspector.get_columns(table_name) existing_column_names = {column['name'] for column in columns} if column_name not in existing_column_names: try: db.execute( text(f""" ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type} COMMENT :comment """), {"comment": comment} ) except Exception as e: db.rollback() raise HTTPException(status_code=400, detail=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已经通过某种方式验证 report = db.query(ReportManagement).filter(ReportManagement.report_id == report_id).first() if not report: raise HTTPException(status_code=404, detail="Report not found") if report.creator_id != creator_id: raise HTTPException(status_code=403, detail="没有权限更新此报告") # 更新字段 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.comments: for column_name, comment in update_data.comments.items(): db.execute( text(f""" ALTER TABLE {report.data_table_name} CHANGE {column_name} {column_name} TEXT COMMENT :comment """), {"comment": comment} ) # 添加新字段 if update_data.new_fields: for field in update_data.new_fields: column_name = field['field_name'] comment = field['comment'] add_column_if_not_exists(report.data_table_name, column_name, 'TEXT', comment, db) 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: raise HTTPException(status_code=404, detail="Report not found") print(report.creator_id,creator_id) # 验证请求者ID if str(report.creator_id) != str(creator_id): raise HTTPException(status_code=403, detail="没有权限更新此报告") if report.issued_status ==2: raise HTTPException(status_code=403, detail="不可重复发布") # 更新issued_status为2 report.issued_status = 2 # 更新create_time为当前时间 report.start_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)) class TaskQuery(BaseModel): # user_id: str submission_status: Optional[List[int]] = None table_name: Optional[str] = None @router.post("/my_filling") @router.get("/my_filling") async def get_user_tasks( db: Session = Depends(get_db), query: TaskQuery = Body(...), user_id = Depends(valid_access_token) ): # 检查用户ID是否提供 if not user_id: raise HTTPException(status_code=400, detail="用户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}%')) # 执行查询 tasks = user_tasks.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, } result_items.append(result_item) return { "code":200, "msg":"查询成功", "data": result_items } @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是必填项") # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter(ReportManagement.report_id == report_id).first() if not report: raise HTTPException(status_code=404, detail="未找到对应的填报ID") data_table_name = report.data_table_name if not data_table_name: raise HTTPException(status_code=404, detail="未找到对应的数据表名称") # 检查用户是否有权限访问填报数据 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="没有权限访问这个填报数据") # 使用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']: result_field = { "field_name": column['name'], "field_comment": column.get('comment', '无注释') } result_fields.append(result_field) # 返回用户ID、填报ID和字段信息 return { "code":200, "msg":"查询成功", "user_id": user_id, "report_id": report_id, "fields": result_fields } 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: raise HTTPException(status_code=400, detail="用户ID和填报ID是必填项") # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter(ReportManagement.report_id == submit_data.report_id).first() if not report: raise HTTPException(status_code=404, detail="未找到对应的填报ID") data_table_name = report.data_table_name if not data_table_name: raise HTTPException(status_code=404, detail="未找到对应的数据表名称") # 检查用户是否有权限填报 submission = db.query(FormSubmission).filter( FormSubmission.report_id == submit_data.report_id, FormSubmission.user_id == str(user_id) # 确保user_id是字符串类型 ).first() if not submission: raise HTTPException(status_code=403, detail="用户没有填报权限") # print(report.creator_id,submit_data.user_id) # if submission. # 将数据写入数据库 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})" print(sql) # 执行插入操作 db.execute(text(sql), item) submission.submission_status = 1 db.add(submission) # 提交事务 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: raise HTTPException(status_code=400, detail="用户ID和填报ID是必填项") # 获取对应填报ID的数据表名称 report = db.query(ReportManagement).filter(ReportManagement.report_id == query.report_id).first() if not report: raise HTTPException(status_code=404, detail="未找到对应的填报ID") data_table_name = report.data_table_name if not data_table_name: 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: raise HTTPException(status_code=404, detail="未找到对应的填报情况") start_time = report.start_time end_time = report.end_time print(start_time,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'] # 构建查询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 get_columns_with_comment_like( inspector: Inspector, table_name: str, comment_like: str ) -> List[str]: 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) # 计算总数 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 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: 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): 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: raise HTTPException(status_code=404, detail="Report not found") # 更新 collection_status report.collection_status = new_status 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) ): creator_id # 查询 ReportManagement 表以获取对应记录 report = db.query(ReportManagement).filter( ReportManagement.creator_id == creator_id, ReportManagement.report_id == query.report_id ).first() # 如果没有找到记录,返回404 if not report: raise HTTPException(status_code=404, detail="Report not found") # 如果没有 data_table_name,返回404 if not report.data_table_name: raise HTTPException(status_code=404, detail="Data table name not found") # 查询工单表所有信息,并关联用户表匹配到用户名字 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() # 使用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']: columns_info.append({ 'prop': column, 'label': column_comments.get(column, '') }) # 构造返回结果 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']} 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 }