ai_member_xiaobian/tmp/format_rich_text.py

#!/usr/bin/env python3
"""
U21 L4 剧本 - 知识点标粗+红输出
格式: type: "text" + textFormatRuns（已验证可行）

- 中互动/核心互动 → 知识点标粗+红色（输出）
- TL行 → 知识点标粗（输入）
"""
import json, re, requests, os, sys

CRED_DIR = "/root/.openclaw/credentials"
BASE_URL = "https://open.feishu.cn/open-apis"

KPS_WORDS = ["flat", "hall", "room", "know"]
KPS_PHRASES = ["do you know", "what is in"]

def get_token(name="xiaobian"):
    config_path = os.path.join(CRED_DIR, name, "config.json")
    with open(config_path) as f:
        config = json.load(f)
    app_id = config["apps"][0]["appId"]
    app_secret = config["apps"][0]["appSecret"]
    resp = requests.post(f"{BASE_URL}/auth/v3/tenant_access_token/internal",
        json={"app_id": app_id, "app_secret": app_secret}, timeout=10)
    return resp.json()["tenant_access_token"]

def parse_kps(text):
    """Find all KP spans. Returns sorted non-overlapping (start, end) list."""
    lower = text.lower()
    spans = []
    
    # Phrase KPs first (take precedence)
    for phrase in KPS_PHRASES:
        idx = 0
        while True:
            idx = lower.find(phrase, idx)
            if idx == -1:
                break
            spans.append((idx, idx + len(phrase)))
            idx += len(phrase)
    
    # Word KPs (word boundary match)
    for word in KPS_WORDS:
        for m in re.finditer(r'\b' + re.escape(word) + r'\b', text, re.IGNORECASE):
            spans.append((m.start(), m.end()))
    
    # Sort and deduplicate overlaps
    spans.sort(key=lambda x: x[0])
    result = []
    last_end = 0
    for s, e in spans:
        if s >= last_end:
            result.append((s, e))
            last_end = e
    return result

def build_rich_text_cell(text, is_output):
    """Build a rich text cell using type:text + textFormatRuns."""
    spans = parse_kps(text)
    if not spans:
        return text
    
    format_runs = []
    pos = 0
    for i, (start, end) in enumerate(spans):
        if pos < start:
            format_runs.append({"format": {}, "startIndex": pos})
        
        f = {"bold": True}
        if is_output:
            f["foreColor"] = {"red": 1.0, "green": 0.0, "blue": 0.0, "alpha": 0.0}
        format_runs.append({"format": f, "startIndex": start})
        pos = end
    
    if pos < len(text):
        format_runs.append({"format": {}, "startIndex": pos})
    
    # Always start with a run at index 0
    if not format_runs or format_runs[0]["startIndex"] != 0:
        format_runs.insert(0, {"format": {}, "startIndex": 0})
    
    return {
        "type": "text",
        "text": text,
        "textFormatRuns": format_runs
    }

def main():
    token = get_token()
    TKN = "NiajsPDjXhQHn8tURCeck8zlndd"
    SHT = "3O2sso"
    
    with open("tmp/u21_l4_merged.json", "r", encoding="utf-8") as f:
        data = json.load(f)
    
    rows = data["rows"]
    
    # Determine type for each row (blank inherits from above)
    current_type = ""
    row_types = []
    for row in rows:
        rt = row[0] if len(row) > 0 and row[0] else ""
        if rt in ("TL", "中互动", "核心互动"):
            current_type = rt
        row_types.append(current_type)
    
    values = []
    format_count = 0
    
    for ri, row in enumerate(rows):
        out_row = []
        is_output = row_types[ri] in ("中互动", "核心互动")
        
        for ci, cell in enumerate(row):
            if ci == 3 and isinstance(cell, str) and cell.strip():
                formatted = build_rich_text_cell(cell, is_output)
                out_row.append(formatted)
                if isinstance(formatted, dict):
                    format_count += 1
            elif ci == 3 and isinstance(cell, str):
                out_row.append(cell)
            else:
                out_row.append(cell if cell else "")
        
        values.append(out_row)
    
    num_rows = len(values)
    num_cols = 4
    col_letter = "D"
    
    print(f"Writing {num_rows} rows, {format_count} formatted cells...")
    
    resp = requests.put(
        f"{BASE_URL}/sheets/v2/spreadsheets/{TKN}/values",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={
            "valueRange": {
                "range": f"{SHT}!A1:{col_letter}{num_rows}",
                "values": values,
            },
        },
        timeout=60,
    )
    
    result = resp.json()
    if result.get("code") != 0:
        print(f"❌ Write failed: {result}", file=sys.stderr)
        sys.exit(1)
    
    print(json.dumps({"status": "ok", "rows": num_rows, "columns": num_cols, "formatted": format_count}))
    
    # Set header style
    sr = requests.put(
        f"{BASE_URL}/sheets/v2/spreadsheets/{TKN}/style",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={
            "appendStyle": {
                "range": f"{SHT}!A1:D1",
                "style": {"font": {"bold": True}, "backColor": "#e8e8e8"},
            },
        },
        timeout=10,
    )
    print(json.dumps({"style_status": "ok"}))

if __name__ == "__main__":
    main()