开始

app/utils/db_utils.py

@@ -0,0 +1,136 @@
+from functools import wraps
+import time
+from flask import current_app
+import logging
+from sqlalchemy.exc import SQLAlchemyError
+from scripts.models import db
+from collections import defaultdict
+
+# 简单的内存缓存实现
+class QueryCache:
+    def __init__(self, max_size=100, ttl=300):  # 默认缓存300秒
+        self.cache = {}
+        self.max_size = max_size
+        self.ttl = ttl
+        self.access_times = defaultdict(int)
+
+    def get(self, key):
+        if key in self.cache:
+            item = self.cache[key]
+            if time.time() - item['time'] < self.ttl:
+                self.access_times[key] += 1
+                return item['value']
+            else:
+                # 缓存过期，删除
+                del self.cache[key]
+                if key in self.access_times:
+                    del self.access_times[key]
+        return None
+
+    def set(self, key, value):
+        # 如果缓存满了，删除最少访问的项
+        if len(self.cache) >= self.max_size:
+            # 找出访问次数最少的键
+            if self.access_times:
+                min_key = min(self.access_times, key=self.access_times.get)
+                if min_key in self.cache:
+                    del self.cache[min_key]
+                del self.access_times[min_key]
+
+        self.cache[key] = {
+            'value': value,
+            'time': time.time()
+        }
+        self.access_times[key] = 1
+        return value
+
+# 创建全局缓存实例
+query_cache = QueryCache()
+
+# 数据库操作装饰器 - 用于缓存结果
+def cache_query(ttl=300):
+    """
+    缓存查询结果的装饰器
+    :param ttl: 缓存有效期（秒）
+    """
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            # 生成缓存键
+            cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
+            
+            # 尝试从缓存获取
+            cached_result = query_cache.get(cache_key)
+            if cached_result is not None:
+                return cached_result
+            
+            # 执行查询
+            result = func(*args, **kwargs)
+            
+            # 设置缓存
+            query_cache.set(cache_key, result)
+            return result
+        return wrapper
+    return decorator
+
+# 数据库操作装饰器 - 用于异常处理
+def db_operation(retries=3, retry_delay=0.5):
+    """
+    数据库操作的装饰器，提供异常处理和重试功能
+    :param retries: 重试次数
+    :param retry_delay: 重试延迟(秒)
+    """
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            last_error = None
+            for attempt in range(retries):
+                try:
+                    return func(*args, **kwargs)
+                except SQLAlchemyError as e:
+                    last_error = e
+                    current_app.logger.warning(f"数据库操作失败: {str(e)}，尝试重试 ({attempt+1}/{retries})")
+                    
+                    # 回滚会话
+                    db.session.rollback()
+                    
+                    # 如果不是最后一次尝试，则等待后重试
+                    if attempt < retries - 1:
+                        time.sleep(retry_delay)
+            
+            # 所有重试都失败
+            current_app.logger.error(f"数据库操作失败(已重试{retries}次): {str(last_error)}")
+            raise last_error
+        return wrapper
+    return decorator
+
+# 辅助函数：清除特定表相关的所有缓存
+def clear_table_cache(table_name):
+    """
+    清除与特定表相关的所有缓存
+    :param table_name: 表名
+    """
+    keys_to_delete = []
+    for key in list(query_cache.cache.keys()):
+        if table_name.lower() in key.lower():
+            keys_to_delete.append(key)
+    
+    for key in keys_to_delete:
+        if key in query_cache.cache:
+            del query_cache.cache[key]
+        if key in query_cache.access_times:
+            del query_cache.access_times[key]
+
+# 通用查询函数，包含异常处理和缓存
+@db_operation()
+@cache_query()
+def get_all_distinct_values(model, column_name):
+    """
+    获取指定模型指定列的所有不同值
+    :param model: 模型类
+    :param column_name: 列名
+    :return: 所有不同值的列表
+    """
+    column = getattr(model, column_name)
+    results = db.session.query(column).distinct().all()
+    return [result[0] for result in results if result[0]] 

app/utils/error_handlers.py

@@ -0,0 +1,150 @@
+from flask import render_template, jsonify, request, flash, redirect, url_for
+from sqlalchemy.exc import SQLAlchemyError, IntegrityError, OperationalError
+from werkzeug.exceptions import HTTPException, NotFound, Forbidden, BadRequest
+import traceback
+
+class ErrorHandler:
+    """
+    全局异常处理器
+    用于注册Flask应用程序的错误处理函数
+    """
+    def __init__(self, app=None):
+        if app:
+            self.init_app(app)
+    
+    def init_app(self, app):
+        """
+        初始化应用程序的错误处理器
+        :param app: Flask应用实例
+        """
+        # 处理SQLAlchemy错误
+        app.register_error_handler(SQLAlchemyError, self.handle_db_error)
+        app.register_error_handler(IntegrityError, self.handle_integrity_error)
+        app.register_error_handler(OperationalError, self.handle_operational_error)
+        
+        # 处理HTTP错误
+        app.register_error_handler(NotFound, self.handle_not_found)
+        app.register_error_handler(Forbidden, self.handle_forbidden)
+        app.register_error_handler(BadRequest, self.handle_bad_request)
+        
+        # 处理通用错误
+        app.register_error_handler(Exception, self.handle_generic_error)
+        
+        # 记录初始化完成
+        app.logger.info("错误处理器初始化完成")
+    
+    def handle_db_error(self, error):
+        """
+        处理一般数据库错误
+        """
+        error_details = str(error)
+        traceback_details = traceback.format_exc()
+        current_app = self._get_current_app()
+        current_app.logger.error(f"数据库错误: {error_details}\n{traceback_details}")
+        
+        return self._handle_error("数据库操作错误", 
+                                 "抱歉，数据库操作失败，请稍后重试。", 
+                                 500)
+    
+    def handle_integrity_error(self, error):
+        """
+        处理数据完整性错误（如违反唯一约束）
+        """
+        error_details = str(error)
+        current_app = self._get_current_app()
+        current_app.logger.error(f"数据完整性错误: {error_details}")
+        
+        # 尝试从错误消息中提取更友好的错误消息
+        friendly_message = "数据冲突，请检查您的输入。"
+        if "Duplicate entry" in error_details:
+            friendly_message = "数据已存在，请检查您的输入。"
+        
+        return self._handle_error("数据完整性错误", friendly_message, 400)
+    
+    def handle_operational_error(self, error):
+        """
+        处理数据库操作错误（如连接问题）
+        """
+        error_details = str(error)
+        current_app = self._get_current_app()
+        current_app.logger.error(f"数据库操作错误: {error_details}")
+        
+        return self._handle_error("数据库连接错误", 
+                                 "抱歉，无法连接到数据库，请稍后重试。", 
+                                 503)
+    
+    def handle_not_found(self, error):
+        """
+        处理404错误
+        """
+        return self._handle_error("页面未找到", 
+                                 "抱歉，您请求的页面不存在。", 
+                                 404)
+    
+    def handle_forbidden(self, error):
+        """
+        处理403错误
+        """
+        return self._handle_error("访问被拒绝", 
+                                 "抱歉，您没有权限访问此页面。", 
+                                 403)
+    
+    def handle_bad_request(self, error):
+        """
+        处理400错误
+        """
+        return self._handle_error("请求错误", 
+                                 "抱歉，服务器无法理解您的请求。", 
+                                 400)
+    
+    def handle_generic_error(self, error):
+        """
+        处理未被其他处理程序捕获的通用错误
+        """
+        error_details = str(error)
+        traceback_details = traceback.format_exc()
+        current_app = self._get_current_app()
+        current_app.logger.error(f"未处理异常: {error_details}\n{traceback_details}")
+        
+        if isinstance(error, HTTPException):
+            return self._handle_error("请求错误", error.description, error.code)
+        
+        return self._handle_error("服务器错误", 
+                                 "抱歉，服务器遇到了未知错误，请稍后重试。", 
+                                 500)
+    
+    def _handle_error(self, title, message, status_code):
+        """
+        根据请求类型返回适当的错误响应
+        :param title: 错误标题
+        :param message: 错误消息
+        :param status_code: HTTP状态码
+        """
+        # 根据接受的内容类型决定返回JSON还是HTML
+        # 检查是否是AJAX请求，is_xhr在新版Flask中已被移除
+        is_ajax_request = request.headers.get('X-Requested-With') == 'XMLHttpRequest'
+        if is_ajax_request or request.headers.get('Accept') == 'application/json':
+            # 返回JSON格式的错误
+            return jsonify({
+                'error': title,
+                'message': message
+            }), status_code
+        else:
+            # 使用Flash消息并重定向到统计页面
+            from flask import current_app
+            flash(message)
+            
+            # 根据错误类型决定重定向页面
+            if status_code == 404:
+                return redirect(url_for('statistics'))
+            elif status_code == 403:
+                return redirect(url_for('statistics'))
+            else:
+                return redirect(url_for('statistics'))
+    
+    def _get_current_app(self):
+        """
+        获取当前Flask应用实例
+        """
+        from flask import current_app
+        return current_app 

app/utils/file_utils.py

@@ -0,0 +1,10 @@
+from PIL import Image
+
+def allowed_file(filename, allowed_extensions):
+    return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowed_extensions
+
+def compress_image(image_path, max_size=(1500, 1500), quality=85):
+    """压缩图片"""
+    img = Image.open(image_path)
+    img.thumbnail(max_size, Image.Resampling.LANCZOS)
+    img.save(image_path, quality=quality)

app/utils/lzmx.py

@@ -0,0 +1,102 @@
+# utils\lzmx.py
+import requests
+import json
+from datetime import datetime
+
+# 推送系统工单处理信息到量子密信
+def send_lzmx_message(webhook_url, handler, device_id, device_type, distance_from_branch, restore_time):
+    headers = {'Content-Type': 'application/json'}
+    
+    message = {
+        "type": "text",
+        "textMsg": {
+            "content": 
+            f"📢 故障处理通知\n"
+            f"故障设备：{device_id}\n"
+            f"处理时间：{restore_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
+            f"故障类型：{device_type}\n"
+            f"距离支局：{distance_from_branch}公里\n"
+            f"处理人：{handler}"
+        }
+    }
+    
+    try:
+        response = requests.post(
+            webhook_url,
+            headers=headers,
+            data=json.dumps(message)
+        )
+        response.raise_for_status()
+        return True
+    except Exception as e:
+        print(f"发送量子密信消息失败: {str(e)}")
+        return False
+
+# 推送手工申报工单信息到量子密信
+def send_manual_lzmx_message(webhook_url, handler, device_id, fault_type, distance_from_branch, fault_time, project_type, device_location):
+    headers = {'Content-Type': 'application/json'}
+    
+    message = {
+        "type": "text",
+        "textMsg": {
+            "content": 
+            f"📝 故障处理通知\n"
+            f"项目类型：{project_type}\n"
+            f"故障设备：{device_id}\n"
+            f"设备位置：{device_location}\n"
+            f"故障时间：{fault_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
+            f"故障类型：{fault_type}\n"
+            f"距离支局：{distance_from_branch}公里\n"
+            f"处理人：{handler}"
+        }
+    }
+    
+    try:
+        response = requests.post(
+            webhook_url,
+            headers=headers,
+            data=json.dumps(message)
+        )
+        response.raise_for_status()
+        return True
+    except Exception as e:
+        print(f"发送量子密信消息失败: {str(e)}")
+        return False
+
+# 每周五统计本周情况推送量子密信
+def send_weekly_stats_message(webhook_url, stats):
+    headers = {'Content-Type': 'application/json'}
+    
+    # 格式化前三名处理人信息
+    top_handlers = []
+    for handler, amount in stats['top_handlers']:
+        top_handlers.append(f"👤 {handler}: {amount}元")
+    
+    message = {
+        "type": "text",
+        "textMsg": {
+            "content": 
+            f"📊 本周工单统计报告\n"
+            f"—————————————————\n"
+            f" 本周派单：{stats['total_orders']}单\n"
+            f" 已处理数：{stats['completed_orders']}单\n"
+            f" 未处理数：{stats['pending_orders']}单\n"
+            f" 手工故障：{stats['manual_orders']}个\n\n"
+            f"🏆 前三名预计奖励：\n"
+            f"{chr(10).join(top_handlers)}\n"
+            f"—————————————————\n"
+            f"统计时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
+        }
+    }
+    
+    try:
+        response = requests.post(
+            webhook_url,
+            headers=headers,
+            data=json.dumps(message)
+        )
+        response.raise_for_status()
+        return True
+    except Exception as e:
+        print(f"发送量子密信消息失败: {str(e)}")
+        return False

app/utils/pagination.py

@@ -0,0 +1,11 @@
+class Pagination:
+    def __init__(self, items, page, per_page, total):
+        self.items = items
+        self.page = page
+        self.per_page = per_page
+        self.total = total
+        self.pages = (total + per_page - 1) // per_page
+        self.has_prev = page > 1
+        self.has_next = page < self.pages
+        self.prev_num = page - 1
+        self.next_num = page + 1

app/utils/reward.py

@@ -0,0 +1,66 @@
+def get_reward_by_distance(project_type, distance):
+    """
+    根据项目类型和距离计算奖励金额
+    
+    Args:
+        project_type (str): 项目类型
+        distance (float): 距离（公里）
+        
+    Returns:
+        float: 奖励金额（元）
+    """
+    # 2025.04.21 by v6ole
+    # 奖励配置说明：
+    # 格式为: '项目类型': [(距离上限1, 奖励金额1), (距离上限2, 奖励金额2), ...]
+    # 距离单位: 公里, 金额单位: 元
+    # 配置按照距离从小到大排序，系统会自动匹配第一个符合条件的奖励金额
+    rewards_config = {
+        # 校园安防"4+N"项目奖励标准：
+        # - 0-5公里: 20元/台
+        # - 5-30公里: 40元/台
+        # - 30-50公里: 50元/台
+        # - 50公里以上: 60元/台
+        '校园安防"4+N"项目': [
+            (5, 20),
+            (30, 40),
+            (50, 50),
+            (float('inf'), 60)
+        ],
+        
+        # 综治视联网奖励标准：
+        # - 0-10公里: 30元/台
+        # - 10-30公里: 40元/台
+        # - 30-50公里: 50元/台
+        # - 50公里以上: 60元/台
+        '综治视联网': [
+            (10, 30),
+            (30, 40),
+            (50, 50),
+            (float('inf'), 60)
+        ],
+        
+        # 教育城域网奖励标准：
+        # - 0-5公里: 20元/台
+        # - 5-30公里: 30元/台
+        # - 30-50公里: 40元/台
+        # - 50公里以上: 50元/台
+        '教育城域网': [
+            (5, 20),
+            (30, 30),
+            (50, 40),
+            (float('inf'), 50)
+        ],
+        
+        # 其他项目统一标准：
+        # - 任何距离: 30元/台
+        '其他': [(float('inf'), 30)]
+    }
+    
+    if project_type not in rewards_config:
+        return 0
+        
+    for max_distance, reward in rewards_config[project_type]:
+        if distance <= max_distance:
+            return reward
+            
+    return 0