#!/usr/bin/env python3 """ daily-oncall-metrics Unified entry point for daily oncall metrics synchronization """ import os import datetime from datetime import timedelta from base import send_feishu_message from failed_job_sync import sync as sync_failed_jobs from jira_ticket_sync import sync as sync_jira # Skip notification flag for debugging SKIP_SEND_NOTIFICATION = os.environ.get("SKIP_SEND_NOTIFICATION", "false").lower() == "true" def build_summary(display_date, failed_jobs, jira): """Build the Feishu summary message.""" backup_br_failed = failed_jobs.get("backup", 0) + failed_jobs.get("br", 0) return f"""📊 Daily Oncall Metrics - {display_date} Failed Jobs: Import {failed_jobs["import"]}, DM {failed_jobs["dm"]}, Backup/BR {backup_br_failed} TCOC: {jira["tcoc"]} Critical Alert: {jira["alert"]} 📄 Docs: • TCOC & Alert: https://pingcap.feishu.cn/wiki/M6UVwB2WKiZhzJkA9NZcxlpknWc?table=tblnZt7BJBTzYMvJ • Failed Jobs: https://pingcap.feishu.cn/wiki/I665wUjMfiiurKkOvPccpAWQnub?table=tblGuNt6JW9eVGvx""" def main(date_arg=None): print("=" * 50) print("daily-oncall-metrics") print("=" * 50) # Parse date argument: if provided, use it; otherwise use yesterday if date_arg: # User provided date like "20260314" date_str = date_arg year = int(date_str[:4]) month = int(date_str[4:6]) day = int(date_str[6:8]) target_date = datetime.date(year, month, day) display_date = target_date.strftime("%Y-%m-%d") else: # Default: yesterday yesterday = datetime.date.today() - timedelta(days=1) date_str = yesterday.strftime("%Y%m%d") display_date = yesterday.strftime("%Y-%m-%d") print(f"Date: {date_str}") print(f"Start time: {datetime.datetime.now()}") results = {} # Run Failed Job Sync print("\n[1/3] Running Failed Job Sync...") results['failed_jobs'] = sync_failed_jobs(date_str) # Run Jira Ticket Sync print("\n[2/3] Running Jira Ticket Sync...") results['jira'] = sync_jira(date_str) print("\n[3/3] Running Alert Sync... (Not implemented)") # Send summary notification summary = build_summary(display_date, results["failed_jobs"], results["jira"]) print("\n" + "=" * 50) print(summary) print("=" * 50) # Send notification unless SKIP_SEND_NOTIFICATION is true if SKIP_SEND_NOTIFICATION: print("\n[SKIPPED] Notification disabled (SKIP_SEND_NOTIFICATION=true)") else: # Send to user directly user_chat_id = "oc_2c586abb5204eca2ddf6ddf5de338bf6" send_feishu_message(user_chat_id, summary) print(f"\nEnd time: {datetime.datetime.now()}") print("Done!") if __name__ == "__main__": import sys date_arg = sys.argv[1] if len(sys.argv) > 1 else None main(date_arg)