#!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from typing import Any from common import DeployError, db_exec, load_env_file, log from file_storage_readiness import ( PROBLEMATIC_STATES, collect_file_storage_checks, sql_string_literal, summarize_results, ) DEFAULT_LIMIT = 1000 DEFAULT_ACTOR = "deploy:file-storage-repair" DEFAULT_PARENT_OBJECT_API_NAME = "Order" DEFAULT_ATTACHMENT_KIND = "archived_order" DEFAULT_STATUS = "Active" DEFAULT_STORAGE_PROVIDER = "filesystem" def is_test_slot_env_name(env_name: str) -> bool: normalized = env_name.strip().lower() return normalized in {"test-test1", "test-test2", "test-test3"} def require_test_slot_env(envv: dict[str, str], *, allow_non_test_env: bool) -> None: if allow_non_test_env: return env_name = str(envv.get("ENV_NAME") or "") if not is_test_slot_env_name(env_name): raise DeployError( "file storage repair is test-slot only by default; " f"got ENV_NAME={env_name!r}" ) def require_repair_scope( *, parent_object_api_name: str, attachment_kind: str, status: str, storage_provider: str, ) -> None: missing = [ name for name, value in [ ("parent_object_api_name", parent_object_api_name), ("attachment_kind", attachment_kind), ("status", status), ("storage_provider", storage_provider), ] if not value ] if missing: raise DeployError(f"refusing broad file storage repair; missing filters: {', '.join(missing)}") def build_soft_delete_query( *, file_ids: list[str], parent_object_api_name: str, attachment_kind: str, status: str, storage_provider: str, actor: str, ) -> str: if not file_ids: return "" id_list = ", ".join(sql_string_literal(file_id) for file_id in file_ids) return f""" UPDATE `_System_File` SET `Status` = 'Deleted', `DeletedDate` = UTC_TIMESTAMP(), `DeletedById` = {sql_string_literal(actor)}, `LastModifiedDate` = UTC_TIMESTAMP() WHERE `Id` IN ({id_list}) AND `ParentObjectApiName` = {sql_string_literal(parent_object_api_name)} AND `AttachmentKind` = {sql_string_literal(attachment_kind)} AND `Status` = {sql_string_literal(status)} AND `StorageProvider` = {sql_string_literal(storage_provider)}; """.strip() def repairable_checks(checks: list[dict[str, Any]]) -> list[dict[str, Any]]: return [ check for check in checks if str(check.get("state") or "") in PROBLEMATIC_STATES and str(check.get("id") or "") ] def summarize_checks( checks: list[dict[str, Any]], *, envv: dict[str, str], limit: int, parent_object_api_name: str, attachment_kind: str, status: str, storage_provider: str, ) -> dict[str, Any]: return summarize_results( checks, envv=envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, status=status, storage_provider=storage_provider, ) def repair_file_storage( envv: dict[str, str], *, apply: bool, limit: int = DEFAULT_LIMIT, parent_object_api_name: str = DEFAULT_PARENT_OBJECT_API_NAME, attachment_kind: str = DEFAULT_ATTACHMENT_KIND, status: str = DEFAULT_STATUS, storage_provider: str = DEFAULT_STORAGE_PROVIDER, actor: str = DEFAULT_ACTOR, allow_non_test_env: bool = False, ) -> dict[str, Any]: require_test_slot_env(envv, allow_non_test_env=allow_non_test_env) require_repair_scope( parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, status=status, storage_provider=storage_provider, ) checks_before = collect_file_storage_checks( envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, status=status, storage_provider=storage_provider, ) before_summary = summarize_checks( checks_before, envv=envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, status=status, storage_provider=storage_provider, ) targets = repairable_checks(checks_before) target_ids = [str(check["id"]) for check in targets] repair = { "mode": "apply" if apply else "dry-run", "actor": actor, "eligible_count": len(targets), "eligible_ids": target_ids, "eligible_samples": targets[:10], "applied_count": 0, } after_summary: dict[str, Any] | None = None if apply and target_ids: sql = build_soft_delete_query( file_ids=target_ids, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, status=status, storage_provider=storage_provider, actor=actor, ) db_exec(envv, sql, database=envv["TIDB_DATABASE"]) repair["applied_count"] = len(target_ids) if apply: checks_after = collect_file_storage_checks( envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, status=status, storage_provider=storage_provider, ) after_summary = summarize_checks( checks_after, envv=envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, status=status, storage_provider=storage_provider, ) final_problem_count = ( int(after_summary.get("problem_count", 0)) if after_summary is not None else int(before_summary.get("problem_count", 0)) ) return { "status": "passed" if final_problem_count == 0 else ("failed" if apply else "would-repair"), "env_name": envv.get("ENV_NAME", ""), "public_base_url": envv.get("PUBLIC_BASE_URL", ""), "file_storage_root": envv.get("FILE_STORAGE_ROOT", ""), "filters": { "status": status, "parent_object_api_name": parent_object_api_name, "attachment_kind": attachment_kind, "storage_provider": storage_provider, "limit": limit, }, "before": before_summary, "repair": repair, "after": after_summary, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Soft-delete test-slot _System_File rows whose filesystem objects are missing." ) parser.add_argument("--env-file", required=True) parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT) parser.add_argument("--parent-object-api-name", default=DEFAULT_PARENT_OBJECT_API_NAME) parser.add_argument("--attachment-kind", default=DEFAULT_ATTACHMENT_KIND) parser.add_argument("--status", default=DEFAULT_STATUS) parser.add_argument("--storage-provider", default=DEFAULT_STORAGE_PROVIDER) parser.add_argument("--actor", default=DEFAULT_ACTOR) parser.add_argument("--apply", action="store_true") parser.add_argument("--allow-non-test-env", action="store_true") parser.add_argument("--summary-json") parser.add_argument("--fail-on-problems", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() envv = load_env_file(Path(args.env_file).expanduser().resolve()) summary = repair_file_storage( envv, apply=args.apply, limit=args.limit, parent_object_api_name=args.parent_object_api_name, attachment_kind=args.attachment_kind, status=args.status, storage_provider=args.storage_provider, actor=args.actor, allow_non_test_env=args.allow_non_test_env, ) output = json.dumps(summary, indent=2, ensure_ascii=False) print(output) if args.summary_json: summary_path = Path(args.summary_json).expanduser().resolve() summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text(output + "\n", encoding="utf-8") log(f"wrote file storage repair summary: {summary_path}") if args.fail_on_problems and summary["status"] != "passed": return 1 return 0 if __name__ == "__main__": raise SystemExit(main())