import argparse import json import sys from lark_common import ( LarkApiError, api_request, configure_output_encoding, get_tenant_access_token, load_dotenv, optional_env, require_env, ) DEFAULT_SHOW_FIELDS = "问题描述,问题详情,严重程度" def parse_args(): parser = argparse.ArgumentParser( description="Search Lark bitable records and optionally update matched records." ) parser.add_argument( "--dry-run", action="store_true", help="Only print matched records without updating them.", ) parser.add_argument( "--limit", type=int, default=200, help="Safety limit for matched records. Default: 200.", ) parser.add_argument( "--page-size", type=int, default=100, help="Records per search request. Default: 100.", ) parser.add_argument( "--field", default="", help="Override LARK_BITABLE_FILTER_FIELD.", ) parser.add_argument( "--from-value", default="", help="Override LARK_BITABLE_FILTER_VALUE.", ) parser.add_argument( "--to-field", default="", help="Override LARK_BITABLE_UPDATE_FIELD.", ) parser.add_argument( "--to-value", default="", help="Override LARK_BITABLE_UPDATE_VALUE.", ) parser.add_argument( "--show-fields", default="", help=( "Comma-separated extra field names to print. " "Defaults to LARK_BITABLE_SHOW_FIELDS." ), ) return parser.parse_args() def bitable_config(args): return { "app_token": require_env("LARK_BITABLE_APP_TOKEN"), "table_id": require_env("LARK_BITABLE_TABLE_ID"), "view_id": optional_env("LARK_BITABLE_VIEW_ID"), "filter_field": args.field or require_env("LARK_BITABLE_FILTER_FIELD"), "filter_value": args.from_value or require_env("LARK_BITABLE_FILTER_VALUE"), "update_field": args.to_field or require_env("LARK_BITABLE_UPDATE_FIELD"), "update_value": args.to_value or require_env("LARK_BITABLE_UPDATE_VALUE"), } def search_records(config, page_size): token = get_tenant_access_token() page_token = "" records = [] path = f"/bitable/v1/apps/{config['app_token']}/tables/{config['table_id']}/records/search" while True: body = { "filter": { "conjunction": "and", "conditions": [ { "field_name": config["filter_field"], "operator": "is", "value": [config["filter_value"]], } ], }, "page_size": page_size, } if config["view_id"]: body["view_id"] = config["view_id"] if page_token: body["page_token"] = page_token payload = api_request("POST", path, token=token, body=body) data = payload.get("data", {}) records.extend(data.get("items", [])) if not data.get("has_more"): break page_token = data.get("page_token", "") if not page_token: break return records def normalize_field_value(value): if isinstance(value, list): return ", ".join(normalize_field_value(item) for item in value) if isinstance(value, dict): for key in ("text", "name", "en_name", "email", "link", "url"): if key in value: return str(value[key]) return json.dumps(value, ensure_ascii=False) if value is None: return "" return str(value) def selected_fields(args, config): fields = [] for field in (config["filter_field"], config["update_field"]): if field not in fields: fields.append(field) show_fields = args.show_fields or optional_env( "LARK_BITABLE_SHOW_FIELDS", DEFAULT_SHOW_FIELDS ) for field in [item.strip() for item in show_fields.split(",") if item.strip()]: if field not in fields: fields.append(field) return fields def print_records(records, fields, config, dry_run): print(f"Matched records: {len(records)}") print( f"Filter: {config['filter_field']} = {config['filter_value']}; " f"Update: {config['update_field']} -> {config['update_value']}" ) if not records: print("No records to update.") return headers = ["#", "record_id"] + fields rows = [] for index, record in enumerate(records, 1): record_fields = record.get("fields", {}) rows.append( [str(index), record.get("record_id", "")] + [normalize_field_value(record_fields.get(field)) for field in fields] ) widths = [ min(max(len(headers[col]), *(len(row[col]) for row in rows)), 60) for col in range(len(headers)) ] print(" ".join(headers[col].ljust(widths[col]) for col in range(len(headers)))) print(" ".join("-" * widths[col] for col in range(len(headers)))) for row in rows: cells = [] for col, cell in enumerate(row): text = cell if len(text) > widths[col]: text = text[: widths[col] - 3] + "..." cells.append(text.ljust(widths[col])) print(" ".join(cells)) if dry_run: print("Dry run only. Remove --dry-run to update matched records.") else: print("Apply mode: matched records will be updated.") def update_record(config, record_id): token = get_tenant_access_token() path = ( f"/bitable/v1/apps/{config['app_token']}/tables/" f"{config['table_id']}/records/{record_id}" ) body = {"fields": {config["update_field"]: config["update_value"]}} return api_request("PUT", path, token=token, body=body) def update_records(records, config): for index, record in enumerate(records, 1): record_id = record.get("record_id") if not record_id: raise RuntimeError(f"Record #{index} has no record_id") update_record(config, record_id) print(f"Updated {index}/{len(records)}: {record_id}") def main(): configure_output_encoding() load_dotenv() args = parse_args() if args.limit <= 0: raise RuntimeError("--limit must be greater than 0") if args.page_size <= 0 or args.page_size > 500: raise RuntimeError("--page-size must be between 1 and 500") config = bitable_config(args) records = search_records(config, args.page_size) fields = selected_fields(args, config) print_records(records, fields, config, args.dry_run) if len(records) > args.limit: raise RuntimeError( f"Matched {len(records)} records, which exceeds safety limit {args.limit}. " "Raise --limit only after confirming the result." ) if not args.dry_run and records: update_records(records, config) if __name__ == "__main__": try: main() except (RuntimeError, LarkApiError) as exc: print(f"ERROR: {exc}", file=sys.stderr) if isinstance(exc, LarkApiError) and exc.log_id: print(f"X-Tt-Logid: {exc.log_id}", file=sys.stderr) sys.exit(1)