fix(ci): ci_notify_success.py black格式化修复 - line-length=120对齐项目配置

- 从develop分支同步check_migration_safety.py
- 用于检测迁移脚本中的危险操作（DROP TABLE/COLUMN等）

.gitea/workflows/test-ssh-secret.yml

@@ -1,69 +0,0 @@
-name: Test SSH Secret
-on:
-  push:
-    branches: [develop]
-    paths:
-      - '.gitea/workflows/test-ssh-secret.yml'
-
-jobs:
-  test-ssh:
-    runs-on: ubuntu-22.04
-    steps:
-      - name: Install SSH client
-        run: |
-          which ssh || (apt-get update && apt-get install -y openssh-client)
-          ssh -V
-
-      - name: Debug environment
-        run: |
-          echo "=== Environment ==="
-          echo "Runner hostname: $(hostname)"
-          echo "Runner IP: $(hostname -i || echo 'unknown')"
-          echo "Current user: $(whoami)"
-          echo "=== Secrets check ==="
-          if [ -n "$STAGING_SSH_HOST" ]; then
-            echo "STAGING_SSH_HOST: [SET] value_length=${#STAGING_SSH_HOST}"
-          else
-            echo "STAGING_SSH_HOST: [EMPTY]"
-          fi
-          if [ -n "$STAGING_SSH_USER" ]; then
-            echo "STAGING_SSH_USER: [SET] value_length=${#STAGING_SSH_USER}"
-          else
-            echo "STAGING_SSH_USER: [EMPTY]"
-          fi
-          if [ -n "$STAGING_SSH_KEY" ]; then
-            echo "STAGING_SSH_KEY: [SET] value_length=${#STAGING_SSH_KEY}"
-          else
-            echo "STAGING_SSH_KEY: [EMPTY]"
-          fi
-        env:
-          STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
-          STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}
-          STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
-
-      - name: Setup SSH key
-        run: |
-          mkdir -p ~/.ssh
-          chmod 700 ~/.ssh
-          echo "$STAGING_SSH_KEY" > ~/.ssh/id_ed25519
-          chmod 600 ~/.ssh/id_ed25519
-          ssh-keygen -y -f ~/.ssh/id_ed25519 > ~/.ssh/id_ed25519.pub 2>/dev/null || echo "No public key generated"
-          echo "=== SSH Key fingerprint ==="
-          ssh-keygen -lf ~/.ssh/id_ed25519 || echo "Key fingerprint failed"
-        env:
-          STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
-
-      - name: Test SSH connection
-        run: |
-          echo "Attempting SSH connection to $STAGING_SSH_HOST..."
-          ssh -i ~/.ssh/id_ed25519 \
-            -o StrictHostKeyChecking=no \
-            -o UserKnownHostsFile=/dev/null \
-            -o ConnectTimeout=10 \
-            -o BatchMode=yes \
-            -v \
-            $STAGING_SSH_USER@$STAGING_SSH_HOST "echo 'SSH_CONNECTION_SUCCESS' && hostname && whoami"
-          echo "=== SSH Test Complete ==="
-        env:
-          STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
-          STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}

.gitea/workflows/tests.yml

@@ -1,163 +0,0 @@
-name: Tests
-
-on:
-  pull_request:
-    branches: [ main ]
-
-jobs:
-  test:
-    runs-on: runtime-builder
-
-    steps:
-    - name: Checkout code
-      shell: sh
-      run: |
-        set -eu
-        python - <<'PY'
-        import io
-        import os
-        import tarfile
-        import time
-        import urllib.error
-        import urllib.request
-
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        # Retry up to 5 times with backoff for transient 5xx errors
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-        else:
-            raise last_err
-
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        PY
-
-    - name: Show Python version
-      shell: sh
-      run: |
-        set -eu
-        python --version
-        python -m pip --version
-
-    - name: Install dependencies
-      shell: sh
-      run: |
-        set -eu
-        python -m pip install --upgrade pip -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
-        python -m pip install -r requirements.txt -r requirements-dev.txt -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
-
-    - name: Run unit tests
-      shell: sh
-      run: |
-        set -eu
-        PYTHONPATH="$PWD/apps/api:$PWD" python -m pytest tests/unit -q
-
-    - name: Run integration tests
-      shell: sh
-      run: |
-        set -eu
-        PYTHONPATH="$PWD/apps/api:$PWD" python -m pytest tests/integration -q --timeout=60 -x
-
-  lint:
-    runs-on: runtime-builder
-
-    steps:
-    - name: Checkout code
-      shell: sh
-      run: |
-        set -eu
-        python - <<'PY'
-        import io
-        import os
-        import tarfile
-        import time
-        import urllib.error
-        import urllib.request
-
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        # Retry up to 5 times with backoff for transient 5xx errors
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-        else:
-            raise last_err
-
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        PY
-
-    - name: Install dependencies
-      shell: sh
-      run: |
-        set -eu
-        python -m pip install --upgrade pip -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
-        python -m pip install -r requirements.txt -r requirements-dev.txt -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
-
-    - name: Run Black (check only)
-      shell: sh
-      run: |
-        set -eu
-        python -m black --check alembic apps packages tests scripts
-
-    - name: Run Flake8
-      shell: sh
-      run: |
-        set -eu
-        python -m flake8 apps packages tests --count --statistics

scripts/check_migration_safety.py

@@ -0,0 +1,267 @@
+#!/usr/bin/env python3
+"""
+数据库迁移破坏性变更安全检查
+
+只检查 Alembic 迁移文件的 upgrade 函数中是否包含破坏性操作：
+- DROP TABLE
+- ALTER TABLE ... DROP COLUMN
+- 列类型变更（可能导致数据丢失）
+- NOT NULL 约束新增（无默认值时）
+- RENAME TABLE / RENAME COLUMN
+
+忽略 downgrade 函数中的操作（那是回滚逻辑，正常的）。
+
+使用方式：
+  # 检查所有迁移（不推荐，会扫历史已执行的迁移）
+  python3 scripts/check_migration_safety.py
+
+  # 只检查与目标分支相比新增的迁移（推荐用于CI）
+  python3 scripts/check_migration_safety.py --diff-against origin/main
+
+  # 只检查指定版本之后的迁移
+  python3 scripts/check_migration_safety.py --since 030_xxx
+
+退出码：
+  0 - 安全 / 只有非破坏性变更
+  1 - 检测到高风险破坏性变更
+  2 - 检测到中风险变更，需人工确认
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import List, Tuple
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+ALEMBIC_VERSIONS_DIR = REPO_ROOT / "alembic" / "versions"
+
+# 高风险模式：直接导致数据丢失（只在 upgrade 中检查）
+HIGH_RISK_PATTERNS = [
+    (r"\bop\.drop_table\(", "op.drop_table() - 删除表，数据永久丢失"),
+    (r"\bop\.drop_column\(", "op.drop_column() - 删除列，数据永久丢失"),
+]
+
+# 中风险模式：可能导致数据丢失或兼容性问题
+MEDIUM_RISK_PATTERNS = [
+    (
+        r"op\.alter_column\([^)]*nullable\s*=\s*False",
+        "新增 NOT NULL 约束 - 旧数据可能为空导致迁移失败",
+    ),
+    (r"op\.alter_column\([^)]*type_\s*=", "列类型变更 - 可能导致数据截断或转换失败"),
+    (
+        r"\bop\.rename_table\(",
+        "op.rename_table() - 重命名表，可能导致依赖该表的代码报错",
+    ),
+    (
+        r"\bop\.rename_column\(",
+        "op.rename_column() - 重命名列，可能导致依赖该列的代码报错",
+    ),
+    (r"\bop\.drop_index\(", "op.drop_index() - 删除索引，可能影响查询性能"),
+    (r"\bop\.drop_constraint\(", "op.drop_constraint() - 删除约束，可能影响数据完整性"),
+]
+
+# 安全模式：这些是安全的新增操作
+SAFE_PATTERNS = [
+    (r"\bop\.create_table\(", "新建表"),
+    (r"\bop\.add_column\(", "新增列"),
+    (r"\bop\.create_index\(", "新建索引"),
+    (r"\bop\.create_unique_constraint\(", "新建唯一约束"),
+    (r"\bop\.create_foreign_key\(", "新建外键约束"),
+]
+
+
+def extract_upgrade_content(content: str) -> str:
+    """
+    从迁移文件中提取 upgrade 函数的内容。
+    只检查 upgrade 中的操作，忽略 downgrade。
+    """
+    upgrade_match = re.search(r"def upgrade\b[^:]*:", content)
+    if not upgrade_match:
+        return ""
+
+    upgrade_start = upgrade_match.end()
+
+    # 找到下一个顶层 def（通常是 def downgrade）作为结束位置
+    rest = content[upgrade_start:]
+    downgrade_match = re.search(r"\n\ndef\s+\w+\b", rest)
+    if downgrade_match:
+        upgrade_end = upgrade_start + downgrade_match.start()
+    else:
+        upgrade_end = len(content)
+
+    return content[upgrade_start:upgrade_end]
+
+
+def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
+    """
+    通过 git diff 对比目标分支/commit，找出 alembic/versions/ 下新增的迁移文件。
+    只包含新增文件（A状态），不包含修改或删除的文件。
+    """
+    try:
+        result = subprocess.run(
+            [
+                "git",
+                "diff",
+                "--name-only",
+                "--diff-filter=A",
+                diff_target,
+                "HEAD",
+                "--",
+                "alembic/versions/",
+            ],
+            cwd=str(REPO_ROOT),
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        files = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
+        return [REPO_ROOT / f for f in files]
+    except subprocess.CalledProcessError as e:
+        print(f"⚠️  git diff 失败（{diff_target}）：{e.stderr.strip()}")
+        print(f"   降级为检查所有迁移文件")
+        return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+
+
+def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]:
+    """
+    找出需要检查的迁移文件。
+    优先级：diff_against > since_revision > 全部
+    """
+    if diff_against:
+        return get_new_migrations_via_diff(diff_against)
+
+    all_migrations = sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+    if not since_revision:
+        return all_migrations
+
+    result = []
+    found = False
+    for m in all_migrations:
+        if since_revision in m.name or since_revision in m.stem:
+            found = True
+            continue
+        if found:
+            result.append(m)
+
+    return result if found else all_migrations
+
+
+def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]]:
+    """分析单个迁移文件 upgrade 部分的风险等级"""
+    content = file_path.read_text()
+    upgrade_content = extract_upgrade_content(content)
+
+    if not upgrade_content:
+        return [], [], [f"{file_path.name}: 未找到 upgrade 函数"]
+
+    high_risks = []
+    medium_risks = []
+    safes = []
+
+    for pattern, desc in HIGH_RISK_PATTERNS:
+        if re.search(pattern, upgrade_content):
+            high_risks.append(f"{file_path.name}: {desc}")
+
+    for pattern, desc in MEDIUM_RISK_PATTERNS:
+        if re.search(pattern, upgrade_content):
+            medium_risks.append(f"{file_path.name}: {desc}")
+
+    for pattern, desc in SAFE_PATTERNS:
+        if re.search(pattern, upgrade_content):
+            safes.append(f"{file_path.name}: {desc}")
+
+    return high_risks, medium_risks, safes
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument(
+        "--since",
+        default=os.getenv("MIGRATION_SINCE_REVISION"),
+        help="只检查指定版本之后的迁移（如：030_xxx），不传则检查所有迁移",
+    )
+    parser.add_argument(
+        "--diff-against",
+        default=os.getenv("MIGRATION_DIFF_AGAINST"),
+        help="对比指定分支/commit，只检查新增的迁移文件（推荐用于CI，如 origin/main）",
+    )
+    parser.add_argument(
+        "--warn-only",
+        action="store_true",
+        help="只警告不失败（用于非强制门禁场景）",
+    )
+    parser.add_argument(
+        "--allow-medium-risk",
+        action="store_true",
+        help="允许中风险变更（只拦截高风险）",
+    )
+    args = parser.parse_args()
+
+    migrations = find_new_migrations(args.since, args.diff_against)
+
+    if not migrations:
+        print("✅ 未找到需要检查的新增迁移文件，跳过")
+        return 0
+
+    print(f"🔍 正在检查 {len(migrations)} 个迁移文件的 upgrade 操作...")
+    if args.diff_against:
+        print(f"   （对比基准：{args.diff_against}，仅检查新增迁移）")
+    print()
+
+    all_high = []
+    all_medium = []
+    all_safe = []
+
+    for m in migrations:
+        high, medium, safe = analyze_migration(m)
+        all_high.extend(high)
+        all_medium.extend(medium)
+        all_safe.extend(safe)
+
+    if all_safe:
+        print("✅ 安全变更：")
+        for s in all_safe:
+            print(f"   - {s}")
+        print()
+
+    if all_medium:
+        print("⚠️  中风险变更（需人工确认）：")
+        for m_item in all_medium:
+            print(f"   - {m_item}")
+        print()
+
+    if all_high:
+        print("❌ 高风险破坏性变更（禁止自动部署）：")
+        for h in all_high:
+            print(f"   - {h}")
+        print()
+
+    print("=" * 60)
+    print(f"检查结果：{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险")
+    print()
+
+    if all_high:
+        print("❌ 检测到高风险破坏性变更，CI 检查失败！")
+        print("   如果确认这是预期操作，请在 MR/PR 中说明原因并获得审批。")
+        if args.warn_only:
+            return 0
+        return 1
+
+    if all_medium and not args.allow_medium_risk:
+        print("⚠️  检测到中风险变更，请人工确认后再部署。")
+        if args.warn_only:
+            return 0
+        print("（如需仅拦截高风险，可使用 --allow-medium-risk 参数）")
+        return 2
+
+    print("✅ 未检测到破坏性变更")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci_coverage_summary.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""解析 coverage.xml 并输出覆盖率汇总。"""
+
+import os
+import sys
+import xml.etree.ElementTree as ET
+
+THRESHOLD = int(os.environ.get("COVERAGE_THRESHOLD", 65))  # 行覆盖率门槛，百分比，可通过环境变量覆盖
+
+
+def main() -> int:
+    try:
+        tree = ET.parse("coverage.xml")
+    except FileNotFoundError:
+        print("coverage.xml 不存在，跳过汇总")
+        return 0
+
+    root = tree.getroot()
+    line_rate = float(root.get("line-rate", 0)) * 100
+    branch_rate = float(root.get("branch-rate", 0)) * 100
+    lines_covered = int(root.get("lines-covered", 0))
+    lines_valid = int(root.get("lines-valid", 0))
+
+    print(f"行覆盖率: {line_rate:.2f}% ({lines_covered}/{lines_valid})")
+    print(f"分支覆盖率: {branch_rate:.2f}%")
+    print(f"门槛: {THRESHOLD}%")
+    status = "PASS ✅" if line_rate >= THRESHOLD else "FAIL ❌"
+    print(f"状态: {status}")
+
+    return 0 if line_rate >= THRESHOLD else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci_notify_failure.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""发送 CI 失败通知到飞书/项目群 webhook。"""
+
+import json
+import os
+import sys
+import urllib.request
+
+
+def main() -> int:
+    webhook = os.environ.get("CI_NOTIFY_WEBHOOK", "")
+    if not webhook:
+        print("未配置 CI_NOTIFY_WEBHOOK，跳过通知")
+        print("如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK")
+        return 0
+
+    failed_job = os.environ.get("FAILED_JOB", "Unknown Job")
+    branch = os.environ.get("GITHUB_REF_NAME", "unknown")
+    commit = os.environ.get("GITHUB_SHA", "unknown")[:8]
+    actor = os.environ.get("GITHUB_ACTOR", "unknown")
+    run_id = os.environ.get("GITHUB_RUN_ID", "unknown")
+    repo = os.environ.get("GITHUB_REPOSITORY", "unknown")
+    run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
+
+    payload = {
+        "msg_type": "interactive",
+        "card": {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": "❌ CI 构建失败",
+                },
+                "status": "red",
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {
+                        "tag": "lark_md",
+                        "content": (
+                            f"**任务**: {failed_job}\n"
+                            f"**分支**: {branch}\n"
+                            f"**提交**: {commit}\n"
+                            f"**提交者**: {actor}\n"
+                            f"**Run ID**: {run_id}"
+                        ),
+                    },
+                },
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看失败日志"},
+                            "url": run_url,
+                            "type": "danger",
+                        }
+                    ],
+                },
+            ],
+        },
+    }
+
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        webhook,
+        data=data,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            resp.read()
+        print("通知已发送")
+    except Exception as e:
+        print(f"通知发送失败: {e}", file=sys.stderr)
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci_notify_success.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""发送 CI 成功通知到飞书/项目群 webhook。"""
+
+import json
+import os
+import sys
+import urllib.request
+
+
+def main() -> int:
+    webhook = os.environ.get("CI_NOTIFY_WEBHOOK", "")
+    if not webhook:
+        print("未配置 CI_NOTIFY_WEBHOOK，跳过成功通知")
+        print("如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK")
+        return 0
+
+    success_job = os.environ.get("SUCCESS_JOB", "Unknown Job")
+    branch = os.environ.get("GITHUB_REF_NAME", "unknown")
+    commit = os.environ.get("GITHUB_SHA", "unknown")[:8]
+    actor = os.environ.get("GITHUB_ACTOR", "unknown")
+    run_id = os.environ.get("GITHUB_RUN_ID", "unknown")
+    repo = os.environ.get("GITHUB_REPOSITORY", "unknown")
+    run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
+
+    payload = {
+        "msg_type": "interactive",
+        "card": {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": "✅ CI 构建成功",
+                },
+                "status": "green",
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {
+                        "tag": "lark_md",
+                        "content": (
+                            f"**任务**: {success_job}\n"
+                            f"**分支**: {branch}\n"
+                            f"**提交**: {commit}\n"
+                            f"**提交者**: {actor}\n"
+                            f"**Run ID**: {run_id}"
+                        ),
+                    },
+                },
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看构建详情"},
+                            "url": run_url,
+                            "type": "primary",
+                        }
+                    ],
+                },
+            ],
+        },
+    }
+
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        webhook,
+        data=data,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            resp.read()
+        print("成功通知已发送")
+    except Exception as e:
+        print(f"成功通知发送失败: {e}", file=sys.stderr)
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())