zyt/app/artifacts/debug_lxh_psb.py

# -*- coding: utf-8 -*-
import json
import sys
from collections import defaultdict
from datetime import datetime

import pymysql

sys.stdout.reconfigure(encoding="utf-8")

conn = pymysql.connect(
    host="127.0.0.1", user="root", password="root", database="zyt", charset="utf8mb4"
)
cur = conn.cursor(pymysql.cursors.DictCursor)

START, END = 1786464000, 1786550399  # 2026-08-12


def q(sql, args=None):
    cur.execute(sql, args or ())
    return cur.fetchall()


def parse_follow(follow):
    if isinstance(follow, str) and follow:
        try:
            return json.loads(follow)
        except Exception:
            return []
    return follow if isinstance(follow, list) else []


def contact_info(ext, user_id):
    rows = q(
        "SELECT id, name, follow_users, delete_time FROM zyt_qywx_external_contact WHERE external_userid=%s ORDER BY id DESC",
        (ext,),
    )
    name = None
    remark = None
    add_way = None
    still = False
    createtime = None
    for c in rows:
        if name is None:
            name = c.get("name")
        for fu in parse_follow(c.get("follow_users")):
            if not isinstance(fu, dict):
                continue
            uid = str(fu.get("userid") or fu.get("UserId") or "")
            if uid == user_id:
                still = True
                remark = fu.get("remark") or fu.get("Remark") or remark
                add_way = fu.get("add_way", fu.get("AddWay"))
                createtime = fu.get("createtime") or fu.get("create_time") or createtime
        if remark:
            break
    return {
        "name": name,
        "remark": remark,
        "add_way": add_way,
        "still_following": still,
        "follow_createtime": createtime,
        "follow_createtime_at": datetime.fromtimestamp(int(createtime)).strftime("%Y-%m-%d %H:%M:%S")
        if createtime
        else None,
        "contact_rows": len(rows),
    }


def events_for(user_id, ext):
    return q(
        """
        SELECT change_type, FROM_UNIXTIME(event_time) AS event_at
        FROM zyt_qywx_external_contact_event
        WHERE user_id=%s AND external_userid=%s
          AND change_type IN ('add_external_contact','del_follow_user','del_external_contact','msg_audit_approved')
        ORDER BY event_time
        """,
        (user_id, ext),
    )


def analyze(label, user_id):
    print(f"\n========== {label} / {user_id} ==========")
    admins = q(
        "SELECT id, name, work_wechat_userid, disable, delete_time FROM zyt_admin WHERE work_wechat_userid=%s OR name LIKE %s",
        (user_id, f"%{label}%"),
    )
    print("admin", json.dumps(admins, ensure_ascii=False, default=str))

    raw = q(
        """
        SELECT e.external_userid, COUNT(*) AS add_events,
               FROM_UNIXTIME(MIN(e.event_time)) AS first_in_day,
               FROM_UNIXTIME(MAX(e.event_time)) AS last_in_day
        FROM zyt_qywx_external_contact_event e
        WHERE e.change_type='add_external_contact'
          AND e.user_id=%s
          AND e.event_time BETWEEN %s AND %s
          AND e.external_userid <> ''
        GROUP BY e.external_userid
        ORDER BY MIN(e.event_time)
        """,
        (user_id, START, END),
    )
    print("raw unique adds", len(raw))

    def filtered(extra_prev=False):
        sql = """
        SELECT e.external_userid, MIN(e.event_time) AS first_add,
               FROM_UNIXTIME(MIN(e.event_time)) AS first_add_at
        FROM zyt_qywx_external_contact_event e
        WHERE e.change_type='add_external_contact'
          AND e.user_id=%s
          AND e.event_time BETWEEN %s AND %s
          AND e.external_userid <> ''
          AND EXISTS (
            SELECT 1 FROM zyt_qywx_external_contact_event audit_e
            WHERE audit_e.user_id=e.user_id
              AND audit_e.external_userid=e.external_userid
              AND audit_e.change_type='msg_audit_approved'
              AND audit_e.event_time >= e.event_time
              AND audit_e.event_time <= %s
          )
          AND NOT EXISTS (
            SELECT 1 FROM zyt_qywx_external_contact_event del_e
            WHERE del_e.user_id=e.user_id
              AND del_e.external_userid=e.external_userid
              AND del_e.change_type='del_external_contact'
              AND del_e.event_time >= e.event_time
              AND del_e.event_time <= %s
          )
        """
        args = [user_id, START, END, END, END]
        if extra_prev:
            sql += """
          AND NOT EXISTS (
            SELECT 1 FROM zyt_qywx_external_contact_event prev_e
            WHERE prev_e.user_id=e.user_id
              AND prev_e.external_userid=e.external_userid
              AND prev_e.change_type='add_external_contact'
              AND prev_e.event_time < %s
          )
            """
            args.append(START)
        sql += " GROUP BY e.external_userid ORDER BY first_add"
        return q(sql, args)

    old_pairs = filtered(False)
    new_pairs = filtered(True)
    print("system OLD (before readd fix)", len(old_pairs))
    print("system NEW (after readd fix)", len(new_pairs))

    old_ids = {p["external_userid"] for p in old_pairs}
    new_ids = {p["external_userid"] for p in new_pairs}
    raw_ids = {p["external_userid"] for p in raw}

    dropped_no_audit_or_del = raw_ids - old_ids
    dropped_readd = old_ids - new_ids

    print("\n--- dropped vs raw (no audit / customer-deleted) ---")
    for ext in sorted(dropped_no_audit_or_del, key=lambda x: next(r["first_in_day"] for r in raw if r["external_userid"] == x)):
        info = contact_info(ext, user_id)
        ev = events_for(user_id, ext)
        print(json.dumps({"ext": ext, **info, "events": ev, "raw": next(r for r in raw if r["external_userid"] == ext)}, ensure_ascii=False, default=str))

    print("\n--- dropped as re-add (old counted, new excluded) ---")
    for ext in sorted(dropped_readd):
        info = contact_info(ext, user_id)
        ev = events_for(user_id, ext)
        print(json.dumps({"ext": ext, **info, "events": ev}, ensure_ascii=False, default=str))

    print("\n--- inheritance add_way 201/202 among NEW pairs ---")
    for p in new_pairs:
        info = contact_info(p["external_userid"], user_id)
        if info["add_way"] in (201, 202):
            print(json.dumps({"ext": p["external_userid"], **info, "first_add_at": p["first_add_at"]}, ensure_ascii=False, default=str))

    print("\n--- NEW counted list ---")
    for p in new_pairs:
        info = contact_info(p["external_userid"], user_id)
        types = [e["change_type"] for e in events_for(user_id, p["external_userid"])]
        print(json.dumps({
            "first_add_at": p["first_add_at"],
            "name": info["name"],
            "remark": info["remark"],
            "add_way": info["add_way"],
            "follow_createtime_at": info["follow_createtime_at"],
            "still_following": info["still_following"],
            "event_types": sorted(set(types)),
        }, ensure_ascii=False, default=str))

    # people in old but check missing audit on the in-day add specifically
    print("\n--- raw adds missing audit after THAT add or del_external ---")
    for r in raw:
        ext = r["external_userid"]
        if ext in old_ids:
            continue
        ev = events_for(user_id, ext)
        print(json.dumps({"ext": ext, "raw": r, "events": ev, **contact_info(ext, user_id)}, ensure_ascii=False, default=str))


analyze("李晓焕", "LiXiaoHuan")
analyze("彭世博", "GeiWoNaHaoDeA")