no message

common/db/db_msg_center.py

@@ -17,6 +17,25 @@ def add_msg(db: Session, msg_type: str, msg_id: str , user_id: int) -> None:
     db.refresh(new_msg)
     return new_msg.id
 
+def add_message(db: Session, msg_type: str, recv_userid: str , title: str, content: str, foreign_key: str = '', from_scenario: str = '') -> str:
+    new_msg = MsgCenter(
+        msg_type = msg_type, 
+        msg_id = new_guid(), 
+        recv_time = datetime.now(), 
+        recv_userid = recv_userid, 
+        recv_status = 0, 
+        update_time = datetime.now(),
+        title = title, 
+        content = content, 
+        foreign_key = foreign_key,
+        from_scenario = from_scenario,
+        del_flag = '0'
+    )
+    db.add(new_msg)
+    db.commit()
+    db.refresh(new_msg)
+    return new_msg.msg_id
+
 def get_unread_msg_count(db: Session, user_id: int, msg_type_list: dict) -> dict:
     data = []
     for msg_type in msg_type_list:
@@ -30,16 +49,26 @@ def get_unread_msg_count(db: Session, user_id: int, msg_type_list: dict) -> dict
             c1 = db.query(EventBase).filter(where).count() 
         else:
             if msg_type == '巡查工作':
-                where = and_(MsgCenter.msg_type == '隐患巡查')
+                where = and_(MsgCenter.msg_type.in_(['隐患巡查']))
+
+            if msg_type == '风险防控':
+                where = and_(MsgCenter.msg_type.in_(['风险排查']))
+
             elif msg_type == '数据管理':
                 where = and_(MsgCenter.msg_type == '数据采集')
 
+            elif msg_type == '值班管理':
+                where = and_(MsgCenter.msg_type == '值班消息')
+                
+            elif msg_type == '工作审批':
+                where = and_(MsgCenter.msg_type == '审批消息')
+
             c1 = db.query(MsgCenter).filter(where).filter(and_(MsgCenter.recv_userid == user_id, MsgCenter.recv_status == 0)).count()
         
         data.append({"name" : msg_type, "num": c1})
     logger.info(data)
     return data
 
-def update_msg_read(db: Session, user_id: int, msg_type: str, msg_id: str) -> None:
-    db.query(MsgCenter).filter(and_(MsgCenter.msg_type == msg_type, MsgCenter.recv_userid == user_id, MsgCenter.msg_id == str(msg_id))).update({"recv_status": 1, "update_time": datetime.now()})
+def update_msg_read(db: Session, user_id: int, msg_type: str, foreign_key: str) -> None:
+    db.query(MsgCenter).filter(and_(MsgCenter.msg_type == msg_type, MsgCenter.recv_userid == user_id, MsgCenter.foreign_key == str(foreign_key))).update({"recv_status": 1, "update_time": datetime.now()})
     db.commit()

jobs/duty_job.py

@@ -77,8 +77,9 @@ def send_yzy_msg(db: Session, to_user_id: int, start_time: str, end_time: str, f
         "detail_url": detail_url,
         "foreign_key": foreign_key,
         "from_scenario": "duty_shift",
-        "title": "值班提醒"
+        "title": "值班消息"
     }
     YzyApi.add_to_msg_queue(db, data)
-    db_msg_center.add_msg(db, "值班提醒", foreign_key, to_user_id)
+    db_msg_center.add_message(db, "值班消息", to_user_id, "值班提醒", description, foreign_key, 'duty_shift')
+
         

jobs/yzy_job.py

@@ -14,6 +14,9 @@ import traceback
 import inspect
 
 def proc():
+    if settings.IS_DEV:
+        return
+    
     lock_key = "yzy_job_proc"
     if redis_lock(lock_key):
         logger.info(datetime.now())

models/base.py

@@ -113,7 +113,11 @@ class MsgCenter(Base):
     recv_userid = Column(Integer, comment='接收用户ID')
     recv_status = Column(Integer, default="0", comment='接收状态 0 未阅 1 已阅')
     update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment='创建时间')
-    
+    del_flag = Column(String, default='0', comment='删除标志（0代表存在 2代表删除）')
+    content = Column(String, default='', comment='消息内容')
+    foreign_key = Column(String(50), comment='文件外键 --技术字段')
+    from_scenario = Column(String(50), comment='对应标识 --技术字段')
+    title = Column(String, default='', comment='标题')
     class Config:
         orm_mode = True
 

routers/api/eventManagement/event.py

@@ -77,7 +77,7 @@ async def create_event(
         db.commit()
 
         # 发送粤政易事件
-        send_yzy_msg(db, event_base, user_id)
+        # send_yzy_msg(db, event_base, user_id)
         
         return {
             "code": 200,
@@ -114,8 +114,7 @@ def send_yzy_msg(db: Session, event_base: EventBase, user_id: int) -> None:
     }
     YzyApi.add_to_msg_queue(db, data)
 
-    # 改为没用关闭的事件统计
-    # db_msg_center.add_msg(db, "事件管理", event_base.id, to_user_id)
+    # db_msg_center.add_message(db, "事件接报", recv_userid, "事件接报提醒", description, event_base.event_code, 'event_base')
 
 @router.get('/list')
 async def get_event_list(
@@ -979,7 +978,7 @@ async def send_emergency_plan_task_by_yzy(
         }
         YzyApi.add_to_msg_queue(db, data)
 
-        db_msg_center.add_msg(db, "预案通知", event_emergency_notify.id, _user_id)
+        db_msg_center.add_message(db, "预案通知", _user_id, f"{plan_name}{response_level}通知", yzy_content, event_emergency_notify.id, "event_emergency_notify")
 
     return {
         "code": 200,

routers/api/infoPublish/me.py

@@ -48,6 +48,7 @@ EXAMINE_SUB_TYPE_DICT = {
 }
     
 # 信息发布分页（我能看的内容，小屏）
+# 因为信息发布被消息中心取代，因为这个接口可能没那么大用了 2024/12/26
 @router.get('/list')
 async def get_publish_list(
     search_keyword: str = Query('', description='信息内容'),

routers/api/infoPublish/msg_center.py

@@ -26,9 +26,11 @@ import base64
 from io import BytesIO
 from PIL import Image
 from config import settings
-
+from utils.riskManagement_uitl import get_task_title_by_type, get_rescue_resources_task_title_by_type
 router = APIRouter()
 
+# 消息中心
+
 @router.post("/unread_msg_count")
 async def msg_count(
     request: Request, 
@@ -95,4 +97,209 @@ async def send_yzy_msg(request: Request,
     return {
         "code": 200,
         "msg": "消息暂存成功"
-    }
+    }
+
+
+@router.get('/list')
+async def get_center_list(
+    search_keyword: str = Query('', description='信息内容'),
+    info_type:  str = Query('', description='类型'),
+    page: int = Query(1, gt=0, description='页码'),
+    begin_time_s:  str = Query(None, description='开始时间'),
+    end_time_s:  str = Query(None, description='结束时间'),
+    info_order:  str = Query("desc", description='时间排序'),
+    page_size: int = Query(10, gt=0, description='pageSize'),
+    db: Session = Depends(get_db),
+    user_id = Depends(valid_access_token)
+):
+    try:
+        # 应用查询条件
+        where = and_(MsgCenter.del_flag == '0', MsgCenter.recv_userid == user_id)
+        if search_keyword != '':
+            where = and_(where, MsgCenter.content.like('%{}%'.format(search_keyword)))
+        if info_type != '':
+            where = and_(where, MsgCenter.msg_type == info_type)
+        if begin_time_s != None:
+            begin_time = datetime.strptime(begin_time_s, "%Y-%m-%d")
+            where = and_(where, MsgCenter.recv_time >= begin_time)
+        if end_time_s != None:
+            end_time = datetime.strptime(end_time_s, "%Y-%m-%d") + timedelta(days=1)
+            where = and_(where, MsgCenter.recv_time < end_time)
+        print(where)
+        
+        # 计算总条目数
+        q = db.query(func.count(MsgCenter.id))
+        q = q.filter(where)
+        total = q.scalar()
+        
+        # 执行分页查询
+        q = db.query(MsgCenter)
+        q = q.filter(where)
+
+        if info_order == 'desc':
+            q = q.order_by(MsgCenter.recv_time.desc())
+        if info_order == 'asc':
+            q = q.order_by(MsgCenter.recv_time.asc())
+        
+        rows = q.offset((page - 1) * page_size).limit(page_size).all()
+
+        data = []
+        for row in rows:
+            foreign_key = row.foreign_key
+            from_scenario = row.from_scenario
+
+            detail = {}
+
+            if row.msg_type == '任务消息':
+                try:
+                    task_info = db.query(TaskRegistration).filter(TaskRegistration.task_id == foreign_key).first()
+                    event_code = task_info.event_code
+
+                    event_info  = db.query(EventBase).filter(EventBase.event_code == event_code).first()
+                    detail = {
+                        "event_title": event_info.event_title,
+                        "creation_time": get_datetime_str(task_info.creation_time),
+                        "task_description": task_info.task_description,
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '预案通知':
+                try:
+                    notify_info = db.query(EventEmergencyNotify).filter(EventEmergencyNotify.id == foreign_key).first()
+                    event_id = notify_info.event_id
+                    plan_id = notify_info.plan_id
+                    sent_time = notify_info.sent_time
+
+                    plan_info = db.query(EmergencyPlan).filter(EmergencyPlan.plan_number == plan_id).first()
+                    plan_name = plan_info.plan_name
+                    
+                    event_info  = db.query(EventBase).filter(EventBase.event_code == event_id).first()
+                    response_level = db_dict.get_dict_label(db, "response_level", event_info.response_level)
+
+                    detail = {
+                        "title": row.title,
+                        "organizing_unit": plan_info.organizing_unit, # 发布单位
+                        "plan_name": plan_name, # 预案名称
+                        "response_level": response_level, # 相应级别
+                        "sent_time": get_datetime_str(sent_time), # 发送时间
+                        "yzy_content": notify_info.yzy_content, # 消息内容
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '在线点名':
+                try:
+                    call_info = db.query(OnlineRollCallDetail).filter(OnlineRollCallDetail.id == foreign_key).first()
+                    detail = {
+                        "call_title": row.content,
+                        "create_time": get_datetime_str(call_info.create_time), # 发送时间
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '事件接报':
+                try:
+                    event_info  = db.query(EventBase).filter(EventBase.event_code == foreign_key).first()
+                    detail = {
+                        "event_title": row.title,
+                        "address": event_info.address,
+                        "event_level": db_dict.get_dict_label(db, "mm_event_level", event_info.event_level),
+                        "event_time": get_datetime_str(event_info.event_time), # 事发时间
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '隐患巡查':
+                try:
+                    # task_info  = db.query(RiskManagementInspectionTask).filter(RiskManagementInspectionTask.task_number == foreign_key).first()
+                    # task_title = get_task_title_by_type(task_info.inspection_business)
+                    detail = {
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '风险排查':
+                try:
+                    # task_info  = db.query(RiskManagementRiskTask).filter(RiskManagementRiskTask.task_number == foreign_key).first()
+                    # task_title = db_dict.get_dict_label(db, 'risk_type', task_info.risk_type)
+                    detail = {
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '数据采集':
+                try:
+                    # task_info  = db.query(RiskManagementRescueResourcesTask).filter(RiskManagementRescueResourcesTask.task_number == foreign_key).first()
+                    # task_title = get_rescue_resources_task_title_by_type(task_info.type)
+                    detail = {
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '值班消息':
+                try:
+                    detail = {
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                   
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '审批消息':
+                try:
+                    detail = {
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+                   
+                except:
+                    traceback.print_exc()
+
+            elif row.msg_type == '系统消息':
+                try:
+                    detail = {
+                        "detail_url": "/infoDetails?id="+str(row.id)
+                    }
+
+                except:
+                    traceback.print_exc()
+
+            # 信息发布
+            elif row.msg_type in ['预警信息', '灾情信息', '灾情信息', '指挥救援', '公众防范']:
+                try:
+                    info = db.query(InfoPublishBase).filter(InfoPublishBase.id == foreign_key).first()
+                    detail = {
+                        "detail_url": "/infoDetails?id="+str(info.id)
+                    }
+                    
+                except:
+                    traceback.print_exc()
+
+            data.append({
+                "id": row.id,
+                "msg_type": row.msg_type,
+                "title": row.title,
+                "content": row.content,
+                "recv_time": get_datetime_str(row.recv_time),
+                "detail": detail
+            })
+
+        # 返回结果
+        return {
+            "code": 200,
+            "msg": "查询成功",
+            "data": data,
+            "total": total
+        }
+
+    except Exception as e:
+        # 处理异常
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=str(e))

routers/api/onlineRollCall/call.py

@@ -594,7 +594,7 @@ def send_yzy_msg(db: Session, detail_entity: OnlineRollCallDetail, user_id: int)
                 "title": "在线点名提醒"
             }
             YzyApi.add_to_msg_queue(db, data)
-            db_msg_center.add_msg(db, "在线点名", detail_entity.id, to_user_id)
+            db_msg_center.add_message(db, "在线点名", to_user_id, "在线点名提醒", "你有一条在线点名通知，请尽快确认", str(detail_entity.id), 'online_roll_call_detail')
 
             user_list.append(yzy_account)
 
@@ -760,10 +760,15 @@ async def get_call_summary(request: Request,
     db: Session = Depends(get_db),
     user_id = Depends(valid_access_token)):
 
+    is_myid_where = or_(OnlineRollCallDetail.standby_staff_id == user_id, 
+                      OnlineRollCallDetail.leader_id == user_id, 
+                      OnlineRollCallDetail.primary_staff_id == user_id, 
+                      OnlineRollCallDetail.secondary_staff_id == user_id)
+    
     try:
-        call_count = db.query(OnlineRollCallDetail).filter(and_(OnlineRollCallDetail.leader_id == user_id, OnlineRollCallDetail.del_flag == '0')).count()
-        ack_count = db.query(OnlineRollCallDetail).filter(and_(OnlineRollCallDetail.leader_id == user_id, OnlineRollCallDetail.del_flag == '0', OnlineRollCallDetail.ack_status == 1)).count()
-        unack_count = db.query(OnlineRollCallDetail).filter(and_(OnlineRollCallDetail.leader_id == user_id, OnlineRollCallDetail.del_flag == '0', OnlineRollCallDetail.ack_status == 0)).count()
+        call_count = db.query(OnlineRollCallDetail).filter(and_(is_myid_where, OnlineRollCallDetail.del_flag == '0')).count()
+        ack_count = db.query(OnlineRollCallDetail).filter(and_(is_myid_where, OnlineRollCallDetail.del_flag == '0', OnlineRollCallDetail.ack_status == 1)).count()
+        unack_count = db.query(OnlineRollCallDetail).filter(and_(is_myid_where, OnlineRollCallDetail.del_flag == '0', OnlineRollCallDetail.ack_status == 0)).count()
 
         data = {
             "call_count": call_count,

routers/api/riskManagement/risk_router.py

@@ -769,8 +769,8 @@ async def create_inspection_task(
         # 标记消息已读
         child_task_info = db.query(RiskManagementRiskTaskChildrenTask).filter(RiskManagementRiskTaskChildrenTask.id == children_task_id).first()
         if child_task_info is not None:
-            logger.info("标记[风险防控任务]已读 {}", str(child_task_info.task_id))
-            db_msg_center.update_msg_read(db, user_id, "风险防控", str(child_task_info.task_id))
+            logger.info("标记[风险排查任务]已读 {}", str(child_task_info.task_id))
+            db_msg_center.update_msg_read(db, user_id, "风险排查", str(child_task_info.task_id))
 
         # 返回创建成功的响应
         return {

routers/api/taskRegistration/__init__.py

@@ -107,7 +107,7 @@ async def create_task(
         raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
 
 # 发送粤政易消息
-def send_yzy_msg(db: Session, task_base: TaskRegistration, user_id: int) -> None:
+def send_yzy_msg(db: Session, task_base: TaskRegistration, recorded_by: int) -> None:
     to_user_id = task_base.executor_id
     user_info = db_user.get_user_info(db, to_user_id)
     yzy_account = user_info.yzy_account
@@ -121,15 +121,16 @@ def send_yzy_msg(db: Session, task_base: TaskRegistration, user_id: int) -> None
         "yzy_userid": yzy_userid,
         "mobile": yzy_account,
         "content": description,
-        "recorded_by": user_id,
+        "recorded_by": recorded_by,
         "detail_url": detail_url,
-        "foreign_key": task_base.id,
+        "foreign_key": task_base.task_id,
         "from_scenario": "task_registrations",
-        "title": "事件任务"
+        "title": f"{event_title}任务"
     }
     YzyApi.add_to_msg_queue(db, data)
 
-    db_msg_center.add_msg(db, "任务消息", task_base.id, task_base.executor_id)
+    db_msg_center.add_message(db, "任务消息", to_user_id, f"{event_title}任务", task_base.task_description, task_base.task_id, 'task_registrations')
+
 
 class TaskQuery(BaseModel):
     task_id: str = None

utils/riskManagement_uitl.py

@@ -194,7 +194,7 @@ def create_children_task(db,task_info,corn_query):
         foreign_key = str(task_info.id)
         from_scenario = "risk_management_inspection_task"
 
-        send_yzy_msg(db, "隐患巡查", to_user_id, task_info.id, yzy_account, description, detail_url, foreign_key, from_scenario)
+        send_yzy_msg(db, "隐患巡查", to_user_id, yzy_account, description, detail_url, foreign_key, from_scenario)
 
 def get_task_title_by_type(type: str) -> str:
     if type == '0':
@@ -209,7 +209,7 @@ def get_task_title_by_type(type: str) -> str:
         return str(type)
 
 # 发送粤政易消息
-def send_yzy_msg(db: Session, msg_type: str, to_user_id: int, task_id: int, yzy_account: str, description: str, detail_url: str, foreign_key:str, from_scenario: str) -> None:
+def send_yzy_msg(db: Session, msg_type: str, to_user_id: int, yzy_account: str, description: str, detail_url: str, foreign_key:str, from_scenario: str) -> None:
     yzy_userid = db_yzy.get_userid_by_account(db, yzy_account)
     
     data = {
@@ -224,9 +224,10 @@ def send_yzy_msg(db: Session, msg_type: str, to_user_id: int, task_id: int, yzy_
     }
     YzyApi.add_to_msg_queue(db, data)
 
-    db_msg_center.add_msg(db, msg_type, str(task_id), to_user_id)
+    db_msg_center.add_message(db, msg_type, to_user_id, "数据报送提醒", description, foreign_key, from_scenario)
 
-# 风险防控
+
+# 风险防控排查
 def create_risk_children_task(db,task_info,corn_query):
     cycle = task_info.task_cycle
     task_range = task_info.task_range
@@ -256,7 +257,7 @@ def create_risk_children_task(db,task_info,corn_query):
         foreign_key = str(task_info.id)
         from_scenario = "risk_management_risk_task"
 
-        send_yzy_msg(db, "风险防控", to_user_id, task_info.id, yzy_account, description, detail_url, foreign_key, from_scenario)
+        send_yzy_msg(db, "风险排查", to_user_id, yzy_account, description, detail_url, foreign_key, from_scenario)
     
 def create_rescue_resources_children_task(db,task_info,corn_query):
     cycle = task_info.task_cycle
@@ -287,7 +288,7 @@ def create_rescue_resources_children_task(db,task_info,corn_query):
         foreign_key = str(task_info.id)
         from_scenario = "risk_management_rescue_resources_task"
 
-        send_yzy_msg(db, "数据采集", to_user_id, task_info.id, yzy_account, description, detail_url, foreign_key, from_scenario)
+        send_yzy_msg(db, "数据采集", to_user_id, yzy_account, description, detail_url, foreign_key, from_scenario)
 
 def get_rescue_resources_task_title_by_type(type: str) -> str:
     if type == '0':