netgescon-day0/scripts/ai/summarize_chat_session_compact.py

#!/usr/bin/env python3
import argparse
import glob
import json
import os
import re
from datetime import datetime


def clean_text(text: str) -> str:
    return re.sub(r"\s+", " ", (text or "")).strip()


def find_latest_quarantine():
    candidates = sorted(glob.glob(os.path.expanduser('~/vscode-chat-backups/quarantine-*')), reverse=True)
    return candidates[0] if candidates else None


def find_largest_chat_json(base_dir):
    files = glob.glob(os.path.join(base_dir, '**', 'chatSessions', '*.json'), recursive=True)
    return max(files, key=lambda p: os.path.getsize(p)) if files else None


def get_request_text(req):
    msg = req.get('message')
    if isinstance(msg, dict):
        return clean_text(msg.get('text') or msg.get('value') or '')
    if isinstance(msg, str):
        return clean_text(msg)
    return ''


def get_response_text(req):
    response = req.get('response', [])
    parts = []
    if isinstance(response, list):
        for item in response:
            if isinstance(item, dict):
                value = item.get('value') or item.get('text') or ''
                if isinstance(value, str) and value.strip():
                    parts.append(clean_text(value))
            elif isinstance(item, str) and item.strip():
                parts.append(clean_text(item))
    return clean_text(' '.join(parts))


def extract_errors(texts):
    code_re = re.compile(r'(SQLSTATE\[[A-Z0-9]+\]|HTTP\s*\d{3}|ERR_[A-Z0-9_]+|Integrity constraint violation|General error)', re.IGNORECASE)
    fileline_re = re.compile(r'([A-Za-z0-9_./-]+\.(?:php|blade\.php|js|ts|sql|py|sh|json|md|conf))(?::(\d+))?')
    page_re = re.compile(r'(https?://[^\s)]+|/admin[^\s)]+|/superadmin[^\s)]+)')
    resolved_re = re.compile(r'\b(risolt[oaie]?|resolved|fixed|ok|funziona|chius[oa])\b', re.IGNORECASE)

    records = {}
    for text in texts:
        if not text:
            continue
        code_match = code_re.search(text)
        if not code_match and ('errore' not in text.lower() and 'exception' not in text.lower()):
            continue

        code = clean_text(code_match.group(1)) if code_match else 'n/a'
        file_match = fileline_re.search(text)
        file_path = file_match.group(1) if file_match else 'n/a'
        line = file_match.group(2) if (file_match and file_match.group(2)) else 'n/a'
        page_match = page_re.search(text)
        page = clean_text(page_match.group(1)) if page_match else 'n/a'

        if code != 'n/a':
            keyword = code.replace(' ', '_').lower()
        elif file_path != 'n/a':
            keyword = os.path.basename(file_path).lower().replace('.', '_')
        elif page != 'n/a':
            keyword = page.split('/')[-1].lower().replace('-', '_') or 'errore_generico'
        else:
            keyword = 'errore_generico'

        note = clean_text(text)[:220]
        status = 'resolved' if resolved_re.search(text) else 'open'

        if code == 'n/a' and file_path == 'n/a' and page == 'n/a':
            continue

        if keyword in {'errore_generico', 'n_a'}:
            continue

        key = (code.lower(), file_path.lower(), line, page.lower())
        prev = records.get(key)
        if prev is None:
            records[key] = {
                'keyword': keyword,
                'code': code,
                'file': file_path,
                'line': line,
                'page': page,
                'note': note,
                'status': status,
            }
        else:
            if prev['status'] != 'open' and status == 'open':
                prev['status'] = 'open'
            if len(note) > len(prev['note']):
                prev['note'] = note

    return list(records.values())


def summarize(chat_file, output_dir, keyword_filter=''):
    with open(chat_file, 'r', encoding='utf-8', errors='replace') as f:
        data = json.load(f)

    requests = data.get('requests', []) if isinstance(data, dict) else []
    session_id = data.get('sessionId', os.path.splitext(os.path.basename(chat_file))[0])
    creation = data.get('creationDate')
    last_date = data.get('lastMessageDate')

    texts = []
    for req in requests:
        if not isinstance(req, dict):
            continue
        ut = get_request_text(req)
        at = get_response_text(req)
        if ut:
            texts.append(ut)
        if at:
            texts.append(at)

    errors = extract_errors(texts)

    kf = keyword_filter.strip().lower()
    if kf:
        errors = [e for e in errors if kf in e['keyword'] or kf in e['code'].lower() or kf in e['file'].lower() or kf in e['page'].lower() or kf in e['note'].lower()]

    open_errors = [e for e in errors if e['status'] == 'open']
    resolved_errors = [e for e in errors if e['status'] == 'resolved']

    def score(err):
        s = 0
        if err['code'] != 'n/a':
            s += 3
        if err['page'] != 'n/a':
            s += 2
        if err['file'] != 'n/a':
            s += 2
        if err['line'] != 'n/a':
            s += 1
        return s

    open_errors = sorted(open_errors, key=score, reverse=True)[:15]

    def fmt_ts(ts):
        if isinstance(ts, (int, float)):
            try:
                return datetime.fromtimestamp(ts / 1000).strftime('%Y-%m-%d %H:%M:%S')
            except Exception:
                return str(ts)
        return str(ts) if ts is not None else 'n/a'

    os.makedirs(output_dir, exist_ok=True)
    out_path = os.path.join(output_dir, f'{session_id}.compact.md')

    with open(out_path, 'w', encoding='utf-8') as out:
        out.write(f'# Chat Compact Summary: {session_id}\n\n')
        out.write(f'- Source: {chat_file}\n')
        out.write(f'- Requests: {len(requests)}\n')
        out.write(f'- Creation: {fmt_ts(creation)}\n')
        out.write(f'- Last message: {fmt_ts(last_date)}\n')
        out.write(f'- Open errors: {len(open_errors)}\n')
        out.write(f'- Resolved errors (omitted): {len(resolved_errors)}\n\n')

        out.write('## Errori essenziali (solo aperti)\n')
        if not open_errors:
            out.write('- Nessun errore aperto rilevato con i pattern correnti.\n')
        else:
            out.write('| Keyword | Codice | Pagina | File | Riga |\n')
            out.write('|---|---|---|---|---|\n')
            for e in open_errors:
                page = e['page'].replace('|', '/')
                filep = e['file'].replace('|', '/')
                out.write(f"| {e['keyword']} | {e['code']} | {page} | {filep} | {e['line']} |\n")

            out.write('\n## Note tecniche minime\n')
            for e in open_errors[:8]:
                out.write(f"- [{e['keyword']}] {e['note']}\n")

        out.write('\n## Prompt bug rapido\n')
        out.write('- Identifica il bug con keyword <KEYWORD> e lavora solo su codice/errore attivo; ometti i bug marcati risolti.\n')

    return out_path


def main():
    parser = argparse.ArgumentParser(description='Create compact essential bug-oriented summary from a chat session')
    parser.add_argument('--chat-file', default='', help='Path to chatSessions/*.json')
    parser.add_argument('--output-dir', default='docs/ai/restart/compact-summaries', help='Compact summaries output folder')
    parser.add_argument('--keyword', default='', help='Optional keyword filter (error code/page/file)')
    args = parser.parse_args()

    chat_file = args.chat_file.strip()
    if not chat_file:
        quarantine = find_latest_quarantine()
        if not quarantine:
            raise SystemExit('No quarantine found and --chat-file missing')
        chat_file = find_largest_chat_json(quarantine)
        if not chat_file:
            raise SystemExit('No chat session JSON found in latest quarantine')

    out = summarize(chat_file, args.output_dir, args.keyword)
    print(f'Compact summary created: {out}')


if __name__ == '__main__':
    main()