chendelian
3f1cde1f99
All checks were successful
技能自动化发布 / release (push) Successful in 6s
Co-authored-by: Cursor <cursoragent@cursor.com>
265 lines
8.9 KiB
Python
265 lines
8.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""平台键 / 展示名校验：引用已有平台 vs 登记新平台。"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any, Iterable, Optional
|
||
|
||
from util.platforms import (
|
||
PLATFORMS,
|
||
PlatformResolveResult,
|
||
_parse_aliases_json,
|
||
is_valid_platform_key_format,
|
||
)
|
||
|
||
_ASCII_TOKEN = re.compile(r"[A-Za-z0-9]")
|
||
|
||
|
||
def msg_empty() -> str:
|
||
return "平台名不能为空。请传平台键（如 douyin、deepseek），不要传空字符串。"
|
||
|
||
|
||
def msg_invalid_key(raw: str) -> str:
|
||
return (
|
||
f"平台键不合法：「{raw}」。\n"
|
||
"平台键必须是小写英文/数字/下划线或连字符，例如 douyin、deepseek。\n"
|
||
"「抖音」「DeepSeek」是展示名，不能当键。请改成已有键，或先登记：\n"
|
||
"platform ensure --key <键> --display-name <中文名>\n"
|
||
"查看已有键：platform list"
|
||
)
|
||
|
||
|
||
def msg_unrecognized(raw: str) -> str:
|
||
return (
|
||
f"无法识别平台：「{raw}」。\n"
|
||
"请传平台键（如 douyin），不要自创中文键。\n"
|
||
"先执行 platform list 核对；若是新平台，用 "
|
||
"platform ensure --key <键> --display-name <中文名> 登记后再调 add-web / ensure-web。"
|
||
)
|
||
|
||
|
||
def msg_not_registered(key: str) -> str:
|
||
return (
|
||
f"平台 `{key}` 尚未注册，已拒绝写入，以免产生不规范平台。\n"
|
||
f"若确认是新平台：platform ensure --key {key} --display-name <中文名> [--url …]\n"
|
||
"若其实是已有站：请改用 platform list 里的平台键。"
|
||
)
|
||
|
||
|
||
def msg_ambiguous(raw: str, keys: Iterable[str]) -> str:
|
||
listed = "、".join(f"`{k}`" for k in keys)
|
||
return f"平台「{raw}」对应多个键（{listed}），请改传明确的平台键。"
|
||
|
||
|
||
def msg_alias_conflict(alias: str, owner: str, key: str) -> str:
|
||
return (
|
||
f"别名「{alias}」已被平台 `{owner}` 占用，不能再登记到 `{key}`。"
|
||
"请换展示名/别名，或复用已有平台键。"
|
||
)
|
||
|
||
|
||
def msg_display_name_required() -> str:
|
||
return (
|
||
"登记平台必须同时提供 --display-name（给人看的名称，可以是中文）。\n"
|
||
"平台键与展示名要分开，例如 --key douyin --display-name 抖音。"
|
||
)
|
||
|
||
|
||
def msg_disabled(key: str) -> str:
|
||
return f"平台 {key} 已禁用。"
|
||
|
||
|
||
def _looks_like_attempted_key(raw: str) -> bool:
|
||
return bool(_ASCII_TOKEN.search(raw or "")) and not is_valid_platform_key_format(raw)
|
||
|
||
|
||
def _lookup_stored_key(conn, key_lower: str) -> Optional[tuple[str, int]]:
|
||
"""Return (platform_key, enabled) for an exact key, builtin or DB."""
|
||
needle = (key_lower or "").strip().lower()
|
||
if not needle:
|
||
return None
|
||
if conn is not None:
|
||
row = conn.execute(
|
||
"SELECT platform_key, enabled FROM platforms WHERE lower(platform_key) = ?",
|
||
(needle,),
|
||
).fetchone()
|
||
if row:
|
||
return str(row[0]), int(row[1] or 0)
|
||
if needle in PLATFORMS:
|
||
return needle, 1
|
||
return None
|
||
|
||
|
||
def _iter_enabled_name_map(conn) -> list[tuple[str, set[str]]]:
|
||
"""Each enabled platform with its display_name + aliases (not the key)."""
|
||
out: list[tuple[str, set[str]]] = []
|
||
seen: set[str] = set()
|
||
if conn is not None:
|
||
rows = conn.execute(
|
||
"SELECT platform_key, enabled, display_name, aliases_json FROM platforms"
|
||
).fetchall()
|
||
for platform_key, enabled, display_name, aliases_json in rows:
|
||
key = str(platform_key)
|
||
seen.add(key)
|
||
if not enabled:
|
||
continue
|
||
names: set[str] = set()
|
||
if display_name:
|
||
names.add(str(display_name).strip())
|
||
for alias in _parse_aliases_json(aliases_json):
|
||
names.add(alias)
|
||
out.append((key, {n for n in names if n}))
|
||
for key, spec in PLATFORMS.items():
|
||
if key in seen:
|
||
continue
|
||
names = {str(spec.get("display_name") or "").strip()}
|
||
for alias in spec.get("aliases") or []:
|
||
names.add(str(alias).strip())
|
||
out.append((key, {n for n in names if n}))
|
||
return out
|
||
|
||
|
||
def _alias_matches(raw: str, conn) -> list[str]:
|
||
needle = (raw or "").strip()
|
||
if not needle:
|
||
return []
|
||
lower = needle.lower()
|
||
matches: list[str] = []
|
||
for key, names in _iter_enabled_name_map(conn):
|
||
lowered = {n.lower() for n in names}
|
||
if needle in names or lower in lowered:
|
||
if key not in matches:
|
||
matches.append(key)
|
||
return matches
|
||
|
||
|
||
def _keys_claiming_label(label: str, conn) -> list[str]:
|
||
"""Keys that already use this string as key, display_name, or alias."""
|
||
text = (label or "").strip()
|
||
if not text:
|
||
return []
|
||
owners: list[str] = []
|
||
stored = _lookup_stored_key(conn, text.lower())
|
||
if stored and stored[0] not in owners:
|
||
owners.append(stored[0])
|
||
for key in _alias_matches(text, conn):
|
||
if key not in owners:
|
||
owners.append(key)
|
||
return owners
|
||
|
||
|
||
def resolve_platform_ref(input_name: str, conn=None) -> PlatformResolveResult:
|
||
"""引用已有平台：已有键优先，其次唯一展示名/别名。不创建平台行。"""
|
||
raw = (input_name or "").strip()
|
||
if not raw:
|
||
return PlatformResolveResult(error_code="INVALID_PLATFORM", message=msg_empty())
|
||
|
||
close_conn = False
|
||
if conn is None:
|
||
from db.connection import get_conn, init_db
|
||
|
||
init_db()
|
||
conn = get_conn()
|
||
close_conn = True
|
||
try:
|
||
if is_valid_platform_key_format(raw):
|
||
stored = _lookup_stored_key(conn, raw)
|
||
if stored:
|
||
key, enabled = stored
|
||
if not enabled:
|
||
return PlatformResolveResult(
|
||
error_code="PLATFORM_DISABLED",
|
||
message=msg_disabled(key),
|
||
)
|
||
return PlatformResolveResult(platform_key=key)
|
||
|
||
folded = raw.lower()
|
||
if is_valid_platform_key_format(folded):
|
||
stored = _lookup_stored_key(conn, folded)
|
||
if stored:
|
||
key, enabled = stored
|
||
if not enabled:
|
||
return PlatformResolveResult(
|
||
error_code="PLATFORM_DISABLED",
|
||
message=msg_disabled(key),
|
||
)
|
||
return PlatformResolveResult(platform_key=key)
|
||
|
||
matches = _alias_matches(raw, conn)
|
||
if len(matches) == 1:
|
||
return PlatformResolveResult(platform_key=matches[0])
|
||
if len(matches) > 1:
|
||
return PlatformResolveResult(
|
||
error_code="AMBIGUOUS_PLATFORM",
|
||
message=msg_ambiguous(raw, matches),
|
||
)
|
||
|
||
if is_valid_platform_key_format(raw):
|
||
return PlatformResolveResult(
|
||
error_code="PLATFORM_NOT_REGISTERED",
|
||
message=msg_not_registered(raw),
|
||
)
|
||
if _looks_like_attempted_key(raw) or is_valid_platform_key_format(folded):
|
||
return PlatformResolveResult(
|
||
error_code="INVALID_PLATFORM_KEY",
|
||
message=msg_invalid_key(raw),
|
||
)
|
||
return PlatformResolveResult(
|
||
error_code="INVALID_PLATFORM",
|
||
message=msg_unrecognized(raw),
|
||
)
|
||
finally:
|
||
if close_conn:
|
||
conn.close()
|
||
|
||
|
||
def validate_platform_registration(
|
||
key: str,
|
||
display_name: str,
|
||
aliases: Optional[Iterable[Any]] = None,
|
||
conn=None,
|
||
) -> PlatformResolveResult:
|
||
"""登记/更新平台：键必须已是规范小写；展示名必填；别名不得抢别人的。"""
|
||
raw_key = (key or "").strip()
|
||
if not raw_key:
|
||
return PlatformResolveResult(error_code="INVALID_PLATFORM", message=msg_empty())
|
||
if not is_valid_platform_key_format(raw_key):
|
||
return PlatformResolveResult(
|
||
error_code="INVALID_PLATFORM_KEY",
|
||
message=msg_invalid_key(raw_key),
|
||
)
|
||
name = (display_name or "").strip()
|
||
if not name:
|
||
return PlatformResolveResult(
|
||
error_code="INVALID_PLATFORM_DISPLAY_NAME",
|
||
message=msg_display_name_required(),
|
||
)
|
||
|
||
close_conn = False
|
||
if conn is None:
|
||
from db.connection import get_conn, init_db
|
||
|
||
init_db()
|
||
conn = get_conn()
|
||
close_conn = True
|
||
try:
|
||
claimed = [name]
|
||
if aliases:
|
||
for item in aliases:
|
||
text = str(item).strip()
|
||
if text:
|
||
claimed.append(text)
|
||
for label in claimed:
|
||
if label.lower() == raw_key:
|
||
continue
|
||
owners = [k for k in _keys_claiming_label(label, conn) if k != raw_key]
|
||
if owners:
|
||
return PlatformResolveResult(
|
||
error_code="PLATFORM_ALIAS_CONFLICT",
|
||
message=msg_alias_conflict(label, owners[0], raw_key),
|
||
)
|
||
return PlatformResolveResult(platform_key=raw_key)
|
||
finally:
|
||
if close_conn:
|
||
conn.close()
|