From 42fc032ec10c39ac2e432dc14dc041387f7c3e55 Mon Sep 17 00:00:00 2001 From: TianyangZhang Date: Tue, 21 Jul 2026 02:26:42 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E7=BF=81=E7=A6=8F(KI?= =?UTF-8?q?O)=E6=95=B0=E6=8D=AE=E9=87=87=E9=9B=86=E3=80=81FastAPI=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E6=8E=A5=E5=8F=A3=E5=8F=8A=E4=BD=8D=E5=8F=B7=E6=98=A0?= =?UTF-8?q?=E5=B0=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - mqttc: 支持KIO datachange偏移使能格式解析，翁福位号白名单映射， HTTP /stats统计接口，邮件报警限流，表名列名重构，MQTT认证 - api: 新增FastAPI数据查询服务(latest/history/mpoints/count/stats) - load_mpoint_name: Excel位号导入脚本 - supervisor_api.conf: API服务supervisor配置 Co-Authored-By: Claude Opus 4.6 Claude-Session: https://claude.ai/code/session_01GhsYr2ZyGxsGw7vnyuEA5Q --- api.py | 193 ++++++++++++++++++++++++++++++ load_mpoint_name.py | 101 ++++++++++++++++ mqttc.py | 283 +++++++++++++++++++++++++++++++++++++++++--- supervisor_api.conf | 13 ++ 4 files changed, 572 insertions(+), 18 deletions(-) create mode 100644 api.py create mode 100644 load_mpoint_name.py create mode 100644 supervisor_api.conf diff --git a/api.py b/api.py new file mode 100644 index 0000000..0d54aa7 --- /dev/null +++ b/api.py @@ -0,0 +1,193 @@ +"""轻量数据查询接口，独立于 mqttc 进程运行。 + +启动: uvicorn api:app --host 0.0.0.0 --port 5900 +""" +import os +import sys + +CUR_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, CUR_DIR) +import conf + +from fastapi import FastAPI, Query, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from sqlalchemy import create_engine, text +from datetime import datetime, timedelta +from typing import Optional + +engine = create_engine(conf.DATABASE_URL, pool_size=5, max_overflow=10, pool_pre_ping=True) + +ALLOWED_TABLES = { + 'mplogx', 'mplogx_xzzl', 'mplogx_tlxn', + 'mplogx_hknf_l1', 'mplogx_hknf_l2', 'mplogx_hknf_l3', + 'mplogx_wengfu', +} + +app = FastAPI(title="MTC 数据查询接口", version="1.0.0") + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.get("/api/latest") +def get_latest( + table: str = Query(..., description="表名，如 mplogx_wengfu"), + mpoint_id: str = Query(..., description="测点ID"), + limit: int = Query(1, ge=1, le=100), +): + """获取指定测点的最新数据""" + if table not in ALLOWED_TABLES: + raise HTTPException(400, f"不允许的表名: {table}") + sql = text(f""" + SELECT timex, mpoint_id, val_float, val_str + FROM "{table}" + WHERE mpoint_id = :mpoint_id + ORDER BY timex DESC + LIMIT :limit + """) + with engine.connect() as conn: + rows = conn.execute(sql, {"mpoint_id": mpoint_id, "limit": limit}).fetchall() + return [ + {"timex": r.timex.isoformat(), "mpoint_id": r.mpoint_id, + "val_float": r.val_float, "val_str": r.val_str} + for r in rows + ] + + +@app.get("/api/history") +def get_history( + table: str = Query(..., description="表名"), + mpoint_id: str = Query(..., description="测点ID"), + start: str = Query(..., description="起始时间 yyyy-MM-dd HH:mm:ss"), + end: Optional[str] = Query(None, description="结束时间，默认当前"), + limit: int = Query(1000, ge=1, le=10000), +): + """按时间范围查询测点历史数据""" + if table not in ALLOWED_TABLES: + raise HTTPException(400, f"不允许的表名: {table}") + try: + start_dt = datetime.strptime(start, "%Y-%m-%d %H:%M:%S") + except ValueError: + raise HTTPException(400, "start 格式错误，需要 yyyy-MM-dd HH:mm:ss") + end_dt = datetime.now() + if end: + try: + end_dt = datetime.strptime(end, "%Y-%m-%d %H:%M:%S") + except ValueError: + raise HTTPException(400, "end 格式错误，需要 yyyy-MM-dd HH:mm:ss") + + sql = text(f""" + SELECT timex, mpoint_id, val_float, val_str + FROM "{table}" + WHERE mpoint_id = :mpoint_id + AND timex >= :start AND timex < :end + ORDER BY timex + LIMIT :limit + """) + with engine.connect() as conn: + rows = conn.execute(sql, { + "mpoint_id": mpoint_id, "start": start_dt, "end": end_dt, "limit": limit + }).fetchall() + return [ + {"timex": r.timex.isoformat(), "mpoint_id": r.mpoint_id, + "val_float": r.val_float, "val_str": r.val_str} + for r in rows + ] + + +@app.get("/api/mpoints") +def list_mpoints( + table: str = Query(..., description="表名"), + keyword: Optional[str] = Query(None, description="模糊搜索测点ID"), + limit: int = Query(200, ge=1, le=1000), +): + """列出表内的测点ID（去重）""" + if table not in ALLOWED_TABLES: + raise HTTPException(400, f"不允许的表名: {table}") + if keyword: + sql = text(f""" + SELECT DISTINCT mpoint_id FROM "{table}" + WHERE mpoint_id LIKE :kw + ORDER BY mpoint_id LIMIT :limit + """) + params = {"kw": f"%{keyword}%", "limit": limit} + else: + sql = text(f""" + SELECT DISTINCT mpoint_id FROM "{table}" + ORDER BY mpoint_id LIMIT :limit + """) + params = {"limit": limit} + with engine.connect() as conn: + rows = conn.execute(sql, params).fetchall() + return [r.mpoint_id for r in rows] + + +@app.get("/api/count") +def get_count(): + """查询每个 mplogx 表的总条数""" + 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": [], "total": 0} + results = [] + total = 0 + for (tablename,) in tables: + count = conn.execute(text( + f'SELECT COUNT(*) FROM "{tablename}"' + )).scalar() + results.append({"table_name": tablename, "row_count": count}) + total += count + return {"tables": results, "total": total} + + +@app.get("/api/stats") +def get_stats(): + """各 mplogx 表的统计信息（复用 mqttc 中的逻辑）""" + 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": {}} + 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 t ORDER BY table_name + """)).fetchall() + + result = [] + for table_name, point_count_1h, approx in rows: + result.append({ + "table_name": table_name, + "point_count_1h": int(point_count_1h or 0), + "approximate_row_count": int(approx or 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), + }, + } diff --git a/load_mpoint_name.py b/load_mpoint_name.py new file mode 100644 index 0000000..9cb14d1 --- /dev/null +++ b/load_mpoint_name.py @@ -0,0 +1,101 @@ +"""一次性脚本：从「翁福 数据远传变量.xlsx」解析点位，建立并灌入 mpoint_name 映射表。 + +- 位号(mpoint_id) 由 mqttc.code_from_name 切出，与线上入库逻辑同一套规则，杜绝漂移。 +- point_name = 全名去掉尾部位号；full_name 保留 Excel 原文便于溯源。 +- 幂等：ON CONFLICT(mpoint_id) DO UPDATE，可反复执行。 + +用法: + python load_mpoint_name.py [Excel路径] [数据库URL] +默认读取桌面上的《数据远传变量.xlsx》，写入 conf.DATABASE_URL 指向的库； +可传第二个参数指定目标库（如线上 10.0.11.52）。 +""" +import os +import sys + +import openpyxl +from sqlalchemy import create_engine, text + +CUR_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, CUR_DIR) +import conf +from mqttc import code_from_name + +DEFAULT_XLSX = r"C:\Users\11825\Desktop\01-文档资料\资料\翁福采集\数据远传变量.xlsx" +TABLE = "wengfu_mpoint_name" + +DDL = f""" +CREATE TABLE IF NOT EXISTS {TABLE} ( + mpoint_id varchar PRIMARY KEY, + point_name text, + full_name text, + dtype varchar +) +""" + +UPSERT = f""" +INSERT INTO {TABLE} (mpoint_id, point_name, full_name, dtype) +VALUES (:mpoint_id, :point_name, :full_name, :dtype) +ON CONFLICT (mpoint_id) DO UPDATE +SET point_name = EXCLUDED.point_name, + full_name = EXCLUDED.full_name, + dtype = EXCLUDED.dtype +""" + + +def read_items(path): + """Excel 为左右两张并排的表：列 A/B 和 D/E 各是 (变量名, 类型)。""" + wb = openpyxl.load_workbook(path, read_only=True, data_only=True) + ws = wb["Sheet1"] + items = [] + for r in list(ws.iter_rows(values_only=True))[1:]: # 跳过表头 + r = list(r) + if r and r[0]: + items.append((str(r[0]).strip(), str(r[1]).strip() if len(r) > 1 and r[1] else '')) + if len(r) > 3 and r[3]: + items.append((str(r[3]).strip(), str(r[4]).strip() if len(r) > 4 and r[4] else '')) + return items + + +def split_row(full_name, dtype): + code = code_from_name(full_name) + point = full_name[:len(full_name) - len(code)].rstrip('_') if code and full_name.endswith(code) else full_name + return { + "mpoint_id": code, + "point_name": point or None, + "full_name": full_name, + "dtype": dtype or None, + } + + +def main(): + path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_XLSX + db_url = sys.argv[2] if len(sys.argv) > 2 else conf.DATABASE_URL + items = read_items(path) + rows = [split_row(name, typ) for name, typ in items] + + # 位号唯一性自检 + seen = {} + collisions = [] + for row in rows: + mid = row["mpoint_id"] + if mid in seen and seen[mid] != row["full_name"]: + collisions.append((mid, seen[mid], row["full_name"])) + seen[mid] = row["full_name"] + if collisions: + print(f"!! 位号冲突 {len(collisions)} 处（不同全名切出同一位号），请先核对：") + for mid, a, b in collisions[:20]: + print(f" {mid}: {a} <> {b}") + sys.exit(1) + + engine = create_engine(db_url, connect_args={'connect_timeout': 10}) + with engine.begin() as conn: + conn.execute(text(DDL)) + for row in rows: + conn.execute(text(UPSERT), row) + total = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar() + + print(f"目标库写入完成：读取 {len(items)} 条，去重后位号 {len(seen)} 个，{TABLE} 表内现有 {total} 行。") + + +if __name__ == '__main__': + main() diff --git a/mqttc.py b/mqttc.py index d1050ee..dd52d03 100644 --- a/mqttc.py +++ b/mqttc.py @@ -1,17 +1,19 @@ 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 +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 +from threading import Thread, Lock +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import traceback import queue @@ -36,20 +38,47 @@ 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_zlwd', - conf.MQTT_TOPIC_tl: 'mplogx_tlsn', + 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_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 @@ -72,7 +101,11 @@ def worker(topic: str): try: payload = q.get(timeout=60) if payload is not None: - save_items(topic, payload) + 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 @@ -85,31 +118,159 @@ def worker(topic: str): ).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 - item_list = json.loads(payload) - sql_str = f"INSERT INTO {table_name} (timex, mpoint_id, val_float, val_str) VALUES (:timex, :mpoint_id, :val_float, :val_str)" + 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 item in item_list: - timex = datetime.strptime(item['time'], "%Y%m%d%H%M%S").replace(tzinfo=tz.gettz('Asia/Shanghai')) - mpoint_id = item['name'] - val_float = None - val_str = None + 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: - try: - val_float = float(item["value"]) - except Exception: - val_str = item["value"] - save_list.append({"timex": timex, "mpoint_id": mpoint_id, "val_float": val_float, "val_str": val_str}) + 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: @@ -131,6 +292,89 @@ def save_items(topic, payload): +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") @@ -153,6 +397,7 @@ def on_disconnect(mqttc: mqtt.Client, userdata, disconnect_flags, reason_code, p 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 @@ -164,9 +409,11 @@ 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) diff --git a/supervisor_api.conf b/supervisor_api.conf new file mode 100644 index 0000000..6fb6697 --- /dev/null +++ b/supervisor_api.conf @@ -0,0 +1,13 @@ +[program:mtc_api] +command=/home/zc/mtc/.venv/bin/uvicorn api:app --host 0.0.0.0 --port 5900 +directory=/home/zc/mtc +user=root +autostart=true +autorestart=true +startsecs=5 +startretries=3 +redirect_stderr=true +stdout_logfile=/var/log/supervisor/mtc_api.log +stdout_logfile_maxbytes=10MB +stdout_logfile_backups=3 +environment=LANG="en_US.UTF-8",LC_ALL="en_US.UTF-8"