#!/usr/bin/env python3 """ Daily Inspection Report Script Generates markdown report locally, upload is handled by separate cronjob """ import os import sys import json import re import subprocess from datetime import datetime, timedelta # Load environment variables from bashrc try: with open(os.path.expanduser('~/.bashrc')) as f: for line in f: if 'CRONJOB_TEAM_ADMIN' in line and line.startswith('export'): match = re.search(r'export (\w+)=\"(.*?)\"', line) if match: key, value = match.groups() os.environ.setdefault(key, value) except Exception: # Ignore parse errors # Best-effort: bashrc loading, missing vars handled elsewhere pass # Path for unmapped open_ids output UNMAPPED_OUTPUT_FILE = os.path.join(os.path.dirname(__file__), "unmapped_open_id_list.json") MAPPING_INPUT_FILE = os.path.join(os.path.dirname(__file__), "open_id_and_user_mapping.json") def load_existing_mappings(): """Load existing open_id to user_name mappings""" if not os.path.exists(MAPPING_INPUT_FILE): return {} try: with open(MAPPING_INPUT_FILE) as f: data = json.load(f) return data.get("all_users", {}) except Exception: # Ignore JSON parse errors return {} def save_unmapped_open_ids(container_unmapped): """Save unmapped open_ids to output file""" with open(UNMAPPED_OUTPUT_FILE, 'w') as f: json.dump(container_unmapped, f, indent=2) print(f"Unmapped open_ids saved to {UNMAPPED_OUTPUT_FILE}") return container_unmapped # TODO: Add user name lookup function when Feishu API permission is granted # def get_user_names(user_ids): # """Get user names from Feishu API""" # ... # Configuration - from environment variables SPACE_ID = os.environ.get("CRONJOB_FEISHU_TEAM_WIKI_SPACE_ID", "7594383226739674063") PARENT_NODE = os.environ.get("CRONJOB_FEISHU_DAILY_INSPECTION_ROOT_NODE", "MEPpwyZ8biP5FrkzahDctohTnJc") WIKI_BASE_URL = "https://pingcap.feishu.cn/wiki" # Get Feishu App credentials APP_ID = os.environ.get("CRONJOB_TEAM_ADMIN_APP_ID", "") APP_SECRET = os.environ.get("CRONJOB_TEAM_ADMIN_APP_SECRET", "") USER_ID = os.environ.get("CRONJOB_FEISHU_DAILY_INSPECTION_NOTIFY_USER", "ou_f442e20fd980363ea486b5a6cd8d2a2f") # md2feishu script path AI_CONFIG_REPO = os.path.expanduser(os.environ.get("AI_CONFIG_REPO", "~/github/data-service-infra-ai-configurations")) MD2FEISHU_SCRIPT = os.path.join(AI_CONFIG_REPO, "skills/team-common/md2feishu/scripts/md2feishu.py") # Upload trigger directory (not used anymore) # TRIGGER_DIR = os.path.expanduser("~/openclaw_local2feishu") def create_wiki_node(title): """Create wiki node and return doc_token and wiki_token""" import requests # Get token first token_resp = requests.post( "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", json={"app_id": APP_ID, "app_secret": APP_SECRET} ) token = token_resp.json()["tenant_access_token"] resp = requests.post( f"https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes", json={ "obj_type": "docx", "parent_node_token": PARENT_NODE, "node_type": "origin", "title": title, "origin": {"token": ""} }, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"} ) result = resp.json() if result.get("code") != 0: raise Exception(f"创建节点失败: {result}") # Return both obj_token (for content) and node_token (for URL) node = result["data"]["node"] return node["obj_token"], node["node_token"] def send_notification(msg): """Send notification via API""" import requests import json resp = requests.post( "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", json={"app_id": APP_ID, "app_secret": APP_SECRET} ) token = resp.json()["tenant_access_token"] resp = requests.post( "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id", json={ "receive_id": USER_ID, "msg_type": "text", "content": json.dumps({"text": msg}) }, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"} ) if resp.json().get("code") != 0: print(f"Notification error: {resp.json()}") return resp.json().get("code") == 0 def calculate_tokens(agent_dir, target_date_str): """Calculate token usage from sessions.jsonl files (including .reset files)""" sessions_dir = os.path.join(agent_dir, "sessions") if not os.path.exists(sessions_dir): return 0 total = 0 try: for f in os.listdir(sessions_dir): if '.jsonl' not in f: continue filepath = os.path.join(sessions_dir, f) try: with open(filepath) as file: for line in file: if target_date_str not in line: continue try: msg = json.loads(line) usage = msg.get('message', {}).get('usage', {}) tokens = usage.get('totalTokens', 0) total += tokens except Exception: # Ignore parse errors continue except Exception: # Ignore parse errors pass except Exception: # Ignore parse errors pass return total def calculate_messages(agent_dir, target_date_str): """Calculate REAL user message count from sessions.jsonl files""" sessions_dir = os.path.join(agent_dir, "sessions") if not os.path.exists(sessions_dir): return 0 total = 0 try: for f in os.listdir(sessions_dir): # Include both .jsonl and .jsonl.reset.* files (they contain all session messages) if '.jsonl' not in f: continue filepath = os.path.join(sessions_dir, f) try: with open(filepath) as file: for line in file: if target_date_str not in line: continue try: msg = json.loads(line) # Only count user role messages - no need to validate sender_id here if msg.get('type') == 'message' and msg.get('message', {}).get('role') == 'user': total += 1 except Exception: # Ignore parse errors continue except Exception: # Ignore parse errors continue except Exception: # Ignore parse errors pass return total def get_user_stats_from_sessions_json(agents_dir, target_date_str): """Get user statistics from ALL jsonl files directly""" user_tokens = {} user_messages = {} if not os.path.exists(agents_dir): return user_tokens, user_messages for agent_name in os.listdir(agents_dir): agent_dir_path = os.path.join(agents_dir, agent_name) if not os.path.isdir(agent_dir_path): continue sessions_dir = os.path.join(agent_dir_path, "sessions") if not os.path.exists(sessions_dir): continue for f in os.listdir(sessions_dir): # Include both .jsonl and .jsonl.reset.* files (they contain all session messages) if '.jsonl' not in f: continue filepath = os.path.join(sessions_dir, f) current_user = None try: with open(filepath) as file: for line in file: if target_date_str not in line: continue try: msg = json.loads(line) # Check for user message if msg.get('type') == 'message' and msg.get('message', {}).get('role') == 'user': content = msg.get('message', {}).get('content', []) if content and isinstance(content, list): text = content[0].get('text', '') # Pattern 1: "sender_id": "ou_..." in conversation info sender_match = re.search(r'"sender_id":\s*"ou_([a-z0-9]{32})"', text) if sender_match: current_user = 'ou_' + sender_match.group(1) user_messages[current_user] = user_messages.get(current_user, 0) + 1 else: # Pattern 2: [message_id: xxx]\nou_xxx: # User ID appears after message_id and before the colon msg_id_match = re.search(r'\[message_id: [^\]]+\]\n(ou_[a-z0-9]{32}):', text) if msg_id_match: current_user = msg_id_match.group(1) user_messages[current_user] = user_messages.get(current_user, 0) + 1 else: # Fallback: try to find any valid 34-char open_id match = re.search(r'ou_[a-z0-9]{32}', text) if match: current_user = match.group(0) user_messages[current_user] = user_messages.get(current_user, 0) + 1 # Get token usage from assistant messages if msg.get('type') == 'message' and msg.get('message', {}).get('role') == 'assistant': usage = msg.get('message', {}).get('usage', {}) tokens = usage.get('totalTokens', 0) if tokens and current_user: user_tokens[current_user] = user_tokens.get(current_user, 0) + tokens except Exception: # Ignore parse errors continue except Exception: # Ignore parse errors pass return user_tokens, user_messages def format_number(num): """Format number to human readable""" if num > 1000000: return f"{num // 1000000}M" elif num > 1000: return f"{num // 1000}k" return str(num) def generate_report_data(target_date, user_mappings=None): """Generate inspection report data and return stats""" if user_mappings is None: user_mappings = {} config_dir = os.path.expanduser("~/openclaw-config") target_date_str = target_date containers = [] if os.path.exists(config_dir): containers = [d for d in os.listdir(config_dir) if os.path.isdir(os.path.join(config_dir, d))] summary_data = [["Container", "Security", "Tokens", "Messages", "#Skills", "Cron"]] details_data = [["Container", "Security Issues", "Skills", "Cron Jobs"]] # Statistics container_tokens = [] container_messages_list = [] total_tokens = 0 total_messages = 0 total_skills = 0 has_security_risk = False security_risk_details = [] # User stats all_user_tokens = {} all_user_messages = {} container_user_map = {} # Track which container each user belongs to for container in sorted(containers): container_path = os.path.join(config_dir, container) config_file = os.path.join(container_path, "openclaw.json") security_issues = [] security_status = "✅ Pass" if os.path.exists(config_file): try: with open(config_file) as f: config = json.load(f) config_str = json.dumps(config) if '"key": "sk-' in config_str: security_issues.append("⚠️ Plaintext API key") security_status = "⚠️ Fail" has_security_risk = True security_risk_details.append(f"{container}: Plaintext API key") if "mode" not in config_str: security_issues.append("⚠️ No gateway auth") security_status = "⚠️ Fail" if container not in [x[0] for x in security_risk_details]: has_security_risk = True security_risk_details.append(f"{container}: No gateway auth") except Exception as e: print(f"Error reading {container}: {e}") if not security_issues: security_issues = ["None"] container_total_tokens = 0 container_messages = 0 skill_count = 0 agents_dir = os.path.join(container_path, "agents") if os.path.exists(agents_dir): try: for agent_name in os.listdir(agents_dir): agent_dir_path = os.path.join(agents_dir, agent_name) if os.path.isdir(agent_dir_path): container_total_tokens += calculate_tokens(agent_dir_path, target_date_str) container_messages += calculate_messages(agent_dir_path, target_date_str) except PermissionError: pass # Calculate skill_count first skills_dir = os.path.join(container_path, "workspace", "skills") skill_count = 0 skill_list = [] if os.path.exists(skills_dir): try: skill_list = sorted([d for d in os.listdir(skills_dir) if os.path.isdir(os.path.join(skills_dir, d))]) skill_count = len(skill_list) except PermissionError: pass # Then add to totals total_tokens += container_total_tokens total_messages += container_messages total_skills += skill_count # Track for top 5 if container_total_tokens > 0: container_tokens.append((container, container_total_tokens)) if container_messages > 0: container_messages_list.append((container, container_messages)) # Collect user stats user_tokens, user_messages = get_user_stats_from_sessions_json(agents_dir, target_date_str) for uid, tokens in user_tokens.items(): all_user_tokens[uid] = all_user_tokens.get(uid, 0) + tokens for uid, msgs in user_messages.items(): all_user_messages[uid] = all_user_messages.get(uid, 0) + msgs # Track which container each user belongs to if container not in container_user_map: container_user_map[container] = set() for uid in user_messages.keys(): container_user_map[container].add(uid) tokens_str = format_number(container_total_tokens) messages_str = format_number(container_messages) cron_count = 0 cron_list = "-" if container == "admin": cron_dir = os.path.join(container_path, "cron") if os.path.exists(cron_dir): try: cron_count = len([f for f in os.listdir(cron_dir) if f.endswith(('.sh', '.py'))]) jobs_file = os.path.join(cron_dir, "jobs.json") if os.path.exists(jobs_file): with open(jobs_file) as f: jobs_data = json.load(f) if isinstance(jobs_data, list): cron_count += len([j for j in jobs_data if j.get("enabled", True)]) cron_list = ", ".join([f for f in os.listdir(cron_dir) if f.endswith(('.sh', '.py'))]) except PermissionError: pass # Parse tokens for sorting (handle M/k suffixes) tokens_num = container_total_tokens skills_str = ", ".join(skill_list) if skill_list else "-" details_data.append([container, ", ".join(security_issues), skills_str, cron_list]) summary_data.append([container, security_status, tokens_str, messages_str, str(skill_count), str(cron_count), tokens_num]) # Sort by tokens (descending), remove the helper column if len(summary_data) > 1: summary_data[1:] = sorted(summary_data[1:], key=lambda x: x[6] if len(x) > 6 else 0, reverse=True) summary_data = [row[:6] for row in summary_data] # Sort by tokens and get top 5 container_tokens.sort(key=lambda x: x[1], reverse=True) top5_tokens = container_tokens[:5] container_messages_list.sort(key=lambda x: x[1], reverse=True) top5_messages = container_messages_list[:5] # Get top users sorted_users = sorted(all_user_messages.items(), key=lambda x: x[1], reverse=True) top5_users = sorted_users[:5] # Build user_by_container mapping user_by_container = {} for cont, users in container_user_map.items(): user_by_container[cont] = {"users": list(users)} stats = { "total_tokens": total_tokens, "total_tokens_str": format_number(total_tokens), "total_messages": total_messages, "total_messages_str": format_number(total_messages), "total_skills": total_skills, "has_security_risk": has_security_risk, "security_risk_details": security_risk_details, "top5_tokens": top5_tokens, "top5_messages": top5_messages, "top5_users": top5_users, "all_user_tokens": all_user_tokens, "all_user_messages": all_user_messages, "user_by_container": user_by_container, } return summary_data, details_data, stats def generate_markdown(target_date, user_mappings=None): """Generate markdown report and return stats""" if user_mappings is None: user_mappings = {} output_dir = os.path.expanduser("~/openclaw-team/skill-scripts") os.makedirs(output_dir, exist_ok=True) report_file = os.path.join(output_dir, f"inspection-{target_date}.md") summary_data, details_data, stats = generate_report_data(target_date, user_mappings) def align_table(data): """Align table columns: calculate max width and pad, no separator lines""" if not data: return [] col_widths = [max(len(str(row[i])) for row in data) for i in range(len(data[0]))] result = [] for row in data: aligned = [] for i, cell in enumerate(row): aligned.append(str(cell).ljust(col_widths[i])) result.append("| " + " | ".join(aligned) + " |") return result # Summary section lines = [ f"# AI Agents Daily Inspection Report - {target_date.replace('-', '')}", "", "## Summary", "", f"- **Total Tokens**: {stats['total_tokens_str']}", f"- **Total Messages**: {stats['total_messages_str']}", f"- **Total Skills**: {stats.get('total_skills', 45)}", "", ] # Top 5 Token (use container data) lines.extend(["", "### Top 5 Token Usage", ""]) top5_token_data = [["Rank", "Container", "Tokens"]] for i, (name, tokens) in enumerate(stats.get("top5_tokens", []), 1): top5_token_data.append([str(i), name, format_number(tokens)]) lines.extend(align_table(top5_token_data)) # Top 5 发消息用户 (by user ID) # Build user_containers mapping user_containers = {} user_by_container = stats.get("user_by_container", {}) for cont, data in user_by_container.items(): for uid in data.get("users", []): user_containers[uid] = cont lines.extend(["", "### Top 5 User Messages", ""]) sorted_users = sorted(stats.get("all_user_messages", {}).items(), key=lambda x: x[1], reverse=True) top5_users_data = [["Rank", "User", "Messages", "Container"]] for i, (uid, msgs) in enumerate(sorted_users[:5], 1): # Try to get user name from mappings, fallback to ID user_name = user_mappings.get(uid, uid) container = user_containers.get(uid, "-") top5_users_data.append([str(i), user_name, str(msgs), container]) lines.extend(align_table(top5_users_data)) # Basic Info lines.extend(["", "## Basic Info", ""]) lines.extend(align_table(summary_data)) # Skills Info lines.extend(["", "## Skills Info", ""]) # Format as simple list: Container: skills for row in details_data[1:]: # Skip header container = row[0] skills = row[2] if len(row) > 2 else "-" lines.append(f"- {container}: {skills}") # User Stats if stats.get("all_user_messages"): # TODO: Add user name mapping from Feishu API or manual mapping # user_ids = list(stats["all_user_messages"].keys()) # user_names = get_user_names(user_ids) # Find which container each user belongs to user_containers = {} user_by_container = stats.get("user_by_container", {}) for cont, data in user_by_container.items(): for uid in data.get("users", []): user_containers[uid] = cont lines.extend(["", "## User Stats", ""]) user_info = [["User", "Tokens", "Messages", "Container"]] # Build user data with raw token count for sorting user_data = [] for uid in stats["all_user_messages"].keys(): # Try to get user name from mappings, fallback to ID user_name = user_mappings.get(uid, uid) tokens = stats["all_user_tokens"].get(uid, 0) msgs = stats["all_user_messages"][uid] container = user_containers.get(uid, "-") user_data.append([user_name, tokens, msgs, container]) # Sort by tokens (descending) - use raw token count user_data.sort(key=lambda x: x[1], reverse=True) # Format tokens for display for row in user_data: row[1] = format_number(row[1]) row[2] = str(row[2]) user_info.extend(user_data) lines.extend(align_table(user_info)) lines.extend(["", "---", f"*Generated by OpenClaw - {target_date}*"]) content = "\n".join(lines) with open(report_file, "w") as f: f.write(content) return report_file, content, stats # ============================================================ def write_to_document(doc_token, content): """Write content to Feishu doc using md2feishu script""" import tempfile # Write content to temp markdown file with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f: f.write(content) temp_file = f.name try: # Call md2feishu script cmd = [ "python3", MD2FEISHU_SCRIPT, "--app-id", APP_ID, "--app-secret", APP_SECRET, "--file", temp_file, "--title", "Daily Inspection Report", "--doc-id", doc_token ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=600) if result.returncode != 0: raise Exception(f"上传失败: {result.stderr}") print("Content uploaded with formatting!") return True finally: # Clean up temp file if os.path.exists(temp_file): os.remove(temp_file) def main(): if len(sys.argv) > 1: target_date = sys.argv[1] else: target_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") report_date = target_date.replace("-", "") print(f"=== Daily Inspection Report - {report_date} ===") # Load existing mappings first existing_mappings = load_existing_mappings() # Collect unmapped open_ids first print("\nCollecting unmapped open_ids...") # Need to get stats to know which users sent messages _, _, stats = generate_report_data(target_date, {}) unmapped = {} user_by_container = stats.get("user_by_container", {}) for container, data in user_by_container.items(): users = data.get("users", []) unmapped_list = [uid for uid in users if uid not in existing_mappings] if unmapped_list: unmapped[container] = unmapped_list save_unmapped_open_ids(unmapped) if unmapped: total_unmapped = sum(len(v) for v in unmapped.values()) print(f"Found {total_unmapped} unmapped open_ids across {len(unmapped)} containers") # Call gen_open_id_mapping.py to query user names print("\nQuerying user names...") import subprocess script_dir = os.path.dirname(os.path.abspath(__file__)) result = subprocess.run( ["python3", os.path.join(script_dir, "gen_open_id_mapping.py")], capture_output=True, text=True ) print(result.stdout) # Read mapped data and merge mapped_file = os.path.join(script_dir, "mapped_open_id_list.json") if os.path.exists(mapped_file): with open(mapped_file) as f: new_mappings = json.load(f) print(f"Loaded {sum(len(v) for v in new_mappings.values())} new mappings") for container, mappings in new_mappings.items(): for open_id, user_name in mappings.items(): existing_mappings[open_id] = user_name # Generate markdown once with all mappings report_file, report_content, stats = generate_markdown(target_date, existing_mappings) print(f"Report: {report_file}") # Create wiki and upload directly enable_upload = os.environ.get("ENABLE_FEISHU_UPLOAD", "true").lower() == "true" wiki_url = "" if enable_upload and APP_ID and APP_SECRET: print("Creating wiki document...") try: doc_token, wiki_token = create_wiki_node(f"AI Agents Daily Inspection - {report_date}") wiki_url = f"{WIKI_BASE_URL}/{wiki_token}" print(f"Wiki: {wiki_url}") # Upload directly using API print("Uploading content...") write_to_document(doc_token, report_content) print("Content uploaded!") except Exception as e: print(f"Error: {e}") # Send notification with stats print("Sending notification...") # Format top 5 top5_lines = [] for i, (name, tokens) in enumerate(stats["top5_tokens"], 1): top5_lines.append(f" {i}. {name}: {format_number(tokens)}") top5_str = "\n".join(top5_lines) if top5_lines else " (无)" # Security status security_status = "⚠️ 发现风险" if stats["has_security_risk"] else "✅ 无风险" if stats["has_security_risk"]: security_status += "\n" + "\n".join([f" - {x}" for x in stats["security_risk_details"]]) msg = f"""📊 Daily Inspection Report - {report_date} 🔢 昨日 Token 使用: {stats['total_tokens_str']} 📧 昨日消息数: {stats['total_messages_str']} 🏆 Token 使用 Top 5: {top5_str} 🔒 安全状态: {security_status}""" if wiki_url: msg += f"\n\n🌐 Wiki: {wiki_url}" send_notification(msg) # Cleanup temporary files (keep mapped_open_id_list.json for caching) script_dir = os.path.dirname(os.path.abspath(__file__)) # Only delete unmapped file, keep mapped file for next run fpath = os.path.join(script_dir, "unmapped_open_id_list.json") if os.path.exists(fpath): os.remove(fpath) print("Cleaned up: unmapped_open_id_list.json") print("\nDone!") if __name__ == "__main__": main()