mescloud/db/mysql/scale_spot_welding_current.py

# -*- coding: utf-8 -*-
"""
一次性把历史点焊电流放大 10 倍，与当前入库规则对齐。
已放大过的记录（电流绝对值 >= 3）会跳过，可重复执行。
"""
import argparse
import json
import sys
from decimal import Decimal

import pymysql
from pymysql.cursors import SSCursor

SCALE_THRESHOLD = Decimal("3")
CURRENT_KEYS = ("robot1Current", "robot2Current")


def to_decimal(value):
    if value is None or value == "":
        return None
    try:
        return Decimal(str(value))
    except Exception:
        return None


def scale_content(content):
    if not content:
        return None
    try:
        payload = json.loads(content)
    except Exception:
        return None
    if not isinstance(payload, dict) or payload.get("weldingType") != "spot":
        return None
    samples = payload.get("samples") or []
    if not isinstance(samples, list) or not samples:
        return None

    max_current = Decimal("0")
    for sample in samples:
        if not isinstance(sample, dict):
            continue
        for key in CURRENT_KEYS:
            current = to_decimal(sample.get(key))
            if current is not None and abs(current) > max_current:
                max_current = abs(current)

    if max_current <= 0 or max_current >= SCALE_THRESHOLD:
        return None

    for sample in samples:
        if not isinstance(sample, dict):
            continue
        for key in CURRENT_KEYS:
            current = to_decimal(sample.get(key))
            if current is None:
                continue
            scaled = current * 10
            if scaled == scaled.to_integral_value():
                sample[key] = int(scaled)
            else:
                sample[key] = float(scaled)
    return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))


def main():
    parser = argparse.ArgumentParser(description="Scale historical spot welding current by 10")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=3306)
    parser.add_argument("--user", default="root")
    parser.add_argument("--password", required=True)
    parser.add_argument("--database", default="mes_cloud_ahdy")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    read_conn = pymysql.connect(
        host=args.host, port=args.port, user=args.user, password=args.password,
        database=args.database, charset="utf8mb4", autocommit=False
    )
    write_conn = pymysql.connect(
        host=args.host, port=args.port, user=args.user, password=args.password,
        database=args.database, charset="utf8mb4", autocommit=False
    )
    scanned = 0
    updated = 0
    skipped = 0
    failed = 0
    try:
        read_cur = read_conn.cursor(SSCursor)
        write_cur = write_conn.cursor()
        read_cur.execute(
            "SELECT id, create_date, content FROM mes_product_arc_welding "
            "WHERE content LIKE %s",
            ('%"weldingType":"spot"%',)
        )
        for row_id, create_date, content in read_cur:
            scanned += 1
            try:
                new_content = scale_content(content)
            except Exception as ex:
                failed += 1
                print("parse fail id=%s date=%s err=%s" % (row_id, create_date, ex))
                continue
            if not new_content:
                skipped += 1
            elif not args.dry_run:
                write_cur.execute(
                    "UPDATE mes_product_arc_welding SET content=%s WHERE id=%s AND create_date=%s",
                    (new_content, row_id, create_date)
                )
                updated += 1
                if updated % 200 == 0:
                    write_conn.commit()
                    print("updated %s / scanned %s" % (updated, scanned))
            else:
                updated += 1
                if updated % 200 == 0:
                    print("dry-run would update %s / scanned %s" % (updated, scanned))
        if not args.dry_run:
            write_conn.commit()
        print("done scanned=%s updated=%s skipped=%s failed=%s dry_run=%s"
              % (scanned, updated, skipped, failed, args.dry_run))
    finally:
        read_conn.close()
        write_conn.close()


if __name__ == "__main__":
    sys.exit(main())