import os import sys import re import paho.mqtt.client as mqtt import json import logging from logging.handlers import RotatingFileHandler from sqlalchemy import create_engine, Engine, text from sqlalchemy.exc import IntegrityError from datetime import datetime, timedelta from dateutil import tz import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from threading import Thread, Lock from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import traceback import queue CUR_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, CUR_DIR) import conf import time LOG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'mqttc.log') logging.basicConfig(level=getattr(logging, conf.LOG_LEVEL.upper(), logging.INFO), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout), # Log to console RotatingFileHandler(LOG_FILE, maxBytes=5*1034*1024, backupCount=1, encoding="utf-8"), # Log to file ]) logger = logging.getLogger(__name__) engine:Engine = None # {table_name: {mpoint_id: last_timex}} 按表分桶，避免跨表互相抑制 mpoint_dict: dict = {} SHANGHAI = tz.gettz('Asia/Shanghai') # topic -> 目标表名 映射 TOPIC_TABLE_MAP = { conf.MQTT_TOPIC: 'mplogx', conf.MQTT_TOPIC_zl: 'mplogx_xzzl', conf.MQTT_TOPIC_tl: 'mplogx_tlxn', conf.MQTT_TOPIC_1: 'mplogx_hknf_l1', conf.MQTT_TOPIC_2: 'mplogx_hknf_l2', conf.MQTT_TOPIC_3: 'mplogx_hknf_l3', conf.MQTT_TOPIC_kio: 'mplogx_wengfu', } # KIO datachange 格式（偏移使能）的 topic，走独立解析逻辑 KIO_TOPICS = {conf.MQTT_TOPIC_kio} # 各表 INSERT 用的值列（除 timex, mpoint_id 外）；未列出的表用默认两列 # mplogx_wengfu(KIO) 按值类型分列存储：bool->val_bool, int->val_int, float->val_float, 文本->val_str TABLE_VALUE_COLS = { 'mplogx_mqtt': ['val_float', 'val_str', 'val_q'], 'mplogx_wengfu': ['val_float', 'val_str', 'val_q', 'val_bool', 'val_int'], } DEFAULT_VALUE_COLS = ['val_float', 'val_str'] # topic -> 独立队列，每个 topic 一个 worker 线程消费 msg_queues: dict = {topic: queue.Queue() for topic in TOPIC_TABLE_MAP} # 错误邮件发送限流：最多 1 小时发一封 EMAIL_INTERVAL = 3600 _last_email_time = 0 _email_lock = Lock() def send_error_email(message, subject='hfnf_mqtt', ): now = time.time() with _email_lock: global _last_email_time if now - _last_email_time < EMAIL_INTERVAL: logger.info(f"距上次报警邮件不足 {EMAIL_INTERVAL}s，跳过本次发送: {message}") return _last_email_time = now msg = MIMEMultipart() msg['From'] = conf.EMAIL_HOST_USER msg['To'] = conf.EMAIL_HOST_USER msg['Subject'] = subject msg.attach(MIMEText(message, 'plain')) server = smtplib.SMTP(conf.EMAIL_HOST, conf.EMAIL_PORT) server.starttls() server.login(conf.EMAIL_HOST_USER, conf.EMAIL_HOST_PASSWORD) server.sendmail(conf.EMAIL_HOST_USER, [conf.EMAIL_HOST_USER, conf.EMAIL_HOST_USER2], msg.as_string()) server.quit() def worker(topic: str): """每个 topic 一个 worker 线程，独立做断流告警""" table_name = TOPIC_TABLE_MAP[topic] q = msg_queues[topic] last_message_time = time.time() # 该 topic 最后一次收到消息时间 last_alert_time = 0 # 该 topic 上一次报警时间 while True: try: payload = q.get(timeout=60) if payload is not None: try: save_items(topic, payload) except Exception as e: logger.error(f"[{table_name}] save_items 异常: {e}", exc_info=True) Thread(target=send_error_email, args=(f"[{table_name}] save_items 异常: {e}",)).start() last_message_time = time.time() last_alert_time = 0 except queue.Empty: now = time.time() if now - last_message_time > 600 and now - last_alert_time > 600: Thread( target=send_error_email, args=(f"time-out 10min [{table_name}] topic={topic} 无数据推送",), ).start() last_alert_time = now def normalize_standard(payload): """现有 6 个 topic 的格式 [{name,value,time,quality?}] -> [(timex, mpoint_id, value, quality)]""" records = [] for item in json.loads(payload): timex = datetime.strptime(item['time'], "%Y%m%d%H%M%S").replace(tzinfo=SHANGHAI) records.append((timex, item['name'], item.get('value'), item.get('quality'))) return records # 翁福(KIO)白名单映射：{点位全名 N -> 位号}，启动时从 wengfu_mpoint_name 加载。 # 只有命中白名单的点位才把 mpoint_id 缩短为位号；其余（裸位号点、以及前/后位 # 限位这类尾串无区分度、缩短会串台的点）一律原样入库，避免不同物理点碰撞合并。 wengfu_name_map: dict = {} def load_wengfu_name_map(eng): """从 wengfu_mpoint_name 载入 full_name -> mpoint_id 映射。失败不影响主流程。""" global wengfu_name_map try: with eng.connect() as conn: rows = conn.execute(text( "SELECT full_name, mpoint_id FROM wengfu_mpoint_name WHERE full_name IS NOT NULL" )).fetchall() wengfu_name_map = {full: mid for full, mid in rows} logger.info(f"载入翁福位号映射 {len(wengfu_name_map)} 条") except Exception as e: logger.error(f"载入 wengfu_mpoint_name 失败，本次将按原样存全名: {e}", exc_info=True) wengfu_name_map = {} def code_from_name(name): """从翁福(KIO)点位全名切出位号（供建表脚本 load_mpoint_name.py 生成白名单用）。 运行时入库不再直接调用本函数，改走 wengfu_name_map 查表白名单，见 normalize_kio。 全名形如 <中文点位名>_<位号>，中文名本身不含下划线——电流点位除外， 其中文名会嵌入形如 PU6301_1001A 的设备号。位号恒为纯 ASCII，故以“最后一个 非 ASCII 字符”（即中文名的结尾）为锚点，锚点之后的 ASCII 尾串即位号。 用非 ASCII 判定而非枚举 CJK 码段，可一并容纳全角符号、罗马数字、生僻字等： 隧道窑液压顶车机B本地和远程_REM_PU5507B -> REM_PU5507B 一号干燥室侧墙温度1_TE_5113 -> TE_5113 (末尾索引数字1归名称) 一号隧道窑布煤机PU6301_1001A电流_PU6301_1001A_I -> PU6301_1001A_I ...前位限位M01_50A_PU_5503_ZAI_ZS5576 -> M01_50A_PU_5503_ZAI_ZS5576 _FLT_PU6302_2009A_22 -> FLT_PU6302_2009A_22 (无中文兜底) 切不出来时原样返回，保证入库不丢数据。 """ if not name: return name pos = 0 for i, ch in enumerate(name): if ord(ch) > 127: # 任意非 ASCII 字符都属于中文名 pos = i + 1 if pos == 0: # 无中文：整体即位号，去掉前导下划线 return name.lstrip('_') or name tail = name[pos:] if tail.startswith('_'): return tail[1:] m = re.match(r'^\d+_(.+)$', tail) # 名称末尾的索引数字（温度1）留给名称 if m: return m.group(1) if tail[:1].isalpha(): # 位号紧跟字母（M01_50A...） return tail return name # 兜底：无法切分则原样返回 def normalize_kio(payload): """KIO datachange（偏移使能）格式 -> [(timex, mpoint_id, value, quality)] 结构: {"PNs":{1:V,2:T,3:Q}, "PVs":{基准值}, "Objs":[{N, 1?, 2?, 3?}]} - 值(1)/质量(3)：缺失即等于 PVs 基准值 - 时间(2)：为毫秒偏移量，真实时间 = PVs 基准时间 + 偏移(ms)；缺失=偏移0 - PVs 时间戳实测为北京时间(Asia/Shanghai)，非手册所称 UTC """ obj = json.loads(payload) pvs = obj.get('PVs', {}) base_v = pvs.get('1') base_q = pvs.get('3') base_ts = pvs.get('2') base_time = None if base_ts: for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S"): try: base_time = datetime.strptime(base_ts, fmt).replace(tzinfo=SHANGHAI) break except ValueError: continue records = [] for o in obj.get('Objs', []): name = o.get('N') if not name or base_time is None: continue value = o.get('1', base_v) try: offset_ms = int(o.get('2', 0)) except (TypeError, ValueError): offset_ms = 0 timex = base_time + timedelta(milliseconds=offset_ms) quality = o.get('3', base_q) # 命中白名单则存位号，否则原样存全名（不丢、不碰撞） mpoint_id = wengfu_name_map.get(name, name) records.append((timex, mpoint_id, value, quality)) return records def route_value(value): """按值的类型分列：返回 (val_float, val_str, val_bool, val_int)。 注意 Python bool 是 int 子类，必须先判 bool 再判 int。""" if isinstance(value, bool): return None, None, value, None if isinstance(value, int): return None, None, None, value if isinstance(value, float): return value, None, None, None # 字符串或其它：能转数字进 val_float，否则当文本 try: return float(value), None, None, None except (TypeError, ValueError): return None, (str(value) if value is not None else None), None, None def save_items(topic, payload): table_name = TOPIC_TABLE_MAP.get(topic) if not table_name: logger.error(f"未知 topic：{topic}，跳过入库") return normalize = normalize_kio if topic in KIO_TOPICS else normalize_standard records = normalize(payload) value_cols = TABLE_VALUE_COLS.get(table_name, DEFAULT_VALUE_COLS) col_list = "timex, mpoint_id, " + ", ".join(value_cols) param_list = ":timex, :mpoint_id, " + ", ".join(f":{c}" for c in value_cols) sql_str = f"INSERT INTO {table_name} ({col_list}) VALUES ({param_list})" table_mpoint_dict = mpoint_dict.setdefault(table_name, {}) save_list = [] for timex, mpoint_id, value, quality in records: last_timex: datetime = table_mpoint_dict.get(mpoint_id, None) if timex.minute not in getattr(conf, "SAVE_MINUTES", [2, 7, 12, 17, 22, 27, 32, 37, 42, 47, 52, 57]) or ( last_timex and last_timex.minute == timex.minute): continue else: val_float, val_str, val_bool, val_int = route_value(value) # 质量戳(Q)：取不到则存 NULL val_q = None if quality is not None: try: val_q = int(quality) except (TypeError, ValueError): val_q = None save_list.append({"timex": timex, "mpoint_id": mpoint_id, "val_float": val_float, "val_str": val_str, "val_q": val_q, "val_bool": val_bool, "val_int": val_int}) if save_list: with engine.connect() as conn: for item in save_list: is_ok = False try: conn.execute(text(sql_str), item) is_ok = True except IntegrityError: logger.error(f"[{table_name}] {item['mpoint_id']}-主键冲突") except Exception as e: logger.error(f"[{table_name}] {item['mpoint_id']}-保存数据失败", exc_info=True) Thread(target=send_error_email, args=(f"[{table_name}] {item['mpoint_id']}-保存数据失败-{e}",)).start() if is_ok: conn.commit() table_mpoint_dict[item['mpoint_id']] = item['timex'] def get_table_stats(): """统计 mplogx% 开头各表的数据量，返回可序列化的 dict""" with engine.connect() as conn: tables = conn.execute(text(""" SELECT tablename FROM pg_tables WHERE schemaname = current_schema() AND tablename LIKE 'mplogx%' ORDER BY tablename """)).fetchall() if not tables: return {"tables": [], "summary": {}} # 表名来自 pg_tables，受数据库控制；用双引号包裹标识符 parts = [] for (tablename,) in tables: parts.append(f""" SELECT '{tablename}' AS table_name, COUNT(DISTINCT mpoint_id)::bigint AS point_count_1h, approximate_row_count('{tablename}')::numeric AS approximate_row_count FROM "{tablename}" WHERE timex >= now() - interval '1 hour' AND timex < now() """) union_sql = "\nUNION ALL\n".join(parts) rows = conn.execute(text(f""" SELECT table_name, point_count_1h, approximate_row_count FROM ({union_sql}) AS per_table ORDER BY table_name """)).fetchall() result = [] for table_name, point_count_1h, approximate_row_count in rows: result.append({ "table_name": table_name, "point_count_1h": int(point_count_1h) if point_count_1h is not None else 0, "approximate_row_count": int(approximate_row_count) if approximate_row_count is not None else 0, }) total_points = sum(r["point_count_1h"] for r in result) total_rows = sum(r["approximate_row_count"] for r in result) return { "tables": result, "summary": { "point_count_1h_total": total_points, "approximate_row_count_total": total_rows, "approximate_row_count_total_yi": round(total_rows / 100000000, 2), }, } class StatsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path.split('?', 1)[0] != '/stats': self._write_json(404, {"error": "not found"}) return try: self._write_json(200, get_table_stats()) except Exception as e: logger.error("获取表统计失败", exc_info=True) self._write_json(500, {"error": str(e)}) def _write_json(self, status, data): body = json.dumps(data, ensure_ascii=False).encode('utf-8') self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, format, *args): logger.info("HTTP %s - %s", self.address_string(), format % args) def start_http_server(): host = getattr(conf, 'HTTP_HOST', '0.0.0.0') port = getattr(conf, 'HTTP_PORT', 5800) server = ThreadingHTTPServer((host, port), StatsHandler) logger.info(f"HTTP stats server listening on http://{host}:{port}/stats") server.serve_forever() def on_connect(mqttc: mqtt.Client, userdata, flags, rc, properties): if rc == 0: logger.info("Connected to MQTT broker") topics = [(t, 1) for t in TOPIC_TABLE_MAP.keys()] mqttc.subscribe(topics) logger.info(f"Subscribed to topics: {[t for t, _ in topics]}") else: logger.error("Failed to connect to MQTT broker") def on_message(mqttc: mqtt.Client, userdata, msg: mqtt.MQTTMessage): topic = msg.topic q = msg_queues.get(topic) if q is not None: q.put(msg.payload) def on_disconnect(mqttc: mqtt.Client, userdata, disconnect_flags, reason_code, properties): logger.error(f"Disconnected from MQTT broker__:{disconnect_flags}, {reason_code}") def start_mqtt(): client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id='hfnf_105') client.username_pw_set(conf.MQTT_USERNAME, conf.MQTT_PASSWORD) client.on_connect = on_connect client.on_message = on_message client.on_disconnect = on_disconnect client.connect(host=conf.MQTT_HOST, port=conf.MQTT_PORT) client.loop_forever() if __name__ == '__main__': try: engine = create_engine(conf.DATABASE_URL) logger.info("Connected to database") load_wengfu_name_map(engine) # 载入翁福位号白名单 for _topic in TOPIC_TABLE_MAP: Thread(target=worker, args=(_topic,), daemon=True).start() logger.info(f"Worker thread started for topic={_topic} -> {TOPIC_TABLE_MAP[_topic]}") Thread(target=start_http_server, daemon=True).start() start_mqtt() except Exception: logger.error("异常退出", exc_info=True) Thread(target=send_error_email, args=(f'{traceback.format_exc()}')).start() raise