dy-dashboard/daily_knowledge_note_from_logs.py

#!/usr/bin/env python3
"""Generate a deterministic daily knowledge note from structured dashboard_logs.

This replaces the old LLM-based whole-conversation replay. It intentionally reads
only the day's structured operation log rows, so a no-op day costs no model tokens.
"""
from __future__ import annotations

import argparse
import os
import sys
from collections import Counter
from datetime import date, datetime, timedelta
from pathlib import Path

import pymysql

ROOT = Path('/home/gxd/蝶雅供应链')
INDEX = ROOT / '知识库' / '索引.md'
TOPDIR = ROOT / '知识库' / '工作指南' / '训练记录'


def db_config():
    sys.path.insert(0, '/home/gxd/.openclaw/workspace/dy-dashboard')
    from config import DB_CONFIG
    return DB_CONFIG


def fetch_rows(day: date):
    start = datetime.combine(day, datetime.min.time())
    end = start + timedelta(days=1)
    cfg = db_config()
    conn = pymysql.connect(**cfg, cursorclass=pymysql.cursors.DictCursor)
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT l.created_at, COALESCE(u.real_name, u.username, '系统') AS user_name,
                       COALESCE(NULLIF(l.page, ''), '未分类') AS page,
                       l.action, COALESCE(l.field, '') AS field,
                       l.result_status, l.risk_level,
                       COALESCE(l.detail, '') AS detail
                FROM dashboard_logs l
                LEFT JOIN dashboard_users u ON u.id = l.user_id
                WHERE l.created_at >= %s AND l.created_at < %s
                  AND l.action <> '登录'
                ORDER BY l.created_at ASC, l.id ASC
                """,
                (start, end),
            )
            return cur.fetchall()
    finally:
        conn.close()


def clip(value, limit=220):
    text = ' '.join(str(value or '').split())
    return text if len(text) <= limit else text[: limit - 1] + '…'


def build_content(day: date, rows):
    title = f'{day.isoformat()} 每日训练笔记（结构化操作日志）'
    lines = [
        f'> 来源：蝶雅采购智能体平台结构化操作日志，{day.isoformat()}',
        '> 类型：系统操作摘要',
        '',
        f'## 当日概况',
        f'- 结构化操作记录：{len(rows)} 条',
    ]
    if not rows:
        lines.append('- 当日没有可归档的业务操作记录，未回放历史会话。')
        return title, '\n'.join(lines) + '\n'

    pages = Counter(r['page'] for r in rows)
    actions = Counter(r['action'] for r in rows)
    results = Counter(r['result_status'] for r in rows)
    risks = Counter(r['risk_level'] for r in rows)
    lines += [
        f"- 涉及页面：{len(pages)} 个",
        f"- 成功/失败/拒绝：{results.get('success', 0)}/{results.get('failed', 0)}/{results.get('denied', 0)}",
        f"- 高风险记录：{risks.get('high', 0)} 条",
        '',
        '## 页面与操作汇总',
        '| 页面 | 操作数 |',
        '|------|------:|',
    ]
    for page, count in pages.most_common():
        lines.append(f'| {page} | {count} |')

    lines += ['', '## 操作类型汇总', '| 操作 | 次数 |', '|------|------:|']
    for action, count in actions.most_common(30):
        lines.append(f'| {action} | {count} |')

    exceptional = [r for r in rows if r['result_status'] != 'success' or r['risk_level'] in ('high', 'medium')]
    if exceptional:
        lines += ['', '## 异常与高风险记录', '| 时间 | 用户 | 页面 | 操作 | 结果 | 说明 |', '|------|------|------|------|------|------|']
        for r in exceptional[:80]:
            ts = r['created_at'].strftime('%H:%M:%S') if hasattr(r['created_at'], 'strftime') else str(r['created_at'])[-8:]
            lines.append(f"| {ts} | {clip(r['user_name'], 24)} | {clip(r['page'], 24)} | {clip(r['action'], 28)} | {r['result_status']} | {clip(r['detail'])} |")
    else:
        lines += ['', '## 异常与高风险记录', '- 无失败、拒绝或中高风险操作记录。']

    lines += ['', '## 结构化操作明细', '| 时间 | 用户 | 页面 | 操作 | 字段 | 结果 |', '|------|------|------|------|------|------|']
    for r in rows[:300]:
        ts = r['created_at'].strftime('%H:%M:%S') if hasattr(r['created_at'], 'strftime') else str(r['created_at'])[-8:]
        lines.append(f"| {ts} | {clip(r['user_name'], 24)} | {clip(r['page'], 24)} | {clip(r['action'], 28)} | {clip(r['field'], 20)} | {r['result_status']} |")
    if len(rows) > 300:
        lines.append(f'- 明细超过300条，以上仅展示前300条；完整记录保留在数据库 `dashboard_logs`。')
    return title, '\n'.join(lines) + '\n'


def write_note(day: date, title: str, content: str):
    TOPDIR.mkdir(parents=True, exist_ok=True)
    path = TOPDIR / f'{day.isoformat()}-{title}.md'
    full = f'# {title}\n\n{content}\n---\n\n*由结构化操作日志任务自动生成*\n'
    path.write_text(full, encoding='utf-8')
    if INDEX.exists():
        index = INDEX.read_text(encoding='utf-8')
        marker = f'[{title}]'
        if marker not in index:
            row = f'| [{title}](工作指南/训练记录/{path.name}) | 基于 dashboard_logs 的结构化操作摘要，不回放历史会话 | {day.isoformat()} |\n'
            anchor = '| 文章 | 摘要 | 更新 |'
            pos = index.find(anchor, index.find('### 训练记录'))
            if pos >= 0:
                header_end = index.find('\n', pos) + 1
                separator_end = index.find('\n', header_end) + 1
                insert_at = separator_end if separator_end > header_end else header_end
                index = index[:insert_at] + row + index[insert_at:]
            else:
                index += f'\n{row}'
            INDEX.write_text(index, encoding='utf-8')
    return path


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--date', default=date.today().isoformat())
    parser.add_argument('--dry-run', action='store_true')
    args = parser.parse_args()
    day = datetime.strptime(args.date, '%Y-%m-%d').date()
    rows = fetch_rows(day)
    title, content = build_content(day, rows)
    if args.dry_run:
        print(f'{day.isoformat()}: {len(rows)} structured log rows')
        print(content[:1200])
        return 0
    path = write_note(day, title, content)
    print(f'STRUCTURED_NOTE_WRITTEN rows={len(rows)} path={path}')
    return 0


if __name__ == '__main__':
    raise SystemExit(main())