#!/usr/bin/env python3 """Sync NG/serverless O11Y alerts into Feishu bitable with dedupe and update support.""" import argparse import base64 import datetime as dt import json import os import re import sys import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, Iterable, List, Optional DEFAULT_WIKI_TOKEN = "SK9dwHO8DitMLqkNXsrcZr9vnCd" DEFAULT_TABLE_ID = "tblKvZKBtuZTrI4e" DEFAULT_VIEW_ID = "vewfa3loWx" DEFAULT_JIRA_BASE_URL = "https://tidb.atlassian.net" O11Y_COMPONENTS = [ "ecosystem-server", "ticdc", "ticdc-kernel", "next-gen-br", "next-gen-br-kernel", "next-gen-import", "lightning", ] TCOC_COMPONENTS = ["BR", "Cloud DM", "Cloud Import", "TiCDC", "TiDB Lightning"] MANAGED_FIELDS = { "Title", "Link", "CreatedAt", "Alert Name", "Cluster Type", "Category", "Scope", "Sub-Category", "Status", "Oncall Confirmed", "Assignee", "Oncall Cycle", } def http_json( method: str, url: str, headers: Optional[Dict[str, str]] = None, payload: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: data = None req_headers = dict(headers or {}) if payload is not None: data = json.dumps(payload).encode("utf-8") req_headers.setdefault("Content-Type", "application/json") req = urllib.request.Request(url, data=data, headers=req_headers, method=method) with urllib.request.urlopen(req) as resp: return json.loads(resp.read().decode("utf-8")) class FeishuClient: def __init__(self, app_id: str, app_secret: str) -> None: self.base = "https://open.feishu.cn/open-apis" self.app_id = app_id self.app_secret = app_secret self._token: Optional[str] = None def headers(self) -> Dict[str, str]: if not self._token: resp = http_json( "POST", f"{self.base}/auth/v3/tenant_access_token/internal", payload={"app_id": self.app_id, "app_secret": self.app_secret}, ) if resp.get("code") != 0: raise RuntimeError(f"feishu auth failed: {resp}") self._token = resp["tenant_access_token"] return {"Authorization": f"Bearer {self._token}"} def resolve_wiki(self, wiki_token: str) -> Dict[str, Any]: url = f"{self.base}/wiki/v2/spaces/get_node?token={urllib.parse.quote(wiki_token)}" resp = http_json("GET", url, headers=self.headers()) if resp.get("code") != 0: raise RuntimeError(f"wiki resolve failed: {resp}") return resp["data"]["node"] def list_fields(self, app_token: str, table_id: str) -> List[Dict[str, Any]]: url = f"{self.base}/bitable/v1/apps/{app_token}/tables/{table_id}/fields" resp = http_json("GET", url, headers=self.headers()) if resp.get("code") != 0: raise RuntimeError(f"list fields failed: {resp}") return resp["data"]["items"] def list_records(self, app_token: str, table_id: str, view_id: str) -> List[Dict[str, Any]]: records: List[Dict[str, Any]] = [] page_token = "" while True: params = {"view_id": view_id, "page_size": 500} if page_token: params["page_token"] = page_token url = ( f"{self.base}/bitable/v1/apps/{app_token}/tables/{table_id}/records?" + urllib.parse.urlencode(params) ) resp = http_json("GET", url, headers=self.headers()) if resp.get("code") != 0: raise RuntimeError(f"list records failed: {resp}") data = resp["data"] records.extend(data.get("items") or []) if not data.get("has_more"): break page_token = data["page_token"] return records def create_record(self, app_token: str, table_id: str, fields: Dict[str, Any]) -> None: url = f"{self.base}/bitable/v1/apps/{app_token}/tables/{table_id}/records" resp = http_json("POST", url, headers=self.headers(), payload={"fields": fields}) if resp.get("code") != 0: raise RuntimeError(f"create record failed: {resp}") def update_record(self, app_token: str, table_id: str, record_id: str, fields: Dict[str, Any]) -> None: url = f"{self.base}/bitable/v1/apps/{app_token}/tables/{table_id}/records/{record_id}" resp = http_json("PUT", url, headers=self.headers(), payload={"fields": fields}) if resp.get("code") != 0: raise RuntimeError(f"update record failed: {resp}") class JiraClient: def __init__(self, base_url: str, email: str, token: str) -> None: auth = base64.b64encode(f"{email}:{token}".encode("utf-8")).decode("utf-8") self.base_url = base_url.rstrip("/") self.headers = {"Authorization": f"Basic {auth}", "Accept": "application/json"} def search(self, jql: str, limit: int) -> List[Dict[str, Any]]: query = urllib.parse.urlencode( { "jql": jql, "maxResults": str(limit), "fields": "summary,created,issuetype,status,project,assignee,components", } ) url = f"{self.base_url}/rest/api/3/search/jql?{query}" return http_json("GET", url, headers=self.headers).get("issues", []) def field_options_by_name(fields: Iterable[Dict[str, Any]]) -> Dict[str, Dict[str, str]]: result: Dict[str, Dict[str, str]] = {} for field in fields: property_meta = field.get("property") or {} options = {} for option in property_meta.get("options", []): options[option["name"]] = option["name"] result[field["field_name"]] = options return result def extract_cycle_dates(option_ids: Dict[str, Dict[str, str]]) -> List[tuple[dt.date, str]]: cycles: List[tuple[dt.date, str]] = [] pattern = re.compile(r"^(\d{4})(\d{2})(\d{2})$") for option_name in option_ids.get("Oncall Cycle", {}): match = pattern.match(option_name) if not match: continue year, month, day = match.groups() cycles.append((dt.date(int(year), int(month), int(day)), option_name)) cycles.sort() return cycles def normalize_date(value: str) -> dt.date: return dt.datetime.strptime(value, "%Y-%m-%d").date() def date_range_for_mode(mode: str, start: Optional[str], end: Optional[str]) -> tuple[dt.date, dt.date]: today = dt.date.today() if mode == "daily": yesterday = today - dt.timedelta(days=1) return yesterday, yesterday if mode == "backfill": if not start: raise ValueError("backfill mode requires start date") return normalize_date(start), today if mode == "range": if not start or not end: raise ValueError("range mode requires start and end date") return normalize_date(start), normalize_date(end) raise ValueError(f"unsupported mode: {mode}") def build_o11y_jql(cluster_type: str, start: dt.date, end: dt.date) -> str: components = ", ".join(f'"{value}"' for value in O11Y_COMPONENTS) date_clause = f'created >= "{start.isoformat()}" and created <= "{end.isoformat()}"' if cluster_type == "next-gen": description_clause = 'description ~ "next-gen"' elif cluster_type == "serverless": description_clause = '(description ~ "serverless" OR description ~ "free-tier")' else: raise ValueError(f"unsupported cluster type: {cluster_type}") return f"""project = O11Y and {date_clause} and component in ({components}) and priority = Critical and {description_clause} ORDER BY priority DESC, created DESC""" def build_tcoc_jql(start: dt.date, end: dt.date) -> str: components = ", ".join(f'"{value}"' for value in TCOC_COMPONENTS) return f"""project = TiDBCloud-OnCall and created >= "{start.isoformat()}" and created <= "{end.isoformat()}" and "Tier[Dropdown]" in ("TiDB X") and component in ({components}) ORDER BY priority DESC, created DESC""" def detect_scope(summary: str, components: List[str]) -> str: component_text = " ".join(components).lower() lowered = summary.lower() if "tidb lightning" in component_text: return "lightning" if "cloud import" in component_text: return "Import" if "ticdc" in component_text: return "CDC" if "cloud dm" in component_text: return "DM" if "br" in component_text: return "BR" if "lightning" in lowered: return "lightning" if "import" in lowered: return "Import" if "changefeed" in lowered or "cdc" in lowered: return "CDC" if "migration" in lowered or "dm" in lowered: return "DM" if "backup" in lowered or "restore" in lowered or "export" in lowered or "br" in lowered: return "BR" return "Dataservice" def to_feishu_ts(created: str) -> int: parsed = dt.datetime.strptime(created, "%Y-%m-%dT%H:%M:%S.%f%z") return int(parsed.timestamp() * 1000) def to_jira_dt(created: str) -> dt.datetime: return dt.datetime.strptime(created, "%Y-%m-%dT%H:%M:%S.%f%z") def detect_oncall_cycle(created: str, cycle_options: List[tuple[dt.date, str]]) -> Optional[str]: created_date = to_jira_dt(created).date() selected: Optional[str] = None for cycle_date, option_name in cycle_options: if cycle_date <= created_date: selected = option_name else: break return selected def build_fields( issue: Dict[str, Any], cluster_type: str, category: str, option_ids: Dict[str, Dict[str, str]], cycle_options: List[tuple[dt.date, str]], ) -> Dict[str, Any]: fields = issue["fields"] summary = fields["summary"] issue_type = fields.get("issuetype", {}).get("name", "") status = fields.get("status", {}).get("name", "") components = [item["name"] for item in fields.get("components", [])] scope = detect_scope(summary, components) alert_name = summary.removeprefix("[PROD]").strip() result: Dict[str, Any] = { "Title": f"{issue['key']} {summary}", "Link": { "text": issue["key"], "link": f"https://tidb.atlassian.net/browse/{issue['key']}", }, "CreatedAt": to_feishu_ts(fields["created"]), } if category == "ALERT": result["Alert Name"] = alert_name if cluster_type in option_ids.get("Cluster Type", {}): result["Cluster Type"] = option_ids["Cluster Type"][cluster_type] if category in option_ids.get("Category", {}): result["Category"] = option_ids["Category"][category] if scope in option_ids.get("Scope", {}): result["Scope"] = option_ids["Scope"][scope] if issue_type in option_ids.get("Sub-Category", {}): result["Sub-Category"] = option_ids["Sub-Category"][issue_type] if status in option_ids.get("Status", {}): result["Status"] = option_ids["Status"][status] if "TODO" in option_ids.get("Oncall Confirmed", {}): result["Oncall Confirmed"] = option_ids["Oncall Confirmed"]["TODO"] assignee = fields.get("assignee") or {} assignee_name = assignee.get("displayName") if assignee_name in option_ids.get("Assignee", {}): result["Assignee"] = option_ids["Assignee"][assignee_name] cycle_name = detect_oncall_cycle(fields["created"], cycle_options) if cycle_name and cycle_name in option_ids.get("Oncall Cycle", {}): result["Oncall Cycle"] = option_ids["Oncall Cycle"][cycle_name] return result def sync(args: argparse.Namespace) -> int: feishu = FeishuClient(args.feishu_app_id, args.feishu_app_secret) jira = JiraClient(args.jira_base_url, args.jira_email, args.jira_token) start_date, end_date = date_range_for_mode(args.mode, args.start_date, args.end_date) node = feishu.resolve_wiki(args.wiki_token) app_token = node["obj_token"] option_ids = field_options_by_name(feishu.list_fields(app_token, args.table_id)) cycle_options = extract_cycle_dates(option_ids) existing_records = {} for record in feishu.list_records(app_token, args.table_id, args.view_id): link = record.get("fields", {}).get("Link") if isinstance(link, str): existing_records[link] = record elif isinstance(link, dict) and isinstance(link.get("link"), str): existing_records[link["link"]] = record creates: List[Dict[str, Any]] = [] updates: List[Dict[str, Any]] = [] query_specs = [ { "name": "next-gen-alert", "jql": build_o11y_jql("next-gen", start_date, end_date), "cluster_type": "next-gen", "category": "ALERT", }, { "name": "serverless-alert", "jql": build_o11y_jql("serverless", start_date, end_date), "cluster_type": "serverless", "category": "ALERT", }, { "name": "tcoc-tidb-x", "jql": build_tcoc_jql(start_date, end_date), "cluster_type": "next-gen", "category": "TCOC", }, ] for spec in query_specs: issues = jira.search(spec["jql"], args.jira_limit) for issue in issues: new_fields = build_fields( issue, spec["cluster_type"], spec["category"], option_ids, cycle_options, ) record_link = new_fields["Link"]["link"] existing = existing_records.get(record_link) if not existing: creates.append({"source": spec["name"], "fields": new_fields}) continue delta = { key: value for key, value in new_fields.items() if existing.get("fields", {}).get(key) != value } for key in MANAGED_FIELDS: if key not in new_fields and existing.get("fields", {}).get(key) is not None: delta[key] = None if delta: updates.append( { "source": spec["name"], "record_id": existing["record_id"], "fields": delta, "link": new_fields["Link"], } ) preview = {"creates": creates[: args.preview_limit], "updates": updates[: args.preview_limit]} print(json.dumps(preview, indent=2, ensure_ascii=False)) print(f"window={start_date.isoformat()}..{end_date.isoformat()}") print(f"prepared_creates={len(creates)}") print(f"prepared_updates={len(updates)}") if not args.apply: print("dry-run only, nothing written") return 0 for item in creates: feishu.create_record(app_token, args.table_id, item["fields"]) for item in updates: feishu.update_record(app_token, args.table_id, item["record_id"], item["fields"]) print(f"created_records={len(creates)}") print(f"updated_records={len(updates)}") return 0 def parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="Sync NG/serverless O11Y alerts into Feishu bitable.") p.add_argument("mode", choices=["daily", "backfill", "range"]) p.add_argument("start_date", nargs="?") p.add_argument("end_date", nargs="?") p.add_argument("--apply", action="store_true") p.add_argument("--dry-run", action="store_true") p.add_argument("--preview-limit", type=int, default=5) p.add_argument("--jira-limit", type=int, default=500) p.add_argument("--feishu-app-id", default=os.environ.get("FEISHU_APP_ID")) p.add_argument("--feishu-app-secret", default=os.environ.get("FEISHU_APP_SECRET")) p.add_argument("--jira-email", default=os.environ.get("JIRA_EMAIL")) p.add_argument("--jira-token", default=os.environ.get("JIRA_TOKEN")) p.add_argument("--jira-base-url", default=os.environ.get("JIRA_BASE_URL", DEFAULT_JIRA_BASE_URL)) p.add_argument("--wiki-token", default=os.environ.get("FEISHU_WIKI_TOKEN", DEFAULT_WIKI_TOKEN)) p.add_argument("--table-id", default=os.environ.get("FEISHU_TABLE_ID", DEFAULT_TABLE_ID)) p.add_argument("--view-id", default=os.environ.get("FEISHU_VIEW_ID", DEFAULT_VIEW_ID)) return p def main() -> int: args = parser().parse_args() if not args.apply: args.dry_run = True required = { "feishu_app_id": args.feishu_app_id, "feishu_app_secret": args.feishu_app_secret, "jira_email": args.jira_email, "jira_token": args.jira_token, } missing = [key for key, value in required.items() if not value] if missing: print(f"missing required args/env: {', '.join(missing)}", file=sys.stderr) return 1 try: return sync(args) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") print(f"http error {exc.code}: {body}", file=sys.stderr) return 1 except Exception as exc: print(str(exc), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())